Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions packages/hflow-server/src/hflow_server/_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
from hflow_server._runtime import (
ResolvedRuntime,
RuntimeResolver,
RuntimeSource,
airflow_failure_refusal,
optional_string,
resolved_runtime_or_refuse,
Expand Down Expand Up @@ -449,11 +450,13 @@ def create_graph_router(pipeline_state: PipelineState, resolver: RuntimeResolver
router = APIRouter(prefix="/api/v1")

def stage_task_instances(
client: AirflowClient, dag_id: str, dag_run_id: str
client: AirflowClient, dag_id: str, dag_run_id: str, *, source: RuntimeSource
) -> list[dict[str, Any]]:
try:
return client.task_instances(dag_id, dag_run_id)
except AirflowClientError:
except AirflowClientError as error:
if error.status != 404:
raise airflow_failure_refusal(error, source=source) from error
# A stage sub-DAG that vanished (or a run Airflow expired) leaves
# that lane without task detail; the master's own state -- the
# page's point -- is already in hand, so this is a thinner
Expand Down Expand Up @@ -523,7 +526,9 @@ def read_run_graph(dag_run_id: str) -> RunGraphResponse:
stage_runs = runtime.client.dag_runs(
stage_dag_id, limit=_STAGE_RUN_SEARCH_LIMIT, order_by="-id"
)
except AirflowClientError:
except AirflowClientError as error:
if error.status != 404:
raise airflow_failure_refusal(error, source=runtime.source) from error
# An unregistered stage sub-DAG (a partial profile, or a
# bundle mid-render) is a stage that never ran here.
stage_runs = []
Expand All @@ -534,7 +539,9 @@ def read_run_graph(dag_run_id: str) -> RunGraphResponse:
stage_run_id = optional_string(matched.run.get("dag_run_id"))
stage_tasks = (
_sorted_task_instances(
stage_task_instances(runtime.client, stage_dag_id, stage_run_id),
stage_task_instances(
runtime.client, stage_dag_id, stage_run_id, source=runtime.source
),
stage_topology.dag,
)
if stage_run_id is not None
Expand Down
4 changes: 3 additions & 1 deletion packages/hflow-server/src/hflow_server/_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -396,7 +396,9 @@ def list_runtime_runs(
recent_runs = runtime.client.dag_runs(
stage_dag_id, limit=_RECENT_STAGE_RUN_LIMIT, order_by="-id"
)
except AirflowClientError:
except AirflowClientError as error:
if error.status != 404:
raise airflow_failure_refusal(error, source=runtime.source) from error
# A stage sub-DAG that has not registered (or errored) is
# an empty strip, not a failed page.
recent_runs = []
Expand Down
66 changes: 66 additions & 0 deletions packages/hflow-server/tests/test_server_run_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -433,6 +433,72 @@ def stage_runs_404(
assert all(stage["dag_run_id"] is None for stage in payload["stages"])


def test_run_graph_maps_stage_dag_listing_failures_to_502(
bundle_api: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
_stubbed_airflow(
monkeypatch,
master_run={
"dag_run_id": MASTER_RUN_ID,
"state": "success",
"start_date": MASTER_STARTED_AT,
},
stage_runs={},
task_instances={(MASTER_DAG_ID, MASTER_RUN_ID): []},
)

def failing_stage_runs(
self: AirflowClient, dag_id: str, *, limit: int = 100, order_by: str | None = None
) -> list[dict[str, Any]]:
if dag_id != MASTER_DAG_ID:
raise AirflowClientError(
"GET /dagRuns failed with HTTP 503: scheduler down", status=503
)
return []

monkeypatch.setattr(AirflowClient, "dag_runs", failing_stage_runs)
response = bundle_api.get(f"/api/v1/runtime/runs/{MASTER_RUN_ID}/graph")
assert response.status_code == 502
assert "scheduler down" in response.json()["detail"]


def test_run_graph_maps_stage_task_listing_failures_to_502(
bundle_api: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
_stubbed_airflow(
monkeypatch,
master_run={
"dag_run_id": MASTER_RUN_ID,
"state": "success",
"start_date": MASTER_STARTED_AT,
},
stage_runs={
"demo_pipeline_sync": [
{
"dag_run_id": "sync__run",
"state": "success",
"start_date": "2026-08-21T10:00:05+00:00",
}
]
},
task_instances={(MASTER_DAG_ID, MASTER_RUN_ID): []},
)

def failing_task_instances(
self: AirflowClient, dag_id: str, dag_run_id: str
) -> list[dict[str, Any]]:
if dag_id != MASTER_DAG_ID:
raise AirflowClientError(
"GET /taskInstances failed with HTTP 401: unauthorized", status=401
)
return []

monkeypatch.setattr(AirflowClient, "task_instances", failing_task_instances)
response = bundle_api.get(f"/api/v1/runtime/runs/{MASTER_RUN_ID}/graph")
assert response.status_code == 502
assert "unauthorized" in response.json()["detail"]


def test_run_graph_unknown_run_is_a_404(
bundle_api: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
Expand Down
16 changes: 16 additions & 0 deletions packages/hflow-server/tests/test_server_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -275,6 +275,22 @@ def fake_dag_runs(
assert all(stage["recent"] == [] for stage in payload["stages"])


def test_runs_maps_stage_dag_listing_failures_to_502(
bundle_api: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
def failing_dag_runs(
self: AirflowClient, dag_id: str, *, limit: int = 100, order_by: str | None = None
) -> list[dict[str, Any]]:
if dag_id == "demo_pipeline_ingest":
return []
raise AirflowClientError("GET /dagRuns failed with HTTP 429: rate limited", status=429)

monkeypatch.setattr(AirflowClient, "dag_runs", failing_dag_runs)
response = bundle_api.get("/api/v1/runtime/runs")
assert response.status_code == 502
assert "rate limited" in response.json()["detail"]


def test_runs_without_a_runtime_is_a_clear_409(
runtime_free_cwd: Path, tmp_path: Path, unbuilt_assets_dir: Path
) -> None:
Expand Down
Loading