From e7285c45e87ff6e5459b1695981d233fa87bc05c Mon Sep 17 00:00:00 2001 From: Evgeny Formanenko Date: Wed, 29 Jul 2026 19:50:24 +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 — insert_fork replaces whole chunks, so a mid-chunk position wedges the dataset on a 60s restart loop — and resuming below fin instead silently rewrites the finalized prefix (GAP-22). Resolution now stops at the stored chunk boundary and the write path verifies the finalized region: a replacement reaching that low must span fin and reproduce every block up to it, hash for hash. Comparing only fin's own hash would admit a source rewriting the interior, since hashes come from the source and nothing ties a reproduced boundary to what leads to it. Payload equality is out of reach by construction (INV-13 Scope). The guard only recovers if the replay reaches fin, which was left as an assumption about the flush triggers. Production disproved it: ethereum-sepolia froze for 8h21m on 2026-07-29 when the 200k-row bound cut the replayed chunk one block below fin. Fork resolution now carries fin to the ingest, which withholds a chunk until it covers that block. A retention change always restarts the ingest epoch. Keeping it alive across a trim let a rollback resolved against the pre-trim window commit afterwards, resurrecting blocks below the new floor and pulling the head under it. Unrelated but folded in: the query slot is released before its caller is woken. The pool thread sent the result and dropped the slot afterwards, so a caller woken in between could be refused admission for a query that had already finished — and the panic-path test flaked on it, 3 failures in 200 local runs. Co-Authored-By: Claude Opus 4.8 Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 1 + crates/data-source/src/standard.rs | 20 +- crates/hotblocks-harness/README.md | 24 +- crates/hotblocks-harness/src/harness.rs | 4 + crates/hotblocks-harness/src/model.rs | 32 +- crates/hotblocks-harness/src/sim.rs | 127 ++++- crates/hotblocks-harness/src/sut.rs | 3 + crates/hotblocks/Cargo.toml | 1 + crates/hotblocks/spec/03-write-path.md | 35 +- crates/hotblocks/spec/06-invariants.md | 15 +- crates/hotblocks/spec/08-failure-model.md | 1 + crates/hotblocks/spec/12-conformance-tdd.md | 42 +- .../dataset_controller/dataset_controller.rs | 10 +- .../src/dataset_controller/ingest_generic.rs | 63 ++- .../dataset_controller/write_controller.rs | 462 ++++++++++++++++-- crates/hotblocks/src/metrics.rs | 11 + crates/hotblocks/src/query/executor.rs | 3 + crates/hotblocks/tests/ct4_finality.rs | 310 ++++++++++++ crates/hotblocks/tests/ct9_source_faults.rs | 1 + crates/storage/src/db/db.rs | 13 + crates/storage/src/db/write/dataset_update.rs | 4 + crates/storage/src/db/write/tx.rs | 107 ++++ 22 files changed, 1163 insertions(+), 126 deletions(-) create mode 100644 crates/hotblocks/tests/ct4_finality.rs diff --git a/Cargo.lock b/Cargo.lock index fc2c0b54..33b1b514 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4894,6 +4894,7 @@ name = "sqd-hotblocks" version = "0.1.0" dependencies = [ "anyhow", + "arrow", "async-stream", "axum", "bytes", diff --git a/crates/data-source/src/standard.rs b/crates/data-source/src/standard.rs index 682c2d77..d1361323 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}; @@ -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 diff --git a/crates/hotblocks-harness/README.md b/crates/hotblocks-harness/README.md index aa26404c..8e4b03c4 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 | @@ -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 diff --git a/crates/hotblocks-harness/src/harness.rs b/crates/hotblocks-harness/src/harness.rs index d0c54b51..bb5e7242 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, @@ -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)) diff --git a/crates/hotblocks-harness/src/model.rs b/crates/hotblocks-harness/src/model.rs index 9c5c36a8..c0003fc8 100644 --- a/crates/hotblocks-harness/src/model.rs +++ b/crates/hotblocks-harness/src/model.rs @@ -198,11 +198,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 ); } @@ -639,10 +653,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::>(); + 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(); diff --git a/crates/hotblocks-harness/src/sim.rs b/crates/hotblocks-harness/src/sim.rs index c37f8fb9..4cf91647 100644 --- a/crates/hotblocks-harness/src/sim.rs +++ b/crates/hotblocks-harness/src/sim.rs @@ -85,13 +85,21 @@ pub struct SimFaults { /// no-data. RP-5b confines the signal to `from == tip + 1`, the one position where the /// assertion is evaluable; a source doing it higher reports a divergence it cannot have /// observed, and a source that is merely behind starts looking like a forked one. - pub fork_signal_above_tip: bool + pub fork_signal_above_tip: bool, + /// Cut every response to at most this many blocks (0 reads as 1), so the service's chunk + /// boundaries stop tracking the source's history. + pub max_blocks_per_response: Option, + /// Report finality no higher than this — a replica lagging behind the one the service already + /// accepted finality from. + pub finality_report_cap: Option } /// Counters a test can assert on (how the SUT actually drove the source). #[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, @@ -166,6 +174,30 @@ 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(()) + } + + /// Fault injection for CT-4/INV-13: rewrite the hash of one block strictly *below* the source's + /// own finalized head, leaving that block and everything above it intact. + /// + /// The chain stays internally linked, so only a comparison against stored history tells this + /// apart from an honest replay. Like [`Self::equivocate_finalized_prefix`] it has no model-side + /// counterpart. + pub fn rewrite_hash_below_finality(&self, dataset: &str, at: BlockNumber) -> Result { + let r = self.try_with(dataset, |d| d.rewrite_hash_below_finality(at))?; + self.bump(); + Ok(r) + } + /// 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))?; @@ -308,22 +340,75 @@ 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 rewrite_hash_below_finality(&mut self, at: BlockNumber) -> Result { + let finalized = self + .fin + .as_ref() + .context("rewriting below finality requires an existing finalized head")? + .number; + ensure!( + at < finalized, + "block {at} is not strictly below the source's finalized head {finalized}" + ); + + let i = self + .chain + .binary_search_by_key(&at, |b| b.number) + .ok() + .with_context(|| format!("the source does not have block {at}"))?; + + let hash = block_hash(at, self.next_fork_id); + self.next_fork_id += 1; + self.chain[i].hash = hash.clone(); + if let Some(child) = self.chain.get_mut(i + 1) { + child.parent_hash = hash.clone(); + } + Ok(BlockRef::new(at, hash)) + } + + 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 { @@ -349,6 +434,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 { @@ -378,23 +464,38 @@ impl DatasetSim { if req.from_block < self.start { self.stats.below_history += 1; - return Reply::NoData(self.fin.clone()); + return Reply::NoData(self.reported_fin()); } let Some(tip) = self.tip_number() else { self.stats.no_data += 1; - return Reply::NoData(self.fin.clone()); + return Reply::NoData(self.reported_fin()); }; if req.from_block > tip { self.stats.no_data += 1; - return Reply::NoData(self.fin.clone()); + return Reply::NoData(self.reported_fin()); } let lo = self.chain.partition_point(|b| b.number < req.from_block); - let blocks = self.chain[lo..].to_vec(); + let mut blocks = self.chain[lo..].to_vec(); + if let Some(cap) = self.faults.max_blocks_per_response { + blocks.truncate((cap as usize).max(1)); + } self.stats.blocks_served += blocks.len() as u64; Reply::Blocks { blocks, - fin: self.fin.clone() + fin: self.reported_fin() + } + } + + /// What this replica admits to having finalized (see [`SimFaults::finality_report_cap`]). + fn reported_fin(&self) -> Option { + let fin = self.fin.clone()?; + match self.faults.finality_report_cap { + Some(cap) if cap < fin.number => self.chain_hash_at(cap).map(|hash| BlockRef { + number: cap, + hash: hash.to_string() + }), + _ => Some(fin) } } } @@ -465,7 +566,7 @@ async fn head(State(shared): State>, Path(ds): Path) -> Resp } async fn finalized_head(State(shared): State>, Path(ds): Path) -> Response { - watermark(&shared, &ds, |d| d.fin.clone()) + watermark(&shared, &ds, |d| d.reported_fin()) } fn watermark(shared: &Shared, ds: &str, f: impl FnOnce(&DatasetSim) -> Option) -> Response { diff --git a/crates/hotblocks-harness/src/sut.rs b/crates/hotblocks-harness/src/sut.rs index 3eee2188..d0b9f68f 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 } @@ -262,6 +264,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/Cargo.toml b/crates/hotblocks/Cargo.toml index 03ad3da4..764496ca 100644 --- a/crates/hotblocks/Cargo.toml +++ b/crates/hotblocks/Cargo.toml @@ -37,6 +37,7 @@ url = { workspace = true, features = ["serde"] } [dev-dependencies] anyhow = { workspace = true } +arrow = { workspace = true } reqwest = { workspace = true } sqd-hotblocks-harness = { path = "../hotblocks-harness" } tempfile = { workspace = true } diff --git a/crates/hotblocks/spec/03-write-path.md b/crates/hotblocks/spec/03-write-path.md index 89580957..cda2a422 100644 --- a/crates/hotblocks/spec/03-write-path.md +++ b/crates/hotblocks/spec/03-write-path.md @@ -91,7 +91,13 @@ position, not a promised block number, DEF-1.) ### 2.3 REPLACE(from, B, f?) — fork application - Pre: `seg ≠ ∅`; `B ≠ ∅` and valid per WP-2; `B[0].number ≥ from`; and - - **fork floor (finality):** `fin = ⊥ ∨ from > fin.number` — MUST (INV-13/14); + - **fork floor (finality):** `fin = ⊥ ∨ from > fin.number`, or `B` reproduces the stored + chain over `[from, fin.number]` block for block — MUST (INV-13/14). The second disjunct + exists because `from` snaps to a stored batch boundary, which sits below `fin` whenever + finality falls inside a batch; INV-13 forbids *altering* a finalized block, not + re-writing it identically. Reproduction MUST be checked over the whole overlap, not at + `fin.number` alone: hashes come from the source, so a reproduced boundary says nothing + about the interior; - **fork floor (window):** `from ≥ first(D)` — MUST (INV-14). A required rollback below `first(D)` is not representable as REPLACE; it MUST be handled as RESET (§2.6) and surfaced as an event ([OB-9](11-observability.md)); @@ -111,21 +117,24 @@ signal contradicts finality and MUST be treated as a source integrity fault the stored chain (or the anchor). If no hint matches stored state, the fallback MUST respect the finality floor: -- `fin ≠ ⊥` → resume from `⟨fin.number + 1, fin.hash⟩`: the finalized block is common to - both chains unless the source contradicts finality, so only the volatile suffix is - replaced (a full-window replacement would violate the fork floor anyway, INV-14). If - the source then repeatedly fails to link at this position, that *is* a finality - contradiction: after `P-SOURCE-STRIKES` consecutive rejections it MUST be classified as - FM-SRC-5 (fault + alarm, keep serving) — never RESET; +- `fin ≠ ⊥` → resume from `⟨fin.number + 1, fin.hash⟩`, or from any lower position + `≥ first(D)` whose replacement reproduces the finalized prefix (INV-14). The finalized + block is common to both chains unless the source contradicts finality, so only the + volatile suffix carries new data either way. An implementation whose REPLACE granularity + is a storage batch MUST take the lower position when `fin` falls inside a batch: `fin + 1` + is then unrepresentable, and retrying it is a wedge rather than a rejection. If the source + repeatedly fails to link at the chosen position, that *is* a finality contradiction: after + `P-SOURCE-STRIKES` consecutive rejections it MUST be classified as FM-SRC-5 (fault + + alarm, keep serving) — never RESET; - `fin = ⊥` → resume from `⟨first(D), anchor.hash⟩` (full-window replacement); WP-6b governs escalation if the divergence turns out to lie below the window. The subsequent commit is `REPLACE(m + 1, …)` (resp. `REPLACE(fin.number + 1, …)`, -`REPLACE(first(D), …)`). The resume point MAY be conservatively deeper than the optimal -`m + 1` by at most one storage batch (an implementation that matches hints only at batch -boundaries): the replacement then re-commits blocks identical to those it replaces, which -is correctness-neutral. It MUST NOT be shallower than `m + 1` and MUST NOT cross the -floors of §2.3. +`REPLACE(first(D), …)`). Both the match and the finality fallback MAY be conservatively +deeper by at most one storage batch (an implementation that commits whole batches): the +replacement then re-commits blocks identical to those it replaces, which is +correctness-neutral, and below `fin` that identity is what INV-14 requires. It MUST NOT be +shallower than `m + 1` and MUST NOT cross the floors of §2.3. **WP-6b (Divergence below the window → RESET).** WP-6's fallback cannot represent a fork deeper than the window; the service MUST detect that case and escalate to RESET (alarmed, @@ -186,7 +195,7 @@ history cannot be re-acquired through RETAIN. symmetric with WP-9. *Finality note:* like every RESET (§2.6) and like upward trims (RS-2), this discards the finalized prefix and clears `fin` — retention dominates finality. It is **not** a rollback below `fin`: the fork floor (INV-13/14) is - untouched — *sources* can never replace anything at or below `fin` — and this path is + untouched — *sources* can never alter anything at or below `fin` — and this path is reachable only through an explicit retention instruction (operator-class actor, FM-OP-4), never through source input. - Case `first(D) < from ≤ next(D)`: diff --git a/crates/hotblocks/spec/06-invariants.md b/crates/hotblocks/spec/06-invariants.md index 32ff3ed7..e428d73a 100644 --- a/crates/hotblocks/spec/06-invariants.md +++ b/crates/hotblocks/spec/06-invariants.md @@ -83,12 +83,21 @@ changes. `fin` may become `⊥` only via RETAIN (window passing above it), RESET **INV-13 — Finalized immutability.** [transition] No `REPLACE` removes or alters blocks at heights `≤ fin.number`. Finalized blocks leave the store only via RETAIN / RESET / DROP. +*Scope:* "alters" is decidable only down to block identity — `⟨number, hash⟩` compared +against stored history. No block hash is re-derived from its payload anywhere in the +system, so a source reproducing a height's hash while serving different content for it is +outside what any write-path check can see; sources are trusted for content at every height, +finalized or not, and this holds equally for the first write of a block. *Check:* CT-1/CT-4 fork storms around the finality boundary. **INV-14 — Fork floor.** [transition] -For every `REPLACE(from, B)`: `from > fin.number` (when `fin ≠ ⊥`) **and** -`from ≥ first(D)`. A deeper divergence is representable only as RESET (explicit, alarmed). -*Why:* silent rollback below the window or below finality corrupts continuation clients. +For every `REPLACE(from, B)`: `from ≥ first(D)` **and** either `from > fin.number` (when +`fin ≠ ⊥`) or `B` reproduces `seg` over `[from, fin.number]` block for block. A deeper +divergence is representable only as RESET (explicit, alarmed). +*Why:* silent rollback below the window or below finality corrupts continuation clients. An +identical replay of the finalized prefix changes nothing, and permitting it is what lets an +implementation whose REPLACE granularity is a storage batch resolve a fork at all: when `fin` +falls inside a batch, `from = fin + 1` is not a representable position. *Check:* CT-4 deep-fork corpus (GAP-3). **INV-15 — Retention trims prefix only.** [transition] diff --git a/crates/hotblocks/spec/08-failure-model.md b/crates/hotblocks/spec/08-failure-model.md index 00d63c91..89e54ec4 100644 --- a/crates/hotblocks/spec/08-failure-model.md +++ b/crates/hotblocks/spec/08-failure-model.md @@ -76,6 +76,7 @@ each. "Required response" uses four verbs: | FM-OP-3 | Two service instances over one store | fail-safe: detect divergence, stop the losing writer per dataset, alarm (WP-15); MUST NOT interleave-corrupt | | FM-OP-4 | Retention mistakes (raise far above head, contradictory instructions) | defined semantics (WP-9/RETAIN cases): destructive outcomes are the documented ones only; observable; idempotent | | FM-OP-5 | Restart with changed parameters (window size, budgets) | mask: state re-converges to policy (trim or backfill-forward per WP-10/WP-5); no invariant violations during convergence | +| FM-OP-6 | Data availability changed for a live dataset — sources swapped or reconfigured so the same blocks arrive with a different section set (the data-availability mask) | **out of scope**: a deliberate operator change, sequenced by the operator. No behaviour across it is promised and none should be relied on. Only the standing integrity line holds: two section sets never share a chunk (INV-7), so where the ingest cannot cut cleanly it stops loudly rather than serving blocks stripped of sections | ## 6. Fault → property cross-reference diff --git a/crates/hotblocks/spec/12-conformance-tdd.md b/crates/hotblocks/spec/12-conformance-tdd.md index 2c738a46..1aad9d33 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-20** and are +Statuses and the gap register reflect the state of knowledge as of **2026-07-29** 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 @@ -95,7 +97,7 @@ model Dataset: replace(from_, B, f=⊥): # WP §2.3 require seg and B and valid_run(B) and B[0].number >= from_ - require fin == ⊥ or from_ > fin.number # INV-13/14 + require fin == ⊥ or from_ > fin.number or reproduces(B, seg, from_, fin.number) # INV-13/14 require from_ >= first() # INV-14 require B[0].parent_hash == hash_at(from_ - 1) # DEF-16 (⊥ accepted at the window edge) seg = [b in seg | b.number < from_] + B; ver += 1; if f: finalize_inline(f) @@ -139,8 +141,11 @@ model Dataset: return RESET((anchor.number, hint hash at that position)) # WP-6b: below-window divergence m = max({x in hints | stored_or_anchor(x)}, default=⊥) if m != ⊥: return (m.number + 1, m.hash) - if fin != ⊥: return (fin.number + 1, fin.hash) # volatile suffix only; repeated - # rejection here = FM-SRC-5, never RESET + if fin != ⊥: return (fin.number + 1, fin.hash) # or any lower position reproducing the + # finalized prefix (INV-14) — a batch-granular + # implementation must, when fin sits inside a + # batch. Repeated rejection here = FM-SRC-5, + # never RESET return (first(), anchor.hash) # full-window replacement *probe*; # repeated WP-2 rejection at first(D) # escalates to RESET per WP-6b @@ -229,9 +234,9 @@ INV-21/22/23 (checks in parentheses): ## 5. Traceability matrix (status @ 2026-07-21) -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 +247,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; three `ct4_finality` scenarios + write-controller regressions pin finalized-prefix immutability (whole overlapping range, not just the boundary at `fin`), 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 +321,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-27 | FINALIZE never checks `e ≥ first(D)`: with `fin = ⊥` (e.g. after a trim passed above it) a lagging source's report below the window commits `fin < first(D)` | WP §2.4, INV-5 | P2 | CT-4: finality below the window on a trimmed dataset; assert the report is ignored | @@ -333,6 +337,7 @@ rare, P3 = polish. **First test** names the cheapest failing-test-first entry po | GAP-39 | A hash-lookup miss and an unknown dataset are both 404 with only a free-text body between them, and IB-7 forbids keying on text. This is worse than the GAP-36 family it belongs to: RP-19 makes "this hash is not indexed" a *deliberately uninformative* answer, so a client that cannot separate it from "this dataset does not exist" cannot tell a misconfiguration from a legitimate miss at all | INV-26, IB-7, RP-19 | P3 | CT-5: unknown dataset vs unknown hash; assert a structured discriminant | | GAP-40 | Hash-index CFs export only engine-wide estimated keys and live SST bytes. OB-12 still lacks per-dataset enabled state / entry count / bytes and hit-vs-miss lookup counts with latency. Because a miss is uninformative by design, an index empty for a structural reason — enabled after the window had filled, wrong kind — remains indistinguishable from one receiving only unknown hashes | OB-12, OB-6 | P3 | CT-1: scrape per-dataset state and exercise hit/miss counters once exported | | GAP-41 | Fork consensus is credulous: `StandardDataSource::poll_next_event` counts fork-signalling endpoints without asking whether a signalling endpoint has any standing at the contested position, and `extract_fork` then adopts the *longest* hint chain among them. RP-5b confines a legitimate signal to `from == tip + 1`, so an in-spec source can only signal where it is; but FM-1 requires surviving one that is not, and a source signalling above its tip is what a source merely *behind* would look like if it answered wrongly. **One** such endpoint out of three suffices, and not by majority: `forks > endpoints.len() / 2` is false at 1-of-3, but the 2 s `fork_consensus_timeout` fires on any poll where every endpoint returned `Pending`, and `accept_new_block` clears that timer only on a commit — so on a chain with block time ≥ 2 s *every inter-block gap is a firing window*. Measured 2026-07-20, 5/5 runs, by instrumenting `poll_next_event`: `forks=1 endpoints=3 active=3 majority=false all_active=false timeout=true`; the adopted hints end at the liar's stale tip, `compute_rollback` rejects them as below `fin`, and ingestion parks per GAP-5 with two healthy sources still offering the chain. The endpoints' own position is known (`Endpoint::last_committed_block`) and unused | FM-1, WP-6, FM-SRC-4/5 | **P1** (was P2 on the premise that it took a majority — measurement overturned it: a lone bad source is enough, on the ordinary inter-block gap rather than a rare coincidence) | `ct4_a_single_source_signalling_a_fork_above_its_tip_does_not_park_ingestion`; the majority path is pinned separately by `ct4_a_fork_signal_majority_above_the_tip_does_not_park_ingestion` so a fix closing only that clause cannot read as green. Both `#[ignore]`d | +| GAP-43 | The withheld-flush alarm (GAP-22) only fires where a flush is attempted — the 200k-row bound. A source that simply stops delivering below the flush floor never gets there: `MaybeOnHead` is suppressed while the position sits below the highest finality any endpoint reported, so no flush is tried, nothing is counted and no epoch restarts. The stall is visible as a growing `hotblocks_last_block_timestamp_ms` but unattributed. Deliberately deferred: the stall pre-dates the flush floor (with no flush trigger there is no chunk either way), so this is an observability hole, not a regression | LIV-2, OB-9, OB-11 | P2 | CT-4: replay below `fin` from a source that stops short; assert an alarmed state within `P-ALARM` independent of the flush triggers | ### 6.1 Closed @@ -341,9 +346,11 @@ 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: resolution resumes at a stored chunk boundary — never a mid-chunk `fin + 1`, which `insert_fork` cannot satisfy and which wedged the dataset on a 60 s loop — and the write path admits a replacement reaching below `fin` only when it reproduces the whole overlap `[chunk.first_block, fin]` block for block (`Tx::validate_finalized_prefix`), else refuses atomically. Comparing `fin`'s own hash alone was the first attempt and let a rewritten interior through, since hashes are source-supplied; payload-level equality stays out of reach by construction (INV-13 *Scope*). The guard also needs the replay to actually reach `fin`, which was assumed of the flush triggers until ethereum-sepolia froze for 8h21m on 2026-07-29 — the 200k-row bound cut the chunk one block short and 374 identical retries were refused — so `Rollback::reach_at_least` now carries `fin` to the ingest, which withholds a chunk until it covers that block. Pinned by five `ct4_finality` scenarios plus write-controller unit tests; the cut-replay and below-`fin`-rewrite cases were verified red. Remaining: below-window anchor and source alarms (GAP-23/5/30), and a withhold that is counted but neither capped nor visible in every shape (GAP-43). Review fallout, same PR: withholding no longer disables the memory spill (the row bound stays crossed once the rows are held, so the byte check must still run) and is warned about once per episode, not once per block; and the no-matching-hint fallback resumes at the *physical* window start, since retention's floor can sit inside a surviving chunk and `insert_fork` refuses that as overlapping — the same wedge in a different disguise. Third round: a refused chunk left its already-materialized tables behind — only a committed `write_chunk` clears the dirty marker and the orphan sweep is startup-only, so a permanently equivocating source leaked a chunk per retry; they are now abandoned to the ordinary purge. The finality rejections also carry `UnapplicableFork` in the error chain, or `report_dataset_epoch_failure` files exactly these divergences under `reason="other"` | | GAP-26 | A block timestamp outside the datetime-conversion range killed the ingest flush — the conversion existed only for a log line | Log formatting is best-effort and the raw value is stored unchanged; evm/solana seconds→millis saturate so an absurd source value neither panics nor wraps (PR #100, 2026-07-20). No regression test — CT-9: serve a block with `time = i64::MAX`; assert the batch commits and serving continues | | 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) | +| GAP-42 | A retention change could invalidate an in-flight fork rollback: a running ingest was kept alive across a `FromBlock` trim whenever the head survived it, so a replay resolved against the pre-trim window committed afterwards, resurrecting blocks below the new floor and leaving the head under `first(D)` — and, where the trim moved the window above `fin`, rewriting the former finalized prefix with the guard disabled | Any retention change now restarts the ingest epoch; the aborted task takes the stale rollback and flush floor with it. Reachable only through the runtime `Api` path. No regression test: driving SET-RETENTION into the replay window needs a harness retention client *and* deterministic control of the replay's duration, so the fix is carried by the control-flow simplification instead (2026-07-29) | ## 7. Build order (recommended) @@ -353,13 +360,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); `ct2_shutdown` exercises the bounded SIGTERM path | the remaining 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`, finalized-prefix and below-finality equivocation faults, `Model::resolve_fork` and follower CONFLICT recovery → CT-4 | below-window RESET, malformed finality, fork-storm 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 e5d14774..2157f9bf 100644 --- a/crates/hotblocks/src/dataset_controller/dataset_controller.rs +++ b/crates/hotblocks/src/dataset_controller/dataset_controller.rs @@ -410,13 +410,11 @@ impl Ctl { match retention { RetentionStrategy::FromBlock { number, parent_hash } => { let (number, parent_hash) = self.clamp_floor(&write, number, parent_hash); - let will_erase_head = write.head().map_or(false, |h| h.number < number) || // FromBlock is greater than current head, so everything is cleared - write.start_block() > number; // FromBlock is less than current front, dropping everything by design blocking_write!(write, write.retain(number, parent_hash))?; - match state { - State::Ingest { .. } if !will_erase_head => {} // Keep ingesting, head is valid - _ => *state = State::Init { head: self.max_blocks } // New ingest needed - } + // Restart unconditionally: a running ingest may hold a rollback resolved against + // the pre-trim window, and committing that replay resurrects blocks below the + // new floor. + *state = State::Init { head: self.max_blocks } } RetentionStrategy::Head(n) => match state { State::HeadProbe { head, .. } => *head = n, diff --git a/crates/hotblocks/src/dataset_controller/ingest_generic.rs b/crates/hotblocks/src/dataset_controller/ingest_generic.rs index c98b5e59..9f1ea831 100644 --- a/crates/hotblocks/src/dataset_controller/ingest_generic.rs +++ b/crates/hotblocks/src/dataset_controller/ingest_generic.rs @@ -10,11 +10,11 @@ use sqd_data_core::{BlockChunkBuilder, ChunkProcessor, PreparedChunk}; use sqd_data_source::{DataEvent, DataSource}; use sqd_primitives::{Block, BlockNumber, BlockRef, DataMask, DisplayBlockRefOption}; use sqd_storage::db::DatasetId; -use tracing::{debug, field::valuable, info}; +use tracing::{debug, field::valuable, info, warn}; use crate::{ dataset_controller::write_controller::Rollback, - metrics::{WriteStage, report_write_duration} + metrics::{WriteStage, report_withheld_flush, report_write_duration} }; pub enum IngestMessage { @@ -121,7 +121,11 @@ pub struct IngestGeneric { last_block: BlockNumber, last_block_hash: String, last_block_time: Option, - data_mask: DataMask + data_mask: DataMask, + /// No chunk may be emitted before it covers this block ([`Rollback::reach_at_least`]). + flush_floor: Option, + /// Whether the current withholding episode has already been warned about and counted. + withholding: bool } impl IngestGeneric @@ -150,7 +154,9 @@ where last_block: 0, last_block_hash: String::new(), last_block_time: None, - data_mask: DataMask::default() + data_mask: DataMask::default(), + flush_floor: None, + withholding: false } } @@ -167,6 +173,12 @@ where let data_mask = block.data_availability_mask(); if self.data_mask != data_mask { self.flush().await?; + // Masks can't share a chunk, and the flush floor can't be waived. + ensure!( + self.buffered_blocks == 0, + "data availability changed at block {} inside the replayed finalized range", + block.number() + ); self.data_mask = data_mask } self.push_block(block, is_final)?; @@ -180,7 +192,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(); @@ -195,16 +211,19 @@ 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, + reach_at_least =? rollback.reach_at_least, "resetting ingest position" ); self.buffered_blocks = 0; self.finalized_head = None; - self.first_block = rollback.first_block; + self.first_block = rollback.resume_from; + self.flush_floor = rollback.reach_at_least; + self.withholding = false; 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(()) } @@ -236,8 +255,10 @@ where async fn maybe_flush(&mut self) -> anyhow::Result<()> { if self.builder_ref().num_rows() > 200_000 { - return self.flush().await; + self.flush().await?; } + // Not `else`: a withheld flush leaves every row in place, so `num_rows` stays above the + // bound and the replay would grow in memory until it reached the floor. if self.builder_ref().in_memory_buffered_bytes() > self.builder_ref().spill_bound_bytes { return self.with_blocking_builder(|b| b.flush_to_processor()).await; } @@ -249,6 +270,28 @@ where return Ok(()); } + // Emitting here would swap the finalized block out of storage; keep buffering. Reported + // once per episode — the row bound is re-crossed by every subsequent block. + // + // FUTURE: this warns only where a flush is attempted. A source that just stops below the + // floor never gets there, since `MaybeOnHead` is suppressed below the highest reported + // finality, so the stall shows up only as a stale last-block timestamp. + if let Some(floor) = self.flush_floor.filter(|floor| self.last_block < *floor) { + if !self.withholding { + self.withholding = true; + warn!( + buffered_blocks = self.buffered_blocks, + last_block = self.last_block, + reach_at_least = floor, + "withholding a chunk until the replay reaches the finalized head" + ); + report_withheld_flush(self.dataset_id); + } + return Ok(()); + } + self.flush_floor = None; + self.withholding = false; + let parent_block_hash = self.parent_block_hash.clone(); let first_block = self.first_block; let last_block = self.last_block; diff --git a/crates/hotblocks/src/dataset_controller/write_controller.rs b/crates/hotblocks/src/dataset_controller/write_controller.rs index 44ce65ed..1e4dbb4c 100644 --- a/crates/hotblocks/src/dataset_controller/write_controller.rs +++ b/crates/hotblocks/src/dataset_controller/write_controller.rs @@ -13,10 +13,21 @@ use crate::{ types::{DBRef, DatasetKind} }; +/// Source position selected after resolving a fork against stored history. +/// +/// `resume_from` lands on a stored chunk boundary, so it may sit at or below `fin` when the +/// common ancestor is inside a finality-straddling chunk; the finalized prefix is guarded on the +/// write path instead ([`WriteController::new_chunk`]). #[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, + /// Lowest block the first replayed chunk must reach, set when `resume_from` sits at or below + /// `fin`. The swap deletes the chunk holding the finalized block; a chunk stopping short never + /// puts it back, and the retry repeats the identical cut forever (GAP-22). + pub reach_at_least: Option } /// Single writer for a dataset. Owns head/finalized-head as its working copy of @@ -137,7 +148,9 @@ impl WriteController { .get_label(self.dataset_id)? .ok_or_else(|| anyhow!("dataset {} no longer exists", self.dataset_id))?; - if let Some(finalized_head) = label.finalized_head() { + let finalized_head = label.finalized_head().cloned(); + + if let Some(finalized_head) = finalized_head.as_ref() { let pos = match prev.iter().position(|b| b.number >= finalized_head.number) { Some(pos) => pos, None => bail!(UnapplicableFork { @@ -155,41 +168,48 @@ impl WriteController { prev = &prev[pos..] } - let existing_chunks = snapshot - .list_chunks(self.dataset_id, 0, Some(prev.last().unwrap().number)) - .into_reversed(); + let (resume_from, expected_parent_hash) = 'resume: { + let existing_chunks = snapshot + .list_chunks(self.dataset_id, 0, Some(prev.last().unwrap().number)) + .into_reversed(); - let mut prev_blocks = prev.iter().rev().peekable(); + let mut prev_blocks = prev.iter().rev().peekable(); - for chunk_result in existing_chunks { - let head = chunk_result?; + for chunk_result in existing_chunks { + let head = chunk_result?; - if prev_blocks.peek().map_or(false, |b| b.number < head.last_block()) { - continue; - } + if prev_blocks.peek().map_or(false, |b| b.number < head.last_block()) { + continue; + } - while prev_blocks.peek().map_or(false, |b| b.number > head.last_block()) { - prev_blocks.next(); - } + while prev_blocks.peek().map_or(false, |b| b.number > head.last_block()) { + prev_blocks.next(); + } - 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()) - }); + if let Some(&b) = prev_blocks.peek() { + if b.number == head.last_block() && b.hash == head.last_block_hash() { + break 'resume (b.number + 1, Some(b.hash.clone())); + } + } else { + break 'resume (head.last_block() + 1, Some(head.last_block_hash().to_string())); } - } else { - return Ok(Rollback { - first_block: head.last_block() + 1, - parent_block_hash: Some(head.last_block_hash().to_string()) - }); } - } + + // Retention trims whole chunks, so `self.first_block` can sit inside the surviving + // one — a position `insert_fork` refuses as overlapping. Fall back to the physical + // start of the window instead. + match snapshot.get_first_chunk(self.dataset_id)? { + Some(chunk) => (chunk.first_block(), Some(chunk.parent_block_hash().to_string())), + None => (self.first_block, self.parent_block_hash.clone()) + } + }; Ok(Rollback { - first_block: self.first_block, - parent_block_hash: self.parent_block_hash.clone() + resume_from, + expected_parent_hash, + reach_at_least: finalized_head + .filter(|fin| resume_from <= fin.number) + .map(|fin| fin.number) }) } @@ -406,23 +426,55 @@ impl WriteController { pub fn new_chunk(&mut self, finalized_head: Option<&BlockRef>, chunk: &StorageChunk) -> anyhow::Result<()> { // FIXME: accept self.first_block rollback limit let dataset_id = self.dataset_id; - let finalized_head = observe_storage_write(dataset_id, WriteStage::Commit, |metrics| { + let commit = observe_storage_write(dataset_id, WriteStage::Commit, |metrics| { self.db .update_dataset_with_hash_index_metrics(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 resuming at a chunk boundary can reach below `fin`. Admit it only if + // it reproduces that region exactly — spanning `fin`, matching every stored + // hash up to it. `fin`'s own hash alone proves nothing: hashes come from the + // source, so a reproduced boundary says nothing about what leads to it. + if let Some(current) = current_finalized_head.as_ref() + && chunk.first_block() <= current.number + { + ensure!( + chunk.last_block() >= current.number, + unapplicable_fork( + "replacement drops the finalized block", + format!( + "chunk {}-{} does not reach finalized block {}", + chunk.first_block(), + chunk.last_block(), + current.number + ) + ) + ); + if let Err(divergence) = tx.validate_finalized_prefix(chunk, current.number)? { + return Err(unapplicable_fork("replacement rewrites finalized history", divergence)); + } + } + + 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, + unapplicable_fork( + "finality hash changed at a fixed height", + format!("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) } @@ -432,7 +484,18 @@ impl WriteController { tx.insert_fork(chunk)?; Ok(new_finalized_head) }) - })?; + }); + + // The caller materialized these tables before we could judge the chunk, and only a + // committed `write_chunk` clears their dirty markers — the orphan sweep runs at startup + // only. A source refused on every retry would leak a chunk a minute. + let finalized_head = match commit { + Ok(head) => head, + Err(err) => { + self.abandon_tables(chunk); + return Err(err); + } + }; debug!(finalized_head = valuable(&finalized_head), "saved new chunk"); @@ -479,6 +542,13 @@ impl WriteController { Ok(()) } + fn abandon_tables(&self, chunk: &StorageChunk) { + let tables = chunk.tables().values().copied().collect::>(); + if let Err(err) = self.db.delete_tables(&tables) { + warn!(reason =? err, "failed to abandon the tables of a refused chunk"); + } + } + fn write_new_chunk(&mut self, mut new_chunk: NewChunk) -> anyhow::Result<()> { let desc = self.dataset_kind().dataset_description(); let started = StdInstant::now(); @@ -524,6 +594,12 @@ impl WriteController { } } +/// Keeps [`UnapplicableFork`] in the chain — `report_dataset_epoch_failure` buckets on the type, +/// and the message can never carry block numbers or hashes into a metric label. +fn unapplicable_fork(reason: &'static str, detail: String) -> anyhow::Error { + anyhow::Error::new(UnapplicableFork { reason }).context(detail) +} + fn get_chunk_head(chunk: &Chunk) -> BlockRef { BlockRef { number: chunk.last_block(), @@ -574,12 +650,19 @@ fn observe_storage_write( mod tests { use std::{collections::BTreeMap, sync::Arc}; + use arrow::{ + array::{RecordBatch, StringArray, UInt64Array}, + datatypes::{DataType, Field, Schema} + }; use sqd_primitives::BlockRef; use sqd_storage::db::{Chunk, DatabaseSettings, DatasetId}; use tokio::sync::watch; - use super::{WriteController, trim_floor}; - use crate::types::{DBRef, DatasetKind}; + use super::{WriteController, get_chunk_head, trim_floor}; + use crate::{ + errors::UnapplicableFork, + types::{DBRef, DatasetKind} + }; #[test] fn nothing_is_trimmed_while_the_window_fits() { @@ -746,4 +829,299 @@ mod tests { eprintln!("chunk commit + publish: {commit_us:.1} us/op"); assert_eq!(*f.head_rx.borrow(), Some(block(last_block, &parent))); } + + /// Blocks carry `{tag}-{number}` hashes: two chunks agree on a range exactly when built with + /// the same tag. + fn hashes(tag: &str, first_block: u64, last_block: u64) -> Vec { + (first_block..=last_block).map(|n| format!("{tag}-{n}")).collect() + } + + /// A chunk with a real `blocks` table — the finalized-prefix guard reads hashes out of + /// storage, so the table-less `chunk` above would exercise nothing. + fn chunk_with_hashes(db: &DBRef, first_block: u64, parent_hash: &str, hashes: &[String]) -> anyhow::Result { + let last_block = first_block + hashes.len() as u64 - 1; + + let schema = Arc::new(Schema::new(vec![ + Field::new("number", DataType::UInt64, false), + Field::new("hash", DataType::Utf8, false), + Field::new("parent_hash", DataType::Utf8, false), + ])); + + let parent_hashes = std::iter::once(parent_hash) + .chain(hashes.iter().map(String::as_str)) + .take(hashes.len()) + .collect::>(); + + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(UInt64Array::from((first_block..=last_block).collect::>())), + Arc::new(StringArray::from(hashes.to_vec())), + Arc::new(StringArray::from(parent_hashes)), + ] + )?; + + let mut builder = db.new_table_builder(schema); + builder.write_record_batch(&batch)?; + let table_id = builder.finish()?; + + Ok(Chunk::V1 { + first_block, + last_block, + last_block_hash: hashes.last().unwrap().clone(), + parent_block_hash: parent_hash.to_string(), + first_block_time: None, + last_block_time: None, + tables: BTreeMap::from([("blocks".to_string(), table_id)]) + }) + } + + fn linked_chunk( + db: &DBRef, + first_block: u64, + last_block: u64, + parent_hash: &str, + tag: &str + ) -> anyhow::Result { + chunk_with_hashes(db, first_block, parent_hash, &hashes(tag, first_block, last_block)) + } + + fn seed_chain(f: &mut Fixture) -> anyhow::Result<(Chunk, Chunk)> { + let first = linked_chunk(&f.db, 0, 5, "genesis", "old")?; + let second = linked_chunk(&f.db, 6, 9, "old-5", "old")?; + f.wc.new_chunk(None, &first)?; + f.wc.new_chunk(None, &second)?; + Ok((first, second)) + } + + fn stored_chunks(f: &Fixture) -> anyhow::Result> { + f.db.snapshot() + .list_chunks(f.dataset_id, 0, None) + .collect::>>() + } + + // INV-12/13: a replacement reaching the finalized height with a different hash there is a + // source equivocating below its own finality; refusing it must leave nothing behind. + #[test] + fn replacement_rewriting_the_finalized_block_is_rejected() -> anyhow::Result<()> { + let mut f = fixture(); + let (first, second) = seed_chain(&mut f)?; + let current_finalized = block(5, "old-5"); + f.wc.finalize(¤t_finalized)?; + + let replacement = linked_chunk(&f.db, 0, 5, "other-genesis", "new")?; + let result = f.wc.new_chunk(Some(&block(5, "new-5")), &replacement); + + let err = result.unwrap_err(); + assert!( + err.chain().any(|e| e.is::()), + "finality rejections must stay in the unapplicable_fork metric bucket: {err:#}" + ); + assert_eq!(*f.fin_rx.borrow(), Some(current_finalized.clone())); + assert_eq!(stored_chunks(&f)?, vec![first, second]); + assert_eq!( + f.db.snapshot() + .get_label(f.dataset_id)? + .and_then(|label| label.finalized_head().cloned()), + Some(current_finalized) + ); + Ok(()) + } + + // INV-12: finality is immutable at a fixed height, even when carried by an ordinary append. + #[test] + fn composed_finality_rejects_hash_change_at_fixed_height() -> anyhow::Result<()> { + let mut f = fixture(); + let (first, second) = seed_chain(&mut f)?; + let current_finalized = block(5, "old-5"); + f.wc.finalize(¤t_finalized)?; + + let append = linked_chunk(&f.db, 10, 12, "old-9", "old")?; + let result = f.wc.new_chunk(Some(&block(5, "other-5")), &append); + + assert!(result.is_err(), "finalized hash changed at a fixed height"); + assert_eq!(*f.fin_rx.borrow(), Some(current_finalized)); + assert_eq!(stored_chunks(&f)?, vec![first, second]); + Ok(()) + } + + // Accepting a replacement that stops below `fin` would leave the finalized block absent until + // some later flush caught up, so it is refused instead. + #[test] + fn replacement_below_finalized_that_misses_it_is_rejected() -> anyhow::Result<()> { + let mut f = fixture(); + let (first, second) = seed_chain(&mut f)?; + let current_finalized = block(7, "old-7"); + f.wc.finalize(¤t_finalized)?; + + let replacement = linked_chunk(&f.db, 6, 6, "old-5", "new")?; + let result = f.wc.new_chunk(None, &replacement); + + assert!( + result.is_err(), + "replacement that drops the finalized block was accepted" + ); + assert_eq!(*f.fin_rx.borrow(), Some(current_finalized)); + assert_eq!(stored_chunks(&f)?, vec![first, second]); + Ok(()) + } + + // INV-13: the replacement carries `fin`'s own hash unchanged and a different one below it — + // what a guard checking only the boundary would admit. + #[test] + fn replacement_reproducing_finality_but_rewriting_below_it_is_rejected() -> anyhow::Result<()> { + let mut f = fixture(); + let (first, second) = seed_chain(&mut f)?; + let current_finalized = block(7, "old-7"); + f.wc.finalize(¤t_finalized)?; + + let replacement = chunk_with_hashes( + &f.db, + 6, + "old-5", + &["fork-6", "old-7", "old-8", "old-9"].map(str::to_string) + )?; + let result = f.wc.new_chunk(None, &replacement); + + let err = result.unwrap_err(); + assert!( + err.chain().any(|e| e.is::()), + "finality rejections must stay in the unapplicable_fork metric bucket: {err:#}" + ); + assert_eq!(*f.fin_rx.borrow(), Some(current_finalized)); + assert_eq!(stored_chunks(&f)?, vec![first, second]); + // The refused chunk's tables were already durable; leaving them would leak one chunk + // per 60-second retry, since the orphan sweep only runs at startup. + assert_eq!(f.db.cleanup()?, 1, "the refused chunk's tables were not abandoned"); + assert_eq!(f.db.purge_orphan_dirty_tables()?, 0); + Ok(()) + } + + // The honest dual: a reorg above `fin` rewrites the whole straddling chunk and must land. + #[test] + fn replacement_reproducing_the_finalized_range_is_accepted() -> anyhow::Result<()> { + let mut f = fixture(); + let (first, _) = seed_chain(&mut f)?; + let current_finalized = block(7, "old-7"); + f.wc.finalize(¤t_finalized)?; + + let replacement = chunk_with_hashes( + &f.db, + 6, + "old-5", + &["old-6", "old-7", "new-8", "new-9"].map(str::to_string) + )?; + f.wc.new_chunk(None, &replacement)?; + + assert_eq!(f.wc.head(), Some(&get_chunk_head(&replacement))); + assert_eq!(*f.fin_rx.borrow(), Some(current_finalized)); + assert_eq!(stored_chunks(&f)?, vec![first, replacement]); + // Exactly one table is collected — the replaced chunk's. The accepted chunk keeps its + // own, so an over-eager abandon would show up here as two. + assert_eq!(f.db.cleanup()?, 1); + Ok(()) + } + + // With no matching boundary the fallback is the window start, below `fin`: the resume position + // is not clamped, the write path guards instead. The anchor comes from the stored chunk, so a + // full-window replay is still linkage-checked. + #[test] + fn rollback_without_matching_hints_resumes_from_window_start() -> anyhow::Result<()> { + let mut f = fixture(); + seed_chain(&mut f)?; + f.wc.finalize(&block(4, "old-4"))?; + let hints = vec![block(5, "fork-5"), block(9, "fork-9")]; + + let rollback = f.wc.compute_rollback(&hints)?; + + assert_eq!(rollback.resume_from, f.wc.start_block()); + assert_eq!(rollback.expected_parent_hash.as_deref(), Some("genesis")); + assert_eq!(rollback.reach_at_least, Some(4)); + Ok(()) + } + + // The wedge regression: clamping to `fin + 1` (8) gave a mid-chunk position `insert_fork` + // could not satisfy, so resolution stops at the chunk boundary below `fin` instead. + #[test] + fn rollback_from_straddling_chunk_resumes_at_chunk_boundary() -> anyhow::Result<()> { + let mut f = fixture(); + seed_chain(&mut f)?; + f.wc.finalize(&block(7, "old-7"))?; + + let rollback = f.wc.compute_rollback(&[block(8, "fork-8")])?; + + assert_eq!(rollback.resume_from, 6); + assert_eq!(rollback.expected_parent_hash.as_deref(), Some("old-5")); + // The replay must carry the chunk back up to 7, or the swap drops the finalized block. + assert_eq!(rollback.reach_at_least, Some(7)); + Ok(()) + } + + // Nothing finalized is being replaced, so the ingest keeps its own flush boundaries. + #[test] + fn rollback_above_finalized_head_sets_no_reach_floor() -> anyhow::Result<()> { + let mut f = fixture(); + seed_chain(&mut f)?; + f.wc.finalize(&block(5, "old-5"))?; + + let rollback = f.wc.compute_rollback(&[block(9, "fork-9")])?; + + assert_eq!(rollback.resume_from, 6); + assert_eq!(rollback.reach_at_least, None); + Ok(()) + } + + // Retention trims whole chunks, so its floor can land inside the surviving one. Resuming at + // that logical floor gives `insert_fork` an overlapping position and wedges the dataset the + // same way the `fin + 1` clamp did (GAP-3). + #[test] + fn rollback_fallback_resumes_at_the_physical_window_start() -> anyhow::Result<()> { + let mut f = fixture(); + seed_chain(&mut f)?; + f.wc.finalize(&block(7, "old-7"))?; + // Drops [0, 5] and keeps [6, 9] whole, with the logical floor at 7. + f.wc.retain(7, None)?; + assert_eq!(f.wc.start_block(), 7); + + let rollback = f.wc.compute_rollback(&[block(8, "fork-8")])?; + + assert_eq!(rollback.resume_from, 6); + assert_eq!(rollback.expected_parent_hash.as_deref(), Some("old-5")); + assert_eq!(rollback.reach_at_least, Some(7)); + Ok(()) + } + + // A fork that cannot reach up to finality is refused, never resumed below it. + #[test] + fn rollback_with_all_hints_below_finalized_head_is_refused() -> anyhow::Result<()> { + let mut f = fixture(); + seed_chain(&mut f)?; + f.wc.finalize(&block(7, "old-7"))?; + + let err = + f.wc.compute_rollback(&[block(3, "fork-3"), block(5, "fork-5")]) + .unwrap_err(); + assert!( + err.to_string().contains("below finalized head"), + "unexpected error: {err}" + ); + Ok(()) + } + + // An equivocation at the finalized height is refused, never absorbed. + #[test] + fn rollback_with_conflicting_hint_at_finalized_height_is_refused() -> anyhow::Result<()> { + let mut f = fixture(); + seed_chain(&mut f)?; + f.wc.finalize(&block(7, "old-7"))?; + + let err = + f.wc.compute_rollback(&[block(7, "fork-7"), block(9, "fork-9")]) + .unwrap_err(); + assert!( + err.to_string().contains("finalized head hash"), + "unexpected error: {err}" + ); + Ok(()) + } } diff --git a/crates/hotblocks/src/metrics.rs b/crates/hotblocks/src/metrics.rs index 10ed3e0f..cbc759b1 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_WITHHELD_FLUSHES: 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> = @@ -177,6 +179,10 @@ pub(crate) fn report_hash_index_write_metrics(dataset_id: DatasetId, metrics: &H } } +pub(crate) fn report_withheld_flush(dataset_id: DatasetId) { + INGEST_WITHHELD_FLUSHES.get_or_create(&dataset_label!(dataset_id)).inc(); +} + pub fn report_query_too_many_tasks_error() { QUERY_ERROR_TOO_MANY_TASKS.inc(); } @@ -557,6 +563,11 @@ pub fn build_metrics_registry() -> Registry { "Number of completed queries", COMPLETED_QUERIES.clone() ); + registry.register( + "ingest_withheld_flushes", + "Fork replays whose chunk was held back until it covered the finalized head", + INGEST_WITHHELD_FLUSHES.clone() + ); top_registry } diff --git a/crates/hotblocks/src/query/executor.rs b/crates/hotblocks/src/query/executor.rs index 13aadff8..e8341e6e 100644 --- a/crates/hotblocks/src/query/executor.rs +++ b/crates/hotblocks/src/query/executor.rs @@ -80,6 +80,9 @@ impl QuerySlot { sqd_polars::POOL.spawn(move || { let slot = self; let result = catch_unwind(AssertUnwindSafe(|| task(&slot))).map_err(|_| QueryTaskPanicked); + // Release before waking the caller: sending first lets it observe the slot still + // taken and be refused admission for a query that has already finished. + drop(slot); let _ = tx.send(result); }); diff --git a/crates/hotblocks/tests/ct4_finality.rs b/crates/hotblocks/tests/ct4_finality.rs new file mode 100644 index 00000000..1e7aa72b --- /dev/null +++ b/crates/hotblocks/tests/ct4_finality.rs @@ -0,0 +1,310 @@ +//! 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 binding. Conflict windows +//! deliberately omit the old finalized block, forcing fork resolution to resume at a stored chunk +//! boundary; the whole-chunk rewrite that follows is verified on the write path. + +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: a legitimate reorg above `fin` whose common ancestor lies inside a +/// finality-straddling chunk must recover, not wedge. Before the 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, false).await { + panic!("honest-reorg recovery failed: {err:?}"); + } + Ok(()) +} + +/// The same reorg, with the source cutting every response short so the replay's first chunk ends +/// one block below `fin`: accepting it drops the finalized block, refusing it parks the epoch. This +/// is the shape that froze `ethereum-sepolia` for 8h21m on 2026-07-29. +#[tokio::test(flavor = "multi_thread")] +async fn ct4_honest_reorg_recovers_when_the_replay_is_cut_below_finality() -> Result<()> { + let mut h = start_harness(true).await?; + + if let Err(err) = run_honest_reorg_recovery(&mut h, true).await { + panic!("cut-replay recovery failed: {err:?}"); + } + Ok(()) +} + +/// The subtle equivocation: the source reproduces the finalized block's own hash but rewrites a +/// block *below* it. The replacement stays internally linked, so a guard checking only that hash +/// admits it and the finalized prefix changes under readers (INV-13). +#[tokio::test(flavor = "multi_thread")] +async fn ct4_replacement_rewriting_below_finality_is_refused() -> Result<()> { + let mut h = start_harness(true).await?; + + if let Err(err) = run_rewrite_below_finality(&mut h).await { + panic!("below-finality rewrite was not refused: {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<()> { + // A finalized head 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<()> { + // Two separate responses commit 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_rewrite_below_finality(h: &mut Harness) -> Result<()> { + // The [prefix] [finality-straddling] layout puts the resume boundary below `fin`, so the + // replay covers finalized ground. + 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 + ); + let baseline_requests = h.sim.stats(&h.dataset).stream_requests; + + let tampered_at = straddling_chunk_start + (fin.number - straddling_chunk_start) / 2; + ensure!( + straddling_chunk_start <= tampered_at && tampered_at < fin.number, + "the tampered block {tampered_at} must lie inside the replayed range and below finality" + ); + let tampered = h.sim.rewrite_hash_below_finality(&h.dataset, tampered_at)?; + ensure!( + h.model.hash_at(tampered_at).is_some_and(|h| h != tampered.hash), + "the fault did not change the stored hash at {tampered_at}" + ); + // A source fault, so the model stays on the accepted chain: reorg through the simulator + // rather than `Harness::fork`. + let reorg_from = fin.number + (head.number - fin.number) / 2; + h.sim.fork(&h.dataset, reorg_from, STRADDLING_CHUNK_BLOCKS)?; + + await_finality_fault_rejection(h, &head, &fin, baseline_requests).await?; + h.assert_conforms().await?; + Ok(()) +} + +async fn run_honest_reorg_recovery(h: &mut Harness, cut_replay: bool) -> Result<()> { + // The same 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 + ); + + // An honest tip reorg above `fin` but inside the straddling chunk: its common ancestor sits + // below `fin`, so the whole chunk is rewritten and must reproduce the finalized block. + 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" + ); + if cut_replay { + // The lagging report is what lets a cut response end a chunk: the client suppresses its + // own end-of-response commit below the finality it has seen, and a restart resets that to + // the lagging replica's view. Production got there through the 200k-row bound instead. + let cut = fin.number - straddling_chunk_start; + ensure!(cut > 0, "the cut response must stop strictly below finality"); + h.sim.inject_fault(&h.dataset, |f| { + f.max_blocks_per_response = Some(cut as u32); + f.finality_report_cap = Some(straddling_chunk_start); + }); + h.sut.restart().await?; + } + h.fork(reorg_from, STRADDLING_CHUNK_BLOCKS)?; + + 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; + + // The source rewrites a suffix including `fin` and claims the new tip final; a source fault, + // so the reference model stays 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" + ); + + 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; + } +} diff --git a/crates/hotblocks/tests/ct9_source_faults.rs b/crates/hotblocks/tests/ct9_source_faults.rs index f80ab139..efa0b2ad 100644 --- a/crates/hotblocks/tests/ct9_source_faults.rs +++ b/crates/hotblocks/tests/ct9_source_faults.rs @@ -79,6 +79,7 @@ async fn ct9_a_total_source_outage_is_counted_before_ingestion_starts() -> Resul kind: Evm.config_kind().to_string(), // `Head` is what routes the dataset through the probe at all. retention: Retention::Head(100), + disable_compaction: false, sources: vec![format!("http://127.0.0.1:{dead}/{DS}")] }] )) diff --git a/crates/storage/src/db/db.rs b/crates/storage/src/db/db.rs index 79f309f5..65098e83 100644 --- a/crates/storage/src/db/db.rs +++ b/crates/storage/src/db/db.rs @@ -13,6 +13,7 @@ use super::{ use crate::db::{ ops::{perform_dataset_compaction, CompactionStatus}, read::datasets::list_all_datasets, + table_id::TableId, write::{ ops as cleanup_ops, table_builder::TableBuilder, @@ -547,6 +548,18 @@ impl Database { Ok(()) } + /// Schedules `tables` for the ordinary purge. Used to abandon tables whose chunk was + /// refused after they were already written: their dirty marker alone is collected by the + /// startup orphan sweep only, so a caller refused on every retry would leak them. + pub fn delete_tables(&self, tables: &[TableId]) -> anyhow::Result<()> { + Tx::new(&self.db).run(|tx| { + for table_id in tables { + tx.delete_table(table_id)?; + } + Ok(()) + }) + } + /// Phase 1 -- logically purge deleted tables (snapshot-safe point deletes). /// Returns the number of tables logically deleted by this call. pub fn cleanup(&self) -> anyhow::Result { diff --git a/crates/storage/src/db/write/dataset_update.rs b/crates/storage/src/db/write/dataset_update.rs index 153a88bd..9672d408 100644 --- a/crates/storage/src/db/write/dataset_update.rs +++ b/crates/storage/src/db/write/dataset_update.rs @@ -47,6 +47,10 @@ impl<'a> DatasetUpdate<'a> { .validate_parent_block_hash(chunk, block_number, expected_parent_hash) } + pub fn validate_finalized_prefix(&self, chunk: &Chunk, up_to: BlockNumber) -> anyhow::Result> { + self.tx.validate_finalized_prefix(self.dataset_id, chunk, up_to) + } + pub fn delete_chunk(&self, chunk: &Chunk) -> anyhow::Result<()> { self.tx.unindex_hashes(self.dataset_id, chunk)?; self.tx.delete_chunk(self.dataset_id, chunk) diff --git a/crates/storage/src/db/write/tx.rs b/crates/storage/src/db/write/tx.rs index f755330f..8b4a93fb 100644 --- a/crates/storage/src/db/write/tx.rs +++ b/crates/storage/src/db/write/tx.rs @@ -521,6 +521,113 @@ impl<'a> Tx<'a> { } } + /// Compares `chunk` against stored history block for block over + /// `[chunk.first_block(), up_to]`, describing the first divergence. + /// + /// `up_to` is the finalized head (INV-13). Checking its hash alone would not + /// do: hashes come from the source, so a reproduced boundary says nothing + /// about the interior. Identical hashes over different payload are likewise + /// invisible here — content is the source's word at every height. + /// + /// Peak memory is one stored chunk's worth of pairs: the replacement streams + /// past a cursor that pulls stored chunks in one at a time. + pub fn validate_finalized_prefix( + &self, + dataset_id: DatasetId, + chunk: &Chunk, + up_to: BlockNumber + ) -> anyhow::Result> { + let from = chunk.first_block(); + + let stored_chunks = self + .list_chunks(dataset_id, from, Some(up_to)) + .collect::>>()?; + let mut stored_chunks = stored_chunks.iter(); + + let mut stored: Vec<(BlockNumber, String)> = Vec::new(); + let mut pos = 0; + let mut divergence = None; + + // `true` while a stored block is available at `stored[pos]`. + let mut seek_stored = |stored: &mut Vec<(BlockNumber, String)>, pos: &mut usize| -> anyhow::Result { + while *pos >= stored.len() { + let Some(next) = stored_chunks.next() else { + return Ok(false); + }; + *stored = self.read_block_hashes(next, from, up_to)?; + *pos = 0; + } + Ok(true) + }; + + let blocks_table_id = chunk + .tables() + .get("blocks") + .copied() + .ok_or_else(|| anyhow!("'blocks' table does not exist in chunk {}", chunk))?; + + let snapshot = ReadSnapshot::new(self.db); + let reader = snapshot.create_table_reader(blocks_table_id)?; + + for_each_block_hash(&reader, |number, hash| { + if divergence.is_some() || number > up_to { + return Ok(()); + } + if !seek_stored(&mut stored, &mut pos)? { + divergence = Some(format!( + "block {}#{} is not part of the stored finalized history", + number, hash + )); + return Ok(()); + } + let (stored_number, stored_hash) = &stored[pos]; + if *stored_number == number && stored_hash == hash { + pos += 1; + } else { + divergence = Some(format!( + "expected finalized block {}#{}, got {}#{}", + stored_number, stored_hash, number, hash + )); + } + Ok(()) + })?; + + if divergence.is_none() && seek_stored(&mut stored, &mut pos)? { + let (stored_number, stored_hash) = &stored[pos]; + divergence = Some(format!( + "finalized block {}#{} is missing from the replacement", + stored_number, stored_hash + )); + } + + Ok(divergence.map_or(Ok(()), Err)) + } + + fn read_block_hashes( + &self, + chunk: &Chunk, + from: BlockNumber, + to: BlockNumber + ) -> anyhow::Result> { + let blocks_table_id = chunk + .tables() + .get("blocks") + .copied() + .ok_or_else(|| anyhow!("'blocks' table does not exist in chunk {}", chunk))?; + + let snapshot = ReadSnapshot::new(self.db); + let reader = snapshot.create_table_reader(blocks_table_id)?; + + let mut hashes = Vec::new(); + for_each_block_hash(&reader, |number, hash| { + if from <= number && number <= to { + hashes.push((number, hash.to_string())); + } + Ok(()) + })?; + Ok(hashes) + } + pub fn list_chunks( &self, dataset_id: DatasetId,