Merge remote-tracking branch 'upstream/master' into pr-osparc-migrate-dy-scheduler-part2

Author: Andrei Neagu

.github/workflows/ci-pact-master.yml

@@ -22,7 +22,7 @@ jobs:
       PACT_BROKER_PASSWORD: ${{ secrets.PACT_BROKER_PASSWORD }}
     steps:
       - name: setup python environment
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v6
         with:
           python-version: "3.11"
       - name: install uv

.github/workflows/ci-testing-deploy.yml

@@ -861,7 +861,7 @@ jobs:
           with:
             driver: docker-container
         - name: setup python environment
-          uses: actions/setup-python@v5
+          uses: actions/setup-python@v6
           with:
             python-version: ${{ matrix.python }}
         - name: install uv
@@ -1851,7 +1851,7 @@ jobs:
         with:
           python-version: ${{ matrix.python }}
           cache-dependency-glob: "**/e2e/requirements/requirements.txt"
-      - uses: actions/setup-node@v4.4.0
+      - uses: actions/setup-node@v5.0.0
         with:
           node-version: ${{ matrix.node }}
           cache: "npm"

.github/workflows/ci-testing-pull-request.yml

@@ -35,7 +35,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: setup python environment
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v6
         with:
           python-version: "3.11"
       - name: install uv
@@ -65,7 +65,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: setup python environment
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v6
         with:
           python-version: "3.11"
       - name: checkout
@@ -93,7 +93,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: setup python environment
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v6
         with:
           python-version: "3.11"
       - name: checkout

api/specs/web-server/_functions.py

@@ -23,6 +23,7 @@
     FunctionGroupPathParams,
 )
 from simcore_service_webserver.functions._controller._functions_rest_schemas import (
+    FunctionDeleteQueryParams,
     FunctionGetQueryParams,
     FunctionPathParams,
     FunctionsListQueryParams,
@@ -80,6 +81,7 @@ async def update_function(
 )
 async def delete_function(
     _path: Annotated[FunctionPathParams, Depends()],
+    _query: Annotated[as_query(FunctionDeleteQueryParams), Depends()],
 ): ...
 
 

packages/aws-library/requirements/_base.txt

@@ -118,6 +118,10 @@ jmespath==1.0.1
     #   aiobotocore
     #   boto3
     #   botocore
+jsonref==1.1.0
+    # via
+    #   -r requirements/../../../packages/models-library/requirements/_base.in
+    #   -r requirements/../../../packages/service-library/requirements/../../../packages/models-library/requirements/_base.in
 jsonschema==4.23.0
     # via
     #   -r requirements/../../../packages/models-library/requirements/_base.in

packages/celery-library/requirements/_base.txt

@@ -109,6 +109,10 @@ idna==3.10
     #   yarl
 importlib-metadata==8.6.1
     # via opentelemetry-api
+jsonref==1.1.0
+    # via
+    #   -r requirements/../../../packages/models-library/requirements/_base.in
+    #   -r requirements/../../../packages/service-library/requirements/../../../packages/models-library/requirements/_base.in
 jsonschema==4.23.0
     # via
     #   -r requirements/../../../packages/models-library/requirements/_base.in

packages/celery-library/src/celery_library/backends/_redis.py

@@ -19,7 +19,7 @@
 _CELERY_TASK_INFO_PREFIX: Final[str] = "celery-task-info-"
 _CELERY_TASK_ID_KEY_ENCODING = "utf-8"
 _CELERY_TASK_ID_KEY_SEPARATOR: Final[str] = ":"
-_CELERY_TASK_SCAN_COUNT_PER_BATCH: Final[int] = 10000
+_CELERY_TASK_SCAN_COUNT_PER_BATCH: Final[int] = 1000
 _CELERY_TASK_METADATA_KEY: Final[str] = "metadata"
 _CELERY_TASK_PROGRESS_KEY: Final[str] = "progress"
 
@@ -51,11 +51,6 @@ async def create_task(
             expiry,
         )
 
-    async def exists_task(self, task_id: TaskID) -> bool:
-        n = await self._redis_client_sdk.redis.exists(_build_key(task_id))
-        assert isinstance(n, int)  # nosec
-        return n > 0
-
     async def get_task_metadata(self, task_id: TaskID) -> TaskMetadata | None:
         raw_result = await self._redis_client_sdk.redis.hget(_build_key(task_id), _CELERY_TASK_METADATA_KEY)  # type: ignore
         if not raw_result:
@@ -131,3 +126,8 @@ async def set_task_progress(self, task_id: TaskID, report: ProgressReport) -> No
             key=_CELERY_TASK_PROGRESS_KEY,
             value=report.model_dump_json(),
         )  # type: ignore
+
+    async def task_exists(self, task_id: TaskID) -> bool:
+        n = await self._redis_client_sdk.redis.exists(_build_key(task_id))
+        assert isinstance(n, int)  # nosec
+        return n > 0

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

@@ -1,6 +1,8 @@
 import base64
 import pickle
 
