diff --git a/discovery/chan_series.go b/discovery/chan_series.go index cf502694c..8ecb3a4c8 100644 --- a/discovery/chan_series.go +++ b/discovery/chan_series.go @@ -2,6 +2,7 @@ package discovery import ( "context" + "iter" "time" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -30,8 +31,8 @@ type ChannelGraphTimeSeries interface { // update timestamp between the start time and end time. We'll use this // to catch up a remote node to the set of channel updates that they // may have missed out on within the target chain. - UpdatesInHorizon(chain chainhash.Hash, - startTime time.Time, endTime time.Time) ([]lnwire.Message, error) + UpdatesInHorizon(chain chainhash.Hash, startTime time.Time, + endTime time.Time) iter.Seq2[lnwire.Message, error] // FilterKnownChanIDs takes a target chain, and a set of channel ID's, // and returns a filtered set of chan ID's. This filtered set of chan @@ -108,140 +109,100 @@ func (c *ChanSeries) HighestChanID(ctx context.Context, // // NOTE: This is part of the ChannelGraphTimeSeries interface. func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, - startTime time.Time, endTime time.Time) ([]lnwire.Message, error) { + startTime, endTime time.Time) iter.Seq2[lnwire.Message, error] { - var updates []lnwire.Message - - // 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( - startTime, endTime, - ) - if err != nil { - return nil, err - } - - // nodesFromChan records the nodes seen from the channels. - nodesFromChan := make(map[[33]byte]struct{}, len(chansInHorizon)*2) - - for _, channel := range chansInHorizon { - // 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 requested. - if channel.Info.AuthProof == nil { - continue - } - - chanAnn, edge1, edge2, err := netann.CreateChanAnnouncement( - channel.Info.AuthProof, channel.Info, channel.Policy1, - channel.Policy2, + 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 := c.graph.ChanUpdatesInHorizon( + startTime, endTime, ) - if err != nil { - return nil, err - } - // Create a slice to hold the `channel_announcement` and - // potentially two `channel_update` msgs. - // - // NOTE: Based on BOLT7, if a channel_announcement has no - // corresponding channel_updates, we must not send the - // channel_announcement. Thus we use this slice to decide we - // want to send this `channel_announcement` or not. By the end - // of the operation, if the len of the slice is 1, we will not - // send the `channel_announcement`. Otherwise, when sending the - // msgs, the `channel_announcement` must be sent prior to any - // corresponding `channel_update` or `node_annoucement`, that's - // why we create a slice here to maintain the order. - chanUpdates := make([]lnwire.Message, 0, 3) - chanUpdates = append(chanUpdates, chanAnn) + 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 + // requested. + if channel.Info.AuthProof == nil { + continue + } + + //nolint:ll + chanAnn, edge1, edge2, err := netann.CreateChanAnnouncement( + channel.Info.AuthProof, channel.Info, + channel.Policy1, channel.Policy2, + ) + if err != nil { + if !yield(nil, err) { + return + } + + continue + } + + if !yield(chanAnn, nil) { + return + } - if edge1 != nil { // We don't want to send channel updates that don't - // conform to the spec (anymore). - err := netann.ValidateChannelUpdateFields(0, edge1) - if err != nil { - log.Errorf("not sending invalid channel "+ - "update %v: %v", edge1, err) - } else { - chanUpdates = append(chanUpdates, edge1) + // conform to the spec (anymore), so check to make sure + // that these channel updates are valid before yielding + // them. + if edge1 != nil { + err := netann.ValidateChannelUpdateFields( + 0, edge1, + ) + if err != nil { + log.Errorf("not sending invalid "+ + "channel update %v: %v", + edge1, err) + } else if !yield(edge1, nil) { + return + } + } + if edge2 != nil { + err := netann.ValidateChannelUpdateFields( + 0, edge2, + ) + if err != nil { + log.Errorf("not sending invalid "+ + "channel update %v: %v", edge2, + err) + } else if !yield(edge2, nil) { + return + } } } - if edge2 != nil { - err := netann.ValidateChannelUpdateFields(0, edge2) + // 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 := c.graph.NodeUpdatesInHorizon( + startTime, endTime, graphdb.WithIterPublicNodesOnly(), + ) + for nodeAnn, err := range nodeAnnsInHorizon { if err != nil { - log.Errorf("not sending invalid channel "+ - "update %v: %v", edge2, err) - } else { - chanUpdates = append(chanUpdates, edge2) + yield(nil, err) + return + } + nodeUpdate, err := nodeAnn.NodeAnnouncement(true) + if err != nil { + if !yield(nil, err) { + return + } + + continue + } + + if !yield(nodeUpdate, nil) { + return } } - - // If there's no corresponding `channel_update` to send, skip - // sending this `channel_announcement`. - if len(chanUpdates) < 2 { - continue - } - - // Append the all the msgs to the slice. - updates = append(updates, chanUpdates...) - - // Record the nodes seen. - nodesFromChan[channel.Info.NodeKey1Bytes] = struct{}{} - nodesFromChan[channel.Info.NodeKey2Bytes] = struct{}{} } - - // 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( - startTime, endTime, - ) - if err != nil { - return nil, err - } - - for _, nodeAnn := range nodeAnnsInHorizon { - // If this node has not been seen in the above channels, we can - // skip sending its NodeAnnouncement. - if _, seen := nodesFromChan[nodeAnn.PubKeyBytes]; !seen { - log.Debugf("Skipping forwarding as node %x not found "+ - "in channel announcement", nodeAnn.PubKeyBytes) - continue - } - - // Ensure we only forward nodes that are publicly advertised to - // prevent leaking information about nodes. - isNodePublic, err := c.graph.IsPublicNode(nodeAnn.PubKeyBytes) - if err != nil { - log.Errorf("Unable to determine if node %x is "+ - "advertised: %v", nodeAnn.PubKeyBytes, err) - continue - } - - if !isNodePublic { - log.Tracef("Skipping forwarding announcement for "+ - "node %x due to being unadvertised", - nodeAnn.PubKeyBytes) - continue - } - - nodeUpdate, err := nodeAnn.NodeAnnouncement(true) - if err != nil { - return nil, err - } - - if err := netann.ValidateNodeAnnFields(nodeUpdate); err != nil { - log.Debugf("Skipping forwarding invalid node "+ - "announcement %x: %v", nodeAnn.PubKeyBytes, err) - - continue - } - - updates = append(updates, nodeUpdate) - } - - return updates, nil } // FilterKnownChanIDs takes a target chain, and a set of channel ID's, and diff --git a/discovery/syncer.go b/discovery/syncer.go index 12e7dbf16..b8c51a9a5 100644 --- a/discovery/syncer.go +++ b/discovery/syncer.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "iter" "math" "math/rand" "sort" @@ -1442,20 +1443,30 @@ func (g *GossipSyncer) ApplyGossipFilter(ctx context.Context, // Now that the remote peer has applied their filter, we'll query the // database for all the messages that are beyond this filter. - newUpdatestoSend, err := g.cfg.channelSeries.UpdatesInHorizon( + newUpdatestoSend := g.cfg.channelSeries.UpdatesInHorizon( g.cfg.chainHash, startTime, endTime, ) - if err != nil { + + // Create a pull-based iterator so we can check if there are any + // updates before launching the goroutine. + next, stop := iter.Pull2(newUpdatestoSend) + + // Check if we have any updates to send by attempting to get the first + // message. + firstMsg, firstErr, ok := next() + if firstErr != nil { + stop() returnSema() - return err + return firstErr } log.Infof("GossipSyncer(%x): applying new remote update horizon: "+ - "start=%v, end=%v, backlog_size=%v", g.cfg.peerPub[:], - startTime, endTime, len(newUpdatestoSend)) + "start=%v, end=%v, has_updates=%v", g.cfg.peerPub[:], + startTime, endTime, ok) // If we don't have any to send, then we can return early. - if len(newUpdatestoSend) == 0 { + if !ok { + stop() returnSema() return nil } @@ -1472,14 +1483,45 @@ func (g *GossipSyncer) ApplyGossipFilter(ctx context.Context, } // We'll conclude by launching a goroutine to send out any updates. + // The goroutine takes ownership of the iterator. g.cg.WgAdd(1) go func() { defer g.cg.WgDone() defer returnSema() defer g.isSendingBacklog.Store(false) + defer stop() - for _, msg := range newUpdatestoSend { - err := g.sendToPeerSync(ctx, msg) + // Send the first message we already pulled. + err := g.sendToPeerSync(ctx, firstMsg) + switch { + case errors.Is(err, ErrGossipSyncerExiting): + return + + case errors.Is(err, lnpeer.ErrPeerExiting): + return + + case err != nil: + log.Errorf("Unable to send message for "+ + "peer catch up: %v", err) + } + + // Continue with the rest of the messages using the same pull + // iterator. + for { + msg, err, ok := next() + if !ok { + return + } + + // If the iterator yielded an error, log it and + // continue. + if err != nil { + log.Errorf("Error fetching update for peer "+ + "catch up: %v", err) + continue + } + + err = g.sendToPeerSync(ctx, msg) switch { case err == ErrGossipSyncerExiting: return diff --git a/discovery/syncer_test.go b/discovery/syncer_test.go index edf91cb69..faef40bbe 100644 --- a/discovery/syncer_test.go +++ b/discovery/syncer_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "iter" "math" "reflect" "sort" @@ -86,13 +87,22 @@ func (m *mockChannelGraphTimeSeries) HighestChanID(_ context.Context, } func (m *mockChannelGraphTimeSeries) UpdatesInHorizon(chain chainhash.Hash, - startTime time.Time, endTime time.Time) ([]lnwire.Message, error) { + startTime, endTime time.Time) iter.Seq2[lnwire.Message, error] { - m.horizonReq <- horizonQuery{ - chain, startTime, endTime, + return func(yield func(lnwire.Message, error) bool) { + m.horizonReq <- horizonQuery{ + chain, startTime, endTime, + } + + // We'll get the response from the channel, then yield it + // immediately. + msgs := <-m.horizonResp + for _, msg := range msgs { + if !yield(msg, nil) { + return + } + } } - - return <-m.horizonResp, nil } func (m *mockChannelGraphTimeSeries) FilterKnownChanIDs(chain chainhash.Hash, diff --git a/docs/release-notes/release-notes-0.20.0.md b/docs/release-notes/release-notes-0.20.0.md index 39a05b236..631ede8c7 100644 --- a/docs/release-notes/release-notes-0.20.0.md +++ b/docs/release-notes/release-notes-0.20.0.md @@ -219,6 +219,12 @@ reader of a payment request. - [Introduced](https://github.com/lightningnetwork/lnd/pull/10136) a wallet interface to decouple the relationship between `lnd` and `btcwallet`. +- [Refactored](https://github.com/lightningnetwork/lnd/pull/10128) channel graph + update iterators to use Go's `iter.Seq2` pattern. The `UpdatesInHorizon`, + `NodeUpdatesInHorizon`, and `ChanUpdatesInHorizon` methods now return lazy + iterators instead of materializing all updates in memory at once, improving + memory efficiency for large graph operations. + ## Breaking Changes ## Performance Improvements @@ -300,6 +306,7 @@ reader of a payment request. * Erick Cestari * Funyug * Mohamed Awnallah +* Olaoluwa Osuntokun * Pins * Torkel Rogstad * Yong Yu diff --git a/fn/go.mod b/fn/go.mod index 89dc280df..abf0cdf4c 100644 --- a/fn/go.mod +++ b/fn/go.mod @@ -1,6 +1,6 @@ module github.com/lightningnetwork/lnd/fn/v2 -go 1.19 +go 1.23 require ( github.com/stretchr/testify v1.8.1 diff --git a/fn/iter.go b/fn/iter.go new file mode 100644 index 000000000..e6636aadb --- /dev/null +++ b/fn/iter.go @@ -0,0 +1,29 @@ +package fn + +import "iter" + +// Collect drains all of the elements from the passed iterator, returning a +// slice of the contents. +func Collect[T any](seq iter.Seq[T]) []T { + + var ret []T + for i := range seq { + ret = append(ret, i) + } + + return ret +} + +// CollectErr drains all of the elements from the passed iterator with error +// handling, returning a slice of the contents and the first error encountered. +func CollectErr[T any](seq iter.Seq2[T, error]) ([]T, error) { + var ret []T + for val, err := range seq { + if err != nil { + return ret, err + } + ret = append(ret, val) + } + + return ret, nil +} diff --git a/go.mod b/go.mod index d1064da1d..9898b4d7f 100644 --- a/go.mod +++ b/go.mod @@ -206,6 +206,9 @@ require ( // store have been included in a tagged sqldb version. replace github.com/lightningnetwork/lnd/sqldb => ./sqldb +// Replace fn package to use local version with iterator Collect function. +replace github.com/lightningnetwork/lnd/fn/v2 => ./fn + // This replace is for https://github.com/advisories/GHSA-25xm-hr59-7c27 replace github.com/ulikunitz/xz => github.com/ulikunitz/xz v0.5.11 diff --git a/go.sum b/go.sum index 189b81c6e..ae6f41da1 100644 --- a/go.sum +++ b/go.sum @@ -374,8 +374,6 @@ github.com/lightningnetwork/lnd/cert v1.2.2 h1:71YK6hogeJtxSxw2teq3eGeuy4rHGKcFf github.com/lightningnetwork/lnd/cert v1.2.2/go.mod h1:jQmFn/Ez4zhDgq2hnYSw8r35bqGVxViXhX6Cd7HXM6U= github.com/lightningnetwork/lnd/clock v1.1.1 h1:OfR3/zcJd2RhH0RU+zX/77c0ZiOnIMsDIBjgjWdZgA0= github.com/lightningnetwork/lnd/clock v1.1.1/go.mod h1:mGnAhPyjYZQJmebS7aevElXKTFDuO+uNFFfMXK1W8xQ= -github.com/lightningnetwork/lnd/fn/v2 v2.0.8 h1:r2SLz7gZYQPVc3IZhU82M66guz3Zk2oY+Rlj9QN5S3g= -github.com/lightningnetwork/lnd/fn/v2 v2.0.8/go.mod h1:TOzwrhjB/Azw1V7aa8t21ufcQmdsQOQMDtxVOQWNl8s= github.com/lightningnetwork/lnd/healthcheck v1.2.6 h1:1sWhqr93GdkWy4+6U7JxBfcyZIE78MhIHTJZfPx7qqI= github.com/lightningnetwork/lnd/healthcheck v1.2.6/go.mod h1:Mu02um4CWY/zdTOvFje7WJgJcHyX2zq/FG3MhOAiGaQ= github.com/lightningnetwork/lnd/kvdb v1.4.16 h1:9BZgWdDfjmHRHLS97cz39bVuBAqMc4/p3HX1xtUdbDI= diff --git a/graph/builder.go b/graph/builder.go index 594093697..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) - oldEdges, 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 oldEdges { 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 6975c9d48..774492786 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -2028,10 +2028,13 @@ 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. - chanUpdates, err := graph.ChanUpdatesInHorizon( + chanIter := graph.ChanUpdatesInHorizon( time.Unix(999, 0), time.Unix(9999, 0), ) + + 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", len(chanUpdates)) @@ -2144,9 +2147,11 @@ func TestChanUpdatesInHorizon(t *testing.T) { }, } for _, queryCase := range queryCases { - resp, 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) } @@ -2189,9 +2194,10 @@ 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. - nodeUpdates, 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") require.Len(t, nodeUpdates, 0) @@ -2253,20 +2259,22 @@ func TestNodeUpdatesInHorizon(t *testing.T) { resp: nodeAnns, }, - // If we reduce the ending time by 10 seconds, then we should - // get all but the last node we inserted. + // If we reduce the ending time by 1 nanosecond before the last + // node's timestamp, then we should get all but the last node. { start: startTime, - end: endTime.Add(-time.Second * 10), + end: endTime.Add(-time.Second*10 - time.Nanosecond), resp: nodeAnns[:9], }, } for _, queryCase := range queryCases { - resp, err := graph.NodeUpdatesInHorizon( + iter := graph.NodeUpdatesInHorizon( queryCase.start, queryCase.end, ) - require.NoError(t, err) + + 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++ { @@ -2275,6 +2283,293 @@ 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) { + t.Parallel() + + ctx := t.Context() + + // Test with various batch sizes to ensure the iterator works correctly + // across batch boundaries. + batchSizes := []int{1, 3, 5, 10, 25, 100} + + for _, batchSize := range batchSizes { + testName := fmt.Sprintf("BatchSize%d", batchSize) + t.Run(testName, func(t *testing.T) { + testNodeUpdatesWithBatchSize(t, ctx, batchSize) + }) + } +} + +// TestNodeUpdatesInHorizonEarlyTermination tests that the iterator properly +// handles early termination when the caller stops iterating. +func TestNodeUpdatesInHorizonEarlyTermination(t *testing.T) { + t.Parallel() + ctx := t.Context() + + graph := MakeTestGraph(t) + + // We'll start by creating 100 nodes, each with an update time spaced + // one hour apart. + startTime := time.Unix(1234567890, 0) + for i := 0; i < 100; i++ { + nodeAnn := createTestVertex(t) + nodeAnn.LastUpdate = startTime.Add(time.Duration(i) * time.Hour) + require.NoError(t, graph.AddNode(ctx, nodeAnn)) + } + + // Test early termination at various points + terminationPoints := []int{0, 1, 5, 10, 23, 50, 99} + + for _, stopAt := range terminationPoints { + t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) { + iter := graph.NodeUpdatesInHorizon( + startTime, startTime.Add(200*time.Hour), + WithNodeUpdateIterBatchSize(10), + ) + + // Collect only up to stopAt nodes, breaking afterwards. + var collected []models.Node + count := 0 + for node := range iter { + if count >= stopAt { + break + } + collected = append(collected, node) + count++ + } + + require.Len( + t, collected, stopAt, + "should have collected exactly %d nodes", + stopAt, + ) + }) + } +} + +// TestChanUpdatesInHorizonBoundaryConditions tests the channel iterator +// boundary conditions. +func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) { + t.Parallel() + ctx := t.Context() + + batchSizes := []int{1, 3, 5, 10} + + for _, batchSize := range batchSizes { + 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) + node1 := createTestVertex(t) + node2 := createTestVertex(t) + require.NoError(t, graph.AddNode(ctx, node1)) + require.NoError(t, graph.AddNode(ctx, node2)) + + // Next, we'll create 25 channels between the two nodes, + // each with increasing timestamps. + startTime := time.Unix(1234567890, 0) + const numChans = 25 + + for i := 0; i < numChans; i++ { + updateTime := startTime.Add( + time.Duration(i) * time.Hour, + ) + + channel, chanID := createEdge( + uint32(i*10), 0, 0, 0, node1, node2, + ) + require.NoError( + t, graph.AddChannelEdge(ctx, &channel), + ) + + edge1 := newEdgePolicy( + chanID.ToUint64(), updateTime.Unix(), + ) + edge1.ChannelFlags = 0 + edge1.ToNode = node2.PubKeyBytes + edge1.SigBytes = testSig.Serialize() + require.NoError( + t, graph.UpdateEdgePolicy(ctx, edge1), + ) + + edge2 := newEdgePolicy( + chanID.ToUint64(), updateTime.Unix(), + ) + edge2.ChannelFlags = 1 + edge2.ToNode = node1.PubKeyBytes + edge2.SigBytes = testSig.Serialize() + 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 := graph.ChanUpdatesInHorizon( + startTime, startTime.Add(26*time.Hour), + WithChanUpdateIterBatchSize(batchSize), + ) + + channels, err := fn.CollectErr(iter) + require.NoError(t, err) + require.Len( + t, channels, numChans, + "expected %d channels, got %d", numChans, + len(channels), + ) + }) + } +} + // TestFilterKnownChanIDsZombieRevival tests that if a ChannelUpdateInfo is // passed to FilterKnownChanIDs that contains a channel that we have marked as // a zombie, then we will mark it as live again if the new ChannelUpdate has @@ -2731,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 }, @@ -3507,11 +3803,12 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) { // update time of our test node. startTime := time.Unix(9, 0) endTime := node1.LastUpdate.Add(time.Minute) - nodesInHorizon, 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, 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)) @@ -3525,7 +3822,8 @@ 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. - nodesInHorizon, 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") if len(nodesInHorizon) != 0 { diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index f90c2da8a..25eb6f5b1 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -2,6 +2,7 @@ package graphdb import ( "context" + "iter" "net" "time" @@ -109,8 +110,8 @@ type V1Store interface { //nolint:interfacebloat // an update timestamp within the passed range. This method can be used // by two nodes to quickly determine if they have the same set of up to // date node announcements. - NodeUpdatesInHorizon(startTime, - endTime time.Time) ([]models.Node, error) + NodeUpdatesInHorizon(startTime, endTime time.Time, + 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 @@ -219,8 +220,8 @@ type V1Store 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) ([]ChannelEdge, - error) + ChanUpdatesInHorizon(startTime, endTime time.Time, + 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 165f51987..cc9a14889 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io" + "iter" "math" "net" "sort" @@ -2039,107 +2040,202 @@ type ChannelEdge struct { Node2 *models.Node } -// 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) ([]ChannelEdge, error) { - - // To ensure we don't return duplicate ChannelEdges, we'll use an - // additional map to keep track of the edges already seen to prevent - // re-adding it. - var edgesSeen map[uint64]struct{} - var edgesToCache map[uint64]ChannelEdge - var edgesInHorizon []ChannelEdge +// updateChanCacheBatch updates the channel cache with multiple edges at once. +// This method acquires the cache lock only once for the entire batch. +func (c *KVStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) { + if len(edgesToCache) == 0 { + return + } c.cacheMu.Lock() defer c.cacheMu.Unlock() - var hits int + for cid, edge := range edgesToCache { + c.chanCache.insert(cid, edge) + } +} + +// isEmptyGraphError returns true if the error indicates the graph database +// is empty (no edges or nodes exist). These errors are expected when the +// graph is first created or has no data. +func isEmptyGraphError(err error) bool { + return errors.Is(err, ErrGraphNoEdgesFound) || + errors.Is(err, ErrGraphNodesNotFound) +} + +// chanUpdatesIterator holds the state for chunked channel update iteration. +type chanUpdatesIterator struct { + // batchSize is the amount of channel updates to read at a single time. + batchSize int + + // startTime is the start time of the iteration request. + startTime time.Time + + // endTime is the end time of the iteration request. + endTime time.Time + + // edgesSeen is used to dedup edges. + edgesSeen map[uint64]struct{} + + // edgesToCache houses all the edges that we read from the disk which + // aren't yet cached. This is used to update the cache after a batch + // chunk. + edgesToCache map[uint64]ChannelEdge + + // lastSeenKey is the last index key seen. This is used to resume + // iteration. + lastSeenKey []byte + + // hits is the number of cache hits. + hits int + + // total is the total number of edges requested. + total int +} + +// newChanUpdatesIterator makes a new chan updates iterator. +func newChanUpdatesIterator(batchSize int, + startTime, endTime time.Time) *chanUpdatesIterator { + + return &chanUpdatesIterator{ + batchSize: batchSize, + startTime: startTime, + endTime: endTime, + edgesSeen: make(map[uint64]struct{}), + edgesToCache: make(map[uint64]ChannelEdge), + lastSeenKey: nil, + } +} + +// fetchNextChanUpdateBatch retrieves the next batch of channel edges within the +// horizon. Returns the batch, whether there are more edges, and any error. +func (c *KVStore) fetchNextChanUpdateBatch( + state *chanUpdatesIterator) ([]ChannelEdge, bool, error) { + + var ( + batch []ChannelEdge + hasMore bool + ) err := kvdb.View(c.db, func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) if edges == nil { return ErrGraphNoEdgesFound } + edgeIndex := edges.NestedReadBucket(edgeIndexBucket) if edgeIndex == nil { return ErrGraphNoEdgesFound } + edgeUpdateIndex := edges.NestedReadBucket(edgeUpdateIndexBucket) if edgeUpdateIndex == nil { return ErrGraphNoEdgesFound } - nodes := tx.ReadBucket(nodeBucket) if nodes == nil { return ErrGraphNodesNotFound } - // We'll now obtain a cursor to perform a range query within - // the index to find all channels within the horizon. + // With all the relevant buckets read, we'll now create a fresh + // read cursor. updateCursor := edgeUpdateIndex.ReadCursor() + // We'll now use the start and end time to create the keys that + // we'll use to seek. var startTimeBytes, endTimeBytes [8 + 8]byte byteOrder.PutUint64( - startTimeBytes[:8], uint64(startTime.Unix()), + startTimeBytes[:8], uint64(state.startTime.Unix()), ) byteOrder.PutUint64( - endTimeBytes[:8], uint64(endTime.Unix()), + endTimeBytes[:8], uint64(state.endTime.Unix()), ) - // With our start and end times constructed, we'll step through - // the index collecting the info and policy of each update of - // each channel that has a last update within the time range. - // - //nolint:ll - for indexKey, _ := updateCursor.Seek(startTimeBytes[:]); indexKey != nil && - bytes.Compare(indexKey, endTimeBytes[:]) <= 0; indexKey, _ = updateCursor.Next() { - // We have a new eligible entry, so we'll slice of the - // chan ID so we can query it in the DB. + var indexKey []byte + + // If we left off earlier, then we'll use that key as the + // starting point. + switch { + case state.lastSeenKey != nil: + // Seek to the last seen key, moving to the key right + // after it. + indexKey, _ = updateCursor.Seek(state.lastSeenKey) + + if bytes.Equal(indexKey, state.lastSeenKey) { + indexKey, _ = updateCursor.Next() + } + + // Otherwise, we'll move to the very start of the time range. + default: + indexKey, _ = updateCursor.Seek(startTimeBytes[:]) + } + + // TODO(roasbeef): iterate the channel graph cache instead w/ a + // treap ordering? + + // Now we'll read items up to the batch size, exiting early if + // we exceed the ending time. + for len(batch) < state.batchSize && indexKey != nil { + // If we're at the end, then we'll break out now. + if bytes.Compare(indexKey, endTimeBytes[:]) > 0 { + break + } + chanID := indexKey[8:] - - // If we've already retrieved the info and policies for - // this edge, then we can skip it as we don't need to do - // so again. chanIDInt := byteOrder.Uint64(chanID) - if _, ok := edgesSeen[chanIDInt]; ok { + + if state.lastSeenKey == nil { + state.lastSeenKey = make([]byte, len(indexKey)) + } + copy(state.lastSeenKey, indexKey) + + // If we've seen this channel ID already, then we'll + // skip it. + if _, ok := state.edgesSeen[chanIDInt]; ok { + indexKey, _ = updateCursor.Next() continue } + // Before we read the edge info, we'll see if this + // element is already in the cache or not. + c.cacheMu.RLock() if channel, ok := c.chanCache.get(chanIDInt); ok { - hits++ - edgesSeen[chanIDInt] = struct{}{} - edgesInHorizon = append(edgesInHorizon, channel) + state.edgesSeen[chanIDInt] = struct{}{} + + batch = append(batch, channel) + + state.hits++ + state.total++ + + indexKey, _ = updateCursor.Next() + + c.cacheMu.RUnlock() continue } + c.cacheMu.RUnlock() - // First, we'll fetch the static edge information. + // The edge wasn't in the cache, so we'll fetch it along + // w/ the edge policies and nodes. edgeInfo, err := fetchChanEdgeInfo(edgeIndex, chanID) if err != nil { - chanID := byteOrder.Uint64(chanID) - return fmt.Errorf("unable to fetch info for "+ - "edge with chan_id=%v: %v", chanID, err) + return fmt.Errorf("unable to fetch info "+ + "for edge with chan_id=%v: %v", + chanIDInt, err) } - - // With the static information obtained, we'll now - // fetch the dynamic policy info. edge1, edge2, err := fetchChanEdgePolicies( edgeIndex, edges, chanID, ) if err != nil { - chanID := byteOrder.Uint64(chanID) - return fmt.Errorf("unable to fetch policies "+ - "for edge with chan_id=%v: %v", chanID, - err) + return fmt.Errorf("unable to fetch "+ + "policies for edge with chan_id=%v: %v", + chanIDInt, err) } - node1, err := fetchLightningNode( nodes, edgeInfo.NodeKey1Bytes[:], ) if err != nil { return err } - node2, err := fetchLightningNode( nodes, edgeInfo.NodeKey2Bytes[:], ) @@ -2147,9 +2243,8 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime, return err } - // Finally, we'll collate this edge with the rest of - // edges to be returned. - edgesSeen[chanIDInt] = struct{}{} + // Now we have all the information we need to build the + // channel edge. channel := ChannelEdge{ Info: &edgeInfo, Policy1: edge1, @@ -2157,94 +2252,274 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime, Node1: &node1, Node2: &node2, } - edgesInHorizon = append(edgesInHorizon, channel) - edgesToCache[chanIDInt] = channel + + state.edgesSeen[chanIDInt] = struct{}{} + state.edgesToCache[chanIDInt] = channel + + batch = append(batch, channel) + + state.total++ + + // Advance the iterator to the next entry. + indexKey, _ = updateCursor.Next() + } + + // If we haven't yet crossed the endTimeBytes, then we still + // have more entries to deliver. + if indexKey != nil && + bytes.Compare(indexKey, endTimeBytes[:]) <= 0 { + + hasMore = true } return nil }, func() { - edgesSeen = make(map[uint64]struct{}) - edgesToCache = make(map[uint64]ChannelEdge) - edgesInHorizon = nil + batch = nil + hasMore = false }) - switch { - case errors.Is(err, ErrGraphNoEdgesFound): - fallthrough - case errors.Is(err, ErrGraphNodesNotFound): - break - - case err != nil: - return nil, err + if err != nil { + return nil, false, err } - // Insert any edges loaded from disk into the cache. - for chanid, channel := range edgesToCache { - c.chanCache.insert(chanid, channel) - } - - if len(edgesInHorizon) > 0 { - log.Debugf("ChanUpdatesInHorizon hit percentage: %.2f (%d/%d)", - float64(hits)*100/float64(len(edgesInHorizon)), hits, - len(edgesInHorizon)) - } else { - log.Debugf("ChanUpdatesInHorizon returned no edges in "+ - "horizon (%s, %s)", startTime, endTime) - } - - return edgesInHorizon, nil + return batch, hasMore, nil } -// NodeUpdatesInHorizon returns all the known lightning node which have an -// update timestamp within the passed range. This method can be used by two -// nodes to quickly determine if they have the same set of up to date node -// announcements. -func (c *KVStore) NodeUpdatesInHorizon(startTime, - endTime time.Time) ([]models.Node, error) { +// 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.Seq2[ChannelEdge, error] { - var nodesInHorizon []models.Node + cfg := defaultIteratorConfig() + for _, opt := range opts { + opt(cfg) + } + + return func(yield func(ChannelEdge, error) bool) { + iterState := newChanUpdatesIterator( + cfg.chanUpdateIterBatchSize, startTime, endTime, + ) + + for { + // At the top of the loop, we'll read the next batch + // chunk from disk. We'll also determine if we have any + // more entries after this or not. + batch, hasMore, err := c.fetchNextChanUpdateBatch( + iterState, + ) + 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 + } + + // 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, nil) { + return + } + } + + // Update cache after successful batch yield. + c.updateChanCacheBatch(iterState.edgesToCache) + iterState.edgesToCache = make(map[uint64]ChannelEdge) + + // If we we're done, then we can just break out here + // now. + if !hasMore || len(batch) == 0 { + break + } + } + + if iterState.total > 0 { + log.Tracef("ChanUpdatesInHorizon hit percentage: "+ + "%.2f (%d/%d)", float64(iterState.hits)*100/ + float64(iterState.total), iterState.hits, + iterState.total) + } else { + log.Tracef("ChanUpdatesInHorizon returned no edges "+ + "in horizon (%s, %s)", startTime, endTime) + } + } +} + +// 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. +type nodeUpdatesIterator struct { + // batchSize is the amount of node updates to read at a single time. + batchSize int + + // startTime is the start time of the iteration request. + startTime time.Time + + // endTime is the end time of the iteration request. + endTime time.Time + + // lastSeenKey is the last index key seen. This is used to resume + // iteration. + lastSeenKey []byte + + // publicNodesOnly filters to only return public nodes if true. + publicNodesOnly bool + + // total tracks total nodes processed. + total int +} + +// newNodeUpdatesIterator makes a new node updates iterator. +func newNodeUpdatesIterator(batchSize int, startTime, endTime time.Time, + publicNodesOnly bool) *nodeUpdatesIterator { + + return &nodeUpdatesIterator{ + batchSize: batchSize, + startTime: startTime, + endTime: endTime, + lastSeenKey: nil, + publicNodesOnly: publicNodesOnly, + } +} + +// fetchNextNodeBatch fetches the next batch of node announcements using the +// iterator state. +func (c *KVStore) fetchNextNodeBatch( + state *nodeUpdatesIterator) ([]models.Node, bool, error) { + + var ( + nodeBatch []models.Node + hasMore bool + ) err := kvdb.View(c.db, func(tx kvdb.RTx) error { nodes := tx.ReadBucket(nodeBucket) if nodes == nil { return ErrGraphNodesNotFound } - + ourPubKey := nodes.Get(sourceKey) + if ourPubKey == nil && state.publicNodesOnly { + // If we're filtering for public nodes only but don't + // have a source node set, we can't determine if nodes + // are public. A node is considered public if it has at + // least one channel with our node (the source node). + return ErrSourceNodeNotSet + } nodeUpdateIndex := nodes.NestedReadBucket(nodeUpdateIndexBucket) if nodeUpdateIndex == nil { return ErrGraphNodesNotFound } - // We'll now obtain a cursor to perform a range query within - // the index to find all node announcements within the horizon. + // We'll now obtain a cursor to perform a range query within the + // index to find all node announcements within the horizon. The + // nodeUpdateIndex key format is: [8 bytes timestamp][33 bytes + // node pubkey] This allows efficient range queries by time + // while maintaining a stable sort order for nodes with the same + // timestamp. updateCursor := nodeUpdateIndex.ReadCursor() var startTimeBytes, endTimeBytes [8 + 33]byte byteOrder.PutUint64( - startTimeBytes[:8], uint64(startTime.Unix()), + startTimeBytes[:8], uint64(state.startTime.Unix()), ) byteOrder.PutUint64( - endTimeBytes[:8], uint64(endTime.Unix()), + endTimeBytes[:8], uint64(state.endTime.Unix()), ) - // With our start and end times constructed, we'll step through - // the index collecting info for each node within the time - // range. - // - //nolint:ll - for indexKey, _ := updateCursor.Seek(startTimeBytes[:]); indexKey != nil && - bytes.Compare(indexKey, endTimeBytes[:]) <= 0; indexKey, _ = updateCursor.Next() { + // If we have a last seen key (existing iteration), then that'll + // be our starting point. Otherwise, we'll seek to the start + // time. + var indexKey []byte + if state.lastSeenKey != nil { + indexKey, _ = updateCursor.Seek(state.lastSeenKey) + + if bytes.Equal(indexKey, state.lastSeenKey) { + indexKey, _ = updateCursor.Next() + } + } else { + indexKey, _ = updateCursor.Seek(startTimeBytes[:]) + } + + // Now we'll read items up to the batch size, exiting early if + // we exceed the ending time. + var lastProcessedKey []byte + for len(nodeBatch) < state.batchSize && indexKey != nil { + // Extract the timestamp from the index key (first 8 + // bytes). Only compare timestamps, not the full key + // with pubkey. + keyTimestamp := byteOrder.Uint64(indexKey[:8]) + endTimestamp := uint64(state.endTime.Unix()) + if keyTimestamp > endTimestamp { + break + } + nodePub := indexKey[8:] node, err := fetchLightningNode(nodes, nodePub) if err != nil { return err } - nodesInHorizon = append(nodesInHorizon, node) + if state.publicNodesOnly { + nodeIsPublic, err := c.isPublic( + tx, node.PubKeyBytes, ourPubKey, + ) + if err != nil { + return err + } + if !nodeIsPublic { + indexKey, _ = updateCursor.Next() + continue + } + } + + nodeBatch = append(nodeBatch, node) + state.total++ + + // Remember the last key we actually processed. We'll + // use this to update the last seen key below. + if lastProcessedKey == nil { + lastProcessedKey = make([]byte, len(indexKey)) + } + copy(lastProcessedKey, indexKey) + + // Advance the iterator to the next entry. + indexKey, _ = updateCursor.Next() + } + + // If we haven't yet crossed the endTime, then we still + // have more entries to deliver. + if indexKey != nil { + keyTimestamp := byteOrder.Uint64(indexKey[:8]) + endTimestamp := uint64(state.endTime.Unix()) + if keyTimestamp <= endTimestamp { + hasMore = true + } + } + + // Update the cursor to the last key we actually processed. + if lastProcessedKey != nil { + if state.lastSeenKey == nil { + state.lastSeenKey = make( + []byte, len(lastProcessedKey), + ) + } + copy(state.lastSeenKey, lastProcessedKey) } return nil }, func() { - nodesInHorizon = nil + nodeBatch = nil }) switch { case errors.Is(err, ErrGraphNoEdgesFound): @@ -2253,10 +2528,55 @@ func (c *KVStore) NodeUpdatesInHorizon(startTime, break case err != nil: - return nil, err + return nil, false, err } - return nodesInHorizon, nil + return nodeBatch, hasMore, nil +} + +// NodeUpdatesInHorizon returns all the known lightning node which have an +// update timestamp within the passed range. +func (c *KVStore) NodeUpdatesInHorizon(startTime, + endTime time.Time, + opts ...IteratorOption) iter.Seq2[models.Node, error] { + + cfg := defaultIteratorConfig() + for _, opt := range opts { + opt(cfg) + } + + return func(yield func(models.Node, error) bool) { + // Initialize iterator state. + state := newNodeUpdatesIterator( + cfg.nodeUpdateIterBatchSize, + startTime, endTime, + cfg.iterPublicNodes, + ) + + for { + nodeAnns, hasMore, err := c.fetchNextNodeBatch(state) + if err != nil { + log.Errorf("unable to read node updates in "+ + "horizon: %v", err) + + yield(models.Node{}, err) + + return + } + + for _, node := range nodeAnns { + if !yield(node, nil) { + return + } + } + + // If we we're done, then we can just break out here + // now. + if !hasMore || len(nodeAnns) == 0 { + break + } + } + } } // FilterKnownChanIDs takes a set of channel IDs and return the subset of chan diff --git a/graph/db/options.go b/graph/db/options.go index 3edda6602..15ea6f4ee 100644 --- a/graph/db/options.go +++ b/graph/db/options.go @@ -20,6 +20,60 @@ const ( DefaultPreAllocCacheNumNodes = 15000 ) +// IteratorOption is a functional option used to change the per-call +// configuration for iterators. +type IteratorOption func(*iterConfig) + +// iterConfig holds the configuration for graph operations. +type iterConfig struct { + // chanUpdateIterBatchSize is the batch size to use when reading out + // channel updates to send a peer a backlog. + chanUpdateIterBatchSize int + + // nodeUpdateIterBatchSize is the batch size to use when reading out + // node updates to send to a peer backlog. + nodeUpdateIterBatchSize int + + // iterPublicNodes is used to make an iterator that only iterates over + // public nodes. + iterPublicNodes bool +} + +// defaultIteratorConfig returns the default configuration. +func defaultIteratorConfig() *iterConfig { + return &iterConfig{ + chanUpdateIterBatchSize: 1_000, + nodeUpdateIterBatchSize: 1_000, + } +} + +// WithChanUpdateIterBatchSize sets the batch size for channel update +// iterators. +func WithChanUpdateIterBatchSize(size int) IteratorOption { + return func(cfg *iterConfig) { + if size > 0 { + cfg.chanUpdateIterBatchSize = size + } + } +} + +// WithNodeUpdateIterBatchSize set the batch size for node ann iterators. +func WithNodeUpdateIterBatchSize(size int) IteratorOption { + return func(cfg *iterConfig) { + if size > 0 { + cfg.nodeUpdateIterBatchSize = size + } + } +} + +// WithIterPublicNodesOnly is used to create an iterator that only iterates over +// public nodes. +func WithIterPublicNodesOnly() IteratorOption { + return func(cfg *iterConfig) { + cfg.iterPublicNodes = true + } +} + // chanGraphOptions holds parameters for tuning and customizing the // ChannelGraph. type chanGraphOptions struct { diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 23fd50482..f13408765 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -7,6 +7,7 @@ import ( "encoding/hex" "errors" "fmt" + "iter" "maps" "math" "net" @@ -552,42 +553,105 @@ func (s *SQLStore) SetSourceNode(ctx context.Context, // announcements. // // NOTE: This is part of the V1Store interface. -func (s *SQLStore) NodeUpdatesInHorizon(startTime, - endTime time.Time) ([]models.Node, error) { +func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time, + opts ...IteratorOption) iter.Seq2[models.Node, error] { - ctx := context.TODO() - - var nodes []models.Node - err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { - dbNodes, err := db.GetNodesByLastUpdateRange( - ctx, sqlc.GetNodesByLastUpdateRangeParams{ - StartTime: sqldb.SQLInt64(startTime.Unix()), - EndTime: sqldb.SQLInt64(endTime.Unix()), - }, - ) - if err != nil { - return fmt.Errorf("unable to fetch nodes: %w", err) - } - - err = forEachNodeInBatch( - ctx, s.cfg.QueryCfg, db, dbNodes, - func(_ int64, node *models.Node) error { - nodes = append(nodes, *node) - - return nil - }, - ) - if err != nil { - return fmt.Errorf("unable to build nodes: %w", err) - } - - return nil - }, sqldb.NoOpReset) - if err != nil { - return nil, fmt.Errorf("unable to fetch nodes: %w", err) + cfg := defaultIteratorConfig() + for _, opt := range opts { + opt(cfg) } - return nodes, nil + return func(yield func(models.Node, error) bool) { + var ( + ctx = context.TODO() + lastUpdateTime sql.NullInt64 + lastPubKey = make([]byte, 33) + hasMore = true + ) + + // Each iteration, we'll read a batch amount of nodes, yield + // them, then decide is we have more or not. + for hasMore { + var batch []models.Node + + //nolint:ll + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + //nolint:ll + params := sqlc.GetNodesByLastUpdateRangeParams{ + StartTime: sqldb.SQLInt64( + startTime.Unix(), + ), + EndTime: sqldb.SQLInt64( + endTime.Unix(), + ), + LastUpdate: lastUpdateTime, + LastPubKey: lastPubKey, + OnlyPublic: sql.NullBool{ + Bool: cfg.iterPublicNodes, + Valid: true, + }, + MaxResults: sqldb.SQLInt32( + cfg.nodeUpdateIterBatchSize, + ), + } + rows, err := db.GetNodesByLastUpdateRange( + ctx, params, + ) + if err != nil { + return err + } + + hasMore = len(rows) == cfg.nodeUpdateIterBatchSize + + err = forEachNodeInBatch( + ctx, s.cfg.QueryCfg, db, rows, + func(_ int64, node *models.Node) error { + batch = append(batch, *node) + + // Update pagination cursors + // based on the last processed + // node. + lastUpdateTime = sql.NullInt64{ + Int64: node.LastUpdate. + Unix(), + Valid: true, + } + lastPubKey = node.PubKeyBytes[:] + + return nil + }, + ) + if err != nil { + return fmt.Errorf("unable to build "+ + "nodes: %w", err) + } + + return nil + }, func() { + batch = []models.Node{} + }) + + if err != nil { + log.Errorf("NodeUpdatesInHorizon batch "+ + "error: %v", err) + + yield(models.Node{}, err) + + return + } + + for _, node := range batch { + if !yield(node, nil) { + return + } + } + + // If the batch didn't yield anything, then we're done. + if len(batch) == 0 { + break + } + } + } } // AddChannelEdge adds a new (undirected, blank) edge to the graph database. An @@ -920,112 +984,254 @@ func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, }, reset) } -// ChanUpdatesInHorizon returns all the known channel edges which have at least -// one edge that has an update timestamp within the specified horizon. -// -// NOTE: This is part of the V1Store interface. -func (s *SQLStore) ChanUpdatesInHorizon(startTime, - endTime time.Time) ([]ChannelEdge, error) { +// extractMaxUpdateTime returns the maximum of the two policy update times. +// This is used for pagination cursor tracking. +func extractMaxUpdateTime( + row sqlc.GetChannelsByPolicyLastUpdateRangeRow) int64 { + + switch { + case row.Policy1LastUpdate.Valid && row.Policy2LastUpdate.Valid: + return max(row.Policy1LastUpdate.Int64, + row.Policy2LastUpdate.Int64) + case row.Policy1LastUpdate.Valid: + return row.Policy1LastUpdate.Int64 + case row.Policy2LastUpdate.Valid: + return row.Policy2LastUpdate.Int64 + default: + return 0 + } +} + +// buildChannelFromRow constructs a ChannelEdge from a database row. +// This includes building the nodes, channel info, and policies. +func (s *SQLStore) buildChannelFromRow(ctx context.Context, db SQLQueries, + row sqlc.GetChannelsByPolicyLastUpdateRangeRow) (ChannelEdge, error) { + + node1, err := buildNode(ctx, s.cfg.QueryCfg, db, row.GraphNode) + if err != nil { + return ChannelEdge{}, fmt.Errorf("unable to build node1: %w", + err) + } + + node2, err := buildNode(ctx, s.cfg.QueryCfg, db, row.GraphNode_2) + if err != nil { + return ChannelEdge{}, fmt.Errorf("unable to build node2: %w", + err) + } + + channel, err := getAndBuildEdgeInfo( + ctx, s.cfg, db, + row.GraphChannel, node1.PubKeyBytes, + node2.PubKeyBytes, + ) + if err != nil { + return ChannelEdge{}, fmt.Errorf("unable to build "+ + "channel info: %w", err) + } + + dbPol1, dbPol2, err := extractChannelPolicies(row) + if err != nil { + return ChannelEdge{}, fmt.Errorf("unable to extract "+ + "channel policies: %w", err) + } + + p1, p2, err := getAndBuildChanPolicies( + ctx, s.cfg.QueryCfg, db, dbPol1, dbPol2, channel.ChannelID, + node1.PubKeyBytes, node2.PubKeyBytes, + ) + if err != nil { + return ChannelEdge{}, fmt.Errorf("unable to build "+ + "channel policies: %w", err) + } + + return ChannelEdge{ + Info: channel, + Policy1: p1, + Policy2: p2, + Node1: node1, + Node2: node2, + }, nil +} + +// updateChanCacheBatch updates the channel cache with multiple edges at once. +// This method acquires the cache lock only once for the entire batch. +func (s *SQLStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) { + if len(edgesToCache) == 0 { + return + } s.cacheMu.Lock() defer s.cacheMu.Unlock() - var ( - ctx = context.TODO() - // To ensure we don't return duplicate ChannelEdges, we'll use - // an additional map to keep track of the edges already seen to - // prevent re-adding it. - edgesSeen = make(map[uint64]struct{}) - edgesToCache = make(map[uint64]ChannelEdge) - edges []ChannelEdge - hits int - ) + for chanID, edge := range edgesToCache { + s.chanCache.insert(chanID, edge) + } +} - err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { - rows, err := db.GetChannelsByPolicyLastUpdateRange( - ctx, sqlc.GetChannelsByPolicyLastUpdateRangeParams{ - Version: int16(ProtocolV1), - StartTime: sqldb.SQLInt64(startTime.Unix()), - EndTime: sqldb.SQLInt64(endTime.Unix()), - }, +// ChanUpdatesInHorizon returns all the known channel edges which have at least +// one edge that has an update timestamp within the specified horizon. +// +// Iterator Lifecycle: +// 1. Initialize state (edgesSeen map, cache tracking, pagination cursors) +// 2. Query batch of channels with policies in time range +// 3. For each channel: check if seen, check cache, or build from DB +// 4. Yield channels to caller +// 5. Update cache after successful batch +// 6. Repeat with updated pagination cursor until no more results +// +// NOTE: This is part of the V1Store interface. +func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time, + opts ...IteratorOption) iter.Seq2[ChannelEdge, error] { + + // Apply options. + cfg := defaultIteratorConfig() + for _, opt := range opts { + opt(cfg) + } + + return func(yield func(ChannelEdge, error) bool) { + var ( + ctx = context.TODO() + edgesSeen = make(map[uint64]struct{}) + edgesToCache = make(map[uint64]ChannelEdge) + hits int + total int + lastUpdateTime sql.NullInt64 + lastID sql.NullInt64 + hasMore = true ) - if err != nil { - return err - } - if len(rows) == 0 { - return nil - } + // Each iteration, we'll read a batch amount of channel updates + // (consulting the cache along the way), yield them, then loop + // back to decide if we have any more updates to read out. + for hasMore { + var batch []ChannelEdge - // We'll pre-allocate the slices and maps here with a best - // effort size in order to avoid unnecessary allocations later - // on. - uncachedRows := make( - []sqlc.GetChannelsByPolicyLastUpdateRangeRow, 0, - len(rows), - ) - edgesToCache = make(map[uint64]ChannelEdge, len(rows)) - edgesSeen = make(map[uint64]struct{}, len(rows)) - edges = make([]ChannelEdge, 0, len(rows)) + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), + func(db SQLQueries) error { + //nolint:ll + params := sqlc.GetChannelsByPolicyLastUpdateRangeParams{ + Version: int16(ProtocolV1), + StartTime: sqldb.SQLInt64( + startTime.Unix(), + ), + EndTime: sqldb.SQLInt64( + endTime.Unix(), + ), + LastUpdateTime: lastUpdateTime, + LastID: lastID, + MaxResults: sql.NullInt32{ + Int32: int32( + cfg.chanUpdateIterBatchSize, + ), + Valid: true, + }, + } + //nolint:ll + rows, err := db.GetChannelsByPolicyLastUpdateRange( + ctx, params, + ) + if err != nil { + return err + } - // Separate cached from non-cached channels since we will only - // batch load the data for the ones we haven't cached yet. - for _, row := range rows { - chanIDInt := byteOrder.Uint64(row.GraphChannel.Scid) + //nolint:ll + hasMore = len(rows) == cfg.chanUpdateIterBatchSize - // Skip duplicates. - if _, ok := edgesSeen[chanIDInt]; ok { - continue - } - edgesSeen[chanIDInt] = struct{}{} + //nolint:ll + for _, row := range rows { + lastUpdateTime = sql.NullInt64{ + Int64: extractMaxUpdateTime(row), + Valid: true, + } + lastID = sql.NullInt64{ + Int64: row.GraphChannel.ID, + Valid: true, + } - // Check cache first. - if channel, ok := s.chanCache.get(chanIDInt); ok { - hits++ - edges = append(edges, channel) - continue + // Skip if we've already + // processed this channel. + chanIDInt := byteOrder.Uint64( + row.GraphChannel.Scid, + ) + _, ok := edgesSeen[chanIDInt] + if ok { + continue + } + + s.cacheMu.RLock() + channel, ok := s.chanCache.get( + chanIDInt, + ) + s.cacheMu.RUnlock() + if ok { + hits++ + total++ + edgesSeen[chanIDInt] = struct{}{} + batch = append(batch, channel) + + continue + } + + chanEdge, err := s.buildChannelFromRow( + ctx, db, row, + ) + if err != nil { + return err + } + + edgesSeen[chanIDInt] = struct{}{} + edgesToCache[chanIDInt] = chanEdge + + batch = append(batch, chanEdge) + + total++ + } + + return nil + }, func() { + batch = nil + edgesSeen = make(map[uint64]struct{}) + edgesToCache = make( + map[uint64]ChannelEdge, + ) + }) + + if err != nil { + log.Errorf("ChanUpdatesInHorizon "+ + "batch error: %v", err) + + yield(ChannelEdge{}, err) + + return } - // Mark this row as one we need to batch load data for. - uncachedRows = append(uncachedRows, row) + for _, edge := range batch { + if !yield(edge, nil) { + return + } + } + + // Update cache after successful batch yield, setting + // the cache lock only once for the entire batch. + s.updateChanCacheBatch(edgesToCache) + edgesToCache = make(map[uint64]ChannelEdge) + + // If the batch didn't yield anything, then we're done. + if len(batch) == 0 { + break + } } - // If there are no uncached rows, then we can return early. - if len(uncachedRows) == 0 { - return nil + if total > 0 { + log.Debugf("ChanUpdatesInHorizon hit percentage: "+ + "%.2f (%d/%d)", + float64(hits)*100/float64(total), hits, total) + } else { + log.Debugf("ChanUpdatesInHorizon returned no edges "+ + "in horizon (%s, %s)", startTime, endTime) } - - // Batch load data for all uncached channels. - newEdges, err := batchBuildChannelEdges( - ctx, s.cfg, db, uncachedRows, - ) - if err != nil { - return fmt.Errorf("unable to batch build channel "+ - "edges: %w", err) - } - - edges = append(edges, newEdges...) - - return nil - }, sqldb.NoOpReset) - if err != nil { - return nil, fmt.Errorf("unable to fetch channels: %w", err) } - - // Insert any edges loaded from disk into the cache. - for chanid, channel := range edgesToCache { - s.chanCache.insert(chanid, channel) - } - - if len(edges) > 0 { - log.Debugf("ChanUpdatesInHorizon hit percentage: %.2f (%d/%d)", - float64(hits)*100/float64(len(edges)), hits, len(edges)) - } else { - log.Debugf("ChanUpdatesInHorizon returned no edges in "+ - "horizon (%s, %s)", startTime, endTime) - } - - return edges, nil } // ForEachNodeCached is similar to forEachNode, but it returns DirectedChannel diff --git a/graph/interfaces.go b/graph/interfaces.go index ec4f42b79..0896a0850 100644 --- a/graph/interfaces.go +++ b/graph/interfaces.go @@ -2,6 +2,7 @@ package graph import ( "context" + "iter" "time" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -152,8 +153,9 @@ type DB interface { // 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) ( - []graphdb.ChannelEdge, error) + ChanUpdatesInHorizon(startTime, endTime time.Time, + 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 diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 4c2be19ea..9c2702737 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -1143,18 +1143,39 @@ WHERE c.version = $1 OR (cp2.last_update >= $2 AND cp2.last_update < $3) ) + -- Pagination using compound cursor (max_update_time, id). + -- We use COALESCE with -1 as sentinel since timestamps are always positive. + AND ( + (CASE + WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0) + THEN COALESCE(cp1.last_update, 0) + ELSE COALESCE(cp2.last_update, 0) + END > COALESCE($4, -1)) + OR + (CASE + WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0) + THEN COALESCE(cp1.last_update, 0) + ELSE COALESCE(cp2.last_update, 0) + END = COALESCE($4, -1) + AND c.id > COALESCE($5, -1)) + ) ORDER BY CASE WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0) THEN COALESCE(cp1.last_update, 0) ELSE COALESCE(cp2.last_update, 0) - END ASC + END ASC, + c.id ASC +LIMIT COALESCE($6, 999999999) ` type GetChannelsByPolicyLastUpdateRangeParams struct { - Version int16 - StartTime sql.NullInt64 - EndTime sql.NullInt64 + Version int16 + StartTime sql.NullInt64 + EndTime sql.NullInt64 + LastUpdateTime sql.NullInt64 + LastID sql.NullInt64 + MaxResults interface{} } type GetChannelsByPolicyLastUpdateRangeRow struct { @@ -1194,7 +1215,14 @@ type GetChannelsByPolicyLastUpdateRangeRow struct { } func (q *Queries) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg GetChannelsByPolicyLastUpdateRangeParams) ([]GetChannelsByPolicyLastUpdateRangeRow, error) { - rows, err := q.db.QueryContext(ctx, getChannelsByPolicyLastUpdateRange, arg.Version, arg.StartTime, arg.EndTime) + rows, err := q.db.QueryContext(ctx, getChannelsByPolicyLastUpdateRange, + arg.Version, + arg.StartTime, + arg.EndTime, + arg.LastUpdateTime, + arg.LastID, + arg.MaxResults, + ) if err != nil { return nil, err } @@ -1969,16 +1997,55 @@ const getNodesByLastUpdateRange = `-- name: GetNodesByLastUpdateRange :many SELECT id, version, pub_key, alias, last_update, color, signature FROM graph_nodes WHERE last_update >= $1 - AND last_update < $2 + AND last_update <= $2 + -- Pagination: We use (last_update, pub_key) as a compound cursor. + -- This ensures stable ordering and allows us to resume from where we left off. + -- We use COALESCE with -1 as sentinel since timestamps are always positive. + AND ( + -- Include rows with last_update greater than cursor (or all rows if cursor is -1) + last_update > COALESCE($3, -1) + OR + -- For rows with same last_update, use pub_key as tiebreaker + (last_update = COALESCE($3, -1) + AND pub_key > $4) + ) + -- Optional filter for public nodes only + AND ( + -- If only_public is false or not provided, include all nodes + COALESCE($5, FALSE) IS FALSE + OR + -- For V1 protocol, a node is public if it has at least one public channel. + -- A public channel has bitcoin_1_signature set (channel announcement received). + EXISTS ( + SELECT 1 + FROM graph_channels c + WHERE c.version = 1 + AND c.bitcoin_1_signature IS NOT NULL + AND (c.node_id_1 = graph_nodes.id OR c.node_id_2 = graph_nodes.id) + ) + ) +ORDER BY last_update ASC, pub_key ASC +LIMIT COALESCE($6, 999999999) ` type GetNodesByLastUpdateRangeParams struct { - StartTime sql.NullInt64 - EndTime sql.NullInt64 + StartTime sql.NullInt64 + EndTime sql.NullInt64 + LastUpdate sql.NullInt64 + LastPubKey []byte + OnlyPublic interface{} + MaxResults interface{} } func (q *Queries) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]GraphNode, error) { - rows, err := q.db.QueryContext(ctx, getNodesByLastUpdateRange, arg.StartTime, arg.EndTime) + rows, err := q.db.QueryContext(ctx, getNodesByLastUpdateRange, + arg.StartTime, + arg.EndTime, + arg.LastUpdate, + arg.LastPubKey, + arg.OnlyPublic, + arg.MaxResults, + ) if err != nil { return nil, err } diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index f2224c00f..19087fc1b 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -164,7 +164,35 @@ ORDER BY node_id, type, position; SELECT * FROM graph_nodes WHERE last_update >= @start_time - AND last_update < @end_time; + AND last_update <= @end_time + -- Pagination: We use (last_update, pub_key) as a compound cursor. + -- This ensures stable ordering and allows us to resume from where we left off. + -- We use COALESCE with -1 as sentinel since timestamps are always positive. + AND ( + -- Include rows with last_update greater than cursor (or all rows if cursor is -1) + last_update > COALESCE(sqlc.narg('last_update'), -1) + OR + -- For rows with same last_update, use pub_key as tiebreaker + (last_update = COALESCE(sqlc.narg('last_update'), -1) + AND pub_key > sqlc.narg('last_pub_key')) + ) + -- Optional filter for public nodes only + AND ( + -- If only_public is false or not provided, include all nodes + COALESCE(sqlc.narg('only_public'), FALSE) IS FALSE + OR + -- For V1 protocol, a node is public if it has at least one public channel. + -- A public channel has bitcoin_1_signature set (channel announcement received). + EXISTS ( + SELECT 1 + FROM graph_channels c + WHERE c.version = 1 + AND c.bitcoin_1_signature IS NOT NULL + AND (c.node_id_1 = graph_nodes.id OR c.node_id_2 = graph_nodes.id) + ) + ) +ORDER BY last_update ASC, pub_key ASC +LIMIT COALESCE(sqlc.narg('max_results'), 999999999); -- name: DeleteNodeAddresses :exec DELETE FROM graph_node_addresses @@ -444,12 +472,30 @@ WHERE c.version = @version OR (cp2.last_update >= @start_time AND cp2.last_update < @end_time) ) + -- Pagination using compound cursor (max_update_time, id). + -- We use COALESCE with -1 as sentinel since timestamps are always positive. + AND ( + (CASE + WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0) + THEN COALESCE(cp1.last_update, 0) + ELSE COALESCE(cp2.last_update, 0) + END > COALESCE(sqlc.narg('last_update_time'), -1)) + OR + (CASE + WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0) + THEN COALESCE(cp1.last_update, 0) + ELSE COALESCE(cp2.last_update, 0) + END = COALESCE(sqlc.narg('last_update_time'), -1) + AND c.id > COALESCE(sqlc.narg('last_id'), -1)) + ) ORDER BY CASE WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0) THEN COALESCE(cp1.last_update, 0) ELSE COALESCE(cp2.last_update, 0) - END ASC; + END ASC, + c.id ASC +LIMIT COALESCE(sqlc.narg('max_results'), 999999999); -- name: GetChannelByOutpointWithPolicies :one SELECT