mirror of
https://github.com/lightninglabs/loop.git
synced 2026-08-15 12:50:28 +02:00
staticaddr/loopin: handle risk rejection notification
Add client handling for the server's static loop-in risk-rejected notification. If the server aborts confirmation-risk waiting before payment, the client fails the local swap instead of waiting for a payment deadline that will never start. Cache rejected notifications by swap hash using the same replay path as accepted notifications, and clear the opposite cached state when a final risk decision is received. This keeps reconnect and subscription-order races from stranding the client in the risk wait.
This commit is contained in:
Slyghtning
parent
aa77682e07
commit
e6404bb867
5 changed files with 620 additions and 9 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue