fix(scheduler): fail running jobs when all executors are lost

[akshaychitneni]

When every executor dies mid-query, the scheduler reset the affected tasks but had nothing to schedule them onto, leaving the job Running forever and hanging the client.

On ExecutorLost, if no live executors remain, wait a bounded, configurable grace period (no_executors_grace_period_seconds, default 30s) for an executor to (re)register, then fail any still-running job with a clear error. The check is scoped to the ExecutorLost path, so jobs merely queued waiting for their first executor (autoscaling cold start) are unaffected.

Adds unit tests for both total loss (job fails) and partial loss (job survives while an executor remains).

Which issue does this PR close?

Closes #. #2029

Rationale for this change

When all executors die mid-query, the scheduler leaves the job Running forever and the client hangs.

What changes are included in this PR?

On ExecutorLost with no live executors left, wait a bounded, configurable grace period (no_executors_grace_period_seconds, default 30s) for an executor to re-register, then
fail any still-running job.

Are there any user-facing changes?

New scheduler config no_executors_grace_period_seconds (default 30s); a total executor loss now fails jobs with a clear error instead of hanging.

[andygrove]

Thanks for picking this up — the diagnosis is right and the shape of the fix (bounded wait, scoped to the ExecutorLost path, config-gated, re-check before failing) is the shape I'd want.

Disclosure: this review was produced with LLM assistance (Claude Code). Code references were checked against the tree; the Scenario G timing analysis below is derived from reading the code, not from an executed run.

First: there's already a regression test for this, and it's waiting on you

#2026 added an HA chaos harness whose whole purpose was to drive these fault-tolerance paths on a real multi-process cluster, and #2029 is one of the three bugs it surfaced. It ships a dedicated reproducer:

chaos-testing/tests/ha.rs → Scenario G, killing_every_executor_terminates_the_job, #[case::aqe_off] / #[case::aqe_on], currently:

#[ignore = "reproduces #2029: killing every executor hangs the job instead of failing it"]

chaos-testing/README.md (Finding 3) says explicitly: "Un-ignore it as the regression test when #2029 is fixed." This PR is that fix, so please enable it here rather than leaving the harness carrying a stale #[ignore].

By my reading it should pass as-is, with no assertion changes:

• ChaosRun::start(aqe, 2) uses executor_timeout_seconds: 5 / expire_dead_executor_interval_seconds: 1, so both SIGKILLed executors are reaped and ExecutorLost posted at ~6s.
• chaos-testing/src/bin/chaos-scheduler.rs builds SchedulerConfig { .., ..Default::default() }, so it inherits the new 30s default → JobRunningFailed at ~36s → client errors out, comfortably inside the scenario's 120s budget.

Suggested follow-through:

1. Remove both #[ignore]s, and update the Scenario G doc comment (it currently reads "Both cases reproduce Job hangs indefinitely instead of failing when all executors are lost #2029") plus the README table row and Finding 3.
2. Plumb the new knob into the harness — CHAOS_NO_EXECUTORS_GRACE_SECONDS in chaos-scheduler.rs and a builder on TestCluster, mirroring the existing executor_timeout_seconds knob and its "defaults would make every executor-kill scenario take three minutes" comment. With a ~1s grace, G costs ~7s instead of ~37s. That matters: rust.yml runs cargo test workspace-wide, so an un-ignored G lands on every PR's critical path.
3. Tighten the assertion. The comment calls it "deliberately weak: this is a hang detector". Now that termination is achievable, assert the outcome is an Err whose message names executor loss — that's the contract this PR advertises, and it's worth locking in.
4. Please double-check the aqe_on case specifically. AQE keeps its own status/graph bookkeeping (state/aqe/mod.rs:181,265); the new guard sits on the shared ExecutorLost path so it should apply, but that's the case I'd least want to assume.
5. The branch may need a rebase — test: add HA chaos harness that drives the scheduler's fault-tolerance paths on a real cluster #2026 merged very recently, so chaos-testing/ may not be on the head branch at all yet.

Worth being explicit about what Scenario G does not cover, because it's blind to finding 1 below: it kills two executors and asserts only that the job terminates, so it passes even with duplicate failure events. It also never exercises the recovery path (executor returns inside the grace window), which is the actual new behaviour being introduced.

Review findings

1. Duplicate grace timers → N JobRunningFailed per job

This misfires in exactly the scenario the PR targets. When the whole cluster dies at once, expire_dead_executors (scheduler_server/mod.rs:326) returns all stale heartbeats in a single pass and calls remove_executor for each. Meanwhile get_alive_executors (executor_manager.rs:466) already filters on executor_timeout_seconds staleness, so by the time the first ExecutorLost is handled every other dying executor is already "not alive". Each of the N ExecutorLost events therefore sees an empty set and arms its own grace timer, and each timer posts JobRunningFailed for every running job.

Per job that means: record_failed N times (inflated failure metrics), abort_job logging "Fail to find job {job_id} in the cache, unable to cancel tasks" for every attempt after the first, and clean_up_failed_job running N times with N delayed state cleanups queued. Nothing dedupes.

Cleanest fix is to stop deciding inside the spawned task and let the event loop decide — which also removes finding 3:

