Skip to content

Resolve VariableInterval deadlines safely at DagRun creation - #68917

Open
seanghaeli wants to merge 8 commits into
apache:mainfrom
aws-mwaa:feature/variable-interval-resolution
Open

Resolve VariableInterval deadlines safely at DagRun creation#68917
seanghaeli wants to merge 8 commits into
apache:mainfrom
aws-mwaa:feature/variable-interval-resolution

Conversation

@seanghaeli

@seanghaeliseanghaeli commented Jun 23, 2026

Copy link
Copy Markdown
Contributor

Why

This re-introduces the VariableInterval resolution portion of #66608.

A DeadlineAlert configured with a VariableInterval is resolved when the scheduler creates a DagRun, inside DAG._process_dagrun_deadline_alerts (which runs under the prohibit_commit guard). Two problems are fixed:

  1. Resolution goes through the full secrets chain (env vars, configured secrets backends, then the metadata DB) via a dedicated _resolve_variable_interval helper, rather than Variable.get / begin_nested. Variable.get and a SAVEPOINT release both commit on the scheduler's session, tripping prohibit_commit (UNEXPECTED COMMIT) and silently dropping deadlines for every scheduled DagRun. The helper passes the scheduler session through to the metastore backend so the DB read does not commit, and reading via the secrets chain (not the variable table directly) means AIRFLOW_VAR_* env vars and secrets-backend-backed Variables resolve too.

  2. Each deadline alert is isolated with a plain try/except (deliberately notbegin_nested, which would commit a SAVEPOINT and trip the same guard). Creating a deadline is auxiliary to creating the DagRun; a single bad alert — a missing/invalid backing Variable, or an undecodable serialized blob — must never abort the DagRun and stop the DAG from scheduling.

VariableInterval.resolve is split into resolve + coerce_to_timedelta so the scheduler-side reader reuses the exact same validation (including the OverflowError -> ValueError translation) without going through Variable.get.

Tests

  • airflow-core/tests/unit/models/test_dagrun.py: VariableInterval resolves from a real Variable row and from an AIRFLOW_VAR_* env var; a missing Variable and an undecodable alert are isolated (DagRun still created, no Deadline row, error logged).
  • task-sdk/tests/task_sdk/definitions/test_deadline.py: coerce_to_timedelta validation (non-integer, <= 0, overflow).

Verified locally in Breeze: 8 passed (dagrun deadline/variable) + 14 passed (SDK TestVariableInterval).

Generated-by: Claude Code (Opus via Claude Code) on behalf of Sean Ghaeli

@seanghaeli
seanghaeliforce-pushed the feature/variable-interval-resolution branch 2 times, most recently from 7aee60c to 94825ecCompareJune 23, 2026 20:54
@seanghaeli
seanghaeli requested a review from potiuk as a code ownerJune 23, 2026 20:54
@seanghaeli
seanghaeli marked this pull request as draft June 23, 2026 20:57
@seanghaeli
seanghaeliforce-pushed the feature/variable-interval-resolution branch from 94825ec to 2342c03CompareJune 23, 2026 21:31
@seanghaeli
seanghaeli marked this pull request as ready for review June 23, 2026 23:00
@potiukpotiuk added the ready for maintainer review Set after triaging when all criteria pass. label Jun 25, 2026

@SameerMesiah97SameerMesiah97 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I did not manage to get to the tests but if I have to be honest, I think this needs a round of polish before a maintainer can review it. Conceptually, it looks correct but there are classic smells of unvetted AI-generated content such as messy code, long commnets/docstrings etc. I would convert it to draft and clean it up before requesting review.

I have left some comments.

Comment threadairflow-core/src/airflow/serialization/definitions/dag.py
Comment threadairflow-core/src/airflow/serialization/definitions/dag.py Outdated
Comment threadairflow-core/src/airflow/serialization/definitions/dag.py
Comment threadairflow-core/src/airflow/serialization/definitions/dag.py Outdated
Comment threadairflow-core/src/airflow/serialization/definitions/dag.py Outdated
@seanghaeli
seanghaeliforce-pushed the feature/variable-interval-resolution branch 2 times, most recently from 56b2666 to 3e4399bCompareJuly 8, 2026 06:46
@seanghaeli
seanghaeliforce-pushed the feature/variable-interval-resolution branch 2 times, most recently from 69649b1 to 4754b7cCompareJuly 28, 2026 22:33
@seanghaeli

