refactor: replace SQL JOIN gating with BatchQuerier + per-handler hc fetch

``DeploymentRepository.get_routes_by_statuses`` (which carried the
``DeploymentRevisionRow.model_definition`` and ``SessionRow.status``
JOINs and the in-memory ``RouteHealthCheckFilter`` post-filter) is
split into two narrower entry points so handlers carry the cost of the
data they actually need:

- ``search_route_datas(querier: BatchQuerier)`` returns bare
  ``RouteData`` from ``RoutingRow`` alone; the coordinator builds a
  ``BatchQuerier`` from each handler's ``RouteTargetStatuses`` (now
  ``lifecycle``/``health``/optional list ``traffic``) at the call site
  with ``RouteConditions``.
- ``fetch_health_check_configs(revision_ids)`` returns a per-revision
  ``ModelHealthCheck | None`` map, called from
  ``HealthCheckRouteHandler``, ``AppProxySyncRouteHandler``,
  ``RouteHealthObserver``, and ``RouteExecutor.check_running_routes``
  to gate their per-route behaviour.

``RouteHealthCheckFilter``, ``RouteHandler.health_check_filter()`` and
every override are removed; ``RouteSessionData`` is dropped and
``RouteData.session_id`` becomes a direct field again.

In ``check_running_routes``, the ``(RouteData, ModelHealthCheck)`` pairs
are bundled into a private ``_RouteWithHealthCheck`` frozen dataclass
that flows down through ``_ensure_health_records`` and
``_initialize_health_records``; ``_populate_replica_info`` now also
mutates ``replica_host``/``replica_port`` on the in-memory routes so
the newly populated rows fall through to the single Phase-4 record
init. ``sync_appproxy`` self-fetches ``session_statuses`` instead of
relying on the dropped SQL join.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: fregataa

src/ai/backend/manager/data/deployment/types.py

@@ -253,23 +253,12 @@ class RouteTargetStatuses:
     """Target statuses for route handler filtering (lifecycle x health x traffic).
 
     ``traffic=None`` skips the filter; otherwise the row's
-    ``traffic_status`` column must match.
+    ``traffic_status`` column must be in the given list.
     """
 
     lifecycle: list[RouteStatus]
     health: list[RouteHealthStatus]
-    traffic: RouteTrafficStatus | None = None
-
-
-@dataclass(frozen=True)
-class RouteHealthCheckFilter:
-    """In-memory gating on a revision's ``health_check_config``.
-
-    ``health_check_required=True`` keeps only routes whose revision has
-    a resolved health-check block; ``False`` (default) skips the filter.
-    """
-
-    health_check_required: bool = False
+    traffic: list[RouteTrafficStatus] | None = None
 
 
 @dataclass(frozen=True)

src/ai/backend/manager/repositories/deployment/db_source/db_source.py

@@ -4,7 +4,7 @@
 import logging
 import uuid
 from collections import Counter, defaultdict
-from collections.abc import AsyncIterator, Mapping, Sequence
+from collections.abc import AsyncIterator, Collection, Mapping, Sequence
 from contextlib import asynccontextmanager as actxmgr
 from dataclasses import dataclass
 from datetime import UTC, datetime
@@ -18,7 +18,7 @@
 from sqlalchemy.ext.asyncio import async_sessionmaker
 from sqlalchemy.orm import selectinload
 
-from ai.backend.common.config import ModelDefinition, ModelHealthCheck
+from ai.backend.common.config import ModelHealthCheck
 from ai.backend.common.data.endpoint.types import EndpointLifecycle
 from ai.backend.common.data.permission.types import RBACElementType
 from ai.backend.common.dto.manager.v2.runtime_variant_preset.types import (
@@ -71,12 +71,10 @@
     ModelDeploymentAutoScalingRuleData,
     ModelRevisionData,
     RevisionSearchResult,
-    RouteHealthCheckFilter,
     RouteHealthStatus,
     RouteInfo,
     RouteSearchResult,
     RouteStatus,
-    RouteTargetStatuses,
     ScalingGroupCleanupConfig,
 )
 from ai.backend.manager.data.deployment_revision_preset.types import ResourceSlotEntryData
