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.

Author: hieblmi

loopd/swapclient_server.go

@@ -1667,18 +1667,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.
@@ -1690,26 +1688,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 {

loopd/swapclient_server_test.go

@@ -1073,9 +1073,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,
@@ -1098,19 +1098,17 @@ 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{}{}
 				// Confirm address string is non-empty and the
 				// 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.
@@ -1138,19 +1136,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})
@@ -1168,13 +1164,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)
 	})
 }

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:

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,180 @@ func getTestNotification(resId []byte) *swapserverrpc.SubscribeNotificationsResp
 	}
 }
 
+func staticLoopInSweepNotification(
+	swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse {
+
+	return &swapserverrpc.SubscribeNotificationsResponse{
+		Notification: &swapserverrpc.
+			SubscribeNotificationsResponse_StaticLoopInSweep{
+			StaticLoopInSweep: &swapserverrpc.
+				ServerStaticLoopInSweepNotification{
+				SwapHash: swapHash[:],
+			},
+		},
+	}
+}
+
+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_StaticLoopInSweepNotificationWaitsForSubscriber verifies that
+// static loop in sweep signing requests are not dropped when the local
+// subscriber is briefly behind.
+func TestManager_StaticLoopInSweepNotificationWaitsForSubscriber(
+	t *testing.T) {
+
+	t.Parallel()
+
+	mgr := NewManager(&Config{})
+
+	subCtx, subCancel := context.WithCancel(t.Context())
+	defer subCancel()
+
+	subChan := mgr.SubscribeStaticLoopInSweepRequests(subCtx)
+
+	swapHashA := lntypes.Hash{0x02, 0x03}
+	swapHashB := lntypes.Hash{0x04, 0x05}
+
+	mgr.handleNotification(staticLoopInSweepNotification(swapHashA))
+
+	done := make(chan struct{})
+	go func() {
+		mgr.handleNotification(staticLoopInSweepNotification(swapHashB))
+		close(done)
+	}()
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHashA[:], received.SwapHash)
+
+	case <-time.After(time.Second):
+		t.Fatal("did not receive first sweep notification")
+	}
+
+	select {
+	case <-done:
+
+	case <-time.After(time.Second):
+		t.Fatal("second sweep notification did not unblock")
+	}
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHashB[:], received.SwapHash)
+
+	case <-time.After(time.Second):
+		t.Fatal("second sweep notification was dropped")
+	}
+}
+
+// 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.