[Feature] Add fault tolerance framework (simplified) for DP+EP external LB deployments

[fangyuchu]

Purpose

Add fault tolerance (FT) framework for DP+EP (Data Parallelism + Expert Parallelism) MoE deployments. When one DP rank dies, the EP all2all operation on surviving ranks blocks indefinitely, causing a full cluster hang. This framework detects faults, aborts in-flight requests, and allows an external orchestrator to trigger coordinated recovery via a REST API.

Key design:

• Sentinel pattern: All FT state lives in dedicated EngineCoreSentinel and WorkerSentinel objects -- EngineCore and Worker hold only a reference
• Externally-driven recovery: Orchestrator can send POST /fault_tolerance/apply with {"instruction": "xxx"} for different recovery policies
• External LB mode only: FT targets the external load balancer topology. Client-side FT logic is minimal -- just forwarding instructions and reporting status
• No extra control flow: FT commands reuse existing call_utility_async (client -> engine) and collective_rpc (engine -> workers) mechanisms. No new communication channels or threads are introduced
• Minimal hot-path intrusion: The fault_tolerant_wrapper decorator on the busy loop is the only structural integration point

This is a simplified version -- redundant abstractions, unused config fields, and unnecessary manual state resets have been removed compared to earlier iterations.

Details

1. External LB mode only. Per community feedback, external LB mode is the recommended topology for production deployments. This PR targets FT exclusively for external LB mode (--data-parallel-external-lb / --data-parallel-rank), which also significantly reduces implementation complexity.

2. Fault detection prerequisites. FT relies on each engine being able to detect peer failures instead of hanging indefinitely. This requires: (a) an FT-capable all2all backend (nixl_ep or deepep_low_latency) that supports timeout + rank masking, and (b) a configured Gloo timeout (--cpu-distributed-timeout-seconds) for the CPU allreduce used in DP batch synchronization.

