Filter jobs API is_alive in SQL so pagination stays correct

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,

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"))

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

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,

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
 */

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
 */

airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts

@@ -1544,7 +1544,6 @@ export const useImportErrorServiceGetImportErrors = <TData = Common.ImportErrorS
 * 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 useImportErrorServiceGetImportErrors = <TData = Common.ImportErrorS
 * @param data.hostname
 * @param data.executorClass
 * @param data.teams
+* @param data.isAlive
 * @returns JobCollectionResponse Successful Response
 * @throws ApiError
 */

airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts

@@ -1544,7 +1544,6 @@ export const useImportErrorServiceGetImportErrorsSuspense = <TData = Common.Impo
 * 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 useImportErrorServiceGetImportErrorsSuspense = <TData = Common.Impo
 * @param data.hostname
 * @param data.executorClass
 * @param data.teams
+* @param data.isAlive
 * @returns JobCollectionResponse Successful Response
 * @throws ApiError
 */

airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts

@@ -3624,7 +3624,6 @@ export class JobService {
      * Get Jobs
      * Get all jobs.
      * @param data The data for the request.
-     * @param data.isAlive
      * @param data.startDateGte
      * @param data.startDateGt
      * @param data.startDateLte
@@ -3642,6 +3641,7 @@ export class JobService {
      * @param data.hostname
      * @param data.executorClass
      * @param data.teams
+     * @param data.isAlive
      * @returns JobCollectionResponse Successful Response
      * @throws ApiError
      */
@@ -3650,7 +3650,6 @@ export class JobService {
             method: 'GET',
             url: '/api/v2/jobs',
             query: {
-                is_alive: data.isAlive,
                 start_date_gte: data.startDateGte,
                 start_date_gt: data.startDateGt,
                 start_date_lte: data.startDateLte,
@@ -3667,7 +3666,8 @@ export class JobService {
                 job_type: data.jobType,
                 hostname: data.hostname,
                 executor_class: data.executorClass,
-                teams: data.teams
+                teams: data.teams,
+                is_alive: data.isAlive
             },
             errors: {
                 400: 'Bad Request',

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py

@@ -16,9 +16,11 @@
 # under the License.
 from __future__ import annotations
 
+from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING
 
 import pytest
+import time_machine
 from sqlalchemy import delete
 from sqlalchemy.orm import Session
 
@@ -239,6 +241,38 @@ def test_get_jobs_filters_by_teams(self, test_client, session: Session, testing_
         assert response_json["total_entries"] == 1
         assert response_json["jobs"][0]["team_names"] == [testing_team.name]
 
+    @time_machine.travel(datetime(2024, 1, 1, tzinfo=timezone.utc), tick=False)
+    def test_get_jobs_is_alive_filter_is_consistent_with_pagination(self, test_client, session: Session):
+        clear_db_jobs()
+        now = datetime(2024, 1, 1, tzinfo=timezone.utc)
+        jobs = [
+            Job(state=JobState.RUNNING, job_type="SchedulerJob", latest_heartbeat=now),
+            Job(state=JobState.RUNNING, job_type="SchedulerJob", latest_heartbeat=now),
+            # RUNNING but the heartbeat is older than the health-check threshold -> 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