Merge branch 'is8102/add-search-api-in-storage' of github.com:giancarloromeo/osparc-simcore into is8102/add-search-api-in-storage

Author: giancarloromeo

packages/celery-library/src/celery_library/errors.py

@@ -26,5 +26,11 @@ def decode_celery_transferrable_error(error: TransferrableCeleryError) -> Except
     return result
 
 
+class TaskSubmissionError(OsparcErrorMixin, Exception):
+    msg_template = (
+        "Unable to submit task {task_name} with id '{task_id}' and params {task_params}"
+    )
+
+
 class TaskNotFoundError(OsparcErrorMixin, Exception):
     msg_template = "Task with id '{task_id}' was not found"

packages/celery-library/src/celery_library/task_manager.py

@@ -4,6 +4,7 @@
 from uuid import uuid4
 
 from celery import Celery  # type: ignore[import-untyped]
+from celery.exceptions import CeleryError  # type: ignore[import-untyped]
 from common_library.async_tools import make_async
 from models_library.progress_bar import ProgressReport
 from servicelib.celery.models import (
@@ -21,7 +22,7 @@
 from servicelib.logging_utils import log_context
 from settings_library.celery import CelerySettings
 
-from .errors import TaskNotFoundError
+from .errors import TaskNotFoundError, TaskSubmissionError
 
 _logger = logging.getLogger(__name__)
 
@@ -50,21 +51,38 @@ async def submit_task(
         ):
             task_uuid = uuid4()
             task_id = task_filter.create_task_id(task_uuid=task_uuid)
-            self._celery_app.send_task(
-                task_metadata.name,
-                task_id=task_id,
-                kwargs={"task_id": task_id} | task_params,
-                queue=task_metadata.queue.value,
-            )
 
             expiry = (
                 self._celery_settings.CELERY_EPHEMERAL_RESULT_EXPIRES
                 if task_metadata.ephemeral
                 else self._celery_settings.CELERY_RESULT_EXPIRES
             )
-            await self._task_info_store.create_task(
-                task_id, task_metadata, expiry=expiry
-            )
+
+            try:
+                await self._task_info_store.create_task(
+                    task_id, task_metadata, expiry=expiry
+                )
+                self._celery_app.send_task(
+                    task_metadata.name,
+                    task_id=task_id,
+                    kwargs={"task_id": task_id} | task_params,
+                    queue=task_metadata.queue.value,
+                )
+            except CeleryError as exc:
+                try:
+                    await self._task_info_store.remove_task(task_id)
+                except CeleryError:
+                    _logger.warning(
+                        "Unable to cleanup task '%s' during error handling",
+                        task_id,
+                        exc_info=True,
+                    )
+                raise TaskSubmissionError(
+                    task_name=task_metadata.name,
+                    task_id=task_id,
+                    task_params=task_params,
+                ) from exc
+
             return task_uuid
 
     async def cancel_task(self, task_filter: TaskFilter, task_uuid: TaskUUID) -> None:

packages/service-library/setup.cfg

@@ -21,6 +21,7 @@ markers =
 	testit: "marks test to run during development"
 	performance_test: "performance test"
 	no_cleanup_check_rabbitmq_server_has_no_errors: "no check in rabbitmq logs"
+	heavy_load: "marks test as heavy load"
 
 [mypy]
 plugins =

packages/service-library/setup.py

@@ -38,7 +38,7 @@ def read_reqs(reqs_path: Path) -> set[str]:
     "python_requires": "~=3.11",
     "install_requires": tuple(PROD_REQUIREMENTS),
     "packages": find_packages(where="src"),
-    "package_data": {"": ["py.typed"]},
+    "package_data": {"": ["py.typed", "redis/lua/*.lua"]},
     "package_dir": {"": "src"},
     "test_suite": "tests",
     "tests_require": tuple(TEST_REQUIREMENTS),

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

@@ -11,7 +11,6 @@
 from common_library.async_tools import cancel_wait_task
 from models_library.api_schemas_long_running_tasks.base import TaskProgress
 from pydantic import NonNegativeFloat, PositiveFloat
-from servicelib.utils import limited_gather
 from settings_library.redis import RedisDatabase, RedisSettings
 from tenacity import (
     AsyncRetrying,
@@ -24,6 +23,7 @@
 from ..logging_errors import create_troubleshootting_log_kwargs
 from ..logging_utils import log_catch, log_context
 from ..redis import RedisClientSDK, exclusive
+from ..utils import limited_gather
 from ._redis_store import RedisStore
 from ._serialization import dumps
 from .errors import (

packages/service-library/src/servicelib/redis/__init__.py

@@ -6,6 +6,8 @@
     CouldNotConnectToRedisError,
     LockLostError,
     ProjectLockError,
+    SemaphoreAcquisitionError,
+    SemaphoreNotAcquiredError,
 )
 from ._models import RedisManagerDBConfig
 from ._project_document_version import (
@@ -18,24 +20,26 @@
     is_project_locked,
     with_project_locked,
 )
+from ._semaphore_decorator import with_limited_concurrency
 from ._utils import handle_redis_returns_union_types
 
 __all__: tuple[str, ...] = (
+    "PROJECT_DB_UPDATE_REDIS_LOCK_KEY",
+    "PROJECT_DOCUMENT_VERSION_KEY",
     "CouldNotAcquireLockError",
     "CouldNotConnectToRedisError",
-    "exclusive",
-    "increment_and_return_project_document_version",
-    "get_project_locked_state",
-    "handle_redis_returns_union_types",
-    "is_project_locked",
     "LockLostError",
-    "PROJECT_DB_UPDATE_REDIS_LOCK_KEY",
-    "PROJECT_DOCUMENT_VERSION_KEY",
     "ProjectLockError",
     "RedisClientSDK",
     "RedisClientsManager",
     "RedisManagerDBConfig",
+    "SemaphoreAcquisitionError",
+    "SemaphoreNotAcquiredError",
+    "exclusive",
+    "get_project_locked_state",
+    "handle_redis_returns_union_types",
+    "increment_and_return_project_document_version",
+    "is_project_locked",
+    "with_limited_concurrency",
     "with_project_locked",
 )
-
-# nopycln: file

packages/service-library/src/servicelib/redis/_constants.py

@@ -6,6 +6,9 @@
 DEFAULT_LOCK_TTL: Final[datetime.timedelta] = datetime.timedelta(seconds=10)
 DEFAULT_SOCKET_TIMEOUT: Final[datetime.timedelta] = datetime.timedelta(seconds=30)
 
+DEFAULT_SEMAPHORE_TTL: Final[datetime.timedelta] = datetime.timedelta(seconds=10)
+SEMAPHORE_KEY_PREFIX: Final[str] = "semaphores:"
+SEMAPHORE_HOLDER_KEY_PREFIX: Final[str] = "semaphores:holders:"
 
 DEFAULT_DECODE_RESPONSES: Final[bool] = True
 DEFAULT_HEALTH_CHECK_INTERVAL: Final[datetime.timedelta] = datetime.timedelta(seconds=5)

packages/service-library/src/servicelib/redis/_errors.py

@@ -4,8 +4,7 @@
 from common_library.errors_classes import OsparcErrorMixin
 
 
-class BaseRedisError(OsparcErrorMixin, RuntimeError):
-    ...
+class BaseRedisError(OsparcErrorMixin, RuntimeError): ...
 
 
 class CouldNotAcquireLockError(BaseRedisError):
@@ -25,3 +24,15 @@ class LockLostError(BaseRedisError):
 
 
 ProjectLockError: TypeAlias = redis.exceptions.LockError  # NOTE: backwards compatible
+
+
+class SemaphoreAcquisitionError(BaseRedisError):
+    msg_template: str = "Could not acquire semaphore '{name}' (capacity: {capacity})"
+
+
+class SemaphoreNotAcquiredError(BaseRedisError):
+    msg_template: str = "Semaphore '{name}' was not acquired by this instance"
+
+
+class SemaphoreLostError(BaseRedisError):
+    msg_template: str = "Semaphore '{name}' was lost by this instance `{instance_id}`"