diff --git a/benchmarks/learn/engine_task_entry.py b/benchmarks/learn/engine_task_entry.py index 965de87..ac4aca8 100644 --- a/benchmarks/learn/engine_task_entry.py +++ b/benchmarks/learn/engine_task_entry.py @@ -15,6 +15,10 @@ import urllib.request from pathlib import Path +class TaskEntryError(RuntimeError): + """A task-entry diagnostic containing only locally controlled text.""" + + def transfer(request, destination=None, *, readiness_budget=None): """Wait for actor egress on startup; bound later transfers to three attempts.""" deadline = time.monotonic() + readiness_budget if readiness_budget is not None else None @@ -23,7 +27,7 @@ def transfer(request, destination=None, *, readiness_budget=None): while True: remaining = deadline - time.monotonic() if deadline is not None else 60 if remaining <= 0: - raise RuntimeError("egress not open after 60 s") + raise TaskEntryError("egress not open after 60 s") attempt += 1 try: with urllib.request.urlopen(request, timeout=min(60, remaining)) as response: @@ -43,7 +47,7 @@ def transfer(request, destination=None, *, readiness_budget=None): if deadline is not None: remaining = deadline - time.monotonic() if remaining <= 0: - raise RuntimeError("egress not open after 60 s") from None + raise TaskEntryError("egress not open after 60 s") from None time.sleep(min(delay, remaining)) delay = min(delay * 2, 8) else: @@ -92,20 +96,26 @@ def main(*, work_root=Path("/tmp"), worker_script=None, readiness_budget=60): failed = False output = work_root / "out" result = output / "result.json" + step = "prepare output" try: output.mkdir(parents=True, exist_ok=True) result.unlink(missing_ok=True) questions = work_root / "questions.json" + step = "GET questions" download(os.environ["QUESTIONS_URL"], questions, readiness_budget=readiness_budget) + step = "parse questions" pin = json.loads(questions.read_text()) if not isinstance(pin, dict): pin = {} raise ValueError("questions object must contain a prepared pin") source = work_root / "source.tar" + step = "GET source" download(os.environ["SOURCE_URL"], source) + step = "unpack source" case = work_root / "case" case.mkdir(parents=True, exist_ok=True) unpack(source, case) + step = "configure worker" deadline = int(os.environ.get("DEADLINE", "900")) command = [ sys.executable, @@ -123,30 +133,39 @@ def main(*, work_root=Path("/tmp"), worker_script=None, readiness_budget=60): "--question-deadline", os.environ.get("QUESTION_DEADLINE", "120"), ] + step = "run worker" completed = subprocess.run(command, timeout=deadline + 5, check=False) if completed.returncode: - raise RuntimeError(f"worker exited {completed.returncode}") + raise TaskEntryError(f"worker exited {completed.returncode}") + step = "read worker result" if not result.is_file() or not json.loads(result.read_text()).get("done"): - raise RuntimeError("worker did not write a completed result") + raise TaskEntryError("worker did not write a completed result") except Exception as exc: failed = True # Avoid including exception text containing a presigned URL in durable output. - reason = str(exc) if isinstance(exc, (RuntimeError, ValueError)) else f"task entry failed: {type(exc).__name__}" + if isinstance(exc, urllib.error.HTTPError): + detail = f"HTTP {exc.code}" + else: + detail = str(exc) if isinstance(exc, TaskEntryError) else type(exc).__name__ + reason = f"{step}: {detail}" try: output.mkdir(parents=True, exist_ok=True) result.write_text(json.dumps(failure_record(pin, reason)) + "\n") except Exception: print("task entry could not write failure result", file=sys.stderr) + step = "pack result" try: archive = work_root / "result.tar" with tarfile.open(archive, "w") as tar: for path in sorted(output.iterdir()): tar.add(path, arcname=path.name) data = archive.read_bytes() + step = "PUT result" request = urllib.request.Request(os.environ["RESULT_URL"], data=data, method="PUT", headers={"Content-Type": "application/x-tar"}) transfer(request, readiness_budget=readiness_budget if failed else None) - except Exception: - print("task entry could not upload result; dispatcher deadline remains authoritative", file=sys.stderr) + except Exception as exc: + detail = f"HTTP {exc.code}" if isinstance(exc, urllib.error.HTTPError) else type(exc).__name__ + print(f"{step}: {detail}; dispatcher deadline remains authoritative", file=sys.stderr) return 0 diff --git a/benchmarks/learn/engine_trace_ax.py b/benchmarks/learn/engine_trace_ax.py index 3a3e539..1c16cf2 100644 --- a/benchmarks/learn/engine_trace_ax.py +++ b/benchmarks/learn/engine_trace_ax.py @@ -24,7 +24,7 @@ import yaml from engine_trace_k8s import pack from openultrasast.config import load_dotenv -from openultrasast.plane.memory import S3Store, open_store +from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store class EgressError(ValueError): @@ -220,7 +220,7 @@ class AX: name = f"ousast-engine-{slug or 'pin'}-{digest}" key_id = hashlib.sha256(str(pin_id).encode()).hexdigest() prefix = f"engine-queue/{self.run}/{key_id}/" - keys = [prefix + suffix for suffix in ("source.tar", "questions.json", "result.tar")] + keys = [prefix + suffix for suffix in (f"source{PRESIGNED_ARCHIVE_SUFFIX}", "questions.json", f"result{PRESIGNED_ARCHIVE_SUFFIX}")] expires = timedelta(seconds=self.args.deadline + 600) worker_ip = None submitted = False diff --git a/benchmarks/learn/engine_trace_k8s.py b/benchmarks/learn/engine_trace_k8s.py index 528e31b..812e2ac 100644 --- a/benchmarks/learn/engine_trace_k8s.py +++ b/benchmarks/learn/engine_trace_k8s.py @@ -12,7 +12,7 @@ import time import uuid from datetime import timedelta -from openultrasast.plane.memory import S3Store, open_store +from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store IMAGE = "ghcr.io/norandom/openultrasast:2.0.1" BOOTSTRAP = r""" @@ -164,7 +164,7 @@ class Kubernetes: name = "engine-" + uuid.uuid4().hex[:24] pin_key = hashlib.sha256(json.dumps([pin.get("repo"), pin.get("pin")]).encode()).hexdigest() prefix = f"engine-jobs/{self.run}/{pin_key}/" - keys = [prefix + file for file in ("source.tar", "analyzer.tar", "result.tar")] + keys = [prefix + name + PRESIGNED_ARCHIVE_SUFFIX for name in ("source", "analyzer", "result")] expiry = timedelta(seconds=self.args.deadline + 900) node = None try: diff --git a/benchmarks/learn/engine_trace_queue.py b/benchmarks/learn/engine_trace_queue.py index 20d1d27..5243bdf 100644 --- a/benchmarks/learn/engine_trace_queue.py +++ b/benchmarks/learn/engine_trace_queue.py @@ -9,6 +9,8 @@ import uuid from engine_worker_service import PENDING, client_from_env, pack, put, read, result_key, status +from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX + class Dispatcher: def __init__(self, args, client=None, *, pause=time.sleep, clock=time.monotonic): @@ -23,7 +25,7 @@ class Dispatcher: id=pin_id, task_id=task_id, pin=pin, - input=f"engine-queue/inputs/{self.run}/{pin_id}/source.tar", + input=f"engine-queue/inputs/{self.run}/{pin_id}/source{PRESIGNED_ARCHIVE_SUFFIX}", image=self.args.image, deadline=self.args.deadline, question_deadline=self.args.question_deadline, diff --git a/src/openultrasast/plane/memory.py b/src/openultrasast/plane/memory.py index e593e59..f601024 100644 --- a/src/openultrasast/plane/memory.py +++ b/src/openultrasast/plane/memory.py @@ -586,6 +586,10 @@ class ObjectClient(Protocol): _VERIFIED: set[str] = set() PROBE_KEY = "_probe/select.jsonl" +# Live finding: files.because-security.com denies presigned GETs ending in .tar, .gz, +# .zip or .bin (HTTP 403 AccessDenied). Use .dat for archive objects; bytes stay tar. +PRESIGNED_ARCHIVE_SUFFIX = ".dat" # engine executors only; the plane keeps output.tar + RUNS_PREFIX = "runs/" RUNS_RULE_ID = "ousast-runs-expiry" RUNS_EXPIRE_DAYS = 30 diff --git a/tests/test_ax_executor.py b/tests/test_ax_executor.py index 9ae4030..978493a 100644 --- a/tests/test_ax_executor.py +++ b/tests/test_ax_executor.py @@ -113,6 +113,10 @@ def test_ax_executor_success_while_running_and_cleanup(setup): assert set(env) == {"SOURCE_URL", "QUESTIONS_URL", "RESULT_URL", "DEADLINE", "QUESTION_DEADLINE"} assert len(json.dumps(manifest["spec"]["env"]).encode()) < 32768 assert "AWS" not in calls[0]["input"] + assert env["SOURCE_URL"].split("?")[0].endswith("/source.dat") + assert env["RESULT_URL"].split("?")[0].endswith("/result.dat") + assert env["QUESTIONS_URL"].split("?")[0].endswith("/questions.json") + assert {key.rsplit("/", 1)[-1] for key in store.deleted} == {"source.dat", "questions.json", "result.dat"} assert store.expiries == [604] * 3 assert len(store.deleted) == 3 and not store.objects diff --git a/tests/test_engine_queue.py b/tests/test_engine_queue.py index 9ac55af..7a406ce 100644 --- a/tests/test_engine_queue.py +++ b/tests/test_engine_queue.py @@ -53,7 +53,7 @@ def task(): run="run", id="pin", task_id="run-pin", - input="engine-queue/inputs/run/pin/source.tar", + input="engine-queue/inputs/run/pin/source.dat", deadline=10, question_deadline=2, pin={"repo": "owner/repo", "pin": "commit", "units": []}, @@ -120,6 +120,7 @@ def test_engine_queue_dispatch_roundtrip_and_timeout(tmp_path): pin = {"repo": "owner/repo", "pin": "sha", "units": [unit], "questions": []} def execute(client, item): + assert item["input"].endswith("/source.dat") unpacked = tmp_path / "unpacked" unpacked.mkdir() service.unpack(client.get(item["input"])[0], unpacked) diff --git a/tests/test_engine_task_entry.py b/tests/test_engine_task_entry.py index 58c00fc..e5bba32 100644 --- a/tests/test_engine_task_entry.py +++ b/tests/test_engine_task_entry.py @@ -37,7 +37,7 @@ def test_engine_cluster_heap_flags(monkeypatch): assert worker.MeasuredBackend(30, 10, lambda: None)._heap_flag() == "-J-Xmx2560m" -@pytest.mark.parametrize("readiness", ["ready", "delayed", "closed"]) +@pytest.mark.parametrize("readiness", ["ready", "delayed", "closed", "source_denied"]) @pytest.mark.parametrize("failure", [False, True]) @pytest.mark.parametrize("transport", ["memory", "loopback"]) def test_task_entry_http_roundtrip(tmp_path, monkeypatch, failure, transport, readiness): @@ -57,6 +57,10 @@ def test_task_entry_http_roundtrip(tmp_path, monkeypatch, failure, transport, re class Handler(BaseHTTPRequestHandler): def do_GET(self): gets.append((self.path, time.monotonic() - started)) + if readiness == "source_denied" and self.path == "/source": + self.send_response(403) + self.end_headers() + return if readiness == "closed" or (readiness == "delayed" and gets[-1][1] < 3): self.send_response(403 if len(gets) % 2 else 503) self.end_headers() @@ -148,7 +152,13 @@ def test_task_entry_http_roundtrip(tmp_path, monkeypatch, failure, transport, re assert result["done"] if readiness == "closed": assert result["status"] == "failed" - assert result["reason"] == "egress not open after 60 s" + assert result["reason"] == "GET questions: egress not open after 60 s" + assert json.loads((tmp_path / "work/out/result.json").read_text()) == result + elif readiness == "source_denied": + assert result["status"] == "failed" + assert result["reason"] == "GET source: HTTP 403" + assert result["units"][0]["reason"] == "GET source: HTTP 403" + assert base not in json.dumps(result) assert json.loads((tmp_path / "work/out/result.json").read_text()) == result elif failure: assert result["units"][0]["status"] == "failed" diff --git a/tests/test_engine_trace.py b/tests/test_engine_trace.py index 80afc24..1385490 100644 --- a/tests/test_engine_trace.py +++ b/tests/test_engine_trace.py @@ -1265,6 +1265,8 @@ def test_kubernetes_host_pin_success_failure_and_cleanup(tmp_path, monkeypatch, manifest = json.loads(kwargs["input"]) if manifest["kind"] == "Secret": assert manifest["metadata"]["ownerReferences"][0]["uid"] == "job-uid" + for variable, filename in [("SOURCE_URL", "source.dat"), ("ANALYZER_URL", "analyzer.dat"), ("RESULT_URL", "result.dat")]: + assert manifest["stringData"][variable].split("?")[0].endswith("/" + filename) if manifest["kind"] == "Job": stdout = json.dumps({"metadata": {"uid": "job-uid"}}) env = manifest["spec"]["template"]["spec"]["containers"][0]["env"] @@ -1276,10 +1278,10 @@ def test_kubernetes_host_pin_success_failure_and_cleanup(tmp_path, monkeypatch, entry = tarfile.TarInfo("result.json") entry.size = len(result_bytes) archive.addfile(entry, io.BytesIO(result_bytes)) - source_key = next(key for key in objects if key.endswith("source.tar")) + source_key = next(key for key in objects if key.endswith("source.dat")) with tarfile.open(fileobj=io.BytesIO(objects[source_key])) as archive: assert archive.extractfile("app.py").read() == b"def run(): pass" - objects[source_key.replace("source.tar", "result.tar")] = stream.getvalue() + objects[source_key.replace("source.dat", "result.dat")] = stream.getvalue() if action[:2] == ["get", "job"]: stdout = json.dumps( {"status": {"conditions": [{"type": "Failed", "status": "True", "reason": "DeadlineExceeded"}]}}