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

staticaddr: enable deposits for swaps

Browse source 
The static address state machine adds states
to reflect the state of a deposit during a
loop-in swap.
This commit is contained in:
Slyghtning 2024-07-30 15:37:37 +02:00
parent 3894370302
commit c11e90138b
No known key found for this signature in database
GPG key ID: F82D456EA023C9BF
6 changed files with 356 additions and 160 deletions
Download patch file Download diff file Expand all files Collapse all files

6
loopd/swapclient_server.go
View file

@ -1471,7 +1471,7 @@ func (s *swapClientServer) GetStaticAddressSummary(ctx context.Context,
"outpoints")
}
allDeposits, err := s.depositManager.GetAllDeposits()
allDeposits, err := s.depositManager.GetAllDeposits(ctx)
if err != nil {
return nil, err
}
@ -1609,7 +1609,7 @@ func toClientState(state fsm.StateType) looprpc.DepositState {
case deposit.Withdrawn:
return looprpc.DepositState_WITHDRAWN
case deposit.PublishExpiredDeposit:
case deposit.PublishExpirySweep:
return looprpc.DepositState_PUBLISH_EXPIRED
case deposit.WaitForExpirySweep:
@ -1635,7 +1635,7 @@ func toServerState(state looprpc.DepositState) fsm.StateType {
return deposit.Withdrawn
case looprpc.DepositState_PUBLISH_EXPIRED:
return deposit.PublishExpiredDeposit
return deposit.PublishExpirySweep
case looprpc.DepositState_WAIT_FOR_EXPIRY_SWEEP:
return deposit.WaitForExpirySweep

14
staticaddr/deposit/actions.go
View file

@ -14,7 +14,7 @@ import (
)
const (
defaultConfTarget = 3
DefaultConfTarget = 3
)
// PublishDepositExpirySweepAction creates and publishes the timeout transaction
@ -39,11 +39,11 @@ func (f *FSM) PublishDepositExpirySweepAction(ctx context.Context,
// Estimate the fee rate of an expiry spend transaction.
feeRateEstimator, err := f.cfg.WalletKit.EstimateFeeRate(
ctx, defaultConfTarget,
ctx, DefaultConfTarget,
)
if err != nil {
return f.HandleError(fmt.Errorf("timeout sweep fee "+
"estimation failed: %v", err))
"estimation failed: %w", err))
}
weight := script.ExpirySpendWeight()
@ -116,7 +116,7 @@ func (f *FSM) WaitForExpirySweepAction(ctx context.Context,
_ fsm.EventContext) fsm.EventType {
spendChan, errSpendChan, err := f.cfg.ChainNotifier.RegisterConfirmationsNtfn( //nolint:lll
ctx, nil, f.deposit.TimeOutSweepPkScript, defaultConfTarget,
ctx, nil, f.deposit.TimeOutSweepPkScript, DefaultConfTarget,
int32(f.deposit.ConfirmationHeight),
)
if err != nil {
@ -124,7 +124,7 @@ func (f *FSM) WaitForExpirySweepAction(ctx context.Context,
}
select {
case err := <-errSpendChan:
case err = <-errSpendChan:
log.Debugf("error while sweeping expired deposit: %v", err)
return fsm.OnError
@ -155,9 +155,9 @@ func (f *FSM) SweptExpiredDepositAction(ctx context.Context,
return fsm.NoOp
}
// WithdrawnDepositAction is the final action after a withdrawal. It signals to
// FinalizeDepositAction is the final action after a withdrawal. It signals to
// the manager that the deposit has been swept and the FSM can be removed.
func (f *FSM) WithdrawnDepositAction(ctx context.Context,
func (f *FSM) FinalizeDepositAction(ctx context.Context,
_ fsm.EventContext) fsm.EventType {
select {

19
staticaddr/deposit/deposit.go
View file

@ -59,20 +59,13 @@ type Deposit struct {
sync.Mutex
}
// IsInPendingState returns true if the deposit is pending.
func (d *Deposit) IsInPendingState() bool {
d.Lock()
defer d.Unlock()
return !d.IsInFinalState()
}
// IsInFinalState returns true if the deposit is final.
func (d *Deposit) IsInFinalState() bool {
d.Lock()
defer d.Unlock()
return d.state == Expired || d.state == Withdrawn || d.state == Failed
return d.state == Expired || d.state == Withdrawn ||
d.state == LoopedIn || d.state == HtlcTimeoutSwept
}
func (d *Deposit) IsExpired(currentHeight, expiry uint32) bool {
@ -96,6 +89,10 @@ func (d *Deposit) SetState(state fsm.StateType) {
d.state = state
}
func (d *Deposit) SetStateNoLock(state fsm.StateType) {
d.state = state
}
func (d *Deposit) IsInState(state fsm.StateType) bool {
d.Lock()
defer d.Unlock()
@ -103,6 +100,10 @@ func (d *Deposit) IsInState(state fsm.StateType) bool {
return d.state == state
}
func (d *Deposit) IsInStateNoLock(state fsm.StateType) bool {
return d.state == state
}
// GetRandomDepositID generates a random deposit ID.
func GetRandomDepositID() (ID, error) {
var id ID

209
staticaddr/deposit/fsm.go
View file

@ -23,34 +23,108 @@ const (
var (
ErrProtocolVersionNotSupported = errors.New("protocol version not " +
"supported")
// Withdrawal and loop-in transitions lock their respective deposits
// themselves. We need to make sure that we don't lock the deposit
// twice. For the events below we expect the deposits already locked.
lockedEvents = map[fsm.EventType]struct{}{
OnLoopInInitiated: {},
OnSweepingHtlcTimeout: {},
OnHtlcTimeoutSwept: {},
OnLoopedIn: {},
fsm.OnError: {},
OnWithdrawInitiated: {},
OnWithdrawn: {},
}
)
// States.
var (
// Deposited signals that funds at a static address have reached the
// confirmation height.
Deposited = fsm.StateType("Deposited")
// Withdrawing signals that the withdrawal transaction has been
// broadcast, awaiting sufficient confirmations.
Withdrawing = fsm.StateType("Withdrawing")
// Withdrawn signals that the withdrawal transaction has been confirmed.
Withdrawn = fsm.StateType("Withdrawn")
PublishExpiredDeposit = fsm.StateType("PublishExpiredDeposit")
// LoopingIn signals that the deposit is locked for a loop in swap.
LoopingIn = fsm.StateType("LoopingIn")
// LoopedIn signals that the loop in swap has been successfully
// completed. It implies that we signed the sweepless sweep tx for the
// server.
LoopedIn = fsm.StateType("LoopedIn")
// SweepHtlcTimeout signals that the htlc timeout path is in the
// process of being swept.
SweepHtlcTimeout = fsm.StateType("SweepHtlcTimeout")
// HtlcTimeoutSwept signals that the htlc timeout path has been swept.
HtlcTimeoutSwept = fsm.StateType("HtlcTimeoutSwept")
// PublishExpirySweep signals that the deposit has expired, and we are
// in the process of publishing the expiry sweep transaction.
PublishExpirySweep = fsm.StateType("PublishExpirySweep")
// WaitForExpirySweep signals that the expiry sweep transaction has been
// published, and we are waiting for it to be confirmed.
WaitForExpirySweep = fsm.StateType("WaitForExpirySweep")
// Expired signals that the deposit has expired and the expiry sweep
// transaction has been confirmed sufficiently.
Expired = fsm.StateType("Expired")
Failed = fsm.StateType("Failed")
)
// Events.
var (
OnStart = fsm.EventType("OnStart")
// OnStart is sent to the fsm once the deposit outpoint has been
// sufficiently confirmed. It transitions the fsm into the Deposited
// state from where we can trigger a withdrawal, a loopin or an expiry.
OnStart = fsm.EventType("OnStart")
// OnWithdrawInitiated is sent to the fsm when a withdrawal has been
// initiated.
OnWithdrawInitiated = fsm.EventType("OnWithdrawInitiated")
OnWithdrawn = fsm.EventType("OnWithdrawn")
OnExpiry = fsm.EventType("OnExpiry")
OnExpiryPublished = fsm.EventType("OnExpiryPublished")
OnExpirySwept = fsm.EventType("OnExpirySwept")
OnRecover = fsm.EventType("OnRecover")
// OnWithdrawn is sent to the fsm when a withdrawal has been confirmed.
OnWithdrawn = fsm.EventType("OnWithdrawn")
// OnLoopInInitiated is sent to the fsm when a loop in has been
// initiated.
OnLoopInInitiated = fsm.EventType("OnLoopInInitiated")
// OnSweepingHtlcTimeout is sent to the fsm when the htlc timeout path
// is being swept. This indicates that the server didn't pay the swap
// invoice, but the htlc tx was published, from we which we need to
// sweep the htlc timeout path.
OnSweepingHtlcTimeout = fsm.EventType("OnSweepingHtlcTimeout")
// OnHtlcTimeoutSwept is sent to the fsm when the htlc timeout path has
// been swept.
OnHtlcTimeoutSwept = fsm.EventType("OnHtlcTimeoutSwept")
// OnLoopedIn is sent to the fsm when the user intents to use the
// deposit for a loop in swap.
OnLoopedIn = fsm.EventType("OnLoopedIn")
// OnExpiry is sent to the fsm when the deposit has expired.
OnExpiry = fsm.EventType("OnExpiry")
// OnExpiryPublished is sent to the fsm when the expiry sweep tx has
// been published.
OnExpiryPublished = fsm.EventType("OnExpiryPublished")
// OnExpirySwept is sent to the fsm when the expiry sweep tx has been
// confirmed.
OnExpirySwept = fsm.EventType("OnExpirySwept")
// OnRecover is sent to the fsm when it should recover from client
// restart.
OnRecover = fsm.EventType("OnRecover")
)
// FSM is the state machine that handles the instant out.
@ -79,12 +153,12 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
params, err := cfg.AddressManager.GetStaticAddressParameters(ctx)
if err != nil {
return nil, fmt.Errorf("unable to get static address "+
"parameters: %v", err)
"parameters: %w", err)
}
address, err := cfg.AddressManager.GetStaticAddress(ctx)
if err != nil {
return nil, fmt.Errorf("unable to get static address: %v", err)
return nil, fmt.Errorf("unable to get static address: %w", err)
}
depoFsm := &FSM{
@ -150,7 +224,7 @@ func (f *FSM) handleBlockNotification(ctx context.Context,
err := f.SendEvent(ctx, OnExpiry, nil)
if err != nil {
log.Debugf("error sending OnExpiry "+
"event: %v", err)
"event: %w", err)
}
}()
}
@ -168,15 +242,22 @@ func (f *FSM) DepositStatesV0() fsm.States {
},
Deposited: fsm.State{
Transitions: fsm.Transitions{
OnExpiry: PublishExpiredDeposit,
OnExpiry: PublishExpirySweep,
OnWithdrawInitiated: Withdrawing,
OnRecover: Deposited,
OnLoopInInitiated: LoopingIn,
// We encounter OnSweepingHtlcTimeout if the
// server published the htlc tx without paying
// us. We then need to monitor for the timeout
// path to open up to sweep it.
OnSweepingHtlcTimeout: SweepHtlcTimeout,
OnRecover: Deposited,
fsm.OnError: Deposited,
},
Action: fsm.NoOpAction,
},
PublishExpiredDeposit: fsm.State{
PublishExpirySweep: fsm.State{
Transitions: fsm.Transitions{
OnRecover: PublishExpiredDeposit,
OnRecover: PublishExpirySweep,
OnExpiryPublished: WaitForExpirySweep,
// If the timeout sweep failed we go back to
// Deposited, hoping that another timeout sweep
@ -190,7 +271,7 @@ func (f *FSM) DepositStatesV0() fsm.States {
Transitions: fsm.Transitions{
OnExpirySwept: Expired,
// Upon recovery, we republish the sweep tx.
OnRecover: PublishExpiredDeposit,
OnRecover: PublishExpirySweep,
// If the timeout sweep failed we go back to
// Deposited, hoping that another timeout sweep
// attempt will be successful. Alternatively,
@ -229,17 +310,49 @@ func (f *FSM) DepositStatesV0() fsm.States {
},
Action: fsm.NoOpAction,
},
LoopingIn: fsm.State{
Transitions: fsm.Transitions{
// This event is triggered when the loop in
// payment has been received. We consider the
// swap to be completed and transition to a
// final state.
OnLoopedIn: LoopedIn,
// If the deposit expires while the loop in is
// still pending, we publish the expiry sweep.
OnExpiry: PublishExpirySweep,
OnLoopInInitiated: LoopingIn,
OnRecover: LoopingIn,
fsm.OnError: Deposited,
},
Action: fsm.NoOpAction,
},
LoopedIn: fsm.State{
Transitions: fsm.Transitions{
OnExpiry: Expired,
},
Action: f.FinalizeDepositAction,
},
SweepHtlcTimeout: fsm.State{
Transitions: fsm.Transitions{
OnHtlcTimeoutSwept: HtlcTimeoutSwept,
OnRecover: SweepHtlcTimeout,
},
Action: fsm.NoOpAction,
},
HtlcTimeoutSwept: fsm.State{
Transitions: fsm.Transitions{
OnExpiry: HtlcTimeoutSwept,
},
Action: f.FinalizeDepositAction,
},
Withdrawn: fsm.State{
Transitions: fsm.Transitions{
OnExpiry: Expired,
},
Action: f.WithdrawnDepositAction,
},
Failed: fsm.State{
Transitions: fsm.Transitions{
OnExpiry: Failed,
},
Action: fsm.NoOpAction,
Action: f.FinalizeDepositAction,
},
}
}
@ -258,25 +371,53 @@ func (f *FSM) updateDeposit(ctx context.Context,
notification.Event,
)
f.deposit.SetState(notification.NextState)
// Don't update the deposit if we are in an initial state or if we
// are transitioning from an initial state to a failed state.
d := f.deposit
if d.IsInState(fsm.EmptyState) || d.IsInState(Deposited) ||
(notification.PreviousState == Deposited && d.IsInState(
Failed,
)) {
type checkStateFunc func(state fsm.StateType) bool
type setStateFunc func(state fsm.StateType)
checkFunc := checkStateFunc(f.deposit.IsInState)
setFunc := setStateFunc(f.deposit.SetState)
if _, ok := lockedEvents[notification.Event]; ok {
checkFunc = f.deposit.IsInStateNoLock
setFunc = f.deposit.SetStateNoLock
}
setFunc(notification.NextState)
if isUpdateSkipped(notification, checkFunc) {
return
}
err := f.cfg.Store.UpdateDeposit(ctx, f.deposit)
if err != nil {
f.Errorf("unable to update deposit: %v", err)
f.Errorf("unable to update deposit: %w", err)
}
}
// isUpdateSkipped returns true if the deposit should not be updated for the given
// notification.
func isUpdateSkipped(notification fsm.Notification,
checkStateFunc func(stateType fsm.StateType) bool) bool {
prevState := notification.PreviousState
// Skip if we are in the empty state because no deposit has been
// persisted yet.
if checkStateFunc(fsm.EmptyState) {
return true
}
// If we transitioned from the empty state to Deposited there's still no
// deposit persisted, so we don't need to update it.
if prevState == fsm.EmptyState && checkStateFunc(Deposited) {
return true
}
// We don't update in self-loops, e.g. in the case of recovery.
if checkStateFunc(prevState) {
return true
}
return false
}
// Infof logs an info message with the deposit outpoint.
func (f *FSM) Infof(format string, args ...interface{}) {
log.Infof(

261
staticaddr/deposit/manager.go
View file

@ -34,7 +34,7 @@ const (
// DefaultTransitionTimeout is the default timeout for transitions in
// the deposit state machine.
DefaultTransitionTimeout = 1 * time.Minute
DefaultTransitionTimeout = 5 * time.Second
)
// ManagerConfig holds the configuration for the address manager.
@ -74,9 +74,8 @@ type ManagerConfig struct {
type Manager struct {
cfg *ManagerConfig
runCtx context.Context
sync.Mutex
// mu guards access to activeDeposits map.
mu sync.Mutex
// initChan signals the daemon that the address manager has completed
// its initialization.
@ -88,9 +87,6 @@ type Manager struct {
// initiationHeight stores the currently best known block height.
initiationHeight uint32
// currentHeight stores the currently best known block height.
currentHeight uint32
// deposits contains all the deposits that have ever been made to the
// static address. This field is used to store and recover deposits. It
// also serves as basis for reconciliation of newly detected deposits by
@ -116,25 +112,21 @@ func NewManager(cfg *ManagerConfig) *Manager {
// Run runs the address manager.
func (m *Manager) Run(ctx context.Context, currentHeight uint32) error {
m.runCtx = ctx
m.initiationHeight = currentHeight
m.Lock()
m.currentHeight, m.initiationHeight = currentHeight, currentHeight
m.Unlock()
newBlockChan, newBlockErrChan, err := m.cfg.ChainNotifier.RegisterBlockEpochNtfn(m.runCtx) //nolint:lll
newBlockChan, newBlockErrChan, err := m.cfg.ChainNotifier.RegisterBlockEpochNtfn(ctx) //nolint:lll
if err != nil {
return err
}
// Recover previous deposits and static address parameters from the DB.
err = m.recover(m.runCtx)
err = m.recoverDeposits(ctx)
if err != nil {
return err
}
// Start the deposit notifier.
m.pollDeposits(m.runCtx)
m.pollDeposits(ctx)
// Communicate to the caller that the address manager has completed its
// initialization.
@ -143,37 +135,33 @@ func (m *Manager) Run(ctx context.Context, currentHeight uint32) error {
for {
select {
case height := <-newBlockChan:
m.Lock()
m.currentHeight = uint32(height)
m.Unlock()
// Inform all active deposits about a new block arrival.
for _, fsm := range m.activeDeposits {
select {
case fsm.blockNtfnChan <- uint32(height):
case <-m.runCtx.Done():
return m.runCtx.Err()
case <-ctx.Done():
return ctx.Err()
}
}
case outpoint := <-m.finalizedDepositChan:
// If deposits notify us about their finalization, we
// update the manager's internal state and flush the
// finalized deposit from memory.
m.finalizeDeposit(outpoint)
delete(m.activeDeposits, outpoint)
case err := <-newBlockErrChan:
case err = <-newBlockErrChan:
return err
case <-m.runCtx.Done():
return m.runCtx.Err()
case <-ctx.Done():
return ctx.Err()
}
}
}
// recover recovers static address parameters, previous deposits and state
// machines from the database and starts the deposit notifier.
func (m *Manager) recover(ctx context.Context) error {
// recoverDeposits recovers static address parameters, previous deposits and
// state machines from the database and starts the deposit notifier.
func (m *Manager) recoverDeposits(ctx context.Context) error {
log.Infof("Recovering static address parameters and deposits...")
// Recover deposits.
@ -195,10 +183,7 @@ func (m *Manager) recover(ctx context.Context) error {
log.Debugf("Recovering deposit %x", d.ID)
// Create a state machine for a given deposit.
fsm, err := NewFSM(
m.runCtx, d, m.cfg,
m.finalizedDepositChan, true,
)
fsm, err := NewFSM(ctx, d, m.cfg, m.finalizedDepositChan, true)
if err != nil {
return err
}
@ -259,12 +244,12 @@ func (m *Manager) reconcileDeposits(ctx context.Context) error {
ctx, MinConfs, MaxConfs,
)
if err != nil {
return fmt.Errorf("unable to list new deposits: %v", err)
return fmt.Errorf("unable to list new deposits: %w", err)
}
newDeposits := m.filterNewDeposits(utxos)
if err != nil {
return fmt.Errorf("unable to filter new deposits: %v", err)
return fmt.Errorf("unable to filter new deposits: %w", err)
}
if len(newDeposits) == 0 {
@ -275,14 +260,14 @@ func (m *Manager) reconcileDeposits(ctx context.Context) error {
for _, utxo := range newDeposits {
deposit, err := m.createNewDeposit(ctx, utxo)
if err != nil {
return fmt.Errorf("unable to retain new deposit: %v",
return fmt.Errorf("unable to retain new deposit: %w",
err)
}
log.Debugf("Received deposit: %v", deposit)
err = m.startDepositFsm(deposit)
err = m.startDepositFsm(ctx, deposit)
if err != nil {
return fmt.Errorf("unable to start new deposit FSM: %v",
return fmt.Errorf("unable to start new deposit FSM: %w",
err)
}
}
@ -332,9 +317,9 @@ func (m *Manager) createNewDeposit(ctx context.Context,
return nil, err
}
m.Lock()
m.mu.Lock()
m.deposits[deposit.OutPoint] = deposit
m.Unlock()
m.mu.Unlock()
return deposit, nil
}
@ -348,7 +333,7 @@ func (m *Manager) getBlockHeight(ctx context.Context,
)
if err != nil {
return 0, fmt.Errorf("couldn't get confirmation height for "+
"deposit, %v", err)
"deposit, %w", err)
}
notifChan, errChan, err := m.cfg.ChainNotifier.RegisterConfirmationsNtfn( //nolint:lll
@ -374,8 +359,8 @@ func (m *Manager) getBlockHeight(ctx context.Context,
// filterNewDeposits filters the given utxos for new deposits that we haven't
// seen before.
func (m *Manager) filterNewDeposits(utxos []*lnwallet.Utxo) []*lnwallet.Utxo {
m.Lock()
defer m.Unlock()
m.mu.Lock()
defer m.mu.Unlock()
var newDeposits []*lnwallet.Utxo
for _, utxo := range utxos {
@ -390,115 +375,152 @@ func (m *Manager) filterNewDeposits(utxos []*lnwallet.Utxo) []*lnwallet.Utxo {
// startDepositFsm creates a new state machine flow from the latest deposit to
// our static address.
func (m *Manager) startDepositFsm(deposit *Deposit) error {
func (m *Manager) startDepositFsm(ctx context.Context, deposit *Deposit) error {
// Create a state machine for a given deposit.
fsm, err := NewFSM(
m.runCtx, deposit, m.cfg, m.finalizedDepositChan, false,
)
fsm, err := NewFSM(ctx, deposit, m.cfg, m.finalizedDepositChan, false)
if err != nil {
return err
}
// Send the start event to the state machine.
go func() {
err = fsm.SendEvent(m.runCtx, OnStart, nil)
err = fsm.SendEvent(ctx, OnStart, nil)
if err != nil {
log.Errorf("Error sending OnStart event: %v", err)
}
}()
err = fsm.DefaultObserver.WaitForState(m.runCtx, time.Minute, Deposited)
err = fsm.DefaultObserver.WaitForState(ctx, time.Minute, Deposited)
if err != nil {
return err
}
// Add the FSM to the active FSMs map.
m.Lock()
m.mu.Lock()
m.activeDeposits[deposit.OutPoint] = fsm
m.Unlock()
m.mu.Unlock()
return nil
}
func (m *Manager) finalizeDeposit(outpoint wire.OutPoint) {
m.Lock()
delete(m.activeDeposits, outpoint)
delete(m.deposits, outpoint)
m.Unlock()
}
// GetActiveDepositsInState returns all active deposits.
// GetActiveDepositsInState returns all active deposits. This function is called
// on a client restart before the manager is fully initialized, hence we don't
// have to lock the deposits.
func (m *Manager) GetActiveDepositsInState(stateFilter fsm.StateType) (
[]*Deposit, error) {
m.Lock()
defer m.Unlock()
m.mu.Lock()
defer m.mu.Unlock()
var deposits []*Deposit
for _, fsm := range m.activeDeposits {
if fsm.deposit.GetState() != stateFilter {
continue
}
deposits = append(deposits, fsm.deposit)
}
sort.Slice(deposits, func(i, j int) bool {
return deposits[i].ConfirmationHeight <
deposits[j].ConfirmationHeight
lockDeposits(deposits)
defer unlockDeposits(deposits)
filteredDeposits := make([]*Deposit, 0, len(deposits))
for _, d := range deposits {
if !d.IsInStateNoLock(stateFilter) {
continue
}
filteredDeposits = append(filteredDeposits, d)
}
sort.Slice(filteredDeposits, func(i, j int) bool {
return filteredDeposits[i].ConfirmationHeight <
filteredDeposits[j].ConfirmationHeight
})
return deposits, nil
}
// GetAllDeposits returns all active deposits.
func (m *Manager) GetAllDeposits() ([]*Deposit, error) {
return m.cfg.Store.AllDeposits(m.runCtx)
return filteredDeposits, nil
}
// AllOutpointsActiveDeposits checks if all deposits referenced by the outpoints
// are active and in the specified state.
// are in our in-mem active deposits map and in the specified state. If
// fsm.EmptyState is set as targetState all deposits are returned regardless of
// their state. Each existent deposit is locked during the check.
func (m *Manager) AllOutpointsActiveDeposits(outpoints []wire.OutPoint,
stateFilter fsm.StateType) ([]*Deposit, bool) {
targetState fsm.StateType) ([]*Deposit, bool) {
m.Lock()
defer m.Unlock()
m.mu.Lock()
defer m.mu.Unlock()
deposits := make([]*Deposit, 0, len(outpoints))
for _, o := range outpoints {
if _, ok := m.activeDeposits[o]; !ok {
_, deposits := m.toActiveDeposits(&outpoints)
if deposits == nil {
return nil, false
}
// If the targetState is empty we return all active deposits regardless
// of state.
if targetState == fsm.EmptyState {
return deposits, true
}
lockDeposits(deposits)
defer unlockDeposits(deposits)
for _, d := range deposits {
if !d.IsInStateNoLock(targetState) {
return nil, false
}
deposit := m.deposits[o]
if deposit.GetState() != stateFilter {
return nil, false
}
deposits = append(deposits, m.deposits[o])
}
return deposits, true
}
// TransitionDeposits allows a caller to transition a set of deposits to a new
// state.
func (m *Manager) TransitionDeposits(deposits []*Deposit, event fsm.EventType,
expectedFinalState fsm.StateType) error {
// AllStringOutpointsActiveDeposits converts outpoint strings of format txid:idx
// to wire outpoints and checks if all deposits referenced by the outpoints are
// active and in the specified state. If fsm.EmptyState is referenced as
// stateFilter all deposits are returned regardless of their state.
func (m *Manager) AllStringOutpointsActiveDeposits(outpoints []string,
stateFilter fsm.StateType) ([]*Deposit, bool) {
for _, d := range deposits {
m.Lock()
sm, ok := m.activeDeposits[d.OutPoint]
m.Unlock()
if !ok {
return fmt.Errorf("deposit not found")
outPoints := make([]wire.OutPoint, len(outpoints))
for i, o := range outpoints {
op, err := wire.NewOutPointFromString(o)
if err != nil {
return nil, false
}
err := sm.SendEvent(m.runCtx, event, nil)
outPoints[i] = *op
}
return m.AllOutpointsActiveDeposits(outPoints, stateFilter)
}
// TransitionDeposits allows a caller to transition a set of deposits to a new
// state.
// Caveat: The action triggered by the state transitions should not compute
// heavy things or call external endpoints that can block for a long time.
// Deposits will be released if a transition takes longer than
// DefaultTransitionTimeout which is set to 5 seconds.
func (m *Manager) TransitionDeposits(ctx context.Context, deposits []*Deposit,
event fsm.EventType, expectedFinalState fsm.StateType) error {
outpoints := make([]wire.OutPoint, len(deposits))
for i, d := range deposits {
outpoints[i] = d.OutPoint
}
m.mu.Lock()
defer m.mu.Unlock()
stateMachines, _ := m.toActiveDeposits(&outpoints)
if stateMachines == nil {
return fmt.Errorf("deposits not found in active deposits")
}
lockDeposits(deposits)
defer unlockDeposits(deposits)
for _, sm := range stateMachines {
err := sm.SendEvent(ctx, event, nil)
if err != nil {
return err
}
err = sm.DefaultObserver.WaitForState(
m.runCtx, DefaultTransitionTimeout, expectedFinalState,
ctx, DefaultTransitionTimeout, expectedFinalState,
)
if err != nil {
return err
@ -508,7 +530,44 @@ func (m *Manager) TransitionDeposits(deposits []*Deposit, event fsm.EventType,
return nil
}
// UpdateDeposit overrides all fields of the deposit with given ID in the store.
func (m *Manager) UpdateDeposit(d *Deposit) error {
return m.cfg.Store.UpdateDeposit(m.runCtx, d)
func lockDeposits(deposits []*Deposit) {
for _, d := range deposits {
d.Lock()
}
}
func unlockDeposits(deposits []*Deposit) {
for _, d := range deposits {
d.Unlock()
}
}
// GetAllDeposits returns all active deposits.
func (m *Manager) GetAllDeposits(ctx context.Context) ([]*Deposit, error) {
return m.cfg.Store.AllDeposits(ctx)
}
// UpdateDeposit overrides all fields of the deposit with given ID in the store.
func (m *Manager) UpdateDeposit(ctx context.Context, d *Deposit) error {
return m.cfg.Store.UpdateDeposit(ctx, d)
}
// toActiveDeposits converts a list of outpoints to a list of FSMs and deposits.
// The caller should call mu.Lock() before calling this function.
func (m *Manager) toActiveDeposits(outpoints *[]wire.OutPoint) ([]*FSM,
[]*Deposit) {
fsms := make([]*FSM, 0, len(*outpoints))
deposits := make([]*Deposit, 0, len(*outpoints))
for _, o := range *outpoints {
sm, ok := m.activeDeposits[o]
if !ok {
return nil, nil
}
fsms = append(fsms, sm)
deposits = append(deposits, m.deposits[o])
}
return fsms, deposits
}

7
staticaddr/deposit/sql_store.go
View file

@ -65,7 +65,7 @@ func (s *SqlStore) UpdateDeposit(ctx context.Context, deposit *Deposit) error {
insertUpdateArgs := sqlc.InsertDepositUpdateParams{
DepositID: deposit.ID[:],
UpdateTimestamp: s.clock.Now().UTC(),
UpdateState: string(deposit.GetState()),
UpdateState: string(deposit.state),
}
var (
@ -242,8 +242,3 @@ func (s *SqlStore) toDeposit(row sqlc.Deposit,
FinalizedWithdrawalTx: finalizedWithdrawalTx,
}, nil
}
// Close closes the database connection.
func (s *SqlStore) Close() {
s.baseDB.DB.Close()
}