From d1cfb47428db23ccff7b50e760e1d42fca77a7c7 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 17 Jun 2025 11:34:58 +0200 Subject: [PATCH 1/5] graph/db: thread context to AddLightningNode --- autopilot/prefattach_test.go | 11 +++- discovery/gossiper.go | 6 +- discovery/gossiper_test.go | 2 +- graph/builder.go | 9 +-- graph/builder_test.go | 13 +++-- graph/db/graph.go | 7 ++- graph/db/graph_test.go | 109 +++++++++++++++++++++-------------- graph/db/interfaces.go | 3 +- graph/db/kv_store.go | 6 +- graph/db/sql_store.go | 6 +- graph/interfaces.go | 5 +- graph/notifications_test.go | 13 +++-- lnrpc/devrpc/dev_server.go | 2 +- routing/pathfind_test.go | 9 ++- routing/router_test.go | 6 +- 15 files changed, 126 insertions(+), 81 deletions(-) diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index e0baf4971..fd94b86cc 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -422,7 +422,9 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey, AuthSigBytes: testSig.Serialize(), } graphNode.AddPubKey(pub) - err := d.db.AddLightningNode(graphNode) + err := d.db.AddLightningNode( + context.Background(), graphNode, + ) if err != nil { return nil, err } @@ -450,7 +452,9 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey, AuthSigBytes: testSig.Serialize(), } dbNode.AddPubKey(nodeKey) - if err := d.db.AddLightningNode(dbNode); err != nil { + if err := d.db.AddLightningNode( + context.Background(), dbNode, + ); err != nil { return nil, err } @@ -554,7 +558,8 @@ func (d *testDBGraph) addRandNode() (*btcec.PublicKey, error) { AuthSigBytes: testSig.Serialize(), } dbNode.AddPubKey(nodeKey) - if err := d.db.AddLightningNode(dbNode); err != nil { + err = d.db.AddLightningNode(context.Background(), dbNode) + if err != nil { return nil, err } diff --git a/discovery/gossiper.go b/discovery/gossiper.go index d6855ae45..3cd3bafaf 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -2022,7 +2022,7 @@ func (d *AuthenticatedGossiper) fetchPKScript(chanID *lnwire.ShortChannelID) ( // addNode processes the given node announcement, and adds it to our channel // graph. -func (d *AuthenticatedGossiper) addNode(_ context.Context, +func (d *AuthenticatedGossiper) addNode(ctx context.Context, msg *lnwire.NodeAnnouncement, op ...batch.SchedulerOption) error { if err := netann.ValidateNodeAnn(msg); err != nil { @@ -2030,7 +2030,9 @@ func (d *AuthenticatedGossiper) addNode(_ context.Context, err) } - return d.cfg.Graph.AddNode(models.NodeFromWireAnnouncement(msg), op...) + return d.cfg.Graph.AddNode( + ctx, models.NodeFromWireAnnouncement(msg), op..., + ) } // isPremature decides whether a given network message has a block height+delta diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 399010991..d337ac4c3 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -109,7 +109,7 @@ func newMockRouter(t *testing.T, height uint32) *mockGraphSource { var _ graph.ChannelGraphSource = (*mockGraphSource)(nil) -func (r *mockGraphSource) AddNode(node *models.LightningNode, +func (r *mockGraphSource) AddNode(_ context.Context, node *models.LightningNode, _ ...batch.SchedulerOption) error { r.mu.Lock() diff --git a/graph/builder.go b/graph/builder.go index 99fc73c17..930091f65 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -1,6 +1,7 @@ package graph import ( + "context" "fmt" "sync" "sync/atomic" @@ -972,10 +973,10 @@ func (b *Builder) ApplyChannelUpdate(msg *lnwire.ChannelUpdate1) bool { // be ignored. // // NOTE: This method is part of the ChannelGraphSource interface. -func (b *Builder) AddNode(node *models.LightningNode, +func (b *Builder) AddNode(ctx context.Context, node *models.LightningNode, op ...batch.SchedulerOption) error { - err := b.addNode(node, op...) + err := b.addNode(ctx, node, op...) if err != nil { logNetworkMsgProcessError(err) @@ -989,7 +990,7 @@ func (b *Builder) AddNode(node *models.LightningNode, // currently have persisted in the graph, and then adds it to the graph. If we // already know about the node, then we only update our DB if the new update // has a newer timestamp than the last one we received. -func (b *Builder) addNode(node *models.LightningNode, +func (b *Builder) addNode(ctx context.Context, node *models.LightningNode, op ...batch.SchedulerOption) error { // Before we add the node to the database, we'll check to see if the @@ -1000,7 +1001,7 @@ func (b *Builder) addNode(node *models.LightningNode, return err } - if err := b.cfg.Graph.AddLightningNode(node, op...); err != nil { + if err := b.cfg.Graph.AddLightningNode(ctx, node, op...); err != nil { return errors.Errorf("unable to add node %x to the "+ "graph: %v", node.PubKeyBytes, err) } diff --git a/graph/builder_test.go b/graph/builder_test.go index fa7fc722d..ca46671ca 100644 --- a/graph/builder_test.go +++ b/graph/builder_test.go @@ -2,6 +2,7 @@ package graph import ( "bytes" + "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -106,7 +107,7 @@ func TestIgnoreNodeAnnouncement(t *testing.T) { } copy(node.PubKeyBytes[:], pub.SerializeCompressed()) - err := ctx.builder.AddNode(node) + err := ctx.builder.AddNode(context.Background(), node) if !IsError(err, ErrIgnored) { t.Fatalf("expected to get ErrIgnore, instead got: %v", err) } @@ -1068,7 +1069,7 @@ func TestIsStaleNode(t *testing.T) { Features: testFeatures, } copy(n1.PubKeyBytes[:], priv1.PubKey().SerializeCompressed()) - if err := ctx.builder.AddNode(n1); err != nil { + if err := ctx.builder.AddNode(context.Background(), n1); err != nil { t.Fatalf("could not add node: %v", err) } @@ -1328,6 +1329,8 @@ func createTestCtxFromFile(t *testing.T, func parseTestGraph(t *testing.T, useCache bool, path string) ( *testGraphInstance, error) { + ctx := context.Background() + graphJSON, err := os.ReadFile(path) if err != nil { return nil, err @@ -1434,7 +1437,7 @@ func parseTestGraph(t *testing.T, useCache bool, path string) ( // With the node fully parsed, add it as a vertex within the // graph. - if err := graph.AddLightningNode(dbNode); err != nil { + if err := graph.AddLightningNode(ctx, dbNode); err != nil { return nil, err } } @@ -1715,6 +1718,8 @@ func createTestGraphFromChannels(t *testing.T, useCache bool, testChannels []*testChannel, source string) (*testGraphInstance, error) { + ctx := context.Background() + // We'll use this fake address for the IP address of all the nodes in // our tests. This value isn't needed for path finding so it doesn't // need to be unique. @@ -1763,7 +1768,7 @@ func createTestGraphFromChannels(t *testing.T, useCache bool, // With the node fully parsed, add it as a vertex within the // graph. - if err := graph.AddLightningNode(dbNode); err != nil { + if err := graph.AddLightningNode(ctx, dbNode); err != nil { return nil, err } diff --git a/graph/db/graph.go b/graph/db/graph.go index a68368e3d..a02e8193f 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -1,6 +1,7 @@ package graphdb import ( + "context" "errors" "fmt" "sync" @@ -260,10 +261,10 @@ func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex, // information. Note that this method is expected to only be called to update an // already present node from a node announcement, or to insert a node found in a // channel update. -func (c *ChannelGraph) AddLightningNode(node *models.LightningNode, - op ...batch.SchedulerOption) error { +func (c *ChannelGraph) AddLightningNode(ctx context.Context, + node *models.LightningNode, op ...batch.SchedulerOption) error { - err := c.V1Store.AddLightningNode(node, op...) + err := c.V1Store.AddLightningNode(ctx, node, op...) if err != nil { return err } diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 18d3e52a5..ee77c1376 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -2,6 +2,7 @@ package graphdb import ( "bytes" + "context" "crypto/sha256" "encoding/hex" "errors" @@ -99,6 +100,7 @@ func createTestVertex(t testing.TB) *models.LightningNode { func TestNodeInsertionAndDeletion(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraphNew(t) @@ -123,7 +125,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { // First, insert the node into the graph DB. This should succeed // without any errors. node := nodeWithAddrs(testAddrs) - if err := graph.AddLightningNode(node); err != nil { + if err := graph.AddLightningNode(ctx, node); err != nil { t.Fatalf("unable to add node: %v", err) } assertNodeInCache(t, graph, node, testFeatures) @@ -187,7 +189,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { // Add the node without any addresses. node = nodeWithAddrs(nil) - require.NoError(t, graph.AddLightningNode(node)) + require.NoError(t, graph.AddLightningNode(ctx, node)) // Fetch the node and assert the empty addresses. dbNode, err = graph.FetchLightningNode(testPub) @@ -214,7 +216,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { testOpaqueAddr, } node = nodeWithAddrs(expAddrs) - require.NoError(t, graph.AddLightningNode(node)) + require.NoError(t, graph.AddLightningNode(ctx, node)) // Fetch the node and assert the updated addresses. dbNode, err = graph.FetchLightningNode(testPub) @@ -235,7 +237,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { testIPV6Addr, } node = nodeWithAddrs(expAddrs) - require.NoError(t, graph.AddLightningNode(node)) + require.NoError(t, graph.AddLightningNode(ctx, node)) // Fetch the node and assert the updated addresses. dbNode, err = graph.FetchLightningNode(testPub) @@ -248,7 +250,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { testOnionV3Addr, } node = nodeWithAddrs(expAddrs) - require.NoError(t, graph.AddLightningNode(node)) + require.NoError(t, graph.AddLightningNode(ctx, node)) // Fetch the node and assert the updated addresses. dbNode, err = graph.FetchLightningNode(testPub) @@ -327,6 +329,7 @@ func TestPartialNode(t *testing.T) { // TestAliasLookup tests the alias lookup functionality of the graph store. func TestAliasLookup(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraphNew(t) @@ -336,7 +339,7 @@ func TestAliasLookup(t *testing.T) { // Add the node to the graph's database, this should also insert an // entry into the alias index for this node. - require.NoError(t, graph.AddLightningNode(testNode)) + require.NoError(t, graph.AddLightningNode(ctx, testNode)) // Next, attempt to lookup the alias. The alias should exactly match // the one which the test node was assigned. @@ -801,18 +804,19 @@ func createChannelEdge(node1, node2 *models.LightningNode, func TestEdgeInfoUpdates(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) // We'd like to test the update of edges inserted into the database, so // we create two vertexes to connect. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } assertNodeInCache(t, graph, node1, testFeatures) node2 := createTestVertex(t) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } assertNodeInCache(t, graph, node2, testFeatures) @@ -1483,6 +1487,8 @@ func TestGraphCacheTraversal(t *testing.T) { func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, numChannels int) (map[uint64]struct{}, []*models.LightningNode) { + ctx := context.Background() + nodes := make([]*models.LightningNode, numNodes) nodeIndex := map[string]struct{}{} for i := 0; i < numNodes; i++ { @@ -1495,7 +1501,7 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, // Add each of the nodes into the graph, they should be inserted // without error. for _, node := range nodes { - require.NoError(t, graph.AddLightningNode(node)) + require.NoError(t, graph.AddLightningNode(ctx, node)) } // Iterate over each node as returned by the graph, if all nodes are @@ -1671,6 +1677,7 @@ func assertChanViewEqualChanPoints(t *testing.T, a []EdgePoint, func TestGraphPruning(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -1687,7 +1694,7 @@ func TestGraphPruning(t *testing.T) { for i := 0; i < numNodes; i++ { node := createTestVertex(t) - if err := graph.AddLightningNode(node); err != nil { + if err := graph.AddLightningNode(ctx, node); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -1919,6 +1926,7 @@ func TestHighestChanID(t *testing.T) { // insertion of a new edge, the edge update index is updated properly. func TestChanUpdatesInHorizon(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -1935,11 +1943,11 @@ func TestChanUpdatesInHorizon(t *testing.T) { // We'll start by creating two nodes which will seed our test graph. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } node2 := createTestVertex(t) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -2077,6 +2085,7 @@ func TestChanUpdatesInHorizon(t *testing.T) { // the most recent node updates within a particular time horizon. func TestNodeUpdatesInHorizon(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraphNew(t) @@ -2108,7 +2117,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) { nodeAnns = append(nodeAnns, *nodeAnn) - require.NoError(t, graph.AddLightningNode(nodeAnn)) + require.NoError(t, graph.AddLightningNode(ctx, nodeAnn)) } queryCases := []struct { @@ -2245,6 +2254,7 @@ func TestFilterKnownChanIDsZombieRevival(t *testing.T) { // know of on disk. func TestFilterKnownChanIDs(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -2277,11 +2287,11 @@ func TestFilterKnownChanIDs(t *testing.T) { // We'll start by creating two nodes which will seed our test graph. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } node2 := createTestVertex(t) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -2422,14 +2432,15 @@ func TestFilterKnownChanIDs(t *testing.T) { // methods that acquire the cache mutex along with the DB mutex. func TestStressTestChannelGraphAPI(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) node1 := createTestVertex(t) - require.NoError(t, graph.AddLightningNode(node1)) + require.NoError(t, graph.AddLightningNode(ctx, node1)) node2 := createTestVertex(t) - require.NoError(t, graph.AddLightningNode(node2)) + require.NoError(t, graph.AddLightningNode(ctx, node2)) require.NoError(t, graph.SetSourceNode(node1)) @@ -2706,16 +2717,17 @@ func TestStressTestChannelGraphAPI(t *testing.T) { // set of short channel ID's for a given block range. func TestFilterChannelRange(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) // We'll first populate our graph with two nodes. All channels created // below will be made between these two nodes. node1 := createTestVertex(t) - require.NoError(t, graph.AddLightningNode(node1)) + require.NoError(t, graph.AddLightningNode(ctx, node1)) node2 := createTestVertex(t) - require.NoError(t, graph.AddLightningNode(node2)) + require.NoError(t, graph.AddLightningNode(ctx, node2)) // If we try to filter a channel range before we have any channels // inserted, we should get an empty slice of results. @@ -2922,17 +2934,18 @@ func TestFilterChannelRange(t *testing.T) { // of ChannelEdge structs for a given set of short channel ID's. func TestFetchChanInfos(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) // We'll first populate our graph with two nodes. All channels created // below will be made between these two nodes. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } node2 := createTestVertex(t) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -3024,16 +3037,17 @@ func TestFetchChanInfos(t *testing.T) { // both sides. func TestIncompleteChannelPolicies(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) // Create two nodes. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } node2 := createTestVertex(t) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -3119,6 +3133,7 @@ func TestIncompleteChannelPolicies(t *testing.T) { // up. func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -3136,11 +3151,11 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) { // We'll first populate our graph with two nodes. All channels created // below will be made between these two nodes. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } node2 := createTestVertex(t) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -3266,6 +3281,7 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) { // PruneSyncState method. func TestPruneGraphNodes(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -3280,15 +3296,15 @@ func TestPruneGraphNodes(t *testing.T) { // channel graph, at the end of the scenario, only two of these nodes // should still be in the graph. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } node2 := createTestVertex(t) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } node3 := createTestVertex(t) - if err := graph.AddLightningNode(node3); err != nil { + if err := graph.AddLightningNode(ctx, node3); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -3331,6 +3347,7 @@ func TestPruneGraphNodes(t *testing.T) { // database, then shell edges are created for each node if needed. func TestAddChannelEdgeShellNodes(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraphNew(t) @@ -3361,7 +3378,7 @@ func TestAddChannelEdgeShellNodes(t *testing.T) { require.ErrorIs(t, err, ErrEdgeAlreadyExist) // Show that updating the shell node to a full node record works. - require.NoError(t, graph.AddLightningNode(node2)) + require.NoError(t, graph.AddLightningNode(ctx, node2)) } // TestNodePruningUpdateIndexDeletion tests that once a node has been removed @@ -3369,13 +3386,14 @@ func TestAddChannelEdgeShellNodes(t *testing.T) { // well. func TestNodePruningUpdateIndexDeletion(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraphNew(t) // We'll first populate our graph with a single node that will be // removed shortly. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -3416,6 +3434,7 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) { // public within the network graph. func TestNodeIsPublic(t *testing.T) { t.Parallel() + ctx := context.Background() // We'll start off the test by creating a small network of 3 // participants with the following graph: @@ -3453,9 +3472,8 @@ func TestNodeIsPublic(t *testing.T) { graphs := []*ChannelGraph{aliceGraph, bobGraph, carolGraph} for _, graph := range graphs { for _, node := range nodes { - if err := graph.AddLightningNode(node); err != nil { - t.Fatalf("unable to add node: %v", err) - } + err := graph.AddLightningNode(ctx, node) + require.NoError(t, err) } for _, edge := range edges { if err := graph.AddChannelEdge(edge); err != nil { @@ -3551,24 +3569,25 @@ func TestNodeIsPublic(t *testing.T) { // DisabledChannelIDs is correct. func TestDisabledChannelIDs(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) // Create first node and add it to the graph. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } // Create second node and add it to the graph. node2 := createTestVertex(t) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } // Adding a new channel edge to the graph. edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -3634,6 +3653,7 @@ func TestDisabledChannelIDs(t *testing.T) { // receive the proper update. func TestEdgePolicyMissingMaxHtcl(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -3646,13 +3666,13 @@ func TestEdgePolicyMissingMaxHtcl(t *testing.T) { // We'd like to test the update of edges inserted into the database, so // we create two vertexes to connect. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { + if err := graph.AddLightningNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } node2 := createTestVertex(t) edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) - if err := graph.AddLightningNode(node2); err != nil { + if err := graph.AddLightningNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } if err := graph.AddChannelEdge(edgeInfo); err != nil { @@ -4037,15 +4057,16 @@ func TestBatchedAddChannelEdge(t *testing.T) { // executes multiple UpdateEdgePolicy requests in a single txn. func TestBatchedUpdateEdgePolicy(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraphNew(t) // We'd like to test the update of edges inserted into the database, so // we create two vertexes to connect. node1 := createTestVertex(t) - require.NoError(t, graph.AddLightningNode(node1)) + require.NoError(t, graph.AddLightningNode(ctx, node1)) node2 := createTestVertex(t) - require.NoError(t, graph.AddLightningNode(node2)) + require.NoError(t, graph.AddLightningNode(ctx, node2)) // Create an edge and add it to the db. edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) @@ -4139,6 +4160,7 @@ func BenchmarkForEachChannel(b *testing.B) { // method works as expected, and is able to handle nil self edges. func TestGraphCacheForEachNodeChannel(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -4147,9 +4169,9 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { graph.graphCache = nil node1 := createTestVertex(t) - require.NoError(t, graph.AddLightningNode(node1)) + require.NoError(t, graph.AddLightningNode(ctx, node1)) node2 := createTestVertex(t) - require.NoError(t, graph.AddLightningNode(node2)) + require.NoError(t, graph.AddLightningNode(ctx, node2)) // Create an edge and add it to the db. edgeInfo, e1, e2 := createChannelEdge(node1, node2) @@ -4305,6 +4327,7 @@ var testNodeAnn = "01012674c2e7ef68c73a086b7de2603f4ef1567358df84bb4edaa06c" + // that the two messages are equal. func TestLightningNodePersistence(t *testing.T) { t.Parallel() + ctx := context.Background() // Create a new test graph instance. graph := MakeTestGraphNew(t) @@ -4323,7 +4346,7 @@ func TestLightningNodePersistence(t *testing.T) { node := models.NodeFromWireAnnouncement(na) // Persist the node to disk. - err = graph.AddLightningNode(node) + err = graph.AddLightningNode(ctx, node) require.NoError(t, err) // Read the node from disk. diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 0f7d499fe..4e7f7eae1 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -1,6 +1,7 @@ package graphdb import ( + "context" "net" "time" @@ -56,7 +57,7 @@ type V1Store interface { //nolint:interfacebloat // update that node's information. Note that this method is expected to // only be called to update an already present node from a node // announcement, or to insert a node found in a channel update. - AddLightningNode(node *models.LightningNode, + AddLightningNode(ctx context.Context, node *models.LightningNode, op ...batch.SchedulerOption) error // AddrsForNode returns all known addresses for the target node public diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index bbac55f5d..bea8fb916 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -957,10 +957,8 @@ func (c *KVStore) SetSourceNode(node *models.LightningNode) error { // channel update. // // TODO(roasbeef): also need sig of announcement. -func (c *KVStore) AddLightningNode(node *models.LightningNode, - opts ...batch.SchedulerOption) error { - - ctx := context.TODO() +func (c *KVStore) AddLightningNode(ctx context.Context, + node *models.LightningNode, opts ...batch.SchedulerOption) error { r := &batch.Request[kvdb.RwTx]{ Opts: batch.NewSchedulerOptions(opts...), diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 2ae133480..148666cb8 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -174,10 +174,8 @@ func NewSQLStore(cfg *SQLStoreConfig, db BatchedSQLQueries, kvStore *KVStore, // information. // // NOTE: part of the V1Store interface. -func (s *SQLStore) AddLightningNode(node *models.LightningNode, - opts ...batch.SchedulerOption) error { - - ctx := context.TODO() +func (s *SQLStore) AddLightningNode(ctx context.Context, + node *models.LightningNode, opts ...batch.SchedulerOption) error { r := &batch.Request[SQLQueries]{ Opts: batch.NewSchedulerOptions(opts...), diff --git a/graph/interfaces.go b/graph/interfaces.go index 52897d8f5..a487c8af5 100644 --- a/graph/interfaces.go +++ b/graph/interfaces.go @@ -1,6 +1,7 @@ package graph import ( + "context" "time" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -22,7 +23,7 @@ type ChannelGraphSource interface { // AddNode is used to add information about a node to the router // database. If the node with this pubkey is not present in an existing // channel, it will be ignored. - AddNode(node *models.LightningNode, + AddNode(ctx context.Context, node *models.LightningNode, op ...batch.SchedulerOption) error // AddEdge is used to add edge/channel to the topology of the router, @@ -202,7 +203,7 @@ type DB interface { // update that node's information. Note that this method is expected to // only be called to update an already present node from a node // announcement, or to insert a node found in a channel update. - AddLightningNode(node *models.LightningNode, + AddLightningNode(ctx context.Context, node *models.LightningNode, op ...batch.SchedulerOption) error // AddChannelEdge adds a new (undirected, blank) edge to the graph diff --git a/graph/notifications_test.go b/graph/notifications_test.go index 2f2ec1b4b..099c2244b 100644 --- a/graph/notifications_test.go +++ b/graph/notifications_test.go @@ -2,6 +2,7 @@ package graph import ( "bytes" + "context" "encoding/hex" "fmt" "image/color" @@ -608,6 +609,7 @@ func TestEdgeUpdateNotification(t *testing.T) { // attributes with new data. func TestNodeUpdateNotification(t *testing.T) { t.Parallel() + ctxb := context.Background() const startingBlockHeight = 101 ctx := createTestCtxSingleNode(t, startingBlockHeight) @@ -665,10 +667,10 @@ func TestNodeUpdateNotification(t *testing.T) { // Change network topology by adding the updated info for the two nodes // to the channel router. - if err := ctx.builder.AddNode(node1); err != nil { + if err := ctx.builder.AddNode(ctxb, node1); err != nil { t.Fatalf("unable to add node: %v", err) } - if err := ctx.builder.AddNode(node2); err != nil { + if err := ctx.builder.AddNode(ctxb, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -763,7 +765,7 @@ func TestNodeUpdateNotification(t *testing.T) { nodeUpdateAnn.LastUpdate = node1.LastUpdate.Add(300 * time.Millisecond) // Add new node topology update to the channel router. - if err := ctx.builder.AddNode(&nodeUpdateAnn); err != nil { + if err := ctx.builder.AddNode(ctxb, &nodeUpdateAnn); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -790,6 +792,7 @@ func TestNodeUpdateNotification(t *testing.T) { // when the client wishes to exit. func TestNotificationCancellation(t *testing.T) { t.Parallel() + ctxb := context.Background() const startingBlockHeight = 101 ctx := createTestCtxSingleNode(t, startingBlockHeight) @@ -845,11 +848,11 @@ func TestNotificationCancellation(t *testing.T) { t.Fatalf("unable to add edge: %v", err) } - if err := ctx.builder.AddNode(node1); err != nil { + if err := ctx.builder.AddNode(ctxb, node1); err != nil { t.Fatalf("unable to add node: %v", err) } - if err := ctx.builder.AddNode(node2); err != nil { + if err := ctx.builder.AddNode(ctxb, node2); err != nil { t.Fatalf("unable to add node: %v", err) } diff --git a/lnrpc/devrpc/dev_server.go b/lnrpc/devrpc/dev_server.go index 71f51b085..a28a2500a 100644 --- a/lnrpc/devrpc/dev_server.go +++ b/lnrpc/devrpc/dev_server.go @@ -260,7 +260,7 @@ func (s *Server) ImportGraph(ctx context.Context, return nil, err } - if err := graphDB.AddLightningNode(node); err != nil { + if err := graphDB.AddLightningNode(ctx, node); err != nil { return nil, fmt.Errorf("unable to add node %v: %w", rpcNode.PubKey, err) } diff --git a/routing/pathfind_test.go b/routing/pathfind_test.go index 361fb83d4..0c483f0d9 100644 --- a/routing/pathfind_test.go +++ b/routing/pathfind_test.go @@ -2,6 +2,7 @@ package routing import ( "bytes" + "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -181,6 +182,8 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph, func parseTestGraph(t *testing.T, useCache bool, path string) ( *testGraphInstance, error) { + ctx := context.Background() + graphJSON, err := os.ReadFile(path) if err != nil { return nil, err @@ -290,7 +293,7 @@ func parseTestGraph(t *testing.T, useCache bool, path string) ( // With the node fully parsed, add it as a vertex within the // graph. - if err := graph.AddLightningNode(dbNode); err != nil { + if err := graph.AddLightningNode(ctx, dbNode); err != nil { return nil, err } } @@ -515,6 +518,8 @@ func createTestGraphFromChannels(t *testing.T, useCache bool, testChannels []*testChannel, source string, sourceFeatureBits ...lnwire.FeatureBit) (*testGraphInstance, error) { + ctx := context.Background() + // We'll use this fake address for the IP address of all the nodes in // our tests. This value isn't needed for path finding so it doesn't // need to be unique. @@ -566,7 +571,7 @@ func createTestGraphFromChannels(t *testing.T, useCache bool, // With the node fully parsed, add it as a vertex within the // graph. - if err := graph.AddLightningNode(dbNode); err != nil { + if err := graph.AddLightningNode(ctx, dbNode); err != nil { return nil, err } diff --git a/routing/router_test.go b/routing/router_test.go index 0f81bea37..a2647800d 100644 --- a/routing/router_test.go +++ b/routing/router_test.go @@ -2,6 +2,7 @@ package routing import ( "bytes" + "context" "fmt" "image/color" "math" @@ -2706,6 +2707,7 @@ func TestNewRouteRequest(t *testing.T) { // announcements for the channel vertexes to be able to use the channel. func TestAddEdgeUnknownVertexes(t *testing.T) { t.Parallel() + ctxb := context.Background() const startingBlockHeight = 101 ctx := createTestCtxFromFile(t, startingBlockHeight, basicGraphFilePath) @@ -2879,7 +2881,7 @@ func TestAddEdgeUnknownVertexes(t *testing.T) { } copy(n1.PubKeyBytes[:], priv1.PubKey().SerializeCompressed()) - require.NoError(t, ctx.graph.AddLightningNode(n1)) + require.NoError(t, ctx.graph.AddLightningNode(ctxb, n1)) n2 := &models.LightningNode{ HaveNodeAnnouncement: true, @@ -2892,7 +2894,7 @@ func TestAddEdgeUnknownVertexes(t *testing.T) { } copy(n2.PubKeyBytes[:], priv2.PubKey().SerializeCompressed()) - require.NoError(t, ctx.graph.AddLightningNode(n2)) + require.NoError(t, ctx.graph.AddLightningNode(ctxb, n2)) // Should still be able to find the route, and the info should be // updated. From 66c5a972025505fc7acb727b9b2b40c6d9bc72ee Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 17 Jun 2025 11:44:33 +0200 Subject: [PATCH 2/5] graph/db: thread context through to FetchLightingNode --- autopilot/prefattach_test.go | 6 ++++-- discovery/gossiper.go | 4 ++-- discovery/gossiper_test.go | 2 +- graph/builder.go | 4 ++-- graph/db/graph_test.go | 27 ++++++++++++++------------- graph/db/interfaces.go | 4 ++-- graph/db/kv_store.go | 8 +++++--- graph/db/sql_store.go | 6 ++---- graph/interfaces.go | 7 ++++--- routing/router_test.go | 4 ++-- rpcserver.go | 4 ++-- server.go | 10 +++++++--- 12 files changed, 47 insertions(+), 39 deletions(-) diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index fd94b86cc..8d785a81a 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -395,6 +395,8 @@ func TestPrefAttachmentSelectSkipNodes(t *testing.T) { func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey, capacity btcutil.Amount) (*ChannelEdge, *ChannelEdge, error) { + ctx := context.Background() + fetchNode := func(pub *btcec.PublicKey) (*models.LightningNode, error) { if pub != nil { vertex, err := route.NewVertexFromBytes( @@ -404,7 +406,7 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey, return nil, err } - dbNode, err := d.db.FetchLightningNode(vertex) + dbNode, err := d.db.FetchLightningNode(ctx, vertex) switch { case errors.Is(err, graphdb.ErrGraphNodeNotFound): fallthrough @@ -737,7 +739,7 @@ func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo, } func (t *testNodeTx) FetchNode(pub route.Vertex) (graphdb.NodeRTx, error) { - node, err := t.db.db.FetchLightningNode(pub) + node, err := t.db.db.FetchLightningNode(context.Background(), pub) if err != nil { return nil, err } diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 3cd3bafaf..3b1d24b90 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -2206,10 +2206,10 @@ func (d *AuthenticatedGossiper) processZombieUpdate(_ context.Context, // fetchNodeAnn fetches the latest signed node announcement from our point of // view for the node with the given public key. -func (d *AuthenticatedGossiper) fetchNodeAnn(_ context.Context, +func (d *AuthenticatedGossiper) fetchNodeAnn(ctx context.Context, pubKey [33]byte) (*lnwire.NodeAnnouncement, error) { - node, err := d.cfg.Graph.FetchLightningNode(pubKey) + node, err := d.cfg.Graph.FetchLightningNode(ctx, pubKey) if err != nil { return nil, err } diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index d337ac4c3..4f8836286 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -294,7 +294,7 @@ func (r *mockGraphSource) GetChannelByID(chanID lnwire.ShortChannelID) ( return &chanInfo, edge1, edge2, nil } -func (r *mockGraphSource) FetchLightningNode( +func (r *mockGraphSource) FetchLightningNode(_ context.Context, nodePub route.Vertex) (*models.LightningNode, error) { for _, node := range r.nodes { diff --git a/graph/builder.go b/graph/builder.go index 930091f65..0eab22033 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -1265,10 +1265,10 @@ func (b *Builder) GetChannelByID(chanID lnwire.ShortChannelID) ( // within the graph. // // NOTE: This method is part of the ChannelGraphSource interface. -func (b *Builder) FetchLightningNode( +func (b *Builder) FetchLightningNode(ctx context.Context, node route.Vertex) (*models.LightningNode, error) { - return b.cfg.Graph.FetchLightningNode(node) + return b.cfg.Graph.FetchLightningNode(ctx, node) } // ForAllOutgoingChannels is used to iterate over all outgoing channels owned by diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index ee77c1376..7e2bb653b 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -132,7 +132,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { // Next, fetch the node from the database to ensure everything was // serialized properly. - dbNode, err := graph.FetchLightningNode(testPub) + dbNode, err := graph.FetchLightningNode(ctx, testPub) require.NoError(t, err, "unable to locate node") _, exists, err := graph.HasLightningNode(dbNode.PubKeyBytes) @@ -170,7 +170,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { // Finally, attempt to fetch the node again. This should fail as the // node should have been deleted from the database. - _, err = graph.FetchLightningNode(testPub) + _, err = graph.FetchLightningNode(ctx, testPub) require.ErrorIs(t, err, ErrGraphNodeNotFound) // Now, we'll specifically test the updating of addresses of a node @@ -192,7 +192,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { require.NoError(t, graph.AddLightningNode(ctx, node)) // Fetch the node and assert the empty addresses. - dbNode, err = graph.FetchLightningNode(testPub) + dbNode, err = graph.FetchLightningNode(ctx, testPub) require.NoError(t, err) require.Empty(t, dbNode.Addresses) @@ -219,7 +219,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { require.NoError(t, graph.AddLightningNode(ctx, node)) // Fetch the node and assert the updated addresses. - dbNode, err = graph.FetchLightningNode(testPub) + dbNode, err = graph.FetchLightningNode(ctx, testPub) require.NoError(t, err) require.Equal(t, expAddrs, dbNode.Addresses) @@ -240,7 +240,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { require.NoError(t, graph.AddLightningNode(ctx, node)) // Fetch the node and assert the updated addresses. - dbNode, err = graph.FetchLightningNode(testPub) + dbNode, err = graph.FetchLightningNode(ctx, testPub) require.NoError(t, err) require.Equal(t, expAddrs, dbNode.Addresses) @@ -253,7 +253,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { require.NoError(t, graph.AddLightningNode(ctx, node)) // Fetch the node and assert the updated addresses. - dbNode, err = graph.FetchLightningNode(testPub) + dbNode, err = graph.FetchLightningNode(ctx, testPub) require.NoError(t, err) require.Equal(t, expAddrs, dbNode.Addresses) } @@ -262,6 +262,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { // only the pubkey is known to the database. func TestPartialNode(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraphNew(t) @@ -282,9 +283,9 @@ func TestPartialNode(t *testing.T) { // Next, fetch the node2 from the database to ensure everything was // serialized properly. - dbNode1, err := graph.FetchLightningNode(pubKey1) + dbNode1, err := graph.FetchLightningNode(ctx, pubKey1) require.NoError(t, err) - dbNode2, err := graph.FetchLightningNode(pubKey2) + dbNode2, err := graph.FetchLightningNode(ctx, pubKey2) require.NoError(t, err) _, exists, err := graph.HasLightningNode(dbNode1.PubKeyBytes) @@ -322,7 +323,7 @@ func TestPartialNode(t *testing.T) { // Finally, attempt to fetch the node again. This should fail as the // node should have been deleted from the database. - _, err = graph.FetchLightningNode(testPub) + _, err = graph.FetchLightningNode(ctx, testPub) require.ErrorIs(t, err, ErrGraphNodeNotFound) } @@ -3338,7 +3339,7 @@ func TestPruneGraphNodes(t *testing.T) { // Finally, we'll ensure that node3, the only fully unconnected node as // properly deleted from the graph and not another node in its place. - _, err := graph.FetchLightningNode(node3.PubKeyBytes) + _, err := graph.FetchLightningNode(ctx, node3.PubKeyBytes) require.NotNil(t, err) } @@ -3364,11 +3365,11 @@ func TestAddChannelEdgeShellNodes(t *testing.T) { // Ensure that node1 was inserted as a full node, while node2 only has // a shell node present. - node1, err := graph.FetchLightningNode(node1.PubKeyBytes) + node1, err := graph.FetchLightningNode(ctx, node1.PubKeyBytes) require.NoError(t, err, "unable to fetch node1") require.True(t, node1.HaveNodeAnnouncement) - node2, err = graph.FetchLightningNode(node2.PubKeyBytes) + node2, err = graph.FetchLightningNode(ctx, node2.PubKeyBytes) require.NoError(t, err, "unable to fetch node2") require.False(t, node2.HaveNodeAnnouncement) @@ -4350,7 +4351,7 @@ func TestLightningNodePersistence(t *testing.T) { require.NoError(t, err) // Read the node from disk. - diskNode, err := graph.FetchLightningNode(node.PubKeyBytes) + diskNode, err := graph.FetchLightningNode(ctx, node.PubKeyBytes) require.NoError(t, err) // Convert it back to a wire message. diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 4e7f7eae1..a03e0a4bf 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -126,8 +126,8 @@ type V1Store interface { //nolint:interfacebloat // FetchLightningNode attempts to look up a target node by its identity // public key. If the node isn't found in the database, then // ErrGraphNodeNotFound is returned. - FetchLightningNode(nodePub route.Vertex) ( - *models.LightningNode, error) + FetchLightningNode(ctx context.Context, + nodePub route.Vertex) (*models.LightningNode, error) // HasLightningNode determines if the graph has a vertex identified by // the target node identity public key. If the node exists in the diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index bea8fb916..7212511f7 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -379,12 +379,14 @@ func initKVStore(db kvdb.Backend) error { func (c *KVStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, error) { + ctx := context.TODO() + pubKey, err := route.NewVertexFromBytes(nodePub.SerializeCompressed()) if err != nil { return false, nil, err } - node, err := c.FetchLightningNode(pubKey) + node, err := c.FetchLightningNode(ctx, pubKey) // We don't consider it an error if the graph is unaware of the node. switch { case err != nil && !errors.Is(err, ErrGraphNodeNotFound): @@ -2995,8 +2997,8 @@ func (c *KVStore) FetchLightningNodeTx(tx kvdb.RTx, nodePub route.Vertex) ( // FetchLightningNode attempts to look up a target node by its identity public // key. If the node isn't found in the database, then ErrGraphNodeNotFound is // returned. -func (c *KVStore) FetchLightningNode(nodePub route.Vertex) ( - *models.LightningNode, error) { +func (c *KVStore) FetchLightningNode(_ context.Context, + nodePub route.Vertex) (*models.LightningNode, error) { return c.fetchLightningNode(nil, nodePub) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 148666cb8..a154fccbd 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -193,10 +193,8 @@ func (s *SQLStore) AddLightningNode(ctx context.Context, // returned. // // NOTE: part of the V1Store interface. -func (s *SQLStore) FetchLightningNode(pubKey route.Vertex) ( - *models.LightningNode, error) { - - ctx := context.TODO() +func (s *SQLStore) FetchLightningNode(ctx context.Context, + pubKey route.Vertex) (*models.LightningNode, error) { var node *models.LightningNode err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { diff --git a/graph/interfaces.go b/graph/interfaces.go index a487c8af5..e54795a98 100644 --- a/graph/interfaces.go +++ b/graph/interfaces.go @@ -84,7 +84,8 @@ type ChannelGraphSource interface { // FetchLightningNode attempts to look up a target node by its identity // public key. channeldb.ErrGraphNodeNotFound is returned if the node // doesn't exist within the graph. - FetchLightningNode(route.Vertex) (*models.LightningNode, error) + FetchLightningNode(context.Context, + route.Vertex) (*models.LightningNode, error) // MarkZombieEdge marks the channel with the given ID as a zombie edge. MarkZombieEdge(chanID uint64) error @@ -242,8 +243,8 @@ type DB interface { // FetchLightningNode attempts to look up a target node by its identity // public key. If the node isn't found in the database, then // ErrGraphNodeNotFound is returned. - FetchLightningNode(nodePub route.Vertex) (*models.LightningNode, - error) + FetchLightningNode(ctx context.Context, + nodePub route.Vertex) (*models.LightningNode, error) // ForEachNodeChannel iterates through all channels of the given node, // executing the passed callback with an edge info structure and the diff --git a/routing/router_test.go b/routing/router_test.go index a2647800d..4a6bebdce 100644 --- a/routing/router_test.go +++ b/routing/router_test.go @@ -2907,12 +2907,12 @@ func TestAddEdgeUnknownVertexes(t *testing.T) { _, _, err = ctx.router.FindRoute(req) require.NoError(t, err, "unable to find any routes") - copy1, err := ctx.graph.FetchLightningNode(pub1) + copy1, err := ctx.graph.FetchLightningNode(ctxb, pub1) require.NoError(t, err, "unable to fetch node") require.Equal(t, n1.Alias, copy1.Alias) - copy2, err := ctx.graph.FetchLightningNode(pub2) + copy2, err := ctx.graph.FetchLightningNode(ctxb, pub2) require.NoError(t, err, "unable to fetch node") require.Equal(t, n2.Alias, copy2.Alias) diff --git a/rpcserver.go b/rpcserver.go index cd2367446..f3a95a77b 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6963,7 +6963,7 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context, // With the public key decoded, attempt to fetch the node corresponding // to this public key. If the node cannot be found, then an error will // be returned. - node, err := graph.FetchLightningNode(pubKey) + node, err := graph.FetchLightningNode(ctx, pubKey) switch { case errors.Is(err, graphdb.ErrGraphNodeNotFound): return nil, status.Error(codes.NotFound, err.Error()) @@ -8040,7 +8040,7 @@ func (r *rpcServer) ForwardingHistory(ctx context.Context, return "", err } - peer, err := r.server.graphDB.FetchLightningNode(vertex) + peer, err := r.server.graphDB.FetchLightningNode(ctx, vertex) if err != nil { return "", err } diff --git a/server.go b/server.go index 9c6a08395..877169579 100644 --- a/server.go +++ b/server.go @@ -4637,6 +4637,8 @@ func (s *server) peerInitializer(p *peer.Brontide) { func (s *server) peerTerminationWatcher(p *peer.Brontide, ready chan struct{}) { defer s.wg.Done() + ctx := context.TODO() + p.WaitForDisconnect(ready) srvrLog.Debugf("Peer %v has been disconnected", p) @@ -4723,7 +4725,7 @@ func (s *server) peerTerminationWatcher(p *peer.Brontide, ready chan struct{}) { // We'll ensure that we locate all the peers advertised addresses for // reconnection purposes. - advertisedAddrs, err := s.fetchNodeAdvertisedAddrs(pubKey) + advertisedAddrs, err := s.fetchNodeAdvertisedAddrs(ctx, pubKey) switch { // We found advertised addresses, so use them. case err == nil: @@ -5216,13 +5218,15 @@ func computeNextBackoff(currBackoff, maxBackoff time.Duration) time.Duration { var errNoAdvertisedAddr = errors.New("no advertised address found") // fetchNodeAdvertisedAddrs attempts to fetch the advertised addresses of a node. -func (s *server) fetchNodeAdvertisedAddrs(pub *btcec.PublicKey) ([]net.Addr, error) { +func (s *server) fetchNodeAdvertisedAddrs(ctx context.Context, + pub *btcec.PublicKey) ([]net.Addr, error) { + vertex, err := route.NewVertexFromBytes(pub.SerializeCompressed()) if err != nil { return nil, err } - node, err := s.graphDB.FetchLightningNode(vertex) + node, err := s.graphDB.FetchLightningNode(ctx, vertex) if err != nil { return nil, err } From dc6259fcc35f37cfc186e10ededa1cbbf03f1a7a Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 17 Jun 2025 11:49:23 +0200 Subject: [PATCH 3/5] graph/db: thread context through to HasLightningNode --- discovery/gossiper.go | 2 +- discovery/gossiper_test.go | 4 +++- graph/builder.go | 10 +++++----- graph/builder_test.go | 7 ++++--- graph/db/graph_test.go | 6 +++--- graph/db/interfaces.go | 4 ++-- graph/db/kv_store.go | 4 ++-- graph/db/sql_store.go | 6 ++---- graph/interfaces.go | 6 ++++-- invoices/testdata/channel.db | Bin 1048576 -> 0 bytes routing/router_test.go | 8 ++++---- rpcserver.go | 2 +- 12 files changed, 31 insertions(+), 28 deletions(-) delete mode 100644 invoices/testdata/channel.db diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 3b1d24b90..2f390da5b 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -2443,7 +2443,7 @@ func (d *AuthenticatedGossiper) handleNodeAnnouncement(ctx context.Context, // We'll quickly ask the router if it already has a newer update for // this node so we can skip validating signatures if not required. - if d.cfg.Graph.IsStaleNode(nodeAnn.NodeID, timestamp) { + if d.cfg.Graph.IsStaleNode(ctx, nodeAnn.NodeID, timestamp) { log.Debugf("Skipped processing stale node: %x", nodeAnn.NodeID) nMsg.err <- nil return nil, true diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 4f8836286..fb219f41c 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -308,7 +308,9 @@ func (r *mockGraphSource) FetchLightningNode(_ context.Context, // IsStaleNode returns true if the graph source has a node announcement for the // target node with a more recent timestamp. -func (r *mockGraphSource) IsStaleNode(nodePub route.Vertex, timestamp time.Time) bool { +func (r *mockGraphSource) IsStaleNode(_ context.Context, + nodePub route.Vertex, timestamp time.Time) bool { + r.mu.Lock() defer r.mu.Unlock() diff --git a/graph/builder.go b/graph/builder.go index 0eab22033..ce8dff179 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -865,7 +865,7 @@ func (b *Builder) updateGraphWithClosedChannels( // timestamp. ErrIgnored will be returned if we already have the node, and // ErrOutdated will be returned if we have a timestamp that's after the new // timestamp. -func (b *Builder) assertNodeAnnFreshness(node route.Vertex, +func (b *Builder) assertNodeAnnFreshness(ctx context.Context, node route.Vertex, msgTimestamp time.Time) error { // If we are not already aware of this node, it means that we don't @@ -873,7 +873,7 @@ func (b *Builder) assertNodeAnnFreshness(node route.Vertex, // node announcements, we will ignore such nodes. If we do know about // this node, check that this update brings info newer than what we // already have. - lastUpdate, exists, err := b.cfg.Graph.HasLightningNode(node) + lastUpdate, exists, err := b.cfg.Graph.HasLightningNode(ctx, node) if err != nil { return errors.Errorf("unable to query for the "+ "existence of node: %v", err) @@ -996,7 +996,7 @@ func (b *Builder) addNode(ctx context.Context, node *models.LightningNode, // Before we add the node to the database, we'll check to see if the // announcement is "fresh" or not. If it isn't, then we'll return an // error. - err := b.assertNodeAnnFreshness(node.PubKeyBytes, node.LastUpdate) + err := b.assertNodeAnnFreshness(ctx, node.PubKeyBytes, node.LastUpdate) if err != nil { return err } @@ -1306,12 +1306,12 @@ func (b *Builder) AddProof(chanID lnwire.ShortChannelID, // target node with a more recent timestamp. // // NOTE: This method is part of the ChannelGraphSource interface. -func (b *Builder) IsStaleNode(node route.Vertex, +func (b *Builder) IsStaleNode(ctx context.Context, node route.Vertex, timestamp time.Time) bool { // If our attempt to assert that the node announcement is fresh fails, // then we know that this is actually a stale announcement. - err := b.assertNodeAnnFreshness(node, timestamp) + err := b.assertNodeAnnFreshness(ctx, node, timestamp) if err != nil { log.Debugf("Checking stale node %x got %v", node, err) return true diff --git a/graph/builder_test.go b/graph/builder_test.go index ca46671ca..8795fc857 100644 --- a/graph/builder_test.go +++ b/graph/builder_test.go @@ -1013,6 +1013,7 @@ func testPruneChannelGraphDoubleDisabled(t *testing.T, assumeValid bool) { // node announcements. func TestIsStaleNode(t *testing.T) { t.Parallel() + ctxb := context.Background() const startingBlockHeight = 101 ctx := createTestCtxSingleNode(t, startingBlockHeight) @@ -1053,7 +1054,7 @@ func TestIsStaleNode(t *testing.T) { // Before we add the node, if we query for staleness, we should get // false, as we haven't added the full node. updateTimeStamp := time.Unix(123, 0) - if ctx.builder.IsStaleNode(pub1, updateTimeStamp) { + if ctx.builder.IsStaleNode(ctxb, pub1, updateTimeStamp) { t.Fatalf("incorrectly detected node as stale") } @@ -1075,14 +1076,14 @@ func TestIsStaleNode(t *testing.T) { // If we use the same timestamp and query for staleness, we should get // true. - if !ctx.builder.IsStaleNode(pub1, updateTimeStamp) { + if !ctx.builder.IsStaleNode(ctxb, pub1, updateTimeStamp) { t.Fatalf("failure to detect stale node update") } // If we update the timestamp and once again query for staleness, it // should report false. newTimeStamp := time.Unix(1234, 0) - if ctx.builder.IsStaleNode(pub1, newTimeStamp) { + if ctx.builder.IsStaleNode(ctxb, pub1, newTimeStamp) { t.Fatalf("incorrectly detected node as stale") } } diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 7e2bb653b..25409df09 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -135,7 +135,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { dbNode, err := graph.FetchLightningNode(ctx, testPub) require.NoError(t, err, "unable to locate node") - _, exists, err := graph.HasLightningNode(dbNode.PubKeyBytes) + _, exists, err := graph.HasLightningNode(ctx, dbNode.PubKeyBytes) if err != nil { t.Fatalf("unable to query for node: %v", err) } else if !exists { @@ -288,7 +288,7 @@ func TestPartialNode(t *testing.T) { dbNode2, err := graph.FetchLightningNode(ctx, pubKey2) require.NoError(t, err) - _, exists, err := graph.HasLightningNode(dbNode1.PubKeyBytes) + _, exists, err := graph.HasLightningNode(ctx, dbNode1.PubKeyBytes) require.NoError(t, err) require.True(t, exists) @@ -302,7 +302,7 @@ func TestPartialNode(t *testing.T) { } compareNodes(t, expectedNode1, dbNode1) - _, exists, err = graph.HasLightningNode(dbNode2.PubKeyBytes) + _, exists, err = graph.HasLightningNode(ctx, dbNode2.PubKeyBytes) require.NoError(t, err) require.True(t, exists) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index a03e0a4bf..9f5eb6c8f 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -134,8 +134,8 @@ type V1Store interface { //nolint:interfacebloat // database, a timestamp of when the data for the node was lasted // updated is returned along with a true boolean. Otherwise, an empty // time.Time is returned with a false boolean. - HasLightningNode(nodePub [33]byte) (time.Time, bool, - error) + HasLightningNode(ctx context.Context, nodePub [33]byte) (time.Time, + bool, error) // IsPublicNode is a helper method that determines whether the node with // the given public key is seen as a public node in the graph from the diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 7212511f7..be8ab2d9d 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -3065,8 +3065,8 @@ func (c *KVStore) fetchLightningNode(tx kvdb.RTx, // timestamp of when the data for the node was lasted updated is returned along // with a true boolean. Otherwise, an empty time.Time is returned with a false // boolean. -func (c *KVStore) HasLightningNode(nodePub [33]byte) (time.Time, bool, - error) { +func (c *KVStore) HasLightningNode(_ context.Context, + nodePub [33]byte) (time.Time, bool, error) { var ( updateTime time.Time diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index a154fccbd..1fd68a7f9 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -217,10 +217,8 @@ func (s *SQLStore) FetchLightningNode(ctx context.Context, // boolean. // // NOTE: part of the V1Store interface. -func (s *SQLStore) HasLightningNode(pubKey [33]byte) (time.Time, bool, - error) { - - ctx := context.TODO() +func (s *SQLStore) HasLightningNode(ctx context.Context, + pubKey [33]byte) (time.Time, bool, error) { var ( exists bool diff --git a/graph/interfaces.go b/graph/interfaces.go index e54795a98..e9f894041 100644 --- a/graph/interfaces.go +++ b/graph/interfaces.go @@ -46,7 +46,8 @@ type ChannelGraphSource interface { // for the target node with a more recent timestamp. This method will // also return true if we don't have an active channel announcement for // the target node. - IsStaleNode(node route.Vertex, timestamp time.Time) bool + IsStaleNode(ctx context.Context, node route.Vertex, + timestamp time.Time) bool // IsPublicNode determines whether the given vertex is seen as a public // node in the graph from the graph's source node's point of view. @@ -238,7 +239,8 @@ type DB interface { // database, a timestamp of when the data for the node was lasted // updated is returned along with a true boolean. Otherwise, an empty // time.Time is returned with a false boolean. - HasLightningNode(nodePub [33]byte) (time.Time, bool, error) + HasLightningNode(ctx context.Context, nodePub [33]byte) (time.Time, + bool, error) // FetchLightningNode attempts to look up a target node by its identity // public key. If the node isn't found in the database, then diff --git a/invoices/testdata/channel.db b/invoices/testdata/channel.db deleted file mode 100644 index 69397f529dd11a2bcab9d1fa8e64e78d18b0d29b..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 1048576 zcmeF)2|N{V|3Ch74nk-lLZw5Bin3JJqO3(ln*}LXf#d{=8oSz zlzhjarcF6BlsTjJzdyNU0C$c$oinlM-RY5GOFn3~SHuJaAOHafKmY;|fB*y_009U< z00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$##AOHafKmY;|fB*y_009U<00Izz00bZa z0SG_<0uX=z1Rwwb2tWV=ofhDx@6-OL`TuI%4cs}o?yUcS19LjPV^o9y1Rwwb2tWV= z5P$##AOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$##AOHaf zKmY;|fB*y_009U<00Izz00bZa0SG{#U4X%zVzfU2fJQSV^OVW3oeWfZe*+mr$iU*p z)?{XHGChk7R5?K6E@VutBeeh4n0fcJ9Ph5XHhoTW`qs81%cLXdsfVT2PZ??Y_1$ra zTEVND`B891y6%>ry|i@Q^gLL)EWJ^l`feU0-8^*-v}mHSOxj9@{VsBJi^(DCds%9k zGnn1V&fe?&mT_l~SmVwCUv5nq(Rbk0p2Ko4oRj$UbZ5(mI0jS6FwN++Z%)~^U2CPr z7kwLjZF+vKC|%0z+O^CZ5&h^fG&g@=w=n7m|5O-k49=1sWJEQ$7a0V$>LI45~fB*y_009U<00Izz00bZa z0SG_<0uX=z1Rwwb2tWV=5P$##AOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z z1Rwwb2tWV=5P$##An==jFgN1Q{}(0mnPl)}dAUXShjVNF%?N4!e?x*95P$##AOHaf zKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$##AOHafKmY;|fB*y_ z009U<00Izz00bZa0SG_<0uX=z1Rwx`e_TL>y9$5)zc`uyU5G{t4vh$8Y5E6w|GiB! zXfo}KwCFS%^*{k8caAw=ZjLcezHLic-aXg;$gb&82c(3x6S|W+IfbU0W@V?w9$RwoOsVM4%a+karV|VY1qI8l z%(ayCKFcC?a_fpES8ug^A5qY4Snz;McezDAqhCx|pg&*g&e!^(78+AXoxC~ELcb2v zoFIR*x_|8LK(*w}ZRvLv*AHI)!O=^j=6(JwQs8vjkKOp z8FR{`JlA#fkzI0hL;DlP)840EOwux(L+TV;>>F99cEZV~BBR(KamA!X8YZ$z$_2vd z8Lh|CjfKlSNS%udKCYZSHb*ye*SCUK*IGmJ8jQ*ZeR}ogkV8ONotTr2FR63shwokQ zCAK^2g$Gp?Em?BU%(^J%)yIO2A@wQ_^lY7s$d$0BnI3^$f0SG_<0uX=z1Rwwb z2tWV=5P$##AOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$## zAOHafKmY;|fB*y_0D*s{Ko{=D88njYPYTJ9r)iNvl?+eF!vUyRg;eSo_mbqq)nvI; zT(_B)YBxnn;?IveY|whNYIe4cckGANuPhRVic55_A*=u0!eA%ga~7MD5tV1$joKk8 zQ^=T#udpK;O^Kw=OOfe4WK5L;lDAJS&m)6}k6WOoM^He3Z+HMJFkI7*6-`y43b!`9 z8JszszudiW=8L_*6HCoz1mAzpSv(%|(7}?1s zJ8qOXDao1)F*zF-Cmr*CJm-K^XiR${Kd`M`&8%ci<4C-U6V)OaHPI( z@FwHom8DEeEprCbLn^YlO@5+YQsKmH+iv%7cws7Xrr(U_rOD1x=d&V3lo?E)P_ZW- zUDsc`+!B@hj=zbtkKB}BB@XWuu_ zD~LMpj4mpHmOT|EW?eff?P^xEbj>I2x>cV#y4RZ{l&!qsVdE9*z6*LkU8>fgt3g$!2}zl<$J{@>Zbi1? zPR)|IIXafJNVJqQCR={~>oY(XqK(yxq01*UnOyhI>JfIpxHUFJbbytAt!u@#l{!5Z zP+MW2HoB0>86A;j=kmXY7JG)2NOZf_fcT^D0}qLgB{kWfXF6m0;X6|X zbUA)%nrlFI!Sq-SQ6cg4w>?gHTyrV7-%`}-MQXB}N*CHxVo>#N$?X1LjZR&B+A!U* zb#Jc(mEPIA4V5o$x}HAnCaKBpEOdtXh?~={ziLQ3?=~)f9Ai@GvF*ckg(ED2emGW* zoYH$*5UEK@XIp)2Wpep<`-gHalfn+o?Okjd7U132y~eoc;Jk{Tk1p!ZPgy*`o*A((pX&7c_dLkfa_3_z{x6fFz zqi3;=X6d+5L03i`N;~OIo{YdLbVlF6r&*?TU+5(P6PGuH&0JvI_kriO$~(^6pGm|o zD4lVzy`~VYude)|t678Y89BWTUA~~OXWG)x$wv-4rBr@c+Pu27o-Rh}QqNPNGpH$c z5n@kCxMc!E3ujUNRw#Dt(0)&y1DCrDj1%j+v9)!0z&Dff zxr<({D4I6ui0Q(9_4)n1&t2;pH}UJOD34(l7n(2hkd=3tI{!)VjmBq{nI`1S{A8D; z4YyYK>Lx6`HQ`xU+T-eHs=HU&#b>I#_38C!@`U^I$(4~^?Z> zQ#-f4j_R4KD|vHqvX`u~@b#Vip<>V| z`_!G?nilTXKIkw-C+glAn=iemm88-b%vlFi3kR!@FEhSmKV!|6uss)+-Dl8+DqY@; zIWVr*@S*qr`8&YBK!^~$Y0wuHe&98*e{`r&sA{9?Bab>dZSLLalSf#LUs)h!&eI~8 zsjob9zfN1KsjZ*UI(fQQiT{Te2Pq+Py8f{{K&MV6bq8R7HEO>DJia-Q zOr17aJA3zx+nf8nV)T2s^KeVy_T5cCET5)*PMv)`+-Njei|Rjgy3oMeC4on67+d25 zuGRTP4e9^(z$k-Fg{ylxy*J2t)|9VEYEgFpQnOq-cW3q??bD{K3|sErIk`J@@T7+6 zn8dQxHo^v@-UT*ksv>pAq6ug^<+PTZjztRR_|y`tb4jfQ{`ku`!6 zlwN=JDLbCFN^QNZ`3Ij*kDa8ePyV=bFxPaKS^FJ;!G8x3%NEfRa0eLv<=IY!(6qhQ zw2Uv2RbJxz{8c9>t&ebY$$WS9rS_A6Q_-f2X5QJN{MIwv)cITDn)tv<_zN90Zj7;C^l~1 zrhKTeu_YOEdwjZlC{B6w6a5z^GrOkGA5nUL>(#5DBpOq8zAV_t1N_tY zPj`UsoJ%BaDj9eD4bY!7vEy%m!vDqJ0KAXt-@OBLs*w$Sdi-_=s2ryCV3_*%kS(pQ z8NQ##c^03|vE*IT$mLH`BqL>BNVaGBvI)NB79KB-u+nXNZ%v%-aQlesi7Xvi>zYia zm6ioVnUVyMBN6!-7Tby5CwDOjXB|^LKq&HD0kn zIx%xvf^E9+C)c0x0n29e*--v{OmFgRsz3Zp;zD=sXVSlbh#1_pit`_yHgmP)cE$#q zu5XXVyu6vvH%qQ|PJRB@r+V&(d0gZca3k-6d)15+hb)7#g>81k3VrfBkxCJfe)W&t z0y=drsapW`ZLa-CfKW(X^25mSrSc_8(Qt>^ zb$;1!--rc==CgCymCx6vEDw;Qa*xOOp4Re>;2`}CL zL3EO6VBwvqqfJG&*44BH`?lW#g#Phc0AsF1u;Gff10jw!L5*<(gKjyj>V3K?E!RKV z>+FW6Rkfi79%EJ89Jf|{OJ71;kWmroe&lhFx%LfBD(CWNZZ~&6*nSJxc6M2v=%(PF zk}H?Dezq&rwt9UcMr62^?VNFkKaCW=UBbHsl(8aT9X2T(Xe_NGUmvk(?8&1J#)VI> z#7o%DxvJ-9%HrJugf^z^JEP(Jo-wbbK)*~`t>v5e5-$<&>2i5G_k;u0PRsFtFopmG zAOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$##AOHafKmY;| zfB*y_0D*szfGBsn4B9~U6qQ40~P1;#i}H$pyTqjB$g!gnB4lAB$nt9 zUm`J;?{R-I`>hB~mm2bx+1GWA& z5>t6gytskHRQ4~gehZ1Iu`HQpa^q$aQ$D=)e{kbz+~t$V&MDt3 z+_)!cEZIf-`Z^@0+Q%fr|4KeS+0LH|gWVd=66#nxH`GVh<; zbEnde{zrRu^Y?WN({zvU@MDGloei|-dbzuWu{v3mlU?O0ur}2RIYo<%nA~Y9bo2D2 zGM@fe`0IK_dTROydH=C4naN+BB>jhl2L-YOZ{I(NoT#vffB?5p-%eKd)MTGBK^@`u z^QB6T(usSRJIJ_W+>Ly;c(I`KMG17K4vqVm4M88&1+OQjeW_j1uqol;yof$Bq8Y^l zjKZ&cZv9kmIc{dyetGhtFZ#zG(7_o0o;pZQAsJ-?p1r`W=T43PdmiFJ7x?4{IxUQr z(5W-ceHsbzV+eZggphaCTbbJlFE6f`;Qy>VEiyMNXn5Ya(aX!yy0tuRS>@p&=yE`v zrOwo0cR!K_E~xui7MAW_DzkEG&k4gshZq_-rT?^lIi}0==Mlm0wmcHwY z!QJ3x!1(KHMEbhOuN+c5X~Yex`~72?$-j3$fk~(TK==C(%3h`mOnGH@zfPU$4&6^s z!mUuK{j$Wy`*WQyKRn^@lvBI+!h-9*GxAGfnx7a%Y@DSy#Dn@a;Heqr;p@pcF@O8F z{hLNmpqFoGfToXcVEA8r%V>WR2F;ahi%AAyGEjy33dtk`FQ$Gkbd0z0#mD&KiydOY zw+VWl;LkEWuQONU&U|4IeB$VNo%!fd-}-{xJpA0e1%Ea34s{Fm(PVjgv-%O4j)k)% zfEDgWero+zI7uhrB`3Cxi`?~p?alQjvMtYL3 zz;;3Yz8>w5$m4B~H|FifH*j8%uLn!>ANfteT_>yuZ)4nX04pqv2$APq1%>{_mv7## zIo)h1S$QRQ1DD8{iZjU|_`4qWizS^wXI$Qs)JyMmQ0&8&#AV-l`Tw{#+O7MNUYje_ zlHQEXXnAmBc=eW>XKM=w&Tzb|rBoVpqkGqog&CD8R}70MPHhcn7$@lR%=;4SRn5FX zr&G_0po-9I&OoD65%rz1^9L5>bT@7m0hDnAI?5z>36TcWvrLXF6l1oOg1&GoFMCJ-}HUjm`}y2 zx-Sj;R#6?+(=FVMYUG~^|IcLV_Xc$_@pKO+fBOgchJ}%*pm_uZhKC0EYlgDIBK-gA zO2yelAS*mFDAbQzoztZR{xapQ8Wm1Ht+IpY5f% zZh@=wbiaPRlWW&Udyk)QPpZ=IFL<%{^wn);pW@AL`v)pVZAl9}WV+(0LE~4ci@sdxTiLZb5sJosa42IELJG*|CuOpX0`tfV})yyej=ce><>A69qPKrBm>pRs^PVl zb*+=ECdi0%8}LKQS7+o7qrR1L<7qm&HO*$L2a+oEFw%8A1>K-|C;iuhS$T%mWE)hs z;Ki;T;!br7^4+mh{esl%RL3Co!XjC$U{V!de|m8xwF*^u{|1}8RnAQEmlw0g0$*I&sYc@M&VpX@ z{hJWp?o4$<`t$aGoc$k^o5}f@bt)qjZ+B)v()cfC=WT0YW5>xMc=fw^rrrsZa zj11Ir5%PvA88T*aV|5aDT;H6;)cfao%RNZk@qO`2`Icvpn0gQ5QL-Ou`_srv4Lj~X zh{V*h0%nqh)bfj>zrU}(l*C=heCjy{)bfXX@e97Vfy8nh>I;j}Xw>_}m}H>VA4X#8 zSp>Xz3W>>o?b~-G0d+h|0?;m*^B;GJj(D(}J((LkI@`ib;dvXfZ`KX7AP;Z#;%=0!n zQ|I5~72b5b!uxjJM85&|<}X>kMrxdGPGz}r7E@0B?!W6RyzvfY!Edzx|LYyf)P1I7 z>C8Kmd9T0Zt@rDl$<$q}WBKztlc{@k$MS#boyosm`0PsBK)oZG_s(Pya`B~N>Yd3{ zxyFro%Y(?3xa0B>l^&fC!YW>c?z8LRZ{@1>9neq7Q;H=Fh4tL6(@4M8jtIT~F z;_P3lc(3BEwx~jl!bn@*O_6&0wZZ6wA!}Etj~jP2&DiV9hrOo*=P;t@i_N)kZ(CW@ z+%a+kbG_bbFBh0I!1LyVV z{6UkqUkkkRQ*XZ}J2l(Y<@|zxL)!->XR8{Wl~g>u@zNqq+wC=aO(E%5cl8gJo;5k; zgQ#Nq7e;MbcE!Gn#v+?)brkLwX06*Fm$^6d&R5YXkvmhIkBp*EpPaS-c)pKYf%VC@ zfL!g*t!qssO?Mg$88+><#G%%)r!wQd52*GlEPQ7edbM)muJuz3$2*8v35C=pKa3n- zDqo@$E$6n^=8a0cZ_c9I-nA{yl_gdMk9f;I6Vz`4awZ0x$WK}4J={kyvgcxU%*Df+^L8H767ia|{6bsZ5O+rNm@{hfYKB2gdYw8G zWM6{LMCT8hoC$&N{H4wW*=dMVM$ho{ty^fh2Sm11?J}9c2u{13UzeiUmodq-`Q0X= zmTFnaD$hsX%be6t9loEDepPh_z3a>3lHlPR``DkEUO}6>Urc1V()=K)i=`i0$~0P| z_Owmk(wbzwLt#;@da-EE!;pUTv_WZdy+*Z4OdB3s72s5=T)igAeAQE<8=mq1?e8I{ zek^i|78xm%L7y8(^TiiB!~(DB4^E!{BCry*Bhu+Ek`bM@nEd&2T;Qjjypz{}t}kuj`KG-{KYSo%-2C9yurQQ%kjsPQ5JsKk*gr z|!7JSTIUab0`)~aM0R$ib0SG_<0uX=z1Rwwb2tWV=5P$##AOHafKmY;|fB*y_ z009U<00Izz00bZa0SG_<0uX=z1pW^SbmboNzhnMClN=kBKzxbhH%O2%FRmdml_z+P z4Ak;1BmuBve*0b$Q|;pAyC3C?bNJ$75_incuOo5C{C^qpZ^}F7|Lc*M${FNs--N`} z@$mBh9m&qA<-Gj=L=t!0{!tQl%>RE#;*R z9rOS9lDK33KlQwUjz0IeF)#mr5Lub>ok<4DGn&L5^Z&P!n5@F~NB+Ob%&zJ4N0i>* zdiCljiN=(jFAFw$__zfIviv6pvjTksy{Cq|g|i6PAIpDP{y)dye?Q^+_lzed*$w&3 zvWq$S%OnFYraCn*mWt~((^BoGNJ;$pafc0Bk5+zml4%k6p*YSwiKrQ6IRD(*x!gbOSQ;InG8sB{)=Z15FY_a&b&vzaZqWr|4T{EG{IlNNVAaS%saPa z%;4Oq*mI~~l1wsCUy`ea*Iw4OPO_RHBhqca4=G=rkvoj~R?3a1>FCxpo2?#5s?fv8 zFB?xyAK$?6FcSW;@NZs~)S@=usS`z`7luW$Si!7N&G0BqAGa_c)?e(1v;Rmp-*ED4 zD>yVL$V*`5KOQn>LI45~fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$## zAOHafKmY;|fB*y_009U<00Izz00jQ82=Ee68SIR2&QbTE(`bXq@NjNfo$-ugR99ze(C4kTZ`A#oU;;{TR{T*ViV=^&RWic8JNL z(zc6p^8XV^K*jw@BRl5vk0&unv8SyjN%7S3+ay1i%6G3J0~H^W=j6*X$+Y12A5tlk z1jJIyoWT@wo%Ma$5eJ_8|Fs#THp746N4#!%-bcbugjvu zZYeD;%$r^2zANh<8N9Y>Sj0iO8+Oy~TG&L|Ssr!S z{@EeCZssUe0$U;IHLC1ZuoEnG%_r@;Ri8S&T;D?JgJFV{t-Rr3;}z<@3wl3Y zs@9;Z!QL(DY*tzp3}xEkE7nEEuNu{|HNHKFJiE&K#jLE4@&{GT{e-IX`WQFPVlZuP z9=-VU-3HsTbwVGz`}>W&CNqBCo%08V$K|M=NmV#k#oHdmD6kp(Z;hFEKg;p%x@*(t zG^cNEJF-kVf}VOgB1HK(d$dj?=X?; zSG;}ejzK2pcMPIkIox0*Q#j%kgE@Xg_r-w&9dp+;epaY?F)GK^&@ymTRNZj#<(JNW z&KTsx+a9+W_JK6s^+HO1yNYmzB%d)#K+pws{U z2V5X=`$wa5&EQ?8R6o_HIYl~!Gu<7wuB+ABm(aFNdG5H_H{QcWJhoYK-JQV{HVK%o zqEcp|Fg$*Ri|%Fd`OiJhv^J>~?>a_H3z6@0j@^omm#M#Mg`kfuyA{-BO7N>&sf#V| zGUcZ&dOvIUty{kH^%zp}6M`!)XT-1D5_|FvZl3r)ha@np3^@QBb!s zVQmeYH7XBVKWSZh_1O7N?aO`aR&=~f{nM?WE>oRr1-S=uy43slehJ#6Qk4g4-XD_t z)oMn_@CV0}TZX*zJnrKCK0y60gXy(;$u^I7uWTY7rQDTNt3H-Iz-+XdxW{|5Q{k+x zH2F#FW{~Z*cd5U+XLRb5!ET12n(W=SUu`e%N$REdIwF zsH?3|rIZ%)Qvv_fD{`StnkR4aHFTS2dVCl^}*FMemY;@8FY zf9WbkJts%gH_($6#p$en`i1z!#_T#V7gVQoi_Yhgk5FU`S=| z!!QIt5r$ir9D1j+U8(Jo4!5=C&6!QY@111blGV;F>i0BVDJOWo&K9k3M{kLO)wdh{IFpuS>3GJCAH)-s|qB zF?r4$o%x@}4Rnimu(e-c;nT}Mlq1ex4_?+@mt4O+eZ$<`{aK!xEKhHiW^j#$MqcI+xUW{6c-dCdHoYn~J}2V{>vH=@{z~%Z~NcNzgG?Am88H>rP z0&kLxgq6Mfg}qDdB63nHDdqGY+S`8l8@)%DRTx@2Kd2w4%M@x1J3UT}W#8W}?O>+u zc-di2GJBUF@A`VyS-Y<}Y5PJYcz0MmF->E?$0bK;Yvu-gS-1Yi@V4Ulu|KCTt}pnx zQf6%u`O-GB;g*Ar^P($CKg2#(7Tw-AZkYCf^wXX&WE#%{#6FzJ4Z&UC7S5ktl^h^@j3%j5;Igt} zaQFUU{S)K*ABkNR%wQUd3BR0kmP0)OQQ z?f~*qS-o~;S~*S|u;j+hA+N?vS85(V{*L=T?FHg*h98J{!}y+YKDElk%%@gFL7|Vx zD)AW)cfCI|^bRBN(NcxigFg=BbpSi%?#avE)mlHuae#&5{3ppP4~vPD);P2r@?%Sg_4_CDEZ?pUt2LsEWS*5jrr9!tj9zN{%V zvRGM+bkIHo(hCnn7tS}B`${^`W3l@`%&3(Uq(N(}6-`{gM^ z$THa=N&TT@lbZdG>K&!6q4Z0#<7QfH__5B~Y2KJMKS&qgSe3-fTG!^>zg=ozwy$XK zynXSvwA?jE7G7&@-qIw0NvV;+ly)_@k`}XGn3R8QoyPw8nHMitO?kbep?|3Ig6xm? zCqCtDr}=*L&qD)>6XZQ}T1}lqGjpmId~(oE`ykzVYWL|UbP@8gn{ee=QLnI-Tm0*% z%x11$o|31&etoQ+;^Hs+GhPdSQs8W-(ptgWG5W5i{qBc;L7BU=PRQKK)-N(zniQM- zN^W3u8~NJryW-^KaoaPW$TA!FZ>J0C-ahkWZT(gdcq;o%7 z?PU#J*4VL~x|?;p3Xn6~xh_C%BZ9ktpzl&b$5tknf46@q=Q1hm(A?g|reOizecfw} ziw@4K`1$Cf{#+s2Sgjbkd_t4Sb?>YmVF!#`V?#s-SoznwR$N=D(_?{qdl%T&SQGPX z=7!bp>NRCc++^?D2e#Ia3-lM7pt3jhh~o?vgK2br;noS#{aN|;y3;bYFJ5JsvPkvQ zd6mOu-nE~nKj_uMCKvQPE(TN36S>%%>0!AuWxMR@;_by+iS@S4eNXwX?a_E3zT!cQ z*W8VxV#tqh_lxhVmVfiokrOpvxoS@Dlv~Ni8T8jb<>y|SY~-q{Ud_1;=xlQ}&D0Lw zRU>bq?I}_qR^{1ft~xGxN@dq*(TB6iwZzizUVqD2FPUA^UDlawYFar#*44i0`?N8i zic@u88uqQ~)YmVn0|-fRelU~s+PMxuzE=tSRxR)+fsl=eGi%#(laK2ozwbjkVXJYo@J?QKEYG&nKd7F=qV??BZlb_f4*!@R@fT z2!A*E^|bUKmrjRYjM=BsN6+KzlQW}ttzW(Dc>Ob}TNxz`rir4?44pWc2GQGrDJ#zV z?Y6mc+-mMTpKWT@Pi8jGIP#7y9^#bIGdz9k7FzBBku6obOlB~G)9&WirD*nLOfqeL zw~73pM*&ZtVsA!qdSJH&iyv2t?Uk{)d1q9^b>ZbLc@GClT6A6Z#c}O~tqi8Or=3e~ z5-Zf&GS|weN8^OFTHj2QD}n1Ylo#Y&6>WLKxsX)QrtTLLS*|ocNa|wghn6yp)~G#g z)3>xHS?^F-6sum$V2b8E4CzNt8S3Nbl;Thkl@5vpn z0OY)Ot^<%Sj)K1k3c3O?B97ZS2pQ;!A3K%udXDl?nH#ruNIjYqZT&T;kJ^J#uX7nw zr%l$*-aX^?=6F%C<3>77KOf=RTCJX0Q~l)h=Elzj|WI-ltwu*Ql0M26VZl zq{cpVX1ltaUl4F;`=I1(Rim?#iibB|TBK>ay+*GoB>n2H{^YBd^sLD#A4C!%iucM!4a)X%DpU4Wcffn9+8t0W@WlRSY{(9amMIc{+!W9o@Wyy8kG{x@Ule;xC_ z$x!=fk30OYC-!o3mHTZu>Ga&VoV}d;V~P5|pX``4Atorm-B(~+IorWs2tWV=5P$## zAOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5ctm#pxz5l6$X_+ zK??Ox%$Q`Ll541c14qT`9bz&`6?$LJnEx+Bntu*r2tWV=5P$##AOHafKmY;|fB*y_ z009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$##AOHafKmY;|fB*y_009U<00Izz z00bZa0SG_<0uX=z1Oy65acgus|DXF(RNiv~+IPTxsi{i)3`{@(0uX=z1Rwwb2tWV= z5P$##AOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$##AOHaf zKmY;|fB*y_009U<00Izz00bb=S%Gfcik;5?=hUb7`ST7KN`Ib@xe$N=1Rwwb2tWV= z5P$##AOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$##AOHaf zKmY;|fB*y_009U<00Izz00bZa0SNqsKzHsk22F@PMJIo?$e>Jy9%P_md%ifdL(I@M zpS0^%ed_dbeG8=zh6z%(@`i_vSE&0g=>2r5T7#}eFH*T!<};H`z|%> zDsx|kIQy3>-m7@4EvitXFw&ND9KTi(3aLwe7&*RFzC zI1}WrD;b2zP(&sFlOoID#CqJ=l%(jZlKGp-d}>y`2#wa0jAxR8ip|NUsI>pt+;|U( zrO5OSGW;s^98$ZD-o-!A-6J$yH$Xc$&{NMlYGiQm$bc{(Pd_(5f3J~&F(X1pdXp$r zXOwvq^E(~yio5zeLs%?-C(Vlc`Uzx5#io3p`-L7wY2=bqgbI@!67joo*@Py z`oSR~Az|JjQ32r*I$8!H5q^3)5dl70`T?Ha;ZZu_^L&DJLqa1AyaWAM`Uai>F_EDm z`nr)}1|vfCBg01e$Ha_^aElIiV`*uz!u3WP=;;QJ^c)fFujj3$72@yb<1tESZm5=z zmO+eL#K?Ie!5-1x{t*U@?vz}&Cn3Gd2ZyTZ)1rjfGupX+{uCDD26PeL6s-%hy~5v_ z5M|fx`80zjMx9bSaWR_sNMF%$G+I~Y{RJ=fp1!)R>{Gn?ZT~>!s4Z!MhfG%-HE8@Q zm3)pVrzjz8N*ha)q|>C@ci4?i;~W>aE6|5ir~KE!SLlDg7WG#L&-f<}UWyz%w{&HS zjH*>()PA2`lJdnaKHE!i-2zwV>3;osC)ciz_8vdqo;vs?zYd-fpbma9eIbomHN5t+ zu62^t1R0TT1Aa*P>WtiB)VES@JWWTprrB)uKsuel;O&aeIT6&Exu>ZkevB3+tSv;N z=W#v~yyKx_(#zWm6_ZZ|uW%Yo9Rn$JLA&R!RY4bH?}uzhh<7wJ&X$Cud)lqbGYcYPT}xA>!J5Gse3PUD{tWUDP5nJwtc3%b~+< z9u03*_k42f_D#@1jg+C+$TF?>GRM5Jj?Qn5wv3iei(S4>8z=VuPHe}< zsYv^s70|dYCEc?9jV1T&6rQosI`ZS7MRjRuHFqUXn`{3tc|S<q-pb zrp64O+w{F^)7sJ=+QYu^ojKAY_#4N6<_1YCF$FYkqE5G0X4NlEZr;h?FgIeKO`kpX zV-+*K_l9|uzqq|%^b0|a^Ycp?E6aT0C>OLKZ*ax8wRWT4+h5<*w=MeF-c%2VBrl`T zt{IHYy_vcBKV~d;8h1A__Q~0|M~;>Ddbzi)cD19?kld%+D?;BN_YIsEZQe(Rn)xV}*zNvwq)K$I6@) ze$AooWz@D+sRf)uEfn~93wDbRU(2%tWdHy4P-uF~`CpvTbw=0d6@>YAiROP6> zpg;fu5P$##AOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5P$## zAOHafKmY;|fB*y_009U<00Izz00bcLUm!r`?Nf!pO~a%TCaFA1aWY&XIgP=g5rHgC z{~&J?|FJM=((Ut^WNj)3lF6Mjf-kn_i)Zu2!F=&jzBrXH&f<#;`QmcE_%&broiFY| z(oZ`czbap>&llVB#jbpD_-`@&bai5m(9`GZcVAt@Os>kR`{)w-wQCo;-*YR6H-|_%k5ry5JZUw0OZK-E+J>7!8jm(kCw0!+-HlbdRM{fG zY>?~#gQdmy5(;Q?i`PVUO^-StC9Iv$oz%%GG}SaKJ2m#$l7nYTMSotlj4m>rU^pl! zSaxNurKI;+7O9h4S1h@DtL6KMf^Nft2V}a-E%F)tV!{Ib`BHbj)(^GNm_q90&3P93 zb(rP^`J2`KV{ZqlC2wv^zpJ=@@bV9iUK%y;^JkGd=hXUt+RVuK*|1A*rB_r$mC+)b z{kNF%>HUhlXhKC@oa{-R{Ix}E{LQ@=pA1qQIBnM3oJK2WBMkUjn;be$~@J?hSVwCKE@#`eyrnn=WR13hlan66@9SzcFxk5%b#?S z%o&xiY!Ina6lyv%=5ce~wV^U8`g_dY^~>>=UpFA=&AoMFWs}b9EOjGwE{xgZ^1^DQ z^_0q(Qy%5HuA`6alA{~ipD>>GKJ{Xfmf;*yr`TfO$U3zXPBs-8#RiEhCN0u1kyTPI z5Khl%J(g}PT<$^YTwL&RbLVoNRnaol8G_ z?|Lt>-BB+*sH$kml6z*>MKP~F7Gw;uXU08KG>)Lo`{hmBtu}3v-&B(1YreI9O=-q7 zjnqM9%kCxB>qk$i$P*>c-=LQ~e|d4Fdz5o=^Rs}YX_r0MtasVkP+GL=p4-PfgXnWo z&rUJa={XgaZZ|q$DKkpF&H$0$;q1FFwW>U*wDL^TluZV(RtN9rOQt@x|(Vu`yq4_gjp7 zf8_iBtMmQ;O|L3AEC@gV0uX=z1Rwwb2tWV=5P$##AOHafKmY;|fB*y_009U<00Izz z00bZa0SG_<0uX=z1Rwwb2tWV=5P$##AOHafKmY;|fB*y_009U<;C~_@!(E9#|6i5N z??DEsxNb8o)ozNE#GfB`*r4@j)$D8?@7NEkUs)s!6_@B`swev9d+s}pmC zo<3i{`|28Ia#dE{N0-pAUAxfzo?AJ*IW*$nXiFN6ex~V&y~?)zx6B5oJa@Btr1EUx zNvqLYvcIL!Hrx!-c(idksdLuuZmim+$`<)$gJcI7EG@p5P(YJgye6`1dei|aVeN$O zq)twusis-ksjmK*dpl4qd2?I(UB&f-mw#~d z(x`c#Ka12kr`G?|W=6)(hFy9qy`mzjj27AKzr~bK?^onS6DsQBWKZhkuPs{RZ|=SL zWRT*(X|vuY_i-)EKQKh<%6UhJbL$K551d5moIg0az~{A0+?AD;11`h`Z#Y?H6=|`$ zuJo)=QnhsN4cm1|or1B;4__^e`)RdsKX%wAPzf=BXw&q)y@X zF%D7jV;#RcZ<{GOH2iI>=!4C-bC$kb{-ld!&ZvZCgGimCP}7+)kDKeR4V6jJ-(&W! zUyir@x&c9N?yVaun{-}hsT-+tVay(v7gi&!r&PwA@+i-B9ere%9Np0Vgz>cZsTY&9 z4CjzK#TNTU)~TIvvZ=@@Hb`7CX_1DBtdeqpaC%1Tv29aA*L#WWj(XuiRYgmd+%vN-ih1?1AY+I< zGwzw9aRfQvM}88B_WB)Dr=y{D1O(cbWtl zUXvOOL&mf?-bNbiekw_$q7P};4ZjVow55*~uexm=vWM3fJDb zbnHgXlWhso8(Fi(HXfO~`_jtvQoV_B=?5!x^FM7_dR8a;l2m;C?Gbu=r56gzTYR_K zciP1_^yr5()jw9H4-_#O-DWd#{-I@M*F5Mt?_I7f|2FVz%8J#W*W5hlJmtq%#SO>5 ztI9u=wo?BPb~p3Kq3w60R{1y`G5#i3cqumS#^fJipCtEmy}E`<{?i)IXqga+7YG=7aZ-J@(A-FfBUN zNA=w*b5umOs4t|=O3AQojgYc1-9APkbfg$hie=+nmQ?w2X18q zxCcJ|)>TzPbib74nibxLSMqnHHCYzKJda)XS-1D(wrabk1d%iPKhvkqrMv0mcm||P zPL`@acCc2hcioCJF`;D__XOI8GIhn52F=^C>Y1D0`Zu?R&#GS9-N@QRbM>>>CBvo~ zNQTPQ1U+eryeVPTucy-Ilp}*q{CND*BQ)dUq*>F`RwRXrRj0jm4KcWS%b~ZD<$6o= z^kJ3#O7>pXe{{R&Lz7YVhslJ1agK*-S1CY)r~HwNGnm+1xg%Bi_!( z%}s35PBzYsh#r%7$E5Y>_|GT0o)_!VZy_zSw5ein^M#8~-*4)7b-(F>MiwKd(g?_Zkp-BAQ$0_0}>^Z)rQUkk)hUk^=R2 z!^KuKYFMYIy6mj~R8;Qc8L&Iz)UJI&H}5CS=+ou?^@dk<%M>*ZS#62krc%(#=yTNB z+$J=!*f~CYzwgZ_8e29lJ~FI1scuci4NHwPmNav}M2j`zvggJR%h|Ik<>bt$<@v*e zb*d$qs&1u?FMXEWkFvd~cU^q>-LzX*U22qd!VJ6D9Y~g?sRz6m5c+6gjrQ*S-Z$)T zYn*#T?{+y^-daLiY53jN`1jX~X``#NT?ZC~4W)0M-1We-`4fCbY)%XA7Z@cZFZ5uj z(4h&k$-*bi&rUd&*r&3Gxb~LVc^3{JF`KnuVd0_GOy_P=9|oUJunK+R<>u?J8O90^ z_h$*d^m8uQUXh-^-8N|4)!Ccsa6jpIk;gOyAOHafKmY;|fB*y_009U<00Izz00bZa z0SG_<0uX=z1Rwwb2tWV=5P$##AOHafKmY;|fB*y_009U<00Izz00bZa0SG_<0ucC* z5K!Q@fK1ZFiCDc^*jJ?ZvB-c z|Gztlxgm{;l}Suhr{-(AGOsjF@ND?cOX=;5BAGbhF zkD!16-|zrdU^x4s0^DlcqiNqQcXoTsJ{;;v0Mr6u!)?Kw3IiNDW36)b@pbC40u6n7 z#4-lSavCT{>s2{S>%lPf?;%@ST{CJOGnnaCX;EUWx-IUWhFK%g;;m5W6aeMA2eM3l=jFK?=0)C z-=F=kV3E7-w-yFd)$!!~U0+s>S8R|@%$$~Bn=bsx^=Ev*vKf6glz$)7d(Iz;I7h+} zq*}|1IW7Mcv*-T{h={>mt2qDRX){+#Zf9(;>H79)%*&ezeY50h=hWwaeX8evn8!sl zZ{%HYubOeYp=|UlO$qysPm&%tYMa#MEwRxiw z@0+vews&pIb7hHD!6V+X)fmoEJUwd6LnGU8r(*Ge^QcExkkW;Elm=Dk^yf5T&LoXN z7kV}|mB!j@(KXXn#wbxe`{xtUwivU126k~Wqx&Y;Pxwr&K?YJ&d_x~sFGblK#>;}- z6a3r`YlMw3kuZLLH-B$p)Tg%-vq&xK(Ho>@RbuVZ)wHaO8uv$h(GE8>o)S}#Sap8c zaNmdphvZ}>>Pao?2uRIW0qK{IXIWlNiWN3Xe{{b1)w=qP7B$n9D~-N1i|lNsp3Omh z92j)QqQzrd$7pODeCAkV+V;s!cHQR3-yT{owlFYVU+jasLmR0@N~c;)qYK%HIJ34r zH~F|e^83!zSIhfsOgo^!a^03P_~t1qo%o5KoUM`V3B}I3S$2KzI7h3xf{IId6(d~^ z8JHyaH@JpOElc(GvoHTuGxz?`mphCSjZ#`&`zX~k#>~1jd&|n34>Sx)CKL_$@Xo(O zP3aV)YcY$yJo}j+(R%7xt@@W2Rco847JrD}ViF!>fAUw&qoR4!hrJzj-7t5lo$%7_ zA4DgK1{U6#I@(lZYh6uSurHllli1xxh(S%Uix7KCEL%iNxP6{SP*AX@hksC*Z=knk zcoe(aQkCiSPIbUuc}X_VJRIl8I5R$HewtdvaU;M}YnDzEc=Xk<#&ugkd3xr|jQ;33Sa%a7aJ5 z&XgmRu)kYa zxMrkVm}V$zUXUM)9?A*`3MUUB2n_S}4rF<1`UiQq`SYGfz`2i5H9g$IeS-qYQg3P_ ze<}?2Cxf$uNk-JU9?1!56S*<>-ol|Eo{}q@McFt8jKT{UBUYpRylznG3 zoZmC%wG`-=DXX=76JO#b;yqn1Pv@R+pxSA3IMSqUee8QoJTziaV5($qdBUP!wr!60y3y z6b~11O)-|93n1Q~YvSeY6Ql@Ky*yVV%+2$U*Xc@a|7uV_i$JQAo<$(d=cwux?b0{W zpdBX*E6bA!3R@Sxo&EX9n*P&Hwz%_UiG8OJ4p&#z&!3XEeT|3HQm-z7y4&QdrS~Dr zG+XaZO{tu9ulw&QN8=WJy3eq+n3oMOD6Aq0swbNx$@`es89K@M<2ArC2(_J|ew($0etH z-`0EdV@$c|_>s4hJZ@iqyeLcS(q5lTz1wMa$IczA(PN;lspEx|zG+JbpEDmZV8)An z9h|l;SW|8Gy4N-v4&Lq1wc77DFWAN=S)Um5(6LkR5ms09dVbsN_N2xu{nZJl>fY|O z+)Z!i)E>vHdpNi7KU#Nh>UM|LowS>-sIhzc>g(QXTHIfn*XB&z=lpIn_kGH15U?+8 z>*yg(w{DuHZ+Wr#tk8j@_Bu6PnLpTgW5lIqeH)y%j84C|_hI{PQx?n8rp{Vmv#()& zn}`#xj;}hTu345aZESpk!@ToHbQ@hv7-o>=(Rc2{(*Y^%rcGT@yTbc=H(#F^bnV3m zBmFTMbsN+^{76w@`>}0~v-F=m@|Yj+IJv87t!rI>8{hHKpbIq;uAbL#K7C2{)hrF& zqbCRN42pmAvf;!{L4#d7SIg=-tE=UPRo2IYS~(BPw%&0($n$k~+w6B2>&JICF!lYd z!E_Jn>l?1Se{N;uTl>VsCa3TP9d zYZF!fYps9|+hn!o45}O)-F%mcvV)CGIdyH*PLmUdxQy?U-zU+%*2Iyi#<^EaG73_| z&GnytH^%aLLUdLCQxU(sO>5Gkv0?XZM?;;K>c6#LXE&n$*0{uwmnXNqGYr0!QeZpG zdBca{86gY9-!Bcn9CK`<#?=?ct4=ZAuu~q=qq4)@4(+RMGJYf<*KP6En0Y5!xee>Q z+aTiUte$(iI4kd%jddCj8JpBa>pQcgoO6%1UZ{Gy(YQY8!OBV#ew}3>GiOJqtNN!l zEs^^jv#aRo+o?vhRcOrX3t^LPKW_6TYUX+`pVLF{#|#|t#pi)tdhB!a(EOeS(_UWd zndY`+jy(H%)`9-l_D>U{F6q*?& z3+&x5+sY{?rXc!AfUWXMg6HG2r^h!wW^rdlcY|J|f1MY0x~kRG(`|j--#xB1d+4O; zn?pa%-);EJVC2GgjUQ`yTla4=$oI^Mp4-j^+Bd0|esyz|`f(N+dEZ;U{Pwkr()et@ zsWUsz^si;Htzy>B&p&uhx{=cP_Qdj!du1Mr)@}Z^myKq^0B47-(LMub1^BhT^6B|9 zyZnA9_t_Z41sm0_t#L3;V?>T3yw|qbw%-j&&G!o5)^@q#VdAYt{YS@tzP>NX>ian{ zZ*(dR%3J-jg-=xa%ri!_0}pQ>sg)KxVb9m-LEer}ZuWj&={w&&s~;x3&0e-7W9#Sv zmlt2@r$5tfM6EV^vm-X%&pj$$_1-spi*B9xdmSe)=rcNc(*m;_jVJD{w$jzBhCxX9 zk1eCCnxB7Q;~)6qi}j};r)ziK>0f30vLOc|*4iBI<6;sXCa=BUai*eoqlBN=kH7wd zlSj;khE4NzQX*}>n8uiVc=NjZf=_AL4esps%duS5^nSwyy(*NCo8e#JceZD{&^xD<-%cwrgR#R_ZUP{9^dZTipEnUh# z2|V-cobKHn$2=^ny}aY)=IvvmR0IY2DAaFKSD%!Zw>C9C6ub9>*Y*9gY&(2-*l1>^7x@vBBGW^DnPKmO~ zy;S-C@^|gOrNqogZ)a<*>lkPGHEPtg&U533RQMs++U<6Bv$yx`X0*xm`>A5ZUIRWo zeqgd<&p8)oZ##$Q84jmMbjURPcyL+LL&KV9W`9yP%-Z1W@gk_!M{~P+lUp`1)ty-* zsgh>SRDb&&nw^*L?N#3YmY)C7*yw|+w07nGX1SzKhJL<9#EglPPmh~#u%mCo&x?X~ z4AUz()q39Is?|pT+S7t7W>#)L^?5yQ^YfREA43z~zV1|eVN~6uKD%{ZmN!ZJ zZF6R1b%Wde8dd459M>aS=ga}eysyc&yY?7cYR5!xj!UYMaqrBXb2NiM5(`iyHF%?!7e@ z_m61V_?K`!=VSTChbs?QdN&|+;-shh&OVwmqNhRAIZHG8*?qPSe%GqY%+<4+4%&Zr zc=N|@dlHplb-FE?@_g%B%lm^@WNxmprr-9y6YIOy%XPfGW9#bedwS?99WC!iwLiP+ zvv+>2l;&-_eizYbOI}Ztw_|)*y8`E}w)XWp>JRJi>RG!@t&>>H#4KP3J^gj+x)4!Bc=+hmd6RW_ zIgCDUD7ToY{1|IG+qp_dr^QVow#tsybJg++S~Ym%i}azFm*GNEW^h3_TMwg05p$|K{-!A#Iz;b!)|ARd8bPZq23K4VL&ZhSZmnvK z3|#llHtlfp7_;8J*GHTip6N5R;?p*dUf%MsI-F5s`h*PG*Pt5##@h#fFbSLaYT)2E zTd%J*cTmJvxpKGD4*Lbat-bRxu$jX^uS2hEX5>HkWqZKVd6g17UT(LugICmxkH31g zwHxO>D(=weALqWkJ7bPVR^RjoM^fdX0f$V#*wolN%H@IOnlm9oEmyZ(Ut!nEclCC~ zn#Y?i46to%6S`&9t(l8{(cAP$)}q$^vD;G*FX*^BBVp{R2Y#Pw`inD^mk|djzMhlP#%s_vUeTt$BH~ zZLK_Rn|)}Pe`f!ScgI%@AGkTotfFO>ysp#8={Kh}t$g^##wMFP9v{;~K2(3|#iaIz z=0`lv_Q?#*GB|Yg@^Let)ycoEYxt~3!)`;KwW)5kYL?!|Hul;Jbey*sSsqH=-00*0 z>sl9rr-ptP^QL8E+d99d+0RSQ_*sq zjmu9t$#tw7ca-lwQDIo6>`HsSGjkVdBnsL2aY*(Mw)dDL1QYCG2%j9Nxda3RRkTpjn-AqrF>5JBmcknwTRCMAbzw1R54KlFPJIpQQZ#?xGlpVk3$auPza1 zmWX?cM(XzR_Py-3Tz|7>=`9S4X;e>s$GG1DAwN{i9bto+GAYiYm?`Pl?hbNlXzwOy_Zthana z?w#-F$g}1tw0iyJGQ^iS9 z8Wk;FF4I(flJfmcMUnFPr8Imgb|?`C7K=;G_irp37avN^+%H^@e@#+qa(s<{wdE56 z2q1s}0tg_000IagfB*srAbJ+YTs+d{v<$n=zB~d?IY>aCA#L5(3o{xxiMZNlY2a`&~VI|@PBCc7i z{bmu@6|tcx()_Q8xO%ZTQ^e&(te$_4h^6(pQnWttqC%?gEQ&P!TMx+#Qi}+)DGZ9PMXDCX!#(p;Xb@fJE*;mcA=NsoU%`XhH7~Q^Q zqlAM$S$>RMUZv9P%EN7%C2A*xpSIp(p1beis6%FB!?G8p_l|5;L9a@xn8n`?a@E!p zR!i$Dx`FB&WsqBtLaVeT>z`;}F>&T{uNp>LS1&*7^4vMiZc6+@Ex#!Ho#l6hZ}R!u z#fpmm*^*~%8(;O)rQNfQx<3B?xz%vpjZ-@$-CuY4@+ZCb(VJhKS^3{zvYU^$o6^KR z*kghs=E)-cR8(C_v%o?KgIa)H&3vQF5%d@$e4x2=)##ar5+4mTaSrOSbh2@ig)A zAFHk|IWF0Jf?|p?$lp&<`uu(T#g0-2`}(>CdY7@er-^EpmDUmdeSb9-tL$8~nw!M9 zcwAW=Tk5#9{UvRAS=z`nPReBBR9@PFI_>4an9mvWGFL^WObV`NptJ2llP*EO-)v( z_V^j;^UgWi?d?9jL19}BcWJDb*fp#$)=4TH%QQ5eNt;nPXlTlMn1*W9j{InKb!=Q! z<&T!RVH0(l^zq4Xy>n%OS=DLMTr{M%8k$x=nQ6~G75_Exg6G8K_{=*6E&D&&t+Tei?XsduC4fxcJb!^I02jRG9xm-6}5C z>uk(Ee(A{O^)DxD$dt0kGVST24UrappmYEAz4i?3)@t_A6S~WNUo4##@H}iylK#4| zD~|(m3j#IAPJU}}Bl5+?dA2@jw_`$1#QC>9d8%tfa!lo%M>z{UJW4wqh=s_?(ni(& z%9Y;zY|bWaonx?Ih`wzbot75n&Ktiuyy#Z$*|Xq)cWWP1i1ko)zekr_#H~(nx_R5> z{^-_WwL6l(KJsrE(|o=eoot5CopDvU)i?L zArHMO?C@ziXlYdN$kDsrU4GH-v2S79!tPhWcv98ExSVtamxkiqPeVgK#Bor&LtBU4 zSlu99-XLZ3o}9BAwtg(Ic^Z?qX4HY8E?uQ5R8wgh-$?M=*I&La-1o|>36onkdi!H1 z^HpaTRd@biemw1CymUt+bw3Ra&D+&ZY(BKR-O=s~Epl$%JiIlqgZ*+(8Im7Ga*>}ub9c!7^Wt(gFM?0R3_$^g(kh-5}nP9%m z+19ZC?kCANUJabk*4Bwo#m6&E~hnmgIeD)Fbv}A5;4VIX`(W_#)eI zK4Ed!xz_di)=8O}>04ga{mQhbi*>*5@lEQ!U%lRF_xprrR#hf7+MxY0x#F>V{VXjv z*K|*bh>|OhC?gK(w*0)%_^76QmW|i@DCcQbBNr|Y8I_kDIM3)%X;&Mub+WRwQFT9Y zRxR!9cmAiw1NQ$~@22}u51-_Ub==Kg`1ZTHSiAn{+6!7}id%x`|^+hd*l(4Gy?Q?k8GSSW(Z_t8R^JmUI2xBPY1+X|3#F zrDyr!R{Zw)lRv#37$;gu-A}Z9I6q^~B3azU*7rMnZWm->IXLvp{D-IKw)GC4w!3Bx zy*Hwj)cr)ubl;7a4#e49UK*y=bK`^47t)u#S=l?)(db^6&so}=v!rXS)cxcd@)e+kzc zFy?aS3EdT%fev%$zrR1;OCwzAvEi;dGN{+bz6E!cLSa#N)K;O zu_1pssLt=A#>3yw%RA85WSqBO&_CQ|h*MzE;3~$3qNpwuiAdeU8AfSbnzx3Q()nvD%~3m9jj@Eofs zx}^|4g+pPDuOi4zoDu$QD6H}G_f)8NtGarex43@{^!9M`DSor}uf18=&fCvB$lJ}Q zXW>1tO^Bzs1-AG1@%EUao=?fKdO6;Hll;9s6ejhk25~cKey`mdS4RM@bid|a0^GpA*l$`l-SWmYgx5AD7^WTy`lcn>ybaL@@4-kJF`Fbmr z-u`~#JQWn^?_(0EPzL+_)0wJp0e*_05dXjlMa>JlRH?sA)ss#R5=ZM4UxigoK0g97};Ik?o7lz0jUQ!#2Piu#*A zCF0S=;!-<{Xe(~GNMon+>N0dfgIdcoqNa@PH`PHj)wnLs6uSeAqVQQv1G3``8+2S8h^J$=j^srY`mG)$AuTYoD6cbCI!VqM;OB z$FsBxw7L@%KA2VN7HY%3;&buXwOCxHZXxa+%hWGKyE1hQ(M}nnPy{FfO@bzyjB`_t zEB&e%;wG+8eq#dy{r$cE;WkBV8Uq3dAbq)~^4iHHrv z_l8CF_9B-4ldwAW5V16^dV1-30LAnFJBxh&;`#saA}*f)pC;mp#iq|JiXFvw()x50 z3ooAkuTlNq^Zy%*nEe00f965i{wuYB{)?exN`;g!B<+*JXCC}la&^f6|1+;TrGxx` zAt7ZJsGkQw{(n&hEcyTB|C9eu{{KHdyr)bTcqxZdT)j(6FfB_GGWq{fMjZM7g*P;R z+{BRo|Mw3JD*X^8|DXJS^8f!A^Z$+hI{#mt+^qg}M!vsHM!rA!{^EEb-(P)FV2l6) z2q1s}0tg_000IagfB*srAbGMU($&q)tNlQw3ijfB*srAb Date: Tue, 17 Jun 2025 11:52:02 +0200 Subject: [PATCH 4/5] graph/db: thread context through to DeleteLightningNode --- graph/db/graph.go | 6 ++++-- graph/db/graph_test.go | 12 ++++++------ graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 4 +++- graph/db/sql_store.go | 4 ++-- 5 files changed, 16 insertions(+), 12 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index a02e8193f..fb2a9975a 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -286,8 +286,10 @@ func (c *ChannelGraph) AddLightningNode(ctx context.Context, // DeleteLightningNode starts a new database transaction to remove a vertex/node // from the database according to the node's public key. -func (c *ChannelGraph) DeleteLightningNode(nodePub route.Vertex) error { - err := c.V1Store.DeleteLightningNode(nodePub) +func (c *ChannelGraph) DeleteLightningNode(ctx context.Context, + nodePub route.Vertex) error { + + err := c.V1Store.DeleteLightningNode(ctx, nodePub) if err != nil { return err } diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 25409df09..a87d66bfc 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -159,13 +159,14 @@ func TestNodeInsertionAndDeletion(t *testing.T) { // Next, delete the node from the graph, this should purge all data // related to the node. - require.NoError(t, graph.DeleteLightningNode(testPub)) + require.NoError(t, graph.DeleteLightningNode(ctx, testPub)) assertNodeNotInCache(t, graph, testPub) // Attempting to delete the node again should return an error since // the node is no longer known. require.ErrorIs( - t, graph.DeleteLightningNode(testPub), ErrGraphNodeNotFound, + t, graph.DeleteLightningNode(ctx, testPub), + ErrGraphNodeNotFound, ) // Finally, attempt to fetch the node again. This should fail as the @@ -318,7 +319,7 @@ func TestPartialNode(t *testing.T) { // Next, delete the node from the graph, this should purge all data // related to the node. - require.NoError(t, graph.DeleteLightningNode(pubKey1)) + require.NoError(t, graph.DeleteLightningNode(ctx, pubKey1)) assertNodeNotInCache(t, graph, testPub) // Finally, attempt to fetch the node again. This should fail as the @@ -3416,9 +3417,8 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) { // We'll now delete the node from the graph, this should result in it // being removed from the update index as well. - if err := graph.DeleteLightningNode(node1.PubKeyBytes); err != nil { - t.Fatalf("unable to delete node: %v", err) - } + err = graph.DeleteLightningNode(ctx, node1.PubKeyBytes) + require.NoError(t, err) // Now that the node has been deleted, we'll again query the nodes in // the horizon. This time we should have no nodes at all. diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 9f5eb6c8f..f45c663f5 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -114,7 +114,7 @@ type V1Store interface { //nolint:interfacebloat // DeleteLightningNode starts a new database transaction to remove a // vertex/node from the database according to the node's public key. - DeleteLightningNode(nodePub route.Vertex) error + DeleteLightningNode(ctx context.Context, nodePub route.Vertex) error // NodeUpdatesInHorizon returns all the known lightning node which have // an update timestamp within the passed range. This method can be used diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index be8ab2d9d..0ce190420 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -1032,7 +1032,9 @@ func (c *KVStore) LookupAlias(pub *btcec.PublicKey) (string, error) { // DeleteLightningNode starts a new database transaction to remove a vertex/node // from the database according to the node's public key. -func (c *KVStore) DeleteLightningNode(nodePub route.Vertex) error { +func (c *KVStore) DeleteLightningNode(_ context.Context, + nodePub route.Vertex) error { + // TODO(roasbeef): ensure dangling edges are removed... return kvdb.Update(c.db, func(tx kvdb.RwTx) error { nodes := tx.ReadWriteBucket(nodeBucket) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 1fd68a7f9..a3be9c8a6 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -291,8 +291,8 @@ func (s *SQLStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, // from the database according to the node's public key. // // NOTE: part of the V1Store interface. -func (s *SQLStore) DeleteLightningNode(pubKey route.Vertex) error { - ctx := context.TODO() +func (s *SQLStore) DeleteLightningNode(ctx context.Context, + pubKey route.Vertex) error { err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { res, err := db.DeleteNodeByPubKey( From e724e1c3e4464048186b5c6daf9db973c171a02b Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 17 Jun 2025 12:06:03 +0200 Subject: [PATCH 5/5] multi: thread context through to AddrsForNode --- chanbackup/backup.go | 16 ++++++++++------ chanbackup/backup_test.go | 15 +++++++++------ chanbackup/pubsub.go | 11 +++++++---- chanbackup/pubsub_test.go | 19 +++++++++++-------- channel_notifier.go | 6 ++++-- channeldb/addr_source.go | 10 ++++++---- channeldb/addr_source_test.go | 26 ++++++++++++++------------ channeldb/db.go | 5 ++++- channeldb/db_test.go | 6 ++++-- graph/db/graph_test.go | 6 +++--- graph/db/interfaces.go | 3 ++- graph/db/kv_store.go | 6 ++---- graph/db/sql_store.go | 6 ++---- rpcserver.go | 7 ++++--- server.go | 10 ++++++---- 15 files changed, 88 insertions(+), 64 deletions(-) diff --git a/chanbackup/backup.go b/chanbackup/backup.go index afffe5a2e..cf7217ae3 100644 --- a/chanbackup/backup.go +++ b/chanbackup/backup.go @@ -1,6 +1,7 @@ package chanbackup import ( + "context" "fmt" "github.com/btcsuite/btcd/wire" @@ -24,7 +25,7 @@ type LiveChannelSource interface { // passed open channel. The backup includes all information required to restore // the channel, as well as addressing information so we can find the peer and // reconnect to them to initiate the protocol. -func assembleChanBackup(addrSource channeldb.AddrSource, +func assembleChanBackup(ctx context.Context, addrSource channeldb.AddrSource, openChan *channeldb.OpenChannel) (*Single, error) { log.Debugf("Crafting backup for ChannelPoint(%v)", @@ -32,7 +33,9 @@ func assembleChanBackup(addrSource channeldb.AddrSource, // First, we'll query the channel source to obtain all the addresses // that are associated with the peer for this channel. - known, nodeAddrs, err := addrSource.AddrsForNode(openChan.IdentityPub) + known, nodeAddrs, err := addrSource.AddrsForNode( + ctx, openChan.IdentityPub, + ) if err != nil { return nil, err } @@ -90,7 +93,8 @@ func buildCloseTxInputs( // FetchBackupForChan attempts to create a plaintext static channel backup for // the target channel identified by its channel point. If we're unable to find // the target channel, then an error will be returned. -func FetchBackupForChan(chanPoint wire.OutPoint, chanSource LiveChannelSource, +func FetchBackupForChan(ctx context.Context, chanPoint wire.OutPoint, + chanSource LiveChannelSource, addrSource channeldb.AddrSource) (*Single, error) { // First, we'll query the channel source to see if the channel is known @@ -104,7 +108,7 @@ func FetchBackupForChan(chanPoint wire.OutPoint, chanSource LiveChannelSource, // Once we have the target channel, we can assemble the backup using // the source to obtain any extra information that we may need. - staticChanBackup, err := assembleChanBackup(addrSource, targetChan) + staticChanBackup, err := assembleChanBackup(ctx, addrSource, targetChan) if err != nil { return nil, fmt.Errorf("unable to create chan backup: %w", err) } @@ -114,7 +118,7 @@ func FetchBackupForChan(chanPoint wire.OutPoint, chanSource LiveChannelSource, // FetchStaticChanBackups will return a plaintext static channel back up for // all known active/open channels within the passed channel source. -func FetchStaticChanBackups(chanSource LiveChannelSource, +func FetchStaticChanBackups(ctx context.Context, chanSource LiveChannelSource, addrSource channeldb.AddrSource) ([]Single, error) { // First, we'll query the backup source for information concerning all @@ -129,7 +133,7 @@ func FetchStaticChanBackups(chanSource LiveChannelSource, // channel. staticChanBackups := make([]Single, 0, len(openChans)) for _, openChan := range openChans { - chanBackup, err := assembleChanBackup(addrSource, openChan) + chanBackup, err := assembleChanBackup(ctx, addrSource, openChan) if err != nil { return nil, err } diff --git a/chanbackup/backup_test.go b/chanbackup/backup_test.go index d36fa385b..d73b3769f 100644 --- a/chanbackup/backup_test.go +++ b/chanbackup/backup_test.go @@ -1,6 +1,7 @@ package chanbackup import ( + "context" "fmt" "net" "testing" @@ -61,8 +62,8 @@ func (m *mockChannelSource) addAddrsForNode(nodePub *btcec.PublicKey, addrs []ne m.addrs[nodeKey] = addrs } -func (m *mockChannelSource) AddrsForNode(nodePub *btcec.PublicKey) (bool, - []net.Addr, error) { +func (m *mockChannelSource) AddrsForNode(_ context.Context, + nodePub *btcec.PublicKey) (bool, []net.Addr, error) { if m.failQuery { return false, nil, fmt.Errorf("fail") @@ -120,7 +121,8 @@ func TestFetchBackupForChan(t *testing.T) { } for i, testCase := range testCases { _, err := FetchBackupForChan( - testCase.chanPoint, chanSource, chanSource, + context.Background(), testCase.chanPoint, chanSource, + chanSource, ) switch { // If this is a valid test case, and we failed, then we'll @@ -141,6 +143,7 @@ func TestFetchBackupForChan(t *testing.T) { // channel source for all channels and construct a Single for each channel. func TestFetchStaticChanBackups(t *testing.T) { t.Parallel() + ctx := context.Background() // First, we'll make the set of channels that we want to seed the // channel source with. Both channels will be fully populated in the @@ -162,7 +165,7 @@ func TestFetchStaticChanBackups(t *testing.T) { // With the channel source populated, we'll now attempt to create a set // of backups for all the channels. This should succeed, as all items // are populated within the channel source. - backups, err := FetchStaticChanBackups(chanSource, chanSource) + backups, err := FetchStaticChanBackups(ctx, chanSource, chanSource) require.NoError(t, err, "unable to create chan back ups") if len(backups) != numChans { @@ -177,7 +180,7 @@ func TestFetchStaticChanBackups(t *testing.T) { copy(n[:], randomChan2.IdentityPub.SerializeCompressed()) delete(chanSource.addrs, n) - _, err = FetchStaticChanBackups(chanSource, chanSource) + _, err = FetchStaticChanBackups(ctx, chanSource, chanSource) if err == nil { t.Fatalf("query with incomplete information should fail") } @@ -186,7 +189,7 @@ func TestFetchStaticChanBackups(t *testing.T) { // source at all, then we'll fail as well. chanSource = newMockChannelSource() chanSource.failQuery = true - _, err = FetchStaticChanBackups(chanSource, chanSource) + _, err = FetchStaticChanBackups(ctx, chanSource, chanSource) if err == nil { t.Fatalf("query should fail") } diff --git a/chanbackup/pubsub.go b/chanbackup/pubsub.go index 8fa1d5f34..43c368aaa 100644 --- a/chanbackup/pubsub.go +++ b/chanbackup/pubsub.go @@ -2,6 +2,7 @@ package chanbackup import ( "bytes" + "context" "fmt" "net" "os" @@ -81,7 +82,8 @@ type ChannelNotifier interface { // synchronization point to ensure that the chanbackup.SubSwapper does // not miss any channel open or close events in the period between when // it's created, and when it requests the channel subscription. - SubscribeChans(map[wire.OutPoint]struct{}) (*ChannelSubscription, error) + SubscribeChans(context.Context, + map[wire.OutPoint]struct{}) (*ChannelSubscription, error) } // SubSwapper subscribes to new updates to the open channel state, and then @@ -119,8 +121,9 @@ type SubSwapper struct { // set of channels, and the required interfaces to be notified of new channel // updates, pack a multi backup, and swap the current best backup from its // storage location. -func NewSubSwapper(startingChans []Single, chanNotifier ChannelNotifier, - keyRing keychain.KeyRing, backupSwapper Swapper) (*SubSwapper, error) { +func NewSubSwapper(ctx context.Context, startingChans []Single, + chanNotifier ChannelNotifier, keyRing keychain.KeyRing, + backupSwapper Swapper) (*SubSwapper, error) { // First, we'll subscribe to the latest set of channel updates given // the set of channels we already know of. @@ -128,7 +131,7 @@ func NewSubSwapper(startingChans []Single, chanNotifier ChannelNotifier, for _, chanBackup := range startingChans { knownChans[chanBackup.FundingOutpoint] = struct{}{} } - chanEvents, err := chanNotifier.SubscribeChans(knownChans) + chanEvents, err := chanNotifier.SubscribeChans(ctx, knownChans) if err != nil { return nil, err } diff --git a/chanbackup/pubsub_test.go b/chanbackup/pubsub_test.go index 32694e5a7..49cfd3fb1 100644 --- a/chanbackup/pubsub_test.go +++ b/chanbackup/pubsub_test.go @@ -1,6 +1,7 @@ package chanbackup import ( + "context" "fmt" "testing" "time" @@ -62,8 +63,8 @@ func newMockChannelNotifier() *mockChannelNotifier { } } -func (m *mockChannelNotifier) SubscribeChans(chans map[wire.OutPoint]struct{}) ( - *ChannelSubscription, error) { +func (m *mockChannelNotifier) SubscribeChans(_ context.Context, + _ map[wire.OutPoint]struct{}) (*ChannelSubscription, error) { if m.fail { return nil, fmt.Errorf("fail") @@ -80,6 +81,7 @@ func (m *mockChannelNotifier) SubscribeChans(chans map[wire.OutPoint]struct{}) ( // channel subscription, then the entire sub-swapper will fail to start. func TestNewSubSwapperSubscribeFail(t *testing.T) { t.Parallel() + ctx := context.Background() keyRing := &lnencrypt.MockKeyRing{} @@ -88,10 +90,8 @@ func TestNewSubSwapperSubscribeFail(t *testing.T) { fail: true, } - _, err := NewSubSwapper(nil, &chanNotifier, keyRing, &swapper) - if err == nil { - t.Fatalf("expected fail due to lack of subscription") - } + _, err := NewSubSwapper(ctx, nil, &chanNotifier, keyRing, &swapper) + require.Error(t, err) } func assertExpectedBackupSwap(t *testing.T, swapper *mockSwapper, @@ -158,7 +158,9 @@ func TestSubSwapperIdempotentStartStop(t *testing.T) { var chanNotifier mockChannelNotifier swapper := newMockSwapper(keyRing) - subSwapper, err := NewSubSwapper(nil, &chanNotifier, keyRing, swapper) + subSwapper, err := NewSubSwapper( + context.Background(), nil, &chanNotifier, keyRing, swapper, + ) require.NoError(t, err, "unable to init subSwapper") if err := subSwapper.Start(); err != nil { @@ -224,7 +226,8 @@ func TestSubSwapperUpdater(t *testing.T) { // With our channel set created, we'll make a fresh sub swapper // instance to begin our test. subSwapper, err := NewSubSwapper( - initialChanSet, chanNotifier, keyRing, swapper, + context.Background(), initialChanSet, chanNotifier, keyRing, + swapper, ) require.NoError(t, err, "unable to make swapper") if err := subSwapper.Start(); err != nil { diff --git a/channel_notifier.go b/channel_notifier.go index 88a05ac4c..8affd48f0 100644 --- a/channel_notifier.go +++ b/channel_notifier.go @@ -1,6 +1,7 @@ package lnd import ( + "context" "fmt" "github.com/btcsuite/btcd/wire" @@ -31,7 +32,8 @@ type channelNotifier struct { // the channel subscription. // // NOTE: This is part of the chanbackup.ChannelNotifier interface. -func (c *channelNotifier) SubscribeChans(startingChans map[wire.OutPoint]struct{}) ( +func (c *channelNotifier) SubscribeChans(ctx context.Context, + startingChans map[wire.OutPoint]struct{}) ( *chanbackup.ChannelSubscription, error) { ltndLog.Infof("Channel backup proxy channel notifier starting") @@ -46,7 +48,7 @@ func (c *channelNotifier) SubscribeChans(startingChans map[wire.OutPoint]struct{ // confirmed channels. sendChanOpenUpdate := func(newOrPendingChan *channeldb.OpenChannel) { _, nodeAddrs, err := c.addrs.AddrsForNode( - newOrPendingChan.IdentityPub, + ctx, newOrPendingChan.IdentityPub, ) if err != nil { pub := newOrPendingChan.IdentityPub diff --git a/channeldb/addr_source.go b/channeldb/addr_source.go index de933ed49..99dc7f7e3 100644 --- a/channeldb/addr_source.go +++ b/channeldb/addr_source.go @@ -1,6 +1,7 @@ package channeldb import ( + "context" "errors" "net" @@ -13,7 +14,8 @@ type AddrSource interface { // AddrsForNode returns all known addresses for the target node public // key. The returned boolean must indicate if the given node is unknown // to the backing source. - AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, error) + AddrsForNode(ctx context.Context, + nodePub *btcec.PublicKey) (bool, []net.Addr, error) } // multiAddrSource is an implementation of AddrSource which gathers all the @@ -38,8 +40,8 @@ func NewMultiAddrSource(sources ...AddrSource) AddrSource { // node. // // NOTE: this implements the AddrSource interface. -func (c *multiAddrSource) AddrsForNode(nodePub *btcec.PublicKey) (bool, - []net.Addr, error) { +func (c *multiAddrSource) AddrsForNode(ctx context.Context, + nodePub *btcec.PublicKey) (bool, []net.Addr, error) { if len(c.sources) == 0 { return false, nil, errors.New("no address sources") @@ -55,7 +57,7 @@ func (c *multiAddrSource) AddrsForNode(nodePub *btcec.PublicKey) (bool, // Iterate over all the address sources and query each one for the // addresses it has for the node in question. for _, src := range c.sources { - isKnown, addrs, err := src.AddrsForNode(nodePub) + isKnown, addrs, err := src.AddrsForNode(ctx, nodePub) if err != nil { return false, nil, err } diff --git a/channeldb/addr_source_test.go b/channeldb/addr_source_test.go index 85ee30bf5..7562a2dd1 100644 --- a/channeldb/addr_source_test.go +++ b/channeldb/addr_source_test.go @@ -1,6 +1,7 @@ package channeldb import ( + "context" "net" "testing" @@ -19,6 +20,7 @@ var ( // deduplicates the results of a set of AddrSource implementations. func TestMultiAddrSource(t *testing.T) { t.Parallel() + ctx := context.Background() var pk1 = newTestPubKey(t) @@ -35,12 +37,12 @@ func TestMultiAddrSource(t *testing.T) { }) // Let source 1 know of 2 addresses (addr 1 and 2) for node 1. - src1.On("AddrsForNode", pk1).Return( + src1.On("AddrsForNode", ctx, pk1).Return( true, []net.Addr{addr1, addr2}, nil, ).Once() // Let source 2 know of 2 addresses (addr 2 and 3) for node 1. - src2.On("AddrsForNode", pk1).Return( + src2.On("AddrsForNode", ctx, pk1).Return( true, []net.Addr{addr2, addr3}, nil, []net.Addr{addr2, addr3}, nil, ).Once() @@ -51,7 +53,7 @@ func TestMultiAddrSource(t *testing.T) { // Query it for the addresses known for node 1. The results // should contain addr 1, 2 and 3. - known, addrs, err := multiSrc.AddrsForNode(pk1) + known, addrs, err := multiSrc.AddrsForNode(ctx, pk1) require.NoError(t, err) require.True(t, known) require.ElementsMatch(t, addrs, []net.Addr{addr1, addr2, addr3}) @@ -70,10 +72,10 @@ func TestMultiAddrSource(t *testing.T) { }) // Let source 1 know of address 1 for node 1. - src1.On("AddrsForNode", pk1).Return( + src1.On("AddrsForNode", ctx, pk1).Return( true, []net.Addr{addr1}, nil, ).Once() - src2.On("AddrsForNode", pk1).Return(false, nil, nil).Once() + src2.On("AddrsForNode", ctx, pk1).Return(false, nil, nil).Once() // Create a multi-addr source that consists of both source 1 // and 2. @@ -81,7 +83,7 @@ func TestMultiAddrSource(t *testing.T) { // Query it for the addresses known for node 1. The results // should contain addr 1. - known, addrs, err := multiSrc.AddrsForNode(pk1) + known, addrs, err := multiSrc.AddrsForNode(ctx, pk1) require.NoError(t, err) require.True(t, known) require.ElementsMatch(t, addrs, []net.Addr{addr1}) @@ -103,13 +105,13 @@ func TestMultiAddrSource(t *testing.T) { // and 2. Neither source known of node 1. multiSrc := NewMultiAddrSource(src1, src2) - src1.On("AddrsForNode", pk1).Return(false, nil, nil).Once() - src2.On("AddrsForNode", pk1).Return(false, nil, nil).Once() + src1.On("AddrsForNode", ctx, pk1).Return(false, nil, nil).Once() + src2.On("AddrsForNode", ctx, pk1).Return(false, nil, nil).Once() // Query it for the addresses known for node 1. It should return // false to indicate that the node is unknown to all backing // sources. - known, addrs, err := multiSrc.AddrsForNode(pk1) + known, addrs, err := multiSrc.AddrsForNode(ctx, pk1) require.NoError(t, err) require.False(t, known) require.Empty(t, addrs) @@ -127,10 +129,10 @@ func newMockAddrSource(t *testing.T) *mockAddrSource { return &mockAddrSource{t: t} } -func (m *mockAddrSource) AddrsForNode(pub *btcec.PublicKey) (bool, []net.Addr, - error) { +func (m *mockAddrSource) AddrsForNode(ctx context.Context, + pub *btcec.PublicKey) (bool, []net.Addr, error) { - args := m.Called(pub) + args := m.Called(ctx, pub) if args.Get(1) == nil { return args.Bool(0), nil, args.Error(2) } diff --git a/channeldb/db.go b/channeldb/db.go index b617d1008..90530459e 100644 --- a/channeldb/db.go +++ b/channeldb/db.go @@ -2,6 +2,7 @@ package channeldb import ( "bytes" + "context" "encoding/binary" "fmt" "net" @@ -1532,7 +1533,9 @@ func (c *ChannelStateDB) RestoreChannelShells(channelShells ...*ChannelShell) er // unknown to the channel DB or not. // // NOTE: this is part of the AddrSource interface. -func (d *DB) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, error) { +func (d *DB) AddrsForNode(_ context.Context, nodePub *btcec.PublicKey) (bool, + []net.Addr, error) { + linkNode, err := d.channelStateDB.linkNodeDB.FetchLinkNode(nodePub) // Only if the error is something other than ErrNodeNotFound do we // return it. diff --git a/channeldb/db_test.go b/channeldb/db_test.go index 9c8c34d58..b010d1b65 100644 --- a/channeldb/db_test.go +++ b/channeldb/db_test.go @@ -1,6 +1,7 @@ package channeldb import ( + "context" "image/color" "math" "math/rand" @@ -180,6 +181,7 @@ func TestFetchClosedChannelForID(t *testing.T) { // channel db and graph db. func TestMultiSourceAddrsForNode(t *testing.T) { t.Parallel() + ctx := context.Background() fullDB, err := MakeTestDB(t) require.NoError(t, err, "unable to make test database") @@ -194,7 +196,7 @@ func TestMultiSourceAddrsForNode(t *testing.T) { testNode := createTestVertex(t) nodePub, err := testNode.PubKey() require.NoError(t, err) - graph.On("AddrsForNode", nodePub).Return( + graph.On("AddrsForNode", ctx, nodePub).Return( true, []net.Addr{testAddr}, nil, ).Once() @@ -212,7 +214,7 @@ func TestMultiSourceAddrsForNode(t *testing.T) { // Now that we've created a link node, as well as a vertex for the // node, we'll query for all its addresses. - known, nodeAddrs, err := addrSource.AddrsForNode(nodePub) + known, nodeAddrs, err := addrSource.AddrsForNode(ctx, nodePub) require.NoError(t, err, "unable to obtain node addrs") require.True(t, known) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index a87d66bfc..b279f5476 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -183,7 +183,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { // Initially, the node is unknown to the graph and there should be no // addresses for it. - known, addrs, err := graph.AddrsForNode(pub) + known, addrs, err := graph.AddrsForNode(ctx, pub) require.NoError(t, err) require.False(t, known) require.Empty(t, addrs) @@ -197,7 +197,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { require.NoError(t, err) require.Empty(t, dbNode.Addresses) - known, addrs, err = graph.AddrsForNode(pub) + known, addrs, err = graph.AddrsForNode(ctx, pub) require.NoError(t, err) require.True(t, known) require.Empty(t, addrs) @@ -224,7 +224,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { require.NoError(t, err) require.Equal(t, expAddrs, dbNode.Addresses) - known, addrs, err = graph.AddrsForNode(pub) + known, addrs, err = graph.AddrsForNode(ctx, pub) require.NoError(t, err) require.True(t, known) require.EqualValues(t, expAddrs, addrs) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index f45c663f5..c1b3dd578 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -63,7 +63,8 @@ type V1Store interface { //nolint:interfacebloat // AddrsForNode returns all known addresses for the target node public // key that the graph DB is aware of. The returned boolean indicates if // the given node is unknown to the graph DB or not. - AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, error) + AddrsForNode(ctx context.Context, + nodePub *btcec.PublicKey) (bool, []net.Addr, error) // ForEachSourceNodeChannel iterates through all channels of the source // node, executing the passed callback on each. The call-back is diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 0ce190420..d722e4fe0 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -376,10 +376,8 @@ func initKVStore(db kvdb.Backend) error { // unknown to the graph DB or not. // // NOTE: this is part of the channeldb.AddrSource interface. -func (c *KVStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, - error) { - - ctx := context.TODO() +func (c *KVStore) AddrsForNode(ctx context.Context, + nodePub *btcec.PublicKey) (bool, []net.Addr, error) { pubKey, err := route.NewVertexFromBytes(nodePub.SerializeCompressed()) if err != nil { diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index a3be9c8a6..1c3c1d116 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -258,10 +258,8 @@ func (s *SQLStore) HasLightningNode(ctx context.Context, // given node is unknown to the graph DB or not. // // NOTE: part of the V1Store interface. -func (s *SQLStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, - error) { - - ctx := context.TODO() +func (s *SQLStore) AddrsForNode(ctx context.Context, + nodePub *btcec.PublicKey) (bool, []net.Addr, error) { var ( addresses []net.Addr diff --git a/rpcserver.go b/rpcserver.go index 63ccfcbd9..3c72d3459 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -8136,7 +8136,7 @@ func (r *rpcServer) ExportChannelBackup(ctx context.Context, // the database. If this channel has been closed, or the outpoint is // unknown, then we'll return an error unpackedBackup, err := chanbackup.FetchBackupForChan( - chanPoint, r.server.chanStateDB, r.server.addrSource, + ctx, chanPoint, r.server.chanStateDB, r.server.addrSource, ) if err != nil { return nil, err @@ -8316,7 +8316,7 @@ func (r *rpcServer) ExportAllChannelBackups(ctx context.Context, // First, we'll attempt to read back ups for ALL currently opened // channels from disk. allUnpackedBackups, err := chanbackup.FetchStaticChanBackups( - r.server.chanStateDB, r.server.addrSource, + ctx, r.server.chanStateDB, r.server.addrSource, ) if err != nil { return nil, fmt.Errorf("unable to fetch all static chan "+ @@ -8451,7 +8451,8 @@ func (r *rpcServer) SubscribeChannelBackups(req *lnrpc.ChannelBackupSubscription // we'll obtains the current set of single channel // backups from disk. chanBackups, err := chanbackup.FetchStaticChanBackups( - r.server.chanStateDB, r.server.addrSource, + updateStream.Context(), r.server.chanStateDB, + r.server.addrSource, ) if err != nil { return fmt.Errorf("unable to fetch all "+ diff --git a/server.go b/server.go index 877169579..2c985b2cc 100644 --- a/server.go +++ b/server.go @@ -555,7 +555,7 @@ func noiseDial(idKey keychain.SingleKeyECDH, // passed listener address. // //nolint:funlen -func newServer(_ context.Context, cfg *Config, listenAddrs []net.Addr, +func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, dbs *DatabaseInstances, cc *chainreg.ChainControl, nodeKeyDesc *keychain.KeyDescriptor, chansToRestore walletunlocker.ChannelsToRecover, @@ -1715,13 +1715,13 @@ func newServer(_ context.Context, cfg *Config, listenAddrs []net.Addr, cfg.BackupFilePath, cfg.NoBackupArchive, ) startingChans, err := chanbackup.FetchStaticChanBackups( - s.chanStateDB, s.addrSource, + ctx, s.chanStateDB, s.addrSource, ) if err != nil { return nil, err } s.chanSubSwapper, err = chanbackup.NewSubSwapper( - startingChans, chanNotifier, s.cc.KeyRing, backupFile, + ctx, startingChans, chanNotifier, s.cc.KeyRing, backupFile, ) if err != nil { return nil, err @@ -2664,6 +2664,8 @@ func (s *server) Stop() error { s.stop.Do(func() { atomic.StoreInt32(&s.stopping, 1) + ctx := context.Background() + close(s.quit) // Shutdown connMgr first to prevent conns during shutdown. @@ -2737,7 +2739,7 @@ func (s *server) Stop() error { // Update channel.backup file. Make sure to do it before // stopping chanSubSwapper. singles, err := chanbackup.FetchStaticChanBackups( - s.chanStateDB, s.addrSource, + ctx, s.chanStateDB, s.addrSource, ) if err != nil { srvrLog.Warnf("failed to fetch channel states: %v",