From 16b2a121c0176c24048502da956861237180310d Mon Sep 17 00:00:00 2001 From: Paulo Alvarado Date: Tue, 15 Sep 2026 15:36:45 +0200 Subject: [PATCH] Keep the exchange client open until the push timer ends The lifespan cancelled the push timer and closed the httpx client at once. A round still under way then met a closed client, the timer died, and no push ran until the next restart. Now the lifespan tells the timer to stop, waits for its round, and only then closes the client. A push that raises anything is logged with its host and traceback, and the timer runs on. Co-Authored-By: Claude Fable 5.1 --- src/secrets_exchange/app.py | 24 ++++----- src/secrets_exchange/tests/test_app.py | 68 +++++++++++++++++++++++++- 2 files changed, 78 insertions(+), 14 deletions(-) diff --git a/src/secrets_exchange/app.py b/src/secrets_exchange/app.py index b4e650d..e28847d 100644 --- a/src/secrets_exchange/app.py +++ b/src/secrets_exchange/app.py @@ -2,7 +2,7 @@ import logging import uuid from collections.abc import AsyncIterator -from contextlib import asynccontextmanager +from contextlib import asynccontextmanager, suppress from datetime import timedelta from typing import Annotated @@ -28,29 +28,27 @@ @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: + """The client closes after the timer ends its round.""" async with httpx.AsyncClient(timeout=10) as client: app.state.secrets = Secrets(client) - timer = asyncio.create_task(push_on_expiry(app.state.secrets)) - timer.add_done_callback(log_stop) + stop = asyncio.Event() + timer = asyncio.create_task(push_on_expiry(app.state.secrets, stop)) try: yield finally: - timer.cancel() + stop.set() + await timer -def log_stop(timer: asyncio.Task[None]) -> None: - if not timer.cancelled() and (failure := timer.exception()): - logger.error("the push timer stopped", exc_info=failure) - - -async def push_on_expiry(secrets: Secrets) -> None: +async def push_on_expiry(secrets: Secrets, stop: asyncio.Event) -> None: """Proxy providers are not visited. Their value refreshes on request.""" - while True: + while not stop.is_set(): try: await push_active_hosts(secrets) except SQLAlchemyError as exc: logger.warning("the hosts could not be read: %s", exc) - await asyncio.sleep(TICK.total_seconds()) + with suppress(TimeoutError): + await asyncio.wait_for(stop.wait(), TICK.total_seconds()) async def push_active_hosts(secrets: Secrets) -> None: @@ -70,6 +68,8 @@ async def push_to_host(secrets: Secrets, host: Host) -> None: await secrets.push(host, service, entry) except (ProviderError, SecretDecryptError) as exc: logger.error("push for host %s failed: %s", host.name, exc) + except Exception: + logger.exception("push for host %s failed", host.name) app = FastAPI(title="Drukbox secrets exchange", lifespan=lifespan) diff --git a/src/secrets_exchange/tests/test_app.py b/src/secrets_exchange/tests/test_app.py index 67af0c9..24d1cfe 100644 --- a/src/secrets_exchange/tests/test_app.py +++ b/src/secrets_exchange/tests/test_app.py @@ -1,3 +1,4 @@ +import asyncio import base64 import logging import uuid @@ -17,8 +18,14 @@ from hosts.tests.conftest import stub_provider # noqa: F401 from providers.exceptions import ProviderTransportError from providers.registry import get_vm_provider -from secrets_exchange.app import UPSTREAM_CREDENTIAL, UPSTREAM_HEADER, app, push_active_hosts -from secrets_exchange.secrets import Secrets +from secrets_exchange.app import ( + UPSTREAM_CREDENTIAL, + UPSTREAM_HEADER, + app, + lifespan, + push_active_hosts, +) +from secrets_exchange.secrets import Secret, Secrets ISSUER = {"url": "https://mint.test/box/anthropic", "headers": {}, "refresh": "1h"} @@ -309,6 +316,63 @@ async def test_one_host_in_trouble_costs_no_other_host_its_value(edge, caplog) - assert f"sb-{troubled.hex[:12]}" in caplog.text +@respx.mock +@pytest.mark.usefixtures("stub_provider") +async def test_a_push_that_fails_unexpectedly_costs_no_other_host_its_value(edge, caplog) -> None: + troubled, healthy = uuid.uuid4(), uuid.uuid4() + + async def push_secret(*, vm: str, name: str, value: str) -> None: + if vm == f"sb-{troubled.hex[:12]}": + raise RuntimeError("the value files went away") + + secrets = MagicMock(needs_value=True, push_secret=AsyncMock(side_effect=push_secret)) + get_vm_provider("stub").secrets = secrets + entry = {"issuer": ISSUER, "placeholder_fingerprint": "a"} + await _create_host(troubled, {"anthropic": entry}, provider="stub") + await _create_host(healthy, {"anthropic": entry}, provider="stub") + respx.get(ISSUER["url"]).respond(json={"value": "sk-ant-fresh"}) + + with caplog.at_level(logging.ERROR): + await push_active_hosts(app.state.secrets) + + secrets.push_secret.assert_any_await( + vm=f"sb-{healthy.hex[:12]}", name="anthropic", value="sk-ant-fresh" + ) + assert f"push for host sb-{troubled.hex[:12]} failed" in caplog.text + assert "the value files went away" in caplog.text + + +@respx.mock +@pytest.mark.usefixtures("stub_provider") +async def test_shutdown_waits_for_the_round_under_way_before_the_client_closes( + monkeypatch, +) -> None: + secrets = MagicMock(needs_value=True, push_secret=AsyncMock()) + get_vm_provider("stub").secrets = secrets + host_id = uuid.uuid4() + await _create_host( + host_id, {"anthropic": {"issuer": ISSUER, "placeholder_fingerprint": "a"}}, provider="stub" + ) + respx.get(ISSUER["url"]).respond(json={"value": "sk-ant-fresh"}) + fetch = Secret.fetch + fetching, release = asyncio.Event(), asyncio.Event() + + async def paused_fetch(cls, issuer, client): + fetching.set() + await release.wait() + return await fetch(issuer, client) + + monkeypatch.setattr(Secret, "fetch", classmethod(paused_fetch)) + + async with lifespan(app): + await asyncio.wait_for(fetching.wait(), 1) + asyncio.get_running_loop().call_soon(release.set) + + secrets.push_secret.assert_awaited_once_with( + vm=f"sb-{host_id.hex[:12]}", name="anthropic", value="sk-ant-fresh" + ) + + @respx.mock @pytest.mark.usefixtures("stub_provider") async def test_a_refresh_order_replaces_the_held_value_at_once(edge) -> None: