Concurrently write Redis tag sets with bounded fan-out

[groupthinking]

Speed up RedisCacheLayer.set() by issuing per-tag SADD writes concurrently while bounding in-flight commands to protect the connection pool.

[linear-code]

GRV-213

[github-actions]

👋 Thanks for opening this issue! It has been automatically labeled for triage.

[coderabbitai]

⚠️ Possible Duplicate Issue(s)

• perf: RedisCacheLayer.set() issues one sequential Redis round trip per tag #1153

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!

[groupthinking]

Triaged this while working the #1162 follow-up. This is already implemented on main — it landed in #1152, which merged 9 seconds before this issue was filed (2026-08-01T20:39:06Z vs 20:39:15Z), so this looks like an automation artifact that raced the merge rather than a genuine gap.

Evidence, all in src/youtube_extension/backend/services/intelligent_cache.py at ad7e2c10:

Concurrent fan-out — set(), ~L430-443. Per-tag SADD is issued concurrently instead of one await per tag:

if tags:
    semaphore = self._get_tag_write_semaphore()

    async def _add_tag(tag: str) -> None:
        async with semaphore:
            await conn.sadd(f"uvai:tag:{tag}", key)

    results = await asyncio.gather(
        *(_add_tag(tag) for tag in tags),
        return_exceptions=True,
    )
    for result in results:
        if isinstance(result, BaseException):
            raise result

Bounded, pool-aware — L38-57:

TAG_WRITE_CONCURRENCY = 8
TAG_WRITE_POOL_RESERVE = 4

def _resolve_tag_write_limit(max_connections: int) -> int:
    return max(1, min(TAG_WRITE_CONCURRENCY, max_connections - TAG_WRITE_POOL_RESERVE))

The cap is derived from the layer's own max_connections (self._tag_write_limit, L337 on my branch), reserving headroom for the SET/SETEX and HSET issued by the same set() call plus concurrent get()/delete() traffic. It degrades to serial writes for small pools rather than saturating them, and never drops below 1.

Checks off both halves of the ask:

• per-tag SADD issued concurrently
• in-flight commands bounded to protect the connection pool
• semaphore is per-instance, so many concurrent set() callers share one budget rather than each getting their own
• return_exceptions=True means no tag write is still in flight when set() returns; the first failure is re-raised to preserve the original False-on-error contract

One thing worth knowing if anyone picks this up anyway: the semaphore is lazily created and rebuilt if its owning loop closes. That's a consequence of the layer-wide event-loop ownership contract added in #1162, not a competing mechanism — see #1179.

Recommend closing as already-shipped. Happy to be corrected if the ask here was actually broader than what #1152 delivered.

[groupthinking]

Closing as already implemented — verified against the code on main, not assumed.

What this issue asked for

Speed up RedisCacheLayer.set() by issuing per-tag SADD writes concurrently while bounding in-flight commands to protect the connection pool.

Where it landed

RedisCacheLayer.set() in src/youtube_extension/backend/services/intelligent_cache.py:434-447:

if tags:
    semaphore = self._get_tag_write_semaphore()

    async def _add_tag(tag: str) -> None:
        async with semaphore:
            await conn.sadd(f"uvai:tag:{tag}", key)

    results = await asyncio.gather(
        *(_add_tag(tag) for tag in tags),
        return_exceptions=True,
    )
    for result in results:
        if isinstance(result, BaseException):
            raise result

Both halves of the request are satisfied:

• Concurrent — per-tag SADD calls are issued under a single asyncio.gather instead of sequentially.
• Bounded — every write acquires a semaphore sized by _resolve_tag_write_limit(max_connections) (line 288), which derives from TAG_WRITE_CONCURRENCY and scales down for small pools. The bound is per layer, not per call, so neither a long tag list nor many concurrent writers can exhaust the connection pool.

Two details worth noting, both intentional and commented in-code:

• return_exceptions=True guarantees no tag write is still in flight when the method returns; the first failure is then re-raised to preserve the original contract (the handler logs and returns False).
• The semaphore is rebuilt if the running event loop changes, so a semaphore bound to a closed loop is never reused.

Landed in ad7e2c1 — perf: issue Redis tag-set writes concurrently on cache set (#1152), merged 2026-08-01.

Test evidence (run just now)

tests/unit/test_intelligent_cache.py — 157 passed, 0 failed, of which 5 target the tag-write bound directly: the observed concurrency peak never exceeds _tag_write_limit, the limit stays below max_connections, it scales down for small pools, and the semaphore is replaced after its loop closes.

Related

TAG_WRITE_CONCURRENCY is being moved onto a shared, validated env-parsing helper in #1180 (PR #1220), which also makes it tunable without a release. That preserves the bounded-fan-out behaviour above exactly — the 157 tests here pass unchanged on that branch.

Why this was still open

Auto-filed alongside its own fix PR and never closed when that PR merged, matching #1191, #1192, #1198, #1199, #1201 and #1206. No product defect remains.