chore: address second round of PR #282 review comments

nv-alicheng · claude · nv-alicheng · commit 731689ccf428 · 2026-04-20T15:41:11.000-07:00
- docs(report_design): add "reports reproducible from event log" principle
- refactor(metrics_table): rename subtract_field -&gt; delta_start_fieldname
- docs(metrics_table): reword ISL token_ids/text comments so SGLang/OpenAI
  are examples, not defining conditions
- test(kv_store): pin empty SeriesStats min/max sentinels; add snapshot
  isolation regression test
- test(aggregator): add explanatory messages to tracking-window asserts
- test(report_builder): pin max/std_dev on empty compute_summary
- test(sample_order): parametrize over [3, 100, 10_000] dataset sizes
- test(zmq_pool_transport): collapse two pool transport classes into one
  parametrized over (num_workers, create_publisher)

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/docs/metrics/report_design.md b/docs/metrics/report_design.md
@@ -38,6 +38,19 @@ this complexity is needed when the input is a `list[float]` from the KVStore.
 The entire rollup is a single function: `compute_summary(values) → dict`.
 It calls numpy for percentiles and histograms. No classes, no state.
 
+**Reports are reproducible from the event log.**
+
+The KVStore is lossy aggregation — it stores per-metric series, not per-sample
+provenance. The authoritative record of what happened during a run is the event
+log written by the `EventLoggerService`. Every number in a `Report` can be
+recomputed by replaying the event log through the same aggregator logic: if a
+production report shows a TTFT spike, the event log is the ground truth a user
+can mine to attribute the spike to specific samples or time windows.
+
+New metrics must preserve this property: the aggregator may only derive values
+from event fields, never from out-of-band state. If a metric cannot be rebuilt
+from the event log alone, it does not belong in the KVStore.
+
 ## Components
 
 ### `compute_summary(values, percentiles, n_histogram_buckets) → dict`
