From 2c22dd365b6e7481e6fb62b040f3b563546dcbab Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 28 Sep 2026 02:01:25 +0000 Subject: [PATCH 1/2] test: validate resource evidence and reject malformed CPU samples --- control_plane/tests/resource_measurements.rs | 40 ++++++++++++++ .../resource-measurement-validation.md | 19 +++++++ tools/empirical-bench/run.py | 14 +++-- .../test_resource_measurements.py | 52 +++++++++++++++++++ 4 files changed, 121 insertions(+), 4 deletions(-) create mode 100644 control_plane/tests/resource_measurements.rs create mode 100644 docs/design_docs/resource-measurement-validation.md create mode 100644 tools/empirical-bench/test_resource_measurements.py diff --git a/control_plane/tests/resource_measurements.rs b/control_plane/tests/resource_measurements.rs new file mode 100644 index 000000000..d887b3fc3 --- /dev/null +++ b/control_plane/tests/resource_measurements.rs @@ -0,0 +1,40 @@ +//! Work multiplicities and units, using explicit synthetic resource coefficients. +#[path = "support/erp_contract.rs"] +mod fixture; +use control_plane::physical::erp::{ErpAccuracyMode, ErpParameterDecision}; +use planner_types::post_asap::{SketchAlgorithm, SketchParams}; + +/// CPU seconds and byte-seconds scale independently; error statistics are unchanged. +#[test] +fn erp_resource_composition_uses_workload_multiplicities() { + for (updates, reads, merges, retention) in [(1000., 100., 0., 60.), (2000., 400., 5., 120.)] { + let mut policy = fixture::policy(ErpAccuracyMode::Empirical); + policy.expected_updates = updates; + policy.expected_queries = reads; + policy.expected_merges = merges; + policy.retention_seconds = retention; + let resources = &policy.artifact.records[0].resources; + let expected = policy.cpu_weight + * (updates * resources.update_cpu_seconds + + reads * resources.query_cpu_seconds + + merges * resources.merge_cpu_seconds) + + policy.byte_second_weight * resources.memory_bytes * retention; + let ErpParameterDecision::Empirical { + estimated_cost, + record_id, + .. + } = policy.select( + SketchAlgorithm::Cms, + 0.01, + SketchParams::Cms { + width: 4096, + depth: 5, + }, + ) + else { + panic!("applicable ERP must be priced") + }; + assert!((estimated_cost - expected).abs() < 1e-12); + assert_eq!(record_id, policy.artifact.records[0].id); + } +} diff --git a/docs/design_docs/resource-measurement-validation.md b/docs/design_docs/resource-measurement-validation.md new file mode 100644 index 000000000..a226bea8f --- /dev/null +++ b/docs/design_docs/resource-measurement-validation.md @@ -0,0 +1,19 @@ +# Resource measurement validation + +This PR validates the existing sketch-bench export path and ERP workload +composition. The Python tests distinguish process CPU from wall time, preserve +paired repetitions, reject missing/invalid inputs and normalize milliseconds to +nanoseconds per operation. The Rust test checks update/read/merge multiplicities +and retained byte-seconds while retaining the ERP record identity. + +The regression first failed because `cpu_batch` silently truncated unmatched +user/system samples and accepted a negative raw component when their sum was +positive. `cpu_per_op` could also turn infinite work into a zero price. Both now +reject these inputs. Missing CPU remains unknown, not zero. + +The tests use synthetic coefficients to validate arithmetic. They do not claim +those coefficients are real measurements. Real resource evidence must retain raw +paired samples, implementation/build/machine identity and measurement scope. +The existing `tools/empirical-bench/run.py` records real benchmark provenance; +its offline microbenchmarks do not independently establish deployment selection +quality. Shared-work costing tests remain in `physical/workload_cost.rs`. diff --git a/tools/empirical-bench/run.py b/tools/empirical-bench/run.py index cd56d756c..fbf9b3c0c 100644 --- a/tools/empirical-bench/run.py +++ b/tools/empirical-bench/run.py @@ -48,9 +48,11 @@ def cpu_per_op(row, op): raise ValueError("unaligned CPU/rate/time samples") values = [] for r, elapsed, user, system in zip(*samples): + if any(not math.isfinite(x) or x < 0 for x in (r, elapsed, user, system)): + raise ValueError("invalid raw CPU/rate/time measurement") work = r * elapsed / 1000 - if work <= 0: - raise ValueError("zero benchmark work") + if not math.isfinite(work) or work <= 0: + raise ValueError("invalid benchmark work") values.append((user + system) * 1_000_000 / work) result = measurement(values) result["method"] = "mean of paired run (process user+system CPU ns)/(throughput*wall seconds); per " + ("binary merge" if op == "merge" else "input item" if op == "insert" else "point-frequency key lookup") @@ -61,8 +63,12 @@ def cpu_batch(row, op): cpu = row.get(op + "_cpu_time_ms") if not cpu: return None - return measurement([(u + s) * 1_000_000 for u, s in - zip(cpu["user_ms"]["samples"], cpu["sys_ms"]["samples"])]) + user, system = cpu["user_ms"]["samples"], cpu["sys_ms"]["samples"] + if len(user) != len(system): + raise ValueError("unaligned user/system CPU samples") + if any(not math.isfinite(x) or x < 0 for x in user + system): + raise ValueError("invalid raw CPU measurement") + return measurement([(u + s) * 1_000_000 for u, s in zip(user, system)]) def export(raw, manifest, operation_reports, memory_rows, resource_rows=None): diff --git a/tools/empirical-bench/test_resource_measurements.py b/tools/empirical-bench/test_resource_measurements.py new file mode 100644 index 000000000..422171d20 --- /dev/null +++ b/tools/empirical-bench/test_resource_measurements.py @@ -0,0 +1,52 @@ +"""Resource normalization contracts; values are synthetic, never calibration data.""" +import math +import unittest +from run import cpu_batch, cpu_per_op + + +class ResourceMeasurementTests(unittest.TestCase): + def test_batch_rejects_unpaired_cpu_samples(self): + for user, system in [([1, 2], [3]), ([1], [2, 3]), ([], [1])]: + with self.subTest(user=user, system=system): + with self.assertRaises(ValueError): + cpu_batch({"query_cpu_time_ms": { + "user_ms": {"samples": user}, "sys_ms": {"samples": system}}}, "query") + + def test_batch_and_per_op_reject_invalid_raw_measurements(self): + for invalid in [-1, math.inf, math.nan]: + with self.subTest(invalid=invalid): + with self.assertRaises(ValueError): + cpu_batch({"query_cpu_time_ms": { + "user_ms": {"samples": [invalid]}, "sys_ms": {"samples": [10]}}}, "query") + for field in ["rate", "elapsed", "user", "system"]: + for invalid in [-1, math.inf, math.nan]: + values = dict(rate=100, elapsed=1000, user=1, system=1) + values[field] = invalid + row = {"query_throughput_items_per_sec": {"samples": [values["rate"]]}, + "query_wall_time_ms": {"samples": [values["elapsed"]]}, + "query_cpu_time_ms": {"user_ms": {"samples": [values["user"]]}, + "sys_ms": {"samples": [values["system"]]}}} + with self.subTest(field=field, invalid=invalid): + with self.assertRaises(ValueError): + cpu_per_op(row, "query") + + def test_cpu_conversion_preserves_repetitions_and_units(self): + row = {"query_cpu_time_ms": {"user_ms": {"samples": [2, 4]}, + "sys_ms": {"samples": [1, 2]}}} + batch = cpu_batch(row, "query") + self.assertEqual(batch["value"], 4_500_000) + self.assertEqual(batch["samples"], 2) + row.update(query_throughput_items_per_sec={"samples": [300, 300]}, + query_wall_time_ms={"samples": [1000, 2000]}) + per_op = cpu_per_op(row, "query") + self.assertEqual(per_op["value"], 10_000) + self.assertEqual(per_op["stddev"], 0) + self.assertEqual(per_op["samples"], 2) + + def test_missing_resource_dimension_stays_unknown(self): + self.assertIsNone(cpu_batch({}, "query")) + self.assertIsNone(cpu_per_op({}, "query")) + + +if __name__ == "__main__": + unittest.main() From bd1a5566af8f22cb507ad92a275b1d7e03f90229 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 28 Sep 2026 02:06:21 +0000 Subject: [PATCH 2/2] ci: run resource measurement regressions --- .github/workflows/mvp-ci.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/mvp-ci.yml b/.github/workflows/mvp-ci.yml index 925b83a98..422aa65e6 100644 --- a/.github/workflows/mvp-ci.yml +++ b/.github/workflows/mvp-ci.yml @@ -55,6 +55,10 @@ jobs: python3 -m unittest discover -s scripts/ci -p 'test_*.py' node scripts/ci/test_planner_merge.js + - name: Test resource measurement normalization + working-directory: ASAPQuery-backend + run: python3 -m unittest discover -s tools/empirical-bench -p 'test_*.py' + - name: Check workspace formatting working-directory: ASAPQuery-backend # `--all` also follows this workspace's sibling path dependencies.