diff --git a/crates/hotblocks-harness/src/compare.rs b/crates/hotblocks-harness/src/compare.rs index 9b7fd203..e149fa68 100644 --- a/crates/hotblocks-harness/src/compare.rs +++ b/crates/hotblocks-harness/src/compare.rs @@ -203,14 +203,15 @@ pub async fn assert_conforms(client: &Client, model: &Model, chain: &dyn Chain, if observed != expected { errs.push(format!("STATUS: expected {expected:?}, got {observed:?}")); } - // INV-5/30: first <= fin <= head, in one consistent view. + // INV-5/30: fin <= head, in one consistent view. `fin` may sit below the window — + // retention can pass finality, and the watermark survives as a claim about the chain. if let Some(data) = &status.data { if let Some(fin) = &data.finalized_head - && !(data.first_block <= fin.number && fin.number <= data.last_block) + && fin.number > data.last_block { errs.push(format!( - "INV-5: finalized head {} outside [{}, {}]", - fin.number, data.first_block, data.last_block + "INV-5: finalized head {} above the head {}", + fin.number, data.last_block )); } let head_ts = model diff --git a/crates/hotblocks-harness/src/model.rs b/crates/hotblocks-harness/src/model.rs index 511e4024..379aa322 100644 --- a/crates/hotblocks-harness/src/model.rs +++ b/crates/hotblocks-harness/src/model.rs @@ -131,18 +131,21 @@ impl Model { } } if let Some(fin) = &self.fin { - let first = self.first().expect("INV-5: finalized watermark without a window"); let head = self.head().expect("INV-5: finalized watermark without a window").number; assert!( - first <= fin.number && fin.number <= head, - "INV-5: finalized watermark {} outside [{first}, {head}]", + fin.number <= head, + "INV-5: finalized watermark {} above the head {head}", fin.number ); - assert_eq!( - self.block_at(fin.number).map(|b| b.hash.as_str()), - Some(fin.hash.as_str()), - "INV-6: finalized watermark does not name the stored block" - ); + // `fin` may sit below the window: a trim keeps it as a claim about the chain + // (finality does not move back), and no stored block backs it there. + if fin.number >= self.first().expect("INV-5: window without a first block") { + assert_eq!( + self.block_at(fin.number).map(|b| b.hash.as_str()), + Some(fin.hash.as_str()), + "INV-6: finalized watermark does not name the stored block" + ); + } } } @@ -377,10 +380,11 @@ impl Model { } self.seg.drain(..drop); self.anchor = Anchor::new(new_anchor.number, Some(new_anchor.hash)); - // RS-2: the finalized watermark cannot survive below the window. - if let Some(fin) = &self.fin - && fin.number < from - { + // `fin` survives below the window: finality does not move back, so the trim + // deletes the finalized blocks but keeps the watermark as a claim about the + // chain. A trim that empties the window is the exception — an empty window + // has no head to bound the watermark (INV-5), and the label clears with it. + if self.seg.is_empty() { self.fin = None; } self.ver += 1; @@ -437,9 +441,19 @@ impl Model { // be replaced (WP-6): the finalized block is common to both chains unless the // source contradicts finality — repeated rejection here is FM-SRC-5, never RESET. None => match &self.fin { - Some(fin) => ForkResolution::Resume { - from: fin.number + 1, - parent_hash: Some(fin.hash.clone()) + Some(fin) => match self.first() { + // WP-6: a trim may leave `fin` below the window (RS-2). Everything + // stored then sits above the watermark, the finality floor is vacuous, + // and the fallback is the window start with its anchor — `fin + 1` + // would be a position INV-14 forbids `replace` from taking. + Some(first) if fin.number + 1 < first => ForkResolution::Resume { + from: first, + parent_hash: self.anchor.hash.clone() + }, + _ => ForkResolution::Resume { + from: fin.number + 1, + parent_hash: Some(fin.hash.clone()) + } }, // Full-window replacement *probe*; repeated WP-2 rejection at `first(D)` // escalates to RESET per WP-6b. @@ -724,7 +738,7 @@ mod tests { } #[test] - fn retain_trims_moves_the_anchor_and_drops_stale_finality() { + fn retain_trims_moves_the_anchor_and_keeps_finality_below_the_window() { let (mut m, blocks) = model_at(100, 10); m.finalize(&BlockRef::new(101, block_hash(101, 0))); @@ -732,7 +746,11 @@ mod tests { assert_eq!(m.first(), Some(105)); assert_eq!(m.anchor, Anchor::new(104, Some(blocks[4].hash.clone())), "INV-18"); - assert_eq!(m.fin, None, "RS-2: finality below the window is dropped"); + assert_eq!( + m.fin, + Some(BlockRef::new(101, block_hash(101, 0))), + "finality does not move back: the watermark survives below the window" + ); assert_eq!(m.head().unwrap().number, 109); assert_eq!( m.finalize(&BlockRef::new(104, blocks[4].hash.clone())), @@ -745,6 +763,29 @@ mod tests { assert_eq!(m.first(), Some(105)); } + #[test] + fn retain_that_empties_the_window_clears_finality() { + let (mut m, blocks) = model_at(100, 10); + m.finalize(&BlockRef::new(101, block_hash(101, 0))); + + // RETAIN at `next` keeps nothing: an empty window has no head to bound the + // watermark (INV-5), and the service clears the label with the data. + m.retain(110, None); + + assert!(m.seg.is_empty()); + assert_eq!(m.fin, None); + assert_eq!(m.anchor, Anchor::new(109, Some(blocks[9].hash.clone())), "INV-18"); + assert_eq!(m.next(), 110); + + // Re-trimming at the same floor changes nothing, and the window re-attaches + // at the anchor the trim left. + m.retain(110, None); + assert!(m.seg.is_empty()); + + m.extend(&run(110, 2, 0, &blocks[9].hash), None).unwrap(); + assert_eq!(m.first(), Some(110)); + } + #[test] fn retain_below_the_window_is_a_destructive_reset() { let (mut m, _) = model_at(100, 10); @@ -827,6 +868,32 @@ mod tests { ); } + #[test] + fn resolve_fork_with_finality_below_the_window_falls_back_to_the_window_start() { + let (mut m, _) = model_at(100, 10); + m.finalize(&BlockRef::new(101, block_hash(101, 0))); + m.retain(105, None); + + // The watermark (101) sits below the window (105..109): `fin + 1` names a + // position INV-14 forbids, so the fallback is the window start with its anchor. + let hints = vec![ + BlockRef::new(107, block_hash(107, 9)), + BlockRef::new(108, block_hash(108, 9)), + ]; + assert_eq!( + m.resolve_fork(&hints).unwrap(), + ForkResolution::Resume { + from: 105, + parent_hash: Some(block_hash(104, 0)) + } + ); + + // The resume point is a position `replace` accepts. + let fork = run(105, 2, 9, &block_hash(104, 0)); + m.replace(105, &fork, None).unwrap(); + assert_eq!(m.head().unwrap().hash, block_hash(106, 9)); + } + #[test] fn resolve_fork_contradicting_the_anchor_resets() { let (m, _) = model_at(100, 5); diff --git a/crates/hotblocks/spec/02-data-model.md b/crates/hotblocks/spec/02-data-model.md index 717f5c5a..b1659652 100644 --- a/crates/hotblocks/spec/02-data-model.md +++ b/crates/hotblocks/spec/02-data-model.md @@ -116,9 +116,11 @@ protocol anchors a client or a source to a position — the anchor (DEF-7), **not** "a block numbered `p` has hash `h`". The two readings coincide exactly on densely-numbered chains. -**DEF-8 (Finalized watermark).** `fin`, when defined, designates a stored block -(`first ≤ fin.number ≤ head.number`, hash matching the stored block — INV-5/6) that the -sources have declared irreversible. Blocks at or below `fin.number` form the **finalized +**DEF-8 (Finalized watermark).** `fin`, when defined, designates a block +(`fin.number ≤ head.number` — INV-5) that the sources have declared irreversible. When +`fin.number ≥ first(D)` it matches the stored block at that height (INV-6); a trim may +leave it below the window, where it survives as a claim about the chain (RS-2) with no +stored block behind it. Blocks at or below `fin.number` form the **finalized prefix**; blocks above it are the **volatile suffix**, subject to replacement by forks. **DEF-9 (Retention policy).** One of: diff --git a/crates/hotblocks/spec/03-write-path.md b/crates/hotblocks/spec/03-write-path.md index a6ba3d82..ca8a20d2 100644 --- a/crates/hotblocks/spec/03-write-path.md +++ b/crates/hotblocks/spec/03-write-path.md @@ -140,7 +140,9 @@ the stored chain (or the anchor). If no hint matches stored state, the fallback respect the finality floor: - `fin ≠ ⊥` → resume from `⟨fin.number + 1, fin.hash⟩`, or from any lower physical position - `≥ chunk_first(D)` whose replacement reproduces the finalized prefix (INV-14). The finalized + `≥ chunk_first(D)` whose replacement reproduces the finalized prefix (INV-14). When + `fin` sits below the window (kept by a trim, RS-2), everything stored is above it, the + finality floor is vacuous, and the fallback position is `chunk_first(D)`. 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 chunk MUST take the lower position when `fin` falls inside a chunk: `fin + 1` @@ -236,9 +238,9 @@ history cannot be re-acquired through RETAIN. destructive re-bootstrap, not a widening, and controllers MUST treat it as such (FM-OP-4). During boot validation of a `Pinned` policy whose `from` lies below the stored window, the service MUST refuse the dataset (INV-43) rather than destroy data — - 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 + symmetric with WP-9. *Finality note:* like every RESET (§2.6), this discards the + finalized prefix and clears `fin` — retention dominates finality (upward trims keep + `fin`, §2.5). It is **not** a rollback below `fin`: the fork floor (INV-13/14) 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. @@ -246,8 +248,10 @@ history cannot be re-acquired through RETAIN. - Post: `seg' = seg[from‥]`; `anchor' = ⟨from − 1, hash of from's preceding block in the pre-trim state⟩` (= `hash_at(from − 1)`, DEF-16). The hash MUST be carried over from previously stored data, never recomputed or dropped (INV-18); - - `fin' = fin` if `fin.number ≥ from`, else `fin' = ⊥` (retention dominates finality, - RS-2). + - `fin' = fin`: finality does not move back, so the watermark survives below the window + as a claim about the chain (RS-2) and keeps anchoring responses until finality climbs + back in. Exception: `from = next(D)` leaves `seg' = ∅`, and an empty window has no head + to bound the watermark (`seg = ∅ ⇒ fin = ⊥`, INV-5) — the label clears with the data. - Case `from > next(D)`: equivalent to `RESET(⟨from − 1, h?⟩)` — the whole window is discarded and the dataset awaits future blocks. - **WP-9 (Mismatch policy).** If `h?` is provided and disagrees with the stored chain's diff --git a/crates/hotblocks/spec/06-invariants.md b/crates/hotblocks/spec/06-invariants.md index 21da44a2..5447826a 100644 --- a/crates/hotblocks/spec/06-invariants.md +++ b/crates/hotblocks/spec/06-invariants.md @@ -43,13 +43,16 @@ enclosing block's number. *Check:* CT-1, CT-5 structural validators. **INV-5 — Watermark bounds.** [state] -If `seg = ∅` then `fin = ⊥`. Otherwise, if `fin ≠ ⊥` then -`first(D) ≤ fin.number ≤ head(D).number`. -*Why:* a finalized pointer outside the window is meaningless and breaks fork-floor logic. -*Check:* CT-1 (assert on every STATUS read). +If `seg = ∅` then `fin = ⊥`. Otherwise, if `fin ≠ ⊥` then `fin.number ≤ head(D).number`; +a trim may leave `fin.number < first(D)` (RS-2). +*Why:* finality does not move back, so a watermark below the window stays a true claim +about the chain and clients read the whole window as unfinalized; above the head it +would be meaningless and break fork-floor logic. +*Check:* CT-1 (assert on every STATUS read); retention passing finality (CT-10). **INV-6 — Finalized-on-chain.** [state] -If `fin ≠ ⊥` then the stored block at height `fin.number` has hash `fin.hash`. +If `fin ≠ ⊥` and `fin.number ≥ first(D)`, then the stored block at height `fin.number` +has hash `fin.hash`. Below the window no stored block backs the watermark. *Why:* finality must describe the chain actually served, else finalized-only reads lie. *Check:* CT-1; CT-4 with equivocating-finality sources; write-controller finality resolver regressions. @@ -77,7 +80,7 @@ previous head. **INV-12 — Finality monotonicity.** [transition] `fin.number` never decreases while defined, and at a fixed height the finalized hash never -changes. `fin` may become `⊥` only via RETAIN (window passing above it), RESET, or DROP. +changes. `fin` may become `⊥` only via a RETAIN that empties the window, RESET, or DROP. *Why:* clients treat finalized data as irreversible-while-retained. *Check:* CT-1 (monitor every FINALIZED-HEAD read); CT-4 stale/regressing finality sources. diff --git a/crates/hotblocks/spec/09-retention-and-space.md b/crates/hotblocks/spec/09-retention-and-space.md index d32121a3..3c2c4ec6 100644 --- a/crates/hotblocks/spec/09-retention-and-space.md +++ b/crates/hotblocks/spec/09-retention-and-space.md @@ -16,10 +16,14 @@ between them. | `Unbounded` | nothing trimmed | never | - **RS-2 (Retention dominates finality).** Trimming ignores `fin`: finalized blocks below - the retention bound are deleted, and `fin` becomes `⊥` when the window passes above it - (03 §2.5). Rationale: this is a bounded hot store, not an archive (NG1). Consequence - for clients: "finalized" means *irreversible while retained*, not *retained forever* - (INV-24 note). + the retention bound are deleted. The watermark itself survives below the window + (03 §2.5): finality does not move back, so `fin` stays a true claim about the chain, + responses keep reporting it, and clients read the whole window as unfinalized. Only a + RESET — or a trim that empties the window, where `seg = ∅` forces `fin = ⊥` (INV-5) — + clears it. Rationale: this is a bounded hot store, not an archive (NG1). + Consequence for clients: "finalized" means *irreversible*, not *retained forever* + (INV-24 note) — blocks at or below `fin.number` may be gone even while `fin` reports + them final. - **RS-3 (Availability floor).** For `Window(k)`: at every committed state, all blocks in `[next(D) − k, next(D) − 1] ∩ [window start after initial fill, ∞)` are present and queryable (an interval of *positions* — DEF-9; on a slot-numbered chain it diff --git a/crates/hotblocks/spec/12-conformance-tdd.md b/crates/hotblocks/spec/12-conformance-tdd.md index 2e5433ab..ff852991 100644 --- a/crates/hotblocks/spec/12-conformance-tdd.md +++ b/crates/hotblocks/spec/12-conformance-tdd.md @@ -123,7 +123,10 @@ model Dataset: if h != ⊥ and h != hash_at(from_ - 1): return reset((from_ - 1, h)) # WP-9 self-heal new_anchor = (from_ - 1, hash_at(from_ - 1)) # INV-18 (DEF-16) seg = [b in seg | b.number >= from_]; anchor = new_anchor - if fin != ⊥ and fin.number < from_: fin = ⊥ # RS-2 + if not seg: fin = ⊥ # from_ = next() emptied the window: + # no head bounds the watermark (INV-5) + # RS-2: `fin` survives below the window — finality does not move back, so the + # watermark stays a true claim about the chain and keeps anchoring responses. ver += 1; wf() else: reset((from_ - 1, h)) # from_ > next() @@ -141,7 +144,12 @@ 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) # or any lower position reproducing the + if fin != ⊥: + if seg and fin.number + 1 < first(): return (first(), anchor.hash) + # fin sits below the window (RS-2): the + # finality floor is vacuous, so the fallback + # is the window start (WP-6) + 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, diff --git a/crates/hotblocks/src/dataset_controller/write_controller.rs b/crates/hotblocks/src/dataset_controller/write_controller.rs index dfb4e9c6..bcad3172 100644 --- a/crates/hotblocks/src/dataset_controller/write_controller.rs +++ b/crates/hotblocks/src/dataset_controller/write_controller.rs @@ -291,15 +291,11 @@ impl WriteController { .next() .expect("bottom chunk can't exist without head chunk")?; - let finalized_head = tx - .label() - .finalized_head() - .filter(|h| chunk.first_block() <= h.number) - .cloned(); - - if finalized_head.is_none() { - tx.set_finalized_head(None) - } + // Finality does not move back, so the head stays true even when + // the trim leaves it below the window. Dropping it would strip + // finality information from every response until the chain's + // finality climbs back into the window. + let finalized_head = tx.label().finalized_head().cloned(); return Ok(Status::Range { first_chunk: chunk, @@ -1668,6 +1664,61 @@ mod tests { Ok(()) } + // Finality does not move back, so a trim that passes the finalized head keeps it as a + // claim about the chain: every block left in the window is then unfinalized, and + // responses keep reporting the head instead of losing finality information. + #[test] + fn a_trim_past_the_finalized_head_keeps_it() -> anyhow::Result<()> { + let mut f = fixture(); + let (_, second) = seed_chain(&mut f)?; + let finalized = block(3, "old-3"); + f.wc.finalize(&finalized)?; + + assert!(f.wc.retain(6, None)?); + + assert_eq!(stored_chunks(&f)?, vec![second]); + assert_eq!(f.wc.start_block(), 6); + assert_eq!(*f.fin_rx.borrow(), Some(finalized.clone())); + assert_eq!( + f.db.snapshot() + .get_label(f.dataset_id)? + .and_then(|label| label.finalized_head().cloned()), + Some(finalized) + ); + Ok(()) + } + + // The kept head survives a writer restart, and while it sits below the window a report + // below the window changes nothing while one inside the window raises it. + #[test] + fn a_kept_head_below_the_window_survives_and_advances() -> anyhow::Result<()> { + let mut f = fixture(); + seed_chain(&mut f)?; + f.wc.finalize(&block(3, "old-3"))?; + f.wc.retain(6, None)?; + drop(f.wc); + + let (head_tx, _head_rx) = watch::channel(None); + let (fin_tx, fin_rx) = watch::channel(None); + let mut wc = WriteController::new(f.db.clone(), f.dataset_id, DatasetKind::Evm, head_tx, fin_tx)?; + assert_eq!(*fin_rx.borrow(), Some(block(3, "old-3"))); + + wc.finalize(&block(4, "old-4"))?; + assert_eq!( + *fin_rx.borrow(), + Some(block(3, "old-3")), + "a below-window report must not move the head" + ); + + wc.finalize(&block(7, "old-7"))?; + assert_eq!( + *fin_rx.borrow(), + Some(block(7, "old-7")), + "an in-window report must raise the head" + ); + Ok(()) + } + // `fin` anchors the finalized-prefix guard and every later `compute_rollback`, so a report // naming a hash the chunk does not carry is an equivocation, not a detail. #[test] diff --git a/crates/hotblocks/src/query/running.rs b/crates/hotblocks/src/query/running.rs index a3b8ec43..ee81f1c3 100644 --- a/crates/hotblocks/src/query/running.rs +++ b/crates/hotblocks/src/query/running.rs @@ -157,13 +157,23 @@ pub struct RunningQuery { } fn finalized_query_last_block( + first_block: BlockNumber, last_block: Option, finalized_head: Option<&BlockRef> ) -> anyhow::Result { let Some(finalized_head) = finalized_head else { bail!(QueryIsAboveTheHead { finalized_head: None }) }; - Ok(last_block.map_or(finalized_head.number, |end| end.min(finalized_head.number))) + let end = last_block.map_or(finalized_head.number, |end| end.min(finalized_head.number)); + // A trim can leave the finalized head below the window, so everything the request + // wants is unfinalized: that is NO_DATA, never an empty 200 or a range error. + ensure!( + end >= first_block, + QueryIsAboveTheHead { + finalized_head: Some(finalized_head.clone()) + } + ); + Ok(end) } impl RunningQuery { @@ -236,7 +246,11 @@ impl RunningQuery { }; let last_block = if only_finalized { - Some(finalized_query_last_block(range.last_block, finalized_head.as_ref())?) + Some(finalized_query_last_block( + range.first_block, + range.last_block, + finalized_head.as_ref() + )?) } else { range.last_block }; @@ -440,13 +454,39 @@ mod tests { #[test] fn finalized_snapshot_without_a_head_is_no_data() { - let query = Query::from_json_value(serde_json::json!({ - "type": "evm", - "fromBlock": 10 - })) - .unwrap(); + let err = finalized_query_last_block(10, None, None).unwrap_err(); + assert!(err.is::()); + } + + // A kept head below the window must read as NO_DATA, not as an empty coverage window. + #[test] + fn finalized_head_below_the_window_is_no_data() { + let fin = BlockRef { + number: 50, + hash: "h50".to_string() + }; + + let err = finalized_query_last_block(100, None, Some(&fin)).unwrap_err(); + let above_the_head = err + .downcast_ref::() + .expect("a head below the window must stay a NO_DATA"); + assert_eq!(above_the_head.finalized_head, Some(fin.clone())); - let err = finalized_query_last_block(query.last_block(), None).unwrap_err(); + let err = finalized_query_last_block(100, Some(120), Some(&fin)).unwrap_err(); assert!(err.is::()); } + + #[test] + fn a_finalized_head_inside_the_range_bounds_the_end() { + let fin = BlockRef { + number: 150, + hash: "h150".to_string() + }; + + let end = finalized_query_last_block(100, Some(120), Some(&fin)).unwrap(); + assert_eq!(end, 120); + + let end = finalized_query_last_block(100, Some(180), Some(&fin)).unwrap(); + assert_eq!(end, 150); + } } diff --git a/crates/hotblocks/src/query/service.rs b/crates/hotblocks/src/query/service.rs index 8a3a8a69..55d4c22f 100644 --- a/crates/hotblocks/src/query/service.rs +++ b/crates/hotblocks/src/query/service.rs @@ -180,7 +180,11 @@ impl QueryService { } }) .await - .map_err(|_| QueryIsAboveTheHead { finalized_head: None })?; + // RP-5: NO_DATA carries the current watermarks, so pollers keep their + // finality information even when the finalized head sits below the window. + .map_err(|_| QueryIsAboveTheHead { + finalized_head: dataset.get_finalized_head() + })?; } QueryResponse::new( diff --git a/crates/hotblocks/tests/ct10_finality_below_window.rs b/crates/hotblocks/tests/ct10_finality_below_window.rs new file mode 100644 index 00000000..66a1cf21 --- /dev/null +++ b/crates/hotblocks/tests/ct10_finality_below_window.rs @@ -0,0 +1,194 @@ +//! CT-10 — retention may pass the finalized head; finality does not move back, so the +//! watermark must survive below the window (RS-2 as revised, INV-5). +//! +//! Covers the three client-visible halves of the behavior: a 200 from the window keeps +//! reporting the head below it, the NO_DATA answers (a live query above the head and a +//! finalized-only query from the window start) carry the watermark too, and a report that +//! climbs back into the window raises it again. + +use std::{ + sync::Arc, + time::{Duration, Instant} +}; + +use anyhow::{Context, Result, bail, ensure}; +use serde_json::json; +use sqd_hotblocks_harness::{ + chain::Evm, + driver::{Outcome, StatusData}, + harness::{Harness, HarnessConfig}, + sut::Retention, + types::{BlockNumber, BlockRef} +}; + +const START: u64 = 1_000; +/// `Head(n)`: small enough that the second run pushes the window past the finalized head. +const WINDOW: u64 = 50; +/// Finality lag while it still sits inside the window: `lag < WINDOW` blocks. +const INSIDE_LAG: u64 = 30; +const FIRST_RUN: u32 = 200; +const SECOND_RUN: u32 = 100; +const POLL: Duration = Duration::from_millis(100); +const CONVERGE: Duration = Duration::from_secs(30); + +#[tokio::test(flavor = "multi_thread")] +async fn ct10_the_finalized_head_survives_a_window_that_passes_it() -> Result<()> { + let cfg = HarnessConfig { + retention: Retention::Api, + // One chunk per source response, so the trim's floor lands on real boundaries. + disable_compaction: true, + ..HarnessConfig::from_block(env!("CARGO_BIN_EXE_sqd-hotblocks"), Arc::new(Evm), START) + }; + let mut h = Harness::start(cfg).await?; + + // The source holds the whole first run and its (lagging) finality before the service + // ingests anything, so the window lands deterministically on top of both. + h.produce(FIRST_RUN)?; + h.finalize_with_lag(INSIDE_LAG)?; + let fin = h + .model + .fin + .clone() + .context("the script finalized nothing before the window was set")?; + + let status = h.client.set_retention(&json!({ "Head": WINDOW })).await?; + ensure!(status == 200, "SET-RETENTION was refused with {status}"); + + // Wait for the first run to commit as a chunk that ends at the tip: the second run + // must not arrive while the ingest still streams the first one, or a single response + // commits one chunk spanning the finalized head. + await_head_and_finality(&h, fin.clone()).await?; + + // Finality stalls while the window advances: the trim passes the finalized head. + h.produce(SECOND_RUN)?; + + let data = await_window_past(&h, fin.number).await?; + ensure!( + fin.number < data.first_block, + "the window [{}, {}] never passed the finalized head {}", + data.first_block, + data.last_block, + fin.number + ); + + // FINALIZED-HEAD itself keeps the pre-trim value. + let observed = h.client.finalized_head().await?; + ensure!( + observed.as_ref() == Some(&fin), + "FINALIZED-HEAD lost the watermark below the window: expected {fin:?}, got {observed:?}" + ); + + // A 200 from inside the window still classifies the whole window as unfinalized. + let from = data.first_block; + let query = h.chain.scan_query(from, Some(from + 10), None); + match h.client.query(&query).await? { + Outcome::Ok { blocks, watermarks } => { + ensure!(!blocks.is_empty(), "the window scan covered nothing"); + ensure!( + watermarks.finalized.as_ref() == Some(&fin), + "a 200 from the window lost the finalized head: {watermarks:?}" + ); + } + other => bail!("the window scan did not answer 200: {other:?}") + } + + // NO_DATA carries the watermark too: a live query above the head, and a finalized-only + // query whose whole range is unfinalized while the head sits below the window. + let above_head = h.chain.scan_query(data.last_block + 1, None, None); + let finalized_from_window = h.chain.scan_query(from, None, None); + let (live, finalized) = tokio::join!( + h.client.query(&above_head), + h.client.query_finalized(&finalized_from_window) + ); + let checks = [ + ("a live query above the head", live?), + ("a finalized-only query from the window", finalized?) + ]; + for (label, outcome) in checks { + match outcome { + Outcome::NoData { watermarks } => ensure!( + watermarks.finalized.as_ref() == Some(&fin), + "{label} lost the finalized head on NO_DATA: {watermarks:?}" + ), + other => bail!("{label} did not answer NO_DATA: {other:?}") + } + } + + // Finality climbing back into the window raises the watermark again. + let recovered_at = data.last_block - 20; + h.finalize(recovered_at)?; + let expected = h + .model + .fin + .clone() + .context("the recovery finalize did not apply in the model")?; + let observed = await_finalized_head(&h, &expected).await?; + ensure!( + observed == Some(expected), + "finality inside the window did not raise the head: got {observed:?}" + ); + + Ok(()) +} + +/// Waits until the service holds the model's head and watermark. The model never trims, +/// so the window itself is not compared here. +async fn await_head_and_finality(h: &Harness, fin: BlockRef) -> Result<()> { + let expected_head = h + .model + .head() + .context("the model has no head before the window was set")?; + let deadline = Instant::now() + CONVERGE; + loop { + let head = h.client.head().await?; + let observed_fin = h.client.finalized_head().await?; + if head.as_ref() == Some(&expected_head) && observed_fin.as_ref() == Some(&fin) { + return Ok(()); + } + ensure!( + Instant::now() < deadline, + "the service did not reach head {expected_head:?} / fin {fin:?}: got {head:?} / {observed_fin:?}" + ); + tokio::time::sleep(POLL).await; + } +} + +/// Waits until the whole chain is in and the window starts above `fin`. +async fn await_window_past(h: &Harness, fin: BlockNumber) -> Result { + let expected_head = h + .model + .head() + .context("the model has no head after the second run")? + .number; + let deadline = Instant::now() + CONVERGE; + loop { + let status = h.client.status().await?; + if let Some(data) = &status.data + && data.last_block == expected_head + && data.first_block > fin + { + return Ok(data.clone()); + } + ensure!( + Instant::now() < deadline, + "the service did not converge on a window past {fin}: {:?}", + status.data + ); + tokio::time::sleep(POLL).await; + } +} + +async fn await_finalized_head(h: &Harness, expected: &BlockRef) -> Result> { + let deadline = Instant::now() + CONVERGE; + loop { + let observed = h.client.finalized_head().await?; + if observed.as_ref() == Some(expected) { + return Ok(observed); + } + ensure!( + Instant::now() < deadline, + "FINALIZED-HEAD never reached {expected:?}: got {observed:?}" + ); + tokio::time::sleep(POLL).await; + } +}