From 2138d4afc3076d18037798316359d88d370e9a66 Mon Sep 17 00:00:00 2001 From: NixReaper Date: Sat, 5 Sep 2026 07:56:00 -0500 Subject: [PATCH] Invalidate movie/series/person detail cache on re-ingestion cached_response() had no write-side invalidation, so re-ingesting an entity kept serving its pre-ingestion cached response for up to CACHE_DEFAULT_TTL. Found live: re-ingesting a movie to test the new company-logo field still showed logo_path: None because the sliding 5-minute cache from earlier testing never expired. Co-Authored-By: Claude Sonnet 5 --- app/services/ingestion/runner.py | 20 ++- docs/architecture.md | 21 ++-- .../test_ingestion_cache_invalidation.py | 114 ++++++++++++++++++ 3 files changed, 146 insertions(+), 9 deletions(-) create mode 100644 tests/integration/test_ingestion_cache_invalidation.py diff --git a/app/services/ingestion/runner.py b/app/services/ingestion/runner.py index d57f468..761037b 100644 --- a/app/services/ingestion/runner.py +++ b/app/services/ingestion/runner.py @@ -14,7 +14,9 @@ from arq import Retry from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import session_scope +from app.core.ids import aurora_id from app.core.logging import get_logger +from app.core.redis import get_redis_pool from app.models.enums import IngestionJobStatus from app.models.provider import IngestionJob from app.providers.base import MetadataProvider @@ -30,6 +32,7 @@ from app.repositories.provider_repository import ( mark_job_succeeded, ) from app.repositories.tv_repository import get_seasons_for_series, get_series_companies +from app.services.cache import build_key, cache_delete from app.services.ingestion.movie_ingestion import ingest_movie from app.services.ingestion.person_ingestion import ingest_person from app.services.ingestion.series_ingestion import ingest_season, ingest_series @@ -76,10 +79,17 @@ async def _run_ingestion( job_id: uuid.UUID, label: str, work: Callable[[AsyncSession, MetadataProvider], Awaitable[uuid.UUID]], + cache_entity_type: str | None = None, ) -> uuid.UUID: """Shared job-lifecycle wrapper: mark running, run `work` in its own session/transaction, mark succeeded/failed. `work` does the actual - fetch+normalize+upsert and returns the resulting Aurora entity id.""" + fetch+normalize+upsert and returns the resulting Aurora entity id. + + `cache_entity_type`, when given, invalidates that entity's cached + `GET .../{aurora_id}` response on success -- `cached_response` + (`app/api/deps.py`) has no invalidate-on-write wiring otherwise, so a + re-ingested movie/series/person would keep serving its pre-ingestion + snapshot for up to `CACHE_DEFAULT_TTL`.""" async with session_scope() as session: job = await _require_job(session, job_id) @@ -92,6 +102,11 @@ async def _run_ingestion( async with session_scope() as session: job = await _require_job(session, job_id) await mark_job_succeeded(session, job, result_id) + if cache_entity_type is not None: + await cache_delete( + get_redis_pool(), + build_key(cache_entity_type, aurora_id(cache_entity_type, str(result_id))), + ) return result_id except Exception as exc: logger.error(f"{label}_failed", job_id=str(job_id), error=str(exc)) @@ -123,6 +138,7 @@ async def run_movie_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID: lambda session, provider: ingest_movie( session, provider, external_id, ingestion_job_id=job_id ), + cache_entity_type="movie", ) @@ -133,6 +149,7 @@ async def run_person_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID lambda session, provider: ingest_person( session, provider, external_id, ingestion_job_id=job_id ), + cache_entity_type="person", ) @@ -143,6 +160,7 @@ async def run_series_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID lambda session, provider: ingest_series( session, provider, external_id, ingestion_job_id=job_id ), + cache_entity_type="series", ) diff --git a/docs/architecture.md b/docs/architecture.md index 7cc978a..4682731 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -139,15 +139,20 @@ titles. Swapping in `OpenSearchProvider` later means implementing the same write function catches Redis errors and behaves as a cache miss / no-op rather than raising, so Redis being down degrades performance, not correctness. `app/api/deps.py:cached_response()` is the read-through helper -used by the movie/person detail endpoints (the read-heaviest, per section -21). Cache keys are versioned (`aurora:v1:...`) so a response-shape change -(bumping `CACHE_VERSION`) doesn't require a manual flush. +used by the movie/series/person detail endpoints (the read-heaviest, per +section 21). Cache keys are versioned (`aurora:v1:...`) so a response-shape +change (bumping `CACHE_VERSION`) doesn't require a manual flush. -Cache invalidation on write: 0.1 has no write API yet (ingestion writes -directly to Postgres via workers, not through the cached read endpoints), so -there is nothing to invalidate today. The moment a write endpoint exists, -it must call `app/services/cache.py:cache_delete` for the affected keys — -tracked as a Known limitation. +Cache invalidation on write: ingestion is the only write path that touches +cached reads (there's no separate write API in 0.1), and it invalidates — +`app/services/ingestion/runner.py:_run_ingestion` takes a `cache_entity_type` +("movie"/"series"/"person") and, on success, deletes that entity's detail +cache key so the next `GET` reflects the just-completed ingestion instead of +serving a pre-ingestion snapshot for up to `CACHE_DEFAULT_TTL`. This was +found live: re-ingesting a movie to pick up a new company-logo field kept +returning the old response because nothing invalidated the cache. Seasons +and episodes aren't cached, so `run_season_ingestion` passes no +`cache_entity_type`. ## Security (section 22) diff --git a/tests/integration/test_ingestion_cache_invalidation.py b/tests/integration/test_ingestion_cache_invalidation.py new file mode 100644 index 0000000..dddd502 --- /dev/null +++ b/tests/integration/test_ingestion_cache_invalidation.py @@ -0,0 +1,114 @@ +"""`GET /api/v1/movies|series|people/{aurora_id}` responses are cached +(`app/api/deps.py:cached_response`) with no write path invalidating them. +Before this, re-ingesting an entity left its cached detail response stale +for up to `CACHE_DEFAULT_TTL` -- discovered live when a company-logo fix +didn't show up in a fresh `GET` right after re-ingestion. +`_run_ingestion`'s `cache_entity_type` param deletes that entity's cache key +on success so the next read reflects the just-completed ingestion. +""" + +from __future__ import annotations + +import uuid + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.ids import aurora_id +from app.repositories.provider_repository import create_ingestion_job, get_or_create_provider +from app.services.cache import build_key +from app.services.ingestion import runner + + +async def _make_job(session: AsyncSession, *, attempts_already_used: int = 0) -> uuid.UUID: + provider = await get_or_create_provider(session, "tmdb", "TMDB") + job = await create_ingestion_job( + session, + provider_id=provider.id, + job_type="movie_sync", + idempotency_key=f"test:{uuid.uuid4()}", + entity_type="movie", + external_id="603", + ) + job.attempts = attempts_already_used + await session.flush() + return job.id + + +@pytest.mark.asyncio +async def test_successful_ingestion_invalidates_the_entity_cache_key( + db_session: AsyncSession, monkeypatch +): + job_id = await _make_job(db_session) + await db_session.commit() + + result_id = uuid.uuid4() + deleted_keys: list[str] = [] + + async def fake_work(session, provider): + return result_id + + async def fake_cache_delete(redis, *keys): + deleted_keys.extend(keys) + + monkeypatch.setattr(runner, "cache_delete", fake_cache_delete) + monkeypatch.setattr(runner, "get_redis_pool", lambda: object()) + + returned_id = await runner._run_ingestion( + job_id, "movie_ingestion", fake_work, cache_entity_type="movie" + ) + + assert returned_id == result_id + assert deleted_keys == [build_key("movie", aurora_id("movie", str(result_id)))] + + +@pytest.mark.asyncio +async def test_ingestion_without_cache_entity_type_does_not_touch_the_cache( + db_session: AsyncSession, monkeypatch +): + job_id = await _make_job(db_session) + await db_session.commit() + + async def fake_work(session, provider): + return uuid.uuid4() + + called = False + + async def fake_cache_delete(redis, *keys): + nonlocal called + called = True + + monkeypatch.setattr(runner, "cache_delete", fake_cache_delete) + + await runner._run_ingestion(job_id, "season_ingestion", fake_work) + + assert called is False + + +@pytest.mark.asyncio +async def test_failed_ingestion_does_not_invalidate_the_cache( + db_session: AsyncSession, monkeypatch +): + # attempts_already_used=4 (default max_attempts is 5) so this failure is + # terminal and raises the original error rather than arq.Retry -- either + # way the cache must not be touched, but this keeps the assertion simple. + job_id = await _make_job(db_session, attempts_already_used=4) + await db_session.commit() + + async def failing_work(session, provider): + raise RuntimeError("simulated failure") + + called = False + + async def fake_cache_delete(redis, *keys): + nonlocal called + called = True + + monkeypatch.setattr(runner, "cache_delete", fake_cache_delete) + + with pytest.raises(RuntimeError): + await runner._run_ingestion( + job_id, "movie_ingestion", failing_work, cache_entity_type="movie" + ) + + assert called is False