diff --git a/staticaddr/deposit/fsm.go b/staticaddr/deposit/fsm.go index 01a7edf8..b4c89ec8 100644 --- a/staticaddr/deposit/fsm.go +++ b/staticaddr/deposit/fsm.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "github.com/btcsuite/btcd/txscript" "github.com/btcsuite/btcd/wire" @@ -160,6 +161,12 @@ type FSM struct { blockNtfnChan chan uint32 + // stopChan requests shutdown of the block notification loop. + stopChan chan struct{} + + // stopOnce ensures Stop is idempotent. + stopOnce sync.Once + // quitChan stops after the FSM stops consuming blockNtfnChan. quitChan chan struct{} @@ -191,6 +198,7 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig, params: params, address: address, blockNtfnChan: make(chan uint32), + stopChan: make(chan struct{}), quitChan: make(chan struct{}), finalizedDepositChan: finalizedDepositChan, } @@ -226,6 +234,9 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig, ctx, currentHeight, ) + case <-fsm.stopChan: + return + case <-ctx.Done(): return } @@ -235,6 +246,17 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig, return depoFsm, nil } +// Stop requests shutdown of the FSM's block notification loop. +func (f *FSM) Stop() { + if f == nil || f.stopChan == nil { + return + } + + f.stopOnce.Do(func() { + close(f.stopChan) + }) +} + // handleBlockNotification inspects the current block height and sends the // OnExpiry event to publish the expiry sweep transaction if the deposit timed // out, or it republishes the expiry sweep transaction if it was not yet swept. diff --git a/staticaddr/deposit/manager.go b/staticaddr/deposit/manager.go index aaadc798..4da45ada 100644 --- a/staticaddr/deposit/manager.go +++ b/staticaddr/deposit/manager.go @@ -146,9 +146,7 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { case outpoint := <-m.finalizedDepositChan: // If deposits notify us about their finalization, flush // the finalized deposit from memory. - m.mu.Lock() - delete(m.activeDeposits, outpoint) - m.mu.Unlock() + m.removeActiveDeposit(outpoint) case err = <-newBlockErrChan: return err @@ -544,6 +542,20 @@ func unlockDeposits(deposits []*Deposit) { } } +// removeActiveDeposit removes and stops the FSM for an active outpoint. +func (m *Manager) removeActiveDeposit(outpoint wire.OutPoint) { + m.mu.Lock() + fsm, ok := m.activeDeposits[outpoint] + if ok { + delete(m.activeDeposits, outpoint) + } + m.mu.Unlock() + + if ok { + fsm.Stop() + } +} + // GetAllDeposits returns all active deposits. func (m *Manager) GetAllDeposits(ctx context.Context) ([]*Deposit, error) { return m.cfg.Store.AllDeposits(ctx)