Merge pull request #9923 from ellemouton/graphCache

graph/db: only fetch required info for graph cache population

docs/release-notes/release-notes-0.20.0.md

@@ -53,6 +53,9 @@ circuit. The indices are only available for forwarding events saved after v0.20.
     byte blobs at the end of gossip messages are valid TLV streams.
   * Various [preparations](https://github.com/lightningnetwork/lnd/pull/9692) 
     of the graph code before the SQL implementation is added.
+  * Only [fetch required 
+    fields](https://github.com/lightningnetwork/lnd/pull/9923) during graph 
+    cache population. 
   * Add graph schemas, queries and CRUD:
     * [1](https://github.com/lightningnetwork/lnd/pull/9866)
 

graph/db/graph.go

@@ -174,13 +174,14 @@ func (c *ChannelGraph) populateCache() error {
 		return err
 	}
 
-	err = c.V1Store.ForEachChannel(func(info *models.ChannelEdgeInfo,
+	err = c.V1Store.ForEachChannelCacheable(
-		policy1, policy2 *models.ChannelEdgePolicy) error {
+		func(info *models.CachedEdgeInfo,
+			policy1, policy2 *models.CachedEdgePolicy) error {
 
-		c.graphCache.AddChannel(info, policy1, policy2)
+			c.graphCache.AddChannel(info, policy1, policy2)
 
-		return nil
+			return nil
-	})
+		})
 	if err != nil {
 		return err
 	}
@@ -312,7 +313,7 @@ func (c *ChannelGraph) AddChannelEdge(edge *models.ChannelEdgeInfo,
 	}
 
 	if c.graphCache != nil {
-		c.graphCache.AddChannel(edge, nil, nil)
+		c.graphCache.AddChannel(models.NewCachedEdge(edge), nil, nil)
 	}
 
 	select {
@@ -347,7 +348,11 @@ func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error {
 
 		info := infos[0]
 
-		c.graphCache.AddChannel(info.Info, info.Policy1, info.Policy2)
+		c.graphCache.AddChannel(
+			models.NewCachedEdge(info.Info),
+			models.NewCachedPolicy(info.Policy1),
+			models.NewCachedPolicy(info.Policy2),
+		)
 	}
 
 	return nil
@@ -566,12 +571,9 @@ func (c *ChannelGraph) UpdateEdgePolicy(edge *models.ChannelEdgePolicy,
 	}
 
 	if c.graphCache != nil {
-		var isUpdate1 bool
+		c.graphCache.UpdatePolicy(
-		if edge.ChannelFlags&lnwire.ChanUpdateDirection == 0 {
+			models.NewCachedPolicy(edge), from, to,
-			isUpdate1 = true
+		)
-		}
-
-		c.graphCache.UpdatePolicy(edge, from, to, isUpdate1)
 	}
 
 	select {

graph/db/graph_cache.go

@@ -114,8 +114,8 @@ func (c *GraphCache) AddNodeFeatures(node route.Vertex,
 // and policy 2 does not matter, the directionality is extracted from the info
 // and policy flags automatically. The policy will be set as the outgoing policy
 // on one node and the incoming policy on the peer's side.
-func (c *GraphCache) AddChannel(info *models.ChannelEdgeInfo,
+func (c *GraphCache) AddChannel(info *models.CachedEdgeInfo,
-	policy1 *models.ChannelEdgePolicy, policy2 *models.ChannelEdgePolicy) {
+	policy1, policy2 *models.CachedEdgePolicy) {
 
 	if info == nil {
 		return
@@ -147,19 +147,17 @@ func (c *GraphCache) AddChannel(info *models.ChannelEdgeInfo,
 	// of node 2 then we have the policy 1 as seen from node 1.
 	if policy1 != nil {
 		fromNode, toNode := info.NodeKey1Bytes, info.NodeKey2Bytes
-		if policy1.ToNode != info.NodeKey2Bytes {
+		if !policy1.IsNode1() {
 			fromNode, toNode = toNode, fromNode
 		}
-		isEdge1 := policy1.ChannelFlags&lnwire.ChanUpdateDirection == 0
+		c.UpdatePolicy(policy1, fromNode, toNode)
-		c.UpdatePolicy(policy1, fromNode, toNode, isEdge1)
 	}
 	if policy2 != nil {
 		fromNode, toNode := info.NodeKey2Bytes, info.NodeKey1Bytes
-		if policy2.ToNode != info.NodeKey1Bytes {
+		if policy2.IsNode1() {
 			fromNode, toNode = toNode, fromNode
 		}
-		isEdge1 := policy2.ChannelFlags&lnwire.ChanUpdateDirection == 0
+		c.UpdatePolicy(policy2, fromNode, toNode)
-		c.UpdatePolicy(policy2, fromNode, toNode, isEdge1)
 	}
 }
 
@@ -177,8 +175,8 @@ func (c *GraphCache) updateOrAddEdge(node route.Vertex, edge *DirectedChannel) {
 // of the from and to node is not strictly important. But we assume that a
 // channel edge was added beforehand so that the directed channel struct already
 // exists in the cache.
-func (c *GraphCache) UpdatePolicy(policy *models.ChannelEdgePolicy, fromNode,
+func (c *GraphCache) UpdatePolicy(policy *models.CachedEdgePolicy, fromNode,
-	toNode route.Vertex, edge1 bool) {
+	toNode route.Vertex) {
 
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
@@ -203,7 +201,7 @@ func (c *GraphCache) UpdatePolicy(policy *models.ChannelEdgePolicy, fromNode,
 		switch {
 		// This is node 1, and it is edge 1, so this is the outgoing
 		// policy for node 1.
-		case channel.IsNode1 && edge1:
+		case channel.IsNode1 && policy.IsNode1():
 			channel.OutPolicySet = true
 			policy.InboundFee.WhenSome(func(fee lnwire.Fee) {
 				channel.InboundFee = fee
@@ -211,7 +209,7 @@ func (c *GraphCache) UpdatePolicy(policy *models.ChannelEdgePolicy, fromNode,
 
 		// This is node 2, and it is edge 2, so this is the outgoing
 		// policy for node 2.
-		case !channel.IsNode1 && !edge1:
+		case !channel.IsNode1 && !policy.IsNode1():
 			channel.OutPolicySet = true
 			policy.InboundFee.WhenSome(func(fee lnwire.Fee) {
 				channel.InboundFee = fee
@@ -220,7 +218,7 @@ func (c *GraphCache) UpdatePolicy(policy *models.ChannelEdgePolicy, fromNode,
 		// The other two cases left mean it's the inbound policy for the
 		// node.
 		default:
-			channel.InPolicy = models.NewCachedPolicy(policy)
+			channel.InPolicy = policy
 		}
 	}
 
@@ -264,34 +262,6 @@ func (c *GraphCache) removeChannelIfFound(node route.Vertex, chanID uint64) {
 	delete(c.nodeChannels[node], chanID)
 }
 
-// UpdateChannel updates the channel edge information for a specific edge. We
-// expect the edge to already exist and be known. If it does not yet exist, this
-// call is a no-op.
-func (c *GraphCache) UpdateChannel(info *models.ChannelEdgeInfo) {
-	c.mtx.Lock()
-	defer c.mtx.Unlock()
-
-	if len(c.nodeChannels[info.NodeKey1Bytes]) == 0 ||
-		len(c.nodeChannels[info.NodeKey2Bytes]) == 0 {
-
-		return
-	}
-
-	channel, ok := c.nodeChannels[info.NodeKey1Bytes][info.ChannelID]
-	if ok {
-		// We only expect to be called when the channel is already
-		// known.
-		channel.Capacity = info.Capacity
-		channel.OtherNode = info.NodeKey2Bytes
-	}
-
-	channel, ok = c.nodeChannels[info.NodeKey2Bytes][info.ChannelID]
-	if ok {
-		channel.Capacity = info.Capacity
-		channel.OtherNode = info.NodeKey1Bytes
-	}
-}
-
 // getChannels returns a copy of the passed node's channels or nil if there
 // isn't any.
 func (c *GraphCache) getChannels(node route.Vertex) []*DirectedChannel {

graph/db/graph_cache_test.go

@@ -43,24 +43,25 @@ func TestGraphCacheAddNode(t *testing.T) {
 			FeeRate: 20,
 		}
 
-		outPolicy1 := &models.ChannelEdgePolicy{
+		outPolicy1 := &models.CachedEdgePolicy{
 			ChannelID:    1000,
 			ChannelFlags: lnwire.ChanUpdateChanFlags(channelFlagA),
-			ToNode:       nodeB,
+			ToNodePubKey: func() route.Vertex {
+				return nodeB
+			},
 			// Define an inbound fee.
 			InboundFee: fn.Some(inboundFee),
-			ExtraOpaqueData: []byte{
-				253, 217, 3, 8, 0, 0, 0, 10, 0, 0, 0, 20,
-			},
 		}
-		inPolicy1 := &models.ChannelEdgePolicy{
+		inPolicy1 := &models.CachedEdgePolicy{
 			ChannelID:    1000,
 			ChannelFlags: lnwire.ChanUpdateChanFlags(channelFlagB),
-			ToNode:       nodeA,
+			ToNodePubKey: func() route.Vertex {
+				return nodeA
+			},
 		}
 		cache := NewGraphCache(10)
 		cache.AddNodeFeatures(nodeA, lnwire.EmptyFeatureVector())
-		cache.AddChannel(&models.ChannelEdgeInfo{
+		cache.AddChannel(&models.CachedEdgeInfo{
 			ChannelID: 1000,
 			// Those are direction independent!
 			NodeKey1Bytes: pubKey1,
@@ -120,7 +121,7 @@ func TestGraphCacheAddNode(t *testing.T) {
 	runTest(pubKey2, pubKey1)
 }
 
-func assertCachedPolicyEqual(t *testing.T, original *models.ChannelEdgePolicy,
+func assertCachedPolicyEqual(t *testing.T, original,
 	cached *models.CachedEdgePolicy) {
 
 	require.Equal(t, original.ChannelID, cached.ChannelID)
@@ -134,7 +135,7 @@ func assertCachedPolicyEqual(t *testing.T, original *models.ChannelEdgePolicy,
 		t, original.FeeProportionalMillionths,
 		cached.FeeProportionalMillionths,
 	)
-	require.Equal(
+	if original.ToNodePubKey != nil {
-		t, route.Vertex(original.ToNode), cached.ToNodePubKey(),
+		require.Equal(t, original.ToNodePubKey(), cached.ToNodePubKey())
-	)
+	}
 }

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

graph/db/models/cached_edge_info.go

@@ -0,0 +1,33 @@
+package models
+
+import "github.com/btcsuite/btcd/btcutil"
+
+// CachedEdgeInfo is a struct that only caches the information of a
+// ChannelEdgeInfo that we actually use for pathfinding and therefore need to
+// store in the cache.
+type CachedEdgeInfo struct {
+	// ChannelID is the unique channel ID for the channel. The first 3
+	// bytes are the block height, the next 3 the index within the block,
+	// and the last 2 bytes are the output index for the channel.
+	ChannelID uint64
+
+	// NodeKey1Bytes is the raw public key of the first node.
+	NodeKey1Bytes [33]byte
+
+	// NodeKey2Bytes is the raw public key of the second node.
+	NodeKey2Bytes [33]byte
+
+	// Capacity is the total capacity of the channel, this is determined by
+	// the value output in the outpoint that created this channel.
+	Capacity btcutil.Amount
+}
+
+// NewCachedEdge creates a new CachedEdgeInfo from the provided ChannelEdgeInfo.
+func NewCachedEdge(edgeInfo *ChannelEdgeInfo) *CachedEdgeInfo {
+	return &CachedEdgeInfo{
+		ChannelID:     edgeInfo.ChannelID,
+		NodeKey1Bytes: edgeInfo.NodeKey1Bytes,
+		NodeKey2Bytes: edgeInfo.NodeKey2Bytes,
+		Capacity:      edgeInfo.Capacity,
+	}
+}

graph/db/models/cached_edge_policy.go

@@ -1,6 +1,7 @@
 package models
 
 import (
+	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/routing/route"
 )
@@ -48,6 +49,9 @@ type CachedEdgePolicy struct {
 	// HTLCs for each millionth of a satoshi forwarded.
 	FeeProportionalMillionths lnwire.MilliSatoshi
 
+	// InboundFee is the fee that the node will charge for incoming HTLCs.
+	InboundFee fn.Option[lnwire.Fee]
+
 	// ToNodePubKey is a function that returns the to node of a policy.
 	// Since we only ever store the inbound policy, this is always the node
 	// that we query the channels for in ForEachChannel(). Therefore, we can
@@ -71,6 +75,18 @@ func (c *CachedEdgePolicy) ComputeFee(
 	return c.FeeBaseMSat + (amt*c.FeeProportionalMillionths)/feeRateParts
 }
 
+// IsDisabled returns true if the channel is disabled in the direction from  the
+// advertising node.
+func (c *CachedEdgePolicy) IsDisabled() bool {
+	return c.ChannelFlags&lnwire.ChanUpdateDisabled != 0
+}
+
+// IsNode1 returns true if this policy was announced by the channel's node_1
+// node.
+func (c *CachedEdgePolicy) IsNode1() bool {
+	return c.ChannelFlags&lnwire.ChanUpdateDirection == 0
+}
+
 // NewCachedPolicy turns a full policy into a minimal one that can be cached.
 func NewCachedPolicy(policy *ChannelEdgePolicy) *CachedEdgePolicy {
 	return &CachedEdgePolicy{
@@ -82,5 +98,6 @@ func NewCachedPolicy(policy *ChannelEdgePolicy) *CachedEdgePolicy {
 		MaxHTLC:                   policy.MaxHTLC,
 		FeeBaseMSat:               policy.FeeBaseMSat,
 		FeeProportionalMillionths: policy.FeeProportionalMillionths,
+		InboundFee:                policy.InboundFee,
 	}
 }

routing/unified_edges.go

@@ -360,9 +360,7 @@ func (u *edgeUnifier) getEdgeNetwork(netAmtReceived lnwire.MilliSatoshi,
 		}
 
 		// For network channels, skip the disabled ones.
-		edgeFlags := edge.policy.ChannelFlags
+		if edge.policy.IsDisabled() {
-		isDisabled := edgeFlags&lnwire.ChanUpdateDisabled != 0
-		if isDisabled {
 			log.Debugf("Skipped edge %v due to it being disabled",
 				edge.policy.ChannelID)
 			continue