Rust #325: verify small-host history order on the server clock alone

benchmarks/server-port/rc1/README.md

@@ -8,6 +8,11 @@ workload, limits, probe, SDK lock and the PHP reference (published Server
 2.6.2, the recorded baseline) are unchanged. Its SHA-256 is the
 `profile_sha256` in every summary.
 
+This profile is frozen with those summaries. It still pins the probe as it was
+on October 30, before #933 moved history verification to the server clock, so
+re-run it from commit `a0f7af002a6af96b219125c54d1ceb1a2be165e5`. On later
+commits the runner refuses it because the probe's hash differs.
+
 The [October 30 observations](observations-2026-10-30.md) are three
 alternating PHP/native-image pairs on one host, run with
 `scripts/perf/run-server-port-matched.sh` and summarized with

benchmarks/server-port/small-host/README.md

@@ -59,9 +59,19 @@ capacity claim. The 4 vCPU comparison host instead runs the matched profile
   the payloads are synthetic and the network is source-restricted. The
   firewall must allow 8080 only from the driver's private address, and the
   private run receipt records that rule.
-- Clock: latency compares the driver's start time with the server's
-  `closed_at`, so both hosts must report `NTPSynchronized=yes`. Each run
-  records the NTP state; check it before trusting a run.
+- Clock: latency and the first history stage
+  (`client_start_to_workflow_started`) compare the driver's clock with the
+  server's, so both hosts must report `NTPSynchronized=yes`. Synchronized is
+  not enough on its own: a 1 vCPU kvm-clock plan measured +39 ms offset and
+  62 ms jitter while synchronized. Each run records offset and jitter, and the
+  summary's `clock.cross_host_timing_reliable` is true only when every start
+  and end capture on both hosts reports both an offset and a jitter, and each
+  host's offset plus jitter is within 5 ms. A missing number makes it false.
+  When it is false, read capacity, resource
+  use and the server-clock history stages, but not latency or the first stage.
+  History order is verified on the server's clock alone, so an offset never
+  fails a probe; each probe's `cross_clock` reports the first stage's minimum
+  and how many were negative.
 - Software on both hosts: Docker Engine with the Compose plugin, `git`,
   `jq` and `curl`, plus `python3` on whichever host summarizes. Check out the
   same Server commit on both hosts, the one that merges this profile.

benchmarks/server-port/small-host/matched-4c.json

@@ -76,7 +76,7 @@
     "adapter": "benchmarks/capacity/v1/bindings/php/capacity_adapter.php",
     "adapter_sha256": "0ad9fd714f325970c18d32cfedf754895cec621bd048593ac04fd8b9a247b926",
     "command": "scripts/perf/standard-workflow-offered.php",
-    "command_sha256": "cac36195aa9d0867516dff7bcb09f7d9df5fc65b9c0a0b7f49f8d7852def055f",
+    "command_sha256": "c0352bbab781df2d3f86f7c96e7b749606e62c19770bdbe85e6c75ef25c58e66",
     "warmup_seconds": 60,
     "offer_seconds": 120,
     "drain_limit_seconds": 120,

benchmarks/server-port/small-host/profile.json

@@ -51,7 +51,7 @@
     "adapter": "benchmarks/capacity/v1/bindings/php/capacity_adapter.php",
     "adapter_sha256": "0ad9fd714f325970c18d32cfedf754895cec621bd048593ac04fd8b9a247b926",
     "command": "scripts/perf/standard-workflow-offered.php",
