fix(BA-7328): stop scheduling past a resource-exhausted session

changes/13707.fix.md

@@ -0,0 +1 @@
+Stop scheduling sessions behind one whose resource group ran out of resources, so lower-priority sessions no longer take what a blocked higher-priority session is waiting for

src/ai/backend/manager/models/scheduling_history/row.py

@@ -78,17 +78,28 @@ class SessionSchedulingHistoryRow(Base):
         onupdate=sa.func.now(),
     )
 
+    def records_an_attempt(self) -> bool:
+        """Whether this record describes an attempt rather than a skip."""
+        return self.result != SchedulingResult.SKIPPED
+
     def should_merge_with(self, new_row: SessionSchedulingHistoryRow) -> bool:
         """Check if a new entry should be merged with this one.
 
         Merge conditions:
         - Same phase, error_code, and to_status -> merge (increment attempts)
-        - from_status and result (success/failure) do not affect merge decision
+        - Both must describe an attempt, or both a skip
+        - from_status and which attempt result (success/failure/give-up) do
+          not affect the merge decision
+
+        Skips are kept apart because ``attempts`` drives the give-up
+        (deprioritization) classification, which may only count attempts a
+        session really got.
         """
         return (
             self.phase == new_row.phase
             and self.error_code == new_row.error_code
             and self.to_status == new_row.to_status
+            and self.records_an_attempt() == new_row.records_an_attempt()
         )
 
     def to_data(self) -> SessionSchedulingHistoryData:

src/ai/backend/manager/sokovan/scheduler/coordinator.py

@@ -39,6 +39,7 @@
 )
 from ai.backend.manager.metrics.scheduler import SchedulerOperationMetricObserver
 from ai.backend.manager.models.kernel.conditions import KernelConditions
+from ai.backend.manager.models.scheduling_history.row import SessionSchedulingHistoryRow
 from ai.backend.manager.models.session.conditions import SessionConditions
 from ai.backend.manager.models.specs.pagination import NoPagination, OffsetPagination
 from ai.backend.manager.repositories.base import BatchQuerier
@@ -58,6 +59,7 @@
 from ai.backend.manager.types import DistributedLockFactory
 from ai.backend.manager.views.sokovan.lifecycle import (
     KernelCreationInfo,
+    LastPhase,
     SessionWithKernels,
 )
 from ai.backend.manager.views.sokovan.result import PromotionSpec
@@ -140,7 +142,8 @@ class FailureClassificationResult:
 
     Classification priority (first match wins):
     1. give_up: per-handler ``max_retry_count`` is set AND
-       ``phase_attempts`` reached it. ``None`` means "no retry limit"
+       the last phase record's ``attempts`` reached it. ``None`` means
+       "no retry limit"
        — give-up never fires.
     2. expired: per-handler ``timeout`` is set AND phase elapsed time
        exceeded it. ``None`` means "no timeout" — expired never fires.
