🐛 Celery tasks list doesn't include submitted, active and scheduled ones

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

@@ -52,7 +52,7 @@ async def start_data_export(
         ) from err
 
     task_uuid = await get_celery_client(app).send_task(
-        "export_data_with_error",
+        "export_data",
         task_context=job_id_data.model_dump(),
         files=data_export_start.file_and_folder_ids,  # ANE: adapt here your signature
     )

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

@@ -27,7 +27,10 @@ def create_app(celery_settings: CelerySettings) -> Celery:
     app.conf.result_expires = celery_settings.CELERY_RESULT_EXPIRES
     app.conf.result_extended = True  # original args are included in the results
     app.conf.result_serializer = "json"
+    app.conf.task_send_sent_event = True
     app.conf.task_track_started = True
+    app.conf.worker_send_task_events = True  # enable tasks monitoring
+
     return app
 
 

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

@@ -18,7 +18,6 @@
 
 _CELERY_INSPECT_TASK_STATUSES: Final[tuple[str, ...]] = (
     "active",
-    "registered",
     "scheduled",
     "revoked",
 )
@@ -131,21 +130,22 @@ def _get_completed_task_uuids(self, task_context: TaskContext) -> set[TaskUUID]:
 
     @make_async()
     def get_task_uuids(self, task_context: TaskContext) -> set[TaskUUID]:
-        all_task_ids = self._get_completed_task_uuids(task_context)
+        task_uuids = self._get_completed_task_uuids(task_context)
 
-        search_key = _CELERY_TASK_META_PREFIX + _build_task_id_prefix(task_context)
+        task_id_prefix = _build_task_id_prefix(task_context)
+        inspect = self._celery_app.control.inspect()
         for task_inspect_status in _CELERY_INSPECT_TASK_STATUSES:
-            if task_ids := getattr(
-                self._celery_app.control.inspect(), task_inspect_status
-            )():
-                for values in task_ids.values():
-                    for value in values:
-                        all_task_ids.add(
-                            TaskUUID(
-                                value.removeprefix(
-                                    search_key + _CELERY_TASK_ID_KEY_SEPARATOR
-                                )
-                            )
-                        )
-
-        return all_task_ids
+            tasks = getattr(inspect, task_inspect_status)() or {}
+
+            task_uuids.update(
+                TaskUUID(
+                    task_info["id"].removeprefix(
+                        task_id_prefix + _CELERY_TASK_ID_KEY_SEPARATOR
+                    )
+                )
+                for tasks_per_worker in tasks.values()
+                for task_info in tasks_per_worker
+                if "id" in task_info
+            )
+
+        return task_uuids