Scheduler deadlocks in pipe write during task dispatch (LocalExecutor) — runtime variant of #67881

### Under which category would you file this issue?

Airflow Core

### Apache Airflow version

3.3.0

### What happened and how to reproduce it?

**Issue Description**

The scheduler permanently deadlocks during **normal task dispatch** (not shutdown), blocked in a pipe `write()` to the LocalExecutor's activity queue. All dispatch halts; the SchedulerJob heartbeat stops while the systemd service stays "active". ~10 minutes later, every running task's supervisor kills its task ("Server indicated the task shouldn't be running anymore"), burning retry attempts.

This appears to be the **runtime/dispatch-path variant of the shutdown deadlock fixed in #67881** (merged 2026-07-02, not yet released — 3.3.0 still ships the pre-fix `end()` and empty `terminate()`).

Occurred **4 times in ~14 hours** on one production host once dispatch volume rose (~150–250 concurrent tasks). Identical signature each time.

**Evidence captured mid-deadlock:**
- `/proc//stack` → blocked in `pipe_write`
- `/proc//syscall` → `write(fd=9)`; fd 9 → `pipe:[]`
- Scanning `/proc/*/fd`: the **same pipe inode held (read fd 7 + write fd 9) by the scheduler and every LocalExecutor worker process**
- Host healthy: 100+ GB free RAM, idle CPU, no OOM, clean kernel log
- One occurrence happened with **zero** api-server errors/timeouts, ruling out API-side causes

**Steps to reproduce:**

Race condition — probabilistic and load-correlated, so not minimally deterministic, but reliably triggered in our environment within minutes-to-hours:

1. LocalExecutor, `parallelism = 256`, default `max_tis_per_query` (512), PostgreSQL backend
2. Generate dispatch bursts of 150–250 mostly lightweight, long-running tasks (ours poll remote Spark jobs), with tasks completing (returning results) concurrently with new dispatch
3. Scheduler eventually blocks forever in `pipe_write`; heartbeat stops

Notably there is no clean load threshold: the same host survived a 30-minute soak at 256 running tasks, then deadlocked at ~40 running during a later dispatch wave — consistent with a timing race on the shared queues, not resource exhaustion.

Suggested synthetic repro: a DAG with ~300 mapped `time.sleep(300)` tasks, cleared repeatedly to force dispatch bursts while earlier tasks are completing.

**Mitigations tried (none sufficient):**

| Change | Result |
|---|---|
| `[api] workers` 4 → 12 | fixed unrelated api saturation; deadlock recurred with 0 timeouts |
| `default_pool` 300 → 150 | reduced frequency; recurred |
| `[scheduler] max_tis_per_query` 512 → 32 | reduced frequency; recurred |
| Pausing the high-volume DAGs | stable (current workaround, at large throughput cost) |

### What you think should happen instead?

The scheduler should never block indefinitely on executor queue writes — a full result/activity queue should trigger draining (or a bounded/non-blocking put with retry), so a slow consumer can't freeze the entire scheduling loop.

**Suspected mechanism** — same as #67881 describes, but hit from the dispatch side:

1. Workers fill `result_queue`'s OS pipe buffer (~64KB) and block on `put()`
2. Blocked workers stop draining `activity_queue`
3. Scheduler blocks in `_process_workloads()` → `self.activity_queue.put(workload)` (`airflow/executors/local_executor.py`)
4. Being single-threaded at that point, the scheduler never reaches `_read_results()`, so `result_queue` never drains → circular wait, permanent

#67881 adds result-queue draining during `end()`/join, but `_process_workloads()` still calls a blocking `put()` with no draining fallback, so the runtime deadlock remains possible on `main` as well.

### Operating System

Amazon Linux 2023

### Deployment

Virtualenv installation

### Apache Airflow Provider(s)

_No response_

### Versions of Apache Airflow Providers

n/a — core issue (apache-airflow-task-sdk 1.3.0 for reference)

### Official Helm Chart version

Not Applicable

### Kubernetes Version

_No response_

### Helm Chart configuration

_No response_

### Docker Image customizations

_No response_

### Anything else?

Standalone deployment: systemd units for scheduler (LocalExecutor), api-server, and dag-processor on a single EC2 host; PostgreSQL metadata DB.

Happy to provide full `/proc` captures, journald excerpts of the supervisor mass-terminations, and timing data from all 4 occurrences on request.

Mainly using Claude to diagnose & type this report out.
- Looked myself for existing GitHub issues, and the one I linked was the closest thing I could find to ours.
- Looked myself at our Airflow server, and found running tasks randomly killed (with the
```
[2026-07-27 13:41:09] INFO - spark app is running...check in again in 120
[2026-07-27 13:43:15] INFO - spark app is running...check in again in 120
[2026-07-27 13:45:20] INFO - spark app is running...check in again in 120
[2026-07-27 13:47:24] INFO - spark app is running...check in again in 120
[2026-07-27 13:49:29] INFO - spark app is running...check in again in 120
[2026-07-27 13:50:40] ERROR - Server indicated the task shouldn't be running anymore. Terminating process detail={"reason":"not_found","message":"Task Instance not found, it may have been moved to the Task Instance History table"`)
```
message). This is what Claude diagnoses as an effect of a scheduler deadlock

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Read airflow/executors/local_executor.py, especially _process_workloads() and _read_results(), alongside the shutdown fix in #67881. Run the suggested mapped-task dispatch reproduction with LocalExecutor and PostgreSQL; done means dispatch and the scheduler heartbeat remain responsive when the result and activity queues fill.

Written by the indexing model from the issue text.