Fix wake-up race: commit dispatch mark before publishing

app.py

@@ -58,8 +58,21 @@ def _log(message: str) -> None:
 def start_redis() -> subprocess.Popen[bytes]:
     Path("/tmp/redis").mkdir(exist_ok=True)
     proc = subprocess.Popen(
-        ["redis-server", "--bind", "127.0.0.1", "--protected-mode", "yes", "--port", "6379",
-         "--dir", "/tmp/redis", "--save", "", "--appendonly", "no"]
+        [
+            "redis-server",
+            "--bind",
+            "127.0.0.1",
+            "--protected-mode",
+            "yes",
+            "--port",
+            "6379",
+            "--dir",
+            "/tmp/redis",
+            "--save",
+            "",
+            "--appendonly",
+            "no",
+        ]
     )
     for _ in range(30):
         ping = subprocess.run(["redis-cli", "-h", "127.0.0.1", "ping"], capture_output=True)
@@ -79,8 +92,17 @@ def run_migrations() -> None:
 
 def start_worker() -> subprocess.Popen[bytes]:
     return subprocess.Popen(
-        [sys.executable, "-m", "celery", "-A", "app.worker.celery_app", "worker",
-         "--pool=solo", "--concurrency=1", "--loglevel=info"]
+        [
+            sys.executable,
+            "-m",
+            "celery",
+            "-A",
+            "app.worker.celery_app",
+            "worker",
+            "--pool=solo",
+            "--concurrency=1",
+            "--loglevel=info",
+        ]
     )
 
 

backend/app/api/routes/reviews.py

