[data][llm] Exit actor on EngineDeadError to enable recovery

[nrghosh]

Why are these changes needed?

When EngineDeadError occurs, the vLLM engine subprocess is dead but the Ray actor process is still alive. Previously, we re-raised the exception, but this causes task retries to go to the SAME actor (actor methods are bound to specific instances), creating an infinite retry loop on the broken actor.

The Problem

vLLM engine subprocess crashes
       ↓
EngineDeadError raised
       ↓
Exception re-raised (actor stays ALIVE)
       ↓
Ray: actor_task_retry_on_errors triggers retry
       ↓
Retry goes to SAME actor (actor methods are bound)
       ↓
Same actor, engine still dead → EngineDeadError
       ↓
Infinite loop (with max_task_retries=-1)

The Fix

Call os._exit(1) to exit the actor. This triggers Ray to:

1. Mark the actor as RESTARTING
2. Create a replacement actor with a fresh vLLM engine
3. Route task retries to healthy actors (Ray Data excludes RESTARTING actors from dispatch)

This leverages Ray Data's existing fault tolerance infrastructure:

• max_restarts=-1 (default) enables actor replacement
• max_task_retries=-1 (default) enables task retry

Why os._exit(1) instead of ray.actor.exit_actor()?

We must use os._exit(1) rather than ray.actor.exit_actor() because they produce different exit types with different retry behavior:

Exit Method	Exit Type	Exception Raised	Retried?
os._exit(1)	SYSTEM_ERROR	RaySystemError	Yes
ray.actor.exit_actor()	INTENDED_USER_EXIT	ActorDiedError	No

The root cause is that Ray Data only adds RaySystemError to its retry_exceptions list (in _add_system_error_to_retry_exceptions()). Since ActorDiedError is NOT a subclass of RaySystemError (they're siblings in the exception hierarchy), tasks that fail due to ray.actor.exit_actor() are not retried.

The semantic gap: Ray currently lacks a "fatal application error" concept - an error where the actor should be restarted AND pending tasks retried. The available options are:

• Clean exit (exit_actor) = "I'm intentionally done" → no retry
• Crash (os._exit) = "Something broke unexpectedly" → retry

We need the "crash" semantics even though this is a deliberate decision, so os._exit(1) is the correct workaround until Ray Core adds explicit support for fatal application errors.

See: #60150

cc @goutamvenkat-anyscale

Validation

We created a minimal reproduction script demonstrating:

1. The problem: All retries go to the same broken actor (same PID)
2. The fix: Actor exits → replacement created → job succeeds with multiple PIDs

# Demo output showing fix works:
[SUCCESS] Processed 20 rows!
PIDs that processed batches: {708593, 708820}
-> Multiple PIDs = replacement actor joined and processed work

Full reproduction: https://gist.github.com/nrghosh/c18e514a975144a238511012774bab8b

Related issue number

Fixes #59522

Checks

• I've signed off every commit
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
• I've added any new APIs to the API Reference. For example, if I added a
  temporary file handling method, I've added it in doc/source/ray-core/api/doc/ray.util.temp_files.rst.
• I've made sure the tests are passing. Note that there might be a few flaky tests.

[gemini-code-assist]

Code Review

This pull request introduces a fix for handling fatal vLLM engine errors by forcefully exiting the Ray actor, which allows Ray's fault tolerance to restart it. The approach is sound and the reasoning is well-documented in the code comments.

My main feedback is to use the idiomatic Ray API ray.actor.exit_actor() instead of the low-level os._exit(1). This improves code clarity and relies on the framework's intended mechanism for actor lifecycle management. This would also involve changing the new import os to import ray.

[nrghosh]

/gemini review

[gemini-code-assist]

Code Review

This pull request addresses a critical issue where a fatal EngineDeadError from the vLLM engine could lead to an infinite retry loop on a broken actor. The proposed solution is to explicitly terminate the actor process upon such an error, enabling Ray's fault tolerance mechanisms to restart it. The implementation is clean and effective. It correctly uses ray.actor.exit_actor() to terminate the actor, which is the recommended public API for this purpose, ensuring a graceful shutdown. The added comments clearly explain the rationale behind the change, and the logging will be helpful for debugging. The change is well-contained and directly solves the described problem. Overall, this is an excellent fix.

[kouroshHakha]

LGTM

[nrghosh]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces a critical fix to prevent infinite retry loops when the vLLM engine encounters a fatal error. By calling os._exit(1) upon an EngineDeadError, the Ray actor is terminated, allowing Ray's fault tolerance to restart it and correctly retry the task on a new, healthy actor. The explanation for using os._exit(1) over ray.actor.exit_actor() is thorough and well-reasoned. The code changes are concise and effective, and the corresponding test has been updated to correctly validate the new behavior by mocking os._exit. I have one suggestion to make the exception handling in the test more specific for improved robustness.

[gemini-code-assist]

While the comment explains why an exception is expected here, catching a broad Exception can mask other potential issues in the test. Since the code path after the mocked os._exit leads to processing a None value which causes a TypeError in the StatefulStageUDF wrapper, it would be more robust to catch the specific TypeError here. This makes the test's intent clearer and less prone to hiding unrelated errors.

Suggested change

	except Exception:
	pass # Code may fail after mock returns None - that's OK for this test
	except TypeError:
	pass # Expect TypeError when processing None from the mocked exit path