diff --git a/src/inference_endpoint/async_utils/services/metrics_aggregator/metrics_table.py b/src/inference_endpoint/async_utils/services/metrics_aggregator/metrics_table.py
@@ -150,18 +150,20 @@ def fire(
 
 
 class TimeDeltaTrigger(EmitTrigger):
-    """Sync trigger: emits ev_rec.timestamp_ns - pre_change[required_field].
+    """Sync trigger: emits ev_rec.timestamp_ns - pre_change[delta_start_fieldname].
 
-    Subclass only needs to set metric_name and the required field name.
-    Skips silently if the required field is None (event hasn't occurred yet).
+    The emitted metric is a time delta: the firing event marks the end of the
+    delta, and ``delta_start_fieldname`` names the SampleField whose timestamp
+    marks the start. Skips silently if the start field is None (the delta has
+    not yet opened for this sample).
     """
 
-    def __init__(self, metric_name: str, kv_store: KVStore, subtract_field: str):
-        super().__init__(metric_name, kv_store, requires=(subtract_field,))
-        self._subtract_field = subtract_field
+    def __init__(self, metric_name: str, kv_store: KVStore, delta_start_fieldname: str):
+        super().__init__(metric_name, kv_store, requires=(delta_start_fieldname,))
+        self._delta_start_fieldname = delta_start_fieldname
 
     def fire(self, ev_rec, row, pre_change):
-        baseline = pre_change.get(self._subtract_field)
+        baseline = pre_change.get(self._delta_start_fieldname)
         if baseline is not None:
             self.kv_store.update(self.metric_name, ev_rec.timestamp_ns - baseline)
         return None
@@ -235,7 +237,9 @@ class TtftTrigger(TimeDeltaTrigger):
 
     def __init__(self, kv_store: KVStore):
         super().__init__(
-            MetricSeriesKey.TTFT_NS, kv_store, subtract_field=SampleField.ISSUED_NS
+            MetricSeriesKey.TTFT_NS,
+            kv_store,
+            delta_start_fieldname=SampleField.ISSUED_NS,
         )
 
 
@@ -249,7 +253,7 @@ def __init__(self, kv_store: KVStore):
         super().__init__(
             MetricSeriesKey.CHUNK_DELTA_NS,
             kv_store,
-            subtract_field=SampleField.LAST_RECV_NS,
+            delta_start_fieldname=SampleField.LAST_RECV_NS,
         )
 
 
@@ -260,7 +264,7 @@ def __init__(self, kv_store: KVStore):
         super().__init__(
             MetricSeriesKey.SAMPLE_LATENCY_NS,
             kv_store,
-            subtract_field=SampleField.ISSUED_NS,
+            delta_start_fieldname=SampleField.ISSUED_NS,
         )
 
 
@@ -281,11 +285,12 @@ def __init__(
         super().__init__(MetricSeriesKey.ISL, kv_store, tokenize_pool, loop)
 
     def fire(self, ev_rec, row, pre_change):
-        # Sync fast path: pre-tokenized IDs (SGLang)
+        # Sync fast path: any backend that pre-populates token_ids (e.g. SGLang).
         if isinstance(ev_rec.data, PromptData) and ev_rec.data.token_ids is not None:
             self.kv_store.update(self.metric_name, len(ev_rec.data.token_ids))
             return None
-        # Async path: tokenize raw text (OpenAI) — handled by base class
+        # Async path: tokenize raw text — used when token_ids are unavailable
+        # (e.g. OpenAI-compatible endpoints). Handled by the base class.
         return super().fire(ev_rec, row, pre_change)
 
     def _extract_text(self, ev_rec, row, pre_change):
diff --git a/tests/unit/async_utils/services/metrics_aggregator/test_aggregator.py b/tests/unit/async_utils/services/metrics_aggregator/test_aggregator.py
@@ -58,9 +58,17 @@ async def test_not_tracked_before_start(self):
                 sample_event(SampleEventType.ISSUED, "s1", ts=100),
             ]
         )
-        assert agg._table.get_row("s1") is None
-        assert store.get_series_values("ttft_ns") == []
-        assert store.get_series_values("sample_latency_ns") == []
+        assert agg._table.get_row("s1") is None, (
+            "Sample issued before START_PERFORMANCE_TRACKING must not create a "
+            "table row — warmup samples should be excluded from the tracked set."
+        )
+        assert (
+            store.get_series_values("ttft_ns") == []
+        ), "No TTFT should be recorded for samples issued before tracking begins."
+        assert store.get_series_values("sample_latency_ns") == [], (
+            "No sample_latency should be recorded for samples issued before "
+            "tracking begins."
+        )
 
     @pytest.mark.asyncio
     async def test_tracked_after_start(self):
@@ -72,7 +80,10 @@ async def test_tracked_after_start(self):
                 sample_event(SampleEventType.ISSUED, "s1", ts=100),
             ]
         )
-        assert agg._table.get_row("s1") is not None
+        assert agg._table.get_row("s1") is not None, (
+            "Sample issued after START_PERFORMANCE_TRACKING must create a table "
+            "row so its metrics are included in the tracked set."
+        )
 
     @pytest.mark.asyncio
     async def test_not_tracked_after_stop(self):
@@ -85,7 +96,10 @@ async def test_not_tracked_after_stop(self):
                 sample_event(SampleEventType.ISSUED, "s1", ts=100),
             ]
         )
-        assert agg._table.get_row("s1") is None
+        assert agg._table.get_row("s1") is None, (
+            "Sample issued after STOP_PERFORMANCE_TRACKING must not create a "
+            "table row — the tracking window has closed."
+        )
 
     @pytest.mark.asyncio
     async def test_inflight_sample_continues_after_stop(self):
diff --git a/tests/unit/async_utils/services/metrics_aggregator/test_kv_store.py b/tests/unit/async_utils/services/metrics_aggregator/test_kv_store.py
@@ -15,6 +15,7 @@
 
 """Tests for the KVStore (BasicKVStore + BasicKVStoreReader)."""
 
+import math
 import multiprocessing
 import struct
 from pathlib import Path
@@ -48,6 +49,10 @@ def test_empty(self):
         stats = SeriesStats()
         assert stats.count == 0
         assert stats.total == 0.0
+        # Sentinel values for an empty series — compute_summary() is responsible
+        # for normalizing these to 0 before exposing them to users.
+        assert stats.min_val == math.inf
+        assert stats.max_val == -math.inf
 
     def test_incremental_rollup(self):
         stats = SeriesStats([1.0, 2.0])
@@ -119,6 +124,28 @@ def test_snapshot(self, tmp_path: Path):
         assert snap["latency"].count == 2
         store.close()
 
+    def test_snapshot_is_isolated_from_later_writes(self, tmp_path: Path):
+        """Mutations after snapshot() must not alter the captured snapshot."""
+        store = BasicKVStore(tmp_path / "kv")
+        store.create_key("n_issued", "counter")
+        store.create_key("latency", "series")
+        store.update("n_issued", 5)
+        store.update("latency", 100)
+        store.update("latency", 200)
+
+        snap = store.snapshot()
+
+        store.update("n_issued", 99)
+        store.update("latency", 300)
+
+        assert snap["n_issued"] == 5
+        latency_snap = snap["latency"]
+        assert isinstance(latency_snap, SeriesStats)
+        assert latency_snap.count == 2
+        assert latency_snap.values == [100, 200]
+        assert latency_snap.total == 300
+        store.close()
+
     def test_update_unknown_key_raises(self, tmp_path: Path):
         store = BasicKVStore(tmp_path / "kv")
         with pytest.raises(KeyError, match="Key not created"):
diff --git a/tests/unit/load_generator/test_sample_order.py b/tests/unit/load_generator/test_sample_order.py
@@ -23,43 +23,52 @@
     WithReplacementSampleOrder,
 )
 
+# Exercise small/medium/large dataset sizes so shuffle-buffer behavior is
+# covered for inputs both much smaller and much larger than typical batches.
+_DATASET_SIZES = [3, 100, 10_000]
+
 
 @pytest.mark.unit
 class TestWithoutReplacementSampleOrder:
-    def test_yields_all_indices(self):
+    @pytest.mark.parametrize("n_samples", _DATASET_SIZES)
+    def test_yields_all_indices(self, n_samples: int):
         order = WithoutReplacementSampleOrder(
-            n_samples_in_dataset=5, rng=random.Random(42)
+            n_samples_in_dataset=n_samples, rng=random.Random(42)
         )
-        indices = [next(order) for _ in range(5)]
-        assert sorted(indices) == [0, 1, 2, 3, 4]
+        indices = [next(order) for _ in range(n_samples)]
+        assert sorted(indices) == list(range(n_samples))
 
-    def test_reshuffles_after_exhaustion(self):
+    @pytest.mark.parametrize("n_samples", _DATASET_SIZES)
+    def test_reshuffles_after_exhaustion(self, n_samples: int):
         order = WithoutReplacementSampleOrder(
-            n_samples_in_dataset=3, rng=random.Random(42)
+            n_samples_in_dataset=n_samples, rng=random.Random(42)
         )
-        first_pass = [next(order) for _ in range(3)]
-        second_pass = [next(order) for _ in range(3)]
-        assert sorted(first_pass) == [0, 1, 2]
-        assert sorted(second_pass) == [0, 1, 2]
+        first_pass = [next(order) for _ in range(n_samples)]
+        second_pass = [next(order) for _ in range(n_samples)]
+        assert sorted(first_pass) == list(range(n_samples))
+        assert sorted(second_pass) == list(range(n_samples))
 
-    def test_never_raises_stop_iteration(self):
+    @pytest.mark.parametrize("n_samples", _DATASET_SIZES)
+    def test_never_raises_stop_iteration(self, n_samples: int):
         order = WithoutReplacementSampleOrder(
-            n_samples_in_dataset=2, rng=random.Random(42)
+            n_samples_in_dataset=n_samples, rng=random.Random(42)
         )
         # Should be able to draw far more than dataset size
-        indices = [next(order) for _ in range(100)]
-        assert len(indices) == 100
-        assert all(0 <= i < 2 for i in indices)
+        draws = max(100, n_samples * 3)
+        indices = [next(order) for _ in range(draws)]
+        assert len(indices) == draws
+        assert all(0 <= i < n_samples for i in indices)
 
-    def test_reproducible_with_seed(self):
+    @pytest.mark.parametrize("n_samples", _DATASET_SIZES)
+    def test_reproducible_with_seed(self, n_samples: int):
         order1 = WithoutReplacementSampleOrder(
-            n_samples_in_dataset=10, rng=random.Random(42)
+            n_samples_in_dataset=n_samples, rng=random.Random(42)
         )
         order2 = WithoutReplacementSampleOrder(
-            n_samples_in_dataset=10, rng=random.Random(42)
+            n_samples_in_dataset=n_samples, rng=random.Random(42)
         )
-        seq1 = [next(order1) for _ in range(20)]
-        seq2 = [next(order2) for _ in range(20)]
+        seq1 = [next(order1) for _ in range(n_samples * 2)]
+        seq2 = [next(order2) for _ in range(n_samples * 2)]
         assert seq1 == seq2
 
     def test_invalid_size_raises(self):
@@ -69,20 +78,22 @@ def test_invalid_size_raises(self):
 
 @pytest.mark.unit
 class TestWithReplacementSampleOrder:
-    def test_yields_valid_indices(self):
+    @pytest.mark.parametrize("n_samples", _DATASET_SIZES)
+    def test_yields_valid_indices(self, n_samples: int):
         order = WithReplacementSampleOrder(
-            n_samples_in_dataset=5, rng=random.Random(42)
+            n_samples_in_dataset=n_samples, rng=random.Random(42)
         )
-        indices = [next(order) for _ in range(100)]
-        assert all(0 <= i < 5 for i in indices)
+        indices = [next(order) for _ in range(max(100, n_samples))]
+        assert all(0 <= i < n_samples for i in indices)
 
-    def test_reproducible_with_seed(self):
+    @pytest.mark.parametrize("n_samples", _DATASET_SIZES)
+    def test_reproducible_with_seed(self, n_samples: int):
         order1 = WithReplacementSampleOrder(
-            n_samples_in_dataset=10, rng=random.Random(42)
+            n_samples_in_dataset=n_samples, rng=random.Random(42)
         )
         order2 = WithReplacementSampleOrder(
-            n_samples_in_dataset=10, rng=random.Random(42)
+            n_samples_in_dataset=n_samples, rng=random.Random(42)
         )
-        seq1 = [next(order1) for _ in range(20)]
-        seq2 = [next(order2) for _ in range(20)]
+        seq1 = [next(order1) for _ in range(n_samples * 2)]
+        seq2 = [next(order2) for _ in range(n_samples * 2)]
         assert seq1 == seq2
diff --git a/tests/unit/metrics/test_report_builder.py b/tests/unit/metrics/test_report_builder.py
@@ -37,6 +37,8 @@ def test_empty(self):
         s = compute_summary(SeriesStats())
         assert s["total"] == 0
         assert s["min"] == 0
+        assert s["max"] == 0
+        assert s["std_dev"] == 0
         assert s["histogram"]["buckets"] == []
 
     def test_single_value(self):
diff --git a/tests/unit/transport/test_zmq_pool_transport.py b/tests/unit/transport/test_zmq_pool_transport.py
@@ -109,60 +109,24 @@ async def test_socket_closed_on_cancellation(self):
 
 @pytest.mark.unit
 @pytest.mark.asyncio
-class TestZmqPoolTransportWithPublisher:
-    """Test pool transport creation with a publisher on the same context."""
-
-    async def _create_publisher_and_pool(
-        self, loop: asyncio.AbstractEventLoop, num_workers: int
-    ):
-        """Helper: create publisher + pool transport, test ready check socket."""
-        sid = uuid.uuid4().hex[:8]
-        zmq_ctx = ManagedZMQContext(io_threads=2)
-        publisher = ZmqEventRecordPublisher(f"ev_pub_{sid}", zmq_ctx, loop=loop)
-
-        pool = ZmqWorkerPoolTransport.create(
-            loop, num_workers, config=ZMQTransportConfig()
-        )
-
-        rc = pool._ready_check
-        assert not rc._sock.closed
-        _ = rc._sock.rcvtimeo
-
-        with pytest.raises(TimeoutError):
-            await pool.wait_for_workers_ready(timeout=0.1)
-
-        pool.cleanup()
-        publisher.close()
-        zmq_ctx.cleanup()
-
-    async def test_2_workers(self):
-        loop = asyncio.get_running_loop()
-        await self._create_publisher_and_pool(loop, 2)
-
-    async def test_3_workers(self):
-        loop = asyncio.get_running_loop()
-        await self._create_publisher_and_pool(loop, 3)
-
-    async def test_4_workers(self):
-        loop = asyncio.get_running_loop()
-        await self._create_publisher_and_pool(loop, 4)
-
-    async def test_8_workers(self):
-        loop = asyncio.get_running_loop()
-        await self._create_publisher_and_pool(loop, 8)
-
-
-@pytest.mark.unit
-@pytest.mark.asyncio
-class TestZmqPoolTransportWithoutPublisher:
-    """Test pool transport creation without a publisher (baseline)."""
+class TestZmqPoolTransport:
+    """Pool transport creation with and without a publisher on the same context."""
 
     @pytest.mark.parametrize("num_workers", [2, 3, 4, 8])
-    async def test_pool_only(self, num_workers: int):
+    @pytest.mark.parametrize("create_publisher", [True, False])
+    async def test_pool(self, num_workers: int, create_publisher: bool):
         loop = asyncio.get_running_loop()
         zmq_ctx = ManagedZMQContext(io_threads=2)
-        dummy = zmq_ctx.socket(zmq.PUB)
-        zmq_ctx.bind(dummy, "dummy")
+
+        publisher = None
+        dummy = None
+        if create_publisher:
+            sid = uuid.uuid4().hex[:8]
+            publisher = ZmqEventRecordPublisher(f"ev_pub_{sid}", zmq_ctx, loop=loop)
+        else:
+            # Baseline: bind an unrelated PUB socket so the context is non-empty.
+            dummy = zmq_ctx.socket(zmq.PUB)
+            zmq_ctx.bind(dummy, "dummy")
 
         pool = ZmqWorkerPoolTransport.create(
             loop, num_workers, config=ZMQTransportConfig()
@@ -176,5 +140,8 @@ async def test_pool_only(self, num_workers: int):
             await pool.wait_for_workers_ready(timeout=0.1)
 
         pool.cleanup()
-        dummy.close()
+        if publisher is not None:
+            publisher.close()
+        if dummy is not None:
+            dummy.close()
         zmq_ctx.cleanup()