@@ -1,115 +1,119 @@
-"""Review submission and polling routes (PRD §14.1–14.2)."""
-
-from __future__ import annotations
-
-import logging
-from datetime import UTC, datetime
-
-from fastapi import APIRouter, Depends, HTTPException, status
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session
-
-from app.api.deps import get_db
-from app.models.job import Job, JobKind
-from app.models.product import Product
-from app.models.review import Review, ReviewStatus
-from app.schemas.review import ReviewCreate, ReviewResponse, ReviewTaskStatus
-from app.worker.outbox import send_wakeup
-
-logger = logging.getLogger(__name__)
-router = APIRouter(prefix="/api/reviews", tags=["reviews"])
-
-STATUS_MESSAGES: dict[str, str] = {
-    ReviewStatus.PENDING.value: "Review is queued for processing",
-    ReviewStatus.PROCESSING.value: "Review is being processed",
-    ReviewStatus.DONE.value: "Review processing completed",
-    ReviewStatus.FAILED.value: "Review processing failed",
-}
-
-
-def _get_review_or_404(db: Session, review_id: int) -> Review:
-    review = db.get(Review, review_id)
-    if review is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Review not found")
-    return review
-
-
-def _ensure_product_shell(db: Session, asin: str, title: str | None) -> None:
-    """Create or update the product row without losing concurrent first reviews."""
-    product = db.get(Product, asin)
-    if product is None:
-        try:
-            # The savepoint lets a concurrent ASIN insert fail locally while
-            # preserving the review/job transaction that follows it.
-            with db.begin_nested():
-                db.add(Product(asin=asin, title=title))
-                db.flush()
-        except IntegrityError:
-            product = db.get(Product, asin)
-            if product is None:
-                raise
-    if product is not None and title and not product.title:
-        product.title = title
-
-
-@router.post("", response_model=ReviewTaskStatus, status_code=status.HTTP_202_ACCEPTED)
-def submit_review(payload: ReviewCreate, db: Session = Depends(get_db)) -> ReviewTaskStatus:
-    """Persist a review and its processing job atomically; return for polling.
-
-    The product shell is created first because ``reviews.product_asin`` is a
-    foreign key (the PRD's schema made every first review of an ASIN fail).
-    After commit a wake-up is sent to Redis; if Redis is down the job is still
-    durable and the outbox dispatcher delivers it later, so this never 503s.
-    """
-    asin = payload.product_asin
-    if asin:
-        _ensure_product_shell(db, asin, payload.product_title)
-
-    review = Review(
-        text=payload.text,
-        star_rating=payload.star_rating,
-        helpful_votes=payload.helpful_votes,
-        verified_purchase=payload.verified_purchase,
-        product_title=payload.product_title,
-        product_asin=asin,
-        status=ReviewStatus.PENDING.value,
-    )
-    db.add(review)
-    db.flush()  # assigns review.id
-    job = Job(kind=JobKind.REVIEW.value, review_id=review.id)
-    db.add(job)
-    db.commit()
-
-    try:
-        send_wakeup(job.id)
-        job.last_dispatched_at = datetime.now(UTC)
-        db.commit()
-    except Exception:
-        db.rollback()
-        logger.warning("Broker unavailable for job %s; outbox will retry", job.id, exc_info=True)
-
-    return ReviewTaskStatus(
-        review_id=review.id,
-        task_id=job.id,
-        status=ReviewStatus.PENDING.value,
-        message="Review queued for processing",
-    )
-
-
-@router.get("/{review_id}", response_model=ReviewResponse)
-def get_review(review_id: int, db: Session = Depends(get_db)) -> Review:
-    """Get a review and, when available, its pipeline output."""
-    return _get_review_or_404(db, review_id)
-
-
-@router.get("/{review_id}/status", response_model=ReviewTaskStatus)
-def get_review_status(review_id: int, db: Session = Depends(get_db)) -> ReviewTaskStatus:
-    """Lightweight lifecycle state for polling clients."""
-    review = _get_review_or_404(db, review_id)
-    job_id = db.query(Job.id).filter(Job.review_id == review.id).scalar()
-    return ReviewTaskStatus(
-        review_id=review.id,
-        task_id=job_id,
-        status=review.status,
-        message=STATUS_MESSAGES.get(review.status, "Unknown status"),
-    )
+"""Review submission and polling routes (PRD §14.1–14.2)."""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+from datetime import UTC, datetime
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_db
+from app.models.job import Job, JobKind
+from app.models.product import Product
+from app.models.review import Review, ReviewStatus
+from app.schemas.review import ReviewCreate, ReviewResponse, ReviewTaskStatus
+from app.worker.outbox import send_wakeup
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/reviews", tags=["reviews"])
+
+STATUS_MESSAGES: dict[str, str] = {
+    ReviewStatus.PENDING.value: "Review is queued for processing",
+    ReviewStatus.PROCESSING.value: "Review is being processed",
+    ReviewStatus.DONE.value: "Review processing completed",
+    ReviewStatus.FAILED.value: "Review processing failed",
+}
+
+
+def _get_review_or_404(db: Session, review_id: int) -> Review:
+    review = db.get(Review, review_id)
+    if review is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Review not found")
+    return review
+
+
+def _ensure_product_shell(db: Session, asin: str, title: str | None) -> None:
+    """Create or update the product row without losing concurrent first reviews."""
+    product = db.get(Product, asin)
+    if product is None:
+        try:
+            # The savepoint lets a concurrent ASIN insert fail locally while
+            # preserving the review/job transaction that follows it.
+            with db.begin_nested():
+                db.add(Product(asin=asin, title=title))
+                db.flush()
+        except IntegrityError:
+            product = db.get(Product, asin)
+            if product is None:
+                raise
+    if product is not None and title and not product.title:
+        product.title = title
+
+
+@router.post("", response_model=ReviewTaskStatus, status_code=status.HTTP_202_ACCEPTED)
+def submit_review(payload: ReviewCreate, db: Session = Depends(get_db)) -> ReviewTaskStatus:
+    """Persist a review and its processing job atomically; return for polling.
+
+    The product shell is created first because ``reviews.product_asin`` is a
+    foreign key (the PRD's schema made every first review of an ASIN fail).
+    After commit a wake-up is sent to Redis; if Redis is down the job is still
+    durable and the outbox dispatcher delivers it later, so this never 503s.
+    """
+    asin = payload.product_asin
+    if asin:
+        _ensure_product_shell(db, asin, payload.product_title)
+
+    review = Review(
+        text=payload.text,
+        star_rating=payload.star_rating,
+        helpful_votes=payload.helpful_votes,
+        verified_purchase=payload.verified_purchase,
+        product_title=payload.product_title,
+        product_asin=asin,
+        status=ReviewStatus.PENDING.value,
+    )
+    db.add(review)
+    db.flush()  # assigns review.id
+    # Mark the job dispatched in the same commit, then publish. Updating the row after
+    # publishing would hold its lock while the worker (skip_locked) receives the message,
+    # so the worker would skip the job and it would wait for the outbox re-send.
+    job = Job(kind=JobKind.REVIEW.value, review_id=review.id, last_dispatched_at=datetime.now(UTC))
+    db.add(job)
+    db.commit()
+
+    try:
+        send_wakeup(job.id)
+    except Exception:
+        logger.warning("Broker unavailable for job %s; outbox will retry", job.id, exc_info=True)
+        with contextlib.suppress(Exception):
+            job.last_dispatched_at = None  # let the outbox re-send on its next poll
+            db.commit()
+
+    return ReviewTaskStatus(
+        review_id=review.id,
+        task_id=job.id,
+        status=ReviewStatus.PENDING.value,
+        message="Review queued for processing",
+    )
+
+
+@router.get("/{review_id}", response_model=ReviewResponse)
+def get_review(review_id: int, db: Session = Depends(get_db)) -> Review:
+    """Get a review and, when available, its pipeline output."""
+    return _get_review_or_404(db, review_id)
+
+
+@router.get("/{review_id}/status", response_model=ReviewTaskStatus)
+def get_review_status(review_id: int, db: Session = Depends(get_db)) -> ReviewTaskStatus:
+    """Lightweight lifecycle state for polling clients."""
+    review = _get_review_or_404(db, review_id)
+    job_id = db.query(Job.id).filter(Job.review_id == review.id).scalar()
+    return ReviewTaskStatus(
+        review_id=review.id,
+        task_id=job_id,
+        status=review.status,
+        message=STATUS_MESSAGES.get(review.status, "Unknown status"),
+    )