@@ -178,7 +176,6 @@
     ProjectDeploymentSearchScope,
     RouteData,
     RouteServiceDiscoveryInfo,
-    RouteSessionData,
 )
 from ai.backend.manager.repositories.scheduler.types.session_creation import (
     ContainerUserContext,
@@ -204,34 +201,6 @@ class EndpointWithRoutesRawData:
 log = BraceStyleAdapter(logging.getLogger(__name__))
 
 
-def _build_session_data(
-    session: uuid.UUID | None,
-    session_status: SessionStatus | None,
-) -> RouteSessionData | None:
-    """Compose ``RouteSessionData`` from a route's joined session columns.
-
-    Returns ``None`` when the route has no session yet (``session IS NULL``)
-    or when the joined session row is missing.
-    """
-    if session is None or session_status is None:
-        return None
-    return RouteSessionData(session_id=SessionId(session), status=session_status)
-
-
-def _project_health_check_config(
-    model_definition: ModelDefinition | None,
-) -> ModelHealthCheck | None:
-    """Project the joined ``model_definition`` to its health-check block.
-
-    ``None`` means the revision opted out of HTTP probing — the manager
-    skips probe scheduling and AppProxy receives traffic as soon as the
-    route reaches RUNNING.
-    """
-    if model_definition is None:
-        return None
-    return model_definition.health_check_config()
-
-
 def _project_preset_slots(
     preset_row: DeploymentRevisionPresetRow | None,
     slot_entries: list[tuple[str, Decimal]],
@@ -1009,17 +978,13 @@ async def get_routes_by_endpoint(
     ) -> list[RouteData]:
         """Get all routes for an endpoint."""
         async with self._begin_readonly_session_read_committed() as db_sess:
-            query = (
-                sa.select(RoutingRow, SessionRow.status)
-                .outerjoin(SessionRow, RoutingRow.session == SessionRow.id)
-                .where(RoutingRow.endpoint == endpoint_id)
-            )
+            query = sa.select(RoutingRow).where(RoutingRow.endpoint == endpoint_id)
             result = await db_sess.execute(query)
             return [
                 RouteData(
                     route_id=row.id,
                     deployment_id=row.endpoint,
-                    session_data=_build_session_data(row.session, session_status),
+                    session_id=SessionId(row.session) if row.session else None,
                     status=row.status,
                     health_status=row.health_status,
                     traffic_ratio=row.traffic_ratio,
@@ -1029,7 +994,7 @@ async def get_routes_by_endpoint(
                     replica_port=row.replica_port,
                     error_data=row.error_data or {},
                 )
-                for row, session_status in result.all()
+                for row in result.scalars().all()
             ]
 
     async def update_route_session(
@@ -1616,61 +1581,63 @@ async def scale_routes(
 
     # Route operations
 
-    async def get_routes_by_statuses(
+    async def search_route_datas(
         self,
-        target: RouteTargetStatuses,
-        health_check_filter: RouteHealthCheckFilter,
+        querier: BatchQuerier,
     ) -> list[RouteData]:
-        """Routes matching ``(lifecycle, health, traffic)`` with
-        revision-level ``health_check_config`` gating applied in memory.
+        """Return :class:`RouteData` rows matched by ``querier`` (no joins).
 
-        ``model_definition`` is selected so the resolved
-        ``ModelHealthCheck`` (or ``None``) is attached to each
-        :class:`RouteData`; ``health_check_filter`` then runs over those
-        materialized rows.
+        Consumers that need session status or revision-level health-check
+        config call :meth:`fetch_session_statuses_by_route_ids` or
+        :meth:`fetch_health_check_configs` separately.
         """
         async with self._begin_readonly_session_read_committed() as db_sess:
-            query = (
-                sa.select(
-                    RoutingRow,
-                    DeploymentRevisionRow.model_definition,
-                    SessionRow.status,
-                )
-                .join(
-                    DeploymentRevisionRow,
-                    DeploymentRevisionRow.id == RoutingRow.revision,
-                )
-                .outerjoin(SessionRow, RoutingRow.session == SessionRow.id)
-                .where(
-                    RoutingRow.status.in_(target.lifecycle),
-                    RoutingRow.health_status.in_(target.health),
+            query = sa.select(RoutingRow)
+            result = await execute_batch_querier(db_sess, query, querier)
+            return [
+                RouteData(
+                    route_id=row.RoutingRow.id,
+                    deployment_id=row.RoutingRow.endpoint,
+                    session_id=SessionId(row.RoutingRow.session)
+                    if row.RoutingRow.session
+                    else None,
+                    status=row.RoutingRow.status,
+                    health_status=row.RoutingRow.health_status,
+                    traffic_ratio=row.RoutingRow.traffic_ratio,
+                    created_at=row.RoutingRow.created_at,
+                    revision_id=DeploymentRevisionID(row.RoutingRow.revision),
+                    replica_host=row.RoutingRow.replica_host,
+                    replica_port=row.RoutingRow.replica_port,
+                    error_data=row.RoutingRow.error_data or {},
                 )
-            )
-            if target.traffic is not None:
-                query = query.where(RoutingRow.traffic_status == target.traffic)
+                for row in result.rows
+            ]
+
+    async def fetch_health_check_configs(
+        self,
+        revision_ids: Collection[DeploymentRevisionID],
+    ) -> Mapping[DeploymentRevisionID, ModelHealthCheck | None]:
+        """Resolve revision-level ``ModelHealthCheck`` for each revision id.
+
+        Missing revisions are omitted from the result; revisions that opted
+        out of ``service.health_check`` return ``None``.
+        """
+        if not revision_ids:
+            return {}
+        unique_ids = {uuid.UUID(str(rid)) for rid in revision_ids}
+        async with self._begin_readonly_session_read_committed() as db_sess:
+            query = sa.select(
+                DeploymentRevisionRow.id,
+                DeploymentRevisionRow.model_definition,
+            ).where(DeploymentRevisionRow.id.in_(unique_ids))
             result = await db_sess.execute(query)
-            routes: list[RouteData] = []
-            for row, model_definition, session_status in result.all():
-                health_check_config = _project_health_check_config(model_definition)
-                if health_check_filter.health_check_required and health_check_config is None:
-                    continue
-                routes.append(
-                    RouteData(
-                        route_id=row.id,
-                        deployment_id=row.endpoint,
-                        session_data=_build_session_data(row.session, session_status),
-                        status=row.status,
-                        health_status=row.health_status,
-                        traffic_ratio=row.traffic_ratio,
-                        created_at=row.created_at,
-                        revision_id=DeploymentRevisionID(row.revision),
-                        replica_host=row.replica_host,
-                        replica_port=row.replica_port,
-                        error_data=row.error_data or {},
-                        health_check_config=health_check_config,
-                    )
+            configs: dict[DeploymentRevisionID, ModelHealthCheck | None] = {}
+            for revision_id, model_definition in result.all():
+                config = (
+                    model_definition.health_check_config() if model_definition is not None else None
                 )
-            return routes
+                configs[DeploymentRevisionID(revision_id)] = config
+            return configs
 
     async def update_route_status_bulk(
         self,

src/ai/backend/manager/repositories/deployment/repository.py

@@ -3,7 +3,7 @@
 import logging
 import uuid
 from collections import defaultdict
-from collections.abc import Mapping, Sequence
+from collections.abc import Collection, Mapping, Sequence
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from decimal import Decimal, DecimalException
@@ -14,7 +14,7 @@
 from ai.backend.common.clients.valkey_client.valkey_live.client import ValkeyLiveClient
 from ai.backend.common.clients.valkey_client.valkey_schedule.client import ValkeyScheduleClient
 from ai.backend.common.clients.valkey_client.valkey_stat.client import ValkeyStatClient
-from ai.backend.common.config import ModelDefinitionDraft
+from ai.backend.common.config import ModelDefinitionDraft, ModelHealthCheck
 from ai.backend.common.data.endpoint.types import EndpointLifecycle
 from ai.backend.common.exception import BackendAIError, InvalidAPIParameters
 from ai.backend.common.identifier.deployment import DeploymentID
@@ -67,11 +67,9 @@
     ModelDeploymentAutoScalingRuleData,
     ModelRevisionData,
     RevisionSearchResult,
-    RouteHealthCheckFilter,
     RouteInfo,
     RouteSearchResult,
     RouteStatus,
-    RouteTargetStatuses,
     ScalingGroupCleanupConfig,
 )
 from ai.backend.manager.data.image.types import ImageIdentifier
@@ -629,15 +627,30 @@ async def scale_routes(
     # Route operations
 
     @deployment_repository_resilience.apply()
-    async def get_routes_by_statuses(
+    async def search_route_datas(
         self,
-        target: RouteTargetStatuses,
-        health_check_filter: RouteHealthCheckFilter,
+        querier: BatchQuerier,
     ) -> list[RouteData]:
-        """Routes matching ``(lifecycle, health, traffic)`` with the
-        in-memory ``health_check_config`` post-filter applied.
+        """Return :class:`RouteData` rows matched by ``querier`` (no joins).
+
+        Session status and revision-level health-check config are not
+        joined — callers fetch them via
+        :meth:`fetch_session_statuses_by_route_ids` and
+        :meth:`fetch_health_check_configs`.
+        """
+        return await self._db_source.search_route_datas(querier)
+
+    @deployment_repository_resilience.apply()
+    async def fetch_health_check_configs(
+        self,
+        revision_ids: Collection[DeploymentRevisionID],
+    ) -> Mapping[DeploymentRevisionID, ModelHealthCheck | None]:
+        """Resolve revision-level ``ModelHealthCheck`` for each id.
+
+        Revisions that opted out of ``service.health_check`` map to
+        ``None``; unknown revision ids are omitted.
         """
-        return await self._db_source.get_routes_by_statuses(target, health_check_filter)
+        return await self._db_source.fetch_health_check_configs(revision_ids)
 
     @deployment_repository_resilience.apply()
     async def update_route_status_bulk(

src/ai/backend/manager/repositories/deployment/types/__init__.py

@@ -7,7 +7,6 @@
     ProjectDeploymentSearchScope,
     RouteData,
     RouteServiceDiscoveryInfo,
-    RouteSessionData,
 )
 
 __all__ = [
@@ -17,5 +16,4 @@
     "ProjectDeploymentSearchScope",
     "RouteData",
     "RouteServiceDiscoveryInfo",
-    "RouteSessionData",
 ]

src/ai/backend/manager/repositories/deployment/types/endpoint.py

@@ -11,13 +11,11 @@
 
 import sqlalchemy as sa
 
-from ai.backend.common.config import ModelHealthCheck
 from ai.backend.common.data.endpoint.types import EndpointLifecycle
 from ai.backend.common.identifier.deployment import DeploymentID
 from ai.backend.common.identifier.deployment_revision import DeploymentRevisionID
 from ai.backend.common.types import SessionId
 from ai.backend.manager.data.deployment.types import RouteHealthStatus, RouteStatus
-from ai.backend.manager.data.session.types import SessionStatus
 from ai.backend.manager.errors.resource import ProjectNotFound
 from ai.backend.manager.models.endpoint.row import EndpointRow
 from ai.backend.manager.models.group.row import GroupRow
@@ -60,21 +58,13 @@ class EndpointData:
     resource_opts: dict[str, Any] = field(default_factory=dict)
 
 
-@dataclass(frozen=True)
-class RouteSessionData:
-    """Session id paired with its current status."""
-
-    session_id: SessionId
-    status: SessionStatus
-
-
 @dataclass
 class RouteData:
     """Data structure for model service route."""
 
     route_id: uuid.UUID
     deployment_id: DeploymentID
-    session_data: RouteSessionData | None
+    session_id: SessionId | None
     status: RouteStatus
     health_status: RouteHealthStatus
     traffic_ratio: float
@@ -84,12 +74,6 @@ class RouteData:
     replica_port: int | None = None
     updated_at: datetime | None = None
     error_data: dict[str, Any] = field(default_factory=dict)
-    health_check_config: ModelHealthCheck | None = None
-
-    @property
-    def session_id(self) -> SessionId | None:
-        """Convenience accessor; reads from ``session_data``."""
-        return self.session_data.session_id if self.session_data else None
 
 
 @dataclass