Fix dataset-triggered DAGs firing without all upstream events#63939
Fix dataset-triggered DAGs firing without all upstream events#63939joseotaviorf wants to merge 1 commit into
Conversation
…rt of apache#62501) Backports three targeted fixes to 2.10.4 to address the race condition reported in GH#56541 and GH#41101, where a dataset-dependent DAG fires with one or more upstream dataset events missing from consumed_dataset_events. Root cause: stale `created_at` timestamps in DatasetDagRunQueue caused the scheduler's event-window query to exclude valid events, yet the presence-based readiness check still considered the DAG ready to run. Changes: - manager.py: replace ON CONFLICT DO NOTHING with ON CONFLICT DO UPDATE (Postgres) and add a WHERE guard to prevent backwards timestamp drift; pass explicit created_at=utcnow() on the non-Postgres merge path so existing rows are always refreshed. - scheduler_job_runner.py: wrap create_dagrun in `if dataset_events:` to prevent phantom runs when the event window is empty; scope the DDRQ DELETE to `created_at <= exec_date` instead of deleting all rows, so events that arrived during processing are preserved for the next cycle. - tests/datasets/test_manager.py: add WHERE-guard regression test for the Postgres upsert path. - dev/BUG_REPORT_DATASET_SCHEDULING.md: detailed bug report with timeline diagrams, related issues (apache#56541, apache#41101, apache#35870), root cause analysis, and explanation of each fix. - reproduce_bug_56541.py: end-to-end reproduction and verification script.
|
@joseotaviorf We should wait for #62501 to get merged, before backporting it. I have reviewed and added some comments on that PR |
| @@ -184,7 +185,7 @@ def _slow_path_queue_dagruns( | |||
| cls, dataset_id: int, dags_to_queue: set[DagModel], session: Session | |||
| ) -> None: | |||
| def _queue_dagrun_if_needed(dag: DagModel) -> str | None: | |||
There was a problem hiding this comment.
The Postgres path has a WHERE guard (created_at < excluded.created_at) that prevents a slower transaction from overwriting a newer timestamp, but session.merge() here still overwrites unconditionally. Two concurrent tasks can race: if the slower one commits last, it writes an older created_at. Using utcnow() makes backward drift unlikely, but it's still possible under clock adjustments or NTP drift.
|
|
||
| values = [{"target_dag_id": dag.dag_id} for dag in dags_to_queue] | ||
| stmt = insert(DatasetDagRunQueue).values(dataset_id=dataset_id).on_conflict_do_nothing() | ||
| values = [{"target_dag_id": dag.dag_id, "created_at": timezone.utcnow()} for dag in dags_to_queue] |
There was a problem hiding this comment.
Each iteration calls timezone.utcnow() separately, so each DAG in dags_to_queue gets a slightly different timestamp. For the common single-DAG case this doesn't matter, but with multiple DAGs the timestamps diverge. Computing the timestamp once before the comprehension would be more consistent:
now = timezone.utcnow()
values = [{"target_dag_id": dag.dag_id, "created_at": now} for dag in dags_to_queue]| delete(DatasetDagRunQueue).where(DatasetDagRunQueue.target_dag_id == dag_run.dag_id) | ||
| delete(DatasetDagRunQueue).where( | ||
| DatasetDagRunQueue.target_dag_id == dag.dag_id, | ||
| DatasetDagRunQueue.created_at <= exec_date, |
There was a problem hiding this comment.
The created_at <= exec_date scoping correctly preserves newer DDRQ rows that arrive during processing. But when dataset_events is empty and DDRQ rows are deleted without creating a DagRun, there's no log message. Operators debugging "my dataset-triggered DAG didn't fire" would have zero visibility into this silent cleanup. A self.log.warning(...) here would help.
|
@joseotaviorf This PR has been converted to draft because it does not yet meet our Pull Request quality criteria. Issues found:
What to do next:
Converting a PR to draft is not a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack. Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you. |
There was a problem hiding this comment.
Pull request overview
Backport of targeted fixes to Airflow 2.x to prevent dataset-triggered DAGs from being scheduled when one or more upstream dataset events are missing, addressing a race around stale DatasetDagRunQueue.created_at and scheduler event-window selection.
Changes:
- Refresh
DatasetDagRunQueue.created_aton conflicts/merges (Postgres and non-Postgres paths) to prevent stale timestamps from excluding valid events. - Guard dataset-triggered DagRun creation on non-empty
dataset_events, and scope DDRQ cleanup bycreated_at <= exec_date. - Add regression tests around timestamp-refresh behavior and the Postgres upsert guard.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
airflow/datasets/manager.py |
Updates DDRQ enqueue logic to refresh created_at on conflict for Postgres and non-Postgres backends. |
airflow/jobs/scheduler_job_runner.py |
Adjusts dataset-triggered DagRun creation and narrows DDRQ deletion window. |
tests/datasets/test_manager.py |
Adds tests for timestamp refresh and Postgres upsert guard behavior. |
Comments suppressed due to low confidence (1)
airflow/datasets/manager.py:194
- In
_slow_path_queue_dagruns,created_at=timezone.utcnow()is evaluated inside_queue_dagrun_if_neededfor every DAG. If multiple DAGs are queued for the same dataset change, this can create slightly differentcreated_atvalues and extra calls. Consider hoistingnow = timezone.utcnow()once per_slow_path_queue_dagrunscall and using it for all queued items.
def _queue_dagrun_if_needed(dag: DagModel) -> str | None:
item = DatasetDagRunQueue(target_dag_id=dag.dag_id, dataset_id=dataset_id, created_at=timezone.utcnow())
# Don't error whole transaction when a single RunQueue item conflicts.
# https://docs.sqlalchemy.org/en/14/orm/session_transaction.html#using-savepoint
try:
with session.begin_nested():
session.merge(item)
except exc.IntegrityError:
| def test_postgres_queue_dagruns_where_guard_prevents_backwards_drift(self, session): | ||
| """ | ||
| The WHERE guard on the Postgres upsert should prevent a slower transaction | ||
| from overwriting a newer created_at with an older one. | ||
| """ | ||
| from datetime import timedelta | ||
|
|
||
| from airflow.utils import timezone | ||
|
|
||
| dsem = DatasetManager() | ||
| ds = DatasetModel(uri="test_postgres_where_guard") | ||
| dag = DagModel(dag_id="test_pg_guard_dag", is_active=True) | ||
| session.add_all([ds, dag]) | ||
| session.flush() | ||
|
|
||
| newer_ts = timezone.utcnow() | ||
| session.add(DatasetDagRunQueue(dataset_id=ds.id, target_dag_id=dag.dag_id, created_at=newer_ts)) | ||
| session.commit() | ||
|
|
||
| dsem._postgres_queue_dagruns(dataset_id=ds.id, dags_to_queue={dag}, session=session) | ||
| session.commit() |
There was a problem hiding this comment.
test_postgres_queue_dagruns_where_guard_prevents_backwards_drift doesn’t currently validate the WHERE-guard behavior it describes. The test sets an existing created_at to newer_ts, then calls _postgres_queue_dagruns(), but that method always uses timezone.utcnow() for the upsert value—i.e. a timestamp newer than newer_ts—so the update is expected to succeed regardless of the WHERE (existing < excluded) guard. To actually regression-test “backwards drift prevention”, freeze/patch timezone.utcnow() (e.g. via the repo’s time_machine fixture) to return an older timestamp for the upsert, then assert created_at remains newer_ts (or unchanged).
| if dataset_events: | ||
| dag_run = dag.create_dagrun( | ||
| run_id=run_id, | ||
| run_type=DagRunType.DATASET_TRIGGERED, | ||
| execution_date=exec_date, | ||
| data_interval=data_interval, | ||
| state=DagRunState.QUEUED, | ||
| external_trigger=False, | ||
| session=session, | ||
| dag_hash=dag_hash, | ||
| creating_job_id=self.job.id, | ||
| ) | ||
| Stats.incr("dataset.triggered_dagruns") | ||
| dag_run.consumed_dataset_events.extend(dataset_events) | ||
| session.execute( | ||
| delete(DatasetDagRunQueue).where(DatasetDagRunQueue.target_dag_id == dag_run.dag_id) | ||
| delete(DatasetDagRunQueue).where( | ||
| DatasetDagRunQueue.target_dag_id == dag.dag_id, | ||
| DatasetDagRunQueue.created_at <= exec_date, | ||
| ) | ||
| ) |
There was a problem hiding this comment.
dag_run is now created only when dataset_events is non-empty, but the subsequent DELETE no longer depends on dag_run. This fixes the previous unbound reference, but it introduces a more serious logic issue: when dataset_events is empty, the code still deletes from DatasetDagRunQueue for the DAG where created_at <= exec_date. That can clear readiness queue entries without creating a DagRun, potentially dropping triggers and preventing future scheduling. Consider moving the DELETE (or at least the created_at <= exec_date cleanup) inside the if dataset_events: block, or otherwise ensure the queue is only pruned when a run is actually created/consumed.
| values = [{"target_dag_id": dag.dag_id, "created_at": timezone.utcnow()} for dag in dags_to_queue] | ||
| stmt = insert(DatasetDagRunQueue).values(dataset_id=dataset_id) | ||
| stmt = stmt.on_conflict_do_update( | ||
| index_elements=["dataset_id", "target_dag_id"], | ||
| set_={"created_at": stmt.excluded.created_at}, | ||
| where=(DatasetDagRunQueue.created_at < stmt.excluded.created_at), | ||
| ) | ||
| session.execute(stmt, values) |
There was a problem hiding this comment.
In _postgres_queue_dagruns, values = [{..., "created_at": timezone.utcnow()} for dag in dags_to_queue] evaluates utcnow() once per DAG. That means created_at can differ slightly across rows from the same dataset event, and it adds unnecessary per-row Python work. Consider computing a single now = timezone.utcnow() once and reusing it for all rows in this batch to keep timestamps consistent and reduce overhead.
|
@joseotaviorf This draft PR has been inactive for 46 days since the last triage comment and no response from the author. Closing to keep the queue clean. You are welcome to reopen this PR when you resume work, or to open a new one addressing the issues previously raised. There is no rush — take your time. Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you. |
Backports #62501 to 2.x
Backports three targeted fixes to 2.11.2 to address the race condition reported in #56750, #56541 and #41101, where a dataset-dependent DAG fires with one or more upstream dataset events missing from consumed_dataset_events.
Root cause: stale
created_attimestamps in DatasetDagRunQueue caused the scheduler's event-window query to exclude valid events, yet the presence-based readiness check still considered the DAG ready to run.Changes:
if dataset_events:to prevent phantom runs when the event window is empty; scope the DDRQ DELETE tocreated_at <= exec_dateinstead of deleting all rows, so events that arrived during processing are preserved for the next cycle.Was generative AI tooling used to co-author this PR?
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.