refactor: no functions inside functions in the worker and the tests

AGENTS.md

@@ -218,7 +218,7 @@ Rules:
 - Use `raise SpecificError(...) from original_exc` to preserve exception chains
 - Never catch `Exception` broadly; catch specific exception types (the two allowed cases follow)
 
-A catch-all is allowed in two cases only. The first records or logs, then re-raises: `UnifiedProducer.produce` in `app/events/core/producer.py` marks the event as failed to publish, `IdempotencyMiddleware` and `DeadLetterMiddleware` in `app/events/middlewares.py` release the idempotency key and save the event to the DLQ, and `ExecutionService` marks the execution as errored before raising `InfrastructureError`. The second is a loop or batch boundary that must survive one bad item: the saga orchestrator's queue loop (`try_schedule_from_queue`), saga step execution and its compensation loop, the pod monitor's watch cycle (`app/worker.py`) and per-event processing, k8s worker pod creation (publishes a pod-creation-failed event instead), notification delivery and the scheduled-notification batch, DLQ batch retry and per-event replay. A new catch-all needs the same justification and must log or record the error; never swallow it silently.
+A catch-all is allowed in two cases only. The first records or logs, then re-raises: `UnifiedProducer.produce` in `app/events/core/producer.py` marks the event as failed to publish, `IdempotencyMiddleware` and `DeadLetterMiddleware` in `app/events/middlewares.py` release the idempotency key and save the event to the DLQ, and `ExecutionService` marks the execution as errored before raising `InfrastructureError`. The second is a loop or batch boundary that must survive one bad item: the saga orchestrator's queue loop (`try_schedule_from_queue`), saga step execution and its compensation loop, the pod monitor's watch cycle (`PodMonitor.watch_cycle`) and per-event processing, k8s worker pod creation (publishes a pod-creation-failed event instead), notification delivery and the scheduled-notification batch, DLQ batch retry and per-event replay. A new catch-all needs the same justification and must log or record the error; never swallow it silently.
 
 ```python
 # Repository: returns None for reads; raises only on write constraints
@@ -544,7 +544,7 @@ Everything outside the API runs in one process, `python -m app.worker` (`backend
 3. `create_worker_container(settings)`
 4. Get `KafkaBroker` from the container and `broker.include_routers(k8s_worker_router(), result_processor_router(), saga_router())`
 5. `setup_dishka(container, broker=broker, auto_inject=True)`
-6. `FastStream(broker, on_startup=[...], on_shutdown=[...]).run()`; APScheduler jobs start in `on_startup`
+6. `start_jobs(container)` prepares the cluster and starts the APScheduler jobs, then `FastStream(broker, on_shutdown=[scheduler.shutdown], after_shutdown=[container.close]).run()`
 
 | Part | Purpose |
 |------|---------|

backend/app/services/pod_monitor/monitor.py

@@ -44,8 +44,7 @@ class PodMonitor:
       and a resource_version cursor.
     - Subsequent calls WATCH from that cursor, receiving only new changes.
 
-    Stateless service — all lifecycle (watch scheduling, error handling,
-    shutdown) is managed by APScheduler in the DI provider.
+    The worker's scheduler runs watch_cycle every 5 seconds.
     """
 
     def __init__(
@@ -86,7 +85,7 @@ async def watch_pod_events(self) -> None:
         that cursor.
 
         Returns normally when the server-side timeout expires.
-        Raises on K8s API errors — the caller (provider) handles retries.
+        Raises on K8s API errors; watch_cycle counts them and resets the cursor on a 410.
         """
         if not self._last_resource_version:
             await self._list_and_process_existing_pods()
@@ -116,6 +115,28 @@ async def watch_pod_events(self) -> None:
             if self._watch.resource_version:
                 self._last_resource_version = self._watch.resource_version
 
