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/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 diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 3710efa8e..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 @@ -108,6 +110,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 @@ -1288,13 +1298,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. @@ -1393,6 +1400,277 @@ 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 +} + +// 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 @@ -1638,9 +1916,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 +2611,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 +3196,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 +} diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 76e61fe8c..fd81114b2 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 @@ -101,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 @@ -168,6 +190,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, @@ -257,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, @@ -888,6 +1070,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 +1744,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/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 +} diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 8174abcd9..629e7beb9 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -13,15 +13,18 @@ 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) + 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) 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) @@ -29,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) @@ -54,6 +58,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 +95,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..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 ───────────────────────────────────────────── @@ -460,3 +509,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;