fix(app): add coroutine retry

Author: emptybutton

src/tgdb/main/server/di.py

@@ -14,7 +14,7 @@
     RelationSchemasFromInMemoryDbAsRelationViews,
 )
 from tgdb.presentation.fastapi.common.app import (
-    FastAPIAppCoroutines,
+    FastAPIAppBackground,
     FastAPIAppRouters,
     FastAPIAppVersion,
 )
@@ -46,10 +46,10 @@ def provide_fast_api_app_coroutines(
         self,
         output_commits_to_tuples: OutputCommitsToTuples,
         output_commits: OutputCommits,
-    ) -> FastAPIAppCoroutines:
-        return FastAPIAppCoroutines((
-            output_commits(),
-            output_commits_to_tuples(),
+    ) -> FastAPIAppBackground:
+        return FastAPIAppBackground((
+            output_commits,
+            output_commits_to_tuples,
         ))
 
     @provide(scope=Scope.APP)

src/tgdb/presentation/fastapi/common/app.py

@@ -1,7 +1,9 @@
 import asyncio
-from collections.abc import AsyncIterator, Coroutine
-from contextlib import asynccontextmanager, suppress
-from typing import Any, NewType, cast
+from collections.abc import AsyncIterator, Callable, Coroutine
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from types import TracebackType
+from typing import Any, NewType, Self, cast
 
 from dishka import AsyncContainer
 from dishka.integrations.fastapi import setup_dishka
@@ -11,24 +13,64 @@
 from tgdb.presentation.fastapi.common.tags import tags_metadata
 
 
-FastAPIAppCoroutines = NewType(
-    "FastAPIAppCoroutines", tuple[Coroutine[Any, Any, Any], ...]
+FastAPIAppBackground = NewType(
+    "FastAPIAppBackground", tuple[Callable[[], Coroutine[Any, Any, Any]], ...]
 )
 FastAPIAppRouters = NewType("FastAPIAppRouters", tuple[APIRouter, ...])
 FastAPIAppVersion = NewType("FastAPIAppVersion", str)
 
 
+@dataclass(frozen=True, unsafe_hash=False)
+class LefespanBackground:
+    _loop: asyncio.AbstractEventLoop = field(
+        default_factory=asyncio.get_running_loop
+    )
+    _tasks: set[asyncio.Task[Any]] = field(init=False, default_factory=set)
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        error_type: type[BaseException] | None,
+        error: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        for task in self._tasks:
+            task.cancel()
+
+    def add(
+        self, func: Callable[[], Coroutine[Any, Any, Any]],
+    ) -> None:
+        decorated_func = self._decorator(func)
+        self._create_task(decorated_func())
+
+    def _decorator(
+        self, func: Callable[[], Coroutine[Any, Any, Any]]
+    ) -> Callable[[], Coroutine[Any, Any, Any]]:
+        async def decorated_func() -> None:
+            try:
+                await func()
+            except Exception as error:
+                self._create_task(decorated_func())
+                raise error from error
+
+        return decorated_func
+
+    def _create_task(self, coro: Coroutine[Any, Any, Any]) -> None:
+        task = self._loop.create_task(coro)
+        self._tasks.add(task)
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
-    with suppress(asyncio.CancelledError):
-        async with asyncio.TaskGroup() as tasks:
-            for coroutine in cast(FastAPIAppCoroutines, app.state.coroutines):
-                tasks.create_task(coroutine)
+    async with LefespanBackground() as background:
+        for func in cast(FastAPIAppBackground, app.state.background):
+            background.add(func)
 
-            yield
+        yield
 
-            await app.state.dishka_container.close()
-            raise asyncio.CancelledError
+        await app.state.dishka_container.close()
 
 
 async def app_from(container: AsyncContainer) -> FastAPI:
@@ -51,7 +93,7 @@ async def app_from(container: AsyncContainer) -> FastAPI:
         docs_url="/",
     )
 
-    app.state.coroutines = await container.get(FastAPIAppCoroutines)
+    app.state.background = await container.get(FastAPIAppBackground)
     routers = await container.get(FastAPIAppRouters)
 
     for router in routers: