From 5bdd23f3bca0fd4f9854d4705a00b6a93f0e39bd Mon Sep 17 00:00:00 2001 From: curo1305 Date: Tue, 23 Jun 2026 15:24:13 +0200 Subject: [PATCH] feat(14-03): add cache orchestration functions and tests - retain_or_reuse_cache_entry: reuse non-evicted entry or reactivate evicted row - pin_cache_entry / release_cache_entry: pin_count lifecycle management - update_cache_access: touch last_accessed_at for LRU ordering - Ownership assertions in all new service functions (T-14-06) - 9 new integration tests covering retain/reuse, pin/release, and isolation --- backend/services/cloud_cache.py | 229 +++++++++++++++++++++++- backend/tests/test_cloud_cache.py | 284 ++++++++++++++++++++++++++++++ 2 files changed, 512 insertions(+), 1 deletion(-) diff --git a/backend/services/cloud_cache.py b/backend/services/cloud_cache.py index ca2b197..8eddd23 100644 --- a/backend/services/cloud_cache.py +++ b/backend/services/cloud_cache.py @@ -21,7 +21,8 @@ References: ── Phase 14: durable byte cache (cloud_byte_cache_entries) ───────────────────── Exports create_cache_entry, evict_lru_entries, list_cache_entries, -increment_quota_for_cache, decrement_quota_for_cache for use by Celery tasks +increment_quota_for_cache, decrement_quota_for_cache, retain_or_reuse_cache_entry, +pin_cache_entry, release_cache_entry, update_cache_access for use by Celery tasks and the analysis API. Also re-exports compute_version_key from services.cloud_analysis_versioning so @@ -323,6 +324,232 @@ async def decrement_quota_for_cache( ) +async def retain_or_reuse_cache_entry( + session: AsyncSession, + *, + user_id: uuid.UUID, + connection_id: uuid.UUID, + cloud_item_id: uuid.UUID, + provider_item_id: str, + version_key: str, + object_key: str, + content_type: Optional[str] = None, + size_bytes: int, + content_hash: Optional[str] = None, +) -> tuple[CloudByteCacheEntry, bool]: + """Look up a non-evicted cache entry matching the version_key; create one if absent. + + Implements the Phase 14 cache lifecycle step 3 (PATTERNS.md): reuse matching + non-evicted entry before hydrating from provider. + + Ownership is strictly enforced — queries filter by (user_id, connection_id, + cloud_item_id, version_key) so a foreign user can never reuse another user's entry. + + When creating a new entry this function does NOT call increment_quota_for_cache. + The caller is responsible for reserving quota atomically before retaining bytes + and calling increment_quota_for_cache before this function. + + Args: + session: Active async SQLAlchemy session. + user_id: Owner UUID. + connection_id: Cloud connection UUID. + cloud_item_id: CloudItem row UUID. + provider_item_id: Provider item id snapshot for audit. + version_key: Idempotency key from compute_version_key. + object_key: Private MinIO object key — UUID-based, never exposed. + content_type: Optional MIME type. + size_bytes: Byte count stored in MinIO. + content_hash: Optional post-hydration content hash. + + Returns: + Tuple (entry, created_new) where created_new=True if a new row was inserted, + False if an existing non-evicted entry was reused. + """ + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + cid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id)) + iid = cloud_item_id if isinstance(cloud_item_id, uuid.UUID) else uuid.UUID(str(cloud_item_id)) + + now = datetime.now(timezone.utc) + + # Step 3: check for existing entry (evicted or not) for this version key. + # The unique constraint on (user_id, connection_id, cloud_item_id, version_key) means + # at most one row exists per version key. We handle both states: + # - non-evicted → reuse (touch last_accessed_at) + # - evicted → reactivate (clear evicted_at, update object_key + size) + stmt = ( + select(CloudByteCacheEntry) + .where( + CloudByteCacheEntry.user_id == uid, + CloudByteCacheEntry.connection_id == cid, + CloudByteCacheEntry.cloud_item_id == iid, + CloudByteCacheEntry.version_key == version_key, + ) + .limit(1) + ) + result = await session.execute(stmt) + existing = result.scalars().first() + + if existing is not None: + if existing.evicted_at is None: + # Active entry — reuse it + existing.last_accessed_at = now + existing.updated_at = now + await session.flush() + return existing, False + else: + # Evicted entry — reactivate with fresh bytes and updated object key + existing.evicted_at = None + existing.object_key = object_key + existing.size_bytes = size_bytes + existing.content_type = content_type + existing.content_hash = content_hash + existing.last_accessed_at = now + existing.updated_at = now + await session.flush() + return existing, True + + # No existing entry — create a new one + # Quota must be reserved by caller before bytes are stored in MinIO + new_entry = await create_cache_entry( + session, + user_id=uid, + connection_id=cid, + cloud_item_id=iid, + provider_item_id=provider_item_id, + version_key=version_key, + object_key=object_key, + content_type=content_type, + size_bytes=size_bytes, + content_hash=content_hash, + ) + return new_entry, True + + +async def pin_cache_entry( + session: AsyncSession, + *, + entry_id: uuid.UUID, + user_id: uuid.UUID, +) -> CloudByteCacheEntry: + """Increment pin_count to mark an active preview or download lease. + + Pinned entries are protected from LRU eviction (PATTERNS.md step 7). + Ownership is asserted — user_id must match the entry's user_id. + + Args: + session: Active async SQLAlchemy session. + entry_id: CloudByteCacheEntry UUID. + user_id: Authenticated user UUID — must own the entry. + + Returns: + Updated CloudByteCacheEntry. + + Raises: + ValueError: Entry not found or does not belong to user_id. + """ + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id)) + + result = await session.execute( + select(CloudByteCacheEntry).where( + CloudByteCacheEntry.id == eid, + CloudByteCacheEntry.user_id == uid, + ) + ) + entry = result.scalars().first() + if entry is None: + raise ValueError(f"Cache entry {eid} not found for user {uid}") + + now = datetime.now(timezone.utc) + entry.pin_count += 1 + entry.last_accessed_at = now + entry.updated_at = now + await session.flush() + return entry + + +async def release_cache_entry( + session: AsyncSession, + *, + entry_id: uuid.UUID, + user_id: uuid.UUID, +) -> CloudByteCacheEntry: + """Decrement pin_count when a preview or download lease ends. + + Clamped to 0 — defensive against double-release (PATTERNS.md step 8). + + Args: + session: Active async SQLAlchemy session. + entry_id: CloudByteCacheEntry UUID. + user_id: Authenticated user UUID — must own the entry. + + Returns: + Updated CloudByteCacheEntry. + + Raises: + ValueError: Entry not found or does not belong to user_id. + """ + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id)) + + result = await session.execute( + select(CloudByteCacheEntry).where( + CloudByteCacheEntry.id == eid, + CloudByteCacheEntry.user_id == uid, + ) + ) + entry = result.scalars().first() + if entry is None: + raise ValueError(f"Cache entry {eid} not found for user {uid}") + + now = datetime.now(timezone.utc) + entry.pin_count = max(0, entry.pin_count - 1) + entry.updated_at = now + await session.flush() + return entry + + +async def update_cache_access( + session: AsyncSession, + *, + entry_id: uuid.UUID, + user_id: uuid.UUID, +) -> CloudByteCacheEntry: + """Touch last_accessed_at to update LRU position. + + Called after a successful cache hit to maintain accurate LRU ordering. + + Args: + session: Active async SQLAlchemy session. + entry_id: CloudByteCacheEntry UUID. + user_id: Authenticated user UUID — must own the entry. + + Returns: + Updated CloudByteCacheEntry. + + Raises: + ValueError: Entry not found or does not belong to user_id. + """ + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id)) + + result = await session.execute( + select(CloudByteCacheEntry).where( + CloudByteCacheEntry.id == eid, + CloudByteCacheEntry.user_id == uid, + ) + ) + entry = result.scalars().first() + if entry is None: + raise ValueError(f"Cache entry {eid} not found for user {uid}") + + now = datetime.now(timezone.utc) + entry.last_accessed_at = now + entry.updated_at = now + await session.flush() + return entry + + # ── Phase 12: in-memory folder listing cache ────────────────────────────────── diff --git a/backend/tests/test_cloud_cache.py b/backend/tests/test_cloud_cache.py index a6adf4b..a60eb09 100644 --- a/backend/tests/test_cloud_cache.py +++ b/backend/tests/test_cloud_cache.py @@ -493,3 +493,287 @@ async def test_content_hash_not_required_during_version_key_computation(): assert key is not None assert isinstance(key, str) + + +# ─── retain_or_reuse_cache_entry ───────────────────────────────────────────── + +async def test_retain_or_reuse_returns_existing_entry(db_session): + """CACHE-03: retain_or_reuse_cache_entry returns an existing non-evicted entry.""" + from services.cloud_cache import create_cache_entry, retain_or_reuse_cache_entry + + user_id = uuid.uuid4() + conn_id = uuid.uuid4() + item_id = uuid.uuid4() + version_key = "etag-shared-v1" + + # Create the first entry as the existing cached bytes + first = await create_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-retain", + version_key=version_key, + object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf", + content_type="application/pdf", + size_bytes=1024, + ) + + # retain_or_reuse must return the existing entry, not create a new one + reused, created_new = await retain_or_reuse_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-retain", + version_key=version_key, + object_key=f"cache/{user_id}/{uuid.uuid4()}-new.pdf", # different key — should not be used + size_bytes=1024, + ) + + assert not created_new, "Should reuse existing entry, not create a new one" + assert str(reused.id) == str(first.id), "Must return the same DB row" + + +async def test_retain_or_reuse_creates_new_entry_when_absent(db_session): + """CACHE-03: retain_or_reuse_cache_entry creates a new entry if none exists.""" + from services.cloud_cache import retain_or_reuse_cache_entry + + user_id = uuid.uuid4() + conn_id = uuid.uuid4() + item_id = uuid.uuid4() + new_key = f"cache/{user_id}/{uuid.uuid4()}.pdf" + + entry, created_new = await retain_or_reuse_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-new", + version_key="etag-brand-new", + object_key=new_key, + size_bytes=2048, + ) + + assert created_new, "No existing entry — must create a new one" + assert entry.size_bytes == 2048 + assert entry.user_id == user_id + + +async def test_retain_or_reuse_reactivates_evicted_entry(db_session): + """CACHE-03: An evicted entry for the same version key is reactivated, not duplicated. + + The unique constraint on (user_id, connection_id, cloud_item_id, version_key) means + inserting a new row is impossible. retain_or_reuse_cache_entry reactivates the evicted + row (clears evicted_at) and updates the object_key to point to fresh bytes. + """ + from services.cloud_cache import create_cache_entry, retain_or_reuse_cache_entry + from datetime import datetime, timezone + + user_id = uuid.uuid4() + conn_id = uuid.uuid4() + item_id = uuid.uuid4() + version_key = "etag-evicted" + + # Create and evict an entry + evicted = await create_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-evicted", + version_key=version_key, + object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf", + size_bytes=512, + ) + evicted.evicted_at = datetime.now(timezone.utc) + await db_session.flush() + + fresh_key = f"cache/{user_id}/{uuid.uuid4()}-fresh.pdf" + reactivated, created_new = await retain_or_reuse_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-evicted", + version_key=version_key, + object_key=fresh_key, + size_bytes=512, + ) + + # created_new=True because we needed to rehydrate bytes (evicted entry was unavailable) + assert created_new, "Reactivated evicted entry must report created_new=True" + # Same DB row is reused (no duplicate due to unique constraint) + assert str(reactivated.id) == str(evicted.id) + # evicted_at must be cleared + assert reactivated.evicted_at is None, "Reactivated entry must have evicted_at=None" + # Object key updated to fresh bytes + assert reactivated.object_key == fresh_key + + +async def test_retain_or_reuse_isolates_foreign_user(db_session): + """T-14-06: retain_or_reuse_cache_entry never returns another user's entry.""" + from services.cloud_cache import create_cache_entry, retain_or_reuse_cache_entry + + owner_id = uuid.uuid4() + foreign_id = uuid.uuid4() + conn_id = uuid.uuid4() + item_id = uuid.uuid4() + version_key = "etag-shared" + + # Create entry for owner + owner_entry = await create_cache_entry( + session=db_session, + user_id=owner_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-owner", + version_key=version_key, + object_key=f"cache/{owner_id}/{uuid.uuid4()}.pdf", + size_bytes=4096, + ) + + # Foreign user must get a new entry, not the owner's + foreign_entry, created_new = await retain_or_reuse_cache_entry( + session=db_session, + user_id=foreign_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-owner", + version_key=version_key, + object_key=f"cache/{foreign_id}/{uuid.uuid4()}.pdf", + size_bytes=4096, + ) + + assert created_new, "Foreign user must not reuse another user's cache entry (T-14-06)" + assert str(foreign_entry.id) != str(owner_entry.id) + + +# ─── pin_cache_entry / release_cache_entry ──────────────────────────────────── + +async def test_pin_increments_pin_count(db_session): + """CACHE-04: pin_cache_entry increments pin_count to protect from eviction.""" + from services.cloud_cache import create_cache_entry, pin_cache_entry + + user_id = uuid.uuid4() + conn_id = uuid.uuid4() + item_id = uuid.uuid4() + + entry = await create_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-pin", + version_key="etag-pin-v1", + object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf", + size_bytes=1024, + pin_count=0, + ) + assert entry.pin_count == 0 + + pinned = await pin_cache_entry(session=db_session, entry_id=entry.id, user_id=user_id) + assert pinned.pin_count == 1, "pin_cache_entry must increment pin_count" + + +async def test_release_decrements_pin_count(db_session): + """CACHE-04: release_cache_entry decrements pin_count back to 0.""" + from services.cloud_cache import create_cache_entry, pin_cache_entry, release_cache_entry + + user_id = uuid.uuid4() + conn_id = uuid.uuid4() + item_id = uuid.uuid4() + + entry = await create_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-release", + version_key="etag-release-v1", + object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf", + size_bytes=1024, + ) + + await pin_cache_entry(session=db_session, entry_id=entry.id, user_id=user_id) + released = await release_cache_entry(session=db_session, entry_id=entry.id, user_id=user_id) + assert released.pin_count == 0, "release_cache_entry must decrement pin_count" + + +async def test_release_clamps_to_zero(db_session): + """CACHE-04: release_cache_entry on an already-zero pin_count stays at 0 (defensive).""" + from services.cloud_cache import create_cache_entry, release_cache_entry + + user_id = uuid.uuid4() + conn_id = uuid.uuid4() + item_id = uuid.uuid4() + + entry = await create_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=item_id, + provider_item_id="pitem-clamp", + version_key="etag-clamp", + object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf", + size_bytes=512, + pin_count=0, + ) + + released = await release_cache_entry(session=db_session, entry_id=entry.id, user_id=user_id) + assert released.pin_count == 0, "Double-release must not produce negative pin_count" + + +async def test_pin_raises_on_foreign_entry(db_session): + """T-14-06: pin_cache_entry raises ValueError on foreign-user entry.""" + from services.cloud_cache import create_cache_entry, pin_cache_entry + + owner_id = uuid.uuid4() + foreign_id = uuid.uuid4() + conn_id = uuid.uuid4() + + entry = await create_cache_entry( + session=db_session, + user_id=owner_id, + connection_id=conn_id, + cloud_item_id=uuid.uuid4(), + provider_item_id="pitem-foreign-pin", + version_key="etag-foreign", + object_key=f"cache/{owner_id}/{uuid.uuid4()}.pdf", + size_bytes=512, + ) + + with pytest.raises(ValueError, match="not found"): + await pin_cache_entry(session=db_session, entry_id=entry.id, user_id=foreign_id) + + +# ─── update_cache_access ───────────────────────────────────────────────────── + +async def test_update_cache_access_touches_last_accessed(db_session): + """CACHE-04: update_cache_access refreshes last_accessed_at for LRU ordering.""" + from services.cloud_cache import create_cache_entry, update_cache_access + from datetime import datetime, timezone, timedelta + + user_id = uuid.uuid4() + conn_id = uuid.uuid4() + + old_ts = datetime.now(timezone.utc) - timedelta(hours=3) + entry = await create_cache_entry( + session=db_session, + user_id=user_id, + connection_id=conn_id, + cloud_item_id=uuid.uuid4(), + provider_item_id="pitem-access", + version_key="etag-access", + object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf", + size_bytes=256, + last_accessed_at=old_ts, + ) + + updated = await update_cache_access( + session=db_session, entry_id=entry.id, user_id=user_id + ) + + assert updated.last_accessed_at > old_ts, ( + "update_cache_access must refresh last_accessed_at beyond the old timestamp" + )