graph/db: thread context through ChanUpdatesInHorizon

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 {

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 {

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)

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.

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)
 

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

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

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