From 102c04daaf326f8a0a0811feba0bf02339cb7ade Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 17:32:52 +0200 Subject: [PATCH 1/4] sqldb/sqlc: prune_log schema Define the schema for the prune log. --- sqldb/sqlc/migrations/000007_graph.down.sql | 3 ++- sqldb/sqlc/migrations/000007_graph.up.sql | 12 ++++++++++++ sqldb/sqlc/models.go | 5 +++++ 3 files changed, 19 insertions(+), 1 deletion(-) diff --git a/sqldb/sqlc/migrations/000007_graph.down.sql b/sqldb/sqlc/migrations/000007_graph.down.sql index a2f519a2c..396471196 100644 --- a/sqldb/sqlc/migrations/000007_graph.down.sql +++ b/sqldb/sqlc/migrations/000007_graph.down.sql @@ -27,4 +27,5 @@ DROP TABLE IF EXISTS node_features; DROP TABLE IF EXISTS node_extra_types; DROP TABLE IF EXISTS nodes; DROP TABLE IF EXISTS channel_policy_extra_types; -DROP TABLE IF EXISTS zombie_channels; \ No newline at end of file +DROP TABLE IF EXISTS zombie_channels; +DROP TABLE IF EXISTS prune_log; diff --git a/sqldb/sqlc/migrations/000007_graph.up.sql b/sqldb/sqlc/migrations/000007_graph.up.sql index a4303bf95..80a58cbd1 100644 --- a/sqldb/sqlc/migrations/000007_graph.up.sql +++ b/sqldb/sqlc/migrations/000007_graph.up.sql @@ -320,3 +320,15 @@ CREATE TABLE IF NOT EXISTS zombie_channels ( ); CREATE UNIQUE INDEX IF NOT EXISTS zombie_channels_channel_id_version_idx ON zombie_channels(scid, version); + +CREATE TABLE IF NOT EXISTS prune_log ( + -- The block height that the prune was performed at. + -- NOTE: we don't use INTEGER PRIMARY KEY here since that would + -- get transformed into an auto-incrementing type by our SQL type + -- replacement logic. We don't want that since this must be the + -- actual block height and not an auto-incrementing value. + block_height BIGINT PRIMARY KEY, + + -- The block hash that the prune was performed at. + block_hash BLOB NOT NULL +); \ No newline at end of file diff --git a/sqldb/sqlc/models.go b/sqldb/sqlc/models.go index b750936d6..2bb61f769 100644 --- a/sqldb/sqlc/models.go +++ b/sqldb/sqlc/models.go @@ -181,6 +181,11 @@ type NodeFeature struct { FeatureBit int32 } +type PruneLog struct { + BlockHeight int64 + BlockHash []byte +} + type SourceNode struct { NodeID int64 } From 2da701cc4faa0f06755c86502ff290254d51fdb9 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 17:41:27 +0200 Subject: [PATCH 2/4] graph/db+sqldb: impl PruneGraphNodes Which lets us run `TestChannelEdgePruningUpdateIndexDeletion` against our SQL backends. --- graph/db/graph_test.go | 2 +- graph/db/sql_store.go | 66 ++++++++++++++++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 55 ++++++++++++++++++++++++++++++ sqldb/sqlc/querier.go | 4 +++ sqldb/sqlc/queries/graph.sql | 20 +++++++++++ 5 files changed, 146 insertions(+), 1 deletion(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index eb04ef20b..cf3ef3deb 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -3291,7 +3291,7 @@ func TestPruneGraphNodes(t *testing.T) { t.Parallel() ctx := context.Background() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) // We'll start off by inserting our source node, to ensure that it's // the only node left after we prune the graph. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 6d2dd85ae..c3f9541c7 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -63,7 +63,9 @@ type SQLQueries interface { ListNodesPaginated(ctx context.Context, arg sqlc.ListNodesPaginatedParams) ([]sqlc.Node, error) ListNodeIDsAndPubKeys(ctx context.Context, arg sqlc.ListNodeIDsAndPubKeysParams) ([]sqlc.ListNodeIDsAndPubKeysRow, error) IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error) + GetUnconnectedNodes(ctx context.Context) ([]sqlc.GetUnconnectedNodesRow, error) DeleteNodeByPubKey(ctx context.Context, arg sqlc.DeleteNodeByPubKeyParams) (sql.Result, error) + DeleteNode(ctx context.Context, id int64) error GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.NodeExtraType, error) UpsertNodeExtraType(ctx context.Context, arg sqlc.UpsertNodeExtraTypeParams) error @@ -2200,6 +2202,70 @@ func (s *SQLStore) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64, return newChanIDs, knownZombies, nil } +// PruneGraphNodes is a garbage collection method which attempts to prune out +// any nodes from the channel graph that are currently unconnected. This ensure +// that we only maintain a graph of reachable nodes. In the event that a pruned +// node gains more channels, it will be re-added back to the graph. +// +// NOTE: this prunes nodes across protocol versions. It will never prune the +// source nodes. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) PruneGraphNodes() ([]route.Vertex, error) { + var ctx = context.TODO() + + var prunedNodes []route.Vertex + err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { + var err error + prunedNodes, err = s.pruneGraphNodes(ctx, db) + + return err + }, func() { + prunedNodes = nil + }) + if err != nil { + return nil, fmt.Errorf("unable to prune nodes: %w", err) + } + + return prunedNodes, nil +} + +// pruneGraphNodes deletes any node in the DB that doesn't have a channel. +// +// NOTE: this prunes nodes across protocol versions. It will never prune the +// source nodes. +func (s *SQLStore) pruneGraphNodes(ctx context.Context, + db SQLQueries) ([]route.Vertex, error) { + + // Fetch all un-connected nodes from the database. + // NOTE: this will not include any nodes that are listed in the + // source table. + nodes, err := db.GetUnconnectedNodes(ctx) + if err != nil { + return nil, fmt.Errorf("unable to fetch unconnected nodes: %w", + err) + } + + prunedNodes := make([]route.Vertex, 0, len(nodes)) + for _, node := range nodes { + // TODO(elle): update to use sqlc.slice() once that works. + if err = db.DeleteNode(ctx, node.ID); err != nil { + return nil, fmt.Errorf("unable to delete "+ + "node(id=%d): %w", node.ID, err) + } + + pubKey, err := route.NewVertexFromBytes(node.PubKey) + if err != nil { + return nil, fmt.Errorf("unable to parse pubkey "+ + "for node(id=%d): %w", node.ID, err) + } + + prunedNodes = append(prunedNodes, pubKey) + } + + return prunedNodes, 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 diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 6463885cd..b25472aa4 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -149,6 +149,16 @@ func (q *Queries) DeleteExtraNodeType(ctx context.Context, arg DeleteExtraNodeTy return err } +const deleteNode = `-- name: DeleteNode :exec +DELETE FROM nodes +WHERE id = $1 +` + +func (q *Queries) DeleteNode(ctx context.Context, id int64) error { + _, err := q.db.ExecContext(ctx, deleteNode, id) + return err +} + const deleteNodeAddresses = `-- name: DeleteNodeAddresses :exec DELETE FROM node_addresses WHERE node_id = $1 @@ -1219,6 +1229,51 @@ func (q *Queries) GetSourceNodesByVersion(ctx context.Context, version int16) ([ return items, nil } +const getUnconnectedNodes = `-- name: GetUnconnectedNodes :many +SELECT n.id, n.pub_key +FROM nodes n +WHERE NOT EXISTS ( + SELECT 1 + FROM channels c + WHERE c.node_id_1 = n.id OR c.node_id_2 = n.id +) +AND NOT EXISTS ( + SELECT 1 + FROM source_nodes sn + WHERE sn.node_id = n.id +) +` + +type GetUnconnectedNodesRow struct { + ID int64 + PubKey []byte +} + +// Select all nodes that do not have any channels. +// Ignore any of our source nodes. +func (q *Queries) GetUnconnectedNodes(ctx context.Context) ([]GetUnconnectedNodesRow, error) { + rows, err := q.db.QueryContext(ctx, getUnconnectedNodes) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetUnconnectedNodesRow + for rows.Next() { + var i GetUnconnectedNodesRow + 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 getV1DisabledSCIDs = `-- name: GetV1DisabledSCIDs :many SELECT c.scid FROM channels c diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 83ee30a8a..4b8b49a2d 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -21,6 +21,7 @@ type Querier interface { DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error DeleteExtraNodeType(ctx context.Context, arg DeleteExtraNodeTypeParams) error DeleteInvoice(ctx context.Context, arg DeleteInvoiceParams) (sql.Result, error) + DeleteNode(ctx context.Context, id int64) error DeleteNodeAddresses(ctx context.Context, nodeID int64) error DeleteNodeByPubKey(ctx context.Context, arg DeleteNodeByPubKeyParams) (sql.Result, error) DeleteNodeFeature(ctx context.Context, arg DeleteNodeFeatureParams) error @@ -60,6 +61,9 @@ type Querier interface { GetPublicV1ChannelsBySCID(ctx context.Context, arg GetPublicV1ChannelsBySCIDParams) ([]Channel, error) GetSCIDByOutpoint(ctx context.Context, arg GetSCIDByOutpointParams) ([]byte, error) GetSourceNodesByVersion(ctx context.Context, version int16) ([]GetSourceNodesByVersionRow, error) + // Select all nodes that do not have any channels. + // Ignore any of our source nodes. + GetUnconnectedNodes(ctx context.Context) ([]GetUnconnectedNodesRow, error) // NOTE: this is V1 specific since for V1, disabled is a // simple, single boolean. The proposed V2 policy // structure will have a more complex disabled bit vector diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index ca40db176..c0ac68121 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -65,11 +65,31 @@ SELECT EXISTS ( AND n.pub_key = $1 ); +-- name: GetUnconnectedNodes :many +SELECT n.id, n.pub_key +FROM nodes n +-- Select all nodes that do not have any channels. +WHERE NOT EXISTS ( + SELECT 1 + FROM channels c + WHERE c.node_id_1 = n.id OR c.node_id_2 = n.id +) +-- Ignore any of our source nodes. +AND NOT EXISTS ( + SELECT 1 + FROM source_nodes sn + WHERE sn.node_id = n.id +); + -- name: DeleteNodeByPubKey :execresult DELETE FROM nodes WHERE pub_key = $1 AND version = $2; +-- name: DeleteNode :exec +DELETE FROM nodes +WHERE id = $1; + /* ───────────────────────────────────────────── node_features table queries ───────────────────────────────────────────── From 9dd0361ed0f6e72fb1919470a73110ede6c0ee33 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 17:46:37 +0200 Subject: [PATCH 3/4] graph/db+sqldb: impl PruneGraph, PruneTip, ChannelView Which lets us run `TestGraphPruning` and `TestBatchedAddChannelEdge` against our SQL backends. --- graph/db/graph_test.go | 4 +- graph/db/sql_store.go | 220 +++++++++++++++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 128 ++++++++++++++++++++ sqldb/sqlc/querier.go | 4 + sqldb/sqlc/queries/graph.sql | 37 ++++++ 5 files changed, 391 insertions(+), 2 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index cf3ef3deb..fb2d2b72e 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1685,7 +1685,7 @@ func TestGraphPruning(t *testing.T) { t.Parallel() ctx := context.Background() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) sourceNode := createTestVertex(t) if err := graph.SetSourceNode(ctx, sourceNode); err != nil { @@ -4006,7 +4006,7 @@ func TestBatchedAddChannelEdge(t *testing.T) { t.Parallel() ctx := context.Background() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) sourceNode := createTestVertex(t) require.Nil(t, graph.SetSourceNode(ctx, sourceNode)) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index c3f9541c7..e03ed273f 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -91,12 +91,14 @@ type SQLQueries interface { */ CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error) GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, error) + GetChannelByOutpoint(ctx context.Context, outpoint string) (sqlc.GetChannelByOutpointRow, error) GetChannelBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelBySCIDWithPoliciesParams) (sqlc.GetChannelBySCIDWithPoliciesRow, error) GetChannelAndNodesBySCID(ctx context.Context, arg sqlc.GetChannelAndNodesBySCIDParams) (sqlc.GetChannelAndNodesBySCIDRow, error) GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]sqlc.GetChannelFeaturesAndExtrasRow, error) HighestSCID(ctx context.Context, version int16) ([]byte, error) ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error) ListChannelsWithPoliciesPaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesPaginatedParams) ([]sqlc.ListChannelsWithPoliciesPaginatedRow, error) + ListChannelsPaginated(ctx context.Context, arg sqlc.ListChannelsPaginatedParams) ([]sqlc.ListChannelsPaginatedRow, error) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg sqlc.GetChannelsByPolicyLastUpdateRangeParams) ([]sqlc.GetChannelsByPolicyLastUpdateRangeRow, error) GetChannelByOutpointWithPolicies(ctx context.Context, arg sqlc.GetChannelByOutpointWithPoliciesParams) (sqlc.GetChannelByOutpointWithPoliciesRow, error) GetPublicV1ChannelsBySCID(ctx context.Context, arg sqlc.GetPublicV1ChannelsBySCIDParams) ([]sqlc.Channel, error) @@ -125,6 +127,12 @@ type SQLQueries interface { CountZombieChannels(ctx context.Context, version int16) (int64, error) DeleteZombieChannel(ctx context.Context, arg sqlc.DeleteZombieChannelParams) (sql.Result, error) IsZombieChannel(ctx context.Context, arg sqlc.IsZombieChannelParams) (bool, error) + + /* + Prune log table queries. + */ + GetPruneTip(ctx context.Context) (sqlc.PruneLog, error) + UpsertPruneLogEntry(ctx context.Context, arg sqlc.UpsertPruneLogEntryParams) error } // BatchedSQLQueries is a version of SQLQueries that's capable of batched @@ -2230,6 +2238,213 @@ func (s *SQLStore) PruneGraphNodes() ([]route.Vertex, error) { return prunedNodes, nil } +// PruneGraph prunes newly closed channels from the channel graph in response +// to a new block being solved on the network. Any transactions which spend the +// funding output of any known channels within he graph will be deleted. +// Additionally, the "prune tip", or the last block which has been used to +// prune the graph is stored so callers can ensure the graph is fully in sync +// with the current UTXO state. A slice of channels that have been closed by +// the target block along with any pruned nodes are returned if the function +// succeeds without error. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint, + blockHash *chainhash.Hash, blockHeight uint32) ( + []*models.ChannelEdgeInfo, []route.Vertex, error) { + + ctx := context.TODO() + + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + + var ( + closedChans []*models.ChannelEdgeInfo + prunedNodes []route.Vertex + ) + err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { + for _, outpoint := range spentOutputs { + // TODO(elle): potentially optimize this by using + // sqlc.slice() once that works for both SQLite and + // Postgres. + // + // NOTE: this fetches channels for all protocol + // versions. + row, err := db.GetChannelByOutpoint( + ctx, outpoint.String(), + ) + if errors.Is(err, sql.ErrNoRows) { + continue + } else if err != nil { + return fmt.Errorf("unable to fetch channel: %w", + err) + } + + node1, node2, err := buildNodeVertices( + row.Node1Pubkey, row.Node2Pubkey, + ) + if err != nil { + return err + } + + info, err := getAndBuildEdgeInfo( + ctx, db, s.cfg.ChainHash, row.Channel.ID, + row.Channel, node1, node2, + ) + if err != nil { + return err + } + + err = db.DeleteChannel(ctx, row.Channel.ID) + if err != nil { + return fmt.Errorf("unable to delete "+ + "channel: %w", err) + } + + closedChans = append(closedChans, info) + } + + err := db.UpsertPruneLogEntry( + ctx, sqlc.UpsertPruneLogEntryParams{ + BlockHash: blockHash[:], + BlockHeight: int64(blockHeight), + }, + ) + if err != nil { + return fmt.Errorf("unable to insert prune log "+ + "entry: %w", err) + } + + // Now that we've pruned some channels, we'll also prune any + // nodes that no longer have any channels. + prunedNodes, err = s.pruneGraphNodes(ctx, db) + if err != nil { + return fmt.Errorf("unable to prune graph nodes: %w", + err) + } + + return nil + }, func() { + prunedNodes = nil + closedChans = nil + }) + if err != nil { + return nil, nil, fmt.Errorf("unable to prune graph: %w", err) + } + + for _, channel := range closedChans { + s.rejectCache.remove(channel.ChannelID) + s.chanCache.remove(channel.ChannelID) + } + + return closedChans, prunedNodes, nil +} + +// ChannelView returns the verifiable edge information for each active channel +// within the known channel graph. The set of UTXOs (along with their scripts) +// returned are the ones that need to be watched on chain to detect channel +// closes on the resident blockchain. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) ChannelView() ([]EdgePoint, error) { + var ( + ctx = context.TODO() + edgePoints []EdgePoint + ) + + handleChannel := func(db SQLQueries, + channel sqlc.ListChannelsPaginatedRow) error { + + pkScript, err := genMultiSigP2WSH( + channel.BitcoinKey1, channel.BitcoinKey2, + ) + if err != nil { + return err + } + + op, err := wire.NewOutPointFromString(channel.Outpoint) + if err != nil { + return err + } + + edgePoints = append(edgePoints, EdgePoint{ + FundingPkScript: pkScript, + OutPoint: *op, + }) + + return nil + } + + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + lastID := int64(-1) + for { + rows, err := db.ListChannelsPaginated( + ctx, sqlc.ListChannelsPaginatedParams{ + Version: int16(ProtocolV1), + ID: lastID, + Limit: pageSize, + }, + ) + if err != nil { + return err + } + + if len(rows) == 0 { + break + } + + for _, row := range rows { + err := handleChannel(db, row) + if err != nil { + return err + } + + lastID = row.ID + } + } + + return nil + }, func() { + edgePoints = nil + }) + if err != nil { + return nil, fmt.Errorf("unable to fetch channel view: %w", err) + } + + return edgePoints, nil +} + +// PruneTip returns the block height and hash of the latest block that has been +// used to prune channels in the graph. Knowing the "prune tip" allows callers +// to tell if the graph is currently in sync with the current best known UTXO +// state. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) PruneTip() (*chainhash.Hash, uint32, error) { + var ( + ctx = context.TODO() + tipHash chainhash.Hash + tipHeight uint32 + ) + err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { + pruneTip, err := db.GetPruneTip(ctx) + if errors.Is(err, sql.ErrNoRows) { + return ErrGraphNeverPruned + } else if err != nil { + return fmt.Errorf("unable to fetch prune tip: %w", err) + } + + tipHash = chainhash.Hash(pruneTip.BlockHash) + tipHeight = uint32(pruneTip.BlockHeight) + + return nil + }, sqldb.NoOpReset) + if err != nil { + return nil, 0, err + } + + return &tipHash, tipHeight, nil +} + // pruneGraphNodes deletes any node in the DB that doesn't have a channel. // // NOTE: this prunes nodes across protocol versions. It will never prune the @@ -3389,6 +3604,11 @@ func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries, chain chainhash.Hash, dbChanID int64, dbChan sqlc.Channel, node1, node2 route.Vertex) (*models.ChannelEdgeInfo, error) { + if dbChan.Version != int16(ProtocolV1) { + return nil, fmt.Errorf("unsupported channel version: %d", + dbChan.Version) + } + fv, extras, err := getChanFeaturesAndExtras( ctx, db, dbChanID, ) diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index b25472aa4..d774a12f7 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -273,6 +273,46 @@ func (q *Queries) GetChannelAndNodesBySCID(ctx context.Context, arg GetChannelAn return i, err } +const getChannelByOutpoint = `-- name: GetChannelByOutpoint :one +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 +FROM channels c + JOIN nodes n1 ON c.node_id_1 = n1.id + JOIN nodes n2 ON c.node_id_2 = n2.id +WHERE c.outpoint = $1 +` + +type GetChannelByOutpointRow struct { + Channel Channel + Node1Pubkey []byte + Node2Pubkey []byte +} + +func (q *Queries) GetChannelByOutpoint(ctx context.Context, outpoint string) (GetChannelByOutpointRow, error) { + row := q.db.QueryRowContext(ctx, getChannelByOutpoint, outpoint) + var i GetChannelByOutpointRow + err := row.Scan( + &i.Channel.ID, + &i.Channel.Version, + &i.Channel.Scid, + &i.Channel.NodeID1, + &i.Channel.NodeID2, + &i.Channel.Outpoint, + &i.Channel.Capacity, + &i.Channel.BitcoinKey1, + &i.Channel.BitcoinKey2, + &i.Channel.Node1Signature, + &i.Channel.Node2Signature, + &i.Channel.Bitcoin1Signature, + &i.Channel.Bitcoin2Signature, + &i.Node1Pubkey, + &i.Node2Pubkey, + ) + return i, err +} + const getChannelByOutpointWithPolicies = `-- name: GetChannelByOutpointWithPolicies :one 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, @@ -1127,6 +1167,20 @@ func (q *Queries) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByL return items, nil } +const getPruneTip = `-- name: GetPruneTip :one +SELECT block_height, block_hash +FROM prune_log +ORDER BY block_height DESC +LIMIT 1 +` + +func (q *Queries) GetPruneTip(ctx context.Context) (PruneLog, error) { + row := q.db.QueryRowContext(ctx, getPruneTip) + var i PruneLog + err := row.Scan(&i.BlockHeight, &i.BlockHash) + return i, err +} + const getPublicV1ChannelsBySCID = `-- name: GetPublicV1ChannelsBySCID :many SELECT id, version, scid, node_id_1, node_id_2, outpoint, capacity, bitcoin_key_1, bitcoin_key_2, node_1_signature, node_2_signature, bitcoin_1_signature, bitcoin_2_signature FROM channels @@ -1650,6 +1704,55 @@ func (q *Queries) ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNo return items, nil } +const listChannelsPaginated = `-- name: ListChannelsPaginated :many +SELECT id, bitcoin_key_1, bitcoin_key_2, outpoint +FROM channels c +WHERE c.version = $1 AND c.id > $2 +ORDER BY c.id +LIMIT $3 +` + +type ListChannelsPaginatedParams struct { + Version int16 + ID int64 + Limit int32 +} + +type ListChannelsPaginatedRow struct { + ID int64 + BitcoinKey1 []byte + BitcoinKey2 []byte + Outpoint string +} + +func (q *Queries) ListChannelsPaginated(ctx context.Context, arg ListChannelsPaginatedParams) ([]ListChannelsPaginatedRow, error) { + rows, err := q.db.QueryContext(ctx, listChannelsPaginated, arg.Version, arg.ID, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListChannelsPaginatedRow + for rows.Next() { + var i ListChannelsPaginatedRow + if err := rows.Scan( + &i.ID, + &i.BitcoinKey1, + &i.BitcoinKey2, + &i.Outpoint, + ); 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 listChannelsWithPoliciesPaginated = `-- name: ListChannelsWithPoliciesPaginated :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, @@ -2033,6 +2136,31 @@ func (q *Queries) UpsertNodeExtraType(ctx context.Context, arg UpsertNodeExtraTy return err } +const upsertPruneLogEntry = `-- name: UpsertPruneLogEntry :exec +/* ───────────────────────────────────────────── + prune_log table queries + ───────────────────────────────────────────── +*/ + +INSERT INTO prune_log ( + block_height, block_hash +) VALUES ( + $1, $2 +) +ON CONFLICT(block_height) DO UPDATE SET + block_hash = EXCLUDED.block_hash +` + +type UpsertPruneLogEntryParams struct { + BlockHeight int64 + BlockHash []byte +} + +func (q *Queries) UpsertPruneLogEntry(ctx context.Context, arg UpsertPruneLogEntryParams) error { + _, err := q.db.ExecContext(ctx, upsertPruneLogEntry, arg.BlockHeight, arg.BlockHash) + return err +} + const upsertZombieChannel = `-- name: UpsertZombieChannel :exec /* ───────────────────────────────────────────── zombie_channels table queries diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 4b8b49a2d..fc9b2128f 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -32,6 +32,7 @@ type Querier interface { FilterInvoices(ctx context.Context, arg FilterInvoicesParams) ([]Invoice, error) GetAMPInvoiceID(ctx context.Context, setID []byte) (int64, error) GetChannelAndNodesBySCID(ctx context.Context, arg GetChannelAndNodesBySCIDParams) (GetChannelAndNodesBySCIDRow, error) + GetChannelByOutpoint(ctx context.Context, outpoint string) (GetChannelByOutpointRow, error) GetChannelByOutpointWithPolicies(ctx context.Context, arg GetChannelByOutpointWithPoliciesParams) (GetChannelByOutpointWithPoliciesRow, error) GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDParams) (Channel, error) GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChannelBySCIDWithPoliciesParams) (GetChannelBySCIDWithPoliciesRow, error) @@ -58,6 +59,7 @@ type Querier interface { GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error) GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]Node, error) + GetPruneTip(ctx context.Context) (PruneLog, error) GetPublicV1ChannelsBySCID(ctx context.Context, arg GetPublicV1ChannelsBySCIDParams) ([]Channel, error) GetSCIDByOutpoint(ctx context.Context, arg GetSCIDByOutpointParams) ([]byte, error) GetSourceNodesByVersion(ctx context.Context, version int16) ([]GetSourceNodesByVersionRow, error) @@ -86,6 +88,7 @@ type Querier interface { IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error) IsZombieChannel(ctx context.Context, arg IsZombieChannelParams) (bool, error) ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error) + ListChannelsPaginated(ctx context.Context, arg ListChannelsPaginatedParams) ([]ListChannelsPaginatedRow, error) ListChannelsWithPoliciesPaginated(ctx context.Context, arg ListChannelsWithPoliciesPaginatedParams) ([]ListChannelsWithPoliciesPaginatedRow, error) ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error) ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]Node, error) @@ -108,6 +111,7 @@ type Querier interface { UpsertEdgePolicy(ctx context.Context, arg UpsertEdgePolicyParams) (int64, error) UpsertNode(ctx context.Context, arg UpsertNodeParams) (int64, error) UpsertNodeExtraType(ctx context.Context, arg UpsertNodeExtraTypeParams) error + UpsertPruneLogEntry(ctx context.Context, arg UpsertPruneLogEntryParams) error UpsertZombieChannel(ctx context.Context, arg UpsertZombieChannelParams) error } diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index c0ac68121..412c4fd15 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -212,6 +212,16 @@ RETURNING id; SELECT * FROM channels WHERE scid = $1 AND version = $2; +-- name: GetChannelByOutpoint :one +SELECT + sqlc.embed(c), + n1.pub_key AS node1_pubkey, + n2.pub_key AS node2_pubkey +FROM channels c + JOIN nodes n1 ON c.node_id_1 = n1.id + JOIN nodes n2 ON c.node_id_2 = n2.id +WHERE c.outpoint = $1; + -- name: GetChannelAndNodesBySCID :one SELECT c.*, @@ -411,6 +421,13 @@ WHERE node_1_signature IS NOT NULL AND scid >= @start_scid AND scid < @end_scid; +-- name: ListChannelsPaginated :many +SELECT id, bitcoin_key_1, bitcoin_key_2, outpoint +FROM channels c +WHERE c.version = $1 AND c.id > $2 +ORDER BY c.id +LIMIT $3; + -- name: ListChannelsWithPoliciesPaginated :many SELECT sqlc.embed(c), @@ -648,3 +665,23 @@ SELECT EXISTS ( WHERE scid = $1 AND version = $2 ) AS is_zombie; + +/* ───────────────────────────────────────────── + prune_log table queries + ───────────────────────────────────────────── +*/ + +-- name: UpsertPruneLogEntry :exec +INSERT INTO prune_log ( + block_height, block_hash +) VALUES ( + $1, $2 +) +ON CONFLICT(block_height) DO UPDATE SET + block_hash = EXCLUDED.block_hash; + +-- name: GetPruneTip :one +SELECT block_height, block_hash +FROM prune_log +ORDER BY block_height DESC +LIMIT 1; From e875183c4f01a304e59fd5d7dbcad0c01b4370af Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 17:50:26 +0200 Subject: [PATCH 4/4] sqldb+graph/db: impl DisconnectBlockAtHeight Which lets us run `TestDisconnectBlockAtHeight` and `TestStressTestChannelGraphAPI` against our SQL backends. --- docs/release-notes/release-notes-0.20.0.md | 1 + graph/db/graph_test.go | 8 +- graph/db/sql_store.go | 94 ++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 77 ++++++++++++++++++ sqldb/sqlc/querier.go | 2 + sqldb/sqlc/queries/graph.sql | 15 ++++ 6 files changed, 195 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 040290f33..19cff1f38 100644 --- a/docs/release-notes/release-notes-0.20.0.md +++ b/docs/release-notes/release-notes-0.20.0.md @@ -88,6 +88,7 @@ circuit. The indices are only available for forwarding events saved after v0.20. * [6](https://github.com/lightningnetwork/lnd/pull/9936) * [7](https://github.com/lightningnetwork/lnd/pull/9937) * [8](https://github.com/lightningnetwork/lnd/pull/9938) + * [9](https://github.com/lightningnetwork/lnd/pull/9939) ## RPC Updates diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index fb2d2b72e..419c96bc3 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -513,7 +513,7 @@ func TestDisconnectBlockAtHeight(t *testing.T) { t.Parallel() ctx := context.Background() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) sourceNode := createTestVertex(t) if err := graph.SetSourceNode(ctx, sourceNode); err != nil { @@ -2440,7 +2440,7 @@ func TestStressTestChannelGraphAPI(t *testing.T) { t.Parallel() ctx := context.Background() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) node1 := createTestVertex(t) require.NoError(t, graph.AddLightningNode(ctx, node1)) @@ -2448,6 +2448,10 @@ func TestStressTestChannelGraphAPI(t *testing.T) { node2 := createTestVertex(t) require.NoError(t, graph.AddLightningNode(ctx, node2)) + // We need to update the node's timestamp since this call to + // SetSourceNode will trigger an upsert which will only be allowed if + // the newest LastUpdate time is greater than the current one. + node1.LastUpdate = node1.LastUpdate.Add(time.Second) require.NoError(t, graph.SetSourceNode(ctx, node1)) type chanInfo struct { diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index e03ed273f..a6c67948c 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -19,6 +19,7 @@ import ( "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/aliasmgr" "github.com/lightningnetwork/lnd/batch" "github.com/lightningnetwork/lnd/fn/v2" "github.com/lightningnetwork/lnd/graph/db/models" @@ -92,6 +93,7 @@ type SQLQueries interface { CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error) GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, error) GetChannelByOutpoint(ctx context.Context, outpoint string) (sqlc.GetChannelByOutpointRow, error) + GetChannelsBySCIDRange(ctx context.Context, arg sqlc.GetChannelsBySCIDRangeParams) ([]sqlc.GetChannelsBySCIDRangeRow, error) GetChannelBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelBySCIDWithPoliciesParams) (sqlc.GetChannelBySCIDWithPoliciesRow, error) GetChannelAndNodesBySCID(ctx context.Context, arg sqlc.GetChannelAndNodesBySCIDParams) (sqlc.GetChannelAndNodesBySCIDRow, error) GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]sqlc.GetChannelFeaturesAndExtrasRow, error) @@ -133,6 +135,7 @@ type SQLQueries interface { */ GetPruneTip(ctx context.Context) (sqlc.PruneLog, error) UpsertPruneLogEntry(ctx context.Context, arg sqlc.UpsertPruneLogEntryParams) error + DeletePruneLogEntriesInRange(ctx context.Context, arg sqlc.DeletePruneLogEntriesInRangeParams) error } // BatchedSQLQueries is a version of SQLQueries that's capable of batched @@ -2481,6 +2484,97 @@ func (s *SQLStore) pruneGraphNodes(ctx context.Context, return prunedNodes, nil } +// DisconnectBlockAtHeight is used to indicate that the block specified +// by the passed height has been disconnected from the main chain. This +// will "rewind" the graph back to the height below, deleting channels +// that are no longer confirmed from the graph. The prune log will be +// set to the last prune height valid for the remaining chain. +// Channels that were removed from the graph resulting from the +// disconnected block are returned. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) DisconnectBlockAtHeight(height uint32) ( + []*models.ChannelEdgeInfo, error) { + + ctx := context.TODO() + + var ( + // Every channel having a ShortChannelID starting at 'height' + // will no longer be confirmed. + startShortChanID = lnwire.ShortChannelID{ + BlockHeight: height, + } + + // Delete everything after this height from the db up until the + // SCID alias range. + endShortChanID = aliasmgr.StartingAlias + + removedChans []*models.ChannelEdgeInfo + ) + + var chanIDStart [8]byte + byteOrder.PutUint64(chanIDStart[:], startShortChanID.ToUint64()) + var chanIDEnd [8]byte + byteOrder.PutUint64(chanIDEnd[:], endShortChanID.ToUint64()) + + err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { + rows, err := db.GetChannelsBySCIDRange( + ctx, sqlc.GetChannelsBySCIDRangeParams{ + StartScid: chanIDStart[:], + EndScid: chanIDEnd[:], + }, + ) + if err != nil { + return fmt.Errorf("unable to fetch channels: %w", err) + } + + for _, row := range rows { + node1, node2, err := buildNodeVertices( + row.Node1PubKey, row.Node2PubKey, + ) + if err != nil { + return err + } + + channel, err := getAndBuildEdgeInfo( + ctx, db, s.cfg.ChainHash, row.Channel.ID, + row.Channel, node1, node2, + ) + if err != nil { + return err + } + + err = db.DeleteChannel(ctx, row.Channel.ID) + if err != nil { + return fmt.Errorf("unable to delete "+ + "channel: %w", err) + } + + removedChans = append(removedChans, channel) + } + + return db.DeletePruneLogEntriesInRange( + ctx, sqlc.DeletePruneLogEntriesInRangeParams{ + StartHeight: int64(height), + EndHeight: int64(endShortChanID.BlockHeight), + }, + ) + }, func() { + removedChans = nil + }) + if err != nil { + return nil, fmt.Errorf("unable to disconnect block at "+ + "height: %w", err) + } + + for _, channel := range removedChans { + s.rejectCache.remove(channel.ChannelID) + s.chanCache.remove(channel.ChannelID) + } + + return removedChans, 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 diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index d774a12f7..c74b4c050 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -200,6 +200,22 @@ func (q *Queries) DeleteNodeFeature(ctx context.Context, arg DeleteNodeFeaturePa return err } +const deletePruneLogEntriesInRange = `-- name: DeletePruneLogEntriesInRange :exec +DELETE FROM prune_log +WHERE block_height >= $1 + AND block_height <= $2 +` + +type DeletePruneLogEntriesInRangeParams struct { + StartHeight int64 + EndHeight int64 +} + +func (q *Queries) DeletePruneLogEntriesInRange(ctx context.Context, arg DeletePruneLogEntriesInRangeParams) error { + _, err := q.db.ExecContext(ctx, deletePruneLogEntriesInRange, arg.StartHeight, arg.EndHeight) + return err +} + const deleteZombieChannel = `-- name: DeleteZombieChannel :execresult DELETE FROM zombie_channels WHERE scid = $1 @@ -943,6 +959,67 @@ func (q *Queries) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg Ge return items, nil } +const getChannelsBySCIDRange = `-- name: GetChannelsBySCIDRange :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_pub_key, + n2.pub_key AS node2_pub_key +FROM channels c + JOIN nodes n1 ON c.node_id_1 = n1.id + JOIN nodes n2 ON c.node_id_2 = n2.id +WHERE scid >= $1 + AND scid < $2 +` + +type GetChannelsBySCIDRangeParams struct { + StartScid []byte + EndScid []byte +} + +type GetChannelsBySCIDRangeRow struct { + Channel Channel + Node1PubKey []byte + Node2PubKey []byte +} + +func (q *Queries) GetChannelsBySCIDRange(ctx context.Context, arg GetChannelsBySCIDRangeParams) ([]GetChannelsBySCIDRangeRow, error) { + rows, err := q.db.QueryContext(ctx, getChannelsBySCIDRange, arg.StartScid, arg.EndScid) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetChannelsBySCIDRangeRow + for rows.Next() { + var i GetChannelsBySCIDRangeRow + if err := rows.Scan( + &i.Channel.ID, + &i.Channel.Version, + &i.Channel.Scid, + &i.Channel.NodeID1, + &i.Channel.NodeID2, + &i.Channel.Outpoint, + &i.Channel.Capacity, + &i.Channel.BitcoinKey1, + &i.Channel.BitcoinKey2, + &i.Channel.Node1Signature, + &i.Channel.Node2Signature, + &i.Channel.Bitcoin1Signature, + &i.Channel.Bitcoin2Signature, + &i.Node1PubKey, + &i.Node2PubKey, + ); 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 getExtraNodeTypes = `-- name: GetExtraNodeTypes :many SELECT node_id, type, value FROM node_extra_types diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index fc9b2128f..f0e1ad3e7 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -25,6 +25,7 @@ type Querier interface { DeleteNodeAddresses(ctx context.Context, nodeID int64) error DeleteNodeByPubKey(ctx context.Context, arg DeleteNodeByPubKeyParams) (sql.Result, error) DeleteNodeFeature(ctx context.Context, arg DeleteNodeFeatureParams) error + DeletePruneLogEntriesInRange(ctx context.Context, arg DeletePruneLogEntriesInRangeParams) error DeleteZombieChannel(ctx context.Context, arg DeleteZombieChannelParams) (sql.Result, error) FetchAMPSubInvoiceHTLCs(ctx context.Context, arg FetchAMPSubInvoiceHTLCsParams) ([]FetchAMPSubInvoiceHTLCsRow, error) FetchAMPSubInvoices(ctx context.Context, arg FetchAMPSubInvoicesParams) ([]AmpSubInvoice, error) @@ -40,6 +41,7 @@ type Querier interface { GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetChannelPolicyByChannelAndNodeParams) (ChannelPolicy, error) GetChannelPolicyExtraTypes(ctx context.Context, arg GetChannelPolicyExtraTypesParams) ([]GetChannelPolicyExtraTypesRow, error) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg GetChannelsByPolicyLastUpdateRangeParams) ([]GetChannelsByPolicyLastUpdateRangeRow, error) + GetChannelsBySCIDRange(ctx context.Context, arg GetChannelsBySCIDRangeParams) ([]GetChannelsBySCIDRangeRow, error) GetDatabaseVersion(ctx context.Context) (int32, error) GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]NodeExtraType, error) // This method may return more than one invoice if filter using multiple fields diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 412c4fd15..67b5dae6b 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -208,6 +208,16 @@ INSERT INTO channels ( ) RETURNING id; +-- name: GetChannelsBySCIDRange :many +SELECT sqlc.embed(c), + n1.pub_key AS node1_pub_key, + n2.pub_key AS node2_pub_key +FROM channels c + JOIN nodes n1 ON c.node_id_1 = n1.id + JOIN nodes n2 ON c.node_id_2 = n2.id +WHERE scid >= @start_scid + AND scid < @end_scid; + -- name: GetChannelBySCID :one SELECT * FROM channels WHERE scid = $1 AND version = $2; @@ -685,3 +695,8 @@ SELECT block_height, block_hash FROM prune_log ORDER BY block_height DESC LIMIT 1; + +-- name: DeletePruneLogEntriesInRange :exec +DELETE FROM prune_log +WHERE block_height >= @start_height + AND block_height <= @end_height;