ferruzzi opened a new issue, #76795: URL: https://github.com/apache/airflow/issues/76795
### Under which category would you file this issue? Airflow Core ### Apache Airflow version main, with #75253 applied (approved, pending rebase) ### What happened and how to reproduce it? Follow-up to #75253 from review threads [on the scheduler](https://github.com/apache/airflow/pull/75253#discussion_r4449076528) and [on the supervisor](https://github.com/apache/airflow/pull/75253#discussion_r4449070205) (thanks @ramitkataria). Part of #62887. Failure 2 exists once #75253 merges; failures 1 and 3 exist on `main` today. A running executor callback has no liveness signal, so nothing can tell a callback that is still running apart from one whose supervisor is gone. **1. A duplicate delivery records FAILED for a callback that succeeded.** To reproduce: 1. Use CeleryExecutor with a Redis or SQS broker, the default `[celery] task_acks_late = True`, and a visibility timeout shorter than the callback, for example `[celery_broker_transport_options] visibility_timeout = 30`. 2. Create a Dag with a `DeadlineAlert` whose callback is a `SyncCallback` that sleeps 60 seconds and then returns. 3. Let the deadline fire. The broker redelivers the message to a second worker while the first is still running. 4. The second supervisor's `PATCH /callbacks/{id}/run` gets 409, because the original already exchanged the token. It fails at once, Celery records FAILURE for the task id both deliveries share, and the scheduler records the callback FAILED. 5. The original finishes, but the executor stopped polling that task id when it reported the failure, so the original's success is never seen. With #75253, the original's `finish(SUCCESS)` is also rejected with 409, because the row is already FAILED. The row ends FAILED although the callback ran to completion. Task instances avoid this because the worker recognizes the redelivered message and discards it (see "Anything else"); callbacks do not. **2. A long callback cannot report its own result.** To reproduce: 1. Set `[execution_api] jwt_expiration_time = 60`. 2. Create a `SyncCallback` that sleeps 120 seconds and makes no Execution API calls (no Connection or Variable reads). 3. Let the deadline fire. The supervisor logs `Failed to report final callback state` with a 403 and carries on. The execution token from `/run` expired: `JWTReissueMiddleware` renews a token only on a request made near expiry, and the supervisor makes none while the callback runs. The row then depends on the executor event alone, so a failed callback loses the output the supervisor captured, and if that event is missing too (see 3) the row stays RUNNING. **3. After a scheduler restart, a callback whose supervisor dies stays active forever.** To reproduce: 1. Use a `SyncCallback` that sleeps 120 seconds. 2. After the row reaches RUNNING, restart the scheduler, then `kill -9` the callback supervisor. On `main` without #75253, restarting the scheduler is enough on its own. The row stays RUNNING indefinitely. A callback still QUEUED at the restart stays QUEUED the same way if its /run fails. Every executor shipped with Airflow that runs callbacks (Local, Celery, ECS, Batch, Lambda and Edge) reports a terminal executor event, but only from the scheduler process that queued the callback. Task instances survive a restart through `adopt_or_reset_orphaned_tasks`; callbacks have no equivalent, and the scheduler only ever selects PENDING callbacks. So after a restart, or when an event is lost as in #69975, nothing resolves a QUEUED or RUNNING callback except the supervisor's own `/state` report, and failures 1 and 2 show that report can be missing or rejected. ### What you think should happen instead? - A callback that succeeded is recorded SUCCESS, even when a duplicate delivery fails first. - A callback's result is recorded however long the callback runs. - A callback whose supervisor is gone is failed after a bounded time, whether QUEUED or RUNNING, on every executor and across scheduler restarts. Task instances already get this through the heartbeat timeout and `[scheduler] task_queued_timeout`. ### Operating System _No response_ ### Deployment None ### Apache Airflow Provider(s) _No response_ ### Versions of Apache Airflow Providers _No response_ ### Official Helm Chart version Not Applicable ### Kubernetes Version _No response_ ### Helm Chart configuration _No response_ ### Docker Image customizations _No response_ ### Anything else? **Proposed direction**, mirroring task-instance liveness: - a `last_heartbeat_at` column on `callback`, with a migration; - a heartbeat endpoint on the Execution API, with a Cadwyn version change; - a heartbeat loop in the callback supervisor; - config mirroring `task_instance_heartbeat_sec`, `task_instance_heartbeat_timeout` and `task_instance_heartbeat_timeout_detection_interval`; - a scheduler reaper mirroring `_find_and_purge_task_instances_without_heartbeats` that fails RUNNING callbacks whose heartbeat is stale; - then let an executor FAILED event move a RUNNING row only once its heartbeat is stale, which closes 1. Two details reduce the scope: - The heartbeat also fixes 2. `JWTReissueMiddleware` renews any execution token that arrives in the last 20% of its lifetime, so a supervisor that heartbeats keeps its token fresh with no separate refresh mechanism. - Heartbeats start after `/run`, so they cannot catch a row stuck in QUEUED. That needs an analogue of `[scheduler] task_queued_timeout`. **A possible first step for 1 on its own.** The task path already handles duplicate delivery: `TaskInstanceOperations.start` turns a `/run` 409 with `previous_state` running into `TaskAlreadyRunningError`, and the Celery worker raises `Ignore()` on it, so the duplicate never records FAILURE. `CallbackOperations.run` has no such mapping, although the callback `/run` route returns the same `invalid_state` / `previous_state` detail. Mapping it, to `TaskAlreadyRunningError` or a subclass the existing handler still catches, would stop Celery duplicates from failing successful callbacks before the heartbeat work lands. It does not help other executors, and does nothing for 2 or 3. **Related:** - #75253, which added the second writer and first-writer-wins - #69975, lost executor callback events on commit failure: the same missing recovery path for QUEUED and RUNNING callbacks - #74784, Edge workers reporting callback terminal state - #62887, executor callback umbrella ### Are you willing to submit PR? - [ ] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md) -- 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]
