From f3a8fd842d11d20ff76c039a1d6a29c5164ab121 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 24 Sep 2025 18:47:26 -0700 Subject: [PATCH] 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 | 20 ++- graph/builder.go | 15 +- graph/db/benchmark_test.go | 7 +- graph/db/graph_test.go | 344 +++++++++++++++++++------------------ graph/db/interfaces.go | 4 +- graph/db/kv_store.go | 33 ++-- graph/db/sql_store.go | 21 ++- graph/interfaces.go | 4 +- 8 files changed, 240 insertions(+), 208 deletions(-) diff --git a/discovery/chan_series.go b/discovery/chan_series.go index 22d6397e0..8ecb3a4c8 100644 --- a/discovery/chan_series.go +++ b/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) { diff --git a/graph/builder.go b/graph/builder.go index c77d0fea2..b446d1a7b 100644 --- a/graph/builder.go +++ b/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 { diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index c9857de2d..7c94db2d4 100644 --- a/graph/db/benchmark_test.go +++ b/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) }, }, diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 1a27cd975..774492786 100644 --- a/graph/db/graph_test.go +++ b/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)) } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index c97b26c5c..25eb6f5b1 100644 --- a/graph/db/interfaces.go +++ b/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 diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 7bd958861..cc9a14889 100644 --- a/graph/db/kv_store.go +++ b/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 diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 2d4d06774..f13408765 100644 --- a/graph/db/sql_store.go +++ b/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 diff --git a/graph/interfaces.go b/graph/interfaces.go index cf660ec50..0896a0850 100644 --- a/graph/interfaces.go +++ b/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