-    "command_sha256": "cac36195aa9d0867516dff7bcb09f7d9df5fc65b9c0a0b7f49f8d7852def055f"
+    "command_sha256": "c0352bbab781df2d3f86f7c96e7b749606e62c19770bdbe85e6c75ef25c58e66"
   },
   "ladder": {
     "warmup_seconds": 30,

benchmarks/server-port/v2/matched.json

@@ -64,7 +64,7 @@
     "adapter": "benchmarks/capacity/v1/bindings/php/capacity_adapter.php",
     "adapter_sha256": "0ad9fd714f325970c18d32cfedf754895cec621bd048593ac04fd8b9a247b926",
     "command": "scripts/perf/standard-workflow-offered.php",
-    "command_sha256": "cac36195aa9d0867516dff7bcb09f7d9df5fc65b9c0a0b7f49f8d7852def055f",
+    "command_sha256": "c0352bbab781df2d3f86f7c96e7b749606e62c19770bdbe85e6c75ef25c58e66",
     "warmup_seconds": 60,
     "offer_seconds": 120,
     "drain_limit_seconds": 120,

benchmarks/server-port/v3/matched.json

@@ -83,7 +83,7 @@
     "adapter": "benchmarks/capacity/v1/bindings/php/capacity_adapter.php",
     "adapter_sha256": "0ad9fd714f325970c18d32cfedf754895cec621bd048593ac04fd8b9a247b926",
     "command": "scripts/perf/standard-workflow-offered.php",
-    "command_sha256": "cac36195aa9d0867516dff7bcb09f7d9df5fc65b9c0a0b7f49f8d7852def055f",
+    "command_sha256": "c0352bbab781df2d3f86f7c96e7b749606e62c19770bdbe85e6c75ef25c58e66",
     "warmup_seconds": 60,
     "offer_seconds": 120,
     "drain_limit_seconds": 120,

scripts/perf/standard-workflow-offered.php

@@ -139,6 +139,7 @@
     }
 }
 
+$negativeFirstStages = 0;
 $lastObservationNs = hrtime(true);
 $lastObservationWall = microtime(true);
 foreach ($completed as $id => $sample) {
@@ -160,16 +161,23 @@
                 $eventTimes[$type] = (float) (new DateTimeImmutable($timestamp))->format('U.u');
             }
         }
