fix(preprod): Prevent duplicate snapshot PR comments under fanout (#117467)

Re-derive the existing snapshot PR comment id under a
`select_for_update` lock that spans the GitHub call, so concurrent post
tasks for the same PR serialize and update the shared comment in place
instead of each creating its own.

Previously `post_snapshot_pr_comment_task` made the GitHub create/update
call with no lock, deciding create-vs-update from a comment id captured
earlier by `create_preprod_snapshot_pr_comment_task`. A snapshot PR
comment is a single PR-wide table with one row per artifact, so when
several artifacts on a commit finished near-simultaneously, each post
task saw no existing comment and created its own top-level comment. The
result was duplicate comments on the PR, one of which became a
permanently orphaned, never-updated copy once the surviving comment id
was persisted.

The fix moves the existing-comment-id resolution and the GitHub call
inside one transaction holding `select_for_update` locks on every
`CommitComparison` row for the PR. The first poster creates and persists
the comment id; later posters block on the lock, observe the persisted
id, and update. This mirrors the existing build-distribution path
(`create_preprod_pr_comment_task`), which already posts inside the lock.
The shared lock-and-resolve logic is extracted into
`lock_pr_comparisons_for_update`.

Holding the GitHub call inside the lock is deliberate: releasing the
lock before the call would reintroduce the race (two posters both
reading "no comment"). Lock hold time is bounded by the integration
client timeout (30s), which matches the task's processing deadline.

Areas worth careful review:
- The lock now spans a network round-trip to GitHub. This is required
for correctness and consistent with the build-distribution sibling, but
reviewers should confirm they're comfortable with the bounded lock-held
I/O.
- `existing_comment_id` is removed from
`post_snapshot_pr_comment_task`'s signature; in-flight queued messages
that still pass it are absorbed by `**kwargs` during deploy.

Author: NicoHinderling

src/sentry/preprod/vcs/pr_comments/snapshot_tasks.py

@@ -19,7 +19,10 @@
     format_solo_snapshot_pr_comment,
     format_waiting_for_base_snapshot_pr_comment,
 )
