fix(data-plane): a max below its own median, and a fraction above 1

Auditing the cross-process e2e output rather than reporting it turned up
two contradictions that shipped.

A maximum below its own median. The e2e printed

  put:  max_ms 133.5   p50_ms 175.0   p99_ms 248.5
  get:  max_ms 0.0063  p50_ms 0.050   p99_ms 0.099

and not one of those four percentiles is data: 100 + 150*0.50 = 175,
100 + 150*0.99 = 248.5, 0 + 0.1*0.50 = 0.05. Two causes. The percentiles
came off the cumulative histogram while the max was step-scoped, so they
described different windows; and interpolation spreads a bucket's samples
uniformly across it, so calls clustered low in a wide bucket read high --
160 calls of exactly 120 ms all land in (100, 250] and interpolate to a p50
of 175, above every call observed.

Both fixed. Percentiles are now differenced per step, so they share a
window with the max, and clamped to it, because a true percentile cannot
exceed the maximum and the maximum is measured exactly. They are withheld
below 50 calls in the window; on a wide DP degree a step clears that
easily, and on a narrow one silence beats bucket geometry.

A fraction above 1. `frac_of_step` read 1.054 -- correct arithmetic on a
`wall_ms` summed over ten processes that ran concurrently, and nonsense as
"105% of the step". Renamed to `busy_frac_mean` and divided by the process
count: the mean fraction of the step a process spent in the data plane,
bounded, and the question people actually ask.

The audit also found three ways the harness was lying, now fixed there:
workers "read" through `lambda: None` so their timings measured nothing;
the driver wrote the whole batch instead of the advantages delta, which is
why bytes_written was exactly 2x bytes_read; and the 131 ms/put denominator
is NoOpDataPlaneClient doing Python bookkeeping, not a wire. With those
corrected the driver's share drops to ~9% of cluster volume and
observability costs 1.2%.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Signed-off-by: Zhiyu Li <zhiyul@oci-aga-slurm-1-dm-02.cm.cluster>

Author: Zhiyu Li

nemo_rl/data_plane/README.md

@@ -456,6 +456,21 @@ affine fit, throughput — is recomputed from the merged totals, never
 averaged across ranks (averaging per-rank percentiles does not give a
 cluster percentile).
 
+Two things read differently in the cluster view and are named to say so:
+
+- **`busy_frac_mean`**, not `frac_of_step`. `wall_ms` is summed over
+  processes that ran concurrently, so dividing it by one step's wall clock
+  exceeds 1 whenever they overlapped (measured 1.054 across ten processes).
+  The mean fraction of the step a process spent in the data plane is
+  bounded and answers the question people ask of it.
+- **Percentiles are per step and clamped to the exact `max_ms`**, and are
+  withheld entirely below 50 calls in the window. Bucket interpolation
+  spreads a bucket's samples uniformly across it, so calls clustered low in
+  a wide bucket read high — 160 calls of 120 ms all land in `(100, 250]`
+  and interpolate to a p50 of 175, above every call observed and above the
+  max reported beside it. The max is measured exactly, so it is the tighter
+  bound.
+
 `grpo_train_sync` fans out to the driver and every policy worker, and logs
 the combined result under `data_plane/cluster/` instead of the driver's
 own. It falls back to `data_plane/driver/` when the fan-out finds only one

nemo_rl/data_plane/observability.py

@@ -95,6 +95,13 @@ class DataPlaneEvent(TypedDict):
 # few lines carry the identity of what broke.
 _MAX_HASH_MISMATCH_LOGS = 20
 
+# Calls needed in a window before a percentile off the histogram means
+# anything. Below this the interpolation returns bucket geometry: one sample
+# in (100, 250] yields p50 = 100 + 150*0.50 = 175 and p99 = 248.5 whatever
+# the call actually took. Reporting that next to an exact ``max_ms`` produced
+# a max below its own median.
+_MIN_SAMPLES_FOR_PERCENTILE = 50
+
 
 class _FieldDigest(NamedTuple):
     """One fingerprint per row, plus how far it can be trusted.
@@ -571,9 +578,18 @@ def cluster_step_metrics(
     """
     wall_ms = merged["total_wall_ms"] - prev.get("total_wall_ms", 0.0)
     overhead_ms = merged["self_ms"] - prev.get("self_ms", 0.0) + collect_ms
