diff --git a/benchmarks/learn/engine_trace_worker.py b/benchmarks/learn/engine_trace_worker.py index e84c8df..eb4b136 100644 --- a/benchmarks/learn/engine_trace_worker.py +++ b/benchmarks/learn/engine_trace_worker.py @@ -36,13 +36,20 @@ def safe_path(root, name): return path +FRONTENDS = {"php2cpg", "pysrc2cpg", "jssrc2cpg", "javasrc2cpg", "c2cpg", "joern-parse"} +MIN_CPG_BYTES = 1024 + + def instrument_failure(proof): if not proof.get("files") or not proof.get("bytes"): return "instrument read zero files or zero bytes" - if not proof.get("cpg_bytes"): - return "instrument produced no CPG bytes" + if proof.get("cpg_bytes", 0) <= MIN_CPG_BYTES: + return "frontend wrote no CPG" for stage in proof.get("jvm", []): - if stage["seconds"] < 5: + if stage["command"] in FRONTENDS: + if stage.get("cpg_bytes", 0) <= MIN_CPG_BYTES: + return "frontend wrote no CPG" + elif stage["command"] == "joern" and stage["seconds"] < 5: return f"implausibly fast JVM step: {stage['command']} in {stage['seconds']:.3f}s (<5s)" if not proof.get("jvm"): return "no JVM step recorded" @@ -73,7 +80,9 @@ class MeasuredBackend(JoernBackend): started = time.monotonic() end = min(self.deadline, started + timeout) question_end = end - jvm = Path(command[0]).name != "php" + binary = Path(command[0]).name + frontend = binary in FRONTENDS + query = binary == "joern" and "--script" in command output = bytearray() pending = b"" process = subprocess.Popen( @@ -117,9 +126,18 @@ class MeasuredBackend(JoernBackend): process.wait(timeout=5) process.stdout.close() elapsed = time.monotonic() - started - if jvm: - self.stages.append({"command": Path(command[0]).name, "seconds": elapsed, "exit": process.returncode}) - if elapsed < 5: + if frontend or query: + stage = {"command": binary, "seconds": elapsed, "exit": process.returncode} + self.stages.append(stage) + if frontend: + # Capture the exact frontend output before build failure cleanup or overlays. + output_arg = next((i + 1 for i, arg in enumerate(command[:-1]) if arg in {"-o", "--output"}), None) + cpg_path = (Path(cwd or Path.cwd()) / command[output_arg]).resolve() if output_arg is not None else None + stage["cpg_path"] = str(cpg_path) if cpg_path is not None else None + stage["cpg_bytes"] = cpg_path.stat().st_size if cpg_path is not None and cpg_path.is_file() else 0 + if stage["cpg_bytes"] <= MIN_CPG_BYTES: + self.fatal = "frontend wrote no CPG" + elif elapsed < 5: self.fatal = f"implausibly fast JVM step: {command[0]} in {elapsed:.3f}s (<5s)" self.checkpoint() return subprocess.CompletedProcess(command, process.returncode if not self.fatal else 1, output.decode(errors="replace"), "") @@ -132,7 +150,7 @@ def unit_record(unit, status, reason="", **extra): "reason": reason, "witness_rows": [], "traces": [], - "instrument": {"files": [], "bytes": 0, "cpg_bytes": 0, "jvm": []}, + "instrument": {"files": [], "bytes": 0, "cpg_path": None, "cpg_bytes": 0, "jvm": []}, "questions": {}, "questions_asked": [], "questions_completed": [], @@ -141,7 +159,7 @@ def unit_record(unit, status, reason="", **extra): def worker(pin, root, out, deadline, question_deadline): - proof = {"files": [], "bytes": 0, "cpg_bytes": 0, "jvm": []} + proof = {"files": [], "bytes": 0, "cpg_path": None, "cpg_bytes": 0, "jvm": []} record = { **pin, "done": False, @@ -152,6 +170,10 @@ def worker(pin, root, out, deadline, question_deadline): def checkpoint(): # Checkpoints retain raw completed rows even if the outer container deadline fires. # They remain timeout until the finished JVM's instrument proof is validated. + for stage in backend.stages: + if "cpg_path" in stage: + proof["cpg_path"] = stage["cpg_path"] + proof["cpg_bytes"] = stage["cpg_bytes"] for row in record["units"]: rid = row["unit"] row["instrument"] = proof @@ -187,7 +209,9 @@ def worker(pin, root, out, deadline, question_deadline): cpg = backend.build(root, language=language) if cpg is None: raise ValueError(backend.fatal or backend.last_failure or ("pin deadline" if backend.timed_out else "CPG build failed")) - proof["cpg_bytes"] = sum(p.stat().st_size for p in cpg.cpg_path.parent.rglob("*.bin")) + if proof["cpg_path"] is None: + proof["cpg_path"] = str(cpg.cpg_path) + proof["cpg_bytes"] = cpg.cpg_path.stat().st_size if cpg.cpg_path.is_file() else 0 failure = instrument_failure(proof) if failure: raise ValueError(failure) @@ -196,6 +220,8 @@ def worker(pin, root, out, deadline, question_deadline): answers = cpg.run_batch_once("taint", requests) or {} record["census"] = answers.get("__census__", []) failure = backend.fatal or instrument_failure(proof) + if not backend.asked and not backend.timed_out: + failure = failure or "query produced no markers" census = record["census"] if census and int(census[0].get("files", 0)) <= 0: failure = failure or "query census read zero files" @@ -230,6 +256,8 @@ def worker(pin, root, out, deadline, question_deadline): status, reason = "failed", failure or "labelled file was dropped or absent from query census" elif rows is None or shard_incomplete: status, reason = ("timeout" if backend.timed_out else "failed"), "question did not complete" + elif not rows and rid not in backend.asked: + status, reason = "failed", "query produced no markers" else: flows = [r for r in rows if isinstance(r, dict) and "kind" not in r and "sink" in r] if any(not row.get("trace") for row in flows): diff --git a/tests/test_engine_trace.py b/tests/test_engine_trace.py index 35352b8..9bd9687 100644 --- a/tests/test_engine_trace.py +++ b/tests/test_engine_trace.py @@ -6,6 +6,7 @@ import importlib.util import json import subprocess import sys +from dataclasses import replace from pathlib import Path import pytest @@ -146,13 +147,13 @@ class FakeBackend: seen = [] def __init__(self, deadline, question_deadline, checkpoint): - self.stages = [{"command": "joern-parse", "seconds": self.seconds, "exit": 0}] + self.stages = [{"command": "joern-parse", "seconds": self.seconds, "exit": 0, "cpg_bytes": 2048}] self.asked = [] self.answers = {} def build(self, root, language): graph = root / "graph.bin" - graph.write_bytes(b"CPG bytes") + graph.write_bytes(b"x" * 2048) def batch(query, requests): self.seen.append(requests) @@ -270,8 +271,8 @@ def test_question_watchdog_retains_completed_rows_without_joern(tmp_path): assert backend.asked == ["a", "b"] assert backend.answers == {"a": []} assert done.returncode != 0 - assert backend.stages[0]["seconds"] < 5 - assert "implausibly fast" in backend.fatal + assert backend.stages == [] + assert not backend.fatal def test_pair_mapping_handles_harvest_normalized_crlf(tmp_path): @@ -293,7 +294,7 @@ def test_missing_batch_answers_are_timeout_not_empty(tmp_path, monkeypatch): class TimeoutBackend(FakeBackend): def build(self, root, language): graph = root / "graph.bin" - graph.write_bytes(b"CPG bytes") + graph.write_bytes(b"x" * 2048) def batch(query, requests): self.timed_out = True @@ -469,3 +470,100 @@ def test_trace_resume_identity_includes_selection_even_if_support_unchanged(tmp_ trace.save_pin(tmp_path / trace.pin_name(first), {**first, "done": True}) assert trace.make_plan([unit()], tmp_path) == [] assert len(trace.make_plan([unit()], tmp_path, include_typescript=True)) == 1 + + +@pytest.mark.parametrize("binary", sorted(worker.FRONTENDS)) +def test_fast_frontend_validates_exact_output(tmp_path, binary): + frontend = tmp_path / binary + frontend.write_text( + f"#!{sys.executable}\n" + "import pathlib, sys, time\n" + "source = pathlib.Path(sys.argv[1])\n" + "print('read bytes:', len(source.read_bytes()), flush=True)\n" + "time.sleep(2 if pathlib.Path(sys.argv[0]).name == 'php2cpg' else 0)\n" + "pathlib.Path(sys.argv[3]).write_bytes(b'x' * 34000)\n" + ) + frontend.chmod(0o755) + (tmp_path / "source.php").write_text("= 2 + assert stage["cpg_path"] == str(tmp_path / "out.graph") + assert stage["cpg_bytes"] == 34000 + proof = {"files": ["source.php"], "bytes": 19, **stage, "jvm": backend.stages} + assert worker.instrument_failure(proof) == "" + + +@pytest.mark.parametrize("size", [None, 0, 1024]) +def test_frontend_without_sufficient_output_fails(tmp_path, size): + frontend = tmp_path / "php2cpg" + frontend.write_text( + f"#!{sys.executable}\nimport pathlib, sys\n" + + (f"pathlib.Path(sys.argv[2]).write_bytes(b'x' * {size})\n" if size is not None else "") + ) + frontend.chmod(0o755) + # A different non-empty CPG must not mask the missing requested output. + (tmp_path / "unrelated.bin").write_bytes(b"x" * 34000) + backend = worker.MeasuredBackend(30, 10, lambda: None) + done = backend._run([str(frontend), "-o", "out.bin"], timeout=10, cwd=tmp_path) + assert done.returncode == 1 + assert backend.stages[0]["exit"] == 0 + assert backend.stages[0]["cpg_bytes"] == (size or 0) + assert backend.fatal == "frontend wrote no CPG" + + +def test_fast_joern_script_is_failed_launch(tmp_path): + joern = tmp_path / "joern" + joern.write_text(f"#!{sys.executable}\n") + joern.chmod(0o755) + backend = worker.MeasuredBackend(30, 10, lambda: None) + done = backend._run([str(joern), "--script", "query.sc"], timeout=10, cwd=tmp_path) + assert done.returncode == 1 + assert backend.stages[0]["exit"] == 0 + assert "implausibly fast JVM step" in backend.fatal + + +@pytest.mark.parametrize("markers", [[], ["a"], ["a", "b"]]) +def test_empty_answers_require_each_question_marker(tmp_path, monkeypatch, markers): + class EmptyBackend(FakeBackend): + def build(self, root, language): + cpg = super().build(root, language) + batch = cpg.run_batch_once + + def empty_batch(query, requests): + answers = batch(query, requests) + self.asked = markers + return {**answers, **dict.fromkeys(requests, [])} + + return replace(cpg, run_batch_once=empty_batch) + + result = run_fake(tmp_path, monkeypatch, EmptyBackend) + for row in result["units"]: + assert row["status"] == ("asked-nothing" if row["unit"] in markers else "failed") + assert row["reason"] == ("" if row["unit"] in markers else "query produced no markers") + + +def test_frontend_measurement_survives_failed_build_cleanup(tmp_path, monkeypatch): + class CleanupBackend(worker.MeasuredBackend): + def build(self, root, language): + frontend = root / "php2cpg" + frontend.write_text( + f"#!{sys.executable}\nimport pathlib, sys\npathlib.Path(sys.argv[2]).write_bytes(b'x' * 34000)\nsys.exit(1)\n" + ) + frontend.chmod(0o755) + self._run([str(frontend), "-o", "actual.graph"], timeout=10, cwd=root) + (root / "actual.graph").unlink() + return None + + result = run_fake(tmp_path, monkeypatch, CleanupBackend) + assert all(row["status"] == "failed" for row in result["units"]) + assert result["instrument"]["cpg_path"] == str(tmp_path / "actual.graph") + assert result["instrument"]["cpg_bytes"] == 34000 + assert not (tmp_path / "actual.graph").exists()