🐛Sticky connection: Ensure emitted socketio messages for logs, progress, status updates and payments are not lost (#7967)

Author: sanderegg

services/docker-compose.yml

@@ -888,10 +888,14 @@ services:
         - traefik.http.routers.${SWARM_STACK_NAME}_webserver_sticky.rule=Path(`/v0/projects`) ||
           Path(`/v0/projects:clone`) ||
           PathRegexp(`^/v0/projects/[0-9a-fA-F-]+/nodes/[0-9a-fA-F-]+:stop`) ||
+          PathRegexp(`^/v0/projects/[0-9a-fA-F-]+/nodes/[0-9a-fA-F-]+:open`) ||
+          PathRegexp(`^/v0/projects/[0-9a-fA-F-]+/nodes/[0-9a-fA-F-]+:close`) ||
           PathRegexp(`^/v0/storage/locations/[0-9]+/paths/.+:size`) ||
           PathRegexp(`^/v0/storage/locations/[0-9]+/-/paths:batchDelete`) ||
           PathRegexp(`^/v0/storage/locations/[0-9]+/export-data`) ||
           PathRegexp(`^/v0/tasks-legacy/.+`)
+        # NOTE: the sticky router must have a higher priority than the webserver router but below dy-proxies
+        - traefik.http.routers.${SWARM_STACK_NAME}_webserver_sticky.priority=8
         - traefik.http.routers.${SWARM_STACK_NAME}_webserver_sticky.entrypoints=http
         - traefik.http.routers.${SWARM_STACK_NAME}_webserver_sticky.service=${SWARM_STACK_NAME}_webserver_sticky
         - traefik.http.routers.${SWARM_STACK_NAME}_webserver_sticky.middlewares=${SWARM_STACK_NAME}_gzip@swarm, ${SWARM_STACK_NAME_NO_HYPHEN}_sslheader@swarm, ${SWARM_STACK_NAME}_webserver_retry

services/web/server/src/simcore_service_webserver/notifications/_rabbitmq_exclusive_queue_consumers.py

@@ -93,7 +93,6 @@ async def _progress_message_parser(app: web.Application, data: bytes) -> bool:
             app,
             rabbit_message.user_id,
             message=message,
-            ignore_queue=True,
         )
     return True
 
@@ -107,7 +106,6 @@ async def _log_message_parser(app: web.Application, data: bytes) -> bool:
             event_type=SOCKET_IO_LOG_EVENT,
             data=rabbit_message.model_dump(exclude={"user_id", "channel_name"}),
         ),
-        ignore_queue=True,
     )
     return True
 
@@ -124,7 +122,6 @@ async def _events_message_parser(app: web.Application, data: bytes) -> bool:
                 "node_id": f"{rabbit_message.node_id}",
             },
         ),
-        ignore_queue=True,
     )
     return True
 
@@ -178,9 +175,10 @@ async def _osparc_credits_message_parser(app: web.Application, data: bytes) -> b
 
 
 async def _unsubscribe_from_rabbitmq(app) -> None:
-    with log_context(
-        _logger, logging.INFO, msg="Unsubscribing from rabbitmq channels"
-    ), log_catch(_logger, reraise=False):
+    with (
+        log_context(_logger, logging.INFO, msg="Unsubscribing from rabbitmq channels"),
+        log_catch(_logger, reraise=False),
+    ):
         rabbit_client: RabbitMQClient = get_rabbitmq_client(app)
         await logged_gather(
             *(

services/web/server/src/simcore_service_webserver/payments/_socketio.py

@@ -29,7 +29,6 @@ async def notify_payment_completed(
             event_type=SOCKET_IO_PAYMENT_COMPLETED_EVENT,
             data=jsonable_encoder(payment, by_alias=True),
         ),
-        ignore_queue=True,
     )
 
 
@@ -46,5 +45,4 @@ async def notify_payment_method_acked(
             event_type=SOCKET_IO_PAYMENT_METHOD_ACKED_EVENT,
             data=jsonable_encoder(payment_method_transaction, by_alias=True),
         ),
-        ignore_queue=True,
     )

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

@@ -1961,7 +1961,6 @@ async def notify_project_state_update(
             app,
             user_id=notify_only_user,
             message=message,
-            ignore_queue=True,
         )
     else:
         rooms_to_notify: Generator[GroupID, None, None] = (

services/web/server/src/simcore_service_webserver/socketio/messages.py

@@ -60,12 +60,12 @@ async def send_message_to_user(
     user_id: UserID,
     message: SocketMessageDict,
     *,
-    ignore_queue: bool,
+    ignore_queue: bool = False,
 ) -> None:
     """
     Keyword Arguments:
-        ignore_queue -- set to False when this message is delivered from a server that has no direct connection to the client (default: {True})
-        An example where this is value is False, is sending messages to a user in the GC
+        ignore_queue -- set to True when this message is delivered from a server that has no direct connection to the user client (default: {False})
+        Be careful with this option, as it can lead to message loss if the user is not connected to this server!!
     """
     sio: AsyncServer = get_socket_server(app)
 

services/web/server/tests/integration/02/notifications/test_rabbitmq_consumers.py

@@ -290,7 +290,6 @@ async def test_log_workflow_only_receives_messages_if_subscribed(
                 "event_type": SOCKET_IO_LOG_EVENT,
                 "data": log_message.model_dump(exclude={"user_id", "channel_name"}),
             },
-            ignore_queue=True,
         ),
     )
     mocked_send_messages.reset_mock()