From e9ee5d4ba542d04b74b74115e7a3e890f742e9bd Mon Sep 17 00:00:00 2001 From: curo1305 Date: Thu, 4 Jun 2026 23:08:16 +0200 Subject: [PATCH] feat(07-04): Celery retry harness + _ClassificationError sentinel (D-09/D-10) - Add _ClassificationError sentinel raised by _run() for classification failures - Add _mark_classification_failed() async helper for final status writeback - Change decorator to bind=True, max_retries=3 - Outer sync task catches _ClassificationError and calls self.retry(countdown=[30,90,270]) - MaxRetriesExceededError caught in nested try/except to avoid re-raise from exc block - Promote test_retry_backoff and test_exhaustion_sets_failed_status from xfail to passing - Tests use push_request(retries=N) + patch.object(task, "retry") to verify countdown values - test_exhaustion_sets_failed_status uses assert_called_once_with (asyncio.run calling convention) --- backend/tasks/document_tasks.py | 91 ++++++++++++++++++++++----- backend/tests/test_document_tasks.py | 93 ++++++++++++++++++++++++---- 2 files changed, 158 insertions(+), 26 deletions(-) diff --git a/backend/tasks/document_tasks.py b/backend/tasks/document_tasks.py index 4d89c33..a86a2a1 100644 --- a/backend/tasks/document_tasks.py +++ b/backend/tasks/document_tasks.py @@ -12,17 +12,82 @@ Flow: 4. Extract text from bytes using services.extractor 5. Persist extracted_text back to the Document row 6. Call services.classifier.classify_document to assign topics - 7. Return a result dict (never raises — classification failures are non-fatal) + 7. Return a result dict; classification failures are retried with exponential backoff + +Celery retry harness (D-09, D-10 — Pitfall 3 from RESEARCH.md): + CRITICAL: self.retry() MUST be raised from the outer sync task, NOT inside + asyncio.run(). _ClassificationError is a sentinel raised by _run() to signal + a retryable classification failure. The outer extract_and_classify catches it + and calls self.retry(countdown=...) in the sync layer. """ import asyncio +from celery.exceptions import MaxRetriesExceededError + from celery_app import celery_app -@celery_app.task(name="tasks.document_tasks.extract_and_classify") -def extract_and_classify(document_id: str) -> dict: - """Synchronous Celery entry-point — delegates to async _run via asyncio.run.""" - return asyncio.run(_run(document_id)) +class _ClassificationError(Exception): + """Sentinel exception raised by _run() to signal a retryable classification failure. + + This exception escapes asyncio.run() and is caught by the outer sync task, + which then calls self.retry() in the Celery sync layer (not inside asyncio.run). + Non-classification failures (extract_failed, invalid_id) are NOT retried — + they return a dict directly from _run() without raising _ClassificationError. + """ + pass + + +async def _mark_classification_failed(document_id: str) -> None: + """Write doc.status = 'classification_failed' after all retries are exhausted. + + D-10: called by extract_and_classify's MaxRetriesExceededError handler via + asyncio.run(_mark_classification_failed(document_id)) so the final status + is written to DB regardless of Celery task context. + """ + import uuid as _uuid + + from db.session import AsyncSessionLocal + from db.models import Document + + async with AsyncSessionLocal() as session: + try: + doc_uuid = _uuid.UUID(document_id) + except ValueError: + return # Silently ignore invalid IDs — nothing to update + doc = await session.get(Document, doc_uuid) + if doc is not None: + doc.status = "classification_failed" + await session.commit() + + +@celery_app.task( + name="tasks.document_tasks.extract_and_classify", + bind=True, + max_retries=3, +) +def extract_and_classify(self, document_id: str) -> dict: + """Synchronous Celery entry-point — delegates to async _run via asyncio.run. + + Retry harness (D-09): classification failures are retried up to 3 times + with exponential backoff: 30s / 90s / 270s. + + Pitfall 3 guard: self.retry() is called HERE in the sync layer, never + inside asyncio.run(). _ClassificationError is the sentinel that escapes + asyncio.run() to trigger the retry. + """ + try: + return asyncio.run(_run(document_id)) + except _ClassificationError as exc: + # Exponential backoff: 30s on first retry, 90s on second, 270s on third + countdowns = [30, 90, 270] + countdown = countdowns[min(self.request.retries, 2)] + try: + raise self.retry(exc=exc, countdown=countdown) + except MaxRetriesExceededError: + # All retries exhausted — write final failure status to DB + asyncio.run(_mark_classification_failed(document_id)) + return {"document_id": document_id, "status": "classification_failed"} async def _run(document_id: str) -> dict: @@ -34,6 +99,9 @@ async def _run(document_id: str) -> dict: Cloud-aware: when doc.storage_backend != 'minio', uses get_storage_backend_for_document() to retrieve bytes from the correct cloud backend instead of hardcoding MinIO. + + Classification failures raise _ClassificationError (D-09 retryable sentinel). + Non-classification failures return a status dict (not retried). """ import uuid as _uuid @@ -105,7 +173,7 @@ async def _run(document_id: str) -> dict: "error": f"Text extraction failed: {e}", } - # ── Step 4: classify document (non-fatal) ────────────────────────────── + # ── Step 4: classify document (retryable via _ClassificationError) ───── try: topics = await classifier.classify_document(session, document_id, ai_provider=ai_provider, ai_model=ai_model) return { @@ -114,14 +182,9 @@ async def _run(document_id: str) -> dict: "topics": topics, } except Exception as e: - # Non-fatal — preserve existing convention from api/documents.py - doc.status = "classification_failed" - await session.commit() - return { - "document_id": document_id, - "status": "classification_failed", - "error": str(e), - } + # Raise sentinel to allow the outer sync layer to call self.retry() + # (D-09 Pitfall 3: self.retry() cannot be called inside asyncio.run) + raise _ClassificationError(str(e)) from e @celery_app.task(name="tasks.document_tasks.cleanup_abandoned_uploads") diff --git a/backend/tests/test_document_tasks.py b/backend/tests/test_document_tasks.py index 3b5d632..f90fa42 100644 --- a/backend/tests/test_document_tasks.py +++ b/backend/tests/test_document_tasks.py @@ -1,25 +1,94 @@ """ -Wave 0 xfail stubs for Phase 7 Celery document task tests. +Phase 7 Plan 04 — Celery retry harness tests. Covers: - D-09: Celery retry with 30s/90s/270s exponential backoff - D-10: After 3 retries doc.status = "classification_failed" -Each function is a placeholder to be promoted in Plan 07-04 once the -bind=True retry harness is implemented in document_tasks.py. +Test contract note (asyncio.run + AsyncMock): + Production code uses asyncio.run(_mark_classification_failed(document_id)). + asyncio.run() invokes the coroutine object returned by calling the factory: + _mark_classification_failed(document_id). This records the invocation as a + __call__ on the mock, NOT as an __await__. Therefore tests must assert via + assert_called_once_with(), NOT assert_awaited_once_with(). -Stub policy (STATE.md decision: xfail(strict=False) for Wave 0): - - Body is a single pytest.xfail() call — no assertion code. - - strict=False so unexpected passes (xpass) never break CI. +Celery bind=True calling convention: + For bind=True tasks, task.run() is the raw underlying function with self=task. + We push a fake request context via push_request(retries=N) to simulate retries, + then patch task.retry to capture the countdown argument rather than letting + Celery's called_directly path re-raise the original exception. """ import pytest +from unittest.mock import patch, AsyncMock +from celery.exceptions import Retry, MaxRetriesExceededError -@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-04") -async def test_retry_backoff(): - pytest.xfail("not implemented yet — Plan 07-04") +def test_retry_backoff(): + """extract_and_classify raises Retry with countdown 30/90/270 on retries 0/1/2. + + D-09: exponential backoff harness. The outer sync task must catch + _ClassificationError and call self.retry(exc=..., countdown=N) where N + comes from [30, 90, 270][min(self.request.retries, 2)]. + """ + from tasks.document_tasks import extract_and_classify, _ClassificationError + + doc_id = "11111111-1111-1111-1111-111111111111" + expected_countdowns = [30, 90, 270] + captured_countdowns = [] + + async def fake_run_raises(document_id): + raise _ClassificationError("fake classification error") + + def capture_retry(exc=None, countdown=None, **kwargs): + """Capture the countdown argument and raise Retry to simulate Celery behaviour.""" + captured_countdowns.append(countdown) + raise Retry(str(exc), exc) + + for retry_num in range(3): + extract_and_classify.push_request(retries=retry_num) + try: + with patch("tasks.document_tasks._run", side_effect=fake_run_raises), \ + patch.object(extract_and_classify, "retry", side_effect=capture_retry): + with pytest.raises(Retry): + extract_and_classify.run(doc_id) + finally: + extract_and_classify.pop_request() + + assert captured_countdowns == expected_countdowns, ( + f"Expected countdowns {expected_countdowns}, got {captured_countdowns}" + ) -@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-04") -async def test_exhaustion_sets_failed_status(): - pytest.xfail("not implemented yet — Plan 07-04") +def test_exhaustion_sets_failed_status(): + """When MaxRetriesExceededError is raised, _mark_classification_failed is called + and the task returns {'document_id': ..., 'status': 'classification_failed'}. + + D-10: final state writeback after exhausting all retries. + + Test contract: assert_called_once_with (not assert_awaited_once_with) because + asyncio.run(_mark_classification_failed(doc_id)) invokes the factory — that is + the __call__ path, not the __await__ path, on the AsyncMock. + """ + from tasks.document_tasks import extract_and_classify, _ClassificationError + + doc_id = "22222222-2222-2222-2222-222222222222" + mock_mark_failed = AsyncMock() + + async def fake_run_raises(document_id): + raise _ClassificationError("fake classification error") + + extract_and_classify.push_request(retries=3) + try: + with patch("tasks.document_tasks._run", side_effect=fake_run_raises), \ + patch("tasks.document_tasks._mark_classification_failed", mock_mark_failed), \ + patch.object(extract_and_classify, "retry", side_effect=MaxRetriesExceededError()): + result = extract_and_classify.run(doc_id) + finally: + extract_and_classify.pop_request() + + # _mark_classification_failed(document_id) was called (factory call, not awaited) + mock_mark_failed.assert_called_once_with(doc_id) + + # Return value reports final failure state + assert result["status"] == "classification_failed" + assert result["document_id"] == doc_id