Measure archive-write responsiveness and specify bounded writer guarantees

.github/workflows/archive-measurement.yml

@@ -0,0 +1,50 @@
+name: Synthetic archive measurement
+on:
+  workflow_dispatch:
+  workflow_call:
+    inputs:
+      runner_tests_only:
+        type: boolean
+        default: false
+permissions:
+  contents: read
+concurrency:
+  group: archive-measurement-${{ github.ref }}
+  cancel-in-progress: false
+jobs:
+  archive-measurement:
+    name: ${{ inputs.runner_tests_only && 'Runner regressions' || 'Archive measurement' }} ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: ${{ inputs.runner_tests_only && 5 || 45 }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-22.04, windows-2022]
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 # v7.1.0
+        with:
+          python-version: '3.12'
+      - if: ${{ !inputs.runner_tests_only }}
+        uses: dtolnay/rust-toolchain@b725eda525362e7cf34af9a7243cac1e3f01a7ec
+        with:
+          toolchain: 1.99.1
+      - name: Runner regressions
+        run: python -m unittest discover -s scripts/tests -p 'test_measure_archive*.py'
+      - name: Evidence detector regressions
+        if: ${{ !inputs.runner_tests_only }}
+        run: |
+          cargo test --locked -p scope-core --lib diagnostics::
+      - name: Measure fresh synthetic archives (debug and release separately)
+        if: ${{ !inputs.runner_tests_only }}
+        run: python scripts/measure-archive.py --profiles debug release --samples 3 --workers 0 2 --output-dir .runtime/archive-measurement
+      - name: Upload numeric synthetic diagnostics only
+        if: ${{ always() && !inputs.runner_tests_only }}
+        uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2
+        with:
+          name: archive-measurement-${{ matrix.os }}-${{ github.sha }}
+          path: .runtime/archive-measurement/measurement.json
+          if-no-files-found: error
+          retention-days: 14

.github/workflows/har-compare-action.yml

@@ -12,16 +12,24 @@ on:
       - 'tools/har-compare/**'
       - '.github/workflows/har-compare-action.yml'
   workflow_dispatch:
+    inputs:
+      diagnostic:
+        description: Explicit diagnostic (archive measurement never runs the HAR consumer)
+        type: choice
+        options: [har-composite, archive-measurement, runner-regressions]
+        required: false
+        default: har-composite
 
 permissions:
   contents: read
 
 concurrency:
-  group: har-composite-${{ github.ref }}
+  group: har-composite-${{ github.event_name }}-${{ inputs.diagnostic }}-${{ github.ref }}
   cancel-in-progress: true
 
 jobs:
   consumer:
+    if: github.event_name != 'workflow_dispatch' || (inputs.diagnostic != 'archive-measurement' && inputs.diagnostic != 'runner-regressions')
     name: Composite ${{ matrix.scenario }}
     runs-on: ubuntu-22.04
     timeout-minutes: 5
@@ -70,3 +78,18 @@ jobs:
           DEFAULT_OUTPUTS: ${{ toJSON(steps.default.outputs) }}
           DEFERRED_OUTPUTS: ${{ toJSON(steps.deferred.outputs) }}
         run: node '.scope action/tools/har-compare/action-test/integration-check.mjs' verify
+
+  archive-measurement:
+    # Registered manual entry point while the dedicated workflow is branch-only.
+    if: github.event_name == 'workflow_dispatch' && inputs.diagnostic == 'archive-measurement'
+    uses: ./.github/workflows/archive-measurement.yml
+    permissions:
+      contents: read
+
+  runner-regressions:
+    if: github.event_name == 'workflow_dispatch' && inputs.diagnostic == 'runner-regressions'
+    uses: ./.github/workflows/archive-measurement.yml
+    with:
+      runner_tests_only: true
+    permissions:
+      contents: read

crates/scope-core/src/diagnostics.rs

