From cee872ad0e29cbef336f31247375548c0d3aff93 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 17:08:36 +0200 Subject: [PATCH 1/5] graph/db: make makeZombiePubkeys reusable So that we can use it in our SQLStore implementation too without needing to fetch the entire ChannelEdgePolicy. --- graph/db/kv_store.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index c2c681965..150e290ad 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -2747,7 +2747,17 @@ func (c *KVStore) delChannelEdgeUnsafe(edges, edgeIndex, chanIndex, nodeKey1, nodeKey2 := edgeInfo.NodeKey1Bytes, edgeInfo.NodeKey2Bytes if strictZombie { - nodeKey1, nodeKey2 = makeZombiePubkeys(&edgeInfo, edge1, edge2) + var e1UpdateTime, e2UpdateTime *time.Time + if edge1 != nil { + e1UpdateTime = &edge1.LastUpdate + } + if edge2 != nil { + e2UpdateTime = &edge2.LastUpdate + } + + nodeKey1, nodeKey2 = makeZombiePubkeys( + &edgeInfo, e1UpdateTime, e2UpdateTime, + ) } return &edgeInfo, markEdgeZombie( @@ -2772,7 +2782,7 @@ func (c *KVStore) delChannelEdgeUnsafe(edges, edgeIndex, chanIndex, // marked with the correct lagging channel since we received an update from only // one side. func makeZombiePubkeys(info *models.ChannelEdgeInfo, - e1, e2 *models.ChannelEdgePolicy) ([33]byte, [33]byte) { + e1, e2 *time.Time) ([33]byte, [33]byte) { switch { // If we don't have either edge policy, we'll return both pubkeys so @@ -2784,7 +2794,7 @@ func makeZombiePubkeys(info *models.ChannelEdgeInfo, // older, we'll return edge1's pubkey and a blank pubkey for edge2. This // means that only an update from edge1 will be able to resurrect the // channel. - case e1 == nil || (e2 != nil && e1.LastUpdate.Before(e2.LastUpdate)): + case e1 == nil || (e2 != nil && e1.Before(*e2)): return info.NodeKey1Bytes, [33]byte{} // Otherwise, we're missing edge2 or edge2 is the older side, so we From c648c7a22db1ae0eda2d6aaa2bbf49e927293552 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 16:59:40 +0200 Subject: [PATCH 2/5] sqldb/sqlc: add zombie index table Note that this table will only contain entries for channels that we have deleted from the `channels` table which is why we cannot use foreign keys. Similarly, we may no longer have node entries for the nodes in the table. --- sqldb/sqlc/migrations/000007_graph.down.sql | 4 ++- sqldb/sqlc/migrations/000007_graph.up.sql | 28 +++++++++++++++++++++ sqldb/sqlc/models.go | 7 ++++++ 3 files changed, 38 insertions(+), 1 deletion(-) diff --git a/sqldb/sqlc/migrations/000007_graph.down.sql b/sqldb/sqlc/migrations/000007_graph.down.sql index 0990705ef..a2f519a2c 100644 --- a/sqldb/sqlc/migrations/000007_graph.down.sql +++ b/sqldb/sqlc/migrations/000007_graph.down.sql @@ -25,4 +25,6 @@ DROP TABLE IF EXISTS source_nodes; DROP TABLE IF EXISTS node_addresses; DROP TABLE IF EXISTS node_features; DROP TABLE IF EXISTS node_extra_types; -DROP TABLE IF EXISTS nodes; \ No newline at end of file +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 diff --git a/sqldb/sqlc/migrations/000007_graph.up.sql b/sqldb/sqlc/migrations/000007_graph.up.sql index 0d179f1d3..a4303bf95 100644 --- a/sqldb/sqlc/migrations/000007_graph.up.sql +++ b/sqldb/sqlc/migrations/000007_graph.up.sql @@ -292,3 +292,31 @@ CREATE TABLE IF NOT EXISTS channel_policy_extra_types ( CREATE UNIQUE INDEX IF NOT EXISTS channel_policy_extra_types_unique ON channel_policy_extra_types ( type, channel_policy_id ); + +/* ───────────────────────────────────────────── + Other graph related tables + ───────────────────────────────────────────── +*/ + +CREATE TABLE IF NOT EXISTS zombie_channels ( + -- The channel id (short channel id) of the channel. + -- NOTE: we don't use a foreign key here to the `channels` + -- table since we may delete the channel record once it + -- is marked as a zombie. + scid BLOB NOT NULL, + + -- The protocol version that this node was gossiped on. + version SMALLINT NOT NULL, + + -- The public key of the node 1 node of the channel. If + -- this is not null, it means an update from this node + -- will be able to resurrect the channel. + node_key_1 BLOB, + + -- The public key of the node 2 node of the channel. If + -- this is not null, it means an update from this node + -- will be able to resurrect the channel. + node_key_2 BLOB +); +CREATE UNIQUE INDEX IF NOT EXISTS zombie_channels_channel_id_version_idx + ON zombie_channels(scid, version); diff --git a/sqldb/sqlc/models.go b/sqldb/sqlc/models.go index 70082eb4d..b750936d6 100644 --- a/sqldb/sqlc/models.go +++ b/sqldb/sqlc/models.go @@ -184,3 +184,10 @@ type NodeFeature struct { type SourceNode struct { NodeID int64 } + +type ZombieChannel struct { + Scid []byte + Version int16 + NodeKey1 []byte + NodeKey2 []byte +} From 137fc092309475071d2ea635fcac32908ecc97d3 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 24 Jun 2025 18:58:52 +0200 Subject: [PATCH 3/5] graph/db: add channelIDToBytes helper --- graph/db/sql_store.go | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 3710efa8e..f8f64f125 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1288,13 +1288,10 @@ func (s *SQLStore) FilterChannelRange(startHeight, endHeight uint32, TxIndex: math.MaxUint32 & 0x00ffffff, TxPosition: math.MaxUint16, } + chanIDStart = channelIDToBytes(startSCID.ToUint64()) + chanIDEnd = channelIDToBytes(endSCID.ToUint64()) ) - var chanIDStart [8]byte - byteOrder.PutUint64(chanIDStart[:], startSCID.ToUint64()) - var chanIDEnd [8]byte - byteOrder.PutUint64(chanIDEnd[:], endSCID.ToUint64()) - // 1) get all channels where channelID is between start and end chan ID. // 2) skip if not public (ie, no channel_proof) // 3) collect that channel. @@ -1638,9 +1635,8 @@ func updateChanEdgePolicy(ctx context.Context, tx SQLQueries, var ( node1Pub, node2Pub route.Vertex isNode1 bool - chanIDB [8]byte + chanIDB = channelIDToBytes(edge.ChannelID) ) - byteOrder.PutUint64(chanIDB[:], edge.ChannelID) // Check that this edge policy refers to a channel that we already // know of. We do this explicitly so that we can return the appropriate @@ -2334,8 +2330,7 @@ func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) { func insertChannel(ctx context.Context, db SQLQueries, edge *models.ChannelEdgeInfo) error { - var chanIDB [8]byte - byteOrder.PutUint64(chanIDB[:], edge.ChannelID) + chanIDB := channelIDToBytes(edge.ChannelID) // Make sure that the channel doesn't already exist. We do this // explicitly instead of relying on catching a unique constraint error @@ -2920,3 +2915,12 @@ func extractChannelPolicies(row any) (*sqlc.ChannelPolicy, *sqlc.ChannelPolicy, "extractChannelPolicies: %T", r) } } + +// channelIDToBytes converts a channel ID (SCID) to a byte array +// representation. +func channelIDToBytes(channelID uint64) [8]byte { + var chanIDB [8]byte + byteOrder.PutUint64(chanIDB[:], channelID) + + return chanIDB +} From 00b6e0204c5fb4775e23489daf4be6ae5bf4e4c8 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 17:04:21 +0200 Subject: [PATCH 4/5] graph/db+sqldb: implement various zombie index methods Here we implement the SQLStore methods: - MarkEdgeZombie - MarkEdgeLive - IsZombieEdge - NumZombies These will be tested in the next commit as one more method implementation is required. --- graph/db/sql_store.go | 162 +++++++++++++++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 85 ++++++++++++++++++ sqldb/sqlc/querier.go | 4 + sqldb/sqlc/queries/graph.sql | 31 +++++++ 4 files changed, 282 insertions(+) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index f8f64f125..7aff00ad6 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -108,6 +108,14 @@ type SQLQueries interface { InsertChanPolicyExtraType(ctx context.Context, arg sqlc.InsertChanPolicyExtraTypeParams) error GetChannelPolicyExtraTypes(ctx context.Context, arg sqlc.GetChannelPolicyExtraTypesParams) ([]sqlc.GetChannelPolicyExtraTypesRow, error) DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error + + /* + Zombie index queries. + */ + UpsertZombieChannel(ctx context.Context, arg sqlc.UpsertZombieChannelParams) error + GetZombieChannel(ctx context.Context, arg sqlc.GetZombieChannelParams) (sqlc.ZombieChannel, error) + CountZombieChannels(ctx context.Context, version int16) (int64, error) + DeleteZombieChannel(ctx context.Context, arg sqlc.DeleteZombieChannelParams) (sql.Result, error) } // BatchedSQLQueries is a version of SQLQueries that's capable of batched @@ -1390,6 +1398,160 @@ func (s *SQLStore) FilterChannelRange(startHeight, endHeight uint32, }), nil } +// MarkEdgeZombie attempts to mark a channel identified by its channel ID as a +// zombie. This method is used on an ad-hoc basis, when channels need to be +// marked as zombies outside the normal pruning cycle. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) MarkEdgeZombie(chanID uint64, + pubKey1, pubKey2 [33]byte) error { + + ctx := context.TODO() + + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + + chanIDB := channelIDToBytes(chanID) + + err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { + return db.UpsertZombieChannel( + ctx, sqlc.UpsertZombieChannelParams{ + Version: int16(ProtocolV1), + Scid: chanIDB[:], + NodeKey1: pubKey1[:], + NodeKey2: pubKey2[:], + }, + ) + }, sqldb.NoOpReset) + if err != nil { + return fmt.Errorf("unable to upsert zombie channel "+ + "(channel_id=%d): %w", chanID, err) + } + + s.rejectCache.remove(chanID) + s.chanCache.remove(chanID) + + return nil +} + +// MarkEdgeLive clears an edge from our zombie index, deeming it as live. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) MarkEdgeLive(chanID uint64) error { + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + + var ( + ctx = context.TODO() + chanIDB = channelIDToBytes(chanID) + ) + + err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { + res, err := db.DeleteZombieChannel( + ctx, sqlc.DeleteZombieChannelParams{ + Scid: chanIDB[:], + Version: int16(ProtocolV1), + }, + ) + if err != nil { + return fmt.Errorf("unable to delete zombie channel: %w", + err) + } + + rows, err := res.RowsAffected() + if err != nil { + return err + } + + if rows == 0 { + return ErrZombieEdgeNotFound + } else if rows > 1 { + return fmt.Errorf("deleted %d zombie rows, "+ + "expected 1", rows) + } + + return nil + }, sqldb.NoOpReset) + if err != nil { + return fmt.Errorf("unable to mark edge live "+ + "(channel_id=%d): %w", chanID, err) + } + + s.rejectCache.remove(chanID) + s.chanCache.remove(chanID) + + return err +} + +// IsZombieEdge returns whether the edge is considered zombie. If it is a +// zombie, then the two node public keys corresponding to this edge are also +// returned. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) IsZombieEdge(chanID uint64) (bool, [33]byte, [33]byte) { + var ( + ctx = context.TODO() + isZombie bool + pubKey1, pubKey2 route.Vertex + chanIDB = channelIDToBytes(chanID) + ) + + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + zombie, err := db.GetZombieChannel( + ctx, sqlc.GetZombieChannelParams{ + Scid: chanIDB[:], + Version: int16(ProtocolV1), + }, + ) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return fmt.Errorf("unable to fetch zombie channel: %w", + err) + } + + copy(pubKey1[:], zombie.NodeKey1) + copy(pubKey2[:], zombie.NodeKey2) + isZombie = true + + return nil + }, sqldb.NoOpReset) + if err != nil { + // TODO(elle): update the IsZombieEdge method to return an + // error. + return false, route.Vertex{}, route.Vertex{} + } + + return isZombie, pubKey1, pubKey2 +} + +// NumZombies returns the current number of zombie channels in the graph. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) NumZombies() (uint64, error) { + var ( + ctx = context.TODO() + numZombies uint64 + ) + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + count, err := db.CountZombieChannels(ctx, int16(ProtocolV1)) + if err != nil { + return fmt.Errorf("unable to count zombie channels: %w", + err) + } + + numZombies = uint64(count) + + return nil + }, sqldb.NoOpReset) + if err != nil { + return 0, fmt.Errorf("unable to count zombies: %w", err) + } + + return numZombies, 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 76e61fe8c..f5f6197a0 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -26,6 +26,19 @@ func (q *Queries) AddSourceNode(ctx context.Context, nodeID int64) error { return err } +const countZombieChannels = `-- name: CountZombieChannels :one +SELECT COUNT(*) +FROM zombie_channels +WHERE version = $1 +` + +func (q *Queries) CountZombieChannels(ctx context.Context, version int16) (int64, error) { + row := q.db.QueryRowContext(ctx, countZombieChannels, version) + var count int64 + err := row.Scan(&count) + return count, err +} + const createChannel = `-- name: CreateChannel :one /* ───────────────────────────────────────────── channels table queries @@ -168,6 +181,21 @@ func (q *Queries) DeleteNodeFeature(ctx context.Context, arg DeleteNodeFeaturePa return err } +const deleteZombieChannel = `-- name: DeleteZombieChannel :execresult +DELETE FROM zombie_channels +WHERE scid = $1 +AND version = $2 +` + +type DeleteZombieChannelParams struct { + Scid []byte + Version int16 +} + +func (q *Queries) DeleteZombieChannel(ctx context.Context, arg DeleteZombieChannelParams) (sql.Result, error) { + return q.db.ExecContext(ctx, deleteZombieChannel, arg.Scid, arg.Version) +} + const getChannelAndNodesBySCID = `-- name: GetChannelAndNodesBySCID :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, @@ -888,6 +916,30 @@ func (q *Queries) GetSourceNodesByVersion(ctx context.Context, version int16) ([ return items, nil } +const getZombieChannel = `-- name: GetZombieChannel :one +SELECT scid, version, node_key_1, node_key_2 +FROM zombie_channels +WHERE scid = $1 +AND version = $2 +` + +type GetZombieChannelParams struct { + Scid []byte + Version int16 +} + +func (q *Queries) GetZombieChannel(ctx context.Context, arg GetZombieChannelParams) (ZombieChannel, error) { + row := q.db.QueryRowContext(ctx, getZombieChannel, arg.Scid, arg.Version) + var i ZombieChannel + err := row.Scan( + &i.Scid, + &i.Version, + &i.NodeKey1, + &i.NodeKey2, + ) + return i, err +} + const highestSCID = `-- name: HighestSCID :one SELECT scid FROM channels @@ -1538,3 +1590,36 @@ func (q *Queries) UpsertNodeExtraType(ctx context.Context, arg UpsertNodeExtraTy _, err := q.db.ExecContext(ctx, upsertNodeExtraType, arg.NodeID, arg.Type, arg.Value) return err } + +const upsertZombieChannel = `-- name: UpsertZombieChannel :exec +/* ───────────────────────────────────────────── + zombie_channels table queries + ───────────────────────────────────────────── +*/ + +INSERT INTO zombie_channels (scid, version, node_key_1, node_key_2) +VALUES ($1, $2, $3, $4) +ON CONFLICT (scid, version) +DO UPDATE SET + -- If a conflict exists for the SCID and version pair, then we + -- update the node keys. + node_key_1 = COALESCE(EXCLUDED.node_key_1, zombie_channels.node_key_1), + node_key_2 = COALESCE(EXCLUDED.node_key_2, zombie_channels.node_key_2) +` + +type UpsertZombieChannelParams struct { + Scid []byte + Version int16 + NodeKey1 []byte + NodeKey2 []byte +} + +func (q *Queries) UpsertZombieChannel(ctx context.Context, arg UpsertZombieChannelParams) error { + _, err := q.db.ExecContext(ctx, upsertZombieChannel, + arg.Scid, + arg.Version, + arg.NodeKey1, + arg.NodeKey2, + ) + return err +} diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 8174abcd9..1796d96f2 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -13,6 +13,7 @@ import ( type Querier interface { AddSourceNode(ctx context.Context, nodeID int64) error ClearKVInvoiceHashIndex(ctx context.Context) error + CountZombieChannels(ctx context.Context, version int16) (int64, error) CreateChannel(ctx context.Context, arg CreateChannelParams) (int64, error) CreateChannelExtraType(ctx context.Context, arg CreateChannelExtraTypeParams) error DeleteCanceledInvoices(ctx context.Context) (sql.Result, error) @@ -22,6 +23,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 + 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) FetchSettledAMPSubInvoices(ctx context.Context, arg FetchSettledAMPSubInvoicesParams) ([]FetchSettledAMPSubInvoicesRow, error) @@ -54,6 +56,7 @@ type Querier interface { GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]Node, error) GetPublicV1ChannelsBySCID(ctx context.Context, arg GetPublicV1ChannelsBySCIDParams) ([]Channel, error) GetSourceNodesByVersion(ctx context.Context, version int16) ([]GetSourceNodesByVersionRow, error) + GetZombieChannel(ctx context.Context, arg GetZombieChannelParams) (ZombieChannel, error) HighestSCID(ctx context.Context, version int16) ([]byte, error) InsertAMPSubInvoice(ctx context.Context, arg InsertAMPSubInvoiceParams) error InsertAMPSubInvoiceHTLC(ctx context.Context, arg InsertAMPSubInvoiceHTLCParams) error @@ -90,6 +93,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 + UpsertZombieChannel(ctx context.Context, arg UpsertZombieChannelParams) error } var _ Querier = (*Queries)(nil) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 5560ec90d..a8a5362d2 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -460,3 +460,34 @@ WHERE cp.id = $1 OR cp.id = $2; -- name: DeleteChannelPolicyExtraTypes :exec DELETE FROM channel_policy_extra_types WHERE channel_policy_id = $1; + +/* ───────────────────────────────────────────── + zombie_channels table queries + ───────────────────────────────────────────── +*/ + +-- name: UpsertZombieChannel :exec +INSERT INTO zombie_channels (scid, version, node_key_1, node_key_2) +VALUES ($1, $2, $3, $4) +ON CONFLICT (scid, version) +DO UPDATE SET + -- If a conflict exists for the SCID and version pair, then we + -- update the node keys. + node_key_1 = COALESCE(EXCLUDED.node_key_1, zombie_channels.node_key_1), + node_key_2 = COALESCE(EXCLUDED.node_key_2, zombie_channels.node_key_2); + +-- name: DeleteZombieChannel :execresult +DELETE FROM zombie_channels +WHERE scid = $1 +AND version = $2; + +-- name: CountZombieChannels :one +SELECT COUNT(*) +FROM zombie_channels +WHERE version = $1; + +-- name: GetZombieChannel :one +SELECT * +FROM zombie_channels +WHERE scid = $1 +AND version = $2; From 2a6e6683eb36f9dc5f66170243fe8274276c47fc Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 11 Jun 2025 17:10:12 +0200 Subject: [PATCH 5/5] graph/db+sqldb: implement DeleteChannelEdges This lets us run TestGraphZombieIndex against the SQL backends. --- docs/release-notes/release-notes-0.20.0.md | 1 + graph/db/graph_test.go | 2 +- graph/db/sql_store.go | 119 ++++++++++++++++ sqldb/sqlc/graph.sql.go | 154 +++++++++++++++++++++ sqldb/sqlc/querier.go | 2 + sqldb/sqlc/queries/graph.sql | 49 +++++++ 6 files changed, 326 insertions(+), 1 deletion(-) diff --git a/docs/release-notes/release-notes-0.20.0.md b/docs/release-notes/release-notes-0.20.0.md index a6c2b0819..dab221f2e 100644 --- a/docs/release-notes/release-notes-0.20.0.md +++ b/docs/release-notes/release-notes-0.20.0.md @@ -86,6 +86,7 @@ circuit. The indices are only available for forwarding events saved after v0.20. * [4](https://github.com/lightningnetwork/lnd/pull/9931) * [5](https://github.com/lightningnetwork/lnd/pull/9935) * [6](https://github.com/lightningnetwork/lnd/pull/9936) + * [7](https://github.com/lightningnetwork/lnd/pull/9937) ## RPC Updates diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 4cce1d9af..bff2f4277 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -3811,7 +3811,7 @@ func TestGraphZombieIndex(t *testing.T) { ctx := context.Background() // We'll start by creating our test graph along with a test edge. - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) node1 := createTestVertex(t) node2 := createTestVertex(t) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 7aff00ad6..722198a66 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -88,6 +88,7 @@ type SQLQueries interface { */ CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error) GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, 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) @@ -95,6 +96,7 @@ type SQLQueries interface { ListChannelsWithPoliciesPaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesPaginatedParams) ([]sqlc.ListChannelsWithPoliciesPaginatedRow, error) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg sqlc.GetChannelsByPolicyLastUpdateRangeParams) ([]sqlc.GetChannelsByPolicyLastUpdateRangeRow, error) GetPublicV1ChannelsBySCID(ctx context.Context, arg sqlc.GetPublicV1ChannelsBySCIDParams) ([]sqlc.Channel, error) + DeleteChannel(ctx context.Context, id int64) error CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error @@ -1552,6 +1554,123 @@ func (s *SQLStore) NumZombies() (uint64, error) { return numZombies, nil } +// DeleteChannelEdges removes edges with the given channel IDs from the +// database and marks them as zombies. This ensures that we're unable to re-add +// it to our database once again. If an edge does not exist within the +// database, then ErrEdgeNotFound will be returned. If strictZombiePruning is +// true, then when we mark these edges as zombies, we'll set up the keys such +// that we require the node that failed to send the fresh update to be the one +// that resurrects the channel from its zombie state. The markZombie bool +// denotes whether to mark the channel as a zombie. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool, + chanIDs ...uint64) ([]*models.ChannelEdgeInfo, error) { + + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + + var ( + ctx = context.TODO() + deleted []*models.ChannelEdgeInfo + ) + err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { + for _, chanID := range chanIDs { + chanIDB := channelIDToBytes(chanID) + + row, err := db.GetChannelBySCIDWithPolicies( + ctx, sqlc.GetChannelBySCIDWithPoliciesParams{ + Scid: chanIDB[:], + Version: int16(ProtocolV1), + }, + ) + if errors.Is(err, sql.ErrNoRows) { + return ErrEdgeNotFound + } else if err != nil { + return fmt.Errorf("unable to fetch channel: %w", + err) + } + + node1, node2, err := buildNodeVertices( + row.Node.PubKey, row.Node_2.PubKey, + ) + 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) + } + + deleted = append(deleted, info) + + if !markZombie { + continue + } + + nodeKey1, nodeKey2 := info.NodeKey1Bytes, + info.NodeKey2Bytes + if strictZombiePruning { + var e1UpdateTime, e2UpdateTime *time.Time + if row.Policy1LastUpdate.Valid { + e1Time := time.Unix( + row.Policy1LastUpdate.Int64, 0, + ) + e1UpdateTime = &e1Time + } + if row.Policy2LastUpdate.Valid { + e2Time := time.Unix( + row.Policy2LastUpdate.Int64, 0, + ) + e2UpdateTime = &e2Time + } + + nodeKey1, nodeKey2 = makeZombiePubkeys( + info, e1UpdateTime, e2UpdateTime, + ) + } + + err = db.UpsertZombieChannel( + ctx, sqlc.UpsertZombieChannelParams{ + Version: int16(ProtocolV1), + Scid: chanIDB[:], + NodeKey1: nodeKey1[:], + NodeKey2: nodeKey2[:], + }, + ) + if err != nil { + return fmt.Errorf("unable to mark channel as "+ + "zombie: %w", err) + } + } + + return nil + }, func() { + deleted = nil + }) + if err != nil { + return nil, fmt.Errorf("unable to delete channel edges: %w", + err) + } + + for _, chanID := range chanIDs { + s.rejectCache.remove(chanID) + s.chanCache.remove(chanID) + } + + return deleted, 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 f5f6197a0..fd81114b2 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -114,6 +114,15 @@ func (q *Queries) CreateChannelExtraType(ctx context.Context, arg CreateChannelE return err } +const deleteChannel = `-- name: DeleteChannel :exec +DELETE FROM channels WHERE id = $1 +` + +func (q *Queries) DeleteChannel(ctx context.Context, id int64) error { + _, err := q.db.ExecContext(ctx, deleteChannel, id) + return err +} + const deleteChannelPolicyExtraTypes = `-- name: DeleteChannelPolicyExtraTypes :exec DELETE FROM channel_policy_extra_types WHERE channel_policy_id = $1 @@ -285,6 +294,151 @@ func (q *Queries) GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDPara return i, err } +const getChannelBySCIDWithPolicies = `-- name: GetChannelBySCIDWithPolicies :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.id, n1.version, n1.pub_key, n1.alias, n1.last_update, n1.color, n1.signature, + n2.id, n2.version, n2.pub_key, n2.alias, n2.last_update, n2.color, n2.signature, + + -- Policy 1 + cp1.id AS policy1_id, + cp1.node_id AS policy1_node_id, + cp1.version AS policy1_version, + cp1.timelock AS policy1_timelock, + cp1.fee_ppm AS policy1_fee_ppm, + cp1.base_fee_msat AS policy1_base_fee_msat, + cp1.min_htlc_msat AS policy1_min_htlc_msat, + cp1.max_htlc_msat AS policy1_max_htlc_msat, + cp1.last_update AS policy1_last_update, + cp1.disabled AS policy1_disabled, + cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat, + cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat, + cp1.signature AS policy1_signature, + + -- Policy 2 + cp2.id AS policy2_id, + cp2.node_id AS policy2_node_id, + cp2.version AS policy2_version, + cp2.timelock AS policy2_timelock, + cp2.fee_ppm AS policy2_fee_ppm, + cp2.base_fee_msat AS policy2_base_fee_msat, + cp2.min_htlc_msat AS policy2_min_htlc_msat, + cp2.max_htlc_msat AS policy2_max_htlc_msat, + cp2.last_update AS policy2_last_update, + cp2.disabled AS policy2_disabled, + cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat, + cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat, + cp2.signature AS policy2_signature + +FROM channels c + JOIN nodes n1 ON c.node_id_1 = n1.id + JOIN nodes n2 ON c.node_id_2 = n2.id + LEFT JOIN channel_policies cp1 + ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version + LEFT JOIN channel_policies cp2 + ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version +WHERE c.scid = $1 + AND c.version = $2 +` + +type GetChannelBySCIDWithPoliciesParams struct { + Scid []byte + Version int16 +} + +type GetChannelBySCIDWithPoliciesRow struct { + Channel Channel + Node Node + Node_2 Node + Policy1ID sql.NullInt64 + Policy1NodeID sql.NullInt64 + Policy1Version sql.NullInt16 + Policy1Timelock sql.NullInt32 + Policy1FeePpm sql.NullInt64 + Policy1BaseFeeMsat sql.NullInt64 + Policy1MinHtlcMsat sql.NullInt64 + Policy1MaxHtlcMsat sql.NullInt64 + Policy1LastUpdate sql.NullInt64 + Policy1Disabled sql.NullBool + Policy1InboundBaseFeeMsat sql.NullInt64 + Policy1InboundFeeRateMilliMsat sql.NullInt64 + Policy1Signature []byte + Policy2ID sql.NullInt64 + Policy2NodeID sql.NullInt64 + Policy2Version sql.NullInt16 + Policy2Timelock sql.NullInt32 + Policy2FeePpm sql.NullInt64 + Policy2BaseFeeMsat sql.NullInt64 + Policy2MinHtlcMsat sql.NullInt64 + Policy2MaxHtlcMsat sql.NullInt64 + Policy2LastUpdate sql.NullInt64 + Policy2Disabled sql.NullBool + Policy2InboundBaseFeeMsat sql.NullInt64 + Policy2InboundFeeRateMilliMsat sql.NullInt64 + Policy2Signature []byte +} + +func (q *Queries) GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChannelBySCIDWithPoliciesParams) (GetChannelBySCIDWithPoliciesRow, error) { + row := q.db.QueryRowContext(ctx, getChannelBySCIDWithPolicies, arg.Scid, arg.Version) + var i GetChannelBySCIDWithPoliciesRow + 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.Node.ID, + &i.Node.Version, + &i.Node.PubKey, + &i.Node.Alias, + &i.Node.LastUpdate, + &i.Node.Color, + &i.Node.Signature, + &i.Node_2.ID, + &i.Node_2.Version, + &i.Node_2.PubKey, + &i.Node_2.Alias, + &i.Node_2.LastUpdate, + &i.Node_2.Color, + &i.Node_2.Signature, + &i.Policy1ID, + &i.Policy1NodeID, + &i.Policy1Version, + &i.Policy1Timelock, + &i.Policy1FeePpm, + &i.Policy1BaseFeeMsat, + &i.Policy1MinHtlcMsat, + &i.Policy1MaxHtlcMsat, + &i.Policy1LastUpdate, + &i.Policy1Disabled, + &i.Policy1InboundBaseFeeMsat, + &i.Policy1InboundFeeRateMilliMsat, + &i.Policy1Signature, + &i.Policy2ID, + &i.Policy2NodeID, + &i.Policy2Version, + &i.Policy2Timelock, + &i.Policy2FeePpm, + &i.Policy2BaseFeeMsat, + &i.Policy2MinHtlcMsat, + &i.Policy2MaxHtlcMsat, + &i.Policy2LastUpdate, + &i.Policy2Disabled, + &i.Policy2InboundBaseFeeMsat, + &i.Policy2InboundFeeRateMilliMsat, + &i.Policy2Signature, + ) + return i, err +} + const getChannelFeaturesAndExtras = `-- name: GetChannelFeaturesAndExtras :many SELECT cf.channel_id, diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 1796d96f2..629e7beb9 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -17,6 +17,7 @@ type Querier interface { CreateChannel(ctx context.Context, arg CreateChannelParams) (int64, error) CreateChannelExtraType(ctx context.Context, arg CreateChannelExtraTypeParams) error DeleteCanceledInvoices(ctx context.Context) (sql.Result, error) + DeleteChannel(ctx context.Context, id int64) error DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error DeleteExtraNodeType(ctx context.Context, arg DeleteExtraNodeTypeParams) error DeleteInvoice(ctx context.Context, arg DeleteInvoiceParams) (sql.Result, error) @@ -31,6 +32,7 @@ type Querier interface { GetAMPInvoiceID(ctx context.Context, setID []byte) (int64, error) GetChannelAndNodesBySCID(ctx context.Context, arg GetChannelAndNodesBySCIDParams) (GetChannelAndNodesBySCIDRow, error) GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDParams) (Channel, error) + GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChannelBySCIDWithPoliciesParams) (GetChannelBySCIDWithPoliciesRow, error) GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]GetChannelFeaturesAndExtrasRow, error) GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetChannelPolicyByChannelAndNodeParams) (ChannelPolicy, error) GetChannelPolicyExtraTypes(ctx context.Context, arg GetChannelPolicyExtraTypesParams) ([]GetChannelPolicyExtraTypesRow, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index a8a5362d2..f315cb335 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -373,6 +373,9 @@ WHERE c.version = $1 AND c.id > $2 ORDER BY c.id LIMIT $3; +-- name: DeleteChannel :exec +DELETE FROM channels WHERE id = $1; + /* ───────────────────────────────────────────── channel_features table queries ───────────────────────────────────────────── @@ -434,6 +437,52 @@ WHERE channel_id = $1 AND node_id = $2 AND version = $3; +-- name: GetChannelBySCIDWithPolicies :one +SELECT + sqlc.embed(c), + sqlc.embed(n1), + sqlc.embed(n2), + + -- Policy 1 + cp1.id AS policy1_id, + cp1.node_id AS policy1_node_id, + cp1.version AS policy1_version, + cp1.timelock AS policy1_timelock, + cp1.fee_ppm AS policy1_fee_ppm, + cp1.base_fee_msat AS policy1_base_fee_msat, + cp1.min_htlc_msat AS policy1_min_htlc_msat, + cp1.max_htlc_msat AS policy1_max_htlc_msat, + cp1.last_update AS policy1_last_update, + cp1.disabled AS policy1_disabled, + cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat, + cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat, + cp1.signature AS policy1_signature, + + -- Policy 2 + cp2.id AS policy2_id, + cp2.node_id AS policy2_node_id, + cp2.version AS policy2_version, + cp2.timelock AS policy2_timelock, + cp2.fee_ppm AS policy2_fee_ppm, + cp2.base_fee_msat AS policy2_base_fee_msat, + cp2.min_htlc_msat AS policy2_min_htlc_msat, + cp2.max_htlc_msat AS policy2_max_htlc_msat, + cp2.last_update AS policy2_last_update, + cp2.disabled AS policy2_disabled, + cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat, + cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat, + cp2.signature AS policy2_signature + +FROM channels c + JOIN nodes n1 ON c.node_id_1 = n1.id + JOIN nodes n2 ON c.node_id_2 = n2.id + LEFT JOIN channel_policies cp1 + ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version + LEFT JOIN channel_policies cp2 + ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version +WHERE c.scid = @scid + AND c.version = @version; + /* ───────────────────────────────────────────── channel_policy_extra_types table queries ─────────────────────────────────────────────