diff --git a/benchmarks/unseen/bank.py b/benchmarks/unseen/bank.py new file mode 100644 index 0000000..ec9cc6a --- /dev/null +++ b/benchmarks/unseen/bank.py @@ -0,0 +1,160 @@ +"""Durable, content-addressed draw decisions; object I/O stays on the coordinator.""" + +from __future__ import annotations + +import hashlib +import json +import os +from collections.abc import Mapping +from datetime import UTC, datetime +from urllib.parse import urlencode, urlsplit + +CORPUS_KEYS = ("CORPUS_S3_ENDPOINT", "CORPUS_S3_BUCKET", "CORPUS_AWS_ACCESS_KEY_ID", "CORPUS_AWS_SECRET_ACCESS_KEY") + + +def corpus_settings(environ: Mapping[str, str] | None = None) -> dict: + """Read corpus-only credentials, naming missing settings without exposing values.""" + if environ is None: + from openultrasast.config import load_dotenv + + load_dotenv() + environ = os.environ + missing = [key for key in CORPUS_KEYS if not environ.get(key)] + if missing: + raise ValueError(f"Corpus bank needs {', '.join(missing)} in .env or the environment") + endpoint = environ["CORPUS_S3_ENDPOINT"].strip() + try: + parsed = urlsplit(endpoint) + valid_endpoint = parsed.scheme in ("http", "https") and bool(parsed.netloc) + except ValueError: + valid_endpoint = False + if not valid_endpoint: + raise ValueError("CORPUS_S3_ENDPOINT must be a URL with a scheme (https://host[:port])") + settings = { + "endpoint": endpoint, + "bucket": environ["CORPUS_S3_BUCKET"], + "access_key": environ["CORPUS_AWS_ACCESS_KEY_ID"], + "secret_key": environ["CORPUS_AWS_SECRET_ACCESS_KEY"], + } + if environ.get("CORPUS_S3_REGION"): + settings["region"] = environ["CORPUS_S3_REGION"] + return settings + + +class CorpusObjects: + """Plain S3 put/get/list, with no bucket administration or capability probes.""" + + def __init__(self, *, endpoint: str, bucket: str, access_key: str, secret_key: str, region: str | None = None): + import boto3 + from botocore.config import Config + + self.bucket = bucket + self.client = boto3.client( + "s3", + endpoint_url=endpoint, + aws_access_key_id=access_key, + aws_secret_access_key=secret_key, + region_name=region or "us-east-1", + # RustFS/MinIO at an IP endpoint needs path-style addressing + # (virtual-host bucket. does not resolve) and SigV4. Verified + # against ousast-corpus @ 10.0.0.54:9001. + config=Config( + signature_version="s3v4", + s3={"addressing_style": "path"}, + retries={"total_max_attempts": 1}, + ), + ) + + def get(self, key: str) -> tuple[bytes, str] | None: + try: + result = self.client.get_object(Bucket=self.bucket, Key=key) + except Exception as error: + if getattr(error, "response", {}).get("Error", {}).get("Code") in {"NoSuchKey", "404", "NotFound"}: + return None + raise + body = result["Body"] + try: + return body.read(), result.get("VersionId") or result.get("ETag", "") + finally: + body.close() + + def put(self, key: str, data: bytes, labels: dict) -> None: + self.client.put_object( + Bucket=self.bucket, + Key=key, + Body=data, + ContentType="application/json", + Tagging=urlencode(labels), + ) + + def keys(self, prefix: str) -> list[str]: + pages = self.client.get_paginator("list_objects_v2").paginate(Bucket=self.bucket, Prefix=prefix) + return [item["Key"] for page in pages for item in page.get("Contents", [])] + + +def _retry(operation, *args, **kwargs): + """Retry one object operation once; a second failure is fatal to durability.""" + try: + return operation(*args, **kwargs) + except Exception: + return operation(*args, **kwargs) + + +class CorpusBank: + def __init__(self, client, *, seed: int, prefix: str = "unseen"): + self.client = client + self.seed = seed + self.decisions_prefix = f"{prefix}/{seed}/decisions/" + + @staticmethod + def key_for(repository: str, fix: str) -> str: + return hashlib.sha256(f"{repository}\n{fix}".encode()).hexdigest() + + def bank(self, repository: str, fix: str, result: dict, *, contract: str) -> None: + decision = { + "repository": repository, + "fix": fix, + "decision": "accept" if "entry" in result else "reject", + "reason": result.get("rejected"), + "entry": result.get("entry"), + "ordinary_exclusions": result.get("ordinary_exclusions", {}), + "counts": result.get("counts", {}), + "contract": contract, + "seed": self.seed, + } + key = self.decisions_prefix + self.key_for(repository, fix) + ".json" + existing = _retry(self.client.get, key) + if existing is not None: + previous = json.loads(existing[0]) + if {field: value for field, value in previous.items() if field != "ts"} == decision: + return # Preserve the first timestamp and avoid an unnecessary object version. + decision["ts"] = datetime.now(UTC).isoformat() + data = json.dumps(decision, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + _retry(self.client.put, key, data, labels={"decision": decision["decision"], "contract": contract}) + + def decided(self) -> dict[str, dict]: + decisions = {} + for key in sorted(self.client.keys(self.decisions_prefix)): + stored = _retry(self.client.get, key) + if stored is None: + raise ValueError("corpus_decision_missing") + decision = json.loads(stored[0]) + digest = self.key_for(decision["repository"], decision["fix"]) + if key != self.decisions_prefix + digest + ".json" or decision["seed"] != self.seed: + raise ValueError("corpus_decision_identity") + if decision["decision"] not in {"accept", "reject"}: + raise ValueError("corpus_decision_invalid") + decisions[digest] = decision + return decisions + + +def split_decided(decisions: Mapping[str, dict], *, contract: str) -> tuple[list[dict], set[str]]: + """Split reusable results, sorted independently of bucket listing order. + + The coordinator merges entries into the full advisory population before seeded + ordering: shuffling only the accepted subset would change the draw's order. + """ + accepted = [row["entry"] for row in decisions.values() if row["decision"] == "accept"] + accepted.sort(key=lambda entry: (entry.get("advisory", ""), entry["repository"], entry["fix"])) + reusable = {key for key, row in decisions.items() if row["decision"] == "accept" or row["contract"] == contract} + return accepted, reusable diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py index 0770126..d90ea09 100644 --- a/benchmarks/unseen/draft.py +++ b/benchmarks/unseen/draft.py @@ -2,7 +2,7 @@ All network and git operations are injected at the run() boundary. Only counts, digests and fixed failure codes go to stdout; identity-bearing output stays in the -private journal outside the repository. No population guard is run here. +private journal or corpus bucket. No population guard is run here. """ from __future__ import annotations @@ -20,12 +20,13 @@ import time from collections import Counter, defaultdict, deque from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait from concurrent.futures.process import BrokenProcessPool -from contextlib import contextmanager +from contextlib import ExitStack, contextmanager from dataclasses import asdict from pathlib import Path from . import eligibility, extract, source -from .extract import Git, InstrumentFailure +from .bank import CorpusBank, CorpusObjects, corpus_settings, split_decided +from .extract import EXTRACTION_CONTRACT, Git, InstrumentFailure from .journal import Journal, ResolvedGitHub, ResumeMismatch, Stopped, atomic, digest, external @@ -367,7 +368,21 @@ def finish_license(result, github): @contextmanager -def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, entries, workers, pool_size): +def ordered_extractions( + rows, + api, + osv, + used, + clones, + fixes, + seed, + checkpoint, + entries, + workers, + pool_size, + bank: CorpusBank | None = None, + decided=None, +): """Bounded in-flight work, completion journaling, and seed-order acceptance. Only running jobs count against `workers`; completed results wait in the @@ -376,6 +391,8 @@ def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, e Graceful STOP/guard/target exits drain started work into the journal for resume; a hard interrupt may lose unjournaled work but cannot change accepted order. """ + decided = dict(decided or {}) + identities = {} pending = deque() # Seed-order acceptance cursor; contains no futures/results. in_flight = {} # Future -> extraction key, independent of the cursor. # Fork shares the coordinator's memory copy-on-write. Jobs receive only the @@ -383,8 +400,22 @@ def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, e context = multiprocessing.get_context("fork") pool = ProcessPoolExecutor(max_workers=workers, mp_context=context) + def remember(repository, fix, result): + bank.bank(repository, fix, result, contract=EXTRACTION_CONTRACT) + # Reuse decisions made during this invocation too: another advisory for + # the same pair must not overwrite an earlier accept with a rejection. + decided[bank.key_for(repository, fix)] = { + "repository": repository, + "fix": fix, + "decision": "accept" if "entry" in result else "reject", + "entry": result.get("entry"), + "reason": result.get("rejected"), + "ordinary_exclusions": result.get("ordinary_exclusions", {}), + "counts": result.get("counts", {}), + } + def collect(future): - key = in_flight[future] + key = in_flight.pop(future) try: result = future.result() except BrokenProcessPool: @@ -401,8 +432,10 @@ def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, e else: clones.counts[name] += value result = finish_license(result, api) + if bank is not None: + repository, fix = identities[key] + remember(repository, fix, result) checkpoint.put("extraction", key, result) - del in_flight[future] def iterate(): nonlocal pool @@ -428,6 +461,53 @@ def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, e exhausted = True break key = row["ghsa_id"] + identity = None + if bank is not None: + links = source.fix_links(row) + if len(links) == 1: + repository, fix = next(iter(links)) + try: + resolved = eligibility.resolve(repository, api) + repository = resolved.full_name + except eligibility.RepositoryNotFound: + resolved = None + except ValueError as exc: + if str(exc) != "http_failure": + raise + resolved = None # decide() records this as a resilient skip. + identity = (repository, fix) + saved = decided.get(bank.key_for(*identity)) + if saved is None and len(fix) < 40: + matches = [ + value for value in decided.values() if value["repository"] == repository and value["fix"].startswith(fix) + ] + if len(matches) == 1: + saved = matches[0] + identity = (repository, saved["fix"]) + if saved is None: + # A duplicate advisory can arrive before its pair's + # worker finishes. Share that pending result as well. + matching = [key for key, pair in identities.items() if pair[0] == repository and pair[1].startswith(fix)] + if len(matching) == 1: + pending.append((row, {"network": resolved.network if resolved else repository}, matching[0])) + break + if saved is not None: + decision = {"network": resolved.network if resolved is not None else repository} + blocked = eligibility.rejection(resolved, used, set()) if resolved is not None else None + if blocked: + decision["rejected"] = "used_" + blocked + pending.append((row, decision, None)) + else: + result = {"ordinary_exclusions": saved["ordinary_exclusions"], "counts": saved["counts"]} + if saved["decision"] == "accept": + result["entry"] = saved["entry"] + else: + result["rejected"] = saved["reason"] + extraction_key = bank.key_for(*identity) + if extraction_key not in checkpoint.rows["extraction"]: + checkpoint.put("extraction", extraction_key, result) + pending.append((row, decision, extraction_key)) + break if key not in checkpoint.rows["eligibility"]: checkpoint.put("eligibility", key, decide(row, api, osv, used)) decision = checkpoint.rows["eligibility"][key] @@ -435,7 +515,10 @@ def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, e if "candidate" in decision: candidate = source.Candidate(**decision["candidate"]) extraction_key = digest({"repository": candidate.repository, "advisory": key, "fix": candidate.fix}) - if extraction_key not in checkpoint.rows["extraction"]: + if bank is not None: + extraction_key = bank.key_for(candidate.repository, candidate.fix) + identities[extraction_key] = (candidate.repository, candidate.fix) + if extraction_key not in checkpoint.rows["extraction"] and extraction_key not in in_flight.values(): checkpoint.boundary() local = copy.copy(clones) local.counts = Counter() @@ -449,6 +532,8 @@ def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, e pool = ProcessPoolExecutor(max_workers=workers, mp_context=context) future = pool.submit(extract_repository, *args) in_flight[future] = extraction_key + elif bank is not None and identity is not None: + remember(*identity, decision) pending.append((row, decision, extraction_key)) # Finalize cached decisions promptly (especially on resume), and # service completed jobs even during a long run of decide skips. @@ -482,7 +567,7 @@ def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, e checkpoint.progress("stopped", reason=stopped) -def publish(journal, assigned, seed): +def publish(journal, assigned, seed, *, destination=None): """Idempotent publication from completed journal data, never inside the checkout.""" public, private = [], [] for row in assigned: @@ -492,7 +577,7 @@ def publish(journal, assigned, seed): private.append(row) public.append({"id": row["id"], "sha256": private_digest(row), "slice": row["slice"], "private": True}) for name, rows in (("private-draft-p1.toml", private), ("draft-p1.toml", public)): - path = journal.path / name + path = (destination or journal.path) / name data = manifest(rows, seed).encode() if path.exists(): if path.read_bytes() != data: @@ -516,6 +601,7 @@ def run( extract_workers=None, pool_size=300, slices=3, + bank: CorpusBank | None = None, ) -> dict: validate_pool(pool_size, slices) workers = extract_workers if extract_workers is not None else getattr(clones, "extract_workers", 6) @@ -530,13 +616,25 @@ def run( report["parse_deadline_seconds"] = getattr(clones, "parse_deadline_seconds", None) counts, rejected, ordinary_counts = Counter(), Counter(), Counter() checkpoint = None + scratch = ExitStack() try: + decided = bank.decided() if bank is not None else {} + accepted, reusable = split_decided(decided, contract=EXTRACTION_CONTRACT) + decided = {key: value for key, value in decided.items() if key in reusable} known_empty = set(known_empty or ()) used.report(known_empty) cache = getattr(clones, "cache", root.parent / (root.name + "-cache")) path = external(root, journal_path or cache / "journal") + journal_directory = path + if bank is not None: + # Local records are a disposable cursor, never a cross-run authority. + path.mkdir(parents=True, exist_ok=True, mode=0o700) + journal_directory = Path(scratch.enter_context(tempfile.TemporaryDirectory(prefix="cursor-", dir=path))) checkpoint = Journal( - path, max_hours=max_hours, clock=clock, rate_limit=lambda: getattr(getattr(github, "transport", None), "rate_limit", {}) + journal_directory, + max_hours=max_hours, + clock=clock, + rate_limit=lambda: getattr(getattr(github, "transport", None), "rate_limit", {}), ) with checkpoint.open(inputs(used, seed, known_empty, clones, pool_size=pool_size, slices=slices)): checkpoint.boundary() @@ -544,7 +642,26 @@ def run( used = copy.deepcopy(used) # canonicalization must not mutate the resume input eligibility.canonicalize_used(used, api) report["used_set"] = used.report(known_empty) - rows = enumerate_advisories(api, checkpoint) + try: + rows = enumerate_advisories(api, checkpoint) + except InstrumentFailure as exc: + if str(exc) != "empty_advisories" or not accepted: + raise + rows = [] + if bank is not None: + # Retain accepted entries even if their advisory disappeared upstream. + present = {row["ghsa_id"] for row in rows} + for entry in accepted: + if entry["advisory"] not in present: + rows.append( + { + "ghsa_id": entry["advisory"], + "published_at": "2026-06-02" if entry["post_cutoff"] else "2026-06-01", + "vulnerabilities": [{"package": {"ecosystem": entry["ecosystem"]}}], + "references": [f"https://github.com/{entry['repository']}/commit/{entry['fix']}"], + } + ) + present.add(entry["advisory"]) report["advisories"] = len(rows) fixes_by_repo = advisory_index(rows, api) selected_networks, entries = set(), [] @@ -561,6 +678,8 @@ def run( entries, workers, pool_size, + bank=bank, + decided=decided, ) as outcomes: for row, decision, result in outcomes: counts[source.ecosystem(row)] += 1 @@ -575,7 +694,9 @@ def run( ordinary_counts.update(result["ordinary_exclusions"]) if "rejected" in result: rejected[result["rejected"]] += 1 - instrument_failures += int(result["rejected"].startswith("extract_")) + instrument_failures += int( + result["rejected"].startswith("extract_") or result["rejected"] == "decide_http_failure" + ) else: selected_networks.add(decision["network"]) entries.append(result["entry"]) @@ -605,7 +726,7 @@ def run( elif checkpoint.rows["completion"]["all"] != completion: raise ValueError("journal_completion_mismatch") checkpoint.boundary() - publish(checkpoint, assigned, seed) + publish(checkpoint, assigned, seed, destination=path if bank is not None else None) report.update( { "status": "draft", @@ -629,6 +750,7 @@ def run( report.update({"status": "instrument_failure", "error_type": type(exc).__name__}) return report finally: + scratch.close() report.update({"candidates": dict(counts), "rejections": dict(rejected), "ordinary_exclusions": dict(ordinary_counts)}) report["instrument"] = dict(clones.counts) report["wall_seconds"] = round(time.monotonic() - started, 3) @@ -658,6 +780,7 @@ def main(argv: list[str] | None = None) -> int: 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)") parser.add_argument("--seed", type=int, default=20260601) + parser.add_argument("--bank", nargs="?", const="unseen", default="", help="Corpus bucket prefix; enables durable decisions") parser.add_argument("--local-store") parser.add_argument("--s3-store", default="s3://") parser.add_argument("--pointer-manifest", type=Path, action="append", default=[]) @@ -686,6 +809,7 @@ def main(argv: list[str] | None = None) -> int: "s3": memory.open_store(args.s3_store, read_only=True), } used = eligibility.build_used(args.root, stores=stores, pointer_manifests=args.pointer_manifest) + bank = CorpusBank(CorpusObjects(**corpus_settings()), seed=args.seed, prefix=args.bank) if args.bank else None report = run( args.root, github=source.GitHub(os.environ.get("GH_TOKEN", "")), @@ -698,6 +822,7 @@ def main(argv: list[str] | None = None) -> int: known_empty=set(args.known_empty), journal_path=clones.cache / "journal", max_hours=args.max_hours, + bank=bank, ) except Exception as exc: report = {"status": "instrument_failure", "error_type": type(exc).__name__} diff --git a/benchmarks/unseen/extract.py b/benchmarks/unseen/extract.py index 6b51800..7e16449 100644 --- a/benchmarks/unseen/extract.py +++ b/benchmarks/unseen/extract.py @@ -19,6 +19,9 @@ from pathlib import Path from .source import Candidate, Rejected +# Bump only when accept/reject outcomes or accepted entry bytes change. +EXTRACTION_CONTRACT = "v1" + LANGUAGES = {".py", ".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx", ".php", ".java"} HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", re.M) SHA = re.compile(r"[a-f0-9]{40}") diff --git a/tests/test_unseen_bank.py b/tests/test_unseen_bank.py new file mode 100644 index 0000000..13f9f02 --- /dev/null +++ b/tests/test_unseen_bank.py @@ -0,0 +1,216 @@ +"""Offline decision-bank contract: durability, identity and plain object transport.""" + +import hashlib +import io +import json +import sys +from datetime import datetime +from types import SimpleNamespace + +import pytest + +from benchmarks.unseen.bank import CorpusBank, CorpusObjects, corpus_settings, split_decided + + +class FakeObjects: + def __init__(self): + self.objects = {} + self.labels = {} + self.calls = [] + self.failures = {} + + def call(self, operation): + self.calls.append(operation) + if self.failures.get(operation, 0): + self.failures[operation] -= 1 + raise OSError("offline failure") + + def get(self, key): + self.call("get") + return (self.objects[key], "version") if key in self.objects else None + + def put(self, key, data, labels): + self.call("put") + self.objects[key] = data + self.labels[key] = labels + + def keys(self, prefix): + return [key for key in reversed(self.objects) if key.startswith(prefix)] + + +def settings_env(): + return { + "CORPUS_S3_ENDPOINT": "https://objects.invalid", + "CORPUS_S3_BUCKET": "ousast-corpus", + "CORPUS_AWS_ACCESS_KEY_ID": "private-access", + "CORPUS_AWS_SECRET_ACCESS_KEY": "private-secret", + } + + +@pytest.mark.parametrize("missing", list(settings_env())) +def test_settings_names_missing_key_without_values(missing): + env = settings_env() + del env[missing] + with pytest.raises(ValueError) as error: + corpus_settings(env) + assert missing in str(error.value) + assert all(value not in str(error.value) for value in env.values()) + + +def test_settings_and_optional_region(): + env = settings_env() + assert corpus_settings(env) == { + "endpoint": env["CORPUS_S3_ENDPOINT"], + "bucket": "ousast-corpus", + "access_key": "private-access", + "secret_key": "private-secret", + } + assert corpus_settings({**env, "CORPUS_S3_REGION": "region-1"})["region"] == "region-1" + + +def test_bank_roundtrip_and_idempotent_bytes(): + objects = FakeObjects() + bank = CorpusBank(objects, seed=42) + entry = {"repository": "org/a", "fix": "abc", "advisory": "GHSA-a", "changes": [{"code": "text"}]} + result = {"entry": entry, "ordinary_exclusions": {"merge": 2}, "counts": {"eligible": 13}} + key = bank.key_for("org/a", "abc") + assert key == hashlib.sha256(b"org/a\nabc").hexdigest() + bank.bank("org/a", "abc", result, contract="v1") + original = objects.objects.copy() + bank.bank("org/a", "abc", result, contract="v1") + assert objects.objects == original + assert list(objects.objects) == [f"unseen/42/decisions/{key}.json"] + row = bank.decided()[key] + assert row == { + "repository": "org/a", + "fix": "abc", + "decision": "accept", + "reason": None, + "entry": entry, + "ordinary_exclusions": {"merge": 2}, + "counts": {"eligible": 13}, + "contract": "v1", + "seed": 42, + "ts": row["ts"], + } + assert datetime.fromisoformat(row["ts"]).utcoffset().total_seconds() == 0 + assert objects.labels[next(iter(objects.labels))] == {"decision": "accept", "contract": "v1"} + + +def test_rejection_contract_update_and_split(): + objects = FakeObjects() + bank = CorpusBank(objects, seed=3, prefix="custom") + for repo, advisory in (("z", "GHSA-z"), ("a", "GHSA-a")): + bank.bank(repo, "fix", {"entry": {"repository": repo, "fix": "fix", "advisory": advisory}}, contract="old") + bank.bank("no", "fix", {"rejected": "reason"}, contract="v1") + bank.bank("stale", "fix", {"rejected": "reason"}, contract="old") + accepted, keys = split_decided(bank.decided(), contract="v1") + assert [entry["repository"] for entry in accepted] == ["a", "z"] + assert keys == {bank.key_for(repo, "fix") for repo in ("a", "z", "no")} + row = bank.decided()[bank.key_for("no", "fix")] + assert row["reason"] == "reason" and row["decision"] == "reject" + assert row["entry"] is None and row["counts"] == row["ordinary_exclusions"] == {} + bank.bank("stale", "fix", {"rejected": "different"}, contract="v1") + assert len(objects.objects) == 4 + assert bank.decided()[bank.key_for("stale", "fix")]["contract"] == "v1" + + +@pytest.mark.parametrize("operation", ["get", "put"]) +@pytest.mark.parametrize("failures", [1, 2]) +def test_bank_retries_once_then_surfaces_failure(operation, failures): + objects = FakeObjects() + objects.failures[operation] = failures + bank = CorpusBank(objects, seed=1) + if failures == 1: + bank.bank("repo", "fix", {"rejected": "reason"}, contract="v1") + assert len(objects.objects) == 1 + else: + with pytest.raises(OSError, match="offline failure"): + bank.bank("repo", "fix", {"rejected": "reason"}, contract="v1") + assert objects.calls.count(operation) == 2 + + +def test_decided_retries_get_and_fails_on_missing_listed_object(): + objects = FakeObjects() + bank = CorpusBank(objects, seed=1) + bank.bank("repo", "fix", {"rejected": "reason"}, contract="v1") + objects.failures["get"] = 1 + assert len(bank.decided()) == 1 + objects.get = lambda key: None + with pytest.raises(ValueError, match="missing"): + bank.decided() + + +def test_plain_s3_adapter_without_bucket_probes(monkeypatch): + calls = [] + + class Missing(Exception): + response = {"Error": {"Code": "NoSuchKey"}} + + class S3: + def put_object(self, **kwargs): + calls.append(("put", kwargs)) + + def get_object(self, **kwargs): + calls.append(("get", kwargs)) + if kwargs["Key"] == "missing": + raise Missing() + return {"Body": io.BytesIO(b"data"), "ETag": "tag"} + + def get_paginator(self, name): + assert name == "list_objects_v2" + return SimpleNamespace(paginate=self.pages) + + def pages(self, **kwargs): + calls.append(("list", kwargs)) + return [{"Contents": [{"Key": "p/a"}]}, {}, {"Contents": [{"Key": "p/b"}]}] + + def client(service, **kwargs): + calls.append((service, kwargs)) + return S3() + + monkeypatch.setitem(sys.modules, "boto3", SimpleNamespace(client=client)) + monkeypatch.setitem(sys.modules, "botocore.config", SimpleNamespace(Config=lambda **kwargs: SimpleNamespace(**kwargs))) + adapter = CorpusObjects(**corpus_settings(settings_env())) + adapter.put("p/a", b"data", {"decision": "accept", "contract": "v1"}) + assert adapter.get("p/a") == (b"data", "tag") + assert adapter.get("missing") is None + assert adapter.keys("p/") == ["p/a", "p/b"] + assert calls[0][0] == "s3" + assert calls[0][1]["endpoint_url"] == "https://objects.invalid" + assert calls[0][1]["config"].retries == {"total_max_attempts": 1} + assert calls[1][1]["Bucket"] == "ousast-corpus" + assert calls[1][1]["Body"] == b"data" + assert calls[-1] == ("list", {"Bucket": "ousast-corpus", "Prefix": "p/"}) + + +def test_decided_rejects_invalid_identity(): + objects = FakeObjects() + bank = CorpusBank(objects, seed=1) + bank.bank("repo", "fix", {"rejected": "reason"}, contract="v1") + key = next(iter(objects.objects)) + row = json.loads(objects.objects[key]) + row["repository"] = "other" + objects.objects[key] = json.dumps(row).encode() + with pytest.raises(ValueError, match="identity"): + bank.decided() + + +@pytest.mark.parametrize("endpoint", ["missing-scheme-private", "https://private[", "http://private/secret"]) +def test_invalid_endpoint_does_not_expose_value(endpoint): + with pytest.raises(ValueError) as error: + corpus_settings({**settings_env(), "CORPUS_S3_ENDPOINT": endpoint}) + assert "CORPUS_S3_ENDPOINT" in str(error.value) + assert endpoint not in str(error.value) + + +def test_import_does_not_require_boto3(monkeypatch): + import importlib.util + + from benchmarks.unseen import bank + + monkeypatch.setitem(sys.modules, "boto3", None) + spec = importlib.util.spec_from_file_location("offline_bank", bank.__file__) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + assert module.CorpusBank.key_for("a", "b") == CorpusBank.key_for("a", "b") diff --git a/tests/test_unseen_draft.py b/tests/test_unseen_draft.py index f29d063..975d9fd 100644 --- a/tests/test_unseen_draft.py +++ b/tests/test_unseen_draft.py @@ -1519,3 +1519,301 @@ def test_pool_configuration_change_refuses_resume_before_io(tmp_path, changed): ) assert report["status"] == "resume_refused" and report["changed_parameters"] == list(changed) assert api.calls == 0 and not clones.counts + + +class CorpusFakeObjects: + def __init__(self): + self.objects = {} + self.coordinator = os.getpid() + + def get(self, key): + assert os.getpid() == self.coordinator + return (self.objects[key], "v1") if key in self.objects else None + + def put(self, key, data, labels): + assert os.getpid() == self.coordinator + self.objects[key] = data + + def keys(self, prefix): + assert os.getpid() == self.coordinator + return list(reversed([key for key in self.objects if key.startswith(prefix)])) + + +@pytest.mark.parametrize("workers", [1, 4]) +@pytest.mark.parametrize("short_refs", [False, True]) +def test_bank_resume_stream_contract_seed_order_and_forks(tmp_path, monkeypatch, workers, short_refs): + from benchmarks.unseen.bank import CorpusBank + + monkeypatch.setattr(importlib.import_module("benchmarks.unseen.journal").os, "fsync", lambda fd: None) + ordered = s.ordered_advisories(list(FakeAPI(10).advisories()), 7) + names = [next(iter(s.fix_links(row)))[0] for row in ordered] + ranks = {name: i for i, name in enumerate(names)} + called = multiprocessing.get_context("fork").Array("i", 10) + changes = entries()[0]["changes"] + selected = [] + original_assign = d.assign_slices + + class API(FakeAPI): + def advisories(self): + for row in super().advisories(): + if short_refs: + row["references"] = [row["references"][0][:-33]] + yield row + + def get_commit(self, name, sha): + return super().get_commit(name, "a" * 40) + + def get_repo(self, name): + result = super().get_repo(name) + if name == names[3]: + result["source"] = {"full_name": names[1]} + return result + + def extract_changes(git, candidate, fixes, *, seed): + rank = ranks[candidate.repository] + called[rank] += 1 + if rank == 0: + raise s.Rejected("no_ordinary") + return copy.deepcopy(changes), {"eligible": 40}, {"bytes": 12, "files": 1} + + def assign(rows, **kwargs): + selected.append([row["repository"] for row in rows]) + return original_assign(rows, **kwargs) + + monkeypatch.setattr(d, "repository_changes", extract_changes) + monkeypatch.setattr(d, "assign_slices", assign) + + def run(path, bank=None): + used = e.UsedSet() + return d.run( + tmp_path / "repo", + github=API(10), + osv=FakeOSV(), + clones=FakeClones(), + used=used, + seed=7, + known_empty=set(used.sources), + journal_path=path, + extract_workers=workers, + pool_size=4, + slices=1, + bank=bank, + ) + + full_objects = CorpusFakeObjects() + full_bank = CorpusBank(full_objects, seed=7) + full = run(tmp_path / "full", full_bank) + assert full["status"] == "draft", full + decisions = full_bank.decided() + assert len(decisions) == sum(called) + assert {obj["decision"] for obj in decisions.values()} == {"accept", "reject"} + resumed_objects = CorpusFakeObjects() + resumed_bank = CorpusBank(resumed_objects, seed=7) + # Partial completion can be out of seed order, including a fork that loses + # to an earlier fresh result. Rejections alone are contract sensitive. + for obj in decisions.values(): + rank = ranks[obj["repository"]] + if rank in {0, 2, 3}: + result = {"entry": obj["entry"]} if obj["decision"] == "accept" else {"rejected": obj["reason"]} + result.update(ordinary_exclusions=obj["ordinary_exclusions"], counts=obj["counts"]) + resumed_bank.bank( + obj["repository"], obj["fix"], result, contract=d.EXTRACTION_CONTRACT if rank == 0 else "old-accepted-contract" + ) + resumed_bank.bank(names[1], "a" * 40, {"rejected": "old_failure"}, contract="old") + before = list(called) + resumed = run(tmp_path / "resume", resumed_bank) + assert resumed["status"] == "draft", resumed + assert selected == [[names[i] for i in (1, 2, 4, 5)]] * 2 + assert resumed["manifest_sha256"] == full["manifest_sha256"] + assert all(called[i] == before[i] for i in (0, 2, 3)) + assert called[1] == before[1] + 1 + for filename in ("draft-p1.toml", "private-draft-p1.toml"): + assert (tmp_path / "resume" / filename).read_bytes() == (tmp_path / "full" / filename).read_bytes() + # A corrupt/stale local journal and code fingerprint must not be cross-run truth. + (tmp_path / "resume" / "inputs.json").write_text('{"design_draft":"old"}') + (tmp_path / "resume" / "extraction.jsonl").write_text("torn") + before = list(called) + again = run(tmp_path / "resume", resumed_bank) + assert again["manifest_sha256"] == full["manifest_sha256"] + assert list(called) == before + + +@pytest.mark.parametrize("failure", ["crash", "persistent_put"]) +def test_bank_crash_resume_and_durability_failure(tmp_path, monkeypatch, failure): + from benchmarks.unseen.bank import CorpusBank + + monkeypatch.setattr(importlib.import_module("benchmarks.unseen.journal").os, "fsync", lambda fd: None) + changes = entries()[0]["changes"] + monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (copy.deepcopy(changes), {}, {"bytes": 1, "files": 1})) + + class Objects(CorpusFakeObjects): + interrupted = False + puts = 0 + + def put(self, key, data, labels): + self.puts += 1 + if failure == "persistent_put" and not self.interrupted: + raise OSError("offline simulated bank outage") + super().put(key, data, labels) + if not self.interrupted: + self.interrupted = True + raise KeyboardInterrupt + + objects = Objects() + bank = CorpusBank(objects, seed=7) + + def run(path, bank): + used = e.UsedSet() + return d.run( + tmp_path / "repo", + github=FakeAPI(8), + osv=FakeOSV(), + clones=FakeClones(), + used=used, + seed=7, + known_empty=set(used.sources), + journal_path=path, + extract_workers=1, + pool_size=3, + slices=1, + bank=bank, + ) + + if failure == "crash": + with pytest.raises(KeyboardInterrupt): + run(tmp_path / "resume", bank) + assert len(bank.decided()) == 1 + else: + report = run(tmp_path / "resume", bank) + assert report["status"] == "instrument_failure" + assert objects.puts == 2 + assert not objects.objects + assert not (tmp_path / "resume" / "draft-p1.toml").exists() + objects.interrupted = True + resumed = run(tmp_path / "resume", bank) + full = run(tmp_path / "full", None) + assert resumed["status"] == full["status"] == "draft" + assert resumed["manifest_sha256"] == full["manifest_sha256"] + + +@pytest.mark.parametrize("bank_args,prefix", [([], None), (["--bank"], "unseen"), (["--bank", "custom"], "custom")]) +def test_cli_corpus_bank_is_opt_in(tmp_path, monkeypatch, capsys, bank_args, prefix): + from openultrasast import config + from openultrasast.plane import memory + + objects = CorpusFakeObjects() + monkeypatch.setattr(config, "load_dotenv", lambda *a: None) + monkeypatch.setattr(memory, "open_store", lambda *a, **kw: object()) + monkeypatch.setattr(d.eligibility, "build_used", lambda *a, **kw: e.UsedSet()) + monkeypatch.setattr(d.source, "GitHub", lambda *a: FakeAPI(0)) + monkeypatch.setattr(d.source, "OSV", FakeOSV) + settings_calls = [] + monkeypatch.setattr(d, "corpus_settings", lambda: settings_calls.append(True) or {"bucket": "ousast-corpus"}) + + def make_objects(**kwargs): + assert kwargs == {"bucket": "ousast-corpus"} + return objects + + def run(*args, **kwargs): + bank = kwargs["bank"] + if prefix is None: + assert bank is None + else: + assert bank.client is objects and bank.decisions_prefix == f"{prefix}/7/decisions/" + return {"status": "draft"} + + monkeypatch.setattr(d, "CorpusObjects", make_objects) + monkeypatch.setattr(d, "run", run) + assert d.main(["--root", str(tmp_path / "repo"), "--cache", str(tmp_path / "cache"), "--seed", "7", *bank_args]) == 0 + assert settings_calls == ([] if prefix is None else [True]) + + +def test_bank_accepts_survive_missing_advisories_and_obey_used_networks(tmp_path, monkeypatch): + from benchmarks.unseen.bank import CorpusBank + + objects = CorpusFakeObjects() + bank = CorpusBank(objects, seed=7) + rows = s.ordered_advisories(list(FakeAPI(4).advisories()), 7) + names = [next(iter(s.fix_links(row)))[0] for row in rows] + for row in rows: + candidate = s.candidate(row, FakeAPI(), FakeOSV()) + entry = {**d.asdict(candidate), "changes": entries()[0]["changes"], "instrument": {"bytes": 12, "files": 1}} + bank.bank(candidate.repository, candidate.fix, {"entry": entry}, contract="older") + + class API(FakeAPI): + def get_repo(self, name): + result = super().get_repo(name) + if name == names[2]: + result["source"] = {"full_name": names[1]} + return result + + def get_commit(self, *args): + pytest.fail("banked accepts must not be re-decided") + + used = e.UsedSet() + used.add("pairs", {names[0]}, size=len(names[0])) + selected = [] + original = d.assign_slices + + def assign(rows, **kwargs): + selected.extend(row["repository"] for row in rows) + return original(rows, **kwargs) + + monkeypatch.setattr(d, "assign_slices", assign) + report = d.run( + tmp_path / "repo", + github=API(0), + osv=FakeOSV(), + clones=FakeClones(), + used=used, + seed=7, + known_empty={key for key, names in used.sources.items() if not names}, + journal_path=tmp_path / "journal", + pool_size=2, + slices=1, + bank=bank, + ) + assert report["status"] == "draft", report + assert selected == [names[1], names[3]] + assert report["rejections"]["fork_network"] == 1 + assert len(bank.decided()) == 4 + + +@pytest.mark.parametrize("workers", [1, 4]) +def test_bank_duplicate_advisory_cannot_overwrite_pair_decision(tmp_path, monkeypatch, workers): + from benchmarks.unseen.bank import CorpusBank + + rows = s.ordered_advisories(list(FakeAPI(6).advisories()), 7) + rows[1]["references"] = rows[0]["references"][:] + rows[1]["cwes"] = [] # Would reject if re-decided under this later advisory. + repository, fix = next(iter(s.fix_links(rows[0]))) + + class API(FakeAPI): + def advisories(self): + return iter(copy.deepcopy(rows)) + + changes = entries()[0]["changes"] + monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (copy.deepcopy(changes), {}, {"bytes": 12, "files": 1})) + bank = CorpusBank(CorpusFakeObjects(), seed=7) + reports = [] + for name in ("first", "resumed"): + used = e.UsedSet() + reports.append( + d.run( + tmp_path / "repo", + github=API(), + osv=FakeOSV(), + clones=FakeClones(), + used=used, + known_empty=set(used.sources), + seed=7, + pool_size=3, + slices=1, + extract_workers=workers, + journal_path=tmp_path / name, + bank=bank, + ) + ) + assert all(report["status"] == "draft" for report in reports) + assert reports[0]["manifest_sha256"] == reports[1]["manifest_sha256"] + assert bank.decided()[bank.key_for(repository, fix)]["decision"] == "accept"