From 482d7fd3eb35337c6e74f06c9cefdf46453c55ef Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 21:30:42 -0500 Subject: [PATCH] perf: shut down container services concurrently ServiceContainer.shutdown is the FastAPI shutdown handler, so it runs inside the SIGTERM grace window on every deploy and restart. It had two defects. Teardown was serial: a for loop awaited each service in turn, making total cost the sum of every close() round-trip rather than the slowest one. The services are independent and no teardown ordering was declared or guaranteed, so they now tear down under a single asyncio.gather. Exceptions are collected via return_exceptions and reported per service, preserving the previous "log and continue" behaviour. Aliased singletons were cleaned up twice. _register_skill_dependency_ aliases registers four aliases whose factories delegate to get_service, which caches into _singletons, so one instance ends up stored under two names. The loop then called cleanup() on it once per name, double closing sessions and pools. Targets are now deduplicated by identity. Teardown hooks may also be synchronous, so the result is awaited only when inspect.isawaitable, matching the idiom in api_cost_worker. This removes a spurious "await NoneType" error previously logged for sync cleanup(). Closes #1322 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/containers/service_container.py | 66 ++++++++++--- tests/unit/test_service_container.py | 95 +++++++++++++++++++ 2 files changed, 148 insertions(+), 13 deletions(-) diff --git a/src/youtube_extension/backend/containers/service_container.py b/src/youtube_extension/backend/containers/service_container.py index e7f9dea05..1bbf42c72 100644 --- a/src/youtube_extension/backend/containers/service_container.py +++ b/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() diff --git a/tests/unit/test_service_container.py b/tests/unit/test_service_container.py index c1aa56c32..8e155f124 100644 --- a/tests/unit/test_service_container.py +++ b/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 # ---------------------------------------------------------------------------