Return 404 when a DagRun disappears during task API writes by 1fanwang · Pull Request #72907 · apache/airflow

1fanwang · 2026-09-10T22:59:17Z

A task-state write can race with DagRun deletion. The backend then finds no DagRun, and the affected API request returned HTTP 500 even though the target no longer existed. The public and execution APIs now return their versioned 404 responses for that race. Unrelated backend failures still return HTTP 500.

The backend now marks a missing run explicitly. Each API turns that signal into its existing 404 response format.

related: #72890

Testing Done

The reproducer sends real HTTP requests through each FastAPI surface, writes rows to SQLite, deletes the DagRun immediately before the backend lookup, confirms the foreign-key cascade, and repeats the write.

breeze run python dev/repro_airflow_72907.py

Before:

HTTP PUT /execution/store/ti/01a099d6-2af0-75f6-aa65-178273e64c5c/job_id (deleted run) -> 404 {"detail":{"reason":"not_found","message":"DagRun with dag_id=dag and run_id=test not found"}}
HTTP PUT /api/v2/dags/dag/dagRuns/test/taskInstances/op1/state-store/job_id (deleted run) -> 404 {"detail":"No DagRun found for dag_id='dag' run_id='test'"}
HTTP PATCH /api/v2/dags/dag/dagRuns/test/taskInstances/op1/state-store/job_id (deleted run) -> 500 Internal Server Error
E   assert 500 == 404
exit_code=1

After:

HTTP PUT /execution/store/ti/01a10fe1-fee5-7713-b7d7-5ead9befe526/job_id (deleted run) -> 404 {"detail":{"reason":"not_found","message":"DagRun with dag_id=dag and run_id=test not found"}}
HTTP PUT /api/v2/dags/dag/dagRuns/test/taskInstances/op1/state-store/job_id (deleted run) -> 404 {"detail":"No DagRun found for dag_id='dag' run_id='test'"}
HTTP PATCH /api/v2/dags/dag/dagRuns/test/taskInstances/op1/state-store/job_id (deleted run) -> 404 {"detail":"No DagRun found for dag_id='dag' run_id='test'"}
exit_code=0

Reproducer source: dev/repro_airflow_72907.py and test_state_store_http_probe_72907.py

# File: dev/repro_airflow_72907.py
from pathlib import Path

import pytest

if __name__ == "__main__":
    test_file = (
        Path(__file__).parents[1]
        / "airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_state_store_http_probe_72907.py"
    )
    raise SystemExit(pytest.main([str(test_file), "-q", "-s"]))

# File: airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_state_store_http_probe_72907.py
from __future__ import annotations

from typing import TYPE_CHECKING

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import Select, event, func, select

from airflow import settings
from airflow.api_fastapi.core_api.routes.public import task_state_store as core_routes
from airflow.api_fastapi.execution_api.routes import task_state_store as execution_routes
from airflow.models.dagrun import DagRun
from airflow.models.task_state_store import TaskStateStoreModel
from airflow.models.taskinstance import TaskInstance

if TYPE_CHECKING:
    from sqlalchemy.engine import Connection
    from sqlalchemy.orm import Session
    from sqlalchemy.sql import Executable

    from tests_common.pytest_plugin import CreateTaskInstance

pytestmark = pytest.mark.db_test


def row_counts(*, dag_id: str, run_id: str) -> dict[str, int]:
    with settings.engine.connect() as connection:
        return {
            model.__tablename__: connection.scalar(
                select(func.count())
                .select_from(model)
                .where(model.dag_id == dag_id, model.run_id == run_id)
            )
            for model in (DagRun, TaskInstance, TaskStateStoreModel)
        }


