mirror of
https://github.com/lightninglabs/loop.git
synced 2026-08-13 12:33:03 +02:00
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.
This commit is contained in:
sputn1ck
parent
7b00baeac6
commit
44306275f7
4 changed files with 39 additions and 47 deletions
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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{
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue