From 2f0d2962dd8c6734805561317732cf6517f5f0ec Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 25 Feb 2026 09:32:33 +0200 Subject: [PATCH] graph/db: thread context through NodeUpdatesInHorizon --- discovery/chan_series.go | 3 ++- graph/db/benchmark_test.go | 2 +- graph/db/graph.go | 5 +++-- graph/db/graph_test.go | 16 ++++++++++------ graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 2 +- graph/db/sql_store.go | 4 ++-- 7 files changed, 20 insertions(+), 14 deletions(-) diff --git a/discovery/chan_series.go b/discovery/chan_series.go index ed64453a4..4d35a4fa1 100644 --- a/discovery/chan_series.go +++ b/discovery/chan_series.go @@ -181,7 +181,8 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, // update within the horizon as well. We send these second to // ensure that they follow any active channels they have. nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon( - startTime, endTime, graphdb.WithIterPublicNodesOnly(), + context.TODO(), startTime, endTime, + graphdb.WithIterPublicNodesOnly(), ) for nodeAnn, err := range nodeAnnsInHorizon { if err != nil { diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 2887b7d22..74d3ba222 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -659,7 +659,7 @@ func BenchmarkGraphReadMethods(b *testing.B) { name: "NodeUpdatesInHorizon", fn: func(b testing.TB, store Store) { iter := store.NodeUpdatesInHorizon( - time.Unix(0, 0), time.Now(), + ctx, time.Unix(0, 0), time.Now(), ) _, err := fn.CollectErr(iter) require.NoError(b, err) diff --git a/graph/db/graph.go b/graph/db/graph.go index acafbcc68..1173c7af1 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -633,10 +633,11 @@ func (c *ChannelGraph) ForEachNodeCacheable(ctx context.Context, // NodeUpdatesInHorizon returns all known lightning nodes with updates in the // range. -func (c *ChannelGraph) NodeUpdatesInHorizon(startTime, endTime time.Time, +func (c *ChannelGraph) NodeUpdatesInHorizon(ctx context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[*models.Node, error] { - return c.db.NodeUpdatesInHorizon(startTime, endTime, opts...) + return c.db.NodeUpdatesInHorizon(ctx, startTime, endTime, opts...) } // HasV1Node determines if the graph has a vertex identified by the target node diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 737376b1a..7a719a285 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -2540,7 +2540,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) { // If we issue an arbitrary query before we insert any nodes into the // database, then we shouldn't get any results back. nodeUpdatesIter := graph.NodeUpdatesInHorizon( - time.Unix(999, 0), time.Unix(9999, 0), + ctx, time.Unix(999, 0), time.Unix(9999, 0), ) nodeUpdates, err := fn.CollectErr(nodeUpdatesIter) require.NoError(t, err, "unable to query for node updates") @@ -2615,7 +2615,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) { } for _, queryCase := range queryCases { iter := graph.NodeUpdatesInHorizon( - queryCase.start, queryCase.end, + ctx, queryCase.start, queryCase.end, ) resp, err := fn.CollectErr(iter) @@ -2743,7 +2743,7 @@ func testNodeUpdatesWithBatchSize(t *testing.T, ctx context.Context, for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { iter := testGraph.NodeUpdatesInHorizon( - tc.start, tc.end, + ctx, tc.start, tc.end, WithNodeUpdateIterBatchSize( batchSize, ), @@ -2816,7 +2816,7 @@ func TestNodeUpdatesInHorizonEarlyTermination(t *testing.T) { for _, stopAt := range terminationPoints { t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) { iter := graph.NodeUpdatesInHorizon( - startTime, startTime.Add(200*time.Hour), + ctx, startTime, startTime.Add(200*time.Hour), WithNodeUpdateIterBatchSize(10), ) @@ -4091,7 +4091,9 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) { // update time of our test node. startTime := time.Unix(9, 0) endTime := node1.LastUpdate.Add(time.Minute) - nodesInHorizonIter := graph.NodeUpdatesInHorizon(startTime, endTime) + nodesInHorizonIter := graph.NodeUpdatesInHorizon( + ctx, startTime, endTime, + ) // We should only have a single node, and that node should exactly // match the node we just inserted. @@ -4107,7 +4109,9 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) { // 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. - nodesInHorizonIter = graph.NodeUpdatesInHorizon(startTime, endTime) + nodesInHorizonIter = graph.NodeUpdatesInHorizon( + ctx, startTime, endTime, + ) nodesInHorizon, err = fn.CollectErr(nodesInHorizonIter) require.NoError(t, err, "unable to fetch nodes in horizon") require.Empty(t, nodesInHorizon) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 897b56aa7..4983c9c95 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -122,7 +122,7 @@ type Store interface { //nolint:interfacebloat // an update timestamp within the passed range. This method can be used // by two nodes to quickly determine if they have the same set of up to // date node announcements. - NodeUpdatesInHorizon(startTime, endTime time.Time, + NodeUpdatesInHorizon(ctx context.Context, startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[*models.Node, error] // FetchNode attempts to look up a target node by its identity diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index ceb2b673e..5a27ee99e 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -2627,7 +2627,7 @@ func (c *KVStore) fetchNextNodeBatch( // NodeUpdatesInHorizon returns all the known lightning node which have an // update timestamp within the passed range. -func (c *KVStore) NodeUpdatesInHorizon(startTime, +func (c *KVStore) NodeUpdatesInHorizon(_ context.Context, startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[*models.Node, error] { diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 77863d71f..023516542 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -610,7 +610,8 @@ func (s *SQLStore) SetSourceNode(ctx context.Context, // announcements. // // NOTE: This is part of the Store interface. -func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time, +func (s *SQLStore) NodeUpdatesInHorizon(ctx context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[*models.Node, error] { cfg := defaultIteratorConfig() @@ -620,7 +621,6 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time, return func(yield func(*models.Node, error) bool) { var ( - ctx = context.TODO() lastUpdateTime sql.NullInt64 lastPubKey = make([]byte, 33) hasMore = true