fix(hotblocks): enforce finalized fork floor

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 <[email protected]>

Author: mo4islona

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())
     }
 }
 

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<F> DataSourceState<F> {
                     }
                     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

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

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)]
             }]
         );

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<BlockNumber>,
     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<BlockRef> {
         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<Vec<Block>> {
+        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<Block> {
         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<BlockRef> {
@@ -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 {

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<String>
 }
 
@@ -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"));