@@ -745,19 +748,11 @@ async def _process_resource_group(
         # Extract session IDs for recorder entity_ids
         session_ids = [s.session_info.identity.id for s in sessions]
 
-        # Populate phase_attempts and phase_started_at from scheduling history for failure classification
+        # Carry the last phase record onto the sessions for failure classification
         # Get last history records (regardless of phase), then compare phase at application level
         history_map = await self._repository.get_last_session_histories(session_ids)
         handler_name = handler.name()
-        for session in sessions:
-            history = history_map.get(session.session_info.identity.id)
-            # Only use history data if the last history is for the current phase
-            if history and history.phase == handler_name:
-                session.phase_attempts = history.attempts
-                session.phase_started_at = history.created_at
-            else:
-                session.phase_attempts = 0
-                session.phase_started_at = None
+        self._populate_phase_history(sessions, history_map, handler_name)
 
         # Create recorder scoped to this resource group
         recorder_scope = f"{schedule_type.value}:{resource_group_id}"
@@ -802,6 +797,30 @@ async def _process_resource_group(
                     resource_group_id,
                 )
 
+    def _populate_phase_history(
+        self,
+        sessions: list[SessionWithKernels],
+        history_map: Mapping[SessionId, SessionSchedulingHistoryRow],
+        handler_name: str,
+    ) -> None:
+        """Carry this phase's last history record onto the sessions.
+
+        Only the last record counts, and only while it is still this phase's;
+        anything else leaves the session without one. Skips are carried like
+        any other record — ``LastPhase.result`` says what was counted, and the
+        give-up rule in :meth:`_classify_failures` is what excludes skips.
+        """
+        for session in sessions:
+            history = history_map.get(session.session_info.identity.id)
+            if history and history.phase == handler_name:
+                session.last_phase = LastPhase(
+                    attempts=history.attempts,
+                    started_at=history.created_at,
+                    result=SchedulingResult(history.result),
+                )
+            else:
+                session.last_phase = None
+
     async def _process_promotion_resource_group(
         self,
         spec: PromotionSpec,
@@ -1120,7 +1139,7 @@ async def _handle_result(
             handler: The lifecycle handler that produced the result
             result: Execution result containing successes, failures, and skipped
             records: Mapping of session IDs to their execution records for sub_steps
-            sessions: Original sessions with phase_attempts for failure classification
+            sessions: Original sessions with last_phase for failure classification
 
         Returns:
             FailureClassificationResult if there were failures, None otherwise.
@@ -1230,13 +1249,15 @@ def _classify_failures(
         is ``None``; ``expired`` never fires when ``timeout`` is ``None``.
 
         Classification priority (first match wins):
-        1. give_up: max_retry_count set AND phase_attempts >= max_retry_count
+        1. give_up: max_retry_count set AND last_phase.attempts >= max_retry_count,
+           and those attempts are not skips (a session queued behind a blocked
+           one was never attempted, so it may not be given up on)
         2. expired: timeout set AND phase elapsed > timeout
         3. need_retry: default
 
         Args:
             failures: Failed session transition info
-            sessions: Original sessions with phase_attempts and phase_started_at populated
+            sessions: Original sessions with last_phase populated
             current_time: Current database time for timeout comparison
             handler_name: ``SessionLifecycleHandler.name()`` for resolving
                 per-handler policy from ``SessionHandlerOptions``
@@ -1257,14 +1278,21 @@ def _classify_failures(
                 continue
 
             policy = session.session_info.handler_options.resolve(handler_name)
-
-            # 1. Check max retries exceeded → give_up
-            if policy.is_retry_exhausted(session.phase_attempts):
+            last_phase = session.last_phase
+
+            # 1. Check max retries exceeded → give_up.
+            # A skip is recorded like any other result but is not an attempt:
+            # a session queued behind a blocked one may not be given up on for
+            # work it never got to do.
+            attempts = last_phase.attempts if last_phase is not None else 0
+            was_skipped = last_phase is not None and last_phase.result == SchedulingResult.SKIPPED
+            if not was_skipped and policy.is_retry_exhausted(attempts):
                 give_up_failures.append(failure)
                 continue
 
             # 2. Check timeout exceeded → expired
-            if policy.is_timed_out(session.phase_started_at, current_time):
+            started_at = last_phase.started_at if last_phase is not None else None
+            if policy.is_timed_out(started_at, current_time):
                 expired_failures.append(failure)
                 continue
 

src/ai/backend/manager/sokovan/scheduler/handlers/lifecycle/schedule_sessions.py

@@ -116,6 +116,10 @@ async def execute(
         - successes: Sessions that were scheduled
         - failures: Sessions whose scheduling attempt failed in the Provisioner
         - skipped: Sessions that were not attempted (priority-based, resource constraints)
+
+        A session left unattempted must stay out of `failures`: the retry
+        pressure that eventually deprioritizes a session may only be charged
+        to attempts that actually happened.
         """
         result = SessionExecutionResult()
 
@@ -141,6 +145,7 @@ async def execute(
         failure_map = {
             failure.session_id: failure for failure in schedule_result.scheduling_failures
         }
+        skip_map = {skip.session_id: skip for skip in schedule_result.scheduling_skips}
 
         # Reservation-backed placements: the sessions hold their resources
         # (kernels already RESERVED) and wait for their victims; the victims
@@ -180,6 +185,9 @@ async def execute(
             elif session_id in failure_map:
                 reason = failure_map[session_id].msg or "scheduling-failed"
                 result.failures.append(self._to_transition_info(session, reason))
+            elif session_id in skip_map:
+                reason = skip_map[session_id].msg or "not-attempted-this-cycle"
+                result.skipped.append(self._to_transition_info(session, reason))
             else:
                 result.skipped.append(self._to_transition_info(session, "not-scheduled-this-cycle"))
 

src/ai/backend/manager/sokovan/scheduler/provisioner/provisioner.py

@@ -19,7 +19,11 @@
 from ai.backend.manager.sokovan.recorder import (
     RecorderContext,
 )
-from ai.backend.manager.sokovan.scheduler.results import PreemptionPlanEntry, ScheduleResult
+from ai.backend.manager.sokovan.scheduler.results import (
+    PreemptionPlanEntry,
+    ScheduleResult,
+    SchedulingSkip,
+)
 from ai.backend.manager.views.sokovan.allocation import (
     KernelAllocation,
     SchedulingFailure,
@@ -30,6 +34,7 @@
 from ai.backend.manager.views.sokovan.snapshot import ScopeVictimCandidates, SystemSnapshot
 from ai.backend.manager.views.sokovan.workload import SessionWorkload
 
+from .selectors.exceptions import BatchAgentSelectionFailedError, NoAvailableAgentError
 from .selectors.selector import (
     AgentSelection,
     AgentSelectionCriteria,
@@ -136,6 +141,10 @@ async def schedule_resource_group(
         3. Agent selection: Select agents using configured strategy
         4. Allocation: Persist allocations to database
 
+        Once a session fails because the group ran out of resources, the
+        sessions behind it are left unattempted and reported as skips, so
+        lower-priority sessions cannot take the resources it is waiting for.
+
         Args:
             scheduling_data: Pre-fetched scheduling data from Handler.
 
@@ -172,8 +181,9 @@ async def schedule_resource_group(
         reserved_allocations: list[SessionAllocation] = []
         claimed_victim_ids: set[SessionId] = set()
         scheduling_failures: list[SchedulingFailure] = []
+        scheduling_skips: list[SchedulingSkip] = []
 
-        for session_workload in sequenced_workloads:
+        for index, session_workload in enumerate(sequenced_workloads):
             try:
                 # Sequencing phase is automatically included via shared phases
                 session_allocation = await self._schedule_workload(
@@ -186,33 +196,43 @@ async def schedule_resource_group(
                     claimed_victim_ids.update(session_allocation.preempting_session_ids)
                 else:
                     session_allocations.append(session_allocation)
-            except Exception as e:
-                log.debug(
-                    "Scheduling failed for workload {}: {}",
-                    session_workload.meta.session_id,
-                    e,
-                )
-                scheduling_failures.append(
-                    SchedulingFailure(
-                        session_id=session_workload.meta.session_id,
-                        msg=str(e),
+            except (BatchAgentSelectionFailedError, NoAvailableAgentError) as e:
+                # The group ran out of resources. This is the one failure the
+                # sessions behind share: they would take exactly what this one
+                # is waiting for, so they are left unattempted.
+                self._record_failure(scheduling_failures, session_workload, e)
+                scheduling_skips.extend(
+                    self._skips_behind(
+                        sequenced_workloads[index + 1 :], session_workload.meta.session_id
                     )
                 )
-                continue
+                break
+            except Exception as e:
+                # Specific to the session that hit it (an architecture mismatch,
+                # an unsatisfied dependency, an exceeded quota, a group with no
+                # agents at all), so it must not stall the sessions behind it.
+                self._record_failure(scheduling_failures, session_workload, e)
 
         log.info(
-            "Processing {} allocations, {} reservations and {} failures in resource group {}",
+            "Processing {} allocations, {} reservations, {} failures"
+            " and {} skips in resource group {}",
             len(session_allocations),
             len(reserved_allocations),
             len(scheduling_failures),
+            len(scheduling_skips),
             resource_group_id,
         )
         with self._phase_metrics.measure_phase("scheduler", resource_group_id, "allocation"):
             scheduled_session_ids = await self._repository.allocate_sessions(session_allocations)
             reserved_session_ids = await self._repository.reserve_sessions(reserved_allocations)
 
-        failure_ids = [f.session_id for f in scheduling_failures]
-        await self._valkey_schedule.set_pending_queue(state.resource_group.name, failure_ids)
+        # The pending queue is the whole still-waiting queue in sequencing
+        # order (the legacy GraphQL resolver re-sorts by this order), so the
+        # unattempted sessions belong in it just as much as the failed ones.
+        pending_ids = [f.session_id for f in scheduling_failures] + [
+            s.session_id for s in scheduling_skips
+        ]
+        await self._valkey_schedule.set_pending_queue(state.resource_group.name, pending_ids)
         reserved_ids = set(reserved_session_ids)
         return ScheduleResult(
             scheduled_session_ids=scheduled_session_ids,
@@ -226,8 +246,44 @@ async def schedule_resource_group(
                 for allocation in reserved_allocations
                 if allocation.session_id in reserved_ids
             ],
+            scheduling_skips=scheduling_skips,
         )
 
+    def _record_failure(
+        self,
+        scheduling_failures: list[SchedulingFailure],
+        session_workload: SessionWorkload,
+        error: Exception,
+    ) -> None:
+        log.debug(
+            "Scheduling failed for workload {}: {}",
+            session_workload.meta.session_id,
+            error,
+        )
+        scheduling_failures.append(
+            SchedulingFailure(
+                session_id=session_workload.meta.session_id,
+                msg=str(error),
+            )
+        )
+
+    def _skips_behind(
+        self,
+        remaining_workloads: Sequence[SessionWorkload],
+        blocking_session_id: SessionId,
+    ) -> list[SchedulingSkip]:
+        """The unattempted tail of the queue, in sequencing order."""
+        return [
+            SchedulingSkip(
+                session_id=workload.meta.session_id,
+                msg=(
+                    "Not attempted: the resource group ran out of resources at the"
+                    f" earlier session {blocking_session_id}"
+                ),
+            )
+            for workload in remaining_workloads
+        ]
+
     async def _schedule_workload(
         self,
         state: SchedulingState,

src/ai/backend/manager/sokovan/scheduler/results.py

@@ -27,6 +27,19 @@ class PreemptionPlanEntry:
     victim_session_ids: tuple[SessionId, ...]
 
 
+@dataclass
+class SchedulingSkip:
+    """A session that was not attempted this pass.
+
+    Distinct from a :class:`SchedulingFailure`: nothing was tried, so no
+    retry pressure (and no eventual deprioritization) may be charged to it.
+    """
+
+    session_id: SessionId
+    # Human-readable skip reason (transition reason / pending queue)
+    msg: str
+
+
 @dataclass
 class ScheduleResult:
     """Result of a scheduling operation."""
@@ -40,6 +53,9 @@ class ScheduleResult:
     reserved_session_ids: list[SessionId]
     # The preemption plans backing those reservations.
     preemption_plan: list[PreemptionPlanEntry]
+    # Sessions left unattempted because an earlier session exhausted the
+    # group's resources, in sequencing order.
+    scheduling_skips: list[SchedulingSkip] = field(default_factory=list)
 
     def success_count(self) -> int:
         """Get the count of successfully scheduled sessions."""