Merge 1fa6afbaf5 into 6d1dbcb599

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.

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,

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{

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
+}

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) {

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)
 

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(

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.
@@ -436,6 +547,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) {
 
@@ -490,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 {

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,

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