fix(hotblocks): enforce finalized fork floor

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};
 
@@ -301,11 +301,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 |
@@ -131,15 +132,18 @@ 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 most of the scripts.
-  `ct4_lagging_source` covers the multi-endpoint shape production actually runs — several
-  sources per dataset, one of them behind (`HarnessConfig::sources`, `Harness::produce_ahead`).
-  Note `Harness::fork()` refuses to run with peers configured: reorging one endpoint of several
-  is a source *disagreement*, and what the service should do with it is the fork-consensus
-  question (`StandardDataSource::poll_next_event` — majority, or all-active, or a 2 s timeout).
-  That deserves a deliberate script, not an accidental one.
+- **CT-4 (fork/finality corpus)** — `ct4_finality` drives equivocation through the real binary at
+  the retained-window floor and at a two-chunk layout with finality inside the second chunk, plus
+  the honest duals: a reorg above `fin` recovers, so does one whose replay is cut below `fin`, and
+  a replay reproducing `fin`'s own hash while rewriting a block below it is refused. Rollback
+  resumes at a stored chunk boundary, which may sit below `fin`; the finalized prefix is verified
+  on the write path instead. `ct4_lagging_source` covers the multi-endpoint shape production
+  actually runs — several sources per dataset, one of them behind (`HarnessConfig::sources`,
+  `Harness::produce_ahead`). Note `Harness::fork()` refuses to run with peers configured: reorging
+  one endpoint of several is a source *disagreement*, and what the service should do with it is the
+  fork-consensus question (`StandardDataSource::poll_next_event` — majority, or all-active, or a
+  2 s timeout). That deserves a deliberate script, not an accidental one. The below-window,
+  malformed-finality, fork-storm and alarm scripts remain.
 - **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,
@@ -59,6 +61,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),
             sources: 1,
@@ -119,6 +122,7 @@ impl Harness {
                 id: cfg.dataset.clone(),
                 kind: cfg.chain.config_kind().to_string(),
                 retention: cfg.retention.clone(),
+                disable_compaction: cfg.disable_compaction,
                 sources: std::iter::once(&sim)
                     .chain(peers.iter())
                     .map(|s| s.base_url(&cfg.dataset))

crates/hotblocks-harness/src/model.rs

@@ -17,9 +17,9 @@ use crate::{
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum Finalize {
     Applied,
-    /// Stale or not-yet-applicable report; state unchanged (WP-7).
+    /// Stale or not-yet-applicable report; state unchanged (WP §2.4/WP-8).
     Ignored,
-    /// Finality contradicting stored data — must be alarmed, never accepted (WP-8, GAP-4/5).
+    /// Finality contradicting stored data — must be alarmed, never accepted (WP-8).
     IntegrityFault
 }
 
@@ -176,9 +176,13 @@ impl Model {
                 blocks[0].number
             );
         }
+        let previous = self.clone();
         self.seg.extend_from_slice(blocks);
         self.ver += 1;
-        self.apply_finality(fin)?;
+        if let Err(err) = self.apply_finality(fin) {
+            *self = previous;
+            return Err(err);
+        }
         self.wf();
         Ok(())
     }
@@ -198,11 +202,25 @@ impl Model {
             from >= first,
             "INV-14: REPLACE at {from} is below the window floor {first}"
         );
-        // ... nor below the finalized prefix (INV-13).
-        if let Some(fin) = &self.fin {
+        // ... nor alter the finalized prefix (INV-13). Reaching below `fin` is allowed only as an
+        // identical replay: a batch-granular implementation cannot start at `fin + 1` when finality
+        // sits inside a batch, and re-committing what is already there changes nothing.
+        if let Some(fin) = &self.fin
+            && from <= fin.number
+        {
+            let stored: Vec<_> = self
+                .seg
+                .iter()
+                .filter(|b| b.number >= from && b.number <= fin.number)
+                .collect();
+            let replayed: Vec<_> = blocks.iter().filter(|b| b.number <= fin.number).collect();
             ensure!(
-                from > fin.number,
-                "INV-13: REPLACE at {from} is at or below the finalized head {}",
+                stored.len() == replayed.len()
+                    && stored
+                        .iter()
+                        .zip(&replayed)
+                        .all(|(a, b)| a.number == b.number && a.hash == b.hash),
+                "INV-13: REPLACE at {from} alters the finalized prefix up to {}",
                 fin.number
             );
         }
@@ -227,14 +245,42 @@ impl Model {
                 }
             }
         }
+        if let Some(report) = fin
+            && self.replacement_evicts_reported_block(from, blocks, report)
+        {
+            bail!("INTEGRITY_FAULT: replacement evicts the block reported final at {report}");
+        }
+        let previous = self.clone();
         self.seg.truncate(keep);
         self.seg.extend_from_slice(blocks);
         self.ver += 1;
-        self.apply_finality(fin)?;
+        if let Err(err) = self.apply_finality(fin) {
+            *self = previous;
+            return Err(err);
+        }
         self.wf();
         Ok(())
     }
 
