rino0601 opened a new issue, #71360:
URL: https://github.com/apache/airflow/issues/71360
### 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**
`GET /execution/asset-events/by-asset` — the Task SDK path behind
`inlet_events[asset]` — returns **500 Internal Server Error** whenever *any*
asset event in that asset's history has a `created_dagrun` whose `start_date`
is `NULL`.
`dag_run.start_date` is nullable and is `NULL` for any run that has not
started yet. Most importantly it is `NULL` **after a `clear`**, which resets
`state` to `queued` and `start_date` to `NULL`. But
`DagRunAssetReference.start_date` is declared as a required `datetime`, so
serializing the response raises:
```
pydantic_core._pydantic_core.ValidationError: 1 validation error for
AssetEventResponse
created_dagruns.0.start_date
Input should be a valid datetime [type=datetime_type, input_value=None,
input_type=NoneType]
```
api-server traceback (tail):
```
File "airflow/api_fastapi/execution_api/routes/asset_events.py", line 106,
in get_asset_event_by_asset_name_uri
return _get_asset_events_through_sql_clauses(
File "airflow/api_fastapi/execution_api/routes/asset_events.py", line 53, in
_get_asset_events_through_sql_clauses
AssetEventResponse(
File "pydantic/main.py", line 263, in __init__
validated_self = self.__pydantic_validator__.validate_python(data,
self_instance=self)
pydantic_core._pydantic_core.ValidationError: 1 validation error for
AssetEventResponse
created_dagruns.0.start_date
Input should be a valid datetime [type=datetime_type, input_value=None,
input_type=NoneType]
```
The offending model — `start_date` is the only date field that is not
nullable:
```python
#
airflow-core/src/airflow/api_fastapi/execution_api/datamodels/asset_event.py
class DagRunAssetReference(StrictBaseModel):
"""DagRun serializer for asset responses."""
run_id: str
dag_id: str
logical_date: datetime | None
start_date: datetime # <-- NULL for any run that never started
end_date: datetime | None
state: str
data_interval_start: datetime | None
data_interval_end: datetime | None
partition_key: str | None
```
An identical `DagRunAssetReference` in `core_api/datamodels/assets.py` has
the same problem, so the public API is affected too: `GET
/api/v2/dags/{dag_id}/dagRuns/{run_id}/upstreamAssetEvents` returns 500 under
the same condition — which is exactly the endpoint needed to explain the
failure from the UI. Both are still unfixed on `main`.
From a task, `inlet_events[asset]` is lazy; the request is issued on first
materialization, so an innocuous `len()` raises:
```
AirflowRuntimeError: API_SERVER_ERROR: {'status_code': 500, 'message':
'Server returned error',
'detail': {'message': 'Internal server error', 'correlation-id': '...'}}
File ".../dags/consumer_dag.py", line 314, in resolve_dies
if len(prev_events):
File "airflow/sdk/execution_time/context.py", line 1160, in __len__
File "airflow/sdk/execution_time/context.py", line 1143, in _asset_events
File "airflow/sdk/execution_time/comms.py", line 239, in send
```
**Steps to reproduce**
Minimal DAG pair:
```python
from airflow.sdk import Asset, dag, task
asset_a = Asset("repro://asset_a")
@dag(schedule=None, catchup=False)
def repro_producer():
@task(outlets=[asset_a])
def emit():
pass
emit()
@dag(schedule=[asset_a], catchup=False)
def repro_consumer():
@task(inlets=[asset_a])
def read_history(**context):
# lazy accessor — the by-asset request is issued by __len__
print(len(context["inlet_events"][asset_a]))
read_history()
repro_producer()
repro_consumer()
```
1. Trigger `repro_producer`. It emits `asset_a` (asset event #1), which
creates and runs `repro_consumer` run #1. The task succeeds.
2. **Clear** `repro_consumer` run #1. Its `state` becomes `queued` and
`start_date` becomes `NULL`.
3. Immediately mark that cleared run **failed**, so it can never acquire a
`start_date`. (This step is not required for the 500 — it just makes the state
permanent instead of racing the scheduler. See the note below.)
4. Trigger `repro_producer` again. Asset event #2 creates `repro_consumer`
run #2, which starts normally.
5. Run #2's task fails with `AirflowRuntimeError: API_SERVER_ERROR:
{'status_code': 500, ...}`. `by-asset` returned the asset's whole history,
event #1 still references run #1, and run #1 has `start_date = NULL`.
6. `GET /api/v2/dags/repro_consumer/dagRuns/{run_1_id}/upstreamAssetEvents`
also returns 500, via the `core_api` copy of the model.
**The state is not limited to `queued`, and it does not always self-heal.**
A run that never started keeps `start_date = NULL` after it leaves `queued`;
marking it `failed` sets `state` and `end_date` but does **not** backfill
`start_date`. Measured on our deployment — these two rows are permanent 500
sources:
```
state=failed type=asset_triggered clear_number=1 consumer_dag /
asset_triggered__...05:22:29Z start_date=NULL
state=failed type=asset_triggered clear_number=1 consumer_dag /
asset_triggered__...05:15:10Z start_date=NULL
-> poisoning 34 asset_event links
```
So the only ways out are to let the run start, or to delete it. The two
obvious operator reflexes — "clear it and retry", "mark it failed" —
respectively reproduce and freeze the breakage.
Other routes to the same state: pausing a DAG that has queued
asset-triggered runs, or a consumer with `max_active_runs=1` whose cleared run
waits behind a running sibling.
To find every currently-affected asset on a live instance:
```sql
SELECT a.name AS asset, dr.dag_id, dr.run_id, dr.state, dr.clear_number,
ae.id AS event_id
FROM dag_run dr
JOIN dagrun_asset_event dae ON dae.dag_run_id = dr.id
JOIN asset_event ae ON ae.id = dae.event_id
JOIN asset a ON a.id = ae.asset_id
WHERE dr.start_date IS NULL
ORDER BY a.name, ae.timestamp;
```
**Two things turn this into an outage rather than a transient blip**
1. `_get_asset_events_through_sql_clauses` is called from `by-asset` without
a `limit`, so it serializes the asset's **entire** event history — ~1800 events
per asset here, ~870 ms per call. One bad row anywhere in that history fails
the whole response; there is no partial success and no way for the caller to
skip it.
2. **The failure drives a self-amplifying loop.** A task reading
`inlet_events` fails → the run is cleared to retry it → the cleared run returns
to `queued` with `start_date = NULL` → that run is itself a `created_dagrun` of
the very asset event the task reads → the retry fails identically. Poisoned
assets grew from 28 to 40 while we were remediating.
**Why this stays invisible in normal operation.** Organically queued runs
pass through `start_date IS NULL` in **p50 20 ms / p90 80 ms**; of 302 runs in
a 6-hour sample only 15 (5%) exceeded 1 s. The bug effectively only bites after
a `clear`, a `pause`, or a `max_active_runs` block — states where a run sits
un-started for minutes or forever.
**Measured impact on our deployment** (Airflow 3.3.0, CeleryExecutor, ~30
asset-driven DAGs)
Every `start_date IS NULL` row we found had `clear_number = 1` — all
produced by a `clear`, none by ordinary scheduling latency:
```
state=failed type=scheduled clear_number=1 paused=True
monitor_dag_a / scheduled__...05:55:00Z
state=failed type=scheduled clear_number=1 paused=True
monitor_dag_a / scheduled__...06:00:00Z
state=failed type=scheduled clear_number=1 paused=True
monitor_dag_a / scheduled__...06:05:00Z
state=failed type=asset_triggered clear_number=1 paused=True
monitor_dag_b / asset_triggered__...05:22:29Z
state=failed type=asset_triggered clear_number=1 paused=True
monitor_dag_b / asset_triggered__...05:15:10Z
state=queued type=asset_triggered clear_number=1 paused=False
consumer_dag / asset_triggered__...06:09:20Z
```
A handful of such rows poisoned **28–40 distinct assets**, the count moving
as runs were cleared and re-cleared. At one sample: 32 assets across 56
asset_event links, from just two source DagRuns:
```
poison source monitor_dag_b[failed] 34 event links
poison source consumer_dag[queued] 22 event links
```
`by-asset` for one asset flipped from 100% success to 100% failure and
stayed there (10-minute buckets, from the api-server access log):
```
01:3x .. 05:2x status_code=200 (58 requests, 100% OK)
05:3x 5 x status_code=500
05:4x 10 x status_code=500
05:5x 5 x status_code=500
06:0x 5 x status_code=500
06:1x 4 x status_code=500
06:2x 4 x status_code=500
```
Across the retained log window, **295 of 295** `by-asset` requests returned
500 after that transition.
Replaying the endpoint's own serialization per event pinpoints one bad event
out of ~1800:
```
asset_x: 1794 events, 1 validation failure
event id=276879 ts=2026-08-10 05:09:53Z
dagrun monitor_dag_b / asset_triggered__2026-08-10T05:15:10Z
state=queued start_date=None
-> created_dagruns.0.start_date
```
Blast radius one hour in: **16 failed + 4 `up_for_retry`** task instances
across 8 asset-consuming DAGs, plus the two DAGs whose cleared runs were the
poison source. Every one failed on the same `inlet_events[...]` call.
### What you think should happen instead?
`by-asset` should return `200` and represent a never-started
`created_dagrun` with `start_date: null`, matching the ORM column (nullable),
the actual DB state, and the scheduler's own behaviour.
Proposed fix — both copies of the model need it. The `execution_api` one
unblocks tasks reading `inlet_events`; the `core_api` one unblocks
`upstreamAssetEvents` so the UI can show what happened.
```diff
---
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/asset_event.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/asset_event.py
@@ class DagRunAssetReference(StrictBaseModel):
run_id: str
dag_id: str
logical_date: datetime | None
- start_date: datetime
+ start_date: datetime | None
end_date: datetime | None
state: str
data_interval_start: datetime | None
data_interval_end: datetime | None
partition_key: str | None
```
```diff
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
@@ class DagRunAssetReference(StrictBaseModel):
run_id: str
dag_id: str
logical_date: datetime | None
- start_date: datetime
+ start_date: datetime | None
end_date: datetime | None
data_interval_start: datetime | None
data_interval_end: datetime | None
partition_key: str | None
```
This is the same defect class that has already been accepted and fixed
elsewhere:
- #46487 / #46570 — viewing a TaskInstance whose DagRun is queued returned
500 (fixed 2025-02)
- #61381 — `dag_run.start_date - Input should be a valid datetime` when
rerunning a **cleared** DagRun (fixed 2026-02) — same trigger as here,
different model
- #68333 — `/eventLogs` 500 when `log.dttm` is NULL — same pattern,
different column (fixed 2026-06)
In particular the sibling `DagRun` model in the *same* package was already
made nullable:
```python
#
airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py —
class DagRun
run_after: UtcDateTime
start_date: UtcDateTime | None # already fixed
end_date: UtcDateTime | None
```
`DagRunAssetReference` appears to have been missed by that sweep, in both
`execution_api` and `core_api`.
Two further points worth considering alongside the type fix, though not
required to close this:
- **Serialization errors in `by-asset` are unrecoverable for the caller.**
One unserializable event out of ~1800 loses the entire history.
Skipping-with-warning, or a `limit`, would keep a single bad row from taking
out the endpoint.
- **The `correlation-id` returned to the task is not logged server-side.**
Emitting it alongside the api-server traceback would make this class of failure
diagnosable from the task log alone, instead of requiring a grep through the
api-server's log files.
### Operating System
Linux, RHEL 8.10-compatible container image, Python 3.12.12
### Deployment
Other 3rd-party Helm chart
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
```
apache-airflow==3.3.0
apache-airflow-core==3.3.0
apache-airflow-task-sdk==1.3.0
apache-airflow-providers-amazon==9.31.0
apache-airflow-providers-apache-hive==9.5.0
apache-airflow-providers-celery==3.21.0
apache-airflow-providers-common-compat==1.15.0
apache-airflow-providers-common-io==1.8.0
apache-airflow-providers-common-sql==2.0.1
apache-airflow-providers-http==6.0.4
apache-airflow-providers-postgres==6.8.0
apache-airflow-providers-redis==4.5.0
apache-airflow-providers-smtp==3.0.1
apache-airflow-providers-standard==1.15.0
apache-airflow-providers-trino==6.6.0
```
### Official Helm Chart version
Not Applicable
### Kubernetes Version
Not Applicable
### Helm Chart configuration
_Not applicable — in-house chart, not the official one. Relevant runtime
topology: CeleryExecutor,
PostgreSQL metadata DB, `airflow api-server` behind a single Service, S3
remote logging._
### Docker Image customizations
_None relevant to this issue. `apache-airflow` and providers are installed
unpatched at the versions
listed above; the defect reproduces against upstream `main` source as shown
in the diff._
### Anything else?
Occurs on **100% of `by-asset` calls** for an affected asset. It does not
recover on its own unless the offending run actually starts — clearing the run
is what *creates* this state, and marking it `failed` makes it permanent. On a
paused DAG, or one whose cleared run is blocked behind a running sibling, the
affected assets stay unreadable indefinitely.
Possibly related, as a way to *reach* this state rather than a cause of the
500: #56050 (most recent asset events ignored when `max_active_runs = 1`).
### 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]
