diff --git a/graph/db/graph.go b/graph/db/graph.go index 7378dcef8..a68368e3d 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -174,17 +174,14 @@ func (c *ChannelGraph) populateCache() error { return err } - err = c.V1Store.ForEachChannel(func(info *models.ChannelEdgeInfo, - policy1, policy2 *models.ChannelEdgePolicy) error { + err = c.V1Store.ForEachChannelCacheable( + func(info *models.CachedEdgeInfo, + policy1, policy2 *models.CachedEdgePolicy) error { - c.graphCache.AddChannel( - models.NewCachedEdge(info), - models.NewCachedPolicy(policy1), - models.NewCachedPolicy(policy2), - ) + c.graphCache.AddChannel(info, policy1, policy2) - return nil - }) + return nil + }) if err != nil { return err } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 25d7257be..0f7d499fe 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -159,6 +159,22 @@ type V1Store interface { //nolint:interfacebloat *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error + // ForEachChannelCacheable iterates through all the channel edges stored + // within the graph and invokes the passed callback for each edge. The + // callback takes two edges as since this is a directed graph, both the + // in/out edges are visited. If the callback returns an error, then the + // transaction is aborted and the iteration stops early. + // + // NOTE: If an edge can't be found, or wasn't advertised, then a nil + // pointer for that particular channel edge routing policy will be + // passed into the callback. + // + // NOTE: this method is like ForEachChannel but fetches only the data + // required for the graph cache. + ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, + *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error) error + // DisabledChannelIDs returns the channel ids of disabled channels. // A channel is disabled when two of the associated ChanelEdgePolicies // have their disabled bit on. diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 51a1d8483..305529bb6 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -453,6 +453,74 @@ func (c *KVStore) ForEachChannel(cb func(*models.ChannelEdgeInfo, }, func() {}) } +// ForEachChannelCacheable iterates through all the channel edges stored within +// the graph and invokes the passed callback for each edge. The callback takes +// two edges as since this is a directed graph, both the in/out edges are +// visited. If the callback returns an error, then the transaction is aborted +// and the iteration stops early. +// +// NOTE: If an edge can't be found, or wasn't advertised, then a nil pointer +// for that particular channel edge routing policy will be passed into the +// callback. +// +// NOTE: this method is like ForEachChannel but fetches only the data required +// for the graph cache. +func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, + *models.CachedEdgePolicy, *models.CachedEdgePolicy) error) error { + + return c.db.View(func(tx kvdb.RTx) error { + edges := tx.ReadBucket(edgeBucket) + if edges == nil { + return ErrGraphNoEdgesFound + } + + // First, load all edges in memory indexed by node and channel + // id. + channelMap, err := c.getChannelMap(edges) + if err != nil { + return err + } + + edgeIndex := edges.NestedReadBucket(edgeIndexBucket) + if edgeIndex == nil { + return ErrGraphNoEdgesFound + } + + // Load edge index, recombine each channel with the policies + // loaded above and invoke the callback. + return kvdb.ForAll( + edgeIndex, func(k, edgeInfoBytes []byte) error { + var chanID [8]byte + copy(chanID[:], k) + + edgeInfoReader := bytes.NewReader(edgeInfoBytes) + info, err := deserializeChanEdgeInfo( + edgeInfoReader, + ) + if err != nil { + return err + } + + policy1 := channelMap[channelMapKey{ + nodeKey: info.NodeKey1Bytes, + chanID: chanID, + }] + + policy2 := channelMap[channelMapKey{ + nodeKey: info.NodeKey2Bytes, + chanID: chanID, + }] + + return cb( + models.NewCachedEdge(&info), + models.NewCachedPolicy(policy1), + models.NewCachedPolicy(policy2), + ) + }, + ) + }, func() {}) +} + // forEachNodeDirectedChannel iterates through all channels of a given node, // executing the passed callback on the directed edge representing the channel // and its incoming policy. If the callback returns an error, then the iteration