tapgarden: reorder Confirmed branch so MarkBatchConfirmed writes last

BatchStateConfirmed on main names two different essences with the
same word: an *input state* for the Confirmed branch of the state
machine, and a *mid-branch durable checkpoint* written by
MarkBatchConfirmed. The Confirmed branch published universe proofs,
emitted the supply-commit event, called MarkBatchConfirmed (which
advanced disk state to Confirmed), and then registered the anchor
tx with the re-org watcher via WatchProofs. A crash between
MarkBatchConfirmed and WatchProofs left disk at Confirmed with no
re-org callback registered by us; on restart the fast-forward
skipped the branch entirely, and the re-org watcher's own Start-
time recovery re-registered the anchor tx with its
DefaultUpdateCallback -- silently dropping the universe re-publish
that updateMintingProofs performs on any subsequent re-org.

Dissolve the identity conflation by making MarkBatchConfirmed the
LAST persistence write in the branch. WatchProofs now runs first;
disk-Confirmed genuinely means "every step the Confirmed branch
owes is done." A crash before MarkBatchConfirmed keeps the batch
at Broadcast so the whole branch re-runs on restart -- universe
publish is idempotent, augmenter side is deduped by event_key
(migration 62), storeMintingProof rides upserts, and WatchProofs
against a fresh watcher instance is a first-time registration.

Also rewrite the surrounding comments to describe the new
ordering, note the semantic justification for the restart fast-
forward, and drop the stale TODO on the Finalized case ("confirmed
should just be the final state?") -- that concern is fully
resolved by the new ordering.

Add a ShouldFail atomic.Bool hook to MockProofWatcher and a
testWatchProofsFailureAbortsConfirmation test that forces
WatchProofs to fail, asserts the batch stays at Broadcast (under
the old ordering this assertion would fail because
MarkBatchConfirmed would have run before the WatchProofs failure),
restarts with the mock succeeding, and asserts the batch advances
to Finalized. Pins the ordering invariant against regression.

Author: jtobin

tapgarden/cultivator.go

@@ -388,8 +388,12 @@ func (b *Cultivator) assetCultivator() {
 	defer b.Wg.Done()
 
 	currentBatchState := b.cfg.Batch.State()
-	// If the batch is already marked as confirmed, then we just need to
-	// advance it one more level to be finalized.
+	// If the batch is already marked as confirmed, then we just
+	// need to advance it one more level to be finalized. Under the
+	// current Confirmed-branch ordering, MarkBatchConfirmed is the
+	// last persistence write, so disk-Confirmed means the full
+	// Confirmed branch (including the re-org watcher registration)
+	// already ran to completion; the skip is semantically justified.
 	if currentBatchState == BatchStateConfirmed {
 		log.Infof("MintingBatch(%x): already confirmed!", b.batchKey[:])
 
@@ -1231,9 +1235,8 @@ func (b *Cultivator) stateStep(currentState BatchState) (BatchState, error) {
 		// confirmation branch re-runs on restart: universe
 		// publish above is idempotent, the event_key dedup
 		// index (migration 62, backfilled by 63) makes the
-		// augmenter side idempotent, and MarkBatchConfirmed
-		// below is what advances state on disk -- so retry is
-		// safe.
+		// augmenter side idempotent, and MarkBatchConfirmed is
+		// the last write below -- so retry is safe.
 		err = b.augmenter().OnBatchConfirmed(
 			ctx, b.cfg.Batch, anchorAssets, nonAnchorAssets,
 			mintingProofs,
@@ -1243,6 +1246,35 @@ func (b *Cultivator) stateStep(currentState BatchState) (BatchState, error) {
 				err)
 		}
 
+		// Register the batch's proofs with the re-org watcher
+		// before advancing state on disk. If this fails, the
+		// batch stays in BatchStateBroadcast and the whole
+		// confirmation branch re-runs on restart, ensuring the
+		// correct updateMintingProofs callback is the one bound
+		// to the anchor tx. If we instead registered after
+		// MarkBatchConfirmed, a crash between the two would
+		// leave disk at Confirmed with no callback registered by
+		// us; the re-org watcher's Start-time recovery would
+		// re-register with its DefaultUpdateCallback, silently
+		// dropping the universe re-publish on any subsequent
+		// re-org.
+		if err := b.cfg.ProofWatcher.WatchProofs(
+			maps.Values(mintingProofs), b.cfg.UpdateMintingProofs,
+		); err != nil {
+			return 0, fmt.Errorf("error watching proof: %w", err)
+		}
+
+		// MarkBatchConfirmed is the last persistence write in
+		// this branch. Under this ordering, disk-Confirmed
+		// genuinely means "every step this branch owes is done."
+		// A crash before this call keeps the batch at
+		// BatchStateBroadcast and the branch re-runs on restart;
+		// a crash after has nothing left to do beyond the
+		// terminal Finalized state advance. The essence-splitting
+		// that produced the older reorg-callback gap (disk-
+		// Confirmed as both an input state and a mid-branch
+		// checkpoint) is dissolved by making the on-disk name
+		// mean what it says.
 		err = b.cfg.BatchStore.MarkBatchConfirmed(
 			ctx, b.cfg.Batch, confInfo.BlockHash,
 			confInfo.BlockHeight, confInfo.TxIndex,
@@ -1252,14 +1284,6 @@ func (b *Cultivator) stateStep(currentState BatchState) (BatchState, error) {
 			return 0, fmt.Errorf("unable to confirm batch: %w", err)
 		}
 
-		// Now that we've confirmed the batch, we'll hand over the
-		// proofs to the re-org watcher.
-		if err := b.cfg.ProofWatcher.WatchProofs(
-			maps.Values(mintingProofs), b.cfg.UpdateMintingProofs,
-		); err != nil {
-			return 0, fmt.Errorf("error watching proof: %w", err)
-		}
-
 		log.Infof("Cultivator(%x): transition states: %v -> %v",
 			b.batchKey[:], BatchStateConfirmed, BatchStateFinalized)
 
@@ -1271,7 +1295,6 @@ func (b *Cultivator) stateStep(currentState BatchState) (BatchState, error) {
 		log.Infof("Cultivator(%x): transition states: %v -> %v",
 			b.batchKey[:], BatchStateFinalized, BatchStateFinalized)
 
-		// TODO(roasbeef): confirmed should just be the final state?
 		ctx, cancel := b.WithCtxQuit()
 		defer cancel()
 		err := b.cfg.BatchStore.UpdateBatchState(

tapgarden/mock.go

@@ -571,11 +571,21 @@ func (m *MockProofArchive) ImportProofs(context.Context,
 }
 
 type MockProofWatcher struct {
+	// ShouldFail, when set, causes WatchProofs to return an error
+	// instead of the usual no-op. Tests that need to observe the
+	// caretaker's handling of a re-org watcher registration failure
+	// flip this on before driving the relevant state transition and
+	// off before the retry.
+	ShouldFail atomic.Bool
 }
 
 func (m *MockProofWatcher) WatchProofs([]*proof.Proof,
 	proof.UpdateCallback) error {
 
+	if m.ShouldFail.Load() {
+		return fmt.Errorf("simulated re-org watcher failure")
+	}
+
 	return nil
 }
 

tapgarden/planter_test.go

@@ -2302,6 +2302,98 @@ func testOnBatchConfirmedFailureAbortsConfirmation(t *mintingTestHarness) {
 	t.assertNumCaretakersActive(0)
 }
 
+// testWatchProofsFailureAbortsConfirmation asserts that a failure
+// returned from ProofWatcher.WatchProofs aborts mint confirmation:
+// the batch stays in BatchStateBroadcast on disk so the whole
+// confirmation branch re-runs on restart. This pins the ordering
+// invariant that MarkBatchConfirmed is the LAST persistence write
+// in the Confirmed branch -- under the pre-A2 ordering
+// (MarkBatchConfirmed before WatchProofs), a WatchProofs failure
+// would leave the batch at BatchStateConfirmed and this assertion
+// would fail. The invariant matters because a batch at disk-
+// Confirmed with no registered re-org callback silently drops
+// universe re-publish on subsequent re-orgs.
+func testWatchProofsFailureAbortsConfirmation(t *mintingTestHarness) {
+	// Wire the harness so the mock re-org watcher rejects the
+	// registration on the first pass; the retry after restart
+	// receives a fresh MockProofWatcher (zero ShouldFail) and
+	// succeeds.
+	t.proofWatcher.ShouldFail.Store(true)
+	t.refreshChainPlanter()
+
+	// Drive Pending -> Frozen -> Committed -> Broadcast.
+	const numSeedlings = 3
+	_ = t.queueInitialBatch(numSeedlings)
+	frozenBatch := t.finalizeBatchAssertFrozen(false)
+	t.assertBatchCommitted(frozenBatch.BatchKey.PubKey)
+	t.assertGenesisPsbtFinalized(nil)
+	tx := t.assertTxPublished()
+
+	// Assemble the confirmation block.
+	merkleTree := blockchain.BuildMerkleTreeStore(
+		[]*btcutil.Tx{btcutil.NewTx(tx)}, false,
+	)
+	merkleRoot := merkleTree[len(merkleTree)-1]
+	blockHeader := wire.NewBlockHeader(
+		0, chaincfg.MainNetParams.GenesisHash, merkleRoot, 0, 0,
+	)
+	block := &wire.MsgBlock{
+		Header:       *blockHeader,
+		Transactions: []*wire.MsgTx{tx},
+	}
+
+	// Deliver the confirmation. WatchProofs will fail and the
+	// Confirmed branch must abort before MarkBatchConfirmed.
+	sendConfNtfn := t.assertConfReqSent(tx, block)
+	sendConfNtfn()
+
+	// The batch on disk must remain at BatchStateBroadcast. If A2's
+	// ordering regressed and MarkBatchConfirmed ran before the
+	// WatchProofs failure, we'd see BatchStateConfirmed here
+	// instead.
+	require.Never(t, func() bool {
+		batches, err := t.store.FetchAllBatches(context.Background())
+		require.NoError(t, err)
+		if len(batches) != 1 {
+			return false
+		}
+		return batches[0].State() != tapgarden.BatchStateBroadcast
+	}, 500*time.Millisecond, 50*time.Millisecond,
+		"batch advanced past Broadcast despite WatchProofs failure")
+
+	// Simulate a restart with a now-succeeding re-org watcher. The
+	// fresh caretaker resumes the Broadcast batch, re-publishes the
+	// tx, and re-registers for a confirmation.
+	t.proofWatcher.ShouldFail.Store(false)
+	t.refreshChainPlanter()
+	select {
+	case <-t.errChan:
+	default:
+	}
+
+	_ = t.assertTxPublished()
+	sendConfNtfn = t.assertConfReqSent(tx, block)
+	sendConfNtfn()
+
+	// With WatchProofs now returning nil, the Confirmed branch
+	// completes and the batch advances to Finalized.
+	err := wait.Predicate(func() bool {
+		batches, err := t.store.FetchAllBatches(
+			context.Background(),
+		)
+		require.NoError(t, err)
+		if len(batches) != 1 {
+			return false
+		}
+		return batches[0].State() == tapgarden.BatchStateFinalized
+	}, defaultTimeout)
+	require.NoError(
+		t, err, "batch never advanced to Finalized on retry",
+	)
+
+	t.assertNumCaretakersActive(0)
+}
+
 // mintingStoreTestCase is used to programmatically run a series of test cases
 // that are parametrized based on a fresh minting store.
 type mintingStoreTestCase struct {
@@ -2351,6 +2443,10 @@ var testCases = []mintingStoreTestCase{
 		name:     "on_batch_confirmed_failure_aborts_confirmation",
 		testFunc: testOnBatchConfirmedFailureAbortsConfirmation,
 	},
+	{
+		name:     "watch_proofs_failure_aborts_confirmation",
+		testFunc: testWatchProofsFailureAbortsConfirmation,
+	},
 }
 
 // TestBatchedAssetIssuance runs a test of tests to ensure that the set of