From e0367e552adfc23e3605387aac5688055b77f6be Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Sat, 1 Aug 2026 01:03:00 -0500 Subject: [PATCH] staticaddr: coordinate concurrent deposit use Withdrawals obtain signatures before deposits enter their durable state, leaving a window where a loop-in can select the same outputs. Coordinate in-flight client operations with an in-memory registry. Persist withdrawal data before publication. Existing recovery resumes it after restart. --- docs/architecture.md | 6 ++ staticaddr/deposit/fsm.go | 5 +- staticaddr/deposit/manager.go | 10 +++ staticaddr/deposit/registry.go | 103 ++++++++++++++++++++++++++++++ staticaddr/loopin/actions_test.go | 7 ++ staticaddr/loopin/interface.go | 4 ++ staticaddr/loopin/manager.go | 8 +++ staticaddr/loopin/manager_test.go | 8 +++ staticaddr/withdraw/interface.go | 4 ++ staticaddr/withdraw/manager.go | 98 +++++++++++++++++----------- 10 files changed, 212 insertions(+), 41 deletions(-) create mode 100644 staticaddr/deposit/registry.go diff --git a/docs/architecture.md b/docs/architecture.md index 5762ef7f..7dde3919 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -111,3 +111,9 @@ then be swapped for an off-chain Lightning payment. HTLC transaction, which then follows a standard Loop-In flow. When the client gets the LN payment, they cooperate with the server to sweep the deposit directly to the server's wallet instead of publishing the HTLC tx. + +- **Concurrent Use:** The client keeps an in-memory registry while a deposit + is being prepared for a Loop-In or withdrawal. This prevents overlapping + operations before the deposit reaches its persisted FSM state. The registry + starts empty after a restart, leaving recovery to persisted deposit state + and wallet reconciliation. diff --git a/staticaddr/deposit/fsm.go b/staticaddr/deposit/fsm.go index c5bb85c3..c3bc1378 100644 --- a/staticaddr/deposit/fsm.go +++ b/staticaddr/deposit/fsm.go @@ -353,8 +353,9 @@ func (f *FSM) DepositStatesV0() fsm.States { OnRecover: Withdrawing, // A precondition for the Withdrawing state is - // that the withdrawal transaction has been - // broadcast. If the deposit expires while the + // that a finalized withdrawal transaction has + // been persisted for automatic publication and + // recovery. If the deposit expires while the // withdrawal isn't confirmed, we can ignore the // expiry. OnExpiry: Withdrawing, diff --git a/staticaddr/deposit/manager.go b/staticaddr/deposit/manager.go index 61fc8e76..3112eb31 100644 --- a/staticaddr/deposit/manager.go +++ b/staticaddr/deposit/manager.go @@ -66,6 +66,10 @@ type ManagerConfig struct { type Manager struct { cfg *ManagerConfig + // useRegistry coordinates concurrent client operations that use the + // same deposits. + useRegistry depositUseRegistry + // mu guards access to the activeDeposits map. mu sync.Mutex @@ -91,6 +95,12 @@ type Manager struct { currentHeight atomic.Uint32 } +// RegisterDepositUse registers the deposits for exclusive use by one client +// operation. The returned function unregisters only that operation's use. +func (m *Manager) RegisterDepositUse(deposits []*Deposit) (func(), error) { + return m.useRegistry.register(deposits) +} + // NewManager creates a new deposit manager. func NewManager(cfg *ManagerConfig) *Manager { return &Manager{ diff --git a/staticaddr/deposit/registry.go b/staticaddr/deposit/registry.go new file mode 100644 index 00000000..0ecf4b68 --- /dev/null +++ b/staticaddr/deposit/registry.go @@ -0,0 +1,103 @@ +package deposit + +import ( + "errors" + "fmt" + "sync" + + "github.com/btcsuite/btcd/wire" +) + +var ( + // ErrDepositInUse is returned when another client operation is already + // using one of the requested deposits. + ErrDepositInUse = errors.New("deposit already in use") +) + +// depositUseRegistration identifies a single use of one or more deposits. +// Its pointer identity prevents delayed cleanup from unregistering a newer +// operation. +type depositUseRegistration struct { + // Give each registration non-zero size so distinct pointer identity is + // guaranteed. + _ byte +} + +// depositUseRegistry coordinates in-flight client operations that use static +// address deposits. It is intentionally kept in memory so a client restart +// clears incomplete registrations and lets persisted deposit state drive +// recovery. +type depositUseRegistry struct { + mu sync.Mutex + + registrations map[wire.OutPoint]*depositUseRegistration +} + +// register records the deposits as being in use and returns an owner-safe +// cleanup function. Either all deposits are registered or none are. +func (r *depositUseRegistry) register(deposits []*Deposit) (func(), error) { + if len(deposits) == 0 { + return nil, errors.New("no deposits selected") + } + + // Copy the outpoints up front so the cleanup closure does not depend on + // caller-owned deposit pointers after this method returns. + outpoints := make([]wire.OutPoint, len(deposits)) + for i, d := range deposits { + if d == nil { + return nil, fmt.Errorf("nil deposit at index %d", i) + } + + outpoints[i] = d.OutPoint + } + + // Reject duplicate inputs before taking the registry lock. A duplicate + // would otherwise make ownership of the cleanup entry ambiguous. + if err := CheckDuplicates(outpoints); err != nil { + return nil, err + } + + // Keep the conflict check and registration under the same lock so a + // request for multiple deposits is registered atomically. + r.mu.Lock() + defer r.mu.Unlock() + + // Check all outpoints before changing the map. This ensures a conflict + // leaves every requested deposit unregistered by this operation. + for _, outpoint := range outpoints { + if _, ok := r.registrations[outpoint]; ok { + return nil, fmt.Errorf("%w: %v", ErrDepositInUse, + outpoint) + } + } + + // Initialize the map lazily because the registry's zero value is ready + // for use and is embedded directly in the deposit manager. + if r.registrations == nil { + r.registrations = make( + map[wire.OutPoint]*depositUseRegistration, + ) + } + + // Use one registration token for the whole request. The token lets the + // cleanup closure prove that it still owns each entry it removes. + registration := &depositUseRegistration{} + for _, outpoint := range outpoints { + r.registrations[outpoint] = registration + } + + return func() { + r.mu.Lock() + defer r.mu.Unlock() + + for _, outpoint := range outpoints { + // Only remove entries still owned by this registration. This + // makes repeated or delayed cleanup safe if a newer operation + // has since registered the same outpoint. + current, ok := r.registrations[outpoint] + if ok && current == registration { + delete(r.registrations, outpoint) + } + } + }, nil +} diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index afc0085d..f035b0bd 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -3354,6 +3354,13 @@ func (n *noopDepositManager) EnsureDepositsFresh(context.Context) error { return n.ensureFreshErr } +// RegisterDepositUse implements DepositManager with a no-op. +func (n *noopDepositManager) RegisterDepositUse( + []*deposit.Deposit) (func(), error) { + + return func() {}, nil +} + // GetAllDeposits implements DepositManager with a no-op. func (n *noopDepositManager) GetAllDeposits(_ context.Context) ( []*deposit.Deposit, error) { diff --git a/staticaddr/loopin/interface.go b/staticaddr/loopin/interface.go index d54355a7..8efcd102 100644 --- a/staticaddr/loopin/interface.go +++ b/staticaddr/loopin/interface.go @@ -48,6 +48,10 @@ type DepositManager interface { // EnsureDepositsFresh reconciles active deposits with the wallet view. EnsureDepositsFresh(ctx context.Context) error + // RegisterDepositUse registers the deposits for exclusive use by this + // client operation and returns a function that unregisters them. + RegisterDepositUse(deposits []*deposit.Deposit) (func(), error) + // GetAllDeposits returns all known deposits from the database store. GetAllDeposits(ctx context.Context) ([]*deposit.Deposit, error) diff --git a/staticaddr/loopin/manager.go b/staticaddr/loopin/manager.go index 47a447fc..794649af 100644 --- a/staticaddr/loopin/manager.go +++ b/staticaddr/loopin/manager.go @@ -694,6 +694,14 @@ func (m *Manager) initiateLoopIn(ctx context.Context, } } + unregister, err := m.cfg.DepositManager.RegisterDepositUse( + selectedDeposits, + ) + if err != nil { + return nil, fmt.Errorf("unable to register deposit use: %w", err) + } + defer unregister() + // Calculate the total deposit amount and check if the selected amount // would leave a dust output. swapAmount, err := DeduceSwapAmount( diff --git a/staticaddr/loopin/manager_test.go b/staticaddr/loopin/manager_test.go index 9fa8587c..af4cde6e 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -436,6 +436,14 @@ func (m *mockDepositManager) EnsureDepositsFresh(context.Context) error { return nil } +// RegisterDepositUse implements DepositManager by accepting all requested +// deposits without tracking their use. +func (m *mockDepositManager) RegisterDepositUse( + []*deposit.Deposit) (func(), error) { + + return func() {}, nil +} + func (m *mockDepositManager) GetAllDeposits(_ context.Context) ( []*deposit.Deposit, error) { diff --git a/staticaddr/withdraw/interface.go b/staticaddr/withdraw/interface.go index 0f32697a..584a79ba 100644 --- a/staticaddr/withdraw/interface.go +++ b/staticaddr/withdraw/interface.go @@ -24,6 +24,10 @@ type DepositManager interface { // EnsureDepositsFresh reconciles active deposits with the wallet view. EnsureDepositsFresh(ctx context.Context) error + // RegisterDepositUse registers the deposits for exclusive use by this + // client operation and returns a function that unregisters them. + RegisterDepositUse(deposits []*deposit.Deposit) (func(), error) + // GetActiveDepositsInState returns all active deposits in the given // state. GetActiveDepositsInState(stateFilter fsm.StateType) ([]*deposit.Deposit, diff --git a/staticaddr/withdraw/manager.go b/staticaddr/withdraw/manager.go index 3a7927a4..f6bc44b1 100644 --- a/staticaddr/withdraw/manager.go +++ b/staticaddr/withdraw/manager.go @@ -386,6 +386,13 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, } } + unregister, err := m.cfg.DepositManager.RegisterDepositUse(deposits) + if err != nil { + return "", "", fmt.Errorf("unable to register deposit use: %w", + err) + } + defer unregister() + for _, d := range deposits { // Deposited now includes mempool outputs for static loop-ins, but // withdrawals still require the deposit input to be confirmed. @@ -439,21 +446,40 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, return "", "", err } - published, err := m.publishFinalizedWithdrawalTx(ctx, finalizedTx) - if err != nil { - return "", "", err - } - - if !published { - return "", "", nil - } - withdrawalPkScript, err := txscript.PayToAddrScript(withdrawalAddress) if err != nil { return "", "", fmt.Errorf("could not get withdrawal "+ "pkscript: %w", err) } + // Attach the finalized transaction before transitioning the deposits. + // The state transition persists it so recovery can republish after a + // restart or an ambiguous publication result. + previousWithdrawalTxns := make([]*wire.MsgTx, len(deposits)) + for i, d := range deposits { + d.Lock() + previousWithdrawalTxns[i] = d.FinalizedWithdrawalTx + d.FinalizedWithdrawalTx = finalizedTx + d.Unlock() + } + + // Transition before publishing so the persisted deposit state takes + // over from the in-memory use registry before the transaction can spend + // the selected deposits. + err = m.cfg.DepositManager.TransitionDeposits( + ctx, deposits, deposit.OnWithdrawInitiated, deposit.Withdrawing, + ) + if err != nil { + for i, d := range deposits { + d.Lock() + d.FinalizedWithdrawalTx = previousWithdrawalTxns[i] + d.Unlock() + } + + return "", "", fmt.Errorf("failed to transition deposits %w", + err) + } + // If this is the first time this cluster of deposits is withdrawn, we // start a goroutine that listens for the spent of the first input of // the withdrawal transaction. @@ -479,49 +505,43 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, // If a previous withdrawal existed across the selected deposits, and // it isn't the same as the new withdrawal, we remove it from the // finalized withdrawals to stop republishing it on block arrivals. - deposits[0].Lock() - prevTx := deposits[0].FinalizedWithdrawalTx - deposits[0].Unlock() + previousWithdrawalTx := previousWithdrawalTxns[0] + if previousWithdrawalTx != nil && + previousWithdrawalTx.TxHash() != finalizedTx.TxHash() { - if prevTx != nil && prevTx.TxHash() != finalizedTx.TxHash() { m.mu.Lock() - delete(m.finalizedWithdrawalTxns, prevTx.TxHash()) + delete(m.finalizedWithdrawalTxns, previousWithdrawalTx.TxHash()) m.mu.Unlock() } - // Attach the finalized withdrawal tx to the deposits. After a client - // restart we can use this address as an indicator to republish the - // withdrawal tx and continue the withdrawal. - // Deposits with the same withdrawal tx are part of the same withdrawal. - for _, d := range deposits { - d.Lock() - d.FinalizedWithdrawalTx = finalizedTx - d.Unlock() - } - // Add the new withdrawal tx to the finalized withdrawals to republish // it on block arrivals. m.mu.Lock() m.finalizedWithdrawalTxns[finalizedTx.TxHash()] = finalizedTx m.mu.Unlock() - // Transition the deposits to the withdrawing state. If the user fee - // bumped a withdrawal this results in a NOOP transition. - err = m.cfg.DepositManager.TransitionDeposits( - ctx, deposits, deposit.OnWithdrawInitiated, deposit.Withdrawing, - ) - if err != nil { - return "", "", fmt.Errorf("failed to transition deposits %w", - err) + // A fee bump is a self-transition, which the deposit FSM doesn't + // persist. Store the replacement transaction explicitly in that case. + if allWithdrawing { + for _, d := range deposits { + err = m.cfg.DepositManager.UpdateDeposit(ctx, d) + if err != nil { + return "", "", fmt.Errorf("failed to update "+ + "deposit %w", err) + } + } } - // Update the deposits in the database. - for _, d := range deposits { - err = m.cfg.DepositManager.UpdateDeposit(ctx, d) - if err != nil { - return "", "", fmt.Errorf("failed to update "+ - "deposit %w", err) - } + // Publish only after the finalized transaction and withdrawal state are + // durable. Keep that preparation on publication errors because the wallet + // RPC outcome can be ambiguous and recovery can retry automatically. + published, err := m.publishFinalizedWithdrawalTx(ctx, finalizedTx) + if err != nil { + return "", "", err + } + + if !published { + return "", "", nil } return finalizedTx.TxID(), withdrawalAddress.String(), nil