diff --git a/docs/redis.md b/docs/redis.md index 150c25e..1ec5984 100644 --- a/docs/redis.md +++ b/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 diff --git a/docs/work-register.md b/docs/work-register.md index 2b3895d..c127b39 100644 --- a/docs/work-register.md +++ b/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). | diff --git a/src/eventful/transports/redis.py b/src/eventful/transports/redis.py index ae07369..88ba457 100644 --- a/src/eventful/transports/redis.py +++ b/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 diff --git a/tests/test_redis_transport.py b/tests/test_redis_transport.py index 581e885..72d377f 100644 --- a/tests/test_redis_transport.py +++ b/tests/test_redis_transport.py @@ -4,6 +4,13 @@ import asyncio import os +import shutil +import socket +import subprocess +import time +import uuid +from collections.abc import Iterator +from pathlib import Path from typing import Any import pytest @@ -14,6 +21,88 @@ from eventful.transports.redis import RedisTransport +class RestartableRedis: + """Own a disposable Redis process for a genuine restart boundary test.""" + + def __init__(self, executable: str, directory: Path, port: int) -> None: + self.executable = executable + self.directory = directory + self.port = port + self.process: subprocess.Popen[bytes] | None = None + + @property + def url(self) -> str: + """Return this isolated server's connection URL.""" + return f"redis://127.0.0.1:{self.port}/0" + + def start(self) -> None: + """Start Redis and wait until its TCP listener is ready.""" + self.process = subprocess.Popen( + [ + self.executable, + "--port", + str(self.port), + "--dir", + str(self.directory), + "--save", + "", + "--appendonly", + "no", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + with socket.socket() as connection: + if connection.connect_ex(("127.0.0.1", self.port)) == 0: + return + time.sleep(0.01) + raise RuntimeError("disposable Redis did not start") + + def stop(self) -> None: + """Stop Redis without persistence and wait for process collection.""" + if self.process is not None: + self.process.terminate() + self.process.wait(timeout=3) + self.process = None + + +@pytest.fixture +def restartable_redis(tmp_path: Path) -> Iterator[RestartableRedis]: + """Provide a real restartable server when redis-server is installed.""" + executable = shutil.which("redis-server") + if executable is None: + pytest.skip("redis-server executable is not installed") + with socket.socket() as reservation: + reservation.bind(("127.0.0.1", 0)) + port = reservation.getsockname()[1] + server = RestartableRedis(executable, tmp_path, port) + server.start() + try: + yield server + finally: + server.stop() + + +def redis_service_url() -> str: + """Return the explicitly configured service URL or skip integration tests.""" + url = os.getenv("EVENTFUL_REDIS_URL") + if url is None: + pytest.skip("EVENTFUL_REDIS_URL is not configured") + return url + + +def redis_service_transport(**options: Any) -> RedisTransport: + """Create an isolated transport against the configured Redis service.""" + return RedisTransport( + redis_service_url(), + channel_prefix=f"eventful-tests:{os.getpid()}:{uuid.uuid4().hex}:", + poll_timeout=0.02, + **options, + ) + + class FakePubSub: """Model the redis-py Pub/Sub methods used by the transport.""" @@ -78,6 +167,43 @@ async def aclose(self) -> None: self.closed = True +class FailingCleanupPubSub(FakePubSub): + """Model failure while closing an established or cancelled subscription.""" + + def __init__(self, client: FakeRedis, *, cancel_subscribe: bool = False) -> None: + super().__init__(client) + self.cancel_subscribe = cancel_subscribe + + async def subscribe(self, channel: str) -> None: + """Optionally model cancellation during subscription setup.""" + if self.cancel_subscribe: + raise asyncio.CancelledError + await super().subscribe(channel) + + async def unsubscribe(self, channel: str) -> None: + """Fail while attempting server-side cleanup.""" + del channel + raise ConnectionError("unsubscribe cleanup failed") + + async def aclose(self) -> None: + """Fail while attempting local Pub/Sub cleanup.""" + raise ConnectionError("Pub/Sub close cleanup failed") + + +class FailingCleanupRedis(FakeRedis): + """Create Pub/Sub resources whose cleanup always fails.""" + + def __init__(self, *, cancel_subscribe: bool = False) -> None: + super().__init__() + self.cancel_subscribe = cancel_subscribe + + def pubsub(self) -> FailingCleanupPubSub: + """Return one failure-configured fake Pub/Sub resource.""" + return FailingCleanupPubSub( + self, cancel_subscribe=self.cancel_subscribe + ) + + async def next_event(consumer: Consumer) -> Event: """Read one event and close the asynchronous iterator deterministically.""" iterator = consumer.consume() @@ -222,18 +348,43 @@ async def test_consumer_cancellation_cleans_up_subscription() -> None: await transport.close() +@pytest.mark.asyncio +async def test_explicit_iterator_close_surfaces_cleanup_failure() -> None: + """Do not let GeneratorExit hide failed unsubscribe and Pub/Sub cleanup.""" + transport, _ = make_transport(FailingCleanupRedis()) + consumer = transport.consumer("sample") + iterator = consumer.consume() + pending = asyncio.create_task(anext(iterator)) + await asyncio.sleep(0) + await transport.publisher().publish(Event("sample")) + assert await pending == Event("sample") + + with pytest.raises(TransportError, match="consumer close failed") as raised: + await iterator.aclose() + assert isinstance(raised.value.__cause__, ConnectionError) + await transport.close() + + +@pytest.mark.asyncio +async def test_cancelled_subscribe_preserves_cancellation_when_cleanup_fails() -> None: + """Keep CancelledError primary if incomplete-subscription cleanup also fails.""" + transport, _ = make_transport(FailingCleanupRedis(cancel_subscribe=True)) + consumer = transport.consumer("sample") + + with pytest.raises(asyncio.CancelledError) as raised: + await consumer.subscribe() + assert raised.value.__notes__ == [ + "Redis subscription cleanup also failed: " + "ConnectionError('Pub/Sub close cleanup failed')" + ] + await transport.close() + + @pytest.mark.redis_integration @pytest.mark.asyncio async def test_real_redis_pubsub_round_trip() -> None: """Validate serialization, subscription, publication, and shutdown end to end.""" - url = os.getenv("EVENTFUL_REDIS_URL") - if url is None: - pytest.skip("EVENTFUL_REDIS_URL is not configured") - prefix = f"eventful-tests:{os.getpid()}:" - - async with RedisTransport( - url, channel_prefix=prefix, poll_timeout=0.05 - ) as transport: + async with redis_service_transport() as transport: consumer = transport.consumer("integration.created") await consumer.subscribe() pending = asyncio.create_task(next_event(consumer)) @@ -242,3 +393,169 @@ async def test_real_redis_pubsub_round_trip() -> None: await transport.publisher().publish(event) assert await asyncio.wait_for(pending, timeout=3) == event + + +@pytest.mark.redis_integration +@pytest.mark.asyncio +async def test_real_redis_connection_loss_during_publish_is_surfaced() -> None: + """Surface an unconfigured publish retry without implying safe replay.""" + transport = redis_service_transport() + publisher = transport.publisher() + original = transport.client.publish + + async def lose_connection(channel: str, payload: bytes) -> int: + del channel, payload + await transport.client.connection_pool.disconnect() + raise ConnectionError("forced publisher connection loss") + + transport.client.publish = lose_connection + with pytest.raises(TransportError, match="publish failed") as raised: + await publisher.publish(Event("publish.loss")) + assert isinstance(raised.value.__cause__, ConnectionError) + + transport.client.publish = original + await publisher.publish(Event("publish.recovered")) + await transport.close() + + +@pytest.mark.redis_integration +@pytest.mark.asyncio +async def test_real_redis_subscription_reconnects_after_connection_restart() -> None: + """Restore a subscription after its server connection is torn down.""" + async with redis_service_transport() as transport: + consumer = transport.consumer("restart") + await consumer.subscribe() + assert consumer._pubsub is not None + await consumer._pubsub.connection.disconnect() + + pending = asyncio.create_task(next_event(consumer)) + # A poll drives redis-py's reconnect and resubscription. Events in the gap + # are intentionally not counted because Pub/Sub remains at-most-once. + await asyncio.sleep(0.1) + expected = Event("restart", {"after": True}) + await transport.publisher().publish(expected) + assert await asyncio.wait_for(pending, timeout=3) == expected + + +@pytest.mark.redis_integration +@pytest.mark.asyncio +async def test_real_redis_consumer_recovers_after_server_restart( + restartable_redis: RestartableRedis, +) -> None: + """Surface the broken read, then recover with a fresh post-restart consumer.""" + async with RedisTransport( + restartable_redis.url, + channel_prefix=f"restart:{uuid.uuid4().hex}:", + poll_timeout=0.02, + ) as transport: + consumer = transport.consumer("server") + await consumer.subscribe() + restartable_redis.stop() + failed = asyncio.create_task(next_event(consumer)) + with pytest.raises(TransportError, match="consume failed"): + await asyncio.wait_for(failed, timeout=3) + restartable_redis.start() + replacement = transport.consumer("server") + await replacement.subscribe() + pending = asyncio.create_task(next_event(replacement)) + expected = Event("server", {"after_restart": True}) + await transport.publisher().publish(expected) + assert await asyncio.wait_for(pending, timeout=3) == expected + + +@pytest.mark.redis_integration +@pytest.mark.asyncio +async def test_real_redis_cancellation_while_reconnecting_cleans_up() -> None: + """Propagate cancellation and leave no consumer after a disconnected poll.""" + async with redis_service_transport() as transport: + consumer = transport.consumer("cancel.reconnect") + await consumer.subscribe() + assert consumer._pubsub is not None + await consumer._pubsub.connection.disconnect() + iterator = consumer.consume() + pending = asyncio.create_task(anext(iterator)) + await asyncio.sleep(0) + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending + assert consumer not in transport._consumers + + +@pytest.mark.redis_integration +@pytest.mark.asyncio +async def test_real_redis_duplicate_subscription_cleanup() -> None: + """Keep one server subscription and remove it on idempotent close.""" + async with redis_service_transport() as transport: + consumer = transport.consumer("duplicate") + await asyncio.gather(consumer.subscribe(), consumer.subscribe()) + channel = transport.channel("duplicate") + assert await transport.client.pubsub_numsub(channel) == [(channel.encode(), 1)] + await consumer.close() + await consumer.close() + assert await transport.client.pubsub_numsub(channel) == [(channel.encode(), 0)] + + +@pytest.mark.redis_integration +@pytest.mark.asyncio +async def test_real_redis_malformed_message_is_surfaced() -> None: + """Reject malformed service data and close the affected subscription.""" + async with redis_service_transport() as transport: + consumer = transport.consumer("malformed") + await consumer.subscribe() + iterator = consumer.consume() + pending = asyncio.create_task(anext(iterator)) + await transport.client.publish(transport.channel("malformed"), b"not-json") + with pytest.raises(TransportError, match="invalid"): + await asyncio.wait_for(pending, timeout=3) + assert consumer not in transport._consumers + + +@pytest.mark.redis_integration +@pytest.mark.asyncio +async def test_real_redis_sustained_concurrent_publishing() -> None: + """Deliver a sustained concurrent burst once subscription readiness is known.""" + async with redis_service_transport() as transport: + consumer = transport.consumer("burst") + await consumer.subscribe() + total = 250 + + async def receive() -> set[int]: + received: set[int] = set() + async for event in consumer.consume(): + received.add(event.payload["sequence"]) + if len(received) == total: + return received + return received + + pending = asyncio.create_task(receive()) + queue = asyncio.Queue[int]() + for index in range(total): + queue.put_nowait(index) + + async def publish_worker() -> None: + while not queue.empty(): + index = queue.get_nowait() + await transport.publisher().publish( + Event("burst", {"sequence": index}) + ) + + # Sustained concurrency is intentionally bounded: redis-py's default + # connection pool rejects an unbounded task fan-out at its connection cap. + await asyncio.gather(*(publish_worker() for _ in range(10))) + assert await asyncio.wait_for(pending, timeout=10) == set(range(total)) + + +@pytest.mark.redis_integration +@pytest.mark.asyncio +async def test_real_redis_shutdown_is_deterministic() -> None: + """Bound shutdown with active consumers and reject all later operations.""" + transport = redis_service_transport() + consumers = [transport.consumer(f"shutdown.{index}") for index in range(20)] + await asyncio.gather(*(consumer.subscribe() for consumer in consumers)) + async with asyncio.timeout(3): + await transport.close() + await transport.close() + assert not transport._consumers + assert all(consumer._closed for consumer in consumers) + with pytest.raises(TransportError, match="closed"): + await transport.publisher().publish(Event("shutdown"))