test: isolate flakes; fewer store opens; cheaper coverage

.github/workflows/ci.yml

@@ -93,13 +93,16 @@ jobs:
           workspaces: ". -> target/dev"
       - name: multi-node IBD (tier A)
         # One TESTNAME per `cargo test` invocation (second free arg is an error on
-        # rustc 1.95+). Run the two tier-A cases as separate commands.
+        # rustc 1.95+). Reconstruct / dead-peer are `#[ignore]` in the default
+        # suite so llvm-cov does not re-pay them; pass `--ignored` here.
         run: |
           set -euo pipefail
           cargo test -p rbitcoin-test --test integration_multinode \
             two_node_header_and_block_sync -- --nocapture
           cargo test -p rbitcoin-test --test integration_multinode \
-            serve_after_restart_via_reconstruct -- --nocapture
+            serve_after_restart_via_reconstruct -- --ignored --nocapture
+          cargo test -p rbitcoin-test --test integration_multinode \
+            ibd_skips_dead_peer -- --ignored --nocapture
   # Line-coverage gate (≥90% first-party LCOV LH/LF). Slow; waits for the
   # fast gates so a red fmt/clippy/test does not start this job. Required.
   coverage:

TESTING.md

@@ -11,6 +11,17 @@
 
 **Fewer scenario functions / store opens, not less coverage** — put more asserts on one carefully designed multi-stage journey.
 
+### Parallel cargo test (same binary)
+
+`cargo test` / `cargo llvm-cov test` run **one process per test binary**. Do not:
+
+- Put HOLD / wait hooks in a shipped function other tests also call (`confirm_scripts_phase`).
+- Assert process-global last-writer meters (`confirm_phase_stats`, `confirm_thr_stats::sample_and_reset`, `last_union_miss` / `last_plan_batch`) as the contract. Use pin/layout, error strings, or a pure formatter / local `AtomicU64`.
+- `std::env::set_var` without the crate lock (or pass the knob as an argument).
+- Bind a fixed port (use `:0`) or share a `/tmp` path (use `TestDatadir` / pid+nanos+seq).
+
+Do **not** “fix” flakes with `RUST_TEST_THREADS=1`.
+
 Shared helpers live in the `rbitcoin-test` crate (`mine`, `chain_fixture`).
 
 ### Third-party deps and compile cost (2026-08)
@@ -60,8 +71,8 @@ Override coverage dir: `CARGO_TARGET_DIR_COV=… ./scripts/coverage.sh`.
 
 | Tier | Command | Contents |
 |------|---------|----------|
-| **Default** (CI / human local full suite) | `cargo test --workspace` | Crate unit tests + scenarios + electrum + consensus_rules + **tier A multi-node IBD** (8-block single-hop + cold reconstruct) + reorg + short IBD error-path smokes. Agents use targeted `-p` tests locally; this suite runs on the PR. |
-| **CI multinode job** | same as tier A filters | Required job after fmt/clippy/test (coverage cadence) |
+| **Default** (CI / human local full suite) | `cargo test --workspace` | Crate unit tests + scenarios + electrum + consensus_rules + **8-block** `two_node` IBD + hub reorgs. Reconstruct / dead-peer are the **multinode** job. Agents use targeted `-p` tests locally; this suite runs on the PR. |
+| **CI multinode job** | named filters + `--ignored` job-only cases | 8-block IBD, reconstruct, slim dead-peer |
 | **Heavy multi-node / IBD** | `./scripts/integration.sh` or `-- --ignored` on `integration_multinode` / `ibd_smoke` | Multi-hop, tip-follow, 48-block dual seeder, mesh, `run_p2p` |
 
 ### Suite speed budgets (default tier)
@@ -91,7 +102,7 @@ Override coverage dir: `CARGO_TARGET_DIR_COV=… ./scripts/coverage.sh`.
 | Remining 100-block maturity pads with `confirm_wire_run` | `pad_empty_from` / `build_mature_regtest_with_spend` once per store |
 | Wall-time multi-round microbenches in default suite | Deterministic structure / chunk-load asserts; demote wall arms to `#[ignore]` |
 
