Merge pull request #4369 from wpaulino/coin-select-lock-sync

rpc: acquire global coin select lock in related RPCs

lnrpc/walletrpc/config_active.go

@@ -38,6 +38,13 @@ type Config struct {
 	// any relevant requests to.
 	Wallet lnwallet.WalletController
 
+	// CoinSelectionLocker allows the caller to perform an operation, which
+	// is synchronized with all coin selection attempts. This can be used
+	// when an operation requires that all coin selection operations cease
+	// forward progress. Think of this as an exclusive lock on coin
+	// selection operations.
+	CoinSelectionLocker sweep.CoinSelectionLocker
+
 	// KeyRing is an interface that the WalletKit will use to derive any
 	// keys due to incoming client requests.
 	KeyRing keychain.KeyRing

lnrpc/walletrpc/walletkit_server.go

@@ -10,6 +10,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/txscript"
@@ -258,7 +259,15 @@ func (w *WalletKit) ListUnspent(ctx context.Context,
 
 	// With our arguments validated, we'll query the internal wallet for
 	// the set of UTXOs that match our query.
-	utxos, err := w.cfg.Wallet.ListUnspentWitness(minConfs, maxConfs)
+	//
+	// We'll acquire the global coin selection lock to ensure there aren't
+	// any other concurrent processes attempting to lock any UTXOs which may
+	// be shown available to us.
+	var utxos []*lnwallet.Utxo
+	err = w.cfg.CoinSelectionLocker.WithCoinSelectLock(func() error {
+		utxos, err = w.cfg.Wallet.ListUnspentWitness(minConfs, maxConfs)
+		return err
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -301,7 +310,13 @@ func (w *WalletKit) LeaseOutput(ctx context.Context,
 		return nil, err
 	}
 
-	expiration, err := w.cfg.Wallet.LeaseOutput(lockID, *op)
+	// Acquire the global coin selection lock to ensure there aren't any
+	// other concurrent processes attempting to lease the same UTXO.
+	var expiration time.Time
+	err = w.cfg.CoinSelectionLocker.WithCoinSelectLock(func() error {
+		expiration, err = w.cfg.Wallet.LeaseOutput(lockID, *op)
+		return err
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -328,7 +343,12 @@ func (w *WalletKit) ReleaseOutput(ctx context.Context,
 		return nil, err
 	}
 
-	if err := w.cfg.Wallet.ReleaseOutput(lockID, *op); err != nil {
+	// Acquire the global coin selection lock to maintain consistency as
+	// it's acquired when we initially leased the output.
+	err = w.cfg.CoinSelectionLocker.WithCoinSelectLock(func() error {
+		return w.cfg.Wallet.ReleaseOutput(lockID, *op)
+	})
+	if err != nil {
 		return nil, err
 	}
 

lnwallet/btcwallet/btcwallet.go

@@ -293,6 +293,8 @@ func (b *BtcWallet) IsOurAddress(a btcutil.Address) bool {
 // the specified outputs. In the case the wallet has insufficient funds, or the
 // outputs are non-standard, a non-nil error will be returned.
 //
+// NOTE: This method requires the global coin selection lock to be held.
+//
 // This is a part of the WalletController interface.
 func (b *BtcWallet) SendOutputs(outputs []*wire.TxOut,
 	feeRate chainfee.SatPerKWeight, label string) (*wire.MsgTx, error) {
@@ -321,6 +323,8 @@ func (b *BtcWallet) SendOutputs(outputs []*wire.TxOut,
 // NOTE: The dryRun argument can be set true to create a tx that doesn't alter
 // the database. A tx created with this set to true SHOULD NOT be broadcasted.
 //
+// NOTE: This method requires the global coin selection lock to be held.
+//
 // This is a part of the WalletController interface.
 func (b *BtcWallet) CreateSimpleTx(outputs []*wire.TxOut,
 	feeRate chainfee.SatPerKWeight, dryRun bool) (*txauthor.AuthoredTx, error) {
@@ -355,6 +359,8 @@ func (b *BtcWallet) CreateSimpleTx(outputs []*wire.TxOut,
 // avoid race conditions when selecting inputs for usage when funding a
 // channel.
 //
+// NOTE: This method requires the global coin selection lock to be held.
+//
 // This is a part of the WalletController interface.
 func (b *BtcWallet) LockOutpoint(o wire.OutPoint) {
 	b.wallet.LockOutpoint(o)
@@ -363,6 +369,8 @@ func (b *BtcWallet) LockOutpoint(o wire.OutPoint) {
 // UnlockOutpoint unlocks a previously locked output, marking it eligible for
 // coin selection.
 //
+// NOTE: This method requires the global coin selection lock to be held.
+//
 // This is a part of the WalletController interface.
 func (b *BtcWallet) UnlockOutpoint(o wire.OutPoint) {
 	b.wallet.UnlockOutpoint(o)
@@ -377,14 +385,25 @@ func (b *BtcWallet) UnlockOutpoint(o wire.OutPoint) {
 // If the output is not known, wtxmgr.ErrUnknownOutput is returned. If the
 // output has already been locked to a different ID, then
 // wtxmgr.ErrOutputAlreadyLocked is returned.
+//
+// NOTE: This method requires the global coin selection lock to be held.
 func (b *BtcWallet) LeaseOutput(id wtxmgr.LockID, op wire.OutPoint) (time.Time,
 	error) {
+
+	// Make sure we don't attempt to double lock an output that's been
+	// locked by the in-memory implementation.
+	if b.wallet.LockedOutpoint(op) {
+		return time.Time{}, wtxmgr.ErrOutputAlreadyLocked
+	}
+
 	return b.wallet.LeaseOutput(id, op)
 }
 
 // ReleaseOutput unlocks an output, allowing it to be available for coin
 // selection if it remains unspent. The ID should match the one used to
 // originally lock the output.
+//
+// NOTE: This method requires the global coin selection lock to be held.
 func (b *BtcWallet) ReleaseOutput(id wtxmgr.LockID, op wire.OutPoint) error {
 	return b.wallet.ReleaseOutput(id, op)
 }
@@ -392,6 +411,8 @@ func (b *BtcWallet) ReleaseOutput(id wtxmgr.LockID, op wire.OutPoint) error {
 // ListUnspentWitness returns a slice of all the unspent outputs the wallet
 // controls which pay to witness programs either directly or indirectly.
 //
+// NOTE: This method requires the global coin selection lock to be held.
+//
 // This is a part of the WalletController interface.
 func (b *BtcWallet) ListUnspentWitness(minConfs, maxConfs int32) (
 	[]*lnwallet.Utxo, error) {

lnwallet/interface.go

@@ -178,6 +178,8 @@ type WalletController interface {
 	// funds, or the outputs are non-standard, an error should be returned.
 	// This method also takes the target fee expressed in sat/kw that should
 	// be used when crafting the transaction.
+	//
+	// NOTE: This method requires the global coin selection lock to be held.
 	SendOutputs(outputs []*wire.TxOut,
 		feeRate chainfee.SatPerKWeight, label string) (*wire.MsgTx, error)
 
@@ -191,6 +193,8 @@ type WalletController interface {
 	// NOTE: The dryRun argument can be set true to create a tx that
 	// doesn't alter the database. A tx created with this set to true
 	// SHOULD NOT be broadcasted.
+	//
+	// NOTE: This method requires the global coin selection lock to be held.
 	CreateSimpleTx(outputs []*wire.TxOut, feeRate chainfee.SatPerKWeight,
 		dryRun bool) (*txauthor.AuthoredTx, error)
 
@@ -201,6 +205,8 @@ type WalletController interface {
 	// 'minconfirms' indicates that even unconfirmed outputs should be
 	// returned. Using MaxInt32 as 'maxconfirms' implies returning all
 	// outputs with at least 'minconfirms'.
+	//
+	// NOTE: This method requires the global coin selection lock to be held.
 	ListUnspentWitness(minconfirms, maxconfirms int32) ([]*Utxo, error)
 
 	// ListTransactionDetails returns a list of all transactions which are
@@ -217,10 +223,14 @@ type WalletController interface {
 	// be deemed as eligible for coin selection. Locking outputs are
 	// utilized in order to avoid race conditions when selecting inputs for
 	// usage when funding a channel.
+	//
+	// NOTE: This method requires the global coin selection lock to be held.
 	LockOutpoint(o wire.OutPoint)
 
 	// UnlockOutpoint unlocks a previously locked output, marking it
 	// eligible for coin selection.
+	//
+	// NOTE: This method requires the global coin selection lock to be held.
 	UnlockOutpoint(o wire.OutPoint)
 
 	// LeaseOutput locks an output to the given ID, preventing it from being
@@ -232,11 +242,15 @@ type WalletController interface {
 	// If the output is not known, wtxmgr.ErrUnknownOutput is returned. If
 	// the output has already been locked to a different ID, then
 	// wtxmgr.ErrOutputAlreadyLocked is returned.
+	//
+	// NOTE: This method requires the global coin selection lock to be held.
 	LeaseOutput(id wtxmgr.LockID, op wire.OutPoint) (time.Time, error)
 
 	// ReleaseOutput unlocks an output, allowing it to be available for coin
 	// selection if it remains unspent. The ID should match the one used to
 	// originally lock the output.
+	//
+	// NOTE: This method requires the global coin selection lock to be held.
 	ReleaseOutput(id wtxmgr.LockID, op wire.OutPoint) error
 
 	// PublishTransaction performs cursory validation (dust checks, etc),

rpcserver.go

@@ -950,7 +950,17 @@ func (r *rpcServer) ListUnspent(ctx context.Context,
 
 	// With our arguments validated, we'll query the internal wallet for
 	// the set of UTXOs that match our query.
-	utxos, err := r.server.cc.wallet.ListUnspentWitness(minConfs, maxConfs)
+	//
+	// We'll acquire the global coin selection lock to ensure there aren't
+	// any other concurrent processes attempting to lock any UTXOs which may
+	// be shown available to us.
+	var utxos []*lnwallet.Utxo
+	err = r.server.cc.wallet.WithCoinSelectLock(func() error {
+		utxos, err = r.server.cc.wallet.ListUnspentWitness(
+			minConfs, maxConfs,
+		)
+		return err
+	})
 	if err != nil {
 		return nil, err
 	}

subrpcserver_config.go

@@ -153,6 +153,9 @@ func (s *subRPCServerConfigs) PopulateDependencies(cfg *Config, cc *chainControl
 			subCfgValue.FieldByName("Wallet").Set(
 				reflect.ValueOf(cc.wallet),
 			)
+			subCfgValue.FieldByName("CoinSelectionLocker").Set(
+				reflect.ValueOf(cc.wallet),
+			)
 			subCfgValue.FieldByName("KeyRing").Set(
 				reflect.ValueOf(cc.keyRing),
 			)