dheerajturaga opened a new pull request, #75342: URL: https://github.com/apache/airflow/pull/75342
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 is where a delivery resumes a waiting task. After a delivery records its topic events and asset route events, the same transaction hands it to the open waits of its webhook that it matches. A parked task resumes at once, a running task's wait is marked received and resumes when the task parks (#75341), and a wait whose attempt can no longer resume is ended and ignored. A delivery with a correlation key that nothing took is kept `early` for a wait registered later (#75340). Points to look at: - Outcomes, first match wins: `resumed_wait` if a wait took the delivery, `accepted` if it recorded events, `early` if it has a correlation key, else `no_subscribers` or `dag_paused`. A delivery that recorded events is never kept early, so a wait registered after it cannot claim it. This changes what earlier layers record: a delivery with a correlation key that no Dag subscribes to, or only paused Dags do, is now `early`, not `no_subscribers` or `dag_paused`. A redelivery of an `early` or `expired` delivery is processed again in full, as one of a `no_subscribers` or `dag_paused` delivery is, so unpausing a Dag and redelivering still starts it. - Payloads: a delivery that goes to a wait or is kept `early` keeps what `store_payload` keeps of its payload, which the resume event carries. An `early` delivery keeps it until its record is removed, so every `github` delivery about a pull request or an issue that no Dag subscribes to now keeps its payload for `delivery_retention`, where it kept none before. `[webhooks] early_delivery_max_per_webhook` caps the live early deliveries of a webhook; past it, the oldest become `expired`, a new outcome, and lose their payload. - Concurrency: the delivery reads only the waits whose match keys its attributes cover, by key digest, or every open wait of the webhook when it has more than 8 attributes. It locks all their task instances in one query, in id order, before it claims or ends any wait, because a retry's registration ends the wait of its earlier try without that try's task instance lock. Each claim is a compare-and-set, which also holds on SQLite. An early delivery looks for waits again after its commit, as a registration does, so exactly one side pairs them. If that second look fails, the delivery is answered 5xx but keeps its record, the one exception to "a 5xx records nothing". `test_waits_concurrency.py` holds each side, on real connections, where a design without the second look would lose the delivery. - When a retry takes over the key of an earlier try's wait, that wait is taken out of the delivery it had received. A delivery that no other wait holds then gets the outcome it would have had without the wait: `accepted` if it recorded events, or `early` again if it has a correlation key, so the new try can claim it. A delivery with neither keeps the ended wait, since its record cannot tell `no_subscribers` from `dag_paused`. The webhooks docs describe the new outcomes and the waits in a delivery's record and audit log, and `security_model.rst` extends the replay residual risk to waits. {{PR-29}} adds labels for the new outcomes to the deliveries page. This is one of the four PRs of the webhook waits unit, with #75340, #75341 and {{PR-28}}. They merge together once all four are approved, so that no release can register a wait that nothing resumes. --- ##### 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]
