diff --git a/notifications/manager.go b/notifications/manager.go index 1b481a44..afe67375 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -30,6 +30,10 @@ const ( // static loop in confirmation risk acceptance. NotificationTypeStaticLoopInRiskAccepted + // NotificationTypeStaticLoopInRiskRejected is the notification type for + // static loop in confirmation risk rejection. + NotificationTypeStaticLoopInRiskRejected + // NotificationTypeUnfinishedSwap is the notification type for unfinished // swap notifications. NotificationTypeUnfinishedSwap @@ -83,6 +87,9 @@ type Manager struct { staticLoopInRiskAccepted map[lntypes.Hash]*swapserverrpc. ServerStaticLoopInRiskAcceptedNotification + + staticLoopInRiskRejected map[lntypes.Hash]*swapserverrpc. + ServerStaticLoopInRiskRejectedNotification } // NewManager creates a new notification manager. @@ -99,12 +106,17 @@ func NewManager(cfg *Config) *Manager { map[lntypes.Hash]*swapserverrpc. ServerStaticLoopInRiskAcceptedNotification, ), + staticLoopInRiskRejected: make( + map[lntypes.Hash]*swapserverrpc. + ServerStaticLoopInRiskRejectedNotification, + ), } } type subscriber struct { subCtx context.Context recvChan any + swapHash *lntypes.Hash } // SubscribeReservations subscribes to the reservation notifications. @@ -167,6 +179,7 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, sub := subscriber{ subCtx: ctx, recvChan: notifChan, + swapHash: &swapHash, } m.Lock() @@ -190,6 +203,43 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, return notifChan } +// 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, + ) + + 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. func (m *Manager) SubscribeUnfinishedSwaps(ctx context.Context, ) <-chan *swapserverrpc.ServerUnfinishedSwapNotification { @@ -376,24 +426,37 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. } case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInRiskAccepted: // nolint: lll - // We'll forward the static loop in risk accepted notification to all - // subscribers. + // 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 + ) if riskAcceptedNtfn != nil { - swapHash, err := lntypes.MakeHash(riskAcceptedNtfn.SwapHash) + hash, err := lntypes.MakeHash(riskAcceptedNtfn.SwapHash) if err != nil { log.Warnf("Received invalid static loop in risk "+ "accepted notification: %v", err) } else { - m.staticLoopInRiskAccepted[swapHash] = + swapHash = hash + hasSwapHash = true + m.staticLoopInRiskAccepted[hash] = 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) @@ -406,6 +469,50 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. } } + 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 + ) + 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] = + 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") + } + } + case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll // We'll forward the unfinished swap notification to all // subscribers. diff --git a/notifications/manager_test.go b/notifications/manager_test.go index 9f8c66d1..2befa3cf 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -203,6 +203,88 @@ func unfinishedSwapNotification( } } +func staticLoopInRiskAcceptedNotification( + swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { + + return &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_StaticLoopInRiskAccepted{ + StaticLoopInRiskAccepted: &swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification{ + SwapHash: swapHash[:], + }, + }, + } +} + +// staticLoopInRiskRejectedNotification builds a risk rejected notification. +func staticLoopInRiskRejectedNotification( + swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { + + return &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_StaticLoopInRiskRejected{ + StaticLoopInRiskRejected: &swapserverrpc. + ServerStaticLoopInRiskRejectedNotification{ + SwapHash: swapHash[:], + }, + }, + } +} + +type staticLoopInRiskNotification interface { + GetSwapHash() []byte +} + +// assertStaticLoopInRiskNotificationSwapScoped checks swap-scoped fanout. +func assertStaticLoopInRiskNotificationSwapScoped[ + T staticLoopInRiskNotification](t *testing.T, + subscribe func(*Manager, context.Context, lntypes.Hash) <-chan T, + notification func(lntypes.Hash) *swapserverrpc. + SubscribeNotificationsResponse, label string, + swapHashA, swapHashB lntypes.Hash) { + + t.Helper() + + mgr := NewManager(&Config{}) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + subChanA := subscribe(mgr, subCtx, swapHashA) + subChanB := subscribe(mgr, subCtx, swapHashB) + + mgr.handleNotification(notification(swapHashA)) + + select { + case received := <-subChanA: + require.Equal(t, swapHashA[:], received.GetSwapHash()) + + case <-time.After(time.Second): + t.Fatalf("did not receive first swap risk %s notification", + label) + } + + select { + case received := <-subChanB: + t.Fatalf("second swap received wrong notification: %x", + received.GetSwapHash()) + + default: + } + + mgr.handleNotification(notification(swapHashB)) + + select { + case received := <-subChanB: + require.Equal(t, swapHashB[:], received.GetSwapHash()) + + case <-time.After(time.Second): + t.Fatalf("did not receive second swap risk %s notification", + label) + } +} + // TestManager_SlowSubscriberDoesNotBlock tests that a subscriber with a full // notification channel does not block delivery to other subscribers. func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { @@ -334,6 +416,22 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) { } } +// TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped verifies that a +// notification for one swap does not occupy another swap's subscriber channel. +func TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped(t *testing.T) { + t.Parallel() + + assertStaticLoopInRiskNotificationSwapScoped( + t, func(m *Manager, ctx context.Context, + swapHash lntypes.Hash) <-chan *swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification { + + return m.SubscribeStaticLoopInRiskAccepted(ctx, swapHash) + }, staticLoopInRiskAcceptedNotification, "accepted", + lntypes.Hash{0x04, 0x05}, lntypes.Hash{0x06, 0x07}, + ) +} + // TestManager_StaticLoopInRiskAcceptedNotificationReplay tests that the Manager // replays a risk accepted notification that arrives before the swap-specific // subscriber is registered. @@ -369,6 +467,92 @@ func TestManager_StaticLoopInRiskAcceptedNotificationReplay(t *testing.T) { } } +// TestManager_StaticLoopInRiskRejectedNotification tests that the Manager +// forwards static loop in risk rejected notifications to subscribers. +func TestManager_StaticLoopInRiskRejectedNotification(t *testing.T) { + t.Parallel() + + mgr := NewManager(&Config{}) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + swapHash := lntypes.Hash{0x08, 0x09} + + subChan := mgr.SubscribeStaticLoopInRiskRejected(subCtx, swapHash) + + mgr.handleNotification( + &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_StaticLoopInRiskRejected{ + StaticLoopInRiskRejected: &swapserverrpc. + ServerStaticLoopInRiskRejectedNotification{ + SwapHash: swapHash[:], + }, + }, + }, + ) + + select { + case received := <-subChan: + require.Equal(t, swapHash[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("did not receive risk rejected notification") + } +} + +// TestManager_StaticLoopInRiskRejectedNotificationSwapScoped verifies that a +// notification for one swap does not occupy another swap's subscriber channel. +func TestManager_StaticLoopInRiskRejectedNotificationSwapScoped(t *testing.T) { + t.Parallel() + + assertStaticLoopInRiskNotificationSwapScoped( + t, func(m *Manager, ctx context.Context, + swapHash lntypes.Hash) <-chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification { + + return m.SubscribeStaticLoopInRiskRejected(ctx, swapHash) + }, staticLoopInRiskRejectedNotification, "rejected", + lntypes.Hash{0x08, 0x09}, lntypes.Hash{0x0a, 0x0b}, + ) +} + +// TestManager_StaticLoopInRiskRejectedNotificationReplay tests that the Manager +// replays a risk rejected notification that arrives before the swap-specific +// subscriber is registered. +func TestManager_StaticLoopInRiskRejectedNotificationReplay(t *testing.T) { + t.Parallel() + + mgr := NewManager(&Config{}) + + swapHash := lntypes.Hash{0x0a, 0x0b} + mgr.handleNotification( + &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_StaticLoopInRiskRejected{ + StaticLoopInRiskRejected: &swapserverrpc. + ServerStaticLoopInRiskRejectedNotification{ + SwapHash: swapHash[:], + }, + }, + }, + ) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + subChan := mgr.SubscribeStaticLoopInRiskRejected(subCtx, swapHash) + + select { + case received := <-subChan: + require.Equal(t, swapHash[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("did not replay risk rejected notification") + } +} + // TestManager_Backoff verifies that repeated failures in // subscribeNotifications cause the Manager to space out subscription attempts // via a predictable incremental backoff. diff --git a/staticaddr/loopin/actions.go b/staticaddr/loopin/actions.go index a5908c3f..7e13dc29 100644 --- a/staticaddr/loopin/actions.go +++ b/staticaddr/loopin/actions.go @@ -690,17 +690,23 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, var ( riskAcceptedChan <-chan *swapserverrpc. ServerStaticLoopInRiskAcceptedNotification - cancelRiskAcceptedSubscription = func() {} + riskRejectedChan <-chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification + cancelRiskNotificationSubscriptions = func() {} ) if f.cfg.NotificationManager != nil { - acceptedCtx, cancel := context.WithCancel(ctx) - cancelRiskAcceptedSubscription = cancel + notificationCtx, cancel := context.WithCancel(ctx) + cancelRiskNotificationSubscriptions = cancel riskAcceptedChan = f.cfg.NotificationManager. SubscribeStaticLoopInRiskAccepted( - acceptedCtx, f.loopIn.SwapHash, + notificationCtx, f.loopIn.SwapHash, + ) + riskRejectedChan = f.cfg.NotificationManager. + SubscribeStaticLoopInRiskRejected( + notificationCtx, f.loopIn.SwapHash, ) } - defer cancelRiskAcceptedSubscription() + defer cancelRiskNotificationSubscriptions() htlcConfirmed := false depositsUnlocked := false @@ -827,6 +833,26 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, startPaymentDeadline("risk accepted notification") + case riskRejected, ok := <-riskRejectedChan: + if !ok { + riskRejectedChan = nil + continue + } + + if !bytes.Equal( + riskRejected.SwapHash, f.loopIn.SwapHash[:], + ) { + + continue + } + + cancelInvoiceSubscription() + f.cancelSwapInvoice(ctx) + + return f.HandleError(errors.New( + "server rejected confirmation risk wait", + )) + case currentHeight := <-blockChan: depositConfirmationHeights := selectedDepositConfirmationHeights(f.loopIn) diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index 7debecb0..40e04b36 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -383,6 +383,284 @@ func TestMonitorInvoiceAndHtlcTxStartsDeadlineOnRiskAccepted(t *testing.T) { } } +// TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected verifies that a server-side +// confirmation risk rejection is terminal for the client monitor action. +func TestMonitorInvoiceAndHtlcTxCancelsOnRiskRejected(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: 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{ + riskRejected: make( + chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification, 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, + } + + 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{ // nolint: lll + SwapHash: swapHash[:], + } + + 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 +// server, the outpoint can disappear because the server published the expected +// HTLC transaction. +func TestMonitorInvoiceAndHtlcTxDoesNotCancelWhenOriginalOutpointVanishes( + 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, 7, 9} + depositOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{10}, + 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, + }) + + txOutChecker := &testTxOutChecker{} + 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, + TxOutChecker: txOutChecker, + } + + 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 should not have been canceled: %v", hash) + + case <-time.After(200 * time.Millisecond): + } + + cancel() + select { + case event := <-resultChan: + require.Equal(t, fsm.OnError, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } + + require.Empty(t, txOutChecker.outpoints) + require.Empty(t, txOutChecker.includeMempool) +} + +// TestMonitorInvoiceAndHtlcTxDoesNotCancelAcceptedInvoiceForMissingOutpoint +// verifies that the outpoint-vanished fallback is only active before payment +// has started. Once the invoice is accepted, the original deposit may disappear +// because the server has moved forward with the swap. +func TestMonitorInvoiceAndHtlcTxDoesNotCancelAcceptedInvoiceForMissingOutpoint( + 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{6, 8, 10} + 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: 3_600, + DepositOutpoints: []string{ + depositOutpoint.String(), + }, + Deposits: []*deposit.Deposit{{ + OutPoint: depositOutpoint, + }}, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractAccepted, + }) + + 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, + TxOutChecker: &testTxOutChecker{}, + } + + 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 should not have been canceled: %v", hash) + + case <-time.After(200 * time.Millisecond): + } + + cancel() + select { + case <-resultChan: + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } +} + // TestMonitorInvoiceAndHtlcTxStartsDeadlineAtLegacyMinConfs verifies that the // monitor action preserves the legacy payment deadline fallback when no risk // notification manager is available. @@ -1030,6 +1308,7 @@ func (r *recordingDepositManager) TransitionDeposits(_ context.Context, // monitor actions. type mockNotificationManager struct { riskAccepted chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification + riskRejected chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification } // SubscribeStaticLoopInSweepRequests implements NotificationManager. @@ -1047,6 +1326,14 @@ func (m *mockNotificationManager) SubscribeStaticLoopInRiskAccepted( return m.riskAccepted } +// SubscribeStaticLoopInRiskRejected implements NotificationManager. +func (m *mockNotificationManager) SubscribeStaticLoopInRiskRejected( + context.Context, lntypes.Hash, +) <-chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification { + + return m.riskRejected +} + type testTxOutChecker struct { txOut *wire.TxOut err error diff --git a/staticaddr/loopin/interface.go b/staticaddr/loopin/interface.go index 146db2a1..f96d08e5 100644 --- a/staticaddr/loopin/interface.go +++ b/staticaddr/loopin/interface.go @@ -129,4 +129,11 @@ type NotificationManager interface { SubscribeStaticLoopInRiskAccepted( ctx context.Context, swapHash lntypes.Hash, ) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification + + // SubscribeStaticLoopInRiskRejected subscribes to static loop in risk + // rejected notifications. These are sent by the server if it aborts the + // confirmation risk wait before payment. + SubscribeStaticLoopInRiskRejected( + ctx context.Context, swapHash lntypes.Hash, + ) <-chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification }