ephraimbuddy commented on code in PR #74794:
URL: https://github.com/apache/airflow/pull/74794#discussion_r4490966410
##########
airflow-core/docs/security/jwt_token_authentication.rst:
##########
@@ -557,14 +587,165 @@ parsing subprocess can without this client.
``Variable.set`` upserts on the key
multi-team mode a write can overwrite another team's Variable with the same
key, as it can with task
tokens.
-The client attempts renewal at 80% of token lifetime. Transient early-renewal
failures leave
-the existing token usable until expiry. A retired registration signals that
its owner must
-restart. Completion retains its original identity and outcome across retries.
-
-File and attempt identity are carried in the token but are not currently used
for access
-checks or request attribution. Archives retain one file identity under the
current processor
-model. These credentials do not isolate hostile code from other credentials
accessible on
-the processor host; the process and deployment limitations above still apply.
+For each ``dag_parse`` token, the manager chooses the bundle, bundle-relative
file location,
+and parsing-attempt UUID.
+
+The manager publishes completed imports through ``POST
/jobs/{job_id}/parse-results``
+using its Job credential. Parsing tokens cannot publish. A request contains
one file or
+container's serialized Dags, diagnostics, source text (or an explicit
unavailable marker),
+and the bundle version captured at dispatch. The API never reads the
processor's source
+paths. It checks bundle access, existing Dag ownership and team-scoped plugin
classes
+before deserializing results and rechecks ownership on the rows used for
persistence.
+A Dag can move between files within its bundle. Recovery from a removed bundle
is
+limited to stale legacy rows with neither a relative file location nor a Dag
version.
+
+Each request carries an attempt UUID and a dispatch sequence within its Job. A
publication
+receipt and the metadata writes commit together. Replaying the latest
identical request returns
+the saved receipt; conflicting payloads and superseded sequences are rejected.
Publication
+receipts retain one entry per Job and source and are removed with Job cleanup.
There is no total
+ordering between different Jobs publishing a source without versioning; an
already-accepted
+request's retry does not overwrite another Job's result. Listener side effects
are not
+transactional and may repeat after a failed transaction.
+
+Publication and inventory request bodies are limited to 16 MiB, and the API
refuses a
+larger body before reading the rest of it. A rejected publication does not
fall back
+to direct database writes, and does not prevent other files from being
processed. Transport
+failures and server errors are retried; exhausting those retries preserves the
last accepted
+result. When the API refuses a result for its content, or the result is over
the limit, the
+processor publishes an import error for the file instead, so the problem is
visible.
+Dags whose definition path matches an import error are marked stale until a
later parse
+publishes them again. Dags missing from an accepted parse are marked stale at
once: the
+publication is atomic, so there is no gap between file and Dag parse times to
wait out. For a
+container-level error, missing member Dags are marked stale only when their
last successful
+parse is older than ``[dag_processor] stale_dag_threshold``, using the
server's clock. Invalid ``access_control`` produces an import error with both
+current and supported older FAB releases. Permission backend failures remain
retryable
+server errors. Empty imports still publish a receipt and clear the file's
earlier import
+errors. A bundle refresh, including one forced by a parse request, waits for
that bundle's
+active imports instead of discarding them; only initializing a bundle again
invalidates and
+requeues older in-flight results from that manager.
+
+Inventories report discovered definitions and sources whose discovery failed
separately.
+Failed sources preserve their existing definitions and diagnostics, including
archive members;
+other sources still undergo deletion reconciliation. Each inventory carries
the previously
+observed server revision; conflicting refreshes must
+rediscover before retrying. Imports carry the accepted revision, so an older
snapshot cannot
+publish after a changed inventory. Only accepted publications establish
missing Dags within
+a file; rejected requests and incomplete discovery do not establish absence.
Every
+``[scheduler] parsing_cleanup_interval``, the scheduler marks the Dags of
inactive bundles stale,
+removes stale warnings, and ends expired processor Jobs, in bounded batches
and in
+both modes.
+
+Callback and priority requests remain stored while claimed. A lost claim
response or
+acknowledgment can be retried. Claims use increasing sequences per Job and
work type; the
+client waits for a claim response before sending the next sequence for that
type.
+Retrying an empty claim remains empty, even if work arrives
+afterward, and retrying an older claim cannot consume newly requested work.
Requests from a
+retired Job become available to another processor. A callback acknowledgment
confirms
+delivery, not successful execution of user code. As with other at-least-once
delivery, a crash
+after a callback's side effect and before acknowledgment can repeat that side
effect. Before the
+processor's Job ends, a graceful shutdown acknowledges callbacks whose
processes have already exited
+cleanly; callbacks that are still running or that the shutdown signal stopped
are delivered again.
+When a callback's bundle
Review Comment:
Updated in `a725ab9fe6`. The paragraph now says the same retry applies after
an Execution API request fails, but only when none of the callbacks processed
together for a file found its Dag or task. When one of them did, a callback
whose Dag or task did not load is acknowledged without running.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -2169,8 +2245,23 @@ def _collect_results(self):
for file in finished:
processor = self._processors.pop(file)
+ if processor.had_callbacks:
+ self._finish_callback_attempt(processor)
processor.close()
+ def _finish_callback_attempt(self, processor: BaseDagFileProcessorProcess)
-> None:
+ # Running a callback again repeats its side effects, so retry only a
process that ran none of them.
+ if processor.api_request_error is None or not
processor.callback_targets_missing:
Review Comment:
Done in `a725ab9fe6`. The two recovery tests are now one table,
`test_callbacks_run_again_after_api_failure_only_when_none_ran`, over both
values of `api_request_error` and `callback_targets_missing`, plus the batched,
persistent-error and timeout rows. The `no-target-loaded-without-api-failure`
row expects an immediate acknowledgment, and fails if the `api_request_error is
None` check is dropped. After the exit-status fix, a removed Dag exits 0, so
that state now only comes from database mode.
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -257,8 +261,11 @@ def _parse_file(msg: DagFileParseRequest, log:
FilteringBoundLogger) -> DagFileP
)
if msg.callback_requests:
+ targets_loaded = any(_is_callback_target_loaded(bag, request) for
request in msg.callback_requests)
Review Comment:
You're right: with `any()`, a batch where one Dag loaded exits 0, and the
callbacks of the Dag that didn't load are acknowledged. Retrying the batch
would run the loaded Dag's callbacks twice, and retrying only the missing ones
needs the child to report which callbacks ran, which changes the supervisor
schema the other language SDKs generate code from. So I kept the limit and
documented it in `a725ab9fe6`: the `job_heartbeat_timeout` description now says
a retry only happens when none of the callbacks processed together for the file
found its Dag or task. The `_parse_file` test (now
`test_parse_file_exits_when_no_callback_target_loaded_after_api_failure`) has a
`one-of-two-dags-loaded` case.
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -257,8 +261,11 @@ def _parse_file(msg: DagFileParseRequest, log:
FilteringBoundLogger) -> DagFileP
)
if msg.callback_requests:
+ targets_loaded = any(_is_callback_target_loaded(bag, request) for
request in msg.callback_requests)
# If the request is for callback, we shouldn't serialize the Dags
_execute_callbacks(bag, msg.callback_requests, log)
+ if not targets_loaded:
+ sys.exit(_CALLBACK_TARGETS_MISSING_EXIT_CODE)
Review Comment:
Fixed in `a725ab9fe6`. The child now exits with 75 only when none of its
callbacks found its Dag or task and the Dag processor also failed to answer one
of its requests through the Execution API, the one case where a retry can help.
Its comms record the `API_SERVER_ERROR` the supervisor sends back for a failed
request. So database mode and a removed Dag exit 0 again, as before this PR. I
left the supervisor's warning alone, since skipping it would change the Task
SDK's `WatchedSubprocess` for every process type. In database mode, 75 can
still follow an in-process request failure; that run already logs the failed
request, and the manager acknowledges it.
--
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]
