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,15 @@ async def process_review(
         raise ValueError(f"Unsupported platform: {review_request.platform}")
 
     if not patch.is_accessible() or not patch.is_public():
+        if (
+            review_request.created_at
+            and datetime.now(UTC) - review_request.created_at < VISIBILITY_TIMEOUT
+        ):
+            raise RevisionNotYetPublicError(
+                f"Revision D{review_request.revision_id} is not yet public. "
+                f"Waiting for phab-bot to update visibility."
+            )
+
         raise ReviewProcessingError(
             "Unable to access the revision. This may be because "
             "the revision is private or has restricted visibility."

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,14 @@ async def create_or_get_review_request(
     db.add(review_request)
     await db.commit()
 
+    schedule_time = (
+        request.revision_created_at + INITIAL_TASK_DELAY
+        if request.revision_created_at
+        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,3 +1,5 @@
+from datetime import datetime
+
 from pydantic import BaseModel
 
 from app.enums import ReviewStatus
@@ -9,6 +11,7 @@ class ReviewRequestCreate(UserActionBase):
 
     revision_id: int
     diff_id: int
+    revision_created_at: datetime | 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)