staticaddr: migrate swap hashes to deposits

This commit is contained in:
Slyghtning 2025-07-30 12:00:03 +02:00
parent 35901d1247
commit 6e7441ba8d
No known key found for this signature in database
GPG key ID: F82D456EA023C9BF
12 changed files with 499 additions and 4 deletions

2
go.mod
View file

@ -16,6 +16,7 @@ require (
github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0
github.com/jackc/pgconn v1.14.3
github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438
github.com/jackc/pgx/v5 v5.6.0
github.com/jessevdk/go-flags v1.4.0
github.com/lib/pq v1.10.9
github.com/lightninglabs/aperture v0.3.13-beta
@ -104,7 +105,6 @@ require (
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
github.com/jackc/pgtype v1.14.0 // indirect
github.com/jackc/pgx/v4 v4.18.2 // indirect
github.com/jackc/pgx/v5 v5.6.0 // indirect
github.com/jackc/puddle v1.3.0 // indirect
github.com/jackc/puddle/v2 v2.2.1 // indirect
github.com/jackpal/gateway v1.0.5 // indirect

View file

@ -626,6 +626,16 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
clock.NewDefaultClock(), d.lnd.ChainParams,
)
// Run the deposit swap hash migration.
err = loopin.MigrateDepositSwapHash(
d.mainCtx, swapDb, depositStore, staticAddressLoopInStore,
)
if err != nil {
errorf("Deposit swap hash migration failed: %v", err)
return err
}
staticLoopInManager = loopin.NewManager(&loopin.Config{
Server: staticAddressClient,
QuoteGetter: swapClient.Server,

View file

@ -0,0 +1 @@
ALTER TABLE deposits DROP COLUMN swap_hash;

View file

@ -0,0 +1 @@
ALTER TABLE deposits ADD swap_hash BLOB;

View file

@ -19,6 +19,7 @@ type Deposit struct {
TimeoutSweepPkScript []byte
ExpirySweepTxid []byte
FinalizedWithdrawalTx sql.NullString
SwapHash []byte
}
type DepositUpdate struct {

View file

@ -19,6 +19,7 @@ type Querier interface {
CreateWithdrawal(ctx context.Context, arg CreateWithdrawalParams) error
CreateWithdrawalDeposit(ctx context.Context, arg CreateWithdrawalDepositParams) error
DepositForOutpoint(ctx context.Context, arg DepositForOutpointParams) (Deposit, error)
DepositIDsForSwapHash(ctx context.Context, swapHash []byte) ([][]byte, error)
FetchLiquidityParams(ctx context.Context) ([]byte, error)
GetAllWithdrawals(ctx context.Context) ([]Withdrawal, error)
GetBatchSweeps(ctx context.Context, batchID int32) ([]Sweep, error)
@ -62,7 +63,9 @@ type Querier interface {
InsertSwap(ctx context.Context, arg InsertSwapParams) error
InsertSwapUpdate(ctx context.Context, arg InsertSwapUpdateParams) error
IsStored(ctx context.Context, swapHash []byte) (bool, error)
MapDepositToSwap(ctx context.Context, arg MapDepositToSwapParams) error
OverrideSwapCosts(ctx context.Context, arg OverrideSwapCostsParams) error
SwapHashForDepositID(ctx context.Context, depositID []byte) ([]byte, error)
UpdateBatch(ctx context.Context, arg UpdateBatchParams) error
UpdateDeposit(ctx context.Context, arg UpdateDepositParams) error
UpdateInstantOut(ctx context.Context, arg UpdateInstantOutParams) error

View file

@ -93,3 +93,30 @@ SELECT EXISTS (
FROM static_address_swaps
WHERE swap_hash = $1
);
-- name: MapDepositToSwap :exec
UPDATE
deposits
SET
swap_hash = $2
WHERE
deposit_id = $1;
-- name: SwapHashForDepositID :one
SELECT
swap_hash
FROM
deposits
WHERE
deposit_id = $1;
-- name: DepositIDsForSwapHash :many
SELECT
deposit_id
FROM
deposits
WHERE
swap_hash = $1;

View file

@ -13,7 +13,7 @@ import (
const allDeposits = `-- name: AllDeposits :many
SELECT
id, deposit_id, tx_hash, out_index, amount, confirmation_height, timeout_sweep_pk_script, expiry_sweep_txid, finalized_withdrawal_tx
id, deposit_id, tx_hash, out_index, amount, confirmation_height, timeout_sweep_pk_script, expiry_sweep_txid, finalized_withdrawal_tx, swap_hash
FROM
deposits
ORDER BY
@ -39,6 +39,7 @@ func (q *Queries) AllDeposits(ctx context.Context) ([]Deposit, error) {
&i.TimeoutSweepPkScript,
&i.ExpirySweepTxid,
&i.FinalizedWithdrawalTx,
&i.SwapHash,
); err != nil {
return nil, err
}
@ -102,7 +103,7 @@ func (q *Queries) CreateDeposit(ctx context.Context, arg CreateDepositParams) er
const depositForOutpoint = `-- name: DepositForOutpoint :one
SELECT
id, deposit_id, tx_hash, out_index, amount, confirmation_height, timeout_sweep_pk_script, expiry_sweep_txid, finalized_withdrawal_tx
id, deposit_id, tx_hash, out_index, amount, confirmation_height, timeout_sweep_pk_script, expiry_sweep_txid, finalized_withdrawal_tx, swap_hash
FROM
deposits
WHERE
@ -129,13 +130,14 @@ func (q *Queries) DepositForOutpoint(ctx context.Context, arg DepositForOutpoint
&i.TimeoutSweepPkScript,
&i.ExpirySweepTxid,
&i.FinalizedWithdrawalTx,
&i.SwapHash,
)
return i, err
}
const getDeposit = `-- name: GetDeposit :one
SELECT
id, deposit_id, tx_hash, out_index, amount, confirmation_height, timeout_sweep_pk_script, expiry_sweep_txid, finalized_withdrawal_tx
id, deposit_id, tx_hash, out_index, amount, confirmation_height, timeout_sweep_pk_script, expiry_sweep_txid, finalized_withdrawal_tx, swap_hash
FROM
deposits
WHERE
@ -155,6 +157,7 @@ func (q *Queries) GetDeposit(ctx context.Context, depositID []byte) (Deposit, er
&i.TimeoutSweepPkScript,
&i.ExpirySweepTxid,
&i.FinalizedWithdrawalTx,
&i.SwapHash,
)
return i, err
}

View file

@ -11,6 +11,38 @@ import (
"time"
)
const depositIDsForSwapHash = `-- name: DepositIDsForSwapHash :many
SELECT
deposit_id
FROM
deposits
WHERE
swap_hash = $1
`
func (q *Queries) DepositIDsForSwapHash(ctx context.Context, swapHash []byte) ([][]byte, error) {
rows, err := q.db.QueryContext(ctx, depositIDsForSwapHash, swapHash)
if err != nil {
return nil, err
}
defer rows.Close()
var items [][]byte
for rows.Next() {
var deposit_id []byte
if err := rows.Scan(&deposit_id); err != nil {
return nil, err
}
items = append(items, deposit_id)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const getLoopInSwapUpdates = `-- name: GetLoopInSwapUpdates :many
SELECT
static_address_swap_updates.id, static_address_swap_updates.swap_hash, static_address_swap_updates.update_state, static_address_swap_updates.update_timestamp
@ -328,6 +360,41 @@ func (q *Queries) IsStored(ctx context.Context, swapHash []byte) (bool, error) {
return exists, err
}
const mapDepositToSwap = `-- name: MapDepositToSwap :exec
UPDATE
deposits
SET
swap_hash = $2
WHERE
deposit_id = $1
`
type MapDepositToSwapParams struct {
DepositID []byte
SwapHash []byte
}
func (q *Queries) MapDepositToSwap(ctx context.Context, arg MapDepositToSwapParams) error {
_, err := q.db.ExecContext(ctx, mapDepositToSwap, arg.DepositID, arg.SwapHash)
return err
}
const swapHashForDepositID = `-- name: SwapHashForDepositID :one
SELECT
swap_hash
FROM
deposits
WHERE
deposit_id = $1
`
func (q *Queries) SwapHashForDepositID(ctx context.Context, depositID []byte) ([]byte, error) {
row := q.db.QueryRowContext(ctx, swapHashForDepositID, depositID)
var swap_hash []byte
err := row.Scan(&swap_hash)
return swap_hash, err
}
const updateStaticAddressLoopIn = `-- name: UpdateStaticAddressLoopIn :exec
UPDATE static_address_swaps
SET

View file

@ -0,0 +1,88 @@
package loopin
import (
"context"
"fmt"
"time"
"github.com/lightninglabs/loop/loopdb"
"github.com/lightninglabs/loop/staticaddr/deposit"
"github.com/lightningnetwork/lnd/lntypes"
)
const (
// depositSwapHashMigrationID is the identifier for the deposit swap
// hash migration.
depositSwapHashMigrationID = "deposit_swap_hash"
)
// MigrateDepositSwapHash will retrieve the comma separated deposit list of
// past and pending swaps and map them to the swap hash in the deposits table.
func MigrateDepositSwapHash(ctx context.Context, db loopdb.SwapStore,
depositStore *deposit.SqlStore, swapStore *SqlStore) error {
migrationDone, err := db.HasMigration(
ctx, depositSwapHashMigrationID,
)
if err != nil {
return fmt.Errorf("unable to check migration status: %w", err)
}
if migrationDone {
log.Infof("Deposit swap hash migration already done, " +
"skipping")
return nil
}
log.Infof("Starting deposit swap hash migration")
startTs := time.Now()
defer func() {
log.Infof("Finished deposit swap hash migration in %v",
time.Since(startTs))
}()
// First we'll fetch all past loop in swaps from the database.
swaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(
ctx, AllStates,
)
if err != nil {
return err
}
// Now we'll map each deposit of a swap to its respective swap hash.
depositsToSwapHashes := make(map[deposit.ID]lntypes.Hash)
for _, swap := range swaps {
for _, outpoint := range swap.DepositOutpoints {
deposit, err := depositStore.DepositForOutpoint(
ctx, outpoint,
)
if err != nil {
return fmt.Errorf("unable to fetch deposit "+
"for outpoint %s: %w", outpoint, err)
}
if deposit == nil {
return fmt.Errorf("deposit for outpoint %s "+
"not found", outpoint)
}
if _, ok := depositsToSwapHashes[deposit.ID]; !ok {
depositsToSwapHashes[deposit.ID] = swap.SwapHash
} else {
log.Warnf("Duplicate deposit ID %s found for "+
"outpoint %s, skipping",
deposit.ID, outpoint)
}
}
}
log.Infof("Batch-mapping %d deposits to swap hashes",
len(depositsToSwapHashes))
err = swapStore.BatchMapDepositsToSwapHashes(ctx, depositsToSwapHashes)
if err != nil {
return err
}
// Finally mark the migration as done.
return db.SetMigration(ctx, depositSwapHashMigrationID)
}

View file

@ -0,0 +1,195 @@
package loopin
import (
"context"
"strings"
"testing"
"time"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/lightninglabs/loop/loopdb"
"github.com/lightninglabs/loop/loopdb/sqlc"
"github.com/lightninglabs/loop/staticaddr/deposit"
"github.com/lightninglabs/loop/test"
"github.com/lightningnetwork/lnd/clock"
"github.com/lightningnetwork/lnd/lntypes"
"github.com/stretchr/testify/require"
)
const (
P2wkhAddr = "bcrt1qq68r6ff4k4pjx39efs44gcyccf7unqnu5qtjjz"
)
// TestDepositSwapHashMigration tests deposit to swap hash migration.
func TestDepositSwapHashMigration(t *testing.T) {
// Set up test context objects.
ctxb := context.Background()
testDb := loopdb.NewTestDB(t)
testClock := clock.NewTestClock(time.Now())
defer testDb.Close()
db := loopdb.NewStoreMock(t)
depositStore := deposit.NewSqlStore(testDb.BaseDB)
swapStore := NewSqlStore(
loopdb.NewTypedStore[Querier](testDb), testClock,
&chaincfg.RegressionNetParams,
)
newID := func() deposit.ID {
did, err := deposit.GetRandomDepositID()
require.NoError(t, err)
return did
}
d1, d2 := &deposit.Deposit{
ID: newID(),
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{0x1a, 0x2b, 0x3c, 0x4d},
Index: 0,
},
Value: btcutil.Amount(100_000),
TimeOutSweepPkScript: []byte{
0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x41,
},
},
&deposit.Deposit{
ID: newID(),
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{0x2a, 0x2b, 0x3c, 0x4e},
Index: 1,
},
Value: btcutil.Amount(200_000),
TimeOutSweepPkScript: []byte{
0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4d,
},
}
err := depositStore.CreateDeposit(ctxb, d1)
require.NoError(t, err)
err = depositStore.CreateDeposit(ctxb, d2)
require.NoError(t, err)
outpoints := []string{
d1.OutPoint.String(),
d2.OutPoint.String(),
}
_, clientPubKey := test.CreateKey(1)
_, serverPubKey := test.CreateKey(2)
addr, err := btcutil.DecodeAddress(P2wkhAddr, nil)
require.NoError(t, err)
swapHash := lntypes.Hash{0x1, 0x2, 0x3, 0x4}
loopIn := StaticAddressLoopIn{
SwapHash: swapHash,
DepositOutpoints: outpoints,
ClientPubkey: clientPubKey,
ServerPubkey: serverPubKey,
HtlcTimeoutSweepAddress: addr,
}
loopIn.SetState(Succeeded)
// Insert the swap without the deposit mapping.
err = swapStore.baseDB.ExecTx(ctxb, loopdb.NewSqlWriteOpts(),
func(q Querier) error {
swapArgs := sqlc.InsertSwapParams{
SwapHash: loopIn.SwapHash[:],
Preimage: loopIn.SwapPreimage[:],
InitiationTime: loopIn.InitiationTime,
AmountRequested: int64(loopIn.TotalDepositAmount()),
CltvExpiry: loopIn.HtlcCltvExpiry,
MaxSwapFee: int64(loopIn.MaxSwapFee),
InitiationHeight: int32(loopIn.InitiationHeight),
ProtocolVersion: int32(loopIn.ProtocolVersion),
Label: loopIn.Label,
}
htlcKeyArgs := sqlc.InsertHtlcKeysParams{
SwapHash: loopIn.SwapHash[:],
SenderScriptPubkey: loopIn.ClientPubkey.SerializeCompressed(),
ReceiverScriptPubkey: loopIn.ServerPubkey.SerializeCompressed(),
ClientKeyFamily: int32(loopIn.HtlcKeyLocator.Family),
ClientKeyIndex: int32(loopIn.HtlcKeyLocator.Index),
}
// Sanity check, if any of the outpoints contain the outpoint separator.
// If so, we reject the loop-in to prevent potential issues with
// parsing.
for _, outpoint := range loopIn.DepositOutpoints {
if strings.Contains(outpoint, outpointSeparator) {
return ErrInvalidOutpoint
}
}
joinedOutpoints := strings.Join(
loopIn.DepositOutpoints, outpointSeparator,
)
staticAddressLoopInParams := sqlc.InsertStaticAddressLoopInParams{
SwapHash: loopIn.SwapHash[:],
SwapInvoice: loopIn.SwapInvoice,
LastHop: loopIn.LastHop,
QuotedSwapFeeSatoshis: int64(loopIn.QuotedSwapFee),
HtlcTimeoutSweepAddress: loopIn.HtlcTimeoutSweepAddress.String(),
HtlcTxFeeRateSatKw: int64(loopIn.HtlcTxFeeRate),
DepositOutpoints: joinedOutpoints,
PaymentTimeoutSeconds: int32(loopIn.PaymentTimeoutSeconds),
}
updateArgs := sqlc.InsertStaticAddressMetaUpdateParams{
SwapHash: loopIn.SwapHash[:],
UpdateTimestamp: testClock.Now(),
UpdateState: string(loopIn.GetState()),
}
err := q.InsertSwap(ctxb, swapArgs)
if err != nil {
return err
}
err = q.InsertHtlcKeys(ctxb, htlcKeyArgs)
if err != nil {
return err
}
err = q.InsertStaticAddressLoopIn(
ctxb, staticAddressLoopInParams,
)
if err != nil {
return err
}
return q.InsertStaticAddressMetaUpdate(ctxb, updateArgs)
},
)
require.NoError(t, err)
depositIDs, err := swapStore.DepositIDsForSwapHash(ctxb, swapHash)
require.NoError(t, err)
require.Len(t, depositIDs, 0)
swapHashes, err := swapStore.SwapHashesForDepositIDs(
ctxb, []deposit.ID{d1.ID, d2.ID},
)
require.NoError(t, err)
require.Len(t, swapHashes, 0)
err = MigrateDepositSwapHash(ctxb, db, depositStore, swapStore)
require.NoError(t, err)
depositIDs, err = swapStore.DepositIDsForSwapHash(ctxb, swapHash)
require.NoError(t, err)
require.Len(t, depositIDs, 2)
require.Contains(t, depositIDs, d1.ID)
require.Contains(t, depositIDs, d2.ID)
swapHashes, err = swapStore.SwapHashesForDepositIDs(
ctxb, []deposit.ID{d1.ID, d2.ID},
)
require.NoError(t, err)
require.Len(t, swapHashes, 1)
require.Len(t, swapHashes[swapHash], 2)
require.Contains(t, swapHashes[swapHash], d1.ID)
require.Contains(t, swapHashes[swapHash], d2.ID)
}

View file

@ -10,9 +10,11 @@ import (
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/jackc/pgx/v5"
"github.com/lightninglabs/loop/fsm"
"github.com/lightninglabs/loop/loopdb"
"github.com/lightninglabs/loop/loopdb/sqlc"
"github.com/lightninglabs/loop/staticaddr/deposit"
"github.com/lightninglabs/loop/staticaddr/version"
"github.com/lightningnetwork/lnd/clock"
"github.com/lightningnetwork/lnd/keychain"
@ -68,6 +70,20 @@ type Querier interface {
// IsStored returns true if a swap with the given hash is stored in the
// database, false otherwise.
IsStored(ctx context.Context, swapHash []byte) (bool, error)
// MapDepositToSwap maps a deposit to a swap in the database.
MapDepositToSwap(ctx context.Context,
arg sqlc.MapDepositToSwapParams) error
// SwapHashForDepositID retrieves the swap hash for the given deposit
// ID.
SwapHashForDepositID(ctx context.Context,
depositID []byte) ([]byte, error)
// DepositIDsForSwapHash retrieves all deposit IDs for a given swap
// hash.
DepositIDsForSwapHash(ctx context.Context,
swapHash []byte) ([][]byte, error)
}
// BaseDB is the interface that contains all the queries generated by sqlc for
@ -306,6 +322,89 @@ func (s *SqlStore) IsStored(ctx context.Context, swapHash lntypes.Hash) (bool,
return s.baseDB.IsStored(ctx, swapHash[:])
}
// BatchMapDepositsToSwapHashes maps multiple deposits to their respective swap
// hashes in a single transaction.
func (s *SqlStore) BatchMapDepositsToSwapHashes(ctx context.Context,
depositsToHashes map[deposit.ID]lntypes.Hash) error {
return s.baseDB.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
func(q Querier) error {
for deposit, swapHash := range depositsToHashes {
err := q.MapDepositToSwap(
ctx, sqlc.MapDepositToSwapParams{
DepositID: deposit[:],
SwapHash: swapHash[:],
},
)
if err != nil {
return err
}
}
return nil
})
}
// SwapHashesForDepositIDs retrieves the swap hashes for the given deposit IDs.
func (s *SqlStore) SwapHashesForDepositIDs(ctx context.Context,
depositIDs []deposit.ID) (map[lntypes.Hash][]deposit.ID, error) {
swapHashes := make(map[lntypes.Hash][]deposit.ID)
for _, id := range depositIDs {
swapHash, err := s.baseDB.SwapHashForDepositID(ctx, id[:])
if err != nil {
if errors.Is(err, sql.ErrNoRows) ||
errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
return nil, err
}
if swapHash == nil {
return nil, nil
}
if len(swapHash) != lntypes.HashSize {
return nil, errors.New("invalid swap hash length")
}
swapHashParsed, err := lntypes.MakeHash(swapHash)
if err != nil {
return nil, err
}
// Place the deposit ID in the map under the
// corresponding swap hash.
swapHashes[swapHashParsed] = append(
swapHashes[swapHashParsed], id,
)
}
return swapHashes, nil
}
// DepositIDsForSwapHash retrieves all deposit IDs for a given swap hash.
func (s *SqlStore) DepositIDsForSwapHash(ctx context.Context,
swapHash lntypes.Hash) ([]deposit.ID, error) {
byteIDs, err := s.baseDB.DepositIDsForSwapHash(ctx, swapHash[:])
if err != nil {
return nil, err
}
depositIDs := make([]deposit.ID, len(byteIDs))
for i, id := range byteIDs {
if len(id) != deposit.IdLength {
return nil, errors.New("invalid deposit ID length")
}
copy(depositIDs[i][:], id)
}
return depositIDs, nil
}
// toStaticAddressLoopIn converts sql rows to an instant out struct.
func toStaticAddressLoopIn(_ context.Context, network *chaincfg.Params,
row sqlc.GetStaticAddressLoopInSwapRow,