graph+discovery: update graph/db gossip backlog interfaces to use iter.Seq2

This lets us emit a rich error if things fail when first creating the
iterator, or if any of the yield attempts fail.

discovery/chan_series.go

@@ -114,15 +114,15 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 	return func(yield func(lnwire.Message, error) bool) {
 		// First, we'll query for all the set of channels that have an
 		// update that falls within the specified horizon.
-		chansInHorizon, err := c.graph.ChanUpdatesInHorizon(
+		chansInHorizon := c.graph.ChanUpdatesInHorizon(
 			startTime, endTime,
 		)
-		if err != nil {
-			yield(nil, err)
-			return
-		}
 
-		for channel := range chansInHorizon {
+		for channel, err := range chansInHorizon {
+			if err != nil {
+				yield(nil, err)
+				return
+			}
 			// If the channel hasn't been fully advertised yet, or
 			// is a private channel, then we'll skip it as we can't
 			// construct a full authentication proof if one is
@@ -181,10 +181,14 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 		// Next, we'll send out all the node announcements that have an
 		// update within the horizon as well. We send these second to
 		// ensure that they follow any active channels they have.
-		nodeAnnsInHorizon, err := c.graph.NodeUpdatesInHorizon(
+		nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon(
 			startTime, endTime, graphdb.WithIterPublicNodesOnly(),
 		)
-		for nodeAnn := range nodeAnnsInHorizon {
+		for nodeAnn, err := range nodeAnnsInHorizon {
+			if err != nil {
+				yield(nil, err)
+				return
+			}
 			nodeUpdate, err := nodeAnn.NodeAnnouncement(true)
 			if err != nil {
 				if !yield(nil, err) {

graph/builder.go

@@ -593,13 +593,14 @@ func (b *Builder) pruneZombieChans() error {
 
 	startTime := time.Unix(0, 0)
 	endTime := time.Now().Add(-1 * chanExpiry)
-	oldEdgesIter, err := b.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime)
-	if err != nil {
-		return fmt.Errorf("unable to fetch expired channel updates "+
-			"chans: %v", err)
-	}
+	oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime)
+
+	for u, err := range oldEdgesIter {
+		if err != nil {
+			return fmt.Errorf("unable to fetch expired "+
+				"channel updates chans: %v", err)
+		}
 
-	for u := range oldEdgesIter {
 		err = filterPruneChans(u.Info, u.Policy1, u.Policy2)
 		if err != nil {
 			return fmt.Errorf("error filtering channels to "+
@@ -619,7 +620,7 @@ func (b *Builder) pruneZombieChans() error {
 		toPrune = append(toPrune, chanID)
 		log.Tracef("Pruning zombie channel with ChannelID(%v)", chanID)
 	}
-	err = b.cfg.Graph.DeleteChannelEdges(
+	err := b.cfg.Graph.DeleteChannelEdges(
 		b.cfg.StrictZombiePruning, true, toPrune...,
 	)
 	if err != nil {

graph/db/benchmark_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btclog/v2"
 	"github.com/lightningnetwork/lnd/batch"
+	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/graph/db/models"
 	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/kvdb/postgres"
@@ -788,9 +789,10 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 		{
 			name: "NodeUpdatesInHorizon",
 			fn: func(b testing.TB, store V1Store) {
-				_, err := store.NodeUpdatesInHorizon(
+				iter := store.NodeUpdatesInHorizon(
 					time.Unix(0, 0), time.Now(),
 				)
+				_, err := fn.CollectErr(iter)
 				require.NoError(b, err)
 			},
 		},
@@ -836,9 +838,10 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 		{
 			name: "ChanUpdatesInHorizon",
 			fn: func(b testing.TB, store V1Store) {
-				_, err := store.ChanUpdatesInHorizon(
+				iter := store.ChanUpdatesInHorizon(
 					time.Unix(0, 0), time.Now(),
 				)
+				_, err := fn.CollectErr(iter)
 				require.NoError(b, err)
 			},
 		},

graph/db/graph_test.go

@@ -2028,12 +2028,12 @@ 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, err := graph.ChanUpdatesInHorizon(
+	chanIter := graph.ChanUpdatesInHorizon(
 		time.Unix(999, 0), time.Unix(9999, 0),
 	)
-	require.NoError(t, err, "unable to updates for updates")
 
-	chanUpdates := fn.Collect(chanIter)
+	chanUpdates, err := fn.CollectErr(chanIter)
+	require.NoError(t, err, "unable to updates for updates")
 
 	if len(chanUpdates) != 0 {
 		t.Fatalf("expected 0 chan updates, instead got %v",
@@ -2147,15 +2147,15 @@ func TestChanUpdatesInHorizon(t *testing.T) {
 		},
 	}
 	for _, queryCase := range queryCases {
-		respIter, err := graph.ChanUpdatesInHorizon(
+		respIter := graph.ChanUpdatesInHorizon(
 			queryCase.start, queryCase.end,
 		)
+
+		resp, err := fn.CollectErr(respIter)
 		if err != nil {
 			t.Fatalf("unable to query for updates: %v", err)
 		}
 
-		resp := fn.Collect(respIter)
-
 		if len(resp) != len(queryCase.resp) {
 			t.Fatalf("expected %v chans, got %v chans",
 				len(queryCase.resp), len(resp))
@@ -2194,11 +2194,11 @@ 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, err := graph.NodeUpdatesInHorizon(
+	nodeUpdatesIter := graph.NodeUpdatesInHorizon(
 		time.Unix(999, 0), time.Unix(9999, 0),
 	)
+	nodeUpdates, err := fn.CollectErr(nodeUpdatesIter)
 	require.NoError(t, err, "unable to query for node updates")
-	nodeUpdates := fn.Collect(nodeUpdatesIter)
 	require.Len(t, nodeUpdates, 0)
 
 	// We'll create 10 node announcements, each with an update timestamp 10
@@ -2269,12 +2269,12 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
 		},
 	}
 	for _, queryCase := range queryCases {
-		iter, err := graph.NodeUpdatesInHorizon(
+		iter := graph.NodeUpdatesInHorizon(
 			queryCase.start, queryCase.end,
 		)
-		require.NoError(t, err, "unable to query for node updates")
 
-		resp := fn.Collect(iter)
+		resp, err := fn.CollectErr(iter)
+		require.NoError(t, err, "unable to query for node updates")
 		require.Len(t, resp, len(queryCase.resp))
 
 		for i := 0; i < len(resp); i++ {
@@ -2283,6 +2283,152 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
 	}
 }
 
+// testNodeUpdatesWithBatchSize is a helper function that tests node updates
+// with a specific batch size to ensure the iterator works correctly across
+// batch boundaries.
+func testNodeUpdatesWithBatchSize(t *testing.T, ctx context.Context,
+	batchSize int) {
+
+	// Create a fresh graph for each test.
+	testGraph := MakeTestGraph(t)
+
+	// Add 25 nodes with increasing timestamps.
+	startTime := time.Unix(1234567890, 0)
+	var nodeAnns []models.Node
+
+	for i := 0; i < 25; i++ {
+		nodeAnn := createTestVertex(t)
+		nodeAnn.LastUpdate = startTime.Add(
+			time.Duration(i) * time.Hour,
+		)
+		nodeAnns = append(nodeAnns, *nodeAnn)
+		require.NoError(
+			t, testGraph.AddNode(ctx, nodeAnn),
+		)
+	}
+
+	testCases := []struct {
+		name  string
+		start time.Time
+		end   time.Time
+		want  int
+	}{
+		{
+			name:  "all nodes",
+			start: startTime,
+			end:   startTime.Add(26 * time.Hour),
+			want:  25,
+		},
+		{
+			name:  "first batch only",
+			start: startTime,
+			end: startTime.Add(
+				time.Duration(
+					min(batchSize, 25)-1,
+				) * time.Hour,
+			),
+			want: min(batchSize, 25),
+		},
+		{
+			name:  "cross batch boundary",
+			start: startTime,
+			end: startTime.Add(
+				time.Duration(
+					min(batchSize, 24),
+				) * time.Hour,
+			),
+			want: min(batchSize+1, 25),
+		},
+		{
+			name: "exact boundary",
+			start: func() time.Time {
+				// Test querying exactly at a
+				// batch boundary.
+				if batchSize <= 25 {
+					return startTime.Add(
+						time.Duration(
+							batchSize-1,
+						) * time.Hour,
+					)
+				}
+
+				// For batch sizes > 25, test
+				// beyond our data range.
+				return startTime.Add(
+					time.Duration(25) * time.Hour,
+				)
+			}(),
+			end: func() time.Time {
+				if batchSize <= 25 {
+					return startTime.Add(
+						time.Duration(
+							batchSize-1,
+						) * time.Hour,
+					)
+				}
+
+				return startTime.Add(
+					time.Duration(25) * time.Hour,
+				)
+			}(),
+			want: func() int {
+				if batchSize <= 25 {
+					return 1
+				}
+
+				// No nodes exist at hour 25 or
+				// beyond.
+				return 0
+			}(),
+		},
+		{
+			name:  "empty range before",
+			start: startTime.Add(-time.Hour),
+			end:   startTime.Add(-time.Minute),
+			want:  0,
+		},
+		{
+			name:  "empty range after",
+			start: startTime.Add(30 * time.Hour),
+			end:   startTime.Add(40 * time.Hour),
+			want:  0,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			iter := testGraph.NodeUpdatesInHorizon(
+				tc.start, tc.end,
+				WithNodeUpdateIterBatchSize(
+					batchSize,
+				),
+			)
+
+			nodes, err := fn.CollectErr(iter)
+			require.NoError(t, err)
+			require.Len(
+				t, nodes, tc.want,
+				"expected %d nodes, got %d",
+				tc.want, len(nodes),
+			)
+
+			// Verify nodes are in the correct time
+			// order.
+			for i := 1; i < len(nodes); i++ {
+				require.True(t,
+					nodes[i-1].LastUpdate.Before(
+						nodes[i].LastUpdate,
+					) || nodes[i-1].LastUpdate.Equal(
+						nodes[i].LastUpdate,
+					),
+					"nodes should be in "+
+						"chronological order",
+				)
+			}
+		})
+	}
+}
+
 // TestNodeUpdatesInHorizonBoundaryConditions tests the iterator boundary
 // conditions, specifically around batch boundaries and edge cases.
 func TestNodeUpdatesInHorizonBoundaryConditions(t *testing.T) {
@@ -2295,144 +2441,9 @@ func TestNodeUpdatesInHorizonBoundaryConditions(t *testing.T) {
 	batchSizes := []int{1, 3, 5, 10, 25, 100}
 
 	for _, batchSize := range batchSizes {
-		t.Run(fmt.Sprintf("BatchSize%d", batchSize), func(t *testing.T) {
-			// Create a fresh graph for each test.
-			testGraph := MakeTestGraph(t)
-
-			// Add 25 nodes with increasing timestamps.
-			startTime := time.Unix(1234567890, 0)
-			var nodeAnns []models.Node
-
-			for i := 0; i < 25; i++ {
-				nodeAnn := createTestVertex(t)
-				nodeAnn.LastUpdate = startTime.Add(
-					time.Duration(i) * time.Hour,
-				)
-				nodeAnns = append(nodeAnns, *nodeAnn)
-				require.NoError(
-					t, testGraph.AddNode(ctx, nodeAnn),
-				)
-			}
-
-			testCases := []struct {
-				name  string
-				start time.Time
-				end   time.Time
-				want  int
-			}{
-				{
-					name:  "all nodes",
-					start: startTime,
-					end:   startTime.Add(26 * time.Hour),
-					want:  25,
-				},
-				{
-					name:  "first batch only",
-					start: startTime,
-					end: startTime.Add(
-						time.Duration(
-							min(batchSize, 25)-1,
-						) * time.Hour,
-					),
-					want:  min(batchSize, 25),
-				},
-				{
-					name:  "cross batch boundary",
-					start: startTime,
-					end: startTime.Add(
-						time.Duration(
-							min(batchSize, 24),
-						) * time.Hour,
-					),
-					want:  min(batchSize+1, 25),
-				},
-				{
-					name: "exact boundary",
-					start: func() time.Time {
-						// Test querying exactly at a
-						// batch boundary.
-						if batchSize <= 25 {
-							return startTime.Add(
-								time.Duration(
-									batchSize-1,
-								) * time.Hour,
-							)
-						}
-
-						// For batch sizes > 25, test
-						// beyond our data range.
-						return startTime.Add(
-							time.Duration(25) * time.Hour,
-						)
-					}(),
-					end: func() time.Time {
-						if batchSize <= 25 {
-							return startTime.Add(
-								time.Duration(
-									batchSize-1,
-								) * time.Hour,
-							)
-						}
-						return startTime.Add(
-							time.Duration(25) * time.Hour,
-						)
-					}(),
-					want: func() int {
-						if batchSize <= 25 {
-							return 1
-						}
-
-						// No nodes exist at hour 25 or
-						// beyond.
-						return 0
-					}(),
-				},
-				{
-					name:  "empty range before",
-					start: startTime.Add(-time.Hour),
-					end:   startTime.Add(-time.Minute),
-					want:  0,
-				},
-				{
-					name:  "empty range after",
-					start: startTime.Add(30 * time.Hour),
-					end:   startTime.Add(40 * time.Hour),
-					want:  0,
-				},
-			}
-
-			for _, tc := range testCases {
-				t.Run(tc.name, func(t *testing.T) {
-					iter, err := testGraph.NodeUpdatesInHorizon(
-						tc.start, tc.end,
-						WithNodeUpdateIterBatchSize(
-							batchSize,
-						),
-					)
-					require.NoError(t, err)
-
-					nodes := fn.Collect(iter)
-					require.Len(
-						t, nodes, tc.want,
-						"expected %d nodes, got %d",
-						tc.want, len(nodes),
-					)
-
-					// Verify nodes are in the correct time
-					// order.
-					for i := 1; i < len(nodes); i++ {
-						require.True(t,
-							nodes[i-1].LastUpdate.Before(
-								nodes[i].LastUpdate,
-							) || nodes[i-1].LastUpdate.Equal(
-								nodes[i].LastUpdate,
-							),
-							"nodes should be in "+
-								"chronological order",
-						)
-					}
-				})
-			}
+		testName := fmt.Sprintf("BatchSize%d", batchSize)
+		t.Run(testName, func(t *testing.T) {
+			testNodeUpdatesWithBatchSize(t, ctx, batchSize)
 		})
 	}
 }
@@ -2459,11 +2470,10 @@ func TestNodeUpdatesInHorizonEarlyTermination(t *testing.T) {
 
 	for _, stopAt := range terminationPoints {
 		t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) {
-			iter, err := graph.NodeUpdatesInHorizon(
+			iter := graph.NodeUpdatesInHorizon(
 				startTime, startTime.Add(200*time.Hour),
 				WithNodeUpdateIterBatchSize(10),
 			)
-			require.NoError(t, err)
 
 			// Collect only up to stopAt nodes, breaking afterwards.
 			var collected []models.Node
@@ -2494,7 +2504,8 @@ func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) {
 	batchSizes := []int{1, 3, 5, 10}
 
 	for _, batchSize := range batchSizes {
-		t.Run(fmt.Sprintf("BatchSize%d", batchSize), func(t *testing.T) {
+		testName := fmt.Sprintf("BatchSize%d", batchSize)
+		t.Run(testName, func(t *testing.T) {
 			// Create a fresh graph for each test, then add two new
 			// nodes to the graph.
 			graph := MakeTestGraph(t)
@@ -2516,7 +2527,9 @@ func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) {
 				channel, chanID := createEdge(
 					uint32(i*10), 0, 0, 0, node1, node2,
 				)
-				require.NoError(t, graph.AddChannelEdge(ctx, &channel))
+				require.NoError(
+					t, graph.AddChannelEdge(ctx, &channel),
+				)
 
 				edge1 := newEdgePolicy(
 					chanID.ToUint64(), updateTime.Unix(),
@@ -2534,18 +2547,20 @@ func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) {
 				edge2.ChannelFlags = 1
 				edge2.ToNode = node1.PubKeyBytes
 				edge2.SigBytes = testSig.Serialize()
-				require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
+				require.NoError(
+					t, graph.UpdateEdgePolicy(ctx, edge2),
+				)
 			}
 
 			// Now we'll run the main query, and verify that we get
 			// back the expected number of channels.
-			iter, err := graph.ChanUpdatesInHorizon(
+			iter := graph.ChanUpdatesInHorizon(
 				startTime, startTime.Add(26*time.Hour),
 				WithChanUpdateIterBatchSize(batchSize),
 			)
-			require.NoError(t, err)
 
-			channels := fn.Collect(iter)
+			channels, err := fn.CollectErr(iter)
+			require.NoError(t, err)
 			require.Len(
 				t, channels, numChans,
 				"expected %d channels, got %d", numChans,
@@ -3011,9 +3026,10 @@ func TestStressTestChannelGraphAPI(t *testing.T) {
 		{
 			name: "ChanUpdateInHorizon",
 			fn: func() error {
-				_, err := graph.ChanUpdatesInHorizon(
+				iter := graph.ChanUpdatesInHorizon(
 					time.Now().Add(-time.Hour), time.Now(),
 				)
+				_, err := fn.CollectErr(iter)
 
 				return err
 			},
@@ -3787,12 +3803,12 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
 	// update time of our test node.
 	startTime := time.Unix(9, 0)
 	endTime := node1.LastUpdate.Add(time.Minute)
-	nodesInHorizonIter, err := graph.NodeUpdatesInHorizon(startTime, endTime)
-	require.NoError(t, err, "unable to fetch nodes in horizon")
+	nodesInHorizonIter := graph.NodeUpdatesInHorizon(startTime, endTime)
 
 	// We should only have a single node, and that node should exactly
 	// match the node we just inserted.
-	nodesInHorizon := fn.Collect(nodesInHorizonIter)
+	nodesInHorizon, err := fn.CollectErr(nodesInHorizonIter)
+	require.NoError(t, err, "unable to fetch nodes in horizon")
 	if len(nodesInHorizon) != 1 {
 		t.Fatalf("should have 1 nodes instead have: %v",
 			len(nodesInHorizon))
@@ -3806,11 +3822,11 @@ 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, err = graph.NodeUpdatesInHorizon(startTime, endTime)
+	nodesInHorizonIter = graph.NodeUpdatesInHorizon(startTime, endTime)
+	nodesInHorizon, err = fn.CollectErr(nodesInHorizonIter)
 	require.NoError(t, err, "unable to fetch nodes in horizon")
-	nodesInHorizon = fn.Collect(nodesInHorizonIter)
 
-	if len(fn.Collect(nodesInHorizonIter)) != 0 {
+	if len(nodesInHorizon) != 0 {
 		t.Fatalf("should have zero nodes instead have: %v",
 			len(nodesInHorizon))
 	}

graph/db/interfaces.go

@@ -111,7 +111,7 @@ type V1Store interface { //nolint:interfacebloat
 	// by two nodes to quickly determine if they have the same set of up to
 	// date node announcements.
 	NodeUpdatesInHorizon(startTime, endTime time.Time,
-		opts ...IteratorOption) (iter.Seq[models.Node], error)
+		opts ...IteratorOption) iter.Seq2[models.Node, error]
 
 	// FetchNode attempts to look up a target node by its identity
 	// public key. If the node isn't found in the database, then
@@ -221,7 +221,7 @@ type V1Store interface { //nolint:interfacebloat
 	// at least one edge that has an update timestamp within the specified
 	// horizon.
 	ChanUpdatesInHorizon(startTime, endTime time.Time,
-		opts ...IteratorOption) (iter.Seq[ChannelEdge], error)
+		opts ...IteratorOption) iter.Seq2[ChannelEdge, error]
 
 	// FilterKnownChanIDs takes a set of channel IDs and return the subset
 	// of chan ID's that we don't know and are not known zombies of the

graph/db/kv_store.go

@@ -2287,14 +2287,14 @@ 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,
-	opts ...IteratorOption) (iter.Seq[ChannelEdge], error) {
+	opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
 
 	cfg := defaultIteratorConfig()
 	for _, opt := range opts {
 		opt(cfg)
 	}
 
-	return func(yield func(ChannelEdge) bool) {
+	return func(yield func(ChannelEdge, error) bool) {
 		iterState := newChanUpdatesIterator(
 			cfg.chanUpdateIterBatchSize, startTime, endTime,
 		)
@@ -2306,14 +2306,15 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 			batch, hasMore, err := c.fetchNextChanUpdateBatch(
 				iterState,
 			)
-			// TODO(roasbeef): yield error here?
 			if err != nil {
 				// These errors just mean the graph is empty,
 				// which is OK.
 				if !isEmptyGraphError(err) {
-
 					log.Errorf("ChanUpdatesInHorizon "+
 						"batch error: %v", err)
+
+					yield(ChannelEdge{}, err)
+
 					return
 				}
 				// Continue with empty batch
@@ -2322,7 +2323,7 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 			// We'll now yield each edge that we just read. If yield
 			// returns false, then that means that we'll exit early.
 			for _, edge := range batch {
-				if !yield(edge) {
+				if !yield(edge, nil) {
 					return
 				}
 			}
@@ -2347,17 +2348,17 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 			log.Tracef("ChanUpdatesInHorizon returned no edges "+
 				"in horizon (%s, %s)", startTime, endTime)
 		}
-	}, nil
+	}
 }
 
 // nodeUpdatesIterator maintains state for iterating through node updates.
 //
 // Iterator Lifecycle:
-// 1. Initialize state with start/end time, batch size, and filtering options
-// 2. Fetch batch using pagination cursor (lastSeenKey)
-// 3. Filter nodes if publicNodesOnly is set
-// 4. Update lastSeenKey to the last processed node's index key
-// 5. Repeat until we exceed endTime or no more nodes exist
+// 1. Initialize state with start/end time, batch size, and filtering options.
+// 2. Fetch batch using pagination cursor (lastSeenKey).
+// 3. Filter nodes if publicNodesOnly is set.
+// 4. Update lastSeenKey to the last processed node's index key.
+// 5. Repeat until we exceed endTime or no more nodes exist.
 type nodeUpdatesIterator struct {
 	// batchSize is the amount of node updates to read at a single time.
 	batchSize int
@@ -2537,14 +2538,14 @@ func (c *KVStore) fetchNextNodeBatch(
 // update timestamp within the passed range.
 func (c *KVStore) NodeUpdatesInHorizon(startTime,
 	endTime time.Time,
-	opts ...IteratorOption) (iter.Seq[models.Node], error) {
+	opts ...IteratorOption) iter.Seq2[models.Node, error] {
 
 	cfg := defaultIteratorConfig()
 	for _, opt := range opts {
 		opt(cfg)
 	}
 
-	return func(yield func(models.Node) bool) {
+	return func(yield func(models.Node, error) bool) {
 		// Initialize iterator state.
 		state := newNodeUpdatesIterator(
 			cfg.nodeUpdateIterBatchSize,
@@ -2558,11 +2559,13 @@ func (c *KVStore) NodeUpdatesInHorizon(startTime,
 				log.Errorf("unable to read node updates in "+
 					"horizon: %v", err)
 
+				yield(models.Node{}, err)
+
 				return
 			}
 
 			for _, node := range nodeAnns {
-				if !yield(node) {
+				if !yield(node, nil) {
 					return
 				}
 			}
@@ -2573,7 +2576,7 @@ func (c *KVStore) NodeUpdatesInHorizon(startTime,
 				break
 			}
 		}
-	}, nil
+	}
 }
 
 // FilterKnownChanIDs takes a set of channel IDs and return the subset of chan

graph/db/sql_store.go

@@ -554,14 +554,14 @@ func (s *SQLStore) SetSourceNode(ctx context.Context,
 //
 // NOTE: This is part of the V1Store interface.
 func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time,
-	opts ...IteratorOption) (iter.Seq[models.Node], error) {
+	opts ...IteratorOption) iter.Seq2[models.Node, error] {
 
 	cfg := defaultIteratorConfig()
 	for _, opt := range opts {
 		opt(cfg)
 	}
 
-	return func(yield func(models.Node) bool) {
+	return func(yield func(models.Node, error) bool) {
 		var (
 			ctx            = context.TODO()
 			lastUpdateTime sql.NullInt64
@@ -634,11 +634,14 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time,
 			if err != nil {
 				log.Errorf("NodeUpdatesInHorizon batch "+
 					"error: %v", err)
+
+				yield(models.Node{}, err)
+
 				return
 			}
 
 			for _, node := range batch {
-				if !yield(node) {
+				if !yield(node, nil) {
 					return
 				}
 			}
@@ -648,7 +651,7 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time,
 				break
 			}
 		}
-	}, nil
+	}
 }
 
 // AddChannelEdge adds a new (undirected, blank) edge to the graph database. An
@@ -1078,7 +1081,7 @@ func (s *SQLStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) {
 //
 // NOTE: This is part of the V1Store interface.
 func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
-	opts ...IteratorOption) (iter.Seq[ChannelEdge], error) {
+	opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
 
 	// Apply options.
 	cfg := defaultIteratorConfig()
@@ -1086,7 +1089,7 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 		opt(cfg)
 	}
 
-	return func(yield func(ChannelEdge) bool) {
+	return func(yield func(ChannelEdge, error) bool) {
 		var (
 			ctx            = context.TODO()
 			edgesSeen      = make(map[uint64]struct{})
@@ -1198,11 +1201,13 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 				log.Errorf("ChanUpdatesInHorizon "+
 					"batch error: %v", err)
 
+				yield(ChannelEdge{}, err)
+
 				return
 			}
 
 			for _, edge := range batch {
-				if !yield(edge) {
+				if !yield(edge, nil) {
 					return
 				}
 			}
@@ -1226,7 +1231,7 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 			log.Debugf("ChanUpdatesInHorizon returned no edges "+
 				"in horizon (%s, %s)", startTime, endTime)
 		}
-	}, nil
+	}
 }
 
 // ForEachNodeCached is similar to forEachNode, but it returns DirectedChannel

graph/interfaces.go

@@ -154,8 +154,8 @@ type DB interface {
 	// at least one edge that has an update timestamp within the specified
 	// horizon.
 	ChanUpdatesInHorizon(startTime, endTime time.Time,
-		opts ...graphdb.IteratorOption) (
-		iter.Seq[graphdb.ChannelEdge], error)
+		opts ...graphdb.IteratorOption,
+	) iter.Seq2[graphdb.ChannelEdge, error]
 
 	// DeleteChannelEdges removes edges with the given channel IDs from the
 	// database and marks them as zombies. This ensures that we're unable to