From a49b15b7207dca35bc88dfc696ac8622f0e80cbf Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 10 Oct 2026 14:46:18 +0200 Subject: [PATCH] unseen: bound the per-repo Python parse phase (--parse-deadline) Git subprocess calls were deadline-bounded (Git.command_timeout uses the remaining per-repo budget), but a pure-Python parse between git calls (repository_changes -> history/blame/diff on a giant file) had no deadline: when it hung, no further git call fired the git deadline, so the worker stalled indefinitely (observed >720s with zero git processes). An external watchdog was papering over it at ~20 min/stall. Add extract.process_deadline(seconds): a SIGALRM/setitimer wall-clock alarm installed on the worker's main thread that raises ExtractionTimeout("wallclock_timeout"), with no-op guards (None/<=0, off the main thread, or no SIGALRM support) and strict timer-cancel + handler-restore in finally. Wrap the whole extract_repository body in it so the timeout propagates to the existing `except source.Rejected` and journals as a clean rejection, not a worker failure. Budget = --extract-deadline (git, 240s) + --parse-deadline (new, 120s) = 360s total ceiling, so git-bound repos keep their full budget (yield unchanged) while a hung parse is rejected well before any external watchdog. Record parse_deadline_seconds in the report. 295 unseen tests pass (incl. timeout + handler-restore + no-op cases), ruff clean. Co-Authored-By: Claude Opus 4.8 --- benchmarks/unseen/draft.py | 23 ++++++++--- benchmarks/unseen/extract.py | 41 +++++++++++++++++++ tests/test_unseen_draft.py | 76 +++++++++++++++++++++++++++++++++++- tests/test_unseen_extract.py | 68 ++++++++++++++++++++++++++++++++ 4 files changed, 201 insertions(+), 7 deletions(-) diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py index 598b978..699aacc 100644 --- a/benchmarks/unseen/draft.py +++ b/benchmarks/unseen/draft.py @@ -144,6 +144,7 @@ class Clones: runner=None, limit_bytes=500 * 1024 * 1024, deadline_seconds: float | None = None, + parse_deadline_seconds: float | None = 120.0, extract_workers: int = 6, ): if extract_workers < 1: @@ -153,6 +154,7 @@ class Clones: self.runner = runner or subprocess.run self.limit_bytes = limit_bytes self.deadline_seconds = deadline_seconds + self.parse_deadline_seconds = parse_deadline_seconds self.counts = Counter() @contextmanager @@ -318,12 +320,15 @@ def decide(row, github, osv, used, selected_networks=()): def extract_repository(candidate, github, clones, fixes, seed): before = dict(clones.counts) + deadline = getattr(clones, "deadline_seconds", None) + total = (deadline or 0) + (getattr(clones, "parse_deadline_seconds", None) or 0) if deadline is not None else None try: - with clones.open(candidate.url) as git: - git.run("fetch", "--quiet", "--filter=blob:none", "origin", candidate.fix) - changes, exclusions, proof = repository_changes(git, candidate, fixes, seed=seed) - entry = {**asdict(candidate), "changes": changes, "instrument": proof} - result = {"entry": entry, "ordinary_exclusions": exclusions} + with extract.process_deadline(total): + with clones.open(candidate.url) as git: + git.run("fetch", "--quiet", "--filter=blob:none", "origin", candidate.fix) + changes, exclusions, proof = repository_changes(git, candidate, fixes, seed=seed) + entry = {**asdict(candidate), "changes": changes, "instrument": proof} + result = {"entry": entry, "ordinary_exclusions": exclusions} except source.Rejected as exc: result = {"rejected": str(exc), "ordinary_exclusions": getattr(exc, "counts", {})} except eligibility.RepositoryNotFound: @@ -517,6 +522,7 @@ def run( # Operational budget only: changing it must not invalidate journal resume inputs. report["extract_workers"] = workers report["extract_deadline_seconds"] = getattr(clones, "deadline_seconds", None) + report["parse_deadline_seconds"] = getattr(clones, "parse_deadline_seconds", None) counts, rejected, ordinary_counts = Counter(), Counter(), Counter() checkpoint = None try: @@ -637,6 +643,12 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--cache", type=Path) parser.add_argument("--max-hours", type=float, help="Stop between units after this many hours per invocation") parser.add_argument("--extract-deadline", type=float, default=240.0, help="Extraction wall-clock budget per repository in seconds") + parser.add_argument( + "--parse-deadline", + type=float, + default=120.0, + help="Extra wall-clock budget in seconds for the Python parse phase beyond --extract-deadline", + ) parser.add_argument("--extract-workers", type=int, default=6, help="Maximum extraction processes, capped to CPU cores (default: 6)") parser.add_argument("--pool-size", type=int, default=300, help="Repositories to accept (default: 300)") parser.add_argument("--slices", type=int, default=3, help="Equal-sized slices (default: 3)") @@ -661,6 +673,7 @@ def main(argv: list[str] | None = None) -> int: args.root, args.cache or results_root() / "unseen-draft", deadline_seconds=args.extract_deadline, + parse_deadline_seconds=args.parse_deadline, extract_workers=args.extract_workers, ) stores = { diff --git a/benchmarks/unseen/extract.py b/benchmarks/unseen/extract.py index b87a781..6b51800 100644 --- a/benchmarks/unseen/extract.py +++ b/benchmarks/unseen/extract.py @@ -10,8 +10,11 @@ from __future__ import annotations import ast import json import re +import signal import subprocess +import threading import time +from contextlib import contextmanager from pathlib import Path from .source import Candidate, Rejected @@ -33,6 +36,42 @@ class ExtractionTimeout(Rejected): """A slow repository is skipped without aborting the draft.""" +@contextmanager +def process_deadline(seconds: float | None): + """Bound Python work in a worker's main thread; cancel the alarm on exit.""" + if ( + seconds is None + or seconds <= 0 + or threading.current_thread() is not threading.main_thread() + or not all(hasattr(signal, name) for name in ("SIGALRM", "ITIMER_REAL", "setitimer")) + ): + yield + return + + def expired(signum, frame): + raise ExtractionTimeout("wallclock_timeout") + + previous = signal.getsignal(signal.SIGALRM) + installed = False + try: + try: + signal.signal(signal.SIGALRM, expired) + installed = True + signal.setitimer(signal.ITIMER_REAL, seconds) + except ValueError: + # Unsupported signal operations must not prevent extraction. + if installed: + signal.signal(signal.SIGALRM, previous) + installed = False + yield + finally: + if installed: + try: + signal.setitimer(signal.ITIMER_REAL, 0) + finally: + signal.signal(signal.SIGALRM, previous) + + class Git: def __init__(self, path: Path, *, runner=None, limit_bytes: int | None = None, deadline_seconds: float | None = None, _now=None): self._now = _now or time.monotonic @@ -66,6 +105,8 @@ class Git: ) except subprocess.TimeoutExpired: raise ExtractionTimeout("extraction_timeout") from None + except ExtractionTimeout: + raise except Exception: raise InstrumentFailure("git_launch_failed") from None if done.returncode not in allowed: diff --git a/tests/test_unseen_draft.py b/tests/test_unseen_draft.py index b0696d7..89912b4 100644 --- a/tests/test_unseen_draft.py +++ b/tests/test_unseen_draft.py @@ -7,6 +7,7 @@ import os import pickle import subprocess import sys +import time import tomllib from collections import Counter from contextlib import contextmanager @@ -111,6 +112,7 @@ def test_default_clone_job_is_picklable(tmp_path): assert local.runner is subprocess.run assert local.cache == clones.cache and local.limit_bytes == clones.limit_bytes assert local.deadline_seconds == 12.5 and local.extract_workers == 6 + assert local.parse_deadline_seconds == 120.0 assert local.counts == {} and fixes == {candidate.fix} and seed == 7 @@ -449,11 +451,12 @@ def test_systemic_zero_source_is_instrument_failure_with_recorded_rejections(tmp @pytest.mark.parametrize( - "deadline_args,deadline,workers", [([], 240.0, 6), (["--extract-deadline", "12.5", "--extract-workers", "2"], 12.5, 2)] + "deadline_args,deadline,parse_deadline,workers", + [([], 240.0, 120.0, 6), (["--extract-deadline", "12.5", "--parse-deadline", "3.5", "--extract-workers", "2"], 12.5, 3.5, 2)], ) @pytest.mark.parametrize("pool_args,pool_size,slices", [([], 300, 3), (["--pool-size", "100", "--slices", "1"], 100, 1)]) def test_cli_uses_environment_and_injected_stores( - tmp_path, monkeypatch, capsys, deadline_args, deadline, workers, pool_args, pool_size, slices + tmp_path, monkeypatch, capsys, deadline_args, deadline, parse_deadline, workers, pool_args, pool_size, slices ): import json @@ -480,6 +483,7 @@ def test_cli_uses_environment_and_injected_stores( assert d.main(args + deadline_args + pool_args) == 1 output = capsys.readouterr().out assert json.loads(output)["extract_deadline_seconds"] == deadline + assert json.loads(output)["parse_deadline_seconds"] == parse_deadline assert json.loads(output)["extract_workers"] == min(workers, os.cpu_count() or 4) assert "instrument_failure" in output and "fake-secret" not in output and "fixture/" not in output assert seen[-1] == "fake-secret" @@ -547,10 +551,78 @@ def test_deadline_is_reported_without_changing_resume_inputs(tmp_path): ) assert d.run(**kwargs)["status"] == "stopped" clones.deadline_seconds = 12.5 + clones.parse_deadline_seconds = 3.5 assert d.inputs(used, 1, set(used.sources), clones) == before report = d.run(**kwargs) assert report["status"] == "stopped" assert report["extract_deadline_seconds"] == 12.5 + assert report["parse_deadline_seconds"] == 3.5 + + +@pytest.mark.parametrize("workers", [1, 2]) +def test_python_parse_timeout_is_journaled_and_next_repository_finishes(tmp_path, monkeypatch, workers): + api, used, clones = FakeAPI(2), e.UsedSet(), FakeClones() + clones.deadline_seconds = 0.05 + clones.parse_deadline_seconds = 0.15 + ordered = s.ordered_advisories(list(api.advisories()), 1) + slow = ordered[0]["references"][0].split("/commit/")[0] + changes = entries()[0]["changes"] + + def parse(git, candidate, fixes, *, seed): + # No further git command: only the process alarm can reject this work. + time.sleep(0.4 if candidate.url == slow else 0.1) + return changes, {}, {"bytes": 12, "files": 1} + + monkeypatch.setattr(d, "repository_changes", parse) + journal = tmp_path / "journal" + report = d.run( + tmp_path / "repo", + github=api, + osv=FakeOSV(), + clones=clones, + used=used, + seed=1, + known_empty=set(used.sources), + journal_path=journal, + extract_workers=workers, + ) + assert report["status"] == "insufficient_repositories" + assert report["rejections"] == {"wallclock_timeout": 1} + assert report["repositories"] == 1 + assert clones.counts["clones"] == 2 + checkpoint = d.Journal(journal) + with checkpoint.open(d.inputs(used, 1, set(used.sources), clones)): + results = list(checkpoint.rows["extraction"].values()) + assert len(results) == 2 + rejected = next(row for row in results if "rejected" in row) + assert rejected == {"rejected": "wallclock_timeout", "ordinary_exclusions": {}, "counts": {"clones": 1}} + accepted = next(row for row in results if "entry" in row) + assert accepted["entry"]["changes"] == changes + assert accepted["entry"]["instrument"] == {"bytes": 12, "files": 1} + + +@pytest.mark.parametrize("deadline,parse_budget,total", [(None, 120, None), (0.1, 0.2, 0.3), (0.1, None, 0.1)]) +def test_repository_process_deadline_budget(monkeypatch, deadline, parse_budget, total): + clones = FakeClones() + clones.deadline_seconds, clones.parse_deadline_seconds = deadline, parse_budget + active = [] + + @contextmanager + def deadline_context(seconds): + assert seconds == pytest.approx(total) if total is not None else seconds is None + active.append(True) + yield + active.pop() + + def parse(*args, **kwargs): + assert active == [True] + return [], {}, {} + + monkeypatch.setattr(d.extract, "process_deadline", deadline_context) + monkeypatch.setattr(d, "repository_changes", parse) + candidate = s.candidate(next(FakeAPI(1).advisories()), FakeAPI(1), FakeOSV()) + assert "entry" in d.extract_repository(candidate, None, clones, set(), 1) + assert not active def test_repository_changes_excludes_all_advisory_sites(monkeypatch): diff --git a/tests/test_unseen_extract.py b/tests/test_unseen_extract.py index fa884d9..888de12 100644 --- a/tests/test_unseen_extract.py +++ b/tests/test_unseen_extract.py @@ -1,8 +1,11 @@ """Offline extraction: an injected local Git client over a synthetic temporary history.""" import importlib +import signal import subprocess import sys +import time +from concurrent.futures import ThreadPoolExecutor from dataclasses import replace from pathlib import Path @@ -13,6 +16,71 @@ e = importlib.import_module("benchmarks.unseen.extract") s = importlib.import_module("benchmarks.unseen.source") +@pytest.mark.parametrize("slow", [False, True]) +def test_process_deadline_restores_alarm(slow): + original = signal.getsignal(signal.SIGALRM) + assert signal.getitimer(signal.ITIMER_REAL) == (0, 0) + if slow: + with pytest.raises(e.ExtractionTimeout, match="^wallclock_timeout$"), e.process_deadline(0.05): + time.sleep(0.2) + else: + with e.process_deadline(0.2): + assert signal.getitimer(signal.ITIMER_REAL)[0] > 0 + assert signal.getsignal(signal.SIGALRM) == original + assert signal.getitimer(signal.ITIMER_REAL) == (0, 0) + + +@pytest.mark.parametrize("seconds", [None, 0, -1]) +def test_process_deadline_disabled(seconds, monkeypatch): + monkeypatch.setattr(signal, "signal", lambda *args: pytest.fail("disabled deadline installed a handler")) + monkeypatch.setattr(signal, "setitimer", lambda *args: pytest.fail("disabled deadline armed a timer")) + with e.process_deadline(seconds): + pass + + +def test_process_deadline_off_main_thread(monkeypatch): + monkeypatch.setattr(signal, "signal", lambda *args: pytest.fail("thread installed a handler")) + monkeypatch.setattr(signal, "setitimer", lambda *args: pytest.fail("thread armed a timer")) + + def work(): + with e.process_deadline(0.01): + time.sleep(0.02) + return "finished" + + with ThreadPoolExecutor(max_workers=1) as pool: + assert pool.submit(work).result() == "finished" + + +@pytest.mark.parametrize("operation", ["signal", "setitimer"]) +def test_process_deadline_unsupported_operation(operation, monkeypatch): + original = signal.getsignal(signal.SIGALRM) + + def unsupported(*args): + raise ValueError("unsupported") + + monkeypatch.setattr(signal, operation, unsupported) + with e.process_deadline(0.01): + time.sleep(0.02) + assert signal.getsignal(signal.SIGALRM) == original + assert signal.getitimer(signal.ITIMER_REAL) == (0, 0) + + +def test_process_deadline_does_not_swallow_body_value_error(): + original = signal.getsignal(signal.SIGALRM) + with pytest.raises(ValueError, match="body failure"), e.process_deadline(0.2): + raise ValueError("body failure") + assert signal.getsignal(signal.SIGALRM) == original + assert signal.getitimer(signal.ITIMER_REAL) == (0, 0) + + +def test_process_deadline_inside_git_remains_rejection(tmp_path): + def runner(*args, **kwargs): + time.sleep(0.2) + + with pytest.raises(e.ExtractionTimeout, match="^wallclock_timeout$"), e.process_deadline(0.05): + e.Git(tmp_path, runner=runner).run("log") + + @pytest.fixture def history(tmp_path): calls = []