Copy link
Copy Markdown
ContributorAuthor

@SameerMesiah97 could I get your inputs on the updated version?

@SameerMesiah97SameerMesiah97 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks good to me. I just left a few more comments.

Comment threadairflow-core/src/airflow/serialization/definitions/dag.py
Comment threadairflow-core/tests/unit/models/test_dagrun.py Outdated
Comment threadairflow-core/tests/unit/models/test_dagrun.py Outdated
Comment threadairflow-core/tests/unit/models/test_dagrun.py Outdated
@seanghaeli
seanghaeliforce-pushed the feature/variable-interval-resolution branch from c176287 to e1a907bCompareAugust 5, 2026 23:14

@ferruzziferruzzi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Apparently I reviewed this last week and forgot to submit. Sorry for the delay

Comment threadairflow-core/src/airflow/serialization/definitions/dag.py Outdated
Comment threadairflow-core/tests/unit/models/test_dagrun.py
Comment threadairflow-core/src/airflow/serialization/definitions/dag.py
Comment threadairflow-core/src/airflow/serialization/definitions/dag.py
Comment threadairflow-core/tests/unit/models/test_dagrun.py
Comment threadtask-sdk/src/airflow/sdk/definitions/deadline.py
Comment threadairflow-core/src/airflow/serialization/definitions/dag.py Outdated
Comment threadairflow-core/src/airflow/serialization/definitions/dag.py
Comment threadairflow-core/tests/unit/models/test_dagrun.py Outdated
A ``DeadlineAlert`` configured with a ``VariableInterval`` is resolved when the
scheduler creates a DagRun, inside ``DAG._process_dagrun_deadline_alerts`` (which
runs under the ``prohibit_commit`` guard). Two problems are fixed:
1. Resolution now goes through the full secrets chain (env vars, configured
secrets backends, then the metadata DB) via a dedicated
``_resolve_variable_interval`` helper, rather than ``Variable.get`` /
``begin_nested``. ``Variable.get`` and a SAVEPOINT release both commit on the
scheduler's session, tripping ``prohibit_commit`` ("UNEXPECTED COMMIT") and
silently dropping deadlines for every scheduled DagRun. The helper passes the
scheduler session through to the metastore backend so the DB read happens
without committing, and reading via the secrets chain (not the variable table
directly) means ``AIRFLOW_VAR_*`` env vars and secrets backends resolve too.
2. Each deadline alert is isolated with a plain ``try``/``except`` (NOT
``begin_nested``). Creating a deadline is auxiliary to creating the DagRun; a
single bad alert -- a missing/invalid backing Variable, or an undecodable
serialized blob -- must never abort the DagRun and stop the DAG scheduling.
``VariableInterval.resolve`` is split into ``resolve`` + ``coerce_to_timedelta``
so the scheduler-side reader reuses the exact same validation (including the
OverflowError -> ValueError translation) without going through ``Variable.get``.
Generated-by: Claude Code (Sonnet/Opus via Claude Code) on behalf of Sean Ghaeli
@seanghaeli
seanghaeliforce-pushed the feature/variable-interval-resolution branch from f9ee1dc to 707ef13CompareAugust 18, 2026 22:36

@ferruzziferruzzi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like my concerns have all been addressed. LGTM

@ferruzziferruzzi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm sorry about this, there are now eight PRs all interacting with the Deadlines code and I'm trying to untangle the conflicts before they get worse. #71802 is approved and needs to be merged before this one because it also blocks a few other fixes, but it is going to kind of pull the rug out from under you a bit on this.

