Merge pull request #1154 from hieblmi/pre-dyn-conf-fixes

confrisk: preparatory changes

notifications/manager.go

@@ -45,6 +45,10 @@ const (
 	// htlc-confirmed subscriber before dropping the notification.
 	htlcConfirmedSubscriberSendTimeout = 200 * time.Millisecond
 
+	// defaultMaxQueuedNotifications is the default number of notifications
+	// we queue per subscriber before dropping new notifications.
+	defaultMaxQueuedNotifications = 1024
+
 	// current_version is the current version of the notification listener.
 	current_version = swapserverrpc.SubscribeNotificationsRequest_V1
 )
@@ -72,6 +76,10 @@ type Config struct {
 	// MinAliveConnTime is the minimum time that the connection to the
 	// server needs to be alive before we consider it a successful.
 	MinAliveConnTime time.Duration
+
+	// MaxQueuedNotifications is the maximum number of notifications that
+	// can wait in each subscriber's delivery queue.
+	MaxQueuedNotifications int
 }
 
 // Manager is a manager for notifications that the swap server sends to the
@@ -92,6 +100,9 @@ func NewManager(cfg *Config) *Manager {
 	if cfg.MinAliveConnTime == 0 {
 		cfg.MinAliveConnTime = defaultMinAliveConnTime
 	}
+	if cfg.MaxQueuedNotifications <= 0 {
+		cfg.MaxQueuedNotifications = defaultMaxQueuedNotifications
+	}
 
 	return &Manager{
 		cfg:         cfg,
@@ -102,6 +113,113 @@ func NewManager(cfg *Config) *Manager {
 type subscriber struct {
 	subCtx   context.Context
 	recvChan any
+	enqueue  func(any)
+}
+
+// newNotificationQueue creates a per-subscriber FIFO delivery function.
+func newNotificationQueue[T any](ctx context.Context,
+	recvChan chan T, maxPending int) func(any) {
+
+	type queue struct {
+		sync.Mutex
+
+		pending []T
+		notify  chan struct{}
+		closed  bool
+	}
+
+	q := &queue{
+		notify: make(chan struct{}, 1),
+	}
+
+	closeQueue := func() {
+		q.Lock()
+		q.closed = true
+		q.pending = nil
+		q.Unlock()
+	}
+
+	go func() {
+		defer close(recvChan)
+		defer closeQueue()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+
+			q.Lock()
+			if len(q.pending) == 0 {
+				q.Unlock()
+
+				select {
+				case <-q.notify:
+					continue
+
+				case <-ctx.Done():
+					return
+				}
+			}
+
+			ntfn := q.pending[0]
+			var zero T
+			q.pending[0] = zero
+			q.pending = q.pending[1:]
+			q.Unlock()
+
+			select {
+			case recvChan <- ntfn:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return func(ntfn any) {
+		typedNtfn, ok := ntfn.(T)
+		if !ok {
+			log.Warnf("unexpected notification type %T", ntfn)
+			return
+		}
+
+		q.Lock()
+		if q.closed {
+			q.Unlock()
+			return
+		}
+		if len(q.pending) >= maxPending {
+			q.Unlock()
+			log.Warnf("dropping notification for slow subscriber: "+
+				"queue depth %d reached", maxPending)
+			return
+		}
+
+		q.pending = append(q.pending, typedNtfn)
+		q.Unlock()
+
+		select {
+		case q.notify <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// queueNotification queues or synchronously sends a must-deliver notification.
+func queueNotification[T any](sub subscriber, recvChan chan T, ntfn T) {
+	if sub.enqueue != nil {
+		sub.enqueue(ntfn)
+		return
+	}
+
+	log.Warnf("subscriber has no notification queue, falling back to " +
+		"blocking send")
+
+	select {
+	case recvChan <- ntfn:
+	case <-sub.subCtx.Done():
+	}
 }
 
 // SubscribeReservations subscribes to the reservation notifications.
@@ -136,6 +254,9 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context,
 	sub := subscriber{
 		subCtx:   ctx,
 		recvChan: notifChan,
+		enqueue: newNotificationQueue(
+			ctx, notifChan, m.cfg.MaxQueuedNotifications,
+		),
 	}
 
 	m.addSubscriber(NotificationTypeStaticLoopInSweepRequest, sub)
@@ -145,7 +266,6 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context,
 			NotificationTypeStaticLoopInSweepRequest,
 			sub,
 		)
-		close(notifChan)
 	})
 
 	return notifChan
@@ -161,12 +281,14 @@ func (m *Manager) SubscribeUnfinishedSwaps(ctx context.Context,
 	sub := subscriber{
 		subCtx:   ctx,
 		recvChan: notifChan,
+		enqueue: newNotificationQueue(
+			ctx, notifChan, m.cfg.MaxQueuedNotifications,
+		),
 	}
 
 	m.addSubscriber(NotificationTypeUnfinishedSwap, sub)
 	context.AfterFunc(ctx, func() {
 		m.removeSubscriber(NotificationTypeUnfinishedSwap, sub)
-		close(notifChan)
 	})
 
 	return notifChan
@@ -332,7 +454,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
@@ -345,7 +473,7 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 			recvChan := sub.recvChan.(chan *swapserverrpc.
 				ServerStaticLoopInSweepNotification)
 
-			recvChan <- staticLoopInSweepRequestNtfn
+			queueNotification(sub, recvChan, staticLoopInSweepRequestNtfn)
 		}
 
 	case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll
@@ -359,7 +487,7 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 			recvChan := sub.recvChan.(chan *swapserverrpc.
 				ServerUnfinishedSwapNotification)
 
-			recvChan <- unfinishedSwapNtfn
+			queueNotification(sub, recvChan, unfinishedSwapNtfn)
 		}
 
 	case *swapserverrpc.SubscribeNotificationsResponse_HtlcConfirmed:
@@ -403,7 +531,7 @@ func (m *Manager) removeSubscriber(notifType NotificationType, sub subscriber) {
 	subs := m.subscribers[notifType]
 	newSubs := make([]subscriber, 0, len(subs))
 	for _, s := range subs {
-		if s != sub {
+		if s.recvChan != sub.recvChan {
 			newSubs = append(newSubs, s)
 		}
 	}

notifications/manager_test.go

@@ -20,7 +20,7 @@ import (
 
 var (
 	testReservationId  = []byte{0x01, 0x02}
-	testReservationId2 = []byte{0x01, 0x02}
+	testReservationId2 = []byte{0x03, 0x04}
 )
 
 // mockNotificationsClient implements the NotificationsClient interface for testing.
@@ -190,6 +190,276 @@ 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[:],
+			},
+		},
+	}
+}
+
+// staticLoopInSweepNotification builds a static loop-in sweep notification.
+func staticLoopInSweepNotification(
+	swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse {
+
+	return &swapserverrpc.SubscribeNotificationsResponse{
+		Notification: &swapserverrpc.
+			SubscribeNotificationsResponse_StaticLoopInSweep{
+			StaticLoopInSweep: &swapserverrpc.
+				ServerStaticLoopInSweepNotification{
+				SwapHash: swapHash[:],
+			},
+		},
+	}
+}
+
+// TestManager_SlowReservationSubscriberDoesNotBlock tests that a reservation
+// subscriber with a full notification channel does not block delivery to other
+// subscribers. Reservation notifications are best-effort, so slow subscribers
+// drop new notifications instead of queueing them.
+func TestManager_SlowReservationSubscriberDoesNotBlock(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)
+
+	select {
+	case received = <-slowChan:
+		require.Equal(t, testReservationId, received.ReservationId)
+
+	case <-time.After(time.Second):
+		t.Fatal("slow subscriber did not receive first notification")
+	}
+
+	select {
+	case received = <-slowChan:
+		t.Fatalf("slow subscriber received dropped notification %x",
+			received.ReservationId)
+
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+// 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()
+
+	assertQueuedSwapHashNotifications(
+		t,
+		func(mgr *Manager, ctx context.Context) <-chan *swapserverrpc.
+			ServerUnfinishedSwapNotification {
+
+			return mgr.SubscribeUnfinishedSwaps(ctx)
+		},
+		unfinishedSwapNotification,
+		func(ntfn *swapserverrpc.ServerUnfinishedSwapNotification) []byte {
+			return ntfn.SwapHash
+		},
+		lntypes.Hash{0x02, 0x03}, lntypes.Hash{0x04, 0x05},
+		"did not receive first unfinished swap notification",
+		"second unfinished swap notification was dropped",
+	)
+}
+
+// TestManager_StaticLoopInSweepNotificationQueuesForSlowSubscriber verifies
+// that a full static-loop-in sweep subscriber channel does not block the global
+// notification receive loop.
+func TestManager_StaticLoopInSweepNotificationQueuesForSlowSubscriber(
+	t *testing.T) {
+
+	t.Parallel()
+
+	assertQueuedSwapHashNotifications(
+		t,
+		func(mgr *Manager, ctx context.Context) <-chan *swapserverrpc.
+			ServerStaticLoopInSweepNotification {
+
+			return mgr.SubscribeStaticLoopInSweepRequests(ctx)
+		},
+		staticLoopInSweepNotification,
+		func(ntfn *swapserverrpc.ServerStaticLoopInSweepNotification) []byte {
+			return ntfn.SwapHash
+		},
+		lntypes.Hash{0x12, 0x13}, lntypes.Hash{0x14, 0x15},
+		"did not receive first sweep notification",
+		"second sweep notification was not queued",
+	)
+}
+
+// TestManager_QueuedNotificationChannelClosesOnCancel verifies that queued
+// subscribers own their channel shutdown even when delivery is blocked.
+func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(&Config{})
+
+	subCtx, subCancel := context.WithCancel(t.Context())
+	subChan := mgr.SubscribeUnfinishedSwaps(subCtx)
+
+	swapHashA := lntypes.Hash{0x21, 0x22}
+	mgr.handleNotification(unfinishedSwapNotification(swapHashA))
+
+	require.Eventually(t, func() bool {
+		return len(subChan) == 1
+	}, time.Second, 10*time.Millisecond)
+
+	swapHashB := lntypes.Hash{0x23, 0x24}
+	done := make(chan struct{})
+	go func() {
+		mgr.handleNotification(unfinishedSwapNotification(swapHashB))
+		close(done)
+	}()
+
+	require.Eventually(t, func() bool {
+		select {
+		case <-done:
+			return true
+		default:
+			return false
+		}
+	}, time.Second, 10*time.Millisecond)
+
+	subCancel()
+
+	select {
+	case received, ok := <-subChan:
+		require.True(t, ok)
+		require.Equal(t, swapHashA[:], received.SwapHash)
+
+	case <-time.After(time.Second):
+		t.Fatal("first unfinished swap notification was not delivered")
+	}
+
+	require.Eventually(t, func() bool {
+		select {
+		case _, ok := <-subChan:
+			return !ok
+		default:
+			return false
+		}
+	}, time.Second, 10*time.Millisecond)
+}
+
+// TestNotificationQueueDropsAtCapacity checks the queue's explicit drop policy
+// once a subscriber reaches its configured backlog limit.
+func TestNotificationQueueDropsAtCapacity(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	recvChan := make(chan int, 1)
+	enqueue := newNotificationQueue(ctx, recvChan, 0)
+
+	enqueue(1)
+
+	select {
+	case ntfn := <-recvChan:
+		t.Fatalf("received dropped notification %d", ntfn)
+
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+// assertQueuedSwapHashNotifications checks queued delivery for swap hashes.
+func assertQueuedSwapHashNotifications[T any](t *testing.T,
+	subscribe func(*Manager, context.Context) <-chan T,
+	notification func(lntypes.Hash) *swapserverrpc.
+		SubscribeNotificationsResponse,
+	swapHash func(T) []byte, swapHashA, swapHashB lntypes.Hash,
+	firstFailureMsg, secondFailureMsg string) {
+
+	t.Helper()
+
+	mgr := NewManager(&Config{})
+
+	subCtx, subCancel := context.WithCancel(t.Context())
+	defer subCancel()
+
+	subChan := subscribe(mgr, subCtx)
+
+	mgr.handleNotification(notification(swapHashA))
+
+	done := make(chan struct{})
+	go func() {
+		mgr.handleNotification(notification(swapHashB))
+		close(done)
+	}()
+
+	require.Eventually(t, func() bool {
+		select {
+		case <-done:
+			return true
+		default:
+			return false
+		}
+	}, time.Second, 10*time.Millisecond)
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHashA[:], swapHash(received))
+
+	case <-time.After(time.Second):
+		t.Fatal(firstFailureMsg)
+	}
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHashB[:], swapHash(received))
+
+	case <-time.After(time.Second):
+		t.Fatal(secondFailureMsg)
+	}
+}
+
 // TestManager_Backoff verifies that repeated failures in
 // subscribeNotifications cause the Manager to space out subscription attempts
 // via a predictable incremental backoff.

staticaddr/deposit/actions.go

@@ -161,14 +161,25 @@ func (f *FSM) WaitForExpirySweepAction(ctx context.Context,
 
 // FinalizeDepositAction is the final action after a withdrawal. It signals to
 // the manager that the deposit has been swept and the FSM can be removed.
-func (f *FSM) FinalizeDepositAction(ctx context.Context,
+func (f *FSM) FinalizeDepositAction(_ context.Context,
 	_ fsm.EventContext) fsm.EventType {
 
-	select {
-	case <-ctx.Done():
-		return fsm.OnError
+	outpoint := f.deposit.OutPoint
 
-	case f.finalizedDepositChan <- f.deposit.OutPoint:
-		return fsm.NoOp
-	}
+	// The finalization notification only tells the manager to remove the
+	// deposit from its active set. Send it asynchronously so a busy manager
+	// loop can't stall withdrawal confirmation while deposit locks are held.
+	go func() {
+		select {
+		case <-f.quitChan:
+			// The deposit is already in a final state. If shutdown wins
+			// this race, startup recovery will skip it instead of
+			// re-adding it to the active set.
+			return
+
+		case f.finalizedDepositChan <- outpoint:
+		}
+	}()
+
+	return fsm.NoOp
 }

staticaddr/deposit/actions_test.go

@@ -0,0 +1,135 @@
+package deposit
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightninglabs/loop/fsm"
+	"github.com/stretchr/testify/require"
+)
+
+// TestFinalizeDepositActionDoesNotBlock ensures the final cleanup notification
+// does not block the withdrawal completion path while the manager loop is busy.
+func TestFinalizeDepositActionDoesNotBlock(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	outpoint := wire.OutPoint{
+		Hash:  chainhash.Hash{1},
+		Index: 1,
+	}
+
+	depositFSM := &FSM{
+		deposit: &Deposit{
+			OutPoint: outpoint,
+		},
+		quitChan:             make(chan struct{}),
+		finalizedDepositChan: make(chan wire.OutPoint),
+	}
+
+	resultChan := make(chan fsm.EventType, 1)
+	go func() {
+		resultChan <- depositFSM.FinalizeDepositAction(ctx, nil)
+	}()
+
+	select {
+	case result := <-resultChan:
+		require.Equal(t, fsm.NoOp, result)
+
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("FinalizeDepositAction blocked on manager cleanup")
+	}
+
+	select {
+	case gotOutpoint := <-depositFSM.finalizedDepositChan:
+		require.Equal(t, outpoint, gotOutpoint)
+
+	case <-time.After(time.Second):
+		t.Fatal("finalization cleanup notification was not delivered")
+	}
+}
+
+// TestFinalizeDepositActionIgnoresRequestCancellation ensures the cleanup
+// notification is tied to the FSM lifetime, not the caller's request context.
+func TestFinalizeDepositActionIgnoresRequestCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	quitChan := make(chan struct{})
+	defer close(quitChan)
+
+	outpoint := wire.OutPoint{
+		Hash:  chainhash.Hash{2},
+		Index: 2,
+	}
+
+	depositFSM := &FSM{
+		deposit: &Deposit{
+			OutPoint: outpoint,
+		},
+		quitChan:             quitChan,
+		finalizedDepositChan: make(chan wire.OutPoint),
+	}
+
+	resultChan := make(chan fsm.EventType, 1)
+	go func() {
+		resultChan <- depositFSM.FinalizeDepositAction(ctx, nil)
+	}()
+
+	select {
+	case result := <-resultChan:
+		require.Equal(t, fsm.NoOp, result)
+
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("FinalizeDepositAction blocked on manager cleanup")
+	}
+
+	cancel()
+
+	select {
+	case gotOutpoint := <-depositFSM.finalizedDepositChan:
+		require.Equal(t, outpoint, gotOutpoint)
+
+	case <-time.After(time.Second):
+		t.Fatal("finalization cleanup notification was dropped after " +
+			"request cancellation")
+	}
+}
+
+// TestFinalizeDepositActionIgnoresCanceledContext ensures the final cleanup
+// notification is still queued even if the caller's context is already done.
+func TestFinalizeDepositActionIgnoresCanceledContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	quitChan := make(chan struct{})
+	defer close(quitChan)
+
+	outpoint := wire.OutPoint{
+		Hash:  chainhash.Hash{3},
+		Index: 3,
+	}
+
+	depositFSM := &FSM{
+		deposit: &Deposit{
+			OutPoint: outpoint,
+		},
+		quitChan:             quitChan,
+		finalizedDepositChan: make(chan wire.OutPoint),
+	}
+
+	result := depositFSM.FinalizeDepositAction(ctx, nil)
+	require.Equal(t, fsm.NoOp, result)
+
+	select {
+	case gotOutpoint := <-depositFSM.finalizedDepositChan:
+		require.Equal(t, outpoint, gotOutpoint)
+
+	case <-time.After(time.Second):
+		t.Fatal("finalization cleanup notification was dropped for " +
+			"an already-canceled request context")
+	}
+}