+from common_library.errors_classes import OsparcErrorMixin
+
 
 class TransferrableCeleryError(Exception):
     def __repr__(self) -> str:
@@ -22,3 +24,7 @@ def decode_celery_transferrable_error(error: TransferrableCeleryError) -> Except
     assert isinstance(error, TransferrableCeleryError)  # nosec
     result: Exception = pickle.loads(base64.b64decode(error.args[0]))  # noqa: S301
     return result
+
+
+class TaskNotFoundError(OsparcErrorMixin, Exception):
+    msg_template = "Task with id '{task_id}' was not found"

packages/celery-library/src/celery_library/rpc/_async_jobs.py

@@ -13,6 +13,7 @@
 from models_library.api_schemas_rpc_async_jobs.exceptions import (
     JobAbortedError,
     JobError,
+    JobMissingError,
     JobNotDoneError,
     JobSchedulerError,
 )
@@ -22,6 +23,7 @@
 from servicelib.rabbitmq import RPCRouter
 
 from ..errors import (
+    TaskNotFoundError,
     TransferrableCeleryError,
     decode_celery_transferrable_error,
 )
@@ -30,7 +32,7 @@
 router = RPCRouter()
 
 
-@router.expose(reraise_if_error_type=(JobSchedulerError,))
+@router.expose(reraise_if_error_type=(JobSchedulerError, JobMissingError))
 async def cancel(
     task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
 ):
@@ -42,11 +44,13 @@ async def cancel(
             task_filter=task_filter,
             task_uuid=job_id,
         )
+    except TaskNotFoundError as exc:
+        raise JobMissingError(job_id=job_id) from exc
     except CeleryError as exc:
         raise JobSchedulerError(exc=f"{exc}") from exc
 
 
-@router.expose(reraise_if_error_type=(JobSchedulerError,))
+@router.expose(reraise_if_error_type=(JobSchedulerError, JobMissingError))
 async def status(
     task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
 ) -> AsyncJobStatus:
@@ -59,6 +63,8 @@ async def status(
             task_filter=task_filter,
             task_uuid=job_id,
         )
+    except TaskNotFoundError as exc:
+        raise JobMissingError(job_id=job_id) from exc
     except CeleryError as exc:
         raise JobSchedulerError(exc=f"{exc}") from exc
 
@@ -71,9 +77,10 @@ async def status(
 
 @router.expose(
     reraise_if_error_type=(
+        JobAbortedError,
         JobError,
+        JobMissingError,
         JobNotDoneError,
-        JobAbortedError,
         JobSchedulerError,
     )
 )
@@ -97,11 +104,11 @@ async def result(
             task_filter=task_filter,
             task_uuid=job_id,
         )
+    except TaskNotFoundError as exc:
+        raise JobMissingError(job_id=job_id) from exc
     except CeleryError as exc:
         raise JobSchedulerError(exc=f"{exc}") from exc
 
-    if _status.task_state == TaskState.ABORTED:
-        raise JobAbortedError(job_id=job_id)
     if _status.task_state == TaskState.FAILURE:
         # fallback exception to report
         exc_type = type(_result).__name__

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

@@ -6,11 +6,7 @@
 from functools import wraps
 from typing import Any, Concatenate, Final, ParamSpec, TypeVar, overload
 
-from celery import Celery  # type: ignore[import-untyped]
-from celery.contrib.abortable import (  # type: ignore[import-untyped]
-    AbortableAsyncResult,
-    AbortableTask,
-)
+from celery import Celery, Task  # type: ignore[import-untyped]
 from celery.exceptions import Ignore  # type: ignore[import-untyped]
 from common_library.async_tools import cancel_wait_task
 from pydantic import NonNegativeInt
@@ -39,42 +35,42 @@ class TaskAbortedError(Exception): ...
 def _async_task_wrapper(
     app: Celery,
 ) -> Callable[
-    [Callable[Concatenate[AbortableTask, P], Coroutine[Any, Any, R]]],
-    Callable[Concatenate[AbortableTask, P], R],
+    [Callable[Concatenate[Task, P], Coroutine[Any, Any, R]]],
+    Callable[Concatenate[Task, P], R],
 ]:
     def decorator(
-        coro: Callable[Concatenate[AbortableTask, P], Coroutine[Any, Any, R]],
-    ) -> Callable[Concatenate[AbortableTask, P], R]:
+        coro: Callable[Concatenate[Task, P], Coroutine[Any, Any, R]],
+    ) -> Callable[Concatenate[Task, P], R]:
         @wraps(coro)
-        def wrapper(task: AbortableTask, *args: P.args, **kwargs: P.kwargs) -> R:
+        def wrapper(task: Task, *args: P.args, **kwargs: P.kwargs) -> R:
             app_server = get_app_server(app)
             # NOTE: task.request is a thread local object, so we need to pass the id explicitly
             assert task.request.id is not None  # nosec
 
-            async def run_task(task_id: TaskID) -> R:
+            async def _run_task(task_id: TaskID) -> R:
                 try:
                     async with asyncio.TaskGroup() as tg:
