graph/db: expose reset in ForEachChannel

Add a `reset` call-back param to the V1Store ForEachChannel method.

graph/db/graph_test.go

@@ -935,7 +935,8 @@ func TestEdgePolicyCRUD(t *testing.T) {
 				)
 
 				return nil
-			})
+			}, func() {},
+		)
 		require.NoError(t, err)
 	}
 
@@ -1385,7 +1386,7 @@ func TestGraphTraversal(t *testing.T) {
 
 		delete(chanIndex, ei.ChannelID)
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 	require.Len(t, chanIndex, 0)
 
@@ -1669,22 +1670,19 @@ func assertPruneTip(t *testing.T, graph *ChannelGraph,
 
 func assertNumChans(t *testing.T, graph *ChannelGraph, n int) {
 	numChans := 0
-	if err := graph.ForEachChannel(context.Background(),
-		func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+	err := graph.ForEachChannel(
+		context.Background(), func(*models.ChannelEdgeInfo,
+			*models.ChannelEdgePolicy,
 			*models.ChannelEdgePolicy) error {
 
 			numChans++
 			return nil
+		}, func() {
+			numChans = 0
 		},
-	); err != nil {
-		_, _, line, _ := runtime.Caller(1)
-		t.Fatalf("line %v: unable to scan channels: %v", line, err)
-	}
-	if numChans != n {
-		_, _, line, _ := runtime.Caller(1)
-		t.Fatalf("line %v: expected %v chans instead have %v", line,
-			n, numChans)
-	}
+	)
+	require.NoError(t, err)
+	require.Equal(t, n, numChans)
 }
 
 func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) {

graph/db/interfaces.go

@@ -159,8 +159,8 @@ type V1Store interface { //nolint:interfacebloat
 	// pointer for that particular channel edge routing policy will be
 	// passed into the callback.
 	ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo,
-		*models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error
+		*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
+		reset func()) error
 
 	// ForEachChannelCacheable iterates through all the channel edges stored
 	// within the graph and invokes the passed callback for each edge. The

graph/db/kv_store.go

@@ -406,9 +406,7 @@ func (c *KVStore) AddrsForNode(ctx context.Context,
 // callback.
 func (c *KVStore) ForEachChannel(_ context.Context,
 	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-	*models.ChannelEdgePolicy) error) error {
-
-	reset := func() {}
+	*models.ChannelEdgePolicy) error, reset func()) error {
 
 	return forEachChannel(c.db, cb, reset)
 }

graph/db/sql_migration_test.go

@@ -475,7 +475,7 @@ func fetchAllChannelsAndPolicies(t *testing.T, store V1Store) chanSet {
 		})
 
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 
 	// Sort the channels by their channel ID to ensure a consistent order.

graph/db/sql_store.go

@@ -1343,9 +1343,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
 // NOTE: part of the V1Store interface.
 func (s *SQLStore) ForEachChannel(ctx context.Context,
 	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error {
-
-	reset := func() {}
+		*models.ChannelEdgePolicy) error, reset func()) error {
 
 	handleChannel := func(db SQLQueries,
 		row sqlc.ListChannelsWithPoliciesPaginatedRow) error {