aunderwood24 opened a new issue, #72061:
URL: https://github.com/apache/airflow/issues/72061
### Apache Airflow version
3.3.1
### What happened and how to reproduce it?
At Patreon, we run Airflow 3.3.1 with a MySQL 8.0 metadata DB. After our 3.x
cutover we started losing `task_instance` rows mid-run: a task shows "no
status" in the grid, and the run either fails with "Task deadlock (no runnable
tasks)" and zero failed tasks, or (worse imo) completes as `success` with that
task's work never executed. Nothing logs the deletion. In one 48h window we
counted 109 affected runs across 55 dags before we understood what was
happening.
The mechanism is in `Trigger.clean_unused()`
(`airflow-core/src/airflow/models/trigger.py`). On MySQL it deletes
unreferenced triggers in two statements: SELECT the candidate ids into a Python
list, then DELETE by id. If a task defers between those two statements, its
trigger is already on the kill list, the DELETE fires anyway, and the `ON
DELETE CASCADE` on `task_instance.trigger_id` takes the task instance row with
it. `clean_unused()` runs on every tick of the triggerer loop, so with steady
deferral traffic this races constantly. We're running #68244's SKIP LOCKED
change (it shipped in 3.3.1) — it locks the SELECT, but nothing can protect a
list that's already been materialized.
Reproducer: 200 tasks that each defer 4 times on short `DateTimeTrigger`s,
so deferral commits land on as many cleanup ticks as possible.
<details>
<summary>repro dag</summary>
```python
from datetime import datetime, timedelta
from airflow.providers.standard.triggers.temporal import DateTimeTrigger
from airflow.sdk import DAG, BaseOperator
from airflow.sdk.timezone import utcnow
class RedeferOperator(BaseOperator):
def __init__(self, *, defers, interval_seconds, initial_delay_seconds,
**kwargs):
super().__init__(**kwargs)
self.defers = defers
self.interval_seconds = interval_seconds
self.initial_delay_seconds = initial_delay_seconds
def execute(self, context):
self._defer_again(self.defers, self.initial_delay_seconds)
def _defer_again(self, remaining, delay_seconds):
self.defer(
trigger=DateTimeTrigger(moment=utcnow() +
timedelta(seconds=delay_seconds)),
method_name="resume_deferral",
kwargs={"remaining": remaining - 1},
)
def resume_deferral(self, context, event=None, remaining=0):
if remaining > 0:
self._defer_again(remaining, self.interval_seconds)
with DAG(
dag_id="ti_vanish_repro",
start_date=datetime(2026, 8, 24),
schedule="*/5 * * * *",
catchup=False,
max_active_runs=1,
max_active_tasks=64,
) as dag:
for i in range(200):
RedeferOperator(
task_id=f"defer_{i:03d}",
defers=4,
interval_seconds=5,
# stagger so deferrals land on as many clean_unused ticks as
possible
initial_delay_seconds=1 + (i % 20),
)
```
Detection (201 tasks would be with a join task; here any run under 200 rows
lost some):
```sql
SELECT run_id, COUNT(*) FROM task_instance
WHERE dag_id='ti_vanish_repro' GROUP BY run_id HAVING COUNT(*) < 200;
```
</details>
Results, same harness across three configs:
- stock 3.3.1 + MySQL 8.0 (plain docker compose): lost 3 rows within the
first run (~800 defer events)
- same versions, Postgres backend: 0 rows lost over 4 runs
- MySQL with the DELETE re-checking references (PR to follow): 0 rows lost
over 21,000+ defer events
### What you think should happen instead?
The DELETE should re-check the reference predicates at delete time. The
two-step exists because of MySQL error 1093 (#38663), but 1093 only forbids
subqueries on the *target* table — correlated NOT EXISTS subqueries against
`task_instance` / `asset` / `callback` are legal inside the DELETE. That gives
the MySQL branch the same atomicity every other dialect already gets from the
single-statement path. I have a fix + regression test ready, PR incoming.
related: #68243 (proposes batching this same DELETE — batching should keep
this guard or it reintroduces the race), #71540 (describes the stranded-run
symptom you see after a TI row disappears mid-run)
### Operating System
Debian 12 (bookworm), official images
### Deployment
Official Apache Airflow Helm Chart / official image on Kubernetes
(KubernetesExecutor); also reproduces with LocalExecutor via docker compose
### Anything else?
Happy to share more of the forensics from production if useful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
