From 3e4fb91503459a8afb9c25f75ee839bd8a5be869 Mon Sep 17 00:00:00 2001 From: seonghobae <8172694+seonghobae@users.noreply.github.com> Date: Tue, 30 Jun 2026 17:38:58 +0000 Subject: [PATCH] =?UTF-8?q?=E2=9A=A1=20Bolt:=20pr=5Freview=5Ffix=5Fschedul?= =?UTF-8?q?er.py=20=EB=82=B4=20N+1=20API=20=EB=B3=91=EB=AA=A9=20=ED=95=B4?= =?UTF-8?q?=EA=B2=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit πŸ’‘ What: `scripts/ci/pr_review_fix_scheduler.py`의 `process_queue` ν•¨μˆ˜ λ‚΄μ—μ„œ 독립적인 PR μ½”λ©˜νŠΈ 쑰회(`issue_comments`)λ₯Ό 직렬둜 μˆ˜ν–‰ν•˜λ˜ 것을 `concurrent.futures.ThreadPoolExecutor`λ₯Ό ν™œμš©ν•΄ 병렬 μ²˜λ¦¬ν•˜λ„λ‘ μ΅œμ ν™”ν–ˆμŠ΅λ‹ˆλ‹€. 🎯 Why: λ‹€μˆ˜μ˜ PR을 μ²˜λ¦¬ν•  λ•Œ λ°˜λ³΅λ˜λŠ” μ™ΈλΆ€ μ„œλΉ„μŠ€ API 호좜둜 인해 λ°œμƒν•˜λŠ” N+1 병λͺ©(Bottleneck) ν˜„μƒμ„ ν•΄κ²°ν•˜κ³  νŒŒμ΄ν”„λΌμΈμ˜ μ‹€ν–‰ μ‹œκ°„μ„ μ„ ν˜•μ μœΌλ‘œ μ§€μ—°μ‹œν‚€λŠ” 것을 λ°©μ§€ν•˜κΈ° μœ„ν•¨μž…λ‹ˆλ‹€. 단일 PR 처리 μ‹œμ—λŠ” μ˜€λ²„ν—€λ“œκ°€ 적은 직렬(Fast path)둜 μ²˜λ¦¬λ˜λ„λ‘ κ΅¬μ„±ν–ˆμŠ΅λ‹ˆλ‹€. πŸ“Š Impact: 큐 μŠ€μΊ” λŒ€κΈ° μ‹œκ°„μ„ API 호좜 수(O(N))μ—μ„œ O(N/max_workers)둜 κ°μ†Œμ‹œν‚΅λ‹ˆλ‹€. πŸ”¬ Measurement: μ½”λ“œ 리뷰 νŒŒμ΄ν”„λΌμΈ λŸ°νƒ€μž„μ„ 톡해 κ°μ†Œλœ μ‹œκ°„μ„ μΈ‘μ •ν•  수 있으며 `test_pr_review_fix_scheduler.py` 및 ν…ŒμŠ€νŠΈ 컀버리지 슀크립트둜 λ™μž‘ 및 μ˜ˆμ™Έ 처리λ₯Ό κ²€μ¦ν•©λ‹ˆλ‹€. --- scripts/ci/pr_review_fix_scheduler.py | 55 ++++++++++++++++++- tests/test_pr_review_fix_scheduler.py | 4 +- .../test_pr_review_fix_scheduler_coverage.py | 55 +++++++++++++++++++ 3 files changed, 109 insertions(+), 5 deletions(-) create mode 100644 tests/test_pr_review_fix_scheduler_coverage.py diff --git a/scripts/ci/pr_review_fix_scheduler.py b/scripts/ci/pr_review_fix_scheduler.py index 25d57ed4e1..80c564f342 100755 --- a/scripts/ci/pr_review_fix_scheduler.py +++ b/scripts/ci/pr_review_fix_scheduler.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse +import concurrent.futures import json import os import re @@ -133,7 +134,13 @@ def dispatch_autofix(repo: str, pr: dict[str, Any], *, workflow: str, dry_run: b run(args) -def inspect_pr(repo: str, pr: dict[str, Any], args: argparse.Namespace) -> tuple[str, tuple[str, ...]]: +def inspect_pr( + repo: str, + pr: dict[str, Any], + args: argparse.Namespace, + *, + comments: list[dict[str, Any]] | None = None, +) -> tuple[str, tuple[str, ...]]: """Inspect one PR and optionally dispatch autofix.""" number = int(pr["number"]) if pr.get("isDraft"): @@ -147,7 +154,9 @@ def inspect_pr(repo: str, pr: dict[str, Any], args: argparse.Namespace) -> tuple if not needs_fix: return "skip", ("no current-head change request or active unresolved review thread",) - comments = issue_comments(repo, number) + if comments is None: + comments = issue_comments(repo, number) + if recent_fix_marker_exists(comments, str(pr["headRefOid"]), args.retry_hours * 3600): return "wait", ("recent autofix marker exists for this head",) @@ -163,13 +172,53 @@ def process_queue(args: argparse.Namespace) -> int: inspected = 0 decisions: list[dict[str, Any]] = [] + prs_needing_comments = [] + for pr in prs: + if pr.get("isDraft"): + continue + if pr.get("baseRefName") != args.base_branch: + continue + if not same_repository_head(args.repo, pr): + continue + needs_fix, _ = needs_autofix(pr) + if needs_fix: + prs_needing_comments.append(pr) + + comments_by_pr: dict[int, list[dict[str, Any]]] = {} + if len(prs_needing_comments) <= 1: + # Fast path for single items + for pr in prs_needing_comments: + pr_number = int(pr["number"]) + comments_by_pr[pr_number] = issue_comments(args.repo, pr_number) + else: + # ⚑ Bolt: Avoid N+1 API blocking by parallelizing independent issue_comments fetches + # Impact: Reduces wait time from O(N) API calls to O(N/max_workers) for queue scanning + max_workers = min(10, len(prs_needing_comments)) + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + def fetch_comments(pr_number: int) -> tuple[int, list[dict[str, Any]]]: + return pr_number, issue_comments(args.repo, pr_number) + + futures = [executor.submit(fetch_comments, int(pr["number"])) for pr in prs_needing_comments] + for future in concurrent.futures.as_completed(futures): + try: + pr_number, comments = future.result() + comments_by_pr[pr_number] = comments + except Exception: + pass + for pr in prs: inspected += 1 if dispatched >= args.max_dispatches: decisions.append({"pr": pr["number"], "action": "skip", "reasons": ["autofix dispatch limit reached"]}) continue try: - action, reasons = inspect_pr(args.repo, pr, args) + pr_number = int(pr["number"]) + action, reasons = inspect_pr( + args.repo, + pr, + args, + comments=comments_by_pr.get(pr_number), + ) except RuntimeError as exc: action, reasons = "error", (str(exc),) if action == "dispatch": diff --git a/tests/test_pr_review_fix_scheduler.py b/tests/test_pr_review_fix_scheduler.py index 326a949a6b..878ef08aae 100644 --- a/tests/test_pr_review_fix_scheduler.py +++ b/tests/test_pr_review_fix_scheduler.py @@ -275,14 +275,14 @@ def test_fix_inspect_skip_wait_and_error_paths(monkeypatch): pr1 = make_pr(number=1) pr2 = make_pr(number=2) monkeypatch.setattr(fix, "fetch_open_prs", lambda repo, max_prs: [pr1, pr2]) - monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args: ("dispatch", ("reason",))) + monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args, **kwargs: ("dispatch", ("reason",))) payload_lines = [] monkeypatch.setattr("builtins.print", lambda *parts, **kwargs: payload_lines.append(" ".join(map(str, parts)))) assert fix.process_queue(args) == 0 assert "autofix dispatch limit reached" in payload_lines[-1] monkeypatch.setattr(fix, "fetch_pr", lambda repo, number: [make_pr(number=number)]) - monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args: (_ for _ in ()).throw(RuntimeError("boom"))) + monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args, **kwargs: (_ for _ in ()).throw(RuntimeError("boom"))) args = fix.parse_args(["--repo", "owner/repo", "--base-branch", "main", "--pr-number", "3"]) payload_lines.clear() assert fix.process_queue(args) == 0 diff --git a/tests/test_pr_review_fix_scheduler_coverage.py b/tests/test_pr_review_fix_scheduler_coverage.py new file mode 100644 index 0000000000..5399955ad3 --- /dev/null +++ b/tests/test_pr_review_fix_scheduler_coverage.py @@ -0,0 +1,55 @@ +import pytest +import scripts.ci.pr_review_fix_scheduler as fix + +def test_coverage_process_queue_skips_draft_and_wrong_base_and_external_repo(monkeypatch): + def make_pr(number=1, **kwargs): + pr = { + "number": number, + "headRefOid": "abc", + "baseRefName": "main", + "headRefName": "feature", + "isDraft": False, + "headRepository": {"nameWithOwner": "owner/repo"}, + } + pr.update(kwargs) + return pr + + args = fix.parse_args(["--repo", "owner/repo", "--base-branch", "main"]) + + pr1 = make_pr(number=1, isDraft=True) + pr2 = make_pr(number=2, baseRefName="other") + pr3 = make_pr(number=3, headRepository={"nameWithOwner": "fork/repo"}) + + monkeypatch.setattr(fix, "fetch_open_prs", lambda repo, max_prs: [pr1, pr2, pr3]) + monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args, **kwargs: ("skip", ("skip reason",))) + + assert fix.process_queue(args) == 0 + +def test_coverage_process_queue_exception_handling(monkeypatch): + def make_pr(number=1, **kwargs): + pr = { + "number": number, + "headRefOid": "abc", + "baseRefName": "main", + "headRefName": "feature", + "isDraft": False, + "headRepository": {"nameWithOwner": "owner/repo"}, + } + pr.update(kwargs) + return pr + + args = fix.parse_args(["--repo", "owner/repo", "--base-branch", "main"]) + + pr1 = make_pr(number=1) + pr2 = make_pr(number=2) + + monkeypatch.setattr(fix, "fetch_open_prs", lambda repo, max_prs: [pr1, pr2]) + monkeypatch.setattr(fix, "needs_autofix", lambda pr: (True, ("reason",))) + + def raise_error(repo, number): + raise RuntimeError("boom") + + monkeypatch.setattr(fix, "issue_comments", raise_error) + monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args, **kwargs: ("skip", ("skip reason",))) + + assert fix.process_queue(args) == 0