sqldb: implement iterator support for NodeUpdatesInHorizon
What the developer wrote
Authored by Olaoluwa Osuntokun
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
What changed, and why it matters
This commit changes how LND's SQL database backend fetches lists of Lightning network nodes that were updated within a time window. Previously it loaded all matching nodes into memory at once; now it pages through them in batches using a database cursor. The change is mostly a performance and memory improvement, but it also slightly alters behavior: the end time is now inclusive rather than exclusive, and an optional filter for 'public nodes only' was added. There is no clear security bug, but the new SQL pagination logic is complex and could in theory skip or duplicate rows if the cursor handling has an edge-case bug.
Review the new SQL query for correctness of the compound cursor, especially edge cases where many nodes share the same last_update timestamp or where the batch size exactly matches the remaining row count. Verify that the inclusive end_time change is intentional and consistent with callers' expectations. Consider adding tests for pagination stability and the public-node filter. No immediate security patch appears required based solely on this commit.
Security signals we found
Changed SQL boundary condition: end_time comparison moved from '<' to '<='
Added cursor-based pagination with compound (last_update, pub_key) cursor
Added optional public-node-only filter using EXISTS subquery on graph_channels
Iterator now executes multiple read transactions instead of one
No input validation changes visible in the diff
Evidence from the diff
The patch reimplements NodeUpdatesInHorizon in graph/db/sql_store.go as an iterator that repeatedly executes GetNodesByLastUpdateRange with cursor-based pagination. The SQL query now uses a (last_update, pub_key) compound cursor, an optional public-node filter via an EXISTS subquery against graph_channels, and a LIMIT. Notable changes: the end-time comparison changed from < to <=, making the horizon inclusive on the upper bound. The cursor advances per yielded node inside the transaction, and hasMore is set based on whether the returned row count equals the configured batch size. The reset callback clears the in-memory batch slice. No explicit security relevance, CVE, or researcher attribution is present in the commit or supplied references.
Changed components
graph/db/sql_store.gosqldb/sqlc/graph.sql.gosqldb/sqlc/queries/graph.sqlNodeUpdatesInHorizon API / SQL storeInspect captured patch +156 / −36
diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go
index 38485f6..55b4b24 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()
+ cfg := defaultIteratorConfig()
+ for _, opt := range opts {
+ opt(cfg)
+ }
- 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()),
- },
+ return func(yield func(models.Node) bool) {
+ var (
+ ctx = context.TODO()
+ lastUpdateTime sql.NullInt64
+ lastPubKey = make([]byte, 33)
+ hasMore = true
)
- 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)
+ // 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
- return nil
- },
- )
- if err != nil {
- return fmt.Errorf("unable to build nodes: %w", err)
- }
+ //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
+ }
- return nil
- }, sqldb.NoOpReset)
- if err != nil {
- return nil, fmt.Errorf("unable to fetch nodes: %w", 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 func(yield func(models.Node) bool) {
- for _, node := range nodes {
- if !yield(node) {
+ 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 4c2be19..8c3f641 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 f2224c0..0621bf9 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
Why this scored 18/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.