[checkpoint_engine] fail fast on hung first NCCL group init (#6967)

[aryanyadav0402]

Motivation

Fixes the silent-infinite-hang failure mode of #6967. The first
ray.util.collective rendezvous during checkpoint-engine weight sync
(NCCLCheckpointEngine.init_process_group → init_collective_group + first
barrier) can hang indefinitely on some setups — a timing race in the
Ray/NCCL layer. When it hangs, both ranks sit at 0% util with no traceback and
the whole job is stuck forever; only the first sync is affected (the group is
reused afterward).

This PR does not claim to fix the underlying race — it adds a robustness
guardrail so a stalled first sync fails fast with a clear, actionable error
instead of hanging silently forever.

Modifications

• New verl/checkpoint_engine/_group_init.py: run_group_init_with_timeout()
  runs the blocking group registration + first barrier in a worker thread
  bounded by a timeout. On timeout it raises CheckpointEngineInitError with a
  message pointing at [fully_async] First NCCL checkpoint-engine group init hangs (timing race) — reproduces single-turn without tools #6967; it re-raises any error the init itself throws.
  Timeout is configurable via VERL_CKPT_ENGINE_INIT_TIMEOUT_S
  (default 600s — generous enough never to trip a slow-but-healthy init).
• NCCLCheckpointEngine.init_process_group wraps its first-rendezvous block
  with this helper.
• A stalled NCCL/collective call cannot be safely interrupted, so this
  deliberately does not retry in place — it surfaces the error and lets the
  launcher tear the job down.

Scope (honest framing)

This is a guardrail against a silent infinite hang, not a root-cause fix for
the race. The exact mechanism is currently unconfirmed and not reproducible:
on H100 NVL I could not reproduce it (natively across staleness_threshold
1/4/16, nor under injected scheduling skew at 4 injection points with delays up
to 120s — the barrier tolerates skew via Rendezvous.meet polling + blocking
ncclCommInitRank), and reporter @chengcuiping reports it no longer reproduces
on the original 2×A100 either, with no path to timing-neutral native stacks
(container ptrace restrictions). Proposing a speculative "fix" would be
unjustified — but this timeout guardrail is independently valuable and endorsed
by the reporter in the issue thread.

Accuracy / Speed Tests

No model-output or perf impact (init-path robustness only).

Test

CPU-only unit test tests/checkpoint_engine/test_group_init_timeout_on_cpu.py
(no GPU/NCCL — exercises the timeout machinery directly):

$ python -m pytest tests/checkpoint_engine/test_group_init_timeout_on_cpu.py -v
tests/checkpoint_engine/test_group_init_timeout_on_cpu.py::test_completes_within_timeout PASSED
tests/checkpoint_engine/test_group_init_timeout_on_cpu.py::test_raises_fast_on_timeout_instead_of_hanging PASSED
tests/checkpoint_engine/test_group_init_timeout_on_cpu.py::test_reraises_init_error PASSED
tests/checkpoint_engine/test_group_init_timeout_on_cpu.py::test_env_var_sets_default_timeout PASSED
4 passed

Existing tests/checkpoint_engine/test_global_steps_on_cpu.py still passes;
ruff check and ruff format --check clean on the changed files.

Not a duplicate

Checked before opening: no open PR references #6967 or touches checkpoint-engine
init timeout.

Notes

/cc @chengcuiping — thanks for the detailed root-cause collaboration on #6967.
Could you validate on your A100 config that the guardrail doesn't false-trip on a
healthy run? Happy to add you as co-author.

AI assistance (Claude) was used in developing this change; the tests above were
run and pass, and I've reviewed the change end-to-end.

[CLAassistant]

All committers have signed the CLA.

[gemini-code-assist]

Code Review

This pull request introduces a timeout guardrail for the first rendezvous initialization of the NCCL checkpoint engine to prevent indefinite hangs (issue #6967). It wraps the collective group initialization and barrier in a background thread with a configurable timeout, failing fast with a clear error if the timeout is exceeded. CPU unit tests are also added to verify this behavior. The reviewer identified a critical thread-safety issue where the ZeroMQ socket connection is executed inside the temporary background thread. Since ZeroMQ sockets are not thread-safe, this connection should be moved to the main thread after the timeout-bounded initialization successfully completes.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Creating and connecting the ZeroMQ socket (self._connect_zmq_client) inside the background thread (_do_init) is unsafe and unnecessary. ZeroMQ contexts and sockets are not thread-safe and should not be created in a temporary thread that then exits, as this can lead to lifetime issues or undefined behavior when the socket is later accessed from other threads (e.g., in BroadcastOperation). Since _connect_zmq_client is non-blocking and does not contribute to the NCCL rendezvous hang, it should be executed in the main thread after the timeout-bounded initialization succeeds.

Suggested change

	def _do_init() -> None:
	if self.rebuild_group or not collective.is_group_initialized(self.group_name):
	collective.init_collective_group(world_size, rank, "nccl", self.group_name)
	self.rank = rank
	self.world_size = world_size
	else:
	assert self.rank == rank, f"rank {rank} is not equal to self.rank {self.rank}"
	assert self.world_size == world_size, (
	f"world_size {world_size} is not equal to self.world_size {self.world_size}"
	)
	if self.rank > 0:
	self._connect_zmq_client(master_metadata)
	collective.barrier(self.group_name)
	if self.rank > 0:
	self._connect_zmq_client(master_metadata)
	collective.barrier(self.group_name)
	# Bound the first-rendezvous init so a hung sync fails fast with a clear
	# error instead of hanging forever (verl issue #6967).
	run_group_init_with_timeout(_do_init, group_name=self.group_name)
	def _do_init() -> None:
	if self.rebuild_group or not collective.is_group_initialized(self.group_name):
	collective.init_collective_group(world_size, rank, "nccl", self.group_name)
	self.rank = rank
	self.world_size = world_size
	else:
	assert self.rank == rank, f"rank {rank} is not equal to self.rank {self.rank}"
	assert self.world_size == world_size, (
	f"world_size {world_size} is not equal to self.world_size {self.world_size}"
	)
	collective.barrier(self.group_name)
	# Bound the first-rendezvous init so a hung sync fails fast with a clear
	# error instead of hanging forever (verl issue #6967).
	run_group_init_with_timeout(_do_init, group_name=self.group_name)
	if self.rank > 0:
	self._connect_zmq_client(master_metadata)

[chengcuiping]

@aryanyadav0402 , Thanks for putting this together and for framing it explicitly as a fail-fast guardrail rather than a confirmed root-cause fix. That scope matches the evidence we currently have, and I’m happy to validate it on the original A100 configuration and to be added as a co-author.

Before I run the GPU validation, I think the current implementation needs one correction and one design check.

First, I agree with the Gemini review that _connect_zmq_client() should not run inside the temporary timeout thread. It should remain outside the timeout-bounded NCCL rendezvous block.

Second, could we consider applying the timeout at CheckpointEngineManager.build_process_group() around the existing controller-side wait for the actor and rollout init futures, rather than moving init_collective_group() and the first barrier into a daemon thread inside every worker? In the current helper, a timeout raises in the actor method but leaves the daemon thread blocked in NCCL/collective with partially initialized process state. The CPU tests demonstrate the timeout mechanism with time.sleep(), but they do not yet prove that the Ray job, actors, communicator state, and GPU resources are actually torn down after a real timeout.

A manager-side bounded wait would preserve the original actor execution context for NCCL/CuPy/ZMQ while still converting the silent infinite wait into an actionable error. It should still avoid any in-place retry.

If you prefer to keep the worker-thread approach, I think we need an end-to-end Ray-level timeout test that verifies the exception reaches the controller and that the affected actors/job are terminated rather than remaining alive with a blocked daemon thread.

After the revision, I can validate:

• several healthy A100 cold starts with no false timeout;
• a deliberately small timeout that deterministically exercises the failure path;
• complete Ray/GPU/process cleanup after that failure;
• a fresh healthy run afterward to rule out stale group or actor state.

[aryanyadav0402]

Thanks for the review — good catch on the ZeroMQ thread-safety. Fixed in da00403: _connect_zmq_client() now runs on the caller thread after the timeout-bounded init, so only the hang-prone NCCL init_collective_group + barrier stay inside the timeout worker. The SUB socket is no longer created inside the temporary thread.

[chengcuiping]

@aryanyadav0402 Following up on my earlier teardown concern, I completed the minimal 2×A100 Ray/NCCL validation against da004031296fef43ac40f3ed69cd686b079fd8c9.

Results:

• Healthy path: 3/3 fresh actor/group attempts passed, with zero false timeouts.
• Forced-timeout path: with one rank deliberately never entering the group and timeout_s=1, the driver received RayTaskError(CheckpointEngineInitError) after approximately one second. The error included the group name, configured timeout, and the reference to [fully_async] First NCCL checkpoint-engine group init hangs (timing race) — reproduces single-turn without tools #6967.
• When the observation driver was intentionally kept alive, the owner-scoped actors also remained alive and responsive. This is normal Ray actor lifetime behavior and, by itself, is not evidence of a teardown leak.
• I therefore ran a separate parent-supervisor/child-driver lifecycle test. The child re-raised the timeout error and exited with code 1. At the first post-exit sample, both actor PIDs, the Raylet, and all GPU compute PIDs had disappeared; both A100s reported 0 MiB, with no ray.kill, ray.shutdown, or other manual cleanup required.
• A subsequent fresh-runtime recovery using new actors and a new NCCL group completed init + barrier successfully.

This resolves my teardown concern for this PR head. The minimal A100 evidence supports the intended design: the helper propagates the timeout to the driver, and normal owner/launcher exit releases the Ray and GPU resources. I no longer think a separate manager-side teardown change is required for this PR.

I could not complete the full fully_async E2E validation because the historical environment is currently missing tensordict, Hydra, and OmegaConf, and I did not modify that environment. Therefore, this result should be scoped to the minimal Ray/NCCL path; it does not prove that the underlying timing race is fixed.

From my side, the fail-fast guardrail at this pinned head looks good.

[aryanyadav0402]

Thanks for the thorough 2×A100 validation — and especially for the parent/child lifecycle test showing Ray reclaims the actors + GPUs cleanly on driver exit.

I've adopted your controller-side suggestion in 0de5324: the timeout now wraps the existing ray.get(...) of the init futures inside CheckpointEngineManager.build_process_group() (wait_for_group_init() → on GetTimeoutError, raise CheckpointEngineInitError). The per-worker daemon thread is gone, so there's no thread left blocked in NCCL with partially-initialized state; it's a single enforcement point and now backend-agnostic (covers every checkpoint-engine backend, not just NCCL). The worker-side change to nccl_checkpoint_engine.py is fully reverted.

Behavior is the same fail-fast guardrail you validated, just enforced on the controller: healthy init passes through, a hung init raises the clear CheckpointEngineInitError (group/timeout/#6967 in the message) after VERL_CKPT_ENGINE_INIT_TIMEOUT_S (default 600s). CPU unit test updated to cover the new helper (mocked ray.get); ruff + format clean.

Whenever convenient, could you re-validate this revision on the A100 config? And happy to add you as co-author — if you'd like the Co-authored-by: trailer, just share the name/email you'd like used. Thanks again for the collaboration.

[chengcuiping]

@aryanyadav0402 Thanks for the update. I validated 0de5324.

The controller-side wait_for_group_init() guard is present and the updated CPU tests plus Ruff checks pass. However, verl/checkpoint_engine/nccl_checkpoint_engine.py still imports and calls run_group_init_with_timeout around NCCL init/barrier, and that file is unchanged in this revision. The checked-out tree therefore still has both worker-side and controller-side timeout mechanisms.

This does not yet match the described worker-side revert / controller-side-only implementation, so I did not run a new 2×A100 validation against the mixed implementation.

Please push a revision that removes the remaining worker-side timeout path (and any now-unused helper/tests as appropriate), then share the exact commit SHA. I will re-run the static audit followed by healthy, forced-timeout, lifecycle-cleanup, and fresh-Ray recovery checks on 2×A100.

Separately, the historical fully_async E2E environment remains blocked by missing tensordict, Hydra, and OmegaConf; I did not modify that environment.

Please push a revision that removes the remaining worker-side timeout path, then share the exact commit SHA. I will re-run the static audit followed by healthy, forced-timeout, lifecycle-cleanup, and fresh-Ray recovery checks on 2×A100.

For the Co-authored-by trailer, please use:
Co-authored-by: chengcuiping chengcuipingswu@163.com