3. Status semantics during fault propagation. After a DP rank fails, all other ranks will eventually raise exceptions too (via Gloo allreduce timeout and/or all2all kernel timeout). If the status API shows some ranks as unhealthy but others still as healthy, the "healthy" ranks are most likely still waiting inside a timeout window (e.g., the Gloo CPU group timeout or the all2all backend's kernel timeout) and have not yet raised their exception. The orchestrator should wait for all ranks to become unhealthy before issuing a recovery command.

4. Retry recovery for transient faults. The current implementation supports retry recovery for transiently recoverable faults such as network blips. On retry, the framework reinitializes the DP process group, cleans worker state (input batch, model runner state), and resets RDMA buffers and mask state in the all2all backend.

Test Plan

Model: Qwen3-30B-A3B (MoE, 128 experts)
Configuration: DP=4, TP=1, EP=4, deepep_low_latency backend

Launching the cluster

Each DP rank runs as a separate vllm serve process. The command below shows rank 0 -- launch one such process per GPU, incrementing CUDA_VISIBLE_DEVICES, --data-parallel-rank, and --port for each rank:

CUDA_VISIBLE_DEVICES=0 python -m vllm.entrypoints.cli.main serve \
    Qwen/Qwen3-30B-A3B \
    --data-parallel-size 4 \
    --data-parallel-size-local 1 \
    --data-parallel-rank 0 \
    --port 8100 \
    --enable-expert-parallel \
    --all2all-backend deepep_low_latency \
    --gpu-memory-utilization 0.5 \
    --max-model-len 1024 \
    --trust-remote-code \
    --cpu-distributed-timeout-seconds 3 \
    --enable-fault-tolerance \
    --fault-tolerance-config '{"engine_recovery_timeout_sec": 500}'

Simulating faults

To simulate a transient device-side failure, inject a RuntimeError in the DP allreduce path (_run_ar in vllm/v1/worker/dp_utils.py). The patch adds a step counter and raises an exception at specific steps (250 and 5000) on a target rank (rank 1), after the allreduce completes:

# After dist.all_reduce(tensor, group=group):
if (step_count in [250, 5000] and dp_rank == 1):
    raise RuntimeError("FAULT INJECTION: exception after allreduce")

Sending inference requests

# Send a chat completion request to rank 0 (repeat for each rank's port: 8100, 8101, ...):
curl -s http://localhost:8100/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model": "Qwen/Qwen3-30B-A3B", "messages": [{"role": "user", "content": "Hello"}], "max_tokens": 64}'

Querying FT status and issuing recovery

# Query engine status (repeat for each rank's port):
curl -s http://localhost:8100/fault_tolerance/status

# Issue retry to all ranks after all become unhealthy:
curl -s -X POST http://localhost:8100/fault_tolerance/apply \
  -H "Content-Type: application/json" \
  -d '{"instruction": "retry", "params": {"timeout": 120}}'

Test Results

Run log:
test_inject_fault_and_retry.log

Tested with DP=4, EP=4, deepep_low_latency backend. Faults were injected at step 250 and step 5000 on DP rank 1. Both fault-recovery cycles completed successfully.

Fault Injection # 1 -- Step 250

Time	Event
16:49:15	All 4 engines fully started. FT routes (/fault_tolerance/apply, /fault_tolerance/status) registered
16:52:40	Fault injected on R1 -- RuntimeError raised after allreduce at step 250
16:52:40	R1 EngineCore catches exception, R1 -> UNHEALTHY
16:52:43	R0, R2, R3 workers hit Gloo allreduce timeout (3s) on the next DP-sync batch (async scheduling overlaps host and device): Timed out waiting 3000ms for recv operation
16:54:21	R0, R2, R3 hit EP all2all kernel timeout -- fault mask: [0, 1, 0, 0] (rank 1 faulted)
16:54:21	R0, R2, R3 -> UNHEALTHY. All 4 engines now unhealthy
16:54:21	GET /fault_tolerance/status confirms all 4 engines are unhealthy
16:54:44	POST /fault_tolerance/apply (retry) sent to all 4 engines. All 4 -> HEALTHY
16:54:45	Normal serving resumes

Key log lines:

[R1] ERROR 16:52:40 [dp_utils.py:67] FAULT INJECTION: exception after allreduce step 250 rank 1
[R1] WARNING 16:52:40 [engine_core_sentinel.py:72] [FT] Busy loop raised KeyError. Waiting for recovery.
[R1] INFO 16:52:40 [engine_core_sentinel.py:83] [FT] Engine 1 status -> UNHEALTHY

[R0] ERROR 16:52:43 RuntimeError: Timed out waiting 3000ms for recv operation to complete

[R0] ERROR 16:54:21 RuntimeError: Fault detected in EP all2all communication:
  one or more ranks timed out during dispatch/combine. Mask: [0, 1, 0, 0]
[R0] WARNING 16:54:21 [engine_core_sentinel.py:72] [FT] Busy loop raised RuntimeError. Waiting for recovery.
[R0] INFO 16:54:21 [engine_core_sentinel.py:83] [FT] Engine 0 status -> UNHEALTHY

[R0] INFO 16:54:44 [engine_core_sentinel.py:111] [FT] Engine 0 status -> HEALTHY
[R1] INFO 16:54:44 [engine_core_sentinel.py:111] [FT] Engine 1 status -> HEALTHY
[R2] INFO 16:54:44 [engine_core_sentinel.py:111] [FT] Engine 2 status -> HEALTHY
[R3] INFO 16:54:44 [engine_core_sentinel.py:111] [FT] Engine 3 status -> HEALTHY

Fault Injection # 2 -- Step 5000

Time	Event
16:54:55	Normal serving continues after first recovery
16:58:48	Fault injected on R1 -- RuntimeError at step 5000
16:58:48	R1 -> UNHEALTHY
16:58:51	R0, R2, R3 hit Gloo allreduce timeout (3s)
17:00:29	R0, R2, R3 hit EP all2all kernel timeout -- mask: [0, 1, 0, 0]
17:00:29	R0, R2, R3 -> UNHEALTHY. All 4 engines unhealthy
17:03:51	POST /fault_tolerance/apply (retry) sent. All 4 -> HEALTHY
17:03:55	Serving resumes. Throughput ramps back to ~19 tok/s within seconds
17:04:55	Steady-state serving confirmed -- all 4 engines at ~19.4 tok/s, all requests returning 200

Summary

• Both fault injection cycles followed the same detection -> status query -> retry -> resume pattern
• The faulted rank (R1) was correctly identified via the all2all fault mask [0, 1, 0, 0]
• Healthy ranks detected the fault via the EP all2all kernel timeout and entered the recovery flow
• After each retry, all 4 engines returned to HEALTHY and inference throughput was fully restored
• The service remained stable after the second recovery with no secondary crashes

---

Essential Elements of an Effective PR Description Checklist

• The purpose of the PR, such as "Fix some issue (link existing issues this PR will resolve)".
• The test plan, such as providing test command.
• The test results, such as pasting the results comparison before and after, or e2e results
• (Optional) The necessary documentation update, such as updating supported_models.md and examples for a new model.

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @fangyuchu.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @fangyuchu.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork