Repository navigation
Enforce the Dag import timeout from the task supervisor side - #74571
Draft
Andrushika wants to merge 4 commits into
Draft
Andrushika wants to merge 4 commits into
Andrushika wants to merge 4 commits into
Conversation
Move the get_dagbag_import_timeout policy lookup, its fallback when airflow settings cannot be imported, and the int/float check into a module-level function, so the task runner can resolve the same value for a Dag file before it is parsed.
The task process needs to drop the supervisor's deadline once a phase it armed is over, so a value of 0 now means "no deadline" instead of being rejected. The supervisor's timeout log line no longer names execution_timeout, since it will also fire for the Dag parse phase. SetExecutionTimeout has not been released yet, so the in-progress supervisor schema version only gets a note instead of a VersionChange.
Andrushika
requested review from
amoghrajesh,
ashb and
jason810496
as code owners
October 12, 2026 05:58
boring-cyborg
Bot
added
area:go-sdk
area:task-sdk
area:ts-sdk
kind:documentation
labels
Before parsing a plain Python Dag file, the task process now sends the file's dagbag_import_timeout to the supervisor and clears it once parsing is done, so a parse stuck in native code is stopped from outside like an execute() that overruns execution_timeout. The value comes from the same get_dagbag_import_timeout policy the importer uses for the file's own alarm. Nothing is sent when the policy disables the timeout. Zipped Dags are not armed: the importer gives each member its own timeout, so a single deadline for the archive could be too short.
The message now carries the Dag import timeout as well as the task's execution_timeout, and 0 clears the deadline, so the old name described only one of its uses. It has not been released yet, so this is the last point where the name can change without a migration.
Andrushika
force-pushed
the
supervisor-parse-timeout
branch
from
October 12, 2026 06:03
cf21b7e to
fcdaddc
Compare
Andrushika
commented
Oct 12, 2026
Comment on lines
+313
to
+328
| what = StartupDetails( | ||
| ti=TaskInstance( | ||
| id=uuid7(), | ||
| task_id="a", | ||
| dag_id="super_basic", | ||
| run_id="c", | ||
| try_number=1, | ||
| dag_version_id=uuid7(), | ||
| queue="default", | ||
| ), | ||
| dag_rel_path="super_basic.py", | ||
| bundle_info=BundleInfo(name="my-bundle", version=None), | ||
| ti_context=make_ti_context(), | ||
| start_date=timezone.utcnow(), | ||
| sentry_integration="", | ||
| ) |
Contributor
Author
There was a problem hiding this comment.
I kinda want to turn the StartupDetails setup into a common fixture, since about 20 tests in this file build it inline.
That would be a separate follow-up though, happy to do it if maintainers think it's worth it
Contributor
Author
|
Another related: #69713 (it's a one-line change, ready for review but was closed due to the PR number limit policy) |
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
Follow-up of #73806. That PR lets the task supervisor stop a task process that runs past
execution_timeout, but the deadline is only armed right beforeexecute().Before that, the task process parses the Dag file under its own
dagbag_import_timeoutalarm. That alarm is the same SIGALRM handler, so it never fires when the import is stuck in native code, and nothing outside the process watches this window.What
Following the
SetTimeout(0)suggestion in #73806, the task process now sends the file'sdagbag_import_timeoutto the supervisor before parsing a plain Python Dag file, and0right after, which clears the deadline.The value comes from the
get_dagbag_import_timeoutcluster policy with the same file path the importer uses, so a file the policy gives more time is not stopped early, and nothing is sent when the policy disables the timeout. Since the message now covers both phases, I renamedSetExecutionTimeouttoSetTimeout. It is not released yet, so no schema migration is needed.Zipped Dags are not armed. Each member is imported under its own timeout, so one deadline for the whole archive could be too short.
related: #73806
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Opus 5.5) following the guidelines