add task deletion

Author: giancarloromeo

services/storage/src/simcore_service_storage/api/rpc/_async_jobs.py

@@ -32,7 +32,7 @@
 
 
 @router.expose(reraise_if_error_type=(JobSchedulerError,))
-async def cancel(app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData):
+async def abort(app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData):
     assert app  # nosec
     assert job_id_data  # nosec
     try:
@@ -44,6 +44,19 @@ async def cancel(app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData
         raise JobSchedulerError(exc=f"{exc}") from exc
 
 
+@router.expose(reraise_if_error_type=(JobSchedulerError,))
+async def deletet(app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData):
+    assert app  # nosec
+    assert job_id_data  # nosec
+    try:
+        await get_celery_client(app).delete_task(
+            task_context=job_id_data.model_dump(),
+            task_uuid=job_id,
+        )
+    except CeleryError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+
+
 @router.expose(reraise_if_error_type=(JobSchedulerError,))
 async def status(
     app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData

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

@@ -78,10 +78,20 @@ async def abort_task(self, task_context: TaskContext, task_uuid: TaskUUID) -> No
         with log_context(
             _logger,
             logging.DEBUG,
-            msg=f"Abort task: {task_context=} {task_uuid=}",
+            msg=f"task abortion: {task_context=} {task_uuid=}",
         ):
             await self._abort_task(task_context, task_uuid)
 
+    async def delete_task(self, task_context: TaskContext, task_uuid: TaskUUID) -> None:
+        with log_context(
+            _logger,
+            logging.DEBUG,
+            msg=f"task deletion: {task_context=} {task_uuid=}",
+        ):
+            task_id = build_task_id(task_context, task_uuid)
+            await self._task_info_store.remove_task(task_id)
+            await self._forget_task(task_id)
+
     @make_async()
     def _forget_task(self, task_id: TaskID) -> None:
         AbortableAsyncResult(task_id, app=self._celery_app).forget()
@@ -100,8 +110,7 @@ async def get_task_result(
             if async_result.ready():
                 task_metadata = await self._task_info_store.get_task_metadata(task_id)
                 if task_metadata is not None and task_metadata.ephemeral:
-                    await self._task_info_store.remove_task(task_id)
-                    await self._forget_task(task_id)
+                    await self.delete_task(task_context, task_uuid)
             return result
 
     async def _get_task_progress_report(