+    /// WP-8's non-hole case: the incoming range owns `e`, omits it, and thereby removes a block
+    /// stored there before the transition. Evaluate only exact, non-regressive reports; a report
+    /// above the resulting head is clamped and does not assert its own height.
+    fn replacement_evicts_reported_block(&self, from: BlockNumber, blocks: &[Block], report: &BlockRef) -> bool {
+        let post_head = blocks.last().expect("REPLACE blocks are non-empty").number;
+        if report.number > post_head || report.number < self.first().expect("REPLACE window is non-empty") {
+            return false;
+        }
+        if let Some(fin) = &self.fin
+            && report.number <= fin.number
+        {
+            return false;
+        }
+        from <= report.number
+            && report.number <= post_head
+            && blocks.iter().all(|b| b.number != report.number)
+            && self.block_at(report.number).is_some()
+    }
+
     /// WP §2.4 — advance `fin`, monotone and clamped to the stored chain.
     pub fn finalize(&mut self, r: &BlockRef) -> Finalize {
         let outcome = self.finalize_inner(r);
@@ -260,10 +306,10 @@ impl Model {
             return Finalize::Ignored;
         }
         let at_e = if e == r.number {
-            // The report names an exact height: a hole there contradicts the stored chain the
-            // same way a hash mismatch does (WP §2.4), and a different stored hash is WP-8.
+            // A slot-numbered chain may genuinely skip this height. There is no block reference to
+            // store as `fin`, so the report is ignored and may re-arrive later (WP-8).
             let Some(stored) = self.block_at(e) else {
-                return Finalize::IntegrityFault;
+                return Finalize::Ignored;
             };
             if stored.hash != r.hash {
                 return Finalize::IntegrityFault;
@@ -610,8 +656,7 @@ mod tests {
         assert_eq!(m.finalize(&BlockRef::new(101, block_hash(101, 0))), Finalize::Ignored);
         assert_eq!(m.fin, Some(blocks[4].as_ref()));
 
-        // A report naming a block we do not have is an integrity fault (WP-8) — this is the
-        // check GAP-4 says the implementation skips below the head.
+        // A report naming an existing block with another hash is an integrity fault (WP-8).
         let (mut m, _) = model_at(100, 5);
         assert_eq!(
             m.finalize(&BlockRef::new(102, block_hash(102, 7))),
@@ -621,16 +666,25 @@ mod tests {
     }
 
     #[test]
-    fn finalize_on_a_hole_is_an_integrity_fault() {
+    fn finalize_on_a_hole_is_ignored() {
         let anchor = block_hash(99, 0);
         let mut m = Model::new(Anchor::new(99, Some(anchor.clone())));
         m.extend(&chain_of(&[100, 103, 104], 0, (99, &anchor)), None).unwrap();
-        // WP §2.4: a report naming a height that is a hole in the stored chain contradicts
-        // the chain exactly like a hash mismatch.
-        assert_eq!(
-            m.finalize(&BlockRef::new(101, block_hash(101, 0))),
-            Finalize::IntegrityFault
-        );
+
+        assert_eq!(m.finalize(&BlockRef::new(101, block_hash(101, 0))), Finalize::Ignored);
+        assert_eq!(m.fin, None);
+    }
+
+    #[test]
+    fn replacement_cannot_evict_the_block_it_reports_finalized() {
+        let (mut m, blocks) = model_at(100, 3);
+        let replacement = chain_of(&[102], 1, (100, &blocks[0].hash));
+        let before = m.seg.clone();
+
+        let err = m.replace(101, &replacement, Some(&blocks[1].as_ref())).unwrap_err();
+
+        assert!(err.to_string().contains("INTEGRITY_FAULT"), "unexpected error: {err:#}");
+        assert_eq!(m.seg, before, "a refused composed transition must be atomic");
         assert_eq!(m.fin, None);
     }
 
@@ -639,10 +693,18 @@ mod tests {
         let (mut m, blocks) = model_at(100, 5);
         assert_eq!(m.finalize(&BlockRef::new(102, block_hash(102, 0))), Finalize::Applied);
 
-        // Below the finalized head — INV-13.
+        // Below the finalized head, on another branch — INV-13.
         let bad = run(102, 2, 1, &blocks[0].hash);
         assert!(m.replace(102, &bad, None).is_err());
 
+        // Reaching that low is allowed when the finalized part is replayed unchanged: a
+        // batch-granular implementation has no `fin + 1` position when finality sits mid-batch.
+        let replay = [blocks[2].clone()]
+            .into_iter()
+            .chain(run(103, 2, 7, &blocks[2].hash))
+            .collect::<Vec<_>>();
+        assert!(m.replace(102, &replay, None).is_ok());
+
         // Above it — accepted, and the head moves back by one.
         let good = run(103, 2, 1, &blocks[2].hash);
         m.replace(103, &good, None).unwrap();
@@ -672,6 +734,11 @@ mod tests {
         assert_eq!(m.anchor, Anchor::new(104, Some(blocks[4].hash.clone())), "INV-18");
         assert_eq!(m.fin, None, "RS-2: finality below the window is dropped");
         assert_eq!(m.head().unwrap().number, 109);
+        assert_eq!(
+            m.finalize(&BlockRef::new(104, blocks[4].hash.clone())),
+            Finalize::Ignored,
+            "WP §2.4: finality below the window stays ignored"
+        );
 
         // Trimming at the floor is a no-op.
         m.retain(105, None);