Merge branch 'master' into 8506-batch-db-requests-in-map-endpoint

Author: bisgaard-itis

packages/pytest-simcore/src/pytest_simcore/helpers/playwright.py

@@ -143,6 +143,7 @@ def to_json(self) -> str:
 
 
 SOCKETIO_MESSAGE_PREFIX: Final[str] = "42"
+_WEBSOCKET_MESSAGE_PREFIX: Final[str] = "📡OSPARC-WEBSOCKET: "
 
 
 @dataclass
@@ -165,20 +166,31 @@ def _configure_websocket_events(self) -> None:
         ) as ctx:
 
             def on_framesent(payload: str | bytes) -> None:
-                ctx.logger.debug("⬇️ Frame sent: %s", payload)
+                ctx.logger.debug(
+                    "%s⬇️ Frame sent: %s", _WEBSOCKET_MESSAGE_PREFIX, payload
+                )
 
             def on_framereceived(payload: str | bytes) -> None:
-                ctx.logger.debug("⬆️ Frame received: %s", payload)
+                ctx.logger.debug(
+                    "%s⬆️ Frame received: %s", _WEBSOCKET_MESSAGE_PREFIX, payload
+                )
 
             def on_close(_: WebSocket) -> None:
                 if self.auto_reconnect:
-                    ctx.logger.warning("⚠️ WebSocket closed. Attempting to reconnect...")
+                    ctx.logger.warning(
+                        "%s⚠️ WebSocket closed. Attempting to reconnect...",
+                        _WEBSOCKET_MESSAGE_PREFIX,
+                    )
                     self._attempt_reconnect(ctx.logger)
                 else:
-                    ctx.logger.warning("⚠️ WebSocket closed.")
+                    ctx.logger.warning(
+                        "%s⚠️ WebSocket closed.", _WEBSOCKET_MESSAGE_PREFIX
+                    )
 
             def on_socketerror(error_msg: str) -> None:
-                ctx.logger.error("❌ WebSocket error: %s", error_msg)
+                ctx.logger.error(
+                    "%s❌ WebSocket error: %s", _WEBSOCKET_MESSAGE_PREFIX, error_msg
+                )
 
             # Attach core event listeners
             self.ws.on("framesent", on_framesent)

packages/pytest-simcore/src/pytest_simcore/helpers/playwright_sim4life.py

@@ -98,6 +98,9 @@ class WaitForS4LDict(TypedDict):
     iframe: FrameLocator
 
 
+_WEBSOCKET_MESSAGE_S4L_PREFIX: Final[str] = "📡S4L-WEBSOCKET: "
+
+
 def wait_for_launched_s4l(
     page: Page,
     node_id,
@@ -137,6 +140,31 @@ def wait_for_launched_s4l(
             )
         s4l_websocket = ws_info.value
         ctx.logger.info("acquired S4L websocket!")
+
+        def on_framesent(payload: str | bytes) -> None:
+            ctx.logger.debug(
+                "%s⬇️ Frame sent: %s", _WEBSOCKET_MESSAGE_S4L_PREFIX, payload
+            )
+
+        def on_framereceived(payload: str | bytes) -> None:
+            ctx.logger.debug(
+                "%s⬆️ Frame received: %s", _WEBSOCKET_MESSAGE_S4L_PREFIX, payload
+            )
+
+        def on_close(_: WebSocket) -> None:
+            ctx.logger.warning("%s⚠️ WebSocket closed.", _WEBSOCKET_MESSAGE_S4L_PREFIX)
+
+        def on_socketerror(error_msg: str) -> None:
+            ctx.logger.error(
+                "%s❌ WebSocket error: %s", _WEBSOCKET_MESSAGE_S4L_PREFIX, error_msg
+            )
+
+        # Attach core event listeners
+        s4l_websocket.on("framesent", on_framesent)
+        s4l_websocket.on("framereceived", on_framereceived)
+        s4l_websocket.on("close", on_close)
+        s4l_websocket.on("socketerror", on_socketerror)
+
         return {
             "websocket": s4l_websocket,
             "iframe": s4l_iframe,

packages/service-library/src/servicelib/long_running_tasks/models.py

@@ -71,6 +71,16 @@ class TaskData(BaseModel):
         ),
     ] = None
 
