From 44306275f7097ca2014a76f1d95acca0d3fcf58c Mon Sep 17 00:00:00 2001 From: sputn1ck Date: Mon, 21 Oct 2024 21:00:35 +0200 Subject: [PATCH] reservation: update package to use new fsm context This commit updates the reservation package to use the new fsm context instead of the old fsm context. This is only a first step in the process of migrating the reservation package to the new fsm context. The next step should be to remove the stored context in the reservation manager. --- instantout/reservation/actions.go | 34 ++++++++++++++------------ instantout/reservation/actions_test.go | 16 ++++++------ instantout/reservation/fsm.go | 17 ++++++------- instantout/reservation/manager.go | 19 +++++--------- 4 files changed, 39 insertions(+), 47 deletions(-) diff --git a/instantout/reservation/actions.go b/instantout/reservation/actions.go index 76cc3f06..1d58cd74 100644 --- a/instantout/reservation/actions.go +++ b/instantout/reservation/actions.go @@ -22,16 +22,16 @@ type InitReservationContext struct { // InitAction is the action that is executed when the reservation state machine // is initialized. It creates the reservation in the database and dispatches the // payment to the server. -func (f *FSM) InitAction(eventCtx fsm.EventContext) fsm.EventType { +func (f *FSM) InitAction(ctx context.Context, + eventCtx fsm.EventContext) fsm.EventType { + // Check if the context is of the correct type. reservationRequest, ok := eventCtx.(*InitReservationContext) if !ok { return f.HandleError(fsm.ErrInvalidContextType) } - keyRes, err := f.cfg.Wallet.DeriveNextKey( - f.ctx, KeyFamily, - ) + keyRes, err := f.cfg.Wallet.DeriveNextKey(ctx, KeyFamily) if err != nil { return f.HandleError(err) } @@ -45,7 +45,7 @@ func (f *FSM) InitAction(eventCtx fsm.EventContext) fsm.EventType { ClientKey: keyRes.PubKey.SerializeCompressed(), } - _, err = f.cfg.ReservationClient.OpenReservation(f.ctx, request) + _, err = f.cfg.ReservationClient.OpenReservation(ctx, request) if err != nil { return f.HandleError(err) } @@ -66,7 +66,7 @@ func (f *FSM) InitAction(eventCtx fsm.EventContext) fsm.EventType { f.reservation = reservation // Create the reservation in the database. - err = f.cfg.Store.CreateReservation(f.ctx, reservation) + err = f.cfg.Store.CreateReservation(ctx, reservation) if err != nil { return f.HandleError(err) } @@ -77,13 +77,15 @@ func (f *FSM) InitAction(eventCtx fsm.EventContext) fsm.EventType { // SubscribeToConfirmationAction is the action that is executed when the // reservation is waiting for confirmation. It subscribes to the confirmation // of the reservation transaction. -func (f *FSM) SubscribeToConfirmationAction(_ fsm.EventContext) fsm.EventType { +func (f *FSM) SubscribeToConfirmationAction(ctx context.Context, + _ fsm.EventContext) fsm.EventType { + pkscript, err := f.reservation.GetPkScript() if err != nil { return f.HandleError(err) } - callCtx, cancel := context.WithCancel(f.ctx) + callCtx, cancel := context.WithCancel(ctx) defer cancel() // Subscribe to the confirmation of the reservation transaction. @@ -141,7 +143,7 @@ func (f *FSM) SubscribeToConfirmationAction(_ fsm.EventContext) fsm.EventType { return OnTimedOut } - case <-f.ctx.Done(): + case <-ctx.Done(): return fsm.NoOp } } @@ -150,10 +152,10 @@ func (f *FSM) SubscribeToConfirmationAction(_ fsm.EventContext) fsm.EventType { // AsyncWaitForExpiredOrSweptAction waits for the reservation to be either // expired or swept. This is non-blocking and can be used to wait for the // reservation to expire while expecting other events. -func (f *FSM) AsyncWaitForExpiredOrSweptAction(_ fsm.EventContext, -) fsm.EventType { +func (f *FSM) AsyncWaitForExpiredOrSweptAction(ctx context.Context, + _ fsm.EventContext) fsm.EventType { - notifCtx, cancel := context.WithCancel(f.ctx) + notifCtx, cancel := context.WithCancel(ctx) blockHeightChan, errEpochChan, err := f.cfg.ChainNotifier. RegisterBlockEpochNtfn(notifCtx) @@ -184,13 +186,13 @@ func (f *FSM) AsyncWaitForExpiredOrSweptAction(_ fsm.EventContext, errSpendChan, ) if err != nil { - f.handleAsyncError(err) + f.handleAsyncError(ctx, err) return } if op == fsm.NoOp { return } - err = f.SendEvent(op, nil) + err = f.SendEvent(ctx, op, nil) if err != nil { f.Errorf("Error sending %s event: %v", op, err) } @@ -229,10 +231,10 @@ func (f *FSM) handleSubcriptions(ctx context.Context, } } -func (f *FSM) handleAsyncError(err error) { +func (f *FSM) handleAsyncError(ctx context.Context, err error) { f.LastActionError = err f.Errorf("Error on async action: %v", err) - err2 := f.SendEvent(fsm.OnError, err) + err2 := f.SendEvent(ctx, fsm.OnError, err) if err2 != nil { f.Errorf("Error sending event: %v", err2) } diff --git a/instantout/reservation/actions_test.go b/instantout/reservation/actions_test.go index d89e526d..3b5adfd2 100644 --- a/instantout/reservation/actions_test.go +++ b/instantout/reservation/actions_test.go @@ -144,7 +144,6 @@ func TestInitReservationAction(t *testing.T) { ).Return(tc.mockStoreErr) reservationFSM := &FSM{ - ctx: ctxb, cfg: &Config{ Wallet: mockLnd.WalletKit, ChainNotifier: mockLnd.ChainNotifier, @@ -154,7 +153,7 @@ func TestInitReservationAction(t *testing.T) { StateMachine: &fsm.StateMachine{}, } - event := reservationFSM.InitAction(tc.eventCtx) + event := reservationFSM.InitAction(ctxb, tc.eventCtx) require.Equal(t, tc.expectedEvent, event) } } @@ -227,10 +226,10 @@ func TestSubscribeToConfirmationAction(t *testing.T) { tc := tc t.Run(tc.name, func(t *testing.T) { chainNotifier := new(MockChainNotifier) - + ctxb := context.Background() // Create the FSM. r := NewFSMFromReservation( - context.Background(), &Config{ + &Config{ ChainNotifier: chainNotifier, }, &Reservation{ @@ -296,7 +295,7 @@ func TestSubscribeToConfirmationAction(t *testing.T) { } }() - eventType := r.SubscribeToConfirmationAction(nil) + eventType := r.SubscribeToConfirmationAction(ctxb, nil) // Assert that the return value is as expected require.Equal(t, tc.expectedEvent, eventType) @@ -335,10 +334,11 @@ func TestAsyncWaitForExpiredOrSweptAction(t *testing.T) { tc := tc t.Run(tc.name, func(t *testing.T) { // Create a mock ChainNotifier and Reservation chainNotifier := new(MockChainNotifier) + ctxb := context.Background() // Define your FSM r := NewFSMFromReservation( - context.Background(), &Config{ + &Config{ ChainNotifier: chainNotifier, }, &Reservation{ @@ -361,7 +361,7 @@ func TestAsyncWaitForExpiredOrSweptAction(t *testing.T) { make(chan error), tc.spendErr, ) - eventType := r.AsyncWaitForExpiredOrSweptAction(nil) + eventType := r.AsyncWaitForExpiredOrSweptAction(ctxb, nil) // Assert that the return value is as expected require.Equal(t, tc.expectedEvent, eventType) }) @@ -415,7 +415,7 @@ func TestHandleSubcriptions(t *testing.T) { // Create the FSM. r := NewFSMFromReservation( - context.Background(), &Config{ + &Config{ ChainNotifier: chainNotifier, }, &Reservation{ diff --git a/instantout/reservation/fsm.go b/instantout/reservation/fsm.go index 86afbbaa..6bf567d2 100644 --- a/instantout/reservation/fsm.go +++ b/instantout/reservation/fsm.go @@ -40,26 +40,21 @@ type FSM struct { cfg *Config reservation *Reservation - - ctx context.Context } // NewFSM creates a new reservation FSM. -func NewFSM(ctx context.Context, cfg *Config) *FSM { +func NewFSM(cfg *Config) *FSM { reservation := &Reservation{ State: fsm.EmptyState, } - return NewFSMFromReservation(ctx, cfg, reservation) + return NewFSMFromReservation(cfg, reservation) } // NewFSMFromReservation creates a new reservation FSM from an existing // reservation recovered from the database. -func NewFSMFromReservation(ctx context.Context, cfg *Config, - reservation *Reservation) *FSM { - +func NewFSMFromReservation(cfg *Config, reservation *Reservation) *FSM { reservationFsm := &FSM{ - ctx: ctx, cfg: cfg, reservation: reservation, } @@ -206,7 +201,9 @@ func (f *FSM) GetReservationStates() fsm.States { // updateReservation updates the reservation in the database. This function // is called after every new state transition. -func (r *FSM) updateReservation(notification fsm.Notification) { +func (r *FSM) updateReservation(ctx context.Context, + notification fsm.Notification) { + if r.reservation == nil { return } @@ -229,7 +226,7 @@ func (r *FSM) updateReservation(notification fsm.Notification) { return } - err := r.cfg.Store.UpdateReservation(r.ctx, r.reservation) + err := r.cfg.Store.UpdateReservation(ctx, r.reservation) if err != nil { r.Errorf("unable to update reservation: %v", err) } diff --git a/instantout/reservation/manager.go b/instantout/reservation/manager.go index 3a20d113..06dedee3 100644 --- a/instantout/reservation/manager.go +++ b/instantout/reservation/manager.go @@ -22,8 +22,6 @@ type Manager struct { // activeReservations contains all the active reservationsFSMs. activeReservations map[ID]*FSM - runCtx context.Context - sync.Mutex } @@ -42,7 +40,6 @@ func (m *Manager) Run(ctx context.Context, height int32) error { runCtx, cancel := context.WithCancel(ctx) defer cancel() - m.runCtx = runCtx currentHeight := height err := m.RecoverReservations(runCtx) @@ -111,9 +108,7 @@ func (m *Manager) newReservation(ctx context.Context, currentHeight uint32, // Create the reservation state machine. We need to pass in the runCtx // of the reservation manager so that the state machine will keep on // running even if the grpc conte - reservationFSM := NewFSM( - ctx, m.cfg, - ) + reservationFSM := NewFSM(m.cfg) // Add the reservation to the active reservations map. m.Lock() @@ -130,7 +125,7 @@ func (m *Manager) newReservation(ctx context.Context, currentHeight uint32, // Send the init event to the state machine. go func() { - err = reservationFSM.SendEvent(OnServerRequest, initContext) + err = reservationFSM.SendEvent(ctx, OnServerRequest, initContext) if err != nil { log.Errorf("Error sending init event: %v", err) } @@ -171,16 +166,14 @@ func (m *Manager) RecoverReservations(ctx context.Context) error { fsmCtx := context.WithValue(ctx, reservation.ID, nil) - reservationFSM := NewFSMFromReservation( - fsmCtx, m.cfg, reservation, - ) + reservationFSM := NewFSMFromReservation(m.cfg, reservation) m.activeReservations[reservation.ID] = reservationFSM // As SendEvent can block, we'll start a goroutine to process // the event. go func() { - err := reservationFSM.SendEvent(OnRecover, nil) + err := reservationFSM.SendEvent(fsmCtx, OnRecover, nil) if err != nil { log.Errorf("FSM %v Error sending recover "+ "event %v, state: %v", @@ -217,7 +210,7 @@ func (m *Manager) LockReservation(ctx context.Context, id ID) error { } // Try to send the lock event to the reservation. - err := reservation.SendEvent(OnLocked, nil) + err := reservation.SendEvent(ctx, OnLocked, nil) if err != nil { return err } @@ -237,7 +230,7 @@ func (m *Manager) UnlockReservation(ctx context.Context, id ID) error { } // Try to send the unlock event to the reservation. - err := reservation.SendEvent(OnUnlocked, nil) + err := reservation.SendEvent(ctx, OnUnlocked, nil) if err != nil && strings.Contains(err.Error(), "config error") { // If the error is a config error, we can ignore it, as the // reservation is already unlocked.