-**Tier A timeouts:** `two_node_header_and_block_sync` 60s wall; `serve_after_restart_via_reconstruct` 90s wall. Confirm pipeline queue depths use saturating counters so teardown races cannot panic on overflow. Heavier paths remain `#[ignore]` (`scripts/integration.sh`).
+**Tier A timeouts:** `two_node_header_and_block_sync` 60s wall (default + job). Reconstruct / dead-peer are **multinode job only** (`#[ignore]`; job passes `--ignored`). `coverage.sh` also `--skip`s those names plus `two_node`. Heavier topology stays `#[ignore]` (`scripts/integration.sh`).
 
 **Speed / reliability (default suite):** prefer `pad_empty_from` / `build_mature_regtest_with_spend` over remine pads; SH run-builder sleeps are 1 ms under `cfg(test)` (40 ms in production). `pin_compose_multi_pack_timed` keeps functional + layout/covered short-circuit gates (multi-ms floor); sticky vs cold assemble is log-only (not a hard timing assert). Schema-13 wire rebuild must stamp create identity from `txid.body` — zero batch identity is treated as missing (regression covered by `reconstruct_and_connect_error_arms` + multi-vout confirm scenarios). Coverage vs speed: prefer **one** scenario at the real entry over N micro-opens that only paint lines; when adding coverage for reduce/materialize, use a **tiny** target, not production stream depth.
 
@@ -195,11 +206,12 @@ Prefer **one high-level scenario** per behavior cluster. Delete lower-level test
 | `scripthash_index_history_balance_and_reorg` | Query | Electrum index + reorg spend clear |
 | `electrum_server_version_history_balance` | Electrum | Protocol fixture: version, history, balance, headers |
 | `electrum_more_methods_and_errors` | Electrum | ping/features/block headers/listunspent/tx get+merkle/fees + error paths |
-| `two_node_header_and_block_sync` | P2P (**default / multinode CI**) | Seeder → peer 8-block IBD |
-| `serve_after_restart_via_reconstruct` | P2P (**default / multinode CI**) | Cold serve via reconstruct |
+| `two_node_header_and_block_sync` | P2P (**default + multinode CI**) | Seeder → peer 8-block IBD. **Not** re-run under `coverage.sh`. |
+| `serve_after_restart_via_reconstruct` | P2P (**multinode job only**) | Cold serve via reconstruct |
+| `ibd_skips_dead_peer` | P2P (**multinode job only**) | Live seeder + `127.0.0.1:1` |
 | `reorg_to_longer_branch` | P2P/chain (default) | Most-work reorg (hub only — no IBD hang risk) |
 | `three_node_relay_path` | P2P (**ignored**) | Hop serve — `scripts/integration.sh` |
-| `ibd_skips_dead_peer` | P2P (**ignored**) | Dial book skips dead address |
+
 | `ibd_two_peers` | P2P (**ignored**) | Dual-seeder 48-block IBD |
 | `tip_follow_after_ibd` / `tip_follow_getheaders_*` / `ibd_to_tip_tracking_*` | P2P (**ignored**) | Tip follow / relay |
 | `node_run_p2p_short` | Node (**ignored**) | Full `run_p2p` entry |
