graph/db: expose reset in ForEachSourceNodeChannel

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)
 }

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

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 {
+	otherNode *models.LightningNode) error, reset func()) error {
-
-	reset := func() {}
 
 	return kvdb.View(c.db, func(tx kvdb.RTx) error {
 		nodes := tx.ReadBucket(nodeBucket)

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 {
+		otherNode *models.LightningNode) error, reset func()) error {
-
-	reset := func() {}
 
 	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		nodeID, nodePub, err := s.getSourceNode(ctx, db, ProtocolV1)

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))