feat(sc): telemetry for rollout checkpointing

docs/guides/single-controller.md

@@ -119,6 +119,8 @@ checkpointing:
 
 rollout_checkpointing:
   snapshot_attempt_interval_s: 120
+  telemetry_interval_s: null
+  max_consecutive_failures: 3
   keep_latest_k: 2
   restore_mode: latest
   extra_fingerprint_excluded_paths: []
@@ -136,6 +138,17 @@ are skipped until the matching trainer checkpoint exists. Before the first
 training step, snapshots are anchored to the initial model and a fingerprint of
 the rollout-semantic configuration.
 
+`telemetry_interval_s` controls an independent wall-clock sampler for rollout
+throughput and checkpoint pressure. It is `null` (disabled) by default; set it
+to a positive number such as `30` to emit one sample every 30 seconds. This does
+not change the checkpoint cadence. See the
+[Single-Controller rollout recovery metrics](../observability/metrics.md#single-controller-rollout-recovery-metrics)
+for the emitted fields.
+
+`max_consecutive_failures` is the number of consecutive retryable periodic-save
+failures tolerated before training aborts. A successful or skipped checkpoint
+attempt resets the count; checkpoint invariant failures still fail immediately.
+
 The bootstrap fingerprint is fail-closed: every configuration value affects
 compatibility unless NeMo-RL's built-in denylist identifies it as operational,
 such as logging, cluster placement, checkpoint location, runtime ports, or

docs/observability/metrics.md

@@ -6,6 +6,72 @@ Metrics are emitted **only when telemetry is exporting** — the driver always e
 
 Training scalars — reward, loss, KL, grad norm, learning rate, throughput — are **not** mirrored to OTel. nemo-lens declares `record_rl_metrics` gauges for most of them, plus `rl.generation.duration_ms` and `rl.rollout.duration_ms` histograms, but NeMo-RL emits none of them: mapping its logger keys onto lens's fixed fields is still being settled with the lens owners. Read those scalars from W&B / TensorBoard, and phase durations from the spans.
 
+## W&B history step axes
+
+W&B's internal `_step` is a monotonically increasing history-row number, not a
+NeMo-RL trainer step. Trainer-correlated metrics are buffered into one history
+row and carry the explicit `nemo_rl/step` field. Use `nemo_rl/step` when plotting
+or joining training metrics by optimizer step; relying on `_step` also counts
+independently committed telemetry rows and can misalign the series.
+
+Independent event streams use their own custom axes and do not carry
+`nemo_rl/step`. In particular, Single-Controller rollout benchmark series under
+`rollout/throughput/*`, `timing/rollout_checkpoint/*`,
+`timing/rollout_recovery/*`, and `rollout/checkpoint_outcome/*` use
+`telemetry/wall_time_seconds`, recorded as Unix wall-clock time. This lets those
+series continue through a long or paused trainer step and across process restart
+without changing the meaning of the trainer-step axis.
+
+## Single-Controller rollout recovery metrics
+
+Wall-clock rollout telemetry sampling is disabled by default. Enable it without
+changing the checkpoint cadence by setting, for example:
+
+```yaml
+rollout_checkpointing:
+  telemetry_interval_s: 30.0
+```
+
+This records one throughput and controller-pressure sample every 30 seconds.
+Set the value back to `null` to disable the sampler. Detailed per-interval
+metric dictionaries are printed to the controller log only when
+`async_rl.diagnostics: true`; configured metric backends receive them either
+way.
+
+The rollout checkpoint benchmark focuses on the following questions:
+
+1. Did checkpointing reduce raw generation or usable rollout throughput?
+2. Which save or restore phase took the time?
+3. How much live data-plane work waited behind the checkpoint barrier?
+4. Did scheduled checkpoint attempts succeed at the intended cadence?
+5. How much completed rollout work was reused after restart?
+
+| Prefix | Important fields | Meaning |
+|---|---|---|
+| `rollout/throughput` | `generation_output_tokens_per_second`, `committed_output_tokens_per_second`, `committed_groups_per_second` | Raw backend decoding throughput compared with output committed for training. The raw metric is absent when the backend does not expose compatible cumulative counters. With token capture, committed tokens are counted exactly from valid staged rows; without token capture, they are estimated by multiplying the reported per-sample mean by the number of completions, so compare like-for-like runs. |
+| `rollout/throughput` | `checkpoint_blocked_mutations`, `checkpoint_mutation_wait_seconds_p95`, `checkpoint_mutation_wait_seconds_max` | Number and latency of live data-plane mutations delayed by an exclusive checkpoint. |
+| `timing/rollout_checkpoint` | `total_save_seconds`, `tq_save_seconds`, `barrier_wait_seconds`, `exclusive_hold_seconds`, `sidecar_save_seconds`, `snapshot_commit_seconds` | End-to-end save latency and its storage, fencing, controller-sidecar, and atomic-publication components. |
+| `timing/rollout_checkpoint` | `snapshot_rows`, `replay_rows`, `staging_rows`, `replay_groups`, `ledger_groups`, `controller_sidecar_bytes` | Logical volume captured by the snapshot. `controller_sidecar_bytes` excludes the native TQ payload because the current TQ checkpoint API does not report bytes written. NeMo-RL deliberately does not recursively scan the shared checkpoint directory because that scan would perturb the benchmark. |
+| `rollout/checkpoint_outcome` | `completed`, `skipped`, `failed`, `reason_*`, `seconds_since_previous_success`, `seconds_since_last_success` | Result, actionable reason, and effective cadence of every scheduled checkpoint attempt. |
+| `timing/rollout_recovery` | `snapshot_resolution_seconds`, `dataloader_load_seconds`, `tq_load_seconds`, `replay_metadata_load_seconds`, `recovery_prepare_seconds`, `total_load_seconds` | Rollout-state restore latency. `total_load_seconds` is the sum of these non-overlapping restore phases. |
+| `timing/rollout_recovery` | `groups_complete_restored`, `groups_unfinished_found`, `siblings_reused`, `siblings_rerun`, `redispatch_schedule_seconds` | Training-ready groups restored without generation, unfinished groups found for redispatch, and sibling work preserved or rerun after restart. |
+
+Recovery counters describe work reconstructed at restore time, not a promise
+that every restored group will eventually train. After recovery, the configured
+sampler may still evict a restored group under its normal staleness rules (for
+example, when a `windowed` sampler finds that the group's policy version has
+fallen outside its valid window).
+
+`barrier_wait_seconds` and mutation wait latency measure opposite sides of the
+same fence. The former is how long the checkpoint waits for already-running
+mutations; the latter is how long rollout or training mutations wait for the
+checkpoint to release its exclusive cut.
+
+The vLLM request, KV-cache, controller-waiter, buffer-occupancy, and per-mutation
+kind fields are diagnostic drill-down signals. They are useful when one of the
+primary throughput or barrier metrics regresses, but do not need to appear on
+the primary rollout checkpoint dashboard.
+
 ## Async efficiency metrics (`rl.efficiency.*`)
 
 Async GRPO measures where wall time goes with a `Timer` and logs the result as `efficiency/*` scalars (`print_efficiency_summary` in `nemo_rl/algorithms/utils.py`). Those same values are teed to OTel as one **dimensioned** gauge rather than one instrument per category, so adding a category needs no instrument change.

examples/configs/grpo_math_1B_megatron_single_controller.yaml

@@ -156,6 +156,8 @@ checkpointing:
 # checkpoint support. save_period=1 provides an anchor after every train step.
 rollout_checkpointing:
   snapshot_attempt_interval_s: null
+  telemetry_interval_s: null
+  max_consecutive_failures: 3
   keep_latest_k: 2
   restore_mode: latest
   # Advanced escape hatch for runtime-only fields from external integrations.

examples/configs/ppo_math_1B_megatron_single_controller.yaml

@@ -161,6 +161,8 @@ checkpointing:
 # checkpoint support. save_period=1 provides an anchor after every train step.
 rollout_checkpointing:
   snapshot_attempt_interval_s: null
+  telemetry_interval_s: null
+  max_consecutive_failures: 3
   keep_latest_k: 2
   restore_mode: latest
   # Advanced escape hatch for runtime-only fields from external integrations.

examples/nemo_gym/grpo_qwen3_30ba3b_instruct.yaml

@@ -53,6 +53,10 @@ policy:
   generation:
     vllm_cfg:
       tensor_parallel_size: 4
+      # Required for rollout-throughput checkpoint A/B telemetry. Keep this
+      # identical in baseline and checkpoint-enabled runs.
+      enable_vllm_metrics_logger: true
+      vllm_metrics_logger_interval: 0.5
       # This is a very low GPU mem utilization. We GPU OOM in two places:
       # Refit after train, refit before validation.
       gpu_memory_utilization: 0.7
@@ -174,3 +178,8 @@ checkpointing:
   # 4. The checkpoint time for this model is around 10 mins.
   checkpoint_must_save_by: "00:03:30:00"
   save_period: 1
+
+# Optional wall-clock benchmark sampling. This is independent of the rollout
+# checkpoint save interval and may be enabled for baseline runs as well.
+rollout_checkpointing:
+  telemetry_interval_s: null

nemo_rl/algorithms/async_utils/replay_buffer.py

@@ -20,10 +20,12 @@
 import math
 import statistics
 import threading as _threading
+import time
 import uuid
-from collections import Counter
+from collections import Counter, deque
 from collections.abc import AsyncIterator, Mapping
 from contextlib import asynccontextmanager
+from dataclasses import dataclass
 from numbers import Integral, Real
 from typing import (
     Any,
@@ -34,6 +36,8 @@
     NotRequired,
     Optional,
     TypedDict,
+    cast,
+    get_args,
 )
 
 import ray
@@ -63,6 +67,35 @@
 REPLAY_BUFFER_METADATA_SCHEMA_VERSION = 1
 REPLAY_BUFFER_METADATA_STORAGE: Literal["tq_checkpoint"] = "tq_checkpoint"
 
+CheckpointMutationKind = Literal[
+    "advantage_writeback",
+    "group_commits",
+    "group_removals",
+    "other",
+    "prompt_reservations",
+    "recovery_restore",
+    "recovery_retries",
+    "sample_clears",
+    "sibling_seals",
+]
+CHECKPOINT_MUTATION_KINDS = cast(
+    tuple[CheckpointMutationKind, ...],
+    get_args(CheckpointMutationKind),
+)
+
+
+@dataclass(frozen=True)
+class DataPlaneCheckpointBarrierTelemetry:
+    """Bounded interval telemetry for checkpoint-induced mutation waits."""
+
+    blocked_by_kind: dict[CheckpointMutationKind, int]
+    wait_durations_s: tuple[float, ...]
+    active_mutations: int
+    waiting_mutations: int
+    max_waiting_mutations: int
+    checkpoint_active: bool
+
+
 # These TypedDicts describe the versioned, plain-mapping checkpoint wire
 # format. They are intentionally not dataclass instances: persisting a
 # dataclass would couple recovery to its Python import path and class layout.
@@ -233,6 +266,10 @@ def __init__(self) -> None:
         self._active_mutations = 0
         self._section_holders: set[asyncio.Task[Any]] = set()
         self._mutation_version = 0
+        self._waiting_mutations = 0
+        self._max_waiting_mutations = 0
+        self._blocked_by_kind: Counter[CheckpointMutationKind] = Counter()
+        self._wait_durations_s: deque[float] = deque(maxlen=10_000)
 
     def _current_task(self) -> asyncio.Task[Any]:
         """Return the task entering a barrier section and reject reentrancy."""
@@ -252,11 +289,31 @@ def mutation_version(self) -> int:
         return self._mutation_version
 
     @asynccontextmanager
-    async def mutation(self) -> AsyncIterator[DataPlaneMutationCut]:
-        """Yield a live cut after any active checkpoint exits."""
+    async def mutation(
+        self, kind: CheckpointMutationKind = "other"
+    ) -> AsyncIterator[DataPlaneMutationCut]:
+        """Yield one task-local live cut after any active checkpoint exits."""
+        if kind not in CHECKPOINT_MUTATION_KINDS:
+            raise ValueError(
+                f"unknown checkpoint mutation kind {kind!r}; expected one of "
+                f"{CHECKPOINT_MUTATION_KINDS!r}"
+            )
         async with self._condition:
             task = self._current_task()
-            await self._condition.wait_for(lambda: not self._checkpoint_active)
+            wait_started: Optional[float] = None
+            if self._checkpoint_active:
+                wait_started = time.monotonic()
+                self._waiting_mutations += 1
+                self._max_waiting_mutations = max(
+                    self._max_waiting_mutations, self._waiting_mutations
+                )
+            try:
+                await self._condition.wait_for(lambda: not self._checkpoint_active)
+            finally:
+                if wait_started is not None:
+                    self._waiting_mutations -= 1
+                    self._blocked_by_kind[kind] += 1
+                    self._wait_durations_s.append(time.monotonic() - wait_started)
             self._active_mutations += 1
             self._section_holders.add(task)
             cut = DataPlaneMutationCut(self)
@@ -273,6 +330,25 @@ async def mutation(self) -> AsyncIterator[DataPlaneMutationCut]:
                 if self._active_mutations == 0:
                     self._condition.notify_all()
 
+    async def drain_telemetry(self) -> DataPlaneCheckpointBarrierTelemetry:
+        """Return and reset interval waits while preserving current state."""
+        async with self._condition:
+            telemetry = DataPlaneCheckpointBarrierTelemetry(
+                blocked_by_kind={
+                    kind: self._blocked_by_kind.get(kind, 0)
+                    for kind in CHECKPOINT_MUTATION_KINDS
+                },
+                wait_durations_s=tuple(self._wait_durations_s),
+                active_mutations=self._active_mutations,
+                waiting_mutations=self._waiting_mutations,
+                max_waiting_mutations=self._max_waiting_mutations,
+                checkpoint_active=self._checkpoint_active,
+            )
+            self._blocked_by_kind.clear()
+            self._wait_durations_s.clear()
+            self._max_waiting_mutations = self._waiting_mutations
+            return telemetry
+
     @asynccontextmanager
     async def checkpoint(self) -> AsyncIterator[DataPlaneMutationCut]:
         """Yield a live capability after blocking and draining all mutations."""
@@ -1184,7 +1260,7 @@ async def commit(
                 "the async message-log flattening path."
             )
         trace_rollout_payload(keys=sample_ids, data=train_batch)
-        async with self._data_plane_checkpoint_barrier.mutation() as cut:
+        async with self._data_plane_checkpoint_barrier.mutation("group_commits") as cut:
             try:
                 await call_data_plane(
                     self._dp_client,
@@ -1264,7 +1340,9 @@ async def remove_group(self, group_id: str, *, remove_in_dp: bool = False) -> in
                 "TQReplayBuffer must be bound to the controller data-plane "
                 "checkpoint barrier before removing a group"
             )
-        async with self._data_plane_checkpoint_barrier.mutation() as cut:
+        async with self._data_plane_checkpoint_barrier.mutation(
+            "group_removals"
+        ) as cut:
             return await self._remove_groups_unlocked(
                 cut, [group_id], clear_data_plane=remove_in_dp
             )
@@ -1435,7 +1513,9 @@ async def remove(self, idxs: list[int], remove_in_dp: bool) -> int:
         # removal may shift every list index while this task waits for the barrier
         # or for DataPlane cleanup.
         drop_group_ids = [self._group_ids[i] for i in drop_idxs]
-        async with self._data_plane_checkpoint_barrier.mutation() as cut:
+        async with self._data_plane_checkpoint_barrier.mutation(
+            "group_removals"
+        ) as cut:
             return await self._remove_groups_unlocked(
                 cut, drop_group_ids, clear_data_plane=remove_in_dp
             )
@@ -1465,7 +1545,9 @@ async def claim_for_training(self, idxs: list[int]) -> int:
                 f"{claim_idxs[0]}; size={len(self.meta_list)}"
             )
         claim_group_ids = [self._group_ids[i] for i in claim_idxs]
