bench(#190): declare the history corpus reading as a phase, and measure the corpus axis

core/benchmark/fs-bench-pro-storage-content/src/ops/history.rs

@@ -65,7 +65,7 @@ use crate::registry::Case;
 use crate::support::{instruments, phases};
 use super::c1;
 use crate::support::trace::Kind;
-use crate::workload::history::{Change, Corpus, HistoryError, Row};
+use crate::workload::history::{read_probe, read_probe_begin, Change, Corpus, HistoryError, Row};
 use crate::workload::providers::TreeStore;
 
 /// The root directory's serial. `build_filesystem` requires it to be one of the
@@ -1537,6 +1537,15 @@ fn perf(_case: &Case, row: Row, context: &mut OpContext<'_>) -> Result<OpOutcome
         ));
     }
 
+    // **The corpus-axis diagnostic.** Residency of the corpus files *before this
+    // invocation first reads them*, and the process's device reads across the
+    // chain. Both are diagnostics: nothing fails on them, nothing is de-warmed, and
+    // neither is a cold claim. They answer the one question the declared cache
+    // stance left open — whether the input pages were already resident when the
+    // operation started — by measuring it rather than assuming it.
+    read_probe_begin();
+    let disk_reads_before = instruments::process_usage();
+
     // **One root, N named children.** The corpus reading sits inside the root and
     // outside every child, so the row's `operation_ns` is the sum of the children
     // and not the root — owner ruling 2. The root is the whole chain including the
@@ -1581,7 +1590,13 @@ fn perf(_case: &Case, row: Row, context: &mut OpContext<'_>) -> Result<OpOutcome
             } else {
                 BTreeMap::new()
             };
-            corpus_read_ns = corpus_read_ns.saturating_add(t_corpus.elapsed().as_nanos() as u64);
+            let corpus_span_ns = t_corpus.elapsed().as_nanos() as u64;
+            corpus_read_ns = corpus_read_ns.saturating_add(corpus_span_ns);
+            // Declared, not left between the phases. The span is inside the
+            // invocation and outside every measured child, so preparation is the
+            // phase that accounts for it, and `history.corpus_read_ns` publishes it
+            // in its own right so a reader can subtract it.
+            phases::add_preparation(corpus_span_ns);
             let ordinal = transition.state.ordinal;
             let mut peak_heap = 0u64;
             let outcome = root
@@ -2023,6 +2038,15 @@ fn perf(_case: &Case, row: Row, context: &mut OpContext<'_>) -> Result<OpOutcome
         Ok(value) => value,
         Err(error) => return Ok(unmeasured(&error, gates)),
     };
+    // The device half of the corpus diagnostic, from the same `rusage` instrument
+    // the cold contract uses. Process-wide, so corpus reads and the Store's own
+    // device traffic are one number here; it is not a gate.
+    let disk_read_bytes_chain = match (disk_reads_before, instruments::process_usage()) {
+        (Some(before), Some(after)) => {
+            Some(after.disk_read_bytes.saturating_sub(before.disk_read_bytes))
+        }
+        _ => None,
+    };
     drop(store);
 
     // The product's own timing tree, byte-verbatim, written **once** for the whole
@@ -2200,8 +2224,46 @@ fn perf(_case: &Case, row: Row, context: &mut OpContext<'_>) -> Result<OpOutcome
         "history.corpus_read_ns",
         corpus_read_ns as i128,
         "ns",
-        "the harness's own corpus reading: inside the root, between the children, untimed",
+        "the harness's own corpus reading: inside the root, between the children, untimed, and attributed to the preparation phase so the declared phases account for the invocation",
     )?;
