tapfreighter: resolve stranded transfers against the chain

docs/release-notes/release-notes-0.8.1.md

@@ -25,6 +25,10 @@
   fixes several failure modes in the handling of force-close sweep
   transactions that could leave transfers stuck in a pending state.
 
+- [PR#2163](https://github.com/lightninglabs/taproot-assets/pull/2163)
+  resolves stranded transfers against the chain instead of leaving them
+  pending forever when their anchor transaction can no longer confirm.
+
 # New Features
 
 ## Functional Enhancements

itest/send_test.go

@@ -509,6 +509,112 @@ func testResumePendingPackageSend(t *harnessTest) {
 	}
 }
 
+// testStrandedTransferRecoversOnRestart exercises the broadcast-state
+// recovery the chain porter takes after a transfer's anchor transaction
+// confirmed on-chain but the porter died before observing the
+// confirmation. On the next startup the parcel is resumed at
+// SendStateBroadcast; the rebroadcast attempt either silently succeeds (if
+// the chain backend ignores re-publishing an already-confirmed tx) or
+// returns ErrDoubleSpend, in which case locateConfirmedInputSpend
+// identifies the parcel's own anchor as the confirmed spender and
+// transitions to SendStateWaitTxConf. Either way the historical
+// confirmation event drives the transfer to completion.
+//
+// Before the stranded-transfer recovery work, the ErrDoubleSpend branch
+// aborted the parcel and re-tried the same broadcast on every startup
+// (issue #1888). This test asserts the operational outcome — that the
+// receiver eventually sees the asset and the sender's transfer is
+// reported as complete — rather than asserting on the specific code path
+// the porter takes through the new logic, since that path varies with
+// the chain backend's handling of an already-mined tx.
+//
+// Not covered here (intentionally; left as a harness-work follow-up):
+//
+//   - The conflicting-spender resolution: where a foreign transaction
+//     reaches SafeDepth confirmations and marks the transfer superseded.
+//     Constructing a key-path P2TR spend of an asset's anchor UTXO from
+//     outside tapd requires harness tooling that doesn't exist today.
+//
+//   - The reorg-rescue path: where a 1-conf foreign spend is reorged out
+//     before reaching SafeDepth. Covered by unit tests at the
+//     waitForConfEventOnce level (see
+//     TestWaitForTransferTxConfReorgRescue in
+//     tapfreighter/chain_porter_test.go) but not end-to-end here.
+func testStrandedTransferRecoversOnRestart(t *harnessTest) {
+	ctxb := context.Background()
+
+	sendTapd := t.tapd
+
+	// Set up a receiver node.
+	recvLnd := t.lndHarness.NewNodeWithCoins("Bob", nil)
+	recvTapd := setupTapdHarness(
+		t.t, t, recvLnd, t.universeServer,
+	)
+	defer func() {
+		require.NoError(t.t, recvTapd.stop(!*noDelete))
+	}()
+
+	// Mint an asset on the sender and sync the universe state so the
+	// receiver can recognise the asset.
+	rpcAssets := MintAssetsConfirmBatch(
+		t.t, t.lndHarness.Miner(), sendTapd,
+		[]*mintrpc.MintAssetRequest{simpleAssets[0]},
+	)
+	genInfo := rpcAssets[0].AssetGenesis
+	t.syncUniverseState(sendTapd, recvTapd, len(rpcAssets))
+
+	// The receiver issues an address for the incoming transfer.
+	recvAddr, err := recvTapd.NewAddr(ctxb, &taprpc.NewAddrRequest{
+		AssetId: genInfo.AssetId,
+		Amt:     10,
+	})
+	require.NoError(t.t, err)
+	AssertAddrCreated(t.t, recvTapd, rpcAssets[0], recvAddr)
+
+	// Send the asset. The send is asynchronous; the porter broadcasts
+	// the anchor tx and then waits for confirmation.
+	label := fmt.Sprintf(
+		"stranded-recovery-%d", time.Now().UnixNano(),
+	)
+	_, _ = sendAssetsToAddrWithLabel(t, sendTapd, label, recvAddr)
+
+	// Wait for the anchor tx to land in the mempool — that confirms
+	// the porter has successfully broadcast and is now in WaitTxConf.
+	t.lndHarness.Miner().AssertNumTxsInMempool(1)
+
+	// Stop the sender. The DB persists the pending transfer; the porter
+	// goroutine that's waiting for the confirmation event is torn down.
+	t.t.Logf("Stopping sender to simulate a crash mid-transfer")
+	require.NoError(t.t, sendTapd.stop(false))
+
+	// Mine the anchor tx into a block while the sender is down. The
+	// confirmation event fires only inside lnd; the sender's porter is
+	// not running to observe it.
+	t.lndHarness.MineBlocksAndAssertNumTxes(6, 1)
+
+	// Restart the sender. resumePendingParcels enqueues the unconfirmed
+	// transfer; the parcel goroutine resumes at SendStateBroadcast and
+	// must recover from the on-chain reality (the tx already confirmed)
+	// rather than aborting.
+	t.t.Logf("Restarting sender to drive the recovery path")
+	require.NoError(t.t, sendTapd.start(false))
+
+	// The operational outcome that matters: the asset eventually
+	// reaches the receiver, and the sender's transfer is reported as
+	// complete — not stuck pending.
+	AssertNonInteractiveRecvComplete(t.t, recvTapd, 1)
+
+	transfers, err := sendTapd.ListTransfers(
+		ctxb, &taprpc.ListTransfersRequest{},
+	)
+	require.NoError(t.t, err)
+	require.Len(t.t, transfers.Transfers, 1)
+	require.NotZero(
+		t.t, transfers.Transfers[0].AnchorTxHeightHint,
+		"transfer must report a confirmed height after recovery",
+	)
+}
+
 // testBasicSendPassiveAsset tests that we can properly send assets which were
 // passive assets during a previous send.
 func testBasicSendPassiveAsset(t *harnessTest) {

itest/test_list_on_test.go

@@ -147,6 +147,10 @@ var allTestCases = []*testCase{
 		test:             testResumePendingPackageSend,
 		proofCourierType: proof.HashmailCourierType,
 	},
+	{
+		name: "stranded transfer recovers on restart",
+		test: testStrandedTransferRecoversOnRestart,
+	},
 	{
 		name:             "reattempt failed send hashmail courier",
 		test:             testReattemptFailedSendHashmailCourier,

lndservices/chain_bridge.go

@@ -94,6 +94,42 @@ func (l *LndRpcChainBridge) RegisterConfirmationsNtfn(ctx context.Context,
 	}, errChan, nil
 }
 
+// RegisterSpendNtfn registers an intent to be notified once the target
+// outpoint is spent by a confirmed transaction. If the outpoint was already
+// spent by a confirmed transaction, the notification is dispatched
+// immediately. The returned SpendEvent's Reorg channel fires when a
+// previously-reported confirmed spend is reorged out; the Spend channel
+// re-fires if and when the input is spent again on the dominant chain.
+func (l *LndRpcChainBridge) RegisterSpendNtfn(ctx context.Context,
+	outpoint *wire.OutPoint, pkScript []byte,
+	heightHint uint32) (*chainntnfs.SpendEvent, chan error, error) {
+
+	// Set up the reorg channel before issuing the registration so it can
+	// be threaded into lndclient. We expose it through the returned
+	// SpendEvent rather than as a caller-supplied parameter (mirroring
+	// chainntnfs.SpendEvent's own shape); the buffer of one matches what
+	// lndclient writes into it on each reorg.
+	reorgChan := make(chan struct{}, 1)
+
+	ctx, cancel := context.WithCancel(ctx) // nolint:govet
+	spendChan, errChan, err := l.lnd.ChainNotifier.RegisterSpendNtfn(
+		ctx, outpoint, pkScript, int32(heightHint),
+		lndclient.WithReOrgChan(reorgChan),
+	)
+	if err != nil {
+		cancel()
+
+		return nil, nil, fmt.Errorf("unable to register for spend: %w",
+			err)
+	}
+
+	return &chainntnfs.SpendEvent{
+		Spend:  spendChan,
+		Reorg:  reorgChan,
+		Cancel: cancel,
+	}, errChan, nil
+}
+
 // RegisterBlockEpochNtfn registers an intent to be notified of each new block
 // connected to the main chain.
 func (l *LndRpcChainBridge) RegisterBlockEpochNtfn(

tapcfg/config.go

@@ -123,6 +123,16 @@ const (
 	// testnet chain.
 	testnetDefaultReOrgSafeDepth = 120
 
+	// defaultSpendQueryTimeout is the default maximum time the chain
+	// porter waits for the chain notifier to report a confirmed spend of
+	// a transfer input when resolving a rejected broadcast. Historical
+	// confirmed spends are dispatched almost immediately after
+	// registration; the timeout only fires when the inputs are unspent
+	// or their spend hasn't confirmed yet. 10s suits a healthy bitcoind
+	// or btcd backend; operators with slow or rescanning backends should
+	// raise it.
+	defaultSpendQueryTimeout = 10 * time.Second
+
 	// defaultUniverseMaxQps is the default maximum number of queries per
 	// second for the universe server. This permis 100 queries per second
 	// by default.
@@ -391,6 +401,8 @@ type Config struct {
 
 	ReOrgSafeDepth int32 `long:"reorgsafedepth" description:"The number of confirmations we'll wait for before considering a transaction safely buried in the chain."`
 
+	SpendQueryTimeout time.Duration `long:"spendquerytimeout" description:"The maximum time the chain porter waits for the chain notifier to report a confirmed spend of a transfer input when resolving a rejected broadcast. Operators with slow or rescanning chain backends may want to raise this. Valid time units are {s, m, h}."`
+
 	// The following options are used to configure the proof courier.
 	DefaultProofCourierAddr string                       `long:"proofcourieraddr" description:"Default proof courier service address."`
 	HashMailCourier         *proof.HashMailCourierCfg    `group:"hashmailcourier" namespace:"hashmailcourier"`
@@ -490,6 +502,7 @@ func DefaultConfig() Config {
 		)...),
 		Prometheus:              monitoring.DefaultPrometheusConfig(),
 		ReOrgSafeDepth:          defaultReOrgSafeDepth,
+		SpendQueryTimeout:       defaultSpendQueryTimeout,
 		DefaultProofCourierAddr: defaultProofCourierAddr,
 		HashMailCourier: &proof.HashMailCourierCfg{
 			ReceiverAckTimeout: defaultProofTransferReceiverAckTimeout,

tapcfg/server.go

@@ -705,6 +705,8 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger,
 			ErrChan:                mainErrChan,
 			BurnCommitter:          supplyCommitManager,
 			DelegationKeyChecker:   addrBook,
+			SafeDepth:              cfg.ReOrgSafeDepth,
+			SpendQueryTimeout:      cfg.SpendQueryTimeout,
 		},
 	)