xchwan commented on code in PR #75708:
URL: https://github.com/apache/airflow/pull/75708#discussion_r4511120037
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -861,6 +862,15 @@ def _process_dagrun_deadline_alerts(
)
stats.incr("deadline_alerts.deadline_creation_failed",
tags=metrics_tags)
+ # An alert can roll the session back, discarding the pending
DagRun. Carrying on would
+ # add Deadlines that violate the dagrun_id foreign key, and
returning normally would let
+ # the caller treat the missing run as created.
+ if orm_dagrun not in session:
Review Comment:
Thanks, this makes sense: evaluating alerts on their own session fixes the
cause, so a rollback in an alert only affects that session and the run (and
other Dags' runs in the same tick) survive.
Plan:
1. Resolve the interval and call `evaluate_with` on a
`create_session(scoped=False)` session, passing a snapshot of the run instead
of the pending ORM object. The snapshot carries `dag_id`, `logical_date` and
`queued_at`, which are the fields `DeadlineDagRunProtocol` guarantees
(`queued_at` is set at construction, so no DB read is needed). `Deadline` rows
are still added on the scheduler's session.
2. Keep the `orm_dagrun not in session` check and `DagRunRolledBackError` as
a backstop.
3. Document the contract in `deadline-alerts.rst`.
Some things I considered:
- **Read-only, no auto-commit.** `create_session` commits on exit, so a
write from a reference would persist even if the run creation later fails. I'd
use a plain `NonScopedSession()` that is always rolled back and closed, one per
Dag run rather than per alert.
- **Locks.** The scheduler's session holds the flushed `DagRun` and the
`DagModel` row lock, so a reference that takes a row lock or writes through the
new session would wait on its own process. Plain reads (`Variable`,
`AverageRuntimeDeadline`) are unaffected. I'd document this rather than enforce
it.
- **Snapshot fields.** A reference that reads something beyond the three
guaranteed fields (`conf`, `start_date`) would hit `AttributeError`, which the
per-alert handler turns into a skipped alert. I'd document the guaranteed
fields rather than widen the snapshot.
- **Bypassing the passed session.** A reference that uses `provide_session`,
`create_session()` or `settings.Session()` gets the scheduler's scoped session.
`commit()` there already fails under `prohibit_commit`; a `rollback()` can't be
isolated from inside this function, so the backstop turns it into a visible
failure instead of an orphaned `Deadline`.
- **Other callers.** `create_dagrun` is also used by API trigger, backfill
and `dags test`, so I'll check those, including SQLite in-memory where a second
connection can't see the first one's data.
This doesn't address #59120 (DB errors on the scheduler's session in the
create loop). That is a separate problem and needs a savepoint-friendly
`prohibit_commit`.
Does this scope work for you, with the rest handled by documentation and the
backstop? I'll update the tests accordingly.
---
Drafted-by: Claude Code (Sonnet 5.5); reviewed by @xchwan
--
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]
