← Watch feed
graph/db: version NodeUpdatesInHorizon and ChanUpdatesInHorizon
Public commit record
73/100 · Adequate
What the developer wrote
Authored by Elle Mouton
graph/db: version NodeUpdatesInHorizon and ChanUpdatesInHorizon
Replace the (startTime, endTime time.Time) parameters on
NodeUpdatesInHorizon and ChanUpdatesInHorizon with
(v GossipVersion, r NodeUpdateRange/ChanUpdateRange). The range
types enforce version-correct bounds at the type level: v1 uses unix
timestamps, v2 will use block heights.
The KV store rejects non-v1 versions since it only stores v1 data.
The SQL store dispatches to version-specific helpers
(nodeUpdatesInHorizonV1, chanUpdatesInHorizonV1); the v2
block-height paths return an error for now and will be wired up in
follow-up commits.
VersionedGraph wrappers supply the version from the embedded field,
so callers only pass the range.
✓ Specific, descriptive subject✓ Names a concrete action or component✓ Provides detailed explanatory context
What changed, and why it matters
This commit refactors how LND queries recent node and channel gossip updates. It changes the function signatures to accept a gossip version and a structured range object instead of raw start/end timestamps. The v1 time-based behavior is preserved, while v2 block-height support is stubbed out to return errors. There is no direct security fix here; it is preparatory plumbing for a future protocol version.
Recommended action
No immediate action required. Treat as routine refactoring. Monitor follow-up commits that wire up v2 block-height queries to ensure validation and indexing remain consistent.
Security signals we found
01
API versioning adds explicit validation and rejects unsupported gossip versions rather than silently interpreting data under the wrong schema
02
Range validation is centralized in validateForVersion, reducing the chance that v2 block-height queries accidentally run against v1 timestamp indexes
03
No memory-safety, cryptographic, or authorization changes are present in the diff
Evidence from the diff
The patch versions the NodeUpdatesInHorizon and ChanUpdatesInHorizon APIs. It introduces NodeUpdateRange and ChanUpdateRange types that carry optional time or block-height ranges, and a validateForVersion helper that enforces version-correct bounds at the type level. The KV store now rejects non-v1 versions with ErrVersionNotSupportedForKVDB. The SQL store dispatches to v1 helpers and returns errors for v2 and unknown versions. Callers in discovery/chan_series.go and graph/builder.go are updated to pass the new range structs, with VersionedGraph wrappers supplying the version implicitly.
Changed components
graph/db/graph.gograph/db/interfaces.gograph/db/kv_store.gograph/db/sql_store.gograph/db/options.godiscovery/chan_series.gograph/builder.goInspect captured patch +234 / −85
diff --git a/discovery/chan_series.go b/discovery/chan_series.go
index 4a9a519..8fa460a 100644
--- a/discovery/chan_series.go
+++ b/discovery/chan_series.go
@@ -6,6 +6,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"
@@ -115,7 +116,10 @@ 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 := c.graph.ChanUpdatesInHorizon(
- context.TODO(), startTime, endTime,
+ context.TODO(), graphdb.ChanUpdateRange{
+ StartTime: fn.Some(startTime),
+ EndTime: fn.Some(endTime),
+ },
)
for channel, err := range chansInHorizon {
@@ -181,7 +185,10 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
// update within the horizon as well. We send these second to
// ensure that they follow any active channels they have.
nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon(
- context.TODO(), startTime, endTime,
+ context.TODO(), graphdb.NodeUpdateRange{
+ StartTime: fn.Some(startTime),
+ EndTime: fn.Some(endTime),
+ },
graphdb.WithIterPublicNodesOnly(),
)
for nodeAnn, err := range nodeAnnsInHorizon {
diff --git a/graph/builder.go b/graph/builder.go
index 614c110..14f5b7a 100644
--- a/graph/builder.go
+++ b/graph/builder.go
@@ -12,6 +12,7 @@ import (
"github.com/btcsuite/btcd/wire"
"github.com/lightningnetwork/lnd/batch"
"github.com/lightningnetwork/lnd/chainntnfs"
+ "github.com/lightningnetwork/lnd/fn/v2"
graphdb "github.com/lightningnetwork/lnd/graph/db"
"github.com/lightningnetwork/lnd/graph/db/models"
"github.com/lightningnetwork/lnd/lnutils"
@@ -647,8 +648,11 @@ func (b *Builder) pruneZombieChans() error {
startTime := time.Unix(0, 0)
endTime := time.Now().Add(-1 * chanExpiry)
- oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon(
- context.TODO(), startTime, endTime,
+ oldEdgesIter := b.v1Graph.ChanUpdatesInHorizon(
+ context.TODO(), graphdb.ChanUpdateRange{
+ StartTime: fn.Some(startTime),
+ EndTime: fn.Some(endTime),
+ },
)
for u, err := range oldEdgesIter {
diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go
index 19d6a13..9a3f76d 100644
--- a/graph/db/benchmark_test.go
+++ b/graph/db/benchmark_test.go
@@ -662,7 +662,13 @@ func BenchmarkGraphReadMethods(b *testing.B) {
name: "NodeUpdatesInHorizon",
fn: func(b testing.TB, store Store) {
iter := store.NodeUpdatesInHorizon(
- ctx, time.Unix(0, 0), time.Now(),
+ ctx, lnwire.GossipVersion1,
+ NodeUpdateRange{
+ StartTime: fn.Some(
+ time.Unix(0, 0),
+ ),
+ EndTime: fn.Some(time.Now()),
+ },
)
_, err := fn.CollectErr(iter)
require.NoError(b, err)
@@ -713,7 +719,13 @@ func BenchmarkGraphReadMethods(b *testing.B) {
name: "ChanUpdatesInHorizon",
fn: func(b testing.TB, store Store) {
iter := store.ChanUpdatesInHorizon(
- ctx, time.Unix(0, 0), time.Now(),
+ ctx, lnwire.GossipVersion1,
+ ChanUpdateRange{
+ StartTime: fn.Some(
+ time.Unix(0, 0),
+ ),
+ EndTime: fn.Some(time.Now()),
+ },
)
_, err := fn.CollectErr(iter)
require.NoError(b, err)
diff --git a/graph/db/graph.go b/graph/db/graph.go
index 6b29bfc..8e2ebc9 100644
--- a/graph/db/graph.go
+++ b/graph/db/graph.go
@@ -721,13 +721,13 @@ func (c *ChannelGraph) ForEachNodeCacheable(ctx context.Context,
return c.db.ForEachNodeCacheable(ctx, v, cb, reset)
}
-// NodeUpdatesInHorizon returns all known lightning nodes with an update
-// timestamp in [startTime, endTime) per BOLT 07.
+// NodeUpdatesInHorizon returns all known lightning nodes with updates within
+// the passed range for the given gossip version.
func (c *ChannelGraph) NodeUpdatesInHorizon(ctx context.Context,
- startTime, endTime time.Time,
+ v lnwire.GossipVersion, r NodeUpdateRange,
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
- return c.db.NodeUpdatesInHorizon(ctx, startTime, endTime, opts...)
+ return c.db.NodeUpdatesInHorizon(ctx, v, r, opts...)
}
// HasV1Node determines if the graph has a vertex identified by the target node
@@ -784,12 +784,12 @@ func (c *ChannelGraph) HighestChanID(ctx context.Context,
}
// ChanUpdatesInHorizon returns all known channel edges with at least one
-// policy update timestamp in [startTime, endTime) per BOLT 07.
+// policy update within the specified range for the given gossip version.
func (c *ChannelGraph) ChanUpdatesInHorizon(ctx context.Context,
- startTime, endTime time.Time,
+ v lnwire.GossipVersion, r ChanUpdateRange,
opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
- return c.db.ChanUpdatesInHorizon(ctx, startTime, endTime, opts...)
+ return c.db.ChanUpdatesInHorizon(ctx, v, r, opts...)
}
// FilterChannelRange returns channel IDs within the passed block height range
@@ -944,13 +944,23 @@ func (c *VersionedGraph) NumZombies(ctx context.Context) (uint64, error) {
return c.db.NumZombies(ctx, c.v)
}
-// NodeUpdatesInHorizon returns all known lightning nodes with an update
-// timestamp in [startTime, endTime) per BOLT 07.
+// NodeUpdatesInHorizon returns all known lightning nodes with updates within
+// the passed range. The version is supplied by the embedded field.
func (c *VersionedGraph) NodeUpdatesInHorizon(ctx context.Context,
- startTime, endTime time.Time,
+ r NodeUpdateRange,
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
- return c.db.NodeUpdatesInHorizon(ctx, startTime, endTime, opts...)
+ return c.db.NodeUpdatesInHorizon(ctx, c.v, r, opts...)
+}
+
+// ChanUpdatesInHorizon returns all known channel edges with at least one
+// policy update within the specified range. The version is supplied by the
+// embedded field.
+func (c *VersionedGraph) ChanUpdatesInHorizon(ctx context.Context,
+ r ChanUpdateRange,
+ opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
+
+ return c.db.ChanUpdatesInHorizon(ctx, c.v, r, opts...)
}
// ChannelView returns the verifiable edge information for each active channel.
diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go
index 36d494e..81aef91 100644
--- a/graph/db/graph_test.go
+++ b/graph/db/graph_test.go
@@ -2440,7 +2440,10 @@ 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 := graph.ChanUpdatesInHorizon(
- ctx, time.Unix(999, 0), time.Unix(9999, 0),
+ ctx, lnwire.GossipVersion1, ChanUpdateRange{
+ StartTime: fn.Some(time.Unix(999, 0)),
+ EndTime: fn.Some(time.Unix(9999, 0)),
+ },
)
chanUpdates, err := fn.CollectErr(chanIter)
@@ -2547,7 +2550,10 @@ func TestChanUpdatesInHorizon(t *testing.T) {
}
for _, queryCase := range queryCases {
respIter := graph.ChanUpdatesInHorizon(
- ctx, queryCase.start, queryCase.end,
+ ctx, lnwire.GossipVersion1, ChanUpdateRange{
+ StartTime: fn.Some(queryCase.start),
+ EndTime: fn.Some(queryCase.end),
+ },
)
resp, err := fn.CollectErr(respIter)
@@ -2584,7 +2590,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.
nodeUpdatesIter := graph.NodeUpdatesInHorizon(
- ctx, time.Unix(999, 0), time.Unix(9999, 0),
+ ctx, lnwire.GossipVersion1, NodeUpdateRange{
+ StartTime: fn.Some(time.Unix(999, 0)),
+ EndTime: fn.Some(time.Unix(9999, 0)),
+ },
)
nodeUpdates, err := fn.CollectErr(nodeUpdatesIter)
require.NoError(t, err, "unable to query for node updates")
@@ -2659,7 +2668,10 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
}
for _, queryCase := range queryCases {
iter := graph.NodeUpdatesInHorizon(
- ctx, queryCase.start, queryCase.end,
+ ctx, lnwire.GossipVersion1, NodeUpdateRange{
+ StartTime: fn.Some(queryCase.start),
+ EndTime: fn.Some(queryCase.end),
+ },
)
resp, err := fn.CollectErr(iter)
@@ -2793,7 +2805,11 @@ func testNodeUpdatesWithBatchSize(t *testing.T, ctx context.Context,
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
iter := testGraph.NodeUpdatesInHorizon(
- ctx, tc.start, tc.end,
+ ctx, lnwire.GossipVersion1,
+ NodeUpdateRange{
+ StartTime: fn.Some(tc.start),
+ EndTime: fn.Some(tc.end),
+ },
WithNodeUpdateIterBatchSize(
batchSize,
),
@@ -2866,7 +2882,13 @@ func TestNodeUpdatesInHorizonEarlyTermination(t *testing.T) {
for _, stopAt := range terminationPoints {
t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) {
iter := graph.NodeUpdatesInHorizon(
- ctx, startTime, startTime.Add(200*time.Hour),
+ ctx, lnwire.GossipVersion1,
+ NodeUpdateRange{
+ StartTime: fn.Some(startTime),
+ EndTime: fn.Some(
+ startTime.Add(200 * time.Hour),
+ ),
+ },
WithNodeUpdateIterBatchSize(10),
)
@@ -2955,7 +2977,13 @@ func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) {
// Now we'll run the main query, and verify that we get
// back the expected number of channels.
iter := graph.ChanUpdatesInHorizon(
- ctx, startTime, startTime.Add(26*time.Hour),
+ ctx, lnwire.GossipVersion1,
+ ChanUpdateRange{
+ StartTime: fn.Some(startTime),
+ EndTime: fn.Some(
+ startTime.Add(26 * time.Hour),
+ ),
+ },
WithChanUpdateIterBatchSize(batchSize),
)
@@ -3037,7 +3065,11 @@ func TestNodeUpdatesInHorizonExclusiveEnd(t *testing.T) {
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
iter := graph.NodeUpdatesInHorizon(
- ctx, tc.start, tc.end,
+ ctx, lnwire.GossipVersion1,
+ NodeUpdateRange{
+ StartTime: fn.Some(tc.start),
+ EndTime: fn.Some(tc.end),
+ },
)
nodes, err := fn.CollectErr(iter)
@@ -3131,7 +3163,11 @@ func TestChanUpdatesInHorizonExclusiveEnd(t *testing.T) {
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
iter := graph.ChanUpdatesInHorizon(
- ctx, tc.start, tc.end,
+ ctx, lnwire.GossipVersion1,
+ ChanUpdateRange{
+ StartTime: fn.Some(tc.start),
+ EndTime: fn.Some(tc.end),
+ },
)
channels, err := fn.CollectErr(iter)
@@ -3603,9 +3639,15 @@ func TestStressTestChannelGraphAPI(t *testing.T) {
{
name: "ChanUpdateInHorizon",
fn: func() error {
+ now := time.Now()
iter := graph.ChanUpdatesInHorizon(
- ctx, time.Now().Add(-time.Hour),
- time.Now(),
+ ctx, lnwire.GossipVersion1,
+ ChanUpdateRange{
+ StartTime: fn.Some(
+ now.Add(-time.Hour),
+ ),
+ EndTime: fn.Some(now),
+ },
)
_, err := fn.CollectErr(iter)
@@ -4465,7 +4507,10 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
startTime := time.Unix(9, 0)
endTime := node1.LastUpdate.Add(time.Minute)
nodesInHorizonIter := graph.NodeUpdatesInHorizon(
- ctx, startTime, endTime,
+ ctx, NodeUpdateRange{
+ StartTime: fn.Some(startTime),
+ EndTime: fn.Some(endTime),
+ },
)
// We should only have a single node, and that node should exactly
@@ -4483,7 +4528,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.
nodesInHorizonIter = graph.NodeUpdatesInHorizon(
- ctx, startTime, endTime,
+ ctx, NodeUpdateRange{
+ StartTime: fn.Some(startTime),
+ EndTime: fn.Some(endTime),
+ },
)
nodesInHorizon, err = fn.CollectErr(nodesInHorizonIter)
require.NoError(t, err, "unable to fetch nodes in horizon")
diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go
index 2c9d5f5..5da4dcd 100644
--- a/graph/db/interfaces.go
+++ b/graph/db/interfaces.go
@@ -121,11 +121,12 @@ type Store interface { //nolint:interfacebloat
nodePub route.Vertex) error
// NodeUpdatesInHorizon returns all the known lightning nodes which have
- // an update timestamp greater than or equal to startTime and less than
- // endTime, i.e. the range [startTime, endTime) per BOLT 07. This
+ // updates within the passed range for the given gossip version. For v1
+ // gossip, the range is time-based with [start, end) per BOLT 07. This
// method can be used by two nodes to quickly determine if they have
// the same set of up to date node announcements.
- NodeUpdatesInHorizon(ctx context.Context, startTime, endTime time.Time,
+ NodeUpdatesInHorizon(ctx context.Context, v lnwire.GossipVersion,
+ r NodeUpdateRange,
opts ...IteratorOption) iter.Seq2[*models.Node, error]
// FetchNode attempts to look up a target node by its identity
@@ -257,11 +258,11 @@ type Store interface { //nolint:interfacebloat
uint64, error)
// ChanUpdatesInHorizon returns all the known channel edges which have
- // at least one edge that has an update timestamp greater than or equal
- // to startTime and less than endTime, i.e. the range
- // [startTime, endTime) per BOLT 07.
- ChanUpdatesInHorizon(ctx context.Context,
- startTime, endTime time.Time,
+ // at least one edge update within the specified range for the given
+ // gossip version. For v1 gossip, the range is time-based with
+ // [start, end) per BOLT 07.
+ ChanUpdatesInHorizon(ctx context.Context, v lnwire.GossipVersion,
+ r ChanUpdateRange,
opts ...IteratorOption) iter.Seq2[ChannelEdge, error]
// FilterKnownChanIDs takes a set of channel IDs and return the subset
diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go
index 951c879..484c864 100644
--- a/graph/db/kv_store.go
+++ b/graph/db/kv_store.go
@@ -2396,10 +2396,10 @@ func (c *KVStore) fetchNextChanUpdateBatch(
}
// ChanUpdatesInHorizon returns all the known channel edges which have at least
-// one edge that has an update timestamp greater than or equal to startTime and
-// less than endTime, i.e. the range [startTime, endTime) per BOLT 07.
+// one edge update within the specified range for the given gossip version. For
+// v1, the range is time-based with [start, end) per BOLT 07.
func (c *KVStore) ChanUpdatesInHorizon(_ context.Context,
- startTime, endTime time.Time,
+ v lnwire.GossipVersion, r ChanUpdateRange,
opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
cfg := defaultIteratorConfig()
@@ -2408,8 +2408,19 @@ func (c *KVStore) ChanUpdatesInHorizon(_ context.Context,
}
return func(yield func(ChannelEdge, error) bool) {
+ if v != lnwire.GossipVersion1 {
+ yield(ChannelEdge{}, ErrVersionNotSupportedForKVDB)
+ return
+ }
+ if err := r.validateForVersion(v); err != nil {
+ yield(ChannelEdge{}, err)
+ return
+ }
+
iterState := newChanUpdatesIterator(
- cfg.chanUpdateIterBatchSize, startTime, endTime,
+ cfg.chanUpdateIterBatchSize,
+ r.StartTime.UnwrapOr(time.Time{}),
+ r.EndTime.UnwrapOr(time.Time{}),
)
for {
@@ -2458,8 +2469,8 @@ func (c *KVStore) ChanUpdatesInHorizon(_ context.Context,
float64(iterState.total), iterState.hits,
iterState.total)
} else {
- log.Tracef("ChanUpdatesInHorizon returned no edges "+
- "in horizon (%s, %s)", startTime, endTime)
+ log.Tracef("ChanUpdatesInHorizon(v%d) returned "+
+ "no edges in horizon", v)
}
}
}
@@ -2649,11 +2660,11 @@ func (c *KVStore) fetchNextNodeBatch(
return nodeBatch, hasMore, nil
}
-// NodeUpdatesInHorizon returns all the known lightning nodes which have an
-// update timestamp greater than or equal to startTime and less than endTime,
-// i.e. the range [startTime, endTime) per BOLT 07.
-func (c *KVStore) NodeUpdatesInHorizon(_ context.Context, startTime,
- endTime time.Time,
+// NodeUpdatesInHorizon returns all the known lightning nodes which have
+// updates within the passed range for the given gossip version. For v1, the
+// range is time-based with [start, end) per BOLT 07.
+func (c *KVStore) NodeUpdatesInHorizon(_ context.Context,
+ v lnwire.GossipVersion, r NodeUpdateRange,
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
cfg := defaultIteratorConfig()
@@ -2662,10 +2673,20 @@ func (c *KVStore) NodeUpdatesInHorizon(_ context.Context, startTime,
}
return func(yield func(*models.Node, error) bool) {
+ if v != lnwire.GossipVersion1 {
+ yield(nil, ErrVersionNotSupportedForKVDB)
+ return
+ }
+ if err := r.validateForVersion(v); err != nil {
+ yield(nil, err)
+ return
+ }
+
// Initialize iterator state.
state := newNodeUpdatesIterator(
cfg.nodeUpdateIterBatchSize,
- startTime, endTime,
+ r.StartTime.UnwrapOr(time.Time{}),
+ r.EndTime.UnwrapOr(time.Time{}),
cfg.iterPublicNodes,
)
diff --git a/graph/db/options.go b/graph/db/options.go
index e33396d..5c876a1 100644
--- a/graph/db/options.go
+++ b/graph/db/options.go
@@ -2,7 +2,6 @@ package graphdb
import (
"fmt"
- "iter"
"time"
"github.com/lightningnetwork/lnd/fn/v2"
@@ -136,13 +135,6 @@ func (r ChanUpdateRange) validateForVersion(v lnwire.GossipVersion) error {
return nil
}
-// chanUpdateRangeErrIter returns an iterator that yields a single error.
-func chanUpdateRangeErrIter(err error) iter.Seq2[ChannelEdge, error] {
- return func(yield func(ChannelEdge, error) bool) {
- _ = yield(ChannelEdge{}, err)
- }
-}
-
// NodeUpdateRange describes a range for node updates. Only one of the time or
// height ranges should be set depending on the gossip version.
type NodeUpdateRange struct {
diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go
index a16fbf9..4d3f635 100644
--- a/graph/db/sql_store.go
+++ b/graph/db/sql_store.go
@@ -606,22 +606,54 @@ func (s *SQLStore) SetSourceNode(ctx context.Context,
}, sqldb.NoOpReset)
}
-// NodeUpdatesInHorizon returns all the known lightning nodes which have an
-// update timestamp greater than or equal to startTime and less than endTime,
-// i.e. the range [startTime, endTime) per BOLT 07. This method can be used by
-// two nodes to quickly determine if they have the same set of up to date node
-// announcements.
+// NodeUpdatesInHorizon returns all the known lightning nodes which have
+// updates within the passed range for the given gossip version. This method
+// can be used by two nodes to quickly determine if they have the same set of
+// up-to-date node announcements.
//
// NOTE: This is part of the Store interface.
func (s *SQLStore) NodeUpdatesInHorizon(ctx context.Context,
- startTime, endTime time.Time,
+ v lnwire.GossipVersion, r NodeUpdateRange,
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
+ if err := r.validateForVersion(v); err != nil {
+ return func(yield func(*models.Node, error) bool) {
+ _ = yield(nil, err)
+ }
+ }
+
cfg := defaultIteratorConfig()
for _, opt := range opts {
opt(cfg)
}
+ switch v {
+ case gossipV1:
+ return s.nodeUpdatesInHorizonV1(ctx, r, cfg)
+
+ case gossipV2:
+ err := fmt.Errorf("v2 node updates in horizon not yet " +
+ "implemented")
+ return func(yield func(*models.Node, error) bool) {
+ _ = yield(nil, err)
+ }
+
+ default:
+ err := fmt.Errorf("unknown gossip version: %v", v)
+ return func(yield func(*models.Node, error) bool) {
+ _ = yield(nil, err)
+ }
+ }
+}
+
+// nodeUpdatesInHorizonV1 implements the v1 time-based node horizon query.
+func (s *SQLStore) nodeUpdatesInHorizonV1(ctx context.Context,
+ r NodeUpdateRange,
+ cfg *iterConfig) iter.Seq2[*models.Node, error] {
+
+ startTime := r.StartTime.UnwrapOr(time.Time{})
+ endTime := r.EndTime.UnwrapOr(time.Time{})
+
return func(yield func(*models.Node, error) bool) {
var (
lastUpdateTime sql.NullInt64
@@ -1162,28 +1194,52 @@ func (s *SQLStore) updateChanCacheBatch(v lnwire.GossipVersion,
}
// ChanUpdatesInHorizon returns all the known channel edges which have at least
-// one edge that has an update timestamp greater than or equal to startTime and
-// less than endTime, i.e. the range [startTime, endTime) per BOLT 07.
-//
-// 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
+// one edge update within the specified range for the given gossip version. For
+// v1, the range is time-based with [start, end) per BOLT 07.
//
// NOTE: This is part of the Store interface.
func (s *SQLStore) ChanUpdatesInHorizon(ctx context.Context,
- startTime, endTime time.Time,
+ v lnwire.GossipVersion, r ChanUpdateRange,
opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
- // Apply options.
+ if err := r.validateForVersion(v); err != nil {
+ return func(yield func(ChannelEdge, error) bool) {
+ _ = yield(ChannelEdge{}, err)
+ }
+ }
+
cfg := defaultIteratorConfig()
for _, opt := range opts {
opt(cfg)
}
+ switch v {
+ case gossipV1:
+ return s.chanUpdatesInHorizonV1(ctx, r, cfg)
+
+ case gossipV2:
+ err := fmt.Errorf("v2 chan updates in horizon not yet " +
+ "implemented")
+ return func(yield func(ChannelEdge, error) bool) {
+ _ = yield(ChannelEdge{}, err)
+ }
+
+ default:
+ err := fmt.Errorf("unknown gossip version: %v", v)
+ return func(yield func(ChannelEdge, error) bool) {
+ _ = yield(ChannelEdge{}, err)
+ }
+ }
+}
+
+// chanUpdatesInHorizonV1 implements the v1 time-based channel horizon query.
+func (s *SQLStore) chanUpdatesInHorizonV1(ctx context.Context,
+ r ChanUpdateRange,
+ cfg *iterConfig) iter.Seq2[ChannelEdge, error] {
+
+ startTime := r.StartTime.UnwrapOr(time.Time{})
+ endTime := r.EndTime.UnwrapOr(time.Time{})
+
return func(yield func(ChannelEdge, error) bool) {
var (
edgesSeen = make(map[uint64]struct{})
@@ -1210,7 +1266,7 @@ func (s *SQLStore) ChanUpdatesInHorizon(ctx context.Context,
func(db SQLQueries) error {
//nolint:ll
params := sqlc.GetChannelsByPolicyLastUpdateRangeParams{
- Version: int16(lnwire.GossipVersion1),
+ Version: int16(gossipV1),
StartTime: sqldb.SQLInt64(
startTime.Unix(),
),
@@ -1261,7 +1317,7 @@ func (s *SQLStore) ChanUpdatesInHorizon(ctx context.Context,
// Check cache (we already hold
// shared read lock).
channel, ok := s.chanCache.get(
- lnwire.GossipVersion1,
+ gossipV1,
chanIDInt,
)
if ok {
@@ -1317,9 +1373,7 @@ func (s *SQLStore) ChanUpdatesInHorizon(ctx context.Context,
// Update cache after successful batch yield, setting
// the cache lock only once for the entire batch.
- s.updateChanCacheBatch(
- lnwire.GossipVersion1, edgesToCache,
- )
+ s.updateChanCacheBatch(gossipV1, edgesToCache)
edgesToCache = make(map[uint64]ChannelEdge)
// If the batch didn't yield anything, then we're done.
@@ -1329,12 +1383,12 @@ func (s *SQLStore) ChanUpdatesInHorizon(ctx context.Context,
}
if total > 0 {
- log.Debugf("ChanUpdatesInHorizon hit percentage: "+
- "%.2f (%d/%d)",
+ log.Debugf("ChanUpdatesInHorizon(v1) 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)
+ log.Debugf("ChanUpdatesInHorizon(v1) returned no "+
+ "edges in horizon")
}
}
}
Risk score
Our methodology →Why this scored 19/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.