fix: Validate Redis operational failure boundaries and document at-most-once limits

Author: aidankhogg

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

@@ -5,6 +5,6 @@ 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.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/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