test(hotblocks): cover the multi-source shape production runs

crates/hotblocks-harness/README.md

@@ -75,6 +75,15 @@ block is emitted (RP-9); a filter-sparse query returning nothing tells the clien
 how far it got (GAP-8). Scanning with `include_all` sidesteps that, so the harness can always
 advance. Do not "optimize" it away.
 
+**A source asked above its tip answers no-data, not a fork.** RP-5b confines the fork signal
+to `from == tip + 1` — the one position where the parent assertion is evaluable against a block
+the source actually holds. The simulator used to signal a fork at *any* position above its tip,
+which made a source that is merely **behind** indistinguishable from one that **disagrees**; with
+several endpoints per dataset that is the difference between a laggard and a reorg. The old
+behavior is kept as an explicit fault (`SimFaults::fork_signal_above_tip`), because a real source
+doing it wedges the service — one such endpoint out of three is enough
+(`ct4_a_single_source_signalling_a_fork_above_its_tip_does_not_park_ingestion`).
+
 **Numbering may be sparse.** Solana numbers blocks by time-based slots and a slot that produced
 nothing leaves a hole, so contiguous *numbering* is not an invariant — being parent and child is
 (`Block::parent_number`). The service agrees: it links batches and chunks by hash and never by
@@ -124,7 +133,13 @@ Fixed in `crates/data-client/src/reqwest/lines.rs`; pinned by a unit test there
   matrix.
 - **CT-4 (fork/finality corpus)** — `Harness::fork()` and the model's `resolve_fork` /
   `Finalize::IntegrityFault` are implemented and unit-tested; the follower implements the
-  normative CONFLICT recovery of 04 §7. What is missing is the scripts.
+  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-5 (error taxonomy)** — `ct5_error_soundness` covers unsupported-dialect containment,
   error classification, and mid-stream worker-panic abort; the anchored check across large
   sparse-number holes is deferred (GAP-21, test `#[ignore]`d). `Model::predict_query` supplies

crates/hotblocks-harness/src/harness.rs

@@ -8,14 +8,14 @@
 
 use std::{path::Path, sync::Arc, time::Duration};
 
-use anyhow::{Context, Result, bail};
+use anyhow::{Context, Result, bail, ensure};
 
 use crate::{
     chain::Chain,
     compare::{self, Quiescence},
     driver::{Client, Follower},
     model::{Finalize, Model},
-    sim::{Numbering, SimConfig, SimDataset, SourceSim},
+    sim::{Numbering, SimConfig, SimDataset, SimStats, SourceSim},
     sut::{DatasetSpec, Retention, Sut, SutConfig},
     types::{Anchor, BlockNumber, block_hash}
 };
