From 23d4bfc1b072dda75f366013638a2baf1b766413 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 18 Jun 2025 15:17:55 +0200 Subject: [PATCH 1/6] graph/db: add comment and switch to length check Use a length check to determine if a bitcoin signature has been set or not. Also add a clarifying comment to explain why we only need to check if one signature field is set to determine if we have the auth proof for the channel or not. --- graph/db/sql_store.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 137c1d2e2..e8cb36dfd 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1702,7 +1702,10 @@ func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries, ExtraOpaqueData: recs, } - if dbChan.Bitcoin1Signature != nil { + // We always set all the signatures at the same time, so we can + // safely check if one signature is present to determine if we have the + // rest of the signatures for the auth proof. + if len(dbChan.Bitcoin1Signature) > 0 { channel.AuthProof = &models.ChannelAuthProof{ NodeSig1Bytes: dbChan.Node1Signature, NodeSig2Bytes: dbChan.Node2Signature, From 973d211083a79c69e21e010582219a30f7795128 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 15:52:23 +0200 Subject: [PATCH 2/6] graph/db: fix DecodeHexColor Ensure that it does a sanity check on the length of its input and also only call it if the nullable SQL string is not null. Expand an existing unit tests to cover the DecodeHexColor such that it would have caught the bug. --- graph/db/notifications.go | 5 +++++ graph/db/sql_store.go | 9 ++++++--- graph/notifications_test.go | 36 ++++++++++++++++++++++++++++-------- 3 files changed, 39 insertions(+), 11 deletions(-) diff --git a/graph/db/notifications.go b/graph/db/notifications.go index 6c0acba88..4af953381 100644 --- a/graph/db/notifications.go +++ b/graph/db/notifications.go @@ -473,6 +473,11 @@ func EncodeHexColor(color color.RGBA) string { // DecodeHexColor takes a hex color string like "#rrggbb" and returns a // color.RGBA. func DecodeHexColor(hex string) (color.RGBA, error) { + if len(hex) != 7 || hex[0] != '#' { + return color.RGBA{}, fmt.Errorf("invalid hex color string: %s", + hex) + } + r, err := strconv.ParseUint(hex[1:3], 16, 8) if err != nil { return color.RGBA{}, fmt.Errorf("invalid red component: %w", diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index e8cb36dfd..ca8569aee 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -934,9 +934,12 @@ func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.Node) ( node.LastUpdate = time.Unix(dbNode.LastUpdate.Int64, 0) var err error - node.Color, err = DecodeHexColor(dbNode.Color.String) - if err != nil { - return nil, fmt.Errorf("unable to decode color: %w", err) + if dbNode.Color.Valid { + node.Color, err = DecodeHexColor(dbNode.Color.String) + if err != nil { + return nil, fmt.Errorf("unable to decode color: %w", + err) + } } // Fetch the node's features. diff --git a/graph/notifications_test.go b/graph/notifications_test.go index 099c2244b..c9096efc1 100644 --- a/graph/notifications_test.go +++ b/graph/notifications_test.go @@ -991,7 +991,9 @@ func TestChannelCloseNotification(t *testing.T) { // TestEncodeHexColor tests that the string used to represent a node color is // correctly encoded. func TestEncodeHexColor(t *testing.T) { - var colorTestCases = []struct { + t.Parallel() + + var tests = []struct { R uint8 G uint8 B uint8 @@ -1006,14 +1008,32 @@ func TestEncodeHexColor(t *testing.T) { {1, 2, 3, "#", false}, } - for _, tc := range colorTestCases { - encoded := graphdb.EncodeHexColor( - color.RGBA{tc.R, tc.G, tc.B, 0}, + for _, test := range tests { + t.Run(fmt.Sprintf("R:%d,G:%d,B:%d", test.R, test.G, test.B), + func(t *testing.T) { + expColor := color.RGBA{ + R: test.R, + G: test.G, + B: test.B, + } + + encoded := graphdb.EncodeHexColor(expColor) + + if test.isValid { + require.Equal(t, test.encoded, encoded) + } + + decoded, err := graphdb.DecodeHexColor( + test.encoded, + ) + if test.isValid { + require.NoError(t, err) + require.Equal(t, expColor, decoded) + } else { + require.Error(t, err) + } + }, ) - if (encoded == tc.encoded) != tc.isValid { - t.Fatalf("incorrect color encoding, "+ - "want: %v, got: %v", tc.encoded, encoded) - } } } From d60761f79c480ceeadc6ba15c0285d31283468ca Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 16:01:54 +0200 Subject: [PATCH 3/6] graph/db: fix error check in test An error in TestChanUpdatesInHorizon was previously not checked correctly. --- graph/db/graph_test.go | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 8dab86611..b54b1fa4c 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -2069,16 +2069,15 @@ func TestChanUpdatesInHorizon(t *testing.T) { assertEdgeInfoEqual(t, chanExp.Info, chanRet.Info) - err := compareEdgePolicies( + err = compareEdgePolicies( chanExp.Policy1, chanRet.Policy1, ) - if err != nil { - t.Fatal(err) - } - compareEdgePolicies(chanExp.Policy2, chanRet.Policy2) - if err != nil { - t.Fatal(err) - } + require.NoError(t, err) + + err = compareEdgePolicies( + chanExp.Policy2, chanRet.Policy2, + ) + require.NoError(t, err) } } } From 8af32951c734bc05aa78a0613f9bdf20d53a6b2f Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 15:59:43 +0200 Subject: [PATCH 4/6] graph/db+sqldb: implement ForEachNode In this commit the `ForEachNode` method is added to the SQLStore. With this, the `TestGraphCacheTraversal` unit test can be run against SQL backends. --- graph/db/graph_test.go | 2 +- graph/db/sql_store.go | 129 +++++++++++++++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 46 ++++++++++++- sqldb/sqlc/querier.go | 1 + sqldb/sqlc/queries/graph.sql | 8 ++- 5 files changed, 183 insertions(+), 3 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index b54b1fa4c..d48e229f3 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1436,7 +1436,7 @@ func TestGraphTraversalCacheable(t *testing.T) { func TestGraphCacheTraversal(t *testing.T) { t.Parallel() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) // We'd like to test some of the graph traversal capabilities within // the DB, so we'll create a series of fake nodes to insert into the diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index ca8569aee..893575154 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -28,6 +28,10 @@ import ( "github.com/lightningnetwork/lnd/tor" ) +// pageSize is the limit for the number of records that can be returned +// in a paginated query. This can be tuned after some benchmarks. +const pageSize = 2000 + // ProtocolVersion is an enum that defines the gossip protocol version of a // message. type ProtocolVersion uint8 @@ -53,6 +57,7 @@ type SQLQueries interface { UpsertNode(ctx context.Context, arg sqlc.UpsertNodeParams) (int64, error) GetNodeByPubKey(ctx context.Context, arg sqlc.GetNodeByPubKeyParams) (sqlc.Node, error) GetNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetNodesByLastUpdateRangeParams) ([]sqlc.Node, error) + ListNodesPaginated(ctx context.Context, arg sqlc.ListNodesPaginatedParams) ([]sqlc.Node, error) DeleteNodeByPubKey(ctx context.Context, arg sqlc.DeleteNodeByPubKeyParams) (sql.Result, error) GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.NodeExtraType, error) @@ -715,6 +720,130 @@ func (s *SQLStore) ForEachSourceNodeChannel(cb func(chanPoint wire.OutPoint, }, sqldb.NoOpReset) } +// ForEachNode iterates through all the stored vertices/nodes in the graph, +// executing the passed callback with each node encountered. If the callback +// returns an error, then the transaction is aborted and the iteration stops +// early. Any operations performed on the NodeTx passed to the call-back are +// executed under the same read transaction and so, methods on the NodeTx object +// _MUST_ only be called from within the call-back. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) ForEachNode(cb func(tx NodeRTx) error) error { + var ( + ctx = context.TODO() + lastID int64 = 0 + ) + + handleNode := func(db SQLQueries, dbNode sqlc.Node) error { + node, err := buildNode(ctx, db, &dbNode) + if err != nil { + return fmt.Errorf("unable to build node(id=%d): %w", + dbNode.ID, err) + } + + err = cb( + newSQLGraphNodeTx(db, s.cfg.ChainHash, dbNode.ID, node), + ) + if err != nil { + return fmt.Errorf("callback failed for node(id=%d): %w", + dbNode.ID, err) + } + + return nil + } + + return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + for { + nodes, err := db.ListNodesPaginated( + ctx, sqlc.ListNodesPaginatedParams{ + Version: int16(ProtocolV1), + ID: lastID, + Limit: pageSize, + }, + ) + if err != nil { + return fmt.Errorf("unable to fetch nodes: %w", + err) + } + + if len(nodes) == 0 { + break + } + + for _, dbNode := range nodes { + err = handleNode(db, dbNode) + if err != nil { + return err + } + + lastID = dbNode.ID + } + } + + return nil + }, sqldb.NoOpReset) +} + +// sqlGraphNodeTx is an implementation of the NodeRTx interface backed by the +// SQLStore and a SQL transaction. +type sqlGraphNodeTx struct { + db SQLQueries + id int64 + node *models.LightningNode + chain chainhash.Hash +} + +// A compile-time constraint to ensure sqlGraphNodeTx implements the NodeRTx +// interface. +var _ NodeRTx = (*sqlGraphNodeTx)(nil) + +func newSQLGraphNodeTx(db SQLQueries, chain chainhash.Hash, + id int64, node *models.LightningNode) *sqlGraphNodeTx { + + return &sqlGraphNodeTx{ + db: db, + chain: chain, + id: id, + node: node, + } +} + +// Node returns the raw information of the node. +// +// NOTE: This is a part of the NodeRTx interface. +func (s *sqlGraphNodeTx) Node() *models.LightningNode { + return s.node +} + +// ForEachChannel can be used to iterate over the node's channels under the same +// transaction used to fetch the node. +// +// NOTE: This is a part of the NodeRTx interface. +func (s *sqlGraphNodeTx) ForEachChannel(cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { + + ctx := context.TODO() + + return forEachNodeChannel(ctx, s.db, s.chain, s.id, cb) +} + +// FetchNode fetches the node with the given pub key under the same transaction +// used to fetch the current node. The returned node is also a NodeRTx and any +// operations on that NodeRTx will also be done under the same transaction. +// +// NOTE: This is a part of the NodeRTx interface. +func (s *sqlGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) { + ctx := context.TODO() + + id, node, err := getNodeByPubKey(ctx, s.db, nodePub) + if err != nil { + return nil, fmt.Errorf("unable to fetch V1 node(%x): %w", + nodePub, err) + } + + return newSQLGraphNodeTx(s.db, s.chain, id, node), nil +} + // forEachNodeChannel iterates through all channels of a node, executing // the passed callback on each. The call-back is provided with the channel's // edge information, the outgoing policy and the incoming policy for the diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 884d5c340..5b7f974a2 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -729,7 +729,6 @@ func (q *Queries) InsertNodeFeature(ctx context.Context, arg InsertNodeFeaturePa } const listChannelsByNodeID = `-- name: ListChannelsByNodeID :many - SELECT c.id, c.version, c.scid, c.node_id_1, c.node_id_2, c.outpoint, c.capacity, c.bitcoin_key_1, c.bitcoin_key_2, c.node_1_signature, c.node_2_signature, c.bitcoin_1_signature, c.bitcoin_2_signature, n1.pub_key AS node1_pubkey, n2.pub_key AS node2_pubkey, @@ -880,6 +879,51 @@ func (q *Queries) ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNo return items, nil } +const listNodesPaginated = `-- name: ListNodesPaginated :many +SELECT id, version, pub_key, alias, last_update, color, signature +FROM nodes +WHERE version = $1 AND id > $2 +ORDER BY id +LIMIT $3 +` + +type ListNodesPaginatedParams struct { + Version int16 + ID int64 + Limit int32 +} + +func (q *Queries) ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]Node, error) { + rows, err := q.db.QueryContext(ctx, listNodesPaginated, arg.Version, arg.ID, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Node + for rows.Next() { + var i Node + if err := rows.Scan( + &i.ID, + &i.Version, + &i.PubKey, + &i.Alias, + &i.LastUpdate, + &i.Color, + &i.Signature, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const upsertEdgePolicy = `-- name: UpsertEdgePolicy :one /* ───────────────────────────────────────────── channel_policies table queries diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index d4a6a9dd0..35d7b2829 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -64,6 +64,7 @@ type Querier interface { InsertNodeAddress(ctx context.Context, arg InsertNodeAddressParams) error InsertNodeFeature(ctx context.Context, arg InsertNodeFeatureParams) error ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error) + ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]Node, error) NextInvoiceSettleIndex(ctx context.Context) (int64, error) OnAMPSubInvoiceCanceled(ctx context.Context, arg OnAMPSubInvoiceCanceledParams) error OnAMPSubInvoiceCreated(ctx context.Context, arg OnAMPSubInvoiceCreatedParams) error diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index c6dbb6e6b..7c98cc60e 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -27,6 +27,13 @@ FROM nodes WHERE pub_key = $1 AND version = $2; +-- name: ListNodesPaginated :many +SELECT * +FROM nodes +WHERE version = $1 AND id > $2 +ORDER BY id +LIMIT $3; + -- name: DeleteNodeByPubKey :execresult DELETE FROM nodes WHERE pub_key = $1 @@ -194,7 +201,6 @@ ORDER BY scid DESC LIMIT 1; -- name: ListChannelsByNodeID :many - SELECT sqlc.embed(c), n1.pub_key AS node1_pubkey, n2.pub_key AS node2_pubkey, From 06079828865a362dbefeda9508ef69b6d1ad4f27 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 16:12:08 +0200 Subject: [PATCH 5/6] graph/db: implement ForEachNodeDirectedChannel and ForEachNodeCacheable Here we add the `ForEachNodeDirectedChannel` and `ForEachNodeCacheable` SQLStore implementations which then lets us run `TestGraphTraversalCacheable` and `TestGraphCacheForEachNodeChannel` against SQL backends. --- graph/db/graph_test.go | 21 ++-- graph/db/sql_store.go | 236 ++++++++++++++++++++++++++++++++++- sqldb/sqlc/graph.sql.go | 61 +++++++++ sqldb/sqlc/querier.go | 2 + sqldb/sqlc/queries/graph.sql | 13 ++ 5 files changed, 324 insertions(+), 9 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index d48e229f3..f681a1f75 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -23,6 +23,7 @@ import ( "github.com/btcsuite/btcd/chaincfg" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/fn/v2" "github.com/lightningnetwork/lnd/graph/db/models" "github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/lnwire" @@ -1366,7 +1367,7 @@ func TestGraphTraversal(t *testing.T) { func TestGraphTraversalCacheable(t *testing.T) { t.Parallel() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) // We'd like to test some of the graph traversal capabilities within // the DB, so we'll create a series of fake nodes to insert into the @@ -4162,7 +4163,7 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { t.Parallel() ctx := context.Background() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) // Unset the channel graph cache to simulate the user running with the // option turned off. @@ -4212,21 +4213,25 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { edge1.ExtraOpaqueData = []byte{ 253, 217, 3, 8, 0, 0, 0, 10, 0, 0, 0, 20, } + inboundFee := lnwire.Fee{ + BaseFee: 10, + FeeRate: 20, + } + edge1.InboundFee = fn.Some(inboundFee) require.NoError(t, graph.UpdateEdgePolicy(edge1)) edge1 = copyEdgePolicy(edge1) // Avoid read/write race conditions. directedChan := getSingleChannel() require.NotNil(t, directedChan) - expectedInbound := lnwire.Fee{ - BaseFee: 10, - FeeRate: 20, - } - require.Equal(t, expectedInbound, directedChan.InboundFee) + require.Equal(t, inboundFee, directedChan.InboundFee) // Set an invalid inbound fee and check that persistence fails. edge1.ExtraOpaqueData = []byte{ 253, 217, 3, 8, 0, } + // We need to update the timestamp so that we don't hit the DB conflict + // error when we try to update the edge policy. + edge1.LastUpdate = edge1.LastUpdate.Add(time.Second) require.ErrorIs( t, graph.UpdateEdgePolicy(edge1), ErrParsingExtraTLVBytes, ) @@ -4235,7 +4240,7 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { // the previous result when we query the channel again. directedChan = getSingleChannel() require.NotNil(t, directedChan) - require.Equal(t, expectedInbound, directedChan.InboundFee) + require.Equal(t, inboundFee, directedChan.InboundFee) } // TestGraphLoading asserts that the cache is properly reconstructed after a diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 893575154..e7bb18904 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -56,8 +56,10 @@ type SQLQueries interface { */ UpsertNode(ctx context.Context, arg sqlc.UpsertNodeParams) (int64, error) GetNodeByPubKey(ctx context.Context, arg sqlc.GetNodeByPubKeyParams) (sqlc.Node, error) + GetNodeIDByPubKey(ctx context.Context, arg sqlc.GetNodeIDByPubKeyParams) (int64, error) GetNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetNodesByLastUpdateRangeParams) ([]sqlc.Node, error) ListNodesPaginated(ctx context.Context, arg sqlc.ListNodesPaginatedParams) ([]sqlc.Node, error) + ListNodeIDsAndPubKeys(ctx context.Context, arg sqlc.ListNodeIDsAndPubKeysParams) ([]sqlc.ListNodeIDsAndPubKeysRow, error) DeleteNodeByPubKey(ctx context.Context, arg sqlc.DeleteNodeByPubKeyParams) (sql.Result, error) GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.NodeExtraType, error) @@ -844,6 +846,224 @@ func (s *sqlGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) { return newSQLGraphNodeTx(s.db, s.chain, id, node), nil } +// ForEachNodeDirectedChannel iterates through all channels of a given node, +// executing the passed callback on the directed edge representing the channel +// and its incoming policy. If the callback returns an error, then the iteration +// is halted with the error propagated back up to the caller. +// +// Unknown policies are passed into the callback as nil values. +// +// NOTE: this is part of the graphdb.NodeTraverser interface. +func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, + cb func(channel *DirectedChannel) error) error { + + var ctx = context.TODO() + + return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + return forEachNodeDirectedChannel(ctx, db, nodePub, cb) + }, sqldb.NoOpReset) +} + +// ForEachNodeCacheable iterates through all the stored vertices/nodes in the +// graph, executing the passed callback with each node encountered. If the +// callback returns an error, then the transaction is aborted and the iteration +// stops early. +// +// NOTE: This is a part of the V1Store interface. +func (s *SQLStore) ForEachNodeCacheable(cb func(route.Vertex, + *lnwire.FeatureVector) error) error { + + ctx := context.TODO() + + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + return forEachNodeCacheable(ctx, db, func(nodeID int64, + nodePub route.Vertex) error { + + features, err := getNodeFeatures(ctx, db, nodeID) + if err != nil { + return fmt.Errorf("unable to fetch node "+ + "features: %w", err) + } + + return cb(nodePub, features) + }) + }, sqldb.NoOpReset) + if err != nil { + return fmt.Errorf("unable to fetch nodes: %w", err) + } + + return nil +} + +// forEachNodeDirectedChannel iterates through all channels of a given +// node, executing the passed callback on the directed edge representing the +// channel and its incoming policy. If the node is not found, no error is +// returned. +func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, + nodePub route.Vertex, cb func(channel *DirectedChannel) error) error { + + toNodeCallback := func() route.Vertex { + return nodePub + } + + dbID, err := db.GetNodeIDByPubKey( + ctx, sqlc.GetNodeIDByPubKeyParams{ + Version: int16(ProtocolV1), + PubKey: nodePub[:], + }, + ) + if errors.Is(err, sql.ErrNoRows) { + return nil + } else if err != nil { + return fmt.Errorf("unable to fetch node: %w", err) + } + + rows, err := db.ListChannelsByNodeID( + ctx, sqlc.ListChannelsByNodeIDParams{ + Version: int16(ProtocolV1), + NodeID1: dbID, + }, + ) + if err != nil { + return fmt.Errorf("unable to fetch channels: %w", err) + } + + // Exit early if there are no channels for this node so we don't + // do the unnecessary feature fetching. + if len(rows) == 0 { + return nil + } + + features, err := getNodeFeatures(ctx, db, dbID) + if err != nil { + return fmt.Errorf("unable to fetch node features: %w", err) + } + + for _, row := range rows { + node1, node2, err := buildNodeVertices( + row.Node1Pubkey, row.Node2Pubkey, + ) + if err != nil { + return fmt.Errorf("unable to build node vertices: %w", + err) + } + + edge, err := buildCacheableChannelInfo( + row.Channel, node1, node2, + ) + if err != nil { + return err + } + + dbPol1, dbPol2, err := extractChannelPolicies(row) + if err != nil { + return err + } + + var p1, p2 *models.CachedEdgePolicy + if dbPol1 != nil { + policy1, err := buildChanPolicy( + *dbPol1, edge.ChannelID, nil, node2, true, + ) + if err != nil { + return err + } + + p1 = models.NewCachedPolicy(policy1) + } + if dbPol2 != nil { + policy2, err := buildChanPolicy( + *dbPol2, edge.ChannelID, nil, node1, false, + ) + if err != nil { + return err + } + + p2 = models.NewCachedPolicy(policy2) + } + + // Determine the outgoing and incoming policy for this + // channel and node combo. + outPolicy, inPolicy := p1, p2 + if p1 != nil && node2 == nodePub { + outPolicy, inPolicy = p2, p1 + } else if p2 != nil && node1 != nodePub { + outPolicy, inPolicy = p2, p1 + } + + var cachedInPolicy *models.CachedEdgePolicy + if inPolicy != nil { + cachedInPolicy = inPolicy + cachedInPolicy.ToNodePubKey = toNodeCallback + cachedInPolicy.ToNodeFeatures = features + } + + directedChannel := &DirectedChannel{ + ChannelID: edge.ChannelID, + IsNode1: nodePub == edge.NodeKey1Bytes, + OtherNode: edge.NodeKey2Bytes, + Capacity: edge.Capacity, + OutPolicySet: outPolicy != nil, + InPolicy: cachedInPolicy, + } + if outPolicy != nil { + outPolicy.InboundFee.WhenSome(func(fee lnwire.Fee) { + directedChannel.InboundFee = fee + }) + } + + if nodePub == edge.NodeKey2Bytes { + directedChannel.OtherNode = edge.NodeKey1Bytes + } + + if err := cb(directedChannel); err != nil { + return err + } + } + + return nil +} + +// forEachNodeCacheable fetches all V1 node IDs and pub keys from the database, +// and executes the provided callback for each node. +func forEachNodeCacheable(ctx context.Context, db SQLQueries, + cb func(nodeID int64, nodePub route.Vertex) error) error { + + var lastID int64 + + for { + nodes, err := db.ListNodeIDsAndPubKeys( + ctx, sqlc.ListNodeIDsAndPubKeysParams{ + Version: int16(ProtocolV1), + ID: lastID, + Limit: pageSize, + }, + ) + if err != nil { + return fmt.Errorf("unable to fetch nodes: %w", err) + } + + if len(nodes) == 0 { + break + } + + for _, node := range nodes { + var pub route.Vertex + copy(pub[:], node.PubKey) + + if err := cb(node.ID, pub); err != nil { + return fmt.Errorf("forEachNodeCacheable "+ + "callback failed for node(id=%d): %w", + node.ID, err) + } + + lastID = node.ID + } + } + + return nil +} + // forEachNodeChannel iterates through all channels of a node, executing // the passed callback on each. The call-back is provided with the channel's // edge information, the outgoing policy and the incoming policy for the @@ -1033,6 +1253,20 @@ func getNodeByPubKey(ctx context.Context, db SQLQueries, return dbNode.ID, node, nil } +// buildCacheableChannelInfo builds a models.CachedEdgeInfo instance from the +// provided database channel row and the public keys of the two nodes +// involved in the channel. +func buildCacheableChannelInfo(dbChan sqlc.Channel, node1Pub, + node2Pub route.Vertex) (*models.CachedEdgeInfo, error) { + + return &models.CachedEdgeInfo{ + ChannelID: byteOrder.Uint64(dbChan.Scid), + NodeKey1Bytes: node1Pub, + NodeKey2Bytes: node2Pub, + Capacity: btcutil.Amount(dbChan.Capacity.Int64), + }, nil +} + // buildNode constructs a LightningNode instance from the given database node // record. The node's features, addresses and extra signed fields are also // fetched from the database and set on the node. @@ -1589,7 +1823,7 @@ func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) { // pass it into the P2P decoding variant. parsedTypes, err := tlvStream.DecodeWithParsedTypesP2P(r) if err != nil { - return nil, err + return nil, fmt.Errorf("%w: %w", ErrParsingExtraTLVBytes, err) } if len(parsedTypes) == 0 { return nil, nil diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 5b7f974a2..606d40217 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -533,6 +533,25 @@ func (q *Queries) GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeatur return items, nil } +const getNodeIDByPubKey = `-- name: GetNodeIDByPubKey :one +SELECT id +FROM nodes +WHERE pub_key = $1 + AND version = $2 +` + +type GetNodeIDByPubKeyParams struct { + PubKey []byte + Version int16 +} + +func (q *Queries) GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error) { + row := q.db.QueryRowContext(ctx, getNodeIDByPubKey, arg.PubKey, arg.Version) + var id int64 + err := row.Scan(&id) + return id, err +} + const getNodesByLastUpdateRange = `-- name: GetNodesByLastUpdateRange :many SELECT id, version, pub_key, alias, last_update, color, signature FROM nodes @@ -879,6 +898,48 @@ func (q *Queries) ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNo return items, nil } +const listNodeIDsAndPubKeys = `-- name: ListNodeIDsAndPubKeys :many +SELECT id, pub_key +FROM nodes +WHERE version = $1 AND id > $2 +ORDER BY id +LIMIT $3 +` + +type ListNodeIDsAndPubKeysParams struct { + Version int16 + ID int64 + Limit int32 +} + +type ListNodeIDsAndPubKeysRow struct { + ID int64 + PubKey []byte +} + +func (q *Queries) ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error) { + rows, err := q.db.QueryContext(ctx, listNodeIDsAndPubKeys, arg.Version, arg.ID, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListNodeIDsAndPubKeysRow + for rows.Next() { + var i ListNodeIDsAndPubKeysRow + if err := rows.Scan(&i.ID, &i.PubKey); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listNodesPaginated = `-- name: ListNodesPaginated :many SELECT id, version, pub_key, alias, last_update, color, signature FROM nodes diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 35d7b2829..c0abde3c2 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -48,6 +48,7 @@ type Querier interface { GetNodeByPubKey(ctx context.Context, arg GetNodeByPubKeyParams) (Node, error) GetNodeFeatures(ctx context.Context, nodeID int64) ([]NodeFeature, error) GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error) + GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]Node, error) GetSourceNodesByVersion(ctx context.Context, version int16) ([]GetSourceNodesByVersionRow, error) HighestSCID(ctx context.Context, version int16) ([]byte, error) @@ -64,6 +65,7 @@ type Querier interface { InsertNodeAddress(ctx context.Context, arg InsertNodeAddressParams) error InsertNodeFeature(ctx context.Context, arg InsertNodeFeatureParams) error ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error) + ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error) ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]Node, error) NextInvoiceSettleIndex(ctx context.Context) (int64, error) OnAMPSubInvoiceCanceled(ctx context.Context, arg OnAMPSubInvoiceCanceledParams) error diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 7c98cc60e..510195906 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -27,6 +27,12 @@ FROM nodes WHERE pub_key = $1 AND version = $2; +-- name: GetNodeIDByPubKey :one +SELECT id +FROM nodes +WHERE pub_key = $1 + AND version = $2; + -- name: ListNodesPaginated :many SELECT * FROM nodes @@ -34,6 +40,13 @@ WHERE version = $1 AND id > $2 ORDER BY id LIMIT $3; +-- name: ListNodeIDsAndPubKeys :many +SELECT id, pub_key +FROM nodes +WHERE version = $1 AND id > $2 +ORDER BY id +LIMIT $3; + -- name: DeleteNodeByPubKey :execresult DELETE FROM nodes WHERE pub_key = $1 From ece157b40b23147153a09d6cab4d982f8b084d4b Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 16:27:09 +0200 Subject: [PATCH 6/6] graph/db: implement ForEachNodeChannel Which lets us then run `TestIncompleteChannelPolicies` and `BenchmarkForEachChannel` against our SQL backends. --- docs/release-notes/release-notes-0.20.0.md | 1 + graph/db/graph_test.go | 4 +-- graph/db/sql_store.go | 35 ++++++++++++++++++++++ 3 files changed, 38 insertions(+), 2 deletions(-) diff --git a/docs/release-notes/release-notes-0.20.0.md b/docs/release-notes/release-notes-0.20.0.md index d5edcab17..496a20475 100644 --- a/docs/release-notes/release-notes-0.20.0.md +++ b/docs/release-notes/release-notes-0.20.0.md @@ -77,6 +77,7 @@ circuit. The indices are only available for forwarding events saved after v0.20. * [2](https://github.com/lightningnetwork/lnd/pull/9869) * [3](https://github.com/lightningnetwork/lnd/pull/9887) * [4](https://github.com/lightningnetwork/lnd/pull/9931) + * [5](https://github.com/lightningnetwork/lnd/pull/9935) ## RPC Updates diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index f681a1f75..789a4c0fb 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -3041,7 +3041,7 @@ func TestIncompleteChannelPolicies(t *testing.T) { t.Parallel() ctx := context.Background() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) // Create two nodes. node1 := createTestVertex(t) @@ -4110,7 +4110,7 @@ func TestBatchedUpdateEdgePolicy(t *testing.T) { // BenchmarkForEachChannel is a benchmark test that measures the number of // allocations and the total memory consumed by the full graph traversal. func BenchmarkForEachChannel(b *testing.B) { - graph := MakeTestGraph(b) + graph := MakeTestGraphNew(b) const numNodes = 100 const numChannels = 4 diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index e7bb18904..22512619c 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -895,6 +895,41 @@ func (s *SQLStore) ForEachNodeCacheable(cb func(route.Vertex, return nil } +// ForEachNodeChannel iterates through all channels of the given node, +// executing the passed callback with an edge info structure and the policies +// of each end of the channel. The first edge policy is the outgoing edge *to* +// the connecting node, while the second is the incoming edge *from* the +// connecting node. If the callback returns an error, then the iteration is +// halted with the error propagated back up to the caller. +// +// Unknown policies are passed into the callback as nil values. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) ForEachNodeChannel(nodePub route.Vertex, + cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy) error) error { + + var ctx = context.TODO() + + return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + dbNode, err := db.GetNodeByPubKey( + ctx, sqlc.GetNodeByPubKeyParams{ + Version: int16(ProtocolV1), + PubKey: nodePub[:], + }, + ) + if errors.Is(err, sql.ErrNoRows) { + return nil + } else if err != nil { + return fmt.Errorf("unable to fetch node: %w", err) + } + + return forEachNodeChannel( + ctx, db, s.cfg.ChainHash, dbNode.ID, cb, + ) + }, sqldb.NoOpReset) +} + // forEachNodeDirectedChannel iterates through all channels of a given // node, executing the passed callback on the directed edge representing the // channel and its incoming policy. If the node is not found, no error is