← Watch feed
graph/db: add gossip version parameter to ForEachChannel
Public commit record
73/100 · Adequate
What the developer wrote
Authored by Elle Mouton
graph/db: add gossip version parameter to ForEachChannel
Update ForEachChannel to accept a gossip version parameter, allowing
callers to specify which gossip version's channels should be iterated.
This change prepares the graph database for supporting multiple gossip
versions while maintaining backward compatibility.
The KVStore implementation validates that only GossipVersion1 is
requested, returning ErrVersionNotSupportedForKVDB for other versions.
The SQLStore implementation validates known versions and passes the
version through to the underlying paginated query.
Update VersionedGraph to include a ForEachChannel method that
automatically uses its configured gossip version, and update the
DescribeGraph RPC handler to use the v1Graph instead of the global
graphDB.
✓ Specific, descriptive subject✓ Names a concrete action or component✓ Provides detailed explanatory context
What changed, and why it matters
This commit is a routine internal refactoring of how the Lightning Network graph database iterates over channels. It adds a 'gossip version' parameter to the ForEachChannel function so the code can later support multiple gossip versions, but currently only version 1 is supported. There is no security fix or vulnerability present in the change itself.
Recommended action
No security action required. Review as normal code-quality change.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The commit modifies the ForEachChannel API across the graph/db package to accept a lnwire.GossipVersion parameter. KVStore rejects non-v1 versions with ErrVersionNotSupportedForKVDB. SQLStore validates known versions and passes the version into the paginated SQL query. A VersionedGraph wrapper is added that uses its configured version, and the DescribeGraph RPC handler is switched from the global graphDB to the v1Graph. This is preparatory work for multi-gossip-version support and maintains backward compatibility.
Changed components
graph/db/Store interfacegraph/db/KVStore.ForEachChannelgraph/db/SQLStore.ForEachChannelgraph/db/ChannelGraph.ForEachChannelgraph/db/VersionedGraph.ForEachChannelrpcserver.DescribeGraphInspect captured patch +94 / −67
diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go
index d7cc3fb..00cfeaf 100644
--- a/graph/db/benchmark_test.go
+++ b/graph/db/benchmark_test.go
@@ -370,8 +370,8 @@ func TestPopulateDBs(t *testing.T) {
numPolicies = 0
)
err := graph.ForEachChannel(
- ctx, func(info *models.ChannelEdgeInfo,
- policy,
+ ctx, lnwire.GossipVersion1,
+ func(info *models.ChannelEdgeInfo, policy,
policy2 *models.ChannelEdgePolicy) error {
numChans++
@@ -497,48 +497,49 @@ func syncGraph(t *testing.T, src, dest *ChannelGraph) {
}
var wgChans sync.WaitGroup
- err = src.ForEachChannel(ctx, func(info *models.ChannelEdgeInfo,
- policy1, policy2 *models.ChannelEdgePolicy) error {
-
- // Add each channel & policy. We do this in a goroutine to
- // take advantage of batch processing.
- wgChans.Add(1)
- go func() {
- defer wgChans.Done()
-
- err := dest.AddChannelEdge(
- ctx, info, batch.LazyAdd(),
- )
- if !errors.Is(err, ErrEdgeAlreadyExist) {
- require.NoError(t, err)
- }
-
- if policy1 != nil {
- err = dest.UpdateEdgePolicy(
- ctx, policy1, batch.LazyAdd(),
+ err = src.ForEachChannel(ctx, lnwire.GossipVersion1,
+ func(info *models.ChannelEdgeInfo,
+ policy1, policy2 *models.ChannelEdgePolicy) error {
+
+ // Add each channel & policy. We do this in a goroutine
+ // to take advantage of batch processing.
+ wgChans.Add(1)
+ go func() {
+ defer wgChans.Done()
+
+ err := dest.AddChannelEdge(
+ ctx, info, batch.LazyAdd(),
)
- require.NoError(t, err)
- }
+ if !errors.Is(err, ErrEdgeAlreadyExist) {
+ require.NoError(t, err)
+ }
- if policy2 != nil {
- err = dest.UpdateEdgePolicy(
- ctx, policy2, batch.LazyAdd(),
- )
- require.NoError(t, err)
- }
+ if policy1 != nil {
+ err = dest.UpdateEdgePolicy(
+ ctx, policy1, batch.LazyAdd(),
+ )
+ require.NoError(t, err)
+ }
- mu.Lock()
- total++
- chunk++
- s.Do(func() {
- reportChanStats()
- chunk = 0
- })
- mu.Unlock()
- }()
+ if policy2 != nil {
+ err = dest.UpdateEdgePolicy(
+ ctx, policy2, batch.LazyAdd(),
+ )
+ require.NoError(t, err)
+ }
- return nil
- }, func() {})
+ mu.Lock()
+ total++
+ chunk++
+ s.Do(func() {
+ reportChanStats()
+ chunk = 0
+ })
+ mu.Unlock()
+ }()
+
+ return nil
+ }, func() {})
require.NoError(t, err)
wgChans.Wait()
@@ -638,7 +639,8 @@ func BenchmarkGraphReadMethods(b *testing.B) {
fn: func(b testing.TB, store Store) {
//nolint:ll
err := store.ForEachChannel(
- ctx, func(_ *models.ChannelEdgeInfo,
+ ctx, lnwire.GossipVersion1,
+ func(_ *models.ChannelEdgeInfo,
_ *models.ChannelEdgePolicy,
_ *models.ChannelEdgePolicy) error {
@@ -821,7 +823,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
//nolint:ll
err = store.ForEachChannel(
- ctx,
+ ctx, lnwire.GossipVersion1,
func(_ *models.ChannelEdgeInfo,
_,
_ *models.ChannelEdgePolicy) error {
diff --git a/graph/db/graph.go b/graph/db/graph.go
index 3aa860c..d7398e0 100644
--- a/graph/db/graph.go
+++ b/graph/db/graph.go
@@ -642,10 +642,11 @@ func (c *ChannelGraph) IsPublicNode(pubKey [33]byte) (bool, error) {
// ForEachChannel iterates through all channel edges stored within the graph.
func (c *ChannelGraph) ForEachChannel(ctx context.Context,
- cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
- *models.ChannelEdgePolicy) error, reset func()) error {
+ v lnwire.GossipVersion, cb func(*models.ChannelEdgeInfo,
+ *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
+ reset func()) error {
- return c.db.ForEachChannel(ctx, cb, reset)
+ return c.db.ForEachChannel(ctx, v, cb, reset)
}
// ForEachChannelCacheable iterates through all channel edges for the cache.
@@ -904,6 +905,14 @@ func (c *VersionedGraph) HasChannelEdge(chanID uint64) (bool, bool, error) {
return c.db.HasChannelEdge(c.v, chanID)
}
+// ForEachChannel iterates through all channel edges stored within the graph.
+func (c *VersionedGraph) ForEachChannel(ctx context.Context,
+ cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+ *models.ChannelEdgePolicy) error, reset func()) error {
+
+ return c.db.ForEachChannel(ctx, c.v, cb, reset)
+}
+
// IsPublicNode determines whether the node is seen as public in the graph.
func (c *VersionedGraph) IsPublicNode(pubKey [33]byte) (bool, error) {
return c.db.IsPublicNode(c.v, pubKey)
diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go
index d1aff61..6a13e32 100644
--- a/graph/db/graph_test.go
+++ b/graph/db/graph_test.go
@@ -1219,7 +1219,8 @@ func TestEdgePolicyCRUD(t *testing.T) {
// assert that the deserialized policies match the original
// ones.
err := graph.ForEachChannel(
- ctx, func(info *models.ChannelEdgeInfo,
+ ctx, lnwire.GossipVersion1,
+ func(info *models.ChannelEdgeInfo,
policy1 *models.ChannelEdgePolicy,
policy2 *models.ChannelEdgePolicy) error {
@@ -1703,13 +1704,14 @@ func TestGraphTraversal(t *testing.T) {
// Iterate through all the known channels within the graph DB, once
// again if the map is empty that indicates that all edges have
// properly been reached.
- err = graph.ForEachChannel(ctx, func(ei *models.ChannelEdgeInfo,
- _ *models.ChannelEdgePolicy,
- _ *models.ChannelEdgePolicy) error {
+ err = graph.ForEachChannel(ctx, lnwire.GossipVersion1,
+ func(ei *models.ChannelEdgeInfo,
+ _ *models.ChannelEdgePolicy,
+ _ *models.ChannelEdgePolicy) error {
- delete(chanIndex, ei.ChannelID)
- return nil
- }, func() {})
+ delete(chanIndex, ei.ChannelID)
+ return nil
+ }, func() {})
require.NoError(t, err)
require.Len(t, chanIndex, 0)
@@ -2005,7 +2007,8 @@ func assertPruneTip(t *testing.T, graph *ChannelGraph,
func assertNumChans(t *testing.T, graph *ChannelGraph, n int) {
numChans := 0
err := graph.ForEachChannel(
- t.Context(), func(*models.ChannelEdgeInfo,
+ t.Context(), lnwire.GossipVersion1,
+ func(*models.ChannelEdgeInfo,
*models.ChannelEdgePolicy,
*models.ChannelEdgePolicy) error {
diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go
index 42f63f9..c650bd2 100644
--- a/graph/db/interfaces.go
+++ b/graph/db/interfaces.go
@@ -162,9 +162,12 @@ type Store interface { //nolint:interfacebloat
// NOTE: If an edge can't be found, or wasn't advertised, then a nil
// pointer for that particular channel edge routing policy will be
// passed into the callback.
- ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo,
- *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
- reset func()) error
+ //
+ // TODO(elle): add a cross-version iteration API and make this iterate
+ // over all versions.
+ ForEachChannel(ctx context.Context, v lnwire.GossipVersion,
+ cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+ *models.ChannelEdgePolicy) error, reset func()) error
// ForEachChannelCacheable iterates through all the channel edges stored
// within the graph and invokes the passed callback for each edge. The
diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go
index 58b227a..03eaf02 100644
--- a/graph/db/kv_store.go
+++ b/graph/db/kv_store.go
@@ -411,10 +411,14 @@ func (c *KVStore) AddrsForNode(ctx context.Context, v lnwire.GossipVersion,
// NOTE: If an edge can't be found, or wasn't advertised, then a nil pointer
// for that particular channel edge routing policy will be passed into the
// callback.
-func (c *KVStore) ForEachChannel(_ context.Context,
+func (c *KVStore) ForEachChannel(_ context.Context, v lnwire.GossipVersion,
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
*models.ChannelEdgePolicy) error, reset func()) error {
+ if v != lnwire.GossipVersion1 {
+ return ErrVersionNotSupportedForKVDB
+ }
+
return forEachChannel(c.db, cb, reset)
}
diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go
index 00be30f..b66e2fe 100644
--- a/graph/db/sql_store.go
+++ b/graph/db/sql_store.go
@@ -1610,11 +1610,16 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
//
// NOTE: part of the Store interface.
func (s *SQLStore) ForEachChannel(ctx context.Context,
- cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
- *models.ChannelEdgePolicy) error, reset func()) error {
+ v lnwire.GossipVersion, cb func(*models.ChannelEdgeInfo,
+ *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
+ reset func()) error {
+
+ if !isKnownGossipVersion(v) {
+ return fmt.Errorf("unsupported gossip version: %d", v)
+ }
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
- return forEachChannelWithPolicies(ctx, db, s.cfg, cb)
+ return forEachChannelWithPolicies(ctx, db, s.cfg, v, cb)
}, reset)
}
@@ -5973,8 +5978,8 @@ func forEachNodePaginated(ctx context.Context, cfg *sqldb.QueryConfig,
// forEachChannelWithPolicies executes a paginated query to process each channel
// with policies in the graph.
func forEachChannelWithPolicies(ctx context.Context, db SQLQueries,
- cfg *SQLStoreConfig, processChannel func(*models.ChannelEdgeInfo,
- *models.ChannelEdgePolicy,
+ cfg *SQLStoreConfig, v lnwire.GossipVersion,
+ processChannel func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
*models.ChannelEdgePolicy) error) error {
type channelBatchIDs struct {
@@ -5988,7 +5993,7 @@ func forEachChannelWithPolicies(ctx context.Context, db SQLQueries,
return db.ListChannelsWithPoliciesPaginated(
ctx, sqlc.ListChannelsWithPoliciesPaginatedParams{
- Version: int16(lnwire.GossipVersion1),
+ Version: int16(v),
ID: lastID,
Limit: limit,
},
diff --git a/rpcserver.go b/rpcserver.go
index 927635a..564d8cc 100644
--- a/rpcserver.go
+++ b/rpcserver.go
@@ -6853,10 +6853,11 @@ func (r *rpcServer) DescribeGraph(ctx context.Context,
}
}
- // Obtain the pointer to the global singleton channel graph, this will
- // provide a consistent view of the graph due to bolt db's
- // transactional model.
- graph := r.server.graphDB
+ // Obtain the pointer to the V1 channel graph. This will provide a
+ // consistent view of the graph due to bolt db's transactional model.
+ //
+ // TODO(elle): switch to a cross-version graph view when available.
+ graph := r.server.v1Graph
// First iterate through all the known nodes (connected or unconnected
// within the graph), collating their current state into the RPC
Risk score
Our methodology →Why this scored 15/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.