fix: Enable exhaustive search for recursive session termination (#1617)

Backported-from: main
Backported-to: 23.03

Author: rapsealk

changes/1617.fix.md

@@ -0,0 +1 @@
+Enable exhaustive search for recursive session termination irrelevant to each session's status

src/ai/backend/manager/api/session.py

@@ -1152,27 +1152,34 @@ async def destroy(request: web.Request, params: Any) -> web.Response:
     if params["recursive"]:
         async with root_ctx.db.begin_readonly_session() as db_sess:
             dependent_session_ids = await find_dependent_sessions(
-                session_name, db_sess, owner_access_key
+                session_name,
+                db_sess,
+                owner_access_key,
+                allow_stale=True,
             )
 
             target_session_references: List[str | uuid.UUID] = [
                 *dependent_session_ids,
                 session_name,
             ]
-            sessions = [
-                await SessionRow.get_session(
-                    db_sess,
-                    name_or_id,
-                    owner_access_key,
-                    kernel_loading_strategy=KernelLoadingStrategy.ALL_KERNELS,
-                )
-                for name_or_id in target_session_references
-            ]
+            sessions: Iterable[SessionRow | Exception] = await asyncio.gather(
+                *[
+                    SessionRow.get_session(
+                        db_sess,
+                        name_or_id,
+                        owner_access_key,
+                        kernel_loading_strategy=KernelLoadingStrategy.ALL_KERNELS,
+                    )
+                    for name_or_id in target_session_references
+                ],
+                return_exceptions=True,
+            )
 
         last_stats = await asyncio.gather(
             *[
                 root_ctx.registry.destroy_session(sess, forced=params["forced"])
                 for sess in sessions
+                if isinstance(sess, SessionRow)
             ],
             return_exceptions=True,
         )
@@ -1570,6 +1577,8 @@ async def find_dependent_sessions(
     root_session_name_or_id: str | uuid.UUID,
     db_session: SASession,
     access_key: AccessKey,
+    *,
+    allow_stale: bool = False,
 ) -> Set[uuid.UUID]:
     async def _find_dependent_sessions(session_id: uuid.UUID) -> Set[uuid.UUID]:
         result = await db_session.execute(
@@ -1588,7 +1597,10 @@ async def _find_dependent_sessions(session_id: uuid.UUID) -> Set[uuid.UUID]:
         return dependent_sessions
 
     root_session = await SessionRow.get_session(
-        db_session, root_session_name_or_id, access_key=access_key
+        db_session,
+        root_session_name_or_id,
+        access_key=access_key,
+        allow_stale=allow_stale,
     )
     return await _find_dependent_sessions(cast(uuid.UUID, root_session.id))
 

src/ai/backend/manager/api/utils.py

@@ -324,7 +324,7 @@ async def call_non_bursty(
     Execute a coroutine once upon max_bursts bursty invocations or max_idle
     milliseconds after bursts smaller than max_bursts.
     """
-    global _burst_last_call, _burst_calls, _burst_counts
+    global _burst_last_call, _burst_times, _burst_counts
     if inspect.iscoroutine(coro):
         # Coroutine objects may not be called before garbage-collected
         # as this function throttles the frequency of invocation.