Delay initial processing and retry via Cloud Tasks when the revision is not yet public

Author: suhaibmujahid

services/reviewhelper-api/app/review_processor.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import AsyncIterator
+from datetime import UTC, datetime, timedelta
 from functools import cache
 from typing import Collection, Iterable
 
@@ -13,11 +14,20 @@
 
 logger = logging.getLogger(__name__)
 
+VISIBILITY_TIMEOUT = timedelta(minutes=5)
+
 
 class ReviewProcessingError(Exception):
     """Custom exception for permanent errors during review processing that should not trigger retries."""
 
 
+class RevisionNotYetPublicError(Exception):
+    """The revision is not yet public but was recently created.
+
+    This is a transient error — Cloud Tasks should retry after backoff.
+    """
+
+
 @cache
 def get_code_review_tool():
     from bugbug.tools.code_review import CodeReviewTool
@@ -50,6 +60,13 @@ async def process_review(
         raise ValueError(f"Unsupported platform: {review_request.platform}")
 
     if not patch.is_accessible() or not patch.is_public():
+        age = datetime.now(UTC) - review_request.created_at
+        if age < VISIBILITY_TIMEOUT:
+            raise RevisionNotYetPublicError(
+                f"Revision D{review_request.revision_id} is not public. "
+                "But the review request was created recently, so this may be a visibility delay."
+            )
+
         raise ReviewProcessingError(
             "Unable to access the revision. This may be because "
             "the revision is private or has restricted visibility."

services/reviewhelper-api/app/routers/internal.py

@@ -13,6 +13,7 @@
 from app.enums import ReviewStatus
 from app.review_processor import (
     ReviewProcessingError,
+    RevisionNotYetPublicError,
     process_review,
     submit_review_to_platform,
 )
@@ -131,6 +132,11 @@ async def _set_retry_pending(db: AsyncSession):
                 # the connection.
                 db.add(review_request)
 
+        except RevisionNotYetPublicError as e:
+            logger.warning("Review request %s: %s", review_request_id, e)
+            review_request.status = ReviewStatus.RETRY_PENDING
+            await db.commit()
+            return Response(status_code=status.HTTP_409_CONFLICT)
         except ReviewProcessingError as e:
             review_request.error = str(e)
             review_request.status = ReviewStatus.FAILED

services/reviewhelper-api/app/routers/request.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import timedelta
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, status
@@ -15,6 +16,12 @@
 
 logger = logging.getLogger(__name__)
 
+# For newly-created revisions, delay the task to give phab-bot time to
+# update visibility before we attempt processing. If the delay has
+# already elapsed, Cloud Tasks dispatches immediately.
+INITIAL_TASK_DELAY = timedelta(seconds=30)
+
+
 router = APIRouter(
     tags=["request"],
     dependencies=[Depends(verify_external_api_key)],
@@ -75,8 +82,17 @@ async def create_or_get_review_request(
     db.add(review_request)
     await db.commit()
 
+    schedule_time = (
+        request.revision_created_at + INITIAL_TASK_DELAY
+        # NOTE: This is a check if for backwards compatibility with older
+        # version of Phabricator, and can be dropped once
+        # https://github.com/mozilla-conduit/phabricator/pull/84 is deployed
+        if request.revision_created_at is not None
+        else None
+    )
+
     # Queue task for processing
-    await create_review_task(review_request.id)
+    await create_review_task(review_request.id, schedule_time=schedule_time)
 
     return JSONResponse(
         {

services/reviewhelper-api/app/schemas/review_request.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, PastDatetime
 
 from app.enums import ReviewStatus
 from app.schemas.base import UserActionBase
@@ -9,6 +9,7 @@ class ReviewRequestCreate(UserActionBase):
 
     revision_id: int
     diff_id: int
+    revision_created_at: PastDatetime | None = None
 
 
 class ReviewRequestResponse(BaseModel):

services/reviewhelper-api/app/tasks.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import datetime
 from functools import cache
 
 from google.cloud.tasks_v2 import (
@@ -19,11 +20,14 @@ def _get_tasks_client():
     return CloudTasksAsyncClient()
 
 
-async def create_review_task(review_request_id: int):
+async def create_review_task(
+    review_request_id: int, schedule_time: datetime | None = None
+):
     """Create a Cloud Task to process a review request.
 
     Args:
         review_request_id: The ID of the review request to process.
+        schedule_time: Optional time to delay task dispatch until.
     """
     client = _get_tasks_client()
 
@@ -44,6 +48,7 @@ async def create_review_task(review_request_id: int):
             },
         ),
         dispatch_deadline=Duration(seconds=30 * 60),
+        schedule_time=schedule_time,
     )
 
     response = await client.create_task(parent=parent, task=task)