zach-overflow opened a new issue, #76558:
URL: https://github.com/apache/airflow/issues/76558
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.3.2
### What happened and how to reproduce it?
## Issue Description
We've run into an issue relating to runs triggered by a large number of
asset events. `PATCH` requests to the
`/execution/task-instances/{task_instance_id}/run` (`ti_run`) takes time
proportional to the number of asset events the DagRun consumes, and it does
that work while holding a `FOR UPDATE` lock on the `task_instance` row.
When an asset-triggered run consumes thousands of events, every task start
in that run is slow enough to hit statement timeouts, while the Task SDK's
retries queue up behind the lock held by the previous attempt.
Code path on `main` ([`task_instances.py` @
bca1e14](https://github.com/apache/airflow/blob/bca1e144da1536eb168d925ac358a91e69ef3654/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py#L157-L376)):
1.
[L182-L212](https://github.com/apache/airflow/blob/bca1e144da1536eb168d925ac358a91e69ef3654/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py#L182-L212):
`SELECT ... FROM task_instance JOIN dag_run JOIN dag WHERE task_instance.id =
:id FOR UPDATE OF task_instance`.
2. The TI is updated to `RUNNING`.
3.
[L322-L334](https://github.com/apache/airflow/blob/bca1e144da1536eb168d925ac358a91e69ef3654/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py#L322-L334):
the DagRun is loaded with `.options(joinedload(DR.consumed_asset_events),
...)`. Only the collection itself is eager-loaded.
4.
[L366-L367](https://github.com/apache/airflow/blob/bca1e144da1536eb168d925ac358a91e69ef3654/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py#L366-L367):
`TIRunContext(dag_run=dr, ...)` is built. All of this runs in one transaction,
so the row lock is held until commit.
This endpoint runs `DagRun` datamodel validations with asset lazy-loading
([`datamodels/taskinstance.py`
L362-L396](https://github.com/apache/airflow/blob/bca1e144da1536eb168d925ac358a91e69ef3654/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py#L362-L396))
which causes numerous reads for `asset` and `source_aliases` on every
`AssetEventDagRunReference`. Because of this the validation issues about one
`source_aliases` SELECT per consumed event, all while the TI row lock is still
held.
The scheduler does not run into the same issue because it avoids the pitfall
(see
[`_eager_load_dag_run_for_validation()`](https://github.com/apache/airflow/blob/bca1e144da1536eb168d925ac358a91e69ef3654/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L192-L214)
)
Semi-related past issues and PRs: #65422, #59693, #66190.
## What We've Observed
- An asset-scheduled DAG (`catchup=False`) stayed paused for ~27 days while
its asset kept receiving ~280 events/day.
- After it was unpaused, the first asset-triggered run consumed 7,512
events. Scheduler log: `Created asset-triggered DagRun for '<dag>': run_id=...,
consumed 7512 asset events`. Other runs of the same DAG consumed 1-11 events
and started normally.
- Each `ti_run` call for the first task of that run held the TI row lock for
~25s on the server. Postgres `log_lock_waits` showed `process N still waiting
for ShareLock on transaction ... after 1000 ms ... while locking tuple ... in
relation "task_instance"`, and each waiter got the lock ~25s after the previous
request started.
- The client got `504 Gateway Timeout`. Each Task SDK retry queued behind
the row lock of the previous, still-running request and was then cancelled by
`statement_timeout` (SQLSTATE 57014).
- All 4 tries of the task failed the same way, and the downstream tasks went
to `upstream_failed`. Every task in such a run pays this cost because every
task start goes through `ti_run`.
## How To Reproduce
This section written by Claude Code Opus 5.5:
1. Create a DAG scheduled on a single asset, then pause it.
2. Emit several thousand events for that asset, e.g. with repeated `POST
/api/v2/assets/events`.
3. Unpause the DAG. The scheduler creates one asset-triggered run that
consumes all of the queued events.
4. Watch the `ti_run` latency for the first task of that run. With
SQLAlchemy echo or Postgres statement logging on, you see one
`asset_alias_asset_event` / `source_aliases` SELECT per consumed event inside
the transaction that holds `FOR UPDATE OF task_instance`. Latency grows with
the number of consumed events.
### What you think should happen instead?
The `ti_run` execution API's endpoint latency, and particularly how long it
holds the TI row lock, should not grow with the number of consumed asset events.
Possible fixes:
1. Eager-load the nested relations in `ti_run` in a similar way that the
scheduler does via `_eager_load_dag_run_for_validation()` (see link in issue
description above). That turns the N+1 queries into a fixed number of queries.
2. Load and serialize the consumed events after the TI state update is
committed, so that work does not run under `FOR UPDATE`. Alternatively, stop
sending every consumed event in `TIRunContext` and have the Task SDK fetch them
lazily when `triggering_asset_events` is accessed.
Fix (1) is small and self-contained. Fix (2) also covers the cost of
serializing large `extra` payloads.
### Operating System
Linux (Debian-based container image)
### Deployment
Other 3rd-party Helm chart
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
This happens every time an asset-triggered run consumes a large number of
events, which is common after a long pause of an asset-scheduled DAG whose
upstream keeps producing events. Runs that consume only a few events are not
noticeably affected.
### Are you willing to submit PR?
- [x] 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)
--
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]
