multi: Register authenticated provisional lineage end to end

arkrpc/indexer.proto

@@ -317,6 +317,27 @@ message AncestryPath {
     // tx anchoring this fragment. Zero means unknown (legacy/unconfirmed),
     // in which case the client falls back to a bounded lookback floor.
     int32 commitment_height = 5;
+
+    // commitment_tx is the serialized commitment transaction. Its hash must
+    // equal commitment_txid and tree_path.batch_outpoint.txid.
+    bytes commitment_tx = 6;
+
+    // commitment_inputs contains one previous-output record for every actual
+    // commitment transaction input, in transaction input order.
+    repeated CommitmentInputEvidence commitment_inputs = 7;
+
+    // commitment_csv_expiry_delta is the batch sweep delay in blocks.
+    int32 commitment_csv_expiry_delta = 8;
+}
+
+// CommitmentInputEvidence binds one commitment input to the full previous
+// output needed for authenticated conflict observation.
+message CommitmentInputEvidence {
+    // outpoint must equal the corresponding commitment transaction input.
+    OutPoint outpoint = 1;
+
+    // prev_out supplies the value and pkScript of the spent output.
+    TxOut prev_out = 2;
 }
 
 // ScriptScope selects a pkScript and carries a proof-of-control for that

batchcanon/admission_test.go

@@ -1,7 +1,9 @@
 package batchcanon
 
 import (
+	"context"
 	"errors"
+	"sync"
 	"testing"
 
 	"github.com/btcsuite/btcd/chainhash/v2"
@@ -152,3 +154,154 @@ func TestAdmissionFailsClosedOnObservationPersistenceError(t *testing.T) {
 		recovered.Token.Lineage[0].Revision,
 	)
 }
