lntest: add new utilities to wait for mempool/block inclusion, then trigger sweep if failed

This implements a generalized pattern where we'll try out assertion,
then if that fails, we'll try a sweep, then try the assertion again.
This uses the new TriggerSweep call that was added earlier.

Author: Roasbeef

lntest/harness_miner.go

@@ -121,6 +121,37 @@ func (h *HarnessTest) MineBlocksAndAssertNumTxes(num uint32,
 	return blocks
 }
 
+// MineBlocksAndAssertNumTxesWithSweep is like MineBlocksAndAssertNumTxes but
+// handles async confirmation notification races by triggering sweeps if needed.
+// Use this for tests that expect sweep transactions after force closes or other
+// events where confirmation notifications may arrive asynchronously.
+func (h *HarnessTest) MineBlocksAndAssertNumTxesWithSweep(num uint32,
+	numTxs int, hn *node.HarnessNode) []*wire.MsgBlock {
+
+	// Update the harness's current height.
+	defer h.updateCurrentHeight()
+
+	// Wait for transactions with sweep triggering support.
+	txids := h.AssertNumTxsInMempoolWithSweepTrigger(numTxs, hn)
+
+	// Mine blocks.
+	blocks := h.miner.MineBlocks(num)
+
+	// Assert that all the transactions were included in the first block.
+	for _, txid := range txids {
+		h.miner.AssertTxInBlock(blocks[0], txid)
+	}
+
+	// Make sure the mempool has been updated.
+	h.miner.AssertTxnsNotInMempool(txids)
+
+	// Finally, make sure all the active nodes are synced.
+	bestBlock := blocks[len(blocks)-1]
+	h.AssertActiveNodesSyncedTo(bestBlock.BlockHash())
+
+	return blocks
+}
+
 // ConnectMiner connects the miner with the chain backend in the network.
 func (h *HarnessTest) ConnectMiner() {
 	err := h.manager.chainBackend.ConnectMiner()
@@ -222,6 +253,16 @@ func (h *HarnessTest) AssertNumTxsInMempool(n int) []chainhash.Hash {
 	return h.miner.AssertNumTxsInMempool(n)
 }
 
+// AssertNumTxsInMempoolWithSweepTrigger waits for N transactions with sweep
+// triggering support to handle async confirmation notification races. If
+// transactions don't appear within a short timeout, it triggers a manual sweep
+// via the provided node's RPC and waits again.
+func (h *HarnessTest) AssertNumTxsInMempoolWithSweepTrigger(n int,
+	hn *node.HarnessNode) []chainhash.Hash {
+
+	return h.miner.AssertNumTxsInMempoolWithSweepTrigger(n, hn.RPC)
+}
+
 // AssertOutpointInMempool asserts a given outpoint can be found in the mempool.
 func (h *HarnessTest) AssertOutpointInMempool(op wire.OutPoint) *wire.MsgTx {
 	return h.miner.AssertOutpointInMempool(op)
@@ -240,6 +281,23 @@ func (h *HarnessTest) GetNumTxsFromMempool(n int) []*wire.MsgTx {
 	return h.miner.GetNumTxsFromMempool(n)
 }
 
+// GetNumTxsFromMempoolWithSweep gets N transactions from mempool with sweep
+// triggering support to handle async confirmation notification races. Use this
+// for tests that expect sweep transactions after force closes.
+func (h *HarnessTest) GetNumTxsFromMempoolWithSweep(n int,
+	hn *node.HarnessNode) []*wire.MsgTx {
+
+	txids := h.AssertNumTxsInMempoolWithSweepTrigger(n, hn)
+
+	var txes []*wire.MsgTx
+	for _, txid := range txids {
+		tx := h.miner.GetRawTransaction(txid)
+		txes = append(txes, tx.MsgTx())
+	}
+
+	return txes
+}
+
 // GetBestBlock makes a RPC request to miner and asserts.
 func (h *HarnessTest) GetBestBlock() (*chainhash.Hash, int32) {
 	return h.miner.GetBestBlock()

lntest/miner/miner.go

@@ -18,6 +18,7 @@ import (
 	"github.com/btcsuite/btcd/rpcclient"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/fn/v2"
+	"github.com/lightningnetwork/lnd/lnrpc/devrpc"
 	"github.com/lightningnetwork/lnd/lntest/node"
 	"github.com/lightningnetwork/lnd/lntest/wait"
 	"github.com/stretchr/testify/require"
@@ -225,6 +226,64 @@ func (h *HarnessMiner) AssertNumTxsInMempool(n int) []chainhash.Hash {
 	return mem
 }
 
+// SweeperTrigger is an interface that allows triggering a manual sweep.
+type SweeperTrigger interface {
+	TriggerSweeper(*devrpc.TriggerSweeperRequest) *devrpc.TriggerSweeperResponse
+}
+
+// AssertNumTxsInMempoolWithSweepTrigger waits for N transactions to appear in
+// the mempool. If they don't appear within a short timeout, it triggers a
+// manual sweep via the provided node's RPC and waits again. This handles the
+// async confirmation notification race where sweeps may be registered after
+// block processing completes.
+func (h *HarnessMiner) AssertNumTxsInMempoolWithSweepTrigger(n int,
+	node SweeperTrigger) []chainhash.Hash {
+
+	// First, try the fast path with a shorter timeout (5 seconds). Most
+	// tests should hit this path when confirmation notifications arrive
+	// quickly.
+	shortTimeout := 5 * time.Second
+	var mem []chainhash.Hash
+
+	err := wait.NoError(func() error {
+		mem = h.GetRawMempool()
+		if len(mem) == n {
+			return nil
+		}
+
+		return fmt.Errorf("want %v, got %v in mempool: %v",
+			n, len(mem), mem)
+	}, shortTimeout)
+
+	// If we found the transactions quickly, we're done.
+	if err == nil {
+		return mem
+	}
+
+	// Second, short timeout expired. This likely means confirmation
+	// notifications haven't arrived yet due to async processing. Trigger a
+	// manual sweep to force broadcast of any registered sweeps.
+	h.Logf("AssertNumTxsInMempoolWithSweepTrigger: timeout after %v, "+
+		"triggering sweep (want %d, got %d)", shortTimeout, n, len(mem))
+
+	node.TriggerSweeper(&devrpc.TriggerSweeperRequest{})
+
+	// Next, we'll wait again with the full timeout. The sweep should now be
+	// broadcast if confirmation notifications have arrived.
+	err = wait.NoError(func() error {
+		mem = h.GetRawMempool()
+		if len(mem) == n {
+			return nil
+		}
+
+		return fmt.Errorf("want %v, got %v in mempool: %v",
+			n, len(mem), mem)
+	}, wait.MinerMempoolTimeout)
+	require.NoError(h, err, "assert tx in mempool timeout after trigger")
+
+	return mem
+}
+
 // AssertTxInBlock asserts that a given txid can be found in the passed block.
 func (h *HarnessMiner) AssertTxInBlock(block *wire.MsgBlock,
 	txid chainhash.Hash) {

lntest/rpc/lnd.go

@@ -816,6 +816,21 @@ func (h *HarnessRPC) Quiesce(
 	return res
 }
 
+// TriggerSweeper triggers an immediate sweep of all pending inputs. This is
+// useful for tests to deterministically control when sweeps are broadcast,
+// especially when handling async confirmation notification races.
+func (h *HarnessRPC) TriggerSweeper(
+	req *devrpc.TriggerSweeperRequest) *devrpc.TriggerSweeperResponse {
+
+	ctx, cancel := context.WithTimeout(h.runCtx, DefaultTimeout)
+	defer cancel()
+
+	res, err := h.Dev.TriggerSweeper(ctx, req)
+	h.NoError(err, "TriggerSweeper returned an error")
+
+	return res
+}
+
 type PeerEventsClient lnrpc.Lightning_SubscribePeerEventsClient
 
 // SubscribePeerEvents makes a RPC call to the node's SubscribePeerEvents and