Merge unseen-parse-deadline: wall-clock deadline for the Python parse phase

Co-Authored-By: Claude Opus 4.8 <[email protected]>

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,7 +320,10 @@ 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 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)
@@ -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 = {

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:

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):

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 = []