+    detected_as_done_at: Annotated[
+        datetime | None,
+        Field(
+            description=(
+                "used to remove the task when it's first detected as done "
+                "if a task was started as fire_and_forget=True"
+            )
+        ),
+    ] = None
+
     is_done: Annotated[
         bool,
         Field(description="True when the task finished running with or without errors"),

packages/service-library/src/servicelib/long_running_tasks/task.py

@@ -113,7 +113,33 @@ async def _get_tasks_to_remove(
 
     for tracked_task in await tracked_tasks.list_tasks_data():
         if tracked_task.fire_and_forget:
-            continue
+            # fire and forget tasks also need to be remove from tracking
+            # when detectes as done, start counting how much time has elapsed
+            # if over stale_task_detect_timeout_s remove the task
+
+            # wait for task to complete
+            if not tracked_task.is_done:
+                continue
+
+            # mark detected as done
+            if tracked_task.detected_as_done_at is None:
+                await tracked_tasks.update_task_data(
+                    tracked_task.task_id,
+                    updates={
+                        "detected_as_done_at": datetime.datetime.now(tz=datetime.UTC)
+                    },
+                )
+                continue
+
+            # if enough time passes remove the task
+            elapsed_since_done = (
+                utc_now - tracked_task.detected_as_done_at
+            ).total_seconds()
+            if elapsed_since_done > stale_task_detect_timeout_s:
+                tasks_to_remove.append(
+                    (tracked_task.task_id, tracked_task.task_context)
+                )
+                continue
 
         if tracked_task.last_status_check is None:
             # the task just added or never received a poll request
@@ -317,6 +343,7 @@ async def _tasks_monitor(self) -> None:  # noqa: C901
         """
         self._started_event_task_tasks_monitor.set()
         task_id: TaskId
+
         for task_id in set(self._created_tasks.keys()):
             if task := self._created_tasks.get(task_id, None):
                 is_done = task.done()

packages/service-library/src/servicelib/rabbitmq/_utils.py

@@ -76,7 +76,7 @@ async def declare_queue(
     if arguments is not None:
         default_arguments.update(arguments)
     queue_parameters: dict[str, Any] = {
-        "durable": True,
+        "durable": not exclusive_queue,
         "exclusive": exclusive_queue,
         "arguments": default_arguments,
         "name": f"{get_rabbitmq_client_unique_name(client_name)}_{queue_name}_exclusive",
@@ -85,19 +85,6 @@ async def declare_queue(
         # NOTE: setting a name will ensure multiple instance will take their data here
         queue_parameters |= {"name": queue_name}
 
-    # avoids deprecated `transient_nonexcl_queues` warning in RabbitMQ
-    if (
-        queue_parameters.get("durable", False) is False
-        and queue_parameters.get("exclusive", False) is False
-    ):
-        msg = (
-            "Queue must be `durable` or `exclusive`, but not both. "
-            "This is to avoid the `transient_nonexcl_queues` warning. "
-            "NOTE: if both `durable` and `exclusive` are missing they are considered False. "
-            f"{queue_parameters=}"
-        )
-        raise ValueError(msg)
-
     # NOTE: if below line raises something similar to ``ChannelPreconditionFailed: PRECONDITION_FAILED``
     # most likely someone changed the signature of the queues (parameters etc...)
     # Safest way to deal with it:

packages/service-library/tests/long_running_tasks/test_long_running_tasks_task.py

@@ -199,7 +199,7 @@ def _get_resutlt(result_field: ResultField) -> Any:
     return loads(result_field.str_result)
 
 
-async def test_fire_and_forget_task_is_not_auto_removed(
+async def test_fire_and_forget_task_is_not_auto_removed_while_running(
     long_running_manager: LongRunningManager, empty_context: TaskContext
 ):
     task_id = await lrt_api.start_task(
@@ -211,19 +211,32 @@ async def test_fire_and_forget_task_is_not_auto_removed(
         fire_and_forget=True,
         task_context=empty_context,
     )
+
     await asyncio.sleep(3 * TEST_CHECK_STALE_INTERVAL_S)
     # the task shall still be present even if we did not check the status before
     status = await long_running_manager.tasks_manager.get_task_status(
         task_id, with_task_context=empty_context
     )
     assert not status.done, "task was removed although it is fire and forget"
-    # the task shall finish
-    await asyncio.sleep(4 * TEST_CHECK_STALE_INTERVAL_S)
-    # get the result
-    task_result = await long_running_manager.tasks_manager.get_task_result(
-        task_id, with_task_context=empty_context
-    )
-    assert _get_resutlt(task_result) == 42
+
+    async for attempt in AsyncRetrying(**_RETRY_PARAMS):
+        with attempt:
+            try:
+                await long_running_manager.tasks_manager.get_task_status(
+                    task_id, with_task_context=empty_context
+                )
+                raise TryAgain
+            except TaskNotFoundError:
+                pass
+
+    with pytest.raises(TaskNotFoundError):
+        await long_running_manager.tasks_manager.get_task_status(
+            task_id, with_task_context=empty_context
+        )
+    with pytest.raises(TaskNotFoundError):
+        await long_running_manager.tasks_manager.get_task_result(
+            task_id, with_task_context=empty_context
+        )
 
 
 async def test_get_result_of_unfinished_task_raises(

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/scheduler/_core/_events_utils.py

@@ -13,6 +13,7 @@
 from models_library.rabbitmq_messages import InstrumentationRabbitMessage
 from models_library.rpc.webserver.auth.api_keys import generate_unique_api_key
 from models_library.service_settings_labels import SimcoreServiceLabels
+from models_library.services_types import ServiceRunID
 from models_library.shared_user_preferences import (
     AllowMetricsCollectionFrontendUserPreference,
 )
@@ -287,17 +288,19 @@ async def service_remove_sidecar_proxy_docker_networks_and_volumes(
         scheduler_data.node_uuid
     )
 
-    await _cleanup_long_running_tasks(app, scheduler_data.node_uuid)
+    await _cleanup_long_running_tasks(app, scheduler_data.run_id)
 
     await task_progress.update(
         message="finished removing resources", percent=ProgressPercent(1)
     )
 
 
-async def _cleanup_long_running_tasks(app: FastAPI, node_id: NodeID) -> None:
+async def _cleanup_long_running_tasks(
+    app: FastAPI, service_run_id: ServiceRunID
+) -> None:
     long_running_client_helper = get_long_running_client_helper(app)
 
-    sidecar_namespace = f"SIMCORE-SERVICE-DYNAMIC-SIDECAR-{node_id}"
+    sidecar_namespace = f"SIMCORE-SERVICE-DYNAMIC-SIDECAR-{service_run_id}"
     await long_running_client_helper.cleanup(sidecar_namespace)
 
 

services/static-webserver/client/source/class/osparc/store/Groups.js

@@ -159,7 +159,8 @@ qx.Class.define("osparc.store.Groups", {
               this.__addMemberToCache(orgMember, groupId);
             });
           }
-        });
+        })
+        .catch(err => osparc.FlashMessenger.logError(err));
     },
 
     fetchGroupsAndMembers: function() {

services/static-webserver/client/source/class/osparc/store/Users.js

@@ -36,10 +36,10 @@ qx.Class.define("osparc.store.Users", {
   members: {
     __unknowns: null,
 
-    __fetchUser: function(groupId) {
+    __fetchUser: function(userGroupId) {
       const params = {
         url: {
-          gid: groupId
+          gid: userGroupId
         }
       };
       return osparc.data.Resources.fetch("users", "get", params)
@@ -48,22 +48,22 @@ qx.Class.define("osparc.store.Users", {
           return user;
         })
         .catch(() => {
-          this.__unknowns.push(groupId);
+          this.__unknowns.push(userGroupId);
           return null;
         });
     },
 
-    getUser: async function(groupId, fetchIfNotFound = true) {
-      if (this.__unknowns.includes(groupId)) {
+    getUser: async function(userGroupId, fetchIfNotFound = true) {
+      if (this.__unknowns.includes(userGroupId)) {
         return null;
       }
-      const userFound = this.getUsers().find(user => user.getGroupId() === groupId);
+      const userFound = this.getUsers().find(user => user.getGroupId() === userGroupId);
       if (userFound) {
         return userFound;
       }
       if (fetchIfNotFound) {
         try {
-          const user = await this.__fetchUser(groupId);
+          const user = await this.__fetchUser(userGroupId);
           if (user) {
             return user;
           }