-        $previousTime = $started[$id]['wall'];
+        // History order is checked on the Server's clock only. The first stage
+        // starts on the driver's clock, so a driver/target offset larger than
+        // that stage (a few ms) can make it negative; it is recorded, not
+        // treated as a history defect.
+        $previousTime = null;
         $intervals = [];
         foreach ($stageNames as $eventType => $stage) {
             $eventTime = $eventTimes[$eventType] ?? null;
-            if ($eventTime === null || $eventTime < $previousTime) {
+            if ($eventTime === null || ($previousTime !== null && $eventTime < $previousTime)) {
                 throw new RuntimeException('Standard history timestamps are missing or out of order: '.$eventType);
             }
-            $intervals[$stage] = $eventTime - $previousTime;
+            $intervals[$stage] = $eventTime - ($previousTime ?? $started[$id]['wall']);
             $previousTime = $eventTime;
         }
+        if ($intervals[$stageNames['WorkflowStarted']] < 0) {
+            $negativeFirstStages++;
+        }
         foreach ($intervals as $stage => $interval) {
             $stageSamples[$stage][] = $interval;
         }
@@ -221,7 +229,12 @@
     'closed_during_offer_window' => $closedInWindow,
     'offer_window_completions_per_second' => $closedInWindow / $duration,
     'observation_seconds_excluding_history_verification' => ($lastObservationNs - $beganNs) / 1_000_000_000,
-    'latency_boundary' => 'Client start request beginning to Server closed_at; shared host clock, SDK describe collection delay excluded.',
+    'latency_boundary' => 'Client start request beginning (driver clock) to Server closed_at (Server clock); SDK describe collection delay excluded. Across hosts it includes their clock offset.',
+    'cross_clock' => [
+        'stage' => $stageNames['WorkflowStarted'],
+        'min_seconds' => ($stageSamples[$stageNames['WorkflowStarted']] ?? []) === [] ? null : min($stageSamples[$stageNames['WorkflowStarted']]),
+        'negative_count' => $negativeFirstStages,
+    ],
     'latency_p50_seconds' => $percentile($latencies, 0.50),
     'latency_p95_seconds' => $percentile($latencies, 0.95),
     'latency_p99_seconds' => $percentile($latencies, 0.99),

scripts/perf/summarize-server-port-small-host.py

@@ -221,6 +221,7 @@ def step_summary(driver, target, samples, marks, name, step):
         "accepted": sum(r.get("accepted_starts", 0) for r in results),
         "completed_by_deadline": sum(r.get("completed_by_observation_deadline", 0) for r in results),
         "errors": sum(len(r.get("errors", [])) for r in results),
+        "cross_clock_negative_first_stages": sum((r.get("cross_clock") or {}).get("negative_count", 0) for r in results),
         "latency_p50_seconds_median_of_probes": p50s[len(p50s) // 2] if p50s else None,
         "latency_p99_seconds_max_of_probes": max(p99s) if p99s else None,
         **observed(driver, target, samples, marks.get(name)),
@@ -250,17 +251,46 @@ def clock_offsets(text):
     return offsets
 
 
+def clock_jitters(text):
+    """Jitter in seconds from timedatectl timesync-status and chronyc's RMS offset."""
+    jitters = [float(value) * OFFSET_UNITS[unit]
+               for value, unit in re.findall(r"^\s*Jitter:\s*([0-9.]+)(ns|us|ms|s)\b", text, re.MULTILINE)]
+    jitters += [float(value) for value in re.findall(r"RMS offset\s*:\s*([0-9.]+) seconds", text)]
+    return jitters
+
+
+# Driver-relative latency and the probe's first history stage compare the
+# driver's clock with the target's. Each host's offset plus jitter must stay
+# within this for those figures to be reliable; a 1 vCPU kvm-clock host was
+# measured at +39 ms offset and 62 ms jitter while NTP-synchronized.
+CROSS_HOST_TOLERANCE_SECONDS = 0.005
+
+
 def clock_records(driver, target):
     """Synchronization state and numeric offsets only, never the raw text."""
     clock = {}
+    reliable = True
     for side, folder, files in (("driver", driver, ("clock-start.txt", "clock-end.txt")),
                                 ("target", target, ("clock-up.txt", "clock-down.txt"))):
         texts = [read_text(folder / f) or "" for f in files]
-        offsets = [offset for text in texts for offset in clock_offsets(text)]
+        offsets = [clock_offsets(text) for text in texts]
+        jitters = [clock_jitters(text) for text in texts]
+        synchronized = all(re.search(r"NTPSynchronized=yes", t) for t in texts)
+        # Every capture, start and end, must carry its own offset and jitter;
+        # one capture's numbers never stand in for another's missing ones.
+        measured = all(offsets) and all(jitters)
+        max_offset = max((abs(o) for capture in offsets for o in capture), default=None)
+        max_jitter = max((j for capture in jitters for j in capture), default=None)
         clock[side] = {
-            "ntp_synchronized_at_both_ends": all(re.search(r"NTPSynchronized=yes", t) for t in texts),
-            "max_abs_offset_seconds": max((abs(o) for o in offsets), default=None),
+            "ntp_synchronized_at_both_ends": synchronized,
+            "offset_and_jitter_in_every_capture": measured,
+            "max_abs_offset_seconds": max_offset,
+            "max_jitter_seconds": max_jitter,
         }
+        reliable = reliable and synchronized and measured \
+            and max_offset + max_jitter <= CROSS_HOST_TOLERANCE_SECONDS
+    clock["cross_host_tolerance_seconds"] = CROSS_HOST_TOLERANCE_SECONDS
+    clock["cross_host_timing_reliable"] = reliable
     return clock
 
 

tests/Unit/Support/small_host_summary_test.py

@@ -57,7 +57,7 @@ def synthetic_run(root):
     clock = ("unix_seconds=1.0\nNTP=yes\nNTPSynchronized=yes\n"
              "Reference ID    : 0A630001 (synthetic-private-ntp)\n"
              "System time     : 0.000250000 seconds fast of NTP time\n"
-             "       Server: 10.99.0.1 (synthetic-private-ntp)\n  Offset: -120us\n")
+             "       Server: 10.99.0.1 (synthetic-private-ntp)\n  Offset: -120us\n  Jitter: 200us\n")
     for name in ("clock-start.txt", "clock-end.txt"):
         write(driver / name, clock)
     for name in ("clock-up.txt", "clock-down.txt"):
@@ -106,10 +106,74 @@ def test_summary_keeps_results_but_no_private_run_metadata(self):
         self.assertEqual(summary["target"]["provider_plan"], "vhp-1c-1gb-amd")
         self.assertTrue(summary["clock"]["driver"]["ntp_synchronized_at_both_ends"])
         self.assertAlmostEqual(summary["clock"]["target"]["max_abs_offset_seconds"], 0.00025)
+        self.assertTrue(summary["clock"]["cross_host_timing_reliable"])
         self.assertEqual(summary["steps"][0]["errors"], 1)
+        self.assertEqual(summary["steps"][0]["cross_clock_negative_first_stages"], 0)
         self.assertEqual(summary["idle"][0]["probe"]["cleanup_errors"], 1)
         self.assertEqual(summary["target_teardown"], "containers=0 volumes=0")
 
+    def test_a_drifting_target_clock_makes_cross_host_timing_unreliable(self):
+        with tempfile.TemporaryDirectory() as directory:
+            driver, target = synthetic_run(pathlib.Path(directory))
+            drifting = ("unix_seconds=1.0\nNTP=yes\nNTPSynchronized=yes\n"
+                        "       Server: 10.99.0.1 (synthetic-private-ntp)\n"
+                        "  Offset: +39.366ms\n  Jitter: 62.124ms\n")
+            for name in ("clock-up.txt", "clock-down.txt"):
+                write(target / name, drifting)
+            probe = json.loads((driver / "rate-4-probe-1.json").read_text())
+            probe["cross_clock"] = {"stage": "client_start_to_workflow_started", "min_seconds": -0.031,
+                                    "negative_count": 12}
+            write(driver / "rate-4-probe-1.json", json.dumps(probe) + "\n")
+            output = subprocess.run([sys.executable, "-I", str(SUMMARIZER), str(driver), str(target)],
+                                    check=True, capture_output=True, text=True).stdout
+        summary = json.loads(output)
+        self.assertTrue(summary["clock"]["target"]["ntp_synchronized_at_both_ends"])
+        self.assertAlmostEqual(summary["clock"]["target"]["max_abs_offset_seconds"], 0.039366)
+        self.assertAlmostEqual(summary["clock"]["target"]["max_jitter_seconds"], 0.062124)
+        self.assertFalse(summary["clock"]["cross_host_timing_reliable"])
+        self.assertEqual(summary["steps"][0]["cross_clock_negative_first_stages"], 12)
+        self.assertNotIn("synthetic-private-ntp", output)
+
+    def summarize_with_target_clocks(self, up, down):
+        with tempfile.TemporaryDirectory() as directory:
+            driver, target = synthetic_run(pathlib.Path(directory))
+            write(target / "clock-up.txt", up)
+            write(target / "clock-down.txt", down)
+            output = subprocess.run([sys.executable, "-I", str(SUMMARIZER), str(driver), str(target)],
+                                    check=True, capture_output=True, text=True).stdout
+        return json.loads(output)["clock"]
+
+    SYNCED = "unix_seconds=1.0\nNTP=yes\nNTPSynchronized=yes\n"
+
+    def test_missing_jitter_at_both_ends_is_unreliable(self):
+        no_jitter = self.SYNCED + "  Offset: -120us\n"
+        clock = self.summarize_with_target_clocks(no_jitter, no_jitter)
+        self.assertIsNone(clock["target"]["max_jitter_seconds"])
+        self.assertFalse(clock["target"]["offset_and_jitter_in_every_capture"])
+        self.assertFalse(clock["cross_host_timing_reliable"])
+
+    def test_missing_jitter_at_one_end_is_not_covered_by_the_other(self):
+        complete = self.SYNCED + "  Offset: -120us\n  Jitter: 200us\n"
+        no_jitter = self.SYNCED + "  Offset: -120us\n"
+        clock = self.summarize_with_target_clocks(complete, no_jitter)
+        self.assertAlmostEqual(clock["target"]["max_jitter_seconds"], 0.0002)
+        self.assertFalse(clock["target"]["offset_and_jitter_in_every_capture"])
+        self.assertFalse(clock["cross_host_timing_reliable"])
+
+    def test_a_missing_offset_while_synchronized_is_unreliable(self):
+        complete = self.SYNCED + "  Offset: -120us\n  Jitter: 200us\n"
+        no_offset = self.SYNCED + "  Jitter: 200us\n"
+        clock = self.summarize_with_target_clocks(complete, no_offset)
+        self.assertTrue(clock["target"]["ntp_synchronized_at_both_ends"])
+        self.assertFalse(clock["target"]["offset_and_jitter_in_every_capture"])
+        self.assertFalse(clock["cross_host_timing_reliable"])
+
+    def test_complete_small_captures_are_reliable(self):
+        complete = self.SYNCED + "  Offset: -120us\n  Jitter: 200us\n"
+        clock = self.summarize_with_target_clocks(complete, complete)
+        self.assertTrue(clock["target"]["offset_and_jitter_in_every_capture"])
+        self.assertTrue(clock["cross_host_timing_reliable"])
+
 
 class TargetAddressTest(unittest.TestCase):
     def run_script(self, arguments, environment):