perf: shut down container services concurrently (#1322)

src/youtube_extension/backend/containers/service_container.py

@@ -7,6 +7,8 @@
 Implements IoC (Inversion of Control) pattern for better testability and modularity.
 """
 
+import asyncio
+import inspect
 import logging
 import os
 from typing import Any, Callable, Optional, TypeVar
@@ -442,6 +444,34 @@ def health_check(self) -> dict[str, Any]:
 
         return health_status
 
+    @staticmethod
+    async def _shutdown_service(name: str, service: Any) -> None:
+        """Run a single service's teardown hook.
+
+        Prefers ``cleanup()`` and falls back to ``close()``. Either hook may be
+        synchronous or a coroutine function, so the result is awaited only when
+        it is actually awaitable.
+        """
+        closer: Optional[Callable[[], Any]] = None
+        completion = ""
+
+        cleanup = getattr(service, "cleanup", None)
+        if callable(cleanup):
+            closer, completion = cleanup, "cleanup completed"
+        else:
+            close = getattr(service, "close", None)
+            if callable(close):
+                closer, completion = close, "closed"
+
+        if closer is None:
+            return
+
+        result = closer()
+        if inspect.isawaitable(result):
+            await result
+
+        logger.info(f"Service {completion}: {name}")
+
     async def shutdown(self):
         """
         Gracefully shutdown all services.
@@ -450,21 +480,31 @@ async def shutdown(self):
 
         shutdown_errors = []
 
+        # Skill-dependency aliases (e.g. ``gemini_service`` ->
+        # ``hybrid_processor_service``) resolve to the *same* instance, so the
+        # singleton map can hold one object under several names. Deduplicate by
+        # identity to avoid tearing the same service down more than once.
+        targets: list[tuple[str, Any]] = []
+        seen: set[int] = set()
         for name, service in self._singletons.items():
-            try:
-                # Call cleanup method if available
-                if hasattr(service, "cleanup"):
-                    if callable(service.cleanup):
-                        await service.cleanup()
-                        logger.info(f"Service cleanup completed: {name}")
-                elif hasattr(service, "close"):
-                    if callable(service.close):
-                        await service.close()
-                        logger.info(f"Service closed: {name}")
+            if id(service) in seen:
+                logger.debug(f"Skipping duplicate service alias: {name}")
+                continue
+            seen.add(id(service))
+            targets.append((name, service))
+
+        # Services are independent, so tear them down concurrently: shutdown runs
+        # inside the SIGTERM grace window, where serial teardown costs the sum of
+        # every close() round-trip instead of the slowest one.
+        results = await asyncio.gather(
+            *(self._shutdown_service(name, service) for name, service in targets),
+            return_exceptions=True,
+        )
 
-            except Exception as e:
-                logger.warning(f"Error during {name} service shutdown: {e}")
-                shutdown_errors.append(f"{name}: {e}")
+        for (name, _), result in zip(targets, results):
+            if isinstance(result, BaseException):
+                logger.warning(f"Error during {name} service shutdown: {result}")
+                shutdown_errors.append(f"{name}: {result}")
 
         # Clear service instances
         self._singletons.clear()

tests/unit/test_service_container.py

@@ -3,6 +3,9 @@
 
 from __future__ import annotations
 
+import asyncio
+import logging
+import time
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -381,6 +384,98 @@ class PlainSvc:
         assert sc._singletons == {}
 
 
+class TestShutdownConcurrency:
+    """Shutdown runs inside the SIGTERM grace window, so teardown is concurrent."""
+
+    async def test_services_are_torn_down_concurrently(self):
+        sc = _bare_container()
+        in_flight = 0
+        peak = 0
+
+        class SlowSvc:
+            async def cleanup(self):
+                nonlocal in_flight, peak
+                in_flight += 1
+                peak = max(peak, in_flight)
+                try:
+                    await asyncio.sleep(0.05)
+                finally:
+                    in_flight -= 1
+
+        for i in range(6):
+            sc._singletons[f"svc{i}"] = SlowSvc()
+
+        started = time.perf_counter()
+        await sc.shutdown()
+        elapsed = time.perf_counter() - started
+
+        assert peak == 6, f"expected all 6 teardowns in flight together, saw {peak}"
+        # Serial teardown would take >= 6 * 50ms = 300ms.
+        assert elapsed < 0.20, f"teardown appears serial ({elapsed * 1000:.0f}ms)"
+
+    async def test_aliased_service_is_torn_down_once(self):
+        """Skill-dependency aliases share an instance; it must not be cleaned twice."""
+        sc = _bare_container()
+        shared = MagicMock()
+        shared.cleanup = AsyncMock()
+
+        sc._singletons["hybrid_processor_service"] = shared
+        sc._singletons["gemini_service"] = shared  # alias -> same object
+
+        await sc.shutdown()
+
+        shared.cleanup.assert_called_once()
+
+    async def test_distinct_services_are_each_torn_down(self):
+        """Deduplication is by identity, not by equality."""
+        sc = _bare_container()
+        first, second = MagicMock(), MagicMock()
+        first.cleanup = AsyncMock()
+        second.cleanup = AsyncMock()
+
+        sc._singletons["a"] = first
+        sc._singletons["b"] = second
+
+        await sc.shutdown()
+
+        first.cleanup.assert_called_once()
+        second.cleanup.assert_called_once()
+
+    async def test_synchronous_cleanup_is_supported(self, caplog):
+        """A non-async cleanup() must run without an 'await NoneType' error."""
+        sc = _bare_container()
+        calls = []
+
+        class SyncSvc:
+            def cleanup(self):
+                calls.append("cleanup")
+
+        sc._singletons["sync"] = SyncSvc()
+
+        with caplog.at_level(logging.WARNING):
+            await sc.shutdown()
+
+        assert calls == ["cleanup"]
+        assert not [r for r in caplog.records if "Error during" in r.message], (
+            "synchronous cleanup should not be reported as a shutdown error"
+        )
+
+    async def test_one_failure_does_not_block_other_services(self):
+        sc = _bare_container()
+        healthy = MagicMock()
+        healthy.cleanup = AsyncMock()
+        broken = MagicMock()
+        broken.cleanup = AsyncMock(side_effect=RuntimeError("boom"))
+
+        sc._singletons["broken"] = broken
+        sc._singletons["healthy"] = healthy
+
+        await sc.shutdown()
+
+        healthy.cleanup.assert_called_once()
+        assert sc._singletons == {}
+
+
 # ---------------------------------------------------------------------------
 # _register_core_services
 # ---------------------------------------------------------------------------