@pytest.mark.parametrize(
    ("surface", "method", "success_code"),
    [("execution", "PUT", 204), ("core", "PUT", 204), ("core", "PATCH", 200)],
)
def test_http_write_racing_run_deletion(
    client: TestClient,
    test_client: TestClient,
    session: Session,
    create_task_instance: CreateTaskInstance,
    surface: str,
    method: str,
    success_code: int,
) -> None:
    ti = create_task_instance()
    dag_id, run_id, ti_id, task_id = ti.dag_id, ti.run_id, ti.id, ti.task_id
    session.commit()
    assert settings.engine.dialect.name == "sqlite"
    print(f"database={settings.engine.url}")
    print(f"core_route_source={core_routes.__file__}")
    print(f"execution_route_source={execution_routes.__file__}")
    with settings.engine.connect() as connection:
        foreign_keys = connection.exec_driver_sql("PRAGMA foreign_keys").scalar_one()
    print(f"sqlite_foreign_keys={foreign_keys}")
    assert foreign_keys == 1

    run_url = f"/api/v2/dags/{dag_id}/dagRuns/{run_id}"
    url = (
        f"/execution/store/ti/{ti_id}/job_id"
        if surface == "execution"
        else f"{run_url}/taskInstances/{task_id}/state-store/job_id"
    )
    original_client = client if surface == "execution" else test_client
    http = TestClient(
        app=original_client.app,
        headers=original_client.headers,
        base_url=str(original_client.base_url),
        raise_server_exceptions=False,
    )
    deleted = False

    def delete_run_before_lookup(
        conn: Connection,
        statement: Executable,
        multiparams: object,
        params: object,
        execution_options: object,
    ) -> None:
        nonlocal deleted
        if (
            deleted
            or not isinstance(statement, Select)
            or statement.column_descriptions[0]["expr"] is not DagRun.id
        ):
            return
        deleted = True
        print(f"backend_lookup={statement}")
        response = test_client.delete(url=run_url)
        print(f"HTTP DELETE {run_url} -> {response.status_code} {response.text}")
        assert response.status_code == 204
        counts = row_counts(dag_id=dag_id, run_id=run_id)
        print(f"rows_after_delete={counts}")
        assert counts == {"dag_run": 0, "task_instance": 0, "task_state_store": 0}

    try:
        response = http.put(url=url, json={"value": "old"})
        print(f"HTTP PUT {url} -> {response.status_code} {response.text}")
        assert response.status_code == 204
        response = http.request(method=method, url=url, json={"value": "success"})
        print(f"HTTP {method} {url} (no deletion) -> {response.status_code} {response.text}")
        assert response.status_code == success_code
        response = http.get(url=url)
        print(f"HTTP GET {url} -> {response.status_code} {response.text}")
        assert response.status_code == 200
        assert response.json()["value"] == "success"
        counts = row_counts(dag_id=dag_id, run_id=run_id)
        print(f"rows_before_delete={counts}")
        assert counts == {"dag_run": 1, "task_instance": 1, "task_state_store": 1}
        event.listen(settings.engine, "before_execute", delete_run_before_lookup)
        try:
            response = http.request(method=method, url=url, json={"value": "after-delete"})
        finally:
            event.remove(settings.engine, "before_execute", delete_run_before_lookup)
        print(f"HTTP {method} {url} (deleted run) -> {response.status_code} {response.text}")
        assert deleted
        assert response.status_code == 404
    finally:
        http.close()

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: GitHub Copilot CLI (GPT-6 Astra) following the guidelines

Read the 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) is needed.
When adding dependency, check compliance with the ASF 3rd Party License Policy.
For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

jason810496

Thanks for the improvement, LGTM overall.

jason810496

LGTM after the update, thanks.

Wait for the change of #72926 (review) get in this PR.

Signed-off-by: 1fanwang <[email protected]>

The conflict merge kept the dedicated DagRunNotFound exception but retained an upstream test that raised ValueError, so current-head CI failed before exercising the route contract. Signed-off-by: Stefan Wang <[email protected]>

Signed-off-by: 1fanwang <[email protected]>

1fanwang · 2026-10-07T19:05:55Z

Thanks again for the review @jason810496, is this good to go? Thnx

henry3260

Thanks for the PR, looks good!

henry3260 · 2026-10-24T19:56:23Z

Need to resolve conflits cc @jason810496

Main reworked the public task state store routes around task coordinates and added more 2026-10-30 Execution API version changes. Keep both sides: the new DynamicRegion import next to DagRunNotFound, and ReturnNotFoundForMissingDagRun alongside the other 2026-10-30 changes. Signed-off-by: 1fanwang <[email protected]>

pierrejeambrun