@@ -211,8 +223,8 @@ Removed (covered by the rows above): `confirm_cross_block_prevout_without_tx_hea
 
 ### Integration / multi-node
 
-Default CI + required **multinode** job run tier A `integration_multinode` cases (not `--ignored` heavies).
-Heavy topology is `#[ignore]` and run periodically:
+Default `cargo test` runs `two_node_header_and_block_sync` (8-block). The required **multinode** job also runs reconstruct and slim dead-peer (`--ignored` filters in `ci.yml`).
+Heavy topology (3-hop, 48-block, mesh, `run_p2p`) stays `#[ignore]` for `scripts/integration.sh`:
 
 ```bash
 ./scripts/integration.sh   # default multinode + --ignored

crates/rbitcoin-consensus/src/confirm_run/mod.rs

@@ -61,7 +61,8 @@ use phases::{assemble_run, script_wave};
 #[cfg(test)]
 use phases::{check_bip34, expected_bits_extending, post_commit};
 use pin::{ensure_spend_abs_layouts, pin_for_wire_batch};
-pub use scripts::scripts_feed_test_sync;
+#[cfg(test)]
+pub use scripts::scripts_stage_from_load_channel_with;
 pub use scripts::{
     confirm_scripts_feed_ahead, confirm_scripts_phase, confirm_scripts_phase_async,
     join_scripts_polling, scripts_stage_from_load_channel, ScriptsBatchMeta, ScriptsPhaseHandle,

crates/rbitcoin-consensus/src/confirm_run/scripts.rs

@@ -5,7 +5,6 @@ use super::*;
 pub fn confirm_scripts_phase(
     mut batch: LoadedBatch,
 ) -> Result<ConfirmScriptOutcome, ConsensusError> {
-    scripts_feed_test_sync::on_phase_enter();
     let t_work = Instant::now();
     script_wave(&batch.prepared, &batch.script_preverified)?;
     for p in &mut batch.prepared {
@@ -51,6 +50,23 @@ impl ScriptsPhaseHandle {
         })
     }
 
+    /// Run `work` on a coordinator. Test-local hold lives in `work`, not in
+    /// [`confirm_scripts_phase`].
+    #[cfg(test)]
+    pub fn spawn_fn(
+        work: impl FnOnce() -> Result<ConfirmScriptOutcome, ConsensusError> + Send + 'static,
+    ) -> Self {
+        let (tx, rx) = std::sync::mpsc::sync_channel(1);
+        let phase_thread = std::sync::Arc::new(std::sync::Mutex::new(None));
+        let slot = std::sync::Arc::clone(&phase_thread);
+        crate::script_pool::spawn_coordinator(move || {
+            *slot.lock().unwrap_or_else(|p| p.into_inner()) =
+                Some(std::thread::current().name().unwrap_or("").to_string());
+            let _ = tx.send(work());
+        });
+        Self { rx, phase_thread }
+    }
+
     /// Join and return the coordinator thread name recorded for **this** handle.
     #[cfg(test)]
     pub fn join_with_phase_thread(self) -> Result<(ConfirmScriptOutcome, String), ConsensusError> {
@@ -81,7 +97,6 @@ impl ScriptsPhaseHandle {
 /// The OS scripts thread must keep claiming N+1 **while** waiting on N’s
 /// [`ScriptsPhaseHandle::recv_timeout`] (not only once before a blocking join).
 pub fn confirm_scripts_phase_async(batch: LoadedBatch) -> ScriptsPhaseHandle {
-    scripts_feed_test_sync::on_async_submit();
     let (tx, rx) = std::sync::mpsc::sync_channel(1);
     #[cfg(test)]
     let phase_thread = std::sync::Arc::new(std::sync::Mutex::new(None));
@@ -125,7 +140,6 @@ where
         match handle.recv_timeout(poll) {
             Ok(r) => return r,
             Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
-                scripts_feed_test_sync::on_recv_timeout();
                 continue;
             }
             Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
@@ -180,19 +194,34 @@ pub fn confirm_scripts_feed_ahead(
 /// `sync_channel(1)` timing.
 pub fn scripts_stage_from_load_channel(
     mat_rx: &std::sync::mpsc::Receiver<(LoadedBatch, u64)>,
+    on_ok: impl FnMut(ConfirmScriptOutcome, ScriptsBatchMeta) -> bool,
+    on_err: impl FnMut(ConsensusError, ScriptsBatchMeta) -> bool,
+    should_stop: impl FnMut() -> bool,
+) {
+    scripts_stage_from_load_channel_with(
+        mat_rx,
+        |batch, mat_ns| {
+            let meta = ScriptsBatchMeta::from_batch(&batch, mat_ns);
+            (confirm_scripts_phase_async(batch), meta)
+        },
+        on_ok,
+        on_err,
+        should_stop,
+    );
+}
+
+/// Same claim/feed-ahead loop as [`scripts_stage_from_load_channel`], with an
+/// injectable start (tests hold the first wave locally).
+pub fn scripts_stage_from_load_channel_with(
+    mat_rx: &std::sync::mpsc::Receiver<(LoadedBatch, u64)>,
+    mut start: impl FnMut(LoadedBatch, u64) -> (ScriptsPhaseHandle, ScriptsBatchMeta),
     mut on_ok: impl FnMut(ConfirmScriptOutcome, ScriptsBatchMeta) -> bool,
     mut on_err: impl FnMut(ConsensusError, ScriptsBatchMeta) -> bool,
     mut should_stop: impl FnMut() -> bool,
 ) {
     let mut current: Option<(ScriptsPhaseHandle, ScriptsBatchMeta)> = None;
     let mut lookahead: Option<(ScriptsPhaseHandle, ScriptsBatchMeta)> = None;
 
-    let start = |batch: LoadedBatch, mat_ns: u64| -> (ScriptsPhaseHandle, ScriptsBatchMeta) {
-        let meta = ScriptsBatchMeta::from_batch(&batch, mat_ns);
-        let handle = confirm_scripts_phase_async(batch);
-        (handle, meta)
-    };
-
     loop {
         if should_stop() {
             break;
@@ -275,81 +304,3 @@ impl ScriptsBatchMeta {
         }
     }
 }
-
-/// Test-only sync so unit tests can prove N+1 was submitted while N’s wave is still open.
-pub mod scripts_feed_test_sync {
-    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-    use std::sync::{Mutex, MutexGuard};
-    use std::time::{Duration, Instant};
-
-    static FEED_TEST_LOCK: Mutex<()> = Mutex::new(());
-    static SUBMIT_COUNT: AtomicU64 = AtomicU64::new(0);
-    static HOLD_FIRST: AtomicBool = AtomicBool::new(false);
-    static HOLD_TAIL: AtomicBool = AtomicBool::new(false);
-    static FIRST_ENTERED: AtomicBool = AtomicBool::new(false);
-    static RECV_TIMEOUTS: AtomicU64 = AtomicU64::new(0);
-
-    /// Hold across a feed-ahead timing test so a parallel `reset()` cannot
-    /// clear HOLD_* while another test is mid-wave (`cargo llvm-cov`).
-    pub fn lock() -> MutexGuard<'static, ()> {
-        FEED_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner())
-    }
-
-    /// Reset counters (call at start of each feed-ahead timing test).
-    pub fn reset() {
-        SUBMIT_COUNT.store(0, Ordering::SeqCst);
-        HOLD_FIRST.store(false, Ordering::SeqCst);
-        HOLD_TAIL.store(false, Ordering::SeqCst);
-        FIRST_ENTERED.store(false, Ordering::SeqCst);
-        RECV_TIMEOUTS.store(0, Ordering::SeqCst);
-    }
-
-    /// When true, the first [`super::confirm_scripts_phase`] waits until
-    /// [`submit_count`] ≥ 2 (second async submit happened mid-wave).
-    pub fn set_hold_first_until_second_submit(hold: bool) {
-        HOLD_FIRST.store(hold, Ordering::SeqCst);
-        FIRST_ENTERED.store(false, Ordering::SeqCst);
-    }
-
-    /// After N+1 is submitted, keep the first wave open ~200 ms so a 200 µs
-    /// join loop would accumulate hundreds of timeouts.
-    pub fn set_hold_tail_after_second(hold: bool) {
-        HOLD_TAIL.store(hold, Ordering::SeqCst);
-    }
-
-    pub fn submit_count() -> u64 {
-        SUBMIT_COUNT.load(Ordering::SeqCst)
-    }
-
-    pub fn recv_timeout_count() -> u64 {
-        RECV_TIMEOUTS.load(Ordering::SeqCst)
-    }
-
-    pub(super) fn on_recv_timeout() {
-        RECV_TIMEOUTS.fetch_add(1, Ordering::Relaxed);
-    }
-
-    pub(super) fn on_async_submit() {
-        SUBMIT_COUNT.fetch_add(1, Ordering::SeqCst);
-    }
-
-    pub(super) fn on_phase_enter() {
-        if !HOLD_FIRST.load(Ordering::SeqCst) {
-            return;
-        }
-        if FIRST_ENTERED.swap(true, Ordering::SeqCst) {
-            return;
-        }
-        let deadline = Instant::now() + Duration::from_secs(5);
-        while submit_count() < 2 {
-            if Instant::now() > deadline {
-                // Avoid hanging the suite if feed-ahead is broken.
-                break;
-            }
-            std::thread::sleep(Duration::from_millis(1));
-        }
-        if HOLD_TAIL.load(Ordering::SeqCst) {
-            std::thread::sleep(Duration::from_millis(200));
-        }
-    }
-}