From fd806303d14ebc5a79f17d1fc6f9a27070b8d390 Mon Sep 17 00:00:00 2001 From: Evgeny Formanenko Date: Thu, 16 Jul 2026 14:26:21 +0300 Subject: [PATCH] fix(hotblocks): enforce finalized fork floor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A fork whose common ancestor lies inside a finality-straddling chunk cannot resume at fin+1: that is a mid-chunk position, and insert_fork only replaces whole chunks, so the commit wedges the dataset on a 60s restart loop. Resuming below fin instead — the fallback's behaviour — silently rewrites the finalized prefix (GAP-22). An honest tip reorg triggers this whenever fin sits inside the head chunk, which is the common case, not just equivocation. Resolve the fork at the stored chunk boundary (which may sit at or below fin) and move the finalized-prefix guard to the write path: a replacement reaching into the finalized region is admitted only when it reproduces the finalized block's hash there, and must span fin so the whole-chunk rewrite never transiently drops it; otherwise it is refused atomically. This keeps the existing whole-chunk replace — no chunk-splitting primitive — and routes every equivocation shape through one loud, bounded 60s loop instead of the clamp's silent retry spin. Honest reorgs above finality now recover. Pinned by seven write-controller unit tests and three ct4_finality black-box scenarios (equivocation refused at both the window floor and a straddling chunk; honest reorg recovers) — the recovery scenario was verified red against the clamp. Co-Authored-By: Claude Opus 4.8 --- crates/data-client/src/reqwest/client.rs | 4 +- crates/data-source/src/standard.rs | 28 +- crates/hotblocks-harness/README.md | 12 +- crates/hotblocks-harness/src/harness.rs | 4 + crates/hotblocks-harness/src/sim.rs | 54 ++- crates/hotblocks-harness/src/sut.rs | 3 + crates/hotblocks/spec/12-conformance-tdd.md | 31 +- .../dataset_controller/dataset_controller.rs | 3 +- .../src/dataset_controller/ingest_generic.rs | 14 +- .../dataset_controller/write_controller.rs | 346 +++++++++++++++++- crates/hotblocks/src/metrics.rs | 11 + crates/hotblocks/tests/ct4_finality.rs | 235 ++++++++++++ 12 files changed, 687 insertions(+), 58 deletions(-) create mode 100644 crates/hotblocks/tests/ct4_finality.rs 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; + } +}