+
+// TestWiredGateFailsClosedViaManagerOverlay proves the WIRED admission gate —
+// which the VTXO manager drives through Manager.GetBatch / LineageBlocked, not
+// the actor QueryLineage path — fails closed after a reorg observation whose
+// durable write failed. Without the manager's GetBatch overlay the gate would
+// read the stale durable row (still Ready) and admit a VTXO whose commitment
+// just left the best chain.
+func TestWiredGateFailsClosedViaManagerOverlay(t *testing.T) {
+	t.Parallel()
+
+	h := newManagerHarness(t, 100)
+	txid := testBatchTxid(0xd7)
+	input := testOutpoint(0xd8, 0)
+	h.registerBatch(t, &RegisterBatchRequest{
+		BatchTxID:            txid,
+		ConfirmationPkScript: []byte{0x51, 0x20, 0xd7},
+		ConsumedInputs:       []ConsumedInput{ci(input)},
+	})
+	h.fireConfirmed(t, txid, 101, testBatchTxid(0xe1))
+	h.fireSpend(t, input, txid, 101)
+
+	// The chain events above are async Tells flowing through the mock
+	// chainsource into the manager actor. Ask the actor (serialized after
+	// them) to synchronize before each direct GetBatch/LineageBlocked call
+	// so the wired-path assertions are not racing that delivery.
+	sync := func() *QueryLineageResponse {
+		t.Helper()
+		resp, err := h.mgrRef.Ask(
+			t.Context(), &QueryLineageRequest{
+				BatchTxIDs: []chainhash.Hash{txid},
+			},
+		).Await(t.Context()).Unpack()
+		require.NoError(t, err)
+		lineage, ok := resp.(*QueryLineageResponse)
+		require.True(t, ok)
+
+		return lineage
+	}
+	require.Equal(t, AvailableProvisional, sync().Availability)
+
+	// Baseline: the wired gate (LineageBlocked over the manager as Reader)
+	// admits the usable provisional lineage.
+	blocked, avail, err := LineageBlocked(t.Context(), h.mgr, txid)
+	require.NoError(t, err)
+	require.False(t, blocked)
+	require.Equal(t, AvailableProvisional, avail)
+
+	// A reorg observation whose durable write fails leaves the durable row
+	// stale (still Ready), but the in-memory overlay knows better.
+	h.store.setApplyError(errors.New("injected durable write failure"))
+	h.fireConfReorged(t, txid)
+
+	// The actor path already fails closed here (it reads the same overlay);
+	// this also synchronizes the reorg before the direct GetBatch below.
+	require.Equal(t, LineageReconciling, sync().Availability)
+
+	durable, err := h.store.GetBatch(t.Context(), txid)
+	require.NoError(t, err)
+	require.True(
+		t, durable.Ready(),
+		"precondition: the durable row stays stale-usable after "+
+			"the failed write",
+	)
+
+	// The wired gate reads through Manager.GetBatch, whose overlay forces
+	// the stale-ready row not-ready, so admission fails closed.
+	overlaid, err := h.mgr.GetBatch(t.Context(), txid)
+	require.NoError(t, err)
+	require.False(
+		t, overlaid.Ready(),
+		"manager overlay must force the stale-ready row not-ready",
+	)
+
+	blocked, avail, err = LineageBlocked(t.Context(), h.mgr, txid)
+	require.NoError(t, err)
+	require.True(t, blocked, "wired gate must refuse admission")
+	require.Equal(t, LineageReconciling, avail)
+}
+
+// storeLockAsserter wraps a Store and fails the test if a durable GetBatch runs
+// without the manager mutex held. Since Receive holds m.mu for the whole of
+// message processing (including the reorg persist), a read taken under the same
+// lock is linearized with the watch snapshot: a reorg cannot land between the
+// read and the snapshot. TryLock returns false only when the mutex is already
+// held, so this deterministically distinguishes the fixed read-under-lock path
+// from the racy read-outside-lock path.
+type storeLockAsserter struct {
+	Store
+
+	mu    *sync.Mutex
+	t     *testing.T
+	reads *int
+}
+
+func (s storeLockAsserter) GetBatch(ctx context.Context, txid chainhash.Hash) (
+	*Record, error) {
+
+	*s.reads++
+	if s.mu.TryLock() {
+		s.mu.Unlock()
+		s.t.Fatal(
+			"GetBatch durable read ran WITHOUT holding m.mu — " +
+				"the read must be linearized with the " +
+				"watch snapshot under one lock, or a " +
+				"concurrent reorg persist can slip between " +
+				"them",
+		)
+	}
+
+	return s.Store.GetBatch(ctx, txid)
+}
+
+// TestGetBatchReadsUnderLock pins the fix for the overlay stale-read race: the
+// durable store read must run under the same m.mu as the watch snapshot, so a
+// reorg that persists a complete ReorgedOut@(r+1) snapshot cannot interleave
+// between a Provisional@r read and the overlay decision and leave the stale
+// record admissible.
+func TestGetBatchReadsUnderLock(t *testing.T) {
+	t.Parallel()
+
+	h := newManagerHarness(t, 100)
+	txid := testBatchTxid(0xf1)
+	input := testOutpoint(0xf2, 0)
+	h.registerBatch(t, &RegisterBatchRequest{
+		BatchTxID:            txid,
+		ConfirmationPkScript: []byte{0x51, 0x20, 0xf1},
+		ConsumedInputs:       []ConsumedInput{ci(input)},
+	})
+	h.fireConfirmed(t, txid, 101, testBatchTxid(0xf3))
+	h.fireSpend(t, input, txid, 101)
+
+	// Drain the actor before swapping cfg.Store. The mailbox is FIFO and
+	// single-threaded, so this Ask is answered only after every prior
+	// message — including the spend observation that reads cfg.Store on the
+	// actor goroutine — has been fully processed. Without this the bare
+	// field swap below races that read.
+	h.mgrRef.Ask(t.Context(), &QueryLineageRequest{
+		BatchTxIDs: []chainhash.Hash{txid},
+	}).Await(t.Context())
+
+	// Swap in the lock-asserting store; the actor is now idle, so there is
+	// no concurrent reader of cfg.Store.
+	var reads int
+	h.mgr.cfg.Store = storeLockAsserter{
+		Store: h.store, mu: &h.mgr.mu, t: t, reads: &reads,
+	}
+
+	_, err := h.mgr.GetBatch(t.Context(), txid)
+	require.NoError(t, err)
+	require.Positive(t, reads, "the durable read must have executed")
+}

batchcanon/manager.go