-        async with self._data_plane_checkpoint_barrier.mutation() as cut:
+        async with self._data_plane_checkpoint_barrier.mutation(
+            "group_removals"
+        ) as cut:
             return await self._remove_groups_unlocked(
                 cut,
                 claim_group_ids,

nemo_rl/algorithms/distillation.py

@@ -1154,7 +1154,12 @@ def distillation_train(
                 metrics["global_valid_toks"] / total_time / total_num_gpus
             )
             logger.log_metrics(metrics, total_steps + 1, prefix="train")
-            logger.log_metrics(timing_metrics, total_steps + 1, prefix="timing/train")
+            logger.log_metrics(
+                timing_metrics,
+                total_steps + 1,
+                prefix="timing/train",
+                step_finished=True,
+            )
 
             timer.reset()
             current_step += 1

nemo_rl/algorithms/dpo.py

@@ -889,7 +889,12 @@ def dpo_train(
                 metrics["global_valid_toks"] / total_time / total_num_gpus
             )
             logger.log_metrics(metrics, total_steps + 1, prefix="train")
-            logger.log_metrics(timing_metrics, total_steps + 1, prefix="timing/train")
+            logger.log_metrics(
+                timing_metrics,
+                total_steps + 1,
+                prefix="timing/train",
+                step_finished=True,
+            )
 
             timer.reset()
             current_step += 1

nemo_rl/algorithms/rm.py

@@ -747,7 +747,12 @@ def rm_train(
                 metrics["global_valid_toks"] / total_time / total_num_gpus
             )
             logger.log_metrics(metrics, total_steps + 1, prefix="train")
-            logger.log_metrics(timing_metrics, total_steps + 1, prefix="timing/train")
+            logger.log_metrics(
+                timing_metrics,
+                total_steps + 1,
+                prefix="timing/train",
+                step_finished=True,
+            )
 
             timer.reset()
             current_step += 1

nemo_rl/algorithms/sft.py

@@ -795,7 +795,12 @@ def sft_train(
             else:
                 timing_metrics["valid_tokens_per_sec_per_gpu"] = 0.0
             logger.log_metrics(metrics, total_steps + 1, prefix="train")
-            logger.log_metrics(timing_metrics, total_steps + 1, prefix="timing/train")
+            logger.log_metrics(
+                timing_metrics,
+                total_steps + 1,
+                prefix="timing/train",
+                step_finished=True,
+            )
 
             timer.reset()
             current_step += 1