fix: absorb concurrent-writer insert races on unique-key models

qb_site/AGENTS.md

@@ -69,6 +69,49 @@ uv run python qb_site/manage.py test zulip_bot
   - or ask the user to run `scripts/repo_check_compose.sh` and share results.
 - When reporting verification, explicitly state what was and was not runnable.
 
+## Concurrent Writers and Unique Keys
+Celery workers overlap: per-PR tasks (`syncer.sync_pr`, `analyzer.process_pr`), periodic
+sweeps, admin actions, and management commands can all write the same rows at the same
+time. Sweeps even *prefer* recently-changed PRs, so they collide with the per-PR task for
+exactly the same row. Never assume your code path is the only writer.
+
+**Rule: for any model with a unique constraint, "check if it exists, then create" is a
+bug.** Between the check and the insert, another worker can create the row, and the
+resulting `IntegrityError` poisons the enclosing transaction (and, in a sweep, kills the
+whole run). This applies equally to `filter(...).first()` + `create(...)` and to
+snapshot-diff loops that collect `to_create` lists for `bulk_create`.
+`select_for_update()` does not help here — it cannot lock a row that does not exist yet.
+
+Use one of these instead (all are in the codebase as reference patterns):
+
+- `get_or_create` / `update_or_create` with lookup kwargs that **exactly cover the unique
+  constraint** (extra values go in `defaults`; constraint fields must not be in
+  `defaults`). Django retries the get under a savepoint on conflict, so this is
+  race-safe. For case-insensitive keys, put the `__iexact` lookup in kwargs and the
+  concrete value in `defaults` (see `syncer/services/sub/labels_sync.py`).
+- `bulk_create(..., ignore_conflicts=True)` for insert-only paths where losing the race
+  needs no follow-up (see `syncer/services/sub/labels_sync.py` PRLabel attach).
+- `bulk_create(..., update_conflicts=True, unique_fields=[...], update_fields=[...])`
+  for upserts where both writers compute the row from the same source data and
+  last-writer-wins is correct (see `analyzer/services/queue_windows.py` and
+  `analyzer/services/queue_window_build_state.py`).
+- Savepoint + catch `IntegrityError` + re-fetch and fall through to the update path,
+  when you need the created-vs-updated distinction or custom conflict handling (see
+  `syncer/services/sub/ci_sync.py:_archive_mode_upsert`,
+  `syncer/services/sub/core_entities_sync.py:upsert_user_from_github`). The
+  `with transaction.atomic():` around the `create()` is mandatory — without the
+  savepoint, catching the error leaves the outer transaction unusable.
+
+Additional expectations:
+- Wrap multi-statement rebuilds (delete + create + update of a derived row set) in
+  `transaction.atomic()` so readers and crashes never observe a half-written state.
+- Sweep tasks must contain per-item failures (`except IntegrityError: log, count,
+  continue`) so one conflicted row cannot abort the rest of the run — see
+  `analyzer/tasks/rebuild_queue_windows_sweep.py` and the `prs_conflict_skipped`
+  counter it reports.
+- When adding a `UniqueConstraint` to an existing model, audit every write site for the
+  patterns above in the same change.
+
 ## Migration and DB Notes
 - Generate migration files on the host and commit them; do not create migrations from inside containers.
 - Running `makemigrations` on host may emit a Postgres connection warning when DB is not running; file generation still works.

qb_site/analyzer/AGENTS.md

@@ -80,3 +80,8 @@ Keep tasks idempotent and resumable; prefer explicit summary payloads to aid adm
 ## Operational Notes
 - Large sweeps can contend with sync tasks on shared worker capacity; tune per-repo limits before broadening cadence.
 - Keep admin/object-tool commands available for targeted per-PR recovery paths.
+- Sweeps and `analyzer.process_pr` rebuild the same derived rows concurrently (sweeps
+  preferentially pick freshly-updated PRs — the same ones process_pr is handling).
+  Follow "Concurrent Writers and Unique Keys" in `qb_site/AGENTS.md`: upsert instead of
+  check-then-create, wrap rebuilds in `transaction.atomic()`, and contain per-PR
+  `IntegrityError` in sweeps so one conflict cannot abort the run.

qb_site/analyzer/services/dependencies.py

