Epic: investigate and upstream Spice AI scheduler/executor reliability fixes

## Overview

The Spice AI fork of Ballista ([spiceai/datafusion-ballista](https://github.com/spiceai/datafusion-ballista)) has accumulated a set of scheduler and executor reliability fixes that are directly relevant to distributed execution stability, most notably the executor heartbeat/hang failures tracked in #2144. Phillip LeBlanc (Spice) noted on #2144 that they have fixes still to be upstreamed.

This epic tracks investigating each change, validating it against our current `main` (the fork sits on a diverged base, so nothing is directly cherry-pickable), and porting the ones we want.

## Screening results (checked against current `main`)

Most of the fork's fixes turned out to be **already present** in upstream, either ported independently or obsoleted by upstream's own evolution (the fork branches sit on an older base). Remaining real work is small.

| Fix | Status |
| --- | --- |
| Keep heartbeating while all task slots are busy | ✅ **Ported — #2159** |
| Ignore stale `TaskStatus` for reset partitions | ✅ Already upstream — append-slot `task_infos` model marks reset tasks `Failed(ResultLost)` and `update_task_info` already guards late updates (regression test `test_update_task_info_after_executor_lost`). The fork's `Option::unwrap` panic is structurally impossible here. |
| Pool shuffle-fetch clients per peer | ✅ Already upstream — `BallistaClientPool` / `ShuffleReaderExec::with_client_pool`. |
| Treat a missing partition file as an empty partition | ✅ Obsolete — the sort-shuffle writer now writes one consolidated `data.arrow` per input partition with a per-partition offset index and a leading schema header, so every partition (including 0-row) resolves; there is no missing-file case. The fork's fix targeted the old per-output-partition lazy-file writer that upstream removed (6540d085). |
| External semaphore for executor task concurrency | ✅ Already upstream — `poll_loop` takes `free_vcores: Option>`. |
| Broadcast channel for job state events | ✅ Already upstream (equivalent) — `ClusterState::job_state_events()` / `JobStateEvent` stream. |
| Persist terminal job status before removing from active cache | ✅ Already upstream — `persist_terminal_and_evict` updates the cached status, saves, then evicts. |
| **Periodic reconciliation sweep for pull-based stage revival** | ❌ **Not present — real gap.** Only `expire_dead_executors` runs periodically; there is no periodic revive/`update_job` sweep. Needs a careful port (new behavioral safety net; confirm the event-driven revive path can still wedge under the current scheduler first). |
| `poll_now_notify` + `on_work_available` callback | ❌ Not present — responsiveness *feature*, not a #2144 root cause. Optional. |
| Serialize execution graphs for cross-scheduler recovery | ❌ Not present — HA *feature*, not a #2144 root cause. Optional. |

## Remaining work

- [ ] **Periodic reconciliation sweep for pull-based stage revival.** Pull-based scheduling resolves downstream stages only via the event-driven path (no periodic revive sweep like push mode), so a single lost/raced revival can wedge a job forever (Running, no available tasks). Port a periodic background sweep that re-runs `update_job` on running jobs (idempotent no-op in the common case). Confirm the wedge is still reachable on current `main` before adding the sweep. (Fork: `fix(scheduler): periodic reconciliation sweep for pull-based stage revival`)
- [ ] **(Optional) `poll_now_notify` + `on_work_available`.** Wake an idle executor immediately via `Arc` instead of waiting for the next poll interval. Reduces scheduling latency; not a correctness fix. (Fork: `upstream/poll-now-notify-on-work-available`)
- [ ] **(Optional) Serialize execution graphs for cross-scheduler recovery.** Persist a job's graph to shared storage so another scheduler can `recover_job` and resume it. HA feature. (Fork: `feat(scheduler): serialize execution graphs for cross-scheduler recovery`)

## Investigate further (lower priority / experimental)

- [ ] In-memory shuffle, broadcast-join enablement, and "increase task parallelism to total available slots" (Peasee WIP branches) — evaluate whether any are worth pursuing independently.
- [ ] Diagnostics from `phillip/diag-pollwork-latency` (heartbeat send-gap vs call-duration, scheduler handler latency) — useful to port temporarily while validating the fixes above. Their notes report that CPU starvation was ruled out as the cause, consistent with the slot-acquire block (now fixed in #2159) being the real culprit.

## Notes

- The fork's `main` is behind apache `main`; the fixes live on feature branches on a diverged base, so each needs to be re-derived against current upstream rather than cherry-picked.
- Each ported change should land as its own focused PR with the reliability rationale preserved from the fork commit messages.
- Credit to Phillip LeBlanc and the Spice AI team for the original fixes.

Related: #2144

Start with related issue #2144 and verify whether a pull-based revival wedge is still reachable on current main. Compare the event-driven revival path with the periodic expire_dead_executors flow, then inspect how update_job could support reconciliation. Done means a focused, tested reliability change, with optional responsiveness and recovery ideas kept separate.

Written by the indexing model from the issue text.