From fc3fba7c2391b646a6ac32280709093ab1dff6bb Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 26 Jun 2026 12:54:30 +0200 Subject: [PATCH] staticaddr/loopin: persist risk decisions Store server confirmation-risk decisions with static loop-in swaps and recover accepted payment-deadline timers after restart. Wire notification persistence through loopd so recovered swaps do not lose pending risk state. Deduplicate notification fanout cache entries by swap hash. --- loopd/daemon.go | 26 +- loopd/swapclient_server_test.go | 8 + ...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 | 285 +++--- notifications/manager_test.go | 184 +++- staticaddr/loopin/actions.go | 305 ++++-- staticaddr/loopin/actions_test.go | 876 +++++++++++++++++- 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 + 17 files changed, 1713 insertions(+), 286 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 51428ce5..25f55487 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/loopd/swapclient_server_test.go b/loopd/swapclient_server_test.go index a79d1950..c28b4d1c 100644 --- a/loopd/swapclient_server_test.go +++ b/loopd/swapclient_server_test.go @@ -535,6 +535,14 @@ func (s *mockStaticAddressLoopInStore) IsStored(_ context.Context, return false, nil } +// RecordStaticAddressRiskDecision satisfies the static loop-in store interface. +func (s *mockStaticAddressLoopInStore) RecordStaticAddressRiskDecision( + _ context.Context, _ lntypes.Hash, + _ loopin.ConfirmationRiskDecision) error { + + return nil +} + // GetLoopInByHash returns the configured loop-in with the given hash. func (s *mockStaticAddressLoopInStore) GetLoopInByHash(_ context.Context, swapHash lntypes.Hash) (*loopin.StaticAddressLoopIn, error) { 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 ed408195..2c8939a7 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -88,6 +88,13 @@ type Config struct { // MaxQueuedNotifications is the maximum number of notifications that // can wait in each subscriber's delivery queue. MaxQueuedNotifications int + + // 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 @@ -106,6 +113,8 @@ type Manager struct { staticLoopInRiskRejected map[lntypes.Hash]*swapserverrpc. ServerStaticLoopInRiskRejectedNotification + + staticLoopInRiskPersisted map[lntypes.Hash]bool } // NewManager creates a new notification manager. @@ -129,14 +138,15 @@ func NewManager(cfg *Config) *Manager { map[lntypes.Hash]*swapserverrpc. ServerStaticLoopInRiskRejectedNotification, ), + staticLoopInRiskPersisted: make(map[lntypes.Hash]bool), } } type subscriber struct { subCtx context.Context recvChan any - enqueue func(any) swapHash *lntypes.Hash + enqueue func(any) } // newNotificationQueue creates a per-subscriber FIFO delivery function. @@ -245,6 +255,19 @@ func queueNotification[T any](sub subscriber, recvChan chan T, ntfn T) { } } +// dropNotification sends a best-effort notification to a subscriber. +func dropNotification[T any](sub subscriber, recvChan chan T, ntfn T, + description string) { + + select { + case recvChan <- ntfn: + case <-sub.subCtx.Done(): + default: + log.Debugf("Dropping %s notification for slow subscriber", + description) + } +} + // SubscribeReservations subscribes to the reservation notifications. func (m *Manager) SubscribeReservations(ctx context.Context, ) <-chan *swapserverrpc.ServerReservationNotification { @@ -294,16 +317,11 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context, return notifChan } -// SubscribeStaticLoopInRiskAccepted subscribes to static loop in risk accepted -// notifications. -func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, - swapHash lntypes.Hash, -) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification { - - notifChan := make( - chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification, 1, - ) +func subscribeStaticLoopInRiskDecision[T any](m *Manager, ctx context.Context, + swapHash lntypes.Hash, notifType NotificationType, + notifications map[lntypes.Hash]T) <-chan T { + notifChan := make(chan T, 1) sub := subscriber{ subCtx: ctx, recvChan: notifChan, @@ -311,19 +329,25 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, } m.Lock() - m.subscribers[NotificationTypeStaticLoopInRiskAccepted] = append( - m.subscribers[NotificationTypeStaticLoopInRiskAccepted], sub, - ) - if ntfn, ok := m.staticLoopInRiskAccepted[swapHash]; ok { + m.subscribers[notifType] = append(m.subscribers[notifType], sub) + if ntfn, ok := notifications[swapHash]; ok { notifChan <- ntfn - delete(m.staticLoopInRiskAccepted, swapHash) + if m.staticLoopInRiskPersisted[swapHash] { + delete(notifications, swapHash) + delete(m.staticLoopInRiskPersisted, swapHash) + } } m.Unlock() context.AfterFunc(ctx, func() { - m.removeSubscriber(NotificationTypeStaticLoopInRiskAccepted, sub) + m.removeSubscriber(notifType, sub) m.Lock() - delete(m.staticLoopInRiskAccepted, swapHash) + if _, ok := notifications[swapHash]; ok && + m.staticLoopInRiskPersisted[swapHash] { + + delete(notifications, swapHash) + delete(m.staticLoopInRiskPersisted, swapHash) + } m.Unlock() close(notifChan) }) @@ -331,41 +355,28 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, return notifChan } +// SubscribeStaticLoopInRiskAccepted subscribes to static loop in risk accepted +// notifications. +func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, + swapHash lntypes.Hash, +) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification { + + return subscribeStaticLoopInRiskDecision( + m, ctx, swapHash, NotificationTypeStaticLoopInRiskAccepted, + m.staticLoopInRiskAccepted, + ) +} + // SubscribeStaticLoopInRiskRejected subscribes to static loop in risk rejected // notifications. func (m *Manager) SubscribeStaticLoopInRiskRejected(ctx context.Context, swapHash lntypes.Hash, ) <-chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification { - notifChan := make( - chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification, 1, + return subscribeStaticLoopInRiskDecision( + m, ctx, swapHash, NotificationTypeStaticLoopInRiskRejected, + m.staticLoopInRiskRejected, ) - - sub := subscriber{ - subCtx: ctx, - recvChan: notifChan, - swapHash: &swapHash, - } - - m.Lock() - m.subscribers[NotificationTypeStaticLoopInRiskRejected] = append( - m.subscribers[NotificationTypeStaticLoopInRiskRejected], sub, - ) - if ntfn, ok := m.staticLoopInRiskRejected[swapHash]; ok { - notifChan <- ntfn - delete(m.staticLoopInRiskRejected, swapHash) - } - m.Unlock() - - context.AfterFunc(ctx, func() { - m.removeSubscriber(NotificationTypeStaticLoopInRiskRejected, sub) - m.Lock() - delete(m.staticLoopInRiskRejected, swapHash) - m.Unlock() - close(notifChan) - }) - - return notifChan } // SubscribeUnfinishedSwaps subscribes to the unfinished swap notifications. @@ -525,7 +536,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 } @@ -535,9 +546,73 @@ func (m *Manager) subscribeNotifications(ctx context.Context) error { } } +// staticLoopInRiskDecisionName returns the log label for a risk decision. +func staticLoopInRiskDecisionName(accepted bool) string { + if accepted { + return "accepted" + } + + return "rejected" +} + +// handleStaticLoopInRiskDecision persists, caches, and forwards a risk +// decision notification to the matching subscriber. +func (m *Manager) handleStaticLoopInRiskDecision(ctx context.Context, + swapHashBytes []byte, accepted bool, notifType NotificationType, + cacheDecision func(lntypes.Hash, bool), + notifySubscriber func(subscriber)) { + + decision := staticLoopInRiskDecisionName(accepted) + persisted := m.cfg.PersistStaticLoopInRiskDecision == nil + + var ( + swapHash lntypes.Hash + hasSwapHash bool + ) + if swapHashBytes != nil { + hash, err := lntypes.MakeHash(swapHashBytes) + if err != nil { + log.Warnf("Received invalid static loop in risk "+ + "%s notification: %v", decision, err) + } else { + swapHash = hash + hasSwapHash = true + } + } + + if hasSwapHash && m.cfg.PersistStaticLoopInRiskDecision != nil { + err := m.cfg.PersistStaticLoopInRiskDecision( + ctx, swapHash, accepted, + ) + if err != nil { + log.Errorf("Unable to persist static loop in risk "+ + "%s notification: %v", decision, err) + } else { + persisted = true + } + } + + m.Lock() + defer m.Unlock() + + if hasSwapHash { + cacheDecision(swapHash, persisted) + } + + for _, sub := range m.subscribers[notifType] { + if !hasSwapHash || sub.swapHash == nil || + *sub.swapHash != swapHash { + + continue + } + + notifySubscriber(sub) + } +} + // 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) { @@ -577,89 +652,57 @@ 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 - ) + var swapHashBytes []byte if riskAcceptedNtfn != nil { - hash, err := lntypes.MakeHash(riskAcceptedNtfn.SwapHash) - if err != nil { - log.Warnf("Received invalid static loop in risk "+ - "accepted notification: %v", err) - } else { - swapHash = hash - hasSwapHash = true - m.staticLoopInRiskAccepted[hash] = + swapHashBytes = riskAcceptedNtfn.SwapHash + } + + m.handleStaticLoopInRiskDecision( + ctx, swapHashBytes, true, + NotificationTypeStaticLoopInRiskAccepted, + func(swapHash lntypes.Hash, persisted bool) { + m.staticLoopInRiskAccepted[swapHash] = riskAcceptedNtfn - delete(m.staticLoopInRiskRejected, hash) - } - } - - for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskAccepted] { // nolint: lll - if !hasSwapHash || sub.swapHash == nil || - *sub.swapHash != swapHash { - - continue - } - - recvChan := sub.recvChan.(chan *swapserverrpc. - ServerStaticLoopInRiskAcceptedNotification) - - select { - case recvChan <- riskAcceptedNtfn: - case <-sub.subCtx.Done(): - default: - log.Debugf("Dropping static loop in risk " + - "accepted notification for slow subscriber") - } - } + m.staticLoopInRiskPersisted[swapHash] = persisted + delete(m.staticLoopInRiskRejected, swapHash) + }, + func(sub subscriber) { + recvChan := sub.recvChan.(chan *swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification) + dropNotification( + sub, recvChan, riskAcceptedNtfn, + "static loop in risk accepted", + ) + }, + ) case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInRiskRejected: // nolint: lll // 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 - ) + var swapHashBytes []byte if riskRejectedNtfn != nil { - hash, err := lntypes.MakeHash(riskRejectedNtfn.SwapHash) - if err != nil { - log.Warnf("Received invalid static loop in risk "+ - "rejected notification: %v", err) - } else { - swapHash = hash - hasSwapHash = true - m.staticLoopInRiskRejected[hash] = + swapHashBytes = riskRejectedNtfn.SwapHash + } + + m.handleStaticLoopInRiskDecision( + ctx, swapHashBytes, false, + NotificationTypeStaticLoopInRiskRejected, + func(swapHash lntypes.Hash, persisted bool) { + m.staticLoopInRiskRejected[swapHash] = riskRejectedNtfn - delete(m.staticLoopInRiskAccepted, hash) - } - } - - for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskRejected] { // nolint: lll - if !hasSwapHash || sub.swapHash == nil || - *sub.swapHash != swapHash { - - continue - } - - recvChan := sub.recvChan.(chan *swapserverrpc. - ServerStaticLoopInRiskRejectedNotification) - - select { - case recvChan <- riskRejectedNtfn: - case <-sub.subCtx.Done(): - default: - log.Debugf("Dropping static loop in risk " + - "rejected notification for slow subscriber") - } - } + m.staticLoopInRiskPersisted[swapHash] = persisted + delete(m.staticLoopInRiskAccepted, swapHash) + }, + func(sub subscriber) { + recvChan := sub.recvChan.(chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification) + dropNotification( + sub, recvChan, riskRejectedNtfn, + "static loop in risk rejected", + ) + }, + ) case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll // We'll forward the unfinished swap notification to all diff --git a/notifications/manager_test.go b/notifications/manager_test.go index b165c494..59ecab78 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -220,6 +220,7 @@ func staticLoopInSweepNotification( } } +// staticLoopInRiskAcceptedNotification builds a risk accepted notification. func staticLoopInRiskAcceptedNotification( swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { @@ -271,7 +272,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: @@ -290,7 +291,7 @@ func assertStaticLoopInRiskNotificationSwapScoped[ default: } - mgr.handleNotification(notification(swapHashB)) + mgr.handleNotification(t.Context(), notification(swapHashB)) select { case received := <-subChanB: @@ -320,7 +321,7 @@ func TestManager_SlowReservationSubscriberDoesNotBlock(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) @@ -328,7 +329,7 @@ func TestManager_SlowReservationSubscriberDoesNotBlock(t *testing.T) { secondNotif := getTestNotification(testReservationId2) done := make(chan struct{}) go func() { - mgr.handleNotification(secondNotif) + mgr.handleNotification(t.Context(), secondNotif) close(done) }() @@ -427,7 +428,7 @@ func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) { subChan := mgr.SubscribeUnfinishedSwaps(subCtx) swapHashA := lntypes.Hash{0x21, 0x22} - mgr.handleNotification(unfinishedSwapNotification(swapHashA)) + mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashA)) require.Eventually(t, func() bool { return len(subChan) == 1 @@ -436,7 +437,7 @@ func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) { swapHashB := lntypes.Hash{0x23, 0x24} done := make(chan struct{}) go func() { - mgr.handleNotification(unfinishedSwapNotification(swapHashB)) + mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashB)) close(done) }() @@ -508,11 +509,11 @@ func assertQueuedSwapHashNotifications[T any](t *testing.T, subChan := subscribe(mgr, subCtx) - mgr.handleNotification(notification(swapHashA)) + mgr.handleNotification(t.Context(), notification(swapHashA)) done := make(chan struct{}) go func() { - mgr.handleNotification(notification(swapHashB)) + mgr.handleNotification(t.Context(), notification(swapHashB)) close(done) }() @@ -557,6 +558,7 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) { subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash) mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskAccepted{ @@ -577,6 +579,169 @@ 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_StaticLoopInRiskDecisionReplaysAfterSubscriberCancel verifies that +// a non-persisted risk decision remains replayable if the subscriber is canceled +// before the FSM has a chance to process it. +func TestManager_StaticLoopInRiskDecisionReplaysAfterSubscriberCancel( + t *testing.T) { + + t.Parallel() + + assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel( + t, + (*Manager).SubscribeStaticLoopInRiskAccepted, + staticLoopInRiskAcceptedNotification, + ) + assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel( + t, + (*Manager).SubscribeStaticLoopInRiskRejected, + staticLoopInRiskRejectedNotification, + ) +} + +func assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel[ + T staticLoopInRiskNotification](t *testing.T, + subscribe func(*Manager, context.Context, lntypes.Hash) <-chan T, + notification func(lntypes.Hash) *swapserverrpc. + SubscribeNotificationsResponse) { + + t.Helper() + + swapHash := lntypes.Hash{0x2a, 0x2b} + mgr := NewManager(&Config{ + PersistStaticLoopInRiskDecision: func(_ context.Context, + _ lntypes.Hash, _ bool) error { + + return errors.New("swap not stored yet") + }, + }) + + subCtx, subCancel := context.WithCancel(t.Context()) + subChan := subscribe(mgr, subCtx, swapHash) + + mgr.handleNotification(t.Context(), notification(swapHash)) + + require.Eventually(t, func() bool { + return len(subChan) == 1 + }, time.Second, 10*time.Millisecond) + + subCancel() + + select { + case <-subChan: + + case <-time.After(time.Second): + t.Fatal("risk decision notification was not delivered before " + + "cancel") + } + + select { + case _, ok := <-subChan: + require.False(t, ok) + + case <-time.After(time.Second): + t.Fatal("risk decision subscription did not close after cancel") + } + + replayCtx, replayCancel := context.WithCancel(t.Context()) + defer replayCancel() + + replayChan := subscribe(mgr, replayCtx, swapHash) + select { + case received := <-replayChan: + require.Equal(t, swapHash[:], received.GetSwapHash()) + + case <-time.After(time.Second): + t.Fatal("cached risk decision was lost after subscriber " + + "cancellation") + } +} + // TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped verifies that a // notification for one swap does not occupy another swap's subscriber channel. func TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped(t *testing.T) { @@ -603,6 +768,7 @@ func TestManager_StaticLoopInRiskAcceptedNotificationReplay(t *testing.T) { swapHash := lntypes.Hash{0x06, 0x07} mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskAccepted{ @@ -643,6 +809,7 @@ func TestManager_StaticLoopInRiskRejectedNotification(t *testing.T) { subChan := mgr.SubscribeStaticLoopInRiskRejected(subCtx, swapHash) mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskRejected{ @@ -689,6 +856,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 4b2d23e3..311cd57a 100644 --- a/staticaddr/loopin/actions.go +++ b/staticaddr/loopin/actions.go @@ -42,6 +42,8 @@ const ( defaultInvoiceCleanupTimeout = 5 * time.Second ) +var paymentDeadlineUnlockRetryDelay = time.Minute + var ( // ErrFeeTooHigh is returned if the server sets a fee rate for the htlc // tx that is too high. We prevent here against a low htlc timeout sweep @@ -797,22 +799,60 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // confirmation risk was accepted. If a server does not support risk // notifications, fall back after the legacy deposit confirmation depth. var ( - deadlineChan <-chan time.Time - deadlineTimer *time.Timer - deadlineStarted bool + deadlineChan <-chan time.Time + deadlineTimer *time.Timer + deadlineStarted bool + unlockRetryChan <-chan time.Time + unlockRetryTimer *time.Timer ) defer func() { if deadlineTimer != nil { deadlineTimer.Stop() } + + if unlockRetryTimer != nil { + unlockRetryTimer.Stop() + } }() - startPaymentDeadline := func(reason string) { + depositsInState := func(state fsm.StateType) bool { + if len(f.loopIn.Deposits) == 0 { + return false + } + + for _, d := range f.loopIn.Deposits { + if d == nil { + return false + } + + d.Lock() + inState := d.IsInStateNoLock(state) + d.Unlock() + if !inState { + return false + } + } + + return true + } + + invoiceCanceledForNonPayment := invoice.State == invoices.ContractCanceled + depositsLockedForHtlcTimeout := depositsInState( + deposit.SweepHtlcTimeout, + ) + + 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) @@ -820,6 +860,74 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, deadlineStarted = true } + scheduleUnlockRetry := func() { + if depositsUnlocked || unlockRetryChan != nil { + return + } + + unlockRetryTimer = time.NewTimer( + paymentDeadlineUnlockRetryDelay, + ) + unlockRetryChan = unlockRetryTimer.C + } + + transitionDepositsToHtlcTimeout := func(reason string) { + if depositsLockedForHtlcTimeout || + depositsInState(deposit.SweepHtlcTimeout) { + + depositsLockedForHtlcTimeout = true + depositsUnlocked = false + return + } + + err = f.cfg.DepositManager.TransitionDeposits( + ctx, f.loopIn.Deposits, + deposit.OnSweepingHtlcTimeout, + deposit.SweepHtlcTimeout, + ) + if err != nil { + f.Errorf("unable to transition deposits to the htlc "+ + "timeout sweeping state after %s: %v", + reason, err) + + return + } + + depositsLockedForHtlcTimeout = true + depositsUnlocked = false + } + + unlockDepositsAfterInvoiceCancel := func(reason string) { + if htlcConfirmed { + transitionDepositsToHtlcTimeout(reason) + return + } + + err = f.unlockDeposits(ctx) + if err != nil { + f.Errorf("unable to unlock deposits after %s: %v, "+ + "retrying in %v", + reason, err, + paymentDeadlineUnlockRetryDelay) + + scheduleUnlockRetry() + + return + } + + depositsUnlocked = true + } + + startLegacyFallback := func(reason string, currentHeight int32) { + if deadlineStarted || invoice.State == invoices.ContractCanceled { + return + } + + if f.shouldStartLegacyConfirmationFallback(ctx, currentHeight) { + startPaymentDeadline(reason, time.Time{}) + } + } + if invoice.State == invoices.ContractCanceled { // If the invoice was canceled previously we end our // subscription to invoice updates. @@ -836,6 +944,108 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // Reuse the same helper as InitHtlcAction so timeout cleanup // follows the same detached-context path as early-init cleanup. f.cancelSwapInvoice() + invoice.State = invoices.ContractCanceled + invoiceCanceledForNonPayment = true + } + + 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 + } + + handleRiskRejected := func(reason string) { + cancelInvoiceSubscription() + f.cancelSwapInvoice() + invoice.State = invoices.ContractCanceled + invoiceCanceledForNonPayment = true + decisionTime := riskDecisionTime( + ConfirmationRiskDecisionRejected, + ) + f.loopIn.ConfirmationRiskDecision = + ConfirmationRiskDecisionRejected + f.loopIn.ConfirmationRiskDecisionTime = decisionTime + riskAcceptedChan = nil + riskRejectedChan = nil + + unlockDepositsAfterInvoiceCancel(reason) + } + + switch f.loopIn.ConfirmationRiskDecision { + case ConfirmationRiskDecisionAccepted: + startPaymentDeadline( + "recovered risk accepted notification", + f.loopIn.ConfirmationRiskDecisionTime, + ) + + case ConfirmationRiskDecisionRejected: + handleRiskRejected("recovered risk rejection") + } + + info, err := f.cfg.LndClient.GetInfo(ctx) + if err != nil { + f.Warnf("unable to query current height for legacy confirmation "+ + "fallback: %v", err) + } else { + startLegacyFallback( + "legacy confirmation fallback", int32(info.BlockHeight), + ) } for { @@ -844,6 +1054,11 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, f.Infof("htlc tx confirmed") htlcConfirmed = true + if invoiceCanceledForNonPayment { + transitionDepositsToHtlcTimeout( + "htlc confirmation after invoice cancellation", + ) + } case err = <-htlcErrConfChan: f.Errorf("htlc tx conf chan error, re-registering: "+ @@ -885,13 +1100,14 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // re-enable them for loop-ins and withdrawals. cancelInvoice() - err = f.unlockDeposits(ctx) - if err != nil { - f.Errorf("unable to unlock deposits after "+ - "payment deadline: %v", err) - continue - } - depositsUnlocked = true + unlockDepositsAfterInvoiceCancel( + "payment deadline expired", + ) + + case <-unlockRetryChan: + unlockRetryChan = nil + + unlockDepositsAfterInvoiceCancel("retry") case riskAccepted, ok := <-riskAcceptedChan: if !ok { @@ -906,7 +1122,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 { @@ -921,47 +1146,12 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, continue } - cancelInvoiceSubscription() - f.cancelSwapInvoice() - - return f.HandleError(errors.New( - "server rejected confirmation risk wait", - )) + handleRiskRejected("risk rejection") case currentHeight := <-blockChan: - if !deadlineStarted && - invoice.State != invoices.ContractCanceled { - - err = f.refreshSelectedDeposits(ctx) - if err != nil { - f.Warnf("unable to refresh selected "+ - "deposits for legacy confirmation "+ - "fallback: %v", err) - } else { - depositConfirmationHeights := - selectedDepositConfirmationHeights( - f.loopIn, - ) - - if legacyMinConfsReached( - f.loopIn.DepositOutpoints, - depositConfirmationHeights, - currentHeight, - ) { - - // This fallback is a compatibility path for - // servers that do not send confirmation-risk - // notifications. Reaching legacy MinConfs is - // treated as synthetic risk acceptance, so the - // payment window starts here just as it would - // when a modern server sends an acceptance - // notification. - startPaymentDeadline( - "legacy confirmation fallback", - ) - } - } - } + startLegacyFallback( + "legacy confirmation fallback", currentHeight, + ) // If the htlc is confirmed but blockChan fires before // htlcConfChan, we would wrongfully assume that the @@ -1000,16 +1190,7 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // If the htlc has confirmed and the timeout path has // opened up we sweep the funds back to us. - err = f.cfg.DepositManager.TransitionDeposits( - ctx, f.loopIn.Deposits, - deposit.OnSweepingHtlcTimeout, - deposit.SweepHtlcTimeout, - ) - if err != nil { - log.Errorf("unable to transition "+ - "deposits to the htlc timeout "+ - "sweeping state: %v", err) - } + transitionDepositsToHtlcTimeout("htlc timeout") return OnSweepHtlcTimeout @@ -1037,7 +1218,7 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, f.Errorf("invoice subscription error: %v", err) case <-ctx.Done(): - return f.HandleError(ctx.Err()) + return fsm.NoOp } } } diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index d8ecd9ad..de54791b 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -17,6 +17,7 @@ import ( "github.com/lightninglabs/loop/swap" "github.com/lightninglabs/loop/swapserverrpc" "github.com/lightninglabs/loop/test" + "github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/invoices" "github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/zpay32" @@ -45,7 +46,7 @@ func TestMonitorInvoiceAndHtlcTxReRegistersOnConfErr(t *testing.T) { SwapHash: swapHash, HtlcCltvExpiry: 2_000, InitiationHeight: uint32(mockLnd.Height), - InitiationTime: time.Now(), + InitiationTime: time.Now().Add(-time.Hour), ProtocolVersion: version.ProtocolVersion_V0, ClientPubkey: clientKey.PubKey(), ServerPubkey: serverKey.PubKey(), @@ -128,6 +129,97 @@ func TestMonitorInvoiceAndHtlcTxReRegistersOnConfErr(t *testing.T) { } } +// TestMonitorInvoiceAndHtlcTxShutdownDoesNotUnlock verifies that daemon +// shutdown exits the monitor action without treating the swap as failed. +func TestMonitorInvoiceAndHtlcTxShutdownDoesNotUnlock(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + runCtx, stop := context.WithCancel(ctx) + + mockLnd := test.NewMockLnd() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + swapHash := lntypes.Hash{1, 2, 4} + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: 2_000, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now().Add(-time.Hour), + ProtocolVersion: version.ProtocolVersion_V0, + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + PaymentTimeoutSeconds: 3_600, + Deposits: []*deposit.Deposit{{ + Value: 200_000, + }}, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + }) + + depositMgr := &recordingDepositManager{ + transitionChan: make(chan depositTransition, 1), + } + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: depositMgr, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + } + + f, err := NewFSM(runCtx, loopIn, cfg, false) + require.NoError(t, err) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(runCtx, nil) + }() + + waitForMonitorSubscriptions(t, ctx, mockLnd) + + stop() + + select { + case event := <-resultChan: + require.Equal(t, fsm.NoOp, event) + + case <-ctx.Done(): + t.Fatalf("monitor action did not exit: %v", ctx.Err()) + } + + require.NoError(t, f.LastActionError) + + select { + case transition := <-depositMgr.transitionChan: + t.Fatalf("deposit transition on shutdown: %v", transition) + + default: + } + + select { + case hash := <-mockLnd.FailInvoiceChannel: + t.Fatalf("invoice canceled on shutdown: %v", hash) + + default: + } +} + // TestInitHtlcActionPreservesRouteHints asserts that static-address loop-in // propagates explicit route hints into the encoded swap invoice sent to the // server. @@ -376,16 +468,17 @@ func TestMonitorInvoiceAndHtlcTxStartsDeadlineOnRiskAccepted(t *testing.T) { cancel() select { case event := <-resultChan: - require.Equal(t, fsm.OnError, event) + require.Equal(t, fsm.NoOp, 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) { +// 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() @@ -396,9 +489,124 @@ func TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected(t *testing.T) { serverKey, err := btcec.NewPrivateKey() require.NoError(t, err) - swapHash := lntypes.Hash{5, 6, 7} + swapHash := lntypes.Hash{4, 5, 7} depositOutpoint := wire.OutPoint{ - Hash: chainhash.Hash{9}, + 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.NoOp, 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, } @@ -426,11 +634,19 @@ func TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected(t *testing.T) { }) notificationMgr := &mockNotificationManager{ - riskRejected: make( + riskAccepted: make( chan *swapserverrpc. - ServerStaticLoopInRiskRejectedNotification, 1, + ServerStaticLoopInRiskAcceptedNotification, 1, ), } + store := &recordingRiskStore{ + mockStore: &mockStore{ + loopIns: map[lntypes.Hash]*StaticAddressLoopIn{ + swapHash: {}, + }, + }, + decisions: make(chan ConfirmationRiskDecision, 1), + } cfg := &Config{ AddressManager: &mockAddressManager{ @@ -446,6 +662,116 @@ func TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected(t *testing.T) { 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.NoOp, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } +} + +// TestMonitorInvoiceAndHtlcTxContinuesAfterRiskRejected verifies that a +// server-side confirmation risk rejection cancels the invoice and unlocks the +// deposits, but keeps monitoring for a possible server-published HTLC until the +// timeout path is resolved. +func TestMonitorInvoiceAndHtlcTxContinuesAfterRiskRejected(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, 7} + depositOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{9}, + Index: 0, + } + + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: mockLnd.Height, + 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{ + riskRejected: make( + chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification, 1, + ), + } + + depositMgr := &recordingDepositManager{ + transitionChan: make(chan depositTransition, 1), + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: depositMgr, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + NotificationManager: notificationMgr, } f, err := NewFSM(ctx, loopIn, cfg, false) @@ -470,6 +796,250 @@ func TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected(t *testing.T) { t.Fatalf("invoice was not canceled: %v", ctx.Err()) } + transition := nextDepositTransition(t, ctx, depositMgr) + require.Equal(t, fsm.OnError, transition.event) + require.Equal(t, deposit.Deposited, transition.state) + + select { + case event := <-resultChan: + t.Fatalf("monitor action exited before HTLC timeout: %v", + event) + + case <-time.After(100 * time.Millisecond): + } + + require.NoError(t, mockLnd.NotifyHeight(mockLnd.Height+1)) + + select { + case event := <-resultChan: + require.Equal(t, OnSwapTimedOut, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } + + require.Len(t, depositMgr.transitions, 1) +} + +// TestMonitorInvoiceAndHtlcTxLocksDepositsWhenHtlcConfirmsAfterRiskRejected +// verifies that deposits unlocked after a risk rejection are moved back into the +// HTLC timeout sweep state if the server later publishes the HTLC. +func TestMonitorInvoiceAndHtlcTxLocksDepositsWhenHtlcConfirmsAfterRiskRejected( + 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{10}, + Index: 0, + } + + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: mockLnd.Height + 2, + 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{ + riskRejected: make( + chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification, 1, + ), + } + depositMgr := &recordingDepositManager{ + transitionChan: make(chan depositTransition, 2), + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: depositMgr, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + NotificationManager: notificationMgr, + } + + 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.riskRejected <- &swapserverrpc. + ServerStaticLoopInRiskRejectedNotification{ + SwapHash: swapHash[:], + } + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, swapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } + + unlockTransition := nextDepositTransition(t, ctx, depositMgr) + require.Equal(t, fsm.OnError, unlockTransition.event) + require.Equal(t, deposit.Deposited, unlockTransition.state) + + htlc, err := loopIn.getHtlc(mockLnd.ChainParams) + require.NoError(t, err) + + htlcTx := wire.NewMsgTx(2) + htlcTx.AddTxOut(&wire.TxOut{ + PkScript: htlc.PkScript, + }) + + select { + case mockLnd.ConfChannel <- &chainntnfs.TxConfirmation{Tx: htlcTx}: + + case <-ctx.Done(): + t.Fatalf("htlc confirmation was not consumed: %v", ctx.Err()) + } + + timeoutTransition := nextDepositTransition(t, ctx, depositMgr) + require.Equal( + t, deposit.OnSweepingHtlcTimeout, timeoutTransition.event, + ) + require.Equal(t, deposit.SweepHtlcTimeout, timeoutTransition.state) + + select { + case event := <-resultChan: + t.Fatalf("monitor action exited before HTLC timeout: %v", + event) + + case <-time.After(100 * time.Millisecond): + } + + require.NoError(t, mockLnd.NotifyHeight(loopIn.HtlcCltvExpiry+1)) + + select { + case event := <-resultChan: + require.Equal(t, OnSweepHtlcTimeout, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } + + require.Len(t, depositMgr.transitions, 2) +} + +// 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.NoOp, event) @@ -479,6 +1049,113 @@ func TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected(t *testing.T) { } } +// TestMonitorInvoiceAndHtlcTxRecoversRejectedRiskDecision verifies that a +// persisted risk rejection still cancels and unlocks after restart without +// dropping HTLC timeout monitoring. +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: mockLnd.Height, + 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, + }) + + depositMgr := &recordingDepositManager{ + transitionChan: make(chan depositTransition, 1), + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: depositMgr, + 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()) + } + + transition := nextDepositTransition(t, ctx, depositMgr) + require.Equal(t, fsm.OnError, transition.event) + require.Equal(t, deposit.Deposited, transition.state) + + select { + case event := <-resultChan: + t.Fatalf("monitor action exited before HTLC timeout: %v", + event) + + case <-time.After(100 * time.Millisecond): + } + + require.NoError(t, mockLnd.NotifyHeight(mockLnd.Height+1)) + + select { + case event := <-resultChan: + require.Equal(t, OnSwapTimedOut, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } + + require.Len(t, depositMgr.transitions, 1) +} + // 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 @@ -745,6 +1422,14 @@ func TestMonitorInvoiceAndHtlcTxStartsDeadlineAtLegacyMinConfs(t *testing.T) { require.NoError(t, mockLnd.NotifyHeight(mockLnd.Height)) + select { + case hash := <-mockLnd.FailInvoiceChannel: + t.Fatalf("invoice canceled immediately after deposit "+ + "confirmation: %v", hash) + + case <-time.After(200 * time.Millisecond): + } + select { case hash := <-mockLnd.FailInvoiceChannel: require.Equal(t, swapHash, hash) @@ -791,7 +1476,7 @@ func TestMonitorInvoiceAndHtlcTxStartsLegacyFallbackWithNotificationManager( SwapHash: swapHash, HtlcCltvExpiry: 2_000, InitiationHeight: uint32(mockLnd.Height), - InitiationTime: time.Now(), + InitiationTime: time.Now().Add(-time.Hour), ProtocolVersion: version.ProtocolVersion_V0, ClientPubkey: clientKey.PubKey(), ServerPubkey: serverKey.PubKey(), @@ -1206,6 +1891,108 @@ func TestMonitorInvoiceAndHtlcTxUnlocksOnHtlcTimeoutWithoutDeadline( require.Equal(t, []fsm.StateType{deposit.Deposited}, depositMgr.states) } +// TestMonitorInvoiceAndHtlcTxRetriesDeadlineUnlock verifies that a temporary +// deposit unlock failure after the payment deadline is retried without waiting +// for the HTLC timeout path or daemon restart. +func TestMonitorInvoiceAndHtlcTxRetriesDeadlineUnlock(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + oldRetryDelay := paymentDeadlineUnlockRetryDelay + paymentDeadlineUnlockRetryDelay = 20 * time.Millisecond + t.Cleanup(func() { + paymentDeadlineUnlockRetryDelay = oldRetryDelay + }) + + mockLnd := test.NewMockLnd() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + swapHash := lntypes.Hash{10, 11, 13} + depositOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{11}, + 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, + }) + + depositMgr := &recordingDepositManager{ + errs: []error{ + errors.New("temporary unlock failure"), + }, + transitionChan: make(chan depositTransition, 2), + } + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: depositMgr, + 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) + + firstTransition := nextDepositTransition(t, ctx, depositMgr) + require.Equal(t, fsm.OnError, firstTransition.event) + require.Equal(t, deposit.Deposited, firstTransition.state) + + secondTransition := nextDepositTransition(t, ctx, depositMgr) + require.Equal(t, fsm.OnError, secondTransition.event) + require.Equal(t, deposit.Deposited, secondTransition.state) + + cancel() + select { + case event := <-resultChan: + require.Equal(t, fsm.NoOp, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } + + require.Len(t, depositMgr.transitions, 2) +} + // waitForMonitorSubscriptions waits until invoice and HTLC watchers are active. func waitForMonitorSubscriptions(t *testing.T, ctx context.Context, mockLnd *test.LndMockServices) { @@ -1225,6 +2012,25 @@ func waitForMonitorSubscriptions(t *testing.T, ctx context.Context, } } +func nextDepositTransition(t *testing.T, ctx context.Context, + depositMgr *recordingDepositManager) depositTransition { + + t.Helper() + + select { + case transition := <-depositMgr.transitionChan: + return transition + + case <-ctx.Done(): + t.Fatalf("deposit transition not observed: %v", ctx.Err()) + + case <-time.After(time.Second): + t.Fatal("deposit transition not observed") + } + + return depositTransition{} +} + // TestOriginalDepositOutpointUnavailableRequiresMissingTxOut verifies that a // present txout does not trigger the RBF cancellation path. func TestOriginalDepositOutpointUnavailableRequiresMissingTxOut(t *testing.T) { @@ -1619,10 +2425,12 @@ type recordingDepositManager struct { noopDepositManager err error + errs []error transitions []depositTransition - events []fsm.EventType - states []fsm.StateType + transitionChan chan depositTransition + events []fsm.EventType + states []fsm.StateType } // TransitionDeposits records the transition and returns the configured error. @@ -1630,17 +2438,57 @@ func (r *recordingDepositManager) TransitionDeposits(_ context.Context, deposits []*deposit.Deposit, event fsm.EventType, state fsm.StateType) error { - r.transitions = append(r.transitions, depositTransition{ + transition := depositTransition{ deposits: deposits, event: event, state: state, - }) + } + + r.transitions = append(r.transitions, transition) r.events = append(r.events, event) r.states = append(r.states, state) + if r.transitionChan != nil { + r.transitionChan <- transition + } + + if len(r.errs) > 0 { + err := r.errs[0] + r.errs = r.errs[1:] + + return err + } + return r.err } +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 f2637885..f9fb01bf 100644 --- a/staticaddr/loopin/interface.go +++ b/staticaddr/loopin/interface.go @@ -91,6 +91,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 cb7eb7e7..08da7205 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -464,6 +464,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 e90d1bdf..0dca396c 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 f0c10968..602d66d0 100644 --- a/staticaddr/loopin/sql_store_test.go +++ b/staticaddr/loopin/sql_store_test.go @@ -349,6 +349,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)