@@ -64,15 +64,21 @@ def rebuild_pr_dependencies(pr: PullRequest) -> DependencyRebuildResult:
         target_pr = targets.get(number)
         key = (pr.repository_id, number)
         dep = existing.get(key)
+        created_now = False
         if dep is None:
-            PRDependency.objects.create(
+            # The per-PR dependency task and the dependencies sweep can rebuild the
+            # same PR concurrently; get_or_create absorbs losing the insert race on
+            # the (pull_request, depends_on_repository, depends_on_number) key and
+            # falls through to the update path with the winner's row.
+            dep, created_now = PRDependency.objects.get_or_create(
                 pull_request=pr,
                 depends_on_repository=pr.repository,
                 depends_on_number=number,
-                depends_on_pull_request=target_pr,
+                defaults={"depends_on_pull_request": target_pr},
             )
-            created += 1
-        else:
+            if created_now:
+                created += 1
+        if not created_now:
             desired_target_id = target_pr.id if target_pr else None
             if dep.depends_on_pull_request_id != desired_target_id:
                 dep.depends_on_pull_request = target_pr

qb_site/analyzer/services/queue_window_build_state.py

@@ -75,7 +75,16 @@ def record_queue_window_build_states(
             to_update.append(row)
 
     if to_create:
-        PRQueueWindowBuildState.objects.bulk_create(to_create, batch_size=200)
+        # analyzer.process_pr and the queue-window sweep both record build state
+        # for the same (PR, ruleset) pairs; upsert so losing the insert race
+        # converges on the other writer's row instead of raising IntegrityError.
+        PRQueueWindowBuildState.objects.bulk_create(
+            to_create,
+            batch_size=200,
+            update_conflicts=True,
+            unique_fields=["pull_request", "rule_set"],
+            update_fields=["revision_version_built", "windows_built_at", "last_status", "last_reason", "updated_at"],
+        )
     if to_update:
         PRQueueWindowBuildState.objects.bulk_update(
             to_update,

qb_site/analyzer/services/queue_windows.py

@@ -4,7 +4,7 @@
 from datetime import datetime
 from typing import Any, Iterable, List, Optional, Tuple
 
-from django.db import models
+from django.db import models, transaction
 from django.utils import timezone
 
 from core.models import Repository
@@ -954,6 +954,29 @@ def _attribution_kwargs(attr: Optional[WindowAttribution], prefix: str) -> dict[
     }
 
 
+# Every PRQueueWindow field the rebuild recomputes. Shared by the bulk_create
+# upsert (conflict update_fields) and bulk_update so the two write paths cannot
+# drift apart.
+QUEUE_WINDOW_MUTABLE_FIELDS = [
+    "to_ts",
+    "cycle_index",
+    "duration_seconds_closed",
+    "cumulative_seconds_closed",
+    "window_count",
+    "first_on_queue_ts",
+    "opened_by_event_type",
+    "opened_by_timeline_event",
+    "opened_by_check_run",
+    "opened_by_status_context",
+    "opened_at_head_sha",
+    "closed_by_event_type",
+    "closed_by_timeline_event",
+    "closed_by_check_run",
+    "closed_by_status_context",
+    "closed_at_head_sha",
+]
+
+
 def rebuild_queue_windows_for_ruleset(
     *,
     pr: PullRequest,
@@ -1015,107 +1038,103 @@ def rebuild_queue_windows_for_ruleset(
     first_on_queue_ts = windows[0].from_ts if windows else None
     cumulative_seconds = 0
 
-    existing_by_start = {obj.from_ts: obj for obj in PRQueueWindow.objects.filter(pull_request=pr, rule_set=rule_set)}
-    to_create: list[PRQueueWindow] = []
-    to_update: list[PRQueueWindow] = []
-
-    for cycle_index, w in enumerate(windows):
-        start, end = w.from_ts, w.to_ts
-        duration_seconds = 0
-        if end is not None:
-            duration_seconds = int((end - start).total_seconds())
-        cumulative_seconds += duration_seconds
-        opened_kwargs = _attribution_kwargs(w.opened_by, "opened_by")
-        closed_kwargs = _attribution_kwargs(w.closed_by, "closed_by")
-
-        obj = existing_by_start.pop(start, None)
-        if obj is None:
-            to_create.append(
-                PRQueueWindow(
-                    pull_request=pr,
-                    rule_set=rule_set,
-                    from_ts=start,
-                    to_ts=end,
-                    cycle_index=cycle_index,
-                    duration_seconds_closed=duration_seconds,
-                    cumulative_seconds_closed=cumulative_seconds,
-                    window_count=window_count,
-                    first_on_queue_ts=first_on_queue_ts,
-                    **opened_kwargs,
-                    **closed_kwargs,
+    # Concurrent writers exist for the same (PR, ruleset) windows: analyzer.process_pr
+    # rebuilds unconditionally after every sync while the staleness sweep independently
+    # picks freshly-updated PRs, plus admin/management-command rebuilds. The atomic
+    # block keeps the create/update/delete trio consistent for readers, and the
+    # bulk_create upsert (update_conflicts) makes losing an insert race converge on the
+    # other writer's row instead of raising IntegrityError on prqwin_pr_ruleset_from_unique.
+    with transaction.atomic():
+        existing_by_start = {obj.from_ts: obj for obj in PRQueueWindow.objects.filter(pull_request=pr, rule_set=rule_set)}
+        to_create: list[PRQueueWindow] = []
+        to_update: list[PRQueueWindow] = []
+
+        for cycle_index, w in enumerate(windows):
+            start, end = w.from_ts, w.to_ts
+            duration_seconds = 0
+            if end is not None:
+                duration_seconds = int((end - start).total_seconds())
+            cumulative_seconds += duration_seconds
+            opened_kwargs = _attribution_kwargs(w.opened_by, "opened_by")
+            closed_kwargs = _attribution_kwargs(w.closed_by, "closed_by")
+
+            obj = existing_by_start.pop(start, None)
+            if obj is None:
+                to_create.append(
+                    PRQueueWindow(
+                        pull_request=pr,
+                        rule_set=rule_set,
+                        from_ts=start,
+                        to_ts=end,
+                        cycle_index=cycle_index,
+                        duration_seconds_closed=duration_seconds,
+                        cumulative_seconds_closed=cumulative_seconds,
+                        window_count=window_count,
+                        first_on_queue_ts=first_on_queue_ts,
+                        **opened_kwargs,
+                        **closed_kwargs,
+                    )
                 )
-            )
-            continue
+                continue
 
-        changed = False
-        if obj.to_ts != end:
-            obj.to_ts = end
-            changed = True
-        if obj.cycle_index != cycle_index:
-            obj.cycle_index = cycle_index
-            changed = True
-        if obj.duration_seconds_closed != duration_seconds:
-            obj.duration_seconds_closed = duration_seconds
-            changed = True
-        if obj.cumulative_seconds_closed != cumulative_seconds:
-            obj.cumulative_seconds_closed = cumulative_seconds
-            changed = True
-        if obj.window_count != window_count:
-            obj.window_count = window_count
-            changed = True
-        if obj.first_on_queue_ts != first_on_queue_ts:
-            obj.first_on_queue_ts = first_on_queue_ts
-            changed = True
-        # Attribution fields: always overwrite since they reflect the current
-        # computation. Compare by FK id to avoid unnecessary updates.
-        for field, val in {**opened_kwargs, **closed_kwargs}.items():
-            if field.endswith("_id"):
-                continue  # skipped; we set the object field, not the _id field
-            cur = getattr(obj, field)
-            # For FK fields Django exposes both .field (object) and .field_id (int).
-            # Compare by id to avoid fetching the related object.
-            if hasattr(obj, f"{field}_id"):
-                cur_id = getattr(obj, f"{field}_id")
-                new_id = val.pk if val is not None else None
-                if cur_id != new_id:
-                    setattr(obj, field, val)
-                    changed = True
-            else:
-                if cur != val:
-                    setattr(obj, field, val)
-                    changed = True
-        if changed:
-            to_update.append(obj)
-
-    if to_create:
-        PRQueueWindow.objects.bulk_create(to_create, batch_size=200)
-    if to_update:
-        PRQueueWindow.objects.bulk_update(
-            to_update,
-            [
-                "to_ts",
-                "cycle_index",
-                "duration_seconds_closed",
-                "cumulative_seconds_closed",
-                "window_count",
-                "first_on_queue_ts",
-                "opened_by_event_type",
-                "opened_by_timeline_event",
-                "opened_by_check_run",
-                "opened_by_status_context",
-                "opened_at_head_sha",
-                "closed_by_event_type",
-                "closed_by_timeline_event",
-                "closed_by_check_run",
-                "closed_by_status_context",
-                "closed_at_head_sha",
-            ],
-            batch_size=200,
-        )
+            changed = False
+            if obj.to_ts != end:
+                obj.to_ts = end
+                changed = True
+            if obj.cycle_index != cycle_index:
+                obj.cycle_index = cycle_index
+                changed = True
+            if obj.duration_seconds_closed != duration_seconds:
+                obj.duration_seconds_closed = duration_seconds
+                changed = True
+            if obj.cumulative_seconds_closed != cumulative_seconds:
+                obj.cumulative_seconds_closed = cumulative_seconds
+                changed = True
+            if obj.window_count != window_count:
+                obj.window_count = window_count
+                changed = True
+            if obj.first_on_queue_ts != first_on_queue_ts:
+                obj.first_on_queue_ts = first_on_queue_ts
+                changed = True
+            # Attribution fields: always overwrite since they reflect the current
+            # computation. Compare by FK id to avoid unnecessary updates.
+            for field, val in {**opened_kwargs, **closed_kwargs}.items():
+                if field.endswith("_id"):
+                    continue  # skipped; we set the object field, not the _id field
+                cur = getattr(obj, field)
+                # For FK fields Django exposes both .field (object) and .field_id (int).
+                # Compare by id to avoid fetching the related object.
+                if hasattr(obj, f"{field}_id"):
+                    cur_id = getattr(obj, f"{field}_id")
+                    new_id = val.pk if val is not None else None
+                    if cur_id != new_id:
+                        setattr(obj, field, val)
+                        changed = True
+                else:
+                    if cur != val:
+                        setattr(obj, field, val)
+                        changed = True
+            if changed:
+                to_update.append(obj)
+
+        if to_create:
+            PRQueueWindow.objects.bulk_create(
+                to_create,
+                batch_size=200,
+                update_conflicts=True,
+                unique_fields=["pull_request", "rule_set", "from_ts"],
+                update_fields=QUEUE_WINDOW_MUTABLE_FIELDS,
+            )
+        if to_update:
+            PRQueueWindow.objects.bulk_update(
+                to_update,
+                QUEUE_WINDOW_MUTABLE_FIELDS,
+                batch_size=200,
+            )
 
-    deleted = 0
-    if existing_by_start:
-        deleted, _ = PRQueueWindow.objects.filter(id__in=[obj.id for obj in existing_by_start.values()]).delete()
+        deleted = 0
+        if existing_by_start:
+            deleted, _ = PRQueueWindow.objects.filter(id__in=[obj.id for obj in existing_by_start.values()]).delete()
 
     return QueueWindowRebuildResult(
         created=len(to_create),

qb_site/analyzer/services/reviewer_opt_out_backfill.py

@@ -104,23 +104,20 @@ def backfill_reviewer_opt_outs(
                     },
                 )
             else:
-                obj = ReviewerOptOut.objects.filter(
+                # This backfill can run while live syncs upsert the same
+                # (repository, pr_number, reviewer_login) row; get_or_create
+                # absorbs losing the insert race and updates the winner's row.
+                obj, was_created = ReviewerOptOut.objects.get_or_create(
                     repository_id=pr.repository_id,
                     pr_number=pr.number,
                     reviewer_login=login,
-                ).first()
-                if obj is None:
-                    obj = ReviewerOptOut.objects.create(
-                        repository_id=pr.repository_id,
-                        pr_number=pr.number,
-                        reviewer_login=login,
-                        active=False,
-                        opted_out_at=occurred_at,
-                        cleared_at=occurred_at,
-                    )
-                    was_created = True
-                else:
-                    was_created = False
+                    defaults={
+                        "active": False,
+                        "opted_out_at": occurred_at,
+                        "cleared_at": occurred_at,
+                    },
+                )
+                if not was_created:
                     if obj.active or obj.cleared_at != occurred_at:
                         obj.active = False
                         obj.cleared_at = occurred_at