From 46ca0a653b1127790b1a50f45e677c08c2e8dfd7 Mon Sep 17 00:00:00 2001 From: Hayden Garvey <154503486+groupthinking@users.noreply.github.com> Date: Tue, 4 Aug 2026 06:11:57 +0000 Subject: [PATCH] fix: give RedisCacheLayer an event-loop ownership contract A redis.asyncio.ConnectionPool caches connections whose transports are bound to the loop that opened them, and this module builds an IntelligentCacheSystem singleton at import time, so one layer instance is reachable from several loops in one process. The layer and its pool are now owned by exactly one event loop, enforced at all pool call sites (connect, disconnect, get, set, delete, clear, invalidate_by_tags). Use from a second live loop raises CacheLoopOwnershipError; a pool whose owner loop closed is discarded loudly and the next loop may reconnect. Ownership transitions are serialized on a threading.Lock, disconnect() releases state only after the pool actually closed (failures propagate), and warm_cache() re-raises the ownership error instead of swallowing it via return_exceptions. Generated with [Linear](https://linear.app/myxstack/issue/GRV-212/rediscachelayer-has-no-event-loop-ownership-contract-across-its-six#agent-session-2f0d8551) Co-authored-by: linear-code[bot] <222613912+linear-code[bot]@users.noreply.github.com> --- .../backend/services/intelligent_cache.py | 265 ++++++++++- tests/unit/test_intelligent_cache.py | 414 +++++++++++++++++- 2 files changed, 657 insertions(+), 22 deletions(-) diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py index 5ac4633f4..0f37c0d2c 100644 --- a/src/youtube_extension/backend/services/intelligent_cache.py +++ b/src/youtube_extension/backend/services/intelligent_cache.py @@ -61,6 +61,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. + """ + + # Hit timestamps retained per key to drive _calculate_adaptive_ttl(). That # consumer reads only the first element, the last element and the length, so the # window only has to be long enough for the ratio between them to be a stable @@ -327,7 +339,52 @@ 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`. Two loops mean two threads, so every + ownership transition is serialized on a ``threading.Lock``; simultaneous + first use from two live loops resolves to one winner, never to both loops + building a pool and leaking the loser's. + + 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. Ownership is released only once the pool actually + closed; a teardown failure propagates and leaves the layer intact so the + caller can retry instead of silently leaking a live pool. + + 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 @@ -335,10 +392,87 @@ 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 + # Serializes ownership transitions. asyncio primitives cannot do this + # job: the contending callers are, by definition, on different loops. + self._ownership_lock = threading.Lock() 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() + + # Fast path: the already-claimed owner re-checking itself needs no lock. + if self._pool_loop is loop: + return loop + + with self._ownership_lock: + 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 -- claiming anyway would lock a later connect() out. + 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 " + f"{owner!r} 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." + ) + + self._discard_pool_of_dead_owner(owner, loop) + return loop + + def _discard_pool_of_dead_owner( + self, + owner: asyncio.AbstractEventLoop, + loop: asyncio.AbstractEventLoop, + ) -> None: + """Drop a pool whose owning loop has closed. Caller holds ``_ownership_lock``. + + The pool's cached connections are attached to transports on the dead + loop, so nothing can use it again -- and it must not be closed from + here either, because ``pool.disconnect()`` would touch those very + transports from the wrong loop. Discarding is the only safe option, + and it is loud because sockets never closed on the owner leak. + """ + 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 + def _get_tag_write_semaphore(self) -> asyncio.Semaphore: """Semaphore shared by every tag fan-out on this layer. @@ -361,20 +495,11 @@ 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()`` and ``clear()``, none of - which this limiter touches -- and to the two paths that do acquire it, - since bounding fan-out is not the same guarantee as owning a loop. - Enforcing a loop-ownership contract is a layer-wide concern tracked in - #1162; guarding only these paths 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, a different + loop can only be seen here after ownership was released to that loop, + so replacement is a consequence of the contract rather than a + competing cross-loop mechanism. """ loop = asyncio.get_running_loop() @@ -385,8 +510,41 @@ 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. The claim + happens under ``_ownership_lock`` *before* the pool is built, so a + simultaneous connect() from a second live loop is rejected instead of + racing this one, overwriting ``self.redis_pool`` and leaking the + losing pool. See the class docstring for the full contract. + """ + loop = asyncio.get_running_loop() + + # Outside the try below: an ownership violation is a programming error + # and must not be downgraded to "Redis is unreachable". + with self._ownership_lock: + owner = self._pool_loop + if owner is not None and owner is not loop: + if not owner.is_closed(): + raise CacheLoopOwnershipError( + f"{self.name}: cannot connect() from {loop!r}; the " + f"layer is owned by live event loop {owner!r}. Await " + "disconnect() on the owning loop first, or use one " + "RedisCacheLayer per event loop." + ) + self._discard_pool_of_dead_owner(owner, loop) + + self._pool_loop = loop + old_pool = self.redis_pool + self.redis_pool = None + self._connected = False + try: + if old_pool is not None: + # Reconnecting on the owning loop: close the previous pool + # instead of silently abandoning its connections. + await old_pool.disconnect() + self.redis_pool = redis.ConnectionPool.from_url( self.redis_url, max_connections=self.max_connections, @@ -403,9 +561,57 @@ async def connect(self): except Exception as e: logger.warning(f"❌ Failed to connect to Redis: {e}") self._connected = False + if self.redis_pool is None: + # Nothing was built, so there is nothing to own; release the + # eager claim so any loop may attempt the next connect(). + with self._ownership_lock: + if self._pool_loop is loop and self.redis_pool is None: + self._pool_loop = None + + 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. It must run on the owning + loop, because closing the pool touches transports bound to it. + + A teardown failure propagates and leaves the layer owned, connected + and holding its pool: ownership is released only once the pool has + actually closed, so a failed disconnect can be retried rather than + silently leaking a live pool. + """ + loop = asyncio.get_running_loop() + + with self._ownership_lock: + owner = self._pool_loop + if owner is not None and owner is not loop: + if not owner.is_closed(): + raise CacheLoopOwnershipError( + f"{self.name}: cannot disconnect() from {loop!r}; the " + f"pool is owned by live event loop {owner!r} and must " + "be closed on that loop." + ) + # The owner died with its transports; discarding is all that + # is left to do, exactly as on any other post-mortem access. + self._discard_pool_of_dead_owner(owner, loop) + return + pool = self.redis_pool + + if pool is not None: + await pool.disconnect() + + with self._ownership_lock: + if self.redis_pool is pool: + self.redis_pool = None + self._connected = False + self._pool_loop = None async def get(self, key: str) -> Optional[Any]: """Get value from Redis cache""" + # Before the try and the _connected check: see the class docstring. + self._require_pool_loop() + if not self._connected: return None @@ -447,6 +653,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 @@ -511,6 +719,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 @@ -534,6 +744,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 @@ -559,6 +771,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 @@ -659,6 +873,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() @@ -734,6 +960,13 @@ async def warm_cache(self, keys_and_values: list[tuple[str, Any]], ttl: Optional tasks.append(self.set(key, value, ttl)) results = await asyncio.gather(*tasks, return_exceptions=True) + + # An ownership violation is a programming error, not a failed warm; + # return_exceptions must not swallow it into the success count. + for result in results: + if isinstance(result, CacheLoopOwnershipError): + raise result + success_count = sum(1 for r in results if r is True) logger.info(f"✅ Cache warming completed: {success_count}/{len(keys_and_values)} successful") diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py index 98330dc38..cd6f25fd5 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 @@ -2240,3 +2247,398 @@ async def test_indexing_and_len_work_over_bounded_history(self): accesses = layer.access_patterns["k"] assert len(accesses) == 5, "len() over the history container is wrong" assert accesses[0] <= accesses[-1], "first/last indexing is not ordered" + + +# --------------------------------------------------------------------------- +# RedisCacheLayer event-loop ownership contract — issue #1162 / GRV-212 +# --------------------------------------------------------------------------- +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 + + # -- ownership transitions are serialized -------------------------------- + + def test_simultaneous_first_connect_from_two_loops_is_serialized(self): + """Two loops racing their first connect() must not both build a pool. + + Ownership is claimed under a lock *before* the pool is created, so the + second loop is rejected while the first is still mid-connect, instead + of overwriting self.redis_pool and leaking the loser's pool. + """ + import types as _types + + layer = RedisCacheLayer("L2") + conn = _make_redis_conn() + pool_a = _make_pool() + + entered = threading.Event() + release = threading.Event() + + def _blocking_from_url(url, **kwargs): + entered.set() + assert release.wait(5), "test deadlock: from_url never released" + return pool_a + + mock_redis_mod = _types.ModuleType("redis.asyncio") + pool_cls = MagicMock() + pool_cls.from_url = MagicMock(side_effect=_blocking_from_url) + mock_redis_mod.ConnectionPool = pool_cls + mock_redis_mod.Redis = MagicMock(return_value=conn) + + owner_loop = asyncio.new_event_loop() + thread = threading.Thread(target=owner_loop.run_forever, daemon=True) + thread.start() + try: + with patch( + "youtube_extension.backend.services.intelligent_cache.redis", + mock_redis_mod, + ): + fut = asyncio.run_coroutine_threadsafe(layer.connect(), owner_loop) + assert entered.wait(5), "owner connect() never started" + + # The owner loop holds the claim but has not finished building + # its pool; the second loop must be rejected, not allowed to + # race it and overwrite the winner. + with pytest.raises(CacheLoopOwnershipError): + asyncio.run(layer.connect()) + + release.set() + fut.result(5) + finally: + release.set() + owner_loop.call_soon_threadsafe(owner_loop.stop) + thread.join(5) + owner_loop.close() + + assert layer._connected is True + assert layer.redis_pool is pool_a + + def test_reconnect_on_the_owner_loop_closes_the_previous_pool(self): + """connect() on the owning loop must not abandon the pool it replaces.""" + layer = RedisCacheLayer("L2") + old_pool = AsyncMock() + layer.redis_pool = old_pool + layer._connected = True + conn = _make_redis_conn() + + async def _reconnect(): + layer._require_pool_loop() + with _patch_redis(conn): + await layer.connect() + + asyncio.run(_reconnect()) + + old_pool.disconnect.assert_awaited_once() + assert layer._connected is True + assert layer.redis_pool is not old_pool + + # -- 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 + + def test_disconnect_failure_propagates_and_keeps_the_pool(self): + """A failed teardown must be loud and must not release the live pool. + + Clearing ownership before the disconnect succeeds — or swallowing its + failure — would leave a still-live pool unreachable: a silent leak. + State is released only once the pool has actually closed, so the + caller can retry. + """ + layer = RedisCacheLayer("L2") + layer._connected = True + pool = AsyncMock() + pool.disconnect = AsyncMock(side_effect=RuntimeError("teardown failed")) + layer.redis_pool = pool + + async def _own_then_fail_then_retry(): + layer._require_pool_loop() + loop = asyncio.get_running_loop() + + with pytest.raises(RuntimeError, match="teardown failed"): + await layer.disconnect() + + # Ownership and the pool survive the failure, so it can be retried. + assert layer.redis_pool is pool + assert layer._connected is True + assert layer._pool_loop is loop + + pool.disconnect = AsyncMock() + await layer.disconnect() + assert layer.redis_pool is None + assert layer._connected is False + assert layer._pool_loop is None + + asyncio.run(_own_then_fail_then_retry()) + + 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_warm_cache_does_not_swallow_the_ownership_error(self): + """warm_cache() gathers set() calls with return_exceptions=True. + + Without a re-raise, an ownership violation would be silently dropped + from the success count and warming would report a partial success + instead of surfacing the programming error. + """ + 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.warm_cache([("k", "v"), ("k2", "v2")]) + + 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