From 325545143c70c8ff072bf095d2711a5280767bc6 Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 15 May 2026 14:54:51 +0200 Subject: [PATCH] staticaddr/loopin: persist risk decisions Store the server's static loop-in confirmation-risk decision and the time it was received. This lets recovered swaps reconstruct whether payment waiting had already started and how much of the payment timeout remains. Wire notification handling to persist accepted and rejected decisions before caching and forwarding them. If the swap row is not present yet, the notification is still cached so the per-swap waiter can replay and store the decision later. Recover accepted decisions by starting the payment deadline from the persisted decision time, and recover rejected decisions by canceling the invoice and failing the swap instead of waiting forever. --- loopd/daemon.go | 26 +- ...00021_static_loopin_risk_decision.down.sql | 3 + .../000021_static_loopin_risk_decision.up.sql | 9 + loopdb/sqlc/models.go | 26 +- loopdb/sqlc/querier.go | 1 + loopdb/sqlc/queries/static_address_loopin.sql | 9 +- loopdb/sqlc/static_address_loopin.sql.go | 152 ++++--- notifications/manager.go | 61 ++- notifications/manager_test.go | 101 +++- staticaddr/loopin/actions.go | 104 ++++- staticaddr/loopin/actions_test.go | 430 ++++++++++++++++++ staticaddr/loopin/interface.go | 5 + staticaddr/loopin/loopin.go | 26 ++ staticaddr/loopin/manager_test.go | 7 + staticaddr/loopin/sql_store.go | 52 +++ staticaddr/loopin/sql_store_test.go | 25 + 16 files changed, 934 insertions(+), 103 deletions(-) create mode 100644 loopdb/sqlc/migrations/000021_static_loopin_risk_decision.down.sql create mode 100644 loopdb/sqlc/migrations/000021_static_loopin_risk_decision.up.sql diff --git a/loopd/daemon.go b/loopd/daemon.go index 5abc78bd..5aeea554 100644 --- a/loopd/daemon.go +++ b/loopd/daemon.go @@ -555,10 +555,30 @@ func (d *Daemon) initialize(withMacaroonService bool) error { } } + // Static address loop-in store setup is needed by the notification + // manager so confirmation-risk decisions are durable before fan-out. + staticAddressLoopInStore := loopin.NewSqlStore( + loopdb.NewTypedStore[loopin.Querier](baseDb), + clock.NewDefaultClock(), d.lnd.ChainParams, + ) + // Start the notification manager. notificationCfg := ¬ifications.Config{ Client: loop_swaprpc.NewSwapServerClient(swapClient.Conn), CurrentToken: swapClient.L402Store.CurrentToken, + PersistStaticLoopInRiskDecision: func(ctx context.Context, + swapHash lntypes.Hash, accepted bool) error { + + decision := loopin.ConfirmationRiskDecisionRejected + if accepted { + decision = loopin.ConfirmationRiskDecisionAccepted + } + + return staticAddressLoopInStore. + RecordStaticAddressRiskDecision( + ctx, swapHash, decision, + ) + }, } notificationManager := notifications.NewManager(notificationCfg) @@ -663,12 +683,6 @@ func (d *Daemon) initialize(withMacaroonService bool) error { } openChannelManager = openchannel.NewManager(openChannelCfg) - // Static address loop-in manager setup. - staticAddressLoopInStore := loopin.NewSqlStore( - loopdb.NewTypedStore[loopin.Querier](baseDb), - clock.NewDefaultClock(), d.lnd.ChainParams, - ) - // Run the deposit swap hash migration. err = loopin.MigrateDepositSwapHash( d.mainCtx, swapDb, depositStore, staticAddressLoopInStore, diff --git a/loopdb/sqlc/migrations/000021_static_loopin_risk_decision.down.sql b/loopdb/sqlc/migrations/000021_static_loopin_risk_decision.down.sql new file mode 100644 index 00000000..3a7313c1 --- /dev/null +++ b/loopdb/sqlc/migrations/000021_static_loopin_risk_decision.down.sql @@ -0,0 +1,3 @@ +-- Drop confirmation-risk decision fields from static address loop-ins. +ALTER TABLE static_address_swaps DROP COLUMN confirmation_risk_decision; +ALTER TABLE static_address_swaps DROP COLUMN confirmation_risk_decision_time; diff --git a/loopdb/sqlc/migrations/000021_static_loopin_risk_decision.up.sql b/loopdb/sqlc/migrations/000021_static_loopin_risk_decision.up.sql new file mode 100644 index 00000000..722fc57a --- /dev/null +++ b/loopdb/sqlc/migrations/000021_static_loopin_risk_decision.up.sql @@ -0,0 +1,9 @@ +-- confirmation_risk_decision records the server's confirmation-risk decision +-- for a static address loop-in. The empty string means no decision has been +-- received yet. +ALTER TABLE static_address_swaps ADD COLUMN confirmation_risk_decision TEXT NOT NULL DEFAULT ''; + +-- confirmation_risk_decision_time records when loopd received and persisted +-- the server's decision, so payment deadlines can be reconstructed after +-- restart. +ALTER TABLE static_address_swaps ADD COLUMN confirmation_risk_decision_time TIMESTAMP; diff --git a/loopdb/sqlc/models.go b/loopdb/sqlc/models.go index 34d8a277..78a75d04 100644 --- a/loopdb/sqlc/models.go +++ b/loopdb/sqlc/models.go @@ -137,18 +137,20 @@ type StaticAddress struct { } type StaticAddressSwap struct { - ID int32 - SwapHash []byte - SwapInvoice string - LastHop []byte - PaymentTimeoutSeconds int32 - QuotedSwapFeeSatoshis int64 - DepositOutpoints string - HtlcTxFeeRateSatKw int64 - HtlcTimeoutSweepTxID sql.NullString - HtlcTimeoutSweepAddress string - SelectedAmount int64 - Fast bool + ID int32 + SwapHash []byte + SwapInvoice string + LastHop []byte + PaymentTimeoutSeconds int32 + QuotedSwapFeeSatoshis int64 + DepositOutpoints string + HtlcTxFeeRateSatKw int64 + HtlcTimeoutSweepTxID sql.NullString + HtlcTimeoutSweepAddress string + SelectedAmount int64 + Fast bool + ConfirmationRiskDecision string + ConfirmationRiskDecisionTime sql.NullTime } type StaticAddressSwapUpdate struct { diff --git a/loopdb/sqlc/querier.go b/loopdb/sqlc/querier.go index 2d8dc137..ba3c35eb 100644 --- a/loopdb/sqlc/querier.go +++ b/loopdb/sqlc/querier.go @@ -67,6 +67,7 @@ type Querier interface { MapDepositToSwap(ctx context.Context, arg MapDepositToSwapParams) error OverrideSelectedSwapAmount(ctx context.Context, arg OverrideSelectedSwapAmountParams) error OverrideSwapCosts(ctx context.Context, arg OverrideSwapCostsParams) error + RecordStaticAddressRiskDecision(ctx context.Context, arg RecordStaticAddressRiskDecisionParams) error SwapHashForDepositID(ctx context.Context, depositID []byte) ([]byte, error) UpdateBatch(ctx context.Context, arg UpdateBatchParams) error UpdateDeposit(ctx context.Context, arg UpdateDepositParams) error diff --git a/loopdb/sqlc/queries/static_address_loopin.sql b/loopdb/sqlc/queries/static_address_loopin.sql index 7ee326a2..21de5bba 100644 --- a/loopdb/sqlc/queries/static_address_loopin.sql +++ b/loopdb/sqlc/queries/static_address_loopin.sql @@ -33,6 +33,14 @@ SET WHERE swap_hash = $1; +-- name: RecordStaticAddressRiskDecision :exec +UPDATE static_address_swaps +SET + confirmation_risk_decision = $2, + confirmation_risk_decision_time = $3 +WHERE + swap_hash = $1; + -- name: InsertStaticAddressMetaUpdate :exec INSERT INTO static_address_swap_updates ( swap_hash, @@ -150,4 +158,3 @@ WHERE - diff --git a/loopdb/sqlc/static_address_loopin.sql.go b/loopdb/sqlc/static_address_loopin.sql.go index 87949cb3..3e440c49 100644 --- a/loopdb/sqlc/static_address_loopin.sql.go +++ b/loopdb/sqlc/static_address_loopin.sql.go @@ -153,7 +153,7 @@ func (q *Queries) GetLoopInSwapUpdates(ctx context.Context, swapHash []byte) ([] const getStaticAddressLoopInSwap = `-- name: GetStaticAddressLoopInSwap :one SELECT swaps.id, swaps.swap_hash, swaps.preimage, swaps.initiation_time, swaps.amount_requested, swaps.cltv_expiry, swaps.max_miner_fee, swaps.max_swap_fee, swaps.initiation_height, swaps.protocol_version, swaps.label, - static_address_swaps.id, static_address_swaps.swap_hash, static_address_swaps.swap_invoice, static_address_swaps.last_hop, static_address_swaps.payment_timeout_seconds, static_address_swaps.quoted_swap_fee_satoshis, static_address_swaps.deposit_outpoints, static_address_swaps.htlc_tx_fee_rate_sat_kw, static_address_swaps.htlc_timeout_sweep_tx_id, static_address_swaps.htlc_timeout_sweep_address, static_address_swaps.selected_amount, static_address_swaps.fast, + static_address_swaps.id, static_address_swaps.swap_hash, static_address_swaps.swap_invoice, static_address_swaps.last_hop, static_address_swaps.payment_timeout_seconds, static_address_swaps.quoted_swap_fee_satoshis, static_address_swaps.deposit_outpoints, static_address_swaps.htlc_tx_fee_rate_sat_kw, static_address_swaps.htlc_timeout_sweep_tx_id, static_address_swaps.htlc_timeout_sweep_address, static_address_swaps.selected_amount, static_address_swaps.fast, static_address_swaps.confirmation_risk_decision, static_address_swaps.confirmation_risk_decision_time, htlc_keys.swap_hash, htlc_keys.sender_script_pubkey, htlc_keys.receiver_script_pubkey, htlc_keys.sender_internal_pubkey, htlc_keys.receiver_internal_pubkey, htlc_keys.client_key_family, htlc_keys.client_key_index FROM swaps @@ -166,36 +166,38 @@ WHERE ` type GetStaticAddressLoopInSwapRow struct { - ID int32 - SwapHash []byte - Preimage []byte - InitiationTime time.Time - AmountRequested int64 - CltvExpiry int32 - MaxMinerFee int64 - MaxSwapFee int64 - InitiationHeight int32 - ProtocolVersion int32 - Label string - ID_2 int32 - SwapHash_2 []byte - SwapInvoice string - LastHop []byte - PaymentTimeoutSeconds int32 - QuotedSwapFeeSatoshis int64 - DepositOutpoints string - HtlcTxFeeRateSatKw int64 - HtlcTimeoutSweepTxID sql.NullString - HtlcTimeoutSweepAddress string - SelectedAmount int64 - Fast bool - SwapHash_3 []byte - SenderScriptPubkey []byte - ReceiverScriptPubkey []byte - SenderInternalPubkey []byte - ReceiverInternalPubkey []byte - ClientKeyFamily int32 - ClientKeyIndex int32 + ID int32 + SwapHash []byte + Preimage []byte + InitiationTime time.Time + AmountRequested int64 + CltvExpiry int32 + MaxMinerFee int64 + MaxSwapFee int64 + InitiationHeight int32 + ProtocolVersion int32 + Label string + ID_2 int32 + SwapHash_2 []byte + SwapInvoice string + LastHop []byte + PaymentTimeoutSeconds int32 + QuotedSwapFeeSatoshis int64 + DepositOutpoints string + HtlcTxFeeRateSatKw int64 + HtlcTimeoutSweepTxID sql.NullString + HtlcTimeoutSweepAddress string + SelectedAmount int64 + Fast bool + ConfirmationRiskDecision string + ConfirmationRiskDecisionTime sql.NullTime + SwapHash_3 []byte + SenderScriptPubkey []byte + ReceiverScriptPubkey []byte + SenderInternalPubkey []byte + ReceiverInternalPubkey []byte + ClientKeyFamily int32 + ClientKeyIndex int32 } func (q *Queries) GetStaticAddressLoopInSwap(ctx context.Context, swapHash []byte) (GetStaticAddressLoopInSwapRow, error) { @@ -225,6 +227,8 @@ func (q *Queries) GetStaticAddressLoopInSwap(ctx context.Context, swapHash []byt &i.HtlcTimeoutSweepAddress, &i.SelectedAmount, &i.Fast, + &i.ConfirmationRiskDecision, + &i.ConfirmationRiskDecisionTime, &i.SwapHash_3, &i.SenderScriptPubkey, &i.ReceiverScriptPubkey, @@ -239,7 +243,7 @@ func (q *Queries) GetStaticAddressLoopInSwap(ctx context.Context, swapHash []byt const getStaticAddressLoopInSwapsByStates = `-- name: GetStaticAddressLoopInSwapsByStates :many SELECT swaps.id, swaps.swap_hash, swaps.preimage, swaps.initiation_time, swaps.amount_requested, swaps.cltv_expiry, swaps.max_miner_fee, swaps.max_swap_fee, swaps.initiation_height, swaps.protocol_version, swaps.label, - static_address_swaps.id, static_address_swaps.swap_hash, static_address_swaps.swap_invoice, static_address_swaps.last_hop, static_address_swaps.payment_timeout_seconds, static_address_swaps.quoted_swap_fee_satoshis, static_address_swaps.deposit_outpoints, static_address_swaps.htlc_tx_fee_rate_sat_kw, static_address_swaps.htlc_timeout_sweep_tx_id, static_address_swaps.htlc_timeout_sweep_address, static_address_swaps.selected_amount, static_address_swaps.fast, + static_address_swaps.id, static_address_swaps.swap_hash, static_address_swaps.swap_invoice, static_address_swaps.last_hop, static_address_swaps.payment_timeout_seconds, static_address_swaps.quoted_swap_fee_satoshis, static_address_swaps.deposit_outpoints, static_address_swaps.htlc_tx_fee_rate_sat_kw, static_address_swaps.htlc_timeout_sweep_tx_id, static_address_swaps.htlc_timeout_sweep_address, static_address_swaps.selected_amount, static_address_swaps.fast, static_address_swaps.confirmation_risk_decision, static_address_swaps.confirmation_risk_decision_time, htlc_keys.swap_hash, htlc_keys.sender_script_pubkey, htlc_keys.receiver_script_pubkey, htlc_keys.sender_internal_pubkey, htlc_keys.receiver_internal_pubkey, htlc_keys.client_key_family, htlc_keys.client_key_index FROM swaps @@ -263,36 +267,38 @@ ORDER BY ` type GetStaticAddressLoopInSwapsByStatesRow struct { - ID int32 - SwapHash []byte - Preimage []byte - InitiationTime time.Time - AmountRequested int64 - CltvExpiry int32 - MaxMinerFee int64 - MaxSwapFee int64 - InitiationHeight int32 - ProtocolVersion int32 - Label string - ID_2 int32 - SwapHash_2 []byte - SwapInvoice string - LastHop []byte - PaymentTimeoutSeconds int32 - QuotedSwapFeeSatoshis int64 - DepositOutpoints string - HtlcTxFeeRateSatKw int64 - HtlcTimeoutSweepTxID sql.NullString - HtlcTimeoutSweepAddress string - SelectedAmount int64 - Fast bool - SwapHash_3 []byte - SenderScriptPubkey []byte - ReceiverScriptPubkey []byte - SenderInternalPubkey []byte - ReceiverInternalPubkey []byte - ClientKeyFamily int32 - ClientKeyIndex int32 + ID int32 + SwapHash []byte + Preimage []byte + InitiationTime time.Time + AmountRequested int64 + CltvExpiry int32 + MaxMinerFee int64 + MaxSwapFee int64 + InitiationHeight int32 + ProtocolVersion int32 + Label string + ID_2 int32 + SwapHash_2 []byte + SwapInvoice string + LastHop []byte + PaymentTimeoutSeconds int32 + QuotedSwapFeeSatoshis int64 + DepositOutpoints string + HtlcTxFeeRateSatKw int64 + HtlcTimeoutSweepTxID sql.NullString + HtlcTimeoutSweepAddress string + SelectedAmount int64 + Fast bool + ConfirmationRiskDecision string + ConfirmationRiskDecisionTime sql.NullTime + SwapHash_3 []byte + SenderScriptPubkey []byte + ReceiverScriptPubkey []byte + SenderInternalPubkey []byte + ReceiverInternalPubkey []byte + ClientKeyFamily int32 + ClientKeyIndex int32 } func (q *Queries) GetStaticAddressLoopInSwapsByStates(ctx context.Context, dollar_1 sql.NullString) ([]GetStaticAddressLoopInSwapsByStatesRow, error) { @@ -328,6 +334,8 @@ func (q *Queries) GetStaticAddressLoopInSwapsByStates(ctx context.Context, dolla &i.HtlcTimeoutSweepAddress, &i.SelectedAmount, &i.Fast, + &i.ConfirmationRiskDecision, + &i.ConfirmationRiskDecisionTime, &i.SwapHash_3, &i.SenderScriptPubkey, &i.ReceiverScriptPubkey, @@ -482,6 +490,26 @@ func (q *Queries) OverrideSelectedSwapAmount(ctx context.Context, arg OverrideSe return err } +const recordStaticAddressRiskDecision = `-- name: RecordStaticAddressRiskDecision :exec +UPDATE static_address_swaps +SET + confirmation_risk_decision = $2, + confirmation_risk_decision_time = $3 +WHERE + swap_hash = $1 +` + +type RecordStaticAddressRiskDecisionParams struct { + SwapHash []byte + ConfirmationRiskDecision string + ConfirmationRiskDecisionTime sql.NullTime +} + +func (q *Queries) RecordStaticAddressRiskDecision(ctx context.Context, arg RecordStaticAddressRiskDecisionParams) error { + _, err := q.db.ExecContext(ctx, recordStaticAddressRiskDecision, arg.SwapHash, arg.ConfirmationRiskDecision, arg.ConfirmationRiskDecisionTime) + return err +} + const swapHashForDepositID = `-- name: SwapHashForDepositID :one SELECT swap_hash diff --git a/notifications/manager.go b/notifications/manager.go index 219737e4..3bfb7c4c 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -80,6 +80,13 @@ type Config struct { // MinAliveConnTime is the minimum time that the connection to the // server needs to be alive before we consider it a successful. MinAliveConnTime time.Duration + + // PersistStaticLoopInRiskDecision durably records static loop-in + // confirmation-risk decisions. If this fails, the notification is still + // cached and forwarded so a later subscriber can process it after the swap + // row exists. + PersistStaticLoopInRiskDecision func(context.Context, lntypes.Hash, + bool) error } // Manager is a manager for notifications that the swap server sends to the @@ -403,7 +410,7 @@ func (m *Manager) subscribeNotifications(ctx context.Context) error { notification, err := notifStream.Recv() if err == nil && notification != nil { log.Tracef("Received notification: %v", notification) - m.handleNotification(notification) + m.handleNotification(ctx, notification) continue } @@ -415,7 +422,7 @@ func (m *Manager) subscribeNotifications(ctx context.Context) error { // handleNotification handles an incoming notification from the server, // forwarding it to the appropriate subscribers. -func (m *Manager) handleNotification(ntfn *swapserverrpc. +func (m *Manager) handleNotification(ctx context.Context, ntfn *swapserverrpc. SubscribeNotificationsResponse) { switch ntfn.Notification.(type) { @@ -458,9 +465,6 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. // We'll forward the static loop in risk accepted notification to the // subscriber for the matching swap. riskAcceptedNtfn := ntfn.GetStaticLoopInRiskAccepted() - m.Lock() - defer m.Unlock() - var ( swapHash lntypes.Hash hasSwapHash bool @@ -473,12 +477,28 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. } else { swapHash = hash hasSwapHash = true - m.staticLoopInRiskAccepted[hash] = - riskAcceptedNtfn - delete(m.staticLoopInRiskRejected, hash) } } + if hasSwapHash && m.cfg.PersistStaticLoopInRiskDecision != nil { + err := m.cfg.PersistStaticLoopInRiskDecision( + ctx, swapHash, true, + ) + if err != nil { + log.Errorf("Unable to persist static loop in "+ + "risk accepted notification: %v", err) + } + } + + m.Lock() + defer m.Unlock() + + if hasSwapHash { + m.staticLoopInRiskAccepted[swapHash] = + riskAcceptedNtfn + delete(m.staticLoopInRiskRejected, swapHash) + } + for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskAccepted] { // nolint: lll if !hasSwapHash || sub.swapHash == nil || *sub.swapHash != swapHash { @@ -502,9 +522,6 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. // We'll forward the static loop in risk rejected notification to the // subscriber for the matching swap. riskRejectedNtfn := ntfn.GetStaticLoopInRiskRejected() - m.Lock() - defer m.Unlock() - var ( swapHash lntypes.Hash hasSwapHash bool @@ -517,12 +534,28 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. } else { swapHash = hash hasSwapHash = true - m.staticLoopInRiskRejected[hash] = - riskRejectedNtfn - delete(m.staticLoopInRiskAccepted, hash) } } + if hasSwapHash && m.cfg.PersistStaticLoopInRiskDecision != nil { + err := m.cfg.PersistStaticLoopInRiskDecision( + ctx, swapHash, false, + ) + if err != nil { + log.Errorf("Unable to persist static loop in "+ + "risk rejected notification: %v", err) + } + } + + m.Lock() + defer m.Unlock() + + if hasSwapHash { + m.staticLoopInRiskRejected[swapHash] = + riskRejectedNtfn + delete(m.staticLoopInRiskAccepted, swapHash) + } + for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskRejected] { // nolint: lll if !hasSwapHash || sub.swapHash == nil || *sub.swapHash != swapHash { diff --git a/notifications/manager_test.go b/notifications/manager_test.go index 87b7aabf..e1b03e4a 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -256,7 +256,7 @@ func assertStaticLoopInRiskNotificationSwapScoped[ subChanA := subscribe(mgr, subCtx, swapHashA) subChanB := subscribe(mgr, subCtx, swapHashB) - mgr.handleNotification(notification(swapHashA)) + mgr.handleNotification(t.Context(), notification(swapHashA)) select { case received := <-subChanA: @@ -275,7 +275,7 @@ func assertStaticLoopInRiskNotificationSwapScoped[ default: } - mgr.handleNotification(notification(swapHashB)) + mgr.handleNotification(t.Context(), notification(swapHashB)) select { case received := <-subChanB: @@ -303,7 +303,7 @@ func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { fastChan := mgr.SubscribeReservations(fastCtx) firstNotif := getTestNotification(testReservationId) - mgr.handleNotification(firstNotif) + mgr.handleNotification(t.Context(), firstNotif) received := <-fastChan require.Equal(t, testReservationId, received.ReservationId) @@ -311,7 +311,7 @@ func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { secondNotif := getTestNotification(testReservationId2) done := make(chan struct{}) go func() { - mgr.handleNotification(secondNotif) + mgr.handleNotification(t.Context(), secondNotif) close(done) }() @@ -351,11 +351,11 @@ func TestManager_UnfinishedSwapNotificationWaitsForSubscriber(t *testing.T) { swapHashA := lntypes.Hash{0x02, 0x03} swapHashB := lntypes.Hash{0x04, 0x05} - mgr.handleNotification(unfinishedSwapNotification(swapHashA)) + mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashA)) done := make(chan struct{}) go func() { - mgr.handleNotification(unfinishedSwapNotification(swapHashB)) + mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashB)) close(done) }() @@ -398,6 +398,7 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) { subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash) mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskAccepted{ @@ -418,6 +419,91 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) { } } +// TestManager_StaticLoopInRiskDecisionPersists verifies that risk decisions are +// handed to the durable callback before they are treated as delivered. +func TestManager_StaticLoopInRiskDecisionPersists(t *testing.T) { + t.Parallel() + + type persistedDecision struct { + swapHash lntypes.Hash + accepted bool + } + + persisted := make(chan persistedDecision, 2) + mgr := NewManager(&Config{ + PersistStaticLoopInRiskDecision: func(_ context.Context, + swapHash lntypes.Hash, accepted bool) error { + + persisted <- persistedDecision{ + swapHash: swapHash, + accepted: accepted, + } + + return nil + }, + }) + + acceptedHash := lntypes.Hash{0x16, 0x17} + rejectedHash := lntypes.Hash{0x18, 0x19} + + mgr.handleNotification( + t.Context(), staticLoopInRiskAcceptedNotification(acceptedHash), + ) + mgr.handleNotification( + t.Context(), staticLoopInRiskRejectedNotification(rejectedHash), + ) + + select { + case decision := <-persisted: + require.Equal(t, acceptedHash, decision.swapHash) + require.True(t, decision.accepted) + + case <-time.After(time.Second): + t.Fatal("accepted risk decision was not persisted") + } + + select { + case decision := <-persisted: + require.Equal(t, rejectedHash, decision.swapHash) + require.False(t, decision.accepted) + + case <-time.After(time.Second): + t.Fatal("rejected risk decision was not persisted") + } +} + +// TestManager_StaticLoopInRiskDecisionReplayOnPersistFailure verifies that an +// early risk notification is still cached if the swap row does not exist yet. +func TestManager_StaticLoopInRiskDecisionReplayOnPersistFailure(t *testing.T) { + t.Parallel() + + swapHash := lntypes.Hash{0x1a, 0x1b} + mgr := NewManager(&Config{ + PersistStaticLoopInRiskDecision: func(_ context.Context, + _ lntypes.Hash, _ bool) error { + + return errors.New("swap not stored yet") + }, + }) + + mgr.handleNotification( + t.Context(), staticLoopInRiskAcceptedNotification(swapHash), + ) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash) + + select { + case received := <-subChan: + require.Equal(t, swapHash[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("did not replay risk notification after persist failure") + } +} + // TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped verifies that a // notification for one swap does not occupy another swap's subscriber channel. func TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped(t *testing.T) { @@ -444,6 +530,7 @@ func TestManager_StaticLoopInRiskAcceptedNotificationReplay(t *testing.T) { swapHash := lntypes.Hash{0x06, 0x07} mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskAccepted{ @@ -484,6 +571,7 @@ func TestManager_StaticLoopInRiskRejectedNotification(t *testing.T) { subChan := mgr.SubscribeStaticLoopInRiskRejected(subCtx, swapHash) mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskRejected{ @@ -530,6 +618,7 @@ func TestManager_StaticLoopInRiskRejectedNotificationReplay(t *testing.T) { swapHash := lntypes.Hash{0x0a, 0x0b} mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskRejected{ diff --git a/staticaddr/loopin/actions.go b/staticaddr/loopin/actions.go index 7e13dc29..649fca1b 100644 --- a/staticaddr/loopin/actions.go +++ b/staticaddr/loopin/actions.go @@ -732,12 +732,18 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, } }() - startPaymentDeadline := func(reason string) { + startPaymentDeadline := func(reason string, startedAt time.Time) { if deadlineStarted || invoice.State == invoices.ContractCanceled { return } timeout := f.loopIn.PaymentTimeoutDuration() + if !startedAt.IsZero() { + timeout -= time.Since(startedAt) + if timeout < 0 { + timeout = 0 + } + } f.Infof("starting payment deadline after %s", reason) deadlineTimer = time.NewTimer(timeout) @@ -763,6 +769,84 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, f.cancelSwapInvoice(ctx) } + riskDecisionTime := func(decision ConfirmationRiskDecision) time.Time { + now := time.Now() + if f.cfg.Store == nil { + return now + } + + storedLoopIn, err := f.cfg.Store.GetLoopInByHash( + ctx, f.loopIn.SwapHash, + ) + if err != nil { + f.Warnf("unable to reload persisted risk decision for "+ + "swap %v: %v", f.loopIn.SwapHash, err) + + return now + } + + if storedLoopIn == nil { + return now + } + + hasPersistedDecision := + storedLoopIn.ConfirmationRiskDecision == decision && + !storedLoopIn.ConfirmationRiskDecisionTime.IsZero() + + if !hasPersistedDecision { + err = f.cfg.Store.RecordStaticAddressRiskDecision( + ctx, f.loopIn.SwapHash, decision, + ) + if err != nil { + f.Warnf("unable to persist replayed risk "+ + "decision for swap %v: %v", + f.loopIn.SwapHash, err) + + return now + } + + storedLoopIn, err = f.cfg.Store.GetLoopInByHash( + ctx, f.loopIn.SwapHash, + ) + if err != nil { + f.Warnf("unable to reload persisted risk "+ + "decision for swap %v: %v", + f.loopIn.SwapHash, err) + + return now + } + if storedLoopIn == nil || + storedLoopIn.ConfirmationRiskDecision != decision || + storedLoopIn.ConfirmationRiskDecisionTime.IsZero() { + + return now + } + } + + f.loopIn.ConfirmationRiskDecision = + storedLoopIn.ConfirmationRiskDecision + f.loopIn.ConfirmationRiskDecisionTime = + storedLoopIn.ConfirmationRiskDecisionTime + + return storedLoopIn.ConfirmationRiskDecisionTime + } + + switch f.loopIn.ConfirmationRiskDecision { + case ConfirmationRiskDecisionAccepted: + startPaymentDeadline( + "recovered risk accepted notification", + f.loopIn.ConfirmationRiskDecisionTime, + ) + + case ConfirmationRiskDecisionRejected: + cancelInvoiceSubscription() + f.cancelSwapInvoice(ctx) + + return f.HandleError(errors.New( + "server rejected confirmation risk wait", + )) + } + for { select { case <-htlcConfChan: @@ -831,7 +915,16 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, continue } - startPaymentDeadline("risk accepted notification") + startedAt := riskDecisionTime( + ConfirmationRiskDecisionAccepted, + ) + f.loopIn.ConfirmationRiskDecision = + ConfirmationRiskDecisionAccepted + f.loopIn.ConfirmationRiskDecisionTime = startedAt + startPaymentDeadline( + "risk accepted notification", + f.loopIn.ConfirmationRiskDecisionTime, + ) case riskRejected, ok := <-riskRejectedChan: if !ok { @@ -848,6 +941,12 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, cancelInvoiceSubscription() f.cancelSwapInvoice(ctx) + decisionTime := riskDecisionTime( + ConfirmationRiskDecisionRejected, + ) + f.loopIn.ConfirmationRiskDecision = + ConfirmationRiskDecisionRejected + f.loopIn.ConfirmationRiskDecisionTime = decisionTime return f.HandleError(errors.New( "server rejected confirmation risk wait", @@ -864,6 +963,7 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, startPaymentDeadline( "legacy confirmation fallback", + time.Time{}, ) } diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index 40e04b36..e97016cf 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -383,6 +383,234 @@ func TestMonitorInvoiceAndHtlcTxStartsDeadlineOnRiskAccepted(t *testing.T) { } } +// TestMonitorInvoiceAndHtlcTxUsesPersistedAcceptedRiskTime verifies that live +// risk notifications use the durable receipt time, not the local channel +// receive time, when reconstructing the payment deadline. +func TestMonitorInvoiceAndHtlcTxUsesPersistedAcceptedRiskTime(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + mockLnd := test.NewMockLnd() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + swapHash := lntypes.Hash{4, 5, 7} + depositOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{8}, + Index: 0, + } + + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: 2_000, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + ProtocolVersion: version.ProtocolVersion_V0, + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + PaymentTimeoutSeconds: 1, + DepositOutpoints: []string{ + depositOutpoint.String(), + }, + Deposits: []*deposit.Deposit{{ + OutPoint: depositOutpoint, + }}, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + }) + + notificationMgr := &mockNotificationManager{ + riskAccepted: make( + chan *swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification, 1, + ), + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: &noopDepositManager{}, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + NotificationManager: notificationMgr, + Store: &mockStore{ + loopIns: map[lntypes.Hash]*StaticAddressLoopIn{ + swapHash: { + ConfirmationRiskDecision: ConfirmationRiskDecisionAccepted, + ConfirmationRiskDecisionTime: time.Now().Add( + -time.Minute, + ), + }, + }, + }, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(ctx, nil) + }() + + waitForMonitorSubscriptions(t, ctx, mockLnd) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + t.Fatalf("invoice canceled before risk acceptance: %v", hash) + + case <-time.After(200 * time.Millisecond): + } + + notificationMgr.riskAccepted <- &swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification{ + SwapHash: swapHash[:], + } + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, swapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } + + cancel() + select { + case event := <-resultChan: + require.Equal(t, fsm.OnError, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } +} + +// TestMonitorInvoiceAndHtlcTxPersistsReplayedRiskAccepted verifies that a risk +// notification replayed after the swap row exists is written back to the store. +func TestMonitorInvoiceAndHtlcTxPersistsReplayedRiskAccepted(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + mockLnd := test.NewMockLnd() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + swapHash := lntypes.Hash{5, 6, 10} + depositOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{14}, + Index: 0, + } + + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: 2_000, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + ProtocolVersion: version.ProtocolVersion_V0, + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + PaymentTimeoutSeconds: 3_600, + DepositOutpoints: []string{ + depositOutpoint.String(), + }, + Deposits: []*deposit.Deposit{{ + OutPoint: depositOutpoint, + }}, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + }) + + notificationMgr := &mockNotificationManager{ + riskAccepted: make( + chan *swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification, 1, + ), + } + store := &recordingRiskStore{ + mockStore: &mockStore{ + loopIns: map[lntypes.Hash]*StaticAddressLoopIn{ + swapHash: {}, + }, + }, + decisions: make(chan ConfirmationRiskDecision, 1), + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: &noopDepositManager{}, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + NotificationManager: notificationMgr, + Store: store, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(ctx, nil) + }() + + waitForMonitorSubscriptions(t, ctx, mockLnd) + + notificationMgr.riskAccepted <- &swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification{ + SwapHash: swapHash[:], + } + + select { + case decision := <-store.decisions: + require.Equal(t, ConfirmationRiskDecisionAccepted, decision) + + case <-ctx.Done(): + t.Fatalf("risk decision was not persisted: %v", ctx.Err()) + } + + stored := store.loopIns[swapHash] + require.Equal(t, ConfirmationRiskDecisionAccepted, + stored.ConfirmationRiskDecision) + require.False(t, stored.ConfirmationRiskDecisionTime.IsZero()) + + cancel() + select { + case event := <-resultChan: + require.Equal(t, fsm.OnError, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } +} + // TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected verifies that a server-side // confirmation risk rejection is terminal for the client monitor action. func TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected(t *testing.T) { @@ -479,6 +707,181 @@ func TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected(t *testing.T) { } } +// TestMonitorInvoiceAndHtlcTxRecoversAcceptedRiskDecision verifies that a +// persisted risk acceptance restarts the payment deadline with elapsed time +// preserved after restart. +func TestMonitorInvoiceAndHtlcTxRecoversAcceptedRiskDecision(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + mockLnd := test.NewMockLnd() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + swapHash := lntypes.Hash{5, 6, 8} + depositOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{12}, + Index: 0, + } + + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: 2_000, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + ProtocolVersion: version.ProtocolVersion_V0, + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + PaymentTimeoutSeconds: 1, + ConfirmationRiskDecision: ConfirmationRiskDecisionAccepted, + ConfirmationRiskDecisionTime: time.Now().Add(-time.Minute), + DepositOutpoints: []string{ + depositOutpoint.String(), + }, + Deposits: []*deposit.Deposit{{ + OutPoint: depositOutpoint, + }}, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + }) + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: &noopDepositManager{}, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(ctx, nil) + }() + + waitForMonitorSubscriptions(t, ctx, mockLnd) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, swapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } + + cancel() + select { + case event := <-resultChan: + require.Equal(t, fsm.OnError, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } +} + +// TestMonitorInvoiceAndHtlcTxRecoversRejectedRiskDecision verifies that a +// persisted risk rejection is terminal after restart without waiting for a +// replayed server notification. +func TestMonitorInvoiceAndHtlcTxRecoversRejectedRiskDecision(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + mockLnd := test.NewMockLnd() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + swapHash := lntypes.Hash{5, 6, 9} + depositOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{13}, + Index: 0, + } + + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: 2_000, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + ProtocolVersion: version.ProtocolVersion_V0, + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + PaymentTimeoutSeconds: 3_600, + ConfirmationRiskDecision: ConfirmationRiskDecisionRejected, + ConfirmationRiskDecisionTime: time.Now(), + DepositOutpoints: []string{ + depositOutpoint.String(), + }, + Deposits: []*deposit.Deposit{{ + OutPoint: depositOutpoint, + }}, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + }) + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: &noopDepositManager{}, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(ctx, nil) + }() + + waitForMonitorSubscriptions(t, ctx, mockLnd) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, swapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } + + select { + case event := <-resultChan: + require.Equal(t, fsm.OnError, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } +} + // TestMonitorInvoiceAndHtlcTxDoesNotCancelWhenOriginalOutpointVanishes // verifies that once the monitor state is reached, a missing original deposit // outpoint does not cancel the invoice. After HTLC signatures are handed to the @@ -1304,6 +1707,33 @@ func (r *recordingDepositManager) TransitionDeposits(_ context.Context, return nil } +type recordingRiskStore struct { + *mockStore + + decisions chan ConfirmationRiskDecision +} + +// RecordStaticAddressRiskDecision records a risk decision in the mock store. +func (s *recordingRiskStore) RecordStaticAddressRiskDecision( + _ context.Context, swapHash lntypes.Hash, + decision ConfirmationRiskDecision) error { + + loopIn, ok := s.loopIns[swapHash] + if !ok { + return ErrLoopInNotFound + } + + loopIn.ConfirmationRiskDecision = decision + loopIn.ConfirmationRiskDecisionTime = time.Now() + + select { + case s.decisions <- decision: + default: + } + + return nil +} + // mockNotificationManager allows tests to push server notifications directly to // monitor actions. type mockNotificationManager struct { diff --git a/staticaddr/loopin/interface.go b/staticaddr/loopin/interface.go index f96d08e5..89de19a2 100644 --- a/staticaddr/loopin/interface.go +++ b/staticaddr/loopin/interface.go @@ -88,6 +88,11 @@ type StaticAddressLoopInStore interface { // IsStored checks if the loop-in is already stored in the database. IsStored(ctx context.Context, swapHash lntypes.Hash) (bool, error) + // RecordStaticAddressRiskDecision persists the server's + // confirmation-risk decision for the loop-in identified by swapHash. + RecordStaticAddressRiskDecision(ctx context.Context, + swapHash lntypes.Hash, decision ConfirmationRiskDecision) error + // GetLoopInByHash returns the loop-in swap with the given hash. GetLoopInByHash(ctx context.Context, swapHash lntypes.Hash) ( *StaticAddressLoopIn, error) diff --git a/staticaddr/loopin/loopin.go b/staticaddr/loopin/loopin.go index 6c1860bd..e77384bf 100644 --- a/staticaddr/loopin/loopin.go +++ b/staticaddr/loopin/loopin.go @@ -31,6 +31,23 @@ import ( "github.com/lightningnetwork/lnd/zpay32" ) +// ConfirmationRiskDecision records the server's decision on whether it accepts +// waiting for low-confirmation deposits before paying a static loop-in invoice. +type ConfirmationRiskDecision string + +const ( + // ConfirmationRiskDecisionNone means no risk decision has been received. + ConfirmationRiskDecisionNone ConfirmationRiskDecision = "" + + // ConfirmationRiskDecisionAccepted means the server accepted waiting for + // deposit confirmations and the payment deadline has started. + ConfirmationRiskDecisionAccepted ConfirmationRiskDecision = "accepted" + + // ConfirmationRiskDecisionRejected means the server stopped waiting for + // deposit confirmations before paying the invoice. + ConfirmationRiskDecisionRejected ConfirmationRiskDecision = "rejected" +) + // StaticAddressLoopIn represents the in-memory loop-in information. type StaticAddressLoopIn struct { // SwapHash is the hashed preimage of the swap invoice. It represents @@ -107,6 +124,15 @@ type StaticAddressLoopIn struct { // LastUpdateTime is the timestamp of the latest persisted state update. LastUpdateTime time.Time + // ConfirmationRiskDecision records the server's persisted decision on + // low-confirmation deposit risk. + ConfirmationRiskDecision ConfirmationRiskDecision + + // ConfirmationRiskDecisionTime is when loopd persisted the server risk + // decision. It is used to reconstruct payment-deadline timeouts after + // restart. + ConfirmationRiskDecisionTime time.Time + // state is the current state of the swap. state fsm.StateType diff --git a/staticaddr/loopin/manager_test.go b/staticaddr/loopin/manager_test.go index 58549747..7cd1f4df 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -460,6 +460,13 @@ func (s *mockStore) IsStored(_ context.Context, _ lntypes.Hash) (bool, error) { return false, nil } +// RecordStaticAddressRiskDecision implements Store for manager tests. +func (s *mockStore) RecordStaticAddressRiskDecision(context.Context, + lntypes.Hash, ConfirmationRiskDecision) error { + + return nil +} + func (s *mockStore) GetLoopInByHash(_ context.Context, swapHash lntypes.Hash) (*StaticAddressLoopIn, error) { diff --git a/staticaddr/loopin/sql_store.go b/staticaddr/loopin/sql_store.go index 8eeb6f07..8c62d1a7 100644 --- a/staticaddr/loopin/sql_store.go +++ b/staticaddr/loopin/sql_store.go @@ -27,6 +27,9 @@ var ( // ErrInvalidOutpoint is returned when an outpoint contains the outpoint // separator. ErrInvalidOutpoint = errors.New("outpoint contains outpoint separator") + + // ErrLoopInNotFound is returned when a loop-in swap is not stored. + ErrLoopInNotFound = errors.New("static address loop-in not found") ) // Querier is the interface that contains all the queries generated by sqlc for @@ -51,6 +54,11 @@ type Querier interface { UpdateStaticAddressLoopIn(ctx context.Context, arg sqlc.UpdateStaticAddressLoopInParams) error + // RecordStaticAddressRiskDecision stores the server's confirmation-risk + // decision for a loop-in swap. + RecordStaticAddressRiskDecision(ctx context.Context, + arg sqlc.RecordStaticAddressRiskDecisionParams) error + // GetStaticAddressLoopInSwap retrieves a loop-in swap by its swap hash. GetStaticAddressLoopInSwap(ctx context.Context, swapHash []byte) (sqlc.GetStaticAddressLoopInSwapRow, error) @@ -361,6 +369,43 @@ func (s *SqlStore) UpdateLoopIn(ctx context.Context, ) } +// RecordStaticAddressRiskDecision stores the server's confirmation-risk +// decision for a static address loop-in. The timestamp is written by the store +// so recovery can reconstruct the remaining payment deadline from one durable +// clock source. +func (s *SqlStore) RecordStaticAddressRiskDecision(ctx context.Context, + swapHash lntypes.Hash, decision ConfirmationRiskDecision) error { + + if decision != ConfirmationRiskDecisionAccepted && + decision != ConfirmationRiskDecisionRejected { + + return errors.New("unknown confirmation risk decision") + } + + params := sqlc.RecordStaticAddressRiskDecisionParams{ + SwapHash: swapHash[:], + ConfirmationRiskDecision: string(decision), + ConfirmationRiskDecisionTime: sql.NullTime{ + Time: s.clock.Now(), + Valid: true, + }, + } + + return s.baseDB.ExecTx(ctx, loopdb.NewSqlWriteOpts(), + func(q Querier) error { + stored, err := q.IsStored(ctx, swapHash[:]) + if err != nil { + return err + } + if !stored { + return ErrLoopInNotFound + } + + return q.RecordStaticAddressRiskDecision(ctx, params) + }, + ) +} + func (s *SqlStore) BatchUpdateSelectedSwapAmounts(ctx context.Context, updateAmounts map[lntypes.Hash]btcutil.Amount) error { @@ -583,6 +628,9 @@ func toStaticAddressLoopIn(_ context.Context, network *chaincfg.Params, DepositOutpoints: depositOutpoints, SelectedAmount: btcutil.Amount(swap.SelectedAmount), Fast: swap.Fast, + ConfirmationRiskDecision: ConfirmationRiskDecision( + swap.ConfirmationRiskDecision, + ), HtlcTxFeeRate: chainfee.SatPerKWeight( swap.HtlcTxFeeRateSatKw, ), @@ -590,6 +638,10 @@ func toStaticAddressLoopIn(_ context.Context, network *chaincfg.Params, HtlcTimeoutSweepTxHash: htlcTimeoutSweepTxHash, Deposits: depositList, } + if swap.ConfirmationRiskDecisionTime.Valid { + loopIn.ConfirmationRiskDecisionTime = + swap.ConfirmationRiskDecisionTime.Time + } if len(updates) > 0 { lastUpdate := updates[len(updates)-1] diff --git a/staticaddr/loopin/sql_store_test.go b/staticaddr/loopin/sql_store_test.go index 7e1ab0a5..72580aa3 100644 --- a/staticaddr/loopin/sql_store_test.go +++ b/staticaddr/loopin/sql_store_test.go @@ -297,6 +297,31 @@ func TestCreateLoopIn(t *testing.T) { require.Equal(t, []string{d1.OutPoint.String(), d2.OutPoint.String()}, swap.DepositOutpoints) require.Equal(t, SignHtlcTx, swap.GetState()) + require.Equal( + t, ConfirmationRiskDecisionNone, + swap.ConfirmationRiskDecision, + ) + + decisionTime := time.Unix(123, 0).UTC() + testClock.SetTime(decisionTime) + err = swapStore.RecordStaticAddressRiskDecision( + ctx, swapHashPending, ConfirmationRiskDecisionAccepted, + ) + require.NoError(t, err) + + swap, err = swapStore.GetLoopInByHash(ctx, swapHashPending) + require.NoError(t, err) + require.Equal( + t, ConfirmationRiskDecisionAccepted, + swap.ConfirmationRiskDecision, + ) + require.True(t, swap.ConfirmationRiskDecisionTime.Equal(decisionTime)) + + err = swapStore.RecordStaticAddressRiskDecision( + ctx, lntypes.Hash{0x9, 0x9, 0x9}, + ConfirmationRiskDecisionRejected, + ) + require.ErrorIs(t, err, ErrLoopInNotFound) require.Len(t, swap.Deposits, 2)