Merge pull request #10128 from Roasbeef/iter-chan-updates

multi: update ChanUpdatesInHorizon and NodeUpdatesInHorizon to return iterators (iter.Seq[T])

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,
+	UpdatesInHorizon(chain chainhash.Hash, startTime time.Time,
-		startTime time.Time, endTime time.Time) ([]lnwire.Message, error)
+		endTime time.Time) iter.Seq2[lnwire.Message, error]
 
 	// FilterKnownChanIDs takes a target chain, and a set of channel ID's,
 	// and returns a filtered set of chan ID's. This filtered set of chan
@@ -108,140 +109,100 @@ func (c *ChanSeries) HighestChanID(ctx context.Context,
 //
 // NOTE: This is part of the ChannelGraphTimeSeries interface.
 func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
-	startTime time.Time, endTime time.Time) ([]lnwire.Message, error) {
+	startTime, endTime time.Time) iter.Seq2[lnwire.Message, error] {
 
-	var updates []lnwire.Message
+	return func(yield func(lnwire.Message, error) bool) {
-
+		// First, we'll query for all the set of channels that have an
-	// First, we'll query for all the set of channels that have an update
+		// update that falls within the specified horizon.
-	// that falls within the specified horizon.
+		chansInHorizon := c.graph.ChanUpdatesInHorizon(
-	chansInHorizon, err := c.graph.ChanUpdatesInHorizon(
+			startTime, endTime,
-		startTime, endTime,
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	// nodesFromChan records the nodes seen from the channels.
-	nodesFromChan := make(map[[33]byte]struct{}, len(chansInHorizon)*2)
-
-	for _, channel := range chansInHorizon {
-		// If the channel hasn't been fully advertised yet, or is a
-		// private channel, then we'll skip it as we can't construct a
-		// full authentication proof if one is requested.
-		if channel.Info.AuthProof == nil {
-			continue
-		}
-
-		chanAnn, edge1, edge2, err := netann.CreateChanAnnouncement(
-			channel.Info.AuthProof, channel.Info, channel.Policy1,
-			channel.Policy2,
 		)
-		if err != nil {
-			return nil, err
-		}
 
-		// Create a slice to hold the `channel_announcement` and
+		for channel, err := range chansInHorizon {
-		// potentially two `channel_update` msgs.
+			if err != nil {
-		//
+				yield(nil, err)
-		// NOTE: Based on BOLT7, if a channel_announcement has no
+				return
-		// corresponding channel_updates, we must not send the
+			}
-		// channel_announcement. Thus we use this slice to decide we
+			// If the channel hasn't been fully advertised yet, or
-		// want to send this `channel_announcement` or not. By the end
+			// is a private channel, then we'll skip it as we can't
-		// of the operation, if the len of the slice is 1, we will not
+			// construct a full authentication proof if one is
-		// send the `channel_announcement`. Otherwise, when sending the
+			// requested.
-		// msgs, the `channel_announcement` must be sent prior to any
+			if channel.Info.AuthProof == nil {
-		// corresponding `channel_update` or `node_annoucement`, that's
+				continue
-		// why we create a slice here to maintain the order.
+			}
-		chanUpdates := make([]lnwire.Message, 0, 3)
+
-		chanUpdates = append(chanUpdates, chanAnn)
+			//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).
+			// conform to the spec (anymore), so check to make sure
-			err := netann.ValidateChannelUpdateFields(0, edge1)
+			// that these channel updates are valid before yielding
-			if err != nil {
+			// them.
-				log.Errorf("not sending invalid channel "+
+			if edge1 != nil {
-					"update %v: %v", edge1, err)
+				err := netann.ValidateChannelUpdateFields(
-			} else {
+					0, edge1,
-				chanUpdates = append(chanUpdates, 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 {
+		// Next, we'll send out all the node announcements that have an
-			err := netann.ValidateChannelUpdateFields(0, edge2)
+		// update within the horizon as well. We send these second to
+		// ensure that they follow any active channels they have.
+		nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon(
+			startTime, endTime, graphdb.WithIterPublicNodesOnly(),
+		)
+		for nodeAnn, err := range nodeAnnsInHorizon {
 			if err != nil {
-				log.Errorf("not sending invalid channel "+
+				yield(nil, err)
-					"update %v: %v", edge2, err)
+				return
-			} else {
+			}
-				chanUpdates = append(chanUpdates, edge2)
+			nodeUpdate, err := nodeAnn.NodeAnnouncement(true)
+			if err != nil {
+				if !yield(nil, err) {
+					return
+				}
+
+				continue
+			}
+
+			if !yield(nodeUpdate, nil) {
+				return
 			}
 		}
-
-		// If there's no corresponding `channel_update` to send, skip
-		// sending this `channel_announcement`.
-		if len(chanUpdates) < 2 {
-			continue
-		}
-
-		// Append the all the msgs to the slice.
-		updates = append(updates, chanUpdates...)
-
-		// Record the nodes seen.
-		nodesFromChan[channel.Info.NodeKey1Bytes] = struct{}{}
-		nodesFromChan[channel.Info.NodeKey2Bytes] = struct{}{}
 	}
-
-	// Next, we'll send out all the node announcements that have an update
-	// within the horizon as well. We send these second to ensure that they
-	// follow any active channels they have.
-	nodeAnnsInHorizon, err := c.graph.NodeUpdatesInHorizon(
-		startTime, endTime,
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, nodeAnn := range nodeAnnsInHorizon {
-		// If this node has not been seen in the above channels, we can
-		// skip sending its NodeAnnouncement.
-		if _, seen := nodesFromChan[nodeAnn.PubKeyBytes]; !seen {
-			log.Debugf("Skipping forwarding as node %x not found "+
-				"in channel announcement", nodeAnn.PubKeyBytes)
-			continue
-		}
-
-		// Ensure we only forward nodes that are publicly advertised to
-		// prevent leaking information about nodes.
-		isNodePublic, err := c.graph.IsPublicNode(nodeAnn.PubKeyBytes)
-		if err != nil {
-			log.Errorf("Unable to determine if node %x is "+
-				"advertised: %v", nodeAnn.PubKeyBytes, err)
-			continue
-		}
-
-		if !isNodePublic {
-			log.Tracef("Skipping forwarding announcement for "+
-				"node %x due to being unadvertised",
-				nodeAnn.PubKeyBytes)
-			continue
-		}
-
-		nodeUpdate, err := nodeAnn.NodeAnnouncement(true)
-		if err != nil {
-			return nil, err
-		}
-
-		if err := netann.ValidateNodeAnnFields(nodeUpdate); err != nil {
-			log.Debugf("Skipping forwarding invalid node "+
-				"announcement %x: %v", nodeAnn.PubKeyBytes, err)
-
-			continue
-		}
-
-		updates = append(updates, nodeUpdate)
-	}
-
-	return updates, nil
 }
 
 // FilterKnownChanIDs takes a target chain, and a set of channel ID's, and

discovery/syncer.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"iter"
 	"math"
 	"math/rand"
 	"sort"
@@ -1442,20 +1443,30 @@ func (g *GossipSyncer) ApplyGossipFilter(ctx context.Context,
 
 	// Now that the remote peer has applied their filter, we'll query the
 	// database for all the messages that are beyond this filter.
-	newUpdatestoSend, err := g.cfg.channelSeries.UpdatesInHorizon(
+	newUpdatestoSend := g.cfg.channelSeries.UpdatesInHorizon(
 		g.cfg.chainHash, startTime, endTime,
 	)
-	if err != nil {
+
+	// Create a pull-based iterator so we can check if there are any
+	// updates before launching the goroutine.
+	next, stop := iter.Pull2(newUpdatestoSend)
+
+	// Check if we have any updates to send by attempting to get the first
+	// message.
+	firstMsg, firstErr, ok := next()
+	if firstErr != nil {
+		stop()
 		returnSema()
-		return err
+		return firstErr
 	}
 
 	log.Infof("GossipSyncer(%x): applying new remote update horizon: "+
-		"start=%v, end=%v, backlog_size=%v", g.cfg.peerPub[:],
+		"start=%v, end=%v, has_updates=%v", g.cfg.peerPub[:],
-		startTime, endTime, len(newUpdatestoSend))
+		startTime, endTime, ok)
 
 	// If we don't have any to send, then we can return early.
-	if len(newUpdatestoSend) == 0 {
+	if !ok {
+		stop()
 		returnSema()
 		return nil
 	}
@@ -1472,14 +1483,45 @@ func (g *GossipSyncer) ApplyGossipFilter(ctx context.Context,
 	}
 
 	// We'll conclude by launching a goroutine to send out any updates.
+	// The goroutine takes ownership of the iterator.
 	g.cg.WgAdd(1)
 	go func() {
 		defer g.cg.WgDone()
 		defer returnSema()
 		defer g.isSendingBacklog.Store(false)
+		defer stop()
 
-		for _, msg := range newUpdatestoSend {
+		// Send the first message we already pulled.
-			err := g.sendToPeerSync(ctx, msg)
+		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

discovery/syncer_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"iter"
 	"math"
 	"reflect"
 	"sort"
@@ -86,13 +87,22 @@ func (m *mockChannelGraphTimeSeries) HighestChanID(_ context.Context,
 }
 
 func (m *mockChannelGraphTimeSeries) UpdatesInHorizon(chain chainhash.Hash,
-	startTime time.Time, endTime time.Time) ([]lnwire.Message, error) {
+	startTime, endTime time.Time) iter.Seq2[lnwire.Message, error] {
 
-	m.horizonReq <- horizonQuery{
+	return func(yield func(lnwire.Message, error) bool) {
-		chain, startTime, endTime,
+		m.horizonReq <- horizonQuery{
+			chain, startTime, endTime,
+		}
+
+		// We'll get the response from the channel, then yield it
+		// immediately.
+		msgs := <-m.horizonResp
+		for _, msg := range msgs {
+			if !yield(msg, nil) {
+				return
+			}
+		}
 	}
-
-	return <-m.horizonResp, nil
 }
 
 func (m *mockChannelGraphTimeSeries) FilterKnownChanIDs(chain chainhash.Hash,

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

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

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
+}

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
 

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=

graph/builder.go

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

graph/db/benchmark_test.go

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

graph/db/graph_test.go

@@ -2028,10 +2028,13 @@ func TestChanUpdatesInHorizon(t *testing.T) {
 
 	// If we issue an arbitrary query before any channel updates are
 	// inserted in the database, we should get zero results.
-	chanUpdates, err := graph.ChanUpdatesInHorizon(
+	chanIter := graph.ChanUpdatesInHorizon(
 		time.Unix(999, 0), time.Unix(9999, 0),
 	)
+
+	chanUpdates, err := fn.CollectErr(chanIter)
 	require.NoError(t, err, "unable to updates for updates")
+
 	if len(chanUpdates) != 0 {
 		t.Fatalf("expected 0 chan updates, instead got %v",
 			len(chanUpdates))
@@ -2144,9 +2147,11 @@ func TestChanUpdatesInHorizon(t *testing.T) {
 		},
 	}
 	for _, queryCase := range queryCases {
-		resp, err := graph.ChanUpdatesInHorizon(
+		respIter := graph.ChanUpdatesInHorizon(
 			queryCase.start, queryCase.end,
 		)
+
+		resp, err := fn.CollectErr(respIter)
 		if err != nil {
 			t.Fatalf("unable to query for updates: %v", err)
 		}
@@ -2189,9 +2194,10 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
 
 	// If we issue an arbitrary query before we insert any nodes into the
 	// database, then we shouldn't get any results back.
-	nodeUpdates, err := graph.NodeUpdatesInHorizon(
+	nodeUpdatesIter := graph.NodeUpdatesInHorizon(
 		time.Unix(999, 0), time.Unix(9999, 0),
 	)
+	nodeUpdates, err := fn.CollectErr(nodeUpdatesIter)
 	require.NoError(t, err, "unable to query for node updates")
 	require.Len(t, nodeUpdates, 0)
 
@@ -2253,20 +2259,22 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
 			resp: nodeAnns,
 		},
 
-		// If we reduce the ending time by 10 seconds, then we should
+		// If we reduce the ending time by 1 nanosecond before the last
-		// get all but the last node we inserted.
+		// node's timestamp, then we should get all but the last node.
 		{
 			start: startTime,
-			end:   endTime.Add(-time.Second * 10),
+			end:   endTime.Add(-time.Second*10 - time.Nanosecond),
 
 			resp: nodeAnns[:9],
 		},
 	}
 	for _, queryCase := range queryCases {
-		resp, err := graph.NodeUpdatesInHorizon(
+		iter := graph.NodeUpdatesInHorizon(
 			queryCase.start, queryCase.end,
 		)
-		require.NoError(t, err)
+
+		resp, err := fn.CollectErr(iter)
+		require.NoError(t, err, "unable to query for node updates")
 		require.Len(t, resp, len(queryCase.resp))
 
 		for i := 0; i < len(resp); i++ {
@@ -2275,6 +2283,293 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
 	}
 }
 
+// testNodeUpdatesWithBatchSize is a helper function that tests node updates
+// with a specific batch size to ensure the iterator works correctly across
+// batch boundaries.
+func testNodeUpdatesWithBatchSize(t *testing.T, ctx context.Context,
+	batchSize int) {
+
+	// Create a fresh graph for each test.
+	testGraph := MakeTestGraph(t)
+
+	// Add 25 nodes with increasing timestamps.
+	startTime := time.Unix(1234567890, 0)
+	var nodeAnns []models.Node
+
+	for i := 0; i < 25; i++ {
+		nodeAnn := createTestVertex(t)
+		nodeAnn.LastUpdate = startTime.Add(
+			time.Duration(i) * time.Hour,
+		)
+		nodeAnns = append(nodeAnns, *nodeAnn)
+		require.NoError(
+			t, testGraph.AddNode(ctx, nodeAnn),
+		)
+	}
+
+	testCases := []struct {
+		name  string
+		start time.Time
+		end   time.Time
+		want  int
+	}{
+		{
+			name:  "all nodes",
+			start: startTime,
+			end:   startTime.Add(26 * time.Hour),
+			want:  25,
+		},
+		{
+			name:  "first batch only",
+			start: startTime,
+			end: startTime.Add(
+				time.Duration(
+					min(batchSize, 25)-1,
+				) * time.Hour,
+			),
+			want: min(batchSize, 25),
+		},
+		{
+			name:  "cross batch boundary",
+			start: startTime,
+			end: startTime.Add(
+				time.Duration(
+					min(batchSize, 24),
+				) * time.Hour,
+			),
+			want: min(batchSize+1, 25),
+		},
+		{
+			name: "exact boundary",
+			start: func() time.Time {
+				// Test querying exactly at a
+				// batch boundary.
+				if batchSize <= 25 {
+					return startTime.Add(
+						time.Duration(
+							batchSize-1,
+						) * time.Hour,
+					)
+				}
+
+				// For batch sizes > 25, test
+				// beyond our data range.
+				return startTime.Add(
+					time.Duration(25) * time.Hour,
+				)
+			}(),
+			end: func() time.Time {
+				if batchSize <= 25 {
+					return startTime.Add(
+						time.Duration(
+							batchSize-1,
+						) * time.Hour,
+					)
+				}
+
+				return startTime.Add(
+					time.Duration(25) * time.Hour,
+				)
+			}(),
+			want: func() int {
+				if batchSize <= 25 {
+					return 1
+				}
+
+				// No nodes exist at hour 25 or
+				// beyond.
+				return 0
+			}(),
+		},
+		{
+			name:  "empty range before",
+			start: startTime.Add(-time.Hour),
+			end:   startTime.Add(-time.Minute),
+			want:  0,
+		},
+		{
+			name:  "empty range after",
+			start: startTime.Add(30 * time.Hour),
+			end:   startTime.Add(40 * time.Hour),
+			want:  0,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			iter := testGraph.NodeUpdatesInHorizon(
+				tc.start, tc.end,
+				WithNodeUpdateIterBatchSize(
+					batchSize,
+				),
+			)
+
+			nodes, err := fn.CollectErr(iter)
+			require.NoError(t, err)
+			require.Len(
+				t, nodes, tc.want,
+				"expected %d nodes, got %d",
+				tc.want, len(nodes),
+			)
+
+			// Verify nodes are in the correct time
+			// order.
+			for i := 1; i < len(nodes); i++ {
+				require.True(t,
+					nodes[i-1].LastUpdate.Before(
+						nodes[i].LastUpdate,
+					) || nodes[i-1].LastUpdate.Equal(
+						nodes[i].LastUpdate,
+					),
+					"nodes should be in "+
+						"chronological order",
+				)
+			}
+		})
+	}
+}
+
+// TestNodeUpdatesInHorizonBoundaryConditions tests the iterator boundary
+// conditions, specifically around batch boundaries and edge cases.
+func TestNodeUpdatesInHorizonBoundaryConditions(t *testing.T) {
+	t.Parallel()
+
+	ctx := t.Context()
+
+	// Test with various batch sizes to ensure the iterator works correctly
+	// across batch boundaries.
+	batchSizes := []int{1, 3, 5, 10, 25, 100}
+
+	for _, batchSize := range batchSizes {
+		testName := fmt.Sprintf("BatchSize%d", batchSize)
+		t.Run(testName, func(t *testing.T) {
+			testNodeUpdatesWithBatchSize(t, ctx, batchSize)
+		})
+	}
+}
+
+// TestNodeUpdatesInHorizonEarlyTermination tests that the iterator properly
+// handles early termination when the caller stops iterating.
+func TestNodeUpdatesInHorizonEarlyTermination(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	graph := MakeTestGraph(t)
+
+	// We'll start by creating 100 nodes, each with an update time spaced
+	// one hour apart.
+	startTime := time.Unix(1234567890, 0)
+	for i := 0; i < 100; i++ {
+		nodeAnn := createTestVertex(t)
+		nodeAnn.LastUpdate = startTime.Add(time.Duration(i) * time.Hour)
+		require.NoError(t, graph.AddNode(ctx, nodeAnn))
+	}
+
+	// Test early termination at various points
+	terminationPoints := []int{0, 1, 5, 10, 23, 50, 99}
+
+	for _, stopAt := range terminationPoints {
+		t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) {
+			iter := graph.NodeUpdatesInHorizon(
+				startTime, startTime.Add(200*time.Hour),
+				WithNodeUpdateIterBatchSize(10),
+			)
+
+			// Collect only up to stopAt nodes, breaking afterwards.
+			var collected []models.Node
+			count := 0
+			for node := range iter {
+				if count >= stopAt {
+					break
+				}
+				collected = append(collected, node)
+				count++
+			}
+
+			require.Len(
+				t, collected, stopAt,
+				"should have collected exactly %d nodes",
+				stopAt,
+			)
+		})
+	}
+}
+
+// TestChanUpdatesInHorizonBoundaryConditions tests the channel iterator
+// boundary conditions.
+func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	batchSizes := []int{1, 3, 5, 10}
+
+	for _, batchSize := range batchSizes {
+		testName := fmt.Sprintf("BatchSize%d", batchSize)
+		t.Run(testName, func(t *testing.T) {
+			// Create a fresh graph for each test, then add two new
+			// nodes to the graph.
+			graph := MakeTestGraph(t)
+			node1 := createTestVertex(t)
+			node2 := createTestVertex(t)
+			require.NoError(t, graph.AddNode(ctx, node1))
+			require.NoError(t, graph.AddNode(ctx, node2))
+
+			// Next, we'll create 25 channels between the two nodes,
+			// each with increasing timestamps.
+			startTime := time.Unix(1234567890, 0)
+			const numChans = 25
+
+			for i := 0; i < numChans; i++ {
+				updateTime := startTime.Add(
+					time.Duration(i) * time.Hour,
+				)
+
+				channel, chanID := createEdge(
+					uint32(i*10), 0, 0, 0, node1, node2,
+				)
+				require.NoError(
+					t, graph.AddChannelEdge(ctx, &channel),
+				)
+
+				edge1 := newEdgePolicy(
+					chanID.ToUint64(), updateTime.Unix(),
+				)
+				edge1.ChannelFlags = 0
+				edge1.ToNode = node2.PubKeyBytes
+				edge1.SigBytes = testSig.Serialize()
+				require.NoError(
+					t, graph.UpdateEdgePolicy(ctx, edge1),
+				)
+
+				edge2 := newEdgePolicy(
+					chanID.ToUint64(), updateTime.Unix(),
+				)
+				edge2.ChannelFlags = 1
+				edge2.ToNode = node1.PubKeyBytes
+				edge2.SigBytes = testSig.Serialize()
+				require.NoError(
+					t, graph.UpdateEdgePolicy(ctx, edge2),
+				)
+			}
+
+			// Now we'll run the main query, and verify that we get
+			// back the expected number of channels.
+			iter := graph.ChanUpdatesInHorizon(
+				startTime, startTime.Add(26*time.Hour),
+				WithChanUpdateIterBatchSize(batchSize),
+			)
+
+			channels, err := fn.CollectErr(iter)
+			require.NoError(t, err)
+			require.Len(
+				t, channels, numChans,
+				"expected %d channels, got %d", numChans,
+				len(channels),
+			)
+		})
+	}
+}
+
 // TestFilterKnownChanIDsZombieRevival tests that if a ChannelUpdateInfo is
 // passed to FilterKnownChanIDs that contains a channel that we have marked as
 // a zombie, then we will mark it as live again if the new ChannelUpdate has
@@ -2731,9 +3026,10 @@ func TestStressTestChannelGraphAPI(t *testing.T) {
 		{
 			name: "ChanUpdateInHorizon",
 			fn: func() error {
-				_, err := graph.ChanUpdatesInHorizon(
+				iter := graph.ChanUpdatesInHorizon(
 					time.Now().Add(-time.Hour), time.Now(),
 				)
+				_, err := fn.CollectErr(iter)
 
 				return err
 			},
@@ -3507,11 +3803,12 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
 	// update time of our test node.
 	startTime := time.Unix(9, 0)
 	endTime := node1.LastUpdate.Add(time.Minute)
-	nodesInHorizon, err := graph.NodeUpdatesInHorizon(startTime, endTime)
+	nodesInHorizonIter := 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, err := fn.CollectErr(nodesInHorizonIter)
+	require.NoError(t, err, "unable to fetch nodes in horizon")
 	if len(nodesInHorizon) != 1 {
 		t.Fatalf("should have 1 nodes instead have: %v",
 			len(nodesInHorizon))
@@ -3525,7 +3822,8 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
 
 	// Now that the node has been deleted, we'll again query the nodes in
 	// the horizon. This time we should have no nodes at all.
-	nodesInHorizon, err = graph.NodeUpdatesInHorizon(startTime, endTime)
+	nodesInHorizonIter = graph.NodeUpdatesInHorizon(startTime, endTime)
+	nodesInHorizon, err = fn.CollectErr(nodesInHorizonIter)
 	require.NoError(t, err, "unable to fetch nodes in horizon")
 
 	if len(nodesInHorizon) != 0 {

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,
+	NodeUpdatesInHorizon(startTime, endTime time.Time,
-		endTime time.Time) ([]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
@@ -219,8 +220,8 @@ type V1Store interface { //nolint:interfacebloat
 	// ChanUpdatesInHorizon returns all the known channel edges which have
 	// at least one edge that has an update timestamp within the specified
 	// horizon.
-	ChanUpdatesInHorizon(startTime, endTime time.Time) ([]ChannelEdge,
+	ChanUpdatesInHorizon(startTime, endTime time.Time,
-		error)
+		opts ...IteratorOption) iter.Seq2[ChannelEdge, error]
 
 	// FilterKnownChanIDs takes a set of channel IDs and return the subset
 	// of chan ID's that we don't know and are not known zombies of the

graph/db/kv_store.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"iter"
 	"math"
 	"net"
 	"sort"
@@ -2039,107 +2040,202 @@ type ChannelEdge struct {
 	Node2 *models.Node
 }
 
-// ChanUpdatesInHorizon returns all the known channel edges which have at least
+// updateChanCacheBatch updates the channel cache with multiple edges at once.
-// one edge that has an update timestamp within the specified horizon.
+// This method acquires the cache lock only once for the entire batch.
-func (c *KVStore) ChanUpdatesInHorizon(startTime,
+func (c *KVStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) {
-	endTime time.Time) ([]ChannelEdge, error) {
+	if len(edgesToCache) == 0 {
-
+		return
-	// 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
 
 	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
+		// With all the relevant buckets read, we'll now create a fresh
-		// the index to find all channels within the horizon.
+		// 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
+		var indexKey []byte
-		// the index collecting the info and policy of each update of
+
-		// each channel that has a last update within the time range.
+		// If we left off earlier, then we'll use that key as the
-		//
+		// starting point.
-		//nolint:ll
+		switch {
-		for indexKey, _ := updateCursor.Seek(startTimeBytes[:]); indexKey != nil &&
+		case state.lastSeenKey != nil:
-			bytes.Compare(indexKey, endTimeBytes[:]) <= 0; indexKey, _ = updateCursor.Next() {
+			// Seek to the last seen key, moving to the key right
-			// We have a new eligible entry, so we'll slice of the
+			// after it.
-			// chan ID so we can query it in the DB.
+			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++
+				state.edgesSeen[chanIDInt] = struct{}{}
-				edgesSeen[chanIDInt] = struct{}{}
+
-				edgesInHorizon = append(edgesInHorizon, channel)
+				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 "+
-				return fmt.Errorf("unable to fetch info for "+
+					"for edge with chan_id=%v: %v",
-					"edge with chan_id=%v: %v", chanID, err)
+					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 "+
-				return fmt.Errorf("unable to fetch policies "+
+					"policies for edge with chan_id=%v: %v",
-					"for edge with chan_id=%v: %v", chanID,
+					chanIDInt, err)
-					err)
 			}
-
 			node1, err := fetchLightningNode(
 				nodes, edgeInfo.NodeKey1Bytes[:],
 			)
 			if err != nil {
 				return err
 			}
-
 			node2, err := fetchLightningNode(
 				nodes, edgeInfo.NodeKey2Bytes[:],
 			)
@@ -2147,9 +2243,8 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime,
 				return err
 			}
 
-			// Finally, we'll collate this edge with the rest of
+			// Now we have all the information we need to build the
-			// edges to be returned.
+			// channel edge.
-			edgesSeen[chanIDInt] = struct{}{}
 			channel := ChannelEdge{
 				Info:    &edgeInfo,
 				Policy1: edge1,
@@ -2157,94 +2252,274 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime,
 				Node1:   &node1,
 				Node2:   &node2,
 			}
-			edgesInHorizon = append(edgesInHorizon, channel)
+
-			edgesToCache[chanIDInt] = channel
+			state.edgesSeen[chanIDInt] = struct{}{}
+			state.edgesToCache[chanIDInt] = channel
+
+			batch = append(batch, channel)
+
+			state.total++
+
+			// Advance the iterator to the next entry.
+			indexKey, _ = updateCursor.Next()
+		}
+
+		// If we haven't yet crossed the endTimeBytes, then we still
+		// have more entries to deliver.
+		if indexKey != nil &&
+			bytes.Compare(indexKey, endTimeBytes[:]) <= 0 {
+
+			hasMore = true
 		}
 
 		return nil
 	}, func() {
-		edgesSeen = make(map[uint64]struct{})
+		batch = nil
-		edgesToCache = make(map[uint64]ChannelEdge)
+		hasMore = false
-		edgesInHorizon = nil
 	})
-	switch {
+	if err != nil {
-	case errors.Is(err, ErrGraphNoEdgesFound):
+		return nil, false, err
-		fallthrough
-	case errors.Is(err, ErrGraphNodesNotFound):
-		break
-
-	case err != nil:
-		return nil, err
 	}
 
-	// Insert any edges loaded from disk into the cache.
+	return batch, hasMore, nil
-	for chanid, channel := range edgesToCache {
-		c.chanCache.insert(chanid, channel)
-	}
-
-	if len(edgesInHorizon) > 0 {
-		log.Debugf("ChanUpdatesInHorizon hit percentage: %.2f (%d/%d)",
-			float64(hits)*100/float64(len(edgesInHorizon)), hits,
-			len(edgesInHorizon))
-	} else {
-		log.Debugf("ChanUpdatesInHorizon returned no edges in "+
-			"horizon (%s, %s)", startTime, endTime)
-	}
-
-	return edgesInHorizon, nil
 }
 
-// NodeUpdatesInHorizon returns all the known lightning node which have an
+// ChanUpdatesInHorizon returns all the known channel edges which have at least
-// update timestamp within the passed range. This method can be used by two
+// one edge that has an update timestamp within the specified horizon.
-// nodes to quickly determine if they have the same set of up to date node
+func (c *KVStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
-// announcements.
+	opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
-func (c *KVStore) NodeUpdatesInHorizon(startTime,
-	endTime time.Time) ([]models.Node, error) {
 
-	var nodesInHorizon []models.Node
+	cfg := defaultIteratorConfig()
+	for _, opt := range opts {
+		opt(cfg)
+	}
+
+	return func(yield func(ChannelEdge, error) bool) {
+		iterState := newChanUpdatesIterator(
+			cfg.chanUpdateIterBatchSize, startTime, endTime,
+		)
+
+		for {
+			// At the top of the loop, we'll read the next batch
+			// chunk from disk. We'll also determine if we have any
+			// more entries after this or not.
+			batch, hasMore, err := c.fetchNextChanUpdateBatch(
+				iterState,
+			)
+			if err != nil {
+				// These errors just mean the graph is empty,
+				// which is OK.
+				if !isEmptyGraphError(err) {
+					log.Errorf("ChanUpdatesInHorizon "+
+						"batch error: %v", err)
+
+					yield(ChannelEdge{}, err)
+
+					return
+				}
+				// Continue with empty batch
+			}
+
+			// We'll now yield each edge that we just read. If yield
+			// returns false, then that means that we'll exit early.
+			for _, edge := range batch {
+				if !yield(edge, nil) {
+					return
+				}
+			}
+
+			// Update cache after successful batch yield.
+			c.updateChanCacheBatch(iterState.edgesToCache)
+			iterState.edgesToCache = make(map[uint64]ChannelEdge)
+
+			// If we we're done, then we can just break out here
+			// now.
+			if !hasMore || len(batch) == 0 {
+				break
+			}
+		}
+
+		if iterState.total > 0 {
+			log.Tracef("ChanUpdatesInHorizon hit percentage: "+
+				"%.2f (%d/%d)", float64(iterState.hits)*100/
+				float64(iterState.total), iterState.hits,
+				iterState.total)
+		} else {
+			log.Tracef("ChanUpdatesInHorizon returned no edges "+
+				"in horizon (%s, %s)", startTime, endTime)
+		}
+	}
+}
+
+// nodeUpdatesIterator maintains state for iterating through node updates.
+//
+// Iterator Lifecycle:
+// 1. Initialize state with start/end time, batch size, and filtering options.
+// 2. Fetch batch using pagination cursor (lastSeenKey).
+// 3. Filter nodes if publicNodesOnly is set.
+// 4. Update lastSeenKey to the last processed node's index key.
+// 5. Repeat until we exceed endTime or no more nodes exist.
+type nodeUpdatesIterator struct {
+	// batchSize is the amount of node updates to read at a single time.
+	batchSize int
+
+	// startTime is the start time of the iteration request.
+	startTime time.Time
+
+	// endTime is the end time of the iteration request.
+	endTime time.Time
+
+	// lastSeenKey is the last index key seen. This is used to resume
+	// iteration.
+	lastSeenKey []byte
+
+	// publicNodesOnly filters to only return public nodes if true.
+	publicNodesOnly bool
+
+	// total tracks total nodes processed.
+	total int
+}
+
+// newNodeUpdatesIterator makes a new node updates iterator.
+func newNodeUpdatesIterator(batchSize int, startTime, endTime time.Time,
+	publicNodesOnly bool) *nodeUpdatesIterator {
+
+	return &nodeUpdatesIterator{
+		batchSize:       batchSize,
+		startTime:       startTime,
+		endTime:         endTime,
+		lastSeenKey:     nil,
+		publicNodesOnly: publicNodesOnly,
+	}
+}
+
+// fetchNextNodeBatch fetches the next batch of node announcements using the
+// iterator state.
+func (c *KVStore) fetchNextNodeBatch(
+	state *nodeUpdatesIterator) ([]models.Node, bool, error) {
+
+	var (
+		nodeBatch []models.Node
+		hasMore   bool
+	)
 
 	err := kvdb.View(c.db, func(tx kvdb.RTx) error {
 		nodes := tx.ReadBucket(nodeBucket)
 		if nodes == nil {
 			return ErrGraphNodesNotFound
 		}
-
+		ourPubKey := nodes.Get(sourceKey)
+		if ourPubKey == nil && state.publicNodesOnly {
+			// If we're filtering for public nodes only but don't
+			// have a source node set, we can't determine if nodes
+			// are public. A node is considered public if it has at
+			// least one channel with our node (the source node).
+			return ErrSourceNodeNotSet
+		}
 		nodeUpdateIndex := nodes.NestedReadBucket(nodeUpdateIndexBucket)
 		if nodeUpdateIndex == nil {
 			return ErrGraphNodesNotFound
 		}
 
-		// We'll now obtain a cursor to perform a range query within
+		// We'll now obtain a cursor to perform a range query within the
-		// the index to find all node announcements within the horizon.
+		// 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
+		// If we have a last seen key (existing iteration), then that'll
-		// the index collecting info for each node within the time
+		// be our starting point. Otherwise, we'll seek to the start
-		// range.
+		// time.
-		//
+		var indexKey []byte
-		//nolint:ll
+		if state.lastSeenKey != nil {
-		for indexKey, _ := updateCursor.Seek(startTimeBytes[:]); indexKey != nil &&
+			indexKey, _ = updateCursor.Seek(state.lastSeenKey)
-			bytes.Compare(indexKey, endTimeBytes[:]) <= 0; indexKey, _ = updateCursor.Next() {
+
+			if bytes.Equal(indexKey, state.lastSeenKey) {
+				indexKey, _ = updateCursor.Next()
+			}
+		} else {
+			indexKey, _ = updateCursor.Seek(startTimeBytes[:])
+		}
+
+		// Now we'll read items up to the batch size, exiting early if
+		// we exceed the ending time.
+		var lastProcessedKey []byte
+		for len(nodeBatch) < state.batchSize && indexKey != nil {
+			// Extract the timestamp from the index key (first 8
+			// bytes). Only compare timestamps, not the full key
+			// with pubkey.
+			keyTimestamp := byteOrder.Uint64(indexKey[:8])
+			endTimestamp := uint64(state.endTime.Unix())
+			if keyTimestamp > endTimestamp {
+				break
+			}
+
 			nodePub := indexKey[8:]
 			node, err := fetchLightningNode(nodes, nodePub)
 			if err != nil {
 				return err
 			}
 
-			nodesInHorizon = append(nodesInHorizon, node)
+			if state.publicNodesOnly {
+				nodeIsPublic, err := c.isPublic(
+					tx, node.PubKeyBytes, ourPubKey,
+				)
+				if err != nil {
+					return err
+				}
+				if !nodeIsPublic {
+					indexKey, _ = updateCursor.Next()
+					continue
+				}
+			}
+
+			nodeBatch = append(nodeBatch, node)
+			state.total++
+
+			// Remember the last key we actually processed. We'll
+			// use this to update the last seen key below.
+			if lastProcessedKey == nil {
+				lastProcessedKey = make([]byte, len(indexKey))
+			}
+			copy(lastProcessedKey, indexKey)
+
+			// Advance the iterator to the next entry.
+			indexKey, _ = updateCursor.Next()
+		}
+
+		// If we haven't yet crossed the endTime, then we still
+		// have more entries to deliver.
+		if indexKey != nil {
+			keyTimestamp := byteOrder.Uint64(indexKey[:8])
+			endTimestamp := uint64(state.endTime.Unix())
+			if keyTimestamp <= endTimestamp {
+				hasMore = true
+			}
+		}
+
+		// Update the cursor to the last key we actually processed.
+		if lastProcessedKey != nil {
+			if state.lastSeenKey == nil {
+				state.lastSeenKey = make(
+					[]byte, len(lastProcessedKey),
+				)
+			}
+			copy(state.lastSeenKey, lastProcessedKey)
 		}
 
 		return nil
 	}, func() {
-		nodesInHorizon = nil
+		nodeBatch = nil
 	})
 	switch {
 	case errors.Is(err, ErrGraphNoEdgesFound):
@@ -2253,10 +2528,55 @@ func (c *KVStore) NodeUpdatesInHorizon(startTime,
 		break
 
 	case err != nil:
-		return nil, err
+		return nil, false, err
 	}
 
-	return nodesInHorizon, nil
+	return nodeBatch, hasMore, nil
+}
+
+// NodeUpdatesInHorizon returns all the known lightning node which have an
+// update timestamp within the passed range.
+func (c *KVStore) NodeUpdatesInHorizon(startTime,
+	endTime time.Time,
+	opts ...IteratorOption) iter.Seq2[models.Node, error] {
+
+	cfg := defaultIteratorConfig()
+	for _, opt := range opts {
+		opt(cfg)
+	}
+
+	return func(yield func(models.Node, error) bool) {
+		// Initialize iterator state.
+		state := newNodeUpdatesIterator(
+			cfg.nodeUpdateIterBatchSize,
+			startTime, endTime,
+			cfg.iterPublicNodes,
+		)
+
+		for {
+			nodeAnns, hasMore, err := c.fetchNextNodeBatch(state)
+			if err != nil {
+				log.Errorf("unable to read node updates in "+
+					"horizon: %v", err)
+
+				yield(models.Node{}, err)
+
+				return
+			}
+
+			for _, node := range nodeAnns {
+				if !yield(node, nil) {
+					return
+				}
+			}
+
+			// If we we're done, then we can just break out here
+			// now.
+			if !hasMore || len(nodeAnns) == 0 {
+				break
+			}
+		}
+	}
 }
 
 // FilterKnownChanIDs takes a set of channel IDs and return the subset of chan

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 {

graph/db/sql_store.go

@@ -7,6 +7,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"iter"
 	"maps"
 	"math"
 	"net"
@@ -552,42 +553,105 @@ func (s *SQLStore) SetSourceNode(ctx context.Context,
 // announcements.
 //
 // NOTE: This is part of the V1Store interface.
-func (s *SQLStore) NodeUpdatesInHorizon(startTime,
+func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time,
-	endTime time.Time) ([]models.Node, error) {
+	opts ...IteratorOption) iter.Seq2[models.Node, error] {
 
-	ctx := context.TODO()
+	cfg := defaultIteratorConfig()
-
+	for _, opt := range opts {
-	var nodes []models.Node
+		opt(cfg)
-	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)
 	}
 
-	return nodes, nil
+	return func(yield func(models.Node, error) bool) {
+		var (
+			ctx            = context.TODO()
+			lastUpdateTime sql.NullInt64
+			lastPubKey     = make([]byte, 33)
+			hasMore        = true
+		)
+
+		// Each iteration, we'll read a batch amount of nodes, yield
+		// them, then decide is we have more or not.
+		for hasMore {
+			var batch []models.Node
+
+			//nolint:ll
+			err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+				//nolint:ll
+				params := sqlc.GetNodesByLastUpdateRangeParams{
+					StartTime: sqldb.SQLInt64(
+						startTime.Unix(),
+					),
+					EndTime: sqldb.SQLInt64(
+						endTime.Unix(),
+					),
+					LastUpdate: lastUpdateTime,
+					LastPubKey: lastPubKey,
+					OnlyPublic: sql.NullBool{
+						Bool:  cfg.iterPublicNodes,
+						Valid: true,
+					},
+					MaxResults: sqldb.SQLInt32(
+						cfg.nodeUpdateIterBatchSize,
+					),
+				}
+				rows, err := db.GetNodesByLastUpdateRange(
+					ctx, params,
+				)
+				if err != nil {
+					return err
+				}
+
+				hasMore = len(rows) == cfg.nodeUpdateIterBatchSize
+
+				err = forEachNodeInBatch(
+					ctx, s.cfg.QueryCfg, db, rows,
+					func(_ int64, node *models.Node) error {
+						batch = append(batch, *node)
+
+						// Update pagination cursors
+						// based on the last processed
+						// node.
+						lastUpdateTime = sql.NullInt64{
+							Int64: node.LastUpdate.
+								Unix(),
+							Valid: true,
+						}
+						lastPubKey = node.PubKeyBytes[:]
+
+						return nil
+					},
+				)
+				if err != nil {
+					return fmt.Errorf("unable to build "+
+						"nodes: %w", err)
+				}
+
+				return nil
+			}, func() {
+				batch = []models.Node{}
+			})
+
+			if err != nil {
+				log.Errorf("NodeUpdatesInHorizon batch "+
+					"error: %v", err)
+
+				yield(models.Node{}, err)
+
+				return
+			}
+
+			for _, node := range batch {
+				if !yield(node, nil) {
+					return
+				}
+			}
+
+			// If the batch didn't yield anything, then we're done.
+			if len(batch) == 0 {
+				break
+			}
+		}
+	}
 }
 
 // AddChannelEdge adds a new (undirected, blank) edge to the graph database. An
@@ -920,112 +984,254 @@ func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex,
 	}, reset)
 }
 
-// ChanUpdatesInHorizon returns all the known channel edges which have at least
+// extractMaxUpdateTime returns the maximum of the two policy update times.
-// one edge that has an update timestamp within the specified horizon.
+// This is used for pagination cursor tracking.
-//
+func extractMaxUpdateTime(
-// NOTE: This is part of the V1Store interface.
+	row sqlc.GetChannelsByPolicyLastUpdateRangeRow) int64 {
-func (s *SQLStore) ChanUpdatesInHorizon(startTime,
+
-	endTime time.Time) ([]ChannelEdge, error) {
+	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 (
+	for chanID, edge := range edgesToCache {
-		ctx = context.TODO()
+		s.chanCache.insert(chanID, edge)
-		// 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 {
+// ChanUpdatesInHorizon returns all the known channel edges which have at least
-		rows, err := db.GetChannelsByPolicyLastUpdateRange(
+// one edge that has an update timestamp within the specified horizon.
-			ctx, sqlc.GetChannelsByPolicyLastUpdateRangeParams{
+//
-				Version:   int16(ProtocolV1),
+// Iterator Lifecycle:
-				StartTime: sqldb.SQLInt64(startTime.Unix()),
+// 1. Initialize state (edgesSeen map, cache tracking, pagination cursors)
-				EndTime:   sqldb.SQLInt64(endTime.Unix()),
+// 2. Query batch of channels with policies in time range
-			},
+// 3. For each channel: check if seen, check cache, or build from DB
+// 4. Yield channels to caller
+// 5. Update cache after successful batch
+// 6. Repeat with updated pagination cursor until no more results
+//
+// NOTE: This is part of the V1Store interface.
+func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
+	opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
+
+	// Apply options.
+	cfg := defaultIteratorConfig()
+	for _, opt := range opts {
+		opt(cfg)
+	}
+
+	return func(yield func(ChannelEdge, error) bool) {
+		var (
+			ctx            = context.TODO()
+			edgesSeen      = make(map[uint64]struct{})
+			edgesToCache   = make(map[uint64]ChannelEdge)
+			hits           int
+			total          int
+			lastUpdateTime sql.NullInt64
+			lastID         sql.NullInt64
+			hasMore        = true
 		)
-		if err != nil {
-			return err
-		}
 
-		if len(rows) == 0 {
+		// Each iteration, we'll read a batch amount of channel updates
-			return nil
+		// (consulting the cache along the way), yield them, then loop
-		}
+		// back to decide if we have any more updates to read out.
+		for hasMore {
+			var batch []ChannelEdge
 
-		// We'll pre-allocate the slices and maps here with a best
+			err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(),
-		// effort size in order to avoid unnecessary allocations later
+				func(db SQLQueries) error {
-		// on.
+					//nolint:ll
-		uncachedRows := make(
+					params := sqlc.GetChannelsByPolicyLastUpdateRangeParams{
-			[]sqlc.GetChannelsByPolicyLastUpdateRangeRow, 0,
+						Version: int16(ProtocolV1),
-			len(rows),
+						StartTime: sqldb.SQLInt64(
-		)
+							startTime.Unix(),
-		edgesToCache = make(map[uint64]ChannelEdge, len(rows))
+						),
-		edgesSeen = make(map[uint64]struct{}, len(rows))
+						EndTime: sqldb.SQLInt64(
-		edges = make([]ChannelEdge, 0, len(rows))
+							endTime.Unix(),
+						),
+						LastUpdateTime: lastUpdateTime,
+						LastID:         lastID,
+						MaxResults: sql.NullInt32{
+							Int32: int32(
+								cfg.chanUpdateIterBatchSize,
+							),
+							Valid: true,
+						},
+					}
+					//nolint:ll
+					rows, err := db.GetChannelsByPolicyLastUpdateRange(
+						ctx, params,
+					)
+					if err != nil {
+						return err
+					}
 
-		// Separate cached from non-cached channels since we will only
+					//nolint:ll
-		// batch load the data for the ones we haven't cached yet.
+					hasMore = len(rows) == cfg.chanUpdateIterBatchSize
-		for _, row := range rows {
-			chanIDInt := byteOrder.Uint64(row.GraphChannel.Scid)
 
-			// Skip duplicates.
+					//nolint:ll
-			if _, ok := edgesSeen[chanIDInt]; ok {
+					for _, row := range rows {
-				continue
+						lastUpdateTime = sql.NullInt64{
-			}
+							Int64: extractMaxUpdateTime(row),
-			edgesSeen[chanIDInt] = struct{}{}
+							Valid: true,
+						}
+						lastID = sql.NullInt64{
+							Int64: row.GraphChannel.ID,
+							Valid: true,
+						}
 
-			// Check cache first.
+						// Skip if we've already
-			if channel, ok := s.chanCache.get(chanIDInt); ok {
+						// processed this channel.
-				hits++
+						chanIDInt := byteOrder.Uint64(
-				edges = append(edges, channel)
+							row.GraphChannel.Scid,
-				continue
+						)
+						_, ok := edgesSeen[chanIDInt]
+						if ok {
+							continue
+						}
+
+						s.cacheMu.RLock()
+						channel, ok := s.chanCache.get(
+							chanIDInt,
+						)
+						s.cacheMu.RUnlock()
+						if ok {
+							hits++
+							total++
+							edgesSeen[chanIDInt] = struct{}{}
+							batch = append(batch, channel)
+
+							continue
+						}
+
+						chanEdge, err := s.buildChannelFromRow(
+							ctx, db, row,
+						)
+						if err != nil {
+							return err
+						}
+
+						edgesSeen[chanIDInt] = struct{}{}
+						edgesToCache[chanIDInt] = chanEdge
+
+						batch = append(batch, chanEdge)
+
+						total++
+					}
+
+					return nil
+				}, func() {
+					batch = nil
+					edgesSeen = make(map[uint64]struct{})
+					edgesToCache = make(
+						map[uint64]ChannelEdge,
+					)
+				})
+
+			if err != nil {
+				log.Errorf("ChanUpdatesInHorizon "+
+					"batch error: %v", err)
+
+				yield(ChannelEdge{}, err)
+
+				return
 			}
 
-			// Mark this row as one we need to batch load data for.
+			for _, edge := range batch {
-			uncachedRows = append(uncachedRows, row)
+				if !yield(edge, nil) {
+					return
+				}
+			}
+
+			// Update cache after successful batch yield, setting
+			// the cache lock only once for the entire batch.
+			s.updateChanCacheBatch(edgesToCache)
+			edgesToCache = make(map[uint64]ChannelEdge)
+
+			// If the batch didn't yield anything, then we're done.
+			if len(batch) == 0 {
+				break
+			}
 		}
 
-		// If there are no uncached rows, then we can return early.
+		if total > 0 {
-		if len(uncachedRows) == 0 {
+			log.Debugf("ChanUpdatesInHorizon hit percentage: "+
-			return nil
+				"%.2f (%d/%d)",
+				float64(hits)*100/float64(total), hits, total)
+		} else {
+			log.Debugf("ChanUpdatesInHorizon returned no edges "+
+				"in horizon (%s, %s)", startTime, endTime)
 		}
-
-		// Batch load data for all uncached channels.
-		newEdges, err := batchBuildChannelEdges(
-			ctx, s.cfg, db, uncachedRows,
-		)
-		if err != nil {
-			return fmt.Errorf("unable to batch build channel "+
-				"edges: %w", err)
-		}
-
-		edges = append(edges, newEdges...)
-
-		return nil
-	}, sqldb.NoOpReset)
-	if err != nil {
-		return nil, fmt.Errorf("unable to fetch channels: %w", err)
 	}
-
-	// Insert any edges loaded from disk into the cache.
-	for chanid, channel := range edgesToCache {
-		s.chanCache.insert(chanid, channel)
-	}
-
-	if len(edges) > 0 {
-		log.Debugf("ChanUpdatesInHorizon hit percentage: %.2f (%d/%d)",
-			float64(hits)*100/float64(len(edges)), hits, len(edges))
-	} else {
-		log.Debugf("ChanUpdatesInHorizon returned no edges in "+
-			"horizon (%s, %s)", startTime, endTime)
-	}
-
-	return edges, nil
 }
 
 // ForEachNodeCached is similar to forEachNode, but it returns DirectedChannel

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) (
+	ChanUpdatesInHorizon(startTime, endTime time.Time,
-		[]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

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
+	Version        int16
-	StartTime sql.NullInt64
+	StartTime      sql.NullInt64
-	EndTime   sql.NullInt64
+	EndTime        sql.NullInt64
+	LastUpdateTime sql.NullInt64
+	LastID         sql.NullInt64
+	MaxResults     interface{}
 }
 
 type GetChannelsByPolicyLastUpdateRangeRow struct {
@@ -1194,7 +1215,14 @@ type GetChannelsByPolicyLastUpdateRangeRow struct {
 }
 
 func (q *Queries) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg GetChannelsByPolicyLastUpdateRangeParams) ([]GetChannelsByPolicyLastUpdateRangeRow, error) {
-	rows, err := q.db.QueryContext(ctx, getChannelsByPolicyLastUpdateRange, arg.Version, arg.StartTime, arg.EndTime)
+	rows, err := q.db.QueryContext(ctx, getChannelsByPolicyLastUpdateRange,
+		arg.Version,
+		arg.StartTime,
+		arg.EndTime,
+		arg.LastUpdateTime,
+		arg.LastID,
+		arg.MaxResults,
+	)
 	if err != nil {
 		return nil, err
 	}
@@ -1969,16 +1997,55 @@ const getNodesByLastUpdateRange = `-- name: GetNodesByLastUpdateRange :many
 SELECT id, version, pub_key, alias, last_update, color, signature
 FROM graph_nodes
 WHERE last_update >= $1
-  AND last_update < $2
+  AND last_update <= $2
+  -- Pagination: We use (last_update, pub_key) as a compound cursor.
+  -- This ensures stable ordering and allows us to resume from where we left off.
+  -- We use COALESCE with -1 as sentinel since timestamps are always positive.
+  AND (
+    -- Include rows with last_update greater than cursor (or all rows if cursor is -1)
+    last_update > COALESCE($3, -1)
+    OR 
+    -- For rows with same last_update, use pub_key as tiebreaker
+    (last_update = COALESCE($3, -1) 
+     AND pub_key > $4)
+  )
+  -- Optional filter for public nodes only
+  AND (
+    -- If only_public is false or not provided, include all nodes
+    COALESCE($5, FALSE) IS FALSE
+    OR 
+    -- For V1 protocol, a node is public if it has at least one public channel.
+    -- A public channel has bitcoin_1_signature set (channel announcement received).
+    EXISTS (
+      SELECT 1
+      FROM graph_channels c
+      WHERE c.version = 1
+        AND c.bitcoin_1_signature IS NOT NULL
+        AND (c.node_id_1 = graph_nodes.id OR c.node_id_2 = graph_nodes.id)
+    )
+  )
+ORDER BY last_update ASC, pub_key ASC
+LIMIT COALESCE($6, 999999999)
 `
 
 type GetNodesByLastUpdateRangeParams struct {
-	StartTime sql.NullInt64
+	StartTime  sql.NullInt64
-	EndTime   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
 	}

sqldb/sqlc/queries/graph.sql

@@ -164,7 +164,35 @@ ORDER BY node_id, type, position;
 SELECT *
 FROM graph_nodes
 WHERE last_update >= @start_time
-  AND last_update < @end_time;
+  AND last_update <= @end_time
+  -- Pagination: We use (last_update, pub_key) as a compound cursor.
+  -- This ensures stable ordering and allows us to resume from where we left off.
+  -- We use COALESCE with -1 as sentinel since timestamps are always positive.
+  AND (
+    -- Include rows with last_update greater than cursor (or all rows if cursor is -1)
+    last_update > COALESCE(sqlc.narg('last_update'), -1)
+    OR 
+    -- For rows with same last_update, use pub_key as tiebreaker
+    (last_update = COALESCE(sqlc.narg('last_update'), -1) 
+     AND pub_key > sqlc.narg('last_pub_key'))
+  )
+  -- Optional filter for public nodes only
+  AND (
+    -- If only_public is false or not provided, include all nodes
+    COALESCE(sqlc.narg('only_public'), FALSE) IS FALSE
+    OR 
+    -- For V1 protocol, a node is public if it has at least one public channel.
+    -- A public channel has bitcoin_1_signature set (channel announcement received).
+    EXISTS (
+      SELECT 1
+      FROM graph_channels c
+      WHERE c.version = 1
+        AND c.bitcoin_1_signature IS NOT NULL
+        AND (c.node_id_1 = graph_nodes.id OR c.node_id_2 = graph_nodes.id)
+    )
+  )
+ORDER BY last_update ASC, pub_key ASC
+LIMIT COALESCE(sqlc.narg('max_results'), 999999999);
 
 -- name: DeleteNodeAddresses :exec
 DELETE FROM graph_node_addresses
@@ -444,12 +472,30 @@ WHERE c.version = @version
        OR
        (cp2.last_update >= @start_time AND cp2.last_update < @end_time)
   )
+  -- Pagination using compound cursor (max_update_time, id).
+  -- We use COALESCE with -1 as sentinel since timestamps are always positive.
+  AND (
+       (CASE
+           WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0)
+               THEN COALESCE(cp1.last_update, 0)
+           ELSE COALESCE(cp2.last_update, 0)
+       END > COALESCE(sqlc.narg('last_update_time'), -1))
+       OR 
+       (CASE
+           WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0)
+               THEN COALESCE(cp1.last_update, 0)
+           ELSE COALESCE(cp2.last_update, 0)
+       END = COALESCE(sqlc.narg('last_update_time'), -1) 
+       AND c.id > COALESCE(sqlc.narg('last_id'), -1))
+  )
 ORDER BY
     CASE
         WHEN COALESCE(cp1.last_update, 0) >= COALESCE(cp2.last_update, 0)
             THEN COALESCE(cp1.last_update, 0)
         ELSE COALESCE(cp2.last_update, 0)
-        END ASC;
+    END ASC,
+    c.id ASC
+LIMIT COALESCE(sqlc.narg('max_results'), 999999999);
 
 -- name: GetChannelByOutpointWithPolicies :one
 SELECT