Once it merges, dag.py checks isinstance(interval, SerializedVariableInterval) and no longer imports VariableInterval at all. That means:

  • You can drop _resolve_variable_interval. #71968 adds an optional keyword-only session to Variable.get, so SerializedVariableInterval.resolve(session=session) covers it without duplicating the backend walk. Your comment there already points at #71801, which is the issue #71968 fixes.

  • coerce_to_timedelta then loses its only caller, so that can be dropped along with the related test_coerce_to_timedelta_invalid. The OverflowError to ValueError translation is a real improvement nobody else hit on, but it needs to live where the live path actually runs, so wrap the return timedelta(seconds=seconds) at the end of the new SerializedVariableInterval.resolve() in airflow-core/src/airflow/serialization/definitions/deadline.py, alongside the int() cast. Then add a "too large" case to test_resolve_invalid which #71802 adds in airflow-core/tests/unit/serialization/definitions/test_deadline.py.

  • The seconds <= 0 guard is the one real conflict. An interval can be negative so that check is a distinct bug. Please drop the guard along with the ("0", "must be > 0") and ("-5", "must be > 0") test cases.

Side note: The per-alert try/except isolation is good and nobody else is doing that. It may turn out to be one of the most valuable additions in your PR.

One thing to check: #71287 is doing failure isolation with savepoints around DagRun creation. Your comment says you chose a plain try/except over session.begin_nested() deliberately, so you may already have looked, but the two haven't been cross-referenced anywhere that I can tell.

Overall, it should result in a smaller PR for you, trimming out some redundant (and conflicting) code, not a rewrite.

I am also submitting this as a "Request changes" so it blocks merging out of order and doesn't screw up the other PRs. Unblock conditions: #71802 must merge first, this PR must be merged on top of that one, and the above changes need to be made. Once this one merges, It will unblock #71968 which will have to rebase on top of this and get some tweaks.

@ferruzziferruzzi added this to the Airflow 3.3.2 milestone Aug 28, 2026
@ferruzziferruzzi removed the ready for maintainer review Set after triaging when all criteria pass. label Aug 29, 2026
Comment on lines +789 to +800
if deadline_time is not None:
session.add(
Deadline(
deadline_time=deadline_time,
callback=deserialized_deadline_alert.callback,
dagrun_id=orm_dagrun.id,
deadline_alert_id=deadline_alert.id,
dag_id=orm_dagrun.dag_id,
bundle_name=orm_dagrun.dag_model.bundle_name,
)
)
)
team_name = (
DagModel.get_team_name(self.dag_id, session=session)
if airflow_conf.getboolean("core", "multi_team")
else None
)
stats.incr(
"deadline_alerts.deadline_created",
tags=prune_dict({"dag_id": self.dag_id, "team_name": team_name}),
)
stats.incr("deadline_alerts.deadline_created", tags=metrics_tags)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are going to end up having to rebase on top of a couple of other PRs that need to land before this one. I'm trying to make that future rebase a bit easier here.

#71767 has an elif branch right after the stats.incr you have on L800. If you can add that here, it should ease your merge pain later:

stats.incr("deadline_alerts.deadline_created", tags=metrics_tags)
elifrequired_dagrun_column:= {
SerializedReferenceModels.DagRunLogicalDateDeadline: "logical_date",
SerializedReferenceModels.DagRunQueuedAtDeadline: "queued_at",
}.get(type(deserialized_deadline_alert.reference)):
log.warning(
"skipping deadline alert because the deadline reference evaluated to None",
dag_id=self.dag_id,
run_id=orm_dagrun.run_id,
deadline_alert_id=deadline_alert.id,
reference_type=deserialized_deadline_alert.reference.reference_name,
required_dagrun_column=required_dagrun_column,
)

That PR also adds two tests in airflow-core/tests/unit/models/test_dagrun.py that you'll want to pull over which cover that branch. That can wait till the merge but may be easier to just paste it over now since you are copying this over.

@ferruzziferruzzi removed this from the Airflow 3.3.2 milestone Aug 29, 2026
@ferruzziferruzzi added the backport-to-v3-3-test Backport to v3-3-test label Aug 29, 2026
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants

@seanghaeli@ferruzzi@SameerMesiah97@potiuk