From b074cf7f8916f97dc22bada5af6b855dcaf0ba0c Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 8 Oct 2026 13:22:22 +0200 Subject: [PATCH] unseen: bound per-repo extraction time; a slow repo skips, never aborts A pathologically slow repository (observed: tensorzero, 183s to clone, then `git blame -C` past the 180s command limit) raised subprocess.TimeoutExpired, which Git.run mislabeled as a fatal InstrumentFailure("git_launch_failed") and aborted the whole 300-repo draft. The same slowness stalled throughput (a 54-minute no-progress gap). Give each clone a per-repo wall-clock budget (--extract-deadline, default 240s, plumbed Clones -> Git, counting the initial clone). A timeout -- a command exceeding the remaining budget, or the budget elapsing -- raises ExtractionTimeout(Rejected), which the existing rejection handler turns into an "extraction_timeout" skip so the draw moves to the next candidate. Genuine launch failures and non-zero git exits stay fatal; timeouts and launch failures are now distinct. The deadline is recorded on the report only, not the journal inputs, so changing it does not refuse resume. 181 unseen tests pass, ruff clean. Co-Authored-By: Claude Opus 4.8 --- benchmarks/unseen/draft.py | 14 +++++-- benchmarks/unseen/extract.py | 23 ++++++++++- tests/test_unseen_draft.py | 75 +++++++++++++++++++++++++++++++++++- tests/test_unseen_extract.py | 38 ++++++++++++++++++ 4 files changed, 142 insertions(+), 8 deletions(-) diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py index 46c28cd..8ff5e0a 100644 --- a/benchmarks/unseen/draft.py +++ b/benchmarks/unseen/draft.py @@ -123,10 +123,11 @@ def write_draft(root: Path, assigned: list[dict], *, seed: int) -> tuple[Path, P class Clones: """One disposable blobless clone at a time; no name in cache paths or subprocess output.""" - def __init__(self, root: Path, cache: Path, *, runner=None, limit_bytes=500 * 1024 * 1024): + def __init__(self, root: Path, cache: Path, *, runner=None, limit_bytes=500 * 1024 * 1024, deadline_seconds: float | None = None): self.cache = external(root, cache) self.runner = runner or subprocess.run self.limit_bytes = limit_bytes + self.deadline_seconds = deadline_seconds self.counts = Counter() @contextmanager @@ -135,7 +136,7 @@ class Clones: started = time.monotonic() with tempfile.TemporaryDirectory(prefix="clone-", dir=self.cache) as directory: path = Path(directory) - git = Git(path, runner=self.runner, limit_bytes=self.limit_bytes) + git = Git(path, runner=self.runner, limit_bytes=self.limit_bytes, deadline_seconds=self.deadline_seconds) try: # Full commit graph is needed for temporal thirds and OSV brackets; blobs # are fetched lazily. This is a partial clone, not a truncated history. @@ -145,7 +146,7 @@ class Clones: stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, - timeout=180, + timeout=git.command_timeout(), env={**os.environ, "GIT_TERMINAL_PROMPT": "0"}, ) if done.returncode: @@ -154,6 +155,8 @@ class Clones: if git.disk_bytes() > self.limit_bytes: raise InstrumentFailure("clone_size_cap") yield git + except subprocess.TimeoutExpired: + raise extract.ExtractionTimeout("extraction_timeout") from None finally: size = git.disk_bytes() self.counts["clone_bytes"] += size @@ -331,6 +334,8 @@ def run( ) -> dict: started = time.monotonic() report = {"status": "sourcing", "seed": seed, "candidates": {}, "rejections": {}, "ordinary_exclusions": {}} + # Operational budget only: changing it must not invalidate journal resume inputs. + report["extract_deadline_seconds"] = getattr(clones, "deadline_seconds", None) counts, rejected, ordinary_counts = Counter(), Counter(), Counter() checkpoint = None try: @@ -440,6 +445,7 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--root", type=Path, default=Path.cwd()) 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("--seed", type=int, default=20260601) parser.add_argument("--local-store") parser.add_argument("--s3-store", default="s3://") @@ -451,7 +457,7 @@ def main(argv: list[str] | None = None) -> int: # Refuse an existing draft before doing any network work. if any((args.root / p).exists() for p in ("benchmarks/unseen/draft-p1.toml", "benchmarks/unseen/private/draft-p1.toml")): raise FileExistsError("draft_exists") - clones = Clones(args.root, args.cache or results_root() / "unseen-draft") + clones = Clones(args.root, args.cache or results_root() / "unseen-draft", deadline_seconds=args.extract_deadline) stores = { "local": memory.open_store(args.local_store or "file://" + str(results_root() / "memory"), read_only=True), "s3": memory.open_store(args.s3_store, read_only=True), diff --git a/benchmarks/unseen/extract.py b/benchmarks/unseen/extract.py index e61c474..b87a781 100644 --- a/benchmarks/unseen/extract.py +++ b/benchmarks/unseen/extract.py @@ -11,6 +11,7 @@ import ast import json import re import subprocess +import time from pathlib import Path from .source import Candidate, Rejected @@ -28,8 +29,15 @@ class InstrumentFailure(ValueError): """A failed read or command must never masquerade as an empty history.""" +class ExtractionTimeout(Rejected): + """A slow repository is skipped without aborting the draft.""" + + class Git: - def __init__(self, path: Path, *, runner=None, limit_bytes: int | None = None): + 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 + self.started = self._now() + self.deadline_seconds = deadline_seconds self.path = path self.runner = runner or subprocess.run self.bytes_read = 0 @@ -37,16 +45,27 @@ class Git: self.last_blob_bytes = 0 self.limit_bytes = limit_bytes + def command_timeout(self) -> float: + if self.deadline_seconds is None: + return 180 + remaining = self.deadline_seconds - (self._now() - self.started) + if remaining <= 0: + raise ExtractionTimeout("extraction_timeout") + return min(180, remaining) + def run(self, *args: str, allowed=(0,)) -> str: + timeout = self.command_timeout() try: done = self.runner( ["git", "-c", "core.quotePath=false", "-C", str(self.path), *args], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - timeout=180, + timeout=timeout, check=False, ) + except subprocess.TimeoutExpired: + raise ExtractionTimeout("extraction_timeout") from None 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 f59db83..dec9c14 100644 --- a/tests/test_unseen_draft.py +++ b/tests/test_unseen_draft.py @@ -224,7 +224,10 @@ def test_zero_source_is_instrument_failure_not_rejection(tmp_path, monkeypatch): assert report["status"] == "instrument_failure" and not report["rejections"] -def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys): +@pytest.mark.parametrize("deadline_args,deadline", [([], 240.0), (["--extract-deadline", "12.5"], 12.5)]) +def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys, deadline_args, deadline): + import json + from openultrasast import config from openultrasast.plane import memory @@ -238,13 +241,81 @@ def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys) args = ["--root", str(tmp_path), "--cache", str(tmp_path.parent / "clone-cache")] for label in e.SOURCES: args += ["--known-empty", label] - assert d.main(args) == 1 + assert d.main(args + deadline_args) == 1 output = capsys.readouterr().out + assert json.loads(output)["extract_deadline_seconds"] == deadline assert "instrument_failure" in output and "fake-secret" not in output and "fixture/" not in output assert seen[-1] == "fake-secret" assert all(kwargs == {"read_only": True} for _, kwargs in seen[:-1]) +@pytest.mark.parametrize("timeout_stage", ["clone", "fetch"]) +def test_timeout_skips_repository_and_draft_continues(tmp_path, monkeypatch, timeout_stage): + import subprocess + + clone_count = 0 + + def runner(cmd, **kwargs): + nonlocal clone_count + if "clone" in cmd: + clone_count += 1 + if clone_count == 1 and timeout_stage in cmd: + raise subprocess.TimeoutExpired(cmd, kwargs["timeout"]) + return subprocess.CompletedProcess(cmd, 0, b"", b"") + + monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (entries()[0]["changes"], {}, {"bytes": 1, "files": 1})) + clones = d.Clones(tmp_path / "repo", tmp_path / "cache", runner=runner, deadline_seconds=240.0) + used = e.UsedSet() + report = d.run(tmp_path / "repo", github=FakeAPI(2), osv=FakeOSV(), clones=clones, used=used, seed=1, known_empty=set(used.sources)) + assert report["status"] == "insufficient_repositories" + assert report["rejections"] == {"extraction_timeout": 1} + assert report["repositories"] == 1 and clone_count == 2 + assert not list(clones.cache.glob("clone-*")) + + +def test_clone_time_counts_and_each_repository_gets_fresh_deadline(tmp_path, monkeypatch): + import subprocess + + now, timeouts = [0.0], [] + monkeypatch.setattr(d.time, "monotonic", lambda: now[0]) + + def runner(cmd, **kwargs): + timeouts.append(kwargs["timeout"]) + now[0] += 7.0 if "clone" in cmd else 1.0 + return subprocess.CompletedProcess(cmd, 0, b"", b"") + + clones = d.Clones(tmp_path / "repo", tmp_path / "cache", runner=runner, deadline_seconds=10.0) + for _ in range(2): + with clones.open("https://github.com/fixture/project") as git: + git.run("log") + now[0] += 3.0 + with pytest.raises(d.extract.ExtractionTimeout, match="^extraction_timeout$"): + git.run("log") + assert timeouts == [10.0, 3.0, 10.0, 3.0] + + +def test_deadline_is_reported_without_changing_resume_inputs(tmp_path): + clones = d.Clones(tmp_path / "repo", tmp_path / "cache", deadline_seconds=240.0) + used = e.UsedSet() + before = d.inputs(used, 1, set(used.sources), clones) + kwargs = dict( + root=tmp_path / "repo", + github=FakeAPI(), + osv=FakeOSV(), + clones=clones, + used=used, + seed=1, + known_empty=set(used.sources), + max_hours=0, + ) + assert d.run(**kwargs)["status"] == "stopped" + clones.deadline_seconds = 12.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 + + def test_repository_changes_excludes_all_advisory_sites(monkeypatch): calls = [] diff --git a/tests/test_unseen_extract.py b/tests/test_unseen_extract.py index b84e2e7..314fd17 100644 --- a/tests/test_unseen_extract.py +++ b/tests/test_unseen_extract.py @@ -118,6 +118,44 @@ def test_git_failures_are_sanitized_and_not_empty(tmp_path): e.Git(tmp_path, runner=failed).run("log") +@pytest.mark.parametrize("deadline,timeout", [(None, 180), (240.0, 180), (12.5, 12.5)]) +def test_git_timeout_is_rejection(tmp_path, deadline, timeout): + def runner(cmd, **kwargs): + assert kwargs["timeout"] == timeout + raise subprocess.TimeoutExpired(cmd, kwargs["timeout"]) + + git = e.Git(tmp_path, runner=runner, deadline_seconds=deadline, _now=lambda: 0.0) + with pytest.raises(e.ExtractionTimeout, match="^extraction_timeout$") as caught: + git.run("log") + assert isinstance(caught.value, s.Rejected) + assert not isinstance(caught.value, e.InstrumentFailure) + + +def test_git_cumulative_deadline_prevents_launch(tmp_path): + now, timeouts = [100.0], [] + + def runner(cmd, **kwargs): + timeouts.append(kwargs["timeout"]) + now[0] += 4.0 + return subprocess.CompletedProcess(cmd, 0, b"read", b"") + + git = e.Git(tmp_path, runner=runner, deadline_seconds=10.0, _now=lambda: now[0]) + assert git.run("log") == "read" + assert git.run("log") == "read" + now[0] += 3.0 # Time between commands counts as well. + with pytest.raises(e.ExtractionTimeout, match="^extraction_timeout$"): + git.run("log") + assert timeouts == [10.0, 6.0] + + +def test_git_launch_failure_stays_fatal(tmp_path): + def runner(cmd, **kwargs): + raise FileNotFoundError("private executable path") + + with pytest.raises(e.InstrumentFailure, match="^git_launch_failed$"): + e.Git(tmp_path, runner=runner, deadline_seconds=240.0).run("log") + + def test_walkback_is_bounded_and_file_cap_falls_back(history): git, _, candidate, _, _, parent, _ = history