backend/app/worker/outbox.py

@@ -1,93 +1,112 @@
-"""Outbox dispatcher: turns unfinished `jobs` rows into Celery wake-ups.
-
-Runs as a daemon thread inside the worker (started on ``worker_ready`` in
-:mod:`app.worker.celery_app`), so a free-tier deployment needs no extra process.
-It can also run standalone::
-
-    cd backend && python -m app.worker.outbox
-
-A job is (re)sent when it is unfinished, due (``available_at <= now``), and was
-never dispatched or was dispatched more than :data:`REDISPATCH_AFTER` ago — this
-recovers messages lost to a Redis outage or a crashed worker. Importing this
-module never loads ML models.
-"""
-
-from __future__ import annotations
-
-import contextlib
-import logging
-import threading
-from datetime import UTC, datetime, timedelta
-from typing import Final
-
-from sqlalchemy import or_
-
-from app.database import SessionLocal
-from app.models.job import Job
-
-logger = logging.getLogger(__name__)
-
-POLL_INTERVAL_SECONDS: Final = 5.0
-REDISPATCH_AFTER: Final = timedelta(seconds=60)
-RUN_JOB_TASK: Final = "vera.run_job"
-
-
-def send_wakeup(job_id: str) -> None:
-    """Publish one wake-up message. Raises if the broker is unreachable."""
-    from app.worker.celery_app import celery_app  # lazy: avoid import cycle
-
-    celery_app.send_task(RUN_JOB_TASK, args=[job_id], retry=False)
-
-
-def dispatch_pending(limit: int = 100) -> int:
-    """Send wake-ups for due jobs. Returns the number sent."""
-    now = datetime.now(UTC)
-    sent = 0
-    with SessionLocal() as db:
-        jobs = (
-            db.query(Job)
-            .filter(
-                Job.finished_at.is_(None),
-                Job.available_at <= now,
-                or_(
-                    Job.last_dispatched_at.is_(None),
-                    Job.last_dispatched_at < now - REDISPATCH_AFTER,
-                ),
-            )
-            .order_by(Job.available_at, Job.id)
-            .with_for_update(skip_locked=True)
-            .limit(limit)
-            .all()
-        )
-        for job in jobs:
-            send_wakeup(job.id)
-            job.last_dispatched_at = now
-            sent += 1
-        db.commit()
-    return sent
-
-
-def run_forever(stop: threading.Event | None = None) -> None:
-    """Poll until ``stop`` is set. Errors are logged; the loop never dies."""
-    stop = stop or threading.Event()
-    while not stop.is_set():
-        try:
-            count = dispatch_pending()
-            if count:
-                logger.info("Outbox dispatched %d job(s)", count)
-        except Exception:
-            logger.exception("Outbox dispatch failed; will retry")
-        stop.wait(POLL_INTERVAL_SECONDS)
-
-
-def start_background_dispatcher() -> threading.Event:
-    """Start :func:`run_forever` in a daemon thread; set the returned event to stop it."""
-    stop = threading.Event()
-    threading.Thread(target=run_forever, args=(stop,), name="vera-outbox", daemon=True).start()
-    return stop
-
-
-if __name__ == "__main__":
-    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
-    with contextlib.suppress(KeyboardInterrupt):
-        run_forever()
+"""Outbox dispatcher: turns unfinished `jobs` rows into Celery wake-ups.
+
+Runs as a daemon thread inside the worker (started on ``worker_ready`` in
+:mod:`app.worker.celery_app`), so a free-tier deployment needs no extra process.
+It can also run standalone::
+
+    cd backend && python -m app.worker.outbox
+
+A job is (re)sent when it is unfinished, due (``available_at <= now``), and was
+never dispatched or was dispatched more than :data:`REDISPATCH_AFTER` ago — this
+recovers messages lost to a Redis outage or a crashed worker. Importing this
+module never loads ML models.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+import threading
+from datetime import UTC, datetime, timedelta
+from typing import Final
+
+from sqlalchemy import or_
+
+from app.database import SessionLocal
+from app.models.job import Job
+
+logger = logging.getLogger(__name__)
+
+POLL_INTERVAL_SECONDS: Final = 5.0
+REDISPATCH_AFTER: Final = timedelta(seconds=60)
+RUN_JOB_TASK: Final = "vera.run_job"
+
+
+def send_wakeup(job_id: str) -> None:
+    """Publish one wake-up message. Raises if the broker is unreachable."""
+    from app.worker.celery_app import celery_app  # lazy: avoid import cycle
+
+    celery_app.send_task(RUN_JOB_TASK, args=[job_id], retry=False)
+
+
+def dispatch_pending(limit: int = 100) -> int:
+    """Send wake-ups for due jobs. Returns the number sent.
+
+    Jobs are marked dispatched and committed BEFORE publishing, so no row lock is held when
+    the worker receives the message (the worker skips locked rows). Jobs whose publish fails
+    are un-marked so the next poll retries them.
+    """
+    now = datetime.now(UTC)
+    with SessionLocal() as db:
+        jobs = (
+            db.query(Job)
+            .filter(
+                Job.finished_at.is_(None),
+                Job.available_at <= now,
+                or_(
+                    Job.last_dispatched_at.is_(None),
+                    Job.last_dispatched_at < now - REDISPATCH_AFTER,
+                ),
+            )
+            .order_by(Job.available_at, Job.id)
+            .with_for_update(skip_locked=True)
+            .limit(limit)
+            .all()
+        )
+        job_ids = [job.id for job in jobs]
+        for job in jobs:
+            job.last_dispatched_at = now
+        db.commit()
+
+    sent = 0
+    failed: list[str] = []
+    for job_id in job_ids:
+        try:
+            send_wakeup(job_id)
+            sent += 1
+        except Exception:
+            failed.append(job_id)
+    if failed:
+        with SessionLocal() as db:
+            db.query(Job).filter(Job.id.in_(failed), Job.last_dispatched_at == now).update(
+                {Job.last_dispatched_at: None}, synchronize_session=False
+            )
+            db.commit()
+        raise RuntimeError(f"Broker unavailable; {len(failed)} wake-up(s) not sent")
+    return sent
+
+
+def run_forever(stop: threading.Event | None = None) -> None:
+    """Poll until ``stop`` is set. Errors are logged; the loop never dies."""
+    stop = stop or threading.Event()
+    while not stop.is_set():
+        try:
+            count = dispatch_pending()
+            if count:
+                logger.info("Outbox dispatched %d job(s)", count)
+        except Exception:
+            logger.exception("Outbox dispatch failed; will retry")
+        stop.wait(POLL_INTERVAL_SECONDS)
+
+
+def start_background_dispatcher() -> threading.Event:
+    """Start :func:`run_forever` in a daemon thread; set the returned event to stop it."""
+    stop = threading.Event()
+    threading.Thread(target=run_forever, args=(stop,), name="vera-outbox", daemon=True).start()
+    return stop
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+    with contextlib.suppress(KeyboardInterrupt):
+        run_forever()