From 7723e07229342d5e06aa73b59dd215db47232bb3 Mon Sep 17 00:00:00 2001 From: Bryan Gin-ge Chen Date: Thu, 2 Jul 2026 03:18:02 -0400 Subject: [PATCH] =?UTF-8?q?fix(syncer):=20stop=20O(n=C2=B2)=20CI-expiry=20?= =?UTF-8?q?query=20from=20pinning=20vacuum=20xmin?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The superseded-row passes of syncer.expire_stale_ci_for_repo used exclude(id__in=), which renders as NOT (id IN (...)). Once the subquery result outgrew work_mem, Postgres re-executed it per outer row — O(n²). In production these queries ran for days (one new orphan per daily beat run surviving worker restarts), pinning the vacuum xmin horizon database-wide: 494k unremovable dead tuples bloated django_celery_results_taskresult to 586 MB and drove the admin "Task results" changelist into Heroku H12 timeouts. Fixes, in layers: - Rewrite passes 3/4 as correlated Exists("newer sibling") anti-joins, which Postgres plans as an index-probe anti-join. - Run all four passes' deletes in id-cursor batches of 5000 (queue window invalidation before each batch per design doc 040 I2), keeping transactions short and memory bounded. - Guard the task with a per-statement timeout, new env-backed setting SYNCER_CI_EXPIRY_STATEMENT_TIMEOUT_SECONDS (default 300s), so any recurrence fails loudly instead of orphaning. - Admin defense-in-depth: TaskResult changelist no longer issues SELECT DISTINCT scans per page load (task_name filter now reads the Celery registry; periodic_task_name/worker AllValues filters and the duplicate task_name filter removed; show_full_result_count off). Also adds the previously-missing SYNCER_CI_EXPIRY_* vars to .env.example and a multi-batch convergence test (batch size 1). Tested: core + syncer suites green against dockerized Postgres (3 pre-existing GH-token-dependent errors excepted, pass with .env); generated SQL verified as correlated EXISTS; ruff check/format clean. Co-Authored-By: Claude Fable 5 --- .env.example | 5 + qb_site/core/admin.py | 34 ++- qb_site/qb_site/settings/base.py | 3 + qb_site/syncer/AGENTS.md | 2 +- qb_site/syncer/tasks/sync_tasks.py | 210 +++++++++++------- .../tests/tasks/test_expire_stale_ci_task.py | 24 ++ 6 files changed, 193 insertions(+), 85 deletions(-) diff --git a/.env.example b/.env.example index 148eca62..7329f42d 100644 --- a/.env.example +++ b/.env.example @@ -161,6 +161,11 @@ SYNCER_CI_SHA_SETTLE_WINDOW_SECONDS=1800 SYNCER_CI_SHA_HARD_CAP_DAYS=400 # CI-by-SHA min attempts before treating empty/filtered/not_found as terminal. SYNCER_CI_SHA_MIN_ATTEMPTS_TERMINAL=2 +# CI row expiry sweep (deletes stale pending + superseded CI rows). Period 0 disables the beat entry. +SYNCER_CI_EXPIRY_PERIOD_SECONDS=86400 +SYNCER_CI_STALE_PENDING_DAYS=30 +# Per-statement Postgres timeout (seconds) inside the expiry task; 0 disables the guard. +SYNCER_CI_EXPIRY_STATEMENT_TIMEOUT_SECONDS=300 # GitHub throttling: minimum spacing between GraphQL requests (ms) and max wait per request (ms). SYNCER_GH_THROTTLE_MS=250 SYNCER_GH_THROTTLE_MAX_WAIT_MS=5000 diff --git a/qb_site/core/admin.py b/qb_site/core/admin.py index bf27532b..3f4c0398 100644 --- a/qb_site/core/admin.py +++ b/qb_site/core/admin.py @@ -1043,6 +1043,33 @@ def run_reviewer_attention_view(self, request, *args, **kwargs): # type: ignore except admin.sites.NotRegistered: # pragma: no cover pass + class RegisteredTaskNameFilter(admin.SimpleListFilter): + """task_name filter fed from the Celery registry instead of the table. + + The stock ``AllValuesFieldListFilter`` runs ``SELECT DISTINCT task_name`` + over the whole table on every changelist load; Postgres has no loose + index scan, so that walks the full index each time. The registry gives + the same choices for free. ``parameter_name = "task_name"`` keeps + existing ``?task_name=...`` links working. + """ + + title = "task name" + parameter_name = "task_name" + + def lookups(self, request, model_admin): + try: + from qb_site.celery import app as celery_app + + names = sorted(name for name in celery_app.tasks.keys() if not name.startswith("celery.")) + except Exception: # pragma: no cover - registry unavailable + names = [] + return [(name, name) for name in names] + + def queryset(self, request, queryset): + if self.value(): + return queryset.filter(task_name=self.value()) + return queryset + @admin.register(TaskResult) class EnhancedTaskResultAdmin(TaskResultAdmin): # type: ignore[misc] change_form_template = "admin/django_celery_results/taskresult/change_form.html" @@ -1057,7 +1084,12 @@ class EnhancedTaskResultAdmin(TaskResultAdmin): # type: ignore[misc] "date_done", ) list_display_links = ("short_id",) - list_filter = getattr(TaskResultAdmin, "list_filter", tuple()) + ("task_name",) + # Deliberately NOT the stock list_filter: its periodic_task_name / + # task_name / worker entries are AllValues filters that each scan the + # whole table per page load (part of an admin-timeout incident). + list_filter = ("status", "date_done", RegisteredTaskNameFilter) + # Skip the unfiltered COUNT(*) when a filter is active. + show_full_result_count = False search_fields = getattr(TaskResultAdmin, "search_fields", tuple()) + ( "task_id", "result", diff --git a/qb_site/qb_site/settings/base.py b/qb_site/qb_site/settings/base.py index 9dac528b..2677a91f 100644 --- a/qb_site/qb_site/settings/base.py +++ b/qb_site/qb_site/settings/base.py @@ -318,6 +318,9 @@ def env_optional_bounded_int(name: str, *, minimum: int, maximum: int) -> int | # CI row expiry / superseded-row cleanup SYNCER_CI_EXPIRY_PERIOD_SECONDS = int(os.getenv("SYNCER_CI_EXPIRY_PERIOD_SECONDS", 86400)) SYNCER_CI_STALE_PENDING_DAYS = int(os.getenv("SYNCER_CI_STALE_PENDING_DAYS", 30)) +# Per-statement Postgres timeout inside the expiry task; a runaway plan once ran for +# days and blocked vacuum database-wide. 0 disables the guard. +SYNCER_CI_EXPIRY_STATEMENT_TIMEOUT_SECONDS = int(os.getenv("SYNCER_CI_EXPIRY_STATEMENT_TIMEOUT_SECONDS", 300)) # Webhook delivery log cleanup SYNCER_WEBHOOK_DELIVERY_RETENTION_DAYS = int(os.getenv("SYNCER_WEBHOOK_DELIVERY_RETENTION_DAYS", 7)) diff --git a/qb_site/syncer/AGENTS.md b/qb_site/syncer/AGENTS.md index 879cb2b5..274c4364 100644 --- a/qb_site/syncer/AGENTS.md +++ b/qb_site/syncer/AGENTS.md @@ -122,7 +122,7 @@ front and expensive to recover from when skipped. count on `SyncerConvergenceSnapshot` via `syncer.collect_convergence`. The shared detector is `syncer.services.consistency.inconsistent_open_prs_queryset`), - `syncer.refresh_pending_ci_for_active_repos` → `syncer.refresh_pending_ci_for_repo`, - - `syncer.expire_stale_ci_for_active_repos` → `syncer.expire_stale_ci_for_repo` (daily; deletes phantom pending and superseded same-SHA+name CI rows), + - `syncer.expire_stale_ci_for_active_repos` → `syncer.expire_stale_ci_for_repo` (daily; deletes phantom pending and superseded same-SHA+name CI rows in bounded id-cursor batches under a per-statement Postgres timeout, `SYNCER_CI_EXPIRY_STATEMENT_TIMEOUT_SECONDS`. The superseded passes must stay written as correlated-`Exists` anti-joins — an `exclude(id__in=)` here once planned as an O(n²) per-row subplan that ran for days and pinned the vacuum xmin horizon database-wide), - `syncer.expire_old_webhook_deliveries` (daily by default; deletes GitHubWebhookDelivery rows older than SYNCER_WEBHOOK_DELIVERY_RETENTION_DAYS), - `syncer.sync_label_catalog_for_active_repos` → `syncer.sync_label_catalog` (hourly by default via SYNCER_LABEL_CATALOG_PERIOD_SECONDS; pages through `repository.labels` and reconciles `LabelDef` rows, deleting labels removed upstream — cascades to `PRLabel`), - `syncer.sync_ci_for_shas` / `syncer.sync_ci_for_repo_shas` — CI-by-SHA ingestion, diff --git a/qb_site/syncer/tasks/sync_tasks.py b/qb_site/syncer/tasks/sync_tasks.py index 7811879a..89d18091 100644 --- a/qb_site/syncer/tasks/sync_tasks.py +++ b/qb_site/syncer/tasks/sync_tasks.py @@ -1,11 +1,12 @@ from __future__ import annotations import logging +from contextlib import contextmanager from typing import Any, Dict, Optional, Sequence from datetime import datetime, timedelta from celery import shared_task -from django.db import models +from django.db import connection, models from django.db.models.functions import Coalesce from dateutil import parser as dtparser from django.utils import timezone @@ -1517,19 +1518,19 @@ def _invalidate_queue_windows_for_ci_rows( *, check_run_ids: list[int], status_context_ids: list[int], -) -> int: +) -> set[int]: """Mark PRQueueWindowBuildState stale and enqueue rebuilds for any PRQueueWindow rows whose attribution FKs reference the given about-to-be-deleted CI row IDs. Must be called *before* the rows are deleted so the FK lookup still resolves. - Returns the number of distinct PR ids affected. + Returns the set of distinct PR ids affected. """ from django.db.models import Q as _Q from analyzer.models import PRQueueWindowBuildState from analyzer.models.queue_window import PRQueueWindow if not check_run_ids and not status_context_ids: - return 0 + return set() fk_filter = _Q() if check_run_ids: @@ -1542,7 +1543,7 @@ def _invalidate_queue_windows_for_ci_rows( affected = PRQueueWindow.objects.filter(fk_filter).values("pull_request_id", "rule_set_id").distinct() affected_rows = list(affected) if not affected_rows: - return 0 + return set() affected_pr_ids = list({int(row["pull_request_id"]) for row in affected_rows}) affected_pairs = [(int(row["pull_request_id"]), int(row["rule_set_id"])) for row in affected_rows] @@ -1562,7 +1563,58 @@ def _invalidate_queue_windows_for_ci_rows( for pr_id in affected_pr_ids: process_pr_task.delay(pr_id) - return len(affected_pr_ids) + return set(affected_pr_ids) + + +_EXPIRE_DELETE_BATCH_SIZE = 5000 + + +@contextmanager +def _ci_expiry_statement_timeout(): + """Cap per-statement runtime for the CI expiry passes. + + A planner regression once left this task's superseded-rows query running + for days server-side, pinning the vacuum xmin horizon database-wide. The + timeout turns any recurrence into a loud task failure instead. + """ + seconds = int(getattr(settings, "SYNCER_CI_EXPIRY_STATEMENT_TIMEOUT_SECONDS", 300)) + if seconds <= 0: + yield + return + with connection.cursor() as cursor: + cursor.execute(f"SET statement_timeout = '{int(seconds)}s'") + try: + yield + finally: + with connection.cursor() as cursor: + cursor.execute("SET statement_timeout = DEFAULT") + + +def _expire_ci_rows_in_batches(qs, *, model, kind: str) -> tuple[int, set[int]]: + """Delete all rows matching ``qs`` in bounded id-cursor batches. + + For each batch: invalidate attributed queue windows first (design doc 040 + invariant I2), then delete by id. Batches keep transactions short and cap + memory; the id cursor guarantees forward progress without rescanning. + Correctness relies on deletions never making a previously-unmatched row + match ``qs`` — true for both the stale-pending and superseded predicates. + Returns (total deleted rows incl. cascades, set of affected PR ids). + """ + deleted_total = 0 + affected_pr_ids: set[int] = set() + last_id = 0 + while True: + batch_ids = list(qs.filter(id__gt=last_id).order_by("id").values_list("id", flat=True)[:_EXPIRE_DELETE_BATCH_SIZE]) + if not batch_ids: + break + if kind == "check_run": + affected_pr_ids |= _invalidate_queue_windows_for_ci_rows(check_run_ids=batch_ids, status_context_ids=[]) + else: + affected_pr_ids |= _invalidate_queue_windows_for_ci_rows(check_run_ids=[], status_context_ids=batch_ids) + deleted, _ = model.objects.filter(id__in=batch_ids).delete() + deleted_total += deleted + last_id = batch_ids[-1] + return deleted_total, affected_pr_ids @shared_task(name="syncer.expire_stale_ci_for_repo") @@ -1583,8 +1635,14 @@ def expire_stale_ci_for_repo_task( # type: ignore[no-redef] Before each deletion pass, any PRQueueWindow rows whose attribution FKs reference the about-to-be-deleted CI row IDs are marked stale and enqueued for rebuild (see docs/design-decisions/040-queue-window-event-attribution.md, invariant I2). + + The superseded passes use a correlated NOT-EXISTS-style anti-join + (``Exists(newer sibling)``), NOT ``exclude(id__in=)``: + the latter renders as ``NOT (id IN (...))``, and once the subquery result + outgrows work_mem Postgres re-executes it per outer row — O(n²), observed + running for days and blocking vacuum database-wide. """ - from django.db.models import Max + from django.db.models import Exists, OuterRef repo = Repository.objects.get(id=int(repo_id)) @@ -1592,80 +1650,66 @@ def expire_stale_ci_for_repo_task( # type: ignore[no-redef] stale_pending_days = int(getattr(settings, "SYNCER_CI_STALE_PENDING_DAYS", 30)) stale_pending_days_int = int(stale_pending_days) - prs_invalidated_stale_cr = 0 - prs_invalidated_stale_sc = 0 - prs_invalidated_superseded_cr = 0 - prs_invalidated_superseded_sc = 0 - - # ------------------------------------------------------------------ # - # Pass 1: stale pending CommitCheckRun rows # - # ------------------------------------------------------------------ # - deleted_stale_cr = 0 - if stale_pending_days_int > 0: - cutoff = timezone.now() - timedelta(days=stale_pending_days_int) - origin = Coalesce("gh_started_at", "gh_completed_at", "created_at") - stale_cr_ids = list( - CommitCheckRun.objects.filter(repository=repo) - .exclude(status="COMPLETED") - .annotate(origin=origin) - .filter(origin__lt=cutoff) - .values_list("id", flat=True) + with _ci_expiry_statement_timeout(): + # -------------------------------------------------------------- # + # Pass 1: stale pending CommitCheckRun rows # + # -------------------------------------------------------------- # + deleted_stale_cr = 0 + prs_invalidated_stale_cr: set[int] = set() + if stale_pending_days_int > 0: + cutoff = timezone.now() - timedelta(days=stale_pending_days_int) + origin = Coalesce("gh_started_at", "gh_completed_at", "created_at") + stale_cr_qs = ( + CommitCheckRun.objects.filter(repository=repo) + .exclude(status="COMPLETED") + .annotate(origin=origin) + .filter(origin__lt=cutoff) + ) + deleted_stale_cr, prs_invalidated_stale_cr = _expire_ci_rows_in_batches( + stale_cr_qs, model=CommitCheckRun, kind="check_run" + ) + + # -------------------------------------------------------------- # + # Pass 2: stale pending CommitStatusContext rows # + # -------------------------------------------------------------- # + deleted_stale_sc = 0 + prs_invalidated_stale_sc: set[int] = set() + if stale_pending_days_int > 0: + cutoff = timezone.now() - timedelta(days=stale_pending_days_int) + stale_sc_qs = CommitStatusContext.objects.filter(repository=repo, state="PENDING").filter(gh_created_at__lt=cutoff) + deleted_stale_sc, prs_invalidated_stale_sc = _expire_ci_rows_in_batches( + stale_sc_qs, model=CommitStatusContext, kind="status_context" + ) + + # -------------------------------------------------------------- # + # Pass 3: superseded CommitCheckRun rows (older non-latest per # + # (head_sha, name) within this repo) # + # -------------------------------------------------------------- # + newer_cr = CommitCheckRun.objects.filter( + repository=repo, + head_sha=OuterRef("head_sha"), + name=OuterRef("name"), + id__gt=OuterRef("id"), ) - prs_invalidated_stale_cr = _invalidate_queue_windows_for_ci_rows(check_run_ids=stale_cr_ids, status_context_ids=[]) - if stale_cr_ids: - deleted_stale_cr, _ = CommitCheckRun.objects.filter(id__in=stale_cr_ids).delete() - - # ------------------------------------------------------------------ # - # Pass 2: stale pending CommitStatusContext rows # - # ------------------------------------------------------------------ # - deleted_stale_sc = 0 - if stale_pending_days_int > 0: - cutoff = timezone.now() - timedelta(days=stale_pending_days_int) - stale_sc_ids = list( - CommitStatusContext.objects.filter(repository=repo, state="PENDING") - .filter(gh_created_at__lt=cutoff) - .values_list("id", flat=True) + superseded_cr_qs = CommitCheckRun.objects.filter(repository=repo).filter(Exists(newer_cr)) + deleted_superseded_cr, prs_invalidated_superseded_cr = _expire_ci_rows_in_batches( + superseded_cr_qs, model=CommitCheckRun, kind="check_run" + ) + + # -------------------------------------------------------------- # + # Pass 4: superseded CommitStatusContext rows (graphql-keyed only)# + # -------------------------------------------------------------- # + newer_sc = CommitStatusContext.objects.filter( + repository=repo, + rest_id__isnull=True, + head_sha=OuterRef("head_sha"), + name=OuterRef("name"), + id__gt=OuterRef("id"), + ) + superseded_sc_qs = CommitStatusContext.objects.filter(repository=repo, rest_id__isnull=True).filter(Exists(newer_sc)) + deleted_superseded_sc, prs_invalidated_superseded_sc = _expire_ci_rows_in_batches( + superseded_sc_qs, model=CommitStatusContext, kind="status_context" ) - prs_invalidated_stale_sc = _invalidate_queue_windows_for_ci_rows(check_run_ids=[], status_context_ids=stale_sc_ids) - if stale_sc_ids: - deleted_stale_sc, _ = CommitStatusContext.objects.filter(id__in=stale_sc_ids).delete() - - # ------------------------------------------------------------------ # - # Pass 3: superseded CommitCheckRun rows (older non-latest per # - # (head_sha, name) within this repo) # - # ------------------------------------------------------------------ # - latest_cr_ids = ( - CommitCheckRun.objects.filter(repository=repo) - .values("head_sha", "name") - .annotate(latest_id=Max("id")) - .values_list("latest_id", flat=True) - ) - superseded_cr_ids = list( - CommitCheckRun.objects.filter(repository=repo).exclude(id__in=latest_cr_ids).values_list("id", flat=True) - ) - prs_invalidated_superseded_cr = _invalidate_queue_windows_for_ci_rows(check_run_ids=superseded_cr_ids, status_context_ids=[]) - deleted_superseded_cr = 0 - if superseded_cr_ids: - deleted_superseded_cr, _ = CommitCheckRun.objects.filter(id__in=superseded_cr_ids).delete() - - # ------------------------------------------------------------------ # - # Pass 4: superseded CommitStatusContext rows (graphql-keyed only) # - # ------------------------------------------------------------------ # - latest_sc_ids = ( - CommitStatusContext.objects.filter(repository=repo, rest_id__isnull=True) - .values("head_sha", "name") - .annotate(latest_id=Max("id")) - .values_list("latest_id", flat=True) - ) - superseded_sc_ids = list( - CommitStatusContext.objects.filter(repository=repo, rest_id__isnull=True) - .exclude(id__in=latest_sc_ids) - .values_list("id", flat=True) - ) - prs_invalidated_superseded_sc = _invalidate_queue_windows_for_ci_rows(check_run_ids=[], status_context_ids=superseded_sc_ids) - deleted_superseded_sc = 0 - if superseded_sc_ids: - deleted_superseded_sc, _ = CommitStatusContext.objects.filter(id__in=superseded_sc_ids, rest_id__isnull=True).delete() return { "repo": f"{repo.owner}/{repo.name}", @@ -1675,10 +1719,10 @@ def expire_stale_ci_for_repo_task( # type: ignore[no-redef] "deleted_stale_pending_status_contexts": deleted_stale_sc, "deleted_superseded_check_runs": deleted_superseded_cr, "deleted_superseded_status_contexts": deleted_superseded_sc, - "prs_invalidated_stale_pending_check_runs": prs_invalidated_stale_cr, - "prs_invalidated_stale_pending_status_contexts": prs_invalidated_stale_sc, - "prs_invalidated_superseded_check_runs": prs_invalidated_superseded_cr, - "prs_invalidated_superseded_status_contexts": prs_invalidated_superseded_sc, + "prs_invalidated_stale_pending_check_runs": len(prs_invalidated_stale_cr), + "prs_invalidated_stale_pending_status_contexts": len(prs_invalidated_stale_sc), + "prs_invalidated_superseded_check_runs": len(prs_invalidated_superseded_cr), + "prs_invalidated_superseded_status_contexts": len(prs_invalidated_superseded_sc), } diff --git a/qb_site/syncer/tests/tasks/test_expire_stale_ci_task.py b/qb_site/syncer/tests/tasks/test_expire_stale_ci_task.py index 0d5b59f0..ddd4b3bc 100644 --- a/qb_site/syncer/tests/tasks/test_expire_stale_ci_task.py +++ b/qb_site/syncer/tests/tasks/test_expire_stale_ci_task.py @@ -172,6 +172,30 @@ def test_pass4_preserves_rest_id_rows(self) -> None: self.assertTrue(CommitStatusContext.objects.filter(pk=rest2.pk).exists()) self.assertEqual(result["deleted_superseded_status_contexts"], 0) + # ------------------------------------------------------------------ # + # Batched deletion # + # ------------------------------------------------------------------ # + + def test_batched_deletion_converges_with_tiny_batch_size(self) -> None: + """All passes drain fully even when each batch holds a single row.""" + from unittest.mock import patch + + # Pass 1 fodder: two stale pending check runs. + _cr(self.repo, node_id="STALE_A", sha="sA", started_delta=timedelta(days=40)) + _cr(self.repo, node_id="STALE_B", sha="sB", started_delta=timedelta(days=41)) + # Pass 3 fodder: three superseded + one latest in the same group. + for i in range(4): + _cr(self.repo, node_id=f"SUP_{i}", sha="shaX", name="build", status="COMPLETED", conclusion="SUCCESS") + + with patch("syncer.tasks.sync_tasks._EXPIRE_DELETE_BATCH_SIZE", 1): + result = expire_stale_ci_for_repo_task(self.repo.id, stale_pending_days=30) + + self.assertEqual(result["deleted_stale_pending_check_runs"], 2) + self.assertEqual(result["deleted_superseded_check_runs"], 3) + remaining = CommitCheckRun.objects.filter(repository=self.repo) + self.assertEqual(remaining.count(), 1) + self.assertEqual(remaining.get().github_node_id, "SUP_3") + # ------------------------------------------------------------------ # # Cross-repo isolation # # ------------------------------------------------------------------ #