diff --git a/qb_site/AGENTS.md b/qb_site/AGENTS.md index 550e984d..5c93cab4 100644 --- a/qb_site/AGENTS.md +++ b/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. diff --git a/qb_site/analyzer/AGENTS.md b/qb_site/analyzer/AGENTS.md index a43f3853..e1267ce8 100644 --- a/qb_site/analyzer/AGENTS.md +++ b/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. diff --git a/qb_site/analyzer/services/dependencies.py b/qb_site/analyzer/services/dependencies.py index 14a20f35..849d82d1 100644 --- a/qb_site/analyzer/services/dependencies.py +++ b/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 diff --git a/qb_site/analyzer/services/queue_window_build_state.py b/qb_site/analyzer/services/queue_window_build_state.py index 0186d00a..c13d7efc 100644 --- a/qb_site/analyzer/services/queue_window_build_state.py +++ b/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, diff --git a/qb_site/analyzer/services/queue_windows.py b/qb_site/analyzer/services/queue_windows.py index 545fe3de..a1c0f5aa 100644 --- a/qb_site/analyzer/services/queue_windows.py +++ b/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), diff --git a/qb_site/analyzer/services/reviewer_opt_out_backfill.py b/qb_site/analyzer/services/reviewer_opt_out_backfill.py index 670e0f21..cab3a240 100644 --- a/qb_site/analyzer/services/reviewer_opt_out_backfill.py +++ b/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 diff --git a/qb_site/analyzer/tasks/rebuild_queue_windows_sweep.py b/qb_site/analyzer/tasks/rebuild_queue_windows_sweep.py index 938b9edc..1a20ab82 100644 --- a/qb_site/analyzer/tasks/rebuild_queue_windows_sweep.py +++ b/qb_site/analyzer/tasks/rebuild_queue_windows_sweep.py @@ -1,7 +1,10 @@ from __future__ import annotations +import logging + from celery import shared_task from datetime import datetime +from django.db import IntegrityError from django.utils import timezone from django.db.models import Count, Exists, F, Min, OuterRef, Q @@ -12,6 +15,8 @@ from core.models import Repository from syncer.models import PullRequest +logger = logging.getLogger(__name__) + def _is_ruleset_stale_for_pr( *, @@ -106,6 +111,7 @@ def rebuild_queue_windows_sweep_task( repos = list(Repository.objects.filter(is_active=True).only("id", "owner", "name")) total_rebuilt = 0 total_prs = 0 + total_prs_conflict_skipped = 0 total_prs_skipped_up_to_date = 0 total_rulesets_skipped_out_of_bounds = 0 total_prs_stale_ruleset = 0 @@ -243,7 +249,7 @@ def rebuild_queue_windows_sweep_task( continue def _process_batch(pr_batch: list[PullRequest]) -> None: - nonlocal repo_limit_hit, repo_rebuilt, repo_prs, total_prs + nonlocal repo_limit_hit, repo_rebuilt, repo_prs, total_prs, total_prs_conflict_skipped if not pr_batch: return @@ -328,7 +334,23 @@ def _process_batch(pr_batch: list[PullRequest]) -> None: repo_prs_skipped_up_to_date.append(pr_num) continue - summary = rebuild_queue_windows_for_pr(pr=pr, rule_sets=stale_rule_sets) + try: + summary = rebuild_queue_windows_for_pr(pr=pr, rule_sets=stale_rule_sets) + except IntegrityError: + # Residual write race with a concurrent per-PR rebuild + # (analyzer.process_pr / admin action). The PR's build state + # was not recorded, so it stays stale and the next sweep + # retries it; one conflicted PR must not abort the run. + logger.warning( + "rebuild_queue_windows_sweep: integrity conflict on pr_id=%s (#%s); skipping this run", + pr.id, + pr.number, + exc_info=True, + ) + total_prs_conflict_skipped += 1 + repo_prs += 1 + total_prs += 1 + continue per_ruleset = summary.get("per_ruleset", {}) or {} pr_num = int(pr.number) if any( @@ -397,6 +419,7 @@ def _process_batch(pr_batch: list[PullRequest]) -> None: "prs_checked": total_prs, "prs_checked_numbers": processed_pr_numbers, "windows_rebuilt": total_rebuilt, + "prs_conflict_skipped": total_prs_conflict_skipped, "prs_skipped_up_to_date": total_prs_skipped_up_to_date, "rulesets_skipped_out_of_bounds": total_rulesets_skipped_out_of_bounds, "prs_stale_ruleset": total_prs_stale_ruleset, diff --git a/qb_site/analyzer/tests/services/test_concurrent_write_races.py b/qb_site/analyzer/tests/services/test_concurrent_write_races.py new file mode 100644 index 00000000..720793e1 --- /dev/null +++ b/qb_site/analyzer/tests/services/test_concurrent_write_races.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +from datetime import datetime, timezone as dt_timezone +from unittest import mock + +from django.test import TestCase + +from django.db.models.query import QuerySet + +from analyzer.models import PRDependency, PRQueueWindow, PRQueueWindowBuildState, QueueRuleSet, ReviewerOptOut +from analyzer.services.dependencies import rebuild_pr_dependencies +from analyzer.services.queue_window_build_state import record_queue_window_build_states +from analyzer.services.queue_windows import rebuild_queue_windows_for_ruleset +from analyzer.services.reviewer_opt_out_backfill import backfill_reviewer_opt_outs +from core.models import Repository +from syncer.models import PRTimelineEvent, PRTimelineEventType, PullRequest + + +def flaky_queryset_get(): + """Patch target for QuerySet.get that misses exactly once, then delegates. + + Simulates the get_or_create race: the initial get sees no row (the winner has + not committed yet), the create then conflicts with the winner's committed row, + and Django's conflict-retry get must recover it. That retry only succeeds when + the lookup kwargs cover the unique constraint — which is the property these + tests pin. + """ + real_get = QuerySet.get + state = {"missed": False} + + def flaky(self, *args, **kwargs): + if not state["missed"]: + state["missed"] = True + raise self.model.DoesNotExist + return real_get(self, *args, **kwargs) + + return flaky + + +def _dt(year: int, month: int, day: int) -> datetime: + return datetime(year, month, day, tzinfo=dt_timezone.utc) + + +class TestConcurrentWriteRaces(TestCase): + """Losing an insert race against a concurrent rebuild must upsert, not raise. + + analyzer.process_pr and the queue-window sweep can rebuild the same PR at the + same time: both snapshot the existing rows, both compute the same new row, and + the loser's bulk_create used to raise IntegrityError (prqwin_pr_ruleset_from_unique) + and abort the whole sweep run. These tests simulate the loser's view by making + the snapshot read return empty while the "winner's" row already exists. + """ + + def setUp(self) -> None: + self.repo = Repository.objects.create(owner="o", name="r", default_branch="master", is_active=True) + self.rules = QueueRuleSet.objects.create( + repository=self.repo, + version=1, + require_open=True, + require_not_draft=True, + require_ci_success=False, + required_label_names=[], + forbidden_label_names=["blocked-by-other-pr"], + ) + self.pr = PullRequest.objects.create( + repository=self.repo, + number=1, + author=None, + state="open", + is_draft=False, + gh_created_at=_dt(2024, 9, 1), + gh_updated_at=_dt(2024, 9, 2), + base_ref_name="master", + head_ref_name="branch", + head_repo_owner_login="o", + head_repo_name="r", + title="t", + body="", + additions=0, + deletions=0, + changed_files_count=0, + timeline_backfill_done=True, + ) + PRTimelineEvent.objects.create( + pull_request=self.pr, + type=PRTimelineEventType.LABELED, + occurred_at=_dt(2024, 9, 1), + label_name="blocked-by-other-PR", + ) + PRTimelineEvent.objects.create( + pull_request=self.pr, + type=PRTimelineEventType.UNLABELED, + occurred_at=_dt(2024, 9, 6), + label_name="blocked-by-other-PR", + ) + + def _empty_first_filter(self, manager): + """Return a filter() side-effect whose first call sees no rows (the loser's + stale snapshot) and delegates to the real manager afterwards.""" + real_filter = manager.filter + state = {"first": True} + + def fake_filter(*args, **kwargs): + if state["first"]: + state["first"] = False + return manager.none() + return real_filter(*args, **kwargs) + + return fake_filter + + def test_queue_window_rebuild_upserts_on_lost_insert_race(self) -> None: + # The "winner" (a concurrent process_pr rebuild) already inserted the + # window this rebuild is about to create, with different rollup values. + PRQueueWindow.objects.create( + pull_request=self.pr, + rule_set=self.rules, + from_ts=_dt(2024, 9, 6), + to_ts=None, + cycle_index=7, + duration_seconds_closed=123, + cumulative_seconds_closed=123, + window_count=9, + first_on_queue_ts=_dt(2024, 9, 1), + ) + + with mock.patch.object( + PRQueueWindow.objects, + "filter", + side_effect=self._empty_first_filter(PRQueueWindow.objects), + ): + res = rebuild_queue_windows_for_ruleset(pr=self.pr, rule_set=self.rules, as_of=_dt(2024, 9, 20)) + + self.assertEqual(res.status, "rebuilt") + windows = list(PRQueueWindow.objects.filter(pull_request=self.pr, rule_set=self.rules)) + self.assertEqual(len(windows), 1) + # The loser's freshly-computed values overwrote the conflicting row. + self.assertEqual(windows[0].from_ts, _dt(2024, 9, 6)) + self.assertIsNone(windows[0].to_ts) + self.assertEqual(windows[0].cycle_index, 0) + self.assertEqual(windows[0].window_count, 1) + self.assertEqual(windows[0].first_on_queue_ts, _dt(2024, 9, 6)) + + def test_build_state_recording_upserts_on_lost_insert_race(self) -> None: + # The "winner" already recorded build state for this (PR, ruleset) pair. + PRQueueWindowBuildState.objects.create( + pull_request=self.pr, + rule_set=self.rules, + revision_version_built=1, + windows_built_at=_dt(2024, 9, 10), + last_status="rebuilt", + last_reason=None, + ) + + with mock.patch.object( + PRQueueWindowBuildState.objects, + "filter", + side_effect=self._empty_first_filter(PRQueueWindowBuildState.objects), + ): + record_queue_window_build_states( + pr=self.pr, + rule_sets=[self.rules], + per_ruleset={int(self.rules.id): {"status": "rebuilt", "reason": None}}, + revision_version=2, + built_at=_dt(2024, 9, 12), + ) + + rows = list(PRQueueWindowBuildState.objects.filter(pull_request=self.pr, rule_set=self.rules)) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0].revision_version_built, 2) + self.assertEqual(rows[0].windows_built_at, _dt(2024, 9, 12)) + + def test_dependency_rebuild_converges_on_lost_insert_race(self) -> None: + self.pr.body = "- [ ] depends on: #99" + self.pr.save(update_fields=["body"]) + # The "winner" (a concurrent rebuild) already inserted the same edge. + PRDependency.objects.create( + pull_request=self.pr, + depends_on_repository=self.repo, + depends_on_number=99, + depends_on_pull_request=None, + ) + + with mock.patch.object( + PRDependency.objects, + "filter", + side_effect=self._empty_first_filter(PRDependency.objects), + ): + res = rebuild_pr_dependencies(self.pr) + + self.assertEqual(res.created, 0) + self.assertEqual(res.deleted, 0) + edges = PRDependency.objects.filter(pull_request=self.pr) + self.assertEqual(edges.count(), 1) + self.assertEqual(edges.get().depends_on_number, 99) + + def test_opt_out_backfill_converges_on_lost_insert_race(self) -> None: + PRTimelineEvent.objects.create( + pull_request=self.pr, + type=PRTimelineEventType.ASSIGNED, + occurred_at=_dt(2024, 9, 10), + assignee_login="Rev", + ) + # The "winner" (a live sync) already inserted the opt-out row. + ReviewerOptOut.objects.create( + repository=self.repo, + pr_number=self.pr.number, + reviewer_login="rev", + active=True, + opted_out_at=_dt(2024, 9, 5), + ) + + with mock.patch.object(QuerySet, "get", flaky_queryset_get()): + res = backfill_reviewer_opt_outs(repository=self.repo, dry_run=False) + + self.assertEqual(res.opt_outs_created, 0) + rows = list(ReviewerOptOut.objects.filter(repository=self.repo, pr_number=self.pr.number)) + self.assertEqual(len(rows), 1) + self.assertFalse(rows[0].active) + self.assertEqual(rows[0].cleared_at, _dt(2024, 9, 10)) diff --git a/qb_site/analyzer/tests/tasks/test_rebuild_queue_windows_sweep_task.py b/qb_site/analyzer/tests/tasks/test_rebuild_queue_windows_sweep_task.py index 8651dbc0..25956c04 100644 --- a/qb_site/analyzer/tests/tasks/test_rebuild_queue_windows_sweep_task.py +++ b/qb_site/analyzer/tests/tasks/test_rebuild_queue_windows_sweep_task.py @@ -925,3 +925,73 @@ def test_no_churn_under_identical_ci_resync(self) -> None: self.assertEqual(res2["windows_rebuilt"], 0) rs_state.refresh_from_db() self.assertEqual(rs_state.windows_built_at, first_built_at) + + +class TestSweepIntegrityErrorContainment(TestCase): + """A write conflict on one PR must not abort the rest of the sweep run.""" + + def setUp(self) -> None: + self.repo = Repository.objects.create(owner="o", name="r", default_branch="master", is_active=True) + self.rule_set = QueueRuleSet.objects.create( + repository=self.repo, + version=1, + require_open=True, + require_not_draft=True, + require_ci_success=False, + ) + + def _mk_stale_pr(self, number: int) -> PullRequest: + now = timezone.now() + pr = PullRequest.objects.create( + repository=self.repo, + number=number, + author=None, + state="open", + is_draft=False, + gh_created_at=now - timezone.timedelta(days=1), + gh_updated_at=now - timezone.timedelta(hours=1), + base_ref_name="master", + head_ref_name="branch", + head_repo_owner_login="o", + head_repo_name="fork", + title="t", + body="", + additions=0, + deletions=0, + changed_files_count=0, + timeline_backfill_done=True, + commits_backfill_done=True, + ) + PRRevision.objects.create(pull_request=pr, head_sha="a1", from_ts=pr.gh_created_at, to_ts=None, seq=0) + PRRevisionBuildState.objects.create(pull_request=pr, revision_version=1) + return pr + + def test_integrity_error_on_one_pr_is_contained(self) -> None: + from unittest import mock + + from django.db import IntegrityError + + from analyzer.services.queue_windows import rebuild_queue_windows_for_pr + + pr_a = self._mk_stale_pr(1) + pr_b = self._mk_stale_pr(2) + + real_rebuild = rebuild_queue_windows_for_pr + + def rebuild_conflicting_on_a(*, pr, rule_sets): + if pr.id == pr_a.id: + raise IntegrityError('duplicate key value violates unique constraint "prqwin_pr_ruleset_from_unique"') + return real_rebuild(pr=pr, rule_sets=rule_sets) + + with mock.patch( + "analyzer.tasks.rebuild_queue_windows_sweep.rebuild_queue_windows_for_pr", + side_effect=rebuild_conflicting_on_a, + ): + res = rebuild_queue_windows_sweep_task.apply(kwargs={"max_prs_per_repo": 5}).get() + + # The conflicting PR was skipped, the other PR was still processed. + self.assertEqual(res["prs_conflict_skipped"], 1) + self.assertEqual(res["prs_checked"], 2) + self.assertTrue(PRQueueWindowBuildState.objects.filter(pull_request=pr_b, rule_set=self.rule_set).exists()) + # The skipped PR recorded no build state, so the next sweep retries it. + self.assertFalse(PRQueueWindowBuildState.objects.filter(pull_request=pr_a, rule_set=self.rule_set).exists()) diff --git a/qb_site/syncer/AGENTS.md b/qb_site/syncer/AGENTS.md index 274c4364..8eb082fa 100644 --- a/qb_site/syncer/AGENTS.md +++ b/qb_site/syncer/AGENTS.md @@ -138,6 +138,11 @@ front and expensive to recover from when skipped. - GraphQL bundle query: `qb_site/syncer/queries/pr_bundle.graphql`. - Sub-sync modules under `qb_site/syncer/services/sub/` should remain narrow and composable. - Preserve boundary: `syncer` stores raw facts; analyzer owns higher-level derived queue/revision semantics. +- Multiple sync paths write the same rows concurrently (live `sync_pr`, archive importer, + admin enqueues, catalog tasks). Follow "Concurrent Writers and Unique Keys" in + `qb_site/AGENTS.md`: never filter-then-create against a unique key — use + `get_or_create`, conflict-aware `bulk_create`, or the savepoint + `IntegrityError` + re-fetch pattern (`ci_sync._archive_mode_upsert`, `core_entities_sync.upsert_user_from_github`). ## Timeline ingest invariants The same logical "page of timeline items" is processed by **three** distinct diff --git a/qb_site/syncer/services/sub/core_entities_sync.py b/qb_site/syncer/services/sub/core_entities_sync.py index acf542d7..ec260e6e 100644 --- a/qb_site/syncer/services/sub/core_entities_sync.py +++ b/qb_site/syncer/services/sub/core_entities_sync.py @@ -2,6 +2,8 @@ from typing import Any, Dict, Optional, Tuple +from django.db import IntegrityError, transaction + from core.models.repository import Repository from core.models.user import User from core.utils.db import update_if_changed @@ -74,33 +76,38 @@ def upsert_user_from_github( created = False updated_fields: list[str] = [] - if gid: - user = User.objects.filter(github_node_id=gid).first() - if user is None and login: - user = User.objects.filter(github_login__iexact=login).first() - if user is None and create_missing and (login or gid): - user = User.objects.create( - github_node_id=gid, - github_login=login, - name=name or None, - avatar_url=avatar or None, - is_active=True, - ) - return user, True, () - elif login: - user = User.objects.filter(github_login__iexact=login).first() - if user is None and create_missing: - user = User.objects.create( - github_login=login, - name=name or None, - avatar_url=avatar or None, - is_active=True, - ) - return user, True, () - else: + def _resolve_existing() -> Optional[User]: + found: Optional[User] = None + if gid: + found = User.objects.filter(github_node_id=gid).first() + if found is None and login: + found = User.objects.filter(github_login__iexact=login).first() + return found + + if not gid and not login: # no usable identity return None, False, () + user = _resolve_existing() + if user is None and create_missing: + try: + # Savepoint: two concurrent syncs ingesting the same new actor race on + # the case-insensitive github_login unique constraint. Losing must not + # poison the caller's transaction; re-resolve and update instead. + with transaction.atomic(): + user = User.objects.create( + github_node_id=gid or None, + github_login=login, + name=name or None, + avatar_url=avatar or None, + is_active=True, + ) + return user, True, () + except IntegrityError: + user = _resolve_existing() + if user is None: + raise + # Update mutable fields if changed if user is None: return None, False, () diff --git a/qb_site/syncer/services/sub/labels_sync.py b/qb_site/syncer/services/sub/labels_sync.py index 60879fc0..fee9fc8d 100644 --- a/qb_site/syncer/services/sub/labels_sync.py +++ b/qb_site/syncer/services/sub/labels_sync.py @@ -35,10 +35,15 @@ def sync_label_catalog(repo: Repository, labels: Iterable[Dict[str, Any]]) -> La if not name: continue color = (node.get("color") or "").strip().lower() - # Case-insensitive match within repo - ld = LabelDef.objects.filter(repository=repo, name__iexact=name).first() - if ld is None: - LabelDef.objects.create(repository=repo, name=name, color=color or "000000") + # Case-insensitive match within repo. Concurrent PR syncs (and the hourly + # catalog task) can race to insert the same new label; get_or_create + # absorbs losing the race on the (repository, lower(name)) unique key. + ld, was_created = LabelDef.objects.get_or_create( + repository=repo, + name__iexact=name, + defaults={"name": name, "color": color or "000000"}, + ) + if was_created: created += 1 else: new_color = (color or ld.color or "").lower() @@ -122,9 +127,18 @@ def sync_full_label_catalog(repo: Repository, labels: Iterable[Dict[str, Any]]) for key, node in desired.items(): ld = existing_by_lower.get(key) if ld is None: - LabelDef.objects.create(repository=repo, name=node["name"], color=node["color"]) - created += 1 - continue + # select_for_update above only locks rows that already exist; a + # concurrent PR-bundle sync can still insert the same new label. + # get_or_create absorbs that race (savepoint inside this atomic + # block) and falls through to the update path with the winner's row. + ld, was_created = LabelDef.objects.get_or_create( + repository=repo, + name__iexact=node["name"], + defaults={"name": node["name"], "color": node["color"]}, + ) + if was_created: + created += 1 + continue updates: Dict[str, Any] = {} if ld.name != node["name"]: updates["name"] = node["name"] diff --git a/qb_site/syncer/services/sub/pull_request_sync.py b/qb_site/syncer/services/sub/pull_request_sync.py index cc88e01b..a05f0b7a 100644 --- a/qb_site/syncer/services/sub/pull_request_sync.py +++ b/qb_site/syncer/services/sub/pull_request_sync.py @@ -5,6 +5,7 @@ from typing import Any, Dict, Tuple from dateutil import parser as dtparser +from django.db import IntegrityError, transaction from django.utils import timezone from core.models.repository import Repository @@ -108,11 +109,21 @@ def upsert_pull_request( } if pr is None: - pr = PullRequest(repository=repo, number=number, **core_values) + new_pr = PullRequest(repository=repo, number=number, **core_values) if not skip_watermark: - pr.last_synced_at = timezone.now() - pr.save() - return PullRequestUpsertResult(pr=pr, created=True, updated_fields=tuple(core_values.keys())) + new_pr.last_synced_at = timezone.now() + try: + # Savepoint: concurrent writers can race to insert the same + # (repository, number) row — live sync_pr vs the archive importer vs + # admin-enqueued syncs. Losing the race falls through to the normal + # update path against the winner's row. + with transaction.atomic(): + new_pr.save() + return PullRequestUpsertResult(pr=new_pr, created=True, updated_fields=tuple(core_values.keys())) + except IntegrityError: + pr = PullRequest.objects.filter(repository=repo, number=number).first() + if pr is None: + raise # Newer-wins guard. Triggered only when callers explicitly pass an # ``if_newer_than`` cutoff (archive ingest does; live sync does not). When diff --git a/qb_site/syncer/tests/services/test_core_entities_sync.py b/qb_site/syncer/tests/services/test_core_entities_sync.py index 81f541aa..6b761189 100644 --- a/qb_site/syncer/tests/services/test_core_entities_sync.py +++ b/qb_site/syncer/tests/services/test_core_entities_sync.py @@ -55,3 +55,40 @@ def test_upsert_user_by_login_then_backfill_node_id(self) -> None: self.assertFalse(created2) self.assertIn("github_node_id", updated2) self.assertEqual(user2.github_node_id, "U2") + + +class TestUserUpsertInsertRace(TestCase): + """Losing the insert race on the case-insensitive login constraint must + re-resolve the winner's row and fall through to the update path.""" + + def test_upsert_user_converges_on_lost_insert_race(self) -> None: + from unittest import mock + + from core.models import User + from syncer.services.sub.core_entities_sync import upsert_user_from_github + + winner = User.objects.create(github_login="Octo") + + # Simulate the loser's view: both resolve lookups (node id, login) miss + # because the winner commits between them and our create. + real_filter = User.objects.filter + calls = {"n": 0} + + def stale_then_real_filter(*args, **kwargs): + calls["n"] += 1 + if calls["n"] <= 2: + return User.objects.none() + return real_filter(*args, **kwargs) + + with mock.patch.object(User.objects, "filter", side_effect=stale_then_real_filter): + user, created, updated_fields = upsert_user_from_github( + {"__typename": "User", "id": "NODE1", "login": "octo", "name": "O", "avatarUrl": "https://x/a.png"} + ) + + self.assertFalse(created) + self.assertIsNotNone(user) + self.assertEqual(user.pk, winner.pk) + winner.refresh_from_db() + self.assertEqual(winner.github_node_id, "NODE1") + self.assertEqual(winner.github_login, "octo") + self.assertEqual(User.objects.count(), 1) diff --git a/qb_site/syncer/tests/services/test_pull_request_sync.py b/qb_site/syncer/tests/services/test_pull_request_sync.py index 15cbfac0..f901b9c7 100644 --- a/qb_site/syncer/tests/services/test_pull_request_sync.py +++ b/qb_site/syncer/tests/services/test_pull_request_sync.py @@ -161,3 +161,55 @@ def test_author_upsert_with_node_id_and_metadata(self) -> None: self.assertEqual(pr.author.github_login, "alice") self.assertEqual(pr.author.name, "Alice A.") self.assertTrue(pr.author.avatar_url.endswith("a.png")) + + +class TestPullRequestUpsertInsertRace(TestCase): + """Losing the insert race on (repository, number) — e.g. live sync vs the + archive importer — must fall through to the update path, not raise.""" + + def setUp(self) -> None: + self.repo = make_repo() + + def test_upsert_converges_on_lost_insert_race(self) -> None: + from unittest import mock + + winner = make_pr(self.repo, 1) + + bundle = { + "number": 1, + "state": "OPEN", + "isDraft": False, + "title": "Racy title", + "body": "body", + "createdAt": "2025-10-20T10:00:00Z", + "updatedAt": "2025-10-20T10:05:00Z", + "baseRefName": "master", + "headRefName": "b", + "headRefOid": "abc123", + "headRepositoryOwner": {"login": "o"}, + "headRepository": {"name": "fork"}, + "additions": 1, + "deletions": 0, + "changedFiles": 1, + "author": None, + } + + # Simulate the loser's stale existence check: the winner's row commits + # between the filter and our insert. + real_filter = PullRequest.objects.filter + state = {"missed": False} + + def stale_then_real_filter(*args, **kwargs): + if not state["missed"]: + state["missed"] = True + return PullRequest.objects.none() + return real_filter(*args, **kwargs) + + with mock.patch.object(PullRequest.objects, "filter", side_effect=stale_then_real_filter): + res = upsert_pull_request(bundle, self.repo) + + self.assertFalse(res.created) + self.assertEqual(res.pr.pk, winner.pk) + self.assertEqual(PullRequest.objects.filter(repository=self.repo, number=1).count(), 1) + winner.refresh_from_db() + self.assertEqual(winner.title, "Racy title") diff --git a/qb_site/syncer/tests/subsystems/test_labels_sync.py b/qb_site/syncer/tests/subsystems/test_labels_sync.py index 413221d3..cd3ddbc4 100644 --- a/qb_site/syncer/tests/subsystems/test_labels_sync.py +++ b/qb_site/syncer/tests/subsystems/test_labels_sync.py @@ -118,3 +118,63 @@ def fetcher(after): nodes = fetch_repo_label_catalog(self.repo, fetcher) self.assertEqual([n["name"] for n in nodes], ["a", "b"]) self.assertEqual(seen_cursors, [None, "c1"]) + + +class TestLabelSyncInsertRace(TestCase): + """Losing the insert race on (repository, lower(name)) must converge, not raise. + + Simulated by making get_or_create's initial get miss once: the create then + conflicts with the pre-existing "winner" row and Django's conflict-retry get + must recover it via the name__iexact lookup — pinning that the lookup covers + the case-insensitive unique constraint. + """ + + def setUp(self) -> None: + self.repo = make_repo() + + def _flaky_get(self): + from django.db.models.query import QuerySet + + real_get = QuerySet.get + state = {"missed": False} + + def flaky(qs_self, *args, **kwargs): + if not state["missed"]: + state["missed"] = True + raise qs_self.model.DoesNotExist + return real_get(qs_self, *args, **kwargs) + + return flaky + + def test_sync_label_catalog_converges_on_lost_insert_race(self) -> None: + from unittest import mock + + from django.db.models.query import QuerySet + + LabelDef.objects.create(repository=self.repo, name="Easy", color="111111") + + with mock.patch.object(QuerySet, "get", self._flaky_get()): + res = sync_label_catalog(self.repo, [{"name": "easy", "color": "abcdef"}]) + + self.assertEqual(res.created, 0) + rows = list(LabelDef.objects.filter(repository=self.repo)) + self.assertEqual(len(rows), 1) + # Fell through to the update path against the winner's row. + self.assertEqual(rows[0].color, "abcdef") + + def test_sync_full_label_catalog_converges_on_lost_insert_race(self) -> None: + from unittest import mock + + LabelDef.objects.create(repository=self.repo, name="Fresh", color="111111") + + # A concurrent PR-bundle sync inserted "Fresh" after this reconcile took + # its select_for_update snapshot: simulate the stale snapshot by making + # it come back empty, so the create path collides with the winner's row. + with mock.patch.object(LabelDef.objects, "select_for_update", return_value=LabelDef.objects.none()): + res = sync_full_label_catalog(self.repo, [{"name": "fresh", "color": "abcdef"}]) + + self.assertEqual(res.created, 0) + rows = list(LabelDef.objects.filter(repository=self.repo)) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0].name, "fresh") + self.assertEqual(rows[0].color, "abcdef") diff --git a/qb_site/zulip_bot/services/registration_linking.py b/qb_site/zulip_bot/services/registration_linking.py index d4a447ec..718aaaa4 100644 --- a/qb_site/zulip_bot/services/registration_linking.py +++ b/qb_site/zulip_bot/services/registration_linking.py @@ -2,7 +2,7 @@ from dataclasses import dataclass -from django.db import transaction +from django.db import IntegrityError, transaction from core.models import User from core.services.github_oauth import GitHubUserIdentity @@ -40,16 +40,34 @@ def link_or_create_user_from_registration( ) if user is None: - created = User.objects.create( - github_node_id=identity.github_node_id, - github_login=identity.github_login, - name=identity.github_name, - avatar_url=identity.github_avatar_url, - zulip_user_id=zulip_user_id, - zulip_full_name=zulip_full_name, - is_active=True, - ) - return RegistrationLinkResult(user=created, outcome="created") + try: + # Savepoint: select_for_update above cannot lock a row that does not + # exist yet, so a concurrent writer (typically the syncer ingesting + # this GitHub user's activity) can insert the same login between our + # lookup and this create. Re-resolve and fall through to the linking + # path instead of failing the registration. + with transaction.atomic(): + created = User.objects.create( + github_node_id=identity.github_node_id, + github_login=identity.github_login, + name=identity.github_name, + avatar_url=identity.github_avatar_url, + zulip_user_id=zulip_user_id, + zulip_full_name=zulip_full_name, + is_active=True, + ) + return RegistrationLinkResult(user=created, outcome="created") + except IntegrityError: + user = User.objects.select_for_update().filter(github_node_id=identity.github_node_id).first() + if user is None: + user = User.objects.select_for_update().filter(github_login__iexact=identity.github_login).first() + if user and user.github_node_id and user.github_node_id != identity.github_node_id: + raise RegistrationLinkConflict( + reason="github_login_bound_to_different_node_id", + message="This GitHub login is already associated with a different GitHub account in Queueboard.", + ) + if user is None: + raise if user.zulip_user_id not in (None, zulip_user_id): raise RegistrationLinkConflict( diff --git a/qb_site/zulip_bot/tests/test_registration_linking.py b/qb_site/zulip_bot/tests/test_registration_linking.py index 964d77b3..200ff943 100644 --- a/qb_site/zulip_bot/tests/test_registration_linking.py +++ b/qb_site/zulip_bot/tests/test_registration_linking.py @@ -81,3 +81,47 @@ def test_conflict_when_login_is_bound_to_different_node_id(self) -> None: zulip_full_name="Reviewer User", identity=self._identity(node_id="U_node_1"), ) + + +class TestRegistrationLinkingInsertRace(TestCase): + """Losing the insert race against the syncer creating the same GitHub user + must fall through to the linking path instead of failing the registration.""" + + def _identity(self) -> GitHubUserIdentity: + return GitHubUserIdentity( + github_user_id=123, + github_node_id="U_node_1", + github_login="reviewer", + github_name="Reviewer User", + github_avatar_url="https://example.com/avatar.png", + ) + + def test_link_converges_on_lost_insert_race(self) -> None: + from unittest import mock + + # The "winner": the syncer ingested this login concurrently (no node id yet). + winner = User.objects.create(github_login="Reviewer", github_node_id=None, zulip_user_id=None) + + # Simulate the loser's stale lookups: both select_for_update lookups miss + # because the winner commits between them and our create. + real_sfu = User.objects.select_for_update + calls = {"n": 0} + + def stale_then_real_sfu(*args, **kwargs): + calls["n"] += 1 + if calls["n"] <= 2: + return User.objects.none() + return real_sfu(*args, **kwargs) + + with mock.patch.object(User.objects, "select_for_update", side_effect=stale_then_real_sfu): + result = link_or_create_user_from_registration( + zulip_user_id=101, + zulip_full_name="Reviewer User", + identity=self._identity(), + ) + + self.assertEqual(result.outcome, "linked_existing") + self.assertEqual(User.objects.count(), 1) + winner.refresh_from_db() + self.assertEqual(winner.github_node_id, "U_node_1") + self.assertEqual(winner.zulip_user_id, 101)