diff --git a/staticaddr/loopin/risk_watcher.go b/staticaddr/loopin/risk_watcher.go new file mode 100644 index 00000000..4da95a20 --- /dev/null +++ b/staticaddr/loopin/risk_watcher.go @@ -0,0 +1,182 @@ +package loopin + +import ( + "bytes" + "context" + "time" + + "github.com/lightningnetwork/lnd/lntypes" +) + +// confirmationRiskUpdate is the normalized result of a server confirmation-risk +// notification. +type confirmationRiskUpdate struct { + decision ConfirmationRiskDecision + reason string +} + +// confirmationRiskWatcher normalizes static loop-in confirmation risk +// notifications and restores the durable decision timestamp. +type confirmationRiskWatcher struct { + swapHash lntypes.Hash + store StaticAddressLoopInStore + notificationManager NotificationManager + logWarnf func(string, ...any) +} + +// newConfirmationRiskWatcher creates a helper that handles confirmation-risk +// notification plumbing for a single static loop-in swap. +func newConfirmationRiskWatcher(cfg *Config, swapHash lntypes.Hash, + warnf func(string, ...any)) *confirmationRiskWatcher { + + return &confirmationRiskWatcher{ + swapHash: swapHash, + store: cfg.Store, + notificationManager: cfg.NotificationManager, + logWarnf: warnf, + } +} + +// warnf logs through the FSM-scoped logger when one is available. +func (w *confirmationRiskWatcher) warnf(format string, args ...any) { + if w.logWarnf != nil { + w.logWarnf(format, args...) + + return + } + + log.Warnf(format, args...) +} + +// subscribe subscribes to accepted and rejected confirmation-risk notifications +// and emits normalized updates for the watcher's swap hash. +func (w *confirmationRiskWatcher) subscribe(ctx context.Context) ( + <-chan confirmationRiskUpdate, func()) { + + if w.notificationManager == nil { + return nil, func() {} + } + + notificationCtx, cancel := context.WithCancel(ctx) + riskAcceptedChan := w.notificationManager.SubscribeStaticLoopInRiskAccepted( + notificationCtx, w.swapHash, + ) + riskRejectedChan := w.notificationManager.SubscribeStaticLoopInRiskRejected( + notificationCtx, w.swapHash, + ) + + riskUpdates := make(chan confirmationRiskUpdate, 1) + go func() { + defer close(riskUpdates) + + for { + select { + case riskAccepted, ok := <-riskAcceptedChan: + if !ok { + riskAcceptedChan = nil + continue + } + + if riskAccepted == nil || !bytes.Equal( + riskAccepted.SwapHash, w.swapHash[:], + ) { + + continue + } + + update := confirmationRiskUpdate{ + decision: ConfirmationRiskDecisionAccepted, + reason: "risk accepted notification", + } + select { + case riskUpdates <- update: + case <-notificationCtx.Done(): + return + } + + case riskRejected, ok := <-riskRejectedChan: + if !ok { + riskRejectedChan = nil + continue + } + + if riskRejected == nil || !bytes.Equal( + riskRejected.SwapHash, w.swapHash[:], + ) { + + continue + } + + update := confirmationRiskUpdate{ + decision: ConfirmationRiskDecisionRejected, + reason: "risk rejection", + } + select { + case riskUpdates <- update: + case <-notificationCtx.Done(): + return + } + + case <-notificationCtx.Done(): + return + } + } + }() + + return riskUpdates, cancel +} + +// decisionTime returns the durable decision timestamp, recording the decision +// first if the notification was replayed before it could be persisted. +func (w *confirmationRiskWatcher) decisionTime(ctx context.Context, + decision ConfirmationRiskDecision) time.Time { + + now := time.Now() + if w.store == nil { + return now + } + + storedLoopIn, err := w.store.GetLoopInByHash(ctx, w.swapHash) + if err != nil { + w.warnf("unable to reload persisted risk decision for swap %v: %v", + w.swapHash, err) + + return now + } + + if storedLoopIn == nil { + return now + } + + hasPersistedDecision := + storedLoopIn.ConfirmationRiskDecision == decision && + !storedLoopIn.ConfirmationRiskDecisionTime.IsZero() + + if !hasPersistedDecision { + err = w.store.RecordStaticAddressRiskDecision( + ctx, w.swapHash, decision, + ) + if err != nil { + w.warnf("unable to persist replayed risk decision for "+ + "swap %v: %v", w.swapHash, err) + + return now + } + + storedLoopIn, err = w.store.GetLoopInByHash(ctx, w.swapHash) + if err != nil { + w.warnf("unable to reload persisted risk decision for "+ + "swap %v: %v", w.swapHash, err) + + return now + } + if storedLoopIn == nil || + storedLoopIn.ConfirmationRiskDecision != decision || + storedLoopIn.ConfirmationRiskDecisionTime.IsZero() { + + return now + } + } + + return storedLoopIn.ConfirmationRiskDecisionTime +} diff --git a/staticaddr/loopin/risk_watcher_test.go b/staticaddr/loopin/risk_watcher_test.go new file mode 100644 index 00000000..4f907615 --- /dev/null +++ b/staticaddr/loopin/risk_watcher_test.go @@ -0,0 +1,121 @@ +package loopin + +import ( + "context" + "testing" + "time" + + "github.com/lightninglabs/loop/swapserverrpc" + "github.com/lightningnetwork/lnd/lntypes" + "github.com/stretchr/testify/require" +) + +// TestConfirmationRiskWatcherSubscribeFiltersSwapHash verifies that the watcher +// only emits normalized decisions for the swap it was created for. +func TestConfirmationRiskWatcherSubscribeFiltersSwapHash(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + + swapHash := lntypes.Hash{1, 2, 3} + otherHash := lntypes.Hash{3, 2, 1} + notificationMgr := &mockNotificationManager{ + riskAccepted: make( + chan *swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification, 1, + ), + riskRejected: make( + chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification, 1, + ), + } + + watcher := newConfirmationRiskWatcher( + &Config{NotificationManager: notificationMgr}, swapHash, + t.Logf, + ) + updates, stop := watcher.subscribe(ctx) + defer stop() + + notificationMgr.riskAccepted <- &swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification{ + SwapHash: otherHash[:], + } + + select { + case update := <-updates: + t.Fatalf("received wrong-hash risk update: %v", update) + + case <-time.After(100 * time.Millisecond): + } + + notificationMgr.riskRejected <- &swapserverrpc. + ServerStaticLoopInRiskRejectedNotification{ + SwapHash: swapHash[:], + } + + select { + case update := <-updates: + require.Equal(t, ConfirmationRiskDecisionRejected, + update.decision) + require.Equal(t, "risk rejection", update.reason) + + case <-ctx.Done(): + t.Fatalf("risk update not received: %v", ctx.Err()) + } +} + +// TestConfirmationRiskWatcherDecisionTimeRestoration verifies that the watcher +// preserves existing persisted decision timestamps and records missing ones. +func TestConfirmationRiskWatcherDecisionTimeRestoration(t *testing.T) { + t.Parallel() + + ctx := t.Context() + swapHash := lntypes.Hash{4, 5, 6} + decisionTime := time.Unix(123, 0).UTC() + store := &recordingRiskStore{ + mockStore: &mockStore{ + loopIns: map[lntypes.Hash]*StaticAddressLoopIn{ + swapHash: { + ConfirmationRiskDecision: ConfirmationRiskDecisionAccepted, + ConfirmationRiskDecisionTime: decisionTime, + }, + }, + }, + decisions: make(chan ConfirmationRiskDecision, 1), + } + + watcher := newConfirmationRiskWatcher(&Config{Store: store}, swapHash, + t.Logf) + restoredTime := watcher.decisionTime( + ctx, ConfirmationRiskDecisionAccepted, + ) + require.True(t, restoredTime.Equal(decisionTime)) + + select { + case decision := <-store.decisions: + t.Fatalf("persisted already-recorded decision: %v", decision) + + default: + } + + store.loopIns[swapHash] = &StaticAddressLoopIn{} + recordedTime := watcher.decisionTime( + ctx, ConfirmationRiskDecisionRejected, + ) + require.False(t, recordedTime.IsZero()) + + select { + case decision := <-store.decisions: + require.Equal(t, ConfirmationRiskDecisionRejected, decision) + + case <-time.After(time.Second): + t.Fatal("missing risk decision was not persisted") + } + require.Equal(t, ConfirmationRiskDecisionRejected, + store.loopIns[swapHash].ConfirmationRiskDecision) + require.True(t, recordedTime.Equal( + store.loopIns[swapHash].ConfirmationRiskDecisionTime, + )) +}