+    let probe = read_probe();
+    for (key, value, unit, basis) in [
+        (
+            "history.corpus.probe.files",
+            probe.files,
+            "files",
+            "distinct corpus files the chain read, probed for residency before their first read",
+        ),
+        (
+            "history.corpus.probe.bytes",
+            probe.length_bytes,
+            "bytes",
+            "the summed lengths of those files (diagnostic, not a gate)",
+        ),
+        (
+            "history.corpus.probe.pages",
+            probe.total_pages,
+            "pages",
+            "pages examined across those files (diagnostic, not a gate)",
+        ),
+        (
+            "history.corpus.probe.resident_pages",
+            probe.resident_pages,
+            "pages",
+            "pages resident before the first read: what an earlier run may have left, measured rather than assumed; nothing is de-warmed and this is not a cold claim",
+        ),
+    ] {
+        context.trace.write_number(Kind::Resource, key, value as i128, unit, basis)?;
+    }
+    if let Some(bytes) = disk_read_bytes_chain {
+        context.trace.write_number(
+            Kind::Resource,
+            "history.operation.disk_read_bytes",
+            bytes as i128,
+            "bytes",
+            "device bytes this process read across the chain, from rusage; process-wide, so corpus and Store reads are one number, and not a gate",
+        )?;
+    }
     context.trace.write_number(
         Kind::Counter,
         "history.children_ns",
@@ -2308,7 +2370,9 @@ fn perf(_case: &Case, row: Row, context: &mut OpContext<'_>) -> Result<OpOutcome
             format!("history_path_states: {}", pins.path_states),
             "measured_region: construct + build/update + save, one named child per state".to_string(),
             "operation_ns: the sum of the named children, not the root".to_string(),
-            "corpus_reading: between the children, untimed".to_string(),
+            "corpus_reading: between the children, untimed, attributed to the preparation phase"
+                .to_string(),
+            "cache_diagnostic: corpus residency before the first read and process device reads; diagnostics, not gates, and not a cold claim".to_string(),
             "prepared: none (Preparation::InProcess)".to_string(),
             // The runner reads this to decide whether the row has a deferred
             // oracle. Without it the verify invocation is never scheduled and a

core/benchmark/fs-bench-pro-storage-content/src/support/phases.rs

@@ -35,6 +35,12 @@ use std::time::Instant;
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
 pub struct Phases {
     /// Setup: build the fixture, or copy and de-warm a prepared one.
+    ///
+    /// A driver whose input assembly happens **between** its measured children
+    /// attributes those spans with [`add_preparation`], because they are real
+    /// invocation time that no other phase accounts for. For such a driver this
+    /// field is the close-of-preparation mark **plus** those spans, so it is not
+    /// contiguous time; the driver publishes the attributed part separately.
     pub preparation_ns: u64,
     /// The per-sample copy and de-warm inside preparation, published separately
     /// because owner decision D2 makes the acquisition cost mandatory.
@@ -110,6 +116,8 @@ struct Clock {
     measured: Option<Instant>,
     verified: Option<Instant>,
     acquisition_ns: u64,
+    /// Harness input assembly attributed to preparation by [`add_preparation`].
+    preparation_added_ns: u64,
     operation_ns: u64,
     timing_json_bytes: u64,
     /// CPU at the close of preparation, and at the close of the measured region.
@@ -140,6 +148,7 @@ fn clock() -> MutexGuard<'static, Clock> {
                 measured: None,
                 verified: None,
                 acquisition_ns: 0,
+                preparation_added_ns: 0,
                 operation_ns: 0,
                 timing_json_bytes: 0,
                 cpu_at_prepared: None,
@@ -163,6 +172,7 @@ pub fn begin(output: &Path) {
     held.measured = None;
     held.verified = None;
     held.acquisition_ns = 0;
+    held.preparation_added_ns = 0;
     held.operation_ns = 0;
     held.timing_json_bytes = 0;
     held.cpu_at_prepared = None;
@@ -252,6 +262,26 @@ pub fn add_acquisition(nanoseconds: u64) {
     clock().acquisition_ns += nanoseconds;
 }
 
+/// Attributes a span of harness **input assembly** to the preparation phase.
+///
+/// A driver whose input assembly happens *between* its measured children has real
+/// invocation time that no declared phase accounts for, and `runner.py`'s
+/// reconciliation fails closed on exactly that (`shared/phases.py`
+/// `reconcile_invocation`: the wall must be covered by the declared phases within
+/// its tolerance). `history.*` reads its corpus between the children, so it
+/// declares those spans here, and the four phases then account for the invocation
+/// instead of leaving a sixth of it unexplained.
+///
+/// The span is outside every measured child by construction, so it cannot be
+/// double-counted, and the same driver publishes it in its own right
+/// (`history.corpus_read_ns`) so a reader can subtract it rather than trust the
+/// attribution. This is a declaration of measured time, not a relaxation: it makes
+/// the budgeted complete command **larger**, never smaller.
+pub fn add_preparation(nanoseconds: u64) {
+    let mut held = clock();
+    held.preparation_added_ns = held.preparation_added_ns.saturating_add(nanoseconds);
+}
+
 /// Runs one piece of harness work, accounting it when the measured region is open.
 ///
 /// This is the per-object handoff: the harness has to hand the product an owned
@@ -317,7 +347,7 @@ pub fn snapshot() -> Phases {
         _ => None,
     };
     Phases {
-        preparation_ns: since_start(held.prepared),
+        preparation_ns: since_start(held.prepared).saturating_add(held.preparation_added_ns),
         acquisition_ns: held.acquisition_ns,
         operation_ns: held.operation_ns,
         verification_ns,

core/benchmark/fs-bench-pro-storage-content/src/workload/history.rs

@@ -62,6 +62,9 @@
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::fmt;
 use std::path::{Path, PathBuf};
+use std::sync::{Mutex, OnceLock};
+
+use crate::support::instruments;
 
 use super::digest::{hex, sha256, Sha256};
 use super::gitoid::{blob_oid, hex_oid, parse_oid};
@@ -1218,7 +1221,85 @@ fn verify_blob(
     Ok(())
 }
 
+/// Residency of the corpus files **before this invocation first read them**.
+///
+/// The corpus axis of the cache stance: the corpus is immutable and identity-pinned
+/// (manifest and tip are checked at open), so whether its pages were already
+/// resident is a fact about the input, never about correctness. This measures that
+/// fact instead of assuming it — `mincore` over a fresh mapping, taken *before* the
+/// first read of each distinct path, so the reading cannot be polluted by the read
+/// it precedes.
+///
+/// It is a **diagnostic**. Nothing fails on it, nothing is de-warmed, and it is not
+/// a cold claim: a resident page here is the previous run's leavings, reported
+/// rather than removed.
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+pub struct ReadProbe {
+    /// Distinct files probed before their first read in this invocation.
+    pub files: u64,
+    /// Summed lengths of those files.
+    pub length_bytes: u64,
+    /// Pages examined across them.
+    pub total_pages: u64,
+    /// Pages the kernel reported resident before the first read.
+    pub resident_pages: u64,
+}
+
+static READ_PROBE: OnceLock<Mutex<ReadProbe>> = OnceLock::new();
+static READ_PROBE_PATHS: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
+
+fn read_probe_state() -> &'static Mutex<ReadProbe> {
+    READ_PROBE.get_or_init(|| Mutex::new(ReadProbe::default()))
+}
+
+fn read_probe_paths() -> &'static Mutex<HashSet<PathBuf>> {
+    READ_PROBE_PATHS.get_or_init(|| Mutex::new(HashSet::new()))
+}
+
+/// Clears the probe, so the next reading covers the reads that follow it.
+pub fn read_probe_begin() {
+    *read_probe_state()
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner()) = ReadProbe::default();
+    read_probe_paths()
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .clear();
+}
+
+/// The probe's reading so far.
+pub fn read_probe() -> ReadProbe {
+    *read_probe_state()
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+/// Reads one path's residency, once per distinct path, before the read itself.
+fn probe_read(path: &Path) {
+    {
+        let mut seen = read_probe_paths()
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if !seen.insert(path.to_path_buf()) {
+            return;
+        }
+    }
+    // An unreadable or unmappable file is not probed and not counted: a diagnostic
+    // that cannot see a file must say nothing about it rather than report zero.
+    let Ok(reading) = instruments::residency(path) else {
+        return;
+    };
+    let mut held = read_probe_state()
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    held.files += 1;
+    held.length_bytes = held.length_bytes.saturating_add(reading.length_bytes);
+    held.total_pages = held.total_pages.saturating_add(reading.total_pages);
+    held.resident_pages = held.resident_pages.saturating_add(reading.resident_pages);
+}
+
 fn read(path: &Path) -> Result<Vec<u8>, HistoryError> {
+    probe_read(path);
     std::fs::read(path).map_err(|error| HistoryError::io(path, error))
 }