Bugfix/concurent opening projects (#1598)

adds usage of redis lock to prevent opening the same study at the same time

Author: sanderegg

services/web/client/source/class/osparc/component/message/FlashMessenger.js

@@ -89,7 +89,10 @@ qx.Class.define("osparc.component.message.FlashMessenger", {
      * @param {*} logMessage.logger IDK
      */
     log: function(logMessage) {
-      let message = ("message" in logMessage.message) ? logMessage.message["message"] : logMessage.message;
+      // TODO: This doesn't look cool
+      let message = osparc.utils.Utils.isObject(logMessage.message) && "message" in logMessage.message ?
+        logMessage.message.message :
+        logMessage.message;
       const level = logMessage.level.toUpperCase(); // "DEBUG", "INFO", "WARNING", "ERROR"
       let logger = logMessage.logger;
       if (logger) {

services/web/client/source/class/osparc/utils/Utils.js

@@ -368,6 +368,10 @@ qx.Class.define("osparc.utils.Utils", {
       const labels = [];
       args.forEach(arg => labels.push(qx.lang.String.firstUp(arg)));
       return labels.join(" ");
+    },
+
+    isObject: function(v) {
+      return typeof v === "object" && v !== null;
     }
   }
 });

services/web/server/requirements/_base.in

@@ -14,6 +14,7 @@ aiohttp_security
 aiohttp-swagger[performance]
 aiopg[sa]
 aioredis
+aioredlock
 aiosmtplib
 asyncpg
 celery

services/web/server/requirements/_base.txt

@@ -12,14 +12,15 @@ aiohttp-session[secure]==2.9.0  # via -r requirements/_base.in
 aiohttp-swagger[performance]==1.0.14  # via -r requirements/_base.in
 aiohttp==3.6.2            # via -r requirements/../../../../packages/service-library/requirements/_base.in, aiohttp-jinja2, aiohttp-security, aiohttp-session, aiohttp-swagger, aiozipkin
 aiopg[sa]==1.0.0          # via -r requirements/../../../../packages/service-library/requirements/_base.in, -r requirements/_base.in
-aioredis==1.3.1           # via -r requirements/_base.in
+aioredis==1.3.1           # via -r requirements/_base.in, aioredlock
+aioredlock==0.4.0         # via -r requirements/_base.in
 aiormq==3.2.2             # via aio-pika
 aiosmtplib==1.1.3         # via -r requirements/_base.in
 aiozipkin==0.6.0          # via -r requirements/../../../../packages/service-library/requirements/_base.in
 amqp==2.6.0               # via kombu
 async-timeout==3.0.1      # via aiohttp, aioredis
 asyncpg==0.20.1           # via -r requirements/_base.in
-attrs==19.3.0             # via -r requirements/../../../../packages/service-library/requirements/_base.in, aiohttp, jsonschema, openapi-core
+attrs==19.3.0             # via -r requirements/../../../../packages/service-library/requirements/_base.in, aiohttp, aioredlock, jsonschema, openapi-core
 billiard==3.6.3.0         # via celery
 celery==4.4.5             # via -r requirements/_base.in
 cffi==1.14.0              # via cryptography

services/web/server/requirements/_test.txt

@@ -12,15 +12,16 @@ aiohttp-session[secure]==2.9.0  # via -r requirements/_base.txt
 aiohttp-swagger[performance]==1.0.14  # via -r requirements/_base.txt
 aiohttp==3.6.2            # via -r requirements/_base.txt, aiohttp-jinja2, aiohttp-security, aiohttp-session, aiohttp-swagger, aiozipkin, pytest-aiohttp
 aiopg[sa]==1.0.0          # via -r requirements/_base.txt
-aioredis==1.3.1           # via -r requirements/_base.txt
+aioredis==1.3.1           # via -r requirements/_base.txt, aioredlock
+aioredlock==0.4.0         # via -r requirements/_base.txt
 aiormq==3.2.2             # via -r requirements/_base.txt, aio-pika
 aiosmtplib==1.1.3         # via -r requirements/_base.txt
 aiozipkin==0.6.0          # via -r requirements/_base.txt
 amqp==2.6.0               # via -r requirements/_base.txt, kombu
 astroid==2.4.2            # via pylint
 async-timeout==3.0.1      # via -r requirements/_base.txt, aiohttp, aioredis
 asyncpg==0.20.1           # via -r requirements/_base.txt
-attrs==19.3.0             # via -r requirements/_base.txt, aiohttp, jsonschema, openapi-core, pytest, pytest-docker
+attrs==19.3.0             # via -r requirements/_base.txt, aiohttp, aioredlock, jsonschema, openapi-core, pytest, pytest-docker
 billiard==3.6.3.0         # via -r requirements/_base.txt, celery
 celery==4.4.5             # via -r requirements/_base.txt
 certifi==2020.6.20        # via requests

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

@@ -3,8 +3,9 @@
 """
 import json
 import logging
-from typing import Set
+from typing import List, Optional, Set
 
+import aioredlock
 from aiohttp import web
 from jsonschema import ValidationError
 
@@ -163,14 +164,14 @@ async def get_project(request: web.Request):
         project = await get_project_for_user(
             request.app,
             project_uuid=project_uuid,
-            user_id=request[RQT_USERID_KEY],
+            user_id=user_id,
             include_templates=True,
         )
 
         return {"data": project}
     except ProjectInvalidRightsError:
         raise web.HTTPForbidden(
-            reason=f"User {user_id} has no right to read {project_uuid}"
+            reason=f"You do not have sufficient rights to read project {project_uuid}"
         )
     except ProjectNotFoundError:
         raise web.HTTPNotFound(reason=f"Project {project_uuid} not found")
@@ -235,7 +236,7 @@ async def replace_project(request: web.Request):
 
     except ProjectInvalidRightsError:
         raise web.HTTPForbidden(
-            reason=f"User {user_id} has no rights to write to project {project_uuid}"
+            reason="You do not have sufficient rights to save the project"
         )
     except ProjectNotFoundError:
         raise web.HTTPNotFound
@@ -256,19 +257,27 @@ async def delete_project(request: web.Request):
             user_id=user_id,
             include_templates=True,
         )
+        project_users: List[int] = []
         with managed_resource(user_id, None, request.app) as rt:
-            other_users = await rt.find_users_of_resource("project_id", project_uuid)
-            if other_users:
-                message = "Project is opened by another user. It cannot be deleted."
-                if user_id in other_users:
-                    message = "Project is still open. It cannot be deleted until it is closed."
-                # we cannot delete that project
-                raise web.HTTPForbidden(reason=message)
+            project_users = await rt.find_users_of_resource("project_id", project_uuid)
+        if project_users:
+            # that project is still in use
+            if user_id in project_users:
+                message = "Project is still open in another tab/browser. It cannot be deleted until it is closed."
+            else:
+                other_users = set(project_users)
+                other_user_names = {
+                    await get_user_name(request.app, x) for x in other_users
+                }
+                message = f"Project is open by {other_user_names}. It cannot be deleted until the project is closed."
+
+            # we cannot delete that project
+            raise web.HTTPForbidden(reason=message)
 
         await projects_api.delete_project(request, project_uuid, user_id)
     except ProjectInvalidRightsError:
         raise web.HTTPForbidden(
-            reason=f"User {user_id} has no rights to delete project"
+            reason="You do not have sufficient rights to delete this project"
         )
     except ProjectNotFoundError:
         raise web.HTTPNotFound(reason=f"Project {project_uuid} not found")
@@ -289,31 +298,41 @@ async def open_project(request: web.Request) -> web.Response:
     project_uuid = request.match_info.get("project_id")
     client_session_id = await request.json()
     try:
-        with managed_resource(user_id, client_session_id, request.app) as rt:
-            # TODO: temporary hidden until get_handlers_from_namespace refactor to seek marked functions instead!
-            from .projects_api import get_project_for_user
-
-            project = await get_project_for_user(
-                request.app,
-                project_uuid=project_uuid,
-                user_id=user_id,
-                include_templates=True,
-            )
+        # TODO: temporary hidden until get_handlers_from_namespace refactor to seek marked functions instead!
+        from .projects_api import get_project_for_user
 
-            # let's check if that project is already opened by someone else
-            other_users: Set[int] = {
-                x
-                for x in await rt.find_users_of_resource("project_id", project_uuid)
-                if x != f"{user_id}"
-            }
+        project = await get_project_for_user(
+            request.app,
+            project_uuid=project_uuid,
+            user_id=user_id,
+            include_templates=True,
+        )
 
-            if other_users:
-                # project is already locked
-                usernames = [
-                    await get_user_name(request.app, uid) for uid in other_users
-                ]
-                raise HTTPLocked(reason=f"Project is already opened by {usernames}")
-            await rt.add("project_id", project_uuid)
+        async def try_add_project() -> Optional[Set[int]]:
+            with managed_resource(user_id, client_session_id, request.app) as rt:
+                try:
+                    async with await rt.get_registry_lock():
+                        other_users: Set[int] = {
+                            x
+                            for x in await rt.find_users_of_resource(
+                                "project_id", project_uuid
+                            )
+                            if x != user_id
+                        }
+
+                        if other_users:
+                            return other_users
+                        await rt.add("project_id", project_uuid)
+                except aioredlock.LockError:
+                    # TODO: this lock is not a good solution for long term
+                    # maybe a project key in redis might improve spped of checking
+                    raise HTTPLocked(reason="Project is locked")
+
+        other_users = await try_add_project()
+        if other_users:
+            # project is already locked
+            usernames = [await get_user_name(request.app, uid) for uid in other_users]
+            raise HTTPLocked(reason=f"Project is already opened by {usernames}")
 
         # user id opened project uuid
         await projects_api.start_project_interactive_services(request, project, user_id)
@@ -381,18 +400,14 @@ async def _close_project_task() -> None:
 async def state_project(request: web.Request) -> web.Response:
     user_id = request[RQT_USERID_KEY]
     project_uuid = request.match_info.get("project_id")
-    with managed_resource(user_id, None, request.app) as rt:
-        # TODO: temporary hidden until get_handlers_from_namespace refactor to seek marked functions instead!
-        from .projects_api import get_project_for_user
-
-        # check that project exists
-        await get_project_for_user(
-            request.app,
-            project_uuid=project_uuid,
-            user_id=user_id,
-            include_templates=True,
-        )
+    # TODO: temporary hidden until get_handlers_from_namespace refactor to seek marked functions instead!
+    from .projects_api import get_project_for_user
 
+    # check that project exists
+    await get_project_for_user(
+        request.app, project_uuid=project_uuid, user_id=user_id, include_templates=True,
+    )
+    with managed_resource(user_id, None, request.app) as rt:
         users_of_project = await rt.find_users_of_resource("project_id", project_uuid)
         usernames = [
             await get_user_name(request.app, uid) for uid in set(users_of_project)
@@ -416,19 +431,20 @@ async def get_active_project(request: web.Request) -> web.Response:
 
     try:
         project = None
+        user_active_projects = []
         with managed_resource(user_id, client_session_id, request.app) as rt:
             # get user's projects
-            list_project_ids = await rt.find("project_id")
-            if list_project_ids:
-                # TODO: temporary hidden until get_handlers_from_namespace refactor to seek marked functions instead!
-                from .projects_api import get_project_for_user
-
-                project = await get_project_for_user(
-                    request.app,
-                    project_uuid=list_project_ids[0],
-                    user_id=user_id,
-                    include_templates=True,
-                )
+            user_active_projects = await rt.find("project_id")
+        if user_active_projects:
+            # TODO: temporary hidden until get_handlers_from_namespace refactor to seek marked functions instead!
+            from .projects_api import get_project_for_user
+
+            project = await get_project_for_user(
+                request.app,
+                project_uuid=user_active_projects[0],
+                user_id=user_id,
+                include_templates=True,
+            )
 
         return web.json_response({"data": project})
     except ProjectNotFoundError:

services/web/server/src/simcore_service_webserver/resource_manager/config.py

@@ -10,6 +10,7 @@
 
 CONFIG_SECTION_NAME = "resource_manager"
 APP_CLIENT_REDIS_CLIENT_KEY = __name__ + ".resource_manager.redis_client"
+APP_CLIENT_REDIS_LOCK_KEY = __name__ + ".resource_manager.redis_lock"
 APP_CLIENT_SOCKET_REGISTRY_KEY = __name__ + ".resource_manager.registry"
 APP_RESOURCE_MANAGER_TASKS_KEY = __name__ + ".resource_manager.tasks.key"
 APP_GARBAGE_COLLECTOR_KEY = __name__ + ".resource_manager.garbage_collector_key"

services/web/server/src/simcore_service_webserver/resource_manager/redis.py

@@ -4,9 +4,14 @@
 from aiohttp import web
 from tenacity import Retrying, before_log, stop_after_attempt, wait_random
 
+from aioredlock import Aioredlock
 from servicelib.application_keys import APP_CONFIG_KEY
 
-from .config import APP_CLIENT_REDIS_CLIENT_KEY, CONFIG_SECTION_NAME
+from .config import (
+    APP_CLIENT_REDIS_CLIENT_KEY,
+    APP_CLIENT_REDIS_LOCK_KEY,
+    CONFIG_SECTION_NAME,
+)
 
 log = logging.getLogger(__name__)
 
@@ -24,20 +29,32 @@ async def redis_client(app: web.Application):
     cfg = app[APP_CONFIG_KEY][CONFIG_SECTION_NAME]
     url = DSN.format(**cfg["redis"])
 
+    # create redis client
+    client = None
     for attempt in Retrying(**retry_upon_init_policy):
         with attempt:
             client = await aioredis.create_redis_pool(url, encoding="utf-8")
+    # create lock manager
+    lock_manager = Aioredlock([url])
 
     assert client  # nosec
     app[APP_CLIENT_REDIS_CLIENT_KEY] = client
 
+    assert lock_manager  # nosec
+    app[APP_CLIENT_REDIS_LOCK_KEY] = lock_manager
+
     yield
 
     if client is not app[APP_CLIENT_REDIS_CLIENT_KEY]:
         log.critical("Invalid redis client in app")
+    if lock_manager is not app[APP_CLIENT_REDIS_LOCK_KEY]:
+        log.critical("Invalid redis lock manager in app")
 
+    # close client
     client.close()
     await client.wait_closed()
+    # delete lock manager
+    await lock_manager.destroy()
 
 
 def setup_redis_client(app: web.Application):
@@ -56,3 +73,7 @@ def setup_redis_client(app: web.Application):
 
 def get_redis_client(app: web.Application) -> aioredis.Redis:
     return app[APP_CLIENT_REDIS_CLIENT_KEY]
+
+
+def get_redis_lock(app: web.Application) -> Aioredlock:
+    return app[APP_CLIENT_REDIS_LOCK_KEY]