fix(db): require asyncpg>=0.32 so a dropped connection can't wedge the worker

hindsight-api-slim/pyproject.toml

@@ -10,10 +10,12 @@ license = "MIT"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-    # 0.30.0 added Connection.get_reset_query(), the only supported way to skip
-    # asyncpg's release-time reset (see engine/db/postgresql.py). On 0.29 the
-    # override would be silently ignored, making the flag a no-op.
-    "asyncpg>=0.30.0",
+    # 0.32.0: before it, a query that timed out on a connection that then dropped
+    # left every later query on that connection (the transaction's ROLLBACK
+    # included) awaiting forever, which wedged the worker's claim loop (#6067).
+    # 0.30.0 also added Connection.get_reset_query(), which the pool's
+    # release-time reset skip relies on (see engine/db/postgresql.py).
+    "asyncpg>=0.32.0",
     "python-dotenv>=1.0.0",
     "openai>=1.66.0",  # Responses API (client.responses.create, reasoning=, text.format) for the openai-responses provider
     "pydantic>=2.0.0",

hindsight-api-slim/tests/test_db_dropped_connection.py

@@ -0,0 +1,94 @@
+"""A transaction whose query timed out must fail, not hang, when its connection drops.
+
+asyncpg <= 0.31 left a timed-out query's cancellation pending until the server
+answered it. If the connection died first, nothing ever resolved it, and every
+later query on that connection awaited forever — including the ROLLBACK that
+``transaction()`` sends on the way out. That is how the worker's claim loop
+wedged silently while /health stayed green (#6067); asyncpg 0.32 fixed it.
+
+This drives our ``PostgresConnection.transaction()`` through a local proxy that
+swallows the server's replies, then cuts the connection, so a lock-file
+downgrade below 0.32 fails here instead of in production.
+"""
+
+import asyncio
+from urllib.parse import urlparse
+
+import asyncpg
+import pytest
+
+from hindsight_api.engine.db.postgresql import PostgresConnection
+
+
+class _BlackholeProxy:
+    """TCP proxy to Postgres that can stop forwarding replies, then drop the client."""
+
+    def __init__(self, pg_host: str, pg_port: int) -> None:
+        self._pg_host = pg_host
+        self._pg_port = pg_port
+        self._client: asyncio.StreamWriter | None = None
+        self.swallow_replies = False
+
+    async def start(self) -> int:
+        self._server = await asyncio.start_server(self._handle, "127.0.0.1", 0)
+        return self._server.sockets[0].getsockname()[1]
+
+    def drop_client(self) -> None:
+        assert self._client is not None
+        self._client.transport.abort()
+
+    async def close(self) -> None:
+        self._server.close()
+        await self._server.wait_closed()
+
+    async def _pipe(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, replies: bool) -> None:
+        try:
+            while data := await reader.read(65536):
+                if replies and self.swallow_replies:
+                    continue
+                writer.write(data)
+                await writer.drain()
+        except (ConnectionError, OSError):
+            pass
+        finally:
+            writer.close()
+
+    async def _handle(self, client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter) -> None:
+        # Only the first client is the session; asyncpg opens a second, short-lived
+        # connection to send the cancel request for the timed-out query.
+        if self._client is None:
+            self._client = client_writer
+        server_reader, server_writer = await asyncio.open_connection(self._pg_host, self._pg_port)
+        await asyncio.gather(
+            self._pipe(client_reader, server_writer, replies=False),
+            self._pipe(server_reader, client_writer, replies=True),
+        )
+
+
+@pytest.mark.asyncio
+async def test_timed_out_query_on_dropped_connection_fails_the_transaction(pg0_db_url):
+    url = urlparse(pg0_db_url)
+    proxy = _BlackholeProxy(url.hostname or "localhost", url.port or 5432)
+    proxy_port = await proxy.start()
+    raw = await asyncpg.connect(pg0_db_url, host="127.0.0.1", port=proxy_port, ssl=False)
+    conn = PostgresConnection(raw)
+
+    async def claim() -> None:
+        async with conn.transaction():
+            proxy.swallow_replies = True
+            asyncio.get_running_loop().call_later(1.0, proxy.drop_client)
+            await conn.execute("SELECT pg_sleep(5)", timeout=0.3)
+
+    task = asyncio.create_task(claim())
+    try:
+        # Before the fix the ROLLBACK never returns; bound the wait so that hang
+        # fails the test instead of stalling the run.
+        done, _ = await asyncio.wait({task}, timeout=10)
+        assert task in done, "transaction is stuck in ROLLBACK on a dropped connection"
+        with pytest.raises((asyncpg.InterfaceError, asyncpg.ConnectionDoesNotExistError)):
+            task.result()
+        assert raw.is_closed()
+    finally:
+        task.cancel()
+        raw.terminate()
+        await proxy.close()

hindsight-docs/docs/developer/configuration.mdx

@@ -92,7 +92,7 @@ A few things to know:
 
 | Variable | Description | Default |
 |----------|-------------|---------|
-| `HINDSIGHT_API_DB_POOL_MIN_SIZE` | Minimum connections in the primary pool | `5` |
+| `HINDSIGHT_API_DB_POOL_MIN_SIZE` | Minimum connections in the primary pool. The pool keeps this many open even when idle and replaces any that close; `0` lets it drain when idle. | `5` |
 | `HINDSIGHT_API_DB_POOL_MAX_SIZE` | Maximum connections in the primary pool | `100` |
 | `HINDSIGHT_API_READ_DB_POOL_MIN_SIZE` | Minimum connections in the read-replica pool (only used when `READ_DATABASE_URL` is set) | Falls back to `DB_POOL_MIN_SIZE` |
 | `HINDSIGHT_API_READ_DB_POOL_MAX_SIZE` | Maximum connections in the read-replica pool (only used when `READ_DATABASE_URL` is set) | Falls back to `DB_POOL_MAX_SIZE` |