perf: invalidate Redis tags concurrently instead of serially

src/youtube_extension/backend/services/intelligent_cache.py

@@ -286,7 +286,12 @@ def __init__(self, name: str = "L2_Redis", redis_url: str = "redis://localhost:6
         self._tag_write_semaphore_loop: Optional[asyncio.AbstractEventLoop] = None
 
     def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
-        """Semaphore shared by every ``set()`` call on this layer.
+        """Semaphore shared by every tag fan-out on this layer.
+
+        Two paths acquire it: ``set()``, which issues one ``sadd`` per tag, and
+        ``invalidate_by_tags()``, which issues an ``smembers``/``delete`` pair
+        per tag. Both draw from the same budget, so a ``set()`` storm and an
+        invalidation storm cannot each claim ``_tag_write_limit`` connections.
 
         The limiter has to be per-instance rather than per-call: all callers
         share ``self.redis_pool``, so a per-call semaphore would let N
@@ -310,11 +315,12 @@ def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
         ``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.
+        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.
         """
         loop = asyncio.get_running_loop()
 
@@ -504,19 +510,51 @@ async def invalidate_by_tags(self, tags: list[str]) -> int:
 
         try:
             async with redis.Redis(connection_pool=self.redis_pool) as conn:
-                total_deleted = 0
+                semaphore = self._get_tag_write_semaphore()
+
+                async def _invalidate_tag(tag: str) -> int:
+                    # One permit is held for the whole smembers->delete pair
+                    # rather than re-acquired per command. This bounds the
+                    # number of tag invalidations in progress at once to the
+                    # permit count and keeps each tag's causally-ordered pair
+                    # (delete operates on the members smembers just returned) as
+                    # one indivisible unit of scheduled work. It also bounds how
+                    # many tags can sit half-invalidated if a delete fails.
+                    #
+                    # It does NOT lower peak pool-connection usage: redis.asyncio
+                    # checks a connection out only for the duration of each
+                    # command and returns it to the pool between the two awaits,
+                    # so acquiring the permit per command would cap in-flight
+                    # commands at the same limit. The reason to hold across the
+                    # pair is scheduling determinism and avoiding permit churn,
+                    # not preventing a doubling of held connections.
+                    async with semaphore:
+                        keys = await conn.smembers(f"uvai:tag:{tag}")
+
+                        if not keys:
+                            return 0
 
-                for tag in tags:
-                    # Get all keys with this tag
-                    keys = await conn.smembers(f"uvai:tag:{tag}")
-
-                    if keys:
                         # Delete cache entries
                         cache_keys = [f"uvai:cache:{key.decode()}" if isinstance(key, bytes) else f"uvai:cache:{key}" for key in keys]
                         stat_keys = [f"uvai:stats:{key.decode()}" if isinstance(key, bytes) else f"uvai:stats:{key}" for key in keys]
 
-                        deleted = await conn.delete(*(cache_keys + stat_keys + [f"uvai:tag:{tag}"]))
-                        total_deleted += deleted
+                        return await conn.delete(*(cache_keys + stat_keys + [f"uvai:tag:{tag}"]))
+
+                # return_exceptions=True so that one failing tag cannot leave
+                # sibling tasks still in flight once this method returns, which
+                # would let them touch conn after the enclosing async with has
+                # closed it. The first failure is re-raised below so the
+                # existing handler still reports 0.
+                results = await asyncio.gather(
+                    *(_invalidate_tag(tag) for tag in tags),
+                    return_exceptions=True,
+                )
+
+                total_deleted = 0
+                for result in results:
+                    if isinstance(result, BaseException):
+                        raise result
+                    total_deleted += result
 
                 logger.info(f"L2 Redis TAG INVALIDATION: {total_deleted} entries for tags {tags}")
                 return total_deleted

tests/unit/test_intelligent_cache.py

@@ -1593,6 +1593,239 @@ async def test_invalidate_exception_returns_zero(self):
 
         assert result == 0
 
+    async def test_invalidate_issues_tags_concurrently(self):
+        """Per-tag work must overlap rather than run one tag at a time.
+
+        This is the non-vacuity guard for the change: a serial ``for`` loop
+        yields a peak of exactly 1, so this assertion fails on the previous
+        implementation.
+        """
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracking_smembers(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            # Yield so sibling tags can start if the fan-out is concurrent.
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return {b"key1"}
+
+        conn.smembers = AsyncMock(side_effect=_tracking_smembers)
+        conn.delete = AsyncMock(return_value=1)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(5)])
+
+        assert peak > 1
+        assert conn.smembers.call_count == 5
+        assert result == 5
+
+    async def test_invalidate_stays_within_concurrency_bound(self):
+        """A large tag list must not fan out past the connection-pool budget."""
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracking_smembers(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return set()
+
+        conn.smembers = AsyncMock(side_effect=_tracking_smembers)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(50)])
+
+        assert result == 0
+        assert conn.smembers.call_count == 50
+        assert peak <= layer._tag_write_limit
+
+    async def test_invalidate_holds_one_permit_across_both_commands(self):
+        """smembers and delete for a tag are scheduled under one permit.
+
+        Holding the permit across the pair keeps each tag's invalidation as one
+        indivisible unit of scheduled work, so with a single permit the pairs
+        run to completion without interleaving. (This does not change peak pool
+        usage -- redis.asyncio returns a connection to the pool between the two
+        awaits -- it pins the per-tag scheduling policy so a later refactor
+        cannot silently split the pair.)
+        """
+        layer = self._connected_layer()
+        layer._tag_write_limit = 1
+        conn = _make_redis_conn()
+
+        order = []
+
+        async def _smembers(name, *_args, **_kwargs):
+            order.append(("smembers", name))
+            await asyncio.sleep(0)
+            return {b"key1"}
+
+        async def _delete(*args, **_kwargs):
+            order.append(("delete", args[-1]))
+            await asyncio.sleep(0)
+            return 1
+
+        conn.smembers = AsyncMock(side_effect=_smembers)
+        conn.delete = AsyncMock(side_effect=_delete)
+
+        with _patch_redis(conn):
+            await layer.invalidate_by_tags(["tag1", "tag2"])
+
+        # With one permit the pairs must not interleave.
+        assert order == [
+            ("smembers", "uvai:tag:tag1"),
+            ("delete", "uvai:tag:tag1"),
+            ("smembers", "uvai:tag:tag2"),
+            ("delete", "uvai:tag:tag2"),
+        ]
+
+    async def test_invalidate_cancellation_drains_before_conn_closes(self):
+        """Cancellation must unwind every child before the connection closes.
+
+        This is the explicit cancellation-parity claim: gather() does not
+        complete its outer future until every cancelled child has finished, so
+        the enclosing ``async with redis.Redis(...)`` cannot close ``conn``
+        while a child could still issue a command on it.
+        """
+        layer = self._connected_layer()
+        layer._tag_write_limit = 4
+        conn = _make_redis_conn()
+
+        events: list[tuple] = []
+        all_blocked = asyncio.Event()
+        entered = 0
+
+        async def _blocking_smembers(name, *_args, **_kwargs):
+            nonlocal entered
+            events.append(("cmd", "smembers", name))
+            entered += 1
+            if entered == 3:
+                all_blocked.set()
+            try:
+                await asyncio.sleep(3600)
+                return {b"key1"}
+            finally:
+                events.append(("unwind", name))
+
+        async def _delete(*args, **_kwargs):
+            events.append(("cmd", "delete", args[-1]))
+            return 1
+
+        async def _aexit(*_args, **_kwargs):
+            events.append(("aexit",))
+            return False
+
+        conn.smembers = AsyncMock(side_effect=_blocking_smembers)
+        conn.delete = AsyncMock(side_effect=_delete)
+        conn.__aexit__ = AsyncMock(side_effect=_aexit)
+
+        with _patch_redis(conn):
+            task = asyncio.create_task(layer.invalidate_by_tags(["t1", "t2", "t3"]))
+            try:
+                await asyncio.wait_for(all_blocked.wait(), timeout=5)
+            except asyncio.TimeoutError:
+                task.cancel()
+                try:
+                    await task
+                except BaseException:
+                    pass
+                pytest.fail(
+                    f"tags were not issued concurrently; only {entered} in flight"
+                )
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+        kinds = [e[0] for e in events]
+        assert "aexit" in kinds, f"connection never closed: {events}"
+        aexit_idx = kinds.index("aexit")
+
+        # Every child finished its finally path before the connection closed.
+        assert kinds.count("unwind") == 3, f"not all children unwound: {events}"
+        assert all(
+            i < aexit_idx for i, e in enumerate(events) if e[0] == "unwind"
+        ), f"a child unwound after conn close: {events}"
+
+        # No Redis command was issued after the connection closed.
+        assert all(
+            i < aexit_idx for i, e in enumerate(events) if e[0] == "cmd"
+        ), f"command issued after conn close: {events}"
+
+    async def test_invalidate_failure_drains_in_flight_work(self):
+        """A failing tag returns 0 with no per-tag task left in flight."""
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        started = 0
+        finished = 0
+
+        async def _flaky_smembers(name, *_args, **_kwargs):
+            nonlocal started, finished
+            started += 1
+            await asyncio.sleep(0)
+            finished += 1
+            if name.endswith("tag3"):
+                raise RuntimeError("redis unavailable")
+            return set()
+
+        conn.smembers = AsyncMock(side_effect=_flaky_smembers)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(6)])
+
+        assert result == 0
+        # Every scheduled tag ran to completion before the method returned.
+        assert started == 6
+        assert finished == started
+
+    async def test_invalidate_shares_tag_write_budget_with_set(self):
+        """set() and invalidate_by_tags() must draw from one shared budget.
+
+        Both hold connections from the same pool, so separate budgets would let
+        a concurrent set storm and invalidation storm each claim the full limit.
+        """
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracked(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return 1
+
+        async def _tracked_smembers(*_args, **_kwargs):
+            await _tracked()
+            return set()
+
+        conn.sadd = AsyncMock(side_effect=_tracked)
+        conn.smembers = AsyncMock(side_effect=_tracked_smembers)
+
+        with _patch_redis(conn):
+            await asyncio.gather(
+                layer.set("k", "v", tags=[f"s{i}" for i in range(40)]),
+                layer.invalidate_by_tags([f"i{i}" for i in range(40)]),
+            )
+
+        assert conn.sadd.call_count == 40
+        assert conn.smembers.call_count == 40
+        assert peak <= layer._tag_write_limit
+
 
 class TestRedisCacheLayerUpdateAvgAccessTime:
     """Tests for RedisCacheLayer._update_avg_access_time() — lines 442-450"""