dheerajturaga opened a new pull request, #75343: URL: https://github.com/apache/airflow/pull/75343
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 lets an operator wait for a webhook delivery. `self.wait_for_webhook(webhook=..., events=..., match=..., method_name=..., timeout=...)`, called from `execute`, registers the wait through the supervisor (#75340) and parks the task in `awaiting_input`, holding no worker slot and no trigger. The task resumes in `method_name` with the delivery, or with the timeout event once the deadline the API server set has passed. `WaitForWebhookOperator` ({{PR-30}}) and `@task.webhook_wait` ({{PR-31}}) build on it. Points to look at: - Against an API server without webhook waits, it raises `WebhookWaitUnsupported`, so an operator with another way to wait, such as a polling trigger, can fall back to it. Only the 404's body tells the two servers apart: the route always answers with a `reason`, and a server without the route answers a bare 404. Any other refusal fails the task with the server's status, reason and message. - The registration is sent once, without the client's retries. The server may have committed it, with an early delivery it claimed, before the response was lost, and a repeat would end that wait and leave the delivery with it. A failed registration fails the try instead, and the next try takes the delivery back when it takes over the wait's key. Letting a repeated registration in the same try get back the delivery its ended wait had received is left as a follow-up. - Only the API server checks events, match and timeout against the match grammar and its bounds, so a worker never refuses a wait that a newer server accepts, such as one using the match operators that dict values are reserved for. The SDK keeps no copy of the grammar: it turns tuples, ranges and lazy XCom sequences into lists, and refuses a timeout that is not a `timedelta`, which the request would read as seconds. - The supervisor schema gains the two messages and the `WEBHOOK_WAIT_UNSUPPORTED` error type. The error type is recorded as a change in the in-progress schema version because it changes `ErrorResponse`, which older runtimes already know. The Go and TypeScript SDK copies of the schema, and the models generated from them, are refreshed. `dag.test()` and the `awaiting_input` docstrings no longer describe that state as Human-in-the-loop only. This is the last of the four PRs of the webhook waits unit, with #75340, #75341 and #75342. They merge together once all four are approved, this one last, so that no release can register a wait that nothing resumes. ### Performance How these were measured: | Path | Base median | Head median | Change | Control base / head | |---|---:|---:|---:|---| | Decode `GetVariable` through `ToSupervisor` | 0.511 µs | 0.516 µs | +1.0% | `Asset(name, uri)` 3.849 / 3.868 µs (+0.5%) | | Decode `GetConnection` through `ToSupervisor` | 0.510 µs | 0.516 µs | +1.1% | (same runs) | | Decode `SetXCom` through `ToSupervisor` | 1.990 µs | 1.989 µs | -0.0% | (same runs) | | Decode `GetXCom` through `ToSupervisor` | 0.819 µs | 0.827 µs | +1.0% | (same runs) | | Decode `VariableResult` through `ToTask` (task side, `CommsDecoder._from_frame`) | 0.808 µs | 0.827 µs | +2.4% | (same runs) | | Decode `ErrorResponse` (variable not found) through `ToTask` | 0.803 µs | 0.824 µs | +2.5% | (same runs) | | Decode `ConnectionResult` through `ToTask` | 0.990 µs | 1.017 µs | +2.8% | (same runs) | | Decode `XComResult` through `ToTask` | 1.841 µs | 1.875 µs | +1.9% | (same runs) | | Supervisor dispatch `GetVariable` (msgpack frame, `ToSupervisor` decode, handler lookup and run, response frame) | 9.861 µs | 9.898 µs | +0.4% | (same runs) | | Supervisor dispatch `GetVariable`, variable not found | 4.063 µs | 4.038 µs | -0.6% | (same runs) | | Supervisor dispatch `GetConnection` | 10.033 µs | 9.958 µs | -0.7% | (same runs) | | Supervisor dispatch `SetXCom` | 5.010 µs | 5.015 µs | +0.1% | (same runs) | | Supervisor dispatch `GetXCom` | 10.408 µs | 10.500 µs | +0.9% | (same runs) | | Supervisor dispatch `GetVariable`, runtime pinned to schema 2026-06-16 (Go/TS SDK path, cadwyn upgrade and downgrade) | 18.982 µs | 19.058 µs | +0.4% | (same runs) | | Same, variable not found (`ErrorResponse` downgraded) | 13.209 µs | 13.479 µs | +2.0% | (same runs) | | Build `TypeAdapter(ToSupervisor)` (once per supervisor process) | 634 µs | 657 µs | +3.7% | (same runs) | | Build `TypeAdapter(ToTask)` (twice per task process) | 390 µs | 402 µs | +3.3% | (same runs) | | Isolated `SchemaVersionMigrator.downgrade(ErrorResponse, "2026-06-16")` | 3.726 µs | 3.716 µs | -0.3% | `timezone.parse` 3.604 / 3.585 µs (-0.5%) | | Isolated `SchemaVersionMigrator.downgrade(VariableResult, "2026-06-16")` | 3.426 µs | 3.462 µs | +1.1% | (same runs) | | Fresh interpreter: `import airflow.sdk.execution_time.supervisor` | 100.8 ms | 99.7 ms | -1.0% | `import airflow.sdk.timezone` 738.0 / 720.2 ms (-2.4%) | | Fresh interpreter: `import ...task_runner` after the supervisor | 222.1 ms | 212.9 ms | -4.1% | (same runs) | | Fresh interpreter: `CommsDecoder[ToTask, ToSupervisor]()` | 1.44 ms | 1.38 ms | -4.2% | (same runs) | Not measurably slower per call. Supervisor dispatch for GetVariable, GetConnection, SetXCom and GetXCom moves between -0.7% and +0.9% while the controls moved +0.5% and +0.7%. Decoding through the larger ToSupervisor/ToTask unions costs at most about 30 ns more per message (+1–3%, at the noise floor), and the pinned-schema error path for Go/TS runtimes costs about 0.27 µs more (+2%). The only steady cost is one-time: building the two unions takes about 23 µs and 12 µs longer per process, which does not show in fresh-interpreter startup. That is acceptable, because every one of these calls also makes an HTTP request to the API server, which takes milliseconds. --- ##### 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]
