From 6e48683d89c6db19c0a1dfe243ed34d55594669e Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Wed, 24 Jun 2026 20:16:18 -0230 Subject: [PATCH 1/7] tapgarden+lndservices: add reorg-aware spend notifications to ChainBridge Add RegisterSpendNtfn to the ChainBridge interface, mirroring the existing RegisterConfirmationsNtfn: registrations for outpoints that were already spent by a confirmed transaction are dispatched immediately, so a spend lookup doubles as a query of historical chain state. The lnd-backed bridge wraps lndclient's RegisterSpendNtfn with a reorg channel internally and exposes it via the returned SpendEvent's Reorg field, so callers can react when a previously reported confirmed spend is reorged out. The supplycommit mock already carried a RegisterSpendNtfn method with a slightly different signature; it is adjusted to match the interface. MockChainBridge gains SetSpend, SendSpendNtfn, SendSpendReorg, SendSpendErr, FailConfFor, and ConfReqForTxid helpers so test code can target spend events, reorgs, and stream errors per outpoint, and target conf-registration failures and lookups per txid, independent of the order the porter happens to issue registrations in. --- lndservices/chain_bridge.go | 36 ++++++++ tapgarden/interface.go | 8 ++ tapgarden/mock.go | 161 +++++++++++++++++++++++++++++++++- universe/supplycommit/mock.go | 14 ++- 4 files changed, 215 insertions(+), 4 deletions(-) diff --git a/lndservices/chain_bridge.go b/lndservices/chain_bridge.go index 68e7146727..809deaaffa 100644 --- a/lndservices/chain_bridge.go +++ b/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( diff --git a/tapgarden/interface.go b/tapgarden/interface.go index 848ad3c83c..de112cc143 100644 --- a/tapgarden/interface.go +++ b/tapgarden/interface.go @@ -324,6 +324,14 @@ type ChainBridge interface { reOrgChan chan struct{}) (*chainntnfs.ConfirmationEvent, chan error, error) + // 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. + RegisterSpendNtfn(ctx context.Context, outpoint *wire.OutPoint, + pkScript []byte, heightHint uint32) (*chainntnfs.SpendEvent, + chan error, error) + // RegisterBlockEpochNtfn registers an intent to be notified of each // new block connected to the main chain. RegisterBlockEpochNtfn(ctx context.Context) (chan int32, chan error, diff --git a/tapgarden/mock.go b/tapgarden/mock.go index f651b5b45f..583afd75a0 100644 --- a/tapgarden/mock.go +++ b/tapgarden/mock.go @@ -585,6 +585,16 @@ type MockChainBridge struct { errConf atomic.Int32 emptyConf atomic.Int32 confErr chan error + + spendDetailsMtx sync.Mutex + spendDetails map[wire.OutPoint]*chainntnfs.SpendDetail + spendReorgs map[wire.OutPoint][]chan struct{} + spendChans map[wire.OutPoint][]chan *chainntnfs.SpendDetail + spendErrChans map[wire.OutPoint][]chan error + + confTxidMtx sync.Mutex + confReqByTx map[chainhash.Hash]int + confFailTx map[chainhash.Hash]bool } func NewMockChainBridge() *MockChainBridge { @@ -596,6 +606,103 @@ func NewMockChainBridge() *MockChainBridge { BlockEpochSignal: make(chan struct{}, 1), NewBlocks: make(chan int32), Blocks: make(map[chainhash.Hash]*wire.MsgBlock), + spendDetails: make( + map[wire.OutPoint]*chainntnfs.SpendDetail, + ), + spendReorgs: make(map[wire.OutPoint][]chan struct{}), + spendChans: make( + map[wire.OutPoint][]chan *chainntnfs.SpendDetail, + ), + spendErrChans: make(map[wire.OutPoint][]chan error), + confReqByTx: make(map[chainhash.Hash]int), + confFailTx: make(map[chainhash.Hash]bool), + } +} + +// FailConfFor arms the bridge so that the next call to +// RegisterConfirmationsNtfn for the given txid fails by returning a +// non-nil error directly from the call. Unlike FailConfOnce (which +// fires on the registration's error channel after the call returns +// successfully), this exercises the registration-failed path and is +// keyed by txid so tests can target a specific registration +// independent of the order in which the porter happens to issue them. +func (m *MockChainBridge) FailConfFor(txid chainhash.Hash) { + m.confTxidMtx.Lock() + defer m.confTxidMtx.Unlock() + + m.confFailTx[txid] = true +} + +// ConfReqForTxid returns the reqNo most recently used to register a +// confirmation notification for the given txid, and whether any +// registration has been seen for it. +func (m *MockChainBridge) ConfReqForTxid(txid chainhash.Hash) (int, bool) { + m.confTxidMtx.Lock() + defer m.confTxidMtx.Unlock() + + reqNo, ok := m.confReqByTx[txid] + return reqNo, ok +} + +// SetSpend registers a spend detail to be dispatched immediately whenever a +// spend notification is requested for the given outpoint. +func (m *MockChainBridge) SetSpend(op wire.OutPoint, + detail *chainntnfs.SpendDetail) { + + m.spendDetailsMtx.Lock() + defer m.spendDetailsMtx.Unlock() + + m.spendDetails[op] = detail +} + +// SendSpendReorg fans a reorg signal to every spend ntfn currently registered +// for the given outpoint, mirroring lndclient's behaviour when a +// previously-reported confirmed spend is reorged out. +func (m *MockChainBridge) SendSpendReorg(op wire.OutPoint) { + m.spendDetailsMtx.Lock() + reorgs := m.spendReorgs[op] + m.spendDetailsMtx.Unlock() + + for _, ch := range reorgs { + select { + case ch <- struct{}{}: + default: + } + } +} + +// SendSpendNtfn dispatches a spend detail to every spend ntfn currently +// registered for the given outpoint. Useful for delivering a replacement +// spender after a reorg. +func (m *MockChainBridge) SendSpendNtfn(op wire.OutPoint, + detail *chainntnfs.SpendDetail) { + + m.spendDetailsMtx.Lock() + chans := m.spendChans[op] + m.spendDetailsMtx.Unlock() + + for _, ch := range chans { + select { + case ch <- detail: + default: + } + } +} + +// SendSpendErr fans an error to every spend ntfn currently registered for +// the given outpoint, simulating a per-input stream failure. Tests use this +// to exercise the porter's per-input recovery path without affecting other +// inputs' watches. +func (m *MockChainBridge) SendSpendErr(op wire.OutPoint, err error) { + m.spendDetailsMtx.Lock() + errChans := m.spendErrChans[op] + m.spendDetailsMtx.Unlock() + + for _, ch := range errChans { + select { + case ch <- err: + default: + } } } @@ -641,7 +748,7 @@ func (m *MockChainBridge) SendConfNtfn(reqNo int, blockHash *chainhash.Hash, } func (m *MockChainBridge) RegisterConfirmationsNtfn(ctx context.Context, - _ *chainhash.Hash, _ []byte, _, _ uint32, _ bool, + txid *chainhash.Hash, _ []byte, _, _ uint32, _ bool, _ chan struct{}) (*chainntnfs.ConfirmationEvent, chan error, error) { select { @@ -663,6 +770,21 @@ func (m *MockChainBridge) RegisterConfirmationsNtfn(ctx context.Context, currentReqCount := m.ReqCount.Load() m.ConfReqs[int(currentReqCount)] = req + if txid != nil { + m.confTxidMtx.Lock() + m.confReqByTx[*txid] = int(currentReqCount) + failByTxid := m.confFailTx[*txid] + if failByTxid { + delete(m.confFailTx, *txid) + } + m.confTxidMtx.Unlock() + + if failByTxid { + return nil, nil, fmt.Errorf("confirmation "+ + "registration error for txid=%v", txid) + } + } + select { case m.ConfReqSignal <- int(currentReqCount): case <-ctx.Done(): @@ -677,6 +799,43 @@ func (m *MockChainBridge) RegisterConfirmationsNtfn(ctx context.Context, return req, m.confErr, nil } +// RegisterSpendNtfn registers an intent to be notified once the target +// outpoint is spent. If a spend detail was set for the outpoint via SetSpend, +// it is dispatched immediately. Further events for the same outpoint can be +// delivered via SendSpendNtfn (replacement spender after a reorg) or +// SendSpendReorg (reorg of the previously reported spender). +func (m *MockChainBridge) RegisterSpendNtfn(ctx context.Context, + outpoint *wire.OutPoint, _ []byte, + _ uint32) (*chainntnfs.SpendEvent, chan error, error) { + + select { + case <-ctx.Done(): + return nil, nil, fmt.Errorf("shutting down") + default: + } + + spendChan := make(chan *chainntnfs.SpendDetail, 1) + reorgChan := make(chan struct{}, 1) + errChan := make(chan error, 1) + + m.spendDetailsMtx.Lock() + if detail, ok := m.spendDetails[*outpoint]; ok { + spendChan <- detail + } + m.spendChans[*outpoint] = append(m.spendChans[*outpoint], spendChan) + m.spendReorgs[*outpoint] = append(m.spendReorgs[*outpoint], reorgChan) + m.spendErrChans[*outpoint] = append( + m.spendErrChans[*outpoint], errChan, + ) + m.spendDetailsMtx.Unlock() + + return &chainntnfs.SpendEvent{ + Spend: spendChan, + Reorg: reorgChan, + Cancel: func() {}, + }, errChan, nil +} + func (m *MockChainBridge) RegisterBlockEpochNtfn( ctx context.Context) (chan int32, chan error, error) { diff --git a/universe/supplycommit/mock.go b/universe/supplycommit/mock.go index c8b8c5c879..fa2275cf7f 100644 --- a/universe/supplycommit/mock.go +++ b/universe/supplycommit/mock.go @@ -192,13 +192,21 @@ func (m *mockChainBridge) RegisterConfirmationsNtfn( func (m *mockChainBridge) RegisterSpendNtfn(ctx context.Context, outpoint *wire.OutPoint, pkScript []byte, - heightHint uint32) (*chainntnfs.SpendEvent, error) { + heightHint uint32) (*chainntnfs.SpendEvent, chan error, error) { args := m.Called(ctx, outpoint, pkScript, heightHint) if args.Get(0) == nil { - return nil, args.Error(1) + return nil, nil, args.Error(2) + } + + // args.Get(1) may legitimately be a nil error channel in tests + // that only exercise the spend path; guard the type assertion so + // a comma-ok form keeps the mock from panicking on that case. + var errChan chan error + if got := args.Get(1); got != nil { + errChan = got.(chan error) } - return args.Get(0).(*chainntnfs.SpendEvent), args.Error(1) + return args.Get(0).(*chainntnfs.SpendEvent), errChan, args.Error(2) } func (m *mockChainBridge) PublishTransaction(ctx context.Context, From db28ff97c815746c8767552dcb3f265161c151ad Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Wed, 24 Jun 2026 20:17:13 -0230 Subject: [PATCH 2/7] tapdb: add MarkTransferSuperseded and FetchAnchorOutputPkScripts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MarkTransferSuperseded marks the unconfirmed transfer anchored by a given transaction as superseded, complementing the existing confirmation-time supersession of conflicting transfers with a way to mark a transfer whose conflict was discovered directly. A transfer whose anchor transaction has confirmed is never eligible; marking an already superseded transfer again succeeds, so concurrent marking through both paths is safe. The transfer's input assets are also marked as spent in the same DB transaction, but only those whose anchor_point appears in the caller-supplied spentOutpoints list — these are the outpoints the conflicting transaction actually consumed on-chain. A multi-input transfer whose conflict only consumed one input leaves the others available, rather than over-aggressively removing them from coin selection. Without this bookkeeping the consumed inputs would resurface as available when their leases expired, leading to ghost balances and a self-reinforcing broadcast-rejection-and-supersede loop on every retry. asset_transfer_inputs only holds rows for real asset inputs (sweep / zero-value inputs are tracked separately), so an ErrNoRows from SetAssetSpent on a listed outpoint is a genuine data anomaly — a script-key mismatch or stale record — not a benign case. Log loudly so the operator sees the inconsistency, but still continue: the on-chain reality is that the input is gone, and blocking the supersession on a stale asset record helps no one. FetchAnchorOutputPkScripts returns the pkScripts of the given anchor outpoints, indexed by outpoint. Source transactions are fetched and deserialized at most once per unique txid, so a transfer whose inputs all spend from the same prior tx incurs one chain_txn fetch rather than one per input. Used by the chain porter to register spend notifications on transfer inputs. --- tapdb/assets_store.go | 180 ++++++++++++++++++++++++++ tapdb/assets_store_test.go | 212 +++++++++++++++++++++++++++++++ tapdb/sqlc/querier.go | 6 + tapdb/sqlc/queries/transfers.sql | 16 +++ tapdb/sqlc/transfers.sql.go | 40 ++++++ tapfreighter/interface.go | 23 ++++ 6 files changed, 477 insertions(+) diff --git a/tapdb/assets_store.go b/tapdb/assets_store.go index 088c7e3b2a..9e825d1aff 100644 --- a/tapdb/assets_store.go +++ b/tapdb/assets_store.go @@ -325,6 +325,12 @@ type ActiveAssetsStore interface { SupersedeConflictingTransfers(ctx context.Context, arg sqlc.SupersedeConflictingTransfersParams) (int64, error) + // MarkTransferSuperseded marks the unconfirmed transfer anchored by + // the given transaction as superseded, returning the IDs of the + // transfers it touched. + MarkTransferSuperseded(ctx context.Context, anchorTxid []byte) ([]int64, + error) + // QuerySupersededTransferIDs returns the IDs of all transfers that // have been marked as superseded. QuerySupersededTransferIDs(ctx context.Context) ([]int64, error) @@ -3846,6 +3852,180 @@ func (a *AssetStore) QueryParcels(ctx context.Context, ) } +// MarkTransferSuperseded marks the unconfirmed transfer anchored by the given +// transaction as superseded: a conflicting transaction has confirmed spending +// (some of) its inputs, so its anchor transaction can never confirm. The +// transfer is then no longer treated as pending and isn't resumed at startup. +// +// Only inputs whose anchor_point appears in spentOutpoints are marked spent: +// these are the outpoints the conflicting transaction actually consumed +// on-chain. A multi-input transfer whose conflict only consumed one input +// leaves the others available, rather than over-aggressively removing them +// from coin selection. Without this bookkeeping the consumed inputs would +// surface as available again when their leases expired, leading to ghost +// balances and a self-reinforcing broadcast-rejection-and-supersede loop on +// every retry. +// +// An error is returned if no unconfirmed transfer with the given anchor +// transaction exists; marking an already superseded transfer again succeeds. +func (a *AssetStore) MarkTransferSuperseded(ctx context.Context, + anchorTxHash chainhash.Hash, + spentOutpoints []wire.OutPoint) error { + + // Build the set of serialized outpoint bytes the caller declared + // spent on-chain. Inputs whose anchor_point falls outside this set + // are left alone — they may still be unspent and usable. + spentSet := make(map[string]struct{}, len(spentOutpoints)) + for _, op := range spentOutpoints { + opBytes, err := encodeOutpoint(op) + if err != nil { + return fmt.Errorf("unable to encode spent outpoint "+ + "%v: %w", op, err) + } + spentSet[string(opBytes)] = struct{}{} + } + + var writeTxOpts AssetStoreTxOptions + return a.db.ExecTx(ctx, &writeTxOpts, func(q ActiveAssetsStore) error { + transferIDs, err := q.MarkTransferSuperseded( + ctx, anchorTxHash[:], + ) + if err != nil { + return fmt.Errorf("unable to mark transfer as "+ + "superseded: %w", err) + } + + if len(transferIDs) == 0 { + return fmt.Errorf("no unconfirmed transfer found "+ + "with anchor txid %v", anchorTxHash) + } + + // Mark just the inputs the conflicting tx actually consumed + // on-chain. SetAssetSpent is idempotent, so re-marking on a + // repeat call is safe. + for _, transferID := range transferIDs { + inputs, err := q.FetchTransferInputs(ctx, transferID) + if err != nil { + return fmt.Errorf("unable to fetch transfer "+ + "inputs (transfer_id=%d): %w", + transferID, err) + } + + for idx := range inputs { + input := inputs[idx] + anchorPoint := string(input.AnchorPoint) + + if _, ok := spentSet[anchorPoint]; !ok { + continue + } + + _, err := q.SetAssetSpent( + ctx, SetAssetSpentParams{ + ScriptKey: input.ScriptKey, + GenAssetID: input.AssetID, + AnchorPoint: input.AnchorPoint, + }, + ) + switch { + // asset_transfer_inputs only holds rows for + // real asset inputs (sweep / zero-value + // inputs are tracked separately), so an + // ErrNoRows here is a data anomaly — a + // script-key mismatch or a stale record — + // not a benign "untracked input" case. Log + // loudly so the operator sees the + // inconsistency, but don't block the + // supersession: the on-chain reality is + // that this input is gone. + case errors.Is(err, sql.ErrNoRows): + log.Errorf("MarkTransferSuperseded: "+ + "no asset row matches input "+ + "(transfer_id=%d, "+ + "anchor_point=%v); "+ + "continuing, but the asset "+ + "record may be out of sync", + transferID, + spew.Sdump(input.AnchorPoint)) + continue + + case err != nil: + return fmt.Errorf("unable to set "+ + "asset spent (transfer_id="+ + "%d, anchor_point=%v): %w", + transferID, + spew.Sdump(input.AnchorPoint), + err) + } + } + } + + return nil + }) +} + +// FetchAnchorOutputPkScripts returns the pkScripts of the given anchor +// outpoints, indexed by outpoint. Source transactions are fetched and +// deserialized at most once per unique txid, so a transfer whose inputs all +// spend from the same prior tx incurs one chain_txn fetch + deserialization +// rather than one per input. +func (a *AssetStore) FetchAnchorOutputPkScripts(ctx context.Context, + anchorPoints []wire.OutPoint) (map[wire.OutPoint][]byte, error) { + + result := make(map[wire.OutPoint][]byte, len(anchorPoints)) + if len(anchorPoints) == 0 { + return result, nil + } + + readOpts := NewAssetStoreReadTx() + dbErr := a.db.ExecTx(ctx, &readOpts, func(q ActiveAssetsStore) error { + txCache := make(map[chainhash.Hash]*wire.MsgTx) + + for _, op := range anchorPoints { + if _, ok := result[op]; ok { + continue + } + + tx, cached := txCache[op.Hash] + if !cached { + chainTx, err := q.FetchChainTx( + ctx, op.Hash[:], + ) + if err != nil { + return fmt.Errorf("unable to fetch "+ + "chain tx for outpoint %v: %w", + op, err) + } + + tx = &wire.MsgTx{} + err = tx.Deserialize( + bytes.NewReader(chainTx.RawTx), + ) + if err != nil { + return fmt.Errorf("unable to "+ + "deserialize chain tx for "+ + "outpoint %v: %w", op, err) + } + txCache[op.Hash] = tx + } + + if op.Index >= uint32(len(tx.TxOut)) { + return fmt.Errorf("output index %d out of "+ + "range for chain tx %v", op.Index, + op.Hash) + } + + result[op] = tx.TxOut[op.Index].PkScript + } + + return nil + }) + if dbErr != nil { + return nil, dbErr + } + + return result, nil +} + // queryParcelsWithFilters returns the set of confirmed or unconfirmed parcels // with optional time, label, and script key filters applied at the database // level. diff --git a/tapdb/assets_store_test.go b/tapdb/assets_store_test.go index b814254acf..e2dc521500 100644 --- a/tapdb/assets_store_test.go +++ b/tapdb/assets_store_test.go @@ -2124,6 +2124,218 @@ func TestAssetExportLog(t *testing.T) { require.Len(t, parcels, 0) } +// TestMarkTransferSuperseded tests that an unconfirmed transfer can be marked +// as superseded by referencing its anchor transaction, after which it is no +// longer reported as pending and its input assets are marked as spent so they +// don't surface as ghost balances, and that a confirmed transfer is never +// eligible. It also tests fetching the pkScript of a transfer input anchor +// outpoint from the stored chain transaction that created it. +func TestMarkTransferSuperseded(t *testing.T) { + t.Parallel() + + _, assetsStore, db := newAssetStore(t) + ctx := context.Background() + + // Generate a single asset, then formulate a pending transfer that + // spends it. + const numAssets = 1 + assetGen := newAssetGenerator(t, numAssets, 1) + assetGen.genAssets(t, assetsStore, []assetDesc{{ + assetGen: assetGen.assetGens[0], + anchorPoint: assetGen.anchorPoints[0], + amt: 16, + }}) + + newAnchorTx := wire.NewMsgTx(2) + newAnchorTx.AddTxIn(&wire.TxIn{}) + newAnchorTx.TxIn[0].SignatureScript = []byte{} + newAnchorTx.AddTxOut(&wire.TxOut{ + PkScript: bytes.Repeat([]byte{0x01}, 34), + Value: 1000, + }) + anchorTxHash := newAnchorTx.TxHash() + + newScriptKey := asset.NewScriptKeyBip86(keychain.KeyDescriptor{ + PubKey: test.RandPubKey(t), + KeyLocator: keychain.KeyLocator{ + Index: uint32(rand.Int31()), + Family: keychain.KeyFamily(rand.Int31()), + }, + }) + + allAssets, err := assetsStore.FetchAllAssets(ctx, true, false, nil) + require.NoError(t, err) + require.Len(t, allAssets, numAssets) + + inputAsset := allAssets[0] + inputAnchorPoint := wire.OutPoint{ + Hash: assetGen.anchorTxs[0].TxHash(), + Index: 0, + } + + newWitness := asset.Witness{ + PrevID: &asset.PrevID{}, + TxWitness: [][]byte{{0x01}, {0x02}}, + } + + spendDelta := &tapfreighter.OutboundParcel{ + AnchorTx: newAnchorTx, + AnchorTxHeightHint: 1450, + ChainFees: int64(100), + Inputs: []tapfreighter.TransferInput{{ + PrevID: asset.PrevID{ + OutPoint: inputAnchorPoint, + ID: inputAsset.ID(), + ScriptKey: asset.ToSerialized( + inputAsset.ScriptKey.PubKey, + ), + }, + Amount: inputAsset.Amount, + }}, + Outputs: []tapfreighter.TransferOutput{{ + Anchor: tapfreighter.Anchor{ + Value: 1000, + OutPoint: wire.OutPoint{ + Hash: anchorTxHash, + Index: 0, + }, + InternalKey: keychain.KeyDescriptor{ + PubKey: test.RandPubKey(t), + KeyLocator: keychain.KeyLocator{ + Family: keychain.KeyFamily( + rand.Int31(), + ), + Index: uint32( + test.RandInt[int32](), + ), + }, + }, + TaprootAssetRoot: bytes.Repeat([]byte{0x1}, 32), + MerkleRoot: bytes.Repeat([]byte{0x1}, 32), + }, + ScriptKey: newScriptKey, + ScriptKeyLocal: true, + Amount: inputAsset.Amount, + WitnessData: []asset.Witness{newWitness}, + AssetVersion: asset.V0, + ProofSuffix: bytes.Repeat([]byte{0x02}, 100), + Position: 0, + }}, + } + + leaseOwner := fn.ToArray[[32]byte](test.RandBytes(32)) + leaseExpiry := time.Now().Add(time.Hour) + require.NoError(t, assetsStore.LogPendingParcel( + ctx, spendDelta, leaseOwner, leaseExpiry, + )) + + parcels, err := assetsStore.PendingParcels(ctx) + require.NoError(t, err) + require.Len(t, parcels, 1) + + // The pkScript of the transfer input's anchor outpoint can be + // extracted from the stored chain transaction that created it. + // Asking for the same source tx twice in one call exercises the + // per-call dedupe (the chain_txn is fetched and deserialized once). + pkScripts, err := assetsStore.FetchAnchorOutputPkScripts( + ctx, []wire.OutPoint{inputAnchorPoint, inputAnchorPoint}, + ) + require.NoError(t, err) + require.Len(t, pkScripts, 1) + require.Equal( + t, assetGen.anchorTxs[0].TxOut[0].PkScript, + pkScripts[inputAnchorPoint], + ) + + // An out-of-range output index in any of the requested outpoints + // is rejected. + _, err = assetsStore.FetchAnchorOutputPkScripts( + ctx, []wire.OutPoint{{ + Hash: inputAnchorPoint.Hash, Index: 10, + }}, + ) + require.ErrorContains(t, err, "out of range") + + // An outpoint of an unknown transaction cannot provide a pkScript. + _, err = assetsStore.FetchAnchorOutputPkScripts( + ctx, []wire.OutPoint{{Hash: test.RandHash()}}, + ) + require.ErrorContains(t, err, "unable to fetch chain tx") + + // An empty input list returns an empty map. + emptyResult, err := assetsStore.FetchAnchorOutputPkScripts( + ctx, nil, + ) + require.NoError(t, err) + require.Empty(t, emptyResult) + + // Marking a transfer with an unknown anchor transaction fails. + err = assetsStore.MarkTransferSuperseded( + ctx, test.RandHash(), []wire.OutPoint{inputAnchorPoint}, + ) + require.ErrorContains(t, err, "no unconfirmed transfer") + + // Before the supersession, the input asset is unspent: an + // exclude-spent fetch (with leased assets included, since + // LogPendingParcel above leased it) still reports it. + unspent, err := assetsStore.FetchAllAssets(ctx, false, true, nil) + require.NoError(t, err) + require.Len(t, unspent, numAssets) + + // Supersede the transfer but pass an outpoint the spender did NOT + // consume. The transfer is still flagged superseded, but no input + // asset is marked spent — only conflicting-spent inputs are. + otherOp := wire.OutPoint{Hash: test.RandHash(), Index: 7} + require.NoError(t, assetsStore.MarkTransferSuperseded( + ctx, anchorTxHash, []wire.OutPoint{otherOp}, + )) + + parcels, err = assetsStore.PendingParcels(ctx) + require.NoError(t, err) + require.Len(t, parcels, 0) + + stillUnspent, err := assetsStore.FetchAllAssets(ctx, false, true, nil) + require.NoError(t, err) + require.Len(t, stillUnspent, numAssets, + "input not in spentOutpoints must NOT be marked spent") + + // Now mark superseded again, this time with the actual input + // outpoint. The asset becomes spent. + require.NoError(t, assetsStore.MarkTransferSuperseded( + ctx, anchorTxHash, []wire.OutPoint{inputAnchorPoint}, + )) + + unspent, err = assetsStore.FetchAllAssets(ctx, false, true, nil) + require.NoError(t, err) + require.Empty(t, unspent, + "input listed in spentOutpoints must be marked spent") + + allAssets, err = assetsStore.FetchAllAssets(ctx, true, true, nil) + require.NoError(t, err) + require.Len(t, allAssets, numAssets) + + // Marking it again succeeds: the operation is idempotent. + require.NoError(t, assetsStore.MarkTransferSuperseded( + ctx, anchorTxHash, []wire.OutPoint{inputAnchorPoint}, + )) + + // Once the anchor transaction confirms, the transfer can no longer be + // marked as superseded: confirmation is final. + randBlockHash := test.RandHash() + err = db.ConfirmChainAnchorTx(ctx, AnchorTxConf{ + Txid: anchorTxHash[:], + BlockHash: randBlockHash[:], + BlockHeight: sqlInt32(441), + TxIndex: sqlInt32(1), + }) + require.NoError(t, err) + + err = assetsStore.MarkTransferSuperseded( + ctx, anchorTxHash, []wire.OutPoint{inputAnchorPoint}, + ) + require.ErrorContains(t, err, "no unconfirmed transfer") +} + // TestAssetGroupWitnessUpsert tests that if you try to insert another asset // group witness with the same asset_gen_id, then only one is actually created. func TestAssetGroupWitnessUpsert(t *testing.T) { diff --git a/tapdb/sqlc/querier.go b/tapdb/sqlc/querier.go index 69f4052ec5..02a6596903 100644 --- a/tapdb/sqlc/querier.go +++ b/tapdb/sqlc/querier.go @@ -175,6 +175,12 @@ type Querier interface { // pre-commitment corresponds to an asset issuance where a remote node acted as // the issuer. MarkPreCommitSpentByOutpoint(ctx context.Context, arg MarkPreCommitSpentByOutpointParams) error + // Mark the unconfirmed transfer anchored by the given transaction as + // superseded, returning the IDs of any rows the update touched. A transfer + // whose anchor transaction has already confirmed on-chain is never eligible: + // confirmation is final. Marking an already superseded transfer again is a + // no-op (the row still matches), so the operation is idempotent. + MarkTransferSuperseded(ctx context.Context, anchorTxid []byte) ([]int64, error) NewMintingBatch(ctx context.Context, arg NewMintingBatchParams) error QueryAddr(ctx context.Context, arg QueryAddrParams) (QueryAddrRow, error) // We use a LEFT JOIN here as not every asset has a group key, so this'll diff --git a/tapdb/sqlc/queries/transfers.sql b/tapdb/sqlc/queries/transfers.sql index 0d3eb45b4c..acc7415b11 100644 --- a/tapdb/sqlc/queries/transfers.sql +++ b/tapdb/sqlc/queries/transfers.sql @@ -109,6 +109,22 @@ SELECT id FROM asset_transfers WHERE superseded = true; +-- name: MarkTransferSuperseded :many +-- Mark the unconfirmed transfer anchored by the given transaction as +-- superseded, returning the IDs of any rows the update touched. A transfer +-- whose anchor transaction has already confirmed on-chain is never eligible: +-- confirmation is final. Marking an already superseded transfer again is a +-- no-op (the row still matches), so the operation is idempotent. +UPDATE asset_transfers +SET superseded = true +WHERE anchor_txn_id IN ( + SELECT txn_id + FROM chain_txns + WHERE txid = @anchor_txid + AND block_hash IS NULL + ) +RETURNING id; + -- name: SupersedeConflictingTransfers :execrows -- Mark all unconfirmed transfers that spend the given anchor point as -- superseded, except for the given (just confirmed) transfer. Once a diff --git a/tapdb/sqlc/transfers.sql.go b/tapdb/sqlc/transfers.sql.go index f4e183907e..7b38bc8eac 100644 --- a/tapdb/sqlc/transfers.sql.go +++ b/tapdb/sqlc/transfers.sql.go @@ -461,6 +461,46 @@ func (q *Queries) LogProofTransferAttempt(ctx context.Context, arg LogProofTrans return err } +const MarkTransferSuperseded = `-- name: MarkTransferSuperseded :many +UPDATE asset_transfers +SET superseded = true +WHERE anchor_txn_id IN ( + SELECT txn_id + FROM chain_txns + WHERE txid = $1 + AND block_hash IS NULL + ) +RETURNING id +` + +// Mark the unconfirmed transfer anchored by the given transaction as +// superseded, returning the IDs of any rows the update touched. A transfer +// whose anchor transaction has already confirmed on-chain is never eligible: +// confirmation is final. Marking an already superseded transfer again is a +// no-op (the row still matches), so the operation is idempotent. +func (q *Queries) MarkTransferSuperseded(ctx context.Context, anchorTxid []byte) ([]int64, error) { + rows, err := q.db.QueryContext(ctx, MarkTransferSuperseded, anchorTxid) + if err != nil { + return nil, err + } + defer rows.Close() + var items []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + items = append(items, id) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const QueryAssetTransfers = `-- name: QueryAssetTransfers :many SELECT id, height_hint, txns.txid, txns.block_hash AS anchor_tx_block_hash, diff --git a/tapfreighter/interface.go b/tapfreighter/interface.go index 6a7b4e9b09..033f4509d5 100644 --- a/tapfreighter/interface.go +++ b/tapfreighter/interface.go @@ -693,6 +693,29 @@ type ExportLog interface { QueryCompletedParcels(ctx context.Context, startTime time.Time, filterLabel string, filterScriptKey *btcec.PublicKey) ( []*OutboundParcel, error) + + // MarkTransferSuperseded marks the unconfirmed transfer anchored by + // the given transaction as superseded: a conflicting transaction has + // confirmed spending (some of) its inputs, so its anchor transaction + // can never confirm. The transfer is then no longer treated as + // pending and isn't resumed at startup. + // + // spentOutpoints lists the outpoints actually consumed on-chain by + // the confirming conflicting transaction; only inputs whose + // anchor_point appears in this list are marked spent. A multi-input + // transfer whose conflict only consumed one input leaves the others + // available for legitimate use. + MarkTransferSuperseded(ctx context.Context, + anchorTxHash chainhash.Hash, + spentOutpoints []wire.OutPoint) error + + // FetchAnchorOutputPkScripts returns the pkScripts of the given + // anchor outpoints, indexed by outpoint. Source transactions are + // fetched at most once per unique txid, so a transfer whose inputs + // all spend from the same prior tx incurs one chain_txn fetch + // rather than one per input. + FetchAnchorOutputPkScripts(ctx context.Context, + anchorPoints []wire.OutPoint) (map[wire.OutPoint][]byte, error) } // ChainBridge aliases into the ChainBridge of the tapgarden package. From 1e29a52babff66df6f75d0a0999fa7113f9521b8 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Wed, 24 Jun 2026 20:23:05 -0230 Subject: [PATCH 3/7] tapfreighter: resolve rejected broadcasts against the chain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When broadcasting a transfer's anchor transaction was rejected as a double spend, the porter previously aborted the parcel and left the transfer pending forever: it was resumed on every startup, rejected again, and aborted again, while also inflating the unconfirmed transfer count indefinitely (the steady state observed in #1888). A double spend rejection means the chain knows something we don't, and the chain can be interrogated directly. Add a chain query (locateConfirmedInputSpend) that registers spend notifications on every transfer input, dispatching historical confirmed spends immediately, and returns the first confirmed spender it sees. Underneath it sits registerInputSpendNtfns / watchInputSpend, which fan per-input spend events, reorg signals, and stream errors into a single per-watch result channel; both helpers are reused by the confirmation-waiting state in a later commit. On ErrDoubleSpend in SendStateBroadcast: if a confirmed spender is located (our own anchor or any other transaction), transition to SendStateWaitTxConf and let that state finalise the outcome. If the lookup is inconclusive (no confirmed spend located within the short query window — the conflict may still be unconfirmed), release any fee inputs and retry the broadcast on next startup, preserving today's behaviour for that case. --- tapfreighter/chain_porter.go | 289 +++++++++++++++++++++++++++++- tapfreighter/chain_porter_test.go | 139 ++++++++++++++ 2 files changed, 420 insertions(+), 8 deletions(-) diff --git a/tapfreighter/chain_porter.go b/tapfreighter/chain_porter.go index 27b0c166db..649e616ffb 100644 --- a/tapfreighter/chain_porter.go +++ b/tapfreighter/chain_porter.go @@ -46,6 +46,13 @@ const ( // confRetryDelayMax is the maximum delay between attempts to // re-register the transfer confirmation notification. confRetryDelayMax = time.Second * 30 + + // spendQueryTimeout is the maximum time we wait for the chain + // notifier to report a confirmed spend of a transfer input. + // Historical spends are dispatched almost immediately after + // registration; the timeout only fires when the inputs are unspent + // or their spend hasn't confirmed yet. + spendQueryTimeout = time.Second * 10 ) // VerifiedProofImporter is used to import verified proofs into the local proof @@ -560,6 +567,249 @@ func (p *ChainPorter) waitForConfEventOnce(ctx context.Context, } } +// inputSpendErr carries a per-input spend notification stream failure +// alongside the outpoint it pertains to, so the caller can target recovery +// to just that input rather than tearing down the entire watch. +type inputSpendErr struct { + op wire.OutPoint + err error +} + +// spendFanin holds the shared channels every per-input spend watcher fans +// its events into. +type spendFanin struct { + spends chan *chainntnfs.SpendDetail + reorgs chan wire.OutPoint + spendErrs chan inputSpendErr +} + +// registerInputSpendNtfns registers a spend notification for each of the +// given parcel's inputs (the asset inputs, as well as any swept zero-value +// UTXOs) and fans the resulting events, reorg signals, and stream errors +// into three shared channels. Registration is all-or-nothing: if any input +// cannot be watched, an error is returned with no registrations remaining +// active. Silently dropping coverage for any input would let a foreign +// confirmed spend of that input go unnoticed, recreating the very stranding +// this machinery exists to prevent. All registrations and forwarding +// goroutines are torn down when the passed context is cancelled. +// +// The per-input goroutines loop so that a reorg of a previously-reported +// spend, followed by re-confirmation in a different block (possibly with a +// different spender) is delivered to the caller as a new spend event after +// a reorg signal. +func (p *ChainPorter) registerInputSpendNtfns(ctx context.Context, + parcel *OutboundParcel, + pkScripts map[wire.OutPoint][]byte) (*spendFanin, error) { + + // Gather the outpoints the anchor transaction spends on behalf of + // the transfer. The caller is responsible for having sourced + // pkScripts already (see pkScriptsForInputs). + numInputs := len(parcel.Inputs) + len(parcel.ZeroValueInputs) + ops := make([]wire.OutPoint, 0, numInputs) + for idx := range parcel.Inputs { + ops = append(ops, parcel.Inputs[idx].OutPoint) + } + for idx := range parcel.ZeroValueInputs { + ops = append(ops, parcel.ZeroValueInputs[idx].OutPoint) + } + + // Buffer each channel to the worst-case "one in-flight per input" + // depth. The main loop drains as it iterates, so a per-input goroutine + // never blocks on a healthy main loop. The factor of 2 absorbs the + // natural spend → reorg → spend cadence without coupling the two + // channels. The reorg channel carries the input's outpoint so the + // caller can target the right pending finality watch. + fanin := &spendFanin{ + spends: make(chan *chainntnfs.SpendDetail, len(ops)*2), + reorgs: make(chan wire.OutPoint, len(ops)*2), + spendErrs: make(chan inputSpendErr, len(ops)), + } + + // The caller (waitForConfEventOnce, locateConfirmedInputSpend) always + // cancels its ctx before returning, which tears down any goroutines + // spawned below — including those already in flight when a later + // input's registration fails and we return early. + for _, op := range ops { + pkScript, ok := pkScripts[op] + if !ok { + return nil, fmt.Errorf("missing pkScript for input "+ + "%v", op) + } + + err := p.watchInputSpend( + ctx, op, pkScript, parcel.AnchorTxHeightHint, fanin, + ) + if err != nil { + return nil, err + } + } + + return fanin, nil +} + +// pkScriptsForInputs returns a map keyed by outpoint of the pkScript of +// each of the parcel's inputs, suitable for passing to +// registerInputSpendNtfns / watchInputSpend. +// +// Asset inputs are batch-fetched from the DB so inputs sharing an anchor +// incur a single chain_txn fetch + deserialization. Zero-value (orphan-UTXO) +// sweep inputs already carry their pkScript in memory and may not even have +// their creating tx tracked in our local chain_txns (the canonical case is +// a sweep input from an external transaction), so they're sourced +// in-memory: this both avoids an unnecessary DB hit and prevents a hard +// registration failure on foreign sweep inputs. +func (p *ChainPorter) pkScriptsForInputs(ctx context.Context, + parcel *OutboundParcel) (map[wire.OutPoint][]byte, error) { + + assetOps := make([]wire.OutPoint, 0, len(parcel.Inputs)) + for idx := range parcel.Inputs { + assetOps = append(assetOps, parcel.Inputs[idx].OutPoint) + } + + pkScripts, err := p.cfg.ExportLog.FetchAnchorOutputPkScripts( + ctx, assetOps, + ) + if err != nil { + return nil, fmt.Errorf("unable to fetch pkScripts for "+ + "transfer inputs: %w", err) + } + + for idx := range parcel.ZeroValueInputs { + zvi := parcel.ZeroValueInputs[idx] + pkScripts[zvi.OutPoint] = zvi.PkScript + } + + return pkScripts, nil +} + +// watchInputSpend registers a single input's spend notification and spawns +// the per-input fan-in goroutine. Used both for initial registration (from +// registerInputSpendNtfns) and for in-flight recovery when the spend stream +// for one input flaps but the rest of the watch is healthy. The caller is +// responsible for sourcing pkScript appropriately (from the DB for asset +// inputs, from memory for zero-value sweep inputs). +func (p *ChainPorter) watchInputSpend(ctx context.Context, + op wire.OutPoint, pkScript []byte, heightHint uint32, + fanin *spendFanin) error { + + spendNtfn, errChan, err := p.cfg.ChainBridge.RegisterSpendNtfn( + ctx, &op, pkScript, heightHint, + ) + if err != nil { + return fmt.Errorf("unable to register for spend of anchor "+ + "outpoint %v: %w", op, err) + } + + go func() { + defer spendNtfn.Cancel() + + for { + select { + case spend, ok := <-spendNtfn.Spend: + if !ok || spend == nil { + return + } + + select { + case fanin.spends <- spend: + case <-ctx.Done(): + return + } + + case _, ok := <-spendNtfn.Reorg: + if !ok { + return + } + + select { + case fanin.reorgs <- op: + case <-ctx.Done(): + return + } + + case err := <-errChan: + sse := inputSpendErr{ + op: op, + err: fmt.Errorf("error whilst "+ + "waiting for spend of "+ + "anchor outpoint %v: %w", + op, err), + } + + select { + case fanin.spendErrs <- sse: + case <-ctx.Done(): + } + + return + + case <-ctx.Done(): + return + } + } + }() + + return nil +} + +// locateConfirmedInputSpend queries the chain for a confirmed transaction +// spending any of the given parcel's inputs. It returns the txid of the +// confirmed spender, which may be the parcel's own anchor transaction. A nil +// return means no confirmed spend could be located before the query timed +// out: the inputs are either unspent, or their spend hasn't confirmed yet. +func (p *ChainPorter) locateConfirmedInputSpend(ctx context.Context, + parcel *OutboundParcel) *chainhash.Hash { + + ctx, cancel := context.WithTimeout(ctx, spendQueryTimeout) + defer cancel() + + // Spends that have already confirmed are dispatched (almost) + // immediately after registration; the first one we see decides the + // fate of the transfer. Reorg signals are ignored here — this is a + // short-lived historical lookup, and any finality concerns are + // handled downstream by the confirmation-waiting state. + pkScripts, err := p.pkScriptsForInputs(ctx, parcel) + if err != nil { + log.Warnf("Unable to source pkScripts for input spend "+ + "watch, the spend lookup is inconclusive: %v", err) + + return nil + } + fanin, err := p.registerInputSpendNtfns(ctx, parcel, pkScripts) + if err != nil { + // If we can't watch every input, we can't decide the + // transfer's fate. The caller treats a nil return as + // inconclusive and falls back to its retry-on-next-startup + // path, which is the safe outcome here too. + log.Warnf("Unable to register input spend notifications, "+ + "the spend lookup is inconclusive: %v", err) + + return nil + } + + for { + select { + case spend := <-fanin.spends: + log.Debugf("Anchor outpoint %v spent by confirmed "+ + "tx %v", spend.SpentOutPoint, + spend.SpenderTxHash) + + return spend.SpenderTxHash + + case sse := <-fanin.spendErrs: + // A spend stream failed; we can no longer claim + // complete coverage, so the lookup is inconclusive. + log.Warnf("Spend stream error for %v, the spend "+ + "lookup is inconclusive: %v", sse.op, sse.err) + + return nil + + case <-ctx.Done(): + return nil + } + } +} + // storeProofs writes the updated sender and receiver proof files to the proof // archive. func (p *ChainPorter) storeProofs(sendPkg *sendPackage) error { @@ -2160,14 +2410,37 @@ func (p *ChainPorter) stateStep(currentPkg sendPackage) (*sendPackage, error) { ) switch { case errors.Is(err, lnwallet.ErrDoubleSpend): - // A double spend error means the transaction will never - // make it into the mempool or chain, so we'll never be - // able to confirm it. At this point we should probably - // put the transfer in a failed state and not re-try on - // next startup... But since we don't have that state - // yet, we just return an error here. But what we can do - // is release any fee sponsoring inputs we selected from - // lnd's wallet to avoid locking up balance. + // The transaction was rejected because an input was + // already spent, or because the transaction itself + // was already mined. The transfer's fate is a fact + // about the chain we can interrogate directly: if a + // confirmed spender (our own anchor or a conflicting + // transaction) is reported for one of our inputs, we + // transition to WaitTxConf and let the + // confirmation-waiting state finalise the outcome — + // in particular, gating supersession on the + // conflicting spender reaching SafeDepth rather than + // acting on a single confirmation here, which would + // be irreversible on a routine reorg. + spender := p.locateConfirmedInputSpend( + ctx, currentPkg.OutboundPkg, + ) + + if spender != nil { + log.Infof("Anchor tx %v: confirmed spender "+ + "of input is %v, transitioning to "+ + "WaitTxConf", txHash, spender) + + currentPkg.SendState = SendStateWaitTxConf + + return ¤tPkg, nil + } + + // No confirmed spender located within the query + // window — the conflict may still be unconfirmed, so + // we don't know the transfer's fate yet. Release any + // fee inputs we locked, and retry the broadcast on + // next startup. // // TODO(guggero): Put this transfer into a failed state // and don't retry on next startup. diff --git a/tapfreighter/chain_porter_test.go b/tapfreighter/chain_porter_test.go index 2c6f658e75..15b25527fe 100644 --- a/tapfreighter/chain_porter_test.go +++ b/tapfreighter/chain_porter_test.go @@ -1,15 +1,21 @@ package tapfreighter import ( + "context" + "fmt" "math/rand" "os" "testing" "time" + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/txscript" "github.com/btcsuite/btcd/wire" "github.com/btcsuite/btclog/v2" "github.com/lightninglabs/taproot-assets/asset" + "github.com/lightninglabs/taproot-assets/tapgarden" "github.com/lightninglabs/taproot-assets/tappsbt" + "github.com/lightningnetwork/lnd/chainntnfs" "github.com/stretchr/testify/require" ) @@ -108,3 +114,136 @@ func TestVerifySplitCommitmentWitnesses(t *testing.T) { }) } } + +// fakeExportLog is a minimal ExportLog implementation for exercising the +// porter's broadcast conflict resolution. Only the methods used by the +// porter's spend resolution are implemented; calling any other method panics +// via the embedded nil interface. +type fakeExportLog struct { + ExportLog + + pkScripts map[wire.OutPoint][]byte + superseded []chainhash.Hash + supersedeOps [][]wire.OutPoint +} + +func (f *fakeExportLog) FetchAnchorOutputPkScripts(_ context.Context, + anchorPoints []wire.OutPoint) (map[wire.OutPoint][]byte, error) { + + result := make(map[wire.OutPoint][]byte, len(anchorPoints)) + for _, op := range anchorPoints { + pkScript, ok := f.pkScripts[op] + if !ok { + return nil, fmt.Errorf("unknown anchor outpoint %v", + op) + } + result[op] = pkScript + } + + return result, nil +} + +func (f *fakeExportLog) MarkTransferSuperseded(_ context.Context, + anchorTxHash chainhash.Hash, spentOutpoints []wire.OutPoint) error { + + f.superseded = append(f.superseded, anchorTxHash) + f.supersedeOps = append(f.supersedeOps, spentOutpoints) + + return nil +} + +// TestLocateConfirmedInputSpend tests the chain query that decides the fate +// of a transfer whose anchor transaction broadcast was rejected as a double +// spend: a confirmed spend of any transfer input identifies the spender, +// while unknown outpoints or absent spends are inconclusive. +func TestLocateConfirmedInputSpend(t *testing.T) { + t.Parallel() + + var ( + spenderHash = chainhash.Hash{0x42} + opAsset = wire.OutPoint{Hash: chainhash.Hash{1}, Index: 0} + opZeroValue = wire.OutPoint{Hash: chainhash.Hash{2}, Index: 1} + ) + + parcel := &OutboundParcel{ + AnchorTxHeightHint: 100, + Inputs: []TransferInput{{ + PrevID: asset.PrevID{OutPoint: opAsset}, + }}, + ZeroValueInputs: []*ZeroValueInput{{ + OutPoint: opZeroValue, + }}, + } + + newPorter := func(exportLog ExportLog, + bridge ChainBridge) *ChainPorter { + + return NewChainPorter(&ChainPorterConfig{ + ExportLog: exportLog, + ChainBridge: bridge, + }) + } + + pkScripts := map[wire.OutPoint][]byte{ + opAsset: {txscript.OP_1}, + opZeroValue: {txscript.OP_1}, + } + + // A confirmed spend of any of the parcel's inputs (here the swept + // zero-value UTXO) identifies the spender. + t.Run("confirmed input spend found", func(t *testing.T) { + bridge := tapgarden.NewMockChainBridge() + bridge.SetSpend(opZeroValue, &chainntnfs.SpendDetail{ + SpentOutPoint: &opZeroValue, + SpenderTxHash: &spenderHash, + }) + + porter := newPorter( + &fakeExportLog{pkScripts: pkScripts}, bridge, + ) + + spender := porter.locateConfirmedInputSpend( + context.Background(), parcel, + ) + require.NotNil(t, spender) + require.Equal(t, spenderHash, *spender) + }) + + // If no spend registration succeeds, the query is inconclusive and + // returns immediately. + t.Run("unknown outpoints are inconclusive", func(t *testing.T) { + bridge := tapgarden.NewMockChainBridge() + bridge.SetSpend(opAsset, &chainntnfs.SpendDetail{ + SpentOutPoint: &opAsset, + SpenderTxHash: &spenderHash, + }) + + // The export log doesn't know any of the parcel's outpoints, + // so no spend registration is ever made. + porter := newPorter(&fakeExportLog{}, bridge) + + spender := porter.locateConfirmedInputSpend( + context.Background(), parcel, + ) + require.Nil(t, spender) + }) + + // If no confirmed spend is dispatched, the query times out and is + // inconclusive. We bound it with a deadline well below the default + // spend query timeout. + t.Run("no confirmed spend times out", func(t *testing.T) { + bridge := tapgarden.NewMockChainBridge() + + porter := newPorter( + &fakeExportLog{pkScripts: pkScripts}, bridge, + ) + + ctx, cancel := context.WithTimeout( + context.Background(), 200*time.Millisecond, + ) + defer cancel() + + spender := porter.locateConfirmedInputSpend(ctx, parcel) + require.Nil(t, spender) + }) +} From f9db14d592956914292b6487a94cf465ea297169 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Wed, 24 Jun 2026 20:24:16 -0230 Subject: [PATCH 4/7] tapfreighter: watch input spends with reorg-safe supersession MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A parcel in the confirmation-waiting state only watched for the confirmation of its own anchor transaction. If a conflicting transaction confirmed spending one of the parcel's inputs, the transfer could never confirm, yet nothing resolved it: the parcel waited forever, and the transfer stayed pending across restarts. For parcels whose anchor transaction is broadcast by an external system (lnd's sweeper for force-close sweeps, the channel peers for commitment transactions), this was the only way to learn of a lost race, since such parcels never attempt a broadcast of their own and so never reach the double spend resolution at the broadcast state. A remote channel party claiming an HTLC output ahead of our sweep, or the remote commitment confirming ahead of ours, left a permanently pending transfer. Register spend notifications for the parcel's inputs alongside the confirmation notification. Registration is all-or-nothing across the input set: silently dropping coverage for any input would let a foreign confirmed spend of the unwatched input go unnoticed, recreating the very stranding this machinery exists to prevent. A 1-confirmation foreign spend is, however, not by itself sufficient to act on. Supersession marks the transfer superseded=true and its input assets spent=true atomically in the DB, and there is no inverse operation. A routine 1-block reorg of the conflicting spender would then convert a recoverable stall into silent, permanent balance loss — strictly worse than the pre-PR behaviour. Instead, the porter registers a SafeDepth confirmation notification on the spender tx (using its own SpendingTx pkScript and SpendingHeight as lookup hints) and only calls supersedeTransfer once that finality depth is reached. SafeDepth is plumbed from the existing --reorgsafedepth config (defaults 6 mainnet, 120 testnet) and clamped to at least 1. Pending finality state is keyed per outpoint, not per parcel: two inputs with distinct conflicting spenders each get their own finality watch, so a reorg of one cannot cancel the other's verdict. Per-input spend stream errors trigger recovery of just that input's spend notification, leaving the conf ntfn and the other inputs undisturbed; tearing down everything on each per-input flap would let a chronically-erroring stream indefinitely delay an imminent own-anchor confirmation. Spend reorg signals cancel any pending finality watch on the affected input; the input's spend channel will re-fire if and when the input is spent again on the dominant chain. ErrTransferSuperseded is exported as a wrapped sentinel returned on the deliberate-supersede path. mainEventLoop checks for it via errors.Is and logs at Info instead of Error; subscribers see the same AssetSendErrorEvent shape but can branch on errors.Is to treat it as a benign terminal state rather than a real failure. Transient DB write failures keep a sentinel-free error. supersedeTransfer marks the transfer superseded in the asset store before unlocking inputs: the DB write is the durable state change; unlocking is transient (lnd's wallet re-derives availability from the chain), and mark-then-unlock preserves the invariant that the durable change always lands first. --- tapcfg/server.go | 1 + tapfreighter/chain_porter.go | 501 +++++++++++++++++++--- tapfreighter/chain_porter_test.go | 691 ++++++++++++++++++++++++++++++ 3 files changed, 1143 insertions(+), 50 deletions(-) diff --git a/tapcfg/server.go b/tapcfg/server.go index ad02714b79..090f8980f3 100644 --- a/tapcfg/server.go +++ b/tapcfg/server.go @@ -705,6 +705,7 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger, ErrChan: mainErrChan, BurnCommitter: supplyCommitManager, DelegationKeyChecker: addrBook, + SafeDepth: cfg.ReOrgSafeDepth, }, ) diff --git a/tapfreighter/chain_porter.go b/tapfreighter/chain_porter.go index 649e616ffb..737a8a2d0f 100644 --- a/tapfreighter/chain_porter.go +++ b/tapfreighter/chain_porter.go @@ -55,6 +55,18 @@ const ( spendQueryTimeout = time.Second * 10 ) +// ErrTransferSuperseded is returned by the chain porter when a transfer's +// anchor transaction can never confirm because a conflicting transaction +// has reached SafeDepth confirmations spending one of the transfer's +// inputs. This is a deliberate terminal outcome — the transfer is marked +// superseded in the asset store, its consumed inputs are marked spent, and +// the parcel goroutine exits — not a bug. Callers receiving the error from +// the porter (state-machine error channels, event subscribers) can identify +// it via errors.Is and treat it as a benign terminal state rather than a +// failure. +var ErrTransferSuperseded = errors.New("transfer superseded by " + + "conflicting confirmed spend") + // VerifiedProofImporter is used to import verified proofs into the local proof // archive after we complete a transfer. type VerifiedProofImporter interface { @@ -138,6 +150,13 @@ type ChainPorterConfig struct { // key for a given asset, which is required for creating supply // commitments. DelegationKeyChecker address.DelegationKeyChecker + + // SafeDepth is the number of confirmations a conflicting spender of a + // transfer input must reach before the porter treats the transfer as + // irreversibly superseded. Acting at a single confirmation would risk + // permanent loss of the transfer's inputs on a routine 1-block reorg. + // A zero value is clamped to 1 at construction time. + SafeDepth int32 } // ChainPorter is the main sub-system of the tapfreighter package. The porter @@ -171,6 +190,15 @@ func NewChainPorter(cfg *ChainPorterConfig) *ChainPorter { subscribers := make( map[uint64]*fn.EventReceiver[fn.Event], ) + + // Clamp the supersession finality depth to at least one confirmation. + // Zero would let RegisterConfirmationsNtfn fire on a 0-conf observation + // and reintroduce exactly the irreversibility the gate exists to + // prevent. + if cfg.SafeDepth < 1 { + cfg.SafeDepth = 1 + } + return &ChainPorter{ cfg: cfg, outboundParcels: make(chan Parcel), @@ -397,8 +425,19 @@ func (p *ChainPorter) advanceState(pkg *sendPackage, kit *parcelKit) { updatedPkg, err := p.stateStep(*pkg) if err != nil { kit.errChan <- err - log.Errorf("Error evaluating state (%v): %v", - pkg.SendState, err) + + // A supersede is a deliberate terminal outcome, not a + // bug — log at Info, not Error. The event still + // carries the wrapped sentinel so subscribers can + // branch via errors.Is(ErrTransferSuperseded) if they + // want to distinguish it from a real failure. + if errors.Is(err, ErrTransferSuperseded) { + log.Infof("Transfer terminally superseded "+ + "(state=%v): %v", pkg.SendState, err) + } else { + log.Errorf("Error evaluating state (%v): %v", + pkg.SendState, err) + } p.publishSubscriberEvent(newAssetSendErrorEvent( err, stateToExecute, *pkg, @@ -448,7 +487,7 @@ func (p *ChainPorter) waitForTransferTxConf(pkg *sendPackage) error { retryDelay := confRetryDelay for attempt := 1; ; attempt++ { - confEvent, terminal, err := p.waitForConfEventOnce( + confEvent, sf, terminal, err := p.waitForConfEventOnce( confCtx, outboundPkg, ) switch { @@ -471,6 +510,14 @@ func (p *ChainPorter) waitForTransferTxConf(pkg *sendPackage) error { return nil + // A conflicting transaction confirmed spending one of the + // parcel's inputs, so the anchor transaction can never + // confirm: the transfer is permanently superseded. + case sf != nil: + return p.supersedeTransfer( + confCtx, pkg, sf.spender, sf.consumed, + ) + // We're shutting down, or the context was cancelled; there's // no point in retrying. case terminal: @@ -511,12 +558,24 @@ func (p *ChainPorter) waitForTransferTxConf(pkg *sendPackage) error { } // waitForConfEventOnce registers a confirmation notification for the parcel's -// anchor transaction and waits for a single outcome. It returns the -// confirmation event on success. If the notification stream fails in a way -// that can be remedied by re-registering, a nil event and terminal=false are -// returned. Shutdown and context cancellation are terminal. +// anchor transaction, as well as spend notifications for the parcel's inputs, +// and waits for a single outcome. It returns the confirmation event on +// success. If a conflicting transaction is reported as the confirmed spender +// of any input and reaches SafeDepth confirmations, the spender's txid is +// returned: the anchor transaction can never confirm. A 1-conf foreign spend +// alone is insufficient — supersession is irreversible in the DB, so we wait +// until the conflicting spender is reorg-safe (see SafeDepth) before +// surfacing it. If a notification stream fails in a way that can be remedied +// by re-registering, all-nil and terminal=false are returned. Shutdown and +// context cancellation are terminal. func (p *ChainPorter) waitForConfEventOnce(ctx context.Context, - outboundPkg *OutboundParcel) (*chainntnfs.TxConfirmation, bool, error) { + outboundPkg *OutboundParcel) (*chainntnfs.TxConfirmation, + *spenderFinality, bool, error) { + + // Make sure all registrations (and their notification streams) are + // torn down once we leave this attempt, whatever the outcome. + ctx, cancel := context.WithCancel(ctx) + defer cancel() txHash := outboundPkg.AnchorTx.TxHash() confNtfn, errChan, err := p.cfg.ChainBridge.RegisterConfirmationsNtfn( @@ -526,45 +585,344 @@ func (p *ChainPorter) waitForConfEventOnce(ctx context.Context, if err != nil { // Registration itself failed, which is generally a transient // RPC issue, so we'll have the caller retry. - return nil, false, fmt.Errorf("unable to register for package "+ - "tx conf: %w", err) + return nil, nil, false, fmt.Errorf("unable to register for "+ + "package tx conf: %w", err) } - - // Make sure the registration (and its notification stream) is torn - // down once we leave this attempt, whatever the outcome. defer confNtfn.Cancel() - select { - case confEvent, ok := <-confNtfn.Confirmed: - if !ok || confEvent == nil { - return nil, false, fmt.Errorf("confirmation event "+ - "channel closed for txid=%v", txHash) + // We also watch the parcel's inputs: a confirmed spend by a + // conflicting transaction decides the fate of the transfer just as + // well as a confirmation does. This is the only exit besides + // confirmation for parcels whose anchor transaction is broadcast by + // an external system (such as the lnd sweeper), where a competing + // transaction version or a third party (such as a remote channel + // party claiming an HTLC output) may win the race for the inputs + // without the porter ever attempting a broadcast itself. + // + // Registration is all-or-nothing: if any input cannot be watched, + // the caller has to re-attempt the whole set, since a partial watch + // could miss a foreign confirmation on the unwatched input and + // strand the parcel exactly as before this fix. + pkScripts, err := p.pkScriptsForInputs(ctx, outboundPkg) + if err != nil { + return nil, nil, false, err + } + fanin, err := p.registerInputSpendNtfns(ctx, outboundPkg, pkScripts) + if err != nil { + return nil, nil, false, fmt.Errorf("unable to register for "+ + "package input spends: %w", err) + } + + // pending tracks the SafeDepth conf watch for each input outpoint + // whose latest reported spend is by a foreign (non-own) spender. + // Keying by outpoint is essential: a single shared pending slot + // would let a later conflicting spender on input B cancel an + // earlier in-flight finality watch on input A, and a subsequent + // reorg of B would then clear all state — losing supersession on + // A even though A's spender is still confirmed. + pending := make(map[wire.OutPoint]*pendingSpenderState) + defer func() { + for _, st := range pending { + st.cancel() } + }() - return confEvent, false, nil - - case err := <-errChan: - return nil, false, fmt.Errorf("error whilst waiting for "+ - "package tx confirmation: %w", err) + // finality fans the result of every per-input SafeDepth conf watch + // into a single channel keyed by (outpoint, spender). The main loop + // drops stale results: a cancel + replace beats the goroutine to the + // send, so a result whose (op, spender) no longer matches the + // pending entry is ignored. + finality := make(chan finalityResult, len(outboundPkg.Inputs)+ + len(outboundPkg.ZeroValueInputs)) - case <-ctx.Done(): - // The context is also cancelled on shutdown, in which case - // both this and the quit case below are ready at the same - // time. Prefer the graceful exit. + for { select { + case confEvent, ok := <-confNtfn.Confirmed: + if !ok || confEvent == nil { + return nil, nil, false, fmt.Errorf( + "confirmation event channel closed "+ + "for txid=%v", txHash) + } + + return confEvent, nil, false, nil + + case err := <-errChan: + return nil, nil, false, fmt.Errorf("error whilst "+ + "waiting for package tx confirmation: %w", err) + + case spend := <-fanin.spends: + spender := spend.SpenderTxHash + + // Our own anchor transaction being the confirmed + // spender means the confirmation event is imminent, + // so we keep waiting for it. + if spender == nil || *spender == txHash { + continue + } + + if spend.SpentOutPoint == nil { + continue + } + op := *spend.SpentOutPoint + + // Already tracking this spender for this input — + // nothing to do. + if st, ok := pending[op]; ok && + st.spender == *spender { + + continue + } + + // A different spender on this input (e.g. a + // post-reorg replacement) supersedes any prior + // pending watch on the same input. Distinct spenders + // on distinct inputs keep their own watches. + if st, ok := pending[op]; ok { + st.cancel() + delete(pending, op) + } + + next, err := p.watchSpenderFinality( + ctx, spend, finality, + ) + if err != nil { + // The spend was already consumed from the + // stream, so it won't re-fire absent another + // reorg. Surface a retryable error so the + // caller re-registers the whole watch set; + // re-registration will redeliver the + // historical spend and reattempt the finality + // watch. Silently logging here would strand + // the transfer indefinitely — exactly the + // failure mode this machinery exists to + // prevent. + return nil, nil, false, fmt.Errorf("unable "+ + "to watch finality of conflicting "+ + "spender %v of anchor input %v: %w", + spender, op, err) + } + + pending[op] = next + + case fr := <-finality: + // A finality watch reported in. Drop stale results + // (the entry may have been cancelled + replaced). + st, ok := pending[fr.op] + if !ok || st.spender != fr.spender { + continue + } + + if fr.err != nil { + // The spender-finality conf watch failed. + // Drop the pending entry and have the caller + // re-attempt the whole watch. + st.cancel() + delete(pending, fr.op) + + return nil, nil, false, fmt.Errorf( + "spender finality watch failed: %w", + fr.err) + } + + // SafeDepth reached: it is now safe to treat the + // transfer as superseded. Hand back the spender along + // with its full input list so MarkTransferSuperseded + // targets only the inputs actually consumed on-chain. + return nil, &spenderFinality{ + spender: fr.spender, + consumed: st.consumed, + }, false, nil + + case op := <-fanin.reorgs: + // The previously-reported confirmed spend of this + // input has been reorged out. Abandon any in-flight + // finality watch on this specific input; other + // inputs' watches are unaffected. lndclient will + // re-fire on the input's Spend channel if and when + // it is spent again on the dominant chain. + if st, ok := pending[op]; ok { + st.cancel() + delete(pending, op) + } + + case sse := <-fanin.spendErrs: + // A single per-input spend stream failed. Recover by + // re-registering only that input's spend ntfn rather + // than tearing down the entire watch — the confNtfn + // and the other inputs are still healthy, and ripping + // them down on every flap would let a chronic + // per-input error indefinitely delay an imminent + // confirmation. A pending finality watch on this + // input is also cancelled: the stream error means we + // no longer trust our coverage of further reorg + // events on it, so the safe move is to start fresh. + log.Warnf("Per-input spend ntfn stream for %v "+ + "failed: %v; re-registering just this input", + sse.op, sse.err) + + if st, ok := pending[sse.op]; ok { + st.cancel() + delete(pending, sse.op) + } + + pkScript, ok := pkScripts[sse.op] + if !ok { + return nil, nil, false, fmt.Errorf("no "+ + "pkScript for failed input %v", + sse.op) + } + err := p.watchInputSpend( + ctx, sse.op, pkScript, + outboundPkg.AnchorTxHeightHint, fanin, + ) + if err != nil { + // Re-registration failed — surface a + // retryable error so the outer backoff loop + // re-attempts the whole watch. + return nil, nil, false, fmt.Errorf("unable "+ + "to re-register spend ntfn for "+ + "%v: %w", sse.op, err) + } + + case <-ctx.Done(): + // The context is also cancelled on shutdown, in which + // case both this and the quit case below are ready at + // the same time. Prefer the graceful exit. + select { + case <-p.Quit: + log.Debugf("Skipping TX confirmation, exiting") + return nil, nil, true, nil + default: + } + + return nil, nil, true, fmt.Errorf("context done "+ + "whilst waiting for package tx confirmation "+ + "of %v", txHash) + case <-p.Quit: log.Debugf("Skipping TX confirmation, exiting") - return nil, true, nil - default: + return nil, nil, true, nil } + } +} - return nil, true, fmt.Errorf("context done whilst waiting "+ - "for package tx confirmation of %v", txHash) +// pendingSpenderState holds the per-input resources for a SafeDepth-conf +// watch on a conflicting spender of one of the parcel's inputs. consumed +// records every outpoint the spender consumed on-chain so we can target +// asset-spent marking to only those inputs rather than the whole transfer. +type pendingSpenderState struct { + spender chainhash.Hash + consumed []wire.OutPoint + cancel func() +} - case <-p.Quit: - log.Debugf("Skipping TX confirmation, exiting") - return nil, true, nil +// spenderFinality is the verdict returned by waitForConfEventOnce when a +// conflicting spender reaches SafeDepth: the spender's txid plus the +// outpoints the spender consumed on-chain. The latter is propagated through +// to MarkTransferSuperseded so only inputs the conflict actually took are +// marked spent, leaving any other inputs of a multi-input transfer +// available. +type spenderFinality struct { + spender chainhash.Hash + consumed []wire.OutPoint +} + +// finalityResult is the outcome of a SafeDepth-conf watch on a conflicting +// spender, keyed by the input outpoint and spender txid so the main loop +// can correlate it against pending state and drop stale results. +type finalityResult struct { + op wire.OutPoint + spender chainhash.Hash + err error +} + +// watchSpenderFinality registers a SafeDepth-conf notification on the spender +// of the given confirmed input spend and spawns a fan-in goroutine that +// forwards the result to the shared finality channel. The spender's own +// SpendingTx pkScript and SpendingHeight are used as the lookup hint so the +// backend can target its rescan tightly. If the SpendDetail lacks a usable +// SpendingTx, registration is skipped and an error is returned: rather than +// supersede on weak evidence, the caller surfaces it as retryable so the +// whole watch set is re-registered (and the historical spend redelivered). +// +// Note that the same spender can spend multiple of our inputs; this +// machinery installs an independent finality watch per (op, spender), which +// duplicates work in that case but stays correct under independent reorgs +// of the individual input spends. +func (p *ChainPorter) watchSpenderFinality(ctx context.Context, + spend *chainntnfs.SpendDetail, + finality chan<- finalityResult) (*pendingSpenderState, error) { + + if spend.SpendingTx == nil || len(spend.SpendingTx.TxOut) == 0 { + return nil, fmt.Errorf("spend detail for spender %v lacks a "+ + "usable spending tx", spend.SpenderTxHash) + } + if spend.SpentOutPoint == nil || spend.SpenderTxHash == nil { + return nil, fmt.Errorf("spend detail is missing outpoint or " + + "spender hash") + } + + op := *spend.SpentOutPoint + spender := *spend.SpenderTxHash + heightHint := uint32(spend.SpendingHeight) + pkScript := spend.SpendingTx.TxOut[0].PkScript + + // Snapshot the spender's inputs. This is the authoritative list of + // outpoints the conflicting transaction consumed; the caller will + // use it to mark only those of our transfer's inputs as spent (not + // the rest, which may still be unspent on-chain). + consumed := make([]wire.OutPoint, len(spend.SpendingTx.TxIn)) + for i, in := range spend.SpendingTx.TxIn { + consumed[i] = in.PreviousOutPoint + } + + subCtx, cancel := context.WithCancel(ctx) + confEvent, errChan, err := p.cfg.ChainBridge.RegisterConfirmationsNtfn( + subCtx, spend.SpenderTxHash, pkScript, + uint32(p.cfg.SafeDepth), heightHint, false, nil, + ) + if err != nil { + cancel() + return nil, fmt.Errorf("unable to register finality conf "+ + "ntfn on spender %v: %w", spend.SpenderTxHash, err) } + + go func() { + defer confEvent.Cancel() + + send := func(res finalityResult) { + select { + case finality <- res: + case <-subCtx.Done(): + } + } + + select { + case ev, ok := <-confEvent.Confirmed: + res := finalityResult{op: op, spender: spender} + if !ok || ev == nil { + res.err = fmt.Errorf("spender finality conf "+ + "channel closed for spender=%v", + spender) + } + send(res) + + case err := <-errChan: + send(finalityResult{ + op: op, + spender: spender, + err: err, + }) + + case <-subCtx.Done(): + } + }() + + return &pendingSpenderState{ + spender: spender, + consumed: consumed, + cancel: cancel, + }, nil } // inputSpendErr carries a per-input spend notification stream failure @@ -810,6 +1168,48 @@ func (p *ChainPorter) locateConfirmedInputSpend(ctx context.Context, } } +// supersedeTransfer finalizes the fate of a transfer whose anchor +// transaction can never confirm because a conflicting transaction confirmed +// spending (some of) its inputs: the transfer is marked superseded so it is +// no longer treated as pending or resumed at startup, and any locked inputs +// are released. spentOutpoints lists the outpoints the conflicting +// transaction actually consumed on-chain; only those of the transfer's +// inputs are marked spent in the asset store. +// +// On success the returned error wraps ErrTransferSuperseded — terminal for +// the parcel's state machine, but a deliberate outcome rather than a bug. +// Callers can branch via errors.Is to distinguish it from a transient DB +// failure (which returns a different, sentinel-free error). +func (p *ChainPorter) supersedeTransfer(ctx context.Context, + pkg *sendPackage, spender chainhash.Hash, + spentOutpoints []wire.OutPoint) error { + + txHash := pkg.OutboundPkg.AnchorTx.TxHash() + + log.Infof("Anchor tx %v superseded by confirmed tx %v spending its "+ + "inputs", txHash, spender) + + // Mark the transfer superseded (and the consumed inputs spent) in + // the asset store first. The DB write is the durable state change; + // unlocking inputs is transient and lnd's wallet will re-derive + // availability from the chain on its own. If we crash between the + // two steps, mark-then-unlock guarantees we never leak transient + // state with the durable change still missing. + err := p.cfg.ExportLog.MarkTransferSuperseded( + ctx, txHash, spentOutpoints, + ) + if err != nil { + return fmt.Errorf("unable to mark transfer as superseded: %w", + err) + } + + p.unlockInputs(ctx, pkg) + + return fmt.Errorf("anchor tx %v superseded by confirmed tx %v "+ + "spending its inputs: %w", txHash, spender, + ErrTransferSuperseded) +} + // storeProofs writes the updated sender and receiver proof files to the proof // archive. func (p *ChainPorter) storeProofs(sendPkg *sendPackage) error { @@ -2413,22 +2813,22 @@ func (p *ChainPorter) stateStep(currentPkg sendPackage) (*sendPackage, error) { // The transaction was rejected because an input was // already spent, or because the transaction itself // was already mined. The transfer's fate is a fact - // about the chain we can interrogate directly: if a - // confirmed spender (our own anchor or a conflicting - // transaction) is reported for one of our inputs, we - // transition to WaitTxConf and let the - // confirmation-waiting state finalise the outcome — - // in particular, gating supersession on the - // conflicting spender reaching SafeDepth rather than - // acting on a single confirmation here, which would - // be irreversible on a routine reorg. + // about the chain that we can interrogate directly: + // if our own anchor transaction is the confirmed + // spender of the inputs, all that's left is to + // process its confirmation. If a different confirmed + // spender is found, we transition to WaitTxConf + // regardless and let the confirmation-waiting state + // gate supersession on the conflicting spender + // reaching SafeDepth — acting on a 1-conf foreign + // spend would be irreversible on a routine reorg. spender := p.locateConfirmedInputSpend( ctx, currentPkg.OutboundPkg, ) if spender != nil { - log.Infof("Anchor tx %v: confirmed spender "+ - "of input is %v, transitioning to "+ + log.Infof("Anchor tx %v: confirmed spender of "+ + "input is %v, transitioning to "+ "WaitTxConf", txHash, spender) currentPkg.SendState = SendStateWaitTxConf @@ -2436,11 +2836,12 @@ func (p *ChainPorter) stateStep(currentPkg sendPackage) (*sendPackage, error) { return ¤tPkg, nil } - // No confirmed spender located within the query - // window — the conflict may still be unconfirmed, so - // we don't know the transfer's fate yet. Release any - // fee inputs we locked, and retry the broadcast on - // next startup. + // We couldn't locate a confirmed spend of our inputs, + // so we can't determine the fate of the transfer yet + // (the conflicting transaction may still be + // unconfirmed). We release any fee sponsoring inputs + // we selected from lnd's wallet to avoid locking up + // balance, and will try again on next startup. // // TODO(guggero): Put this transfer into a failed state // and don't retry on next startup. diff --git a/tapfreighter/chain_porter_test.go b/tapfreighter/chain_porter_test.go index 15b25527fe..b8ef4544a7 100644 --- a/tapfreighter/chain_porter_test.go +++ b/tapfreighter/chain_porter_test.go @@ -1,6 +1,7 @@ package tapfreighter import ( + "bytes" "context" "fmt" "math/rand" @@ -247,3 +248,693 @@ func TestLocateConfirmedInputSpend(t *testing.T) { require.Nil(t, spender) }) } + +// TestWaitForTransferTxConfSpendResolution tests that a spend of one of the +// parcel's inputs by the parcel's own anchor transaction is ignored in favour +// of the confirmation event (so the parcel completes normally). The +// conflicting-spend supersede path is covered separately by +// TestWaitForTransferTxConfSupersedesAtSafeDepth (positive) and +// TestWaitForTransferTxConfReorgRescue (negative) below. +func TestWaitForTransferTxConfSpendResolution(t *testing.T) { + t.Parallel() + + newParcelPkg := func() (*sendPackage, wire.OutPoint) { + anchorTx := wire.NewMsgTx(2) + anchorTx.AddTxIn(&wire.TxIn{}) + anchorTx.AddTxOut(&wire.TxOut{ + PkScript: bytes.Repeat([]byte{0x01}, 34), + Value: 1000, + }) + + inputOp := wire.OutPoint{ + Hash: chainhash.Hash{1}, + Index: uint32(rand.Int31n(10)), + } + + return &sendPackage{ + SendState: SendStateWaitTxConf, + OutboundPkg: &OutboundParcel{ + AnchorTx: anchorTx, + AnchorTxHeightHint: 100, + Inputs: []TransferInput{{ + PrevID: asset.PrevID{ + OutPoint: inputOp, + }, + }}, + }, + }, inputOp + } + + // serviceConfReqs consumes the mock bridge's confirmation + // registration signals for the duration of the test. + serviceConfReqs := func(t *testing.T, + bridge *tapgarden.MockChainBridge) chan int { + + reqNums := make(chan int, 1) + done := make(chan struct{}) + t.Cleanup(func() { close(done) }) + + go func() { + for { + select { + case reqNo := <-bridge.ConfReqSignal: + select { + case reqNums <- reqNo: + case <-done: + return + } + + case <-done: + return + } + } + }() + + return reqNums + } + + // A spend by the parcel's own anchor transaction is not a conflict: + // the parcel keeps waiting, and completes once the confirmation + // event arrives. + t.Run("own spend awaits confirmation", func(t *testing.T) { + pkg, inputOp := newParcelPkg() + anchorTx := pkg.OutboundPkg.AnchorTx + anchorTxHash := anchorTx.TxHash() + + bridge := tapgarden.NewMockChainBridge() + bridge.SetSpend(inputOp, &chainntnfs.SpendDetail{ + SpentOutPoint: &inputOp, + SpenderTxHash: &anchorTxHash, + }) + reqNums := serviceConfReqs(t, bridge) + + exportLog := &fakeExportLog{ + pkScripts: map[wire.OutPoint][]byte{ + inputOp: {txscript.OP_1}, + }, + } + + porter := NewChainPorter(&ChainPorterConfig{ + ExportLog: exportLog, + ChainBridge: bridge, + }) + + // Dispatch the confirmation event once the conf notification + // has been registered. + blockHash := chainhash.Hash{0x07} + go func() { + reqNo := <-reqNums + bridge.SendConfNtfn(reqNo, &blockHash, 123, 1, nil, + anchorTx) + }() + + err := porter.waitForTransferTxConf(pkg) + require.NoError(t, err) + require.Empty(t, exportLog.superseded) + + require.Equal( + t, SendStateStorePostAnchorTxConf, pkg.SendState, + ) + require.NotNil(t, pkg.TransferTxConfEvent) + require.Equal( + t, blockHash, + pkg.OutboundPkg.AnchorTxBlockHash.UnwrapOr( + chainhash.Hash{}, + ), + ) + }) +} + +// TestWaitForConfEventOnceRequiresFullSpendCoverage tests that a registration +// failure for any input surfaces as a retryable error from +// waitForConfEventOnce, so the caller re-attempts the whole watch through +// its existing backoff loop. Silently dropping coverage of an input would +// recreate the original stranding bug: a foreign confirmed spend of the +// unwatched input would go unnoticed. +func TestWaitForConfEventOnceRequiresFullSpendCoverage(t *testing.T) { + t.Parallel() + + anchorTx := wire.NewMsgTx(2) + anchorTx.AddTxIn(&wire.TxIn{}) + anchorTx.AddTxOut(&wire.TxOut{ + PkScript: bytes.Repeat([]byte{0x01}, 34), + Value: 1000, + }) + + knownInput := wire.OutPoint{Hash: chainhash.Hash{1}} + unknownInput := wire.OutPoint{Hash: chainhash.Hash{2}} + + parcel := &OutboundParcel{ + AnchorTx: anchorTx, + AnchorTxHeightHint: 100, + Inputs: []TransferInput{ + {PrevID: asset.PrevID{OutPoint: knownInput}}, + {PrevID: asset.PrevID{OutPoint: unknownInput}}, + }, + } + + bridge := tapgarden.NewMockChainBridge() + go func() { + for range bridge.ConfReqSignal { + } + }() + + // Only the first input has a pkScript on record. The second + // input's pkScript lookup will fail, which must surface as an + // error rather than a silent partial watch. + exportLog := &fakeExportLog{ + pkScripts: map[wire.OutPoint][]byte{ + knownInput: {txscript.OP_1}, + }, + } + + porter := NewChainPorter(&ChainPorterConfig{ + ExportLog: exportLog, + ChainBridge: bridge, + }) + + confEvent, spender, terminal, err := porter.waitForConfEventOnce( + context.Background(), parcel, + ) + require.Error(t, err) + require.Nil(t, confEvent) + require.Nil(t, spender) + require.False( + t, terminal, "registration failure must be retryable, not "+ + "terminal — otherwise the parcel aborts instead of "+ + "being re-watched", + ) +} + +// newSpenderPkg returns a freshly populated sendPackage in the +// waiting-for-conf state along with the single transfer input outpoint and a +// minimal valid spender tx (one input, one output) that the porter can use +// when registering the spender-finality conf watch. +func newSpenderPkg(t *testing.T) (*sendPackage, wire.OutPoint, *wire.MsgTx) { + t.Helper() + + anchorTx := wire.NewMsgTx(2) + anchorTx.AddTxIn(&wire.TxIn{}) + anchorTx.AddTxOut(&wire.TxOut{ + PkScript: bytes.Repeat([]byte{0x01}, 34), + Value: 1000, + }) + + inputOp := wire.OutPoint{ + Hash: chainhash.Hash{1}, + Index: uint32(rand.Int31n(10)), + } + + spenderTx := wire.NewMsgTx(2) + spenderTx.AddTxIn(&wire.TxIn{PreviousOutPoint: inputOp}) + spenderTx.AddTxOut(&wire.TxOut{ + PkScript: bytes.Repeat([]byte{0x02}, 34), + Value: 900, + }) + + pkg := &sendPackage{ + SendState: SendStateWaitTxConf, + OutboundPkg: &OutboundParcel{ + AnchorTx: anchorTx, + AnchorTxHeightHint: 100, + Inputs: []TransferInput{{ + PrevID: asset.PrevID{OutPoint: inputOp}, + }}, + }, + } + + return pkg, inputOp, spenderTx +} + +// drainConfReqs spawns a goroutine that forwards every confirmation +// registration the mock bridge receives to the returned channel, until the +// test ends. +func drainConfReqs(t *testing.T, + bridge *tapgarden.MockChainBridge) chan int { + + t.Helper() + out := make(chan int, 4) + done := make(chan struct{}) + t.Cleanup(func() { close(done) }) + + go func() { + for { + select { + case reqNo := <-bridge.ConfReqSignal: + select { + case out <- reqNo: + case <-done: + return + } + + case <-done: + return + } + } + }() + + return out +} + +// TestWaitForTransferTxConfReorgRescue exercises the finality gate on +// supersession: a 1-confirmation foreign spend of a transfer input must NOT +// immediately mark the transfer superseded. If that spend is then reorged +// out, the porter must abandon the supersession and let the parcel's own +// anchor confirmation drive the transfer to completion. +// +// Without the finality gate, the parcel was superseded the moment a +// 1-conf foreign spend was seen, and any subsequent reorg permanently +// stranded a transfer that could otherwise have completed normally. +func TestWaitForTransferTxConfReorgRescue(t *testing.T) { + t.Parallel() + + pkg, inputOp, spenderTx := newSpenderPkg(t) + anchorTx := pkg.OutboundPkg.AnchorTx + anchorTxHash := anchorTx.TxHash() + spenderHash := spenderTx.TxHash() + + const safeDepth = 3 + bridge := tapgarden.NewMockChainBridge() + bridge.SetSpend(inputOp, &chainntnfs.SpendDetail{ + SpentOutPoint: &inputOp, + SpenderTxHash: &spenderHash, + SpendingTx: spenderTx, + SpendingHeight: 200, + }) + reqNums := drainConfReqs(t, bridge) + + exportLog := &fakeExportLog{ + pkScripts: map[wire.OutPoint][]byte{ + inputOp: {txscript.OP_1}, + }, + } + + porter := NewChainPorter(&ChainPorterConfig{ + ExportLog: exportLog, + ChainBridge: bridge, + SafeDepth: safeDepth, + }) + + errCh := make(chan error, 1) + go func() { errCh <- porter.waitForTransferTxConf(pkg) }() + + // The porter registers two conf ntfns: one on its own anchor, and + // one on the spender for finality. Drain both before proceeding. + var ownAnchorReq, spenderReq int + select { + case ownAnchorReq = <-reqNums: + case <-time.After(time.Second): + t.Fatal("own anchor conf ntfn was never registered") + } + select { + case spenderReq = <-reqNums: + case <-time.After(time.Second): + t.Fatal("spender finality conf ntfn was never registered; " + + "supersession is being decided on a single conf") + } + _ = spenderReq + + // At this point the 1-conf foreign spend has been observed and the + // finality watch is in place; supersession must not have happened. + require.Empty(t, exportLog.superseded) + + // Reorg the spend. The porter should drop the finality watch and + // go back to waiting for its own anchor. + bridge.SendSpendReorg(inputOp) + + // Now dispatch the own anchor's confirmation event. + blockHash := chainhash.Hash{0x07} + bridge.SendConfNtfn(ownAnchorReq, &blockHash, 123, 1, nil, anchorTx) + + select { + case err := <-errCh: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("waitForTransferTxConf did not return after own " + + "anchor confirmed post-reorg") + } + + require.Empty(t, exportLog.superseded) + require.Equal(t, SendStateStorePostAnchorTxConf, pkg.SendState) + require.NotNil(t, pkg.TransferTxConfEvent) + require.Equal( + t, anchorTxHash, pkg.TransferTxConfEvent.Tx.TxHash(), + ) +} + +// TestWaitForTransferTxConfSupersedesAtSafeDepth is the positive complement +// of the reorg-rescue test: when a conflicting spender does reach SafeDepth +// confirmations without being reorged out, supersession does fire. +func TestWaitForTransferTxConfSupersedesAtSafeDepth(t *testing.T) { + t.Parallel() + + pkg, inputOp, spenderTx := newSpenderPkg(t) + anchorTxHash := pkg.OutboundPkg.AnchorTx.TxHash() + spenderHash := spenderTx.TxHash() + + const safeDepth = 3 + bridge := tapgarden.NewMockChainBridge() + bridge.SetSpend(inputOp, &chainntnfs.SpendDetail{ + SpentOutPoint: &inputOp, + SpenderTxHash: &spenderHash, + SpendingTx: spenderTx, + SpendingHeight: 200, + }) + reqNums := drainConfReqs(t, bridge) + + exportLog := &fakeExportLog{ + pkScripts: map[wire.OutPoint][]byte{ + inputOp: {txscript.OP_1}, + }, + } + + porter := NewChainPorter(&ChainPorterConfig{ + ExportLog: exportLog, + ChainBridge: bridge, + SafeDepth: safeDepth, + }) + + errCh := make(chan error, 1) + go func() { errCh <- porter.waitForTransferTxConf(pkg) }() + + // Drain the own-anchor and spender-finality conf registrations. + var spenderReq int + select { + case <-reqNums: + case <-time.After(time.Second): + t.Fatal("own anchor conf ntfn was never registered") + } + select { + case spenderReq = <-reqNums: + case <-time.After(time.Second): + t.Fatal("spender finality conf ntfn was never registered") + } + + // Fire the spender's SafeDepth-conf event. + spenderBlock := chainhash.Hash{0x09} + bridge.SendConfNtfn( + spenderReq, &spenderBlock, 203, 1, nil, spenderTx, + ) + + select { + case err := <-errCh: + require.ErrorIs(t, err, ErrTransferSuperseded) + case <-time.After(2 * time.Second): + t.Fatal("waitForTransferTxConf did not return after spender " + + "reached SafeDepth") + } + + require.Equal( + t, []chainhash.Hash{anchorTxHash}, exportLog.superseded, + ) + require.Equal( + t, [][]wire.OutPoint{{inputOp}}, exportLog.supersedeOps, + "only the outpoint actually consumed on-chain by the "+ + "spender must be forwarded for spent-marking", + ) +} + +// TestWaitForTransferTxConfRetriesOnSpenderRegistrationFailure verifies that +// a transient failure registering the SafeDepth conf watch on a conflicting +// spender does not silently strand the transfer. The spend event has +// already been consumed from the stream and will not re-fire absent another +// reorg, so the function must return a retryable (non-terminal) error so +// the outer backoff loop re-registers the whole watch set and redelivers +// the historical spend. +func TestWaitForTransferTxConfRetriesOnSpenderRegistrationFailure( + t *testing.T) { + + t.Parallel() + + pkg, inputOp, spenderTx := newSpenderPkg(t) + spenderHash := spenderTx.TxHash() + + const safeDepth = 3 + bridge := tapgarden.NewMockChainBridge() + bridge.SetSpend(inputOp, &chainntnfs.SpendDetail{ + SpentOutPoint: &inputOp, + SpenderTxHash: &spenderHash, + SpendingTx: spenderTx, + SpendingHeight: 200, + }) + + // Arm the bridge to fail the next conf registration for the + // spender specifically. Targeting by txid (rather than by count) + // avoids racing against the porter's registration order. + bridge.FailConfFor(spenderHash) + + exportLog := &fakeExportLog{ + pkScripts: map[wire.OutPoint][]byte{ + inputOp: {txscript.OP_1}, + }, + } + + porter := NewChainPorter(&ChainPorterConfig{ + ExportLog: exportLog, + ChainBridge: bridge, + SafeDepth: safeDepth, + }) + + // Drain conf registration signals in the background so the porter's + // registrations don't block. + go func() { + for range bridge.ConfReqSignal { + } + }() + + confEvent, spender, terminal, err := porter.waitForConfEventOnce( + context.Background(), pkg.OutboundPkg, + ) + require.Error(t, err) + require.ErrorContains(t, err, "watch finality") + require.Nil(t, confEvent) + require.Nil(t, spender) + require.False( + t, terminal, "spender registration failure must be "+ + "retryable, not terminal — silent stranding is the "+ + "failure mode the finality gate exists to prevent", + ) + require.Empty(t, exportLog.superseded) +} + +// TestWaitForTransferTxConfPerInputFinality covers the multi-input case +// where two inputs have distinct conflicting spenders. A reorg of one +// spender must not cancel the in-flight finality watch on the other, and +// the surviving spender must still drive supersession when it reaches +// SafeDepth. A single shared "pending spender" slot would lose this case: +// the second spend would cancel the first watch, and the subsequent reorg +// of the second would clear all state — leaving the first spender's +// supersession opportunity on the floor until restart. +func TestWaitForTransferTxConfPerInputFinality(t *testing.T) { + t.Parallel() + + anchorTx := wire.NewMsgTx(2) + anchorTx.AddTxIn(&wire.TxIn{}) + anchorTx.AddTxOut(&wire.TxOut{ + PkScript: bytes.Repeat([]byte{0x01}, 34), + Value: 1000, + }) + anchorTxHash := anchorTx.TxHash() + + opA := wire.OutPoint{Hash: chainhash.Hash{1}, Index: 0} + opB := wire.OutPoint{Hash: chainhash.Hash{2}, Index: 1} + + mkSpenderTx := func(in wire.OutPoint, fillByte byte) *wire.MsgTx { + tx := wire.NewMsgTx(2) + tx.AddTxIn(&wire.TxIn{PreviousOutPoint: in}) + tx.AddTxOut(&wire.TxOut{ + PkScript: bytes.Repeat([]byte{fillByte}, 34), + Value: 900, + }) + return tx + } + spenderTxA := mkSpenderTx(opA, 0x0a) + spenderTxB := mkSpenderTx(opB, 0x0b) + spenderHashA := spenderTxA.TxHash() + spenderHashB := spenderTxB.TxHash() + + pkg := &sendPackage{ + SendState: SendStateWaitTxConf, + OutboundPkg: &OutboundParcel{ + AnchorTx: anchorTx, + AnchorTxHeightHint: 100, + Inputs: []TransferInput{ + {PrevID: asset.PrevID{OutPoint: opA}}, + {PrevID: asset.PrevID{OutPoint: opB}}, + }, + }, + } + + const safeDepth = 3 + bridge := tapgarden.NewMockChainBridge() + bridge.SetSpend(opA, &chainntnfs.SpendDetail{ + SpentOutPoint: &opA, + SpenderTxHash: &spenderHashA, + SpendingTx: spenderTxA, + SpendingHeight: 200, + }) + bridge.SetSpend(opB, &chainntnfs.SpendDetail{ + SpentOutPoint: &opB, + SpenderTxHash: &spenderHashB, + SpendingTx: spenderTxB, + SpendingHeight: 200, + }) + drainConfReqs(t, bridge) + + exportLog := &fakeExportLog{ + pkScripts: map[wire.OutPoint][]byte{ + opA: {txscript.OP_1}, + opB: {txscript.OP_1}, + }, + } + + porter := NewChainPorter(&ChainPorterConfig{ + ExportLog: exportLog, + ChainBridge: bridge, + SafeDepth: safeDepth, + }) + + errCh := make(chan error, 1) + go func() { errCh <- porter.waitForTransferTxConf(pkg) }() + + // Wait until both finality watches have been registered. They are + // independent so the order is not deterministic; poll the per-txid + // map until both spenders show up. + require.Eventually(t, func() bool { + _, okA := bridge.ConfReqForTxid(spenderHashA) + _, okB := bridge.ConfReqForTxid(spenderHashB) + return okA && okB + }, time.Second, 10*time.Millisecond, + "both spender finality conf ntfns were never registered") + + // Reorg input B's spend. The porter must cancel only B's watch and + // leave A's intact. + bridge.SendSpendReorg(opB) + + // Give the porter a moment to process the reorg before firing A's + // SafeDepth conf — the cancellation of B's watch is purely a + // background side effect, not observable from the outside, so a + // brief sleep is the most direct sync we have. + time.Sleep(50 * time.Millisecond) + + reqA, ok := bridge.ConfReqForTxid(spenderHashA) + require.True(t, ok) + + blockA := chainhash.Hash{0x0a} + bridge.SendConfNtfn(reqA, &blockA, 203, 1, nil, spenderTxA) + + select { + case err := <-errCh: + require.ErrorIs(t, err, ErrTransferSuperseded) + case <-time.After(2 * time.Second): + t.Fatal("waitForTransferTxConf did not return after spender " + + "A reached SafeDepth post-reorg of spender B") + } + + require.Equal( + t, []chainhash.Hash{anchorTxHash}, exportLog.superseded, + ) + // Only A's outpoint must be forwarded for spent-marking: B's + // spender was reorged out, so B's input is presumed unspent. + require.Equal( + t, [][]wire.OutPoint{{opA}}, exportLog.supersedeOps, + "only the input consumed by the spender that reached "+ + "SafeDepth must be marked spent", + ) +} + +// TestWaitForTransferTxConfRecoversPerInputSpendError verifies that a +// single per-input spend ntfn stream error is recovered by re-registering +// only that input's spend ntfn, without tearing down the confirmation +// watch or the other inputs' watches. A chronically-flapping per-input +// stream should not be able to block an imminent own-anchor confirmation, +// which would happen under the old "drop everything on any spend stream +// error" behaviour. +func TestWaitForTransferTxConfRecoversPerInputSpendError(t *testing.T) { + t.Parallel() + + anchorTx := wire.NewMsgTx(2) + anchorTx.AddTxIn(&wire.TxIn{}) + anchorTx.AddTxOut(&wire.TxOut{ + PkScript: bytes.Repeat([]byte{0x01}, 34), + Value: 1000, + }) + anchorTxHash := anchorTx.TxHash() + + opA := wire.OutPoint{Hash: chainhash.Hash{1}, Index: 0} + opB := wire.OutPoint{Hash: chainhash.Hash{2}, Index: 1} + + pkg := &sendPackage{ + SendState: SendStateWaitTxConf, + OutboundPkg: &OutboundParcel{ + AnchorTx: anchorTx, + AnchorTxHeightHint: 100, + Inputs: []TransferInput{ + {PrevID: asset.PrevID{OutPoint: opA}}, + {PrevID: asset.PrevID{OutPoint: opB}}, + }, + }, + } + + bridge := tapgarden.NewMockChainBridge() + reqNums := drainConfReqs(t, bridge) + + exportLog := &fakeExportLog{ + pkScripts: map[wire.OutPoint][]byte{ + opA: {txscript.OP_1}, + opB: {txscript.OP_1}, + }, + } + + porter := NewChainPorter(&ChainPorterConfig{ + ExportLog: exportLog, + ChainBridge: bridge, + SafeDepth: 3, + }) + + errCh := make(chan error, 1) + go func() { errCh <- porter.waitForTransferTxConf(pkg) }() + + // Wait for the own-anchor conf registration. + var ownAnchorReq int + select { + case ownAnchorReq = <-reqNums: + case <-time.After(time.Second): + t.Fatal("own anchor conf ntfn was never registered") + } + + // Fan a stream error into input A's spend ntfn. The porter must + // re-register A's spend ntfn (without disturbing input B or the + // own-anchor conf), so the test can then complete normally by + // dispatching the own-anchor conf event. + bridge.SendSpendErr(opA, fmt.Errorf("transient stream blip")) + + // Give the porter a moment to react to the per-input error. The + // re-registration is purely a background side effect — no + // observable signal — so a brief sleep is the most direct sync. + time.Sleep(50 * time.Millisecond) + + // Dispatch the own-anchor conf event. If the per-input error had + // torn down the entire watch (as in the pre-fix behaviour), this + // SendConfNtfn would block forever because the registered req has + // been cancelled. + blockHash := chainhash.Hash{0x07} + bridge.SendConfNtfn(ownAnchorReq, &blockHash, 123, 1, nil, anchorTx) + + select { + case err := <-errCh: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("waitForTransferTxConf did not return after own " + + "anchor confirmed; per-input spend error likely " + + "tore down the conf watch") + } + + require.Empty(t, exportLog.superseded, + "the spend stream error must not cause supersession") + require.Equal(t, SendStateStorePostAnchorTxConf, pkg.SendState) + require.NotNil(t, pkg.TransferTxConfEvent) + require.Equal( + t, anchorTxHash, pkg.TransferTxConfEvent.Tx.TxHash(), + ) +} From 234add6102e09910734f4601366e020040417804 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Wed, 24 Jun 2026 21:32:33 -0230 Subject: [PATCH 5/7] tapfreighter+tapcfg: make spend-query timeout configurable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit locateConfirmedInputSpend's 10-second budget on every ErrDoubleSpend broadcast rejection used to be a hardcoded constant. On slow or rescanning chain backends 10s can be insufficient; the lookup returns nil (inconclusive), the porter unlocks and waits for the next startup, the broadcast hits ErrDoubleSpend again, repeat — without any visible signal that the timeout is the bottleneck. Lift the constant to ChainPorterConfig.SpendQueryTimeout, expose it via a new --spendquerytimeout config flag (defaults to 10s in tapcfg, the same value as before), and emit a Warnf when the timeout fires with no confirmed spend observed — so operators seeing repeated inconclusive cycles can correlate them with the setting and raise it. A zero value at the porter layer falls back to defaultSpendQueryTimeout. --- tapcfg/config.go | 13 ++++++++++ tapcfg/server.go | 1 + tapfreighter/chain_porter.go | 50 +++++++++++++++++++++++++++++++----- 3 files changed, 57 insertions(+), 7 deletions(-) diff --git a/tapcfg/config.go b/tapcfg/config.go index 47a688ea73..32f3a0843b 100644 --- a/tapcfg/config.go +++ b/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, diff --git a/tapcfg/server.go b/tapcfg/server.go index 090f8980f3..6c2156a97f 100644 --- a/tapcfg/server.go +++ b/tapcfg/server.go @@ -706,6 +706,7 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger, BurnCommitter: supplyCommitManager, DelegationKeyChecker: addrBook, SafeDepth: cfg.ReOrgSafeDepth, + SpendQueryTimeout: cfg.SpendQueryTimeout, }, ) diff --git a/tapfreighter/chain_porter.go b/tapfreighter/chain_porter.go index 737a8a2d0f..244577d45d 100644 --- a/tapfreighter/chain_porter.go +++ b/tapfreighter/chain_porter.go @@ -47,12 +47,14 @@ const ( // re-register the transfer confirmation notification. confRetryDelayMax = time.Second * 30 - // spendQueryTimeout is the maximum time we wait for the chain - // notifier to report a confirmed spend of a transfer input. - // Historical spends are dispatched almost immediately after - // registration; the timeout only fires when the inputs are unspent - // or their spend hasn't confirmed yet. - spendQueryTimeout = time.Second * 10 + // defaultSpendQueryTimeout is the value used for + // ChainPorterConfig.SpendQueryTimeout if the caller leaves it at + // zero. 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 can raise it via config. + defaultSpendQueryTimeout = time.Second * 10 ) // ErrTransferSuperseded is returned by the chain porter when a transfer's @@ -157,6 +159,18 @@ type ChainPorterConfig struct { // permanent loss of the transfer's inputs on a routine 1-block reorg. // A zero value is clamped to 1 at construction time. SafeDepth int32 + + // SpendQueryTimeout caps how long locateConfirmedInputSpend waits + // for the chain notifier to report a confirmed spend of any of a + // transfer's inputs when the porter is resolving a rejected + // broadcast. Historical confirmed spends usually dispatch + // sub-second; the timeout matters when the inputs are unspent or + // their spend hasn't yet confirmed (the lookup returns inconclusive + // and the porter retries on next startup), or when the chain + // backend is slow (a too-tight timeout can cause spurious + // inconclusives and re-broadcast cycles). A zero value uses the + // defaultSpendQueryTimeout. + SpendQueryTimeout time.Duration } // ChainPorter is the main sub-system of the tapfreighter package. The porter @@ -199,6 +213,13 @@ func NewChainPorter(cfg *ChainPorterConfig) *ChainPorter { cfg.SafeDepth = 1 } + // Default the chain-query timeout if the caller left it at zero; + // otherwise an unconfigured field would degenerate into an instant + // "inconclusive" return on every double-spend resolution. + if cfg.SpendQueryTimeout <= 0 { + cfg.SpendQueryTimeout = defaultSpendQueryTimeout + } + return &ChainPorter{ cfg: cfg, outboundParcels: make(chan Parcel), @@ -1118,7 +1139,7 @@ func (p *ChainPorter) watchInputSpend(ctx context.Context, func (p *ChainPorter) locateConfirmedInputSpend(ctx context.Context, parcel *OutboundParcel) *chainhash.Hash { - ctx, cancel := context.WithTimeout(ctx, spendQueryTimeout) + ctx, cancel := context.WithTimeout(ctx, p.cfg.SpendQueryTimeout) defer cancel() // Spends that have already confirmed are dispatched (almost) @@ -1163,6 +1184,21 @@ func (p *ChainPorter) locateConfirmedInputSpend(ctx context.Context, return nil case <-ctx.Done(): + // The timeout fired before any of the transfer's + // inputs reported a confirmed spend. The conflict may + // still be unconfirmed (or our backend may be slow to + // rescan); either way, the broadcast-state caller + // falls back to unlock-and-retry-on-next-startup. + // Log loudly so the operator can correlate repeated + // inconclusive cycles with the SpendQueryTimeout + // setting. + log.Warnf("Spend lookup timed out after %v with no "+ + "confirmed spend observed for any of the "+ + "transfer's inputs; treating as inconclusive "+ + "(the transfer will be retried on next "+ + "startup). If this recurs, consider raising "+ + "--spendquerytimeout.", p.cfg.SpendQueryTimeout) + return nil } } From 85632835b992c78815d781547e9839a4d13e92e8 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Wed, 24 Jun 2026 21:44:50 -0230 Subject: [PATCH 6/7] itest: stranded transfer recovers on restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an integration test for the broadcast-state recovery path the chain porter takes when a transfer's anchor transaction has 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). The test asserts the operational outcome — that the receiver eventually sees the asset and the sender's transfer is reported as complete with a confirmed anchor height. Two related deficiencies remain uncovered at the itest level and are documented inline as follow-up: - The conflicting-spender resolution: where a foreign transaction reaches SafeDepth 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 at the waitForConfEventOnce unit-test level (TestWaitForTransferTxConfReorgRescue) but not end-to-end here. --- itest/send_test.go | 106 +++++++++++++++++++++++++++++++++++++ itest/test_list_on_test.go | 4 ++ 2 files changed, 110 insertions(+) diff --git a/itest/send_test.go b/itest/send_test.go index 6db23d530a..f91d2c63cc 100644 --- a/itest/send_test.go +++ b/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) { diff --git a/itest/test_list_on_test.go b/itest/test_list_on_test.go index 070dcd8e92..42b10183d2 100644 --- a/itest/test_list_on_test.go +++ b/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, From 024ce9b8b1741809163d381551821b792fbc5f34 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Wed, 24 Jun 2026 20:24:41 -0230 Subject: [PATCH 7/7] docs: add release note --- docs/release-notes/release-notes-0.8.1.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/release-notes/release-notes-0.8.1.md b/docs/release-notes/release-notes-0.8.1.md index 6632c9abdd..07df9686d1 100644 --- a/docs/release-notes/release-notes-0.8.1.md +++ b/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