Thanks for this, and for addressing the earlier feedback! I tested it against a running API server by deleting the Dag run in the middle of a write: the Execution API now returns 404 for 2026-10-30 clients and keeps 500 for older ones. One note on the description: on current main the public API PUT/PATCH already return 404 for this race (they caught ValueError), so the user-visible fix is in the Execution API, and the public API change narrows which errors map to 404. Could you update the "Before" section? A couple of small things inline.

… API On main the Execution API route never caught ValueError, so the test there passed without this change. The public API used to turn any ValueError into a 404.

pierrejeambrun

I just pushed a small adjustment commit to address those.

github-actions · 2026-11-06T12:25:49Z

Backport successfully created: v3-4-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-4-test

apache#72907) * Return 404 instead of 500 when DagRun is missing in Execution API In task_instances.py validate_inlets_and_outlets and task_state_store.py set_task_state_store, missing DagRun instances triggered unhandled AttributeError and ValueError exceptions that surfaced as HTTP 500 Internal Server Errors. Catch the missing DagRun condition and return a clean 404 Not Found response matching the pattern established in PR 72890. Signed-off-by: 1fanwang <[email protected]> * Distinguish missing runs from other state backend errors Unrelated backend ValueErrors must not be reported as missing runs. Exercise writes racing actual run deletion instead of manufacturing impossible ORM results. Signed-off-by: 1fanwang <[email protected]> * Return 404 on task-state writes after a DagRun is deleted mid-request Signed-off-by: Stefan Wang <[email protected]> * Fix task state store 404 regression coverage The conflict merge kept the dedicated DagRunNotFound exception but retained an upstream test that raised ValueError, so current-head CI failed before exercising the route contract. Signed-off-by: Stefan Wang <[email protected]> * Test the narrowed ValueError handling where it changed, in the public API On main the Execution API route never caught ValueError, so the test there passed without this change. The public API used to turn any ValueError into a 404. --------- (cherry picked from commit f90d260) Co-authored-by: Stefan Wang <[email protected]> Signed-off-by: 1fanwang <[email protected]> Signed-off-by: Stefan Wang <[email protected]> Co-authored-by: pierrejeambrun <[email protected]>

#72907) (#76599) * Return 404 instead of 500 when DagRun is missing in Execution API In task_instances.py validate_inlets_and_outlets and task_state_store.py set_task_state_store, missing DagRun instances triggered unhandled AttributeError and ValueError exceptions that surfaced as HTTP 500 Internal Server Errors. Catch the missing DagRun condition and return a clean 404 Not Found response matching the pattern established in PR 72890. * Distinguish missing runs from other state backend errors Unrelated backend ValueErrors must not be reported as missing runs. Exercise writes racing actual run deletion instead of manufacturing impossible ORM results. * Return 404 on task-state writes after a DagRun is deleted mid-request * Fix task state store 404 regression coverage The conflict merge kept the dedicated DagRunNotFound exception but retained an upstream test that raised ValueError, so current-head CI failed before exercising the route contract. * Test the narrowed ValueError handling where it changed, in the public API On main the Execution API route never caught ValueError, so the test there passed without this change. The public API used to turn any ValueError into a 404. --------- (cherry picked from commit f90d260) Signed-off-by: 1fanwang <[email protected]> Signed-off-by: Stefan Wang <[email protected]> Co-authored-by: Stefan Wang <[email protected]> Co-authored-by: pierrejeambrun <[email protected]>

1fanwang requested review from amoghrajesh, ashb and kaxil as code owners September 10, 2026 22:59

boring-cyborg Bot added area:API Airflow's REST/HTTP API area:task-sdk labels Sep 10, 2026

jason810496 reviewed Sep 11, 2026

View reviewed changes

Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_state_store.py Resolved

Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py Resolved

eladkal added this to the Airflow 3.3.2 milestone Sep 11, 2026

eladkal added type:bug-fix Changelog: Bug Fixes backport-to-v3-3-test labels Sep 11, 2026

vatsrahul1001 modified the milestones: Airflow 3.3.2, Airflow 3.4.0 Sep 11, 2026

1fanwang force-pushed the 1fannnw/execution-api-missing-dagrun-404 branch from 3a0ede1 to c3aba24 Compare September 11, 2026 17:47