@@ -214,6 +214,9 @@ func (m *Manager) Receive(ctx context.Context,
 	case *inputSpendDoneMsg:
 		m.handleInputSpendDone(ctx, v)
 
+	case *ConsumerForfeitPersistedMsg:
+		m.handleConsumerForfeitPersisted(ctx, v.ConsumerBatch)
+
 	default:
 		return fn.Err[ManagerResp](
 			fmt.Errorf("unknown batchcanon message: %T", msg),
@@ -327,6 +330,9 @@ func validateRegistration(req *RegisterBatchRequest,
 	if len(req.ConfirmationPkScript) == 0 {
 		return fmt.Errorf("batch confirmation pkScript is required")
 	}
+	if req.CSVExpiryDelta <= 0 {
+		return fmt.Errorf("batch CSV expiry delta must be positive")
+	}
 	if len(req.ConsumedInputs) == 0 {
 		return fmt.Errorf("batch must register every consumed input")
 	}
@@ -1050,10 +1056,53 @@ func observationComplete(w *batchWatch) bool {
 	return true
 }
 
+// GetBatch implements batchcanon.Reader with the manager's in-memory overlay so
+// the wired admission gate fails closed the instant a reorg/conflict
+// observation cannot be persisted. The gate reads availability through this
+// method; when an in-memory watch is not ready — most importantly after an
+// ApplyObservation write failed in persistObservation — the returned record is
+// forced not-ready, so LineageAvailability derives LineageReconciling (never
+// usable) even though the stale durable row still reads ready. The overlay only
+// ever downgrades ready->not-ready, never the reverse, so it is strictly
+// fail-closed. Without it the gate would admit a VTXO whose commitment just
+// left the best chain until a restart durably reconciled the row.
+//
+// The durable read and the watch snapshot are taken under the SAME m.mu that
+// Receive holds for the whole of message processing, so they observe one
+// consistent point in time relative to any concurrent persist. Reading the
+// store outside the lock races a reorg that lands between the read and the
+// snapshot: the read returns Provisional@r, the reorg then persists a complete
+// ReorgedOut@(r+1) snapshot (leaving w.ready=true), and the overlay below would
+// not downgrade the already-read stale record — admitting it.
+func (m *Manager) GetBatch(ctx context.Context, txid chainhash.Hash) (*Record,
+	error) {
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	record, err := m.cfg.Store.GetBatch(ctx, txid)
+	if err != nil {
+		return nil, err
+	}
+
+	if w, watched := m.watches[txid]; watched && !w.ready &&
+		record.Ready() {
+
+		notReady := *record
+		notReady.ReadyGeneration = fn.None[uint64]()
+
+		return &notReady, nil
+	}
+
+	return record, nil
+}
+
 // persistObservation atomically writes the complete in-memory view. On any
-// error the manager's overlay closes admission immediately, even if the old
-// durable row was usable; restart reconciliation closes it durably before
-// re-arming watches.
+// error the in-memory watch is marked not-ready; the wired admission gate reads
+// through Manager.GetBatch, whose overlay forces such a batch not-ready even
+// though the stale durable row still reads ready, so admission closes
+// immediately. Restart reconciliation then closes it durably before re-arming
+// watches.
 func (m *Manager) persistObservation(ctx context.Context, w *batchWatch) {
 	inputs := make([]InputObservation, 0, len(w.inputs))
 	for outpoint, input := range w.inputs {
@@ -1314,6 +1363,36 @@ func (m *Manager) redriveConsumersForCreator(ctx context.Context,
 	return nil
 }
 
+// handleConsumerForfeitPersisted redrives terminal consumer-edge resolution for
+// a batch whose consumed-VTXO forfeiture marker has just become durable. It is
+// the event-driven complement to the restart-time redrive
+// (redriveTerminalConsumerLifecycles): when a consumer batch reaches a terminal
+// state before its ForfeitedBy(consumer) marker exists, the original resolution
+// defers on the revision compare-and-swap, and no further batchcanon event is
+// guaranteed. The marker's arrival is exactly the evidence change that can now
+// let the terminal restore succeed. Resolution is gated on the consumer being
+// Ready and terminal, so a not-yet-final consumer (the common case, since the
+// marker is usually written well before finality) is a harmless no-op.
+func (m *Manager) handleConsumerForfeitPersisted(ctx context.Context,
+	consumerBatch chainhash.Hash) {
+
+	record, err := m.cfg.Store.GetBatch(ctx, consumerBatch)
+	if err != nil {
+		// No record yet (or a transient read error): the restart-time
+		// redrive remains the backstop, so treat this as a no-op.
+		m.logger(ctx).DebugS(ctx, "No batch record to redrive on "+
+			"forfeit persist",
+			slog.String("batch", consumerBatch.String()))
+
+		return
+	}
+	if !record.Ready() || !terminalState(record.State) {
+		return
+	}
+
+	m.handleConsumerLifecycle(ctx, consumerBatch, record.State)
+}
+
 // releaseSpendWatches unregisters the per-input spend watches for a batch,
 // called once the batch finalizes.
 func (m *Manager) releaseSpendWatches(ctx context.Context, w *batchWatch) {