-from sentry.preprod.vcs.pr_comments.tasks import find_existing_comment_id, save_pr_comment_result
+from sentry.preprod.vcs.pr_comments.tasks import (
+    lock_pr_comparisons_for_update,
+    save_pr_comment_result,
+)
 from sentry.shared_integrations.exceptions import ApiError
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task
@@ -113,23 +116,14 @@ def create_preprod_snapshot_pr_comment_task(
     db_alias = router.db_for_write(CommitComparison)
 
     with transaction.atomic(db_alias):
-        all_for_pr = list(
-            CommitComparison.objects.select_for_update()
-            .filter(
-                organization_id=commit_comparison.organization_id,
-                head_repo_name=commit_comparison.head_repo_name,
-                pr_number=commit_comparison.pr_number,
-            )
-            .order_by("id")
+        cc, existing_comment_id = lock_pr_comparisons_for_update(
+            organization_id=commit_comparison.organization_id,
+            head_repo_name=commit_comparison.head_repo_name,
+            pr_number=commit_comparison.pr_number,
+            target_id=commit_comparison.id,
+            comment_type="snapshots",
         )
 
-        try:
-            cc = next(c for c in all_for_pr if c.id == commit_comparison.id)
-        except StopIteration:
-            raise CommitComparison.DoesNotExist(
-                f"CommitComparison {commit_comparison.id} was deleted before lock acquisition"
-            )
-
         all_artifacts = list(artifact.get_sibling_artifacts_for_commit())
 
         artifact_ids = [a.id for a in all_artifacts]
@@ -165,7 +159,6 @@ def create_preprod_snapshot_pr_comment_task(
 
         is_solo = not base_artifact_map
 
-        existing_comment_id = find_existing_comment_id(all_for_pr, "snapshots")
         cc_id = cc.id
 
         if is_solo:
@@ -241,7 +234,6 @@ def create_preprod_snapshot_pr_comment_task(
         commit_comparison_id=cc_id,
         artifact_id=artifact.id,
         comment_body=comment_body,
-        existing_comment_id=existing_comment_id,
     )
 
 
@@ -261,7 +253,6 @@ def post_snapshot_pr_comment_task(
     commit_comparison_id: int,
     artifact_id: int | None = None,
     comment_body: str,
-    existing_comment_id: str | None,
     **kwargs: Any,
 ) -> None:
     try:
@@ -283,38 +274,53 @@ def post_snapshot_pr_comment_task(
 
     comment_id: str | None = None
     api_error: Exception | None = None
+    db_alias = router.db_for_write(CommitComparison)
 
     try:
-        if existing_comment_id:
-            client.update_comment(
-                repo=repo_name,
-                issue_id=str(pr_number),
-                comment_id=str(existing_comment_id),
-                data={"body": comment_body},
-            )
-            comment_id = existing_comment_id
-        else:
-            resp = client.create_comment(
-                repo=repo_name,
-                issue_id=str(pr_number),
-                data={"body": comment_body},
+        # The comment_id is re-derived under the lock instead of trusting the
+        # value passed from the create task: when several artifacts on a commit
+        # post at once, each create task reads no existing comment, so the
+        # first post here would otherwise create a duplicate comment instead of
+        # updating the shared one. The GitHub call is held inside the lock (as
+        # in create_preprod_pr_comment_task) so concurrent posters serialize on
+        # the decision; lock hold is bounded by the client timeout, which
+        # matches this task's processing deadline.
+        with transaction.atomic(db_alias):
+            cc, comment_id = lock_pr_comparisons_for_update(
+                organization_id=organization.id,
+                head_repo_name=repo_name,
+                pr_number=pr_number,
+                target_id=commit_comparison_id,
+                comment_type="snapshots",
             )
-            comment_id = str(resp["id"])
-    except Exception as e:
-        extra: dict[str, Any] = {
-            "commit_comparison_id": commit_comparison_id,
-            "organization_id": organization_id,
-            "error_type": type(e).__name__,
-        }
-        if isinstance(e, ApiError):
-            extra["status_code"] = e.code
-        logger.exception("preprod.snapshot_pr_comments.post.failed", extra=extra)
-        api_error = e
+            is_update = comment_id is not None
+
+            try:
+                if comment_id:
+                    client.update_comment(
+                        repo=repo_name,
+                        issue_id=str(pr_number),
+                        comment_id=str(comment_id),
+                        data={"body": comment_body},
+                    )
+                else:
+                    resp = client.create_comment(
+                        repo=repo_name,
+                        issue_id=str(pr_number),
+                        data={"body": comment_body},
+                    )
+                    comment_id = str(resp["id"])
+            except Exception as e:
+                extra: dict[str, Any] = {
+                    "commit_comparison_id": commit_comparison_id,
+                    "organization_id": organization_id,
+                    "error_type": type(e).__name__,
+                }
+                if isinstance(e, ApiError):
+                    extra["status_code"] = e.code
+                logger.exception("preprod.snapshot_pr_comments.post.failed", extra=extra)
+                api_error = e
 
-    db_alias = router.db_for_write(CommitComparison)
-    try:
-        with transaction.atomic(db_alias):
-            cc = CommitComparison.objects.select_for_update().get(id=commit_comparison_id)
             if api_error is not None:
                 save_pr_comment_result(cc, "snapshots", success=False, error=api_error)
             else:
@@ -328,7 +334,7 @@ def post_snapshot_pr_comment_task(
                         "comment_id": comment_id,
                         "repo_name": repo_name,
                         "pr_number": pr_number,
-                        "is_update": existing_comment_id is not None,
+                        "is_update": is_update,
                     },
                 )
     except CommitComparison.DoesNotExist:
@@ -338,6 +344,9 @@ def post_snapshot_pr_comment_task(
         )
         return
 
+    # Re-raised outside the transaction so the failure record is committed
+    # before the retry fires. Terminal 4xx (except 429) are swallowed; 429,
+    # 5xx, and network errors re-raise to trigger the task's retry policy.
     if api_error is not None:
         if (
             isinstance(api_error, ApiError)

src/sentry/preprod/vcs/pr_comments/tasks.py

@@ -194,6 +194,39 @@ def find_existing_comment_id(
     return None
 
 
+def lock_pr_comparisons_for_update(
+    *,
+    organization_id: int,
+    head_repo_name: str,
+    pr_number: int,
+    target_id: int,
+    comment_type: str,
+) -> tuple[CommitComparison, str | None]:
+    """Lock every CommitComparison row for the PR and return the target row
+    along with the existing comment id for ``comment_type``.
+
+    Must be called inside an open transaction. The ``SELECT ... FOR UPDATE``
+    serializes tasks operating on the same PR (ordered by id for a stable lock
+    order), so the comment id is read from a consistent committed view rather
+    than a value captured earlier by another task.
+    """
+    all_for_pr = list(
+        CommitComparison.objects.select_for_update()
+        .filter(
+            organization_id=organization_id,
+            head_repo_name=head_repo_name,
+            pr_number=pr_number,
+        )
+        .order_by("id")
+    )
+    cc = next((c for c in all_for_pr if c.id == target_id), None)
+    if cc is None:
+        raise CommitComparison.DoesNotExist(
+            f"CommitComparison {target_id} was deleted before lock acquisition"
+        )
+    return cc, find_existing_comment_id(all_for_pr, comment_type)
+
+
 def save_pr_comment_result(
     commit_comparison: CommitComparison,
     comment_type: str,