diff --git a/.jules/bolt.md b/.jules/bolt.md new file mode 100644 index 0000000..f5a2e6c --- /dev/null +++ b/.jules/bolt.md @@ -0,0 +1,3 @@ +## 2024-07-12 - Prevent N+1 query problem with parallel subprocesses for PR checks +**Learning:** Checking the state of multiple PRs in a list sequentially via `_gh_json` leads to N+1 delays, as each check spawns a subprocess call to `gh` CLI which blocks the loop. +**Action:** Use `concurrent.futures.ThreadPoolExecutor` to run `_pr_is_still_open` asynchronously. Using a thread pool speeds up list checking and doesn't block the main thread. diff --git a/ralph_loop/cli.py b/ralph_loop/cli.py index da6a240..987a0c3 100644 --- a/ralph_loop/cli.py +++ b/ralph_loop/cli.py @@ -1,6 +1,8 @@ """Command-line interface and top-level orchestration.""" + from __future__ import annotations +import concurrent.futures import argparse import os import re @@ -335,9 +337,7 @@ def _validate_pr_metadata( ) -> str: if pr_data.get("state") != "OPEN": raise CommandError( - "PR {} is not open (state={}).".format( - pr_number, pr_data.get("state") - ) + "PR {} is not open (state={}).".format(pr_number, pr_data.get("state")) ) if pr_data.get("isDraft"): raise CommandError( @@ -529,6 +529,15 @@ def _spawn_child( return proc, log_path, log_handle +def _check_pr_state_worker(pr: int) -> Tuple[int, bool, Optional[Exception]]: + try: + return pr, _pr_is_still_open(pr), None + except CommandError as exc: + return pr, True, exc # keep on transient error + except Exception as exc: + return pr, True, exc + + def _filter_to_still_open_prs(pr_numbers: List[int]) -> List[int]: """Drop PRs that are no longer OPEN/non-draft via a targeted gh pr view. @@ -543,24 +552,33 @@ def _filter_to_still_open_prs(pr_numbers: List[int]) -> List[int]: not-OPEN. This matches the behaviour callers expect: do not silently swallow stale PRs because of a flaky network. """ + if not pr_numbers: + return [] + + # ⚡ Bolt Optimization: Prevent N+1 query problem. + # We use ThreadPoolExecutor because sequential `gh pr view` calls + # create a major bottleneck when scanning large numbers of PRs. + # The `executor.map` ensures PR order is preserved in the output. kept: List[int] = [] - for pr in pr_numbers: - try: - still_open = _pr_is_still_open(pr) - except CommandError as exc: - _print_step( - "Could not confirm PR #{} open state ({}); keeping it in the " - "fan-out set.".format(pr, exc) - ) - kept.append(pr) - continue - if still_open: - kept.append(pr) - else: - _print_step( - "PR #{} is no longer open (per gh pr view); skipping " - "fan-out spawn.".format(pr) - ) + max_workers = min(10, len(pr_numbers)) + + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + results = executor.map(_check_pr_state_worker, pr_numbers) + + for pr, still_open, exc in results: + if exc is not None: + _print_step( + "Could not confirm PR #{} open state ({}); keeping it in the " + "fan-out set.".format(pr, exc) + ) + kept.append(pr) + elif still_open: + kept.append(pr) + else: + _print_step( + "PR #{} is no longer open (per gh pr view); skipping " + "fan-out spawn.".format(pr) + ) return kept @@ -605,9 +623,7 @@ def _fan_out_all_prs( if args.fan_out_log_dir: log_root = os.path.abspath(os.path.expanduser(args.fan_out_log_dir)) else: - log_root = os.path.join( - os.path.dirname(script_path), ".ralph-logs", "fan-out" - ) + log_root = os.path.join(os.path.dirname(script_path), ".ralph-logs", "fan-out") os.makedirs(log_root, exist_ok=True) stuck_timeout = max(60, args.fan_out_stuck_timeout_seconds) respawn_backoff = max(1, args.fan_out_respawn_backoff_seconds) @@ -700,9 +716,7 @@ def _request_reload(_signum, _frame): reason = "ordinary exit" _print_step( "PR #{} loop exited with code {} ({}); respawning after " - "{}s (log: {})".format( - pr, rc, reason, backoff_for_pr, log_path - ) + "{}s (log: {})".format(pr, rc, reason, backoff_for_pr, log_path) ) try: with open(log_path, "ab", buffering=0) as marker: @@ -796,15 +810,11 @@ def _request_reload(_signum, _frame): # ordinary-exit and stuck-timeout branches above, which both # rewrite pending_backoff to ``respawn_backoff``. _print_step( - "Respawned PR #{} pid={} (log: {})".format( - pr, proc.pid, log_path - ) + "Respawned PR #{} pid={} (log: {})".format(pr, proc.pid, log_path) ) finally: if reload_requested["flag"]: - _print_step( - "Reload requested via SIGHUP; re-exec'ing supervisor" - ) + _print_step("Reload requested via SIGHUP; re-exec'ing supervisor") for pr, (proc, _log_path, log_handle, _spawned_at) in list( children.items() ): @@ -822,9 +832,7 @@ def _request_reload(_signum, _frame): signal.signal(signal.SIGHUP, previous_hup) sys.stdout.flush() sys.stderr.flush() - os.execv( - sys.executable, [sys.executable, script_path] + sys.argv[1:] - ) + os.execv(sys.executable, [sys.executable, script_path] + sys.argv[1:]) for pr, (proc, _log_path, log_handle, _spawned_at) in list(children.items()): try: proc.terminate() @@ -854,9 +862,7 @@ def _request_reload(_signum, _frame): return 0 -def _resolve_target_directories( - raw_dirs: List[str], recursive: bool -) -> List[str]: +def _resolve_target_directories(raw_dirs: List[str], recursive: bool) -> List[str]: """Expand the user-supplied directory args into concrete repo paths. - If ``recursive`` is set, each input path is treated as a parent and we @@ -878,9 +884,7 @@ def _push(path: str) -> None: path = os.path.abspath(os.path.expanduser(raw)) if not os.path.isdir(path): raise CommandError( - "Target directory does not exist or is not a directory: {}".format( - raw - ) + "Target directory does not exist or is not a directory: {}".format(raw) ) if not recursive: _push(path) @@ -933,7 +937,10 @@ def _fan_out_across_directories( base_args.append(token) procs: List[Tuple[str, subprocess.Popen, str, Any]] = [] for target_dir in target_dirs: - slug = re.sub(r"[^A-Za-z0-9._-]+", "-", os.path.basename(target_dir.rstrip("/"))) or "repo" + slug = ( + re.sub(r"[^A-Za-z0-9._-]+", "-", os.path.basename(target_dir.rstrip("/"))) + or "repo" + ) log_path = os.path.join(log_root, "{}.log".format(slug)) log_handle = open(log_path, "ab", buffering=0) cmd = [sys.executable, script_path] + base_args + [target_dir] @@ -1127,9 +1134,7 @@ def _handle_shutdown(signum, _frame): ) pr_target = str(pr_number) - _print_step( - "Using PR #{} {}".format(pr_number, pr_data.get("url", "")) - ) + _print_step("Using PR #{} {}".format(pr_number, pr_data.get("url", ""))) _mark_pr_needs_review(pr_target) if not args.skip_rebase: _print_step("Initial rebase before review/fix loop") @@ -1345,9 +1350,7 @@ def _handle_shutdown(signum, _frame): if not args.skip_merge: fresh_pr_data = _pr_view(str(pr_number)) - fresh_branch = _validate_pr_metadata( - fresh_pr_data, pr_number, args.base - ) + fresh_branch = _validate_pr_metadata(fresh_pr_data, pr_number, args.base) if fresh_branch != branch: raise CommandError( "PR #{} head branch changed from '{}' to '{}' during the run.".format(