diff --git a/instantout/actions.go b/instantout/actions.go index d1c405fd..25842652 100644 --- a/instantout/actions.go +++ b/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 { diff --git a/instantout/cleanup_test.go b/instantout/cleanup_test.go new file mode 100644 index 00000000..45640079 --- /dev/null +++ b/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) +} diff --git a/instantout/reservation/interfaces.go b/instantout/reservation/interfaces.go index 04bf830d..23658d5a 100644 --- a/instantout/reservation/interfaces.go +++ b/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. diff --git a/instantout/reservation/manager.go b/instantout/reservation/manager.go index 930dff7b..bccdfa1e 100644 --- a/instantout/reservation/manager.go +++ b/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. diff --git a/instantout/reservation/manager_test.go b/instantout/reservation/manager_test.go index 0437955e..ca3dec45 100644 --- a/instantout/reservation/manager_test.go +++ b/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 {