Test PostgreSQL cancellation, transactions and pool recovery across event loops

.github/workflows/async-ecosystem.yml

@@ -65,3 +65,52 @@ jobs:
           uv run --no-sync python -c "import aiohttp, aiosqlite, anyio, httpx, sqlalchemy, websockets; from importlib.metadata import version; print({p: version(p) for p in ('aiohttp', 'aiosqlite', 'anyio', 'httpx', 'sqlalchemy', 'websockets')})"
       - name: Compare real application lifecycles across event loops
         run: uv run --no-sync python scripts/run_python_tests.py tests/ecosystem -m ecosystem
+
+  postgresql:
+    name: PostgreSQL / Linux / Python ${{ matrix.python }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    strategy:
+      fail-fast: false
+      matrix:
+        python: ["3.10", "3.14"]
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: rsloop
+          POSTGRES_PASSWORD: rsloop
+          POSTGRES_DB: rsloop_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U rsloop -d rsloop_test"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      RSLOOP_POSTGRES_DSN: postgresql://rsloop:rsloop@127.0.0.1:5432/rsloop_test
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b
+        with:
+          python-version: ${{ matrix.python }}
+          cache-suffix: postgres-${{ matrix.python }}
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: nightly-2026-09-25
+      - uses: Swatinem/rust-cache@v2
+        with:
+          cache-targets: false
+      - name: Install locked clients and build rsloop
+        run: |
+          uv sync --locked --no-default-groups --group ecosystem --no-install-project
+          uv run --no-sync maturin develop --locked
+      - name: Require clients and record environment
+        run: |
+          uv run --no-sync python -c "import asyncpg, sqlalchemy, uvloop, platform; from importlib.metadata import version; print(platform.platform(), platform.python_version()); print({p: version(p) for p in ('asyncpg', 'sqlalchemy', 'uvloop')})"
+          docker exec ${{ job.services.postgres.id }} postgres --version
+      - name: Compare PostgreSQL lifecycles
+        run: uv run --no-sync python scripts/run_python_tests.py tests/ecosystem/test_postgresql.py -m ecosystem -ra

docs/async-ecosystem-tests.md

@@ -3,8 +3,8 @@
 The downstream suite exercises actual clients, protocols, SQLite worker threads,
 and event loops. It extends the existing framework smoke scripts beyond a single
 successful request. HTTP clients run against loopback servers; databases use
-temporary SQLite files. No public endpoint, database service, or credentials are
-required at test time.
+temporary SQLite files. The SQLite/HTTP suite requires no external service. PostgreSQL tests opt in
+with `RSLOOP_POSTGRES_DSN` pointing to an isolated disposable database.
 
 ## Selected projects and contracts
 
@@ -38,7 +38,7 @@ Contract sources:
 
 Existing Redis/hiredis and AnyIO CI jobs remain useful complements. Framework
 smokes for FastAPI, Starlette, Uvicorn and other packages remain under
-tests/integration/packages. This suite does not establish PostgreSQL, Redis
+tests/integration/packages. This suite does not establish Redis
 cluster, HTTP/2, proxy, or production load compatibility.
 
 ## Run the suite
@@ -84,3 +84,60 @@ HTTPX 0.28.1, HTTPcore 1.0.9, AnyIO 4.14.2, websockets 17.0.1, SQLAlchemy 2.0.52
 aiosqlite 0.22.1 and uvloop 0.23.0. These cases did not reproduce a new rsloop
 defect; no production implementation change was needed. See the PR checks for
 platform-specific execution results.
+
+## PostgreSQL contracts
+
+`tests/ecosystem/test_postgresql.py` adds 36 cases: asyncpg and SQLAlchemy
+AsyncEngine, each compared on asyncio, rsloop and uvloop. Transactions are
+interrupted by cancellation, timeout or application exceptions; rollback is
+verified from an independent connection before a new transaction commits.
+Concurrent transactions remain invisible until both are released by an event.
+A held limit-one pool guarantees exhaustion until acquisition times out;
+backend PIDs prove subsequent connection reuse.
+
+For in-flight interruption, an observer holds a PostgreSQL advisory lock and
+polls `pg_stat_activity` until the target backend reports a lock wait. Thus
+cancellation occurs during real database I/O, without arbitrary sleeps. The
+20-second scenario deadline bounds server-state polling. Pool timeouts expire
+while the sole slot is held; no scheduler speed assumption is required.
+
+TCP recovery terminates only the test backend with `pg_terminate_backend`
+while its query is blocked. This tests a real server-side socket close and
+replacement connection, not packet loss, a TCP RST proxy, or a full server
+restart. The test user must be able to inspect and terminate its own backends.
+Unique table names and advisory keys isolate cases; tables, locks, connections,
+pools and tasks are cleaned up. Use only a disposable test database.
+
+```bash
+export RSLOOP_POSTGRES_DSN=postgresql://rsloop:rsloop@127.0.0.1:5432/rsloop_test
+uv run --no-sync python scripts/run_python_tests.py tests/ecosystem/test_postgresql.py -m ecosystem -ra
+```
+
+A separate Linux GitHub Actions job provisions PostgreSQL 16 with a health
+check and runs Python 3.10/3.14 with all three loops. It requires imports and
+records Python, platform, client and server versions. Existing SQLite and
+HTTP/WebSocket OS jobs remain in place; without the DSN PostgreSQL cases skip.
+PostgreSQL execution on macOS/Windows is not claimed.
+
+Local validation for this addition: Linux, CPython 3.14.7, release rsloop build;
+66 existing ecosystem cases passed, 36 PostgreSQL cases skipped (no runnable
+PostgreSQL service; container UID restrictions prevent starting the downloaded
+server). Ruff and Pyright passed for the added test module. Skips do not establish PostgreSQL compatibility.
+
+[GitHub Actions run 38062646332](https://github.com/RustedBytes/rsloop/actions/runs/38062646332)
+executed the added PostgreSQL suite at code commit `40a9ad8`:
+
+| Actual environment | Result |
+| --- | --- |
+| Linux x86_64, CPython 3.10.22, PostgreSQL 16.15 | 36 passed (asyncio, rsloop, uvloop) |
+| Linux x86_64, CPython 3.14.8, PostgreSQL 16.15 | 36 passed (asyncio, rsloop, uvloop) |
+
+Both jobs used asyncpg 0.32.0, SQLAlchemy 2.0.52 and uvloop 0.23.0.
+These scenarios reproduced no rsloop implementation defect; no production-code
+change was needed. PostgreSQL ran in CI, not locally. This is lifecycle
+compatibility evidence under the listed conditions, not a production-load test.
+
+Primary contracts: [asyncpg pools and transactions](https://magicstack.github.io/asyncpg/current/api/index.html),
+[SQLAlchemy asyncio](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html),
+[PostgreSQL activity monitoring](https://www.postgresql.org/docs/16/monitoring-stats.html),
+[backend termination](https://www.postgresql.org/docs/16/functions-admin.html).

pyproject.toml

@@ -79,6 +79,7 @@ integration = [
 ]
 # Reproducible downstream lifecycle tests, separate from framework smoke scripts.
 ecosystem = [
+  "asyncpg>=0.31.0",
   { include-group = "test" },
   "maturin>=1.14.1",
   "aiohttp>=3.14.3",

tests/ecosystem/test_postgresql.py

@@ -0,0 +1,257 @@
+"""Real PostgreSQL contracts; opt in with a dedicated RSLOOP_POSTGRES_DSN."""
+
+import asyncio
+import os
+import uuid
+from contextlib import asynccontextmanager
+
+import pytest
+
+asyncpg = pytest.importorskip("asyncpg")
+sa = pytest.importorskip("sqlalchemy")
+sa_async = pytest.importorskip("sqlalchemy.ext.asyncio")
+pytestmark = pytest.mark.ecosystem
+
+
+@pytest.fixture
+def postgres_dsn():
+    dsn = os.environ.get("RSLOOP_POSTGRES_DSN")
+    if not dsn:
+        pytest.skip("set RSLOOP_POSTGRES_DSN to an isolated test database")
+    return dsn
+
+
+class Client:
+    """Same observable contracts through each driver's public API."""
+
+    def __init__(self, dsn, driver, size=1):
+        self.dsn, self.driver, self.size = dsn, driver, size
+
+    async def open(self):
+        if self.driver == "asyncpg":
+            self.pool = await asyncpg.create_pool(
+                self.dsn, min_size=self.size, max_size=self.size
+            )
+        else:
+            self.pool = sa_async.create_async_engine(
+                self.dsn.replace("postgresql://", "postgresql+asyncpg://", 1),
+                pool_size=self.size,
+                max_overflow=0,
+                pool_timeout=0.05,
+            )
+        return self
+
+    @asynccontextmanager
+    async def connection(self):
+        if self.driver == "asyncpg":
+            async with self.pool.acquire() as connection:
+                yield connection
+        else:
+            async with self.pool.connect() as connection:
+                yield connection
+
+    async def scalar(self, connection, query):
+        if self.driver == "asyncpg":
+            return await connection.fetchval(query)
+        return await connection.scalar(sa.text(query))
+
+    def transaction(self, connection):
+        if self.driver == "asyncpg":
+            return connection.transaction()
+        return connection.begin()
+
+    async def close(self):
+        if self.driver == "asyncpg":
+            await asyncio.wait_for(self.pool.close(), 5)
+        else:
+            await asyncio.wait_for(self.pool.dispose(), 5)
+
+
+async def wait_for_lock(observer, pid):
+    # Observe the server state, not an assumed duration of query execution.
+    while not await observer.fetchval(
+        "SELECT EXISTS (SELECT FROM pg_stat_activity "
+        "WHERE pid=$1 AND wait_event_type='Lock')",
+        pid,
+    ):
+        await asyncio.sleep(0)
+
+
+@pytest.mark.parametrize("driver", ["asyncpg", "sqlalchemy"])
+@pytest.mark.parametrize("interruption", ["cancel", "timeout", "exception"])
+def test_postgres_interrupted_transaction(loop, postgres_dsn, driver, interruption):
+    async def main():
+        client = await Client(postgres_dsn, driver).open()
+        observer = await asyncpg.connect(postgres_dsn)
+        table = "rsloop_" + uuid.uuid4().hex
+        key = uuid.uuid4().int % (2**63)
+        task = None
+        pid_ready = asyncio.Future()
+        try:
+            await observer.execute(f"CREATE TABLE {table} (value integer)")
+            await observer.execute(f"SELECT pg_advisory_lock({key})")
+
+            async def transaction():
+                async with (
+                    client.connection() as connection,
+                    client.transaction(connection),
+                ):
+                    await client.scalar(
+                        connection,
+                        f"INSERT INTO {table} VALUES (1) RETURNING value",
+                    )
+                    pid_ready.set_result(
+                        await client.scalar(connection, "SELECT pg_backend_pid()")
+                    )
+                    if interruption == "exception":
+                        raise ValueError("application failure")
+                    await client.scalar(
+                        connection, f"SELECT pg_advisory_xact_lock({key})"
+                    )
+
+            task = asyncio.create_task(transaction())
+            pid = await pid_ready
+            if interruption == "exception":
+                with pytest.raises(ValueError, match="application failure"):
+                    await task
+            else:
+                await wait_for_lock(observer, pid)
+                if interruption == "cancel":
+                    task.cancel()
+                    with pytest.raises(asyncio.CancelledError):
+                        await task
+                else:
+                    with pytest.raises(asyncio.TimeoutError):
+                        await asyncio.wait_for(task, 0)
+            assert await observer.fetchval(f"SELECT count(*) FROM {table}") == 0
+            # Acquiring the only slot proves pool return. Query proves protocol recovery.
+            async with (
+                client.connection() as connection,
+                client.transaction(connection),
+            ):
+                assert (
+                    await client.scalar(
+                        connection,
+                        f"INSERT INTO {table} VALUES (2) RETURNING value",
+                    )
+                    == 2
+                )
+            assert await observer.fetchval(f"SELECT sum(value) FROM {table}") == 2
+        finally:
+            if task is not None:
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+            await observer.execute("SELECT pg_advisory_unlock_all()")
+            await client.close()
+            await observer.execute(f"DROP TABLE IF EXISTS {table}")
+            await observer.close()
+
+    loop.run_until_complete(asyncio.wait_for(main(), 20))
+
+
+@pytest.mark.parametrize("driver", ["asyncpg", "sqlalchemy"])
+def test_postgres_pool_exhaustion_and_reuse(loop, postgres_dsn, driver):
+    async def main():
+        client = await Client(postgres_dsn, driver).open()
+        try:
+            async with client.connection() as held:
+                pid = await client.scalar(held, "SELECT pg_backend_pid()")
+                if driver == "asyncpg":
+                    with pytest.raises(asyncio.TimeoutError):
+                        await client.pool.acquire(timeout=0.05)
+                else:
+                    with pytest.raises(sa.exc.TimeoutError):
+                        async with client.connection():
+                            pytest.fail("exhausted pool returned a connection")
+                assert await client.scalar(held, "SELECT 42") == 42
+            async with client.connection() as reused:
+                assert await client.scalar(reused, "SELECT pg_backend_pid()") == pid
+        finally:
+            await client.close()
+
+    loop.run_until_complete(asyncio.wait_for(main(), 20))
+
+
+@pytest.mark.parametrize("driver", ["asyncpg", "sqlalchemy"])
+def test_postgres_concurrent_transactions(loop, postgres_dsn, driver):
+    async def main():
+        client = await Client(postgres_dsn, driver, size=2).open()
+        observer = await asyncpg.connect(postgres_dsn)
+        table = "rsloop_" + uuid.uuid4().hex
+        ready = [asyncio.Event(), asyncio.Event()]
+        release = asyncio.Event()
+        tasks = []
+        try:
+            await observer.execute(f"CREATE TABLE {table} (value integer)")
+
+            async def writer(index):
+                async with (
+                    client.connection() as connection,
+                    client.transaction(connection),
+                ):
+                    await client.scalar(
+                        connection,
+                        f"INSERT INTO {table} VALUES ({index}) RETURNING value",
+                    )
+                    ready[index].set()
+                    await release.wait()
+
+            tasks = [asyncio.create_task(writer(i)) for i in range(2)]
+            await asyncio.gather(*(event.wait() for event in ready))
+            assert await observer.fetchval(f"SELECT count(*) FROM {table}") == 0
+            release.set()
+            await asyncio.gather(*tasks)
+            assert await observer.fetchval(
+                f"SELECT array_agg(value ORDER BY value) FROM {table}"
+            ) == [0, 1]
+        finally:
+            release.set()
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            await client.close()
+            await observer.execute(f"DROP TABLE IF EXISTS {table}")
+            await observer.close()
+
+    loop.run_until_complete(asyncio.wait_for(main(), 20))
+
+
+@pytest.mark.parametrize("driver", ["asyncpg", "sqlalchemy"])
+def test_postgres_tcp_disconnect_and_pool_recovery(loop, postgres_dsn, driver):
+    async def main():
+        client = await Client(postgres_dsn, driver).open()
+        observer = await asyncpg.connect(postgres_dsn)
+        key = uuid.uuid4().int % (2**63)
+        task = None
+        try:
+            await observer.execute(f"SELECT pg_advisory_lock({key})")
+            async with client.connection() as connection:
+                pid = await client.scalar(connection, "SELECT pg_backend_pid()")
+                task = asyncio.create_task(
+                    client.scalar(connection, f"SELECT pg_advisory_xact_lock({key})")
+                )
+                await wait_for_lock(observer, pid)
+                # Kill only our backend: the real server closes the TCP socket
+                # while the client is receiving a response. No private transport API.
+                assert await observer.fetchval("SELECT pg_terminate_backend($1)", pid)
+                expected = (
+                    asyncpg.PostgresError if driver == "asyncpg" else sa.exc.DBAPIError
+                )
+                with pytest.raises(expected):
+                    await task
+                if driver == "sqlalchemy":
+                    await connection.rollback()
+            async with client.connection() as replacement:
+                assert (
+                    await client.scalar(replacement, "SELECT pg_backend_pid()") != pid
+                )
+                assert await client.scalar(replacement, "SELECT 42") == 42
+        finally:
+            if task is not None:
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+            await observer.execute("SELECT pg_advisory_unlock_all()")
+            await client.close()
+            await observer.close()
+
+    loop.run_until_complete(asyncio.wait_for(main(), 20))