add expiration

Author: giancarloromeo

packages/settings-library/src/settings_library/celery.py

@@ -22,6 +22,12 @@ class CelerySettings(BaseCustomSettings):
             description="Time after which task results will be deleted (default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formating)."
         ),
     ] = timedelta(days=7)
+    CELERY_EPHEMERAL_RESULT_EXPIRES: Annotated[
+        timedelta,
+        Field(
+            description="Time after which ephemeral task results will be deleted (default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formating)."
+        ),
+    ] = timedelta(hours=1)
     CELERY_RESULT_PERSISTENT: Annotated[
         bool,
         Field(

services/storage/src/simcore_service_storage/modules/celery/__init__.py

@@ -29,7 +29,9 @@ async def on_startup() -> None:
         )
 
         app.state.celery_client = CeleryTaskQueueClient(
-            celery_app, RedisTaskMetadataStore(redis_client_sdk)
+            celery_app,
+            celery_settings,
+            RedisTaskMetadataStore(redis_client_sdk),
         )
 
         register_celery_types()

services/storage/src/simcore_service_storage/modules/celery/backends/_redis.py

@@ -1,3 +1,5 @@
+import logging
+from datetime import timedelta
 from typing import Final
 
 from celery.result import AsyncResult
@@ -11,6 +13,8 @@
 _CELERY_TASK_ID_KEY_SEPARATOR: Final[str] = ":"
 _CELERY_TASK_SCAN_COUNT_PER_BATCH: Final[int] = 10000
 
+_logger = logging.getLogger(__name__)
+
 
 class RedisTaskMetadataStore:
     def __init__(self, redis_client_sdk: RedisClientSDK) -> None:
@@ -30,7 +34,11 @@ async def get(self, task_id: TaskID) -> TaskMetadata | None:
         return TaskMetadata.model_validate_json(result) if result else None
 
     async def get_uuids(self, task_context: TaskContext) -> set[TaskUUID]:
-        search_key = build_task_id_prefix(task_context) + _CELERY_TASK_ID_KEY_SEPARATOR
+        search_key = (
+            _CELERY_TASK_METADATA_PREFIX
+            + build_task_id_prefix(task_context)
+            + _CELERY_TASK_ID_KEY_SEPARATOR
+        )
         keys = set()
         async for key in self._redis_client_sdk.redis.scan_iter(
             match=search_key + "*", count=_CELERY_TASK_SCAN_COUNT_PER_BATCH
@@ -50,8 +58,11 @@ async def remove(self, task_id: TaskID) -> None:
         )
         AsyncResult(_CELERY_TASK_META_PREFIX + task_id).forget()
 
-    async def set(self, task_id: TaskID, task_data: TaskMetadata) -> None:
+    async def set(
+        self, task_id: TaskID, task_data: TaskMetadata, expiry: timedelta
+    ) -> None:
         await self._redis_client_sdk.redis.set(
             _CELERY_TASK_METADATA_PREFIX + task_id,
             task_data.model_dump_json(),
+            ex=expiry,
         )

services/storage/src/simcore_service_storage/modules/celery/client.py

@@ -12,6 +12,7 @@
 from models_library.progress_bar import ProgressReport
 from pydantic import ValidationError
 from servicelib.logging_utils import log_context
+from settings_library.celery import CelerySettings
 
 from .models import (
     TaskContext,
@@ -43,6 +44,7 @@
 @dataclass
 class CeleryTaskQueueClient:
     _celery_app: Celery
+    _celery_settings: CelerySettings
     _task_store: TaskMetadataStore
 
     async def send_task(
@@ -67,7 +69,13 @@ async def send_task(
                 kwargs=task_params,
                 queue=task_metadata.queue,
             )
-            await self._task_store.set(task_id, task_metadata)
+
+            expiry = (
+                self._celery_settings.CELERY_EPHEMERAL_RESULT_EXPIRES
+                if task_metadata.ephemeral
+                else self._celery_settings.CELERY_RESULT_EXPIRES
+            )
+            await self._task_store.set(task_id, task_metadata, expiry=expiry)
             return task_uuid
 
     @staticmethod

services/storage/src/simcore_service_storage/modules/celery/models.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from enum import StrEnum, auto
 from typing import Any, Final, Protocol, TypeAlias
 from uuid import UUID
@@ -49,7 +50,9 @@ async def get_uuids(self, task_context: TaskContext) -> set[TaskUUID]: ...
 
     async def remove(self, task_id: TaskID) -> None: ...
 
-    async def set(self, task_id: TaskID, task_data: TaskMetadata) -> None: ...
+    async def set(
+        self, task_id: TaskID, task_data: TaskMetadata, expiry: timedelta
+    ) -> None: ...
 
 
 class TaskStatus(BaseModel):