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

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

@@ -77,17 +77,28 @@ class SessionSchedulingHistoryRow(Base):  # type: ignore[misc]
         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/data/lifecycle.py

@@ -18,8 +18,14 @@
     SessionId,
     SessionTypes,
 )
+<<<<<<< HEAD:src/ai/backend/manager/sokovan/data/lifecycle.py
 from ai.backend.manager.data.kernel.types import KernelInfo
 from ai.backend.manager.data.session.types import SessionInfo
+=======
+from ai.backend.manager.data.kernel.types import KernelInfo, KernelStatus
+from ai.backend.manager.data.network.types import NetworkType
+from ai.backend.manager.data.session.types import SchedulingResult, SessionInfo
+>>>>>>> 99e8e59f (fix(BA-7328): stop scheduling past a resource-exhausted session (#13707)):src/ai/backend/manager/views/sokovan/lifecycle.py
 from ai.backend.manager.defs import DEFAULT_ROLE
 from ai.backend.manager.errors.kernel import MainKernelNotFound, TooManyKernelsFound
 from ai.backend.manager.models.kernel import KernelStatus
@@ -300,6 +306,26 @@ class SessionRunningData:
     occupying_slots: ResourceSlot
 
 
+@dataclass(frozen=True)
+class LastPhase:
+    """The session's last scheduling-history record of the phase in progress.
+
+    Absent when the session has no record of that phase yet. Read by the
+    coordinator's failure classification: ``attempts`` against the retry
+    budget, ``started_at`` against the timeout, and ``result`` to tell an
+    attempt from a skip.
+
+    Attributes:
+        attempts: How many times the phase was recorded, skips included
+        started_at: When the phase was first recorded
+        result: What the record ended in
+    """
+
+    attempts: int
+    started_at: datetime
+    result: SchedulingResult
+
+
 @dataclass
 class SessionWithKernels:
     """
@@ -311,16 +337,13 @@ class SessionWithKernels:
     Attributes:
         session_info: Session information including lifecycle data
         kernel_infos: List of kernels belonging to this session
-        phase_attempts: Number of attempts for current phase from scheduling history
-                       (used for failure classification: give_up when >= max_retries)
-        phase_started_at: When the current phase started from scheduling history
-                         (used for failure classification: expired when timeout exceeded)
+        last_phase: The session's last record of the phase being processed,
+                   or None when it has none yet
     """
 
     session_info: SessionInfo
     kernel_infos: list[KernelInfo]
-    phase_attempts: int = 0
-    phase_started_at: datetime | None = None
+    last_phase: LastPhase | None = None
 
     @property
     def main_kernel(self) -> KernelInfo:

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

@@ -38,6 +38,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.repositories.base import BatchQuerier
 from ai.backend.manager.repositories.base.creator import BulkCreator
@@ -60,6 +61,15 @@
 from ai.backend.manager.sokovan.scheduler.types import ScheduleType
 from ai.backend.manager.sokovan.scheduling_controller import SchedulingController
 from ai.backend.manager.types import DistributedLockFactory
+<<<<<<< HEAD
+=======
+from ai.backend.manager.views.sokovan.lifecycle import (
+    KernelCreationInfo,
+    LastPhase,
+    SessionWithKernels,
+)
+from ai.backend.manager.views.sokovan.result import PromotionSpec
+>>>>>>> 99e8e59f (fix(BA-7328): stop scheduling past a resource-exhausted session (#13707))
 
 from .factory import CoordinatorHandlers
 from .handlers import SessionLifecycleHandler
@@ -139,7 +149,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.
@@ -740,19 +751,11 @@ async def _process_scaling_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 scaling group
         recorder_scope = f"{schedule_type.value}:{scaling_group}"
@@ -797,7 +800,35 @@ async def _process_scaling_group(
                     scaling_group,
                 )
 
+<<<<<<< HEAD
     async def _process_promotion_scaling_group(
+=======
+    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(
+>>>>>>> 99e8e59f (fix(BA-7328): stop scheduling past a resource-exhausted session (#13707))
         self,
         spec: PromotionSpec,
         schedule_type: ScheduleType,
@@ -1115,7 +1146,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.
@@ -1225,13 +1256,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``
@@ -1252,14 +1285,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

@@ -105,6 +105,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()
 
@@ -133,6 +137,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}
 
         # Allocated sessions transition to SCHEDULED; failed attempts are reported
         # as failures so the coordinator can classify them (need_retry/expired/
@@ -146,6 +151,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"))