Skip to content

Publish Dag results and inventories through the Execution API - #74790

Merged
ephraimbuddy merged 10 commits into
dag-processor-source-consistencyfrom
dag-processor-publication-api
Nov 6, 2026
Merged

ephraimbuddy merged 10 commits into
dag-processor-source-consistencyfrom
dag-processor-publication-api

Conversation

@ephraimbuddy

@ephraimbuddy ephraimbuddy commented Oct 15, 2026 •

Copy link
Copy Markdown
Contributor

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:

  1. FAB validation
  2. Source consistency
  3. Publication API
  4. Durable work delivery
  5. HTTP manager
  6. Complete processor API mode
  7. Helm and Compose deployment

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?
  • Yes — Codex (GPT-6)

Generated-by: Codex (GPT-6) following the guidelines

@ephraimbuddy
ephraimbuddy added this pull request to stack #74792 October 15, 2026 10:04
@ephraimbuddy
ephraimbuddy force-pushed the dag-processor-publication-api branch from a769ccc to 5f00fff Compare October 16, 2026 15:35
@ephraimbuddy
ephraimbuddy marked this pull request as ready for review October 16, 2026 15:37
@ephraimbuddy
ephraimbuddy force-pushed the dag-processor-publication-api branch from 5f00fff to 62cd459 Compare October 16, 2026 16:59
@ephraimbuddy
ephraimbuddy force-pushed the dag-processor-publication-api branch 2 times, most recently from 021b1e6 to 354bcfb Compare October 20, 2026 14:15
@ephraimbuddy
ephraimbuddy force-pushed the dag-processor-publication-api branch from 354bcfb to 93da855 Compare October 20, 2026 17:22
@ephraimbuddy
ephraimbuddy force-pushed the dag-processor-publication-api branch from 0d0ee05 to fa11b27 Compare October 28, 2026 20:16
@github-actions

github-actions Bot commented Oct 28, 2026 •

Copy link
Copy Markdown
Contributor

The re-run of the jobs stopped by EC2 Spot interruptions finished: attempt 2 passed.

@kaxil

kaxil commented Nov 4, 2026

Copy link
Copy Markdown
Member

@ephraimbuddy Conflicts

@kaxil kaxil left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

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's relative_fileloc, which is still allowed up to 2000. A 1025+ character path that fails validation hits DataError at flush and comes back as a retryable 500. Could every path in paths check against IMPORT_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 and DagModel locks. Could it skip Dags whose hash matches the stored dag_hash, or run before the locks?
  • routes/dag_parsing.py#L176: dag_ids includes 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: token is untyped in all three routes and _validate_job. jobs.py uses token: DagProcessorToken = CurrentDagProcessorToken, which also checks the type at runtime.
  • routes/dag_parsing.py#L59: this is a new runtime airflow.sdk import in the API server. The other core users keep DagSourceCode under TYPE_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
ephraimbuddy force-pushed the dag-processor-publication-api branch from b90c256 to ca49ca3 Compare November 5, 2026 11:44
@ephraimbuddy
ephraimbuddy force-pushed the dag-processor-publication-api branch from ca49ca3 to e153354 Compare November 5, 2026 13:11
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
ephraimbuddy force-pushed the dag-processor-publication-api branch from e153354 to 8d65973 Compare November 6, 2026 09:07
@ephraimbuddy
ephraimbuddy merged commit dac96bc into main Nov 6, 2026
172 of 189 checks passed
@ephraimbuddy
ephraimbuddy deleted the dag-processor-publication-api branch November 6, 2026 16:11
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants