From d35e8056b3d58cf8253722eebf6fb2e07a8f7647 Mon Sep 17 00:00:00 2001 From: vincbeck Date: Thu, 24 Sep 2026 14:34:11 -0400 Subject: [PATCH] Filter jobs API is_alive in SQL so pagination stays correct The /jobs endpoint evaluated liveness in Python on the page returned by paginated_select, so total_entries counted rows the filter would drop and a page could come back short whenever is_alive was combined with another filter. The same post-filter also kept only alive jobs for every value of the flag, so is_alive=false returned alive jobs instead of dead ones. Resolving liveness in SQL lets the count and the page agree and makes the flag mean what it says. Job.alive_filter() mirrors Job._is_alive and reads its thresholds from health_check_threshold, so the endpoint and the health check cannot drift apart. --- .../api_fastapi/common/parameters/__init__.py | 1 + .../api_fastapi/common/parameters/misc.py | 43 +++++++++++++++++++ .../openapi/v2-rest-api-generated.yaml | 16 +++---- .../api_fastapi/core_api/routes/public/job.py | 7 ++- .../ui/openapi-gen/queries/ensureQueryData.ts | 2 +- .../ui/openapi-gen/queries/prefetch.ts | 2 +- .../airflow/ui/openapi-gen/queries/queries.ts | 2 +- .../ui/openapi-gen/queries/suspense.ts | 2 +- .../ui/openapi-gen/requests/services.gen.ts | 6 +-- .../core_api/routes/public/test_job.py | 34 +++++++++++++++ 10 files changed, 96 insertions(+), 19 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py b/airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py index deae80a25acc9..8ec274d007889 100644 --- a/airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py +++ b/airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py @@ -105,6 +105,7 @@ QueryHITLDetailTaskIdPrefixPatternSearch as QueryHITLDetailTaskIdPrefixPatternSearch, QueryIncludeDownstream as QueryIncludeDownstream, QueryIncludeUpstream as QueryIncludeUpstream, + QueryJobIsAliveFilter as QueryJobIsAliveFilter, QueryParseImportErrorBundleNameFilter as QueryParseImportErrorBundleNameFilter, QueryParseImportErrorFilenameFilter as QueryParseImportErrorFilenameFilter, QueryParseImportErrorFilenamePatternSearch as QueryParseImportErrorFilenamePatternSearch, diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters/misc.py b/airflow-core/src/airflow/api_fastapi/common/parameters/misc.py index d718d2e132ee9..6a5b2a44bdccf 100644 --- a/airflow-core/src/airflow/api_fastapi/common/parameters/misc.py +++ b/airflow-core/src/airflow/api_fastapi/common/parameters/misc.py @@ -86,6 +86,49 @@ def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActi QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)] +class _IsAliveFilter(BaseParam[bool]): + """Filter jobs by liveness, resolved in SQL so it agrees with pagination counts.""" + + def to_orm(self, select: Select) -> Select: + if self.value is None and self.skip_none: + return select + + from datetime import timedelta + + from sqlalchemy import and_, case, or_ + + from airflow._shared.timezones import timezone + from airflow.jobs.job import Job, JobState, health_check_threshold + + now = timezone.utcnow() + # Only the concrete core job types are persisted here and their thresholds come from config + # rather than the per-row heartrate, so the cutoff is a bound timestamp and the predicate stays + # portable. Kept in sync with Job._is_alive: alive means RUNNING with a heartbeat newer than the + # job type's health-check threshold. The case wraps the result so it is always True/False (never + # NULL), keeping the negated is_alive=False query exact. + fresh_per_type = or_( + *( + and_( + Job.job_type == job_type, + Job.latest_heartbeat > now - timedelta(seconds=health_check_threshold(job_type, 0)), + ) + for job_type in ("SchedulerJob", "TriggererJob", "DagProcessorJob") + ) + ) + alive = case( + (and_(Job.state == JobState.RUNNING, Job.latest_heartbeat.is_not(None), fresh_per_type), True), + else_=False, + ) + return select.where(alive if self.value else ~alive) + + @classmethod + def depends(cls, is_alive: bool | None = Query(None)) -> _IsAliveFilter: + return cls().set_value(is_alive) + + +QueryJobIsAliveFilter = Annotated[_IsAliveFilter, Depends(_IsAliveFilter.depends)] + + # Variables QueryVariableKeyPatternSearch = Annotated[ _SearchParam, Depends(search_param_factory(Variable.key, "variable_key_pattern")) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index 22f1afe6903e5..027c4274fee50 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -5352,14 +5352,6 @@ paths: - OAuth2PasswordBearer: [] - HTTPBearer: [] parameters: - - name: is_alive - in: query - required: false - schema: - anyOf: - - type: boolean - - type: 'null' - title: Is Alive - name: start_date_gte in: query required: false @@ -5514,6 +5506,14 @@ paths: items: type: string title: Teams + - name: is_alive + in: query + required: false + schema: + anyOf: + - type: boolean + - type: 'null' + title: Is Alive responses: '200': description: Successful Response diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py index 3760b461b233e..35fc993cac057 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py @@ -28,6 +28,7 @@ ) from airflow.api_fastapi.common.parameters import ( FilterParam, + QueryJobIsAliveFilter, QueryJobTeamsFilter, QueryLimit, QueryOffset, @@ -103,7 +104,7 @@ def get_jobs( Depends(filter_param_factory(Job.executor_class, str | None, filter_name="executor_class")), ], teams: QueryJobTeamsFilter, - is_alive: bool | None = None, + is_alive: QueryJobIsAliveFilter, ) -> JobCollectionResponse: """Get all jobs.""" base_select = select(Job).order_by(Job.latest_heartbeat.desc()).options(joinedload(Job.dag_model)) @@ -119,6 +120,7 @@ def get_jobs( hostname, executor_class, teams, + is_alive, ], order_by=order_by, limit=limit, @@ -128,9 +130,6 @@ def get_jobs( ) jobs = session.scalars(jobs_select).all() - if is_alive is not None: - jobs = [job for job in jobs if job.is_alive()] - return JobCollectionResponse( jobs=jobs, total_entries=total_entries, diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts index 58c12d2a67055..c19b3eea84850 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts @@ -1544,7 +1544,6 @@ export const ensureUseImportErrorServiceGetImportErrorsData = (queryClient: Quer * Get Jobs * Get all jobs. * @param data The data for the request. -* @param data.isAlive * @param data.startDateGte * @param data.startDateGt * @param data.startDateLte @@ -1562,6 +1561,7 @@ export const ensureUseImportErrorServiceGetImportErrorsData = (queryClient: Quer * @param data.hostname * @param data.executorClass * @param data.teams +* @param data.isAlive * @returns JobCollectionResponse Successful Response * @throws ApiError */ diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts index b04f0e549d1c7..fdcd48fbd35ed 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts @@ -1544,7 +1544,6 @@ export const prefetchUseImportErrorServiceGetImportErrors = (queryClient: QueryC * Get Jobs * Get all jobs. * @param data The data for the request. -* @param data.isAlive * @param data.startDateGte * @param data.startDateGt * @param data.startDateLte @@ -1562,6 +1561,7 @@ export const prefetchUseImportErrorServiceGetImportErrors = (queryClient: QueryC * @param data.hostname * @param data.executorClass * @param data.teams +* @param data.isAlive * @returns JobCollectionResponse Successful Response * @throws ApiError */ diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts index df1286bea2557..b71f083d3db18 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts @@ -1544,7 +1544,6 @@ export const useImportErrorServiceGetImportErrors = not alive. + Job( + state=JobState.RUNNING, + job_type="SchedulerJob", + latest_heartbeat=now - timedelta(days=1), + ), + # Not RUNNING -> not alive. + Job(state=JobState.FAILED, job_type="SchedulerJob", latest_heartbeat=now), + ] + session.add_all(jobs) + session.commit() + + alive = test_client.get("/jobs", params={"is_alive": True}).json() + assert alive["total_entries"] == 2 + assert {job["id"] for job in alive["jobs"]} == {jobs[0].id, jobs[1].id} + + not_alive = test_client.get("/jobs", params={"is_alive": False}).json() + assert not_alive["total_entries"] == 2 + assert {job["id"] for job in not_alive["jobs"]} == {jobs[2].id, jobs[3].id} + + # total_entries counts the filtered set in SQL, so it stays correct once a page limit applies. + paged = test_client.get("/jobs", params={"is_alive": True, "limit": 1}).json() + assert paged["total_entries"] == 2 + assert len(paged["jobs"]) == 1 + def test_should_raises_401_unauthenticated(self, unauthenticated_test_client): response = unauthenticated_test_client.get("/jobs") assert response.status_code == 401