From 16126d77031eac0dabbe676e791bebeff3b14222 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 10:23:59 +0200 Subject: [PATCH] graph/db: expose reset in ForEachSourceNodeChannel --- graph/db/graph_test.go | 2 +- graph/db/interfaces.go | 3 ++- graph/db/kv_store.go | 4 +--- graph/db/sql_store.go | 4 +--- server.go | 18 ++++++++++++++---- 5 files changed, 19 insertions(+), 12 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 662eba77f..50d288cdc 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1331,7 +1331,7 @@ func TestForEachSourceNodeChannel(t *testing.T) { delete(expectedSrcChans, chanPoint) return nil - }) + }, func() {}) require.NoError(t, err) require.Empty(t, expectedSrcChans) } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 388ba3ca6..652e7bd34 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -72,7 +72,8 @@ type V1Store interface { //nolint:interfacebloat // the channel and the channel peer's node information. ForEachSourceNodeChannel(ctx context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.LightningNode) error) error + otherNode *models.LightningNode) error, + reset func()) error // ForEachNodeChannel iterates through all channels of the given node, // executing the passed callback with an edge info structure and the diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index cd56bb41e..2a52848bf 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -3291,9 +3291,7 @@ func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex, // peer's node information. func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.LightningNode) error) error { - - reset := func() {} + otherNode *models.LightningNode) error, reset func()) error { return kvdb.View(c.db, func(tx kvdb.RTx) error { nodes := tx.ReadBucket(nodeBucket) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index b241a3fa4..5a8c75a4f 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -727,9 +727,7 @@ func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy, // NOTE: part of the V1Store interface. func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.LightningNode) error) error { - - reset := func() {} + otherNode *models.LightningNode) error, reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { nodeID, nodePub, err := s.getSourceNode(ctx, db, ProtocolV1) diff --git a/server.go b/server.go index 4ad5c09e3..2c42dab6c 100644 --- a/server.go +++ b/server.go @@ -1254,7 +1254,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, DefaultRoutingPolicy: cc.RoutingPolicy, ForAllOutgoingChannels: func(ctx context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy) error) error { return s.graphDB.ForEachNodeChannel(ctx, selfVertex, func(c *models.ChannelEdgeInfo, @@ -3599,7 +3599,7 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { // nodeAddrsMap stores the combination of node public keys and addresses // that we'll attempt to reconnect to. PubKey strings are used as keys // since other PubKey forms can't be compared. - nodeAddrsMap := map[string]*nodeAddresses{} + nodeAddrsMap := make(map[string]*nodeAddresses) // Iterate through the list of LinkNodes to find addresses we should // attempt to connect to based on our set of previous connections. Set @@ -3623,6 +3623,7 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { // that have been added via NodeAnnouncement messages. // TODO(roasbeef): instead iterate over link nodes and query graph for // each of the nodes. + graphAddrs := make(map[string]*nodeAddresses) forEachSrcNodeChan := func(chanPoint wire.OutPoint, havePolicy bool, channelPeer *models.LightningNode) error { @@ -3688,10 +3689,14 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { return err } - nodeAddrsMap[pubStr] = n + graphAddrs[pubStr] = n return nil } - err = s.graphDB.ForEachSourceNodeChannel(ctx, forEachSrcNodeChan) + err = s.graphDB.ForEachSourceNodeChannel( + ctx, forEachSrcNodeChan, func() { + clear(graphAddrs) + }, + ) if err != nil { srvrLog.Errorf("Failed to iterate over source node channels: "+ "%v", err) @@ -3703,6 +3708,11 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { } } + // Combine the addresses from the link nodes and the channel graph. + for pubStr, nodeAddr := range graphAddrs { + nodeAddrsMap[pubStr] = nodeAddr + } + srvrLog.Debugf("Establishing %v persistent connections on start", len(nodeAddrsMap))