-                        main_task = tg.create_task(
+                        async_io_task = tg.create_task(
                             coro(task, *args, **kwargs),
                         )
 
-                        async def abort_monitor():
-                            abortable_result = AbortableAsyncResult(task_id, app=app)
-                            while not main_task.done():
-                                if abortable_result.is_aborted():
+                        async def _abort_monitor():
+                            while not async_io_task.done():
+                                if not await app_server.task_manager.task_exists(
+                                    task_id
+                                ):
                                     await cancel_wait_task(
-                                        main_task,
+                                        async_io_task,
                                         max_delay=_DEFAULT_CANCEL_TASK_TIMEOUT.total_seconds(),
                                     )
-                                    AbortableAsyncResult(task_id, app=app).forget()
                                     raise TaskAbortedError
                                 await asyncio.sleep(
                                     _DEFAULT_ABORT_TASK_TIMEOUT.total_seconds()
                                 )
 
-                        tg.create_task(abort_monitor())
+                        tg.create_task(_abort_monitor())
 
-                    return main_task.result()
+                    return async_io_task.result()
                 except BaseExceptionGroup as eg:
                     task_aborted_errors, other_errors = eg.split(TaskAbortedError)
 
@@ -88,7 +84,7 @@ async def abort_monitor():
                     raise other_errors.exceptions[0] from eg
 
             return asyncio.run_coroutine_threadsafe(
-                run_task(task.request.id),
+                _run_task(task.request.id),
                 app_server.event_loop,
             ).result()
 
@@ -102,14 +98,14 @@ def _error_handling(
     delay_between_retries: timedelta,
     dont_autoretry_for: tuple[type[Exception], ...],
 ) -> Callable[
-    [Callable[Concatenate[AbortableTask, P], R]],
-    Callable[Concatenate[AbortableTask, P], R],
+    [Callable[Concatenate[Task, P], R]],
+    Callable[Concatenate[Task, P], R],
 ]:
     def decorator(
-        func: Callable[Concatenate[AbortableTask, P], R],
-    ) -> Callable[Concatenate[AbortableTask, P], R]:
+        func: Callable[Concatenate[Task, P], R],
+    ) -> Callable[Concatenate[Task, P], R]:
         @wraps(func)
-        def wrapper(task: AbortableTask, *args: P.args, **kwargs: P.kwargs) -> R:
+        def wrapper(task: Task, *args: P.args, **kwargs: P.kwargs) -> R:
             try:
                 return func(task, *args, **kwargs)
             except TaskAbortedError as exc:
@@ -144,7 +140,7 @@ def wrapper(task: AbortableTask, *args: P.args, **kwargs: P.kwargs) -> R:
 @overload
 def register_task(
     app: Celery,
-    fn: Callable[Concatenate[AbortableTask, TaskID, P], Coroutine[Any, Any, R]],
+    fn: Callable[Concatenate[Task, TaskID, P], Coroutine[Any, Any, R]],
     task_name: str | None = None,
     timeout: timedelta | None = _DEFAULT_TASK_TIMEOUT,
     max_retries: NonNegativeInt = _DEFAULT_MAX_RETRIES,
@@ -156,7 +152,7 @@ def register_task(
 @overload
 def register_task(
     app: Celery,
-    fn: Callable[Concatenate[AbortableTask, P], R],
+    fn: Callable[Concatenate[Task, P], R],
     task_name: str | None = None,
     timeout: timedelta | None = _DEFAULT_TASK_TIMEOUT,
     max_retries: NonNegativeInt = _DEFAULT_MAX_RETRIES,
@@ -168,8 +164,8 @@ def register_task(
 def register_task(  # type: ignore[misc]
     app: Celery,
     fn: (
-        Callable[Concatenate[AbortableTask, TaskID, P], Coroutine[Any, Any, R]]
-        | Callable[Concatenate[AbortableTask, P], R]
+        Callable[Concatenate[Task, TaskID, P], Coroutine[Any, Any, R]]
+        | Callable[Concatenate[Task, P], R]
     ),
     task_name: str | None = None,
     timeout: timedelta | None = _DEFAULT_TASK_TIMEOUT,
@@ -186,7 +182,7 @@ def register_task(  # type: ignore[misc]
         delay_between_retries -- dealy between each attempt in case of error (default: {_DEFAULT_WAIT_BEFORE_RETRY})
         dont_autoretry_for -- exceptions that should not be retried when raised by the task
     """
-    wrapped_fn: Callable[Concatenate[AbortableTask, P], R]
+    wrapped_fn: Callable[Concatenate[Task, P], R]
     if asyncio.iscoroutinefunction(fn):
         wrapped_fn = _async_task_wrapper(app)(fn)
     else:
@@ -202,7 +198,6 @@ def register_task(  # type: ignore[misc]
     app.task(
         name=task_name or fn.__name__,
         bind=True,
-        base=AbortableTask,
         time_limit=None if timeout is None else timeout.total_seconds(),
         pydantic=True,
     )(wrapped_fn)