dheerajturaga opened a new pull request, #75576: URL: https://github.com/apache/airflow/pull/75576
Part of [AIP-125 (Inbound Webhooks)](https://lists.apache.org/thread/p3dpnor4zwko8tldpncj6ftj1bpp18q7), which lets external services such as GitHub trigger Dags by sending webhooks to Airflow. This PR lets a Dag or a task declare which webhook deliveries cancel it, with `cancel_on=[WebhookMatch(...)]`. For example, a preview environment's run fails when its pull request is closed, and its end-to-end tests are skipped when a new commit is pushed to the pull request, while the runs for other pull requests go on. It has one commit per part and is best reviewed commit by commit: commits 1 and 2 cover the shared grammar and the Task SDK, commit 3 covers serialization and the Dag processor, and commits 4 and 5 cover webhooks and the scheduler. This is Phase 3 of AIP-125, cancelling runs from webhooks. It is not part of the vote on Phases 1 and 2, and targets Airflow 3.6. A delivery only records what to cancel, and the scheduler does the cancelling. The webhooks process faces the internet and has to answer within the sender's timeout: 10 seconds for GitHub, 3 for Slack. Cancelling inside the delivery's transaction would take longer as Dags and runs grow. It would also need Dags loaded in that process and would wait on run locks that the scheduler holds. Instead, a delivery with a correlation key reads the conditions that the Dag processor stored for its webhook. It records one cancel request per matching Dag, in the same transaction as the delivery's record. About once a second, the scheduler applies the requests to the queued and running runs that are about that key. A run is about the key if it was started from an event that an earlier delivery to the same webhook with the key recorded, or if it was queued before the delivery with the key in its conf. ### Commit 1: AIP-125: Move the webhook match grammar into the shared library This commit moves the match grammar of webhook waits, with its evaluator and digests, from `airflow.webhooks.waits` into the `webhook_naming` shared library, as its `matching` module. The Task SDK has to refuse a malformed `cancel_on` condition when the Dag file is parsed, and core may not import the SDK, so both need the same copy. Webhook waits behave as before. Points to look at: - Three names change. `normalize_events` and `normalize_match` become public, because a condition declared in the SDK keeps their canonical form. `validate_wait` becomes `validate_match`, because it now checks any condition, not only a wait. - `airflow.webhooks.waits` re-exports nothing. It keeps only the resume and timeout events, which depend on serde. Core imports the grammar from `airflow._shared.webhook_naming.matching`, and the SDK from `airflow.sdk._shared.webhook_naming.matching`. - The grammar's tests move with it, to `shared/webhook_naming/tests/webhook_naming/test_matching.py`. ### Commit 2: AIP-125: Add WebhookMatch and cancel_on to the Task SDK This commit adds three things to the SDK, all marked experimental: - `WebhookMatch(webhook, *, events=None, match=None, correlation_key_conf=None)`, exported from `airflow.sdk`; - `DAG(cancel_on=...)`, which fails the runs; - `BaseOperator(cancel_on=..., cancel_state=...)`, which ends a task as `failed` (the default) or `skipped`. A condition is checked against the shared grammar when it is created. A typo is therefore an import error rather than a Dag that never cancels. Points to look at: - Canonical form. A condition keeps `events` and `match` sorted, without repeats and with values as strings. Conditions written differently are therefore equal, and a Dag file parsed twice gives the same Dag. - `correlation_key_conf` names a top-level conf key of at most 250 characters. The bound is defined once in the shared library, so the SDK check and the column that stores conditions (commit 3) agree. - Refused when the Dag file is parsed: - `cancel_on` on a teardown, which runs however its run ends; - `cancel_on` or `cancel_state` passed to `expand()`, since the scheduler reads them from `partial_kwargs` and they cannot differ between map indexes; - `cancel_state` without `cancel_on`. Given through `expand_kwargs()`, whose keys are known only at run time, they fail the task when it is unmapped. - Both fields can come from the Dag's or a task group's `default_args`. A teardown then passes `cancel_on=None`, and a task without `cancel_on` ignores an inherited `cancel_state`. - `cancel_state` defaults to None, which means `failed`. That way, a `cancel_state` given without `cancel_on` can be told apart from the default. - The new fields stay out of generic serialization, which would write each condition with `str()`. Commit 3 serializes them explicitly. ### Commit 3: AIP-125: Serialize cancel_on and store Dag cancel conditions The scheduler and the API server only see Dags in serialized form, so this commit adds `cancel_on` and `cancel_state` to it. The Dag processor also stores each condition in a new `dag_cancel_condition` table, indexed by webhook. The webhooks endpoint has to find the conditions that a delivery matches without loading any Dag. Points to look at: - Hashes and versions: - Conditions are written sorted, without repeats, and left out when unset. A Dag without `cancel_on` keeps its hash, and conditions written in another order make no new Dag version. - A `cancel_state` that a task inherits without having `cancel_on` is not stored. - The version diff treats a change to `cancel_on` as a change to how runs execute. - Lang-SDK Dags. Core decodes the conditions into `SerializedWebhookMatch` and checks them against the grammar. A Dag from another language SDK with a malformed condition is therefore refused when it is validated, instead of being stored. The schema's `webhook_match` is closed, so a condition with a field that this version does not know is refused rather than read as a wider condition. The Go, TypeScript and Java SDKs get the new schema, and keep the fields out of their authoring types until they model `cancel_on`. - `SerializedDAG.has_cancel_on` lets the scheduler cheaply skip Dags that have no conditions (commit 5). - The table's primary key includes a digest of the whole condition. `correlation_key_conf` is nullable, and NULLs never collide in a unique key, so a key over the columns would let one condition be stored twice. A Dag's own conditions use the task id `""`. - Cost on the Dag processor: - The stored rows are loaded through the Dag model's relationship. That takes one more query per batch, whether or not any Dag has `cancel_on`, so `FIXED_PER_CALL` in `test_manager.py` goes from 9 to 10. - Unchanged conditions cost no writes, and a Dag that drops its last condition loses its rows. - Rows are deleted with their Dag. A stale Dag keeps its rows, since a cancel only reduces work. ### Commit 4: AIP-125: Record cancel requests from webhook deliveries This commit is where a delivery asks for a cancel. A delivery with a correlation key reads the `dag_cancel_condition` rows of its webhook and evaluates them with the shared grammar. It then inserts one `webhook_cancel_request` row per matching Dag, in the delivery's transaction. It loads no Dag, looks for no run, and writes no run or task instance state. Commit 5 applies the requests. Points to look at: - Outcomes, first match wins: `resumed_wait` > `accepted` > `cancel_requested` (new) > `early` > `dag_paused` > `no_subscribers`. - This order differs from the first design, which ranked `cancel_requested` above `accepted` and answered a redelivery 200 duplicate. That order would have broken behaviour of the earlier phases: a wait registered later could not claim the delivery, and redelivering after unpausing a Dag would not start it. - A `cancel_requested` delivery is kept as an `early` one is. A wait registered later can claim it. It counts toward `[webhooks] early_delivery_max_per_webhook`, and it expires at that cap but keeps its requests. - A retry can take a delivery back from an earlier try's wait. If that delivery recorded no events and has a correlation key, it becomes `cancel_requested` again if it recorded requests, and `early` if it did not. - A delivery that recorded events or went to a wait keeps `accepted` or `resumed_wait`. Its `actions` also list its requests, under `cancel_requests`. - Redeliveries: - A `cancel_requested` delivery is processed again in full, as an `early` one is, and answered 202. Unpausing a Dag and redelivering therefore still starts it, and a Dag that declared `cancel_on` since then gets its request. - A delivery never requests twice for the same Dag. A unique key on `(inbound_delivery_id, dag_id)` and a read of the requests that the delivery already has make the insert idempotent, also between concurrent copies. - Only `accepted` and `resumed_wait` redeliveries are answered 200 duplicate. - Targets: `{"run": {"correlation_key_confs": [...]} | null, "tasks": [{"task_id", "cancel_state", "correlation_key_confs"}]}`, with both keys always present. A request records every target that matched. Commit 5 decides for each run whether the Dag-level target or the task targets apply, because a Dag-level condition need not apply to every run that a task's condition applies to. - Cost at the endpoint: - one indexed read of the webhook's conditions; - one evaluation per distinct `events` and `match`, because a condition from `default_args` is stored once per task; - one insert per matching Dag. A delivery without a correlation key reads nothing. Paused, draining and stale Dags still get requests. - Locks: the requests are inserted after the delivery's row, so each copy takes its locks in the order task instances, waits, delivery, requests. - Retention: - Requests are kept until their delivery's `retain_until`, whatever their status, because a run that the scheduler creates later from an older event still has to find them (commit 5). - `airflow db clean` ages them on that column and selects them by Dag, and they are deleted with their Dag. - `inbound_delivery_id` is not a foreign key, as for `inbound_wait.delivery_id`, because `db clean` ages each table on its own column. - Audit log: each copy of a delivery that records requests writes one entry that lists them, rather than one entry per Dag. Most closed pull requests find no run to cancel, and the scheduler writes an entry for each run and task instance that it changes. - Multi-team: Dags with a team get no requests, since webhooks have no team yet. - `test_pipeline_concurrency.py` checks that concurrent copies of a delivery that requests a cancel record one request. With `test_waits_concurrency.py`, it passed 50 runs out of 50 on Postgres and on MySQL. ### Commit 5: AIP-125: Apply webhook cancel requests in the scheduler This commit is where a cancel happens. The scheduler looks for pending requests about once a second, on a timer of its own rather than on the 15-second awaiting-input sweep. It applies each request to the unfinished runs of its Dag that are about its key and older than the delivery. It applies up to `[webhooks] cancel_requests_per_loop` requests per look (default 20), each committed on its own. A cancel thus typically lands within a second or two of the delivery's answer, later under load or while no scheduler runs, and a running task is killed at its next heartbeat. When nothing is pending, a look costs one indexed query. Points to look at: - A cancel is a failure, since Airflow has no cancelled state. A Dag-level cancel fails the run as marking it failed does: running and parked task instances fail, the others are skipped, and finished ones are left alone. - The Dag's `on_failure_callback` runs with `context["reason"] == "cancelled"`, as it runs with `timed_out` for a run that timed out. - If the run has no note, it gets "Cancelled by webhook <name> delivery <id>". - The Dag's `max_active_runs` flag is recomputed, so that its next run can start. - It does not reuse `set_dag_run_state_to_failed`, which selects running tasks by task id and would also fail the finished map indexes of a running mapped task. - A task-level cancel gives the unfinished task instances of the matched tasks their `cancel_state`, and the run goes on. - Either way: - the open webhook waits of the changed task instances are ended; - one `webhook_cancel` audit entry is written per cancelled run or task instance; - listeners are called after commit; - `webhooks.cancels` counts the cancels by webhook and scope. - Teardowns are left to run. A run that a teardown holds ends when its teardowns finish, as its leaf tasks make it end, and gets no `cancelled` callback. - Which runs: - A run is about the key if it consumed an earlier event that a delivery to the same webhook with the key recorded. - It is also about the key if it was queued before the delivery with the key in its conf, under a conf key that a matched condition names in `correlation_key_conf`. - A provenance rule leaves out events from task outlets, the REST API and triggers, because whoever creates those can give them the same extra. - A run that only its conf makes about the key, and that was cleared after the delivery, is queued again, so the older request leaves it alone. A run that an event started is cancelled even if it was cleared after the delivery. - Claiming. Each request is claimed with `FOR UPDATE SKIP LOCKED` in a transaction of its own. Locking a batch and committing after each request would release the other rows' locks at the first commit, and another scheduler could apply them too. - A request that raises is marked `failed`, with the exception's class name in its result, so that it leaves the head of the queue. - Only an `OperationalError`, which may be transient, leaves the request pending for the next look. - Only what changes is recorded, so applying a request to a run again is harmless. - Runs are locked before their task instances, in the order that the scheduler and the Execution API use. - Dag versions: - Each run is cancelled with its own Dag version, so a target task missing from that version is left alone. - A run whose version cannot be loaded is listed as skipped (`dag_not_found`) in the request's result. - The targets were fixed when the delivery arrived (commit 4). - Runs created later. The scheduler holds an asset-triggered run back while its Dag is at `max_active_runs`, and creates an asset-gated run only when its schedule is due. A request can therefore be applied before the run it is about exists. When the scheduler creates an asset-triggered run (partitioned or not) or an asset-gated run for a Dag with `cancel_on`, it applies the pending and applied requests received after the run's delivery events. - This happens in a savepoint, so that a failure never stops run creation. A transient database error propagates, so that the creation is retried. - The scheduler records the run on a request only if no other scheduler holds that request, so that creating runs never waits for one. - The docstring of `airflow.webhooks.cancel_application` names the two races that this leaves open. First, with more than one scheduler: a request committed after a creating scheduler read the requests, and applied by another scheduler before the new run commits. Second: a request applied while the API transaction of a run triggered with the key in its conf is still open. - Multi-team: the runs of a Dag with a team are left alone and listed as skipped (`dag_has_team`), as the endpoint records no request for them. - The scheduler runs no Dag code to apply a request. The `on_failure_callback` goes to the Dag processor as a callback request, as for a run that timed out. Two helpers become shared: `build_task_instance_coordinates`, used by both the delivery actions and the cancel results, and `RUNNING_STATES` in `mark_tasks`. #### Tried end to end With the whole stack applied, including the next PR in this stack and the demo kit (#75168, which gains `demo.sh 3` with this stack): - the affected test suites (webhooks, serialization, Dag processing, the new models, db clean and the scheduler job) passed on SQLite, Postgres and MySQL, and the migration checks on Postgres and MySQL; - `demo.sh 3` checks cancelling on Postgres and on MySQL. A live run also sent the deliveries of one real GitHub pull request through an ngrok tunnel to a separate `--apps webhooks` process: - the label started `preview_environment`; - a new commit skipped its `run_e2e_tests`; - the merge resumed the webhook wait of the `merge_gate_operator` Dag and cancelled the preview run. The run got its note, `on_kill` ran, and `on_failure_callback` got the reason `cancelled`. Not done yet: - measuring the scheduler's cost under load; - a characterization test of a trigger's `on_kill` for a deferred task; - a characterization test of `on_failure_callback` after `SERVER_TERMINATED`. #### Performance Base is layer 32 of this stack; head is this PR with the next PR in this stack. The runs alternate base and head on the same host, report medians, and come with controls: `timezone.parse`, `json.dumps` and a fixed SQLite `SELECT`. A round whose control moved more than 10% was dropped. Wall time on this machine cannot resolve differences under about 2–3%: an A/A run of two identical base worktrees differed by up to ±1% in medians, and by up to 6% per round. So the serialization rows also give the deterministic instruction count per call. | Path, for a Dag without `cancel_on` | Base | Head | Change | |---|---:|---:|---| | `DagSerialization.from_dict`, 500 tasks | 12.57 ms | 12.65 ms | +0.6% wall (+0.6% to +2.5% across five runs); instructions +0.69% | | `DagSerialization.from_dict`, 100 tasks | 2.701 ms | 2.682 ms | -0.7% | | `DagSerialization.to_dict`, 500 tasks | 46.81 ms | 46.48 ms | -0.7%; instructions +0.07% | | Building the Dag at parse time, 500 tasks, with `DAG.validate()` | 24.20 ms | 24.39 ms | +0.8% (+0.5% to +2.2% across five runs); instructions +0.91% | | `SerializedDAG.bulk_write_to_db`, rewriting 50 unchanged Dags (SQLite) | 104 statements | 105 statements | +1 `SELECT dag_cancel_condition` per call (the time is within noise) | | Asset-triggered run creation, 10 Dags | 152 statements | 152 statements | +0.5% to +0.9% time; the `cancel_on` gate costs 66 ns per run | | Webhook delivery with a correlation key | 7 statements | 8 statements | +1 indexed `SELECT dag_cancel_condition`, pinned by a test; 0.12 ms in-process | | The scheduler's new cancel step, nothing pending | — | 0.17 ms | 1 `SELECT` (and its transaction) per call, once a second | Profiling an earlier version of this PR showed that the Task SDK and the serializer did work for every task, even without `cancel_on`: - from_dict on 500 tasks executed 4.7% more instructions; - building a 500-task Dag executed 2.3% more. The fix does that work only for tasks that declare `cancel_on`, and leaves the small costs above. The opt-in costs grow with what is declared: - `default_args` `cancel_on` on 500 tasks makes the Dag 13% to 14% slower to serialize and deserialize, and its `dag_hash` 35% slower; - its 1,000 condition rows make the first write 20% slower and a rewrite 10% slower. Once any Dag names a webhook in `cancel_on`, every keyed delivery to that webhook reads its conditions. With 1,000 distinct conditions that do not match, a delivery took about 8 ms longer, whichever Dag it was for. Of that, the query took 2.7 ms and the evaluation 5.4 ms. This was measured only at 1,000 conditions. The query grows with the condition rows (a `default_args` condition is stored once per task), and the evaluation with the distinct conditions. These are acceptable for an opt-in feature: each delivery stays far inside the 3-second response window of the strictest sender. All of these were measured on SQLite. On Postgres and MySQL, each statement also adds a round trip. ### Merging This PR is the cancel unit: its commits merge together, so that no release accepts a `cancel_on` that does nothing. It sits above #75347 and below the demo kit (#75168), so it merges after everything below it in the stack, #75347 last. Its two migrations, `dag_cancel_condition` and `webhook_cancel_request`, are stamped 3.6.0, the release the AIP targets for Phase 3, while the earlier AIP-125 migrations are stamped 3.5.0. If this phase ships in 3.5 instead, they are restamped. The next PR in this stack does three things: - shows the conditions and the requests in the UI, the REST API and `airflow webhooks check`; - documents `cancel_on`, including its grants and the security model's new residual risks; - holds the newsfragment, which says that a Dag using `cancel_on` needs the Task SDK of the Dag processor and of the workers upgraded with core. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Opus 5.5) Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) --- * Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged. * For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed. * When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x). * For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number. -- 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]
