Test real HTTP, WebSocket and database async lifecycles across event loops

.github/workflows/async-ecosystem.yml

@@ -0,0 +1,67 @@
+name: Async ecosystem
+
+on:
+  pull_request:
+    paths:
+      - "Cargo.toml"
+      - "Cargo.lock"
+      - "build.rs"
+      - "src/**"
+      - "python/**"
+      - "tests/ecosystem/**"
+      - "tests/conftest.py"
+      - "scripts/generate_test_tls_certs.py"
+      - "scripts/run_python_tests.py"
+      - "pyproject.toml"
+      - "uv.lock"
+      - ".github/workflows/async-ecosystem.yml"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: async-ecosystem-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  lifecycle:
+    name: ecosystem / ${{ matrix.os }} / Python ${{ matrix.python }}
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 20
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - os: ubuntu-latest
+            python: "3.10"
+          - os: ubuntu-latest
+            python: "3.14"
+          - os: macos-15
+            python: "3.14"
+          - os: windows-latest
+            python: "3.14"
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b
+        with:
+          python-version: ${{ matrix.python }}
+          cache-suffix: ecosystem-${{ matrix.os }}-${{ 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 downstream dependencies and build rsloop
+        run: |
+          uv sync --locked --no-default-groups --group ecosystem --no-install-project
+          uv run --no-sync maturin develop --locked
+          uv run --no-sync python scripts/generate_test_tls_certs.py
+      - name: Record versions and require downstream imports
+        run: |
+          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

docs/async-ecosystem-tests.md

@@ -0,0 +1,86 @@
+# Async application lifecycle tests
+
+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.
+
+## Selected projects and contracts
+
+The selection targets common integration surfaces in Python async applications:
+HTTP pools, streaming protocols, cancellation scopes, database transactions,
+and cross-thread result delivery. It builds on dependencies already used by the
+repository's integration scripts and adds direct HTTPX coverage.
+
+| Project | Application scenario | Assertions |
+| --- | --- | --- |
+| HTTPX / HTTPcore / AnyIO | Cancel or time out an HTTP/HTTPS download while the only pool connection is occupied | Stream closes; the same client can open a replacement connection and complete another request |
+| HTTPX | Pool exhaustion while another response is held open | PoolTimeout; no extra server connection; completing the first response permits keep-alive reuse |
+| aiohttp | Cancel or time out a request queued behind a limit-one connector | The queued request never reaches the server; the active response remains readable; the connection is reused |
+| HTTPX and aiohttp | Streaming upload and response with Unicode bytes split across arbitrary chunks | Exact payload preservation in both directions |
+| HTTPX and aiohttp | Upstream closes before Content-Length bytes arrive | Library-specific truncation error; a subsequent request succeeds |
+| websockets | Cancel recv while a fragmented message is incomplete, with and without compression | The next recv returns the entire message; subsequent binary messages remain ordered |
+| websockets | Concurrent receivers and graceful close | Second receiver is rejected; the pending receiver observes normal closure |
+| SQLAlchemy / aiosqlite | Cancel, time out, or raise after an insert in a transaction | Rollback removes the uncommitted row; the single pool slot is returned; the next transaction commits |
+| SQLAlchemy / aiosqlite | Concurrent independent sessions stream rows through a two-connection pool | Complete ordered results; cursors and pool slots released |
+| aiosqlite | Cancel a query while its SQLite worker is still executing | Late cross-thread completion does not corrupt the cancelled Future or the next query |
+
+Contract sources:
+
+- [HTTPX async streaming and closing](https://www.python-httpx.org/async/)
+- [HTTPX timeouts](https://www.python-httpx.org/advanced/timeouts/)
+- [aiohttp pool lifecycle](https://docs.aiohttp.org/en/stable/client_advanced.html)
+- [aiohttp queue tracing](https://docs.aiohttp.org/en/stable/tracing_reference.html)
+- [websockets recv cancellation and concurrency](https://websockets.readthedocs.io/en/stable/reference/asyncio/connection.html)
+- [SQLAlchemy async sessions and connection disposal](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)
+- [aiosqlite worker-thread design](https://aiosqlite.omnilib.dev/en/stable/)
+
+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
+cluster, HTTP/2, proxy, or production load compatibility.
+
+## Run the suite
+
+From the repository root:
+
+```bash
+uv sync --locked --no-default-groups --group ecosystem --no-install-project
+uv run --no-sync maturin develop --locked
+uv run --no-sync python scripts/generate_test_tls_certs.py
+uv run --no-sync python scripts/run_python_tests.py tests/ecosystem -m ecosystem
+```
+
+The ecosystem marker is excluded from the default core test run. Its dedicated
+workflow installs the locked dependency group, requires all downstream imports,
+prints dependency versions and runs the suite on Linux/Python 3.10 and 3.14,
+macOS/Python 3.14, and Windows/Python 3.14. Every available platform compares
+asyncio with rsloop; Unix also compares uvloop. Windows intentionally skips the
+uvloop variants. Downstream native extension support is separate from rsloop's
+core free-threaded matrix.
+
+Optional HTTP/database modules skip collection when their dependencies are
+absent in a core-only environment. The dedicated workflow's import check makes
+missing dependencies a job failure rather than an apparently successful suite.
+
+## Test design
+
+Events establish the interrupted state before cancellation. aiohttp's public
+queue trace confirms a request is waiting for a pool slot. A WebSocket ping/pong
+round trip accompanies an incomplete fragmented message. A threading event holds
+a real SQLite worker until its caller has been cancelled. Timeout tests hold the
+resource until the timeout occurs; they do not rely on the scheduler beating a
+sleep of a particular duration.
+
+Network assertions operate on accumulated bytes, not TCP packet boundaries.
+Handlers propagate their failures to the parent test. Sessions, responses,
+cursors, sockets and engines are closed; task and loop-exception checks detect
+orphaned work. Per-scenario deadlines bound waits.
+
+Initial Linux validation on CPython 3.14.7 passed all 66 parameterized cases
+against asyncio, rsloop and uvloop with the locked dependencies: aiohttp 3.14.3,
+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.

docs/development.md

@@ -253,6 +253,10 @@ exercise prolonged pause deterministically without depending on kernel buffer
 sizes. Weak references check that cancelled waits are released before resume.
 Tests also check for leaked tasks and unhandled loop exceptions.
 
+For real HTTP, WebSocket and database cancellation/recovery scenarios, see
+[Async application lifecycle tests](async-ecosystem-tests.md). The optional
+ecosystem dependency group has a dedicated cross-platform CI matrix.
+
 CI and `just test` use `scripts/run_python_tests.py`, which forwards pytest
 arguments and prints recurring stack dumps when tests stop making progress.
 For the same diagnostics locally, run `uv run python scripts/run_python_tests.py`.

mkdocs.yml

@@ -17,4 +17,5 @@ nav:
   - How It Works: how-it-works.md
   - Project Structure: project-structure.md
   - Development: development.md
+  - Async Ecosystem Tests: async-ecosystem-tests.md
   - Verification: verification.md

pyproject.toml

@@ -77,6 +77,16 @@ integration = [
   "tortoise-orm>=1.0.0",
   "uvicorn>=0.33.0",
 ]
+# Reproducible downstream lifecycle tests, separate from framework smoke scripts.
+ecosystem = [
+  { include-group = "test" },
+  "maturin>=1.14.1",
+  "aiohttp>=3.14.3",
+  "httpx>=0.28.1",
+  "sqlalchemy[asyncio]>=2.0.38",
+  "aiosqlite>=0.20.0",
+  "uvloop>=0.23.0; sys_platform != 'win32'",
+]
 # Core pytest dependencies, including on free-threaded Python.
 test = [
   "pytest>=8.4",
@@ -104,8 +114,9 @@ module-name = "rsloop._loop"
 python-source = "python"
 
 [tool.pytest.ini_options]
-addopts = ["-m", "not tooling"]
+addopts = ["-m", "not tooling and not ecosystem"]
 markers = [
+  "ecosystem: downstream async library lifecycle contracts (requires ecosystem group)",
   "slow_network: real operating-system network behavior that may be slow",
   "stress: correctness tests with an optional high-repetition profile",
   "tooling: tests for repository benchmark and test infrastructure",

tests/ecosystem/conftest.py

@@ -0,0 +1,34 @@
+"""Real event loops and bounded teardown for downstream library contracts."""
+
+import asyncio
+
+import pytest
+import rsloop
+
+
+@pytest.fixture(params=["asyncio", "rsloop", "uvloop"])
+def loop(request):
+    if request.param == "uvloop":
+        factory = pytest.importorskip("uvloop").new_event_loop
+    elif request.param == "rsloop":
+        factory = rsloop.new_event_loop
+    else:
+        factory = asyncio.new_event_loop
+    event_loop = factory()
+    errors = []
+    event_loop.set_exception_handler(lambda _, context: errors.append(context))
+    yield event_loop
+    try:
+        pending = asyncio.all_tasks(event_loop)
+        for task in pending:
+            task.cancel()
+        if pending:
+            event_loop.run_until_complete(
+                asyncio.wait_for(asyncio.gather(*pending, return_exceptions=True), 5)
+            )
+        event_loop.run_until_complete(event_loop.shutdown_asyncgens())
+        event_loop.run_until_complete(event_loop.shutdown_default_executor())
+        assert not pending, "test leaked tasks"
+        assert not errors, errors
+    finally:
+        event_loop.close()

tests/ecosystem/test_database_clients.py

@@ -0,0 +1,164 @@
+"""Transaction ownership and pool recovery through SQLAlchemy and aiosqlite."""
+
+import asyncio
+from typing import Any, cast
+
+import pytest
+
+sqlalchemy = pytest.importorskip("sqlalchemy")
+pytest.importorskip("aiosqlite")
+sqlalchemy_asyncio = pytest.importorskip("sqlalchemy.ext.asyncio")
+URL = sqlalchemy.URL
+text = sqlalchemy.text
+async_sessionmaker = sqlalchemy_asyncio.async_sessionmaker
+create_async_engine = sqlalchemy_asyncio.create_async_engine
+
+pytestmark = pytest.mark.ecosystem
+
+
+@pytest.mark.parametrize("interruption", ["cancel", "timeout", "exception"])
+def test_interrupted_transaction_rolls_back_and_returns_connection(
+    loop, tmp_path, interruption
+):
+    async def main():
+        engine = create_async_engine(
+            URL.create("sqlite+aiosqlite", database=str(tmp_path / "transactions.db")),
+            pool_size=1,
+            max_overflow=0,
+            pool_timeout=2,
+        )
+        sessions = async_sessionmaker(engine)
+        inserted = asyncio.Event()
+        blocked = asyncio.Event()
+
+        async def transaction():
+            async with sessions.begin() as session:
+                await session.execute(
+                    text("INSERT INTO messages VALUES (1, 'discard')")
+                )
+                inserted.set()
+                if interruption == "exception":
+                    raise ValueError("application failed after insert")
+                await blocked.wait()
+
+        task = None
+        try:
+            async with engine.begin() as connection:
+                await connection.execute(
+                    text("CREATE TABLE messages (id INTEGER PRIMARY KEY, body TEXT)")
+                )
+            task = asyncio.create_task(transaction())
+            await inserted.wait()
+            if interruption == "exception":
+                with pytest.raises(ValueError, match="application failed"):
+                    await task
+            elif interruption == "cancel":
+                task.cancel()
+                with pytest.raises(asyncio.CancelledError):
+                    await task
+            else:
+                with pytest.raises(asyncio.TimeoutError):
+                    await asyncio.wait_for(task, 0)
+            assert cast(Any, engine.pool).checkedout() == 0
+            async with sessions.begin() as session:
+                assert await session.scalar(text("SELECT count(*) FROM messages")) == 0
+                await session.execute(text("INSERT INTO messages VALUES (2, 'commit')"))
+            async with sessions() as session:
+                result = await session.execute(text("SELECT id, body FROM messages"))
+                assert result.all() == [(2, "commit")]
+        finally:
+            if task is not None:
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+            await asyncio.wait_for(engine.dispose(), 5)
+
+    loop.run_until_complete(asyncio.wait_for(main(), 20))
+
+
+def test_concurrent_sessions_stream_results_and_release_pool(loop, tmp_path):
+    async def main():
+        engine = create_async_engine(
+            URL.create("sqlite+aiosqlite", database=str(tmp_path / "readers.db")),
+            pool_size=2,
+            max_overflow=0,
+            pool_timeout=5,
+        )
+        sessions = async_sessionmaker(engine)
+        tasks = []
+        try:
+            async with engine.begin() as connection:
+                await connection.execute(text("CREATE TABLE numbers (value INTEGER)"))
+                await connection.execute(
+                    text("INSERT INTO numbers VALUES (:value)"),
+                    [{"value": value} for value in range(128)],
+                )
+
+            async def read(offset):
+                # Each concurrent task owns its own session and result stream.
+                async with sessions() as session:
+                    result = await session.stream(
+                        text(
+                            "SELECT value FROM numbers WHERE value >= :offset ORDER BY value"
+                        ),
+                        {"offset": offset},
+                        execution_options={"yield_per": 7},
+                    )
+                    try:
+                        values = [row[0] async for row in result]
+                        assert values == list(range(offset, 128))
+                    finally:
+                        await result.close()
+                return len(values)
+
+            tasks = [asyncio.create_task(read(offset)) for offset in range(8)]
+            assert await asyncio.gather(*tasks) == [128 - offset for offset in range(8)]
+            assert cast(Any, engine.pool).checkedout() == 0
+        finally:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            await asyncio.wait_for(engine.dispose(), 5)
+
+    loop.run_until_complete(asyncio.wait_for(main(), 20))
+
+
+@pytest.mark.parametrize("interruption", ["cancel", "timeout"])
+def test_aiosqlite_cancel_inflight_thread_operation_then_reuse(loop, interruption):
+    import threading
+
+    aiosqlite = pytest.importorskip("aiosqlite")
+
+    async def main():
+        entered = asyncio.Event()
+        release = threading.Event()
+
+        def blocked_query():
+            loop.call_soon_threadsafe(entered.set)
+            if not release.wait(5):
+                raise RuntimeError("test did not release the SQLite worker")
+            return 7
+
+        async with aiosqlite.connect(":memory:") as connection:
+            await connection.create_function("blocked_query", 0, blocked_query)
+            task = asyncio.ensure_future(connection.execute("SELECT blocked_query()"))
+            try:
+                await entered.wait()
+                assert not task.done()
+                if interruption == "cancel":
+                    task.cancel()
+                    with pytest.raises(asyncio.CancelledError):
+                        await task
+                else:
+                    with pytest.raises(asyncio.TimeoutError):
+                        await asyncio.wait_for(task, 0)
+                release.set()
+                # The cancelled operation completes on its worker first. Its
+                # late result must not poison the next queued operation.
+                async with connection.execute("SELECT 42") as cursor:
+                    assert await cursor.fetchone() == (42,)
+            finally:
+                release.set()
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+
+    loop.run_until_complete(asyncio.wait_for(main(), 20))