Repository navigation
Publish Dag results and inventories through the Execution API - #74790
Merged
ephraimbuddy merged 10 commits intoNov 6, 2026
Merged
ephraimbuddy merged 10 commits into
ephraimbuddy merged 10 commits into
Conversation
1 task done
This was referenced Oct 15, 2026
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
October 16, 2026 15:35
a769ccc to
5f00fff
Compare
ephraimbuddy
marked this pull request as ready for review
October 16, 2026 15:37
ephraimbuddy
requested review from
XD-DENG,
amoghrajesh,
ashb,
bolkedebruin and
jedcunningham
as code owners
October 16, 2026 15:37
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
October 16, 2026 16:59
5f00fff to
62cd459
Compare
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
2 times, most recently
from
October 20, 2026 14:15
021b1e6 to
354bcfb
Compare
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
October 20, 2026 17:22
354bcfb to
93da855
Compare
kaxil
reviewed
Oct 28, 2026
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
October 28, 2026 20:16
0d0ee05 to
fa11b27
Compare
Contributor
|
The re-run of the jobs stopped by EC2 Spot interruptions finished: attempt 2 passed. |
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
October 29, 2026 16:55
fa11b27 to
1232175
Compare
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
October 30, 2026 15:40
1232175 to
b90c256
Compare
1 task done
Member
|
@ephraimbuddy Conflicts |
kaxil
approved these changes
Nov 4, 2026
kaxil
left a comment
Member
There was a problem hiding this comment.
Thanks, the per-Dag drop reads well and everything I raised earlier is in. A few smaller things on the current head:
datamodels/dag_parsing.py#L104: the 1024 bound covers the keys the processor sends, but the_reject_*helpers key errors by the Dag'srelative_fileloc, which is still allowed up to 2000. A 1025+ character path that fails validation hitsDataErrorat flush and comes back as a retryable 500. Could every path inpathscheck againstIMPORT_ERROR_FILENAME_MAX_LENGTH?collection.py#L706: full validation (schema, deepcopy,from_dict) runs on every Dag in every publication, unchanged ones included, while the route holds the Job, bundle andDagModellocks. Could it skip Dags whosehashmatches the storeddag_hash, or run before the locks?routes/dag_parsing.py#L176:dag_idsincludes Dags that get dropped as invalid later. A Dag that moves between zip members and fails validation in the same publication keeps its old row live on the old version. Should the sweep use the accepted ids?routes/dag_parsing.py#L143:tokenis untyped in all three routes and_validate_job.jobs.pyusestoken: DagProcessorToken = CurrentDagProcessorToken, which also checks the type at runtime.routes/dag_parsing.py#L59: this is a new runtimeairflow.sdkimport in the API server. The other core users keepDagSourceCodeunderTYPE_CHECKING. Is there a way to do the same here?test_dag_parsing.py#L376: could this also cover both Dags in one plain file (the sibling goes stale, as in DB mode) and an invalid Dag that already has a stored row?
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
November 5, 2026 11:44
b90c256 to
ca49ca3
Compare
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
November 5, 2026 13:11
ca49ca3 to
e153354
Compare
Isolated processors need to submit complete parsing results without access to the metadata database or server filesystem. Replayed requests and concurrent refreshes must preserve bundle ownership and persist metadata together with their receipts.
A complete inventory left Dags upgraded from Airflow 2 active after their files were gone, because rows never reparsed since the upgrade were skipped, while DB mode deactivates them on the next refresh. Publication receipts also outlived their Job: they are only read while the Job is open, yet stayed until db clean purged it, growing with every processor restart. Processor clients need the documented 403 reasons to tell a closed Job from a retryable failure, and listener authors need to know that import-error filenames are bundle-relative when results arrive through the API server. Combining the unreleased migrations keeps the 3.5.0 chain shorter without changing the head revision later layers build on.
A Dag can already be stale when its source disappears. Skipping those rows during inventory reconciliation leaves its asset references active indefinitely.
The schema allows a serialized Dag with no timetable, a single deadline alert, or the alert UUIDs a stored Dag keeps, but none of these can be stored. Through the Execution API they surfaced as server errors, which a Dag processor retries and then drops, so the author never saw an import error. Rejecting them as invalid gives the processor a reason it can report. Lang-SDK processors validate with the same check before storing, so they get the same answer.
Several guards in the publication API could be removed without a test failing: the revision check on parse results, the Job and bundle checks on inventory and the bundle catalog, the source-index check that keeps the API server out of Dag files, the grace period for members of a failed container, and the 2000-character limit on Dag locations. A regression in any of them would go unnoticed until it rejected valid publications, let a replaced processor reconcile a bundle, sent the API server to the processor's files, or staled Dags at the wrong time. Each now has a test that fails without it. Rejections also assert the reason, since processor clients use it to decide whether to retry. Import-error listeners that accept bundle_name and relative_fileloc load only on Airflow 3.5.0 and later. Earlier versions reject the hook, and the error stops every plugin listener from loading, so the docs say which version added the arguments and what a listener that must load on both should accept. A publication stales missing Dags without waiting for stale_dag_threshold. That looks inconsistent with DB mode, which waits only because its manager compares timestamps written separately; the route comment now says so. The remaining checkpoint names in the publication code and tests become receipt, to match the table they refer to, and the source_codes description says who deduplicates the text. The migration comment explains why the SQLite downgrade turns foreign keys off: rebuilding dag_bundle would fail on the dag rows that reference it and delete its team rows.
One Dag that failed validation, or used another team's plugin, made the API server reject its whole file or container. DB mode drops only that Dag and records the error under its own path, so in a zip archive the other members' Dags stay active. Through the Execution API the error landed on the archive, and every member went stale after stale_dag_threshold. Validation runs after the plugin and timetable checks DB mode already has, so a timetable that raises on load keeps the traceback DB mode records. Problems with the publication itself still return a 422.
A Dag processor with no result for a file, because the parse timed out or crashed or the API server refused the result, publishes an import error so users can see why. For a plain file that error marked every Dag in it stale at once, including Dags another processor had published seconds earlier. DB mode records no import error for a timeout and keeps the Dags until they outlast stale_dag_threshold, so these reports are now flagged and their Dags keep that threshold, judged by each Dag's last parse. A genuine import error still deactivates the file's Dags at once. The stale sweep also skipped every Dag a publication named, including the ones dropped later as invalid. When such a Dag had moved to another member of the same container, its previous version stayed live. Each publication validates all of its Dags, unchanged ones included, and did so while holding the Job, bundle and Dag row locks. Other publications for the bundle waited on it, and a scheduler that skips locked Dag rows passed these over, though the checks need none of the locks. The Dag parsing routes took an untyped execution token, while the jobs routes refuse any other token type before the handler runs. Main added migrations 0152 and 0153, so the publication-state migration is now 0154.
A report that a Dag processor has no result for a file kept the file's Dags from going stale but still set has_import_errors on them. That flag also stops the scheduler from creating their runs and refuses manual and operator triggers, so the Dags stopped running at once, which a DB-mode timeout never does. It also broke the scheduler's assumption that has_import_errors is only set together with is_stale, which the partitioned-asset path relies on. The report now records the import error without touching the Dags, so they keep running until they outlast stale_dag_threshold. Checking Dags before taking the publication's locks also moved the read of the bundle's team, used to refuse another team's plugin classes, out from under the bundle lock. On MySQL that lock waits for a team reassignment to commit, so the team check now runs again after it.
Main added migration 0154 (idempotency key on hitl_detail), so the Dag processor publication-state migration moves to 0155 and revises it.
ephraimbuddy
force-pushed
the
dag-processor-publication-api
branch
from
November 6, 2026 09:07
e153354 to
8d65973
Compare
1 task done
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.
Add transactional publication receipts, bundle revisions, complete inventory reconciliation, and parse-result persistence to the Execution API. Publication validates bundle ownership and preserves atomicity across authorization callbacks without reading the processor's source files on the API server.
Layer 3 of the seven-part replacement for #74233, in GitHub native stack #74792. Depends on #74789. Base:
dag-processor-source-consistency.GitHub's Files changed tab shows this layer. Compare this layer.
Merge order:
Focused SQLite tests: 262 passed, 6 skipped. PostgreSQL: 378 passed, 3 skipped, followed by 9 passing targeted cases covering initial timeouts and session-call updates. Regular and manual static checks passed. The complete migration round trip also passed with the following work-delivery layer.
Was generative AI tooling used to co-author this PR?
Generated-by: Codex (GPT-6) following the guidelines