@@ -0,0 +1,166 @@
+//! Compiled only in the core unit-test binary, never in shipping applications.
+use super::*;
+use std::collections::BTreeSet;
+use std::time::{Duration, Instant};
+
+mod tests;
+
+const CLIENTS: usize = 32;
+const SAVE_CAP: usize = 256;
+
+#[derive(Default, Serialize)]
+struct SaveSample {
+    serialization_ns: u64,
+    mutex_wait_ns: u64,
+    statement_ns: u64,
+    serialized_bytes: usize,
+    success: bool,
+}
+#[derive(Default, Serialize)]
+pub(super) struct SaveSamples {
+    samples: Vec<SaveSample>,
+    dropped: usize,
+}
+#[derive(Default)]
+pub(super) struct SaveMetrics {
+    enabled: AtomicBool,
+    values: Mutex<SaveSamples>,
+}
+impl SaveMetrics {
+    pub(super) fn start(&self) -> Option<Instant> {
+        self.enabled.load(Ordering::Relaxed).then(Instant::now)
+    }
+    pub(super) fn record(
+        &self,
+        start: Option<Instant>,
+        serialized: Instant,
+        acquired: Instant,
+        finished: Instant,
+        bytes: usize,
+        success: bool,
+    ) {
+        if let Some(start) = start {
+            self.push(SaveSample {
+                serialization_ns: ns(serialized.duration_since(start)),
+                mutex_wait_ns: ns(acquired.duration_since(serialized)),
+                statement_ns: ns(finished.duration_since(acquired)),
+                serialized_bytes: bytes,
+                success,
+            });
+        }
+    }
+    fn push(&self, sample: SaveSample) {
+        // This lock is outside the DB guard; the harness reads only after the batch.
+        let mut values = self.values.lock().unwrap();
+        if values.samples.len() < SAVE_CAP {
+            values.samples.push(sample);
+        } else {
+            values.dropped += 1;
+        }
+    }
+    fn take(&self) -> SaveSamples {
+        std::mem::take(&mut *self.values.lock().unwrap())
+    }
+}
+fn ns(duration: Duration) -> u64 {
+    duration.as_nanos().min(u64::MAX as u128) as u64
+}
+
+fn valid_evidence(rows: &[Exchange], session: &str, rules: &[String; 3]) -> bool {
+    let mut ids = BTreeSet::new();
+    let mut indexes = BTreeSet::new();
+    rows.len() == CLIENTS
+        && rows.iter().all(|e| {
+            let index = url::Url::parse(&e.url).ok().and_then(|url| {
+                let values: Vec<_> = url
+                    .query_pairs()
+                    .filter(|(key, _)| key == "i")
+                    .map(|(_, v)| v.into_owned())
+                    .collect();
+                (values.len() == 1)
+                    .then(|| values[0].parse::<usize>().ok())
+                    .flatten()
+            });
+            let expected = [
+                ("rewriteRequest", "request"),
+                ("mockResponse", "request"),
+                ("rewriteResponse", "response"),
+            ];
+            ids.insert(&e.id)
+                && index.is_some_and(|i| i < CLIENTS && indexes.insert(i))
+                && e.session_id == session
+                && e.method == "GET"
+                && e.completed
+                && e.error.is_none()
+                && e.status == Some(201)
+                && e.provenance == TrafficProvenance::Mock
+                && e.request_body == "request rule"
+                && !e.request_body_base64
+                && e.response_body == "response rule"
+                && !e.response_body_base64
+                && e.request_headers.iter().any(|h| {
+                    h.name.eq_ignore_ascii_case("x-measure-index")
+                        && h.value.parse::<usize>().ok() == index
+                })
+                && e.original_request.as_ref().is_some_and(|v| {
+                    v["body"] == ""
+                        && v["method"] == "GET"
+                        && v["url"] == e.url
+                        && v["headers"].as_array().is_some_and(|headers| {
+                            headers.iter().any(|h| {
+                                h["name"] == "x-measure-index"
+                                    && h["value"].as_str().and_then(|s| s.parse::<usize>().ok())
+                                        == index
+                            })
+                        })
+                })
+                && e.original_response
+                    .as_ref()
+                    .is_some_and(|v| v["body"] == "mock body" && v["status"] == 201)
+                && e.applied_rules.len() == 3
+                && e.applied_rules.iter().enumerate().all(|(n, r)| {
+                    r.rule_id == rules[n]
+                        && r.name == format!("Rule {}", n + 1)
+                        && r.action == expected[n].0
+                        && r.stage == expected[n].1
+                })
+        })
+}
+
+mod probes;
+mod workload;
+
+// Explicitly ignored: manually launched by scripts/measure-archive.py, never a CI timing gate.
+#[test]
+#[ignore = "opt-in bounded synthetic measurement"]
+fn archive_responsiveness() {
+    let workers: usize = std::env::var("SCOPE_MEASURE_WORKERS")
+        .unwrap()
+        .parse()
+        .unwrap();
+    assert!(workers == 0 || workers == 2, "unsupported runtime");
+    let samples: usize = std::env::var("SCOPE_MEASURE_SAMPLES")
+        .unwrap()
+        .parse()
+        .unwrap();
+    assert!((1..=5).contains(&samples), "unsupported sample count");
+    let mut builder = if workers == 0 {
+        tokio::runtime::Builder::new_current_thread()
+    } else {
+        let mut b = tokio::runtime::Builder::new_multi_thread();
+        b.worker_threads(workers);
+        b
+    };
+    let runtime = builder
+        .enable_all()
+        .max_blocking_threads(2)
+        .build()
+        .unwrap();
+    let report = runtime.block_on(workload::run(samples, workers));
+    let output = std::env::var_os("SCOPE_MEASURE_OUTPUT").expect("output required");
+    std::fs::write(output, serde_json::to_vec_pretty(&report).unwrap()).unwrap();
+    assert_eq!(
+        report["status"], "PASS",
+        "synthetic measurement/evidence/control failure; inspect numeric report"
+    );
+}