+    async def watch_cycle(self) -> None:
+        """Run one watch and count the error that ended it, so the next cycle starts over cleanly.
+
+        A 410 Gone means the cursor expired, so the next cycle lists again before watching.
+        """
+        try:
+            await self.watch_pod_events()
+            return
+        except ApiException as e:
+            if e.status == 410:
+                self.logger.warning("Resource version expired, resetting watch cursor")
+                self._last_resource_version = None
+                error_type = ErrorType.RESOURCE_VERSION_EXPIRED
+            else:
+                self.logger.error("API error in watch", status=e.status, reason=e.reason)
+                error_type = ErrorType.API_ERROR
+        except Exception:
+            self.logger.error("Unexpected error in watch", exc_info=True)
+            error_type = ErrorType.UNEXPECTED
+        self._metrics.record_pod_monitor_watch_error(error_type)
+        self._metrics.increment_pod_monitor_watch_reconnects()
+
     async def _list_and_process_existing_pods(self) -> None:
         """LIST all matching pods to bootstrap state and get a resource_version cursor."""
         kwargs: dict[str, Any] = {

backend/app/worker.py

@@ -1,112 +1,76 @@
 import asyncio
 from typing import Any
 
+import structlog
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from beanie import init_beanie
+from dishka import AsyncContainer
 from dishka_faststream import setup_dishka
 from faststream import FastStream
 from faststream.kafka import KafkaBroker
-from kubernetes_asyncio.client.rest import ApiException
 from pymongo import AsyncMongoClient
 
 from app.core.container import create_worker_container
 from app.core.logging import setup_log_exporter, setup_logger
-from app.core.metrics import KubernetesMetrics
 from app.core.middlewares import setup_metrics
 from app.db.docs import ALL_DOCUMENTS
 from app.dlq.manager import DLQManager
 from app.events.handlers import k8s_worker_router, result_processor_router, saga_router
 from app.services.k8s_worker import KubernetesWorker
-from app.services.pod_monitor import ErrorType, PodMonitor
+from app.services.pod_monitor import PodMonitor
 from app.services.saga import SagaOrchestrator
 from app.settings import Settings
 
 
 def main() -> None:
     """Run everything outside the API: the Kafka consumers, the pod watch and the scheduled jobs."""
     settings = Settings(override_path="config.worker.toml")
-
     logger = setup_logger(settings.LOG_LEVEL)
     setup_log_exporter(settings, logger)
     setup_metrics(settings, logger)
-
     logger.info("Starting worker...")
-
-    async def run() -> None:
-        # Initialize Beanie with tz_aware client (so MongoDB returns aware datetimes)
-        client: AsyncMongoClient[dict[str, Any]] = AsyncMongoClient(settings.MONGODB_URL, tz_aware=True)
-        await init_beanie(
-            database=client.get_default_database(default=settings.MONGO_DB),
-            document_models=ALL_DOCUMENTS,
-        )
-        logger.info("MongoDB initialized via Beanie")
-
-        container = create_worker_container(settings)
-        broker: KafkaBroker = await container.get(KafkaBroker)
-
-        # Include the subscribers, then set up DI integration
-        broker.include_routers(k8s_worker_router(), result_processor_router(), saga_router())
-        setup_dishka(container, broker=broker, auto_inject=True)
-
-        scheduler = AsyncIOScheduler()
-
-        async def start() -> None:
-            k8s_worker = await container.get(KubernetesWorker)
-            await k8s_worker.ensure_namespace_security()
-            await k8s_worker.ensure_image_pre_puller_daemonset()
-
-            monitor = await container.get(PodMonitor)
-            kubernetes_metrics = await container.get(KubernetesMetrics)
-
-            async def watch_pods() -> None:
-                error_type: ErrorType | None = None
-                try:
-                    await monitor.watch_pod_events()
-                except ApiException as e:
-                    if e.status == 410:
-                        logger.warning("Resource version expired, resetting watch cursor")
-                        monitor._last_resource_version = None
-                        error_type = ErrorType.RESOURCE_VERSION_EXPIRED
-                    else:
-                        logger.error("API error in watch", status=e.status, reason=e.reason)
-                        error_type = ErrorType.API_ERROR
-                except Exception:
-                    logger.error("Unexpected error in watch", exc_info=True)
-                    error_type = ErrorType.UNEXPECTED
-
-                if error_type is not None:
-                    kubernetes_metrics.record_pod_monitor_watch_error(error_type)
-                    kubernetes_metrics.increment_pod_monitor_watch_reconnects()
-
-            orchestrator = await container.get(SagaOrchestrator)
-            dlq_manager = await container.get(DLQManager)
-
-            jobs = (
-                ("pod_monitor_watch", watch_pods, 5, 60),
-                ("saga_try_schedule", orchestrator.try_schedule_from_queue, 10, 30),
-                ("saga_check_timeouts", orchestrator.check_timeouts, 30, 60),
-                ("dlq_monitor_retries", dlq_manager.process_monitoring_cycle, 10, 60),
-            )
-            for job_id, job, seconds, grace in jobs:
-                scheduler.add_job(
-                    job, trigger="interval", seconds=seconds, id=job_id, max_instances=1, misfire_grace_time=grace
-                )
-            scheduler.start()
-            logger.info("Worker started", jobs=[job_id for job_id, *_ in jobs])
-
-        async def stop_scheduler() -> None:
-            scheduler.shutdown(wait=False)
-
-        app = FastStream(
-            broker,
-            on_startup=[start],
-            on_shutdown=[stop_scheduler],
-            after_shutdown=[container.close],
+    asyncio.run(run(settings, logger))
+
+
+async def run(settings: Settings, logger: structlog.stdlib.BoundLogger) -> None:
+    # A tz-aware client, so MongoDB returns aware datetimes
+    client: AsyncMongoClient[dict[str, Any]] = AsyncMongoClient(settings.MONGODB_URL, tz_aware=True)
+    await init_beanie(database=client.get_default_database(default=settings.MONGO_DB), document_models=ALL_DOCUMENTS)
+
+    container = create_worker_container(settings)
+    broker: KafkaBroker = await container.get(KafkaBroker)
+    broker.include_routers(k8s_worker_router(), result_processor_router(), saga_router())
+    setup_dishka(container, broker=broker, auto_inject=True)
+
+    scheduler = await start_jobs(container)
+    # FastStream stops the scheduler before the broker and closes the container after it
+    await FastStream(broker, on_shutdown=[scheduler.shutdown], after_shutdown=[container.close]).run()
+    logger.info("Worker shutdown complete")
+
+
+async def start_jobs(container: AsyncContainer) -> AsyncIOScheduler:
+    """Prepare the cluster, then schedule the pod watch, the saga queue and timeouts, and DLQ retries."""
+    k8s_worker = await container.get(KubernetesWorker)
+    await k8s_worker.ensure_namespace_security()
+    await k8s_worker.ensure_image_pre_puller_daemonset()
+
+    monitor = await container.get(PodMonitor)
+    orchestrator = await container.get(SagaOrchestrator)
+    dlq_manager = await container.get(DLQManager)
+
+    scheduler = AsyncIOScheduler()
+    jobs = (
+        ("pod_monitor_watch", monitor.watch_cycle, 5, 60),
+        ("saga_try_schedule", orchestrator.try_schedule_from_queue, 10, 30),
+        ("saga_check_timeouts", orchestrator.check_timeouts, 30, 60),
+        ("dlq_monitor_retries", dlq_manager.process_monitoring_cycle, 10, 60),
+    )
+    for job_id, job, seconds, grace in jobs:
+        scheduler.add_job(
+            job, trigger="interval", seconds=seconds, id=job_id, max_instances=1, misfire_grace_time=grace
         )
-        await app.run()
-        logger.info("Worker shutdown complete")
-
-    asyncio.run(run())
+    scheduler.start()
+    return scheduler
 
 
 if __name__ == "__main__":

backend/pyproject.toml

@@ -103,7 +103,6 @@ packages = ["app"]
 [dependency-groups]
 test = [
     "coverage==7.16.2",
-    "hypothesis==6.168.5",
     "iniconfig==2.3.1",
     "pluggy==1.6.0",
     "pytest==9.1.1",

backend/tests/e2e/conftest.py

@@ -11,7 +11,7 @@
 from dishka import AsyncContainer
 from fastapi import FastAPI
 from httpx2 import AsyncClient
-from starlette.types import ASGIApp, Message
+from starlette.types import ASGIApp, Message, Scope
 
 from app.core.metrics import DLQMetrics, ExecutionMetrics, RateLimitMetrics
 from app.core.security import SecurityService
@@ -49,20 +49,51 @@
 from app.services.user_settings_service import UserSettingsService
 
 
+class _Exchange:
+    """One request's ASGI side: the body to hand over, the response chunks and the client's disconnect."""
+
+    def __init__(self, body: bytes) -> None:
+        self.body = body
+        self.request_sent = False
+        self.chunks: asyncio.Queue[bytes | None] = asyncio.Queue()
+        self.started: asyncio.Future[Message] = asyncio.get_running_loop().create_future()
+        self.disconnect = asyncio.Event()
+
+    async def receive(self) -> Message:
+        if not self.request_sent:
+            self.request_sent = True
+            return {"type": "http.request", "body": self.body, "more_body": False}
+        await self.disconnect.wait()
+        return {"type": "http.disconnect"}
+
+    async def send(self, message: Message) -> None:
+        if message["type"] == "http.response.start":
+            self.started.set_result(message)
+        elif message["type"] == "http.response.body":
+            await self.chunks.put(message.get("body", b""))
+            if not message.get("more_body", False):
+                await self.chunks.put(None)
+
+    async def run(self, app: ASGIApp, scope: Scope) -> None:
+        await app(scope, self.receive, self.send)
+
+    def end_body(self, _app_task: asyncio.Task[None]) -> None:
+        self.chunks.put_nowait(None)
+
+
 class _QueuedBody(httpx2.AsyncByteStream):
-    def __init__(self, chunks: asyncio.Queue[bytes | None], disconnect: asyncio.Event, app_task: asyncio.Task[None]):
-        self._chunks = chunks
-        self._disconnect = disconnect
+    def __init__(self, exchange: _Exchange, app_task: asyncio.Task[None]) -> None:
+        self._exchange = exchange
         self._app_task = app_task
 
     async def __aiter__(self) -> AsyncIterator[bytes]:
-        while (chunk := await self._chunks.get()) is not None:
+        while (chunk := await self._exchange.chunks.get()) is not None:
             yield chunk
         if self._app_task.done() and not self._app_task.cancelled() and (error := self._app_task.exception()):
             raise error
 
     async def aclose(self) -> None:
-        self._disconnect.set()
+        self._exchange.disconnect.set()
         await asyncio.wait({self._app_task}, timeout=1.0)
         self._app_task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
@@ -76,29 +107,8 @@ def __init__(self, app: ASGIApp) -> None:
         self._app = app
 
     async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response:
-        body = await request.aread()
-        chunks: asyncio.Queue[bytes | None] = asyncio.Queue()
-        started: asyncio.Future[Message] = asyncio.get_running_loop().create_future()
-        disconnect = asyncio.Event()
-        request_sent = False
-
-        async def receive() -> Message:
-            nonlocal request_sent
-            if not request_sent:
-                request_sent = True
-                return {"type": "http.request", "body": body, "more_body": False}
-            await disconnect.wait()
-            return {"type": "http.disconnect"}
-
-        async def send(message: Message) -> None:
-            if message["type"] == "http.response.start":
-                started.set_result(message)
-            elif message["type"] == "http.response.body":
-                await chunks.put(message.get("body", b""))
-                if not message.get("more_body", False):
-                    await chunks.put(None)
-
-        scope = {
+        exchange = _Exchange(await request.aread())
+        scope: Scope = {
             "type": "http",
             "asgi": {"version": "3.0"},
             "http_version": "1.1",
@@ -112,22 +122,18 @@ async def send(message: Message) -> None:
             "client": ("127.0.0.1", 123),
             "root_path": "",
         }
-
-        async def run_app() -> None:
-            await self._app(scope, receive, send)
-
-        app_task = asyncio.create_task(run_app())
-        app_task.add_done_callback(lambda _: chunks.put_nowait(None))
-        first: set[asyncio.Future[Any]] = {app_task, started}
+        app_task = asyncio.create_task(exchange.run(self._app, scope))
+        app_task.add_done_callback(exchange.end_body)
+        first: set[asyncio.Future[Any]] = {app_task, exchange.started}
         await asyncio.wait(first, return_when=asyncio.FIRST_COMPLETED)
-        if not started.done():
+        if not exchange.started.done():
             app_task.result()
             raise RuntimeError("ASGI app returned without starting a response")
-        start = started.result()
+        start = exchange.started.result()
         return httpx2.Response(
             start["status"],
             headers=start.get("headers", []),
-            stream=_QueuedBody(chunks, disconnect, app_task),
+            stream=_QueuedBody(exchange, app_task),
             request=request,
         )
 

backend/tests/e2e/test_admin_users_routes.py

@@ -107,15 +107,11 @@ async def test_list_users_by_active_status(self, test_admin: AsyncClient) -> Non
         update = UserUpdate(is_active=False)
         await test_admin.put(f"/api/v1/admin/users/{user_id}", json=update.model_dump(exclude_none=True))
 
-        async def matches(is_active: bool) -> list[str]:
-            response = await test_admin.get(
-                "/api/v1/admin/users/", params={"search": request.username, "is_active": is_active}
-            )
-            assert response.status_code == 200
-            return [u.user_id for u in UserListResponse.model_validate(response.json()).users]
-
-        assert await matches(False) == [user_id]
-        assert await matches(True) == []
+        disabled = await test_admin.get("/api/v1/admin/users/", params={"search": request.username, "is_active": False})
+        active = await test_admin.get("/api/v1/admin/users/", params={"search": request.username, "is_active": True})
+
+        assert [u.user_id for u in UserListResponse.model_validate(disabled.json()).users] == [user_id]
+        assert UserListResponse.model_validate(active.json()).users == []
 
     @pytest.mark.parametrize(
         ("limits", "params", "listed"),