perf: issue Redis tag-set writes concurrently on cache set - #1152
Conversation
RedisCacheLayer.set() awaited one sadd per tag in a sequential loop, so a value written with N tags paid N serial Redis round trips on top of the set/setex and hset calls. This is a per-write hot path for every cached entry, so the added latency scales linearly with tag count. Issue the sadd calls with asyncio.gather so they are pipelined onto the connection concurrently, reducing the tag phase from N round trips to roughly one. Behaviour is unchanged: the same keys and members are written, and the untagged case now short-circuits instead of entering the loop. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
Important Review skippedAuto reviews are limited based on label configuration. 🏷️ Required labels (at least one) (1)
Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Repository YAML (base), Repository UI (inherited), Organization UI (inherited) Review profile: ASSERTIVE Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughSummary by CodeRabbit
WalkthroughRedis cache tag updates now execute concurrently through bounded tasks. The limit is eight active Redis tag writes. The cache write awaits all tag updates and routes failures through the existing error handler. ChangesRedis tag-write concurrency
Estimated code review effort: 2 (Simple) | ~10 minutes Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 4❌ Failed checks (2 warnings, 2 inconclusive)
✅ Passed checks (3 passed)
✨ Finishing Touches 💡 2📝 Generate docstrings 💡
🛠️ Fix failing CI checks 💡
🧪 Generate unit tests (beta)
✨ Simplify code
Warning Review ran into problems🔥 ProblemsThese MCP integrations need to be re-authenticated in the Integrations settings: Sentry Linked repositories: Public OSS repositories can only analyze public repositories installed in this organization. No linked repositories were analyzed; skipped Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Snapshot WarningsEnsure that dependencies are being submitted on PR branches and consider enabling retry-on-snapshot-warnings. See the documentation for more information and troubleshooting advice. Scanned FilesNone |
|
@coderabbitai review Focused perf change: Please confirm:
Note: I deliberately did not convert this to a Redis pipeline (which would collapse all 3+N round trips into one) because the existing test suite mocks individual command methods and has no |
Agent Completion Truth Gate: NOT_APPLICABLEEvidence agrees. Machine-readable verdict{
"details": {},
"reasons": [],
"verdict": "not_applicable"
} |
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
echo '=== changed files ==='
git diff --name-only HEAD^ HEAD
echo '=== intelligent_cache outline ==='
file=$(fd -a '^intelligent_cache\.py$' . | head -n 1)
printf 'file=%s\n' "$file"
ast-grep outline "$file" --items all --match 'RedisCacheLayer|set' || true
echo '=== PR diff for cache file ==='
git diff HEAD^ HEAD -- "$file"
echo '=== RedisCacheLayer implementation and nearby client lifecycle ==='
rg -n -A100 -B25 'class RedisCacheLayer\b|async def set\s*\(' "$file"
echo '=== cache tests and their tag-write assertions ==='
fd -a '^test_intelligent_cache.*\.py$' . -x sh -c 'echo "--- $1"; rg -n -A12 -B8 "RedisCacheLayer|sadd|\\.set\\(" "$1" || true' sh {}
echo '=== Redis package declarations ==='
rg -n -i -C2 'redis([<=>!~ ]|$)|redis-py|redis\.asyncio' \
pyproject.toml poetry.lock requirements.txt requirements/*.txt setup.cfg setup.py 2>/dev/null || trueLength of output: 22541 🤖 web_search executed: `For redis-py 5.x redis.asyncio.Redis constructed with connection_pool, are concurrent execute_command calls from the same Redis client safe? Does each command acquire a connection from the pool by default, and what is the behavior of Redis.aclose() when the pool was supplied by the caller? Python asyncio.gather default failure semantics: when one awaited coroutine raises, are the remaining coroutines cancelled or do they continue running?` Length of output: 14968
The However, The default I recommend a bounded-concurrency implementation, or preferably a Redis pipeline as a follow-up. A pipeline uses one pool connection for the tag phase and removes the pool-exhaustion regression. The test-mock churn is worth it if tag counts can be large or are caller-controlled. For exception handling, the surrounding If I would not block this focused performance change if You are interacting with an AI system. |
There was a problem hiding this comment.
🟡 Not ready to approve
Concurrent tag writes can exhaust the 20-connection Redis pool and leave partially written cache metadata.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.
Pull request overview
Optimizes Redis cache tag writes, but the unbounded concurrency can exhaust the configured connection pool.
Changes:
- Replaces sequential tag writes with
asyncio.gather. - Skips tag processing when no tags exist.
File summaries
| File | Description |
|---|---|
src/youtube_extension/backend/services/intelligent_cache.py |
Concurrently writes Redis tag memberships. |
Review details
- Files reviewed: 1/1 changed files
- Comments generated: 1
- Review effort level: Balanced
We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.
groupthinking
left a comment
groupthinking
left a comment
There was a problem hiding this comment.
Review — perf: issue Redis tag-set writes concurrently on cache set
Verdict: looks correct. The change is behaviour-preserving and the concurrency is safe. One low-severity trade-off worth a note.
Concurrency safety (the part that looks risky) ✅
The obvious concern with asyncio.gather over redis-py commands is protocol corruption from interleaving multiple commands on one socket. That does not apply here: conn is a standard redis.Redis(connection_pool=self.redis_pool) client, not a pipeline or a pinned single-connection context. In redis-py asyncio, a standard client's execute_command acquires its own connection from the pool per command and releases it afterward, so the gathered sadds run on separate pooled connections — no shared read/write buffer. The concurrency is safe and the "N round trips → ~1" saving is real.
Failure semantics are unchanged: gather propagates the first exception, the existing try/except catches it and returns False, and a partial set of tags could already be written — exactly as the sequential loop behaved.
One trade-off to consider (low severity) ⚠️
Peak connection usage per set() goes from 1 (sequential) to len(tags) (concurrent), against a pool created with max_connections=20 (intelligent_cache.py:257,267). Because this is a non-blocking ConnectionPool, a set() whose tag count approaches/exceeds the available pool slots (a very-high-tag entry, or fewer tags under concurrent Redis load elsewhere) could raise ConnectionError("Too many connections"). It degrades gracefully — caught by the same try/except, so the write returns False rather than erroring out — but that is a new way for a write to fail that the sequential loop couldn't hit.
For typical small tag counts this is a non-issue. If high-tag-count entries are expected, either bound the fan-out with an asyncio.Semaphore or move to the Redis pipeline the PR explicitly (and reasonably) deferred. Not a blocker.
CI note
The red combined status is only the canceled Vercel deployment, which the PR correctly documents as N/A for this backend Redis path. CodeRabbit was skipped by label config; the agent-completion truth-gate passed.
Verification
Verified by static analysis of the diff and surrounding set() method. I could not re-run the 172-test suite in the review sandbox (pytest not installed here), so I'm relying on that static review plus the author's reported 172 passed and the required test check.
Generated by Claude Code
Address review feedback on the concurrent tag-set write path. redis-py's async connection pool defaults to max_connections=20 and each in-flight command holds one connection, so an unbounded asyncio.gather over the tag list could exhaust the pool and raise ConnectionError on writes with many tags - a failure mode the original sequential loop did not have. - Bound fan-out with a semaphore (TAG_WRITE_CONCURRENCY = 8), well below the pool default so concurrent callers retain capacity. - Use return_exceptions=True and await every task, so no tag write is left in flight when set() returns; the first failure is re-raised to preserve the existing return-False-on-error contract. - Add regression tests asserting peak concurrency stays within the bound and that a failing tag write drains all siblings before returning. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
|
@coderabbitai Thanks — the pool-exhaustion point is correct and I've addressed it in You're right that What changed:
Regression tests added (
I verified the first test is non-vacuous: with the bound raised to an effectively-unbounded value, peak in-flight connections reaches 50 against a pool of 20 — exactly the exhaustion you flagged. With the bound in place it stays at 8. Local: On the pipeline suggestion — agreed it's the better long-term shape, but it's out of scope here: the existing tests mock individual commands ( Please re-review. |
🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
cache_file="src/youtube_extension/backend/services/intelligent_cache.py"
test_file="tests/unit/test_intelligent_cache.py"
echo '=== current commit and relevant diff ==='
git rev-parse --short HEAD
git show --stat --oneline --decorate HEAD
git diff HEAD^ HEAD -- "$cache_file" "$test_file"
echo '=== implementation outline ==='
ast-grep outline "$cache_file" --items all --match 'RedisCacheLayer|TAG_WRITE_CONCURRENCY'
echo '=== RedisCacheLayer.set implementation and concurrency constant ==='
rg -n -A150 -B20 \
'TAG_WRITE_CONCURRENCY|class RedisCacheLayer\b|async def set\s*\(' \
"$cache_file"
echo '=== regression test definitions and nearby fixtures ==='
rg -n -A110 -B20 \
'async def test_set_tag_writes_stay_within_concurrency_bound\b|async def test_set_tag_write_failure_returns_false_and_drains\b|TAG_WRITE_CONCURRENCY' \
"$test_file"
echo '=== all direct Redis pool configuration and cache-layer construction sites ==='
rg -n -C3 \
'ConnectionPool\s*\(|max_connections|RedisCacheLayer\s*\(' \
src testsLength of output: 35052 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo '=== locate relevant test files ==='
fd -a -i '^test_intelligent_cache.*\.py$' .
echo '=== locate all references to the new concurrency constant ==='
rg -n -C4 '\bTAG_WRITE_CONCURRENCY\b' .
echo '=== RedisCacheLayer construction and set call sites ==='
rg -n -C3 '\bRedisCacheLayer\s*\(|\.\s*set\s*\(' src tests 2>/dev/null || true
echo '=== relevant test blocks and support fixtures ==='
fd -a -i '^test_intelligent_cache.*\.py$' . -x sh -c '
echo "--- $1"
rg -n -A115 -B35 \
"def _connected_layer\b|def _make_redis_conn\b|def _patch_redis\b|test_set_tag_writes_stay_within_concurrency_bound\b|test_set_tag_write_failure_returns_false_and_drains\b" \
"$1"
' sh {}
echo '=== Redis client context-manager behavior in the installed dependency source, if present ==='
python - <<'PY'
from pathlib import Path
for root in (Path('.'), Path('/usr/local/lib')):
if root.exists():
for candidate in root.rglob('redis/asyncio/client.py'):
print(candidate)
PYLength of output: 16919
A blocking pool-exhaustion path remains.
For example, three concurrent Move the semaphore to The current tests validate per-call fan-out and drain behavior. They do not validate the shared-pool bound. ✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/youtube_extension/backend/services/intelligent_cache.py`:
- Line 378: Update RedisCacheLayer.connect() so ConnectionPool.from_url
configures both socket_timeout and socket_connect_timeout alongside the existing
pool options. Add coverage proving a set() timeout returns False and that
concurrent tag tasks have completed or been cancelled when the call path
returns.
- Around line 38-43: Move the tag-write semaphore from per-call construction in
IntelligentCacheSystem.set() into RedisCacheLayer.__init__, sizing its capacity
from the Redis pool’s max_connections while reserving capacity for non-tag
commands and never exceeding the pool limit. Reuse this shared limiter for all
set() tag writes so concurrent calls are collectively bounded, and add
regression coverage for concurrent set() calls and pools smaller than eight.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Repository UI (inherited), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 1fbb3d45-830e-444e-b6c0-437c85ab7e1a
⛔ Files ignored due to path filters (1)
tests/unit/test_intelligent_cache.pyis excluded by!tests/**
📒 Files selected for processing (1)
src/youtube_extension/backend/services/intelligent_cache.py
📜 Review details
⏰ Context from checks skipped due to timeout. (5)
- GitHub Check: Generate and Upload Coverage
- GitHub Check: Security Scan - python
- GitHub Check: build
- GitHub Check: test
- GitHub Check: trivy
⚠️ CI failures not shown inline (5)
GitHub Actions: Secret Scan / gitleaks (working tree): perf: issue Redis tag-set writes concurrently on cache set
Conclusion: failure
##[group]Run gitleaks detect --no-git --config .gitleaks.toml --redact --verbose --exit-code 1
�[36;1mgitleaks detect --no-git --config .gitleaks.toml --redact --verbose --exit-code 1�[0m
shell: /usr/bin/bash -e {0}
##[endgroup]
○
│╲
│ ○
○ ░
░ gitleaks
Finding: ...gz", hash = "sha256:�[1;3;mREDACTED�[0m, size = 401824, upl...
***REDACTED_SECRET_ASSIGNMENT***
RuleID: square-access-token
Entropy: 3.884400
File: uv.lock
Line: 5129
Fingerprint: uv.lock:square-access-***REDACTED_SECRET_ASSIGNMENT***
�[90m7:50PM�[0m �[32mINF�[0m scan completed in 5.81s
�[90m7:50PM�[0m �[31mWRN�[0m leaks found: 1
##[error]Process completed with exit code 1.
GitHub Actions: Secret Scan / 0_gitleaks (working tree).txt: perf: issue Redis tag-set writes concurrently on cache set
Conclusion: failure
##[group]Run gitleaks detect --no-git --config .gitleaks.toml --redact --verbose --exit-code 1
�[36;1mgitleaks detect --no-git --config .gitleaks.toml --redact --verbose --exit-code 1�[0m
shell: /usr/bin/bash -e {0}
##[endgroup]
○
│╲
│ ○
○ ░
░ gitleaks
Finding: ...gz", hash = "sha256:�[1;3;mREDACTED�[0m, size = 401824, upl...
***REDACTED_SECRET_ASSIGNMENT***
RuleID: square-access-token
Entropy: 3.884400
File: uv.lock
Line: 5129
Fingerprint: uv.lock:square-access-***REDACTED_SECRET_ASSIGNMENT***
�[90m7:50PM�[0m �[32mINF�[0m scan completed in 5.81s
�[90m7:50PM�[0m �[31mWRN�[0m leaks found: 1
##[error]Process completed with exit code 1.
GitHub Actions: Agent completion enforcement / 0_Agent completion enforcement.txt: perf: issue Redis tag-set writes concurrently on cache set
Conclusion: failure
##[group]Run actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3
with:
script: const fs = require('fs');
const pull = await github.rest.pulls.get({
owner: context.repo.owner,
repo: context.repo.repo,
pull_number: Number(process.env.PR)
});
let verdict = {
conclusion: 'failure',
reason: 'verifier_did_not_publish',
details: {}
};
try {
verdict = JSON.parse(fs.readFileSync(
'enforcement-verdict.json', 'utf8'
));
} catch (error) {
core.warning(error.message);
}
const conclusion = verdict.conclusion === 'success'
? 'success'
: 'failure';
const summary = JSON.stringify(verdict);
await github.rest.checks.create({
owner: context.repo.owner,
repo: context.repo.repo,
name: 'Agent completion enforcement',
head_sha: pull.data.head.sha,
status: 'completed',
conclusion,
output: {
title: conclusion === 'success'
? 'Trusted evidence verified'
: 'Trusted evidence blocked',
summary: summary.slice(0, 60000)
}
});
if (conclusion !== 'success') {
core.setFailed(verdict.reason || 'trusted evidence blocked');
}
github-***REDACTED_SECRET_ASSIGNMENT***
debug: false
user-agent: actions/github-script
result-encoding: json
retries: 0
retry-exempt-status-codes: 400,401,403,404,422
env:
PR: 1152
##[endgroup]
##[error]missing_trusted_publication
GitHub Actions: Agent completion enforcement / Agent completion enforcement: perf: issue Redis tag-set writes concurrently on cache set
Conclusion: failure
##[group]Run actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3
with:
script: const fs = require('fs');
const pull = await github.rest.pulls.get({
owner: context.repo.owner,
repo: context.repo.repo,
pull_number: Number(process.env.PR)
});
let verdict = {
conclusion: 'failure',
reason: 'verifier_did_not_publish',
details: {}
};
try {
verdict = JSON.parse(fs.readFileSync(
'enforcement-verdict.json', 'utf8'
));
} catch (error) {
core.warning(error.message);
}
const conclusion = verdict.conclusion === 'success'
? 'success'
: 'failure';
const summary = JSON.stringify(verdict);
await github.rest.checks.create({
owner: context.repo.owner,
repo: context.repo.repo,
name: 'Agent completion enforcement',
head_sha: pull.data.head.sha,
status: 'completed',
conclusion,
output: {
title: conclusion === 'success'
? 'Trusted evidence verified'
: 'Trusted evidence blocked',
summary: summary.slice(0, 60000)
}
});
if (conclusion !== 'success') {
core.setFailed(verdict.reason || 'trusted evidence blocked');
}
github-***REDACTED_SECRET_ASSIGNMENT***
debug: false
user-agent: actions/github-script
result-encoding: json
retries: 0
retry-exempt-status-codes: 400,401,403,404,422
env:
PR: 1152
##[endgroup]
##[error]missing_trusted_publication
Commit Status: Vercel: Vercel
Conclusion: failure
Canceled from the Vercel Dashboard
🧰 Additional context used
📓 Path-based instructions (10)
**/*.{py,js,jsx,ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{py,js,jsx,ts,tsx}: Use Python 3.9+ and Node 18+ for development
Never hardcode API keys, database URLs, or secrets in code
Make minimal, surgical changes and avoid deleting working code unless fixing security issues
Files:
src/youtube_extension/backend/services/intelligent_cache.py
**/*.py
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.py: Always use type hints for Python functions
Use Black formatter with 88 character line length for Python code
Follow PEP 8 conventions for Python code
Use descriptive variable names and add docstrings to all public functions in Python
Group imports in Python: standard library, third-party, local
Use SQLAlchemy ORM and never write raw SQL queries
Use environment variables via os.getenv() or pydantic-settings to access configuration
Wrap database operations in try-except blocks and use context managers or FastAPI dependencies for connection cleanup
Implement comprehensive error handling with proper logging in all functions
Version APIs using /api/v1/ prefix for stability
Use JSON-RPC 2.0 protocol for all MCP communication
Follow the single-flow workflow: YouTube link → context extraction → agent dispatch → outputs
Use context managers for resource management in Python code
Implement comprehensive input validation and sanitize outputs for security
Use parameterized queries and SQLAlchemy ORM to prevent SQL injection
Define API request/response models using Pydantic for FastAPI endpoints
Store the single unified workflow as the only workflow; never introduce alternate flows or manual triggers
Use SQLAlchemy with connection pooling for database connections and manage sessions with context managers
Provide sensible defaults for non-sensitive configuration in Python settings
Maintain backward compatibility and do not break existing API endpoints
Include comprehensive logging for debugging in MCP implementations
Files:
src/youtube_extension/backend/services/intelligent_cache.py
⚙️ CodeRabbit configuration file
Python backend code. Check for type hints, proper exception handling, async context manager usage, and potential blocking calls in async functions. Flag any bare except clauses or missing timeout parameters on network calls. CRITICAL: Flag any file that contains placeholder/stub implementations — especially in unified_ai_sdk. Any class or function that says "TODO: Replace with production implementation" or returns mock/fake data must be flagged as a blocking issue. Flag any code generation output that reaches users without AST validation or syntax checking.
Files:
src/youtube_extension/backend/services/intelligent_cache.py
**/*.{py,js,ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Maintain >80% code coverage for new features
Files:
src/youtube_extension/backend/services/intelligent_cache.py
**/*.{py,ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{py,ts,tsx}: Keep frontend and backend data models synchronized using matching Pydantic (backend) and TypeScript (frontend) interfaces
Use type-safe interfaces for backend-frontend data exchange
**/*.{py,ts,tsx}: Production code must use real behavior only: no mock delays, fake data, or simulated responses.
Maintain strict type safety: mypy strict mode for Python and TypeScript strict mode for the frontend.
Name events using the<domain>.<entity>.<action>format.
Files:
src/youtube_extension/backend/services/intelligent_cache.py
**/*
📄 CodeRabbit inference engine (Custom checks)
**/*: Strictly verify that GitHub Copilot has explicitly reviewed and approved the pull request; human approvals alone must not satisfy this check.
Before allowing a merge, require thecopilot-rabbitlabel and AI-generated unit tests committed alongside the code changes; fail the check if either is missing.For Vercel-specific work, include
https://vercel.com/docs/llms-full.txtin the AI assistant context set.
Files:
src/youtube_extension/backend/services/intelligent_cache.py
**/*.{py,pyw}
📄 CodeRabbit inference engine (AGENTS.md)
Write Python code to remain compatible with Linux and Windows where possible, including correct handling of
asyncioevent loops.
Files:
src/youtube_extension/backend/services/intelligent_cache.py
src/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
src/**/*.py: Format Python code with Black using an 88-character line length.
Sort Python imports with isort using the Black profile.
Run Ruff with E, W, F, I, B, C4, and UP rules; E501 is ignored.
Use mypy strict mode; untyped function definitions are disallowed.
Target Python 3.9 or newer.
Validate backend inputs with Pydantic and sanitize subprocess arguments.
Use Anthropic SDK featuresthinking={"type": "adaptive"}andoutput_config={"effort": "..."}withanthropic>=0.105.0; do not addTypeErrorfallbacks for these parameters.Use the service container dependency-injection pattern in
backend/containers/.
Files:
src/youtube_extension/backend/services/intelligent_cache.py
**/*.{py,ts,tsx,js,jsx}
📄 CodeRabbit inference engine (CLAUDE.md)
Do not commit secrets; store keys and credentials in gitignored
.envfiles.
Files:
src/youtube_extension/backend/services/intelligent_cache.py
**/*.{py,pyi}
📄 CodeRabbit inference engine (GEMINI.md)
**/*.{py,pyi}: Use Black formatting with an 88-character line length for Python code.
Use Ruff with rules E, W, F, I, B, C4, and UP for Python linting.
Use strict mypy type checking; do not define untyped functions.
Target Python 3.9 or newer.
Validate Python inputs with Pydantic.
Sanitize subprocess arguments before execution.
Do not add mock delays or fake data to production Python code; production runs in REAL_MODE_ONLY.
Keep secrets out of Python source code; load keys and credentials from environment variables instead.
Use absolute imports that resolve withPYTHONPATH=srcin the Python backend.
Files:
src/youtube_extension/backend/services/intelligent_cache.py
**/*.{py,pyi,ts,tsx}
📄 CodeRabbit inference engine (GEMINI.md)
**/*.{py,pyi,ts,tsx}: Preserve the single workflow: YouTube link → transcript → events → agents → outputs; do not introduce alternative flows or manual triggers that bypass it.
Use event names following<domain>.<entity>.<action>, such asyoutube.video.captured.
Make surgical, precise changes and do not delete working code without justification.
Files:
src/youtube_extension/backend/services/intelligent_cache.py
🔍 Remote MCP GitHub Copilot
Relevant review context
- The semaphore is created inside each
RedisCacheLayer.set()call, so it limits only per-call fan-out.warm_cache()already launches multipleset()calls viaasyncio.gather; three concurrent writes could issue up to 24SADDs against the shared default pool of 20 connections. max_connectionsis configurable, butTAG_WRITE_CONCURRENCYis fixed at 8. A custom pool configured below 8 may still be exhausted by one call.- Added tests cover one call with 50 tags and failure draining, but do not test aggregate concurrency across concurrent
set()calls or custom pool sizes. - The latest review explicitly identifies the shared-pool issue and recommends moving the semaphore to
__init__plus adding an aggregate-concurrency regression test. - At retrieval time, live checks were not uniformly green:
gitleaksand Agent Completion Enforcement had failed, while build/test/security/coverage checks were still in progress.
🔇 Additional comments (1)
src/youtube_extension/backend/services/intelligent_cache.py (1)
369-372: LGTM!Also applies to: 380-386
The previous bound was created per set() call, so it limited fan-out within one call but not across callers. Every caller draws from the same redis_pool, so N concurrent writers could each run TAG_WRITE_CONCURRENCY commands and still exhaust it. warm_cache() gathers set() calls, so this was reachable rather than hypothetical. Move the semaphore to the layer so all set() calls share one budget, and scale the cap from max_connections so small pools degrade to serial tag writes instead of saturating themselves. The semaphore is created lazily rather than in __init__ because this module builds an IntelligentCacheSystem singleton at import time, outside any event loop; an asyncio.Semaphore binds to the first loop that uses it and raises for every other one. Measured with 5 concurrent set() calls x 20 tags against a pool of 20: per-call budget peaked at 40 in-flight writes, shared budget peaks at 8. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
|
@coderabbitai You're right, and thanks for catching it — the per-call semaphore bounded fan-out within a call but not across callers, and Fixed in Shared budget. The semaphore now lives on the layer, so every Sized from return max(1, min(TAG_WRITE_CONCURRENCY, max_connections - TAG_WRITE_POOL_RESERVE))The reserve keeps connections free for the One wrinkle worth flagging. I could not build the semaphore in Regression tests — 3 added, and the shared-bound one is measurably non-vacuous. Simulating the old per-call semaphore under 5 concurrent
New tests:
|
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
cache_file="src/youtube_extension/backend/services/intelligent_cache.py"
test_file="tests/unit/test_intelligent_cache.py"
echo '=== current revision and target commit ==='
git rev-parse --short HEAD
git show --no-patch --format='%H%n%s' 26543a3
echo '=== changes in target commit ==='
git show --format=fuller --stat 26543a3
git diff 26543a3^ 26543a3 -- "$cache_file" "$test_file"
echo '=== cache layer structure ==='
ast-grep outline "$cache_file" --items all --match 'RedisCacheLayer|IntelligentCacheSystem|TAG_WRITE'
echo '=== limiter, pool setup, constructor, and set implementation ==='
rg -n -A150 -B30 \
'TAG_WRITE_CONCURRENCY|TAG_WRITE_POOL_RESERVE|_resolve_tag_write_limit|_get_tag_write_semaphore|class RedisCacheLayer\b|def __init__\s*\(|async def set\s*\(|ConnectionPool\s*\(' \
"$cache_file"
echo '=== warm_cache concurrent set call context ==='
rg -n -A70 -B25 'async def warm_cache\s*\(|asyncio\.gather\s*\(' "$cache_file"
echo '=== named regression tests and fixtures ==='
rg -n -A140 -B35 \
'test_concurrent_sets_share_one_tag_write_budget|test_tag_write_limit_scales_down_for_small_pools|test_tag_write_semaphore_rebinds_across_event_loops|def _connected_layer\b|def _make_redis_conn\b|def _patch_redis\b' \
"$test_file"
echo '=== all cache layer instantiation and pool-size configuration ==='
rg -n -C4 'RedisCacheLayer\s*\(|max_connections\s*=|ConnectionPool\s*\(' src testsLength of output: 52010 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo '=== locate cache tests ==='
fd -a -i 'test_intelligent_cache.py' .
echo '=== inspect named tests at their actual path ==='
test_file=$(fd -a -i 'test_intelligent_cache.py' . | head -n 1)
if [ -n "$test_file" ]; then
printf 'test_file=%s\n' "$test_file"
rg -n -A145 -B35 \
'test_concurrent_sets_share_one_tag_write_budget|test_tag_write_limit_scales_down_for_small_pools|test_tag_write_semaphore_rebinds_across_event_loops|def _connected_layer\b|def _make_redis_conn\b|def _patch_redis\b' \
"$test_file"
fi
echo '=== redis dependency declaration ==='
rg -n -i -C3 '(^|[^A-Za-z])redis([<=>!~ ]|$)|redis-py' \
pyproject.toml poetry.lock requirements.txt requirements setup.cfg setup.py 2>/dev/null || true
echo '=== installed redis asyncio pool/client source locations ==='
python - <<'PY'
import sysconfig
from pathlib import Path
roots = [Path(sysconfig.get_paths()["purelib"]), Path(sysconfig.get_paths()["platlib"])]
seen = set()
for root in roots:
for rel in ("redis/asyncio/connection.py", "redis/asyncio/client.py"):
p = root / rel
if p.is_file() and p not in seen:
print(p)
seen.add(p)
PY
echo '=== relevant event-loop ownership code in installed redis source ==='
python - <<'PY'
import sysconfig
from pathlib import Path
roots = [Path(sysconfig.get_paths()["purelib"]), Path(sysconfig.get_paths()["platlib"])]
for root in roots:
for rel in ("redis/asyncio/connection.py", "redis/asyncio/client.py"):
p = root / rel
if p.is_file():
print(f"--- {p} ---")
text = p.read_text(errors="replace").splitlines()
for i, line in enumerate(text, 1):
if any(term in line for term in ("get_running_loop", "_available_connections", "_in_use_connections", "asyncio.Lock", "asyncio.Queue")):
start, end = max(1, i - 8), min(len(text), i + 16)
for n in range(start, end + 1):
print(f"{n}:{text[n-1]}")
print()
PYLength of output: 2985 🏁 Script executed: #!/bin/bash
set -euo pipefail
commit="26543a3"
echo '=== tracked test paths in commit ==='
git ls-tree -r --name-only "$commit" | rg -i '(^|/)test_.*intelligent.*cache.*\.py$|(^|/)intelligent_cache.*test.*\.py$' || true
echo '=== exact files changed in commit ==='
git diff-tree --no-commit-id --name-status -r "$commit"
echo '=== test file content from commit, if tracked ==='
git ls-tree -r --name-only "$commit" | while IFS= read -r path; do
case "$path" in
*test_intelligent_cache.py)
echo "--- $path ---"
git show "${commit}:${path}" | rg -n -A145 -B35 \
'test_concurrent_sets_share_one_tag_write_budget|test_tag_write_limit_scales_down_for_small_pools|test_tag_write_semaphore_rebinds_across_event_loops|def _connected_layer\b|def _make_redis_conn\b|def _patch_redis\b'
;;
esac
done
echo '=== current sparse-checkout configuration, if any ==='
git sparse-checkout list 2>/dev/null || true
echo '=== tracked test directory top-level entries ==='
git ls-tree -r --name-only "$commit" | rg '^tests/' | head -n 80 || trueLength of output: 27306
One production concern remains in When a second event loop calls the layer, Lines 305–312 replace The aggregate bound is therefore not preserved across overlapping event loops. The cross-loop test only performs sequential rebinding. It does not test overlapping use. Do not claim that this layer safely supports one shared Redis pool across event loops unless the implementation enforces one cross-loop budget. A safer design is to bind each 🐇✅ You are interacting with an AI system. |
|
@coderabbitai The shared-pool concern is correct and I've resolved it — verified below. Root cause confirmed. The semaphore was created inside Fix: move the semaphore to the layer instance (shared across every # __init__
self._tag_write_concurrency = min(
TAG_WRITE_CONCURRENCY, max(1, self.max_connections // 2)
)
self._tag_write_semaphore: Optional[asyncio.Semaphore] = None
def _tag_write_gate(self) -> "asyncio.Semaphore":
if self._tag_write_semaphore is None:
self._tag_write_semaphore = asyncio.Semaphore(self._tag_write_concurrency)
return self._tag_write_semaphore
# set(): was `semaphore = asyncio.Semaphore(TAG_WRITE_CONCURRENCY)`
semaphore = self._tag_write_gate()Regression test ( Local: Note on delivery: this session's push scope is a separate branch, not Generated by Claude Code |
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
cache_file="src/youtube_extension/backend/services/intelligent_cache.py"
test_file="tests/unit/test_intelligent_cache.py"
echo '=== current PR revision ==='
git rev-parse --short HEAD
git status --short
echo '=== availability of reported commit ==='
git show --no-patch --format='%H%n%P%n%s' 572361e 2>&1 || true
echo '=== branches and refs containing reported commit ==='
git branch -a --contains 572361e 2>&1 || true
git tag --contains 572361e 2>&1 || true
echo '=== files changed by reported commit ==='
git diff-tree --no-commit-id --name-status -r 572361e 2>&1 || true
echo '=== cache layer structure ==='
ast-grep outline "$cache_file" --items all --match 'RedisCacheLayer|_tag_write_gate|_get_tag_write_semaphore|TAG_WRITE' || true
echo '=== limiter and set implementation on current branch ==='
rg -n -A115 -B25 \
'TAG_WRITE_CONCURRENCY|TAG_WRITE_POOL_RESERVE|_resolve_tag_write_limit|_tag_write_gate|_get_tag_write_semaphore|class RedisCacheLayer\b|def __init__\s*\(|async def set\s*\(' \
"$cache_file" || true
echo '=== staged implementation from reported commit ==='
git show "572361e:$cache_file" 2>/dev/null | \
rg -n -A115 -B25 \
'TAG_WRITE_CONCURRENCY|TAG_WRITE_POOL_RESERVE|_resolve_tag_write_limit|_tag_write_gate|_get_tag_write_semaphore|class RedisCacheLayer\b|def __init__\s*\(|async def set\s*\(' \
|| true
echo '=== relevant tests on current branch and reported commit ==='
for rev in HEAD 572361e; do
echo "--- $rev ---"
git show "$rev:$test_file" 2>/dev/null | \
rg -n -A100 -B25 \
'test_concurrent_set_calls_share_tag_write_bound|test_concurrent_sets_share_one_tag_write_budget|test_tag_write_semaphore_rebinds_across_event_loops|test_tag_write_limit_scales_down_for_small_pools' \
|| true
doneLength of output: 37703
The PR branch is at In the current code,
The proposed Apply the change to PR
Option 1 is the smallest safe change. It also matches the event-loop affinity that 🐇 You are interacting with an AI system. |
The lazy tag-write semaphore rebound whenever the running loop changed. That is correct for sequential reuse (asyncio.run called more than once, or one loop per test), but it also silently rebound while the first loop was still running. Two live loops sharing one RedisCacheLayer therefore got a full _tag_write_limit budget each against the same redis_pool. Measured with two concurrent loops issuing 20 tag writes each against a pool of 20: peak in-flight SADD was 16 (2 x 8) with the unconditional rebind, versus 8 with the guard. Rebinding is now allowed only once the previously bound loop has finished, so sequential reuse keeps working while a second live loop is refused with a clear RuntimeError. Concurrent cross-loop use is unsupported at the pool level too: redis.asyncio caches connections whose transports are bound to the loop that opened them. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
|
@coderabbitai Confirmed and fixed in The finding
Reproduced with two concurrent loops issuing 20 tag writes each against a pool of 20:
Why not strict one-loop-forever bindingYour Option 1 is the right intent, but a permanent binding is not implementable here: line 721 is So the contract distinguishes the two cases, which I believe is what Option 1 actually wants:
if self._tag_write_semaphore is not None and bound is loop:
return self._tag_write_semaphore
if bound is not None and not bound.is_closed() and bound.is_running():
raise RuntimeError(...) # message names the layer and the fixOn the pool itselfAgreed, and this is the deeper point: Scope of the guard — stated plainlyIt is best-effort, not thread-safe. Two threads could still race inside the accessor. It detects a programming error early rather than making cross-loop use safe; the supported model is one layer per event loop, and the error message says so. TestsReplaced the old test, which used
334 passed locally. Each test was proven non-vacuous by restoring the old behaviour behind a monkeypatch and re-measuring. One note: this round referenced commit Ready for another look. |
|
@coderabbitai review |
✅ Action performedReview finished.
|
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
cache_file="src/youtube_extension/backend/services/intelligent_cache.py"
test_file="tests/unit/test_intelligent_cache.py"
echo '=== current PR revision and reported commit ==='
git rev-parse --short HEAD
git show --no-patch --format='%H%n%P%n%s' ec4b910
echo '=== focused commit diff ==='
git diff ec4b910^ ec4b910 -- "$cache_file" "$test_file"
echo '=== implementation outline ==='
ast-grep outline "$cache_file" --items all --match 'RedisCacheLayer|IntelligentCacheSystem|TAG_WRITE'
echo '=== limiter, Redis lifecycle, and set() ==='
rg -n -A145 -B30 \
'TAG_WRITE_CONCURRENCY|TAG_WRITE_POOL_RESERVE|_resolve_tag_write_limit|_get_tag_write_semaphore|_tag_write_gate|class RedisCacheLayer\b|def __init__\s*\(|async def set\s*\(|ConnectionPool\s*\(' \
"$cache_file"
echo '=== warm_cache concurrent set() call path ==='
rg -n -A80 -B25 'async def warm_cache\s*\(|asyncio\.gather\s*\(' "$cache_file"
echo '=== named loop-ownership and concurrency tests ==='
rg -n -A145 -B35 \
'test_tag_write_semaphore_rebinds_across_sequential_event_loops|test_tag_write_semaphore_rejects_concurrent_event_loops|test_set_from_a_second_live_loop_fails_closed|test_concurrent_set_calls_share_tag_write_bound|test_concurrent_sets_share_one_tag_write_budget|def _connected_layer\b|def _make_redis_conn\b|def _patch_redis\b' \
"$test_file"
echo '=== all RedisCacheLayer construction sites ==='
rg -n -C4 '\bRedisCacheLayer\s*\(|ConnectionPool\s*\(|max_connections\s*=' src testsLength of output: 50962 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo '=== tracked intelligent-cache test paths ==='
git ls-tree -r --name-only HEAD | rg -i '(^|/).*intelligent.*cache.*test.*\.py$|(^|/)test_.*intelligent.*cache.*\.py$' || true
echo '=== test references to RedisCacheLayer loop ownership and tag limits ==='
git grep -n -E \
'test_tag_write_semaphore_rebinds_across_sequential_event_loops|test_tag_write_semaphore_rejects_concurrent_event_loops|test_set_from_a_second_live_loop_fails_closed|test_concurrent_set_calls_share_tag_write_bound|_get_tag_write_semaphore|_tag_write_semaphore_loop|_tag_write_limit' \
HEAD -- ':*.py' || true
echo '=== Redis dependency declarations ==='
for file in pyproject.toml poetry.lock requirements.txt setup.cfg setup.py; do
if [ -f "$file" ]; then
echo "--- $file ---"
rg -n -i -C3 '(^|[^A-Za-z])redis([<=>!~ ;\[]|$)|redis-py' "$file" || true
fi
done
fd -a -t f -i 'requirements*.txt' . -x sh -c \
'echo "--- $1 ---"; rg -n -i -C3 "(^|[^A-Za-z])redis([<=>!~ ;\[]|$)|redis-py" "$1" || true' sh {}
echo '=== installed redis asyncio source candidates ==='
python - <<'PY'
import sysconfig
from pathlib import Path
roots = {Path(sysconfig.get_paths()[key]) for key in ("purelib", "platlib")}
for root in sorted(roots):
for relative in ("redis/asyncio/connection.py", "redis/asyncio/client.py"):
candidate = root / relative
if candidate.is_file():
print(candidate)
PY
echo '=== Redis pool loop-affinity implementation, if installed ==='
python - <<'PY'
import sysconfig
from pathlib import Path
roots = {Path(sysconfig.get_paths()[key]) for key in ("purelib", "platlib")}
for root in roots:
path = root / "redis/asyncio/connection.py"
if not path.is_file():
continue
print(f"--- {path} ---")
lines = path.read_text(errors="replace").splitlines()
needles = (
"class ConnectionPool",
"asyncio.Lock",
"asyncio.Queue",
"get_running_loop",
"get_event_loop",
"get_connection",
"release(",
"disconnect(",
)
for index, line in enumerate(lines, 1):
if any(needle in line for needle in needles):
start, end = max(1, index - 8), min(len(lines), index + 30)
print(f"@@ {index} @@")
for number in range(start, end + 1):
print(f"{number}:{lines[number - 1]}")
PYLength of output: 6662 🏁 Script executed: #!/bin/bash
set -euo pipefail
test_file="tests/unit/test_intelligent_cache.py"
cache_file="src/youtube_extension/backend/services/intelligent_cache.py"
echo '=== Redis test fixtures and focused test blocks ==='
sed -n '1180,1495p' "$test_file"
echo '=== RedisCacheLayer public pool consumers ==='
rg -n -A65 -B10 \
'async def connect\s*\(|async def get\s*\(|async def set\s*\(|async def delete\s*\(|async def clear\s*\(|async def invalidate_by_tags\s*\(' \
"$cache_file"
echo '=== direct uses of self.redis_pool in RedisCacheLayer ==='
rg -n -C4 'self\.redis_pool' "$cache_file"
echo '=== test coverage for cross-loop get/delete/untagged set ==='
rg -n -i -A10 -B10 \
'second.*loop|event.*loop|asyncio\.to_thread|new_event_loop|untagged|tags=\[\]|\.get\(|\.delete\(' \
"$test_file"Length of output: 1234 🤖 web_search executed:
Length of output: 20943
A production issue remains.
The new tests use mocked Redis clients. They do not test the retained real pool connections. Therefore, the sequential-rebind test does not prove that sequential cross-loop Redis use works. The loop-ownership policy is also incomplete. Please select one supported contract and enforce it for every Redis operation:
Do not allow sequential rebinding against the existing pool unless its lifecycle guarantees that no connections from the prior loop remain. 🐇 ✏️ Learnings added
If you found this review helpful, would you consider giving us a shout-out on X? Thank you for using CodeRabbit! You are interacting with an AI system. |
|
Tip For best results, initiate chat on the files or code changes.
The wording no longer claims that Thank you for correcting the PR description as well. This resolves the finding. 🐇 ✅ ✏️ Learnings added
You are interacting with an AI system. |
|
@coderabbitai resolve Your
You confirmed in round 5 that "the scope split is correct" and "removing the partial event-loop guard is also correct", and that All five gates are green at Please dismiss or approve so this can merge. |
|
The bounded tag-write concurrency, aggregate per-layer limit, task draining, and documentation scope are now consistent with the reviewed contract. Issue ✅ Action performedComments resolved and changes approved. |
|
Remediation triage for head
Net: the code change itself looks sound and the review feedback is handled — the gate is provenance/enforcement + draft state, plus the spurious secret-scan that #1165 resolves. Generated by Claude Code |
Canonical issue
Closes #1153
Outcome
Redis cache writes no longer pay one sequential network round trip per tag.
RedisCacheLayer.set()previously awaitedsaddonce per tag in a loop, so avalue written with N tags cost N serial round trips on top of the
set/setexand
hsetcalls. The tag writes are now issued concurrently withasyncio.gather, collapsing that phase from N round trips to roughly one.This is the per-write hot path for every cached entry, so the saving applies to
all tagged cache writes.
Scope
RedisCacheLayer.set()(src/youtube_extension/backend/services/intelligent_cache.py).set/hsetround trips). The existing unit tests mock individual Redis command methods and provide nopipelinemock, so that change would require test rework and is better handled separately.Risk
saddcalls now run concurrently on one pooled connection. If one fails,asyncio.gatherpropagates the first exception; the pre-existingtry/exceptaround the method catches it and returnsFalse, exactly as the sequential loop did. A partial set of tags could be written before the failure — this was already true of the sequential loop, which would also stop partway through.Verification
Head
d093a854394594e2281f218d613234a01d43f466.Non-vacuity proof — 5 concurrent
set()calls x 20 tags,max_connections=20.The limiter is swapped for an effectively-unbounded one to reproduce pre-fix behaviour;
everything else is identical. Redis is mocked (
AsyncMock), so this measures thesemaphore's aggregate bound, not real pool behaviour.
saddmain)Regression tests in
tests/unit/test_intelligent_cache.py:test_set_tag_writes_stay_within_concurrency_boundtest_concurrent_sets_share_one_tag_write_budgettest_tag_write_limit_scales_down_for_small_poolstest_tag_write_semaphore_is_replaced_after_its_loop_closesCommands
->
332 passedOut of scope: layer-wide event-loop ownership across all six
self.redis_poolcall sites is pre-existing on
mainand tracked in #1162.Production evidence
Not applicable. This change touches a backend Redis cache path that is not
exercised by the Vercel preview deployment, and it is behaviour-preserving —
the same keys and members are written, only the scheduling of the awaits
changes. Correctness is covered by the 172 focused unit tests above rather
than by a runtime deployment.
Agent handoff