@@ -34,6 +34,11 @@ pub struct HarnessConfig {
     pub anchored: bool,
     /// How long the simulator holds a `/stream` request that has nothing to serve.
     pub source_poll: Duration,
+    /// How many independent simulators serve the dataset. Production configures several
+    /// endpoints per dataset, so anything above 1 exercises `StandardDataSource`'s
+    /// multi-endpoint path: the fixed-order poll race, the fork consensus, and the
+    /// per-endpoint `MaybeOnHead` flush trigger.
+    pub sources: usize,
     pub rust_log: String,
     pub quiescence: Quiescence,
     pub sut_args: Vec<String>
@@ -56,6 +61,7 @@ impl HarnessConfig {
             },
             anchored: true,
             source_poll: Duration::from_millis(200),
+            sources: 1,
             rust_log: "info".to_string(),
             quiescence: Quiescence::default(),
             sut_args: sut.args
@@ -64,39 +70,59 @@ impl HarnessConfig {
 }
 
 pub struct Harness {
+    /// The primary source — the one a script drives by default.
     pub sim: SourceSim,
+    /// The remaining endpoints serving the same dataset. Block hashes are a pure function of
+    /// ⟨number, fork⟩, so a peer fed the same script holds a byte-identical chain and one fed
+    /// less holds an exact prefix: a source that is *behind*, not one that disagrees.
+    pub peers: Vec<SourceSim>,
     pub sut: Sut,
     pub client: Client,
     pub model: Model,
     pub chain: Arc<dyn Chain>,
     pub dataset: String,
     pub start_block: BlockNumber,
     pub numbering: Numbering,
-    pub quiescence: Quiescence
+    pub quiescence: Quiescence,
+    /// Per peer, the blocks it owes the primary. Per peer because production's laggard is a
+    /// *minority* — one endpoint of three, not all of them.
+    peer_deficits: Vec<u32>
 }
 
 impl Harness {
     pub async fn start(cfg: HarnessConfig) -> Result<Self> {
-        let sim = SourceSim::start(SimConfig {
-            datasets: vec![SimDataset {
-                id: cfg.dataset.clone(),
-                chain: cfg.chain.clone(),
-                start_block: cfg.start_block,
-                base_timestamp_ms: cfg.base_timestamp_ms,
-                numbering: cfg.numbering
-            }],
-            poll_timeout: cfg.source_poll
-        })
-        .await
-        .context("failed to start the source simulator")?;
+        ensure!(cfg.sources >= 1, "a dataset needs at least one source");
+
+        let mut sims = Vec::with_capacity(cfg.sources);
+        for _ in 0..cfg.sources {
+            sims.push(
+                SourceSim::start(SimConfig {
+                    datasets: vec![SimDataset {
+                        id: cfg.dataset.clone(),
+                        chain: cfg.chain.clone(),
+                        start_block: cfg.start_block,
+                        base_timestamp_ms: cfg.base_timestamp_ms,
+                        numbering: cfg.numbering
+                    }],
+                    poll_timeout: cfg.source_poll
+                })
+                .await
+                .context("failed to start the source simulator")?
+            );
+        }
+        let peers = sims.split_off(1);
+        let sim = sims.pop().expect("at least one source");
 
         let mut sut_cfg = SutConfig::new(
             &cfg.bin,
             vec![DatasetSpec {
                 id: cfg.dataset.clone(),
                 kind: cfg.chain.config_kind().to_string(),
                 retention: cfg.retention.clone(),
-                sources: vec![sim.base_url(&cfg.dataset)]
+                sources: std::iter::once(&sim)
+                    .chain(peers.iter())
+                    .map(|s| s.base_url(&cfg.dataset))
+                    .collect()
             }]
         );
         sut_cfg.args = cfg.sut_args;
@@ -108,7 +134,9 @@ impl Harness {
         let model = Model::new(Anchor::new(cfg.start_block - 1, anchor_hash));
 
         Ok(Self {
+            peer_deficits: vec![0; peers.len()],
             sim,
+            peers,
             sut,
             client,
             model,
@@ -120,21 +148,105 @@ impl Harness {
         })
     }
 
-    /// The source mints `n` more blocks; the model EXTENDs by the same run.
+    /// Every source mints `n` more blocks; the model EXTENDs by the same run.
     pub fn produce(&mut self, n: u32) -> Result<()> {
+        self.produce_lagging(&[], n)
+    }
+
+    /// Only the primary mints: *every* peer falls `n` blocks behind.
+    pub fn produce_ahead(&mut self, n: u32) -> Result<()> {
+        let all: Vec<usize> = (0..self.peers.len()).collect();
+        self.produce_lagging(&all, n)
+    }
+
+    /// The primary and every peer outside `behind` mint `n` blocks; those named hold and fall
+    /// `n` further back.
+    ///
+    /// A peer left behind holds an exact prefix of the canonical chain: healthy, agreeing, merely
+    /// not caught up — the one state `StandardDataSource` cannot name, since an endpoint leaves
+    /// the rotation by erroring and never by being slow.
+    pub fn produce_lagging(&mut self, behind: &[usize], n: u32) -> Result<()> {
+        for &i in behind {
+            ensure!(
+                i < self.peers.len(),
+                "peer {i} does not exist — {} configured",
+                self.peers.len()
+            );
+        }
         let blocks = self.sim.produce(&self.dataset, n);
-        self.model.extend(&blocks, None)
+        self.model.extend(&blocks, None)?;
+        for i in 0..self.peers.len() {
+            if behind.contains(&i) {
+                self.peer_deficits[i] += n;
+            } else {
+                self.mint_on_peer(i, n)?;
+            }
+        }
+        Ok(())
+    }
+
+    /// Bring every peer back up to the primary's tip.
+    pub fn catch_up_peers(&mut self) -> Result<()> {
+        for i in 0..self.peers.len() {
+            let owed = std::mem::take(&mut self.peer_deficits[i]);
+            self.mint_on_peer(i, owed)?;
+        }
+        Ok(())
+    }
+
+    /// Mint on a peer and check it reproduced the primary's chain. Determinism is the premise
+    /// of every multi-source script: the moment it breaks, a lag test silently becomes a fork
+    /// test and stops testing what it claims to.
+    fn mint_on_peer(&self, i: usize, n: u32) -> Result<()> {
+        let minted = self.peers[i].produce(&self.dataset, n);
+        let canonical = self.model.blocks_in(
+            minted.first().map_or(0, |b| b.number),
+            minted.last().map_or(0, |b| b.number)
+        );
+        // `zip` stops at the shorter side: without this a length mismatch passes vacuously.
+        ensure!(
+            minted.len() == canonical.len(),
+            "peer {i} minted {} blocks over a range the canonical chain fills with {}",
+            minted.len(),
+            canonical.len()
+        );
+        for (got, want) in minted.iter().zip(canonical) {
+            ensure!(
+                got.hash == want.hash && got.number == want.number,
+                "peer {i} diverged from the primary at block {}: {} vs {}",
+                got.number,
+                got.hash,
+                want.hash
+            );
+        }
+        Ok(())
     }
 
     /// The source reorgs: the suffix at `from` is replaced by `len` freshly-minted blocks.
+    ///
+    /// Single-source only. Reorging one endpoint of several is not "a fork" but a 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 needs its own script, not an accidental one.
     pub fn fork(&mut self, from: BlockNumber, len: u32) -> Result<()> {
+        ensure!(
+            self.peers.is_empty(),
+            "fork() drives the primary only; with peers configured it would script a source \
+             disagreement, not a reorg"
+        );
         let blocks = self.sim.fork(&self.dataset, from, len)?;
         self.model.replace(from, &blocks, None)
     }
 
-    /// The source declares block `number` final.
+    /// The sources declare block `number` final. A peer that has not reached it yet stays
+    /// silent — a lagging source cannot declare a block it does not hold.
     pub fn finalize(&mut self, number: BlockNumber) -> Result<()> {
         let r = self.sim.finalize(&self.dataset, number)?;
+        for peer in &self.peers {
+            if peer.tip(&self.dataset).is_some_and(|t| t.number >= number) {
+                peer.finalize(&self.dataset, number)?;
+            }
+        }
         match self.model.finalize(&r) {
             Finalize::Applied | Finalize::Ignored => Ok(()),
             Finalize::IntegrityFault => bail!("the script declared a finality that contradicts the model: {r}")
@@ -157,7 +269,16 @@ impl Harness {
     pub async fn settle(&self) -> Result<()> {
         compare::await_quiescence(&self.client, &self.model, &self.quiescence)
             .await
-            .with_context(|| format!("the source saw: {:?}", self.sim.stats(&self.dataset)))
+            .with_context(|| format!("the sources saw: {:?}", self.source_stats()))
+    }
+
+    /// What each endpoint saw, primary first — in a multi-source script the endpoint that
+    /// explains a failure is rarely the driven one.
+    pub fn source_stats(&self) -> Vec<SimStats> {
+        std::iter::once(&self.sim)
+            .chain(self.peers.iter())
+            .map(|s| s.stats(&self.dataset))
+            .collect()
     }
 
     /// Diff every observable against the model.

crates/hotblocks-harness/src/sim.rs

@@ -80,7 +80,12 @@ impl Numbering {
 #[derive(Clone, Copy, Debug, Default)]
 pub struct SimFaults {
     /// Serve a JSONL body whose final record is not newline-terminated.
-    pub unterminated_final_line: bool
+    pub unterminated_final_line: bool,
+    /// Answer a parent assertion naming a position above the tip with a fork signal instead of
+    /// 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
 }
 
 /// Counters a test can assert on (how the SUT actually drove the source).
@@ -179,6 +184,8 @@ impl SourceSim {
     /// Turn a source-side fault on or off (FM-SRC-*).
     pub fn inject_fault(&self, dataset: &str, f: impl FnOnce(&mut SimFaults)) {
         self.with(dataset, |d| f(&mut d.faults));
+        // Without it a peer parked in the long poll serves one more request under the old behavior.
+        self.bump();
     }
 
     fn with<T>(&self, dataset: &str, f: impl FnOnce(&mut DatasetSim) -> T) -> T {
@@ -352,14 +359,19 @@ impl DatasetSim {
                     return Reply::Fork(self.hints_ending_at(parent_pos));
                 }
                 None => {
-                    // Above our tip: it claims blocks we do not have — disagree, hint at our tip.
+                    // Strictly above our tip, so the assertion names a position we have not
+                    // reached: not evaluable, and RP-5b confines the fork signal to the one
+                    // position where it is (`from == tip + 1`, handled above). A source that is
+                    // merely *behind* must say it has nothing yet — answering "you are on the
+                    // wrong chain" is how a lagging endpoint gets mistaken for a forked one.
+                    // Below our history: unverifiable, serve as-is (the `⊥`-anchor case).
                     if let Some(tip) = self.tip_number()
                         && parent_pos > tip
+                        && self.faults.fork_signal_above_tip
                     {
                         self.stats.fork_signals += 1;
                         return Reply::Fork(self.hints_ending_at(tip));
                     }
-                    // Below our history: unverifiable, serve as-is (the `⊥`-anchor case).
                 }
             }
         }
@@ -570,6 +582,38 @@ mod tests {
         );
     }
 
+    /// RP-5b confines the fork signal to `from == tip + 1`; above it the honest answer is "nothing
+    /// yet". Answering "wrong chain" makes a source that is *behind* look like one that
+    /// *disagrees* — the distinction every multi-source script rests on, so the default is pinned
+    /// here too, not just the fault.
+    #[tokio::test]
+    async fn answers_no_data_above_its_tip_unless_the_fault_is_injected() {
+        let sim = sim().await;
+        sim.produce(DS, 5);
+        let url = format!("{}/stream", sim.base_url(DS));
+        let req = json!({"fromBlock": START + 10, "parentBlockHash": "0xnot-our-block"});
+
+        let res = reqwest::Client::new().post(&url).json(&req).send().await.unwrap();
+        assert_eq!(
+            res.status(),
+            StatusCode::NO_CONTENT,
+            "a source below the asked position is behind, not disagreeing"
+        );
+        assert_eq!(sim.stats(DS).fork_signals, 0);
+
+        sim.inject_fault(DS, |f| f.fork_signal_above_tip = true);
+
+        let res = reqwest::Client::new().post(&url).json(&req).send().await.unwrap();
+        assert_eq!(res.status(), StatusCode::CONFLICT);
+        let body: serde_json::Value = res.json().await.unwrap();
+        assert_eq!(
+            body["previousBlocks"].as_array().unwrap().last().unwrap()["number"],
+            START + 4,
+            "the fault reports a divergence at a tip it never reached past"
+        );
+        assert_eq!(sim.stats(DS).fork_signals, 1);
+    }
+
     #[tokio::test]
     async fn validates_parent_hash_across_a_sparse_numbering_hole() {
         let sim = SourceSim::start(SimConfig {