fix test createing/deleteing nodes in high throuput conditions

Author: sanderegg

services/web/server/src/simcore_service_webserver/projects/_projects_repository_legacy.py

@@ -5,7 +5,6 @@
 
 """
 
-import datetime
 import logging
 from contextlib import AsyncExitStack
 from typing import Any, Self, cast
@@ -449,7 +448,6 @@ def _create_shared_workspace_query(
         is_search_by_multi_columns: bool,
         user_groups: list[GroupID],
     ) -> sql.Select | None:
-
         if workspace_query.workspace_scope is not WorkspaceScope.PRIVATE:
             assert workspace_query.workspace_scope in (  # nosec
                 WorkspaceScope.SHARED,
@@ -951,27 +949,32 @@ async def _update_project_workbench_with_lock_and_notify(
             thread-safe operations on the project document.
         """
 
+        # Get user's primary group ID for notification
+        async with self.engine.acquire() as conn:
+            user_primary_gid = await self._get_user_primary_group_gid(conn, user_id)
+
+        # 10 concurrent calls
         @exclusive(
             get_redis_lock_manager_client_sdk(self._app),
             lock_key=PROJECT_DB_UPDATE_REDIS_LOCK_KEY.format(project_uuid),
             blocking=True,
-            blocking_timeout=datetime.timedelta(seconds=30),
+            blocking_timeout=None,  # NOTE: this is a blocking call, a timeout has undefined effects
         )
         async def _update_workbench_and_notify() -> (
-            tuple[ProjectDict, dict[NodeIDStr, Any]]
+            tuple[ProjectDict, dict[NodeIDStr, Any], ProjectDocument, int]
         ):
-            # Update the workbench
+            """This function is protected because
+            - the project document and its version must be kept in sync
+            """
+            # Update the workbench work since it's atomic
             updated_project, changed_entries = await self._update_project_workbench(
                 partial_workbench_data,
                 user_id=user_id,
                 project_uuid=f"{project_uuid}",
                 product_name=product_name,
                 allow_workbench_changes=allow_workbench_changes,
             )
-
-            # Get user's primary group ID for notification
-            async with self.engine.acquire() as conn:
-                user_primary_gid = await self._get_user_primary_group_gid(conn, user_id)
+            # the update project with last_modified timestamp latest is the last
 
             # Get the full project with workbench for document creation
             project_with_workbench = (
@@ -1004,17 +1007,23 @@ async def _update_workbench_and_notify() -> (
             document_version = await increment_and_return_project_document_version(
                 redis_client=redis_client_sdk, project_uuid=project_uuid
             )
-            await notify_project_document_updated(
-                app=self._app,
-                project_id=project_uuid,
-                user_primary_gid=user_primary_gid,
-                version=document_version,
-                document=project_document,
-            )
 
-            return updated_project, changed_entries
-
-        return await _update_workbench_and_notify()
+            return updated_project, changed_entries, project_document, document_version
+
+        (
+            updated_project,
+            changed_entries,
+            project_document,
+            document_version,
+        ) = await _update_workbench_and_notify()
+        await notify_project_document_updated(
+            app=self._app,
+            project_id=project_uuid,
+            user_primary_gid=user_primary_gid,
+            version=document_version,
+            document=project_document,
+        )
+        return updated_project, changed_entries
 
     async def _update_project_workbench(
         self,

services/web/server/src/simcore_service_webserver/projects/_projects_service.py

@@ -207,9 +207,12 @@ async def patch_project_and_notify_users(
         get_redis_lock_manager_client_sdk(app),
         lock_key=PROJECT_DB_UPDATE_REDIS_LOCK_KEY.format(project_uuid),
         blocking=True,
-        blocking_timeout=datetime.timedelta(seconds=30),
+        blocking_timeout=None,  # NOTE: this is a blocking call, a timeout has undefined effects
     )
-    async def _patch_and_notify() -> None:
+    async def _patch_and_create_project_document() -> tuple[ProjectDocument, int]:
+        """This function is protected because
+        - the project document and its version must be kept in sync
+        """
         await _projects_repository.patch_project(
             app=app,
             project_uuid=project_uuid,
@@ -239,15 +242,16 @@ async def _patch_and_notify() -> None:
         document_version = await increment_and_return_project_document_version(
             redis_client=redis_client_sdk, project_uuid=project_uuid
         )
-        await notify_project_document_updated(
-            app=app,
-            project_id=project_uuid,
-            user_primary_gid=user_primary_gid,
-            version=document_version,
-            document=project_document,
-        )
+        return project_document, document_version
 
-    await _patch_and_notify()
+    project_document, document_version = await _patch_and_create_project_document()
+    await notify_project_document_updated(
+        app=app,
+        project_id=project_uuid,
+        user_primary_gid=user_primary_gid,
+        version=document_version,
+        document=project_document,
+    )
 
 
 def _is_node_dynamic(node_key: str) -> bool: