vatsrahul1001 opened a new issue, #74860:
URL: https://github.com/apache/airflow/issues/74860
### Apache Airflow version
3.4.0 (confirmed on `3.4-1-alpha2`)
### What happened
Any DAG that uses dynamic task mapping (`.expand()`) **and** has a real
`schedule` (not `schedule=None`) never gets its initial TaskInstances created
when the **scheduler itself** creates the DagRun. The scheduler logs `Error
scheduling DAG run ... of <dag_id>` on every scheduler loop, forever — the
DagRun never progresses and no mapped task ever runs.
```
[error] Error scheduling DAG run scheduled__2026-10-16T14:50:00+00:00 of
dynamic_region_scheduler_repro
Traceback (most recent call last):
File ".../airflow/jobs/scheduler_job_runner.py", line 3219, in
_schedule_all_dag_runs
callback = self._schedule_dag_run(run, session=session)
File ".../airflow/jobs/scheduler_job_runner.py", line 3327, in
_schedule_dag_run
if not dag_run.bundle_version and not
self._verify_integrity_if_dag_changed(
File ".../airflow/jobs/scheduler_job_runner.py", line 3394, in
_verify_integrity_if_dag_changed
dag_run.verify_integrity(dag_version_id=latest_dag_version.id,
session=session)
File ".../airflow/models/dagrun.py", line 2046, in verify_integrity
tis_to_create = list(
File ".../airflow/models/dagrun.py", line 2390, in _create_initial_tasks
yield from self._create_tasks(ordinary_tasks, task_creator,
session=session, expand_literals=True)
File ".../airflow/models/dagrun.py", line 2466, in _create_tasks
regions = DynamicRegion.get_or_create_many(
File ".../airflow/models/dynamic_region.py", line 302, in
get_or_create_many
with session.begin_nested():
File ".../sqlalchemy/engine/util.py", line 147, in __exit__
with util.safe_reraise():
...
File ".../airflow/utils/sqlalchemy.py", line 700, in _validate_commit
raise RuntimeError(PROHIBIT_COMMIT_ERROR_MESSAGE)
RuntimeError: UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!
```
### What you think should happen instead
The DagRun should get its initial (mapped) TaskInstances created and run
normally, same as any other scheduled DAG.
### How to reproduce
```python
from airflow.sdk import DAG
from airflow.decorators import task
with DAG(
dag_id="dynamic_region_scheduler_repro",
schedule="*/2 * * * *",
catchup=False,
max_active_runs=1,
is_paused_upon_creation=False,
) as dag:
@task
def say(i: int):
return i
say.expand(i=[1, 2, 3])
```
Deploy this DAG (must have a real `schedule`, not `schedule=None` — see root
cause below) and watch the scheduler logs; the failure above repeats every loop
indefinitely.
### Root cause
`DynamicRegion.get_or_create_many()` (new in 3.4 — #74275, #74346, #74344,
#74348, #74349) wraps its batch insert in `session.begin_nested()` (a
SAVEPOINT):
```python
try:
with session.begin_nested():
session.add_all(slots.values())
except IntegrityError:
...
```
When `verify_integrity()` is invoked from inside the scheduler's own
`prohibit_commit`-guarded session (`_schedule_all_dag_runs` →
`_schedule_dag_run` → `_verify_integrity_if_dag_changed`), releasing that
SAVEPOINT dispatches a `before_commit` event, which
`CommitProhibitorGuard._validate_commit` treats the same as a real commit and
rejects — it has no way to distinguish a SAVEPOINT release from an actual
commit.
This is the exact same `CommitProhibitorGuard` vs. nested-transaction
mechanics already documented at length in #59120, but a **different** code
path: #59120 is about `_create_dag_runs`'s own (never-merged) proposed
savepoint fix, ultimately resolved there via a different approach (#74761,
propagating `DBAPIError` to the existing retry boundary instead).
`DynamicRegion.get_or_create_many()` independently hits the same guard
limitation and doesn't appear to be tracked anywhere.
### Why this doesn't show up when a DAG is triggered manually/via the API
`trigger_dag_run` (API) calls `DAG.create_dagrun()` → `verify_integrity()`
in the **API server's own process/session**, which never has
`CommitProhibitorGuard` installed — only the scheduler's dedicated session does
(`with prohibit_commit(session) as guard:`, `scheduler_job_runner.py:2180`).
The identical `DynamicRegion` code runs there without issue. Only a DagRun the
**scheduler itself** creates from a cron `schedule` goes through the guarded
session, which is why this only reproduces with a real schedule, not
`schedule=None` + manual/API trigger.
### Are you willing to submit a PR?
- [ ] Yes I am willing to submit a PR!
cc @ashb — tagging since you authored the DynamicRegion PRs (#74275 et al.)
--
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]
