diff --git a/src/sturnus/entrypoints/link.py b/src/sturnus/entrypoints/link.py index 8f6e0b5..ac85b0a 100644 --- a/src/sturnus/entrypoints/link.py +++ b/src/sturnus/entrypoints/link.py @@ -98,7 +98,6 @@ async def _run() -> None: engine = create_async_engine(settings.database_url) session_factory = async_sessionmaker(engine, expire_on_commit=False) - await _wait_for_schema(engine) states = LinkStateStore(session_factory) links = AccountLinkRepository(session_factory) @@ -109,11 +108,22 @@ async def _run() -> None: redirect_uri=settings.outline_redirect_uri, ) + # The wait for the worker's migrations happens *after* this server is + # listening, not before. Waiting first leaves the health port closed for + # as long as the wait takes, and the liveness probe kills the pod while + # it is doing exactly what it should -- which is what happened on the + # first deployment, where the tables did not exist until the worker had + # run. `/healthz` now answers immediately and `/readyz` reports 503 + # until the schema is there, so Kubernetes holds traffic back without + # restarting anything. + schema_ready = False + app = build_app( oauth=oauth, states=states, links=links, now=lambda: datetime.now(UTC), + schema_ready=lambda: schema_ready, ) runner = web.AppRunner(app) @@ -126,6 +136,10 @@ async def _run() -> None: await site.start() log.info("Link service listening on port %d", settings.health_port) + await _wait_for_schema(engine) + schema_ready = True + log.info("Database schema is present; ready to serve account links") + stop = asyncio.Event() loop = asyncio.get_running_loop() for sig in (signal.SIGTERM, signal.SIGINT): diff --git a/src/sturnus/infrastructure/linkserver.py b/src/sturnus/infrastructure/linkserver.py index b3f107c..cf17a3b 100644 --- a/src/sturnus/infrastructure/linkserver.py +++ b/src/sturnus/infrastructure/linkserver.py @@ -105,11 +105,17 @@ def build_app( states: StateStore, links: LinkRepository, now: Callable[[], datetime], + schema_ready: Callable[[], bool], ) -> web.Application: """Builds the aiohttp application the link deployment serves. `now` is injected rather than read from the wall clock directly so a test can pin it -- the same reason `SystemClock` exists for the bot. + + `schema_ready` reports whether the database tables the worker creates + have appeared yet. The caller starts this server *before* waiting for + them, so `/healthz` answers from the first moment while `/readyz` stays + 503 until the wait finishes -- see `sturnus.entrypoints.link`. """ async def healthz(_request: web.Request) -> web.Response: @@ -119,10 +125,12 @@ async def healthz(_request: web.Request) -> web.Response: return web.json_response({"status": "ok"}) async def readyz(_request: web.Request) -> web.Response: - # This process holds no connection of its own beyond `states` and - # `links`, both backed by the same database the callback route - # already exercises on every real request; there is no separate - # dependency to probe here that `/healthz` does not already cover. + # Beyond the schema, this process holds no connection of its own: + # `states` and `links` are backed by the same database the callback + # route exercises on every real request, so there is nothing further + # to probe that `/healthz` does not already cover. + if not schema_ready(): + return web.json_response({"status": "waiting for database schema"}, status=503) return web.json_response({"status": "ready"}) async def oauth_callback(request: web.Request) -> web.Response: diff --git a/tests/infrastructure/test_linkserver.py b/tests/infrastructure/test_linkserver.py index c07c596..360ce51 100644 --- a/tests/infrastructure/test_linkserver.py +++ b/tests/infrastructure/test_linkserver.py @@ -83,7 +83,13 @@ async def save(self, discord_user_id: int, provider: str, external_id: str, name @pytest.fixture async def client(aiohttp_client: AiohttpClientFactory) -> TestClient[web.Request, web.Application]: return await aiohttp_client( - build_app(oauth=FakeOAuth(), states=FakeStates(), links=FakeLinks(), now=lambda: T0) + build_app( + oauth=FakeOAuth(), + states=FakeStates(), + links=FakeLinks(), + now=lambda: T0, + schema_ready=lambda: True, + ) ) @@ -94,7 +100,13 @@ async def test_healthz_is_served(client: TestClient[web.Request, web.Application async def test_a_valid_callback_stores_the_link(aiohttp_client: AiohttpClientFactory) -> None: links = FakeLinks() c = await aiohttp_client( - build_app(oauth=FakeOAuth(), states=FakeStates(), links=links, now=lambda: T0) + build_app( + oauth=FakeOAuth(), + states=FakeStates(), + links=links, + now=lambda: T0, + schema_ready=lambda: True, + ) ) response = await c.get("/oauth/callback", params={"code": "c", "state": "good-state"}) assert response.status == 200 @@ -107,7 +119,13 @@ async def test_an_unknown_state_is_refused_and_stores_nothing( """A forged callback must not link anything.""" links = FakeLinks() c = await aiohttp_client( - build_app(oauth=FakeOAuth(), states=FakeStates(), links=links, now=lambda: T0) + build_app( + oauth=FakeOAuth(), + states=FakeStates(), + links=links, + now=lambda: T0, + schema_ready=lambda: True, + ) ) response = await c.get("/oauth/callback", params={"code": "c", "state": "forged"}) assert response.status == 400 @@ -117,7 +135,13 @@ async def test_an_unknown_state_is_refused_and_stores_nothing( async def test_a_replayed_state_is_refused(aiohttp_client: AiohttpClientFactory) -> None: links = FakeLinks() c = await aiohttp_client( - build_app(oauth=FakeOAuth(), states=FakeStates(), links=links, now=lambda: T0) + build_app( + oauth=FakeOAuth(), + states=FakeStates(), + links=links, + now=lambda: T0, + schema_ready=lambda: True, + ) ) params = {"code": "c", "state": "good-state"} assert (await c.get("/oauth/callback", params=params)).status == 200 @@ -137,7 +161,13 @@ async def test_a_failed_exchange_reports_an_error_and_stores_nothing( ) -> None: links = FakeLinks() c = await aiohttp_client( - build_app(oauth=FakeOAuth(fail=True), states=FakeStates(), links=links, now=lambda: T0) + build_app( + oauth=FakeOAuth(fail=True), + states=FakeStates(), + links=links, + now=lambda: T0, + schema_ready=lambda: True, + ) ) response = await c.get("/oauth/callback", params={"code": "c", "state": "good-state"}) assert response.status >= 400 @@ -147,10 +177,59 @@ async def test_a_failed_exchange_reports_an_error_and_stores_nothing( async def test_the_error_page_does_not_echo_the_input(aiohttp_client: AiohttpClientFactory) -> None: """Reflecting user input into HTML is how a callback becomes an XSS sink.""" c = await aiohttp_client( - build_app(oauth=FakeOAuth(), states=FakeStates(), links=FakeLinks(), now=lambda: T0) + build_app( + oauth=FakeOAuth(), + states=FakeStates(), + links=FakeLinks(), + now=lambda: T0, + schema_ready=lambda: True, + ) ) response = await c.get( "/oauth/callback", params={"code": "c", "state": ""} ) body = await response.text() assert "