Repository navigation
Scheduler exception handling in dagrun create loop doesn't work #59120
Description
Body
In SchedulerJobRunner, in _create_dag_runs, we do except Exception: ... continue but this does not work when the exception is the result of a db error; in that case you need to roll the transaction back.
to fix this we need either smaller transactions (e.g. one per dag run create attempt) or savepoints. either way it's a bit of a refactor.
Committer
- I acknowledge that I am a maintainer/committer of the Apache Airflow project.
Activity
Add todo re exception handling in dagrun create loop
Arunodoy18 commentedon Dec 6, 2025
Add todo re exception handling in dagrun create loop (#59121)
justinpakzad commentedon Jan 7, 2026
Gonna give this one a go. Cheers.
Edit: This one might be a bit out of reach for me (someone else can pick it up) as I think it would require a pretty big change. We currently use prohibit_commit which wouldn't allow us to do any sort of commits or savepoints from within the
_create_dag_runsmethod , so the only way to patch the fix would be to move it out of the prohibit_commit block (I might be wrong though)Add todo re exception handling in dagrun create loop (apache#59121)
Add todo re exception handling in dagrun create loop (apache#59121)
SEPURI-SAI-KRISHNA commentedon Jul 25, 2026
I looked into the savepoint approach mentioned earlier in this thread. The concern raised above — that
prohibit_commitwon't allow savepoints — is correct, and it's worth recording exactly why, plus what does work.Why
session.begin_nested()is blockedIt isn't an Airflow quirk.
SessionTransaction._prepare_implin SQLAlchemy 2.0 dispatches the session-level event for nested transactions too:CommitProhibitorGuardlistens onbefore_commit, so releasing aSAVEPOINTraisesUNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!exactly as a real commit would. Verified against SQLAlchemy 2.0.51.Dropping to the connection doesn't help either
session.connection().begin_nested()sidesteps the Session event layer, but mixing connection-level savepoints with ORM flush leaves the Session inPendingRollbackErrorafter the first failure. Dead end — noting it so nobody else spends time there.What does work: let the guard tolerate savepoint releases
With that, wrapping each iteration of the
_create_dag_runsloop insession.begin_nested()behaves as intended — a failing iteration rolls back only its own savepoint and the loop continues:and a rogue
session.commit()inside the guard is still rejected, so the guard keeps doing its job.The rationale for allowing this: releasing a savepoint neither ends the outer transaction nor drops row locks, so it cannot break the HA locks the guard exists to protect. Only the outer commit can, and that remains guarded.
Caveat
I ran this against SQLite, which does not reproduce Postgres's aborted-transaction state. So this establishes the guard mechanics and the loop pattern, not that the Postgres failure mode in the issue description is fixed.
SAVEPOINT/ROLLBACK TO SAVEPOINTis the documented Postgres recovery path, so I'd expect it to hold, but it needs confirming against a real Postgres backend before anyone trusts it.Question before I put up a PR
This makes the change smaller than "smaller transactions or savepoints" implies — no restructuring of the scheduler loop — but it does mean touching
CommitProhibitorGuard, which is a core primitive rather than something local to_create_dag_runs.Is widening the guard to permit savepoint releases acceptable in principle? If so I'm happy to open a PR with the guard change, the loop wrapped in
begin_nested(), and a test that a failing iteration no longer poisons the rest of the batch. If the preference is to keep the guard strict and restructure the transaction boundaries instead, I'd rather know before writing it.Drafted-by: Claude Code (Opus 5); reviewed by @SEPURI-SAI-KRISHNA before posting
Add todo re exception handling in dagrun create loop (#59121)
Add todo re exception handling in dagrun create loop (#59121)
goyaladitay11 commentedon Oct 1, 2026
Hey, I've been reading through scheduler_job_runner.py and the issue description. Here's what I've understood and how I'm planning to approach this:
What's happening now
The dagrun creation loop processes multiple DAGs inside a single database transaction. When one DAG raises an exception (say a DB constraint violation or a bad timetable evaluation), the current except Exception: ... continue handler catches the Python-level exception — but the underlying SQLAlchemy session is now in a broken state. The transaction is poisoned.
┌─────────────────────────────────────┐
│ Single DB Transaction │
│ │
│ ┌─ DAG_A ──── create dagrun ✅ ──┐ │
│ ├─ DAG_B ──── DB error 💥 ──────┤ │ ← exception caught, continue...
│ ├─ DAG_C ──── uses same session ┤ │ ← but session is now INVALID
│ └─ DAG_D ──── also fails ───────┘ │ ← all subsequent DAGs fail
│ │
│ session.commit() 💥 │ ← entire batch lost
└─────────────────────────────────────┘
The except Exception: continue only catches the Python exception. It doesn't roll back the failed DB operation, so the session accumulates dirty state. Every DAG processed after the failure either silently produces corrupt data or raises another exception.
Why it's a real problem
In production with hundreds of DAGs, one misconfigured DAG (bad timetable, broken serialization, constraint violation) can prevent ALL other healthy DAGs from getting new DagRuns created. The scheduler silently stops scheduling everything in that batch.
Proposed fix
The issue description mentions two approaches — I'm leaning toward savepoints because they're less disruptive:
┌──────────────────────────────────────┐
│ Outer DB Transaction │
│ │
│ ┌─ SAVEPOINT ──────────────────┐ │
│ │ DAG_A ── create dagrun ✅ │ │ ← savepoint released
│ └──────────────────────────────┘ │
│ ┌─ SAVEPOINT ──────────────────┐ │
│ │ DAG_B ── DB error 💥 │ │ ← savepoint ROLLED BACK
│ └──────────────────────────────┘ │
│ ┌─ SAVEPOINT ──────────────────┐ │
│ │ DAG_C ── create dagrun ✅ │ │ ← unaffected, works fine
│ └──────────────────────────────┘ │
│ │
│ session.commit() ✅ │ ← A and C committed, B skipped
└──────────────────────────────────────┘
In SQLAlchemy, this would be session.begin_nested() which creates a SAVEPOINT. If the operation inside fails, we roll back only that savepoint, leaving the rest of the transaction intact.
Rough sketch of the approach:
python
for dag_model in dag_models:
try:
with session.begin_nested(): # SAVEPOINT
# create dagrun for this dag
...
except Exception:
self.log.exception(
"Failed to create DagRun for DAG '%s', skipping",
dag_model.dag_id,
)
# savepoint already rolled back, session is clean
continue
My plan
Find the exact loop in scheduler_job_runner.py where DagRuns are created in batch
Wrap each per-DAG operation in session.begin_nested() to create a savepoint
Update the except Exception handler to log and continue cleanly
Write a test that verifies: one bad DAG in a batch doesn't prevent other DAGs from getting their runs created
Run the scheduler test suite locally to make sure nothing else breaks
Will have a PR up soon. Let me know if the savepoint approach aligns with what you had in mind, or if you'd prefer the smaller-transactions route instead.
SEPURI-SAI-KRISHNA commentedon Oct 1, 2026
_create_dag_runsruns insideprohibit_commit:_do_schedulingopens the guard and calls_create_dagruns_for_dags(guard, session), which calls_create_dag_runs. On main today that isscheduler_job_runner.py2107, 2109 and 2583.CommitProhibitorGuardlistens onbefore_commit, and SQLAlchemy dispatches that for nested transactions too:so releasing the savepoint raises
RuntimeError: UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!. Re-checked on SQLAlchemy 2.0.52.session.connection().begin_nested()sidesteps the Session event layer, but mixing connection-level savepoints with ORM flush leaves the Session inPendingRollbackErrorafter the first failure, so that is also a dead end.What works is returning early from
_validate_commitwhensession.in_nested_transaction(). Releasing a savepoint neither ends the outer transaction nor drops row locks, so it cannot break the HA locks the guard protects, and a real commit inside the guard is still rejected. Full writeup, including the loop pattern and a test result: #59120 (comment)Two things are unsettled. It changes
CommitProhibitorGuard, a core primitive rather than anything local to this loop, and whether that is acceptable is still unanswered on this issue. And verification was on SQLite, which does not reproduce the Postgres aborted-transaction state, so the loop mechanics are established but the failure mode in the description is not confirmed fixed.Allow nested transactions in CommitProhibitorGuard and use savepoints…
boqiang commentedon Oct 14, 2026
Hi maintainers — is anyone currently working on a fix for this issue, or is there an active PR I should coordinate with? I would be happy to contribute a MySQL deadlock regression test and help with a fix. I noticed the prior session-isolation/savepoint proposals and would like to avoid duplicating work.
Below is a sanitized technical analysis, without organization-specific data.
Conclusion
A MySQL deadlock during DAG metadata autoflush can become a fatal scheduler error when the original database exception is caught and execution continues with a failed SQLAlchemy Session. The next ORM operation raises
PendingRollbackError, which can escape the scheduling loop.Failure mechanism
The relevant path is
_create_dag_runsand its enclosing transaction/retry boundary. Concurrent updates to parsing fields (last_parsed_time,last_parse_duration) and scheduling fields (next_dagrun*) can contend ondagrows. The deadlock is the trigger; reusing the failed Session turns it into a scheduler failure. The DAG named in a creation-error message need not be the DAG whose pending UPDATE was autoflushed.sequenceDiagram participant S as Scheduler participant O as SQLAlchemy Session participant D as MySQL S->>O: Create DagRun / autoflush metadata O->>D: UPDATE dag amid concurrent metadata writes D-->>O: Deadlock (1213) O-->>S: Database exception Note over S,O: Broad handler catches error and continues S->>O: Next ORM operation on failed Session O-->>S: PendingRollbackError Note over S: Exception escapes scheduling loopSupporting code evidence
_create_dag_runscatchesExceptionaround DagRun creation and continues.retry_db_transaction/run_with_db_retriesinairflow/utils/retries.pyhandlesDBAPIError, including rollback before retry. Swallowing the original error prevents that path; the laterPendingRollbackErroris not the configured retry category.Proposed direction — not implemented or validated
Could database exceptions be propagated to the existing owning transaction/retry boundary instead of being swallowed by per-DAG handlers? For example, explicitly re-raising
DBAPIErrorbefore a generic handler is one candidate to evaluate, including other broad handlers on the same path. The owner would roll back the failed unit, discard/reload ORM state and retry with bounded backoff, re-querying candidates and reacquiring required HA locks.sequenceDiagram participant S as Scheduler participant T as Owning transaction/retry boundary participant O as SQLAlchemy Session participant D as MySQL S->>T: Schedule a batch T->>O: Query / flush O->>D: Metadata transaction D-->>O: Deadlock (1213) O-->>T: Propagate original DBAPIError T->>O: Roll back failed unit and reload state Note over T: Bounded backoff, re-query and reacquire HA locks T->>O: Retry the transaction O->>D: Retry metadata work D-->>O: Success T-->>S: Continue schedulingThis is a proposal for discussion, not a claim that adding one exception clause is sufficient. It needs real MySQL testing for full-transaction deadlock rollback, HA lock correctness, duplicate-run prevention, retry exhaustion and continued scheduler heartbeats. Simply increasing retry counts or adding
PendingRollbackErrorto retry eligibility would not address the swallowed original error. Savepoint approaches also need to account forprohibit_commitand MySQL deadlock semantics.Would this recovery boundary be an acceptable direction, or is there a preferred approach / existing PR I could help validate?
SEPURI-SAI-KRISHNA commentedon Oct 14, 2026
Not working on it, so nothing to coordinate around on my side.
On the savepoint route you referenced: it is blocked, and the reason is worth knowing before you spend time there.
_create_dag_runsruns insideprohibit_commit, andCommitProhibitorGuardlistens onbefore_commit, which SQLAlchemy dispatches for nested transactions too (if self._parent is None or self.nested). So releasing aSAVEPOINTraisesUNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!. Dropping tosession.connection().begin_nested()sidesteps the Session event layer but leaves the Session inPendingRollbackErrorafter the first failure, so that is a dead end too. Details and a variant that works are in #59120 (comment), re-checked on SQLAlchemy 2.0.52.That variant needs a change to
CommitProhibitorGuarditself, and whether widening a core primitive is acceptable is the question I left open here and never got answered. Your direction avoids that entirely: propagatingDBAPIErrorto the existingrun_with_db_retriesboundary reuses machinery that already rolls back before retrying, and leaves the guard alone. I think that is the better shape for exactly that reason.The piece this issue has never had is a reproduction on a real MySQL backend. My own check was on SQLite, which does not reproduce the aborted-transaction state, so it established the guard mechanics and nothing about the actual failure. If you add that regression test it would settle the part that analysis cannot.
boqiang commentedon Oct 14, 2026
I've prepared a scoped fix and regression tests in #74761.
The change lets
DBAPIErrorescape both serialized-Dag loading and scheduled DagRun creation, so the existing transaction owner can roll back and retry the batch, re-querying candidates and reacquiring their row locks.CommitProhibitorGuardand ordinary per-Dag exception handling stay unchanged.The MySQL regression uses two real InnoDB transactions to trigger error 1213 after a DagRun has already been created in the batch. It passes with autoflush both enabled and disabled, verifies a fresh locked candidate query, exactly one run per scheduled Dag, and preservation of the concurrent metadata update. Reverting the production change makes all six new MySQL cases fail with
PendingRollbackErrorfollowing the original database error. The complete MySQL scheduler test file passes (566 passed, 2 skipped).This addresses recovery from transient database errors. Persistent errors still propagate when the configured retry limit is exhausted, so the broader per-Dag isolation work in this issue remains open. The tests exercise the transaction boundary directly; they do not establish multi-scheduler process acceptance. The PR includes the broader validation results and remaining limitations.
boqiang commentedon Oct 15, 2026
@SEPURI-SAI-KRISHNA Thanks again for your earlier feedback on the transaction retry boundary. I have opened #74761 and added a regression test using two real MySQL/InnoDB transactions to reproduce deadlock 1213, with autoflush both enabled and disabled. The fix propagates the original DBAPIError to the existing rollback/retry boundary and leaves CommitProhibitorGuard unchanged.
When you have time, could you please take a look at the approach and tests? Any feedback would be appreciated. The PR covers transient-error recovery; persistent per-Dag failure isolation remains outside its scope.