Add durable task instance launch records (RFC)

[wolvery]

Summary

This is a draft/RFC vertical slice that prevents expected stale executor launches from surfacing as opaque /run 404 failures. It introduces durable executor launch records and a typed 409 stale_executor_launch response, so that a launch attempt from an executor whose task instance row has already been superseded (reset, rescheduled, cleared, or re-adopted) can be distinguished from a genuinely unknown task.

Refs #69760.

Motivation

When a task instance transitions out of its original launch (scheduler failover, adoption, stuck-queued reschedule, orphan reset, or a manual clear), the executor that was originally assigned may still attempt to /run the task. Because the launch identity lived only on the mutable TaskInstance row, that late launch produced an opaque 404 not_found, which is indistinguishable from a truly unknown task and consumes a retry / blocks clean requeueing.

Approach

• Durable launch records. New TaskInstanceLaunch ORM persists an immutable executor launch token independently of the mutable/deleted TaskInstance row, with an active -> consumed | superseded lifecycle, guarded state transitions, lookup helpers, indexes, and a state check constraint.
• Atomic creation. A launch record is written during the SCHEDULED -> QUEUED transition in the scheduler.
• Supersession. Reschedule of stuck-queued tasks, failed adoption, orphan reset, and TaskInstance clear/next-try all supersede any active launch so an old token cannot be reused if the old executor comes back. Successful adoption preserves the token.
• Typed conflict. /run validates the launch token:
  • Known terminal token (consumed/superseded) -> 409 stale_executor_launch.
  • Unknown token -> remains 404 not_found (legacy tokens predating this change still return 404).
  • Successful /run guarded-consumes the launch; duplicate-running remains idempotent.
• Worker behavior. New TaskInstanceSupersededError in the Task SDK; a superseded worker logs once and exits successfully instead of failing.
• Executor-agnostic. Works across Kubernetes, Celery, and Edge executors. No authorization changes and no fallback by logical task key. Two enabling changes were required for KubernetesExecutor (see below): opting the executor into launch-token pre-assignment, and letting the token survive serialization into the worker pod.

Changes

• New model airflow/models/task_instance_launch.py + migration 3c5f8e9a1d2b (down-revision 7a98f1b7dbd3), registered with DB cleanup.
• Scheduler: launch creation on queueing; supersession on reschedule / failed adoption / orphan reset.
• TaskInstance clear/next-try: supersede active launch.
• Execution API /run: durable-token validation + typed 409; Cadwyn version adds optional external_executor_id.
• Task SDK: propagate token on start, map 409 stale_executor_launch to TaskInstanceSupersededError; supervisor exits 0 on supersession.
• KubernetesExecutor enablement:
  • KubernetesExecutor.pre_assigns_external_executor_id = True so the scheduler pre-assigns a durable launch token (UUID) at QUEUED time and writes the TaskInstanceLaunch row. Without this the launch table stayed empty under K8s and the whole mechanism was inert (Celery-only).
  • Removed exclude=True from external_executor_id on the executor→worker workload DTO (airflow/executors/workloads/task.py) so the token survives model_dump_json() into the pod's --json-string and the worker can echo it back on /run. Previously the field was silently dropped for every containerized executor; Celery only worked because it re-injects the token out-of-band as the Celery task_id.
• Tests: model, migration, scheduler, execution API, SDK client, and supervisor.

Rollout / compatibility

• Additive migration; no backfill required.
• Unknown legacy tokens (launched before this change) still return 404 — behavior only changes for tokens that have a durable record.

Testing

Automated tests

• Launch model + migration: test_task_instance_launch.py and test_0128_task_instance_launch_table.py (5 passed) — lifecycle, guarded transitions, and the additive migration (revision 3c5f8e9a1d2b, down-revision 7a98f1b7dbd3).
• Execution API + cleartasks: 228 passed, 4 skipped — /run durable-token validation, typed 409 stale_executor_launch, unknown-token 404, guarded consume, and clear/next-try supersession.
• Scheduler: supersession on stuck-queued reschedule, failed adoption, and orphan reset; test_failed_adoption_marks_launch_record_superseded covers the failed-adoption path (successful adoption preserves the token).
• Task SDK: test_task_instance_start_maps_409_stale_executor_launch (client maps 409 → TaskInstanceSupersededError) and test_supervise_handles_superseded_task (superseded worker logs once and exits 0).
• Execution API versioning: TestExternalExecutorIdFieldBackwardCompat asserts the new behavior is opt-in per Cadwyn version — a pre-v2026_06_30 client (no external_executor_id in its request schema) still gets 404 not_found for a stale launch, while a head-version client gets 409 stale_executor_launch.
• Migration round-trip: upgrade → downgrade (-r 7a98f1b7dbd3) → upgrade verified against Postgres; task_instance_launch is cleanly dropped and recreated (matches CI's migration_tests action).

Note: two pre-existing Task SDK supervisor tests (test_run_simple_dag, test_supervise_handles_deferred_task) fail on a clean checkout in this environment and are unrelated to this change (confirmed by git stash on a clean tree).

Validation on a live KubernetesExecutor cluster

The end-to-end path was validated on a local kind cluster (Airflow Helm chart, --executor KubernetesExecutor, image built from this branch, migration head 3c5f8e9a1d2b) driven by a synthetic high-fan-out DAG load (~131 DAGs):

• Before the two K8s enablement changes: task_instance_launch stayed at 0 rows while K8s tasks ran — the mechanism was inert under KubernetesExecutor (token never pre-assigned; and even if it were, it was dropped by exclude=True before reaching the worker).
• After: durable launch rows populate (100+), each carrying a proper 36-char UUID token. A worker pod's --json-string argument was confirmed to carry external_executor_id=<uuid>, and the same token was traced end-to-end: scheduler pre-assign at QUEUED → serialized into the pod → echoed back on /run → guarded-consumed (state=consumed, which only fires on a token match).
• Guarded state machine exercised deterministically against the deployed image + live Postgres: active → superseded succeeds; a subsequent consume on the now-terminal token is refused by the WHERE state == ACTIVE guard; get_by_token still returns the terminal row (this is what lets /run return the typed 409 stale_executor_launch instead of 404); repeated supersession is idempotent.

Status

Draft / RFC for review of the approach before hardening. Feedback welcome on the launch-record lifecycle and the 409 contract.

AI disclosure

This change was developed with AI assistance (Anthropic Claude). The design, implementation, migration, tests, and the live-cluster validation were authored with an AI coding agent and reviewed by the submitter, who takes responsibility for the contents.