fix: startup

Author: giancarloromeo

packages/celery-library/src/celery_library/signals.py

@@ -24,16 +24,19 @@ def on_worker_init(
     sender: WorkController,
     **_kwargs,
 ) -> None:
-    def _init() -> None:
+    startup_complete_event = threading.Event()
+
+    def _init(startup_complete_event: threading.Event) -> None:
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
 
         app_server.event_loop = loop
 
-        async def setup_task_manager():
+        async def _setup():
             assert sender.app  # nosec
             assert isinstance(sender.app, Celery)  # nosec
 
+            set_app_server(sender.app, app_server)
             set_task_manager(
                 sender.app,
                 create_task_manager(
@@ -42,18 +45,20 @@ async def setup_task_manager():
                 ),
             )
 
-        set_app_server(sender.app, app_server)
-        loop.run_until_complete(setup_task_manager())
-        loop.run_until_complete(app_server.startup())
+        loop.run_until_complete(_setup())
+        loop.run_until_complete(app_server.startup(startup_complete_event))
 
     thread = threading.Thread(
         group=None,
         target=_init,
         name="app_server_init",
+        args=(startup_complete_event,),
         daemon=True,
     )
     thread.start()
 
+    startup_complete_event.wait()
+
 
 def on_worker_shutdown(sender, **_kwargs) -> None:
     with log_context(_logger, logging.INFO, "Worker shutdown"):

packages/service-library/src/servicelib/base_app_server.py

@@ -1,3 +1,4 @@
+import threading
 from abc import ABC, abstractmethod
 from asyncio import AbstractEventLoop
 from contextlib import suppress
@@ -20,7 +21,7 @@ def aiohttp_app(self) -> "Application":
         raise NotImplementedError
 
     @abstractmethod
-    async def startup(self):
+    async def startup(self, completed: threading.Event):
         pass
 
     @property

packages/service-library/src/servicelib/fastapi/app_server.py

@@ -22,13 +22,14 @@ def fastapi_app(self) -> FastAPI:
         assert isinstance(self._app, FastAPI)  # nosec
         return self._app
 
-    async def startup(self):
+    async def startup(self, completed: asyncio.Event):
         self._lifespan_manager = LifespanManager(
             self.fastapi_app,
             startup_timeout=_STARTUP_TIMEOUT,
             shutdown_timeout=_SHUTDOWN_TIMEOUT,
         )
         await self._lifespan_manager.__aenter__()
+        completed.set()
         await self._shutdown_event.wait()
 
     async def shutdown(self):