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)