GlcoinCore/loop
1
0
Fork 0
mirror of https://github.com/lightninglabs/loop.git synced 2026-08-13 12:33:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #839 from sputn1ck/fsm_ctx

Browse source 
FSM: add ctx to SendEvent and Actions
This commit is contained in:
Konstantin Nick 2024-10-22 16:49:45 +02:00 • committed by GitHub
commit 78fafbaeb4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 188 additions and 154 deletions
Download patch file Download diff file Expand all files Collapse all files

131
instantout/actions.go
View file

@ -65,7 +65,9 @@ type InitInstantOutCtx struct {
// InitInstantOutAction is the first action that is executed when the instant
// out FSM is started. It will send the instant out request to the server.
func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
func (f *FSM) InitInstantOutAction(ctx context.Context,
eventCtx fsm.EventContext) fsm.EventType {
initCtx, ok := eventCtx.(*InitInstantOutCtx)
if !ok {
return f.HandleError(fsm.ErrInvalidContextType)
@ -86,9 +88,7 @@ func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
// The requested amount needs to be full reservation amounts.
for _, reservationId := range initCtx.reservations {
resId := reservationId
res, err := f.cfg.ReservationManager.GetReservation(
f.ctx, resId,
)
res, err := f.cfg.ReservationManager.GetReservation(ctx, resId)
if err != nil {
return f.HandleError(err)
}
@ -120,7 +120,7 @@ func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
}
// Create the keys for the swap.
keyRes, err := f.cfg.Wallet.DeriveNextKey(f.ctx, KeyFamily)
keyRes, err := f.cfg.Wallet.DeriveNextKey(ctx, KeyFamily)
if err != nil {
return f.HandleError(err)
}
@ -128,7 +128,7 @@ func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
swapHash := preimage.Hash()
// Create a high fee rate so that the htlc will be confirmed quickly.
feeRate, err := f.cfg.Wallet.EstimateFeeRate(f.ctx, urgentConfTarget)
feeRate, err := f.cfg.Wallet.EstimateFeeRate(ctx, urgentConfTarget)
if err != nil {
f.Infof("error estimating fee rate: %v", err)
return f.HandleError(err)
@ -136,7 +136,7 @@ func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
// Send the instantout request to the server.
instantOutResponse, err := f.cfg.InstantOutClient.RequestInstantLoopOut(
f.ctx,
ctx,
&swapserverrpc.InstantLoopOutRequest{
ReceiverKey: keyRes.PubKey.SerializeCompressed(),
SwapHash: swapHash[:],
@ -151,7 +151,7 @@ func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
}
// Decode the invoice to check if the hash is valid.
payReq, err := f.cfg.LndClient.DecodePaymentRequest(
f.ctx, instantOutResponse.SwapInvoice,
ctx, instantOutResponse.SwapInvoice,
)
if err != nil {
return f.HandleError(err)
@ -170,7 +170,7 @@ func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
sweepAddress := initCtx.sweepAddress
if sweepAddress == nil {
sweepAddress, err = f.cfg.Wallet.NextAddr(
f.ctx, lnwallet.DefaultAccountName,
ctx, lnwallet.DefaultAccountName,
walletrpc.AddressType_TAPROOT_PUBKEY, false,
)
if err != nil {
@ -196,7 +196,7 @@ func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
sweepAddress: sweepAddress,
}
err = f.cfg.Store.CreateInstantLoopOut(f.ctx, instantOut)
err = f.cfg.Store.CreateInstantLoopOut(ctx, instantOut)
if err != nil {
return f.HandleError(err)
}
@ -208,21 +208,23 @@ func (f *FSM) InitInstantOutAction(eventCtx fsm.EventContext) fsm.EventType {
// PollPaymentAcceptedAction locks the reservations, sends the payment to the
// server and polls the server for the payment status.
func (f *FSM) PollPaymentAcceptedAction(_ fsm.EventContext) fsm.EventType {
func (f *FSM) PollPaymentAcceptedAction(ctx context.Context,
_ fsm.EventContext) fsm.EventType {
// Now that we're doing the swap, we first lock the reservations
// so that they can't be used for other swaps.
for _, reservation := range f.InstantOut.Reservations {
err := f.cfg.ReservationManager.LockReservation(
f.ctx, reservation.ID,
ctx, reservation.ID,
)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
}
// Now we send the payment to the server.
payChan, paymentErrChan, err := f.cfg.RouterClient.SendPayment(
f.ctx,
ctx,
lndclient.SendPaymentRequest{
Invoice: f.InstantOut.swapInvoice,
Timeout: defaultSendpaymentTimeout,
@ -232,7 +234,7 @@ func (f *FSM) PollPaymentAcceptedAction(_ fsm.EventContext) fsm.EventType {
)
if err != nil {
f.Errorf("error sending payment: %v", err)
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
// We'll continuously poll the server for the payment status.
@ -246,20 +248,20 @@ func (f *FSM) PollPaymentAcceptedAction(_ fsm.EventContext) fsm.EventType {
f.Debugf("payment result: %v", payRes)
if payRes.State == lnrpc.Payment_FAILED {
return f.handleErrorAndUnlockReservations(
fmt.Errorf("payment failed: %v",
ctx, fmt.Errorf("payment failed: %v",
payRes.FailureReason),
)
}
case err := <-paymentErrChan:
f.Errorf("error sending payment: %v", err)
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
case <-f.ctx.Done():
return f.handleErrorAndUnlockReservations(nil)
case <-ctx.Done():
return f.handleErrorAndUnlockReservations(ctx, nil)
case <-timer.C:
res, err := f.cfg.InstantOutClient.PollPaymentAccepted(
f.ctx,
ctx,
&swapserverrpc.PollPaymentAcceptedRequest{
SwapHash: f.InstantOut.SwapHash[:],
},
@ -267,7 +269,7 @@ func (f *FSM) PollPaymentAcceptedAction(_ fsm.EventContext) fsm.EventType {
if err != nil {
pollPaymentTries++
if pollPaymentTries > 20 {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
}
if res != nil && res.Accepted {
@ -280,74 +282,76 @@ func (f *FSM) PollPaymentAcceptedAction(_ fsm.EventContext) fsm.EventType {
// BuildHTLCAction creates the htlc transaction, exchanges nonces with
// the server and sends the htlc signatures to the server.
func (f *FSM) BuildHTLCAction(eventCtx fsm.EventContext) fsm.EventType {
func (f *FSM) BuildHTLCAction(ctx context.Context,
eventCtx fsm.EventContext) fsm.EventType {
htlcSessions, htlcClientNonces, err := f.InstantOut.createMusig2Session(
f.ctx, f.cfg.Signer,
ctx, f.cfg.Signer,
)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
f.htlcMusig2Sessions = htlcSessions
// Send the server the client nonces.
htlcInitRes, err := f.cfg.InstantOutClient.InitHtlcSig(
f.ctx,
ctx,
&swapserverrpc.InitHtlcSigRequest{
SwapHash: f.InstantOut.SwapHash[:],
HtlcClientNonces: htlcClientNonces,
},
)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
if len(htlcInitRes.HtlcServerNonces) != len(f.InstantOut.Reservations) {
return f.handleErrorAndUnlockReservations(
errors.New("invalid number of server nonces"),
ctx, errors.New("invalid number of server nonces"),
)
}
htlcServerNonces, err := toNonces(htlcInitRes.HtlcServerNonces)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
// Now that our nonces are set, we can create and sign the htlc
// transaction.
htlcTx, err := f.InstantOut.createHtlcTransaction(f.cfg.Network)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
// Next we'll get our sweep tx signatures.
htlcSigs, err := f.InstantOut.signMusig2Tx(
f.ctx, f.cfg.Signer, htlcTx, f.htlcMusig2Sessions,
ctx, f.cfg.Signer, htlcTx, f.htlcMusig2Sessions,
htlcServerNonces,
)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
// Send the server the htlc signatures.
htlcRes, err := f.cfg.InstantOutClient.PushHtlcSig(
f.ctx,
ctx,
&swapserverrpc.PushHtlcSigRequest{
SwapHash: f.InstantOut.SwapHash[:],
ClientSigs: htlcSigs,
},
)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
// We can now finalize the htlc transaction.
htlcTx, err = f.InstantOut.finalizeMusig2Transaction(
f.ctx, f.cfg.Signer, f.htlcMusig2Sessions, htlcTx,
ctx, f.cfg.Signer, f.htlcMusig2Sessions, htlcTx,
htlcRes.ServerSigs,
)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
f.InstantOut.finalizedHtlcTx = htlcTx
@ -359,25 +363,27 @@ func (f *FSM) BuildHTLCAction(eventCtx fsm.EventContext) fsm.EventType {
// sweepless sweep transaction and sends the signatures to the server. Finally,
// it publishes the sweepless sweep transaction. If any of the steps after
// pushing the preimage fail, the htlc timeout transaction will be published.
func (f *FSM) PushPreimageAction(eventCtx fsm.EventContext) fsm.EventType {
func (f *FSM) PushPreimageAction(ctx context.Context,
eventCtx fsm.EventContext) fsm.EventType {
// First we'll create the musig2 context.
coopSessions, coopClientNonces, err := f.InstantOut.createMusig2Session(
f.ctx, f.cfg.Signer,
ctx, f.cfg.Signer,
)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
f.sweeplessSweepSessions = coopSessions
// Get the feerate for the coop sweep.
feeRate, err := f.cfg.Wallet.EstimateFeeRate(f.ctx, normalConfTarget)
feeRate, err := f.cfg.Wallet.EstimateFeeRate(ctx, normalConfTarget)
if err != nil {
return f.handleErrorAndUnlockReservations(err)
return f.handleErrorAndUnlockReservations(ctx, err)
}
pushPreImageRes, err := f.cfg.InstantOutClient.PushPreimage(
f.ctx,
ctx,
&swapserverrpc.PushPreimageRequest{
Preimage: f.InstantOut.swapPreimage[:],
ClientNonces: coopClientNonces,
@ -408,7 +414,7 @@ func (f *FSM) PushPreimageAction(eventCtx fsm.EventContext) fsm.EventType {
// Next we'll get our sweep tx signatures.
_, err = f.InstantOut.signMusig2Tx(
f.ctx, f.cfg.Signer, sweepTx, f.sweeplessSweepSessions,
ctx, f.cfg.Signer, sweepTx, f.sweeplessSweepSessions,
coopServerNonces,
)
if err != nil {
@ -418,7 +424,7 @@ func (f *FSM) PushPreimageAction(eventCtx fsm.EventContext) fsm.EventType {
// Now we'll finalize the sweepless sweep transaction.
sweepTx, err = f.InstantOut.finalizeMusig2Transaction(
f.ctx, f.cfg.Signer, f.sweeplessSweepSessions, sweepTx,
ctx, f.cfg.Signer, f.sweeplessSweepSessions, sweepTx,
pushPreImageRes.Musig2SweepSigs,
)
if err != nil {
@ -430,7 +436,7 @@ func (f *FSM) PushPreimageAction(eventCtx fsm.EventContext) fsm.EventType {
f.InstantOut.swapPreimage.Hash())
// Publish the sweepless sweep transaction.
err = f.cfg.Wallet.PublishTransaction(f.ctx, sweepTx, txLabel)
err = f.cfg.Wallet.PublishTransaction(ctx, sweepTx, txLabel)
if err != nil {
f.LastActionError = err
return OnErrorPublishHtlc
@ -446,7 +452,7 @@ func (f *FSM) PushPreimageAction(eventCtx fsm.EventContext) fsm.EventType {
// WaitForSweeplessSweepConfirmedAction waits for the sweepless sweep
// transaction to be confirmed.
func (f *FSM) WaitForSweeplessSweepConfirmedAction(
func (f *FSM) WaitForSweeplessSweepConfirmedAction(ctx context.Context,
eventCtx fsm.EventContext) fsm.EventType {
pkscript, err := txscript.PayToAddrScript(f.InstantOut.sweepAddress)
@ -456,7 +462,7 @@ func (f *FSM) WaitForSweeplessSweepConfirmedAction(
confChan, confErrChan, err := f.cfg.ChainNotifier.
RegisterConfirmationsNtfn(
f.ctx, f.InstantOut.SweepTxHash, pkscript,
ctx, f.InstantOut.SweepTxHash, pkscript,
1, f.InstantOut.initiationHeight,
)
if err != nil {
@ -483,11 +489,13 @@ func (f *FSM) WaitForSweeplessSweepConfirmedAction(
// PublishHtlcAction publishes the htlc transaction and the htlc sweep
// transaction.
func (f *FSM) PublishHtlcAction(eventCtx fsm.EventContext) fsm.EventType {
func (f *FSM) PublishHtlcAction(ctx context.Context,
eventCtx fsm.EventContext) fsm.EventType {
// Publish the htlc transaction.
label := fmt.Sprintf("htlc-%v", f.InstantOut.swapPreimage.Hash())
err := f.cfg.Wallet.PublishTransaction(
f.ctx, f.InstantOut.finalizedHtlcTx,
fmt.Sprintf("htlc-%v", f.InstantOut.swapPreimage.Hash()),
ctx, f.InstantOut.finalizedHtlcTx, label,
)
if err != nil {
return f.HandleError(err)
@ -499,7 +507,7 @@ func (f *FSM) PublishHtlcAction(eventCtx fsm.EventContext) fsm.EventType {
// We'll now wait for the htlc to be confirmed.
confChan, confErrChan, err := f.cfg.ChainNotifier.
RegisterConfirmationsNtfn(
f.ctx, &txHash,
ctx, &txHash,
f.InstantOut.finalizedHtlcTx.TxOut[0].PkScript,
1, f.InstantOut.initiationHeight,
)
@ -518,21 +526,23 @@ func (f *FSM) PublishHtlcAction(eventCtx fsm.EventContext) fsm.EventType {
}
// PublishHtlcSweepAction publishes the htlc sweep transaction.
func (f *FSM) PublishHtlcSweepAction(eventCtx fsm.EventContext) fsm.EventType {
func (f *FSM) PublishHtlcSweepAction(ctx context.Context,
eventCtx fsm.EventContext) fsm.EventType {
// Create a feerate that will confirm the htlc quickly.
feeRate, err := f.cfg.Wallet.EstimateFeeRate(f.ctx, urgentConfTarget)
feeRate, err := f.cfg.Wallet.EstimateFeeRate(ctx, urgentConfTarget)
if err != nil {
return f.HandleError(err)
}
getInfo, err := f.cfg.LndClient.GetInfo(f.ctx)
getInfo, err := f.cfg.LndClient.GetInfo(ctx)
if err != nil {
return f.HandleError(err)
}
// We can immediately publish the htlc sweep transaction.
htlcSweepTx, err := f.InstantOut.generateHtlcSweepTx(
f.ctx, f.cfg.Signer, feeRate, f.cfg.Network, getInfo.BlockHeight,
ctx, f.cfg.Signer, feeRate, f.cfg.Network, getInfo.BlockHeight,
)
if err != nil {
return f.HandleError(err)
@ -540,7 +550,7 @@ func (f *FSM) PublishHtlcSweepAction(eventCtx fsm.EventContext) fsm.EventType {
label := fmt.Sprintf("htlc-sweep-%v", f.InstantOut.swapPreimage.Hash())
err = f.cfg.Wallet.PublishTransaction(f.ctx, htlcSweepTx, label)
err = f.cfg.Wallet.PublishTransaction(ctx, htlcSweepTx, label)
if err != nil {
log.Errorf("error publishing htlc sweep tx: %v", err)
return f.HandleError(err)
@ -555,7 +565,7 @@ func (f *FSM) PublishHtlcSweepAction(eventCtx fsm.EventContext) fsm.EventType {
// WaitForHtlcSweepConfirmedAction waits for the htlc sweep transaction to be
// confirmed.
func (f *FSM) WaitForHtlcSweepConfirmedAction(
func (f *FSM) WaitForHtlcSweepConfirmedAction(ctx context.Context,
eventCtx fsm.EventContext) fsm.EventType {
sweepPkScript, err := txscript.PayToAddrScript(
@ -566,7 +576,7 @@ func (f *FSM) WaitForHtlcSweepConfirmedAction(
}
confChan, confErrChan, err := f.cfg.ChainNotifier.RegisterConfirmationsNtfn(
f.ctx, f.InstantOut.SweepTxHash, sweepPkScript,
ctx, f.InstantOut.SweepTxHash, sweepPkScript,
1, f.InstantOut.initiationHeight,
)
if err != nil {
@ -592,10 +602,11 @@ func (f *FSM) WaitForHtlcSweepConfirmedAction(
// handleErrorAndUnlockReservations handles an error and unlocks the
// reservations.
func (f *FSM) handleErrorAndUnlockReservations(err error) fsm.EventType {
func (f *FSM) handleErrorAndUnlockReservations(ctx context.Context,
err 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(context.Background(), time.Second*30)
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
defer cancel()
// Unlock the reservations.
@ -613,7 +624,7 @@ func (f *FSM) handleErrorAndUnlockReservations(err error) fsm.EventType {
// 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(context.Background(), time.Second*30)
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
defer cancel()
_, cancelErr := f.cfg.InstantOutClient.CancelInstantSwap(
ctx, &swapserverrpc.CancelInstantSwapRequest{

19
instantout/fsm.go
View file

@ -176,8 +176,6 @@ type Config struct {
type FSM struct {
*fsm.StateMachine
ctx context.Context
// cfg contains all the services that the reservation manager needs to
// operate.
cfg *Config
@ -195,24 +193,19 @@ type FSM struct {
}
// NewFSM creates a new instant out FSM.
func NewFSM(ctx context.Context, cfg *Config,
protocolVersion ProtocolVersion) (*FSM, error) {
func NewFSM(cfg *Config, protocolVersion ProtocolVersion) (*FSM, error) {
instantOut := &InstantOut{
State: fsm.EmptyState,
protocolVersion: protocolVersion,
}
return NewFSMFromInstantOut(ctx, cfg, instantOut)
return NewFSMFromInstantOut(cfg, instantOut)
}
// NewFSMFromInstantOut creates a new instantout FSM from an existing instantout
// recovered from the database.
func NewFSMFromInstantOut(ctx context.Context, cfg *Config,
instantOut *InstantOut) (*FSM, error) {
func NewFSMFromInstantOut(cfg *Config, instantOut *InstantOut) (*FSM, error) {
instantOutFSM := &FSM{
ctx: ctx,
cfg: cfg,
InstantOut: instantOut,
}
@ -328,7 +321,9 @@ func (f *FSM) GetV1ReservationStates() fsm.States {
// updateInstantOut is called after every action and updates the reservation
// in the db.
func (f *FSM) updateInstantOut(notification fsm.Notification) {
func (f *FSM) updateInstantOut(ctx context.Context,
notification fsm.Notification) {
f.Infof("Previous: %v, Event: %v, Next: %v", notification.PreviousState,
notification.Event, notification.NextState)
@ -349,7 +344,7 @@ func (f *FSM) updateInstantOut(notification fsm.Notification) {
return
}
err := f.cfg.Store.UpdateInstantLoopOut(f.ctx, f.InstantOut)
err := f.cfg.Store.UpdateInstantLoopOut(ctx, f.InstantOut)
if err != nil {
log.Errorf("Error updating instant out: %v", err)
return

10
instantout/manager.go
View file

@ -111,7 +111,7 @@ func (m *Manager) recoverInstantOuts(ctx context.Context) error {
log.Debugf("Recovering instantout %v", instantOut.SwapHash)
instantOutFSM, err := NewFSMFromInstantOut(
ctx, m.cfg, instantOut,
m.cfg, instantOut,
)
if err != nil {
return err
@ -122,7 +122,7 @@ func (m *Manager) recoverInstantOuts(ctx context.Context) error {
// As SendEvent can block, we'll start a goroutine to process
// the event.
go func() {
err := instantOutFSM.SendEvent(OnRecover, nil)
err := instantOutFSM.SendEvent(ctx, OnRecover, nil)
if err != nil {
log.Errorf("FSM %v Error sending recover "+
"event %v, state: %v",
@ -162,9 +162,7 @@ func (m *Manager) NewInstantOut(ctx context.Context,
sweepAddress: sweepAddr,
}
instantOut, err := NewFSM(
m.runCtx, m.cfg, ProtocolVersionFullReservation,
)
instantOut, err := NewFSM(m.cfg, ProtocolVersionFullReservation)
if err != nil {
m.Unlock()
return nil, err
@ -174,7 +172,7 @@ func (m *Manager) NewInstantOut(ctx context.Context,
// Start the instantout FSM.
go func() {
err := instantOut.SendEvent(OnStart, request)
err := instantOut.SendEvent(m.runCtx, OnStart, request)
if err != nil {
log.Errorf("Error sending event: %v", err)
}

34
instantout/reservation/actions.go
View file

@ -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)
}

16
instantout/reservation/actions_test.go
View file

@ -146,7 +146,6 @@ func TestInitReservationAction(t *testing.T) {
).Return(tc.mockStoreErr)
reservationFSM := &FSM{
ctx: ctxb,
cfg: &Config{
Wallet: mockLnd.WalletKit,
ChainNotifier: mockLnd.ChainNotifier,
@ -156,7 +155,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)
}
}
@ -236,10 +235,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{
@ -305,7 +304,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)
@ -344,10 +343,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{
@ -370,7 +370,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)
})
@ -424,7 +424,7 @@ func TestHandleSubcriptions(t *testing.T) {
// Create the FSM.
r := NewFSMFromReservation(
context.Background(), &Config{
&Config{
ChainNotifier: chainNotifier,
},
&Reservation{

17
instantout/reservation/fsm.go
View file

@ -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)
}

26
instantout/reservation/manager.go
View file

@ -22,8 +22,6 @@ type Manager struct {
// activeReservations contains all the active reservationsFSMs.
activeReservations map[ID]*FSM
runCtx context.Context
sync.Mutex
}
@ -36,13 +34,14 @@ func NewManager(cfg *Config) *Manager {
}
// Run runs the reservation manager.
func (m *Manager) Run(ctx context.Context, height int32) error {
func (m *Manager) Run(ctx context.Context, height int32,
initChan chan struct{}) error {
log.Debugf("Starting reservation manager")
runCtx, cancel := context.WithCancel(ctx)
defer cancel()
m.runCtx = runCtx
currentHeight := height
err := m.RecoverReservations(runCtx)
@ -58,6 +57,9 @@ func (m *Manager) Run(ctx context.Context, height int32) error {
ntfnChan := m.cfg.NotificationManager.SubscribeReservations(runCtx)
// Signal that the manager has been initialized.
close(initChan)
for {
select {
case height := <-newBlockChan:
@ -111,9 +113,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 +130,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 +171,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 +215,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 +235,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.

6
instantout/reservation/manager_test.go
View file

@ -25,12 +25,16 @@ func TestManager(t *testing.T) {
testContext := newManagerTestContext(t)
initChan := make(chan struct{})
// Start the manager.
go func() {
err := testContext.manager.Run(ctxb, testContext.mockLnd.Height)
err := testContext.manager.Run(ctxb, testContext.mockLnd.Height, initChan)
require.NoError(t, err)
}()
// We'll now wait for the manager to be initialized.
<-initChan
// Create a new reservation.
reservationFSM, err := testContext.manager.newReservation(
ctxb, uint32(testContext.mockLnd.Height),