Skip to content

graph/db: cross-version graph Store queries #11175

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ViktorT-11 wants to merge 12 commits into
base: master
Choose a base branch
from
Open

graph/db: cross-version graph Store queries #11175

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
4b66417
graph/db: fix zero ChannelID in zombie ChannelEdgeInfo
ellemouton May 7, 2026
9dc6e7a
sqldb: add preferred-node and preferred-channel mapping tables
ellemouton Apr 7, 2026
c60d9e2
graph/db: add preferred channel fetch methods
ellemouton May 7, 2026
411e3f0
graph/db: wire preferred-table maintenance into write paths
ellemouton Apr 7, 2026
45e4dbb
graph/db: make ForEachNode and ForEachChannel cross-version
ellemouton Apr 7, 2026
c8f6343
graph/db: implement cross-version node traversal
ellemouton Apr 7, 2026
5ba2018
graph/db: move ForEachNode/ForEachChannel off VersionedGraph
ellemouton May 7, 2026
cb9e6b8
graph/db: thread version through ForEachNodeCached channel lookup
ellemouton May 7, 2026
da63500
graph/db: drop redundant VersionedGraph.GraphSession override
ellemouton May 14, 2026
d2caf97
graph/db: document why sqlNodeTraverser hardcodes v1
ellemouton May 14, 2026
08733f4
graph/db: detect preferred mapping divergence
ViktorT-11 Sep 22, 2026
6462add
docs: add release note for cross-version graph Store
ellemouton Apr 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions docs/release-notes/release-notes-0.22.0.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,13 @@

## Database

* Make the [graph `Store` interface
cross-version](https://github.com/lightningnetwork/lnd/pull/11175) so that
`ForEachNode`, `ForEachChannel`, and `ForEachNodeDirectedChannel` work across
gossip v1 and v2. Add `Preferred` fetch helpers and `GetVersions` queries
so callers can retrieve channels without knowing which gossip version
announced them.

## Code Health

## Tooling and Documentation
Expand All @@ -165,5 +172,7 @@
* bitromortac
* Boris Nagaev
* Erick Cestari
* Elle Mouton
* Jared Tobin
* Nishant Bansal
* ViktorT-11
20 changes: 9 additions & 11 deletions graph/db/benchmark_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -347,8 +347,7 @@ func TestPopulateDBs(t *testing.T) {
// graph.
countNodes := func(graph *ChannelGraph) int {
numNodes := 0
v1Graph := NewVersionedGraph(graph, lnwire.GossipVersion1)
err := v1Graph.ForEachNode(
err := graph.ForEachNode(

@bitromortac bitromortac Sep 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Whole-graph iteration moves from a single-table cursor scan to a join through the new mapping tables, and every node, shell node, channel and policy write now issues an extra upsert that multiplies during initial gossip sync. Could you post before and after BenchmarkGraphReadMethods numbers for ForEachNode and ForEachChannel?

ctx,
func(node *models.Node) error {
numNodes++
Expand All @@ -371,7 +370,7 @@ func TestPopulateDBs(t *testing.T) {
numPolicies = 0
)
err := graph.ForEachChannel(
ctx, lnwire.GossipVersion1,
ctx,
func(info *models.ChannelEdgeInfo, policy,
policy2 *models.ChannelEdgePolicy) error {

Expand Down Expand Up @@ -499,7 +498,7 @@ func syncGraph(t *testing.T, src, dest *ChannelGraph) {
}

var wgChans sync.WaitGroup
err = src.ForEachChannel(ctx, lnwire.GossipVersion1,
err = src.ForEachChannel(ctx,
func(info *models.ChannelEdgeInfo,
policy1, policy2 *models.ChannelEdgePolicy) error {

Expand Down Expand Up @@ -623,7 +622,7 @@ func BenchmarkGraphReadMethods(b *testing.B) {
name: "ForEachNode",
fn: func(b testing.TB, store Store) {
err := store.ForEachNode(
ctx, lnwire.GossipVersion1,
ctx,
func(_ *models.Node) error {
// Increment the counter to
// ensure the callback is doing
Expand All @@ -639,12 +638,11 @@ func BenchmarkGraphReadMethods(b *testing.B) {
{
name: "ForEachChannel",
fn: func(b testing.TB, store Store) {
//nolint:ll
err := store.ForEachChannel(
ctx, lnwire.GossipVersion1,
err := store.ForEachChannel(ctx,
func(_ *models.ChannelEdgeInfo,
_,
_ *models.ChannelEdgePolicy,
_ *models.ChannelEdgePolicy) error {
) error {

// Increment the counter to
// ensure the callback is doing
Expand Down Expand Up @@ -994,7 +992,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
)

err := store.ForEachNode(
ctx, lnwire.GossipVersion1,
ctx,
func(_ *models.Node) error {
numNodes++

Expand All @@ -1005,7 +1003,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {

//nolint:ll
err = store.ForEachChannel(
ctx, lnwire.GossipVersion1,
ctx,
func(_ *models.ChannelEdgeInfo,
_,
_ *models.ChannelEdgePolicy) error {
Expand Down
140 changes: 87 additions & 53 deletions graph/db/graph.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,8 @@ func (c *ChannelGraph) Start() error {
log.Criticalf("Could not populate the "+
"graph cache: %v", err)
}

c.checkPreferredMappings(ctx)
}()
} else {
if err := c.populateCache(ctx); err != nil {
Expand All @@ -138,9 +140,37 @@ func (c *ChannelGraph) Start() error {
c.wg.Add(1)
go c.handleTopologySubscriptions(ctx)

// The synchronous cache path has finished its full graph read. Run the
// detection-only integrity check in the background so it doesn't delay
// startup. The asynchronous path runs it after cache population above.
if !c.opts.asyncGraphCachePopulation {
c.wg.Add(1)
go func() {
defer c.wg.Done()

c.checkPreferredMappings(ctx)
}()
}

return nil
}

// checkPreferredMappings asks stores with cross-version preferred mappings to
// verify their derived indexes. The check only logs failures and divergence.
func (c *ChannelGraph) checkPreferredMappings(ctx context.Context) {
checker, ok := c.db.(interface {
checkPreferredMappings(context.Context) error
})
if !ok {
return
}

err := checker.checkPreferredMappings(ctx)
if err != nil && !errors.Is(err, context.Canceled) {
log.ErrorS(ctx, "Unable to check preferred graph mappings", err)
}
}

// Stop signals any active goroutines for a graceful closure.
func (c *ChannelGraph) Stop() error {
if !c.stopped.CompareAndSwap(false, true) {
Expand Down Expand Up @@ -214,6 +244,18 @@ func (c *ChannelGraph) handleTopologySubscriptions(ctx context.Context) {
}
}

// cacheableFeatures reports whether a node's feature vector may overwrite the
// vector the graph cache already holds for that node. The cache keys features
// by pub key alone, so an empty v2 vector would otherwise shadow a non-empty
// v1 one. This is the same rule the no-cache FetchNodeFeatures fallback
// applies, and every path that feeds the cache must apply it, otherwise the
// cache disagrees with itself before and after a restart.
func cacheableFeatures(v lnwire.GossipVersion,
features *lnwire.FeatureVector) bool {

return v != gossipV2 || !features.IsEmpty()
}

// populateCache loads the entire channel graph into the in-memory graph cache.
func (c *ChannelGraph) populateCache(ctx context.Context) error {
if c.cache == nil {
Expand All @@ -238,12 +280,21 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error {
for _, v := range []lnwire.GossipVersion{
gossipV1, gossipV2,
} {
// TODO(elle): If we have both v1 and v2 entries for the same
// node/channel, prefer v2 when merging.
// We iterate v1 first, then v2. AddNodeFeatures overwrites on
// key collision, so v2 features take precedence when both
// versions exist, except for the entries that
// cacheableFeatures rejects. AddChannel ranks a colliding SCID
// itself, using the same rule as the preferred lookup tables,
// so the iteration order here does not decide which version a
// channel is cached from.
err := c.db.ForEachNodeCacheable(ctx, v,
func(node route.Vertex,
features *lnwire.FeatureVector) error {

if !cacheableFeatures(v, features) {
return nil
}

cache.AddNodeFeatures(node, features)

return nil
Expand Down Expand Up @@ -299,9 +350,8 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(ctx context.Context,
return c.cache.graphCache.ForEachChannel(node, cb)
}

// TODO(elle): once the no-cache path needs to support
// pathfinding across gossip versions, this should iterate
// across all versions rather than defaulting to v1.
// The no-cache path only runs against the KV backend, which is
// v1-only.
return c.db.ForEachNodeDirectedChannel(
ctx, gossipV1, node, cb, reset,
)
Expand All @@ -320,13 +370,16 @@ func (c *ChannelGraph) FetchNodeFeatures(ctx context.Context,
return c.cache.graphCache.GetFeatures(node), nil
}

return c.db.FetchNodeFeatures(ctx, lnwire.GossipVersion1, node)
// The no-cache path only runs against the KV backend, which is

@bitromortac bitromortac Sep 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: graph/db: simplify no-cache fallback paths to v1-only removes the v2-then-v1 fallback loop that graph/db: implement cross-version node traversal adds, so the loop only ever exists between those two commits. Could you squash them so the dead loop never enters history?

// v1-only.
return c.db.FetchNodeFeatures(ctx, gossipV1, node)
}

// GraphSession will provide the call-back with access to a NodeTraverser
// instance which can be used to perform queries against the channel graph. If
// the graph cache is not enabled, then the call-back will be provided with
// access to the graph via a consistent read-only transaction.
// access to the graph via a consistent read-only transaction; the no-cache
// path only runs against the KV backend, which is v1-only.
func (c *ChannelGraph) GraphSession(ctx context.Context,
cb func(graph NodeTraverser) error, reset func()) error {

Expand Down Expand Up @@ -373,7 +426,7 @@ func (c *ChannelGraph) AddNode(ctx context.Context,
return err
}

if c.cache != nil {
if c.cache != nil && cacheableFeatures(node.Version, node.Features) {
c.cache.applyUpdate(func(cache *GraphCache) {
cache.AddNodeFeatures(
node.PubKeyBytes, node.Features,
Expand Down Expand Up @@ -679,13 +732,22 @@ func (c *ChannelGraph) HasV1Node(ctx context.Context,
return c.db.HasV1Node(ctx, nodePub)
}

// ForEachChannel iterates through all channel edges stored within the graph.
// ForEachChannel iterates through all channel edges stored within the graph
// across all gossip versions.
func (c *ChannelGraph) ForEachChannel(ctx context.Context,
v lnwire.GossipVersion, cb func(*models.ChannelEdgeInfo,
*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
*models.ChannelEdgePolicy) error,
reset func()) error {

return c.db.ForEachChannel(ctx, v, cb, reset)
return c.db.ForEachChannel(ctx, cb, reset)

@bitromortac bitromortac Sep 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Promoting these off VersionedGraph means the embedded *ChannelGraph methods now serve callers holding a version-pinned graph, so the pin is ignored with no signal at the call site. autopilot and the discovery bootstrapper both receive v1Graph and will iterate every gossip version once v2 data exists. From the docstrings I see it's intentional, but maybe we can find a better solution?

}

// ForEachNode iterates through all stored vertices/nodes in the graph across
// all gossip versions.
func (c *ChannelGraph) ForEachNode(ctx context.Context,
cb func(*models.Node) error, reset func()) error {

return c.db.ForEachNode(ctx, cb, reset)
}

// DisabledChannelIDs returns the channel ids of disabled channels.
Expand Down Expand Up @@ -817,26 +879,23 @@ func (c *ChannelGraph) FetchChanInfos(ctx context.Context,
}

// FetchChannelEdgesByOutpoint attempts to lookup directed edges by funding
// outpoint.
// outpoint, returning the highest available gossip version.
func (c *ChannelGraph) FetchChannelEdgesByOutpoint(ctx context.Context,
op *wire.OutPoint) (
*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
*models.ChannelEdgePolicy, error) {

return c.db.FetchChannelEdgesByOutpoint(
ctx, lnwire.GossipVersion1, op,
)
return c.db.FetchChannelEdgesByOutpointPreferred(ctx, op)
}

// FetchChannelEdgesByID attempts to lookup directed edges by channel ID.
// FetchChannelEdgesByID attempts to lookup directed edges by channel ID,

@bitromortac bitromortac Sep 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This fetch now spans gossip versions, but deleteAliasEdge in server.go still calls a v1-scoped delete with the SCID it returns. A v2 alias edge is therefore reported to the funding manager as deleted while its row stays in the graph. server.go sits outside this diff, which is what makes it easy to miss. Could you delete the version the fetch resolved to, using the Version on the returned ChannelEdgeInfo?

// returning the highest available gossip version.
func (c *ChannelGraph) FetchChannelEdgesByID(ctx context.Context,
chanID uint64) (
*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
*models.ChannelEdgePolicy, error) {

return c.db.FetchChannelEdgesByID(
ctx, lnwire.GossipVersion1, chanID,
)
return c.db.FetchChannelEdgesByIDPreferred(ctx, chanID)
}

// PutClosedScid stores a SCID for a closed channel in the database.
Expand Down Expand Up @@ -869,6 +928,14 @@ func (c *ChannelGraph) PruneTip(ctx context.Context) (*chainhash.Hash,

// VersionedGraph is a wrapper around ChannelGraph that will call underlying
// Store methods with a specific gossip version.
//
// NOTE: ForEachNode and ForEachChannel are the exception. They are promoted
// from the embedded ChannelGraph and iterate every gossip version, yielding
// the preferred record per pub key or SCID. Both callers that reach them
// through this wrapper want exactly that: autopilot ranks every candidate node
// for channel opening, and the discovery bootstrapper dials every known
// address, so neither has a reason to ignore a v2 announcement. A caller that
// does need one version must use a version-taking Store method instead.
type VersionedGraph struct {
*ChannelGraph
v lnwire.GossipVersion
Expand Down Expand Up @@ -926,13 +993,6 @@ func (c *VersionedGraph) ForEachNodeCached(ctx context.Context,
return c.ChannelGraph.ForEachNodeCached(ctx, c.v, cb, reset)
}

// ForEachNode iterates through all stored vertices/nodes in the graph.
func (c *VersionedGraph) ForEachNode(ctx context.Context,
cb func(*models.Node) error, reset func()) error {

return c.db.ForEachNode(ctx, c.v, cb, reset)
}

// NumZombies returns the current number of zombie channels in the graph.
func (c *VersionedGraph) NumZombies(ctx context.Context) (uint64, error) {
return c.db.NumZombies(ctx, c.v)
Expand Down Expand Up @@ -964,24 +1024,6 @@ func (c *VersionedGraph) ChannelView(ctx context.Context) ([]EdgePoint,
return c.db.ChannelView(ctx, c.v)
}

// GraphSession provides the callback with access to a NodeTraverser instance
// for performing queries against the channel graph. If the graph cache is
// enabled, the callback receives the VersionedGraph directly (which implements
// NodeTraverser using the cache). Otherwise a read-only database session is
// used.
func (c *VersionedGraph) GraphSession(ctx context.Context,
cb func(graph NodeTraverser) error, reset func()) error {

if c.cache != nil && c.cache.isLoaded() {
return cb(c)
}

// TODO(elle): the underlying GraphSession currently creates a
// NodeTraverser that is hardcoded to GossipVersion1. This needs to be
// updated to pass the version through for v2 support.
return c.db.GraphSession(ctx, cb, reset)
}

// FetchNode attempts to look up a target node by its identity public key.
func (c *VersionedGraph) FetchNode(ctx context.Context,
nodePub route.Vertex) (*models.Node, error) {
Expand Down Expand Up @@ -1106,14 +1148,6 @@ func (c *VersionedGraph) ForEachNodeChannel(ctx context.Context,
return c.db.ForEachNodeChannel(ctx, c.v, nodePub, cb, reset)
}

// 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)
}

// ForEachNodeCacheable iterates through all stored vertices/nodes in the graph.
func (c *VersionedGraph) ForEachNodeCacheable(ctx context.Context,
cb func(route.Vertex, *lnwire.FeatureVector) error,
Expand Down
Loading
Loading