// in the ExecutorLost arm
if self.state.executor_manager.get_alive_executors().is_empty()
    && !self.no_executor_check_pending.swap(true, Ordering::SeqCst)
{
    let sender = event_sender.clone();
    let grace = Duration::from_secs(self.state.config.no_executors_grace_period_seconds);
    let lost_at = timestamp_millis();
    tokio::spawn(async move {
        tokio::time::sleep(grace).await;
        let _ = sender.post_event(QueryStageSchedulerEvent::NoExecutorsCheck { lost_at }).await;
    });
}

…with a new arm that clears the flag and does the alive-check plus fail loop inline. A plain AtomicBool on QueryStageScheduler is enough if a full event variant feels heavy.

2. Jobs submitted during the grace window get a truncated grace and a misleading message

A job reaches Status::Running as soon as planning succeeds (execution_graph.rs:365) — executor availability doesn't enter into it. So the comment's claim that "jobs merely queued waiting for their first executor (autoscaling cold start) are never affected" only holds for jobs submitted before the timer was armed. A job submitted at t=29s of a 30s window is failed ~1s later, told that no executor re-registered within 30s. That's both a wrong message and the opposite of the autoscaling tolerance the PR is aiming for.

Capturing lost_at (as above) and skipping jobs whose running.started_at > lost_at handles this precisely.

3. Snapshot-then-post race (low severity, but grace = 0 is a documented setting)

The task snapshots the running-job cache and then awaits a post_event per job. With grace = 0, sleep(0) merely yields, so a JobFinished already queued behind the ExecutorLost can be processed after the snapshot but before the JobRunningFailed lands. The job then takes record_failed and clean_up_failed_job → clean_up_job_data despite having succeeded. Moving the check into the event loop makes this impossible; otherwise re-verify job_info.status is still Running immediately before posting.

4. Test coverage

• Both new tests use with_no_executors_grace_period_seconds(0), so the grace behaviour — the core of the new logic — is never exercised. The highest-value missing test: grace > 0, lose the only executor, re-register one inside the window, assert the job is not failed. #[tokio::test(start_paused = true)] keeps it fast.
• Nothing asserts the failure reason reaches the client. Asserting the FailedJob message mentions executor loss locks in the user-visible half of this fix.
• Finding 1 is unit-testable and cheap: TestMetricsCollector::job_events (test_utils.rs:716) lets you assert exactly one MetricEvent::Failed per job after losing N executors at once.
• test_running_job_survives_partial_executor_loss leans on a fixed tokio::time::sleep(500ms); a paused clock is both faster and less of a timing guess.
• "Wait until the job is actually running with tasks in flight" over-claims — Running is set at planning time, so the condition can be satisfied before any task is launched.

5. Nits

• _ => timestamp_millis() for queued_at is unreachable: get_running_job_cache already filters to Running (task_manager.rs:398). An else { continue } reads better. (scheduler_server/mod.rs:222 does the same thing, so this is cosmetic.)
• for pair in ... { let (job_id, job_info) = pair; } → for (job_id, job_info) in ....
• fail_message starts with "Job {job_id} failed: ", but the handler already logs "Job failed: [{job_id}]" and other producers (execution_graph.rs:430) don't prefix the id — dropping it reads better in the client-facing error.
• Comment convention in this tree leans toward the full URL (// See https://github.com/apache/datafusion-ballista/issues/2029) over a bare #2029.
• Naming: the neighbouring field is executor_termination_grace_period. Something like executor_loss_grace_period_seconds sits closer to both the trigger and existing naming, though no_executors_... is fine if you prefer it.
• The spawned timer isn't tied to scheduler shutdown, so a stopping scheduler can hold a task for up to grace. Harmless, just worth knowing.

No security concerns, and the performance cost is negligible (one get_alive_executors() snapshot per ExecutorLost).

Overall: worth landing. I'd want finding 1 fixed before merge since it fires on the primary scenario, finding 2 is a small addition on the same change, and I'd like either the grace-period recovery unit test or un-ignored Scenario G (ideally both) so the new config isn't shipping untested.

[andygrove]

Thanks for the contribution @akshaychitneni. Sorry for the long AI review above. I would start with enabling the existing test and see if that passes. We can address other points in follow on PRs potentially. Let me know what you think.

[villebro]

Minor design related question: Is there a strong reason to distinguish between a cluster that has not registered its first executor and one that has lost all registered executors?

From the perspective of a running job, both cases appear equivalent: there are no executors available and the job cannot make progress. Would it be cleaner to apply the same configurable grace period in both cases, rather than allowing jobs submitted before the first executor appears to wait indefinitely?

This would also bound failures caused by executor startup or configuration problems.

[akshaychitneni]

I would start with enabling the existing test and see if that passes.

Sure. I will work in integrating with test harness.

[akshaychitneni]

Is there a strong reason to distinguish between a cluster that has not registered its first executor and one that has lost all registered executors?

This is a good callout. From a running job's view they are equivalent. I would keep them separate for now because "lost all executors" has an ExecutorLost event to trigger on, while "no first executor yet" has no event and would need a different mechanism (like a startup timeout). This is also already mitigated on k8s by the existing /readyz probe, which keeps an executor-less scheduler out of the k8s service and not accept jobs from clients