Merge pull request #7800 from ziggie1984/neutrino-remove-sweeptx

neutrino remove sweeptx

cmd/lncli/walletrpc_active.go

@@ -75,6 +75,7 @@ func walletCommands() []cli.Command {
 				labelTxCommand,
 				publishTxCommand,
 				getTxCommand,
+				removeTxCommand,
 				releaseOutputCommand,
 				leaseOutputCommand,
 				listLeasesCommand,
@@ -545,7 +546,6 @@ func publishTransaction(ctx *cli.Context) error {
 
 	req := &walletrpc.Transaction{
 		TxHex: tx,
-		Label: ctx.String("label"),
 	}
 
 	_, err = walletClient.PublishTransaction(ctxc, req)
@@ -599,6 +599,66 @@ func getTransaction(ctx *cli.Context) error {
 	return nil
 }
 
+var removeTxCommand = cli.Command{
+	Name: "removetx",
+	Usage: "Attempts to remove the unconfirmed transaction with the " +
+		"specified txid and all its children from the underlying " +
+		"internal wallet.",
+	ArgsUsage: "txid",
+	Description: `
+	Removes the transaction with the specified txid from the underlying 
+	wallet which must still be unconfirmmed (in mempool). This command is 
+	useful when a transaction is RBFed by another transaction. The wallet 
+	will only resolve this conflict when the other transaction is mined 
+	(which can take time). If a transaction was removed erronously a simple
+	rebroadcast of the former transaction with the "publishtx" cmd will
+	register the relevant outputs of the raw tx again with the wallet
+	(if there are no errors broadcasting this transaction due to an RBF
+	replacement sitting in the mempool). As soon as a removed transaction
+	is confirmed funds will be registered with the wallet again.`,
+	Flags:  []cli.Flag{},
+	Action: actionDecorator(removeTransaction),
+}
+
+func removeTransaction(ctx *cli.Context) error {
+	ctxc := getContext()
+
+	// Display the command's help message if we do not have the expected
+	// number of arguments/flags.
+	if ctx.NArg() != 1 {
+		return cli.ShowCommandHelp(ctx, "removetx")
+	}
+
+	// Fetch the only cmd argument which must be a valid txid.
+	txid := ctx.Args().First()
+	txHash, err := chainhash.NewHashFromStr(txid)
+	if err != nil {
+		return err
+	}
+
+	walletClient, cleanUp := getWalletClient(ctx)
+	defer cleanUp()
+
+	req := &walletrpc.GetTransactionRequest{
+		Txid: txHash.String(),
+	}
+
+	resp, err := walletClient.RemoveTransaction(ctxc, req)
+	if err != nil {
+		return err
+	}
+
+	printJSON(&struct {
+		Status string `json:"status"`
+		TxID   string `json:"txid"`
+	}{
+		Status: resp.GetStatus(),
+		TxID:   txHash.String(),
+	})
+
+	return nil
+}
+
 // utxoLease contains JSON annotations for a lease on an unspent output.
 type utxoLease struct {
 	ID         string   `json:"id"`

docs/code_formatting_rules.md

@@ -203,6 +203,9 @@ func foo(a, b,
 
 func baz(a, b, c) (d,
 	error) {
+
+func longFunctionName(
+	a, b, c) (d, error) {
 ```
 
 If a function declaration spans multiple lines the body should start with an

docs/release-notes/release-notes-0.18.0.md

@@ -44,6 +44,14 @@
  
 * [Ensure that a valid SCID](https://github.com/lightningnetwork/lnd/pull/8171) 
   is used when marking a zombie edge as live.
+  
+* [Remove sweep transactions of the
+  same exclusive group](https://github.com/lightningnetwork/lnd/pull/7800).
+  When using neutrino as a backend unconfirmed transactions have to be
+  removed from the wallet when a conflicting tx is confirmed. For other backends
+  these unconfirmed transactions are already removed. In addition a new 
+  walletrpc endpoint `RemoveTransaction` is introduced which let one easily
+  remove unconfirmed transaction manually.
 
 # New Features
 ## Functional Enhancements

itest/list_on_test.go

@@ -558,4 +558,8 @@ var allTestCases = []*lntest.TestCase{
 		Name:     "query blinded route",
 		TestFunc: testQueryBlindedRoutes,
 	},
+	{
+		Name:     "removetx",
+		TestFunc: testRemoveTx,
+	},
 }

itest/lnd_onchain_test.go

@@ -218,7 +218,7 @@ func testCPFP(ht *lntest.HarnessTest) {
 	runCPFP(ht, ht.Alice, ht.Bob)
 }
 
-// runCPFP ensures that the daemon can bump an unconfirmed  transaction's fee
+// runCPFP ensures that the daemon can bump an unconfirmed transaction's fee
 // rate by broadcasting a Child-Pays-For-Parent (CPFP) transaction.
 func runCPFP(ht *lntest.HarnessTest, alice, bob *node.HarnessNode) {
 	// Skip this test for neutrino, as it's not aware of mempool
@@ -731,3 +731,104 @@ func genAnchorSweep(ht *lntest.HarnessTest,
 
 	return btcutil.NewTx(tx)
 }
+
+// testRemoveTx tests that we are able to remove an unconfirmed transaction
+// from the internal wallet as long as the tx is still unconfirmed. This test
+// also verifies that after the tx is removed (while unconfirmed) it will show
+// up as confirmed as soon as the original transaction is mined.
+func testRemoveTx(ht *lntest.HarnessTest) {
+	// Create a new node so that we start with no funds on the internal
+	// wallet.
+	alice := ht.NewNode("Alice", nil)
+
+	const initialWalletAmt = btcutil.SatoshiPerBitcoin
+
+	// Funding the node with an initial balance.
+	ht.FundCoins(initialWalletAmt, alice)
+
+	// Create an address for Alice to send the coins to.
+	req := &lnrpc.NewAddressRequest{
+		Type: lnrpc.AddressType_WITNESS_PUBKEY_HASH,
+	}
+	resp := alice.RPC.NewAddress(req)
+
+	// We send half the amount to that address generating two unconfirmed
+	// outpoints in our internal wallet.
+	sendReq := &lnrpc.SendCoinsRequest{
+		Addr:   resp.Address,
+		Amount: initialWalletAmt / 2,
+	}
+	alice.RPC.SendCoins(sendReq)
+	txID := ht.Miner.AssertNumTxsInMempool(1)[0]
+
+	// Make sure the unspent number of utxos is 2 and the unconfirmed
+	// balances add up.
+	unconfirmed := ht.GetUTXOsUnconfirmed(
+		alice, lnwallet.DefaultAccountName,
+	)
+	require.Lenf(ht, unconfirmed, 2, "number of unconfirmed tx")
+
+	// Get the raw transaction to calculate the exact fee.
+	tx := ht.Miner.GetNumTxsFromMempool(1)[0]
+
+	// Calculate the tx fee so we can compare the end amounts. We are
+	// sending from the internal wallet to the internal wallet so only
+	// the tx fee applies when calucalting the final amount of the wallet.
+	txFee := ht.CalculateTxFee(tx)
+
+	// All of alice's balance is unconfirmed and equals the initial amount
+	// minus the tx fee.
+	aliceBalResp := alice.RPC.WalletBalance()
+	expectedAmt := btcutil.Amount(initialWalletAmt) - txFee
+	require.EqualValues(ht, expectedAmt, aliceBalResp.UnconfirmedBalance)
+
+	// Now remove the transaction. We should see that the wallet state
+	// equals the amount prior to sending the transaction. It is important
+	// to understand that we do not remove any transaction from the mempool
+	// (thats not possible in reality) we just remove it from our local
+	// store.
+	var buf bytes.Buffer
+	require.NoError(ht, tx.Serialize(&buf))
+	alice.RPC.RemoveTransaction(&walletrpc.GetTransactionRequest{
+		Txid: txID.String(),
+	})
+
+	// Verify that the balance equals the initial state.
+	confirmed := ht.GetUTXOsConfirmed(
+		alice, lnwallet.DefaultAccountName,
+	)
+	require.Lenf(ht, confirmed, 1, "number confirmed tx")
+
+	// Alice's balance should be the initial balance now because all the
+	// unconfirmed tx got removed.
+	aliceBalResp = alice.RPC.WalletBalance()
+	expectedAmt = btcutil.Amount(initialWalletAmt)
+	require.EqualValues(ht, expectedAmt, aliceBalResp.ConfirmedBalance)
+
+	// Mine a block and make sure the transaction previously broadcasted
+	// shows up in alice's wallet although we removed the transaction from
+	// the wallet when it was unconfirmed.
+	block := ht.Miner.MineBlocks(1)[0]
+	ht.Miner.AssertTxInBlock(block, txID)
+
+	// Verify that alice has 2 confirmed unspent utxos in her default
+	// wallet.
+	err := wait.NoError(func() error {
+		confirmed = ht.GetUTXOsConfirmed(
+			alice, lnwallet.DefaultAccountName,
+		)
+		if len(confirmed) != 2 {
+			return fmt.Errorf("expected 2 confirmed tx, "+
+				" got %v", len(confirmed))
+		}
+
+		return nil
+	}, lntest.DefaultTimeout)
+	require.NoError(ht, err, "timeout checking for confirmed utxos")
+
+	// The remaining balance should equal alice's starting balance minus the
+	// tx fee.
+	aliceBalResp = alice.RPC.WalletBalance()
+	expectedAmt = btcutil.Amount(initialWalletAmt) - txFee
+	require.EqualValues(ht, expectedAmt, aliceBalResp.ConfirmedBalance)
+}

lnrpc/walletrpc/walletkit.pb.gw.go

@@ -602,6 +602,40 @@ func local_request_WalletKit_PublishTransaction_0(ctx context.Context, marshaler
 
 }
 
+func request_WalletKit_RemoveTransaction_0(ctx context.Context, marshaler runtime.Marshaler, client WalletKitClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var protoReq GetTransactionRequest
+	var metadata runtime.ServerMetadata
+
+	newReader, berr := utilities.IOReaderFactory(req.Body)
+	if berr != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", berr)
+	}
+	if err := marshaler.NewDecoder(newReader()).Decode(&protoReq); err != nil && err != io.EOF {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+
+	msg, err := client.RemoveTransaction(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+
+}
+
+func local_request_WalletKit_RemoveTransaction_0(ctx context.Context, marshaler runtime.Marshaler, server WalletKitServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var protoReq GetTransactionRequest
+	var metadata runtime.ServerMetadata
+
+	newReader, berr := utilities.IOReaderFactory(req.Body)
+	if berr != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", berr)
+	}
+	if err := marshaler.NewDecoder(newReader()).Decode(&protoReq); err != nil && err != io.EOF {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+
+	msg, err := server.RemoveTransaction(ctx, &protoReq)
+	return msg, metadata, err
+
+}
+
 func request_WalletKit_SendOutputs_0(ctx context.Context, marshaler runtime.Marshaler, client WalletKitClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
 	var protoReq SendOutputsRequest
 	var metadata runtime.ServerMetadata
@@ -1309,6 +1343,29 @@ func RegisterWalletKitHandlerServer(ctx context.Context, mux *runtime.ServeMux,
 
 	})
 
+	mux.Handle("POST", pattern_WalletKit_RemoveTransaction_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		rctx, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/walletrpc.WalletKit/RemoveTransaction", runtime.WithHTTPPathPattern("/v2/wallet/removetx"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_WalletKit_RemoveTransaction_0(rctx, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		ctx = runtime.NewServerMetadataContext(ctx, md)
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+
+		forward_WalletKit_RemoveTransaction_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+	})
+
 	mux.Handle("POST", pattern_WalletKit_SendOutputs_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
 		ctx, cancel := context.WithCancel(req.Context())
 		defer cancel()
@@ -1897,6 +1954,26 @@ func RegisterWalletKitHandlerClient(ctx context.Context, mux *runtime.ServeMux,
 
 	})
 
+	mux.Handle("POST", pattern_WalletKit_RemoveTransaction_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		rctx, err := runtime.AnnotateContext(ctx, mux, req, "/walletrpc.WalletKit/RemoveTransaction", runtime.WithHTTPPathPattern("/v2/wallet/removetx"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_WalletKit_RemoveTransaction_0(rctx, inboundMarshaler, client, req, pathParams)
+		ctx = runtime.NewServerMetadataContext(ctx, md)
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+
+		forward_WalletKit_RemoveTransaction_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+	})
+
 	mux.Handle("POST", pattern_WalletKit_SendOutputs_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
 		ctx, cancel := context.WithCancel(req.Context())
 		defer cancel()
@@ -2115,6 +2192,8 @@ var (
 
 	pattern_WalletKit_PublishTransaction_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v2", "wallet", "tx"}, ""))
 
+	pattern_WalletKit_RemoveTransaction_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v2", "wallet", "removetx"}, ""))
+
 	pattern_WalletKit_SendOutputs_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v2", "wallet", "send"}, ""))
 
 	pattern_WalletKit_EstimateFee_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 1, 0, 4, 1, 5, 3}, []string{"v2", "wallet", "estimatefee", "conf_target"}, ""))
@@ -2169,6 +2248,8 @@ var (
 
 	forward_WalletKit_PublishTransaction_0 = runtime.ForwardResponseMessage
 
+	forward_WalletKit_RemoveTransaction_0 = runtime.ForwardResponseMessage
+
 	forward_WalletKit_SendOutputs_0 = runtime.ForwardResponseMessage
 
 	forward_WalletKit_EstimateFee_0 = runtime.ForwardResponseMessage

lnrpc/walletrpc/walletkit.pb.json.go

@@ -447,6 +447,31 @@ func RegisterWalletKitJSONCallbacks(registry map[string]func(ctx context.Context
 		callback(string(respBytes), nil)
 	}
 
+	registry["walletrpc.WalletKit.RemoveTransaction"] = func(ctx context.Context,
+		conn *grpc.ClientConn, reqJSON string, callback func(string, error)) {
+
+		req := &GetTransactionRequest{}
+		err := marshaler.Unmarshal([]byte(reqJSON), req)
+		if err != nil {
+			callback("", err)
+			return
+		}
+
+		client := NewWalletKitClient(conn)
+		resp, err := client.RemoveTransaction(ctx, req)
+		if err != nil {
+			callback("", err)
+			return
+		}
+
+		respBytes, err := marshaler.Marshal(resp)
+		if err != nil {
+			callback("", err)
+			return
+		}
+		callback(string(respBytes), nil)
+	}
+
 	registry["walletrpc.WalletKit.SendOutputs"] = func(ctx context.Context,
 		conn *grpc.ClientConn, reqJSON string, callback func(string, error)) {
 

lnrpc/walletrpc/walletkit.proto

@@ -208,6 +208,13 @@ service WalletKit {
     */
     rpc PublishTransaction (Transaction) returns (PublishResponse);
 
+    /* lncli: `wallet removetx`
+    RemoveTransaction attempts to remove the provided transaction from the
+    internal transaction store of the wallet.
+    */
+    rpc RemoveTransaction (GetTransactionRequest)
+        returns (RemoveTransactionResponse);
+
     /*
     SendOutputs is similar to the existing sendmany call in Bitcoind, and
     allows the caller to create a transaction that sends to several outputs at
@@ -751,6 +758,7 @@ message Transaction {
     */
     string label = 2;
 }
+
 message PublishResponse {
     /*
     If blank, then no error occurred and the transaction was successfully
@@ -762,6 +770,11 @@ message PublishResponse {
     string publish_error = 1;
 }
 
+message RemoveTransactionResponse {
+    // The status of the remove transaction operation.
+    string status = 1;
+}
+
 message SendOutputsRequest {
     /*
     The number of satoshis per kilo weight that should be used when crafting

lnrpc/walletrpc/walletkit.swagger.json

@@ -506,6 +506,39 @@
         ]
       }
     },
+    "/v2/wallet/removetx": {
+      "post": {
+        "summary": "lncli: `wallet removetx`\nRemoveTransaction attempts to remove the provided transaction from the\ninternal transaction store of the wallet.",
+        "operationId": "WalletKit_RemoveTransaction",
+        "responses": {
+          "200": {
+            "description": "A successful response.",
+            "schema": {
+              "$ref": "#/definitions/walletrpcRemoveTransactionResponse"
+            }
+          },
+          "default": {
+            "description": "An unexpected error response.",
+            "schema": {
+              "$ref": "#/definitions/rpcStatus"
+            }
+          }
+        },
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/walletrpcGetTransactionRequest"
+            }
+          }
+        ],
+        "tags": [
+          "WalletKit"
+        ]
+      }
+    },
     "/v2/wallet/reserve": {
       "get": {
         "summary": "lncli: `wallet requiredreserve`\nRequiredReserve returns the minimum amount of satoshis that should be kept\nin the wallet in order to fee bump anchor channels if necessary. The value\nscales with the number of public anchor channels but is capped at a maximum.",
@@ -1432,6 +1465,15 @@
         }
       }
     },
+    "walletrpcGetTransactionRequest": {
+      "type": "object",
+      "properties": {
+        "txid": {
+          "type": "string",
+          "description": "The txid of the transaction."
+        }
+      }
+    },
     "walletrpcImportAccountRequest": {
       "type": "object",
       "properties": {
@@ -1775,6 +1817,15 @@
     "walletrpcReleaseOutputResponse": {
       "type": "object"
     },
+    "walletrpcRemoveTransactionResponse": {
+      "type": "object",
+      "properties": {
+        "status": {
+          "type": "string",
+          "description": "The status of the remove transaction operation."
+        }
+      }
+    },
     "walletrpcRequiredReserveResponse": {
       "type": "object",
       "properties": {

lnrpc/walletrpc/walletkit.yaml

@@ -73,3 +73,6 @@ http:
     - selector: walletrpc.WalletKit.VerifyMessageWithAddr
       post: "/v2/wallet/address/verifymessage"
       body: "*"
+    - selector: walletrpc.WalletKit.RemoveTransaction
+      post: "/v2/wallet/removetx"
+      body: "*"

lnrpc/walletrpc/walletkit_grpc.pb.go

@@ -156,6 +156,10 @@ type WalletKitClient interface {
 	// attempt to re-broadcast the transaction on start up, until it enters the
 	// chain.
 	PublishTransaction(ctx context.Context, in *Transaction, opts ...grpc.CallOption) (*PublishResponse, error)
+	// lncli: `wallet removetx`
+	// RemoveTransaction attempts to remove the provided transaction from the
+	// internal transaction store of the wallet.
+	RemoveTransaction(ctx context.Context, in *GetTransactionRequest, opts ...grpc.CallOption) (*RemoveTransactionResponse, error)
 	// SendOutputs is similar to the existing sendmany call in Bitcoind, and
 	// allows the caller to create a transaction that sends to several outputs at
 	// once. This is ideal when wanting to batch create a set of transactions.
@@ -420,6 +424,15 @@ func (c *walletKitClient) PublishTransaction(ctx context.Context, in *Transactio
 	return out, nil
 }
 
+func (c *walletKitClient) RemoveTransaction(ctx context.Context, in *GetTransactionRequest, opts ...grpc.CallOption) (*RemoveTransactionResponse, error) {
+	out := new(RemoveTransactionResponse)
+	err := c.cc.Invoke(ctx, "/walletrpc.WalletKit/RemoveTransaction", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *walletKitClient) SendOutputs(ctx context.Context, in *SendOutputsRequest, opts ...grpc.CallOption) (*SendOutputsResponse, error) {
 	out := new(SendOutputsResponse)
 	err := c.cc.Invoke(ctx, "/walletrpc.WalletKit/SendOutputs", in, out, opts...)
@@ -641,6 +654,10 @@ type WalletKitServer interface {
 	// attempt to re-broadcast the transaction on start up, until it enters the
 	// chain.
 	PublishTransaction(context.Context, *Transaction) (*PublishResponse, error)
+	// lncli: `wallet removetx`
+	// RemoveTransaction attempts to remove the provided transaction from the
+	// internal transaction store of the wallet.
+	RemoveTransaction(context.Context, *GetTransactionRequest) (*RemoveTransactionResponse, error)
 	// SendOutputs is similar to the existing sendmany call in Bitcoind, and
 	// allows the caller to create a transaction that sends to several outputs at
 	// once. This is ideal when wanting to batch create a set of transactions.
@@ -800,6 +817,9 @@ func (UnimplementedWalletKitServer) ImportTapscript(context.Context, *ImportTaps
 func (UnimplementedWalletKitServer) PublishTransaction(context.Context, *Transaction) (*PublishResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method PublishTransaction not implemented")
 }
+func (UnimplementedWalletKitServer) RemoveTransaction(context.Context, *GetTransactionRequest) (*RemoveTransactionResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method RemoveTransaction not implemented")
+}
 func (UnimplementedWalletKitServer) SendOutputs(context.Context, *SendOutputsRequest) (*SendOutputsResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method SendOutputs not implemented")
 }
@@ -1146,6 +1166,24 @@ func _WalletKit_PublishTransaction_Handler(srv interface{}, ctx context.Context,
 	return interceptor(ctx, in, info, handler)
 }
 
+func _WalletKit_RemoveTransaction_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(GetTransactionRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(WalletKitServer).RemoveTransaction(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/walletrpc.WalletKit/RemoveTransaction",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(WalletKitServer).RemoveTransaction(ctx, req.(*GetTransactionRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _WalletKit_SendOutputs_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(SendOutputsRequest)
 	if err := dec(in); err != nil {
@@ -1383,6 +1421,10 @@ var WalletKit_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "PublishTransaction",
 			Handler:    _WalletKit_PublishTransaction_Handler,
 		},
+		{
+			MethodName: "RemoveTransaction",
+			Handler:    _WalletKit_RemoveTransaction_Handler,
+		},
 		{
 			MethodName: "SendOutputs",
 			Handler:    _WalletKit_SendOutputs_Handler,

lnrpc/walletrpc/walletkit_server.go

@@ -172,6 +172,10 @@ var (
 			Entity: "onchain",
 			Action: "write",
 		}},
+		"/walletrpc.WalletKit/RemoveTransaction": {{
+			Entity: "onchain",
+			Action: "write",
+		}},
 	}
 
 	// DefaultWalletKitMacFilename is the default name of the wallet kit
@@ -672,6 +676,64 @@ func (w *WalletKit) PublishTransaction(ctx context.Context,
 	return &PublishResponse{}, nil
 }
 
+// RemoveTransaction attempts to remove the transaction and all of its
+// descendants resulting from further spends of the outputs of the provided
+// transaction id.
+// NOTE: We do not remove the transaction from the rebroadcaster which might
+// run in the background rebroadcasting not yet confirmed transactions. We do
+// not have access to the rebroadcaster here nor should we. This command is not
+// a way to remove transactions from the network. It is a way to shortcircuit
+// wallet utxo housekeeping while transactions are still unconfirmed and we know
+// that a transaction will never confirm because a replacement already pays
+// higher fees.
+func (w *WalletKit) RemoveTransaction(_ context.Context,
+	req *GetTransactionRequest) (*RemoveTransactionResponse, error) {
+
+	// If the client doesn't specify a hash, then there's nothing to
+	// return.
+	if req.Txid == "" {
+		return nil, fmt.Errorf("must provide a transaction hash")
+	}
+
+	txHash, err := chainhash.NewHashFromStr(req.Txid)
+	if err != nil {
+		return nil, err
+	}
+
+	// Query the tx store of our internal wallet for the specified
+	// transaction.
+	res, err := w.cfg.Wallet.GetTransactionDetails(txHash)
+	if err != nil {
+		return nil, fmt.Errorf("transaction with txid=%v not found "+
+			"in the internal wallet store", txHash)
+	}
+
+	// Only allow unconfirmed transactions to be removed because as soon
+	// as a transaction is confirmed it will be evaluated by the wallet
+	// again and the wallet state would be updated in case the user had
+	// removed the transaction accidentally.
+	if res.NumConfirmations > 0 {
+		return nil, fmt.Errorf("transaction with txid=%v is already "+
+			"confirmed (numConfs=%d) cannot be removed", txHash,
+			res.NumConfirmations)
+	}
+
+	tx := &wire.MsgTx{}
+	txReader := bytes.NewReader(res.RawTx)
+	if err := tx.Deserialize(txReader); err != nil {
+		return nil, err
+	}
+
+	err = w.cfg.Wallet.RemoveDescendants(tx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &RemoveTransactionResponse{
+		Status: "Successfully removed transaction",
+	}, nil
+}
+
 // SendOutputs is similar to the existing sendmany call in Bitcoind, and allows
 // the caller to create a transaction that sends to several outputs at once.
 // This is ideal when wanting to batch create a set of transactions.

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"
 )
@@ -693,6 +694,27 @@ func (h *HarnessTest) AssertStreamChannelForceClosed(hn *node.HarnessNode,
 	closingTxid := h.WaitForChannelCloseEvent(stream)
 	h.Miner.AssertTxInBlock(block, closingTxid)
 
+	// This makes sure that we do not have any lingering unconfirmed anchor
+	// cpfp transactions blocking some of our utxos. Especially important
+	// in case of a neutrino backend.
+	if anchors {
+		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)

lntest/rpc/wallet_kit.go

@@ -210,6 +210,22 @@ func (h *HarnessRPC) GetTransaction(
 	return resp
 }
 
+// RemoveTransaction makes an RPC call to the node's WalletKitClient and
+// asserts.
+//
+//nolint:lll
+func (h *HarnessRPC) RemoveTransaction(
+	req *walletrpc.GetTransactionRequest) *walletrpc.RemoveTransactionResponse {
+
+	ctxt, cancel := context.WithTimeout(h.runCtx, DefaultTimeout)
+	defer cancel()
+
+	resp, err := h.WalletKit.RemoveTransaction(ctxt, req)
+	h.NoError(err, "RemoveTransaction")
+
+	return resp
+}
+
 // BumpFee makes a RPC call to the node's WalletKitClient and asserts.
 func (h *HarnessRPC) BumpFee(
 	req *walletrpc.BumpFeeRequest) *walletrpc.BumpFeeResponse {

sweep/sweeper.go

@@ -506,13 +506,23 @@ 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
+// removeConflictSweepDescendants removes any transactions from the wallet that
+// spend outputs included in the passed outpoint set. 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 {
+// after 16 blocks. Moreover this is also needed for wallets which use neutrino
+// as a backend when a channel is force closed and anchor cpfp txns are
+// created to bump the initial commitment transaction. In this case an anchor
+// 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). For neutrino backends the corresponding BIP 157 serving full
+// nodes do not signal invalid transactions anymore.
+func (s *UtxoSweeper) removeConflictSweepDescendants(
+	outpoints map[wire.OutPoint]struct{}) 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
@@ -525,16 +535,6 @@ func (s *UtxoSweeper) removeLastSweepDescendants(spendingTx *wire.MsgTx) error {
 		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
@@ -561,28 +561,30 @@ func (s *UtxoSweeper) removeLastSweepDescendants(spendingTx *wire.MsgTx) error {
 		// same inputs as spendingTx.
 		var isConflicting bool
 		for _, txIn := range sweepTx.TxIn {
-			if _, ok := inputsSpent[txIn.PreviousOutPoint]; ok {
+			if _, ok := outpoints[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)
-			}
-
-			// If this transaction was conflicting, then we'll stop
-			// rebroadcasting it in the background.
-			s.cfg.Wallet.CancelRebroadcast(sweepHash)
+		if !isConflicting {
+			continue
 		}
+
+		// If it is conflicting, then we'll signal the wallet to remove
+		// all the transactions that are descendants of outputs created
+		// by the sweepTx and the sweepTx itself.
+		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)
 	}
 
 	return nil
@@ -661,7 +663,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 from the wallet.
 func (s *UtxoSweeper) removeExclusiveGroup(group uint64) {
 	for outpoint, input := range s.pendingInputs {
 		outpoint := outpoint
@@ -680,6 +683,17 @@ func (s *UtxoSweeper) removeExclusiveGroup(group uint64) {
 		s.signalAndRemove(&outpoint, Result{
 			Err: ErrExclusiveGroupSpend,
 		})
+
+		// Remove all unconfirmed transactions from the wallet which
+		// spend the passed outpoint of the same exclusive group.
+		outpoints := map[wire.OutPoint]struct{}{
+			outpoint: {},
+		}
+		err := s.removeConflictSweepDescendants(outpoints)
+		if err != nil {
+			log.Warnf("Unable to remove conflicting sweep tx from "+
+				"wallet for outpoint %v : %v", outpoint, err)
+		}
 	}
 }
 
@@ -1575,17 +1589,30 @@ func (s *UtxoSweeper) handleInputSpent(spend *chainntnfs.SpendDetail) {
 	// as well as justice transactions. In this case, we'll notify the
 	// wallet to remove any spends that descent from this output.
 	if !isOurTx {
-		err := s.removeLastSweepDescendants(spend.SpendingTx)
+		// Construct a map of the inputs this transaction spends.
+		spendingTx := spend.SpendingTx
+		inputsSpent := make(
+			map[wire.OutPoint]struct{}, len(spendingTx.TxIn),
+		)
+		for _, txIn := range spendingTx.TxIn {
+			inputsSpent[txIn.PreviousOutPoint] = struct{}{}
+		}
+
+		log.Debugf("Attempting to remove descendant txns invalidated "+
+			"by (txid=%v): %v", spendingTx.TxHash(),
+			spew.Sdump(spendingTx))
+
+		err := s.removeConflictSweepDescendants(inputsSpent)
 		if err != nil {
 			log.Warnf("unable to remove descendant transactions "+
 				"due to tx %v: ", spendHash)
 		}
 
 		log.Debugf("Detected third party spend related to in flight "+
-			"inputs (is_ours=%v): %v",
+			"inputs (is_ours=%v): %v", isOurTx,
 			newLogClosure(func() string {
 				return spew.Sdump(spend.SpendingTx)
-			}), isOurTx,
+			}),
 		)
 	}