diff --git a/src/youtube_extension/backend/real_api_endpoints.py b/src/youtube_extension/backend/real_api_endpoints.py index be7ae7868..788895e05 100644 --- a/src/youtube_extension/backend/real_api_endpoints.py +++ b/src/youtube_extension/backend/real_api_endpoints.py @@ -15,7 +15,7 @@ from pathlib import Path from typing import Any, Final, Optional -from fastapi import BackgroundTasks, FastAPI, HTTPException +from fastapi import BackgroundTasks, FastAPI, HTTPException, Response from pydantic import BaseModel, Field from .services.api_cost_monitor import cost_monitor @@ -28,11 +28,20 @@ # Configure logging logger = logging.getLogger(__name__) -# Distinguishes "no cache entry" from an entry that parses to ``None`` -- which is -# what a file holding the JSON literal ``null`` yields. A plain ``None`` return -# conflates the two and turns a stored ``null`` analysis into a 404. +# Distinguishes "no cache entry" from any entry content. The read helper returns +# raw bytes today, but its historical contract returned parsed JSON, where a file +# holding the literal ``null`` yields ``None`` -- a plain ``None`` return conflates +# that with absence and turns a stored ``null`` analysis into a 404. The sentinel +# keeps "miss" impossible to confuse with any payload, parsed or raw. _CACHE_MISS: Final = object() +# Cache entries at or below this size are JSON-validated (parsed and discarded) +# before being served, so a damaged small entry still surfaces as a 500. The +# validation parse holds the GIL at roughly 3 ms/MB, so this threshold *is* the +# event-loop stall bound: ~6 ms worst case, independent of video duration. Real +# entries sit well under it (~0.15 MB for an hour of video, ~1.6 MB long-form). +_VALIDATION_MAX_BYTES: Final = 2 * 1024 * 1024 + def _collect_processed_videos_sync(cache_dir: Path) -> list[dict[str, Any]]: """Scan the processing cache directory and parse every cached result. @@ -91,34 +100,45 @@ def _collect_processed_videos_sync(cache_dir: Path) -> list[dict[str, Any]]: def _read_video_analysis_sync(cache_path: Path) -> Any: - """Read and parse a single cached video analysis. - - This performs blocking filesystem work (``open()`` and a full - ``json.load()`` of the analysis payload) and is therefore intended to be - executed in a worker thread via :func:`asyncio.to_thread` rather than - directly on the event loop. - - What that offload does and does not buy is worth stating precisely, because - the two halves of this function behave differently under the GIL: - - * ``open()``/``read()`` release the GIL, so moving them off the loop removes - the caller's exposure to filesystem latency entirely. This is the part - that is unbounded -- a cold page cache, a networked mount or a contended - disk can stall for hundreds of milliseconds. - * ``json.load()`` is CPU-bound C code that *holds* the GIL for its whole - duration. Running it in a worker thread does not stop it blocking the - loop; it only relocates it. Measured stall tracks payload size at roughly - 3 ms/MB (~0.4 ms for a typical one-hour transcript, ~3 ms for long-form). - - So this converts an unbounded, environment-dependent stall into a bounded, - payload-proportional one. It does not make the read non-blocking. On a warm - page cache with a small payload the executor hop is measurably *worse* than - reading inline (~0.5 ms of dispatch overhead against a ~0.4 ms parse); the - change earns its keep when the filesystem is slow, which is precisely the - case that cannot be predicted from inside the handler. - - Removing the residual parse stall needs a different fix (streaming/incremental - parse, a size cap, or a process pool) and is tracked in issue #1306. + """Read a single cached video analysis and return its raw JSON bytes. + + This performs blocking filesystem work (``open()`` and a full ``read()`` + of the analysis payload) and is therefore intended to be executed in a + worker thread via :func:`asyncio.to_thread` rather than directly on the + event loop. + + Guarantee: the event-loop stall this read can cause is *bounded and + independent of payload size*. That holds because of two decisions: + + * ``open()``/``read()`` release the GIL, so the loop is shielded from + filesystem latency entirely -- a cold page cache, a networked mount or a + contended disk can stall this thread for hundreds of milliseconds + without the loop noticing, at any payload size. + * The payload is *not* parsed into Python objects. ``json.load()`` is + CPU-bound C code that holds the GIL for its whole duration (~3 ms/MB), + so relocating it to a worker thread only moved the stall, and entry size + is unbounded -- it tracks video duration because the full transcript is + persisted into the cache entry. The only consumer of this helper streams + the bytes back to the client verbatim, which also removes FastAPI's + re-serialisation of the parsed object -- work that ran *on* the loop and + likewise scaled with payload size. The parse/re-encode round-trip was + pure overhead: the cache entry already is the response body. + + Entries at or below :data:`_VALIDATION_MAX_BYTES` are still parse-validated + here (the parsed result is discarded) so a damaged entry keeps surfacing as + a 500 rather than being handed to clients as garbage. That validation is + the sole remaining GIL-held, size-proportional work, and the threshold caps + it at ~6 ms. Larger entries skip validation: the writer + (``RealVideoProcessor._write_cache_file``) publishes atomically via a temp + file and ``os.replace``, so a torn half-written entry cannot be observed; + only out-of-band corruption of an oversized entry would reach a client + unflagged, and that is accepted in exchange for the bounded stall. + + The executor dispatch hop (~0.5 ms) still costs more than a warm-cache + read of a small entry. That fixed overhead is accepted deliberately: it is + the insurance premium against unbounded filesystem latency, which cannot + be predicted from inside the handler -- and unlike before, no deferred + parse or on-loop re-serialisation is added on top of it. Opening directly and treating :class:`FileNotFoundError` as the miss replaces a separate ``Path.exists()`` probe. That is one syscall instead of @@ -138,17 +158,23 @@ def _read_video_analysis_sync(cache_path: Path) -> Any: treats anything older than ``_CACHE_TTL_SECONDS`` (24 hours) as a miss; reusing it here would turn every analysis over a day old into a 404. - Returns the parsed JSON payload, which may legitimately be ``None`` when the - entry holds the literal ``null``. Absence is reported as the distinct - :data:`_CACHE_MISS` sentinel so the two cannot be confused. + Returns the entry's UTF-8 JSON bytes -- ``b"null"`` for an entry holding + the literal ``null``, which must still be served as a 200. Absence is + reported as the distinct :data:`_CACHE_MISS` sentinel so the two cannot be + confused. """ try: - handle = open(cache_path, encoding="utf-8") + handle = open(cache_path, "rb") except (FileNotFoundError, ValueError): return _CACHE_MISS with handle as f: - return json.load(f) + raw = f.read() + + if len(raw) <= _VALIDATION_MAX_BYTES: + json.loads(raw) + + return raw # Pydantic models for API requests/responses @@ -334,16 +360,16 @@ async def get_video_analysis(video_id: str): processor = get_real_video_processor() cache_path = processor._get_cache_path(video_id) - # Reading an entry opens and JSON-parses a file whose size is set - # by the stored analysis payload, so the parse cost scales with the - # video rather than being bounded. Run it in a worker thread so a - # large analysis cannot stall the event loop for every other - # in-flight request. Building the path stays here: it is pure - # string arithmetic and touches no filesystem. + # The entry is read in a worker thread; ``open()``/``read()`` + # release the GIL, so the loop is shielded from filesystem latency + # at any payload size. The bytes come back *unparsed* -- see + # _read_video_analysis_sync for why that bounds the loop stall + # independently of video duration. Building the path stays here: + # it is pure string arithmetic and touches no filesystem. video_data = await asyncio.to_thread(_read_video_analysis_sync, cache_path) - # Identity check against the sentinel, not a truthiness or ``is None`` - # test: an entry holding the JSON literal ``null`` parses to ``None`` + # Identity check against the sentinel, not a truthiness test: an + # entry holding the JSON literal ``null`` comes back as ``b"null"`` # and has always been served as a 200, so it must not 404 here. if video_data is _CACHE_MISS: raise HTTPException( @@ -351,7 +377,11 @@ async def get_video_analysis(video_id: str): detail=f"Video analysis not found: {video_id}" ) - return video_data + # Raw passthrough: the cache entry already is the response JSON. + # Returning a Response skips FastAPI's jsonable_encoder/json.dumps + # round-trip, which would otherwise re-serialise the payload on + # the event loop in proportion to its size. + return Response(content=video_data, media_type="application/json") except HTTPException: raise diff --git a/tests/unit/test_real_api_endpoints.py b/tests/unit/test_real_api_endpoints.py index efc2918b8..42d57a460 100644 --- a/tests/unit/test_real_api_endpoints.py +++ b/tests/unit/test_real_api_endpoints.py @@ -46,6 +46,7 @@ # --------------------------------------------------------------------------- from youtube_extension.backend.real_api_endpoints import ( # noqa: E402 _CACHE_MISS, + _VALIDATION_MAX_BYTES, BatchProcessingRequest, VideoAnalysisResponse, VideoProcessingRequest, @@ -1150,11 +1151,11 @@ async def test_event_loop_stays_responsive_during_cache_read( Scope note: ``_SlowReadPath`` sleeps in ``__fspath__``, which models *filesystem* latency -- and ``time.sleep`` releases the GIL, exactly as - a real blocking syscall does. So this test covers the I/O half of the - read only. It is not vacuous: reverting the ``asyncio.to_thread`` hop - drives ``heartbeats`` to 0. But it deliberately says nothing about the - ``json.load()`` half, which holds the GIL and still stalls the loop -- - see ``test_parse_still_stalls_the_loop_in_proportion_to_payload``. + a real blocking syscall does. It is not vacuous: reverting the + ``asyncio.to_thread`` hop drives ``heartbeats`` to 0. Since the helper + stopped parsing the payload, filesystem latency is essentially the + whole read; the size-proportional stall that used to remain is pinned + by ``test_read_stall_no_longer_scales_with_payload``. """ tmp_cache.mkdir(parents=True, exist_ok=True) cache_file = _write_cache_file(tmp_cache, "auJzb1D-fag") @@ -1206,30 +1207,32 @@ async def _heartbeat(): heartbeats >= 5 ), f"event loop was starved during the cache read (ticks={heartbeats})" - async def test_parse_still_stalls_the_loop_in_proportion_to_payload( - self, tmp_path - ): - """Document the residual: ``json.load`` blocks the loop even off-thread. - - ``open()``/``read()`` release the GIL, so the ``asyncio.to_thread`` hop - genuinely removes filesystem latency from the loop. ``json.load()`` does - not -- it is CPU-bound C code that holds the GIL for its full duration, - so relocating it to a worker thread does not stop it blocking. - - This is a *characterisation* test. It exists so the weaker guarantee - stays honest: if someone later claims this endpoint's read is fully - non-blocking, this test is the counter-example. Asserted as a ratio - between a small and a large payload so it does not depend on the - absolute speed of the runner. + async def test_read_stall_no_longer_scales_with_payload(self, tmp_path): + """The read must not stall the loop in proportion to payload size. + + Successor to ``test_parse_still_stalls_the_loop_in_proportion_to_ + payload``, which characterised the residual left by the ``to_thread`` + offload: ``json.load()`` held the GIL in the worker, so the loop still + stalled ~3 ms/MB of payload. The helper no longer parses entries above + the validation threshold -- it returns raw bytes, and ``read()`` + releases the GIL -- so that proportional stall must be gone. + + Self-calibrating rather than absolute-time based: the baseline is the + same payload pushed through ``json.loads`` on a worker thread, which + is exactly the old behaviour. The helper's read must stall the loop + for a small fraction of what the parse does, whatever the runner's + speed. If this fails, size-proportional GIL-held work has crept back + into the read path and the documented bounded-stall guarantee in + ``_read_video_analysis_sync`` no longer holds. """ - small = tmp_path / "small.json" + payload = json.dumps({"transcript": [{"t": i} for i in range(400_000)]}) large = tmp_path / "large.json" - small.write_text(json.dumps({"transcript": [{"t": i} for i in range(200)]})) - large.write_text( - json.dumps({"transcript": [{"t": i} for i in range(400_000)]}) - ) + large.write_text(payload) + # The point of the threshold is that entries above it skip the + # validation parse; the fixture must actually exercise that path. + assert large.stat().st_size > _VALIDATION_MAX_BYTES - async def _max_loop_gap(path): + async def _max_loop_gap(work, arg): gaps: list[float] = [] stop = asyncio.Event() @@ -1244,28 +1247,35 @@ async def _ticker(): task = asyncio.create_task(_ticker()) await asyncio.sleep(0.02) gaps.clear() - await asyncio.to_thread(_read_video_analysis_sync, path) + await asyncio.to_thread(work, arg) stop.set() await task return max(gaps) - small_gap = await _max_loop_gap(small) - large_gap = await _max_loop_gap(large) - - # The large payload is ~2000x the small one. Even allowing for executor - # dispatch overhead dominating the small case, the parse must show up as - # a materially longer stall -- that is the point being documented. - assert large_gap > small_gap * 5, ( - "expected json.load to stall the loop in proportion to payload size " - f"(small={small_gap * 1000:.2f}ms, large={large_gap * 1000:.2f}ms); " - "if this now passes trivially, the parse may have been made " - "incremental -- update the endpoint's documented guarantee" + parse_gap = await _max_loop_gap(json.loads, payload) + # Take the best of a few runs for the read: a single scheduling hiccup + # on a loaded CI runner must not masquerade as a proportional stall. + read_gap = min( + [await _max_loop_gap(_read_video_analysis_sync, large) for _ in range(3)] + ) + + assert read_gap < parse_gap / 4, ( + "expected the raw-bytes read to stall the loop far less than " + "parsing the same payload " + f"(read={read_gap * 1000:.2f}ms, parse={parse_gap * 1000:.2f}ms); " + "size-proportional GIL-held work appears to be back on the read " + "path -- re-narrow the guarantee documented in " + "_read_video_analysis_sync if that is intentional" ) def test_offloaded_read_returns_same_payload( self, client, mock_processor, tmp_cache ): - """Offloading must not change the response contract.""" + """Offloading must not change the response contract. + + Stronger than semantic equality: the raw-passthrough response body is + the cache entry byte-for-byte, and it still declares itself as JSON. + """ tmp_cache.mkdir(parents=True, exist_ok=True) cache_file = _write_cache_file(tmp_cache, "auJzb1D-fag") mock_processor._get_cache_path.return_value = cache_file @@ -1273,7 +1283,9 @@ def test_offloaded_read_returns_same_payload( response = client.get("/api/v2/videos/auJzb1D-fag") assert response.status_code == 200 - assert response.json() == _read_video_analysis_sync(cache_file) + assert response.headers["content-type"].startswith("application/json") + assert response.content == cache_file.read_bytes() + assert response.json() == json.loads(cache_file.read_text(encoding="utf-8")) class TestReadVideoAnalysisSync: @@ -1284,21 +1296,22 @@ def test_missing_file_returns_the_miss_sentinel(self, tmp_path): _read_video_analysis_sync(tmp_path / "absent_processed.json") is _CACHE_MISS ) - def test_existing_file_returns_parsed_payload(self, tmp_cache): + def test_existing_file_returns_raw_bytes_verbatim(self, tmp_cache): tmp_cache.mkdir(parents=True, exist_ok=True) cache_file = _write_cache_file(tmp_cache, "auJzb1D-fag") result = _read_video_analysis_sync(cache_file) - assert result is not None - assert result["video_id"] == "auJzb1D-fag" - assert result == json.loads(cache_file.read_text(encoding="utf-8")) + assert isinstance(result, bytes) + assert result == cache_file.read_bytes() + assert json.loads(result)["video_id"] == "auJzb1D-fag" def test_null_content_is_a_payload_not_a_miss(self, tmp_cache): - """A stored JSON ``null`` parses to ``None`` but is *not* a cache miss. + """A stored JSON ``null`` is a payload, *not* a cache miss. - Returning a bare ``None`` for absence would conflate the two and turn - such an entry into a 404. The sentinel keeps them distinguishable. + It comes back as the raw bytes ``b"null"`` and must be served as a + 200. The sentinel keeps absence distinguishable from any content -- + including the parsed-``None`` form the helper used to return. """ tmp_cache.mkdir(parents=True, exist_ok=True) cache_file = tmp_cache / "nullish_processed.json" @@ -1306,11 +1319,11 @@ def test_null_content_is_a_payload_not_a_miss(self, tmp_cache): result = _read_video_analysis_sync(cache_file) - assert result is None + assert result == b"null" assert result is not _CACHE_MISS def test_falsy_payloads_are_not_misses(self, tmp_cache): - """Neither is any other falsy JSON value the helper can legally parse.""" + """Neither is any other JSON value that parses to something falsy.""" tmp_cache.mkdir(parents=True, exist_ok=True) for index, raw in enumerate(("{}", "[]", '""', "0", "false")): @@ -1338,6 +1351,12 @@ def test_corrupt_entry_raises_rather_than_reporting_a_miss(self, tmp_cache): Only "the path names no readable entry" -- ``FileNotFoundError`` or a path the OS rejects outright -- means "no such analysis"; anything else is a real fault and has to keep propagating. + + This guarantee is now scoped to entries at or below + ``_VALIDATION_MAX_BYTES``; validating above the threshold would + reintroduce the unbounded GIL-held parse the raw-bytes read removed. + See ``test_oversized_corrupt_entry_is_returned_unvalidated`` for the + other side of that line. """ tmp_cache.mkdir(parents=True, exist_ok=True) cache_file = tmp_cache / "bad_processed.json" @@ -1346,6 +1365,34 @@ def test_corrupt_entry_raises_rather_than_reporting_a_miss(self, tmp_cache): with pytest.raises(json.JSONDecodeError): _read_video_analysis_sync(cache_file) + def test_corrupt_entry_at_the_threshold_still_raises(self, tmp_cache): + """The validation boundary is inclusive: exactly-at-cap entries parse.""" + tmp_cache.mkdir(parents=True, exist_ok=True) + cache_file = tmp_cache / "boundary_processed.json" + blob = b"{invalid json" + b" " * (_VALIDATION_MAX_BYTES - len(b"{invalid json")) + assert len(blob) == _VALIDATION_MAX_BYTES + cache_file.write_bytes(blob) + + with pytest.raises(json.JSONDecodeError): + _read_video_analysis_sync(cache_file) + + def test_oversized_corrupt_entry_is_returned_unvalidated(self, tmp_cache): + """Entries above the validation threshold are served verbatim. + + Validating them would mean a GIL-held parse proportional to payload + size -- exactly the unbounded loop stall this helper exists to avoid. + Integrity above the threshold is delegated to the writer, which + publishes entries atomically (temp file + ``os.replace``), so a torn + entry cannot be observed; only out-of-band corruption slips through, + and it reaches the client as-is rather than as a 500. + """ + tmp_cache.mkdir(parents=True, exist_ok=True) + cache_file = tmp_cache / "huge_processed.json" + blob = b"{not json at all" + b"x" * _VALIDATION_MAX_BYTES + cache_file.write_bytes(blob) + + assert _read_video_analysis_sync(cache_file) == blob + def test_directory_path_raises_rather_than_reporting_a_miss(self, tmp_cache): """Opening a directory is an ``OSError`` but not a missing entry.""" tmp_cache.mkdir(parents=True, exist_ok=True) @@ -1380,8 +1427,8 @@ def test_helper_returns_entry_older_than_processor_ttl(self, tmp_cache): result = _read_video_analysis_sync(cache_file) - assert result is not None - assert result["video_id"] == "auJzb1D-fag" + assert result is not _CACHE_MISS + assert json.loads(result)["video_id"] == "auJzb1D-fag" def test_endpoint_serves_entry_older_than_processor_ttl( self, client, mock_processor, tmp_cache