reservation: bound and prune active state machines

Limit newly accepted reservation state machines, resume all persisted
reservations, remove terminal entries from memory, and make cleanup
resilient to observer-driven eviction.

instantout/actions.go

@@ -617,20 +617,23 @@ func (f *FSM) WaitForHtlcSweepConfirmedAction(ctx context.Context,
 // handleErrorAndUnlockReservations handles an error and unlocks the
 // reservations.
 func (f *FSM) handleErrorAndUnlockReservations(ctx context.Context,
-	err error) fsm.EventType {
+	actionErr error) fsm.EventType {
 	// We might get here from a canceled context, we create a new context
 	// with a timeout to unlock the reservations.
-	ctx, cancel := context.WithTimeout(ctx, time.Second*30)
+	cleanupCtx, cancel := context.WithTimeout(
+		context.WithoutCancel(ctx), time.Second*30,
+	)
 	defer cancel()
 
 	// Unlock the reservations.
+	var unlockErr error
 	for _, reservation := range f.InstantOut.Reservations {
 		err := f.cfg.ReservationManager.UnlockReservation(
-			ctx, reservation.ID,
+			cleanupCtx, reservation.ID,
 		)
 		if err != nil {
 			f.Errorf("error unlocking reservation: %v", err)
-			return f.HandleError(err)
+			unlockErr = errors.Join(unlockErr, err)
 		}
 	}
 
@@ -638,10 +641,12 @@ func (f *FSM) handleErrorAndUnlockReservations(ctx context.Context,
 	// release the reservations. This can be done in a goroutine as we
 	// wan't to fail the fsm early.
 	go func() {
-		ctx, cancel := context.WithTimeout(ctx, time.Second*30)
+		cancelCtx, cancel := context.WithTimeout(
+			context.WithoutCancel(ctx), time.Second*30,
+		)
 		defer cancel()
 		_, cancelErr := f.cfg.InstantOutClient.CancelInstantSwap(
-			ctx, &swapserverrpc.CancelInstantSwapRequest{
+			cancelCtx, &swapserverrpc.CancelInstantSwapRequest{
 				SwapHash: f.InstantOut.SwapHash[:],
 			},
 		)
@@ -652,7 +657,13 @@ func (f *FSM) handleErrorAndUnlockReservations(ctx context.Context,
 		}
 	}()
 
-	return f.HandleError(err)
+	// Preserve the action failure when cleanup also fails. If cleanup was
+	// the only failure, report it to the state machine.
+	if actionErr != nil {
+		return f.HandleError(actionErr)
+	}
+
+	return f.HandleError(unlockErr)
 }
 
 func getMaxRoutingFee(amt btcutil.Amount) btcutil.Amount {

instantout/cleanup_test.go

@@ -0,0 +1,77 @@
+package instantout
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/lightninglabs/loop/fsm"
+	"github.com/lightninglabs/loop/instantout/reservation"
+	"github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+type cleanupTestReservationManager struct {
+	ReservationManager
+
+	unlockErr error
+}
+
+func (m *cleanupTestReservationManager) UnlockReservation(context.Context,
+	reservation.ID) error {
+
+	return m.unlockErr
+}
+
+type cleanupTestInstantOutClient struct {
+	swapserverrpc.InstantSwapServerClient
+
+	canceled chan struct{}
+}
+
+func (c *cleanupTestInstantOutClient) CancelInstantSwap(context.Context,
+	*swapserverrpc.CancelInstantSwapRequest, ...grpc.CallOption) (
+	*swapserverrpc.CancelInstantSwapResponse, error) {
+
+	close(c.canceled)
+	return &swapserverrpc.CancelInstantSwapResponse{}, nil
+}
+
+// TestCleanupPreservesActionError verifies that an unlock failure doesn't
+// replace the action failure or prevent the cancellation notification.
+func TestCleanupPreservesActionError(t *testing.T) {
+	actionErr := errors.New("action failed")
+	cancelClient := &cleanupTestInstantOutClient{
+		canceled: make(chan struct{}),
+	}
+	instantOutFSM := &FSM{
+		StateMachine: &fsm.StateMachine{},
+		cfg: &Config{
+			ReservationManager: &cleanupTestReservationManager{
+				unlockErr: errors.New("unlock failed"),
+			},
+			InstantOutClient: cancelClient,
+		},
+		InstantOut: &InstantOut{
+			Reservations: []*reservation.Reservation{
+				{ID: reservation.ID{1}},
+			},
+		},
+	}
+
+	event := instantOutFSM.handleErrorAndUnlockReservations(
+		t.Context(), actionErr,
+	)
+	require.Equal(t, fsm.OnError, event)
+	require.ErrorIs(t, instantOutFSM.LastActionError, actionErr)
+	require.Eventually(t, func() bool {
+		select {
+		case <-cancelClient.canceled:
+			return true
+		default:
+			return false
+		}
+	}, time.Second, time.Millisecond)
+}

instantout/reservation/interfaces.go

@@ -8,14 +8,18 @@ import (
 )
 
 var (
-	ErrReservationAlreadyExists = fmt.Errorf("reservation already exists")
-	ErrReservationNotFound      = fmt.Errorf("reservation not found")
+	ErrReservationAlreadyExists  = fmt.Errorf("reservation already exists")
+	ErrReservationNotFound       = fmt.Errorf("reservation not found")
+	ErrTooManyActiveReservations = fmt.Errorf(
+		"too many active reservations",
+	)
 )
 
 const (
-	KeyFamily         = int32(42068)
-	DefaultConfTarget = int32(3)
-	IdLength          = 32
+	KeyFamily             = int32(42068)
+	DefaultConfTarget     = int32(3)
+	IdLength              = 32
+	maxActiveReservations = 1000
 )
 
 // Store is the interface that stores the reservations.

instantout/reservation/manager.go

@@ -14,6 +14,11 @@ import (
 	reservationrpc "github.com/lightninglabs/loop/swapserverrpc"
 )
 
+var (
+	reservationStateWaitTimeout = 5 * time.Second
+	reservationStatePollDelay   = time.Second
+)
+
 // Manager manages the reservation state machines.
 type Manager struct {
 	sync.Mutex
@@ -26,6 +31,28 @@ type Manager struct {
 	activeReservations map[ID]*FSM
 }
 
+// finalStateObserver removes a reservation FSM from the active set once it
+// reaches a terminal state.
+type finalStateObserver struct {
+	manager *Manager
+	id      ID
+	fsm     *FSM
+}
+
+// Notify implements the fsm.Observer interface.
+func (o *finalStateObserver) Notify(notification fsm.Notification) {
+	if !isFinalState(notification.NextState) {
+		return
+	}
+
+	o.manager.Lock()
+	defer o.manager.Unlock()
+
+	if o.manager.activeReservations[o.id] == o.fsm {
+		delete(o.manager.activeReservations, o.id)
+	}
+}
+
 // NewManager creates a new reservation manager.
 func NewManager(cfg *Config) *Manager {
 	return &Manager{
@@ -134,9 +161,19 @@ func (m *Manager) newReservation(ctx context.Context, currentHeight uint32,
 		m.Unlock()
 		return nil, ErrReservationAlreadyExists
 	}
+	if len(m.activeReservations) >= maxActiveReservations {
+		m.Unlock()
+		return nil, ErrTooManyActiveReservations
+	}
 	m.activeReservations[reservationID] = reservationFSM
 	m.Unlock()
 
+	reservationFSM.RegisterObserver(&finalStateObserver{
+		manager: m,
+		id:      reservationID,
+		fsm:     reservationFSM,
+	})
+
 	initContext := &InitReservationContext{
 		reservationID: reservationID,
 		serverPubkey:  serverKey,
@@ -158,16 +195,10 @@ func (m *Manager) newReservation(ctx context.Context, currentHeight uint32,
 	// We'll now wait for the reservation to be in the state where it is
 	// waiting to be confirmed.
 	err = reservationFSM.DefaultObserver.WaitForState(
-		ctx, 5*time.Second, WaitForConfirmation,
-		fsm.WithWaitForStateOption(time.Second),
+		ctx, reservationStateWaitTimeout, WaitForConfirmation,
+		fsm.WithWaitForStateOption(reservationStatePollDelay),
 	)
 	if err != nil {
-		m.Lock()
-		if m.activeReservations[reservationID] == reservationFSM {
-			delete(m.activeReservations, reservationID)
-		}
-		m.Unlock()
-
 		if reservationFSM.LastActionError != nil {
 			return nil, fmt.Errorf("error waiting for "+
 				"state: %v, last action error: %v",
@@ -198,7 +229,14 @@ func (m *Manager) RecoverReservations(ctx context.Context) error {
 
 		reservationFSM := NewFSMFromReservation(m.cfg, reservation)
 
+		m.Lock()
 		m.activeReservations[reservation.ID] = reservationFSM
+		m.Unlock()
+		reservationFSM.RegisterObserver(&finalStateObserver{
+			manager: m,
+			id:      reservation.ID,
+			fsm:     reservationFSM,
+		})
 
 		// As SendEvent can block, we'll start a goroutine to process
 		// the event.
@@ -236,7 +274,7 @@ func (m *Manager) LockReservation(ctx context.Context, id ID) error {
 	m.Unlock()
 
 	if !ok {
-		return fmt.Errorf("reservation not found")
+		return ErrReservationNotFound
 	}
 
 	// Try to send the lock event to the reservation.
@@ -256,7 +294,20 @@ func (m *Manager) UnlockReservation(ctx context.Context, id ID) error {
 	m.Unlock()
 
 	if !ok {
-		return fmt.Errorf("reservation not found")
+		storedReservation, err := m.cfg.Store.GetReservation(ctx, id)
+		if err != nil {
+			return err
+		}
+
+		// Terminal reservations are removed from the active set. Treat an
+		// unlock after that removal as idempotent, while still surfacing a
+		// missing active FSM for reservations that should be running.
+		if isFinalState(storedReservation.State) {
+			return nil
+		}
+
+		return fmt.Errorf("%w: reservation %x is in state %v",
+			ErrReservationNotFound, id, storedReservation.State)
 	}
 
 	// Try to send the unlock event to the reservation.

instantout/reservation/manager_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/lightninglabs/loop/swapserverrpc"
 	"github.com/lightninglabs/loop/test"
 	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/keychain"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
@@ -97,6 +98,11 @@ func TestManager(t *testing.T) {
 	// We'll now expect the reservation to be expired.
 	err = reservationFSM.DefaultObserver.WaitForState(ctxb, 5*time.Second, Spent)
 	require.NoError(t, err)
+
+	testContext.manager.Lock()
+	_, ok := testContext.manager.activeReservations[defaultReservationId]
+	testContext.manager.Unlock()
+	require.False(t, ok)
 }
 
 // TestManagerContinuesAfterInvalidNotification verifies that a malformed
@@ -173,6 +179,166 @@ func TestManagerRejectsDuplicateReservation(t *testing.T) {
 	)
 }
 
+// TestManagerLimitsActiveReservations verifies that server notifications
+// cannot grow the active FSM set without bound.
+func TestManagerLimitsActiveReservations(t *testing.T) {
+	testContext := newManagerTestContext(t)
+
+	for i := range maxActiveReservations {
+		var id ID
+		id[0] = byte(i)
+		id[1] = byte(i >> 8)
+		testContext.manager.activeReservations[id] = NewFSM(
+			testContext.manager.cfg,
+		)
+	}
+
+	reservationFSM, err := testContext.manager.newReservation(
+		t.Context(), uint32(testContext.mockLnd.Height),
+		&swapserverrpc.ServerReservationNotification{
+			ReservationId: defaultReservationId[:],
+			Value:         uint64(defaultValue),
+			ServerKey:     defaultPubkeyBytes,
+			Expiry: uint32(testContext.mockLnd.Height) +
+				defaultExpiry,
+		},
+	)
+	require.ErrorIs(t, err, ErrTooManyActiveReservations)
+	require.Nil(t, reservationFSM)
+	require.Len(
+		t, testContext.manager.activeReservations,
+		maxActiveReservations,
+	)
+}
+
+// TestManagerKeepsReservationAfterWaitTimeout verifies that a caller-side
+// wait timeout doesn't evict a reservation FSM that is still initializing.
+func TestManagerKeepsReservationAfterWaitTimeout(t *testing.T) {
+	testContext := newManagerTestContext(t)
+
+	originalWaitTimeout := reservationStateWaitTimeout
+	originalPollDelay := reservationStatePollDelay
+	reservationStateWaitTimeout = 20 * time.Millisecond
+	reservationStatePollDelay = time.Millisecond
+	t.Cleanup(func() {
+		reservationStateWaitTimeout = originalWaitTimeout
+		reservationStatePollDelay = originalPollDelay
+	})
+
+	releaseOpen := make(chan struct{})
+	testContext.mockReservationClient.ExpectedCalls = nil
+	testContext.mockReservationClient.On(
+		"OpenReservation", mock.Anything, mock.Anything, mock.Anything,
+	).Run(func(mock.Arguments) {
+		<-releaseOpen
+	}).Return(
+		&swapserverrpc.ServerOpenReservationResponse{}, nil,
+	)
+
+	reservationFSM, err := testContext.manager.newReservation(
+		t.Context(), uint32(testContext.mockLnd.Height),
+		&swapserverrpc.ServerReservationNotification{
+			ReservationId: defaultReservationId[:],
+			Value:         uint64(defaultValue),
+			ServerKey:     defaultPubkeyBytes,
+			Expiry: uint32(testContext.mockLnd.Height) +
+				defaultExpiry,
+		},
+	)
+	require.Error(t, err)
+	require.Nil(t, reservationFSM)
+
+	testContext.manager.Lock()
+	activeFSM := testContext.manager.activeReservations[defaultReservationId]
+	testContext.manager.Unlock()
+	require.NotNil(t, activeFSM)
+
+	close(releaseOpen)
+	require.NoError(t, activeFSM.DefaultObserver.WaitForState(
+		t.Context(), 5*time.Second, WaitForConfirmation,
+	))
+}
+
+// TestManagerRecoversAllPersistedReservations verifies that the cap applied to
+// new notifications doesn't prevent the manager from resuming obligations
+// already recorded in the database. The terminal transitions also exercise
+// concurrent observer-driven removal from the active map.
+func TestManagerRecoversAllPersistedReservations(t *testing.T) {
+	reservations := make([]*Reservation, maxActiveReservations+1)
+	for i := range reservations {
+		reservations[i] = &Reservation{
+			ID: ID{
+				byte(i), byte(i >> 8), byte(i >> 16),
+			},
+			State:           Init,
+			ProtocolVersion: ProtocolVersionServerInitiated,
+		}
+	}
+
+	manager := NewManager(&Config{
+		Store: &recoveryStore{reservations: reservations},
+	})
+	require.NoError(t, manager.RecoverReservations(t.Context()))
+	require.Eventually(t, func() bool {
+		manager.Lock()
+		defer manager.Unlock()
+
+		return len(manager.activeReservations) == 0
+	}, 5*time.Second, time.Millisecond)
+}
+
+// TestUnlockTerminalReservationIsIdempotent verifies that cleanup can safely
+// race with terminal-state eviction without masking the original swap result.
+func TestUnlockTerminalReservationIsIdempotent(t *testing.T) {
+	testContext := newManagerTestContext(t)
+	storedReservation := &Reservation{
+		ID:              defaultReservationId,
+		State:           Init,
+		ClientPubkey:    defaultPubkey,
+		ServerPubkey:    defaultPubkey,
+		Value:           defaultValue,
+		Expiry:          defaultExpiry,
+		ProtocolVersion: ProtocolVersionServerInitiated,
+		KeyLocator: keychain.KeyLocator{
+			Family: keychain.KeyFamily(KeyFamily),
+			Index:  1,
+		},
+	}
+
+	require.NoError(t, testContext.manager.cfg.Store.CreateReservation(
+		t.Context(), storedReservation,
+	))
+	storedReservation.State = TimedOut
+	require.NoError(t, testContext.manager.cfg.Store.UpdateReservation(
+		t.Context(), storedReservation,
+	))
+
+	require.NoError(t, testContext.manager.UnlockReservation(
+		t.Context(), defaultReservationId,
+	))
+	require.ErrorIs(t, testContext.manager.UnlockReservation(
+		t.Context(), ID{1},
+	), ErrReservationNotFound)
+}
+
+type recoveryStore struct {
+	Store
+
+	reservations []*Reservation
+}
+
+func (s *recoveryStore) ListReservations(context.Context) ([]*Reservation,
+	error) {
+
+	return s.reservations, nil
+}
+
+func (s *recoveryStore) UpdateReservation(context.Context,
+	*Reservation) error {
+
+	return nil
+}
+
 // ManagerTestContext is a helper struct that contains all the necessary
 // components to test the reservation manager.
 type ManagerTestContext struct {