From 015875e455f1a71b0dcf1319ec08316511f78d84 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 4 Aug 2025 15:54:59 -0700 Subject: [PATCH 01/13] fn: add Collect+CollertErr function for iterators In this commit, we introduce a new utility function `Collect` to the fn package. This function drains all elements from an iterator and returns them as a slice. This is particularly useful when transitioning from iterator-based APIs to code that expects slices, allowing for gradual migration to the new iterator patterns. The fn module's go.mod is also updated to require Go 1.23, which is necessary for the built-in iter.Seq type support. The replace directive will be removed once the fn package changes are merged and a new version is tagged. --- fn/go.mod | 2 +- fn/iter.go | 29 +++++++++++++++++++++++++++++ go.mod | 3 +++ go.sum | 2 -- 4 files changed, 33 insertions(+), 3 deletions(-) create mode 100644 fn/iter.go 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= From f8ce00b84a20f0b3ca44866a895bfb732216705b Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 4 Aug 2025 15:55:19 -0700 Subject: [PATCH 02/13] graph/db: add options infrastructure for iterator configuration In this commit, we introduce a new options pattern for configuring iterator behavior in the graph database. This includes configuration for batch sizes when iterating over channel and node updates, as well as an option to filter for public nodes only. The new functional options pattern allows callers to customize iterator behavior without breaking existing APIs. Default batch sizes are set to 1000 entries for both channel and node updates, which provides a good balance between memory usage and performance. --- graph/db/options.go | 54 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) 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 { From 1d6d54e5db7c8aa5b9ad2eb4a73122c0670785b1 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 4 Aug 2025 17:08:30 -0700 Subject: [PATCH 03/13] graph/db: convert NodeUpdatesInHorizon to use iterators In this commit, we refactor the NodeUpdatesInHorizon method to return an iterator instead of a slice. This change significantly reduces memory usage when dealing with large result sets by allowing callers to process items incrementally rather than loading everything into memory at once. The new implementation uses Go 1.23's iter.Seq type to provide a standard iterator interface. The method now supports configurable batch sizes through functional options, allowing fine-tuned control over memory usage and performance characteristics. Rather than reading all the entries from disk into memory (before this commit, we did consult the cache for most entries, skipping the disk hits), we now expose a chunked iterator instead. We also make the process of filtering out public nodes first class. This saves many newly created db transactions later. --- discovery/chan_series.go | 2 +- graph/db/graph_test.go | 11 ++- graph/db/interfaces.go | 5 +- graph/db/kv_store.go | 208 ++++++++++++++++++++++++++++++++++----- graph/db/sql_store.go | 12 ++- 5 files changed, 206 insertions(+), 32 deletions(-) diff --git a/discovery/chan_series.go b/discovery/chan_series.go index cf502694c..1fcb13d55 100644 --- a/discovery/chan_series.go +++ b/discovery/chan_series.go @@ -201,7 +201,7 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, return nil, err } - for _, nodeAnn := range nodeAnnsInHorizon { + 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 { diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 6975c9d48..9a25f2292 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -2263,9 +2263,11 @@ func TestNodeUpdatesInHorizon(t *testing.T) { }, } for _, queryCase := range queryCases { - resp, err := graph.NodeUpdatesInHorizon( + respIter, err := graph.NodeUpdatesInHorizon( queryCase.start, queryCase.end, ) + + resp := fn.Collect(respIter) require.NoError(t, err) require.Len(t, resp, len(queryCase.resp)) @@ -3507,11 +3509,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) + nodesInHorizonIter, err := graph.NodeUpdatesInHorizon(startTime, endTime) require.NoError(t, err, "unable to fetch nodes in horizon") // We should only have a single node, and that node should exactly // match the node we just inserted. + nodesInHorizon := fn.Collect(nodesInHorizonIter) if len(nodesInHorizon) != 1 { t.Fatalf("should have 1 nodes instead have: %v", len(nodesInHorizon)) @@ -3525,10 +3528,10 @@ 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, err = graph.NodeUpdatesInHorizon(startTime, endTime) require.NoError(t, err, "unable to fetch nodes in horizon") - if len(nodesInHorizon) != 0 { + if len(fn.Collect(nodesInHorizonIter)) != 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 f90c2da8a..fb77120e1 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.Seq[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 diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 165f51987..080fd3812 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" @@ -2194,57 +2195,175 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime, return edgesInHorizon, 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) { +// 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 - var nodesInHorizon []models.Node + // 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 +2372,53 @@ 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.Seq[models.Node], error) { + + cfg := defaultIteratorConfig() + for _, opt := range opts { + opt(cfg) + } + + return func(yield func(models.Node) 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) + + return + } + + for _, node := range nodeAnns { + if !yield(node) { + return + } + } + + // If we we're done, then we can just break out here + // now. + if !hasMore || len(nodeAnns) == 0 { + 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 23fd50482..b617aa3b8 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" @@ -553,7 +554,8 @@ func (s *SQLStore) SetSourceNode(ctx context.Context, // // NOTE: This is part of the V1Store interface. func (s *SQLStore) NodeUpdatesInHorizon(startTime, - endTime time.Time) ([]models.Node, error) { + endTime time.Time, + opts ...IteratorOption) (iter.Seq[models.Node], error) { ctx := context.TODO() @@ -587,7 +589,13 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, return nil, fmt.Errorf("unable to fetch nodes: %w", err) } - return nodes, nil + return func(yield func(models.Node) bool) { + for _, node := range nodes { + if !yield(node) { + return + } + } + }, nil } // AddChannelEdge adds a new (undirected, blank) edge to the graph database. An From 069888b51a033dab3de7e3f249a6f3694a3a5808 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 4 Aug 2025 17:16:21 -0700 Subject: [PATCH 04/13] graph/db: convert ChanUpdatesInHorizon to use iterators In this commit, we refactor the ChanUpdatesInHorizon method to return an iterator instead of a slice. This change significantly reduces memory usage when dealing with large result sets by allowing callers to process items incrementally rather than loading everything into memory at once. --- discovery/chan_series.go | 4 +- graph/builder.go | 4 +- graph/db/graph_test.go | 6 +- graph/db/interfaces.go | 4 +- graph/db/kv_store.go | 305 +++++++++++++++++++++++++++++---------- graph/db/sql_store.go | 11 +- graph/interfaces.go | 6 +- 7 files changed, 254 insertions(+), 86 deletions(-) diff --git a/discovery/chan_series.go b/discovery/chan_series.go index 1fcb13d55..6c75635be 100644 --- a/discovery/chan_series.go +++ b/discovery/chan_series.go @@ -5,6 +5,7 @@ import ( "time" "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/lightningnetwork/lnd/fn/v2" graphdb "github.com/lightningnetwork/lnd/graph/db" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/netann" @@ -114,12 +115,13 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, // First, we'll query for all the set of channels that have an update // that falls within the specified horizon. - chansInHorizon, err := c.graph.ChanUpdatesInHorizon( + chansInHorizonIter, err := c.graph.ChanUpdatesInHorizon( startTime, endTime, ) if err != nil { return nil, err } + chansInHorizon := fn.Collect(chansInHorizonIter) // nodesFromChan records the nodes seen from the channels. nodesFromChan := make(map[[33]byte]struct{}, len(chansInHorizon)*2) diff --git a/graph/builder.go b/graph/builder.go index 594093697..c77d0fea2 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -593,13 +593,13 @@ func (b *Builder) pruneZombieChans() error { startTime := time.Unix(0, 0) endTime := time.Now().Add(-1 * chanExpiry) - oldEdges, err := b.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime) + oldEdgesIter, err := b.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime) if err != nil { return fmt.Errorf("unable to fetch expired channel updates "+ "chans: %v", err) } - for _, u := range oldEdges { + for u := range oldEdgesIter { err = filterPruneChans(u.Info, u.Policy1, u.Policy2) if err != nil { return fmt.Errorf("error filtering channels to "+ diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 9a25f2292..730fd8363 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -2028,10 +2028,11 @@ 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( + chanUpdatesIter, err := graph.ChanUpdatesInHorizon( time.Unix(999, 0), time.Unix(9999, 0), ) require.NoError(t, err, "unable to updates for updates") + chanUpdates := fn.Collect(chanUpdatesIter) if len(chanUpdates) != 0 { t.Fatalf("expected 0 chan updates, instead got %v", len(chanUpdates)) @@ -2144,12 +2145,13 @@ func TestChanUpdatesInHorizon(t *testing.T) { }, } for _, queryCase := range queryCases { - resp, err := graph.ChanUpdatesInHorizon( + respIter, err := graph.ChanUpdatesInHorizon( queryCase.start, queryCase.end, ) 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", diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index fb77120e1..c97b26c5c 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -220,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.Seq[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 080fd3812..7bd958861 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -2040,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[:], ) @@ -2148,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, @@ -2158,41 +2252,102 @@ 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) + return batch, hasMore, nil +} + +// 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) { + + cfg := defaultIteratorConfig() + for _, opt := range opts { + opt(cfg) } - 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 func(yield func(ChannelEdge) bool) { + iterState := newChanUpdatesIterator( + cfg.chanUpdateIterBatchSize, startTime, endTime, + ) - return edgesInHorizon, nil + 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, + ) + // 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) + 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) { + 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) + } + }, nil } // nodeUpdatesIterator maintains state for iterating through node updates. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index b617aa3b8..38485f6c1 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -933,7 +933,8 @@ func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, // // NOTE: This is part of the V1Store interface. func (s *SQLStore) ChanUpdatesInHorizon(startTime, - endTime time.Time) ([]ChannelEdge, error) { + endTime time.Time, + opts ...IteratorOption) (iter.Seq[ChannelEdge], error) { s.cacheMu.Lock() defer s.cacheMu.Unlock() @@ -1033,7 +1034,13 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, "horizon (%s, %s)", startTime, endTime) } - return edges, nil + return func(yield func(ChannelEdge) bool) { + for _, edge := range edges { + if !yield(edge) { + return + } + } + }, nil } // ForEachNodeCached is similar to forEachNode, but it returns DirectedChannel diff --git a/graph/interfaces.go b/graph/interfaces.go index ec4f42b79..cf660ec50 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.Seq[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 From c69971c20b63a613adeb6151ed4f9ce619d6974e Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 4 Aug 2025 17:36:45 -0700 Subject: [PATCH 05/13] sqldb: implement iterator support for NodeUpdatesInHorizon In this commit, we update the SQL store implementation to support the new iterator-based API for NodeUpdatesInHorizon. This includes adding a new SQL query that supports efficient pagination through result sets. The SQL implementation uses cursor-based pagination with configurable batch sizes, allowing efficient iteration over large result sets without loading everything into memory. The query is optimized to use indexes effectively and minimize database round trips. New SQL query GetNodesByLastUpdateRange is updated to support: * Cursor-based pagination using (last_update, pub_key) compound cursor * Optional filtering for public nodes only * Configurable batch sizes via MaxResults parameter --- graph/db/sql_store.go | 121 +++++++++++++++++++++++++---------- sqldb/sqlc/graph.sql.go | 47 ++++++++++++-- sqldb/sqlc/queries/graph.sql | 30 ++++++++- 3 files changed, 159 insertions(+), 39 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 38485f6c1..55b4b2450 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -553,47 +553,100 @@ func (s *SQLStore) SetSourceNode(ctx context.Context, // announcements. // // NOTE: This is part of the V1Store interface. -func (s *SQLStore) NodeUpdatesInHorizon(startTime, - endTime time.Time, +func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time, opts ...IteratorOption) (iter.Seq[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 func(yield func(models.Node) bool) { - for _, node := range nodes { - if !yield(node) { + 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) return } + + for _, node := range batch { + if !yield(node) { + return + } + } + + // If the batch didn't yield anything, then we're done. + if len(batch) == 0 { + break + } } }, nil } diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 4c2be19ea..8c3f641ed 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -1969,16 +1969,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..0621bf9f8 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 From 31ab2ae4b63470f963e157f0997348da4a99e74e Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 4 Aug 2025 17:45:19 -0700 Subject: [PATCH 06/13] sqldb: implement iterator support for ChanUpdatesInHorizon In this commit, we update the SQL store implementation to support the new iterator-based API for ChanUpdatesInHorizon. This includes adding SQL query pagination support and helper functions for efficient batch processing. The SQL implementation uses cursor-based pagination with configurable batch sizes, allowing efficient iteration over large result sets without loading everything into memory. The query is optimized to use indexes effectively and minimize database round trips. New SQL query GetChannelsByPolicyLastUpdateRange is updated to support: - Cursor-based pagination using (max_update_time, id) compound cursor - Configurable batch sizes via MaxResults parameter - Efficient batch caching with updateChanCacheBatch helper --- graph/db/sql_store.go | 331 ++++++++++++++++++++++++----------- sqldb/sqlc/graph.sql.go | 38 +++- sqldb/sqlc/queries/graph.sql | 20 ++- 3 files changed, 284 insertions(+), 105 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 55b4b2450..2d4d06774 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -981,117 +981,250 @@ 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, - opts ...IteratorOption) (iter.Seq[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 - ) - - 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()), - }, - ) - if err != nil { - return err - } - - if len(rows) == 0 { - return nil - } - - // 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)) - - // 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) - - // Skip duplicates. - if _, ok := edgesSeen[chanIDInt]; ok { - continue - } - edgesSeen[chanIDInt] = struct{}{} - - // Check cache first. - if channel, ok := s.chanCache.get(chanIDInt); ok { - hits++ - edges = append(edges, channel) - continue - } - - // Mark this row as one we need to batch load data for. - uncachedRows = append(uncachedRows, row) - } - - // If there are no uncached rows, then we can return early. - if len(uncachedRows) == 0 { - return nil - } - - // 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) + for chanID, edge := range edgesToCache { + s.chanCache.insert(chanID, edge) } +} - // Insert any edges loaded from disk into the cache. - for chanid, channel := range edgesToCache { - s.chanCache.insert(chanid, channel) - } +// 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.Seq[ChannelEdge], error) { - 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) + // Apply options. + cfg := defaultIteratorConfig() + for _, opt := range opts { + opt(cfg) } return func(yield func(ChannelEdge) bool) { - for _, edge := range edges { - if !yield(edge) { + 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 + ) + + // 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 + + 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 + } + + //nolint:ll + hasMore = len(rows) == cfg.chanUpdateIterBatchSize + + //nolint:ll + for _, row := range rows { + lastUpdateTime = sql.NullInt64{ + Int64: extractMaxUpdateTime(row), + Valid: true, + } + lastID = sql.NullInt64{ + Int64: row.GraphChannel.ID, + Valid: true, + } + + // 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) + return } + + for _, edge := range batch { + if !yield(edge) { + 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 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) } }, nil } diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 8c3f641ed..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 } diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 0621bf9f8..19087fc1b 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -472,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 From 32528afa0a3a1ba8b07aa56cf661ebc7c02a9224 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 4 Aug 2025 16:01:50 -0700 Subject: [PATCH 07/13] graph/db: add tests for iterator implementations --- graph/db/graph_test.go | 295 +++++++++++++++++++++++++++++++++++++++-- 1 file changed, 286 insertions(+), 9 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 730fd8363..1a27cd975 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -2028,11 +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. - chanUpdatesIter, err := graph.ChanUpdatesInHorizon( + chanIter, err := graph.ChanUpdatesInHorizon( time.Unix(999, 0), time.Unix(9999, 0), ) require.NoError(t, err, "unable to updates for updates") - chanUpdates := fn.Collect(chanUpdatesIter) + + chanUpdates := fn.Collect(chanIter) + if len(chanUpdates) != 0 { t.Fatalf("expected 0 chan updates, instead got %v", len(chanUpdates)) @@ -2151,6 +2153,7 @@ func TestChanUpdatesInHorizon(t *testing.T) { if err != nil { t.Fatalf("unable to query for updates: %v", err) } + resp := fn.Collect(respIter) if len(resp) != len(queryCase.resp) { @@ -2191,10 +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. - nodeUpdates, err := graph.NodeUpdatesInHorizon( + nodeUpdatesIter, err := graph.NodeUpdatesInHorizon( time.Unix(999, 0), time.Unix(9999, 0), ) 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 @@ -2255,22 +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 { - respIter, err := graph.NodeUpdatesInHorizon( + iter, err := graph.NodeUpdatesInHorizon( queryCase.start, queryCase.end, ) + require.NoError(t, err, "unable to query for node updates") - resp := fn.Collect(respIter) - require.NoError(t, err) + resp := fn.Collect(iter) require.Len(t, resp, len(queryCase.resp)) for i := 0; i < len(resp); i++ { @@ -2279,6 +2283,278 @@ func TestNodeUpdatesInHorizon(t *testing.T) { } } +// 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 { + 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", + ) + } + }) + } + }) + } +} + +// 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, err := 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 + 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 { + t.Run(fmt.Sprintf("BatchSize%d", batchSize), 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, err := graph.ChanUpdatesInHorizon( + startTime, startTime.Add(26*time.Hour), + WithChanUpdateIterBatchSize(batchSize), + ) + require.NoError(t, err) + + channels := fn.Collect(iter) + 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 @@ -3532,6 +3808,7 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) { // the horizon. This time we should have no nodes at all. nodesInHorizonIter, err = graph.NodeUpdatesInHorizon(startTime, endTime) require.NoError(t, err, "unable to fetch nodes in horizon") + nodesInHorizon = fn.Collect(nodesInHorizonIter) if len(fn.Collect(nodesInHorizonIter)) != 0 { t.Fatalf("should have zero nodes instead have: %v", From fda989da9c7bf4b95a006b274ee2169f295323fe Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 4 Aug 2025 16:05:57 -0700 Subject: [PATCH 08/13] discovery+graph: update callers to use new iterator APIs In this commit, we update all callers of NodeUpdatesInHorizon and ChanUpdatesInHorizon to use the new iterator-based APIs. The changes use fn.Collect to maintain existing behavior while benefiting from the memory efficiency of iterators when possible. --- discovery/chan_series.go | 37 ++----------------------------------- 1 file changed, 2 insertions(+), 35 deletions(-) diff --git a/discovery/chan_series.go b/discovery/chan_series.go index 6c75635be..081133dfc 100644 --- a/discovery/chan_series.go +++ b/discovery/chan_series.go @@ -5,7 +5,6 @@ import ( "time" "github.com/btcsuite/btcd/chaincfg/chainhash" - "github.com/lightningnetwork/lnd/fn/v2" graphdb "github.com/lightningnetwork/lnd/graph/db" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/netann" @@ -121,12 +120,8 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, if err != nil { return nil, err } - chansInHorizon := fn.Collect(chansInHorizonIter) - // nodesFromChan records the nodes seen from the channels. - nodesFromChan := make(map[[33]byte]struct{}, len(chansInHorizon)*2) - - for _, channel := range chansInHorizon { + for channel := range chansInHorizonIter { // 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. @@ -187,47 +182,19 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, // 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, + startTime, endTime, graphdb.WithIterPublicNodesOnly(), ) 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 From d8f6fd29f7f9f3cb8c094fe154ac4a65ffbfef5f Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 10 Sep 2025 18:23:00 -0700 Subject: [PATCH 09/13] discovery: convert UpdatesInHorizon to return iter.Seq2[lnwire.Message, error] In this commit, we complete the iterator conversion work started in PR 10128 by threading the iterator pattern through to the higher-level UpdatesInHorizon method. This change converts the method from returning a fully materialized slice of messages to returning a lazy iterator that yields messages on demand. The new signature uses iter.Seq2 to allow error propagation during iteration, eliminating the need for a separate error return value. This approach enables callers to handle errors as they occur during iteration rather than failing upfront. The implementation now lazily processes channel and node updates, yielding them as they're generated rather than accumulating them in memory. This maintains the same ordering guarantees (channels before nodes) while significantly reducing memory pressure when dealing with large update sets during gossip synchronization. --- discovery/chan_series.go | 170 ++++++++++++++++++--------------------- discovery/syncer.go | 17 +--- discovery/syncer_test.go | 18 +++-- 3 files changed, 95 insertions(+), 110 deletions(-) diff --git a/discovery/chan_series.go b/discovery/chan_series.go index 081133dfc..22d6397e0 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,109 +109,96 @@ 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. - chansInHorizonIter, err := c.graph.ChanUpdatesInHorizon( - startTime, endTime, - ) - if err != nil { - return nil, err - } - - for channel := range chansInHorizonIter { - // 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, err := c.graph.ChanUpdatesInHorizon( + startTime, endTime, ) if err != nil { - return nil, err + yield(nil, err) + return } - // 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 := 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 + } + + //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, err := c.graph.NodeUpdatesInHorizon( + startTime, endTime, graphdb.WithIterPublicNodesOnly(), + ) + for nodeAnn := range nodeAnnsInHorizon { + nodeUpdate, err := nodeAnn.NodeAnnouncement(true) if err != nil { - log.Errorf("not sending invalid channel "+ - "update %v: %v", edge2, err) - } else { - chanUpdates = append(chanUpdates, edge2) + 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...) } - - // 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, graphdb.WithIterPublicNodesOnly(), - ) - if err != nil { - return nil, err - } - - for nodeAnn := range nodeAnnsInHorizon { - 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..478e7acd3 100644 --- a/discovery/syncer.go +++ b/discovery/syncer.go @@ -1442,23 +1442,12 @@ 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 { - returnSema() - return err - } log.Infof("GossipSyncer(%x): applying new remote update horizon: "+ - "start=%v, end=%v, backlog_size=%v", g.cfg.peerPub[:], - startTime, endTime, len(newUpdatestoSend)) - - // If we don't have any to send, then we can return early. - if len(newUpdatestoSend) == 0 { - returnSema() - return nil - } + "start=%v, end=%v", g.cfg.peerPub[:], startTime, endTime) // Set the atomic flag to indicate we're starting to send the backlog. // If the swap fails, it means another goroutine is already active, so @@ -1478,7 +1467,7 @@ func (g *GossipSyncer) ApplyGossipFilter(ctx context.Context, defer returnSema() defer g.isSendingBacklog.Store(false) - for _, msg := range newUpdatestoSend { + for msg := range newUpdatestoSend { err := g.sendToPeerSync(ctx, msg) switch { case err == ErrGossipSyncerExiting: diff --git a/discovery/syncer_test.go b/discovery/syncer_test.go index edf91cb69..72ad2ce47 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,20 @@ 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, + } + + msgs := <-m.horizonResp + for _, msg := range msgs { + if !yield(msg, nil) { + return + } + } } - - return <-m.horizonResp, nil } func (m *mockChannelGraphTimeSeries) FilterKnownChanIDs(chain chainhash.Hash, From 77f3b356408753b5286cb90bf95e2fbcffd1118c Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 10 Sep 2025 18:23:15 -0700 Subject: [PATCH 10/13] discovery: update ApplyGossipFilter to use lazy iterator with Pull2 In this commit, we update ApplyGossipFilter to leverage the new iterator-based UpdatesInHorizon method. The key innovation here is using iter.Pull2 to create a pull-based iterator that allows us to check if any updates exist before launching the background goroutine. This approach provides several benefits over the previous implementation. First, we avoid the overhead of launching a goroutine when there are no updates to send, which was previously unavoidable without materializing the entire result set. Second, we maintain lazy loading throughout the sending process, only pulling messages from the database as they're needed for transmission. The implementation uses Pull2 to peek at the first message, determining whether to proceed with sending updates. If updates exist, ownership of the iterator is transferred to the goroutine, which continues pulling and sending messages until exhausted. This design ensures memory usage remains bounded regardless of the number of updates being synchronized. --- discovery/syncer.go | 59 ++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 56 insertions(+), 3 deletions(-) diff --git a/discovery/syncer.go b/discovery/syncer.go index 478e7acd3..b8c51a9a5 100644 --- a/discovery/syncer.go +++ b/discovery/syncer.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "iter" "math" "math/rand" "sort" @@ -1446,8 +1447,29 @@ func (g *GossipSyncer) ApplyGossipFilter(ctx context.Context, g.cfg.chainHash, startTime, endTime, ) + // 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 firstErr + } + log.Infof("GossipSyncer(%x): applying new remote update horizon: "+ - "start=%v, end=%v", g.cfg.peerPub[:], startTime, endTime) + "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 !ok { + stop() + returnSema() + return nil + } // Set the atomic flag to indicate we're starting to send the backlog. // If the swap fails, it means another goroutine is already active, so @@ -1461,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 From 0b3816afb9479e88338898b39be7f07171492e58 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 10 Sep 2025 18:23:28 -0700 Subject: [PATCH 11/13] discovery/test: update mock to support iterator-based UpdatesInHorizon In this commit, we update the mockChannelGraphTimeSeries to implement the new iterator-based UpdatesInHorizon interface. The mock maintains its existing behavior of receiving messages through a channel and returning them to the caller, but now wraps this in an iterator function. The implementation creates an iterator that pulls the entire message slice from the mock's response channel, then yields each message individually. This preserves the test semantics while conforming to the new interface, ensuring all existing tests continue to pass without modification. --- discovery/syncer_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/discovery/syncer_test.go b/discovery/syncer_test.go index 72ad2ce47..faef40bbe 100644 --- a/discovery/syncer_test.go +++ b/discovery/syncer_test.go @@ -94,6 +94,8 @@ func (m *mockChannelGraphTimeSeries) UpdatesInHorizon(chain chainhash.Hash, 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) { From f3a8fd842d11d20ff76c039a1d6a29c5164ab121 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 24 Sep 2025 18:47:26 -0700 Subject: [PATCH 12/13] 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 From 506c3d5b0d17ffe37206a5435f566636a2bef44c Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 26 Sep 2025 15:05:53 -0700 Subject: [PATCH 13/13] docs/release-notes: add release notes entry --- docs/release-notes/release-notes-0.20.0.md | 7 +++++++ 1 file changed, 7 insertions(+) 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