+    # Not ``frac_of_step``: ``wall_ms`` here is the SUM over processes that
+    # ran concurrently, so dividing by one step's wall clock gives a number
+    # that exceeds 1 whenever they overlapped (measured 1.054 across ten
+    # processes) -- correct arithmetic, but it reads as "105% of the step".
+    # The mean fraction of the step a process spent in the data plane is
+    # bounded and answers the question people ask of it.
+    n_procs = max(merged.get("n_processes", 1), 1)
     metrics: dict[str, float] = {
         "wall_ms": wall_ms,
-        "frac_of_step": (wall_ms / 1e3 / step_time_s) if step_time_s > 0 else 0.0,
+        "busy_frac_mean": (
+            wall_ms / 1e3 / (step_time_s * n_procs) if step_time_s > 0 else 0.0
+        ),
         "comm_volume_mb": (
             merged["comm_volume_bytes"] - prev.get("comm_volume_bytes", 0)
         )
@@ -582,7 +598,7 @@ def cluster_step_metrics(
         / 1e6,
         "bytes_read_mb": (merged["bytes_read"] - prev.get("bytes_read", 0)) / 1e6,
         "bytes_outstanding_mb": merged["bytes_outstanding"] / 1e6,
-        "n_processes": merged.get("n_processes", 0),
+        "n_processes": n_procs,
         "observability_overhead_ms": overhead_ms,
         "observability_overhead_frac": (overhead_ms / wall_ms if wall_ms > 0 else 0.0),
     }
@@ -595,11 +611,33 @@ def cluster_step_metrics(
         metrics[f"{op}/calls"] = calls
         metrics[f"{op}/wall_ms"] = stats["wall_ms"] - prev_op.get("wall_ms", 0.0)
         metrics[f"{op}/max_ms"] = stats["max_ms"]
-        # Percentiles are worth reporting here and not per process: this
-        # histogram is the sum over every rank, so it is the real cluster
-        # distribution rather than one process's handful of calls.
-        metrics[f"{op}/p50_ms"] = stats["p50_ms"]
-        metrics[f"{op}/p99_ms"] = stats["p99_ms"]
+        # Percentiles over THIS step's calls, summed across ranks, not over
+        # the lifetime: a cumulative percentile beside a step-scoped max is
+        # two different windows on one chart, and it showed a max below its
+        # own median. Emitted only when the window holds enough calls to
+        # out-resolve the buckets -- with a wide DP degree a step easily
+        # clears it, and on a narrow one the honest answer is silence.
+        step_hist = [
+            now - was
+            for now, was in zip(
+                stats["latency_hist"],
+                prev_op.get("latency_hist") or [0] * len(stats["latency_hist"]),
+            )
+        ]
+        if sum(step_hist) >= _MIN_SAMPLES_FOR_PERCENTILE:
+            # Clamped to the exact max. Interpolation spreads a bucket's
+            # samples uniformly across it, so calls clustered low in a wide
+            # bucket read high -- 160 calls of 120 ms all land in (100, 250]
+            # and interpolate to a p50 of 175, above every call observed. A
+            # true percentile cannot exceed the maximum, and the maximum is
+            # measured exactly, so it is the tighter bound.
+            ceiling = stats["max_ms"]
+            metrics[f"{op}/p50_ms"] = min(
+                percentile_from_hist(step_hist, 0.50), ceiling
+            )
+            metrics[f"{op}/p99_ms"] = min(
+                percentile_from_hist(step_hist, 0.99), ceiling
+            )
     return metrics
 
 

tests/unit/data_plane/test_observability.py

@@ -743,6 +743,23 @@ def test_cluster_step_metrics_report_their_own_cost():
     client.close()
 
 
+def test_cluster_busy_fraction_is_bounded():
+    """``wall_ms`` is summed over processes that ran concurrently, so
+    dividing it by one step's wall clock exceeds 1 whenever they
+    overlapped — measured 1.054 across ten processes, which reads as
+    '105% of the step'. The mean per-process fraction is bounded."""
+    # Physical fixture: each rank spends 500 ms inside a 1 s step. A rank
+    # cannot spend longer in the data plane than the step lasted, so a
+    # fixture that implies it would be testing the arithmetic on impossible
+    # input rather than the metric.
+    ranks = [_rank_with([100.0] * 5) for _ in range(10)]
+    metrics = cluster_step_metrics(merge_snapshots(ranks), {}, 1.0)
+
+    assert "frac_of_step" not in metrics
+    assert 0.0 <= metrics["busy_frac_mean"] <= 1.0, metrics["busy_frac_mean"]
+    assert metrics["n_processes"] == 10
+
+
 def test_cluster_overhead_includes_the_collection_fan_out():
     """The fan-out is the larger half of the bill. Reporting only the per-op
     wrapper understated the real cost by ~19x in the cross-process e2e
@@ -765,3 +782,37 @@ def test_cluster_overhead_includes_the_collection_fan_out():
     )
     assert delta == pytest.approx(2.31, rel=1e-6)
     client.close()
+
+
+def _rank_with(latencies_ms):
+    client = MetricsDataPlaneClient(NoOpDataPlaneClient())
+    now = monotonic()
+    for ms in latencies_ms:
+        client._emit("put", "p", 1, 1_000, now - ms / 1e3, "ok")
+    return client.snapshot()
+
+
+def test_cluster_percentiles_never_exceed_the_measured_max():
+    """Bucket interpolation spreads a bucket's samples uniformly across it,
+    so calls clustered low in a wide bucket read high: 160 calls of 120 ms
+    all land in (100, 250] and interpolate to a p50 of 175 — above every
+    call observed, and above the exact max reported beside it. The max is
+    the tighter bound, so the percentiles are clamped to it."""
+    merged = merge_snapshots([_rank_with([120.0] * 20) for _ in range(8)])
+    metrics = cluster_step_metrics(merged, {}, 1.0)
+
+    assert metrics["put/max_ms"] == pytest.approx(120.0, abs=2.0)
+    assert metrics["put/p50_ms"] <= metrics["put/max_ms"]
+    assert metrics["put/p99_ms"] <= metrics["put/max_ms"]
+    assert metrics["put/p50_ms"] <= metrics["put/p99_ms"]
+
+
+def test_cluster_percentiles_withheld_below_a_useful_sample_count():
+    """A percentile off a handful of calls is bucket geometry, not data.
+    Silence beats a number that looks like an answer."""
+    few = merge_snapshots([_rank_with([120.0] * 3) for _ in range(2)])  # 6 calls
+    many = merge_snapshots([_rank_with([120.0] * 20) for _ in range(8)])  # 160
+
+    assert "put/p50_ms" not in cluster_step_metrics(few, {}, 1.0)
+    assert "put/max_ms" in cluster_step_metrics(few, {}, 1.0), "max always works"
+    assert "put/p50_ms" in cluster_step_metrics(many, {}, 1.0)