Merge pull request #6274 from Roasbeef/anchor-utxos-unconf

lnrpc+sweep: properly remove any unconfirmed descendant chains a to-be-swept input is spent

docs/release-notes/release-notes-0.15.0.md

@@ -51,6 +51,9 @@
 * [Fixed deadlock in invoice
   registry](https://github.com/lightningnetwork/lnd/pull/6332).
 
+* [Fixed an issue that would cause wallet UTXO state to be incorrect if a 3rd
+  party sweeps our anchor
+  output](https://github.com/lightningnetwork/lnd/pull/6274).
 
 ## Misc
 

lnrpc/lightning.proto

@@ -2502,6 +2502,10 @@ message WalletBalanceResponse {
     // The unconfirmed balance of a wallet(with 0 confirmations)
     int64 unconfirmed_balance = 3;
 
+    // The total amount of wallet UTXOs held in outputs that are locked for
+    // other usage.
+    int64 locked_balance = 5;
+
     // A mapping of each wallet account's name to its balance.
     map<string, WalletAccountBalance> account_balance = 4;
 }

lnrpc/lightning.swagger.json

@@ -6540,6 +6540,11 @@
           "format": "int64",
           "title": "The unconfirmed balance of a wallet(with 0 confirmations)"
         },
+        "locked_balance": {
+          "type": "string",
+          "format": "int64",
+          "description": "The total amount of wallet UTXOs held in outputs that are locked for\nother usage."
+        },
         "account_balance": {
           "type": "object",
           "additionalProperties": {

lntest/itest/assertions.go

@@ -1789,3 +1789,70 @@ func assertChannelPolicyUpdate(t *testing.T, node *lntest.HarnessNode,
 		), "error while waiting for channel update",
 	)
 }
+
+func transactionInWallet(node *lntest.HarnessNode, txid chainhash.Hash) bool {
+	txStr := txid.String()
+
+	txResp, err := node.GetTransactions(
+		context.Background(), &lnrpc.GetTransactionsRequest{},
+	)
+	if err != nil {
+		return false
+	}
+
+	for _, txn := range txResp.Transactions {
+		if txn.TxHash == txStr {
+			return true
+		}
+	}
+
+	return false
+}
+
+func assertTransactionInWallet(t *testing.T, node *lntest.HarnessNode, txID chainhash.Hash) {
+	t.Helper()
+
+	err := wait.Predicate(func() bool {
+		return transactionInWallet(node, txID)
+	}, defaultTimeout)
+	require.NoError(
+		t, err, fmt.Sprintf("transaction %v not found in wallet", txID),
+	)
+}
+
+func assertTransactionNotInWallet(t *testing.T, node *lntest.HarnessNode,
+	txID chainhash.Hash) {
+
+	t.Helper()
+
+	err := wait.Predicate(func() bool {
+		return !transactionInWallet(node, txID)
+	}, defaultTimeout)
+	require.NoError(
+		t, err, fmt.Sprintf("transaction %v found in wallet", txID),
+	)
+}
+
+func assertAnchorOutputLost(t *harnessTest, node *lntest.HarnessNode,
+	chanPoint wire.OutPoint) {
+
+	pendingChansRequest := &lnrpc.PendingChannelsRequest{}
+	err := wait.Predicate(func() bool {
+		resp, pErr := node.PendingChannels(
+			context.Background(), pendingChansRequest,
+		)
+		if pErr != nil {
+			return false
+		}
+
+		for _, pendingChan := range resp.PendingForceClosingChannels {
+			if pendingChan.Channel.ChannelPoint == chanPoint.String() {
+				return (pendingChan.Anchor ==
+					lnrpc.PendingChannelsResponse_ForceClosedChannel_LOST)
+			}
+		}
+
+		return false
+	}, defaultTimeout)
+	require.NoError(t.t, err, "anchor doesn't show as being lost")
+}

lntest/itest/lnd_channel_force_close_test.go

@@ -526,7 +526,7 @@ func channelForceClosureTest(net *lntest.NetworkHarness, t *harnessTest,
 		aliceReports[aliceAnchor.OutPoint.String()] = &lnrpc.Resolution{
 			ResolutionType: lnrpc.ResolutionType_ANCHOR,
 			Outcome:        lnrpc.ResolutionOutcome_CLAIMED,
-			SweepTxid:      aliceAnchor.SweepTx,
+			SweepTxid:      aliceAnchor.SweepTx.TxHash().String(),
 			Outpoint: &lnrpc.OutPoint{
 				TxidBytes:   aliceAnchor.OutPoint.Hash[:],
 				TxidStr:     aliceAnchor.OutPoint.Hash.String(),
@@ -632,7 +632,7 @@ func channelForceClosureTest(net *lntest.NetworkHarness, t *harnessTest,
 		carolReports[carolAnchor.OutPoint.String()] = &lnrpc.Resolution{
 			ResolutionType: lnrpc.ResolutionType_ANCHOR,
 			Outcome:        lnrpc.ResolutionOutcome_CLAIMED,
-			SweepTxid:      carolAnchor.SweepTx,
+			SweepTxid:      carolAnchor.SweepTx.TxHash().String(),
 			AmountSat:      anchorSize,
 			Outpoint: &lnrpc.OutPoint{
 				TxidBytes:   carolAnchor.OutPoint.Hash[:],
@@ -770,7 +770,7 @@ func channelForceClosureTest(net *lntest.NetworkHarness, t *harnessTest,
 			OutputIndex: carolCommit.OutPoint.Index,
 		},
 		AmountSat: uint64(pushAmt),
-		SweepTxid: carolCommit.SweepTx,
+		SweepTxid: carolCommit.SweepTx.TxHash().String(),
 	}
 
 	// Check that we can find the commitment sweep in our set of known
@@ -1337,7 +1337,7 @@ func padCLTV(cltv uint32) uint32 {
 
 type sweptOutput struct {
 	OutPoint wire.OutPoint
-	SweepTx  string
+	SweepTx  *wire.MsgTx
 }
 
 // findCommitAndAnchor looks for a commitment sweep and anchor sweep in the
@@ -1364,7 +1364,7 @@ func findCommitAndAnchor(t *harnessTest, net *lntest.NetworkHarness,
 		if len(inputs) == 1 {
 			commitSweep = &sweptOutput{
 				OutPoint: inputs[0].PreviousOutPoint,
-				SweepTx:  txHash.String(),
+				SweepTx:  tx,
 			}
 		} else {
 			// Since we have more than one input, we run through
@@ -1375,7 +1375,7 @@ func findCommitAndAnchor(t *harnessTest, net *lntest.NetworkHarness,
 				if outpointStr == closeTx {
 					anchorSweep = &sweptOutput{
 						OutPoint: txin.PreviousOutPoint,
-						SweepTx:  txHash.String(),
+						SweepTx:  tx,
 					}
 				}
 			}

lntest/itest/lnd_onchain_test.go

@@ -5,9 +5,12 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/btcsuite/btcd/btcutil"
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnrpc/walletrpc"
 	"github.com/lightningnetwork/lnd/lntest"
@@ -379,3 +382,239 @@ func testAnchorReservedValue(net *lntest.NetworkHarness, t *harnessTest) {
 		t.Fatalf("expected 1 output instead have %v", len(sweepTx.TxOut))
 	}
 }
+
+// genAnchorSweep generates a "3rd party" anchor sweeping from an existing one.
+// In practice, we just re-use the existing witness, and track on our own
+// output producing a 1-in-1-out transaction.
+func genAnchorSweep(t *harnessTest, net *lntest.NetworkHarness,
+	aliceAnchor *sweptOutput, anchorCsv uint32) *btcutil.Tx {
+
+	// At this point, we have the transaction that Alice used to try to
+	// sweep her anchor. As this is actually just something anyone can
+	// spend, just need to find the input spending the anchor output, then
+	// we can swap the output address.
+	aliceAnchorTxIn := func() wire.TxIn {
+		sweepCopy := aliceAnchor.SweepTx.Copy()
+		for _, txIn := range sweepCopy.TxIn {
+			if txIn.PreviousOutPoint == aliceAnchor.OutPoint {
+				return *txIn
+			}
+		}
+
+		t.Fatalf("anchor op not found")
+		return wire.TxIn{}
+	}()
+
+	// We'll set the signature on the input to nil, and then set the
+	// sequence to 16 (the anchor CSV period).
+	aliceAnchorTxIn.Witness[0] = nil
+	aliceAnchorTxIn.Sequence = anchorCsv
+
+	minerAddr, err := net.Miner.NewAddress()
+	if err != nil {
+		t.Fatalf("unable to get miner addr: %v", err)
+	}
+	addrScript, err := txscript.PayToAddrScript(minerAddr)
+	if err != nil {
+		t.Fatalf("unable to gen addr script: %v", err)
+	}
+
+	// Now that we have the txIn, we can just make a new transaction that
+	// uses a different script for the output.
+	tx := wire.NewMsgTx(2)
+	tx.AddTxIn(&aliceAnchorTxIn)
+	tx.AddTxOut(&wire.TxOut{
+		PkScript: addrScript,
+		Value:    anchorSize - 1,
+	})
+
+	return btcutil.NewTx(tx)
+}
+
+// testAnchorThirdPartySpend tests that if we force close a channel, but then
+// don't sweep the anchor in time and a 3rd party spends it, that we remove any
+// transactions that are a descendent of that sweep.
+func testAnchorThirdPartySpend(net *lntest.NetworkHarness, t *harnessTest) {
+	// First, we'll create two new nodes that both default to anchor
+	// channels.
+	//
+	// NOTE: The itests differ here as anchors is default off vs the normal
+	// lnd binary.
+	args := nodeArgsForCommitType(lnrpc.CommitmentType_ANCHORS)
+	alice := net.NewNode(t.t, "Alice", args)
+	defer shutdownAndAssert(net, t, alice)
+
+	bob := net.NewNode(t.t, "Bob", args)
+	defer shutdownAndAssert(net, t, bob)
+
+	ctxb := context.Background()
+	net.ConnectNodes(t.t, alice, bob)
+
+	// We'll fund our Alice with coins, as she'll be opening the channel.
+	// We'll fund her with *just* enough coins to open the channel.
+	const (
+		firstChanSize   = 1_000_000
+		anchorFeeBuffer = 500_000
+	)
+	net.SendCoins(t.t, firstChanSize, alice)
+
+	// We'll give Alice another spare UTXO as well so she can use it to
+	// help sweep all coins.
+	net.SendCoins(t.t, anchorFeeBuffer, alice)
+
+	// Open the channel between the two nodes and wait for it to confirm
+	// fully.
+	aliceChanPoint1 := openChannelAndAssert(
+		t, net, alice, bob, lntest.OpenChannelParams{
+			Amt: firstChanSize,
+		},
+	)
+
+	// With the channel open, we'll actually immediately force close it. We
+	// don't care about network announcements here since there's no routing
+	// in this test.
+	_, _, err := net.CloseChannel(alice, aliceChanPoint1, true)
+	if err != nil {
+		t.Fatalf("unable to execute force channel closure: %v", err)
+	}
+
+	// Now that the channel has been force closed, it should show up in the
+	// PendingChannels RPC under the waiting close section.
+	pendingChansRequest := &lnrpc.PendingChannelsRequest{}
+	ctxt, _ := context.WithTimeout(ctxb, defaultTimeout)
+	pendingChanResp, err := alice.PendingChannels(ctxt, pendingChansRequest)
+	if err != nil {
+		t.Fatalf("unable to query for pending channels: %v", err)
+	}
+	err = checkNumWaitingCloseChannels(pendingChanResp, 1)
+	if err != nil {
+		t.Fatalf(err.Error())
+	}
+
+	// Get the normal channel outpoint so we can track it in the set of
+	// channels that are waiting to be closed.
+	fundingTxID, err := lnrpc.GetChanPointFundingTxid(aliceChanPoint1)
+	if err != nil {
+		t.Fatalf("unable to get txid: %v", err)
+	}
+	chanPoint := wire.OutPoint{
+		Hash:  *fundingTxID,
+		Index: aliceChanPoint1.OutputIndex,
+	}
+	waitingClose, err := findWaitingCloseChannel(pendingChanResp, &chanPoint)
+	if err != nil {
+		t.Fatalf(err.Error())
+	}
+
+	// At this point, the channel is waiting close, and we have both the
+	// commitment transaction and anchor sweep in the mempool.
+	const expectedTxns = 2
+	sweepTxns, err := getNTxsFromMempool(
+		net.Miner.Client, expectedTxns, minerMempoolTimeout,
+	)
+	require.NoError(t.t, err, "no sweep txns in miner mempool")
+	aliceCloseTx := waitingClose.Commitments.LocalTxid
+	_, aliceAnchor := findCommitAndAnchor(t, net, sweepTxns, aliceCloseTx)
+
+	// We'll now mine _only_ the commitment force close transaction, as we
+	// want the anchor sweep to stay unconfirmed.
+	var emptyTime time.Time
+	forceCloseTxID, _ := chainhash.NewHashFromStr(aliceCloseTx)
+	commitTxn, err := net.Miner.Client.GetRawTransaction(
+		forceCloseTxID,
+	)
+	if err != nil {
+		t.Fatalf("unable to get transaction: %v", err)
+	}
+	_, err = net.Miner.GenerateAndSubmitBlock(
+		[]*btcutil.Tx{commitTxn}, -1, emptyTime,
+	)
+	if err != nil {
+		t.Fatalf("unable to generate block: %v", err)
+	}
+
+	// With the anchor output located, and the main commitment mined we'll
+	// instruct the wallet to send all coins in the wallet to a new address
+	// (to the miner), including unconfirmed change.
+	minerAddr, err := net.Miner.NewAddress()
+	if err != nil {
+		t.Fatalf("unable to create new miner addr: %v", err)
+	}
+	sweepReq := &lnrpc.SendCoinsRequest{
+		Addr:             minerAddr.String(),
+		SendAll:          true,
+		MinConfs:         0,
+		SpendUnconfirmed: true,
+	}
+	ctxt, _ = context.WithTimeout(ctxb, defaultTimeout)
+	sweepAllResp, err := alice.SendCoins(ctxt, sweepReq)
+	if err != nil {
+		t.Fatalf("unable to sweep coins: %v", err)
+	}
+
+	// Both the original anchor sweep transaction, as well as the
+	// transaction we created to sweep all the coins from Alice's wallet
+	// should be found in her transaction store.
+	sweepAllTxID, _ := chainhash.NewHashFromStr(sweepAllResp.Txid)
+	assertTransactionInWallet(t.t, alice, aliceAnchor.SweepTx.TxHash())
+	assertTransactionInWallet(t.t, alice, *sweepAllTxID)
+
+	// Next, we'll shutdown Alice, and allow 16 blocks to pass so that the
+	// anchor output can be swept by anyone. Rather than use the normal API
+	// call, we'll generate a series of _empty_ blocks here.
+	aliceRestart, err := net.SuspendNode(alice)
+	if err != nil {
+		t.Fatalf("unable to shutdown alice: %v", err)
+	}
+	const anchorCsv = 16
+	for i := 0; i < anchorCsv; i++ {
+		_, err := net.Miner.GenerateAndSubmitBlock(nil, -1, emptyTime)
+		if err != nil {
+			t.Fatalf("unable to generate block: %v", err)
+		}
+	}
+
+	// Before we sweep the anchor, we'll restart Alice.
+	if err := aliceRestart(); err != nil {
+		t.Fatalf("unable to restart alice: %v", err)
+	}
+
+	// Now that the channel has been closed, and Alice has an unconfirmed
+	// transaction spending the output produced by her anchor sweep, we'll
+	// mine a transaction that double spends the output.
+	thirdPartyAnchorSweep := genAnchorSweep(t, net, aliceAnchor, anchorCsv)
+	_, err = net.Miner.GenerateAndSubmitBlock(
+		[]*btcutil.Tx{thirdPartyAnchorSweep}, -1, emptyTime,
+	)
+	if err != nil {
+		t.Fatalf("unable to generate block: %v", err)
+	}
+
+	// At this point, we should no longer find Alice's transaction that
+	// tried to sweep the anchor in her wallet.
+	assertTransactionNotInWallet(t.t, alice, aliceAnchor.SweepTx.TxHash())
+
+	// In addition, the transaction she sent to sweep all her coins to the
+	// miner also should no longer be found.
+	assertTransactionNotInWallet(t.t, alice, *sweepAllTxID)
+
+	// The anchor should now show as being "lost", while the force close
+	// response is still present.
+	assertAnchorOutputLost(t, alice, chanPoint)
+
+	// At this point Alice's CSV output should already be fully spent and
+	// the channel marked as being resolved. We mine a block first, as so
+	// far we've been generating custom blocks this whole time..
+	commitSweepOp := wire.OutPoint{
+		Hash:  *forceCloseTxID,
+		Index: 1,
+	}
+	assertSpendingTxInMempool(
+		t, net.Miner.Client, minerMempoolTimeout, commitSweepOp,
+	)
+	_, err = net.Miner.Client.Generate(1)
+	if err != nil {
+		t.Fatalf("unable to generate block: %v", err)
+	}
+	assertNumPendingChannels(t, alice, 0, 0)
+}

lntest/itest/lnd_test_list_on_test.go

@@ -379,4 +379,8 @@ var allTestCases = []*testCase{
 		name: "remote signer",
 		test: testRemoteSigner,
 	},
+	{
+		name: "3rd party anchor spend",
+		test: testAnchorThirdPartySpend,
+	},
 }

lntest/mock/walletcontroller.go

@@ -237,3 +237,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/btcwallet/btcwallet.go

@@ -56,6 +56,10 @@ var (
 	// stored within the top-level waleltdb buckets of btcwallet.
 	waddrmgrNamespaceKey = []byte("waddrmgr")
 
+	// wtxmgrNamespaceKey is the namespace key that the wtxmgr state is
+	// stored within the top-level waleltdb buckets of btcwallet.
+	wtxmgrNamespaceKey = []byte("wtxmgr")
+
 	// lightningAddrSchema is the scope addr schema for all keys that we
 	// derive. We'll treat them all as p2wkh addresses, as atm we must
 	// specify a particular type.
@@ -1400,3 +1404,46 @@ func (b *BtcWallet) GetRecoveryInfo() (bool, float64, error) {
 
 	return isRecoveryMode, progress, nil
 }
+
+// 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.
+func (b *BtcWallet) FetchTx(txHash chainhash.Hash) (*wire.MsgTx, error) {
+	var targetTx *wtxmgr.TxDetails
+	err := walletdb.View(b.db, func(tx walletdb.ReadTx) error {
+		wtxmgrNs := tx.ReadBucket(wtxmgrNamespaceKey)
+		txDetails, err := b.wallet.TxStore.TxDetails(wtxmgrNs, &txHash)
+		if err != nil {
+			return err
+		}
+
+		targetTx = txDetails
+
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	if targetTx == nil {
+		return nil, nil
+	}
+
+	return &targetTx.TxRecord.MsgTx, 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.
+func (b *BtcWallet) RemoveDescendants(tx *wire.MsgTx) error {
+	txRecord, err := wtxmgr.NewTxRecordFromMsgTx(tx, time.Now())
+	if err != nil {
+		return err
+	}
+
+	return walletdb.Update(b.db, func(tx walletdb.ReadWriteTx) error {
+		wtxmgrNs := tx.ReadWriteBucket(wtxmgrNamespaceKey)
+		return b.wallet.TxStore.RemoveUnminedTx(wtxmgrNs, txRecord)
+	})
+}

lnwallet/interface.go

@@ -346,6 +346,17 @@ type WalletController interface {
 	// is 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

rpcserver.go

@@ -3087,6 +3087,27 @@ func (r *rpcServer) WalletBalance(ctx context.Context,
 		}
 	}
 
+	// Now that we have the base balance accounted for with each account,
+	// we'll look at the set of locked UTXOs to tally that as well. If we
+	// don't display this, then anytime we attempt a funding reservation,
+	// the outputs will chose as being "gone" until they're confirmed on
+	// chain.
+	var lockedBalance btcutil.Amount
+	leases, err := r.server.cc.Wallet.ListLeasedOutputs()
+	if err != nil {
+		return nil, err
+	}
+	for _, leasedOutput := range leases {
+		utxoInfo, err := r.server.cc.Wallet.FetchInputInfo(
+			&leasedOutput.Outpoint,
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		lockedBalance += utxoInfo.Value
+	}
+
 	rpcsLog.Debugf("[walletbalance] Total balance=%v (confirmed=%v, "+
 		"unconfirmed=%v)", totalBalance, confirmedBalance,
 		unconfirmedBalance)
@@ -3095,6 +3116,7 @@ func (r *rpcServer) WalletBalance(ctx context.Context,
 		TotalBalance:       int64(totalBalance),
 		ConfirmedBalance:   int64(confirmedBalance),
 		UnconfirmedBalance: int64(unconfirmedBalance),
+		LockedBalance:      int64(lockedBalance),
 		AccountBalance:     rpcAccountBalances,
 	}, nil
 }

sweep/backend_mock_test.go

@@ -151,3 +151,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/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/lnwallet"
 )
@@ -25,4 +26,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

@@ -510,6 +510,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) {
@@ -619,12 +694,29 @@ 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 {
-					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 {
+						return spew.Sdump(spend.SpendingTx)
+					}), isOurTx,
+				)
+			}
 
 			// Signal sweep results for inputs in this confirmed
 			// tx.

sweep/txgenerator.go

@@ -220,7 +220,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