GlcoinCore/loop
1
0
Fork 0
mirror of https://github.com/lightninglabs/loop.git synced 2026-08-13 12:33:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

staticaddr/loopin: persist risk decisions

Browse source 
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.
This commit is contained in:
Slyghtning 2026-06-26 12:54:30 +02:00
parent d045d5ecd2
commit fc3fba7c23
No known key found for this signature in database
GPG key ID: F82D456EA023C9BF
17 changed files with 1713 additions and 286 deletions
Download patch file Download diff file Expand all files Collapse all files

26
loopd/daemon.go
View file

@ -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 := &notifications.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,

8
loopd/swapclient_server_test.go
View file

@ -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) {

3
loopdb/sqlc/migrations/000021_static_loopin_risk_decision.down.sql Normal file
View file

@ -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;

9
loopdb/sqlc/migrations/000021_static_loopin_risk_decision.up.sql Normal file
View file

@ -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;

26
loopdb/sqlc/models.go
View file

@ -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 {

1
loopdb/sqlc/querier.go
View file

@ -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

9
loopdb/sqlc/queries/static_address_loopin.sql
View file

@ -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

152
loopdb/sqlc/static_address_loopin.sql.go
View file

@ -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

285
notifications/manager.go
View file

@ -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

184
notifications/manager_test.go
View file

@ -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{

305
staticaddr/loopin/actions.go
View file

@ -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
}
}
}

876
staticaddr/loopin/actions_test.go
View file

@ -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 {

5
staticaddr/loopin/interface.go
View file

@ -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)

26
staticaddr/loopin/loopin.go
View file

@ -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

7
staticaddr/loopin/manager_test.go
View file

@ -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) {

52
staticaddr/loopin/sql_store.go
View file

@ -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]

25
staticaddr/loopin/sql_store_test.go
View file

@ -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)