diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0975347..dc259f0 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/docs/frameworks.md b/docs/frameworks.md index fc9546c..2776716 100644 --- a/docs/frameworks.md +++ b/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. 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..9924b5d 100644 --- a/docs/work-register.md +++ b/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). | diff --git a/src/eventful/adapters/starlette.py b/src/eventful/adapters/starlette.py index 5f6dd00..42a5870 100644 --- a/src/eventful/adapters/starlette.py +++ b/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,12 +59,28 @@ 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()`.""" @@ -72,14 +88,16 @@ async def close(self) -> None: 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( 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_adapters.py b/tests/test_adapters.py index be708f1..9c12cd9 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -8,6 +8,9 @@ from typing import Any from fastapi import Depends, FastAPI, Request +from starlette.applications import Starlette +from starlette.requests import Request as StarletteRequest +from starlette.responses import JSONResponse from eventful import EventBus from eventful.adapters.fastapi import event_bus_dependency, install_eventful @@ -21,10 +24,12 @@ def __init__(self) -> None: """Create an open local bus.""" super().__init__() self.closed = False + self.close_calls = 0 async def close(self) -> None: """Mark the test bus closed.""" self.closed = True + self.close_calls += 1 async def terminal_app(scope: dict[str, Any], receive: Any, send: Any) -> None: @@ -139,6 +144,154 @@ def test_lifespan_respects_external_and_explicit_ownership() -> None: assert owned.closed is True +def test_repeated_lifespan_shutdown_closes_owned_bus_once() -> None: + """Make repeated server lifespan cycles safe and cleanup idempotent.""" + bus = CloseableBus() + middleware = EventfulMiddleware(terminal_app, bus_factory=lambda: bus) + + assert asyncio.run(invoke_lifespan(middleware))[-1] == "lifespan.shutdown.complete" + assert asyncio.run(invoke_lifespan(middleware))[-1] == "lifespan.shutdown.complete" + assert bus.close_calls == 1 + + +def test_concurrent_application_instances_have_distinct_owned_buses() -> None: + """Never share implicitly created buses between application instances.""" + first = EventfulMiddleware(terminal_app, bus_factory=CloseableBus) + second = EventfulMiddleware(terminal_app, bus_factory=CloseableBus) + + async def exercise() -> None: + await asyncio.gather(invoke_lifespan(first), invoke_lifespan(second)) + + asyncio.run(exercise()) + assert first.bus is not second.bus + assert first.bus.closed is True + assert second.bus.closed is True + + +def test_injected_bus_scope_is_shared_by_caller_choice() -> None: + """Attach one caller-owned bus to both apps when it is injected twice.""" + bus = CloseableBus() + first = EventfulMiddleware(terminal_app, bus=bus) + second = EventfulMiddleware(terminal_app, bus=bus) + + async def exercise() -> tuple[dict[str, Any], dict[str, Any]]: + return await asyncio.gather(invoke_http(first), invoke_http(second)) + + first_state, second_state = asyncio.run(exercise()) + assert first_state["eventful_bus"] is bus + assert second_state["eventful_bus"] is bus + + +def test_owned_bus_is_cleaned_up_after_startup_failure() -> None: + """Release adapter resources when startup fails or raises.""" + async def failed_app(scope: dict[str, Any], receive: Any, send: Any) -> None: + await receive() + await send({"type": "lifespan.startup.failed", "message": "nope"}) + + failed_bus = CloseableBus() + failed = EventfulMiddleware(failed_app, bus_factory=lambda: failed_bus) + assert asyncio.run(invoke_lifespan(failed)) == ["lifespan.startup.failed"] + assert failed_bus.close_calls == 1 + + async def raising_app(scope: dict[str, Any], receive: Any, send: Any) -> None: + await receive() + raise RuntimeError("startup exploded") + + raised_bus = CloseableBus() + raised = EventfulMiddleware(raising_app, bus_factory=lambda: raised_bus) + try: + asyncio.run(invoke_lifespan(raised)) + except RuntimeError as exc: + assert str(exc) == "startup exploded" + else: + raise AssertionError("startup exception should propagate") + assert raised_bus.close_calls == 1 + + +def test_startup_and_cleanup_failures_are_both_preserved() -> None: + """Report cleanup failure without replacing the startup root cause.""" + class FailingCloseBus(EventBus): + async def close(self) -> None: + raise OSError("cleanup exploded") + + async def raising_app(scope: dict[str, Any], receive: Any, send: Any) -> None: + await receive() + raise RuntimeError("startup exploded") + + middleware = EventfulMiddleware(raising_app, bus_factory=FailingCloseBus) + try: + asyncio.run(invoke_lifespan(middleware)) + except BaseExceptionGroup as exc: + assert [str(error) for error in exc.exceptions] == [ + "startup exploded", + "cleanup exploded", + ] + assert isinstance(exc.exceptions[0], RuntimeError) + assert isinstance(exc.exceptions[1], OSError) + else: + raise AssertionError("both lifespan failures should propagate") + + +def test_startup_failure_preserves_application_owned_bus() -> None: + """Do not clean up an injected bus merely because application startup fails.""" + async def failed_app(scope: dict[str, Any], receive: Any, send: Any) -> None: + await receive() + await send({"type": "lifespan.startup.failed"}) + + bus = CloseableBus() + middleware = EventfulMiddleware(failed_app, bus=bus) + asyncio.run(invoke_lifespan(middleware)) + assert bus.close_calls == 0 + + +def test_owned_bus_is_cleaned_up_after_shutdown_failure() -> None: + """Treat a failed shutdown message as a terminal lifespan outcome.""" + async def failed_shutdown( + scope: dict[str, Any], receive: Any, send: Any + ) -> None: + await receive() + await send({"type": "lifespan.startup.complete"}) + await receive() + await send({"type": "lifespan.shutdown.failed", "message": "nope"}) + + bus = CloseableBus() + middleware = EventfulMiddleware(failed_shutdown, bus_factory=lambda: bus) + assert asyncio.run(invoke_lifespan(middleware)) == [ + "lifespan.startup.complete", + "lifespan.shutdown.failed", + ] + assert bus.close_calls == 1 + + +def test_concurrent_close_waits_for_cleanup() -> None: + """Do not let a second close return while the first cleanup is in progress.""" + class BlockingCloseBus(EventBus): + def __init__(self) -> None: + super().__init__() + self.started = asyncio.Event() + self.release = asyncio.Event() + self.close_calls = 0 + + async def close(self) -> None: + self.close_calls += 1 + self.started.set() + await self.release.wait() + + async def exercise() -> None: + bus = BlockingCloseBus() + middleware = EventfulMiddleware(terminal_app, bus_factory=lambda: bus) + first = asyncio.create_task(middleware.close()) + await bus.started.wait() + second = asyncio.create_task(middleware.close()) + await asyncio.sleep(0) + assert second.done() is False + bus.release.set() + await asyncio.gather(first, second) + assert bus.close_calls == 1 + + asyncio.run(exercise()) + + def test_fastapi_installs_middleware_and_dependency_uses_request_state() -> None: """Use FastAPI's middleware registry and Request annotation contract.""" app = FastAPI() @@ -157,6 +310,19 @@ def bus_endpoint(resolved: EventBus = Depends(dependency)): assert asyncio.run(invoke_fastapi(app, "/bus")) == (200, {"same": True}) +def test_supported_starlette_application_uses_request_state() -> None: + """Exercise the public middleware API on the supported Starlette release.""" + app = Starlette() + bus = CloseableBus() + app.add_middleware(EventfulMiddleware, bus=bus) + + async def endpoint(request: StarletteRequest) -> JSONResponse: + return JSONResponse({"same": request_event_bus(request) is bus}) + + app.add_route("/bus", endpoint) + assert asyncio.run(invoke_fastapi(app, "/bus")) == (200, {"same": True}) + + def test_request_event_bus_requires_middleware_state() -> None: """Avoid silently leaking the process-global bus into unconfigured requests.""" request = SimpleNamespace(state=SimpleNamespace()) 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"))