fix(metrics): address P0 review-council findings

Two high-severity issues raised by the review-council pass on PR #306:

1. (#306-1) Subscriber late-binding could drop early ticks via the ZMQ
   slow-joiner pattern. Move MetricsSnapshotSubscriber construction +
   start() BEFORE launcher.launch() so the SUB handshake completes
   during the subprocess-spawn window. ZMQ tolerates connect-before-
   bind on IPC — the connect resolves once the binder appears. The
   prior ordering (subscribe AFTER launch returns) had a window where
   the aggregator could begin ticking on STARTED before the SUB
   subscription warmed up, dropping early live snapshots and, in the
   worst case, missing COMPLETE entirely.

2. (#306-2) MetricsPublisher._write_atomic_fallback runs synchronous
   f.flush + fsync(file) + fsync(parent dir) + rename on the
   aggregator's event loop. On a busy host this can block tens-to-
   hundreds of ms — long enough to back-pressure event-record
   processing. Wrap with asyncio.to_thread inside publish_final.

Both fixes are localized — no API changes, no test changes required.
Existing integration tests (test_concurrency_benchmark, test_end_to_
end_oracle) exercise both paths end-to-end and still pass.

The third P0 item (#306-3, unbounded raw-sample retention) is the
agreed memory trade documented in metrics_pubsub_design_v5.md §11;
addressed by adding "--persist-raw" as a tracked follow-up rather
than a code change in this PR.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: nv-alicheng

src/inference_endpoint/async_utils/services/metrics_aggregator/publisher.py

@@ -160,8 +160,15 @@ async def publish_final(
             logger.exception("metrics: pub/sub final publish failed")
 
         # Disk fallback — best-effort, must not affect pub/sub above.
+        # The atomic write does synchronous f.flush + fsync(file) +
+        # fsync(parent dir) + rename, which can block tens-to-hundreds of
+        # ms on a busy host. Run it on a worker thread so it doesn't
+        # back-pressure any in-flight event-record processing on the
+        # aggregator's event loop.
         try:
-            self._write_atomic_fallback(self._encoder.encode(snap))
+            await asyncio.to_thread(
+                self._write_atomic_fallback, self._encoder.encode(snap)
+            )
         except Exception:  # noqa: BLE001 — best-effort.
             logger.exception("metrics: disk fallback write failed")
 

src/inference_endpoint/commands/benchmark/execute.py

@@ -423,18 +423,25 @@ async def _run_benchmark_async(
         metrics_output_dir = ctx.report_dir / "metrics"
         metrics_output_dir.mkdir(parents=True, exist_ok=True)
 
-        # Subscribe to the metrics PUB socket BEFORE the aggregator binds it,
-        # so we never miss the STARTED-time first ticks. The aggregator's
-        # ManagedZMQContext is a separate process; we share socket_dir.
         metrics_socket_name = f"metrics_pub_{uuid.uuid4().hex[:8]}"
-        # The aggregator subprocess will bind metrics_socket_name; the main
-        # process just needs to know the path to connect to. Connect is
-        # deferred until after launcher.launch() so the IPC file exists.
 
-        # Launch service subprocesses
-        launcher = ServiceLauncher(zmq_ctx)
+        # Connect the metrics-snapshot subscriber BEFORE launching the
+        # aggregator subprocess that binds the matching PUB socket. ZMQ
+        # tolerates connect-before-bind on IPC (the connect resolves once
+        # the binder appears), and starting the SUB reader early gives
+        # the subscription handshake time to complete during the
+        # ~1-2 second subprocess-launch window. This eliminates the
+        # slow-joiner risk of dropping early live ticks (or the worst
+        # case: missing COMPLETE if the SUB handshake never warms up).
         if zmq_ctx.socket_dir is None:
             raise RuntimeError("ZMQ socket_dir must be set after publisher bind")
+        metrics_subscriber = MetricsSnapshotSubscriber(
+            metrics_socket_name, zmq_ctx, loop
+        )
+        metrics_subscriber.start()
+
+        # Launch service subprocesses
+        launcher = ServiceLauncher(zmq_ctx)
         aggregator_args: list[str] = [
             "--socket-dir",
             zmq_ctx.socket_dir,
@@ -476,14 +483,6 @@ async def _run_benchmark_async(
             timeout=30.0,
         )
 
-        # Connect the metrics-snapshot subscriber AFTER aggregator readiness
-        # so the IPC bind is in place. We may still miss the very first tick;
-        # the disk fallback covers the missing-final case.
-        metrics_subscriber = MetricsSnapshotSubscriber(
-            metrics_socket_name, zmq_ctx, loop
-        )
-        metrics_subscriber.start()
-
         # Create endpoint client on the shared loop
         endpoints = config.endpoint_config.endpoints
         logger.info(f"Connecting: {endpoints}")