sweep: remove all unconfirmed descendant transactions when a sweep co…

…nflicts

Before this commit, we we were trying to sweep an anchor output, and
that output was spent by someone else (not the sweeper), then we would
report this back to the original resolver (allowing it to be cleaned
up), and also remove the set of inputs spent by that transaction from
the set we need to sweep.

However, it's possible that if a user is spending unconfirmed outputs,
then the wallet is holding onto an invalid transaction, as the outputs
that were used as inputs have been double spent elsewhere.

In this commit, we fix this issue by recursively removing all descendant
transactions of our past sweeps that have an intersecting input set as
the spending transaction. In cases where a user spent an unconfirmed
output to funding a channel, and that output was a descendant of the now
swept anchor output, the funds will now properly be marked as available.

Fixes lightningnetwork#6241

lntest/mock/walletcontroller.go

@@ -251,3 +251,11 @@ func (w *WalletController) Start() error {
 func (w *WalletController) Stop() error {
 	return nil
 }
+
+func (w *WalletController) FetchTx(chainhash.Hash) (*wire.MsgTx, error) {
+	return nil, nil
+}
+
+func (w *WalletController) RemoveDescendants(*wire.MsgTx) error {
+	return nil
+}

lnwallet/dcrwallet/wallet.go

@@ -1034,6 +1034,34 @@ func (b *DcrWallet) GetRecoveryInfo() (bool, float64, error) {
 	return false, 0, fmt.Errorf("unimplemented")
 }
 
+// FetchTx attempts to fetch a transaction in the wallet's database
+// identified by the passed transaction hash. If the transaction can't
+// be found, then a nil pointer is returned.
+//
+// This is a part of the WalletController interface.
+func (b *DcrWallet) FetchTx(txid chainhash.Hash) (*wire.MsgTx, error) {
+	txs, _, err := b.wallet.GetTransactionsByHashes(b.ctx, []*chainhash.Hash{&txid})
+	if err != nil {
+		return nil, err
+	}
+
+	if len(txs) < 1 {
+		return nil, fmt.Errorf("tx %s not found", txid)
+	}
+
+	return txs[0], nil
+}
+
+// RemoveDescendants attempts to remove any transaction from the
+// wallet's tx store (that may be unconfirmed) that spends outputs
+// created by the passed transaction. This remove propagates
+// recursively down the chain of descendent transactions.
+//
+// This is a part of the WalletController interface.
+func (b *DcrWallet) RemoveDescendants(*wire.MsgTx) error {
+	return fmt.Errorf("RemoveDescendants is unimplemented")
+}
+
 // ListAccount lists existing wallet accounts.
 //
 // This is a part of the WalletController interface.

lnwallet/interface.go

@@ -347,6 +347,17 @@ type WalletController interface {
 	// set. Labels must not be empty, and they are limited to 500 chars.
 	LabelTransaction(hash chainhash.Hash, label string, overwrite bool) error
 
+	// FetchTx attempts to fetch a transaction in the wallet's database
+	// identified by the passed transaction hash. If the transaction can't
+	// be found, then a nil pointer is returned.
+	FetchTx(chainhash.Hash) (*wire.MsgTx, error)
+
+	// RemoveDescendants attempts to remove any transaction from the
+	// wallet's tx store (that may be unconfirmed) that spends outputs
+	// created by the passed transaction. This remove propagates
+	// recursively down the chain of descendent transactions.
+	RemoveDescendants(*wire.MsgTx) error
+
 	// FundPsbt creates a fully populated PSBT packet that contains enough
 	// inputs to fund the outputs specified in the passed in packet with the
 	// specified fee rate. If there is change left, a change output from the

lnwallet/remotedcrwallet/wallet.go

@@ -1227,6 +1227,35 @@ func (b *DcrWallet) GetRecoveryInfo() (bool, float64, error) {
 	return false, 0, fmt.Errorf("unimplemented")
 }
 
+// FetchTx attempts to fetch a transaction in the wallet's database
+// identified by the passed transaction hash. If the transaction can't
+// be found, then a nil pointer is returned.
+//
+// This is a part of the WalletController interface.
+func (b *DcrWallet) FetchTx(txid chainhash.Hash) (*wire.MsgTx, error) {
+	req := &pb.GetTransactionRequest{TransactionHash: txid[:]}
+	res, err := b.wallet.GetTransaction(b.ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	tx := wire.NewMsgTx()
+	err = tx.Deserialize(bytes.NewBuffer(res.Transaction.Transaction))
+	if err != nil {
+		return nil, err
+	}
+	return tx, err
+}
+
+// RemoveDescendants attempts to remove any transaction from the
+// wallet's tx store (that may be unconfirmed) that spends outputs
+// created by the passed transaction. This remove propagates
+// recursively down the chain of descendent transactions.
+//
+// This is a part of the WalletController interface.
+func (b *DcrWallet) RemoveDescendants(*wire.MsgTx) error {
+	return fmt.Errorf("RemoveDescendants is unimplemented")
+}
+
 // ListAccount lists existing wallet accounts.
 //
 // This is a part of the WalletController interface.

sweep/backend_mock_test.go

@@ -154,3 +154,11 @@ func (b *mockBackend) mine() {
 func (b *mockBackend) isDone() bool {
 	return len(b.unconfirmedTxes) == 0
 }
+
+func (b *mockBackend) RemoveDescendants(*wire.MsgTx) error {
+	return nil
+}
+
+func (b *mockBackend) FetchTx(chainhash.Hash) (*wire.MsgTx, error) {
+	return nil, nil
+}

sweep/interface.go

@@ -1,6 +1,7 @@
 package sweep
 
 import (
+	"github.com/decred/dcrd/chaincfg/chainhash"
 	"github.com/decred/dcrd/wire"
 	"github.com/decred/dcrlnd/lnwallet"
 )
@@ -32,4 +33,13 @@ type Wallet interface {
 	// ability to execute a function closure under an exclusive coin
 	// selection lock.
 	WithCoinSelectLock(f func() error) error
+
+	// RemoveDescendants removes any wallet transactions that spends
+	// outputs created by the specified transaction.
+	RemoveDescendants(*wire.MsgTx) error
+
+	// FetchTx returns the transaction that corresponds to the transaction
+	// hash passed in. If the transaction can't be found then a nil
+	// transaction pointer is returned.
+	FetchTx(chainhash.Hash) (*wire.MsgTx, error)
 }

sweep/sweeper.go

@@ -524,6 +524,81 @@ func (s *UtxoSweeper) feeRateForPreference(
 	return feeRate, nil
 }
 
+// removeLastSweepDescendants removes any transactions from the wallet that
+// spend outputs produced by the passed spendingTx. This needs to be done in
+// cases where we're not the only ones that can sweep an output, but there may
+// exist unconfirmed spends that spend outputs created by a sweep transaction.
+// The most common case for this is when someone sweeps our anchor outputs
+// after 16 blocks.
+func (s *UtxoSweeper) removeLastSweepDescendants(spendingTx *wire.MsgTx) error {
+	// Obtain all the past sweeps that we've done so far. We'll need these
+	// to ensure that if the spendingTx spends any of the same inputs, then
+	// we remove any transaction that may be spending those inputs from the
+	// wallet.
+	//
+	// TODO(roasbeef): can be last sweep here if we remove anything confirmed
+	// from the store?
+	pastSweepHashes, err := s.cfg.Store.ListSweeps()
+	if err != nil {
+		return err
+	}
+
+	log.Debugf("Attempting to remove descendant txns invalidated by "+
+		"(txid=%v): %v", spendingTx.TxHash(), spew.Sdump(spendingTx))
+
+	// Construct a map of the inputs this transaction spends for each look
+	// up.
+	inputsSpent := make(map[wire.OutPoint]struct{}, len(spendingTx.TxIn))
+	for _, txIn := range spendingTx.TxIn {
+		inputsSpent[txIn.PreviousOutPoint] = struct{}{}
+	}
+
+	// We'll now go through each past transaction we published during this
+	// epoch and cross reference the spent inputs. If there're any inputs
+	// in common with the inputs the spendingTx spent, then we'll remove
+	// those.
+	//
+	// TODO(roasbeef): need to start to remove all transaction hashes after
+	// every N blocks (assumed point of no return)
+	for _, sweepHash := range pastSweepHashes {
+		sweepTx, err := s.cfg.Wallet.FetchTx(sweepHash)
+		if err != nil {
+			return err
+		}
+
+		// Transaction wasn't found in the wallet, may have already
+		// been replaced/removed.
+		if sweepTx == nil {
+			continue
+		}
+
+		// Check to see if this past sweep transaction spent any of the
+		// same inputs as spendingTx.
+		var isConflicting bool
+		for _, txIn := range sweepTx.TxIn {
+			if _, ok := inputsSpent[txIn.PreviousOutPoint]; ok {
+				isConflicting = true
+				break
+			}
+		}
+
+		// If it did, then we'll signal the wallet to remove all the
+		// transactions that are descendants of outputs created by the
+		// sweepTx.
+		if isConflicting {
+			log.Debugf("Removing sweep txid=%v from wallet: %v",
+				sweepTx.TxHash(), spew.Sdump(sweepTx))
+
+			err := s.cfg.Wallet.RemoveDescendants(sweepTx)
+			if err != nil {
+				log.Warnf("unable to remove descendants: %v", err)
+			}
+		}
+	}
+
+	return nil
+}
+
 // collector is the sweeper main loop. It processes new inputs, spend
 // notifications and counts down to publication of the sweep tx.
 func (s *UtxoSweeper) collector(blockEpochs <-chan *chainntnfs.BlockEpoch) {
@@ -645,13 +720,30 @@ func (s *UtxoSweeper) collector(blockEpochs <-chan *chainntnfs.BlockEpoch) {
 				continue
 			}
 
-			log.Debugf("Detected spend related to in flight inputs "+
-				"(is_ours=%v): %v",
-				newLogClosure(func() string {
-					spend.SpendingTx.CachedTxHash()
-					return spew.Sdump(spend.SpendingTx)
-				}), isOurTx,
-			)
+			// If this isn't our transaction, it means someone else
+			// swept outputs that we were attempting to sweep. This
+			// can happen for anchor outputs as well as justice
+			// transactions. In this case, we'll notify the wallet
+			// to remove any spends that a descent from this
+			// output.
+			if !isOurTx {
+				err := s.removeLastSweepDescendants(
+					spend.SpendingTx,
+				)
+				if err != nil {
+					log.Warnf("unable to remove descendant "+
+						"transactions due to tx %v: ",
+						spendHash)
+				}
+
+				log.Debugf("Detected spend related to in flight inputs "+
+					"(is_ours=%v): %v",
+					newLogClosure(func() string {
+						spend.SpendingTx.CachedTxHash()
+						return spew.Sdump(spend.SpendingTx)
+					}), isOurTx,
+				)
+			}
 
 			// Clear out the last published tx since it either
 			// already got confirmed or something else made it

sweep/txgenerator.go

@@ -223,7 +223,9 @@ func createSweepTx(inputs []input.Input, outputs []*wire.TxOut,
 	}
 
 	if requiredOutput+txFee > totalInput {
-		return nil, fmt.Errorf("insufficient input to create sweep tx")
+		return nil, fmt.Errorf("insufficient input to create sweep "+
+			"tx: input_sum=%v, output_sum=%v", totalInput,
+			requiredOutput+txFee)
 	}
 
 	// The value remaining after the required output and fees, go to