apache / apache/airflow

Trigger.clean_unused() on MySQL can cascade-delete the task_instance row of a task that defers during the sweep

Open
#72,061 1 comment 0 reactions 0 assignees View on GitHub
area:core area:MetaDB kind:bug priority:high
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### 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 (we want to move to postgres and be more standard -- I promise). 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.

repro dag

```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;
```

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.

Contributor guide

Open the contributing guide

Research direction

Start with airflow-core/src/airflow/models/trigger.py and inspect Trigger.clean_unused(), then run the supplied MySQL reproducer and compare it with PostgreSQL. Add a regression test for cleanup racing with task deferral, and verify that repeated deferrals do not remove task_instance rows while unused triggers are deleted safely.

Written by the indexing model from the issue text.

Assessment

Tech stack
kubernetes, mysql, postgresql, python
Domain
backend, data-engineering, databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.