From b60beb927945c7844157435a2c631efd26f1c90d Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:10:55 -0500 Subject: [PATCH] fix: give RedisCacheLayer an event-loop ownership contract (#1162) A redis.asyncio ConnectionPool caches connections whose transports are bound to the loop that opened them (redis/redis-py#3351), but RedisCacheLayer held one pool in self.redis_pool and reached it from six methods with no notion of which loop owned it. This module also builds an IntelligentCacheSystem singleton at import time, outside any loop, so one layer instance really is reachable from several loops in a single process. #1152 added a partial loop guard covering only the tagged-set() path and then reverted it, because guarding one of six call sites implies a safety property the layer does not have. This adds the layer-wide contract instead. The layer and its pool are now owned by exactly one event loop. Ownership is claimed the first time the pool is touched and verified at every call site that reaches it: connect, disconnect, get, set, delete, clear and invalidate_by_tags. - non-owning loop, owner alive -> raise CacheLoopOwnershipError - non-owning loop, owner closed -> the pool is unusable by anyone, so drop it, mark the layer disconnected, log a WARNING, and let the new loop re-claim via connect(). We deliberately do not await pool.disconnect() here: that would touch the very transports bound to the dead loop. - disconnect() is the supported clean handoff, with IntelligentCacheSystem.shutdown() as its facade counterpart to initialize(). Without that the recovery path documented on the class would only be reachable by indexing into system.layers. Every guard runs outside its method's `except Exception` block. Those blocks return None/False/0 on failure, so a guard placed inside one would disguise cross-loop transport misuse as an ordinary cache miss. The closed-owner-loop discard is a judgement call: #1162 left the recovery behaviour open, and a pure hard reject would permanently brick the import-time singleton for any process that calls asyncio.run() more than once. Discarding is strictly safer than the status quo, which silently reuses a dead pool, and it is loud. Tests: 17 added, covering rejection from a second live loop at all seven call sites, non-swallowing of the error through both the layer and the IntelligentCacheSystem facade, the closed-owner discard and its warning, the disconnect()/connect() handoff, and the import-time singleton. Verified non-vacuous: with the guard neutered, 12 of them fail (the other 5 exercise disconnect()/shutdown(), which do not exist without this fix). test_tag_write_semaphore_is_replaced_after_its_loop_closes now reconnects between loops. Its docstring already disclaimed cross-loop pool safety and cited this issue; under the contract the second loop must re-establish the pool before writing, which is exactly why the semaphore must be replaced. Closes #1162 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/intelligent_cache.py | 185 ++++++++++- tests/unit/test_intelligent_cache.py | 288 +++++++++++++++++- 2 files changed, 452 insertions(+), 21 deletions(-) diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py index a0281de08..95bde9f21 100644 --- a/src/youtube_extension/backend/services/intelligent_cache.py +++ b/src/youtube_extension/backend/services/intelligent_cache.py @@ -56,6 +56,18 @@ def _resolve_tag_write_limit(max_connections: int) -> int: """ return max(1, min(TAG_WRITE_CONCURRENCY, max_connections - TAG_WRITE_POOL_RESERVE)) + +class CacheLoopOwnershipError(RuntimeError): + """A cache layer's connection pool was used from a non-owning event loop. + + Raised by :class:`RedisCacheLayer` when the running loop is not the loop + that owns ``self.redis_pool``. This is a programming error, not a transient + Redis fault, so it is deliberately *not* folded into the ``None``/``False``/ + ``0`` fallbacks that the layer returns for connection failures -- a silent + fallback here would hide cross-loop transport misuse behind what looks like + an ordinary cache miss. + """ + class DateTimeEncoder(json.JSONEncoder): """JSON encoder that handles datetime objects""" def default(self, obj): @@ -273,7 +285,47 @@ def _update_avg_access_time(self, access_time_ms: float): (1 - alpha) * self.stats.avg_access_time_ms) class RedisCacheLayer(IntelligentCacheLayer): - """L2 Cache: Redis distributed cache""" + """L2 Cache: Redis distributed cache. + + Event-loop ownership contract + ----------------------------- + A ``redis.asyncio.ConnectionPool`` caches connections whose transports are + bound to the event loop that opened them, so this layer and its + ``self.redis_pool`` are **owned by exactly one event loop**. + + Ownership is claimed the first time the pool is touched -- normally inside + :meth:`connect`, which is what creates the pool -- and is then verified at + every call site that reaches ``self.redis_pool``: :meth:`connect`, + :meth:`disconnect`, :meth:`get`, :meth:`set`, :meth:`delete`, :meth:`clear` + and :meth:`invalidate_by_tags`. + + The rules are: + + * Called from the owning loop -> allowed. + * Called from a different loop while the owning loop is still **alive** -> + :class:`CacheLoopOwnershipError`. Two live loops sharing one pool is + exactly the cross-loop transport misuse this contract exists to prevent + (see redis/redis-py#3351), so it is rejected rather than papered over. + * Called from a different loop after the owning loop has **closed** -> the + pool can no longer be used by anyone, because every connection it cached + is attached to a dead loop. It is dropped, the layer is marked + disconnected, and the new loop may claim the layer by calling + :meth:`connect` again. This is logged at WARNING because connections that + were never closed on the owning loop leak their sockets. + * :meth:`disconnect` is the supported clean handoff: it closes the pool on + the owning loop and releases ownership, after which any loop may + :meth:`connect` again. + + Guards run *before* each method's ``try``/``except Exception`` block, so a + :class:`CacheLoopOwnershipError` is never swallowed into a cache-miss-shaped + return value. + + Why this matters here specifically: this module builds an + ``IntelligentCacheSystem`` singleton at import time, outside any event loop, + so a single process (a test suite giving each test a fresh loop, or any code + calling ``asyncio.run()`` more than once) can reach one layer instance from + several loops. + """ def __init__(self, name: str = "L2_Redis", redis_url: str = "redis://localhost:6379", max_connections: int = 20): super().__init__(name, max_size=100000) # Logical limit for Redis @@ -281,10 +333,67 @@ def __init__(self, name: str = "L2_Redis", redis_url: str = "redis://localhost:6 self.max_connections = max_connections self.redis_pool = None self._connected = False + self._pool_loop: Optional[asyncio.AbstractEventLoop] = None self._tag_write_limit = _resolve_tag_write_limit(max_connections) self._tag_write_semaphore: Optional[asyncio.Semaphore] = None self._tag_write_semaphore_loop: Optional[asyncio.AbstractEventLoop] = None + def _require_pool_loop(self) -> asyncio.AbstractEventLoop: + """Enforce the event-loop ownership contract for ``self.redis_pool``. + + Ownership is only meaningful while a pool exists, so a layer with no + pool claims nothing and any loop may go on to :meth:`connect` it. + + Returns: + The running loop. On return it is guaranteed to be a loop that may + safely use ``self.redis_pool``. + + Raises: + CacheLoopOwnershipError: another, still-running, loop owns the pool. + """ + loop = asyncio.get_running_loop() + owner = self._pool_loop + + if owner is loop: + return loop + + if owner is None: + # A pool assigned without going through connect() is claimed by the + # first loop that touches it. With no pool there is nothing to own. + if self.redis_pool is not None: + self._pool_loop = loop + return loop + + if not owner.is_closed(): + raise CacheLoopOwnershipError( + f"{self.name}: RedisCacheLayer is owned by event loop {owner!r} " + f"but was used from {loop!r}. A redis.asyncio ConnectionPool " + "caches connections bound to the loop that opened them and " + "cannot be shared across live loops. Use one RedisCacheLayer " + "per event loop, or await disconnect() on the owning loop " + "before reconnecting from another one." + ) + + # The owning loop is gone. Nothing can use its pool again, so drop it + # rather than hand out connections attached to a dead loop. We must not + # await pool.disconnect() from here: that would touch those very + # transports from the wrong loop. + if self.redis_pool is not None: + logger.warning( + "%s: event loop %r that owned the Redis connection pool has " + "closed without disconnect(); discarding the pool (its " + "connections may have leaked) and releasing ownership. Call " + "connect() to re-establish the pool on loop %r.", + self.name, + owner, + loop, + ) + + self.redis_pool = None + self._connected = False + self._pool_loop = None + return loop + def _get_tag_write_semaphore(self) -> asyncio.Semaphore: """Semaphore shared by every ``set()`` call on this layer. @@ -302,19 +411,10 @@ def _get_tag_write_semaphore(self) -> asyncio.Semaphore: singleton. It is therefore replaced whenever a different loop is seen, so that the - semaphore itself never outlives the loop it bound to. - - Scope note: this limiter bounds tag-write fan-out within one loop. It is - deliberately *not* a cross-loop safety mechanism, and replacing the - semaphore does **not** make the layer reusable across loops. A - ``redis.asyncio`` pool caches connections whose transports are bound to - the loop that opened them, so a ``RedisCacheLayer`` is already - event-loop-affine through ``self.redis_pool`` -- and that affinity - applies equally to ``get()``, ``delete()``, ``clear()`` and - ``invalidate_by_tags()``, none of which this limiter touches. Enforcing - a loop-ownership contract is a layer-wide concern tracked in #1162; - guarding only this one path would give a misleading partial guarantee. - Use one layer per event loop. + semaphore itself never outlives the loop it bound to. Under the + layer-wide ownership contract documented on this class, that can only + happen after ownership was released to a new loop, so replacement is a + consequence of the contract rather than a competing mechanism. """ loop = asyncio.get_running_loop() @@ -325,13 +425,22 @@ def _get_tag_write_semaphore(self) -> asyncio.Semaphore: return self._tag_write_semaphore async def connect(self): - """Connect to Redis""" + """Connect to Redis. + + Claims event-loop ownership of the pool this creates. See the class + docstring for the full contract. + """ + # Outside the try: an ownership violation is a programming error and + # must not be downgraded to "Redis is unreachable". + loop = self._require_pool_loop() + try: self.redis_pool = redis.ConnectionPool.from_url( self.redis_url, max_connections=self.max_connections, decode_responses=False # We handle binary data ) + self._pool_loop = loop # Test connection async with redis.Redis(connection_pool=self.redis_pool) as conn: @@ -344,8 +453,34 @@ async def connect(self): logger.warning(f"❌ Failed to connect to Redis: {e}") self._connected = False + async def disconnect(self) -> None: + """Close the pool on its owning loop and release ownership. + + This is the supported way to hand a layer from one event loop to + another: ``await layer.disconnect()`` on the owning loop, then + ``await layer.connect()`` on the next one. Must be called from the + owning loop, because closing the pool touches transports bound to it. + """ + # Outside any try/except for the same reason as connect(). + self._require_pool_loop() + + pool = self.redis_pool + self.redis_pool = None + self._connected = False + self._pool_loop = None + + if pool is None: + return + + try: + await pool.disconnect() + except Exception as e: # pragma: no cover - defensive + logger.warning(f"Redis pool disconnect error: {e}") + async def get(self, key: str) -> Optional[Any]: """Get value from Redis cache""" + self._require_pool_loop() + if not self._connected: return None @@ -387,6 +522,8 @@ async def get(self, key: str) -> Optional[Any]: async def set(self, key: str, value: Any, ttl: Optional[int] = None, tags: list[str] = None) -> bool: """Set value in Redis cache""" + self._require_pool_loop() + if not self._connected: return False @@ -451,6 +588,8 @@ async def _add_tag(tag: str) -> None: async def delete(self, key: str) -> bool: """Delete value from Redis cache""" + self._require_pool_loop() + if not self._connected: return False @@ -474,6 +613,8 @@ async def delete(self, key: str) -> bool: async def clear(self) -> int: """Clear all Redis cache entries""" + self._require_pool_loop() + if not self._connected: return 0 @@ -499,6 +640,8 @@ async def clear(self) -> int: async def invalidate_by_tags(self, tags: list[str]) -> int: """Invalidate cache entries by tags""" + self._require_pool_loop() + if not self._connected: return 0 @@ -567,6 +710,18 @@ async def initialize(self): if hasattr(layer, 'connect'): await layer.connect() + async def shutdown(self) -> None: + """Release every layer's connections on the loop that owns them. + + The counterpart to :meth:`initialize`. Call this before the running + event loop closes: ``RedisCacheLayer`` binds its connection pool to the + loop that created it, and this is the supported way to release that + binding so the next loop can call :meth:`initialize` cleanly. + """ + for layer in self.layers: + if hasattr(layer, 'disconnect'): + await layer.disconnect() + async def get(self, key: str) -> Optional[Any]: """Get value from cache layers (L1 → L2 → L3)""" start_time = time.time() diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py index 591cb0c12..ffc0a289d 100644 --- a/tests/unit/test_intelligent_cache.py +++ b/tests/unit/test_intelligent_cache.py @@ -1046,6 +1046,7 @@ async def test_cache_invalidate_tags_returns_dict(self): from youtube_extension.backend.services.intelligent_cache import ( TAG_WRITE_CONCURRENCY, + CacheLoopOwnershipError, IntelligentCacheSystem, # noqa: E402 RedisCacheLayer, ) @@ -1388,11 +1389,11 @@ def test_tag_write_semaphore_is_replaced_after_its_loop_closes(self): time, outside any loop, so that is reachable -- e.g. a process calling asyncio.run() more than once, or a suite giving each test a fresh loop. - This does NOT establish that reusing a RedisCacheLayer or its - redis.asyncio ConnectionPool across loops is safe. The pool caches - connections whose transports are bound to the loop that opened them and - has no ownership contract; see issue #1162. Redis is mocked here, so - only semaphore replacement is exercised. + Pool reuse across loops is a separate concern governed by the + event-loop ownership contract (issue #1162): the first write's loop + closes, so the second loop must re-establish the pool via connect() + before it may write. That reconnect is what this test performs, and it + is exactly why the semaphore has to be replaced too. """ layer = self._connected_layer() conn = _make_redis_conn() @@ -1401,13 +1402,19 @@ async def _write(key, tag): with _patch_redis(conn): return await layer.set(key, "v", tags=[tag]) + async def _reconnect_and_write(key, tag): + with _patch_redis(conn): + # The owning loop is gone, so the layer released its pool. + await layer.connect() + return await layer.set(key, "v", tags=[tag]) + assert asyncio.run(_write("k", "a")) is True first = layer._tag_write_semaphore first_loop = layer._tag_write_semaphore_loop assert first is not None assert first_loop is not None and first_loop.is_closed() - assert asyncio.run(_write("k2", "b")) is True + assert asyncio.run(_reconnect_and_write("k2", "b")) is True assert layer._tag_write_semaphore is not first assert layer._tag_write_semaphore_loop is not first_loop @@ -1608,3 +1615,272 @@ def test_subsequent_call_uses_ema(self): layer._update_avg_access_time(20.0) # EMA: 0.1 * 20 + 0.9 * 10 = 11.0 assert layer.stats.avg_access_time_ms == pytest.approx(11.0, rel=0.01) + + +# --------------------------------------------------------------------------- +# RedisCacheLayer event-loop ownership contract — issue #1162 +# --------------------------------------------------------------------------- +import contextlib +import logging +import threading + + +@contextlib.contextmanager +def _foreign_owner_loop(layer): + """Hand ``layer`` to an owner that is a *different, still running* loop. + + The loop runs on a background thread and stays alive for the duration of + the ``with`` block, so the ownership check inside the test's own loop hits + the "owner is still live" branch rather than the closed-loop branch. + """ + + async def _claim(): + layer._require_pool_loop() + + loop = asyncio.new_event_loop() + ready = threading.Event() + + def _run(): + asyncio.set_event_loop(loop) + loop.call_soon(ready.set) + loop.run_forever() + + thread = threading.Thread(target=_run, daemon=True) + thread.start() + assert ready.wait(5), "owner loop failed to start" + asyncio.run_coroutine_threadsafe(_claim(), loop).result(5) + assert layer._pool_loop is loop + + try: + yield loop + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(5) + loop.close() + + +class TestRedisCacheLayerLoopOwnership: + """RedisCacheLayer + its ConnectionPool are owned by one event loop. + + redis.asyncio.ConnectionPool caches connections whose transports are bound + to the loop that opened them, so a pool reached from a second live loop + hands out connections that loop cannot drive. This module also builds an + IntelligentCacheSystem singleton at import time, outside any loop, so a + single layer instance really is reachable from several loops in one + process. See issue #1162. + """ + + def _connected_layer(self): + layer = RedisCacheLayer("L2") + layer._connected = True + layer.redis_pool = _make_pool() + return layer + + # -- rejection from a second live loop, at every pool call site --------- + + @pytest.mark.parametrize( + "call", + [ + pytest.param(lambda layer: layer.connect(), id="connect"), + pytest.param(lambda layer: layer.disconnect(), id="disconnect"), + pytest.param(lambda layer: layer.get("k"), id="get"), + pytest.param(lambda layer: layer.set("k", "v"), id="set"), + pytest.param(lambda layer: layer.delete("k"), id="delete"), + pytest.param(lambda layer: layer.clear(), id="clear"), + pytest.param(lambda layer: layer.invalidate_by_tags(["t"]), id="invalidate_by_tags"), + ], + ) + async def test_live_foreign_loop_is_rejected(self, call): + layer = self._connected_layer() + conn = _make_redis_conn() + + with _foreign_owner_loop(layer): + with _patch_redis(conn): + with pytest.raises(CacheLoopOwnershipError): + await call(layer) + + async def test_rejection_is_not_swallowed_into_a_fallback_value(self): + """Every pool method wraps its body in ``except Exception``. + + If the guard ran inside that block the ownership error would surface as + an ordinary cache miss (None / False / 0) and the bug would stay + invisible. Assert the error escapes instead. + """ + layer = self._connected_layer() + conn = _make_redis_conn() + + with _foreign_owner_loop(layer): + with _patch_redis(conn): + for coro_factory in ( + lambda: layer.get("k"), + lambda: layer.set("k", "v"), + lambda: layer.delete("k"), + lambda: layer.clear(), + lambda: layer.invalidate_by_tags(["t"]), + ): + with pytest.raises(CacheLoopOwnershipError): + await coro_factory() + + async def test_rejection_leaves_the_layer_untouched(self): + """A rejected access must not tear down the owner's working pool.""" + layer = self._connected_layer() + pool = layer.redis_pool + + with _foreign_owner_loop(layer) as owner: + with pytest.raises(CacheLoopOwnershipError): + await layer.get("k") + + assert layer.redis_pool is pool + assert layer._connected is True + assert layer._pool_loop is owner + + async def test_no_pool_means_no_ownership_to_violate(self): + """A layer that never connected is claimed by nobody.""" + layer = RedisCacheLayer("L2") + + assert layer._pool_loop is None + assert await layer.get("k") is None + # Reading a pool-less layer must not claim it for this loop, or the + # next loop to call connect() would be locked out. + assert layer._pool_loop is None + + # -- closed owner loop ------------------------------------------------- + + def test_closed_owner_loop_releases_the_pool(self, caplog): + """The pool dies with its loop, so drop it instead of reusing it. + + We deliberately do not ``await pool.disconnect()`` here: that would + touch the very transports bound to the dead loop. + """ + layer = self._connected_layer() + pool = layer.redis_pool + + async def _claim(): + layer._require_pool_loop() + + asyncio.run(_claim()) + first_loop = layer._pool_loop + assert first_loop is not None and first_loop.is_closed() + assert layer.redis_pool is pool + + with caplog.at_level(logging.WARNING): + asyncio.run(_claim()) + + assert layer.redis_pool is None + assert layer._connected is False + assert layer._pool_loop is None + assert any( + "closed without disconnect()" in r.getMessage() for r in caplog.records + ), "the discarded pool must be reported, not dropped silently" + + def test_new_loop_can_reconnect_after_the_owner_closed(self): + layer = self._connected_layer() + conn = _make_redis_conn() + + async def _use(): + with _patch_redis(conn): + return await layer.get("k") + + async def _reconnect_and_use(): + with _patch_redis(conn): + await layer.connect() + return await layer.get("k") + + asyncio.run(_use()) + assert asyncio.run(_reconnect_and_use()) is None # miss, but no raise + assert layer._connected is True + + # -- disconnect() is the supported handoff ----------------------------- + + def test_disconnect_releases_ownership_for_the_next_loop(self): + layer = self._connected_layer() + pool = AsyncMock() + layer.redis_pool = pool + conn = _make_redis_conn() + + async def _own_then_release(): + layer._require_pool_loop() + await layer.disconnect() + + asyncio.run(_own_then_release()) + + pool.disconnect.assert_awaited_once() + assert layer.redis_pool is None + assert layer._connected is False + assert layer._pool_loop is None + + async def _adopt(): + with _patch_redis(conn): + await layer.connect() + return layer._connected + + # No warning path, no error: this is the clean handoff. + assert asyncio.run(_adopt()) is True + + async def test_disconnect_without_a_pool_is_a_noop(self): + layer = RedisCacheLayer("L2") + await layer.disconnect() + assert layer.redis_pool is None + assert layer._pool_loop is None + + # -- the import-time singleton is the real-world trigger --------------- + + async def test_import_time_singleton_honours_the_contract(self): + """`intelligent_cache` is built at import time, outside any loop. + + That single instance is reachable from every loop in the process, which + is precisely the situation issue #1162 describes. + """ + from youtube_extension.backend.services import intelligent_cache as ic_module + + layer = ic_module.intelligent_cache.layers[1] + assert isinstance(layer, RedisCacheLayer) + + original_pool, original_connected, original_loop = ( + layer.redis_pool, + layer._connected, + layer._pool_loop, + ) + try: + layer.redis_pool = _make_pool() + layer._connected = True + layer._pool_loop = None + + with _foreign_owner_loop(layer): + with pytest.raises(CacheLoopOwnershipError): + await ic_module.intelligent_cache.get("k") + finally: + layer.redis_pool = original_pool + layer._connected = original_connected + layer._pool_loop = original_loop + + async def test_facade_does_not_swallow_the_ownership_error(self): + """IntelligentCacheSystem must surface the error, not mask it.""" + system = IntelligentCacheSystem() + layer = system.layers[1] + layer._connected = True + layer.redis_pool = _make_pool() + + with _foreign_owner_loop(layer): + with pytest.raises(CacheLoopOwnershipError): + await system.set("k", "v") + + async def test_system_shutdown_releases_the_redis_layer(self): + """`shutdown()` is how a process performs the clean handoff. + + Without it the documented recovery path would only be reachable by + reaching into `system.layers[1]`. + """ + system = IntelligentCacheSystem() + layer = system.layers[1] + pool = AsyncMock() + layer._connected = True + layer.redis_pool = pool + layer._require_pool_loop() + + await system.shutdown() + + pool.disconnect.assert_awaited_once() + assert layer.redis_pool is None + assert layer._connected is False + assert layer._pool_loop is None