fix(tasks): add async_exit_stack protection to the worker

diracx-db/src/diracx/db/sql/dummy/db.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from sqlalchemy import insert
+from sqlalchemy.orm import DeclarativeBase
 from uuid_utils import UUID
 
 from diracx.core.models.search import SearchSpec
@@ -25,18 +26,22 @@ class DummyDB(BaseSQLDB):
     metadata = DummyDBBase.metadata
 
     async def summary(
-        self, group_by: list[str], search: list[SearchSpec]
+        self,
+        group_by: list[str],
+        search: list[SearchSpec],
+        table: type[DeclarativeBase] = Cars,
     ) -> list[dict[str, str | int]]:
         """Get a summary of the cars.
 
         Args:
-            group_by: Car fields used to group the summary.
+            group_by: Fields used to group the summary.
             search: Search conditions to apply before summarizing.
+            table: Optional table name to be queried. Default: Cars.
 
         Returns:
             Summary rows containing grouped values and counts.
         """
-        return await self._summary(table=Cars, group_by=group_by, search=search)
+        return await self._summary(table=table, group_by=group_by, search=search)
 
     async def insert_owner(self, name: str) -> int:
         """Insert a car owner.

diracx-tasks/src/diracx/tasks/plumbing/worker/worker.py

@@ -4,6 +4,7 @@
 import logging
 import weakref
 from collections.abc import Iterable
+from contextlib import AsyncExitStack
 from datetime import UTC, datetime, timedelta
 from time import time
 from typing import Any, Awaitable, Callable
@@ -739,7 +740,9 @@ async def _execute_task(
         found_exception: BaseException | None = None
 
         dep_kwargs: dict[str, Any] = {}
-        async_exit_stack = None
+        # async_exit_stack must be initialized here since a task without
+        # "_dependant" would otherwise hit 'async with None' and crash
+        async_exit_stack = AsyncExitStack()
 
         try:
             if hasattr(task_func, "_dependant"):
@@ -748,16 +751,17 @@ async def _execute_task(
                     dependency_overrides=self.broker.dependency_overrides,
                 )
 
-            # Obtain a Redis connection for lock acquisition
-            redis = await self._get_redis()
-            async with redis:
-                all_kwargs = {
-                    **dep_kwargs,
-                    **task_message.task_kwargs,
-                }
-                returned = await task_func(
-                    *task_message.task_args, _redis=redis, **all_kwargs
-                )
+            async with async_exit_stack:
+                # Obtain a Redis connection for lock acquisition
+                redis = await self._get_redis()
+                async with redis:
+                    all_kwargs = {
+                        **dep_kwargs,
+                        **task_message.task_kwargs,
+                    }
+                    returned = await task_func(
+                        *task_message.task_args, _redis=redis, **all_kwargs
+                    )
 
         except UnableToAcquireLockError:
             logger.info(
@@ -785,13 +789,6 @@ async def _execute_task(
                 exc_info=True,
             )
 
-        finally:
-            if async_exit_stack:
-                try:
-                    await async_exit_stack.aclose()
-                except Exception:
-                    logger.debug("Error closing exit stack", exc_info=True)
-
         execution_time = time() - start_time
 
         if found_exception is not None: