prune trash

Author: pcrespov

packages/service-library/src/servicelib/background_task.py

@@ -38,11 +38,14 @@ def periodic(
     *,
     interval: datetime.timedelta,
     raise_on_error: bool = False,
-    early_wake_up_event: asyncio.Event | None = None,
+    early_wake_up_event: (
+        asyncio.Event | None
+    ) = None,  # TODO: i would argue that this should be a different decorator instead of an optional arguments since with this event,
+    # the funciton is not periodic anymore but rahter repeatedly triggered by the event
 ) -> Callable[
     [Callable[P, Coroutine[Any, Any, None]]], Callable[P, Coroutine[Any, Any, None]]
 ]:
-    """Calls the function periodically with a given interval.
+    """Calls the function periodically with a given interval or triggered by an early wake-up event.
 
     Arguments:
         interval -- the interval between calls
@@ -58,7 +61,7 @@ def periodic(
     """
 
     def _decorator(
-        func: Callable[P, Coroutine[Any, Any, None]],
+        coro: Callable[P, Coroutine[Any, Any, None]],
     ) -> Callable[P, Coroutine[Any, Any, None]]:
         class _InternalTryAgain(TryAgain):
             # Local exception to prevent reacting to similarTryAgain exceptions raised by the wrapped func
@@ -82,10 +85,10 @@ class _InternalTryAgain(TryAgain):
             ),
             before_sleep=before_sleep_log(_logger, logging.DEBUG),
         )
-        @functools.wraps(func)
+        @functools.wraps(coro)
         async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> None:
             with log_catch(_logger, reraise=True):
-                await func(*args, **kwargs)
+                await coro(*args, **kwargs)
             raise _InternalTryAgain
 
         return _wrapper

packages/service-library/src/servicelib/background_task_utils.py

@@ -43,7 +43,7 @@ def _decorator(
             # Replicas will raise CouldNotAcquireLockError
             # SEE https://github.com/ITISFoundation/osparc-simcore/issues/7574
             (CouldNotAcquireLockError,),
-            reason="Multiple instances of the periodic task `{coro.__module__}.{coro.__name__}` are running.",
+            reason=f"Multiple instances of the periodic task `{coro.__module__}.{coro.__name__}` are running.",
         )
         @exclusive(
             redis_client,

services/web/server/src/simcore_service_webserver/garbage_collector/_tasks_trash.py

@@ -1,17 +1,18 @@
 """
-    Scheduled tasks addressing users
+Scheduled tasks addressing users
 
 """
 
 import asyncio
 import logging
 from collections.abc import AsyncIterator, Callable
+from datetime import timedelta
 
 from aiohttp import web
+from servicelib.async_utils import cancel_wait_task
+from servicelib.background_task_utils import exclusive_periodic
 from servicelib.logging_utils import log_context
-from tenacity import retry
-from tenacity.before_sleep import before_sleep_log
-from tenacity.wait import wait_exponential
+from simcore_service_webserver.redis import get_redis_lock_manager_client_sdk
 
 from ..trash import trash_service
 
@@ -20,45 +21,36 @@
 CleanupContextFunc = Callable[[web.Application], AsyncIterator[None]]
 
 
-_PERIODIC_TASK_NAME = f"{__name__}"
-_APP_TASK_KEY = f"{_PERIODIC_TASK_NAME}.task"
+def create_background_task_to_prune_trash(wait_s: float) -> CleanupContextFunc:
 
+    async def _cleanup_ctx_fun(app: web.Application) -> AsyncIterator[None]:
 
-@retry(
-    wait=wait_exponential(min=5, max=20),
-    before_sleep=before_sleep_log(_logger, logging.WARNING),
-)
-async def _run_task(app: web.Application):
-    with log_context(_logger, logging.INFO, "Deleting expired trashed items"):
-        await trash_service.safe_delete_expired_trash_as_admin(app)
-
-
-async def _run_periodically(app: web.Application, wait_interval_s: float):
-    while True:
-        await _run_task(app)
-        await asyncio.sleep(wait_interval_s)
-
+        @exclusive_periodic(
+            # Function-exclusiveness is required to avoid multiple tasks like thisone running concurrently
+            get_redis_lock_manager_client_sdk(app),
+            task_interval=timedelta(seconds=wait_s),
+            retry_after=timedelta(minutes=5),
+        )
+        async def _prune_trash_periodically() -> None:
+            with log_context(_logger, logging.INFO, "Deleting expired trashed items"):
+                await trash_service.safe_delete_expired_trash_as_admin(app)
 
-def create_background_task_to_prune_trash(
-    wait_s: float, task_name: str = _PERIODIC_TASK_NAME
-) -> CleanupContextFunc:
-    async def _cleanup_ctx_fun(
-        app: web.Application,
-    ) -> AsyncIterator[None]:
         # setup
+        task_name = _prune_trash_periodically.__name__
+
         task = asyncio.create_task(
-            _run_periodically(app, wait_s),
+            _prune_trash_periodically(),
             name=task_name,
         )
-        app[_APP_TASK_KEY] = task
+
+        # prevents premature garbage collection of the task
+        app_task_key = f"tasks.{task_name}"
+        app[app_task_key] = task
 
         yield
 
         # tear-down
-        task.cancel()
-        try:
-            await task
-        except asyncio.CancelledError:
-            assert task.cancelled()  # nosec
+        await cancel_wait_task(task)
+        app.pop(app_task_key, None)
 
     return _cleanup_ctx_fun