From a7bff02fd8cdbe3c3efe5c6c4dc59ee776ee3755 Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Mon, 27 Apr 2026 14:50:53 +0200 Subject: [PATCH] staticaddr: harden client deposit readiness ListUnspentDeposits now reports only wallet UTXOs that have an active Deposited record. That matches the static loop-in admission path and avoids exposing wallet-seen outputs that are not ready for loop-in selection. Make local notification fan-out non-blocking for best-effort categories so a slow subscriber cannot stall the notification manager while it holds the subscriber lock. Static loop-in sweep signing requests remain blocking because they are work requests required for sweepbatcher presigning and must not be dropped. --- loopd/swapclient_server.go | 22 ++----- loopd/swapclient_server_test.go | 37 ++++------- notifications/manager.go | 18 ++++- notifications/manager_test.go | 113 +++++++++++++++++++++++++++++++- 4 files changed, 145 insertions(+), 45 deletions(-) diff --git a/loopd/swapclient_server.go b/loopd/swapclient_server.go index 40296d94..312e9895 100644 --- a/loopd/swapclient_server.go +++ b/loopd/swapclient_server.go @@ -1701,18 +1701,16 @@ func (s *swapClientServer) ListUnspentDeposits(ctx context.Context, } // ListUnspentRaw returns the unspent wallet view of the backing lnd - // wallet. It might be that deposits show up there that are actually - // not spendable because they already have been used but not yet spent - // by the server. We filter out such deposits here. + // wallet. Static loop-in initiation requires an active deposit record, + // so only deposits that are both wallet-visible and tracked as + // Deposited are returned here. var ( - outpoints []string - isUnspent = make(map[wire.OutPoint]struct{}) - knownUtxos = make(map[wire.OutPoint]struct{}) + outpoints []string + isUnspent = make(map[wire.OutPoint]struct{}) ) for _, utxo := range utxos { outpoints = append(outpoints, utxo.OutPoint.String()) - knownUtxos[utxo.OutPoint] = struct{}{} } // Check the spent status of the deposits by looking at their states. @@ -1724,26 +1722,16 @@ func (s *swapClientServer) ListUnspentDeposits(ctx context.Context, return nil, err } - knownDeposits := make(map[wire.OutPoint]struct{}, len(deposits)) for _, d := range deposits { if d == nil { continue } - knownDeposits[d.OutPoint] = struct{}{} if d.IsInState(deposit.Deposited) { isUnspent[d.OutPoint] = struct{}{} } } - // Any wallet outpoints that are unknown to the deposit store are new - // deposits and therefore still available. - for op := range knownUtxos { - if _, ok := knownDeposits[op]; !ok { - isUnspent[op] = struct{}{} - } - } - // Prepare the list of unspent deposits for the rpc response. var respUtxos []*looprpc.Utxo for _, u := range utxos { diff --git a/loopd/swapclient_server_test.go b/loopd/swapclient_server_test.go index faac0ec1..92120d47 100644 --- a/loopd/swapclient_server_test.go +++ b/loopd/swapclient_server_test.go @@ -1142,9 +1142,9 @@ func TestListUnspentDeposits(t *testing.T) { return deposit.NewManager(&deposit.ManagerConfig{Store: store}) } - // Unknown deposits are available, Deposited is available and known - // non-Deposited states are excluded. - t.Run("unknown and Deposited included, locked states excluded", + // Only known Deposited records are available. Unknown deposits and + // known non-Deposited states are excluded. + t.Run("only known Deposited included", func(t *testing.T) { mock.SetListUnspent([]*lnwallet.Utxo{ utxoUnknown, utxoDeposited, utxoWithdrawn, @@ -1167,8 +1167,8 @@ func TestListUnspentDeposits(t *testing.T) { ) require.NoError(t, err) - // Expect the unknown utxo and the Deposited utxo only. - require.Len(t, resp.Utxos, 2) + // Expect the Deposited utxo only. + require.Len(t, resp.Utxos, 1) got := map[string]struct{}{} for _, u := range resp.Utxos { got[u.Outpoint] = struct{}{} @@ -1176,10 +1176,8 @@ func TestListUnspentDeposits(t *testing.T) { // same across utxos. require.NotEmpty(t, u.StaticAddress) } - _, ok1 := got[utxoUnknown.OutPoint.String()] - _, ok2 := got[utxoDeposited.OutPoint.String()] - require.True(t, ok1) - require.True(t, ok2) + _, ok := got[utxoDeposited.OutPoint.String()] + require.True(t, ok) }) // Confirmation depth no longer changes availability; state does. @@ -1207,19 +1205,17 @@ func TestListUnspentDeposits(t *testing.T) { ) require.NoError(t, err) - require.Len(t, resp.Utxos, 2) + require.Len(t, resp.Utxos, 1) got := map[string]struct{}{} for _, u := range resp.Utxos { got[u.Outpoint] = struct{}{} } - _, ok1 := got[utxoUnknown.OutPoint.String()] - _, ok2 := got[utxoDeposited.OutPoint.String()] - require.True(t, ok1) - require.True(t, ok2) + _, ok := got[utxoDeposited.OutPoint.String()] + require.True(t, ok) }) - // Confirmed UTXO not present in store should be included. - t.Run("confirmed utxo not in store is included", func(t *testing.T) { + // Confirmed UTXO not present in store should be excluded. + t.Run("confirmed utxo not in store is excluded", func(t *testing.T) { // Only return a confirmed UTXO from lnd and make sure the // deposit manager/store doesn't know about it. mock.SetListUnspent([]*lnwallet.Utxo{utxoConfirmedUnknown}) @@ -1237,13 +1233,6 @@ func TestListUnspentDeposits(t *testing.T) { ) require.NoError(t, err) - // We expect the confirmed UTXO to be included even though it - // doesn't exist in the store yet. - require.Len(t, resp.Utxos, 1) - require.Equal( - t, utxoConfirmedUnknown.OutPoint.String(), - resp.Utxos[0].Outpoint, - ) - require.NotEmpty(t, resp.Utxos[0].StaticAddress) + require.Empty(t, resp.Utxos) }) } diff --git a/notifications/manager.go b/notifications/manager.go index bd1ac417..0d59a183 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -303,7 +303,13 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerReservationNotification) - recvChan <- reservationNtfn + select { + case recvChan <- reservationNtfn: + case <-sub.subCtx.Done(): + default: + log.Debugf("Dropping reservation " + + "notification for slow subscriber") + } } case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInSweep: // nolint: lll // We'll forward the static loop in sweep request to all @@ -316,7 +322,10 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerStaticLoopInSweepNotification) - recvChan <- staticLoopInSweepRequestNtfn + select { + case recvChan <- staticLoopInSweepRequestNtfn: + case <-sub.subCtx.Done(): + } } case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll @@ -330,7 +339,10 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerUnfinishedSwapNotification) - recvChan <- unfinishedSwapNtfn + select { + case recvChan <- unfinishedSwapNtfn: + case <-sub.subCtx.Done(): + } } default: diff --git a/notifications/manager_test.go b/notifications/manager_test.go index 9a06503e..8ba6f9e5 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -18,7 +18,7 @@ import ( var ( testReservationId = []byte{0x01, 0x02} - testReservationId2 = []byte{0x01, 0x02} + testReservationId2 = []byte{0x03, 0x04} ) // mockNotificationsClient implements the NotificationsClient interface for testing. @@ -188,6 +188,117 @@ func getTestNotification(resId []byte) *swapserverrpc.SubscribeNotificationsResp } } +// unfinishedSwapNotification builds an unfinished swap notification. +func unfinishedSwapNotification( + swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { + + return &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_UnfinishedSwap{ + UnfinishedSwap: &swapserverrpc. + ServerUnfinishedSwapNotification{ + SwapHash: swapHash[:], + }, + }, + } +} + +// TestManager_SlowSubscriberDoesNotBlock tests that a subscriber with a full +// notification channel does not block delivery to other subscribers. +func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { + t.Parallel() + + mgr := NewManager(&Config{}) + + slowCtx, slowCancel := context.WithCancel(t.Context()) + defer slowCancel() + slowChan := mgr.SubscribeReservations(slowCtx) + + fastCtx, fastCancel := context.WithCancel(t.Context()) + defer fastCancel() + fastChan := mgr.SubscribeReservations(fastCtx) + + firstNotif := getTestNotification(testReservationId) + mgr.handleNotification(firstNotif) + + received := <-fastChan + require.Equal(t, testReservationId, received.ReservationId) + + secondNotif := getTestNotification(testReservationId2) + done := make(chan struct{}) + go func() { + mgr.handleNotification(secondNotif) + close(done) + }() + + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + select { + case received = <-fastChan: + require.Equal(t, testReservationId2, received.ReservationId) + + case <-time.After(time.Second): + t.Fatal("fast subscriber did not receive notification") + } + + require.Len(t, slowChan, 1) +} + +// TestManager_UnfinishedSwapNotificationWaitsForSubscriber verifies that +// unfinished swap recovery notifications are not dropped when the local +// subscriber is briefly behind. +func TestManager_UnfinishedSwapNotificationWaitsForSubscriber(t *testing.T) { + t.Parallel() + + mgr := NewManager(&Config{}) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + subChan := mgr.SubscribeUnfinishedSwaps(subCtx) + + swapHashA := lntypes.Hash{0x02, 0x03} + swapHashB := lntypes.Hash{0x04, 0x05} + + mgr.handleNotification(unfinishedSwapNotification(swapHashA)) + + done := make(chan struct{}) + go func() { + mgr.handleNotification(unfinishedSwapNotification(swapHashB)) + close(done) + }() + + select { + case received := <-subChan: + require.Equal(t, swapHashA[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("did not receive first unfinished swap notification") + } + + select { + case <-done: + + case <-time.After(time.Second): + t.Fatal("second unfinished swap notification did not unblock") + } + + select { + case received := <-subChan: + require.Equal(t, swapHashB[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("second unfinished swap notification was dropped") + } +} + // TestManager_Backoff verifies that repeated failures in // subscribeNotifications cause the Manager to space out subscription attempts // via a predictable incremental backoff.