Add an end-to-end HA chaos harness: the scheduler's fault-tolerance paths are never tested on a real cluster

[andygrove]

Is your feature request related to a problem or challenge?

Ballista has a substantial high-availability state machine in ballista/scheduler/src/state/execution_graph.rs (update_task_status):

• failures are classified retryable / count_to_failures (ballista/core/src/error.rs)
• tasks retry up to task_max_failures, stages up to stage_max_failures
• a FetchPartitionError rolls back the running stage and resubmits the map stage
• a lost executor (heartbeat expiry → ExecutorLost) resets its running tasks and removes the shuffle output it produced, so map stages re-run

None of this is tested end to end. Every existing test of these paths fabricates TaskStatus protobuf messages by hand (the tests in execution_graph.rs, and scheduler/src/test_utils.rs with its VirtualExecutor / mock_failed_task). No test has ever driven the HA state machine from a real query on a real cluster, and no test has ever killed an executor.

That gap is not theoretical. Running these paths for real immediately surfaced three bugs, two of which mean Ballista fails queries it is designed to recover from:

• Shuffle-fetch failures lose their type, so the map-stage resubmit never fires and the job fails #2027 — shuffle-fetch failures lose their type, so the map-stage resubmit never fires
• Retryable IO errors are misclassified as non-retryable when wrapped in DataFusionError::Shared #2028 — retryable IO errors are misclassified when wrapped in DataFusionError::Shared
• Job hangs indefinitely instead of failing when all executors are lost #2029 — killing every executor hangs the job instead of failing it

Describe the solution you'd like

A test harness that launches a real multi-process cluster and injects faults deterministically, run under both AQE on and AQE off.

Proposed in #2026 as a new non-published workspace crate (chaos-testing), with no changes to any production crate:

• Fault injection via UDFs rather than the planner. chaos_fail(guard, mode, budget_dir) and chaos_delay(guard, ms) are pass-through scalar UDFs spliced into the query text, so the identical injection works with AQE on and off with zero planner wiring. mode selects which failure classification is exercised: io (retryable), exec (non-retryable), panic (caught by the executor's catch_unwind).
• Determinism without RNG. A guard predicate over fixed data selects which partitions fault; a filesystem token budget bounds how many attempts fault cluster-wide, across task retries and executor restarts.
• Real processes. Executors are spawned as OS processes (injecting the UDFs through the existing override_function_registry / override_session_builder hooks) so they can be SIGKILLed and restarted.
• Kills land at a known point — the scheduler's REST API is polled so a kill happens while a stage genuinely has running tasks, rather than after a guessed sleep.
• Correctness, not just liveness. Every recovery scenario asserts the result equals a chaos-free baseline run. "It didn't error" proves nothing: a re-run stage that drops or duplicates a partition still returns a result.

Describe alternatives you've considered

Reusing the existing ChaosExec (ballista/core/src/execution_plans/chaos_exec.rs). It is not usable as a regression harness:

• it is wired only into the AQE planner (ballista/scheduler/src/state/aqe/planner.rs), so it cannot inject faults with AQE off;
• ChaosCreatingRule wraps a randomly chosen plan node;
• the failure decision is probabilistic per partition.

Random + probabilistic + AQE-only is the opposite of what a regression test needs. (No test currently uses the ballista.testing.chaos_execution.* config keys at all.)

Separately, making ChaosExec work with the static distributed planner would still be worthwhile, but it is orthogonal to this.

An in-process cluster (extending the standalone pattern used by ballista/client/tests). Cheaper, but you cannot kill -9 an in-process executor, so the executor-loss paths — the ones that turned out to be broken — are exactly the ones it cannot reach.

Additional context

The harness currently reports 11 passed, 5 failed. The failures are the three bugs above, reproduced end to end; they are deliberately left failing rather than suppressed. Once those issues are fixed, these scenarios become the regression tests for them.