1fanwang requested review from bugraoz93, choo121600, ephraimbuddy, henry3260, pierrejeambrun, rawwar and shubhamraj-git as code owners September 11, 2026 17:47

jason810496 previously approved these changes Sep 12, 2026

View reviewed changes

potiuk mentioned this pull request Sep 26, 2026

Support rendered_task_instance_fields in airflow db clean #73745

Closed

1 task done

vatsrahul1001 removed the backport-to-v3-3-test label Oct 5, 2026

1fanwang added 2 commits October 5, 2026 12:21

Merge main and preserve missing DagRun 404 handling

b55d095

Signed-off-by: 1fanwang <[email protected]>

Fix task state store 404 regression coverage

231f48f

The conflict merge kept the dedicated DagRunNotFound exception but retained an upstream test that raised ValueError, so current-head CI failed before exercising the route contract. Signed-off-by: Stefan Wang <[email protected]>

1fanwang changed the title ~~Return 404 instead of 500 when DagRun is missing in Execution API~~ Return 404 when a DagRun disappears during task API writes Oct 6, 2026

1fanwang added 2 commits October 7, 2026 11:50

Merge main and preserve missing DagRun 404 handling

34b04b9

Signed-off-by: 1fanwang <[email protected]>

Merge latest main into missing DagRun 404 fix

cda1309

Signed-off-by: 1fanwang <[email protected]>

Merge branch 'main' into 1fannnw/execution-api-missing-dagrun-404

87d0e22

vatsrahul1001 added the backport-to-v3-4-test Backport to v3-4-test label Oct 18, 2026

vatsrahul1001 modified the milestones: Airflow 3.4.0, Airflow 3.4.1 Oct 23, 2026

henry3260 approved these changes Oct 24, 2026

View reviewed changes

github-actions Bot added need workflow approval and removed need workflow approval labels Nov 6, 2026

pierrejeambrun reviewed Nov 6, 2026

View reviewed changes

Comment thread airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_state_store.py Resolved

Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_state_store.py Resolved

Test the narrowed ValueError handling where it changed, in the public…

691ebdb

… API On main the Execution API route never caught ValueError, so the test there passed without this change. The public API used to turn any ValueError into a 404.

github-actions Bot added the need workflow approval label Nov 6, 2026

pierrejeambrun approved these changes Nov 6, 2026

View reviewed changes

pierrejeambrun enabled auto-merge (squash) November 6, 2026 11:25

github-actions Bot removed the need workflow approval label Nov 6, 2026

pierrejeambrun merged commit f90d260 into apache:main Nov 6, 2026
78 checks passed

Dev-iL mentioned this pull request Nov 14, 2026

Make task state store Execution API routes async #75381

Open

1 task done

Navigation Menu

Repository navigation

Return 404 when a DagRun disappears during task API writes - #72907

Return 404 when a DagRun disappears during task API writes#72907
pierrejeambrun merged 10 commits into
apache:mainfrom
1fanwang:1fannnw/execution-api-missing-dagrun-404

1fanwang commented Sep 10, 2026 •

edited

Loading

Uh oh!

jason810496 left a comment

Uh oh!

Uh oh!

Uh oh!

jason810496 left a comment

Uh oh!

1fanwang commented Oct 7, 2026

Uh oh!

henry3260 left a comment

Uh oh!

henry3260 commented Oct 24, 2026

Uh oh!

pierrejeambrun left a comment

Uh oh!

Uh oh!

Uh oh!

pierrejeambrun left a comment

Uh oh!

Uh oh!

github-actions Bot commented Nov 6, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

Conversation

1fanwang commented Sep 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Testing Done

Was generative AI tooling used to co-author this PR?

Uh oh!

jason810496 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

jason810496 left a comment

Choose a reason for hiding this comment

Uh oh!

1fanwang commented Oct 7, 2026

Uh oh!

henry3260 left a comment

Choose a reason for hiding this comment

Uh oh!

henry3260 commented Oct 24, 2026

Uh oh!

pierrejeambrun left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

pierrejeambrun left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

github-actions Bot commented Nov 6, 2026

Backport successfully created: v3-4-test

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

1fanwang commented Sep 10, 2026 •

edited

Loading