Expand ASGI adapter lifecycle coverage: close on failed startup, tests, and docs

.github/workflows/ci.yml

@@ -30,6 +30,29 @@ jobs:
             dist/*.tar.gz
           if-no-files-found: error
 
+  framework-compatibility:
+    name: Frameworks ${{ matrix.frameworks }} / Python ${{ matrix.python-version }}
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.13', '3.14']
+        frameworks: [minimum, current]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with: {python-version: '${{ matrix.python-version }}'}
+      - name: Install adapter test dependencies
+        shell: bash
+        run: |
+          python -m pip install -e . pytest pytest-asyncio
+          if [ '${{ matrix.frameworks }}' = minimum ]; then
+            python -m pip install 'fastapi>=0.110,<0.111' 'starlette>=0.37,<0.38'
+          else
+            python -m pip install 'fastapi>=0.110' 'starlette>=0.37'
+          fi
+      - run: python -m pytest tests/test_adapters.py
+
   installation:
     name: Python ${{ matrix.python-version }} / ${{ matrix.installation }}
     needs: build

docs/frameworks.md

@@ -66,11 +66,17 @@ fallback is intentional.
 - An injected `bus=` remains caller-owned by default.
 - A bus created by `bus_factory=` or by the middleware default is adapter-owned.
 - `close_on_shutdown=True` opts an injected bus into adapter ownership.
-- On `lifespan.shutdown.complete`, the adapter awaits `bus.close()` when the owned
-  bus exposes a synchronous or asynchronous close method.
-- Cleanup completes before shutdown success is forwarded to the ASGI server.
-- Startup failures do not claim that shutdown cleanup occurred; applications should
-  manage resources created outside the adapter in their own lifespan handler.
+- On `lifespan.shutdown.complete` or `lifespan.shutdown.failed`, the adapter awaits
+  `bus.close()` when the owned bus exposes a synchronous or asynchronous close
+  method.
+- Cleanup completes before the terminal shutdown result is forwarded to the ASGI
+  server, including when terminal lifespan calls overlap.
+- A failed startup (either `lifespan.startup.failed` or an exception escaping the
+  lifespan application) also closes an adapter-owned bus. Caller-owned buses remain
+  untouched in every failure path.
+- Cleanup is idempotent: repeated startup/shutdown cycles against the same middleware
+  instance do not close a bus more than once. Create a new application/middleware
+  instance to obtain a fresh adapter-owned bus after shutdown.
 
 The local `EventBus` has no resources to close. The generic close behavior exists
 for application bus subclasses that coordinate transports, stores, or plugins.
@@ -81,3 +87,22 @@ The bus itself is application-scoped. Request state isolates access paths, not b
 registrations. Use separate application instances or a custom `bus_factory` when
 tests or tenants require distinct registration state. `state_key=` supports
 coexistence with another request-state convention.
+
+Each middleware instance using the default bus or `bus_factory=` creates its own bus,
+so those concurrently running application instances are isolated. Injected buses
+have caller-defined scope: injecting the same `bus=` into multiple applications
+intentionally shares registrations and delivery state, and isolation is the caller's
+responsibility. This is process-local isolation only: pre-fork and multi-worker
+deployments create one adapter-owned bus per worker. Eventful does not coordinate
+registrations or delivery between workers; inject a caller-managed transport-backed
+bus when cross-process behavior is required. Do not share one adapter-owned
+middleware instance between event loops.
+
+## Supported versions
+
+The declared dependency floors are FastAPI 0.110 and Starlette 0.37 on Eventful's
+supported Python versions. CI exercises those minor-version ranges through the
+public ASGI and dependency APIs, and a separate matrix leg resolves the latest
+available releases. Versions between the floor and latest tested releases are
+expected to work; compatibility with a future major release is not promised until
+that release is separately validated.

docs/redis.md

@@ -81,15 +81,56 @@ lazily when explicit coordination is unnecessary.
 
 ## Reconnection and cancellation
 
-The transport relies on redis-py's connection retry and Pub/Sub resubscription
-behavior. Redis client options may be passed as keyword arguments to
-`RedisTransport` and are forwarded to `redis.asyncio.Redis.from_url`. Terminal
-client errors become `TransportError` with the original exception retained as the
-cause. Task cancellation remains `asyncio.CancelledError` and triggers consumer
-cleanup.
+Eventful does **not** implement a retry queue or replay loop. It makes one public
+method call and relies only on retry behavior configured in redis-py:
+
+- `publish()` calls redis-py's `publish()` once. A reconnect/retry completed by the
+  client can make that call succeed; otherwise Eventful raises `TransportError`.
+  Eventful never resends after the call returns or raises, because after a lost
+  response it cannot know whether Redis received the publication.
+- `subscribe()` calls redis-py's `subscribe()` once. An unrecovered connection
+  failure is a `TransportError`; cancellation remains `asyncio.CancelledError` and
+  closes the incomplete Pub/Sub resource.
+- `consume()` lets redis-py reconnect and restore its current Pub/Sub subscriptions
+  while polling. An unrecovered read failure is a `TransportError` and permanently
+  closes that consumer. Start a new consumer to try again.
+- `close()` does not retry failed unsubscribe or close operations. It attempts all
+  consumer closes before surfacing a shutdown `TransportError`, and repeated calls
+  remain safe.
+
+Redis client options may be passed as keyword arguments to `RedisTransport` and
+are forwarded to `redis.asyncio.Redis.from_url`. Original client exceptions are
+retained as causes. Task cancellation is never translated to `TransportError`.
 
 Automatic reconnection does not change the at-most-once delivery model: messages
-sent while disconnected may be lost.
+sent while disconnected may be lost. A successful `publish()` means Redis reported
+the number of live subscribers; it does not mean a consumer processed, persisted,
+or can replay the event. A clean connection reset can be recovered internally by
+redis-py, including resubscription. In the validated default configuration, a hard
+server restart surfaced the broken read as `TransportError`; callers had to create
+a fresh consumer after the server returned. Either outcome leaves an unavoidable
+delivery gap.
+
+## Validated operational limits
+
+Service-backed scenarios exercise forced publisher and subscriber connection
+loss, reconnection after the service becomes available again, cancellation during
+reconnection, duplicate-subscription cleanup, malformed broker messages,
+concurrent bursts, and bounded deterministic shutdown. These checks establish
+lifecycle and error behavior, not a durability guarantee:
+
+- Connection recovery is bounded by redis-py's configured retry policy and the
+  caller's own timeout or cancellation.
+- No event count is asserted across a disconnect or restart boundary. Only events
+  published after subscription readiness is re-established are expected. A hard
+  restart may terminate the old consumer before redis-py's reconnect path runs.
+- Concurrent publishing is safe when callers bound concurrency to the configured
+  redis-py pool. A task fan-out beyond `max_connections` is surfaced as a publish
+  `TransportError`; Eventful supplies no global ordering or unbounded buffering.
+- Malformed data terminates only the affected consumer; it is not skipped or sent
+  to a dead-letter channel.
+- Shutdown is deterministic for responsive Redis connections. A stalled network
+  operation still needs an application-level deadline such as `asyncio.timeout()`.
 
 ## Serialization and validation
 

docs/work-register.md

@@ -4,7 +4,7 @@ All intentional incompleteness must use an annotation listed in `docs/annotation
 
 | Module | Status | Deferred work |
 | --- | --- | --- |
-| `eventful.adapters` | provisional ASGI lifecycle | Validate multi-worker ownership and framework-version compatibility (Issue: Forebase/Eventful#1). |
-| `eventful.transports.redis` | provisional Pub/Sub | Validate operational reconnect and load behavior; Redis Streams durability remains deferred (Issue: Forebase/Eventful#2). |
+| `eventful.adapters` | validated in-process ASGI lifecycle | Multi-worker delivery remains deployment-managed; major framework versions require separate validation (Issue: Forebase/Eventful#1). |
+| `eventful.transports.redis` | provisional, service-validated Pub/Sub | Connection loss/restart gaps, cancellation cleanup, malformed input, concurrent publishing, and shutdown limits validated; delivery remains live at-most-once with no Eventful retry/replay or durability. Redis Streams remains deferred (Forebase/Eventful#2). |
 | `eventful.persistence.postgres_persistence` | provisional durable store | Validate migration upgrades, retention, replication, and operational load behavior (Issue: Forebase/Eventful#3). |
 | `eventful.contracts` | provisional, reference-validated | Validate async lifecycle, delivery, and durability semantics against real Redis/PostgreSQL integrations (Issue: Forebase/Eventful#4). |

src/eventful/adapters/starlette.py

@@ -50,7 +50,7 @@ def __init__(
         self._close_lock = asyncio.Lock()
 
     async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None:
-        """Attach state and wait for application shutdown before closing resources."""
+        """Attach state and close owned resources when a lifespan terminates."""
         if scope["type"] in {"http", "websocket"}:
             scope.setdefault("state", {})[self.state_key] = self.bus
 
@@ -59,27 +59,45 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None
             return
 
         async def lifespan_send(message: dict[str, Any]) -> None:
-            """Close owned resources before reporting successful shutdown."""
-            if message["type"] == "lifespan.shutdown.complete":
+            """Close resources before reporting shutdown or failed startup."""
+            if message["type"] in {
+                "lifespan.startup.failed",
+                "lifespan.shutdown.complete",
+                "lifespan.shutdown.failed",
+            }:
                 await self.close()
             await send(message)
 
-        await self.app(scope, receive, lifespan_send)
+        try:
+            await self.app(scope, receive, lifespan_send)
+        except BaseException as application_error:
+            # A lifespan exception may prevent the application from sending either
+            # terminal message. Do not strand a bus that this adapter created.
+            try:
+                await self.close()
+            except BaseException as cleanup_error:
+                raise BaseExceptionGroup(
+                    "lifespan application and Eventful cleanup failed",
+                    [application_error, cleanup_error],
+                ) from None
+            raise
 
     async def close(self) -> None:
         """Close an owned/opted-in bus once when it exposes `close()`."""
         async with self._close_lock:
             if self._closed:
                 return
             self._closed = True
-        if not self.close_on_shutdown:
-            return
-        close = getattr(self.bus, "close", None)
-        if close is None:
-            return
-        result = close()
-        if inspect.isawaitable(result):
-            await result
+            if not self.close_on_shutdown:
+                return
+            close = getattr(self.bus, "close", None)
+            if close is None:
+                return
+            result = close()
+            if inspect.isawaitable(result):
+                # Keep the lock until cleanup finishes so a concurrent terminal
+                # lifespan message cannot be forwarded ahead of resource cleanup.
+                await result
 
 
 def request_event_bus(

src/eventful/transports/redis.py

@@ -3,13 +3,15 @@
 Events publish to channels derived from their event type. Redis Pub/Sub provides
 live, at-most-once fan-out: it has no replay, acknowledgement, or durability. The
 transport owns clients it creates, while injected clients remain caller-owned.
-redis-py's connection retry and Pub/Sub resubscription behavior handle transient
-reconnections; terminal client errors propagate to the consumer.
+redis-py's configured connection retry and Pub/Sub resubscription behavior may
+handle transient reconnects. Eventful adds no retry loop: failures left by the
+client are surfaced, and publications are never replayed.
 """
 
 from __future__ import annotations
 
 import asyncio
+import sys
 from collections.abc import AsyncIterator
 from typing import Any
 
@@ -29,7 +31,7 @@ def __init__(self, transport: RedisTransport) -> None:
         self._closed = False
 
     async def publish(self, event: Event) -> None:
-        """Publish one event to its prefixed event-type channel."""
+        """Publish once, surfacing any failure not retried by redis-py."""
         if self._closed:
             raise TransportError("Redis publisher is closed")
         self._transport._ensure_open()
@@ -72,11 +74,11 @@ def topic(self) -> str:
         return self._topic
 
     async def subscribe(self) -> None:
-        """Eagerly establish the subscription when readiness must be explicit."""
+        """Subscribe once, surfacing any failure not recovered by redis-py."""
         await self._subscribe()
 
     async def consume(self) -> AsyncIterator[Event]:
-        """Yield live events sequentially until cancellation, failure, or close."""
+        """Yield live events, surfacing unrecovered reads and invalid payloads."""
         async with self._lifecycle_lock:
             if self._closed:
                 raise TransportError("Redis consumer is closed")
@@ -121,7 +123,16 @@ async def consume(self) -> AsyncIterator[Event]:
         finally:
             async with self._lifecycle_lock:
                 self._consuming = False
-            await self.close()
+            active_failure = sys.exception()
+            try:
+                await self.close()
+            except TransportError:
+                # Cleanup must not replace cancellation, malformed-message, or
+                # connection errors already escaping the iterator. GeneratorExit
+                # denotes an explicit/normal iterator close, so its cleanup error
+                # remains observable to the caller.
+                if active_failure is None or isinstance(active_failure, GeneratorExit):
+                    raise
 
     async def close(self) -> None:
         """Unsubscribe and close the Pub/Sub resource idempotently."""
@@ -131,13 +142,20 @@ async def close(self) -> None:
             self._closed = True
             pubsub, self._pubsub = self._pubsub, None
         if pubsub is not None:
+            failure: BaseException | None = None
             try:
                 await pubsub.unsubscribe(self._transport.channel(self.topic))
+            except Exception as exc:
+                failure = exc
+            try:
                 await pubsub.aclose()
             except Exception as exc:
+                failure = failure or exc
+            if failure is not None:
+                self._transport._discard_consumer(self)
                 raise TransportError(
                     f"Redis consumer close failed for topic {self.topic!r}"
-                ) from exc
+                ) from failure
         self._transport._discard_consumer(self)
 
     async def _subscribe(self) -> Any:
@@ -151,8 +169,21 @@ async def _subscribe(self) -> Any:
             pubsub = self._transport.client.pubsub()
             try:
                 await pubsub.subscribe(self._transport.channel(self.topic))
-            except Exception as exc:
-                await pubsub.aclose()
+            except BaseException as exc:
+                # Cancellation is deliberately included: a connection attempt may
+                # be cancelled while redis-py is reconnecting, but its Pub/Sub
+                # object must not retain a duplicate server-side subscription.
+                try:
+                    await asyncio.shield(pubsub.aclose())
+                except BaseException as cleanup_exc:
+                    # In particular, task cancellation must remain cancellation
+                    # even when a broken connection also makes cleanup fail.
+                    exc.add_note(
+                        "Redis subscription cleanup also failed: "
+                        f"{cleanup_exc!r}"
+                    )
+                if isinstance(exc, asyncio.CancelledError):
+                    raise exc
                 raise TransportError(
                     f"Redis subscribe failed for topic {self.topic!r}"
                 ) from exc