diff --git a/discovery/chan_series.go b/discovery/chan_series.go index 4d35a4fa1..75654a6ea 100644 --- a/discovery/chan_series.go +++ b/discovery/chan_series.go @@ -115,7 +115,7 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, // First, we'll query for all the set of channels that have an // update that falls within the specified horizon. chansInHorizon := c.graph.ChanUpdatesInHorizon( - startTime, endTime, + context.TODO(), startTime, endTime, ) for channel, err := range chansInHorizon { diff --git a/graph/builder.go b/graph/builder.go index eec97b37b..25dfdf062 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -600,7 +600,9 @@ func (b *Builder) pruneZombieChans() error { startTime := time.Unix(0, 0) endTime := time.Now().Add(-1 * chanExpiry) - oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime) + oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon( + context.TODO(), startTime, endTime, + ) for u, err := range oldEdgesIter { if err != nil { diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 74d3ba222..229aad407 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -709,7 +709,7 @@ func BenchmarkGraphReadMethods(b *testing.B) { name: "ChanUpdatesInHorizon", fn: func(b testing.TB, store Store) { iter := store.ChanUpdatesInHorizon( - 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 51284d172..cb8061b50 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -703,10 +703,11 @@ func (c *ChannelGraph) HighestChanID(ctx context.Context, // ChanUpdatesInHorizon returns all known channel edges with updates in the // horizon. -func (c *ChannelGraph) ChanUpdatesInHorizon(startTime, endTime time.Time, +func (c *ChannelGraph) ChanUpdatesInHorizon(ctx context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[ChannelEdge, error] { - return c.db.ChanUpdatesInHorizon(startTime, endTime, opts...) + return c.db.ChanUpdatesInHorizon(ctx, startTime, endTime, opts...) } // FilterChannelRange returns channel IDs within the passed block height range. diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 097e182a1..9c017aafd 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -2396,7 +2396,7 @@ func TestChanUpdatesInHorizon(t *testing.T) { // If we issue an arbitrary query before any channel updates are // inserted in the database, we should get zero results. chanIter := graph.ChanUpdatesInHorizon( - time.Unix(999, 0), time.Unix(9999, 0), + ctx, time.Unix(999, 0), time.Unix(9999, 0), ) chanUpdates, err := fn.CollectErr(chanIter) @@ -2503,7 +2503,7 @@ func TestChanUpdatesInHorizon(t *testing.T) { } for _, queryCase := range queryCases { respIter := graph.ChanUpdatesInHorizon( - queryCase.start, queryCase.end, + ctx, queryCase.start, queryCase.end, ) resp, err := fn.CollectErr(respIter) @@ -2905,7 +2905,7 @@ func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) { // Now we'll run the main query, and verify that we get // back the expected number of channels. iter := graph.ChanUpdatesInHorizon( - startTime, startTime.Add(26*time.Hour), + ctx, startTime, startTime.Add(26*time.Hour), WithChanUpdateIterBatchSize(batchSize), ) @@ -3370,7 +3370,8 @@ func TestStressTestChannelGraphAPI(t *testing.T) { name: "ChanUpdateInHorizon", fn: func() error { iter := graph.ChanUpdatesInHorizon( - time.Now().Add(-time.Hour), time.Now(), + ctx, time.Now().Add(-time.Hour), + time.Now(), ) _, err := fn.CollectErr(iter) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 568145ca2..db0250760 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -255,7 +255,8 @@ type Store interface { //nolint:interfacebloat // ChanUpdatesInHorizon returns all the known channel edges which have // at least one edge that has an update timestamp within the specified // horizon. - ChanUpdatesInHorizon(startTime, endTime time.Time, + ChanUpdatesInHorizon(ctx context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[ChannelEdge, error] // FilterKnownChanIDs takes a set of channel IDs and return the subset diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 2e3496a5e..ad330c8c3 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -2377,7 +2377,8 @@ func (c *KVStore) fetchNextChanUpdateBatch( // ChanUpdatesInHorizon returns all the known channel edges which have at least // one edge that has an update timestamp within the specified horizon. -func (c *KVStore) ChanUpdatesInHorizon(startTime, endTime time.Time, +func (c *KVStore) ChanUpdatesInHorizon(_ context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[ChannelEdge, error] { cfg := defaultIteratorConfig() diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index e195fc14e..fe8ca379c 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1170,7 +1170,8 @@ func (s *SQLStore) updateChanCacheBatch(v lnwire.GossipVersion, // 6. Repeat with updated pagination cursor until no more results // // NOTE: This is part of the Store interface. -func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time, +func (s *SQLStore) ChanUpdatesInHorizon(ctx context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[ChannelEdge, error] { // Apply options. @@ -1181,7 +1182,6 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time, return func(yield func(ChannelEdge, error) bool) { var ( - ctx = context.TODO() edgesSeen = make(map[uint64]struct{}) edgesToCache = make(map[uint64]ChannelEdge) hits int