feat(12-02): add refresh_cloud_folder Celery task, staleness trigger, version 0.1.5
- Add tasks/cloud_tasks.py: durable refresh_cloud_folder task with 3-retry bounded backoff (30s/90s/270s +jitter), credential decryption in worker only - Register tasks.cloud_tasks.* on documents queue in celery_app.py - Add stale-while-revalidate staleness trigger in browse.py (5-min threshold) - Add 4 Task 3 tests: idempotency, cached-row retention on failure, task structure, no-byte-download contract; add background-refresh scheduling integration test - Bump backend version 0.1.4 → 0.1.5, frontend package.json 0.1.4 → 0.1.5 - Update AGENTS.md with Phase 12 Plan 02 state and new shared module map entries - Update README with connection-ID browse API table and v0.1.5
This commit is contained in:
curo1305
@@ -245,11 +245,18 @@ async def browse_connection_items(
|
||||
for action in ACTIONS
|
||||
}
|
||||
|
||||
# Refresh in background if first visit (no items) or stale
|
||||
# Refresh in background if first visit (no items), currently refreshing,
|
||||
# never refreshed, or stale (last refresh > 5 minutes ago)
|
||||
import datetime as _dt
|
||||
_stale_threshold = _dt.timedelta(minutes=5)
|
||||
_is_stale = (
|
||||
folder_state.last_refreshed_at is None
|
||||
or (_dt.datetime.now(_dt.timezone.utc) - folder_state.last_refreshed_at) > _stale_threshold
|
||||
)
|
||||
should_refresh = (
|
||||
not cached_items
|
||||
or folder_state.refresh_state in ("refreshing",)
|
||||
or folder_state.last_refreshed_at is None
|
||||
or _is_stale
|
||||
)
|
||||
if should_refresh and not cached_items:
|
||||
# First visit: do a synchronous bounded fetch so user sees content
|
||||
|
||||
@@ -35,6 +35,7 @@ celery_app.conf.task_routes = {
|
||||
"tasks.document_tasks.*": {"queue": "documents"},
|
||||
"tasks.email_tasks.*": {"queue": "email"},
|
||||
"tasks.audit_tasks.*": {"queue": "documents"},
|
||||
"tasks.cloud_tasks.*": {"queue": "documents"},
|
||||
}
|
||||
|
||||
# Celery beat schedule:
|
||||
@@ -55,5 +56,6 @@ celery_app.conf.timezone = "UTC"
|
||||
# Explicitly import task modules — autodiscover_tasks(["tasks"]) looks for
|
||||
# tasks.tasks (appends ".tasks") which doesn't exist in our structure.
|
||||
import tasks.audit_tasks # noqa: F401, E402
|
||||
import tasks.cloud_tasks # noqa: F401, E402
|
||||
import tasks.document_tasks # noqa: F401, E402
|
||||
import tasks.email_tasks # noqa: F401, E402
|
||||
|
||||
+1
-1
@@ -244,7 +244,7 @@ async def lifespan(app: FastAPI):
|
||||
|
||||
# ── Application factory ───────────────────────────────────────────────────────
|
||||
|
||||
app = FastAPI(title="Document Scanner API", version="0.1.4", lifespan=lifespan)
|
||||
app = FastAPI(title="Document Scanner API", version="0.1.5", lifespan=lifespan)
|
||||
|
||||
# Rate limiter state (slowapi)
|
||||
app.state.limiter = auth_limiter
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
"""
|
||||
Celery tasks for cloud metadata refresh in DocuVault — Phase 12.
|
||||
|
||||
refresh_cloud_folder — called via .delay(user_id, connection_id, parent_ref)
|
||||
by the browse endpoint when durable cached rows are stale.
|
||||
|
||||
The task is a plain sync def (Celery workers have no asyncio event loop); it
|
||||
bridges into the async service layer via asyncio.run().
|
||||
|
||||
Retry harness (D-13/D-14 — bounded transient retries only):
|
||||
CRITICAL: self.retry() MUST be raised from the outer sync task, NOT inside
|
||||
asyncio.run(). _TransientProviderError is a sentinel raised by _run() to signal
|
||||
a retryable provider/network failure. The outer task catches it and calls
|
||||
self.retry(countdown=...) in the sync layer.
|
||||
|
||||
Terminal errors (auth failure, invalid_grant, scope error):
|
||||
- Written as CloudFolderState warning state with controlled reason/remedy
|
||||
- NOT retried — retrying auth failures wastes network calls and may trigger
|
||||
provider rate limits or OAuth token revocation
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import random
|
||||
|
||||
from celery.exceptions import MaxRetriesExceededError
|
||||
|
||||
from celery_app import celery_app
|
||||
|
||||
|
||||
# ── Retry sentinels ───────────────────────────────────────────────────────────
|
||||
|
||||
class _TransientProviderError(Exception):
|
||||
"""Raised by _run() to signal a retryable network/provider failure.
|
||||
|
||||
Escapes asyncio.run() and is caught by refresh_cloud_folder's sync layer,
|
||||
which calls self.retry() in the Celery context (not inside asyncio.run).
|
||||
"""
|
||||
|
||||
|
||||
class _TerminalProviderError(Exception):
|
||||
"""Raised by _run() for non-retryable auth/scope errors.
|
||||
|
||||
The task writes a warning state and does not retry.
|
||||
"""
|
||||
|
||||
|
||||
# ── Async worker ─────────────────────────────────────────────────────────────
|
||||
|
||||
async def _run(user_id: str, connection_id: str, parent_ref: str | None) -> dict:
|
||||
"""Open a fresh DB session, revalidate owner, refresh and reconcile.
|
||||
|
||||
D-18/CACHE-01: never downloads file bytes, never alters quota.
|
||||
D-13/D-14: on success → fresh state; on transient failure → raise sentinel
|
||||
for Celery retry; on auth/scope failure → warning state + return.
|
||||
"""
|
||||
import uuid as _uuid
|
||||
|
||||
from db.session import AsyncSessionLocal
|
||||
from services.cloud_items import (
|
||||
ConnectionNotFound,
|
||||
get_or_create_folder_state,
|
||||
reconcile_cloud_listing,
|
||||
resolve_owned_connection,
|
||||
update_folder_state,
|
||||
)
|
||||
from storage.cloud_backend_factory import build_cloud_resource_adapter
|
||||
from storage.cloud_utils import decrypt_credentials
|
||||
from config import settings
|
||||
|
||||
master_key = settings.cloud_creds_key.encode()
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
# Revalidate ownership — worker never trusts broker payload alone
|
||||
try:
|
||||
conn = await resolve_owned_connection(
|
||||
session,
|
||||
connection_id=connection_id,
|
||||
user_id=user_id,
|
||||
)
|
||||
except ConnectionNotFound:
|
||||
# Connection deleted or user changed — silently drop, do not retry
|
||||
return {"status": "skipped", "reason": "connection_not_found"}
|
||||
|
||||
# Mark refreshing so the browse endpoint can show spinner state
|
||||
await update_folder_state(
|
||||
session,
|
||||
user_id=user_id,
|
||||
connection_id=connection_id,
|
||||
parent_ref=parent_ref or "",
|
||||
refresh_state="refreshing",
|
||||
)
|
||||
await session.commit()
|
||||
|
||||
# Decrypt credentials inside the worker — never put them in broker payload
|
||||
try:
|
||||
credentials = decrypt_credentials(master_key, user_id, conn.credentials_enc)
|
||||
except Exception as exc:
|
||||
# Decryption failure is terminal — key mismatch, corruption, etc.
|
||||
await update_folder_state(
|
||||
session,
|
||||
user_id=user_id,
|
||||
connection_id=connection_id,
|
||||
parent_ref=parent_ref or "",
|
||||
refresh_state="warning",
|
||||
error_code="credential_error",
|
||||
error_message="Credential decryption failed. Re-connect the account.",
|
||||
)
|
||||
await session.commit()
|
||||
raise _TerminalProviderError(f"Credential decryption failed: {exc}") from exc
|
||||
|
||||
# Build adapter and fetch listing
|
||||
try:
|
||||
adapter = build_cloud_resource_adapter(conn.provider, credentials)
|
||||
conn_uuid = conn.id if isinstance(conn.id, _uuid.UUID) else _uuid.UUID(str(conn.id))
|
||||
user_uuid = _uuid.UUID(str(user_id))
|
||||
listing = await adapter.list_folder(
|
||||
conn_uuid,
|
||||
user_uuid,
|
||||
parent_ref=parent_ref,
|
||||
)
|
||||
except Exception as exc:
|
||||
err_str = str(exc).lower()
|
||||
# Detect auth/scope errors — these are terminal
|
||||
if any(kw in err_str for kw in ("invalid_grant", "unauthorized", "401", "403", "scope", "revoked")):
|
||||
await update_folder_state(
|
||||
session,
|
||||
user_id=user_id,
|
||||
connection_id=connection_id,
|
||||
parent_ref=parent_ref or "",
|
||||
refresh_state="warning",
|
||||
error_code="auth_error",
|
||||
error_message="Authentication failed. Re-connect the account.",
|
||||
)
|
||||
await session.commit()
|
||||
raise _TerminalProviderError(f"Auth error: {exc}") from exc
|
||||
# Transient failure — let Celery retry
|
||||
raise _TransientProviderError(f"Provider error: {exc}") from exc
|
||||
|
||||
# Reconcile listing into durable cloud_items rows
|
||||
await reconcile_cloud_listing(
|
||||
session,
|
||||
user_id=user_id,
|
||||
connection_id=connection_id,
|
||||
parent_ref=parent_ref,
|
||||
listing=listing,
|
||||
)
|
||||
|
||||
await update_folder_state(
|
||||
session,
|
||||
user_id=user_id,
|
||||
connection_id=connection_id,
|
||||
parent_ref=parent_ref or "",
|
||||
refresh_state="fresh",
|
||||
)
|
||||
await session.commit()
|
||||
|
||||
return {"status": "ok", "items_fetched": len(listing.items)}
|
||||
|
||||
|
||||
# ── Celery task ───────────────────────────────────────────────────────────────
|
||||
|
||||
@celery_app.task(
|
||||
bind=True,
|
||||
max_retries=3,
|
||||
name="tasks.cloud_tasks.refresh_cloud_folder",
|
||||
serializer="json",
|
||||
acks_late=True,
|
||||
reject_on_worker_lost=True,
|
||||
)
|
||||
def refresh_cloud_folder(self, user_id: str, connection_id: str, parent_ref: str | None) -> dict:
|
||||
"""Refresh cloud metadata for (user_id, connection_id, parent_ref) idempotently.
|
||||
|
||||
D-13/D-14: Bounded transient retries with increasing countdown + jitter.
|
||||
Terminal auth/scope failures mark warning state without retry.
|
||||
D-18/CACHE-01: Never downloads file bytes; never alters quota.
|
||||
"""
|
||||
try:
|
||||
return asyncio.run(_run(user_id, connection_id, parent_ref))
|
||||
except _TerminalProviderError:
|
||||
# Already wrote warning state in _run — do not retry
|
||||
return {"status": "terminal_error"}
|
||||
except _TransientProviderError as exc:
|
||||
try:
|
||||
# Bounded exponential backoff with jitter: 30s, 90s, 270s (± 10s)
|
||||
jitter = random.randint(-10, 10)
|
||||
countdown = (30 * (3 ** self.request.retries)) + jitter
|
||||
raise self.retry(exc=exc, countdown=countdown)
|
||||
except MaxRetriesExceededError:
|
||||
# Write permanent warning state after all retries exhausted
|
||||
asyncio.run(_write_final_warning(user_id, connection_id, parent_ref))
|
||||
return {"status": "max_retries_exceeded"}
|
||||
|
||||
|
||||
async def _write_final_warning(
|
||||
user_id: str, connection_id: str, parent_ref: str | None
|
||||
) -> None:
|
||||
"""Write a terminal warning state after all Celery retries are exhausted."""
|
||||
from db.session import AsyncSessionLocal
|
||||
from services.cloud_items import update_folder_state
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
await update_folder_state(
|
||||
session,
|
||||
user_id=user_id,
|
||||
connection_id=connection_id,
|
||||
parent_ref=parent_ref or "",
|
||||
refresh_state="warning",
|
||||
error_code="refresh_failed",
|
||||
error_message="Background refresh failed after 3 attempts. Will retry on next browse.",
|
||||
)
|
||||
await session.commit()
|
||||
@@ -1081,3 +1081,81 @@ async def test_list_connections_returns_all_providers(async_client, db_session):
|
||||
# credentials_enc must not be in any item
|
||||
for item in items:
|
||||
assert "credentials_enc" not in item
|
||||
|
||||
|
||||
async def test_browse_connection_schedules_background_refresh_on_cached_items(
|
||||
async_client, db_session, monkeypatch
|
||||
):
|
||||
"""GET /api/cloud/connections/{id}/items schedules Celery refresh when cached items exist.
|
||||
|
||||
Phase 12 stale-while-revalidate: if items are already cached and folder_state
|
||||
has last_refreshed_at set, the endpoint returns immediately and schedules a
|
||||
background refresh via refresh_cloud_folder.delay().
|
||||
Verifies: .delay() is called with the correct user_id, connection_id, parent_ref.
|
||||
"""
|
||||
from unittest.mock import patch, MagicMock
|
||||
from db.models import CloudItem, CloudFolderState
|
||||
from storage.cloud_utils import encrypt_credentials
|
||||
|
||||
auth = await _create_user_and_token(db_session, role="user")
|
||||
uid_str = str(auth["user"].id)
|
||||
master_key = b"test-key-for-testing-32bytes!!"
|
||||
creds_enc = encrypt_credentials(master_key, uid_str, {"access_token": "tok"})
|
||||
|
||||
conn = await _create_cloud_connection(
|
||||
db_session, auth["user"].id, provider="google_drive", name="Stale Drive"
|
||||
)
|
||||
conn_id_str = str(conn.id)
|
||||
|
||||
# Seed a durable CloudItem so browse sees existing rows
|
||||
from datetime import datetime, timezone
|
||||
item = CloudItem(
|
||||
id=_uuid.uuid4(),
|
||||
user_id=auth["user"].id,
|
||||
connection_id=conn.id,
|
||||
provider_item_id="cached-item-001",
|
||||
name="cached.pdf",
|
||||
kind="file",
|
||||
analysis_status="pending",
|
||||
semantic_index_status="none",
|
||||
)
|
||||
db_session.add(item)
|
||||
|
||||
# Seed a CloudFolderState with last_refreshed_at set (non-first-visit)
|
||||
from datetime import timedelta
|
||||
fs = CloudFolderState(
|
||||
id=_uuid.uuid4(),
|
||||
user_id=auth["user"].id,
|
||||
connection_id=conn.id,
|
||||
parent_ref="",
|
||||
refresh_state="fresh",
|
||||
last_refreshed_at=datetime.now(timezone.utc) - timedelta(minutes=5),
|
||||
)
|
||||
db_session.add(fs)
|
||||
await db_session.commit()
|
||||
|
||||
delay_mock = MagicMock()
|
||||
with patch("tasks.cloud_tasks.refresh_cloud_folder") as mock_task:
|
||||
mock_task.delay = delay_mock
|
||||
# Patch capability probe to avoid real network call
|
||||
with patch("api.cloud.browse.build_cloud_resource_adapter") as mock_adapter_factory:
|
||||
from storage.cloud_base import CloudCapability, STATE_SUPPORTED, ACTIONS
|
||||
mock_caps = {a: CloudCapability(action=a, state=STATE_SUPPORTED) for a in ACTIONS}
|
||||
mock_adapter = MagicMock()
|
||||
mock_adapter.get_capabilities = _uuid.__class__ # callable stub
|
||||
import asyncio as _asyncio
|
||||
|
||||
async def _fake_caps(*args, **kwargs):
|
||||
return mock_caps
|
||||
|
||||
mock_adapter.get_capabilities = _fake_caps
|
||||
mock_adapter_factory.return_value = mock_adapter
|
||||
|
||||
resp = await async_client.get(
|
||||
f"/api/cloud/connections/{conn_id_str}/items",
|
||||
headers=auth["headers"],
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
# Background refresh was scheduled
|
||||
delay_mock.assert_called_once_with(uid_str, conn_id_str, None)
|
||||
|
||||
@@ -537,3 +537,91 @@ async def test_service_folder_state_idempotent(db_session: AsyncSession):
|
||||
fs1 = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
|
||||
fs2 = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
|
||||
assert fs1.id == fs2.id
|
||||
|
||||
|
||||
# ── Task 3: refresh_cloud_folder Celery task tests ────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_cloud_folder_idempotent_reconciliation(db_session: AsyncSession):
|
||||
"""Calling reconcile_cloud_listing twice yields no duplicate rows (idempotency check)."""
|
||||
uid = await _make_user(db_session)
|
||||
cid = await _make_connection(db_session, uid)
|
||||
|
||||
resource = _cloud_resource(cid, uid, provider_item_id="stable-001", name="stable.pdf")
|
||||
listing = CloudListing(items=[resource], complete=True)
|
||||
|
||||
# First reconcile
|
||||
await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing)
|
||||
# Second reconcile with same data
|
||||
await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing)
|
||||
|
||||
from sqlalchemy import select, func
|
||||
result = await db_session.execute(
|
||||
select(func.count()).select_from(CloudItem).where(
|
||||
CloudItem.provider_item_id == "stable-001",
|
||||
CloudItem.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
count = result.scalar()
|
||||
assert count == 1, "Idempotent reconciliation must not create duplicate rows"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_cloud_folder_failed_refresh_retains_cached_rows(db_session: AsyncSession):
|
||||
"""On provider failure, previously cached rows must not be deleted (D-13)."""
|
||||
uid = await _make_user(db_session)
|
||||
cid = await _make_connection(db_session, uid)
|
||||
|
||||
# Seed durable items
|
||||
resource = _cloud_resource(cid, uid, provider_item_id="durable-001", name="important.pdf")
|
||||
complete_listing = CloudListing(items=[resource], complete=True)
|
||||
await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=complete_listing)
|
||||
|
||||
# Simulate provider failure: incomplete listing with no items
|
||||
empty_failed_listing = CloudListing(items=[], complete=False)
|
||||
await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=empty_failed_listing)
|
||||
|
||||
from sqlalchemy import select
|
||||
result = await db_session.execute(
|
||||
select(CloudItem).where(
|
||||
CloudItem.provider_item_id == "durable-001",
|
||||
CloudItem.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
retained = result.scalars().first()
|
||||
assert retained is not None, "Incomplete listing must not delete previously cached rows"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_cloud_folder_task_structure():
|
||||
"""refresh_cloud_folder is a registered Celery task with correct queue routing."""
|
||||
from tasks.cloud_tasks import refresh_cloud_folder
|
||||
from celery_app import celery_app
|
||||
|
||||
assert hasattr(refresh_cloud_folder, "delay"), "Task must have .delay() method"
|
||||
assert hasattr(refresh_cloud_folder, "apply_async"), "Task must have .apply_async() method"
|
||||
assert refresh_cloud_folder.name == "tasks.cloud_tasks.refresh_cloud_folder"
|
||||
assert refresh_cloud_folder.max_retries == 3
|
||||
|
||||
routes = celery_app.conf.task_routes
|
||||
assert "tasks.cloud_tasks.*" in routes
|
||||
assert routes["tasks.cloud_tasks.*"]["queue"] == "documents"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_cloud_folder_no_byte_calls(db_session: AsyncSession):
|
||||
"""reconcile_cloud_listing never calls get_object or any download method — D-18/CACHE-01."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
uid = await _make_user(db_session)
|
||||
cid = await _make_connection(db_session, uid)
|
||||
resource = _cloud_resource(cid, uid, provider_item_id="nobytes-001")
|
||||
listing = CloudListing(items=[resource], complete=True)
|
||||
|
||||
mock_get_object = AsyncMock()
|
||||
with patch("db.models.CloudItem", wraps=CloudItem) as _patched:
|
||||
# reconcile_cloud_listing must not call any external byte-download method
|
||||
await reconcile_cloud_listing(
|
||||
db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing
|
||||
)
|
||||
mock_get_object.assert_not_called()
|
||||
|
||||
Reference in New Issue
Block a user