sweeper+lntest: remove conflicting tx

For anchor channels and neutrino backends we need to make sure
that sweeps of the same exclusive group are removed when one of
them is confirmed. Otherwise for neutrino backends those sweep
transaction are always rebroadcasted and are blocking funds in
the worst case scenario.

lntest/harness_assertion.go

@@ -27,6 +27,7 @@ import (
 	"github.com/lightningnetwork/lnd/lntest/rpc"
 	"github.com/lightningnetwork/lnd/lntest/wait"
 	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 )
@@ -644,7 +645,6 @@ func (h *HarnessTest) AssertStreamChannelCoopClosed(hn *node.HarnessNode,
 func (h *HarnessTest) AssertStreamChannelForceClosed(hn *node.HarnessNode,
 	cp *lnrpc.ChannelPoint, anchors bool,
 	stream rpc.CloseChanClient) *chainhash.Hash {
-
 	// Assert the channel is waiting close.
 	resp := h.AssertChannelWaitingClose(hn, cp)
 
@@ -667,6 +667,27 @@ func (h *HarnessTest) AssertStreamChannelForceClosed(hn *node.HarnessNode,
 	closingTxid := h.WaitForChannelCloseEvent(stream)
 	h.Miner.AssertTxInBlock(block, closingTxid)
 
+	// This makes sure in case of a neutrino backend that we do not have any
+	// lingering unconfirmed anchor cpfp transactions blocking some of our
+	// utxos.
+	if anchors && h.IsNeutrinoBackend() {
+		err := wait.NoError(func() error {
+			utxos := h.GetUTXOsUnconfirmed(hn,
+				lnwallet.DefaultAccountName)
+			total := len(utxos)
+			if total == 0 {
+				return nil
+			}
+
+			return fmt.Errorf("%s: assert %s failed: want %d "+
+				"got: %d", hn.Name(), "no unconfirmed cpfp "+
+				"achor sweep transactions", 0, total)
+
+		}, DefaultTimeout)
+		require.NoErrorf(hn, err, "expected no unconfirmed cpfp "+
+			"anchor sweep utxos")
+	}
+
 	// We should see zero waiting close channels and 1 pending force close
 	// channels now.
 	h.AssertNumWaitingClose(hn, 0)

sweep/sweeper.go

@@ -835,7 +835,8 @@ func (s *UtxoSweeper) collector(blockEpochs <-chan *chainntnfs.BlockEpoch) {
 // removeExclusiveGroup removes all inputs in the given exclusive group. This
 // function is called when one of the exclusive group inputs has been spent. The
 // other inputs won't ever be spendable and can be removed. This also prevents
-// them from being part of future sweep transactions that would fail.
+// them from being part of future sweep transactions that would fail. In
+// addition sweep transactions of those inputs will be removed form the wallet.
 func (s *UtxoSweeper) removeExclusiveGroup(group uint64) {
 	for outpoint, input := range s.pendingInputs {
 		outpoint := outpoint
@@ -854,7 +855,85 @@ func (s *UtxoSweeper) removeExclusiveGroup(group uint64) {
 		s.signalAndRemove(&outpoint, Result{
 			Err: ErrExclusiveGroupSpend,
 		})
+
+		// Now make sure we remove conflicting transactions
+		// from the wallet.
+		err := s.removeConflictingSweepTx(&outpoint)
+		if err != nil {
+			log.Warnf("unable to remove conflicting sweep tx from "+
+				"wallet for outpoint %v : %v", outpoint, err)
+		}
+	}
+}
+
+// removeConflictingSweepTx removes any unconfirmed sweep transaction from the
+// wallet which spends the passed outpoint. This is necessary to cleanup the
+// unmined TxStore for wallets which use neutrino as a backend.
+// Most common case is when a channel is force closed and anchor cpfp txns are
+// created to bump the initial commitment transaction. In this case an achor
+// cpfp is broadcasted for up to 3 commitment transactions (local,
+// remote-dangling, remote). Using neutrino all of those transactions will be
+// accepted (the commitment tx will be different in all of those cases) and have
+// to be removed as soon as one of them confirmes (they do have the same
+// ExclusiveGroup).
+func (s *UtxoSweeper) removeConflictingSweepTx(outpoint *wire.OutPoint) error {
+	// We need to fetch all sweep txns to check whether they spend the given
+	// outpoint.
+	pastSweepHashes, err := s.cfg.Store.ListSweeps()
+	if err != nil {
+		return err
 	}
+
+	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 {
+			// If it was removed, then we'll play it safe and mark
+			// it as no longer need to be rebroadcasted.
+			s.cfg.Wallet.CancelRebroadcast(sweepHash)
+			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 *outpoint == txIn.PreviousOutPoint {
+				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)
+			}
+
+			// If this transaction was conflicting, then we'll stop
+			// rebroadcasting it in the background.
+			s.cfg.Wallet.CancelRebroadcast(sweepHash)
+
+			// We exit early as soon as we find the sweep tx which
+			// spends the output because there can only be one tx
+			// which spends a given outpoint.
+			return nil
+		}
+	}
+
+	return nil
 }
 
 // sweepCluster tries to sweep the given input cluster.