staticaddr/loopin/actions.go

@@ -36,6 +36,8 @@ const (
 	defaultConfTarget = 3
 
 	DefaultPaymentTimeoutSeconds = 60
+
+	defaultInvoiceCleanupTimeout = 5 * time.Second
 )
 
 var (
@@ -57,6 +59,24 @@ var (
 func (f *FSM) InitHtlcAction(ctx context.Context,
 	_ fsm.EventContext) fsm.EventType {
 
+	var event fsm.EventType
+	invoiceNeedsCleanup := false
+	defer func() {
+		// If we created the private invoice but failed before persisting the
+		// swap, cancel it so retries do not accumulate orphan invoices.
+		if !invoiceNeedsCleanup || event != fsm.OnError {
+			return
+		}
+
+		f.cancelSwapInvoice()
+	}()
+
+	returnError := func(err error) fsm.EventType {
+		event = f.HandleError(err)
+
+		return event
+	}
+
 	// Lock the deposits and transition them to the LoopingIn state.
 	err := f.cfg.DepositManager.TransitionDeposits(
 		ctx, f.loopIn.Deposits, deposit.OnLoopInInitiated,
@@ -65,7 +85,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 	if err != nil {
 		err = fmt.Errorf("unable to loop-in deposits: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 
 	// Calculate the swap invoice amount. The server needs to pay us the
@@ -88,7 +108,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 		err = fmt.Errorf("unable to create random swap preimage: %w",
 			err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 	f.loopIn.SwapPreimage = swapPreimage
 	f.loopIn.SwapHash = swapPreimage.Hash()
@@ -100,7 +120,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 	if err != nil {
 		err = fmt.Errorf("unable to derive client htlc key: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 	f.loopIn.ClientPubkey = keyDesc.PubKey
 	f.loopIn.HtlcKeyLocator = keyDesc.KeyLocator
@@ -119,10 +139,14 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 	if err != nil {
 		err = fmt.Errorf("unable to create swap invoice: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 	f.loopIn.SwapInvoice = swapInvoice
 
+	// From here until CreateLoopIn succeeds, any error path would otherwise
+	// leave behind a live invoice with no persisted swap to recover it.
+	invoiceNeedsCleanup = true
+
 	f.loopIn.ProtocolVersion = version.AddressProtocolVersion(
 		version.CurrentRPCProtocolVersion(),
 	)
@@ -149,7 +173,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 		err = fmt.Errorf("unable to initiate the loop-in with the "+
 			"server: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 
 	// Pushing empty sigs signals the server that we abandoned the swap
@@ -171,7 +195,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 		pushEmptySigs()
 		err = fmt.Errorf("unable to parse server pubkey: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 	f.loopIn.ServerPubkey = serverPubkey
 
@@ -185,7 +209,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 		err = fmt.Errorf("server response parameters are outside "+
 			"our allowed range: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 
 	f.loopIn.HtlcCltvExpiry = loopInResp.HtlcExpiry
@@ -194,7 +218,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 		pushEmptySigs()
 		err = fmt.Errorf("unable to convert server nonces: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 	f.htlcServerNoncesHighFee, err = toNonces(
 		loopInResp.HighFeeHtlcInfo.Nonces,
@@ -202,7 +226,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 	if err != nil {
 		pushEmptySigs()
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 	f.htlcServerNoncesExtremelyHighFee, err = toNonces(
 		loopInResp.ExtremeFeeHtlcInfo.Nonces,
@@ -210,7 +234,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 	if err != nil {
 		pushEmptySigs()
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 
 	// We need to defend against the server setting high fees for the htlc
@@ -232,7 +256,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 		log.Errorf("server htlc tx fee is higher than the configured "+
 			"allowed maximum: %v > %v", fee, maxHtlcTxFee)
 
-		return f.HandleError(ErrFeeTooHigh)
+		return returnError(ErrFeeTooHigh)
 	}
 	f.loopIn.HtlcTxFeeRate = feeRate
 
@@ -246,7 +270,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 			"configured allowed maximum: %v > %v", fee,
 			maxHtlcTxBackupFee)
 
-		return f.HandleError(ErrFeeTooHigh)
+		return returnError(ErrFeeTooHigh)
 	}
 	f.loopIn.HtlcTxHighFeeRate = highFeeRate
 
@@ -262,7 +286,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 			"configured allowed maximum: %v > %v", fee,
 			maxHtlcTxBackupFee)
 
-		return f.HandleError(ErrFeeTooHigh)
+		return returnError(ErrFeeTooHigh)
 	}
 	f.loopIn.HtlcTxExtremelyHighFeeRate = extremelyHighFeeRate
 
@@ -276,7 +300,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 		err = fmt.Errorf("unable to derive htlc timeout sweep "+
 			"address: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 	f.loopIn.HtlcTimeoutSweepAddress = sweepAddress
 
@@ -286,10 +310,34 @@ func (f *FSM) InitHtlcAction(ctx context.Context,
 		pushEmptySigs()
 		err = fmt.Errorf("unable to store loop-in in db: %w", err)
 
-		return f.HandleError(err)
+		return returnError(err)
 	}
 
-	return OnHtlcInitiated
+	// Once the swap is stored, restart/recovery code owns invoice lifecycle.
+	invoiceNeedsCleanup = false
+
+	event = OnHtlcInitiated
+
+	return event
+}
+
+// cancelSwapInvoice best-effort cancels the current swap invoice using a
+// detached timeout-limited context.
+func (f *FSM) cancelSwapInvoice() {
+	if f.loopIn.SwapInvoice == "" {
+		return
+	}
+
+	cleanupCtx, cancel := context.WithTimeout(
+		context.Background(), defaultInvoiceCleanupTimeout,
+	)
+	defer cancel()
+
+	err := f.cfg.InvoicesClient.CancelInvoice(cleanupCtx, f.loopIn.SwapHash)
+	if err != nil {
+		f.Warnf("unable to cancel invoice for swap %v: %v",
+			f.loopIn.SwapHash, err)
+	}
 }
 
 // SignHtlcTxAction is called if the htlc was initialized and the server
@@ -557,11 +605,9 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context,
 		// Cancel the lndclient invoice subscription.
 		cancelInvoiceSubscription()
 
-		err = f.cfg.InvoicesClient.CancelInvoice(ctx, f.loopIn.SwapHash)
-		if err != nil {
-			f.Warnf("unable to cancel invoice "+
-				"for swap hash: %v", err)
-		}
+		// Reuse the same helper as InitHtlcAction so timeout cleanup
+		// follows the same detached-context path as early-init cleanup.
+		f.cancelSwapInvoice()
 	}
 
 	for {
@@ -609,10 +655,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context,
 			// re-enable them for loop-ins and withdrawals.
 			cancelInvoice()
 
-			event := f.UnlockDepositsAction(ctx, nil)
-			if event != fsm.OnError {
-				f.Errorf("unable to unlock deposits after " +
-					"payment deadline")
+			err = f.unlockDeposits(ctx)
+			if err != nil {
+				f.Errorf("unable to unlock deposits after "+
+					"payment deadline: %v", err)
 			}
 
 		case currentHeight := <-blockChan:
@@ -824,18 +870,27 @@ func (f *FSM) PaymentReceivedAction(ctx context.Context,
 func (f *FSM) UnlockDepositsAction(ctx context.Context,
 	_ fsm.EventContext) fsm.EventType {
 
-	err := f.cfg.DepositManager.TransitionDeposits(
-		ctx, f.loopIn.Deposits, fsm.OnError, deposit.Deposited,
-	)
-	if err != nil {
-		err = fmt.Errorf("unable to unlock deposits: %w", err)
+	f.cancelSwapInvoice()
 
+	err := f.unlockDeposits(ctx)
+	if err != nil {
 		return f.HandleError(err)
 	}
 
 	return fsm.OnError
 }
 
+func (f *FSM) unlockDeposits(ctx context.Context) error {
+	err := f.cfg.DepositManager.TransitionDeposits(
+		ctx, f.loopIn.Deposits, fsm.OnError, deposit.Deposited,
+	)
+	if err != nil {
+		return fmt.Errorf("unable to unlock deposits: %w", err)
+	}
+
+	return nil
+}
+
 // createAndPublishHtlcTimeoutSweepTx creates and publishes the htlc timeout
 // sweep transaction.
 func (f *FSM) createAndPublishHtlcTimeoutSweepTx(ctx context.Context) error {

staticaddr/loopin/actions_test.go

@@ -270,6 +270,192 @@ func testValidateLoopInContract(_ int32, _ int32) error {
 	return nil
 }
 
+// TestInitHtlcActionCancelsInvoiceOnServerError verifies that an invoice
+// created before a server-side rejection is canceled immediately.
+func TestInitHtlcActionCancelsInvoiceOnServerError(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+
+	mockLnd := test.NewMockLnd()
+
+	loopIn := &StaticAddressLoopIn{
+		Deposits: []*deposit.Deposit{{
+			Value: 200_000,
+		}},
+		InitiationHeight:      uint32(mockLnd.Height),
+		InitiationTime:        time.Now(),
+		PaymentTimeoutSeconds: DefaultPaymentTimeoutSeconds,
+		ProtocolVersion:       version.ProtocolVersion_V0,
+	}
+
+	cfg := &Config{
+		AddressManager: &mockAddressManager{
+			params: &script.Parameters{
+				ProtocolVersion: version.ProtocolVersion_V0,
+			},
+		},
+		DepositManager: &noopDepositManager{},
+		WalletKit:      mockLnd.WalletKit,
+		LndClient:      mockLnd.Client,
+		InvoicesClient: mockLnd.LndServices.Invoices,
+		Server: &initHtlcTestServer{
+			loopInErr: errors.New("server rejected swap"),
+		},
+	}
+
+	f, err := NewFSM(ctx, loopIn, cfg, false)
+	require.NoError(t, err)
+
+	// The init step should fail and synchronously trigger deferred invoice
+	// cleanup.
+	event := f.InitHtlcAction(ctx, nil)
+	require.Equal(t, fsm.OnError, event)
+
+	select {
+	case hash := <-mockLnd.FailInvoiceChannel:
+		require.Equal(t, loopIn.SwapHash, hash)
+
+	case <-ctx.Done():
+		t.Fatalf("invoice was not canceled: %v", ctx.Err())
+	}
+}
+
+// TestInitHtlcActionCancelsInvoiceOnFeeGuardFailure verifies that the early
+// fee guard also cancels the pre-created invoice before returning an error.
+func TestInitHtlcActionCancelsInvoiceOnFeeGuardFailure(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+
+	mockLnd := test.NewMockLnd()
+	serverKey, err := btcec.NewPrivateKey()
+	require.NoError(t, err)
+
+	loopIn := &StaticAddressLoopIn{
+		Deposits: []*deposit.Deposit{{
+			Value: 200_000,
+		}},
+		InitiationHeight:      uint32(mockLnd.Height),
+		InitiationTime:        time.Now(),
+		PaymentTimeoutSeconds: DefaultPaymentTimeoutSeconds,
+		ProtocolVersion:       version.ProtocolVersion_V0,
+	}
+
+	cfg := &Config{
+		AddressManager: &mockAddressManager{
+			params: &script.Parameters{
+				ProtocolVersion: version.ProtocolVersion_V0,
+			},
+		},
+		DepositManager: &noopDepositManager{},
+		WalletKit:      mockLnd.WalletKit,
+		LndClient:      mockLnd.Client,
+		InvoicesClient: mockLnd.LndServices.Invoices,
+		Server: &initHtlcTestServer{
+			loopInResp: &swapserverrpc.ServerStaticAddressLoopInResponse{
+				HtlcServerPubKey: serverKey.PubKey().
+					SerializeCompressed(),
+				HtlcExpiry: mockLnd.Height +
+					DefaultLoopInOnChainCltvDelta,
+				StandardHtlcInfo: &swapserverrpc.ServerHtlcSigningInfo{
+					FeeRate: 1_000_000,
+				},
+				HighFeeHtlcInfo: &swapserverrpc.ServerHtlcSigningInfo{},
+				ExtremeFeeHtlcInfo: &swapserverrpc.
+					ServerHtlcSigningInfo{},
+			},
+		},
+		ValidateLoopInContract: func(int32, int32) error {
+			return nil
+		},
+		MaxStaticAddrHtlcFeePercentage:       0,
+		MaxStaticAddrHtlcBackupFeePercentage: 1,
+	}
+
+	f, err := NewFSM(ctx, loopIn, cfg, false)
+	require.NoError(t, err)
+
+	// The fee guard runs before persistence, so the deferred cleanup must
+	// cancel the invoice on this error path as well.
+	event := f.InitHtlcAction(ctx, nil)
+	require.Equal(t, fsm.OnError, event)
+
+	select {
+	case hash := <-mockLnd.FailInvoiceChannel:
+		require.Equal(t, loopIn.SwapHash, hash)
+
+	case <-ctx.Done():
+		t.Fatalf("invoice was not canceled: %v", ctx.Err())
+	}
+}
+
+// TestUnlockDepositsActionCancelsInvoice verifies that stored swaps that enter
+// the generic error unlock path also clean up their swap invoice.
+func TestUnlockDepositsActionCancelsInvoice(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+
+	mockLnd := test.NewMockLnd()
+	dep := &deposit.Deposit{
+		Value: 200_000,
+	}
+	swapHash := lntypes.Hash{0x44, 0x55}
+	depositMgr := &recordingDepositManager{}
+
+	f := &FSM{
+		StateMachine: &fsm.StateMachine{},
+		cfg: &Config{
+			DepositManager: depositMgr,
+			InvoicesClient: mockLnd.LndServices.Invoices,
+		},
+		loopIn: &StaticAddressLoopIn{
+			SwapHash:    swapHash,
+			SwapInvoice: "lnbc1test",
+			Deposits:    []*deposit.Deposit{dep},
+		},
+	}
+
+	event := f.UnlockDepositsAction(ctx, nil)
+	require.Equal(t, fsm.OnError, event)
+	require.NoError(t, f.LastActionError)
+
+	select {
+	case hash := <-mockLnd.FailInvoiceChannel:
+		require.Equal(t, swapHash, hash)
+
+	case <-ctx.Done():
+		t.Fatalf("invoice was not canceled: %v", ctx.Err())
+	}
+
+	require.Len(t, depositMgr.transitions, 1)
+	require.Equal(t, []*deposit.Deposit{dep}, depositMgr.transitions[0].deposits)
+	require.Equal(t, fsm.OnError, depositMgr.transitions[0].event)
+	require.Equal(t, deposit.Deposited, depositMgr.transitions[0].state)
+}
+
+// TestUnlockDepositsActionReportsTransitionError ensures the unlock path
+// preserves the real deposit transition failure for callers that need to log it.
+func TestUnlockDepositsActionReportsTransitionError(t *testing.T) {
+	depositMgr := &recordingDepositManager{
+		err: errors.New("transition failed"),
+	}
+	f := &FSM{
+		StateMachine: &fsm.StateMachine{},
+		cfg: &Config{
+			DepositManager: depositMgr,
+		},
+		loopIn: &StaticAddressLoopIn{
+			Deposits: []*deposit.Deposit{{Value: 200_000}},
+		},
+	}
+
+	event := f.UnlockDepositsAction(t.Context(), nil)
+	require.Equal(t, fsm.OnError, event)
+	require.ErrorContains(
+		t, f.LastActionError, "unable to unlock deposits",
+	)
+	require.ErrorContains(t, f.LastActionError, "transition failed")
+}
+
 // mockAddressManager is a minimal AddressManager implementation used by the
 // test FSM setup.
 type mockAddressManager struct {
@@ -327,3 +513,56 @@ func (n *noopDepositManager) GetActiveDepositsInState(fsm.StateType) (
 
 	return nil, nil
 }
+
+type depositTransition struct {
+	deposits []*deposit.Deposit
+	event    fsm.EventType
+	state    fsm.StateType
+}
+
+type recordingDepositManager struct {
+	noopDepositManager
+
+	err         error
+	transitions []depositTransition
+}
+
+// TransitionDeposits records the transition and returns the configured error.
+func (r *recordingDepositManager) TransitionDeposits(_ context.Context,
+	deposits []*deposit.Deposit, event fsm.EventType,
+	state fsm.StateType) error {
+
+	r.transitions = append(r.transitions, depositTransition{
+		deposits: deposits,
+		event:    event,
+		state:    state,
+	})
+
+	return r.err
+}
+
+// initHtlcTestServer lets InitHtlcAction tests inject a deterministic server
+// response without standing up the full gRPC client.
+type initHtlcTestServer struct {
+	swapserverrpc.StaticAddressServerClient
+
+	loopInResp *swapserverrpc.ServerStaticAddressLoopInResponse
+	loopInErr  error
+}
+
+// ServerStaticAddressLoopIn returns the canned response configured by the test.
+func (s *initHtlcTestServer) ServerStaticAddressLoopIn(context.Context,
+	*swapserverrpc.ServerStaticAddressLoopInRequest, ...grpc.CallOption,
+) (*swapserverrpc.ServerStaticAddressLoopInResponse, error) {
+
+	return s.loopInResp, s.loopInErr
+}
+
+// PushStaticAddressHtlcSigs accepts the abandonment signal used by error-path
+// tests without adding additional assertions.
+func (s *initHtlcTestServer) PushStaticAddressHtlcSigs(context.Context,
+	*swapserverrpc.PushStaticAddressHtlcSigsRequest, ...grpc.CallOption,
+) (*swapserverrpc.PushStaticAddressHtlcSigsResponse, error) {
+
+	return &swapserverrpc.PushStaticAddressHtlcSigsResponse{}, nil
+}

staticaddr/loopin/sql_store.go

@@ -56,8 +56,8 @@ type Querier interface {
 		swapHash []byte) (sqlc.GetStaticAddressLoopInSwapRow, error)
 
 	// GetStaticAddressLoopInSwapsByStates retrieves all swaps with the
-	// given states. The states string is an input for the IN primitive in
-	// sqlite, hence the format needs to be '{State1,State2,...}'.
+	// given states. The states string is comma-separated so the query can
+	// match complete state names by wrapping it with comma sentinels.
 	GetStaticAddressLoopInSwapsByStates(ctx context.Context,
 		states sql.NullString) ([]sqlc.GetStaticAddressLoopInSwapsByStatesRow,
 		error)
@@ -203,7 +203,7 @@ func (s *SqlStore) GetStaticAddressLoopInSwapsByStates(ctx context.Context,
 }
 
 func toJointStringStates(states []fsm.StateType) string {
-	return "{" + strings.Join(toStrings(states), ",") + "}"
+	return strings.Join(toStrings(states), ",")
 }
 
 func toStrings(states []fsm.StateType) []string {

staticaddr/loopin/sql_store_test.go

@@ -41,8 +41,10 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) {
 	}
 
 	loopingDepositID := newID()
+	timeoutDepositID := newID()
 	loopedInDepositID := newID()
-	d1, d2 := &deposit.Deposit{
+	failedDepositID := newID()
+	d1, d2, d3, d4 := &deposit.Deposit{
 		ID: loopingDepositID,
 		OutPoint: wire.OutPoint{
 			Hash:  chainhash.Hash{0x1a, 0x2b, 0x3c, 0x4d},
@@ -54,7 +56,7 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) {
 		},
 	},
 		&deposit.Deposit{
-			ID: loopedInDepositID,
+			ID: timeoutDepositID,
 			OutPoint: wire.OutPoint{
 				Hash:  chainhash.Hash{0x2a, 0x2b, 0x3c, 0x4e},
 				Index: 1,
@@ -63,29 +65,67 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) {
 			TimeOutSweepPkScript: []byte{
 				0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4d,
 			},
+		},
+		&deposit.Deposit{
+			ID: loopedInDepositID,
+			OutPoint: wire.OutPoint{
+				Hash:  chainhash.Hash{0x3a, 0x2b, 0x3c, 0x4e},
+				Index: 2,
+			},
+			Value: btcutil.Amount(300_000),
+			TimeOutSweepPkScript: []byte{
+				0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4f,
+			},
+		},
+		&deposit.Deposit{
+			ID: failedDepositID,
+			OutPoint: wire.OutPoint{
+				Hash:  chainhash.Hash{0x4a, 0x2b, 0x3c, 0x4e},
+				Index: 3,
+			},
+			Value: btcutil.Amount(400_000),
+			TimeOutSweepPkScript: []byte{
+				0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x50,
+			},
 		}
 
 	err := depositStore.CreateDeposit(ctxb, d1)
 	require.NoError(t, err)
 	err = depositStore.CreateDeposit(ctxb, d2)
 	require.NoError(t, err)
+	err = depositStore.CreateDeposit(ctxb, d3)
+	require.NoError(t, err)
+	err = depositStore.CreateDeposit(ctxb, d4)
+	require.NoError(t, err)
 
 	// Add two updates per deposit, expect the last to be retrieved.
 	d1.SetState(deposit.Deposited)
 	d2.SetState(deposit.Deposited)
+	d3.SetState(deposit.Deposited)
+	d4.SetState(deposit.Deposited)
 
 	err = depositStore.UpdateDeposit(ctxb, d1)
 	require.NoError(t, err)
 	err = depositStore.UpdateDeposit(ctxb, d2)
 	require.NoError(t, err)
+	err = depositStore.UpdateDeposit(ctxb, d3)
+	require.NoError(t, err)
+	err = depositStore.UpdateDeposit(ctxb, d4)
+	require.NoError(t, err)
 
 	d1.SetState(deposit.LoopingIn)
-	d2.SetState(deposit.LoopedIn)
+	d2.SetState(deposit.HtlcTimeoutSwept)
+	d3.SetState(deposit.LoopedIn)
+	d4.SetState(deposit.Deposited)
 
 	err = depositStore.UpdateDeposit(ctxb, d1)
 	require.NoError(t, err)
 	err = depositStore.UpdateDeposit(ctxb, d2)
 	require.NoError(t, err)
+	err = depositStore.UpdateDeposit(ctxb, d3)
+	require.NoError(t, err)
+	err = depositStore.UpdateDeposit(ctxb, d4)
+	require.NoError(t, err)
 
 	_, clientPubKey := test.CreateKey(1)
 	_, serverPubKey := test.CreateKey(2)
@@ -108,13 +148,30 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) {
 	err = swapStore.CreateLoopIn(ctxb, &swapPending)
 	require.NoError(t, err)
 
+	// Create htlc-timeout-swept swap. HtlcTimeoutSwept is the first final
+	// state, so this exercises the state-list query boundary.
+	swapHashTimeoutSwept := lntypes.Hash{0x4, 0x2, 0x3, 0x5}
+	swapTimeoutSwept := StaticAddressLoopIn{
+		SwapHash:                swapHashTimeoutSwept,
+		SwapPreimage:            lntypes.Preimage{0x4, 0x2, 0x3, 0x5},
+		DepositOutpoints:        []string{d2.OutPoint.String()},
+		Deposits:                []*deposit.Deposit{d2},
+		ClientPubkey:            clientPubKey,
+		ServerPubkey:            serverPubKey,
+		HtlcTimeoutSweepAddress: addr,
+	}
+	swapTimeoutSwept.SetState(HtlcTimeoutSwept)
+
+	err = swapStore.CreateLoopIn(ctxb, &swapTimeoutSwept)
+	require.NoError(t, err)
+
 	// Create succeeded swap.
 	swapHashSucceeded := lntypes.Hash{0x2, 0x2, 0x3, 0x5}
 	swapSucceeded := StaticAddressLoopIn{
 		SwapHash:                swapHashSucceeded,
 		SwapPreimage:            lntypes.Preimage{0x2, 0x2, 0x3, 0x5},
-		DepositOutpoints:        []string{d2.OutPoint.String()},
-		Deposits:                []*deposit.Deposit{d2},
+		DepositOutpoints:        []string{d3.OutPoint.String()},
+		Deposits:                []*deposit.Deposit{d3},
 		ClientPubkey:            clientPubKey,
 		ServerPubkey:            serverPubKey,
 		HtlcTimeoutSweepAddress: addr,
@@ -124,6 +181,23 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) {
 	err = swapStore.CreateLoopIn(ctxb, &swapSucceeded)
 	require.NoError(t, err)
 
+	// Create failed swap. Failed is the last final state, so this
+	// exercises the state-list query boundary.
+	swapHashFailed := lntypes.Hash{0x3, 0x2, 0x3, 0x5}
+	swapFailed := StaticAddressLoopIn{
+		SwapHash:                swapHashFailed,
+		SwapPreimage:            lntypes.Preimage{0x3, 0x2, 0x3, 0x5},
+		DepositOutpoints:        []string{d4.OutPoint.String()},
+		Deposits:                []*deposit.Deposit{d4},
+		ClientPubkey:            clientPubKey,
+		ServerPubkey:            serverPubKey,
+		HtlcTimeoutSweepAddress: addr,
+	}
+	swapFailed.SetState(Failed)
+
+	err = swapStore.CreateLoopIn(ctxb, &swapFailed)
+	require.NoError(t, err)
+
 	pendingSwaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(ctxb, PendingStates)
 	require.NoError(t, err)
 
@@ -142,16 +216,33 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) {
 	finalizedSwaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(ctxb, FinalStates)
 	require.NoError(t, err)
 
-	require.Len(t, finalizedSwaps, 1)
-	require.Equal(t, swapHashSucceeded, finalizedSwaps[0].SwapHash)
-	require.Equal(t, []string{d2.OutPoint.String()}, finalizedSwaps[0].DepositOutpoints)
-	require.Equal(t, Succeeded, finalizedSwaps[0].GetState())
+	require.Len(t, finalizedSwaps, 3)
+	finalizedByState := make(map[string]*StaticAddressLoopIn)
+	for _, swap := range finalizedSwaps {
+		finalizedByState[string(swap.GetState())] = swap
+	}
 
-	finalizedDeposits := finalizedSwaps[0].Deposits
+	timeoutSweptSwap := finalizedByState[string(HtlcTimeoutSwept)]
+	require.NotNil(t, timeoutSweptSwap)
+	require.Equal(t, swapHashTimeoutSwept, timeoutSweptSwap.SwapHash)
+	require.Equal(t, HtlcTimeoutSwept, timeoutSweptSwap.GetState())
+
+	succeededSwap := finalizedByState[string(Succeeded)]
+	require.NotNil(t, succeededSwap)
+	require.Equal(t, swapHashSucceeded, succeededSwap.SwapHash)
+	require.Equal(t, []string{d3.OutPoint.String()}, succeededSwap.DepositOutpoints)
+	require.Equal(t, Succeeded, succeededSwap.GetState())
+
+	failedSwap := finalizedByState[string(Failed)]
+	require.NotNil(t, failedSwap)
+	require.Equal(t, swapHashFailed, failedSwap.SwapHash)
+	require.Equal(t, Failed, failedSwap.GetState())
+
+	finalizedDeposits := succeededSwap.Deposits
 	require.Len(t, finalizedDeposits, 1)
-	require.Equal(t, d2.ID, finalizedDeposits[0].ID)
-	require.Equal(t, d2.OutPoint, finalizedDeposits[0].OutPoint)
-	require.Equal(t, d2.Value, finalizedDeposits[0].Value)
+	require.Equal(t, d3.ID, finalizedDeposits[0].ID)
+	require.Equal(t, d3.OutPoint, finalizedDeposits[0].OutPoint)
+	require.Equal(t, d3.Value, finalizedDeposits[0].Value)
 	require.Equal(t, deposit.LoopedIn, finalizedDeposits[0].GetState())
 }