test: add HA chaos harness that drives the scheduler's fault-tolerance paths on a real cluster

[andygrove]

Which issue does this PR close?

Closes #2030.

This harness reproduces the following, each of which is filed separately:

• 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 (fixed on main by Retryable IO errors misclassified when wrapped in DataFusionError::Shared #2119, but see the test status section below)
• Job hangs indefinitely instead of failing when all executors are lost #2029 — killing every executor hangs the job instead of failing it

Rationale for this change

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, tasks retry up to task_max_failures, a FetchPartitionError
rolls back the running stage and resubmits the map stage, and a lost executor resets its running
tasks and removes the shuffle output it produced so map stages re-run.

None of it was tested end to end. Every existing test of these paths fabricates TaskStatus
protobuf messages by hand (execution_graph.rs tests, scheduler/src/test_utils.rs). No test ever
drove the HA state machine from a real query on a real cluster, and no test ever killed an executor.

Running the state machine for real immediately surfaced three bugs, two of which mean Ballista
fails queries that it is designed to recover from. See the linked issues.

There is an existing chaos operator (ChaosExec), but it is unusable as a regression harness: it is
wired only into the AQE planner (state/aqe/planner.rs), so it cannot inject faults with AQE off; it
wraps a randomly chosen plan node; and it fires probabilistically. No test uses the
ballista.testing.chaos_execution.* keys today.

What changes are included in this PR?

A new non-published workspace crate, chaos-testing (ballista-chaos). No production crate is
modified — the only change outside chaos-testing/ is adding the member to the workspace
Cargo.toml.

• Fault injection via UDFs, not 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 AQE off with zero planner wiring. mode selects which of Ballista's
  three failure classifications is exercised: io (retryable), exec (non-retryable), panic
  (caught by the executor's catch_unwind, non-retryable).
• Determinism without RNG. The 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 (consuming a token is fs::remove_file, which is atomic across processes). Budget 1
  means exactly one attempt faults anywhere, so a retry must succeed; budget ≫ task_max_failures
  means retries must exhaust.
• A real multi-process cluster. TestCluster spawns real chaos-scheduler / chaos-executor
  binaries as OS processes (they inject the UDFs via the existing override_function_registry /
  override_session_builder hooks, following examples/custom-executor.rs), so executors can be
  SIGKILLed and restarted. It tunes executor_timeout_seconds / heartbeat down from the production
  defaults of 180s/60s, without which a killed executor would not be noticed for three minutes.
  Clusters serialize themselves via a process-wide mutex plus a machine-wide flock, and a
  registration timeout attaches every child process's log tail to the error so CI failures are
  diagnosable from test output alone.
• Kills land at a known point. The scheduler's REST API is polled (/api/job/{id}/stages) so a kill
  happens while a stage genuinely has running tasks, rather than after a guessed sleep.
• Seven scenarios, each run under both AQE settings: retryable fault recovered; retries exhausted;
  panicking task; executor killed mid-stage; executor killed after writing shuffle output; executor
  killed and restarted; every executor killed.

The load-bearing assertion in every recovery scenario is result equality against a chaos-free baseline
run, not merely "it did not error" — re-running a stage is exactly where duplicated or dropped
partitions would appear, and only a correctness check catches that.

Current test status

cargo test -p ballista-chaos → lib: 18 passed. ha scenarios: 8 passed, 8 ignored, 0 failed.

Every scenario that reproduces a known bug is #[ignore]d against the issue it reproduces, with its
original assertions intact — nothing is relaxed to manufacture a pass. Run them with
cargo test -p ballista-chaos -- --ignored to see the failures. Once the linked issues are fixed,
un-ignoring these scenarios turns them into the regression tests for those fixes. Each is documented
in chaos-testing/README.md with its root cause:

• executor_killed_mid_stage_is_recovered and executor_killed_after_shuffle_write_is_recovered
  (both AQE settings) reproduce Shuffle-fetch failures lose their type, so the map-stage resubmit never fires and the job fails #2027: the typed FetchFailed is flattened before the scheduler can
  classify it, so the map stage is never resubmitted and the job fails instead of recovering. The
  shuffle-write case failed in CI with FetchFailed buried inside DataFusionError::Shared; it only
  passes when the kill happens to land after the reduce stage already fetched its input.
• retryable_fault_is_retried_and_result_is_correct_{aqe_off,aqe_on} originally demonstrated Retryable IO errors are misclassified as non-retryable when wrapped in DataFusionError::Shared #2028
  (Shared(IoError) missing the retryability classifier's shallow match). Retryable IO errors misclassified when wrapped in DataFusionError::Shared #2119 fixed that by
  classifying on find_root() — but the sort-shuffle writer refactor (Multi-partition tasks: partition_slice plumbing + SortShuffleWriter refactor #2038/Remove the hash-partitioning shuffle writer #2106) now flattens any
  task error into DataFusionError::Execution(format!("{e:?}")) in the shuffle-write coordinator, so
  the IoError reaches the classifier as inert text and both cases fail again, this time under both
  AQE settings. This is the same type-erasing mechanism as Shuffle-fetch failures lose their type, so the map-stage resubmit never fires and the job fails #2027, and arguably makes that issue more
  urgent: every transient IO error in any stage that writes shuffle output is now non-retryable.
• killing_every_executor_terminates_the_job (both AQE settings) reproduces Job hangs indefinitely instead of failing when all executors are lost #2029: the job hangs
  instead of failing once no executor can ever satisfy the remaining tasks.

Passing: baseline correctness (both AQE settings), retry exhaustion, panicking task (job fails
cleanly, executor survives), and executor kill + restart.

Are there any user-facing changes?

No. The crate is test-only and publish = false; no production crate is touched.

[andygrove]

cc @phillipleblanc

[phillipleblanc]

This is great, looking forward to integrating this into our project's ballista testing setup.

[andygrove]

Thanks for the review @phillipleblanc. I'll go ahead and merge so that others can iterate on improving this.