diff --git a/crates/data-client/src/reqwest/client.rs b/crates/data-client/src/reqwest/client.rs index 93b662ec..cc371d6a 100644 --- a/crates/data-client/src/reqwest/client.rs +++ b/crates/data-client/src/reqwest/client.rs @@ -36,9 +36,7 @@ pub struct ReqwestDataClient { impl Debug for ReqwestDataClient { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - f.debug_struct("ReqwestDataClient") - .field("url", &self.url.as_str()) - .finish() + f.write_str(self.url.as_str()) } } diff --git a/crates/data-source/src/standard.rs b/crates/data-source/src/standard.rs index f896c2dd..7b8b5c1c 100644 --- a/crates/data-source/src/standard.rs +++ b/crates/data-source/src/standard.rs @@ -5,7 +5,7 @@ use futures::{future::BoxFuture, stream::BoxStream, FutureExt, Stream, StreamExt use sqd_data_client::{BlockStreamRequest, BlockStreamResponse, DataClient}; use sqd_primitives::{Block, BlockNumber, BlockRef}; use tokio::time::Sleep; -use tracing::warn; +use tracing::{info, warn}; use crate::types::{DataEvent, DataSource}; @@ -78,6 +78,14 @@ impl DataSourceState { } Poll::Ready(Ok(BlockStreamResponse::Fork(prev_blocks))) => { ep.error_counter = 0; + info!( + data_source =? ep.client, + stream_from = req.first_block, + hint_count = prev_blocks.len(), + oldest_hint =? prev_blocks.first().map(|b| b.number), + newest_hint =? prev_blocks.last().map(|b| b.number), + "upstream reported a fork" + ); ep.state = EndpointState::Fork { req: req.clone(), prev_blocks @@ -289,11 +297,19 @@ where let forks = self.endpoints.iter().filter(|ep| ep.is_on_fork()).count(); if forks > 0 { - if forks > self.endpoints.len() / 2 - || forks == self.endpoints.iter().filter(|ep| ep.is_active()).count() - || self.fork_consensus_timeout(cx) - { - return Poll::Ready(DataEvent::Fork(self.extract_fork())); + let active = self.endpoints.iter().filter(|ep| ep.is_active()).count(); + if forks > self.endpoints.len() / 2 || forks == active || self.fork_consensus_timeout(cx) { + let chain = self.extract_fork(); + info!( + forked_endpoints = forks, + active_endpoints = active, + total_endpoints = self.endpoints.len(), + hint_count = chain.len(), + oldest_hint =? chain.first().map(|b| b.number), + newest_hint =? chain.last().map(|b| b.number), + "fork consensus reached" + ); + return Poll::Ready(DataEvent::Fork(chain)); } } else { self.state.fork_consensus_timeout = None diff --git a/crates/hotblocks-harness/README.md b/crates/hotblocks-harness/README.md index 167ac9ff..05347b46 100644 --- a/crates/hotblocks-harness/README.md +++ b/crates/hotblocks-harness/README.md @@ -22,6 +22,7 @@ is what makes the crash/restart and shutdown classes expressible at all. ```bash cargo test -p sqd-hotblocks-harness # the harness's own unit tests (model, chain, simulator) cargo test -p sqd-hotblocks --test ct1_happy_path # CT-1 — the Phase 0 exit criterion +cargo test -p sqd-hotblocks --test ct4_finality # CT-4 — finalized-prefix equivocation cargo test -p sqd-hotblocks --test ct9_source_faults ``` @@ -33,7 +34,7 @@ reusable lives here, so a future soak or benchmark runner can use it outside `ca | Module | What it is | Spec | |---|---|---| -| [`sim`](src/sim.rs) | source simulator: scripted chain, fork signals, finality headers, fault knobs | 13 §7, DEF-12 | +| [`sim`](src/sim.rs) | source simulator: scripted chain, fork signals, finality headers, fault knobs including explicit finalized-prefix equivocation | 13 §7, DEF-12 | | [`model`](src/model.rs) | the reference model — the oracle. Block-exact, well-formedness asserted after every transition | 12 §2 | | [`driver`](src/driver.rs) | client: the read binding, the structural validators, the anchored follower and backfill scanner | 04 §7, 12 §4 | | [`compare`](src/compare.rs) | quiescence comparator: diffs every observable, collects *all* violations before failing | 12 §1 | @@ -122,9 +123,12 @@ Fixed in `crates/data-client/src/reqwest/lines.rs`; pinned by a unit test there - **CT-2 (crash/restart)** — `Sut::crash()`, `Sut::stop()`, `Sut::restart()` already exist and keep the same database directory and port across boots. What is missing is the kill-point matrix. -- **CT-4 (fork/finality corpus)** — `Harness::fork()` and the model's `resolve_fork` / - `Finalize::IntegrityFault` are implemented and unit-tested; the follower implements the - normative CONFLICT recovery of 04 §7. What is missing is the scripts. +- **CT-4 (fork/finality corpus)** — `ct4_finality` drives source-equivocation faults through the + real binary at both the retained-window floor and a deterministic two-chunk layout with finality + strictly inside the second chunk. Both prove rollback resumes at `fin + 1` and the accepted + finalized prefix remains unchanged. `Harness::fork()`, the model's `resolve_fork` / + `Finalize::IntegrityFault`, and the follower's normative CONFLICT recovery of 04 §7 support the + remaining successful-reorg, below-window, malformed-finality, fork-storm and alarm scripts. - **CT-5 (error taxonomy)** — `ct5_error_soundness` covers unsupported-dialect containment, error classification, and mid-stream worker-panic abort; the anchored check across large sparse-number holes is deferred (GAP-21, test `#[ignore]`d). `Model::predict_query` supplies diff --git a/crates/hotblocks-harness/src/harness.rs b/crates/hotblocks-harness/src/harness.rs index 676198e6..a6a8c083 100644 --- a/crates/hotblocks-harness/src/harness.rs +++ b/crates/hotblocks-harness/src/harness.rs @@ -29,6 +29,8 @@ pub struct HarnessConfig { /// Dense (evm, hyperliquid) or sparse (Solana slots) block numbering. pub numbering: Numbering, pub retention: Retention, + /// Keep physical ingest chunks separate when a test needs a deterministic storage layout. + pub disable_compaction: bool, /// Whether the service is told the anchor hash. If not, the anchor is `⊥` (DEF-7) and the /// first block's parent is unverifiable. pub anchored: bool, @@ -54,6 +56,7 @@ impl HarnessConfig { number: start_block, parent_hash: Some(block_hash(start_block - 1, 0)) }, + disable_compaction: false, anchored: true, source_poll: Duration::from_millis(200), rust_log: "info".to_string(), @@ -96,6 +99,7 @@ impl Harness { id: cfg.dataset.clone(), kind: cfg.chain.config_kind().to_string(), retention: cfg.retention.clone(), + disable_compaction: cfg.disable_compaction, sources: vec![sim.base_url(&cfg.dataset)] }] ); diff --git a/crates/hotblocks-harness/src/sim.rs b/crates/hotblocks-harness/src/sim.rs index 375fe46b..b67f3719 100644 --- a/crates/hotblocks-harness/src/sim.rs +++ b/crates/hotblocks-harness/src/sim.rs @@ -87,6 +87,8 @@ pub struct SimFaults { #[derive(Clone, Copy, Debug, Default)] pub struct SimStats { pub stream_requests: u64, + /// `fromBlock` on the most recent request, for rollback-position assertions. + pub last_stream_from: Option, pub blocks_served: u64, pub fork_signals: u64, pub no_data: u64, @@ -161,6 +163,18 @@ impl SourceSim { Ok(blocks) } + /// Fault injection for CT-4/FM-SRC-5: replace a suffix that includes the source's own + /// finalized head and claim the replacement tip as final. + /// + /// Unlike [`Self::fork`], this deliberately violates source finality. It has no model-side + /// counterpart: the last accepted model state remains the oracle while the SUT rejects the + /// equivocating source. + pub fn equivocate_finalized_prefix(&self, dataset: &str, from: BlockNumber, len: u32) -> Result<()> { + self.try_with(dataset, |d| d.equivocate_finalized_prefix(from, len))?; + self.bump(); + Ok(()) + } + /// Declare `number` (and everything below it) final. pub fn finalize(&self, dataset: &str, number: BlockNumber) -> Result { let r = self.try_with(dataset, |d| d.finalize(number))?; @@ -301,22 +315,49 @@ impl DatasetSim { } fn fork(&mut self, from: BlockNumber, len: u32) -> Result> { + self.validate_fork_position(from)?; + ensure!( + self.fin.as_ref().is_none_or(|f| f.number < from), + "the script forks at or below the source's own finalized head — an equivocating source \ + belongs to the CT-4 fault corpus, not to a well-formed script" + ); + Ok(self.replace_suffix(from, len)) + } + + fn equivocate_finalized_prefix(&mut self, from: BlockNumber, len: u32) -> Result<()> { + self.validate_fork_position(from)?; + ensure!(len > 0, "a finality-equivocation fault must mint a replacement tip"); + let finalized = self + .fin + .as_ref() + .context("a finality-equivocation fault requires an existing finalized head")?; + ensure!( + from <= finalized.number, + "equivocation at {from} does not replace finalized block {}", + finalized.number + ); + + let replacement = self.replace_suffix(from, len); + self.fin = Some(replacement.last().expect("a non-empty replacement has a tip").as_ref()); + Ok(()) + } + + fn validate_fork_position(&self, from: BlockNumber) -> Result<()> { ensure!( from >= self.start, "fork at {from} is below the source's first block {}", self.start ); ensure!(from <= self.next_number(), "fork at {from} is above the source's chain"); - ensure!( - self.fin.as_ref().is_none_or(|f| f.number < from), - "the script forks at or below the source's own finalized head — an equivocating source \ - belongs to the CT-4 fault corpus, not to a well-formed script" - ); + Ok(()) + } + + fn replace_suffix(&mut self, from: BlockNumber, len: u32) -> Vec { let keep = self.chain.partition_point(|b| b.number < from); self.chain.truncate(keep); self.fork_id = self.next_fork_id; self.next_fork_id += 1; - Ok(self.produce(len)) + self.produce(len) } fn finalize(&mut self, number: BlockNumber) -> Result { @@ -342,6 +383,7 @@ impl DatasetSim { fn respond(&mut self, req: &StreamReq) -> Reply { self.stats.stream_requests += 1; + self.stats.last_stream_from = Some(req.from_block); let parent_pos = req.from_block.saturating_sub(1); if let Some(asserted) = &req.parent_block_hash { diff --git a/crates/hotblocks-harness/src/sut.rs b/crates/hotblocks-harness/src/sut.rs index 876eb1d0..354c8574 100644 --- a/crates/hotblocks-harness/src/sut.rs +++ b/crates/hotblocks-harness/src/sut.rs @@ -39,6 +39,8 @@ pub struct DatasetSpec { pub id: String, pub kind: String, pub retention: Retention, + /// Preserve response-aligned chunks for tests that exercise physical layout boundaries. + pub disable_compaction: bool, pub sources: Vec } @@ -235,6 +237,7 @@ impl Sut { Retention::Api => yaml.push_str(" retention_strategy: Api\n"), Retention::None => yaml.push_str(" retention_strategy: None\n") } + yaml.push_str(&format!(" disable_compaction: {}\n", ds.disable_compaction)); yaml.push_str(" data_sources:\n"); for src in &ds.sources { yaml.push_str(&format!(" - \"{src}\"\n")); diff --git a/crates/hotblocks/spec/12-conformance-tdd.md b/crates/hotblocks/spec/12-conformance-tdd.md index 49441293..7b1e8a2f 100644 --- a/crates/hotblocks/spec/12-conformance-tdd.md +++ b/crates/hotblocks/spec/12-conformance-tdd.md @@ -4,11 +4,13 @@ This document turns the spec into a test program: the reference model (oracle), harness architecture, the test-class taxonomy, the traceability matrix, and the dated gap register that seeds the hardening backlog. -Statuses and the gap register reflect the state of knowledge as of **2026-07-12** and are +Statuses and the gap register reflect the state of knowledge as of **2026-07-15** and are expected to change; everything else in this document is stable methodology. The harness described here exists: [`crates/hotblocks-harness`](../../hotblocks-harness). -Phase 0 of §7 is done — CT-1 runs a happy-path script green against the real binary. +Phase 0 of §7 is done — CT-1 runs a happy-path script green against the real binary, and CT-4 +now includes finality-equivocation regressions for both deep-window and straddling-chunk fallback, +plus an honest-reorg recovery that must converge rather than wedge. ## 1. Harness architecture @@ -229,9 +231,9 @@ INV-21/22/23 (checks in parentheses): ## 5. Traceability matrix (status @ 2026-07-15) -Legend: **C** covered, **P** partial (some storage-layer or fixture coverage exists; -service-level black-box coverage absent), **U** untested. Rows that changed with Phase 0 name -the test that moved them; unless a row says otherwise, "covered" means *on the happy path* — +Legend: **C** covered, **P** partial (some paths or layers are covered; the full class is not), +**U** untested. Rows that changed name the test that moved them; unless a row says otherwise, +"covered" means *on the happy path* — the same property under forks, crashes and retention is the business of CT-2/CT-4. | Property | CT class | Status | Note | @@ -242,15 +244,15 @@ the same property under forks, crashes and retention is the business of CT-2/CT- | INV-7 provenance | CT-1/6 | **C** | `ct1_happy_path`: what the source served is read back, payload included | | INV-10 atomic transitions | CT-2/3 | U | | | INV-11 append | CT-1 | **C** | | -| INV-12/13 finality monotone/immutable | CT-1/4 | **P — known-violated** | monotone advance observed; composed-finality REPLACE below `fin` admitted (GAP-22); regression/conflict paths await CT-4 | -| INV-14 fork floor | CT-4 | **U — known-violated** | GAP-3; composed-finality bypass (GAP-22) | +| INV-12/13 finality monotone/immutable | CT-1/4 | **P** | monotone advance in CT-1; two `ct4_finality` scenarios + write-controller regressions pin finalized-prefix immutability, fixed-height hash immutability and atomic rejection; stale/regressing finality and fork-storm boundary cases remain | +| INV-14 fork floor | CT-4 | **P — known-violated** | `ct4_finality` covers all-mismatching-hint fallback at both the retained-window floor and a finality-straddling chunk; below-window RESET and trimmed-anchor cases remain (GAP-3/23) | | INV-15/18 retention trim/anchor | CT-1 | **U — known-violated** | INV-18: trims drop the anchor hash (GAP-23), restart rebuilds it wrong (GAP-2); comparator needs RS-4 slack first (see §7) | | INV-16 frame | CT-1/7 | U | | | INV-17 maintenance transparency | CT-7 | P | merge-equivalence tested storage-level | | INV-20 snapshot isolation | CT-3 | P | single-threaded snapshot test only | | INV-21/22 response shape/completeness | CT-1/5/6 | P | structural validators + emission diff under `include_all`; coverage cuts and filtered emission await CT-5; comparator must implement the RP-9 marker exemption (spec change 2026-07-12) | -| INV-23 anchored ancestry | CT-1/4 | P | anchored continuation across responses covered; the CONFLICT path awaits CT-4 | -| INV-24 finalized-only | CT-4 | U | | +| INV-23 anchored ancestry | CT-1/4 | P | anchored continuation across responses, finality-conflict rejection, and honest reorg recovery above finality covered; the broader fork corpus remains | +| INV-24 finalized-only | CT-4 | P | `ct4_finality` pins once-finalized content across source equivocation with a public full-window scan; a dedicated `QUERY-FINALIZED` poller and the broader fork corpus remain | | INV-25 progress | CT-1/6 | P | a successful response must cover ≥ 1 block — asserted by the scanner | | INV-26 error soundness | CT-5 | **P — known-violated** | `ct5_error_soundness`: unsupported dialect containment/accounting and mid-stream worker-panic abort pinned; finalized-snapshot race pinned unit-level. Anchored eval across large holes (GAP-21) reverted; shared-status families keep free-text discrimination (GAP-36/39) | | INV-27 range honesty | CT-1 | **C** | validator: no block outside `[from, min(to, head)]` | @@ -316,8 +318,7 @@ rare, P3 = polish. **First test** names the cheapest failing-test-first entry po | GAP-18 | Dual-writer detection exists only on some paths (finality/head updates), not all mutations | WP-15, FM-OP-3 | P3 | CT-5: two harness-driven writers, assert loser stops on every mutation type | | GAP-20 | `parent_number` linkage is never validated on any layer (the block trait exposes it; nothing reads it): a hash-linked run can claim an arbitrarily higher number for the next block, storing a false hole on a densely-numbered chain — a silent data gap served as if it were a slot gap. (The originally-filed non-monotonic-numbers scenario is unreachable today: the source-position advance forces ascending numbers.) | WP-2, DEF-4, INV-1 | P2 | CT-4/CT-9: hash-linked run with a number jump on a dense chain; the run MUST be rejected with no state change | | GAP-21 | An anchored query whose `from` sits mid-chunk above a number gap larger than the conflict-check lookback (a hard-coded 100 positions in the plan's base-block check) fails `INTERNAL` instead of evaluating the assertion. A >100-position hole with an anchor landing just above it is probably unrealistic, hence low priority. A correct all-predecessors scan was tried and reverted 2026-07-15 — it regressed `check_parent_block` into an unbounded per-chunk scan+sort; needs a lazy sort-desc + limit(100) | RP-11, INV-26 | P3 | CT-5 `ct5_anchor_is_evaluated_across_a_large_number_hole` (`#[ignore]` until fixed): >100-position hole in one chunk; anchored query just above must yield OK/CONFLICT, never 500 | -| GAP-22 | Deep-fork handling can silently replace the finalized prefix: the fork-resolution fallback ignores `fin` (resumes from the window start instead of `⟨fin + 1, fin.hash⟩`), the composed-finality guard admits a REPLACE whose base lies at/below `fin` whenever the pack carries a finality mark ≥ current, and Window trims have dropped the anchor hash (GAP-23) so the replacement attaches unchecked. If the first replacement batch reaches past the old `fin`, the finalized prefix is replaced with no RESET event and no alarm — finalized-only clients observe two hashes at one height (INV-24 broken); otherwise the commit trips the fork-floor check ("can't fork safely") and the epoch parks on the blind 60 s retry loop. Fix: fallback → `fin + 1`; enforce the fork floor at commit unconditionally; carry the anchor hash | WP-6, INV-13/14, INV-24, FM-SRC-5, LIV-9 | **P1** | CT-4: fork with all-mismatching hints on a dataset with `fin` defined; assert REPLACE from `fin + 1` (or alarmed fault) — never a commit whose base ≤ `fin` | -| GAP-23 | Window trims drop the anchor hash: the automatic trim passes no hash and the retained state stores `⊥`, though the correct value sits unused in the first batch's `parent_block_hash`. Disables below-window divergence detection (WP-6b has nothing to contradict) and feeds GAP-22 | INV-18, DEF-7, WP-6b | P1 | CT-1: CONFLICT hints / STATUS at the window edge after a trim; CT-4: below-window fork after a trim must RESET, not absorb silently | +| GAP-23 | Window trims drop the anchor hash: the automatic trim passes no hash and the retained state stores `⊥`, though the correct value sits unused in the first batch's `parent_block_hash`. Disables below-window divergence detection (WP-6b has nothing to contradict); it was also one precondition of the now-closed GAP-22 | INV-18, DEF-7, WP-6b | P1 | CT-1: CONFLICT hints / STATUS at the window edge after a trim; CT-4: below-window fork after a trim must RESET, not absorb silently | | GAP-24 | One dataset's init failure aborts the whole service: startup propagates the first controller error (kind mismatch, retention bail, corrupt state) instead of alarming that dataset and serving the rest | CN-10, FM-OP-1, INV-36, INV-43 | P1 | CT-5 boot matrix: corrupt one dataset's persisted state; assert the others serve and the broken one alarms | | GAP-25 | Downward retention (`from < first(D)`) executes as an *implicit, unobservable* RESET (WP §2.5 as amended 2026-07-12 legalizes the destruction, but requires OB-9 observability) — no event, indistinguishable from a trim; and the boot-time `Pinned` equivalent aborts the entire service (via GAP-24) instead of a dataset-level refusal | WP §2.5, OB-9, INV-43 | P2 | CT-1: SET-RETENTION below `first`; assert a RESET observable + serving continuity; CT-5: boot with lowered `Pinned.from` | | GAP-26 | A block timestamp outside the datetime-conversion range kills the ingest flush — the conversion exists only for a log line — and the dataset enters a permanent crash-loop on redelivery | FM-1, WP-18, CN-8 | P1 | CT-9: serve a block with `time = i64::MAX`; the dataset must ingest it and keep serving | @@ -341,6 +342,7 @@ rare, P3 = polish. **First test** names the cheapest failing-test-first entry po | GAP-11 | Unsupported `substrate` / `fuel` queries and query-worker panics escaped the HTTP error taxonomy by panicking their request task | Typed `UNSUPPORTED_QUERY` admission errors, panic containment in the query executor, CT-5 dialect requests, and an executor unit test (2026-07-15) | | GAP-16 | No service-level automated tests | [`crates/hotblocks-harness`](../../hotblocks-harness) + `ct1_happy_path` (Phase 0, 2026-07-12) | | GAP-19 | A source response whose final JSONL record carried no trailing newline panicked the line reader (`LineStream::take_final_line` left its scan position past the emptied buffer). The ingest task died, its buffered batch was lost, and the dataset parked for `P-EPOCH-RETRY` — then crash-looped, since the source served the same body on retry. Violated FM-1, LIV-2 | Found by CT-1 on the harness's first run; fixed in `crates/data-client/src/reqwest/lines.rs`; pinned by a unit test there and by `ct9_source_faults` (2026-07-12) | +| GAP-22 | Fork-resolution fallback and composed finality could silently replace the finalized prefix | PR #96: fork resolution resumes at a stored chunk boundary (never a mid-chunk `fin + 1`), and the write path admits a replacement reaching to/below `fin` only when it reproduces the finalized block's hash there — otherwise it refuses atomically. This closes the silent replace while avoiding the wedge a `fin + 1` clamp causes when finality sits inside a chunk (`insert_fork` cannot split it). Pinned by three `ct4_finality` scenarios (deep-window and straddling-chunk equivocation refused; honest reorg above finality recovers) plus seven write-controller unit tests (2026-07-16). Remaining below-window anchor and source-alarm work is tracked by GAP-23/GAP-5/GAP-30. The write-path guard assumes re-ingest never flushes a chunk ending strictly below `fin` — else it refuses the whole-chunk rewrite and the dataset takes the bounded 60 s loop; today's 200k-row / mask-change / `MaybeOnHead` flushers satisfy this on any source with homogeneous per-block data availability | | GAP-32 | A finalized-head trim/reset race could turn an admitted finalized query into `INTERNAL` when its snapshot no longer had a finalized head | Snapshot-time absence now maps to `NO_DATA`; pinned by `finalized_snapshot_without_a_head_is_no_data` (2026-07-15) | | GAP-38 | `TX-BY-HASH` / `tidx` absent; fork re-inclusion ordering unimplemented | Transaction hash CF + independent flag + HTTP binding; storage transition suite and black-box ingest/reorg/re-inclusion tests (2026-07-15) | @@ -352,13 +354,12 @@ rare, P3 = polish. **First test** names the cheapest failing-test-first entry po Delivered as [`crates/hotblocks-harness`](../../hotblocks-harness), driving the real binary as a child process over the binding of 13. Its README records the design decisions the - next phases must not undo. Three things the later phases need are built but unexercised, and - three are missing: + next phases must not undo. The current harness support and remaining corpus are: - | Built, awaiting scripts | Missing | + | Harness support | Remaining | |---|---| | `Sut::crash/stop/restart` (same db, same port) → CT-2 | the CT-2 kill-point matrix | - | `Harness::fork` + `Model::resolve_fork` + the follower's CONFLICT recovery → CT-4 | the CT-4 fork/finality corpus | + | `Harness::fork`, explicit finalized-prefix equivocation, `Model::resolve_fork` and follower CONFLICT recovery → CT-4 | `ct4_finality` is the first script; successful reorg, below-window RESET, malformed finality and alarm cases remain | | `Model::predict_query` + initial `ct5_error_soundness` matrix | remaining CT-5 binding, boot, and overload rows | | `SimFaults` injection point → CT-9 | the rest of the FM-SRC repertoire | diff --git a/crates/hotblocks/src/dataset_controller/dataset_controller.rs b/crates/hotblocks/src/dataset_controller/dataset_controller.rs index eb107ca0..3d8220d0 100644 --- a/crates/hotblocks/src/dataset_controller/dataset_controller.rs +++ b/crates/hotblocks/src/dataset_controller/dataset_controller.rs @@ -18,7 +18,7 @@ use crate::{ ingest_generic::{IngestMessage, NewChunk}, write_controller::WriteController }, - metrics::{WriteStage, report_hash_index_write_metrics, report_write_duration}, + metrics::{WriteStage, report_hash_index_write_metrics, report_ingest_fork, report_write_duration}, types::{DBRef, DatasetKind, RetentionStrategy} }; @@ -206,6 +206,7 @@ impl WriteCtx { prev_blocks, rollback_sender } => { + report_ingest_fork(self.dataset_id); self.write.compute_rollback(&prev_blocks).map(|rollback| { let _ = rollback_sender.send(rollback); })?; diff --git a/crates/hotblocks/src/dataset_controller/ingest_generic.rs b/crates/hotblocks/src/dataset_controller/ingest_generic.rs index 621fd113..16e7eddf 100644 --- a/crates/hotblocks/src/dataset_controller/ingest_generic.rs +++ b/crates/hotblocks/src/dataset_controller/ingest_generic.rs @@ -167,7 +167,11 @@ where } async fn handle_fork(&mut self, prev_blocks: Vec) -> anyhow::Result<()> { - info!(upstream_blocks = valuable(&prev_blocks), "fork received"); + info!( + stream_from = self.first_block, + upstream_blocks = valuable(&prev_blocks), + "fork received" + ); let (rollback_sender, rollback_recv) = tokio::sync::oneshot::channel(); @@ -182,16 +186,16 @@ where let rollback = rollback_recv.await?; info!( - block_number = rollback.first_block, - parent_block_hash =? rollback.parent_block_hash, + resume_from = rollback.resume_from, + expected_parent_hash =? rollback.expected_parent_hash, "resetting ingest position" ); self.buffered_blocks = 0; self.finalized_head = None; - self.first_block = rollback.first_block; + self.first_block = rollback.resume_from; self.data_source - .set_position(rollback.first_block, rollback.parent_block_hash.as_deref()); + .set_position(rollback.resume_from, rollback.expected_parent_hash.as_deref()); Ok(()) } diff --git a/crates/hotblocks/src/dataset_controller/write_controller.rs b/crates/hotblocks/src/dataset_controller/write_controller.rs index 7f315635..f434c167 100644 --- a/crates/hotblocks/src/dataset_controller/write_controller.rs +++ b/crates/hotblocks/src/dataset_controller/write_controller.rs @@ -5,10 +5,18 @@ use tracing::{debug, field::valuable, info, instrument, warn}; use crate::types::{DBRef, DatasetKind}; +/// Source position selected after resolving a fork against stored history. +/// +/// `resume_from` always lands on a stored chunk boundary, so it may sit at or below the +/// finalized head when the fork's common ancestor is inside a finality-straddling chunk. The +/// finalized prefix is protected on the write path instead (see [`WriteController::new_chunk`]), +/// which rejects a replacement that would rewrite an already-finalized block. #[derive(Debug)] pub struct Rollback { - pub first_block: BlockNumber, - pub parent_block_hash: Option + /// Lowest block number the source may return after resolving the fork. + pub resume_from: BlockNumber, + /// Hash that must anchor the first returned block, when an anchor is known. + pub expected_parent_hash: Option } #[derive(Debug)] @@ -99,7 +107,13 @@ impl WriteController { None => bail!("all passed prev blocks lie below finalized head") }; if prev[pos].number == finalized_head.number { - ensure!(prev[pos].hash == finalized_head.hash); + ensure!( + prev[pos].hash == finalized_head.hash, + "fork hint at finalized block {} conflicts: expected {}, got {}", + finalized_head.number, + finalized_head.hash, + prev[pos].hash + ); } prev = &prev[pos..] } @@ -124,21 +138,21 @@ impl WriteController { if let Some(&b) = prev_blocks.peek() { if b.number == head.last_block() && b.hash == head.last_block_hash() { return Ok(Rollback { - first_block: b.number + 1, - parent_block_hash: Some(b.hash.clone()) + resume_from: b.number + 1, + expected_parent_hash: Some(b.hash.clone()) }); } } else { return Ok(Rollback { - first_block: head.last_block() + 1, - parent_block_hash: Some(head.last_block_hash().to_string()) + resume_from: head.last_block() + 1, + expected_parent_hash: Some(head.last_block_hash().to_string()) }); } } Ok(Rollback { - first_block: self.first_block, - parent_block_hash: self.parent_block_hash.clone() + resume_from: self.first_block, + expected_parent_hash: self.parent_block_hash.clone() }) } @@ -366,20 +380,54 @@ impl WriteController { let finalized_head = self .db .update_dataset_with_hash_index_metrics(self.dataset_id, metrics, |tx| { - let new_finalized_head = match (finalized_head, tx.label().finalized_head()) { - (Some(new), None) => Some(new), - (Some(new), Some(current)) if new.number >= current.number => Some(new), - (_, Some(current)) if current.number < chunk.first_block() => Some(current), - (_, Some(_)) => bail!( - "can't fork safely, because fork base is below the current finalized head \ - and finalized head of the data pack is below the current" - ), + let current_finalized_head = tx.label().finalized_head().cloned(); + + // A fork whose common ancestor sits inside a finality-straddling chunk resumes at + // that chunk's boundary, at or below `fin`, so the replacement can reach into the + // finalized region. Permit it only when it reproduces the finalized block exactly: + // it must span `fin` (else the swap would momentarily drop the finalized block) and + // carry `fin`'s hash unchanged. A mismatch is a source equivocating below its own + // finality and is refused, leaving the accepted prefix intact (INV-12/13/14). + if let Some(current) = current_finalized_head.as_ref() + && chunk.first_block() <= current.number + { + ensure!( + chunk.last_block() >= current.number, + "replacement chunk {}-{} would drop finalized block {} without reproducing it", + chunk.first_block(), + chunk.last_block(), + current.number + ); + if let Err(actual) = tx.validate_parent_block_hash(chunk, current.number + 1, ¤t.hash)? { + bail!( + "replacement would rewrite finalized block {}: expected hash {}, got {}", + current.number, + current.hash, + actual + ); + } + } + + let new_finalized_head = match (finalized_head, current_finalized_head.as_ref()) { + (Some(new), Some(current)) if new.number < current.number => Some(current.clone()), + (Some(new), Some(current)) if new.number == current.number => { + ensure!( + new.hash == current.hash, + "finalized hash mismatch at block {}: expected {}, got {}", + current.number, + current.hash, + new.hash + ); + Some(current.clone()) + } + (Some(new), _) => Some(new.clone()), + (None, Some(current)) => Some(current.clone()), (None, None) => None }; let new_finalized_head = new_finalized_head.map(|head| { if head.number < chunk.last_block() { - head.clone() + head } else { get_chunk_head(&chunk) } @@ -412,3 +460,265 @@ fn get_chunk_head(chunk: &Chunk) -> BlockRef { hash: chunk.last_block_hash().to_string() } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use sqd_storage::db::{DatabaseSettings, HashIndexWriteMetrics}; + + use super::*; + + fn chunk(first_block: BlockNumber, last_block: BlockNumber, parent_hash: &str, last_hash: &str) -> Chunk { + Chunk::V1 { + first_block, + last_block, + last_block_hash: last_hash.to_string(), + parent_block_hash: parent_hash.to_string(), + first_block_time: None, + last_block_time: None, + tables: Default::default() + } + } + + fn setup() -> anyhow::Result<(tempfile::TempDir, DBRef, DatasetId, WriteController)> { + let db_dir = tempfile::tempdir()?; + let db = Arc::new(DatabaseSettings::default().open(db_dir.path())?); + let dataset_id = DatasetId::from_str("finality-regression"); + let write = WriteController::new(db.clone(), dataset_id, DatasetKind::Evm)?; + Ok((db_dir, db, dataset_id, write)) + } + + fn seed_chain(write: &mut WriteController, metrics: &mut HashIndexWriteMetrics) -> anyhow::Result<(Chunk, Chunk)> { + let first = chunk(0, 5, "genesis", "old-5"); + let second = chunk(6, 9, "old-5", "old-9"); + write.new_chunk(None, &first, metrics)?; + write.new_chunk(None, &second, metrics)?; + Ok((first, second)) + } + + #[test] + fn replacement_rewriting_the_finalized_block_is_rejected() -> anyhow::Result<()> { + // Arrange: block 5 is finalized on the original chain. + let (_db_dir, db, dataset_id, mut write) = setup()?; + let mut metrics = HashIndexWriteMetrics::default(); + let (first, second) = seed_chain(&mut write, &mut metrics)?; + let current_finalized = BlockRef { + number: 5, + hash: "old-5".to_string() + }; + write.finalize(¤t_finalized)?; + + // Act: an honest reorg would resume at a boundary below fin, but this replacement reaches + // the finalized height carrying a different hash there — a source equivocating below finality. + let replacement = chunk(0, 5, "other-genesis", "new-5"); + let new_finalized = BlockRef { + number: 5, + hash: "new-5".to_string() + }; + let result = write.new_chunk(Some(&new_finalized), &replacement, &mut metrics); + + // Assert (INV-12/13/14): the finalized block is immutable and rejection is atomic. + assert!( + result.is_err(), + "replacement rewriting the finalized block was accepted" + ); + assert_eq!(write.finalized_head(), Some(¤t_finalized)); + + let snapshot = db.snapshot(); + let stored = snapshot + .list_chunks(dataset_id, 0, None) + .collect::>>()?; + assert_eq!(stored, vec![first, second]); + assert_eq!( + snapshot + .get_label(dataset_id)? + .and_then(|label| label.finalized_head().cloned()), + Some(current_finalized) + ); + Ok(()) + } + + #[test] + fn composed_finality_rejects_hash_change_at_fixed_height() -> anyhow::Result<()> { + // Arrange: the finalized block is the boundary of the first stored chunk. + let (_db_dir, db, dataset_id, mut write) = setup()?; + let mut metrics = HashIndexWriteMetrics::default(); + let (first, second) = seed_chain(&mut write, &mut metrics)?; + let current_finalized = BlockRef { + number: 5, + hash: "old-5".to_string() + }; + write.finalize(¤t_finalized)?; + + // Act: a normal append carries a conflicting hash at the already-finalized height. + let append = chunk(10, 12, "old-9", "old-12"); + let conflicting_finality = BlockRef { + number: 5, + hash: "other-5".to_string() + }; + let result = write.new_chunk(Some(&conflicting_finality), &append, &mut metrics); + + // Assert (INV-12): fixed-height finality is immutable and the append is not partially committed. + assert!(result.is_err(), "finalized hash changed at a fixed height"); + assert_eq!(write.finalized_head(), Some(¤t_finalized)); + + let snapshot = db.snapshot(); + let stored = snapshot + .list_chunks(dataset_id, 0, None) + .collect::>>()?; + assert_eq!(stored, vec![first, second]); + assert_eq!( + snapshot + .get_label(dataset_id)? + .and_then(|label| label.finalized_head().cloned()), + Some(current_finalized) + ); + Ok(()) + } + + #[test] + fn replacement_below_finalized_that_misses_it_is_rejected() -> anyhow::Result<()> { + // Arrange: finality lies strictly inside [6, 9]. + let (_db_dir, db, dataset_id, mut write) = setup()?; + let mut metrics = HashIndexWriteMetrics::default(); + let (first, second) = seed_chain(&mut write, &mut metrics)?; + let current_finalized = BlockRef { + number: 7, + hash: "old-7".to_string() + }; + write.finalize(¤t_finalized)?; + + // Act: a partial replacement reaches below the finalized height but stops before reproducing + // it — accepting it would drop the finalized block until a later flush caught up. + let replacement = chunk(6, 6, "old-5", "new-6"); + let result = write.new_chunk(None, &replacement, &mut metrics); + + // Assert: refuse so the finalized block is never transiently absent, atomically. + assert!( + result.is_err(), + "replacement that drops the finalized block was accepted" + ); + assert_eq!(write.finalized_head(), Some(¤t_finalized)); + let snapshot = db.snapshot(); + let stored = snapshot + .list_chunks(dataset_id, 0, None) + .collect::>>()?; + assert_eq!(stored, vec![first, second]); + Ok(()) + } + + #[test] + fn rollback_without_matching_hints_resumes_from_window_start() -> anyhow::Result<()> { + // Arrange: finality lies inside a storage chunk and every supplied fork hint mismatches. + let (_db_dir, _db, _dataset_id, mut write) = setup()?; + let mut metrics = HashIndexWriteMetrics::default(); + seed_chain(&mut write, &mut metrics)?; + write.finalize(&BlockRef { + number: 4, + hash: "old-4".to_string() + })?; + let hints = vec![ + BlockRef { + number: 5, + hash: "fork-5".to_string() + }, + BlockRef { + number: 9, + hash: "fork-9".to_string() + }, + ]; + + // Act. + let rollback = write.compute_rollback(&hints)?; + + // Assert: with no matching boundary the fallback is the window start; the finalized prefix is + // guarded on the write path (new_chunk), not by clamping the resume position. + assert_eq!(rollback.resume_from, write.start_block()); + assert_eq!(rollback.expected_parent_hash, None); + Ok(()) + } + + #[test] + fn rollback_from_straddling_chunk_resumes_at_chunk_boundary() -> anyhow::Result<()> { + // Arrange: finality lies strictly inside [6, 9], and the lowest hint is in that chunk. + let (_db_dir, _db, _dataset_id, mut write) = setup()?; + let mut metrics = HashIndexWriteMetrics::default(); + seed_chain(&mut write, &mut metrics)?; + write.finalize(&BlockRef { + number: 7, + hash: "old-7".to_string() + })?; + let hints = vec![BlockRef { + number: 8, + hash: "fork-8".to_string() + }]; + + // Act: the hint mismatches inside the chunk that straddles finality. + let rollback = write.compute_rollback(&hints)?; + + // Assert: resume at the straddling chunk's lower boundary (block 6, below fin); new_chunk then + // verifies the finalized block survives the whole-chunk rewrite. Previously this clamped to 8 + // (fin + 1), a mid-chunk position insert_fork could not satisfy — the wedge. + assert_eq!(rollback.resume_from, 6); + assert_eq!(rollback.expected_parent_hash.as_deref(), Some("old-5")); + Ok(()) + } + + #[test] + fn rollback_with_all_hints_below_finalized_head_is_refused() -> anyhow::Result<()> { + // Arrange: finality at 7; every fork hint lies strictly below it. + let (_db_dir, _db, _dataset_id, mut write) = setup()?; + let mut metrics = HashIndexWriteMetrics::default(); + seed_chain(&mut write, &mut metrics)?; + write.finalize(&BlockRef { + number: 7, + hash: "old-7".to_string() + })?; + let hints = vec![ + BlockRef { + number: 3, + hash: "fork-3".to_string() + }, + BlockRef { + number: 5, + hash: "fork-5".to_string() + }, + ]; + + // Act + Assert: a fork that cannot reach up to finality is refused, never resumed below it. + let err = write.compute_rollback(&hints).unwrap_err(); + assert!( + err.to_string().contains("below finalized head"), + "unexpected error: {err}" + ); + Ok(()) + } + + #[test] + fn rollback_with_conflicting_hint_at_finalized_height_is_refused() -> anyhow::Result<()> { + // Arrange: finality at 7; a hint sits exactly at 7 but disagrees on its hash. + let (_db_dir, _db, _dataset_id, mut write) = setup()?; + let mut metrics = HashIndexWriteMetrics::default(); + seed_chain(&mut write, &mut metrics)?; + write.finalize(&BlockRef { + number: 7, + hash: "old-7".to_string() + })?; + let hints = vec![ + BlockRef { + number: 7, + hash: "fork-7".to_string() + }, + BlockRef { + number: 9, + hash: "fork-9".to_string() + }, + ]; + + // Act + Assert: an equivocation at the finalized height is refused, never absorbed. + let err = write.compute_rollback(&hints).unwrap_err(); + assert!(err.to_string().contains("finalized block 7"), "unexpected error: {err}"); + Ok(()) + } +} diff --git a/crates/hotblocks/src/metrics.rs b/crates/hotblocks/src/metrics.rs index f60a04ca..f3c7a7d8 100644 --- a/crates/hotblocks/src/metrics.rs +++ b/crates/hotblocks/src/metrics.rs @@ -60,6 +60,8 @@ pub static QUERY_ERROR_WORKER_PANIC: LazyLock = LazyLock::new(Default:: pub static COMPLETED_QUERIES: LazyLock = LazyLock::new(Default::default); +static INGEST_FORKS: LazyLock> = LazyLock::new(Default::default); + pub static STREAM_DURATIONS: LazyLock> = LazyLock::new(|| Family::new_with_constructor(|| Histogram::new(exponential_buckets(0.01, 2.0, 20)))); pub static STREAM_BYTES: LazyLock> = @@ -154,6 +156,10 @@ pub(crate) fn report_hash_index_write_metrics(dataset_id: DatasetId, metrics: &H } } +pub(crate) fn report_ingest_fork(dataset_id: DatasetId) { + INGEST_FORKS.get_or_create(&dataset_label!(dataset_id)).inc(); +} + pub fn report_query_too_many_tasks_error() { QUERY_ERROR_TOO_MANY_TASKS.inc(); } @@ -509,6 +515,11 @@ pub fn build_metrics_registry() -> Registry { "Number of completed queries", COMPLETED_QUERIES.clone() ); + registry.register( + "ingest_forks", + "Fork/rollback events resolved during ingestion", + INGEST_FORKS.clone() + ); top_registry } diff --git a/crates/hotblocks/tests/ct4_finality.rs b/crates/hotblocks/tests/ct4_finality.rs new file mode 100644 index 00000000..8472db9b --- /dev/null +++ b/crates/hotblocks/tests/ct4_finality.rs @@ -0,0 +1,235 @@ +//! CT-4 — an equivocating source must not rewrite the accepted finalized prefix, and an honest +//! reorg above finality must recover rather than wedge. +//! +//! Covers INV-12/13/14/24, WP-6 and FM-SRC-5 through the public service binding. Conflict windows +//! deliberately omit the old finalized block, forcing fork resolution to resume at a stored chunk +//! boundary — both at the retained-window floor and from a chunk that straddles finality. The +//! whole-chunk rewrite that follows is verified on the write path: an equivocating source reproduces +//! a different hash at the finalized height and is refused; an honest source reproduces it and the +//! service converges on the reorged chain. + +use std::{ + sync::Arc, + time::{Duration, Instant} +}; + +use anyhow::{Context, Result, ensure}; +use sqd_hotblocks_harness::{ + P_CONFLICT_WINDOW, + chain::HlFills, + harness::{Harness, HarnessConfig}, + types::BlockRef +}; + +const START: u64 = 1_000; +const DEEP_FORK_BLOCKS: u32 = (P_CONFLICT_WINDOW + 50) as u32; +const PREFIX_CHUNK_BLOCKS: u32 = 50; +const STRADDLING_CHUNK_BLOCKS: u32 = (P_CONFLICT_WINDOW + 50) as u32; +const FINALITY_LAG: u64 = P_CONFLICT_WINDOW + 20; +const REJECTION_TIMEOUT: Duration = Duration::from_secs(10); +const POLL: Duration = Duration::from_millis(50); + +#[tokio::test(flavor = "multi_thread")] +async fn ct4_finality_equivocation_does_not_replace_finalized_prefix() -> Result<()> { + let mut h = start_harness(false).await?; + + if let Err(err) = run_deep_fork(&mut h).await { + panic!("CT-4 failed: {err:?}"); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn ct4_straddling_chunk_rollback_respects_finalized_floor() -> Result<()> { + let mut h = start_harness(true).await?; + + if let Err(err) = run_straddling_chunk_fork(&mut h).await { + panic!("CT-4 straddling-chunk scenario failed: {err:?}"); + } + Ok(()) +} + +/// The honest dual of the fault scenarios: a legitimate reorg above `fin` whose common ancestor +/// lies inside a finality-straddling chunk must RECOVER, not wedge. Fork resolution resumes at the +/// chunk boundary (below `fin`) and rewrites the whole chunk; the write path verifies the finalized +/// block is reproduced and accepts. Before the finalized-floor fix this clamped to `fin + 1`, a +/// mid-chunk position `insert_fork` rejected, freezing the dataset on a 60-second restart loop. +#[tokio::test(flavor = "multi_thread")] +async fn ct4_honest_reorg_into_straddling_chunk_recovers() -> Result<()> { + let mut h = start_harness(true).await?; + + if let Err(err) = run_honest_reorg_recovery(&mut h).await { + panic!("honest-reorg recovery failed: {err:?}"); + } + Ok(()) +} + +async fn start_harness(disable_compaction: bool) -> Result { + let mut cfg = HarnessConfig::from_block(env!("CARGO_BIN_EXE_sqd-hotblocks"), Arc::new(HlFills), START); + cfg.disable_compaction = disable_compaction; + Harness::start(cfg).await +} + +async fn run_deep_fork(h: &mut Harness) -> Result<()> { + // Arrange: accept a chain whose finalized head is deeper than one conflict-hint window. + h.produce(DEEP_FORK_BLOCKS)?; + h.finalize_with_lag(FINALITY_LAG)?; + h.settle().await?; + h.assert_conforms().await?; + + assert_finality_equivocation_rejected(h, START, DEEP_FORK_BLOCKS).await +} + +async fn run_straddling_chunk_fork(h: &mut Harness) -> Result<()> { + // Arrange: commit two separate responses as [prefix] [finality-straddling chunk]. + h.produce(PREFIX_CHUNK_BLOCKS)?; + h.settle().await?; + h.assert_conforms().await?; + + let straddling_chunk_start = START + .checked_add(u64::from(PREFIX_CHUNK_BLOCKS)) + .context("the second chunk start overflows")?; + h.produce(STRADDLING_CHUNK_BLOCKS)?; + h.finalize_with_lag(FINALITY_LAG)?; + h.settle().await?; + h.assert_conforms().await?; + + let head = h.model.head().context("the accepted model has no head")?; + let fin = h + .model + .fin + .as_ref() + .context("the accepted model has no finalized head")?; + ensure!( + straddling_chunk_start < fin.number && fin.number < head.number, + "finalized block {} is not strictly inside the second chunk [{straddling_chunk_start}, {}]", + fin.number, + head.number + ); + + assert_finality_equivocation_rejected(h, straddling_chunk_start, STRADDLING_CHUNK_BLOCKS).await +} + +async fn run_honest_reorg_recovery(h: &mut Harness) -> Result<()> { + // Arrange: the same [prefix] [finality-straddling] layout, with `fin` strictly inside chunk 2. + h.produce(PREFIX_CHUNK_BLOCKS)?; + h.settle().await?; + let straddling_chunk_start = START + .checked_add(u64::from(PREFIX_CHUNK_BLOCKS)) + .context("the second chunk start overflows")?; + h.produce(STRADDLING_CHUNK_BLOCKS)?; + h.finalize_with_lag(FINALITY_LAG)?; + h.settle().await?; + h.assert_conforms().await?; + + let head = h.model.head().context("the accepted model has no head")?; + let fin = h + .model + .fin + .clone() + .context("the accepted model has no finalized head")?; + ensure!( + straddling_chunk_start < fin.number && fin.number < head.number, + "finalized block {} is not strictly inside the second chunk [{straddling_chunk_start}, {}]", + fin.number, + head.number + ); + + // Act: an honest reorg of the tip, above `fin` but inside the straddling chunk. Its common + // ancestor sits below `fin`, so resolution must resume at the chunk boundary and rewrite the + // whole chunk — reproducing the finalized block unchanged. + let reorg_from = fin.number + (head.number - fin.number) / 2; + ensure!( + fin.number < reorg_from && reorg_from <= head.number, + "the reorg point {reorg_from} must lie strictly above finality and on the chain" + ); + h.fork(reorg_from, STRADDLING_CHUNK_BLOCKS)?; + + // Assert: the service recovers onto the reorged chain and finality is preserved, not rewound. + h.settle().await?; + h.assert_conforms().await?; + let recovered_fin = h + .client + .finalized_head() + .await + .context("failed to read FINALIZED-HEAD after recovery")?; + ensure!( + recovered_fin.as_ref() == Some(&fin), + "finality moved during an honest recovery: expected {fin:?}, got {recovered_fin:?}" + ); + Ok(()) +} + +async fn assert_finality_equivocation_rejected(h: &Harness, fork_from: u64, replacement_blocks: u32) -> Result<()> { + let expected_head = h.model.head().context("the accepted model has no head")?; + let expected_fin = h + .model + .fin + .clone() + .context("the accepted model has no finalized head")?; + ensure!( + expected_head.number.saturating_sub(expected_fin.number) > P_CONFLICT_WINDOW, + "the first conflict window would include finalized block {}", + expected_fin.number + ); + let baseline_requests = h.sim.stats(&h.dataset).stream_requests; + + // Act: the source rewrites a suffix including `fin` and claims the new tip final. + // This is a source fault, so the reference model intentionally remains on the accepted fork. + h.sim + .equivocate_finalized_prefix(&h.dataset, fork_from, replacement_blocks)?; + let conflicting_source_head = h.sim.tip(&h.dataset).context("the faulty source has no head")?; + assert_ne!( + conflicting_source_head.hash, expected_head.hash, + "the fault did not mint a distinct source branch" + ); + + // Assert: fork resolution resumes at a chunk boundary and the write path refuses the rewrite + // (it reproduces a different hash at the finalized height), so every public watermark stays on + // the last accepted state throughout. + await_finality_fault_rejection(h, &expected_head, &expected_fin, baseline_requests).await?; + h.assert_conforms().await?; + Ok(()) +} + +async fn await_finality_fault_rejection( + h: &Harness, + expected_head: &BlockRef, + expected_fin: &BlockRef, + baseline_requests: u64 +) -> Result<()> { + // Hold the accepted watermarks still for the whole window. Adopting the equivocation would move + // HEAD/FINALIZED-HEAD off the accepted chain within a poll or two; refusing it keeps them fixed. + let deadline = Instant::now() + REJECTION_TIMEOUT; + loop { + let observed_head = h + .client + .head() + .await + .context("failed to read HEAD during fork recovery")?; + let observed_fin = h + .client + .finalized_head() + .await + .context("failed to read FINALIZED-HEAD during fork recovery")?; + ensure!( + observed_head.as_ref() == Some(expected_head), + "finality equivocation changed HEAD: expected {expected_head:?}, got {observed_head:?}" + ); + ensure!( + observed_fin.as_ref() == Some(expected_fin), + "finality equivocation changed FINALIZED-HEAD: expected {expected_fin:?}, got {observed_fin:?}" + ); + + if Instant::now() > deadline { + // Liveness: the SUT actually re-engaged the faulty source rather than idling. + let stats = h.sim.stats(&h.dataset); + ensure!( + stats.stream_requests > baseline_requests, + "the SUT never re-engaged the faulty source after the fault: {stats:?}" + ); + return Ok(()); + } + tokio::time::sleep(POLL).await; + } +}