From e0367e552adfc23e3605387aac5688055b77f6be Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Sat, 1 Aug 2026 01:03:00 -0500 Subject: [PATCH 1/2] 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 | 92 +++++++++++++++----------- 10 files changed, 209 insertions(+), 38 deletions(-) create mode 100644 staticaddr/deposit/registry.go diff --git a/docs/architecture.md b/docs/architecture.md index 5762ef7fe..7dde39194 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 c5bb85c30..c3bc13786 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 61fc8e769..3112eb311 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 000000000..0ecf4b68e --- /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 afc0085d8..f035b0bd5 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 d54355a7b..8efcd1025 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 47a447fc3..794649af7 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 9fa8587c2..af4cde6eb 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 0f32697a9..584a79bab 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 3a7927a45..f6bc44b18 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,19 +446,38 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, return "", "", err } - published, err := m.publishFinalizedWithdrawalTx(ctx, finalizedTx) + withdrawalPkScript, err := txscript.PayToAddrScript(withdrawalAddress) if err != nil { - return "", "", err + return "", "", fmt.Errorf("could not get withdrawal "+ + "pkscript: %w", err) } - if !published { - return "", "", nil + // 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() } - withdrawalPkScript, err := txscript.PayToAddrScript(withdrawalAddress) + // 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 { - return "", "", fmt.Errorf("could not get withdrawal "+ - "pkscript: %w", err) + 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 @@ -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, - ) + // 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) + } + } + } + + // 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 "", "", fmt.Errorf("failed to transition deposits %w", - err) + return "", "", 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) - } + if !published { + return "", "", nil } return finalizedTx.TxID(), withdrawalAddress.String(), nil From 1fa6afbaf5a4b8a94d58ef7b78bbb51d8969ef02 Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Sat, 1 Aug 2026 01:07:30 -0500 Subject: [PATCH 2/2] staticaddr: test concurrent withdrawal and loop-in Hold a withdrawal while it prepares a confirmed deposit, then start a loop-in with the same outpoint. It must fail before reaching its quote request, and cleanup must make the deposit available again. --- staticaddr/loopin/manager_test.go | 206 ++++++++++++++++++++++++++++++ 1 file changed, 206 insertions(+) diff --git a/staticaddr/loopin/manager_test.go b/staticaddr/loopin/manager_test.go index af4cde6eb..aa4cb496c 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -4,20 +4,25 @@ import ( "bytes" "context" "errors" + "sync" "testing" + "time" "github.com/btcsuite/btcd/btcec/v2/schnorr/musig2" "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/btcutil/psbt" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/lndclient" "github.com/lightninglabs/loop" "github.com/lightninglabs/loop/fsm" "github.com/lightninglabs/loop/labels" "github.com/lightninglabs/loop/staticaddr/deposit" "github.com/lightninglabs/loop/staticaddr/script" + "github.com/lightninglabs/loop/staticaddr/withdraw" "github.com/lightninglabs/loop/swap" "github.com/lightninglabs/loop/swapserverrpc" + "github.com/lightningnetwork/lnd/lnrpc/walletrpc" "github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/zpay32" @@ -245,6 +250,112 @@ func TestInitiateLoopInAllowsReservedAutoloopLabel(t *testing.T) { require.Equal(t, selectedDeposit.Value, quoteGetter.amount) } +// TestWithdrawalExcludesConcurrentLoopIn verifies that a withdrawal keeps its +// deposits unavailable to a loop-in while it is still being prepared. Without +// the deposit use registry, both operations observe the deposit as Deposited +// and the loop-in reaches its quote request. +func TestWithdrawalExcludesConcurrentLoopIn(t *testing.T) { + ctx := t.Context() + + // Use one deposited output and share a real deposit-use registry between + // the withdrawal and loop-in managers. + selectedDeposit := makeDeposit(3, 0, 9_000, 100) + selectedDeposit.SetState(deposit.Deposited) + selectedOutpoint := selectedDeposit.OutPoint.String() + depositManager := &conflictDepositManager{ + mockDepositManager: &mockDepositManager{ + byOutpoint: map[string]*deposit.Deposit{ + selectedOutpoint: selectedDeposit, + }, + }, + useRegistry: deposit.NewManager(&deposit.ManagerConfig{}), + } + + // Block destination address creation. Withdrawal registers the deposit + // before this point, but has not yet changed its persisted state, which + // creates the precise overlap this test needs. + withdrawalStarted := make(chan struct{}, 1) + releaseWithdrawal := make(chan struct{}) + var releaseOnce sync.Once + release := func() { + releaseOnce.Do(func() { + close(releaseWithdrawal) + }) + } + defer release() + + withdrawalErr := errors.New("stop withdrawal after registration") + walletKit := &blockingNextAddrWallet{ + started: withdrawalStarted, + release: releaseWithdrawal, + err: withdrawalErr, + } + withdrawalManager, err := withdraw.NewManager( + &withdraw.ManagerConfig{ + DepositManager: depositManager, + WalletKit: walletKit, + }, 200, + ) + require.NoError(t, err) + + // Start the withdrawal asynchronously so the test can attempt a loop-in + // while withdrawal preparation is paused. + withdrawalDone := make(chan error, 1) + go func() { + _, _, err := withdrawalManager.WithdrawDeposits( + ctx, []wire.OutPoint{selectedDeposit.OutPoint}, "", 1, 0, + ) + withdrawalDone <- err + }() + + // Reaching NextAddr proves the withdrawal has already registered its use + // of the deposit. + select { + case <-withdrawalStarted: + + case <-time.After(5 * time.Second): + t.Fatal("withdrawal did not reach destination address request") + } + + // A sentinel quote error makes it clear if the concurrent loop-in gets + // past the registry check, as it would without the fix. + quoteErr := errors.New("loop-in reached quote request") + loopInManager, err := NewManager(&Config{ + DepositManager: depositManager, + QuoteGetter: &mockQuoteGetter{ + err: quoteErr, + }, + NodePubkey: route.Vertex{2}, + }, 200) + require.NoError(t, err) + + request := &loop.StaticAddressLoopInRequest{ + DepositOutpoints: []string{selectedOutpoint}, + SelectedAmount: selectedDeposit.Value, + MaxSwapFee: 1_000, + } + + // The loop-in must fail on local deposit ownership before requesting a + // quote or starting any swap work. + _, err = loopInManager.initiateLoopIn(ctx, request) + require.ErrorContains(t, err, "deposit already in use") + require.NotErrorIs(t, err, quoteErr) + + // Let the withdrawal fail before signing and verify its registration is + // cleaned up. A subsequent loop-in must reach the quote request. + release() + select { + case err = <-withdrawalDone: + require.ErrorIs(t, err, withdrawalErr) + + case <-time.After(5 * time.Second): + t.Fatal("withdrawal did not release its deposit use") + } + + _, err = loopInManager.initiateLoopIn(ctx, request) + require.ErrorIs(t, err, quoteErr) +} + // TestUpdateLoopInSendsUpdateAfterSuccessfulStoreUpdate protects the // notification contract: after a successful database update, the manager must // publish the stored loop-in state to listeners. @@ -498,6 +609,101 @@ func (m *mockDepositManager) GetActiveDepositsInState(_ fsm.StateType) ( return m.activeDeposits, nil } +// depositUseRegistrar describes the production registration method exercised +// by this regression test. The local interface keeps the test buildable when +// the implementation under test is reverted. +type depositUseRegistrar interface { + // RegisterDepositUse claims the requested deposits until the returned + // cleanup function is called. + RegisterDepositUse([]*deposit.Deposit) (func(), error) +} + +// conflictDepositManager supplies both loop-in and withdrawal test methods +// while delegating deposit-use coordination to a real deposit manager. +type conflictDepositManager struct { + // mockDepositManager supplies the standard loop-in deposit lookups. + *mockDepositManager + + // useRegistry owns the production registry shared by both operations. + useRegistry *deposit.Manager +} + +// RegisterDepositUse delegates registration to the production deposit +// manager when the method is available. +func (m *conflictDepositManager) RegisterDepositUse( + deposits []*deposit.Deposit) (func(), error) { + + registrar, ok := any(m.useRegistry).(depositUseRegistrar) + if !ok { + return nil, errors.New("deposit use registry unavailable") + } + + unregister, err := registrar.RegisterDepositUse(deposits) + if err != nil { + return nil, err + } + + return unregister, nil +} + +// AllOutpointsActiveDeposits adapts withdrawal's wire outpoints to the string +// outpoint lookup supplied by mockDepositManager. +func (m *conflictDepositManager) AllOutpointsActiveDeposits( + outpoints []wire.OutPoint, state fsm.StateType) ([]*deposit.Deposit, bool) { + + stringOutpoints := make([]string, len(outpoints)) + for i, outpoint := range outpoints { + stringOutpoints[i] = outpoint.String() + } + + return m.AllStringOutpointsActiveDeposits(stringOutpoints, state) +} + +// UpdateDeposit implements withdraw.DepositManager without persisting state +// because the test stops withdrawal before a deposit update is needed. +func (m *conflictDepositManager) UpdateDeposit(context.Context, + *deposit.Deposit) error { + + return nil +} + +// blockingNextAddrWallet pauses withdrawal at destination address creation so +// the test can start a concurrent loop-in at a deterministic point. +type blockingNextAddrWallet struct { + // WalletKitClient supplies methods not exercised by this test. + lndclient.WalletKitClient + + // started signals that withdrawal reached destination address creation. + started chan<- struct{} + + // release unblocks the destination address request. + release <-chan struct{} + + // err is returned after the destination address request is released. + err error +} + +// NextAddr signals that withdrawal reached address creation, waits for the +// test to release it, and then returns the configured error. +func (w *blockingNextAddrWallet) NextAddr(ctx context.Context, _ string, + _ walletrpc.AddressType, _ bool) (btcutil.Address, error) { + + select { + case w.started <- struct{}{}: + + case <-ctx.Done(): + return nil, ctx.Err() + } + + select { + case <-w.release: + return nil, w.err + + case <-ctx.Done(): + return nil, ctx.Err() + } +} + // mockQuoteGetter records the inputs to quote requests and returns a fixed // loop-in quote. type mockQuoteGetter struct {