GlcoinCore/lnd
1
0
Fork 0
mirror of https://github.com/lightningnetwork/lnd.git synced 2026-08-17 13:06:14 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge pull request #10796 from ellemouton/for-each-node-cached-autopilot

Browse source 
autopilot+graph/db: drop unused address loading from cached node iteration
This commit is contained in:
Elle 2026-05-12 16:00:53 -07:00 • committed by GitHub
commit 1c5df12ffa
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 78 additions and 147 deletions
Download patch file Download diff file Expand all files Collapse all files

60
autopilot/graph.go
View file

@ -101,25 +101,23 @@ func (d *databaseChannelGraph) ForEachNode(ctx context.Context,
}, reset)
}
// ForEachNodesChannels iterates through all connected nodes, and for each node,
// all the channels that connect to it. The passed callback will be called with
// the context, the Node itself, and a slice of ChannelEdge that connect to the
// node.
// ForEachNodesChannels iterates through all connected nodes, and for each
// node, all the channels that connect to it. The passed callback will be
// called with the context, the node's pubkey, and a slice of ChannelEdge
// that connect to the node.
//
// NOTE: Part of the autopilot.ChannelGraph interface.
func (d *databaseChannelGraph) ForEachNodesChannels(ctx context.Context,
cb func(context.Context, Node, []*ChannelEdge) error,
cb func(context.Context, NodeID, []*ChannelEdge) error,
reset func()) error {
// The channel-scoring callers only need topology data here. Address
// filtering happens through ForEachNode before connecting to peers.
return d.db.ForEachNodeCached(
ctx, true, func(ctx context.Context, node route.Vertex,
addrs []net.Addr,
ctx, func(ctx context.Context, node route.Vertex,
chans map[uint64]*graphdb.DirectedChannel) error {
// We'll skip over any node that doesn't have any
// advertised addresses. As we won't be able to reach
// them to actually open any channels.
if len(addrs) == 0 {
if len(chans) == 0 {
return nil
}
@ -134,10 +132,7 @@ func (d *databaseChannelGraph) ForEachNodesChannels(ctx context.Context,
})
}
return cb(ctx, &dbNode{
pub: node,
addrs: addrs,
}, edges)
return cb(ctx, NodeID(node), edges)
}, reset,
)
}
@ -196,8 +191,8 @@ func (nc dbNodeCached) Addrs() []net.Addr {
func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context,
cb func(context.Context, Node) error, reset func()) error {
return dc.db.ForEachNodeCached(ctx, false, func(ctx context.Context,
n route.Vertex, _ []net.Addr,
return dc.db.ForEachNodeCached(ctx, func(ctx context.Context,
n route.Vertex,
channels map[uint64]*graphdb.DirectedChannel) error {
if len(channels) > 0 {
@ -213,20 +208,24 @@ func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context,
}, reset)
}
// ForEachNodesChannels iterates through all connected nodes, and for each node,
// all the channels that connect to it. The passed callback will be called with
// the context, the Node itself, and a slice of ChannelEdge that connect to the
// node.
// ForEachNodesChannels iterates through all connected nodes, and for each
// node, all the channels that connect to it. The passed callback will be
// called with the context, the node's pubkey, and a slice of ChannelEdge
// that connect to the node.
//
// NOTE: Part of the autopilot.ChannelGraph interface.
func (dc *databaseChannelGraphCached) ForEachNodesChannels(ctx context.Context,
cb func(context.Context, Node, []*ChannelEdge) error,
cb func(context.Context, NodeID, []*ChannelEdge) error,
reset func()) error {
return dc.db.ForEachNodeCached(ctx, false, func(ctx context.Context,
n route.Vertex, _ []net.Addr,
return dc.db.ForEachNodeCached(ctx, func(ctx context.Context,
n route.Vertex,
channels map[uint64]*graphdb.DirectedChannel) error {
if len(channels) == 0 {
return nil
}
edges := make([]*ChannelEdge, 0, len(channels))
for cid, channel := range channels {
edges = append(edges, &ChannelEdge{
@ -236,18 +235,7 @@ func (dc *databaseChannelGraphCached) ForEachNodesChannels(ctx context.Context,
})
}
if len(channels) > 0 {
node := dbNodeCached{
node: n,
channels: channels,
}
if err := cb(ctx, node, edges); err != nil {
return err
}
}
return nil
return cb(ctx, NodeID(n), edges)
}, reset)
}

7
autopilot/interface.go
View file

@ -84,10 +84,10 @@ type ChannelGraph interface {
// ForEachNodesChannels iterates through all connected nodes, and for
// each node, all the channels that connect to it. The passed callback
// will be called with the context, the Node itself, and a slice of
// will be called with the context, the node's pubkey, and a slice of
// ChannelEdge that connect to the node.
ForEachNodesChannels(ctx context.Context,
cb func(context.Context, Node, []*ChannelEdge) error,
cb func(context.Context, NodeID, []*ChannelEdge) error,
reset func()) error
}
@ -237,9 +237,8 @@ type GraphSource interface {
// channel graph cache if one is available. It is less consistent than
// ForEachNode since any further calls are made across multiple
// transactions.
ForEachNodeCached(ctx context.Context, withAddrs bool,
ForEachNodeCached(ctx context.Context,
cb func(ctx context.Context, node route.Vertex,
addrs []net.Addr,
chans map[uint64]*graphdb.DirectedChannel) error,
reset func()) error
}

13
autopilot/prefattach.go
View file

@ -90,7 +90,7 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph,
seenChans = make(map[uint64]struct{})
)
err := g.ForEachNodesChannels(
ctx, func(_ context.Context, node Node,
ctx, func(_ context.Context, node NodeID,
channels []*ChannelEdge) error {
for _, e := range channels {
@ -121,7 +121,7 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph,
var maxChans int
nodeChanNum := make(map[NodeID]int)
err = g.ForEachNodesChannels(
ctx, func(ctx context.Context, node Node,
ctx, func(ctx context.Context, node NodeID,
edges []*ChannelEdge) error {
var nodeChans int
@ -154,17 +154,16 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph,
// If this node is not among our nodes to score, we can
// return early.
nID := NodeID(node.PubKey())
if _, ok := nodes[nID]; !ok {
if _, ok := nodes[node]; !ok {
log.Tracef("Node %x not among nodes to score, "+
"ignoring", nID[:])
"ignoring", node[:])
return nil
}
// Otherwise we'll record the number of channels.
nodeChanNum[nID] = nodeChans
nodeChanNum[node] = nodeChans
log.Tracef("Counted %v channels for node %x", nodeChans,
nID[:])
node[:])
return nil
}, func() {

17
autopilot/prefattach_test.go
View file

@ -246,11 +246,11 @@ func TestPrefAttachmentSelectGreedyAllocation(t *testing.T) {
twoChans := false
nodes := make(map[NodeID]struct{})
err = graph.ForEachNodesChannels(
ctx, func(_ context.Context, node Node,
ctx, func(_ context.Context, node NodeID,
edges []*ChannelEdge) error {
numNodes++
nodes[node.PubKey()] = struct{}{}
nodes[node] = struct{}{}
numChans := 0
for range edges {
@ -619,14 +619,15 @@ func (m *memChannelGraph) ForEachNode(ctx context.Context,
return nil
}
// ForEachNodesChannels iterates through all connected nodes, and for each node,
// all the channels that connect to it. The passed callback will be called with
// the context, the Node itself, and a slice of ChannelEdge that connect to the
// node.
// ForEachNodesChannels iterates through all connected nodes, and for each
// node, all the channels that connect to it. The passed callback will be
// called with the context, the node's pubkey, and a slice of ChannelEdge
// that connect to the node.
//
// NOTE: Part of the autopilot.ChannelGraph interface.
func (m *memChannelGraph) ForEachNodesChannels(ctx context.Context,
cb func(context.Context, Node, []*ChannelEdge) error, _ func()) error {
cb func(context.Context, NodeID, []*ChannelEdge) error,
_ func()) error {
for _, node := range m.graph {
edges := make([]*ChannelEdge, 0, len(node.chans))
@ -634,7 +635,7 @@ func (m *memChannelGraph) ForEachNodesChannels(ctx context.Context,
edges = append(edges, &node.chans[i])
}
if err := cb(ctx, node, edges); err != nil {
if err := cb(ctx, NewNodeID(node.pub), edges); err != nil {
return err
}
}

15
autopilot/simple_graph.go
View file

@ -2,8 +2,6 @@ package autopilot
import (
"context"
"github.com/lightningnetwork/lnd/routing/route"
)
// diameterCutoff is used to discard nodes in the diameter calculation.
@ -35,12 +33,11 @@ func NewSimpleGraph(ctx context.Context, g ChannelGraph) (*SimpleGraph, error) {
// The returned index is then used to create a simplified adjacency list
// where each node is identified by its index instead of its pubkey, and
// also to create a mapping from node index to node pubkey.
getNodeIndex := func(node route.Vertex) int {
key := NodeID(node)
nodeIndex, ok := nodes[key]
getNodeIndex := func(node NodeID) int {
nodeIndex, ok := nodes[node]
if !ok {
nodes[key] = nextIndex
nodes[node] = nextIndex
nodeIndex = nextIndex
nextIndex++
}
@ -51,12 +48,12 @@ func NewSimpleGraph(ctx context.Context, g ChannelGraph) (*SimpleGraph, error) {
// Iterate over each node and each channel and update the adj and the
// node index.
err := g.ForEachNodesChannels(ctx, func(_ context.Context,
node Node, channels []*ChannelEdge) error {
node NodeID, channels []*ChannelEdge) error {
u := getNodeIndex(node.PubKey())
u := getNodeIndex(node)
for _, edge := range channels {
v := getNodeIndex(edge.Peer)
v := getNodeIndex(NodeID(edge.Peer))
adj[u] = append(adj[u], v)
}

7
docs/release-notes/release-notes-0.21.0.md
View file

@ -265,6 +265,13 @@
is fully populated. This new behaviour can be opted out of via the new
`--db.sync-graph-cache-load` option.
* Autopilot's graph-wide channel scoring traversal [no longer requests node
addresses](https://github.com/lightningnetwork/lnd/pull/10796) from the
graph backend, since the scoring code does not consume them. This removes
an unnecessary address batch-load on the SQL backend, and lets the kvdb
backend serve the traversal from the in-memory graph cache when it is
loaded.
* [Invoice pagination queries no longer use
`OFFSET`](https://github.com/lightningnetwork/lnd/pull/10700). The five
invoice filter queries previously used `LIMIT+OFFSET` for internal batching,

4
graph/db/benchmark_test.go
View file

@ -5,7 +5,6 @@ import (
"database/sql"
"errors"
"fmt"
"net"
"path"
"sync"
"testing"
@ -698,10 +697,9 @@ func BenchmarkGraphReadMethods(b *testing.B) {
fn: func(b testing.TB, store Store) {
//nolint:ll
err := store.ForEachNodeCached(
ctx, lnwire.GossipVersion1, false,
ctx, lnwire.GossipVersion1,
func(context.Context,
route.Vertex,
[]net.Addr,
map[uint64]*DirectedChannel) error {
// Increment the counter to

15
graph/db/graph.go
View file

@ -342,21 +342,21 @@ func (c *ChannelGraph) GraphSession(ctx context.Context,
//
// NOTE: The callback contents MUST not be modified.
func (c *ChannelGraph) ForEachNodeCached(ctx context.Context,
v lnwire.GossipVersion, withAddrs bool,
cb func(ctx context.Context, node route.Vertex, addrs []net.Addr,
v lnwire.GossipVersion,
cb func(ctx context.Context, node route.Vertex,
chans map[uint64]*DirectedChannel) error, reset func()) error {
if !withAddrs && c.cache != nil && c.cache.isLoaded() {
if c.cache != nil && c.cache.isLoaded() {
return c.cache.graphCache.ForEachNode(
func(node route.Vertex,
channels map[uint64]*DirectedChannel) error {
return cb(ctx, node, nil, channels)
return cb(ctx, node, channels)
},
)
}
return c.db.ForEachNodeCached(ctx, v, withAddrs, cb, reset)
return c.db.ForEachNodeCached(ctx, v, cb, reset)
}
// AddNode adds a vertex/node to the graph database. If the node is not
@ -919,12 +919,11 @@ func (c *VersionedGraph) ForEachNodeDirectedChannel(ctx context.Context,
// ForEachNodeCached iterates through all stored vertices/nodes in the graph,
// delegating to the embedded ChannelGraph.
func (c *VersionedGraph) ForEachNodeCached(ctx context.Context,
withAddrs bool, cb func(ctx context.Context, node route.Vertex,
addrs []net.Addr,
cb func(ctx context.Context, node route.Vertex,
chans map[uint64]*DirectedChannel) error,
reset func()) error {
return c.ChannelGraph.ForEachNodeCached(ctx, c.v, withAddrs, cb, reset)
return c.ChannelGraph.ForEachNodeCached(ctx, c.v, cb, reset)
}
// ForEachNode iterates through all stored vertices/nodes in the graph.

25
graph/db/graph_test.go
View file

@ -449,24 +449,6 @@ func testNodeInsertionAndDeletion(t *testing.T, v lnwire.GossipVersion) {
dbNode, err = graph.FetchNode(ctx, testPub)
require.NoError(t, err)
require.Equal(t, expAddrs, dbNode.Addresses)
// Also check that the withAddr param of ForEachNodeCached correctly
// returns the addresses we expect for this node.
err = graph.ForEachNodeCached(
ctx, true, func(ctx context.Context, node route.Vertex,
addrs []net.Addr,
chans map[uint64]*DirectedChannel) error {
if node != dbNode.PubKeyBytes {
return nil
}
require.Equal(t, expAddrs, addrs)
return nil
}, func() {},
)
require.NoError(t, err)
}
// testPartialNode tests that partial/shell nodes are correctly created when
@ -1799,8 +1781,8 @@ func TestGraphTraversal(t *testing.T) {
nodeIndex[node.PubKeyBytes] = struct{}{}
}
err := graph.ForEachNodeCached(ctx, lnwire.GossipVersion1, false,
func(_ context.Context, node route.Vertex, _ []net.Addr,
err := graph.ForEachNodeCached(ctx, lnwire.GossipVersion1,
func(_ context.Context, node route.Vertex,
chans map[uint64]*DirectedChannel) error {
if _, ok := nodeIndex[node]; !ok {
@ -6026,9 +6008,8 @@ func TestAsyncGraphCache(t *testing.T) {
// assert that we get the expected number of nodes and
// channels.
err := graph.ForEachNodeCached(
ctx, lnwire.GossipVersion1, false,
ctx, lnwire.GossipVersion1,
func(_ context.Context, node route.Vertex,
_ []net.Addr,
chans map[uint64]*DirectedChannel) error {
numNodes++

8
graph/db/interfaces.go
View file

@ -81,17 +81,11 @@ type Store interface { //nolint:interfacebloat
*models.ChannelEdgePolicy) error, reset func()) error
// ForEachNodeCached is similar to forEachNode, but it returns
// DirectedChannel data to the call-back. If withAddrs is true, then
// the call-back will also be provided with the addresses associated
// with the node. The address retrieval will likely result in an
// additional round-trip to the database, so it should only be used if
// the addresses are actually needed.
// DirectedChannel data to the call-back.
//
// NOTE: The callback contents MUST not be modified.
ForEachNodeCached(ctx context.Context, v lnwire.GossipVersion,
withAddrs bool,
cb func(ctx context.Context, node route.Vertex,
addrs []net.Addr,
chans map[uint64]*DirectedChannel) error,
reset func()) error

11
graph/db/kv_store.go
View file

@ -703,8 +703,8 @@ func (c *KVStore) FetchNodeFeatures(_ context.Context, v lnwire.GossipVersion,
//
// NOTE: The callback contents MUST not be modified.
func (c *KVStore) ForEachNodeCached(ctx context.Context,
v lnwire.GossipVersion, withAddrs bool,
cb func(ctx context.Context, node route.Vertex, addrs []net.Addr,
v lnwire.GossipVersion,
cb func(ctx context.Context, node route.Vertex,
chans map[uint64]*DirectedChannel) error, reset func()) error {
if v != lnwire.GossipVersion1 {
@ -769,12 +769,7 @@ func (c *KVStore) ForEachNodeCached(ctx context.Context,
return err
}
var addrs []net.Addr
if withAddrs {
addrs = node.Addresses
}
return cb(ctx, node.PubKeyBytes, addrs, channels)
return cb(ctx, node.PubKeyBytes, channels)
}, reset)
}

34
graph/db/sql_store.go
View file

@ -1719,20 +1719,16 @@ func (s *SQLStore) chanUpdatesInHorizonV2(ctx context.Context,
}
// ForEachNodeCached is similar to forEachNode, but it returns DirectedChannel
// data to the call-back. If withAddrs is true, then the call-back will also be
// provided with the addresses associated with the node. The address retrieval
// result in an additional round-trip to the database, so it should only be used
// if the addresses are actually needed.
// data to the call-back.
//
// NOTE: part of the Store interface.
func (s *SQLStore) ForEachNodeCached(ctx context.Context,
v lnwire.GossipVersion, withAddrs bool,
cb func(ctx context.Context, node route.Vertex, addrs []net.Addr,
v lnwire.GossipVersion,
cb func(ctx context.Context, node route.Vertex,
chans map[uint64]*DirectedChannel) error, reset func()) error {
type nodeCachedBatchData struct {
features map[int64][]int
addrs map[int64][]nodeAddress
chanBatchData *batchChannelData
chanMap map[int64][]sqlc.ListChannelsForNodeIDsRow
}
@ -1765,19 +1761,6 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context,
"node features: %w", err)
}
// Maybe fetch the node's addresses if requested.
var nodeAddrs map[int64][]nodeAddress
if withAddrs {
nodeAddrs, err = batchLoadNodeAddressesHelper(
ctx, s.cfg.QueryCfg, db, nodeIDs,
)
if err != nil {
return nil, fmt.Errorf("unable to "+
"batch load node "+
"addresses: %w", err)
}
}
// Batch load ALL unique channels for ALL nodes in this
// page.
allChannels, err := db.ListChannelsForNodeIDs(
@ -1866,7 +1849,6 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context,
return &nodeCachedBatchData{
features: nodeFeatures,
addrs: nodeAddrs,
chanBatchData: channelBatchData,
chanMap: nodeChannelMap,
}, nil
@ -1910,15 +1892,7 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context,
channels[directedChan.ChannelID] = directedChan
}
addrs, err := buildNodeAddresses(
batchData.addrs[nodeData.ID],
)
if err != nil {
return fmt.Errorf("unable to build node "+
"addresses: %w", err)
}
return cb(ctx, nodePub, addrs, channels)
return cb(ctx, nodePub, channels)
}
return sqldb.ExecuteCollectAndBatchWithSharedDataQuery(

5
itest/lnd_graph_migration_test.go
View file

@ -3,7 +3,6 @@ package itest
import (
"context"
"database/sql"
"net"
graphdb "github.com/lightningnetwork/lnd/graph/db"
"github.com/lightningnetwork/lnd/lntest"
@ -66,9 +65,9 @@ func testGraphMigration(ht *lntest.HarnessTest) {
numNodes int
edges = make(map[uint64]bool)
)
err := db.ForEachNodeCached(ctx, lnwire.GossipVersion1, false,
err := db.ForEachNodeCached(ctx, lnwire.GossipVersion1,
func(_ context.Context,
_ route.Vertex, _ []net.Addr,
_ route.Vertex,
chans map[uint64]*graphdb.DirectedChannel,
) error {

4
rpcserver.go
View file

@ -7439,8 +7439,8 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context,
// network, tallying up the total number of nodes, and also gathering
// each node so we can measure the graph diameter and degree stats
// below.
err := graph.ForEachNodeCached(ctx, false, func(ctx context.Context,
node route.Vertex, _ []net.Addr,
err := graph.ForEachNodeCached(ctx, func(ctx context.Context,
node route.Vertex,
edges map[uint64]*graphdb.DirectedChannel) error {
// Increment the total number of nodes with each iteration.