diff --git a/docs/release-notes/release-notes-0.20.0.md b/docs/release-notes/release-notes-0.20.0.md index 451537870..476bca1f9 100644 --- a/docs/release-notes/release-notes-0.20.0.md +++ b/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) diff --git a/graph/db/graph.go b/graph/db/graph.go index cf3b3ab70..a68368e3d 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -174,13 +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(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 - if edge.ChannelFlags&lnwire.ChanUpdateDirection == 0 { - isUpdate1 = true - } - - c.graphCache.UpdatePolicy(edge, from, to, isUpdate1) + c.graphCache.UpdatePolicy( + models.NewCachedPolicy(edge), from, to, + ) } select { diff --git a/graph/db/graph_cache.go b/graph/db/graph_cache.go index 141f610b2..a691361a2 100644 --- a/graph/db/graph_cache.go +++ b/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, - policy1 *models.ChannelEdgePolicy, policy2 *models.ChannelEdgePolicy) { +func (c *GraphCache) AddChannel(info *models.CachedEdgeInfo, + 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, isEdge1) + c.UpdatePolicy(policy1, fromNode, toNode) } 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, isEdge1) + c.UpdatePolicy(policy2, fromNode, toNode) } } @@ -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, - toNode route.Vertex, edge1 bool) { +func (c *GraphCache) UpdatePolicy(policy *models.CachedEdgePolicy, fromNode, + 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 { diff --git a/graph/db/graph_cache_test.go b/graph/db/graph_cache_test.go index 96318366d..43c35862e 100644 --- a/graph/db/graph_cache_test.go +++ b/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( - t, route.Vertex(original.ToNode), cached.ToNodePubKey(), - ) + if original.ToNodePubKey != nil { + require.Equal(t, original.ToNodePubKey(), cached.ToNodePubKey()) + } } 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 diff --git a/graph/db/models/cached_edge_info.go b/graph/db/models/cached_edge_info.go new file mode 100644 index 000000000..eab12fbdd --- /dev/null +++ b/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, + } +} diff --git a/graph/db/models/cached_edge_policy.go b/graph/db/models/cached_edge_policy.go index b770ec1fb..40b0d9212 100644 --- a/graph/db/models/cached_edge_policy.go +++ b/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, } } diff --git a/routing/unified_edges.go b/routing/unified_edges.go index a18547485..9e28a53c0 100644 --- a/routing/unified_edges.go +++ b/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 - isDisabled := edgeFlags&lnwire.ChanUpdateDisabled != 0 - if isDisabled { + if edge.policy.IsDisabled() { log.Debugf("Skipped edge %v due to it being disabled", edge.policy.ChannelID) continue