re-use same closing function

sanderegg · sanderegg · commit 8e59daa45b00 · 2025-07-25T15:44:34.000+02:00
diff --git a/services/web/server/src/simcore_service_webserver/garbage_collector/_core.py b/services/web/server/src/simcore_service_webserver/garbage_collector/_core.py
@@ -5,7 +5,7 @@
 from aiohttp import web
 from servicelib.logging_utils import log_catch, log_context
 
-from ..resource_manager.registry import RedisResourceRegistry, get_registry
+from ..resource_manager.registry import get_registry
 from ._core_disconnected import remove_disconnected_user_resources
 from ._core_guests import remove_users_manually_marked_as_guests
 from ._core_orphans import remove_orphaned_services
@@ -34,24 +34,31 @@ async def collect_garbage(app: web.Application):
     The field `garbage_collection_interval_seconds` defines the interval at which this
     function will be called.
     """
-    registry: RedisResourceRegistry = get_registry(app)
+    registry = get_registry(app)
 
-    with log_catch(_logger, reraise=False), log_context(
-        _logger, logging.INFO, "Step 1: Removes disconnected user sessions"
+    with (
+        log_catch(_logger, reraise=False),
+        log_context(
+            _logger, logging.INFO, "Step 1: Removes disconnected user sessions"
+        ),
     ):
         # Triggers signal to close possible pending opened projects
         # Removes disconnected GUEST users after they finished their sessions
         await remove_disconnected_user_resources(registry, app)
 
-    with log_catch(_logger, reraise=False), log_context(
-        _logger, logging.INFO, "Step 2: Removes users manually marked for removal"
+    with (
+        log_catch(_logger, reraise=False),
+        log_context(
+            _logger, logging.INFO, "Step 2: Removes users manually marked for removal"
+        ),
     ):
         # if a user was manually marked as GUEST it needs to be
         # removed together with all the associated projects
         await remove_users_manually_marked_as_guests(registry, app)
 
-    with log_catch(_logger, reraise=False), log_context(
-        _logger, logging.INFO, "Step 3: Removes orphaned services"
+    with (
+        log_catch(_logger, reraise=False),
+        log_context(_logger, logging.INFO, "Step 3: Removes orphaned services"),
     ):
         # For various reasons, some services remain pending after
         # the projects are closed or the user was disconencted.
diff --git a/services/web/server/src/simcore_service_webserver/garbage_collector/_core_disconnected.py b/services/web/server/src/simcore_service_webserver/garbage_collector/_core_disconnected.py
@@ -1,17 +1,14 @@
 import logging
 
 from aiohttp import web
-from redis.asyncio import Redis
 from servicelib.common_headers import UNDEFINED_DEFAULT_SIMCORE_USER_AGENT_VALUE
 from servicelib.utils import logged_gather
 
-from ..projects._projects_service import remove_project_dynamic_services
+from ..projects import _projects_service
 from ..projects.exceptions import ProjectLockError, ProjectNotFoundError
 from ..redis import get_redis_lock_manager_client
 from ..resource_manager.registry import (
     RedisResourceRegistry,
-    ResourcesDict,
-    UserSessionDict,
 )
 from ._core_guests import remove_guest_user_with_all_its_resources
 from .settings import GUEST_USER_RC_LOCK_FORMAT
@@ -22,7 +19,7 @@
 async def remove_disconnected_user_resources(
     registry: RedisResourceRegistry, app: web.Application
 ) -> None:
-    lock_manager: Redis = get_redis_lock_manager_client(app)
+    lock_manager = get_redis_lock_manager_client(app)
 
     #
     # In redis jargon, every entry is denoted as "key"
@@ -40,10 +37,9 @@ async def remove_disconnected_user_resources(
 
     # clean up all resources of expired keys
     for dead_session in all_sessions_dead:
-
         try:
             user_id = int(dead_session["user_id"])
-        except (KeyError, ValueError):  # noqa: PERF203
+        except (KeyError, ValueError):
             continue
 
         if await lock_manager.lock(
@@ -56,7 +52,7 @@ async def remove_disconnected_user_resources(
             continue
 
         # (0) If key has no resources => remove from registry and continue
-        resources: ResourcesDict = await registry.get_resources(dead_session)
+        resources = await registry.get_resources(dead_session)
         if not resources:
             await registry.remove_key(dead_session)
             continue
@@ -81,7 +77,7 @@ async def remove_disconnected_user_resources(
             # In that case, the resource is released by THE LAST DYING KEY
             # (we could call this the "last-standing-man" pattern! :-) )
             #
-            other_sessions_with_this_resource: list[UserSessionDict] = [
+            other_sessions_with_this_resource = [
                 k
                 for k in await registry.find_keys((resource_name, f"{resource_value}"))
                 if k != dead_session
@@ -107,17 +103,15 @@ async def remove_disconnected_user_resources(
                     #
                     try:
                         _logger.info(
-                            "Closing services for project '%s'", resource_value
+                            "Closing project '%s' of user %s", resource_value, user_id
                         )
-                        await remove_project_dynamic_services(
-                            user_id=user_id,
-                            project_uuid=f"{resource_value}",
-                            app=app,
+                        await _projects_service.try_close_project_for_user(
+                            user_id,
+                            f"{resource_value}",
+                            dead_session["client_session_id"],
+                            app,
                             simcore_user_agent=UNDEFINED_DEFAULT_SIMCORE_USER_AGENT_VALUE,
-                            user_name={
-                                "first_name": "garbage",
-                                "last_name": "collector",
-                            },
+                            wait_for_service_closed=True,
                         )
 
                     except (ProjectNotFoundError, ProjectLockError) as err:
diff --git a/services/web/server/src/simcore_service_webserver/projects/_projects_service.py b/services/web/server/src/simcore_service_webserver/projects/_projects_service.py
@@ -1590,6 +1590,8 @@ async def try_close_project_for_user(
     client_session_id: str,
     app: web.Application,
     simcore_user_agent: str,
+    *,
+    wait_for_service_closed: bool = False,
 ):
     with managed_resource(user_id, client_session_id, app) as user_session:
         current_user_session = user_session.get_id()
@@ -1609,23 +1611,33 @@ async def try_close_project_for_user(
             return
 
         # remove the project from our list of opened ones
-        with log_context(
-            log, logging.DEBUG, f"removing {user_id=} session for {project_uuid=}"
-        ):
-            await user_session.remove(key=PROJECT_ID_KEY)
+        await user_session.remove(key=PROJECT_ID_KEY)
 
     # check it is not opened by someone else
     all_user_sessions_with_project.remove(current_user_session)
     log.debug("remaining user_to_session_ids: %s", all_user_sessions_with_project)
     if not all_user_sessions_with_project:
         # NOTE: depending on the garbage collector speed, it might already be removing it
-        fire_and_forget_task(
-            remove_project_dynamic_services(
-                user_id, project_uuid, app, simcore_user_agent
-            ),
-            task_suffix_name=f"remove_project_dynamic_services_{user_id=}_{project_uuid=}",
-            fire_and_forget_tasks_collection=app[APP_FIRE_AND_FORGET_TASKS_KEY],
+        remove_services_task = remove_project_dynamic_services(
+            user_id, project_uuid, app, simcore_user_agent
         )
+        if wait_for_service_closed:
+            # wait for the task to finish
+            await remove_services_task
+        else:
+            fire_and_forget_task(
+                remove_services_task,
+                task_suffix_name=f"remove_project_dynamic_services_{user_id=}_{project_uuid=}",
+                fire_and_forget_tasks_collection=app[APP_FIRE_AND_FORGET_TASKS_KEY],
+            )
+    # notify users that project is now closed
+    project = await get_project_for_user(
+        app,
+        project_uuid,
+        user_id,
+        include_state=True,
+    )
+    await notify_project_state_update(app, project)
 
 
 async def _get_project_share_state(
diff --git a/services/web/server/src/simcore_service_webserver/resource_manager/registry.py b/services/web/server/src/simcore_service_webserver/resource_manager/registry.py
@@ -14,6 +14,7 @@
 """
 
 import logging
+from typing import TypeAlias
 
 import redis.asyncio as aioredis
 from aiohttp import web
@@ -55,6 +56,10 @@ class ResourcesDict(TypedDict, total=False):
     socket_id: str
 
 
+AliveSessions: TypeAlias = list[UserSessionDict]
+DeadSessions: TypeAlias = list[UserSessionDict]
+
+
 class RedisResourceRegistry:
     """Keeps a record of connected sockets per user
 
@@ -158,9 +163,7 @@ async def remove_key(self, key: UserSessionDict) -> None:
             f"{self._hash_key(key)}:{_ALIVE_SUFFIX}",
         )
 
-    async def get_all_resource_keys(
-        self,
-    ) -> tuple[list[UserSessionDict], list[UserSessionDict]]:
+    async def get_all_resource_keys(self) -> tuple[AliveSessions, DeadSessions]:
         alive_keys = [
             self._decode_hash_key(hash_key)
             async for hash_key in self.client.scan_iter(match=f"*:{_ALIVE_SUFFIX}")
