feat(library): P2 rebased — YouTube + Web processors with streaming, content-type gate, timeout guards - #2177
Conversation
…content-type gate, timeout guards Rebase feat/library-p2 onto origin/dev (c5b1a6f) keeping ONLY P2 delta: - YouTubeProcessor: metadata, thumbnail, transcript, chapters via yt-dlp - WebProcessor: HTML fetch, readability extraction, SSRF-guarded redirects - _extract_readable_text() helper (readability-lxml with tag-stripping fallback) - url:youtube + url:web entries in _PROCESSORS - run_pipeline reference-artifact removal (URL items now get processed) - _store_init_lock: asyncio.Lock on lazy LibraryStore init (TOCTOU fix) - 'chapters' added to _TEXT_ARTIFACT_KINDS Fixes from jaylfc review of jaylfc#2068: - Streaming read via client.stream() — OOM-safe, cap enforced before buffering - Content-type gate: non-text/* responses raise ValueError (→ error status) - 60s wall-clock deadline on WebProcessor fetch - _cleanup_procs() in youtube.py: kill tracked subprocesses on timeout - 120s asyncio.wait_for on YouTubeProcessor fetch with cleanup on TimeoutError Preserves all P1 hardening from dev: - try_update_item_status CAS (library_store.py untouched) - reprocess 409 guard + artifact cleanup + rollback-to-ready - taosmd HTTP collections flow (memory_url, admin-token, idempotent collection_id) - logger.warning OSError handlers (not bare pass) - path='' on metadata artifacts (unlink guard) - test_library_page_gone (404), test_unauth_library_endpoints, test_reprocess_idempotent, test_reprocess_while_processing_returns_409 Tests: 54/54 pass (42 existing + 12 new: 3 YouTube, 4 web happy-path, 2 SSRF block, 1 redirect-hop validation, 1 size cap, 1 content-type gate)
|
Warning Review limit reached
Next review available in: 33 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (5)
✨ Finishing Touches🧪 Generate unit tests (beta)
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 |
|
Reviewed against every blocking finding from #2068. This is the right rebase and I will merge it once CI is green. Thank you for doing it properly rather than patching the old branch.
The silent revert is gone, and I checked it the same way I caught it, by diffing symbols against Both survive. And the shape of the diff tells the story on its own: 769 additions, 21 deletions, against a branch that previously deleted the CAS, the reprocess guards, the collections flow and six tests. This is now additive. The streaming fix is correct, not merely different. async with client.stream("GET", current_url) as resp:
content_type = resp.headers.get("content-type", "")
if ct_base and not ct_base.startswith("text/"):
raise ...
async for chunk in resp.aiter_bytes(8192):
total += len(chunk)
if total > self._MAX_WEB_BYTES:
raise ...Three things right at once: the client stays open while reading, the content-type is gated before the body is consumed, and the cap is checked inside the chunk loop against a running total. That closes the OOM vector I measured at 130MB peak heap for a claimed 10MB cap. The tests are real, which is the part I care about most. All five I asked for are present ( Two notes, neither blocking: The
Merging when the remaining shards finish. Good work on the SSRF chain surviving the rebase intact, since that was always the strongest part of the original PR and the reason it was worth rebasing rather than abandoning. |
PR Summary by QodoLibrary: add YouTube + Web URL processors with streaming and timeout guards
AI Description
Diagram
High-Level Assessment
Files changed (5)
|
Code Review by Qodo
1. Global proc tracker race
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
| class YouTubeProcessor(Processor): | ||
| """YouTube URL processor — cheap tier: metadata, thumbnail, transcript, chapters. | ||
|
|
||
| Uses yt-dlp via the knowledge_fetchers.youtube module to fetch video | ||
| metadata and captions without downloading the video file. Produces | ||
| artifacts that flow into taosmd collections for agent querying. | ||
|
|
||
| The cheap tier (per the design doc, docs/design/library-app.md section 4) | ||
| covers steps 1-4: canonical link, title, channel, description, thumbnail, | ||
| duration, upload date, subtitles/transcript, chapters. | ||
| """ | ||
|
|
||
| _YTDLP_TIMEOUT = 120 # seconds | ||
|
|
||
| async def process(self, item: dict) -> list[dict]: | ||
| item_id = item["id"] | ||
| source_url = item.get("source_url", "") | ||
| artifacts: list[dict] = [] | ||
|
|
||
| if not source_url: | ||
| return artifacts | ||
|
|
||
| # Only catch ImportError (missing yt-dlp). Let fetch errors | ||
| # propagate so run_pipeline marks the item as "error" — a failed | ||
| # yt-dlp invocation must not silently look successful. |
There was a problem hiding this comment.
1. Em dashes in library_pipeline.py 📜 Skill insight ✧ Quality
New docstrings/comments and a raised error message include em dashes (—), which the checklist forbids in public-facing text. This can leak into logs/UI/error surfaces and violates the project style requirement.
Agent Prompt
## Issue description
The PR introduces em dashes (`—`) in docstrings/comments and in a user-visible exception message.
## Issue Context
Compliance rule PR Compliance ID 2212258 forbids em dashes in public-facing text; use `-` (hyphen) or rewrite.
## Fix Focus Areas
- tinyagentos/library_pipeline.py[323-347]
- tinyagentos/library_pipeline.py[483-507]
- tinyagentos/library_pipeline.py[614-616]
- tests/test_library.py[642-643]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
| # Tracked subprocesses so they can be killed on timeout or cancel. | ||
| _tracked_procs: list[asyncio.subprocess.Process] = [] | ||
|
|
||
|
|
||
| def _cleanup_procs() -> None: | ||
| """Kill all tracked subprocesses (called on timeout or cancel).""" | ||
| for p in _tracked_procs: | ||
| try: | ||
| p.kill() | ||
| except Exception: | ||
| pass | ||
| _tracked_procs.clear() |
There was a problem hiding this comment.
2. Global proc tracker race 🐞 Bug ☼ Reliability
youtube.py tracks yt-dlp subprocesses in a module-global list and _cleanup_procs() kills *all* tracked processes, so one timed-out YouTube fetch can terminate other concurrent fetches and the list grows unbounded across successful runs. _cleanup_procs() also kills without awaiting process exit, risking zombie/orphaned processes on timeout/cancel paths.
Agent Prompt
### Issue description
`tinyagentos.knowledge_fetchers.youtube` stores every spawned `yt-dlp` subprocess in a module-global `_tracked_procs` list and `_cleanup_procs()` kills everything in that global list. This can (a) kill unrelated concurrent fetches and (b) leak references to completed processes indefinitely. Additionally, `kill()` is not followed by `await proc.wait()`, which can leave zombies/orphans.
### Issue Context
- `YouTubeProcessor` wraps `fetch()` in `asyncio.wait_for()` and calls `_cleanup_procs()` on timeout.
- `fetch()`/`download_video()` append subprocesses to `_tracked_procs` but never remove them on normal completion.
### Fix Focus Areas
- tinyagentos/knowledge_fetchers/youtube.py[17-29]
- tinyagentos/knowledge_fetchers/youtube.py[140-146]
- tinyagentos/knowledge_fetchers/youtube.py[177-211]
- tinyagentos/knowledge_fetchers/youtube.py[266-280]
- tinyagentos/library_pipeline.py[348-363]
### What to change
- Replace the module-global `_tracked_procs` with **per-invocation** tracking (e.g., local `procs: list[Process]` inside `fetch()` / `download_video()`), and ensure each proc is removed/cleared in a `finally` block.
- Make cleanup only kill processes belonging to that invocation. (If the processor needs cleanup access, return a cleanup callback/object from `fetch()` or manage the timeout inside `fetch()`.)
- In cleanup, after `proc.kill()` also `await proc.wait()` (or `communicate()` if appropriate) to avoid zombies.
- Handle `asyncio.CancelledError` as well as timeout, ensuring subprocesses are terminated on cancellation.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
| for _hop in range(self._MAX_WEB_REDIRECTS + 1): | ||
| validate_url_or_raise(current_url) | ||
|
|
||
| async with httpx.AsyncClient( | ||
| timeout=httpx.Timeout(30), | ||
| follow_redirects=False, | ||
| ) as client: | ||
| async with client.stream("GET", current_url) as resp: | ||
| status_code = resp.status_code |
There was a problem hiding this comment.
3. Ssrf check not pinned 🐞 Bug ⛨ Security
WebProcessor validates each redirect hop with validate_url_or_raise() but then performs the actual request via httpx using the original hostname URL, allowing a DNS-rebinding TOCTOU window where the connection can resolve to a different (potentially private/loopback) IP than was validated. This undermines the intended SSRF defense for the new library URL-fetching pipeline.
Agent Prompt
### Issue description
`WebProcessor` calls `validate_url_or_raise(current_url)` (which resolves DNS and validates the resulting IPs) but then passes `current_url` to `httpx.AsyncClient.stream()`. Because `validate_url_or_raise()` returns no resolved address information and `httpx` will resolve DNS again at connection time, there is a DNS TOCTOU window (DNS rebinding) where the validated hostname can later resolve to a blocked/internal address.
### Issue Context
- This PR introduces a new server-side fetching path (library WebProcessor) that relies on SSRF checks.
- The SSRF helper validates by DNS resolution (`socket.getaddrinfo`) but does not bind the subsequent connection to the validated resolution.
### Fix Focus Areas
- tinyagentos/library_pipeline.py[486-496]
- tinyagentos/routes/desktop_browser/ssrf.py[58-119]
### What to change
Implement a fetch path where **the IP address that is validated is the IP address that is connected to**, e.g.:
- Introduce a custom httpx/httpcore transport or resolver that performs resolution + `validate_resolved_addr()` inside the connection routine and rejects any selected address that is blocked.
- Alternatively, resolve+validate in `WebProcessor`, then connect to a validated address explicitly while preserving HTTP `Host` and (for HTTPS) SNI/cert verification in a safe way.
- At minimum, add a second validation step as close to connection establishment as possible (though true pinning is preferred).
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
2 participants
Rebase of #2068 onto origin/dev (c5b1a6f)
This is the re-land jaylfc requested. Keeps ONLY the P2 delta on top of current dev.
What was changed from the old PR
P2 delta (kept):
_extract_readable_text()helperurl:youtube+url:webin_PROCESSORS_store_init_lock(asyncio.Lock on lazy LibraryStore init)"chapters"in_TEXT_ARTIFACT_KINDSP1 re-application (dropped):
Fixes from jaylfc review:
client.stream()instead ofclient.get()text/*→ValueError(→ error status)_cleanup_procs()kill tracked subprocesses on timeoutasyncio.wait_foron YouTubeProcessor with cleanupPreserved from dev (not reverted):
try_update_item_statusCASlogger.warningOSError handlerspath=""on metadata artifacts (unlink guard)test_library_page_gone,test_unauth_library_endpoints,test_reprocess_idempotent,test_reprocess_while_processing_returns_409Tests: 54/54 pass (42 existing dev tests + 12 new): 3 YouTube, 4 web happy-path, 2 SSRF block, 1 redirect-hop validation, 1 size cap, 1 content-type gate.
Closes #2068.