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]

Reply via email to