Description
Should a normal task retry after WORKER_DIED prefer a different feasible node from the immediately preceding attempt?
Today, when two equivalent nodes are available, a task whose worker exits can be retried on the same still-alive node. This is correct for a deterministic application crash, but it can repeat a node-local failure such as a broken device, corrupted local environment, or unhealthy worker runtime even when another eligible node is idle.
The desired behavior would be a soft preference, not a hard exclusion:
- Preserve the task's resources, label selectors, affinity, and fallback order.
- Within each existing scheduling tier, prefer a node other than the previous failed node.
- Fall back immediately to the original tier if no alternative is currently available.
- Apply this only to the next attempt and remember at most one node.
I am intentionally limiting the question to WORKER_DIED. For NODE_DIED, the old NodeID is already unavailable to the scheduler, so an additional exclusion is generally redundant.
The main ambiguity is that WORKER_DIED is broad: it can represent node-local trouble, but it can also be caused by deterministic user native code. Before preparing a PR, I would like feedback on whether Ray Core should:
- keep the current behavior;
- apply this soft preference by default; or
- expose an opt-in retry placement policy.
Use case
Long-running task workloads often rely on max_retries to tolerate transient worker or hardware failures. If a node remains registered but has a local fault, immediately retrying there can consume another attempt for the same reason. A one-attempt soft preference can use an already-idle peer without reducing schedulability in single-node, hard-affinity, or temporarily busy-cluster cases.
Minimal reproduction:
import os
import ray
from ray.cluster_utils import Cluster
from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
cluster = Cluster()
cluster.add_node(num_cpus=0, resources={"control": 1})
cluster.add_node(num_cpus=1, resources={"first": 1})
cluster.add_node(num_cpus=1, resources={"second": 1})
ray.init(address=cluster.address)
cluster.wait_for_nodes()
nodes = ray.nodes()
first = next(n["NodeID"] for n in nodes if n["Resources"].get("first") == 1)
second = next(n["NodeID"] for n in nodes if n["Resources"].get("second") == 1)
@ray.remote(num_cpus=0, resources={"control": 0.01})
class Recorder:
def __init__(self):
self.nodes = []
def record(self, node_id):
self.nodes.append(node_id)
return len(self.nodes)
def get(self):
return self.nodes
@ray.remote(max_retries=1)
def task(recorder):
node_id = ray.get_runtime_context().get_node_id()
if ray.get(recorder.record.remote(node_id)) == 1:
os._exit(1)
return node_id
recorder = Recorder.remote()
ray.get(task.options(
scheduling_strategy=NodeAffinitySchedulingStrategy(first, soft=True)
).remote(recorder))
print(ray.get(recorder.get.remote()))On a master build at 2b30448d6b, I observed [first, first] in 5/5 fresh three-raylet runs. The relevant scheduling and retry files are unchanged at the current master tip, c76067e02c. A prototype implementing the soft preference produced [first, second] in 10/10 runs on one host and 10/10 runs across three physical Kubernetes hosts.
The prototype also preserved these boundaries in repeated tests:
- hard node affinity: retry remained on the pinned node;
- no free capacity on the alternative node: retry fell back to the original node immediately;
- application exceptions: scheduling behavior was unchanged;
- the retry hint did not create a new process-global scheduling class.
Related but distinct issues: #39861 asks to change node-pool configuration after machine failure; #59522 concerns Ray Data/LLM actor resiliency; #48996 concerns scheduling stalls while detecting node removal.
AI assistance was used to explore the code path and validate the prototype. I will review and defend every changed line before proposing any implementation PR.
Description
Should a normal task retry after
WORKER_DIEDprefer a different feasible node from the immediately preceding attempt?Today, when two equivalent nodes are available, a task whose worker exits can be retried on the same still-alive node. This is correct for a deterministic application crash, but it can repeat a node-local failure such as a broken device, corrupted local environment, or unhealthy worker runtime even when another eligible node is idle.
The desired behavior would be a soft preference, not a hard exclusion:
I am intentionally limiting the question to
WORKER_DIED. ForNODE_DIED, the old NodeID is already unavailable to the scheduler, so an additional exclusion is generally redundant.The main ambiguity is that
WORKER_DIEDis broad: it can represent node-local trouble, but it can also be caused by deterministic user native code. Before preparing a PR, I would like feedback on whether Ray Core should:Use case
Long-running task workloads often rely on
max_retriesto tolerate transient worker or hardware failures. If a node remains registered but has a local fault, immediately retrying there can consume another attempt for the same reason. A one-attempt soft preference can use an already-idle peer without reducing schedulability in single-node, hard-affinity, or temporarily busy-cluster cases.Minimal reproduction:
On a
masterbuild at2b30448d6b, I observed[first, first]in 5/5 fresh three-raylet runs. The relevant scheduling and retry files are unchanged at the currentmastertip,c76067e02c. A prototype implementing the soft preference produced[first, second]in 10/10 runs on one host and 10/10 runs across three physical Kubernetes hosts.The prototype also preserved these boundaries in repeated tests:
Related but distinct issues: #39861 asks to change node-pool configuration after machine failure; #59522 concerns Ray Data/LLM actor resiliency; #48996 concerns scheduling stalls while detecting node removal.
AI assistance was used to explore the code path and validate the prototype. I will review and defend every changed line before proposing any implementation PR.