test(zakura): restore the dropped test suite and cover the reorg stack end-to-end

CHANGELOG.md

@@ -18,6 +18,15 @@ and this project adheres to [Semantic Versioning](https://semver.org).
 - Verified-commitment-trees fast sync is now enabled by default when checkpoint
   sync is enabled. Operators can keep checkpoint sync but opt out of the new
   path by setting `consensus.vct_fast_sync = false`.
+- All Zakura header-store writers (header-range commits, best-chain seeds, and
+  body-commit releases) now change chain membership through a single
+  suffix-replacement primitive that deletes everything above the fork point by
+  scanning the actual on-disk rows of all five column families — so rows can no
+  longer be stranded by bounded delete loops — and verifies linkage as a hard
+  precondition. Any batch that replaces a header suffix (a header reorg) is
+  followed by the store coherence audit, repairing and reporting
+  (`state.zakura.header_store.incoherent`) any residual violation at the moment
+  it happens instead of leaving a latent on-disk fault.
 - Refreshed the default Testnet Zakura bootstrap peer identities
   (`DEFAULT_TESTNET_ZAKURA_BOOTSTRAP_PEERS`) after the Testnet fleet's iroh node
   keys were rotated. The previous hardcoded node IDs were stale, so a fresh node
@@ -35,6 +44,29 @@ and this project adheres to [Semantic Versioning](https://semver.org).
 
 ### Fixed
 
+- Zakura header-store consensus reads now verify store invariants as they
+  walk: the difficulty-context walk checks that every stored header is the
+  block its hash row names and links to the row below it, and the anchor
+  lookup distinguishes a corrupted index round-trip from a genuinely unknown
+  anchor. A corrupted store now surfaces as an explicit local
+  `StoreIncoherent` error (never scored against peers) instead of poisoning
+  difficulty validation and rejecting honest headers with
+  `InvalidDifficultyThreshold`.
+- Fixed three Zakura header-store write paths that could leave the on-disk
+  header store internally incoherent after chain forks, causing nodes to
+  reject valid headers from honest peers (`InvalidDifficultyThreshold` /
+  `UnknownAnchor`) and wedge below the network tip until manual intervention.
+  Header ranges must now link to their anchor and be internally contiguous
+  (rejected with the new `UnlinkedRange` error otherwise, which is classified
+  as a local, non-peer-scoring failure), header ranges re-delivered over
+  heights that already have committed block bodies no longer re-insert
+  provisional header rows below the body tip, and the header row seeded from
+  a committed best-chain block is skipped when it does not link to the stored
+  row below it (header-range sync converges the store instead).
+- Fixed dual-stack Zakura fallback shutting down the Zakura serving layer. When
+  the stall watchdog resumes legacy `ChainSync`, Zakura now keeps its header-
+  and block-sync reactors alive as a serving/advertising bridge while an apply
+  gate drains in-flight Zakura body applies before legacy sync drives commits.
 - Fixed legacy peers being disconnected for returning empty `FindBlocks` or
   `FindHeaders` responses when Zebra is at or near the network tip.
 - Kept an already-active mempool and `getblocktemplate` mining RPCs running
@@ -44,6 +76,88 @@ and this project adheres to [Semantic Versioning](https://semver.org).
 - Fixed a near-tip sync restart loop when a timed-out `AwaitUtxo` lookup in the
   transaction verifier was converted to `InternalDowncastError` instead of a
   missing transparent input.
+- Fixed a permanent header-sync wedge where a fork-recovery re-commit left
+  stale rows in the stored difficulty-validation window: every extension
+  range was then rejected with `InvalidDifficultyThreshold`, each rejection
+  scored and disconnected the (honest) serving peer, and the wedge survived
+  restarts because the stale rows are on disk (observed live at height
+  4148005 for hours). Contextual validation failures are now classified as
+  `ContextMismatch`: they no longer score peers, and repeated identical
+  failures from independent peers feed the fork-recovery walk-back — each
+  deepening re-commits a longer range, rewriting the poisoned window until
+  the chain extends again.
+- Zakura block-sync liveness disconnects are now deferred while the local
+  apply pipeline holds unfinished bodies: "no accepted block progress" cannot
+  be blamed on a peer while the node's own commits are not landing. During a
+  ~45-minute local commit stall the old behavior exiled healthy peers one by
+  one into the no-progress cooldown. A genuinely dead peer is still caught as
+  soon as the pipeline drains. Deferrals are counted in
+  `sync.block.liveness.deferred_local_stall`.
+- Fixed a chain-tip reset (fork-recovery invalidation, `invalidateblock`)
+  being silently discarded by the Zakura chain-tip mirror. The mirror maxed
+  the reset tip against a `latest_chain_tip` watch that may not have observed
+  the reset yet, republished the stale higher tip, and turned the downstream
+  `VerifiedReset` into a no-op — the block-sync sequencer then pinned one
+  block above the real tip, never requested the missing parent, and every
+  body commit timed out until restart. On `TipAction::Reset` the action's tip
+  is now published verbatim; the anti-regression maxing only applies to
+  `Grow`.
+- Fixed Zakura block-sync peers being serially disconnected and parked at the
+  chain tip by a stale liveness deadline. When the download floor passed a
+  request because other peers' deliveries satisfied its heights, the floor GC
+  removed the request but only cleared the liveness deadline for peers whose
+  last delivery was newer than their last request — a healthy peer with an
+  older delivery kept a deadline it could no longer answer, was disconnected
+  for "no accepted block progress" with zero outstanding requests, and parked
+  in the 180-second no-progress cooldown. At tip this exiled peers one by one
+  until the block-sync peer set collapsed (reproduced live: 5 → 0 peers within
+  ~25 minutes of a fleet-wide restart, wedging body sync when the next gap
+  formed). Heights satisfied below the floor now always clear an idle peer's
+  deadline; genuinely unresponsive peers are still caught by request timeouts,
+  which deliberately keep the deadline armed. The liveness disconnect log was
+  also promoted to info.
+- After a legacy fallback, the dual-stack watchdog now re-promotes Zakura to
+  the body-sync driver once legacy `ChainSync` is caught up and stable (three
+  consecutive exhausted sync rounds with the tip advancing by at most two
+  blocks). Fallback was previously permanent per process lifetime, so every
+  min-difficulty burst demoted one more node's Zakura sync to a serve-only
+  bridge until restart. The hand-back reuses the apply-gate commit barrier in
+  reverse and is counted in `sync.zakura.repromoted`.
+- The dual-stack Zakura stall watchdog no longer shuts down the Zakura header-
+  and block-sync reactors when it falls back to legacy `ChainSync`. Killing
+  them turned every fallback into a fleet-wide Zakura outage: the fallback
+  node — often the only peer with working block ingest — stopped advertising
+  statuses, serving headers and bodies, and forwarding `NewBlock`s, which
+  pinned every Zakura peer's frontier and starved zakura-only nodes until a
+  manual restart. Legacy `ChainSync` now resumes as the body-sync driver while
+  the Zakura reactors stay alive, follow local commits through the chain-tip
+  mirror, and keep serving the Zakura network as a bridge. Fallback engagement
+  is counted in `sync.zakura.legacy_fallback.engaged`. Inbound block-gossip
+  ingest failures and queue drops — previously silent at default log levels,
+  which made tip-ingest freezes unattributable — are now logged at info and
+  counted (`gossip.ingest.failed.count`, `gossip.ingest.duplicate.count`).
+- Fixed Zakura nodes gossiping and following side-chain blocks. An inbound
+  `NewBlock` that committed as a side chain (for example a testnet
+  min-difficulty branch) still advanced the node's Zakura header and verified
+  frontiers and was forwarded to peers, so a whole fleet could advertise and
+  propagate a losing branch over Zakura while each node's own chain stayed on
+  the best one — stranding zakura-only peers that followed the gossip. An
+  accepted `NewBlock` is now checked against the best chain before it advances
+  any frontier or is forwarded; side-chain commits are remembered for dedup
+  only and counted in `sync.header.tip.new_block.side_chain`.
+- Fixed Zakura header sync permanently stranding a node whose committed chain
+  suffix ends up on a branch the network abandoned. Header sync only requested
+  headers forward from its own frontier hash and treated every non-linking
+  response as peer misbehavior, so after a reorg a zakura-only node rejected
+  all honest headers forever (`first_header_does_not_link`), flagged the
+  honest peers, and never caught up — restarts did not help. Non-linking
+  forward responses are no longer scored; after repeated failures from
+  independent peers the request anchor now walks back below the verified block
+  tip (exponentially deeper per round, at most `MAX_BLOCK_REORG_HEIGHT`, never
+  below the finalized height) until responses link, the higher-work header
+  range commits through the fork point, and the stranded committed body suffix
+  is invalidated so block sync re-downloads the new branch. Recovery progress
+  is logged and counted (`sync.header.fork_recovery.*`).
 - Fixed dual-stack nodes (`v2_p2p` and `legacy_p2p` both enabled) permanently
   shutting down their own Zakura header- and block-sync drivers when a legacy
   peer on a foreign fork answered the body-sync stall watchdog's cross-check

zebra-consensus/src/block.rs

@@ -303,6 +303,7 @@ where
             let mut block_miner_fees = Ok(Amount::zero());
 
             use futures::StreamExt;
+            let tx_checks_started = std::time::Instant::now();
             while let Some(result) = async_checks.next().await {
                 tracing::trace!(?result, remaining = async_checks.len());
                 let response = result
@@ -323,6 +324,18 @@ where
                 }
             }
 
+            // A slow drain here means transaction verification (usually a UTXO
+            // await on an out-of-order parent) held the block; this stage log
+            // attributes driver-side commit timeouts to their pipeline stage.
+            if tx_checks_started.elapsed() > std::time::Duration::from_secs(20) {
+                tracing::warn!(
+                    ?height,
+                    ?hash,
+                    elapsed = ?tx_checks_started.elapsed(),
+                    "slow transaction verification stage while committing block"
+                );
+            }
+
             // Check the summed block totals
 
             if sigops > MAX_BLOCK_SIGOPS {
@@ -378,13 +391,25 @@ where
                 };
             }
 
-            match state_service
+            let state_commit_started = std::time::Instant::now();
+            let state_commit_result = state_service
                 .ready()
                 .await
                 .map_err(|source| VerifyBlockError::StateService { source, hash })?
                 .call(zs::Request::CommitSemanticallyVerifiedBlock(prepared_block))
-                .await
-            {
+                .await;
+            // A slow response here means the block sat in the state's
+            // parent-waiting queue or behind the write task; this stage log
+            // attributes driver-side commit timeouts to their pipeline stage.
+            if state_commit_started.elapsed() > std::time::Duration::from_secs(20) {
+                tracing::warn!(
+                    ?height,
+                    ?hash,
+                    elapsed = ?state_commit_started.elapsed(),
+                    "slow state commit stage while committing block"
+                );
+            }
+            match state_commit_result {
                 Ok(zs::Response::Committed(committed_hash)) => {
                     assert_eq!(committed_hash, hash, "state must commit correct hash");
                     Ok(hash)

zebra-consensus/src/transaction.rs

@@ -743,14 +743,33 @@ where
 
                     utxo
                 } else {
+                    let lookup_started = std::time::Instant::now();
                     let response = state
                         .clone()
                         .oneshot(zebra_state::Request::AwaitUtxo(*outpoint))
                         .await
                         .map_err(|boxed_error| match boxed_error.downcast::<Elapsed>() {
-                            Ok(_) => TransactionError::TransparentInputNotFound,
+                            Ok(_) => {
+                                // A timed-out lookup held this transaction's whole
+                                // block for UTXO_LOOKUP_TIMEOUT; chains of these
+                                // serialize into multi-minute commit stalls that
+                                // are otherwise unattributable from logs.
+                                tracing::warn!(
+                                    ?outpoint,
+                                    elapsed = ?lookup_started.elapsed(),
+                                    "transparent input UTXO lookup timed out"
+                                );
+                                TransactionError::TransparentInputNotFound
+                            }
                             Err(boxed_error) => TransactionError::from(boxed_error),
                         })?;
+                    if lookup_started.elapsed() > std::time::Duration::from_secs(20) {
+                        tracing::warn!(
+                            ?outpoint,
+                            elapsed = ?lookup_started.elapsed(),
+                            "slow transparent input UTXO lookup delayed transaction verification"
+                        );
+                    }
 
                     if let zebra_state::Response::Utxo(utxo) = response {
                         utxo