graph/db: introduce ForEachChannelCacheable

Implement ForEachChannelCacheable which is like ForEachChannel but its
call-back takes the cached versions of channel info & policies. This is
then used during graph cache population. This will be useful once the
SQL implementation is added so that we can reduce the number of DB trips
on cache population.

Author: ellemouton

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
 	}

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.

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