From 55bb3c700cd28318f28d48f77f976437d9f7fd72 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 12:43:08 +0800 Subject: [PATCH] fix(benchmark): enforce shared deadlines within Docker archives Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 11 ++ benchmark/edgebench/checkpoint_worker.py | 64 ++++++++-- .../test_checkpoint_archive_transport.py | 116 +++++++++++++++++- benchmark/tests/test_checkpoint_worker.py | 97 ++++++++++++++- 4 files changed, 269 insertions(+), 19 deletions(-) diff --git a/benchmark/README.md b/benchmark/README.md index 28396d14e8..b1d9d6fbce 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -632,6 +632,17 @@ staging directory instead of ambient temporary storage. If that filesystem fills capture still fails and runs its existing reader/barrier cleanup without publishing an outer artifact; it does not retry on another volume or substitute a later point. +The controller checks its original shared deadline during download progress, tar +header and file reads, safety filtering and permission restoration, including +standalone SQLite companions, the held request and the phase file. It retains +HTTP read timeouts instead of using Docker's process-stream reader, which disables +them. These checks are cooperative between IO operations, not preemption of a +blocked filesystem call or a new allowance per member. On expiry the daemon reader +closes before this operation's writer cleanup; no native point or outer artifact +is published from the expired export. Safe extraction still rejects the complete +invalid inventory before writing member payloads, uses Python's data filter and +restores ordinary source modes without owners or special permission bits. + The container must initially be running and unpaused. Cancellation, failed file reads, rejected native capture and a lost pause response attempt to release only this operation's barrier. Failure to resume the worker is reported and publishes diff --git a/benchmark/edgebench/checkpoint_worker.py b/benchmark/edgebench/checkpoint_worker.py index 4061361a42..db3fcfe565 100644 --- a/benchmark/edgebench/checkpoint_worker.py +++ b/benchmark/edgebench/checkpoint_worker.py @@ -64,7 +64,22 @@ def _guest_path(value: str) -> PurePosixPath: return path -def _archive(container, remote: str, destination: Path) -> Path: +class _ArchiveReader: + """Check the shared clock during tar headers and large-file copy reads.""" + def __init__(self, raw, deadline): + self.raw, self.deadline = raw, deadline + + def __getattr__(self, name): + return getattr(self.raw, name) + + def read(self, *args): + _remaining(self.deadline) + block = self.raw.read(*args) + _remaining(self.deadline) + return block + + +def _archive(container, remote: str, destination: Path, *, deadline: float | None = None) -> Path: """Read through the daemon, rejecting links/devices and archive traversal. Docker's archive reader works while paused, unlike exec-based file readers. @@ -72,13 +87,20 @@ def _archive(container, remote: str, destination: Path) -> Path: """ from docker.utils import decode_json_header + def check(): + if deadline is not None: + _remaining(deadline) + + check() # The SDK's get_archive hides its Response behind a lazy generator. Closing # that generator alone need not close an unread HTTP body. Own the response # so every early exit releases the daemon reader before writer cleanup. api = container.client.api response = api._get(api._url("/containers/{0}/archive", container.id), - params={"path": remote}, stream=True, headers={"Accept-Encoding": "identity"}) + params={"path": remote}, stream=True, headers={"Accept-Encoding": "identity"}, + **({"timeout": _remaining(deadline)} if deadline is not None else {})) with closing(response): + check() api._raise_for_status(response) stat = decode_json_header(response.headers["x-docker-container-path-stat"]) if stat.get("linkTarget"): @@ -87,32 +109,48 @@ def _archive(container, remote: str, destination: Path) -> Path: # The explicitly provisioned export volume owns both the raw transfer # and extracted files; a full ambient /tmp must not defeat that choice. with tempfile.TemporaryFile(dir=destination) as raw: - for chunk in api._stream_raw_result(response, 2 * 1024 * 1024, False): + # _stream_raw_result is Docker's long-running-process reader: it + # disables socket timeouts. Archive IO must keep the request's read + # timeout as well as check the shared absolute clock on progress. + for chunk in response.iter_content(64 * 1024, decode_unicode=False): + check() raw.write(chunk) + check() raw.seek(0) - with tarfile.open(fileobj=raw) as archive: - members = archive.getmembers() + with tarfile.open(fileobj=_ArchiveReader(raw, deadline) if deadline is not None else raw) as archive: + members = [] root = PurePosixPath(remote).name names = set() - for member in members: + for member in archive: + check() path = PurePosixPath(member.name) if (path.is_absolute() or ".." in path.parts or not path.parts or path.parts[0] != root or str(path) in names or not (member.isfile() or member.isdir())): raise ValueError("Worker archive contains undeclared, linked or special members") names.add(str(path)) - archive.extractall(destination, members=members, filter="data") + members.append(member) + + def filtered(member, path): + check() + result = tarfile.data_filter(member, path) + check() + return result + + archive.extractall(destination, members=members, filter=filtered) # The data filter intentionally drops directory modes. This is a # private raw transport witness, so restore only ordinary permission # bits after safe extraction; never restore owners or special bits. # Otherwise a private 0700 authority directory becomes 0755 and a # matching frozen source is falsely rejected as having changed. for member in sorted(members, key=lambda entry: len(PurePosixPath(entry.name).parts), reverse=True): + check() (destination / member.name).chmod(member.mode & 0o777) + check() return _physical(destination / root) -def _archive_component(container, remote: str, destination: Path) -> Path: +def _archive_component(container, remote: str, destination: Path, *, deadline: float | None = None) -> Path: """Transport a standalone SQLite participant's committed storage too. A paused process can still have committed data only in its WAL. Reading @@ -123,7 +161,7 @@ def _archive_component(container, remote: str, destination: Path) -> Path: """ from docker.errors import NotFound - copied = _archive(container, remote, destination) + copied = _archive(container, remote, destination, deadline=deadline) if not copied.is_file(): return copied with copied.open("rb") as source: @@ -135,7 +173,7 @@ def _archive_component(container, remote: str, destination: Path) -> Path: for suffix in ("-wal", "-journal"): with tempfile.TemporaryDirectory(prefix=".sqlite-transport-", dir=destination) as stage: try: - companion = _archive(container, remote + suffix, Path(stage) / "member") + companion = _archive(container, remote + suffix, Path(stage) / "member", deadline=deadline) except NotFound: continue if not companion.is_file(): @@ -220,7 +258,7 @@ def capture_worker(controller_file: Path, *, token: str, capture_id: str, if not container.attrs["State"]["Paused"]: raise RuntimeError("Worker writer barrier was not established") if "handoff" in request: - held = _archive(container, request["handoff"]["source_path"], stage / "held") + held = _archive(container, request["handoff"]["source_path"], stage / "held", deadline=deadline) raw = held.read_bytes() if hashlib.sha256(raw).hexdigest() != request["handoff"]["sha256"]: raise ValueError("Worker handoff changed before its writer barrier") @@ -235,7 +273,7 @@ def capture_worker(controller_file: Path, *, token: str, capture_id: str, deadline = min(deadline, time.monotonic() + prepared["deadline_epoch"] - time.time()) for entry in request["components"]: client.api.timeout = _remaining(deadline) - copied = _archive_component(container, entry["source_path"], stage / "workers" / entry["id"]) + copied = _archive_component(container, entry["source_path"], stage / "workers" / entry["id"], deadline=deadline) components.append({"id": entry["id"], "source_path": str(copied)}) if prepared is not None: observed = effect_runtime_result("runtime.checkpoint_artifacts", { @@ -251,7 +289,7 @@ def capture_worker(controller_file: Path, *, token: str, capture_id: str, raise ValueError("Worker source changed after native/research preparation: " + ", ".join(changed)) if time.time() >= prepared["deadline_epoch"]: raise TimeoutError("Worker handoff expired before evaluator capture") - phase = _archive(container, request["phase_deadline_path"], stage / "phase") + phase = _archive(container, request["phase_deadline_path"], stage / "phase", deadline=deadline) native = capture(controller_file, token=token, capture_id=capture_id, destination=stage / "native", timeout=_remaining(deadline), phase_deadline_file=phase) container.reload() diff --git a/benchmark/tests/test_checkpoint_archive_transport.py b/benchmark/tests/test_checkpoint_archive_transport.py index 9e9d035ef5..bf363f7503 100644 --- a/benchmark/tests/test_checkpoint_archive_transport.py +++ b/benchmark/tests/test_checkpoint_archive_transport.py @@ -24,6 +24,9 @@ def __iter__(self): raise self.error yield b"incomplete archive" + def iter_content(self, *args, **kwargs): + return iter(self) + def close(self): self.closed = True @@ -38,7 +41,7 @@ def test_failed_archive_always_closes_daemon_reader(tmp_path, monkeypatch, failu base64.b64encode(json.dumps(stat).encode()).decode()} api = SimpleNamespace(_url=lambda *args: "archive", _get=lambda *args, **kwargs: stream, _raise_for_status=lambda response: None, - _stream_raw_result=lambda response, *args: iter(response)) + _stream_raw_result=lambda *args: pytest.fail("archive must retain its socket timeout")) class Container: id = "synthetic" @@ -74,9 +77,10 @@ def test_export_storage_does_not_require_ambient_temporary_storage(tmp_path, mon response = SimpleNamespace(headers={"x-docker-container-path-stat": base64.b64encode(json.dumps({"linkTarget": ""}).encode()).decode()}, closed=False) response.close = lambda: setattr(response, "closed", True) + response.iter_content = lambda *args, **kwargs: iter([raw.getvalue()]) api = SimpleNamespace(_url=lambda *args: "archive", _get=lambda *args, **kwargs: response, _raise_for_status=lambda value: None, - _stream_raw_result=lambda *args: iter([raw.getvalue()])) + _stream_raw_result=lambda *args: pytest.fail("archive must retain its socket timeout")) container = SimpleNamespace(id="synthetic", client=SimpleNamespace(api=api)) destination = tmp_path / "provisioned-export" @@ -86,3 +90,111 @@ def test_export_storage_does_not_require_ambient_temporary_storage(tmp_path, mon assert list(destination.iterdir()) == [copied], "no raw transfer file may escape its lifetime" assert response.closed assert unavailable.read_bytes() == b"not a directory" + + +def _tar_transport(payload): + response = SimpleNamespace(headers={"x-docker-container-path-stat": + base64.b64encode(json.dumps({"linkTarget": ""}).encode()).decode()}, closed=False) + response.close = lambda: setattr(response, "closed", True) + response.iter_content = lambda *args, **kwargs: iter([payload]) + api = SimpleNamespace(_url=lambda *args: "archive", _get=lambda *args, **kwargs: response, + _raise_for_status=lambda value: None, + _stream_raw_result=lambda *args: pytest.fail("archive must retain its socket timeout")) + return SimpleNamespace(id="synthetic", client=SimpleNamespace(api=api)), response + + +def _research_tar(): + raw = io.BytesIO() + with tarfile.open(fileobj=raw, mode="w") as archive: + directory = tarfile.TarInfo("data") + directory.type, directory.mode = tarfile.DIRTYPE, 0o700 + archive.addfile(directory) + for name in ("first", "second"): + member = tarfile.TarInfo("data/" + name) + member.size, member.mode = 256 * 1024, 0o600 + archive.addfile(member, io.BytesIO(b"x" * member.size)) + return raw.getvalue() + + +@pytest.mark.parametrize("stage", ["download", "headers", "file_read", "filter", "modes"]) +def test_single_archive_stops_at_shared_deadline_and_closes_reader(tmp_path, monkeypatch, stage): + """Progress in one component cannot renew the original capture clock.""" + container, response = _tar_transport(_research_tar()) + now = [100.0] + monkeypatch.setattr(worker.time, "monotonic", lambda: now[0]) + triggered = [] + if stage == "download": + def download(*args, **kwargs): + yield b"first block" + triggered.append(True) + now[0] = 102 + yield b"must not be written" + pytest.fail("expired stream must not be consumed further") + response.iter_content = download + elif stage in {"headers", "file_read"}: + original = worker.tempfile.TemporaryFile + @contextmanager + def delayed_read(**kwargs): + with original(**kwargs) as raw: + class SlowRead: + def __getattr__(self, name): + return getattr(raw, name) + def read(self, size=-1): + # Header scan reads 512-byte records; file extraction + # reads blocks after validating the complete inventory. + data = raw.read(size) + if (stage == "headers" and size == 512 or + stage == "file_read" and size > 512): + triggered.append(True) + now[0] = 102 + return data + yield SlowRead() + monkeypatch.setattr(worker.tempfile, "TemporaryFile", delayed_read) + elif stage == "filter": + original = worker.tarfile.data_filter + def slow_filter(*args): + member = original(*args) + triggered.append(True) + now[0] = 102 + return member + monkeypatch.setattr(worker.tarfile, "data_filter", slow_filter) + else: + original = worker.tarfile.TarFile.extractall + def finished_extraction(*args, **kwargs): + original(*args, **kwargs) + triggered.append(True) + now[0] = 102 + monkeypatch.setattr(worker.tarfile.TarFile, "extractall", finished_extraction) + with pytest.raises(TimeoutError, match="capture window closed"): + worker._archive(container, "/data", tmp_path / "expired", deadline=101) + assert triggered and response.closed + + +def test_expired_archive_does_not_open_a_daemon_reader(tmp_path, monkeypatch): + container, response = _tar_transport(_research_tar()) + container.client.api._get = lambda *args, **kwargs: pytest.fail("expired operation must not start IO") + monkeypatch.setattr(worker.time, "monotonic", lambda: 2) + with pytest.raises(TimeoutError, match="capture window closed"): + worker._archive(container, "/data", tmp_path / "expired", deadline=1) + assert not (tmp_path / "expired").exists() + + +@pytest.mark.parametrize("unsafe", ["absolute", "parent", "foreign", "duplicate", "symlink", "hardlink", "fifo"]) +def test_archive_rejects_unsafe_inventory_before_extracting_any_file(tmp_path, unsafe): + raw = io.BytesIO() + with tarfile.open(fileobj=raw, mode="w") as archive: + first = tarfile.TarInfo("data/retained") + first.size = 4 + archive.addfile(first, io.BytesIO(b"safe")) + second = tarfile.TarInfo({"absolute": "/escape", "parent": "data/../escape", + "foreign": "elsewhere/file", "duplicate": "data/retained"}.get(unsafe, "data/unsafe")) + if unsafe in {"symlink", "hardlink", "fifo"}: + second.type = {"symlink": tarfile.SYMTYPE, "hardlink": tarfile.LNKTYPE, + "fifo": tarfile.FIFOTYPE}[unsafe] + second.linkname = "data/retained" + archive.addfile(second) + container, response = _tar_transport(raw.getvalue()) + destination = tmp_path / "rejected" + with pytest.raises(ValueError, match="undeclared, linked or special"): + worker._archive(container, "/data", destination) + assert response.closed and not (destination / "data").exists() diff --git a/benchmark/tests/test_checkpoint_worker.py b/benchmark/tests/test_checkpoint_worker.py index 1372b68502..0ad833b373 100644 --- a/benchmark/tests/test_checkpoint_worker.py +++ b/benchmark/tests/test_checkpoint_worker.py @@ -277,10 +277,10 @@ def test_sqlite_companion_failure_cannot_publish_an_older_valid_database( request.write_text(json.dumps(payload)) if fault == "transport": original = checkpoint_worker._archive - def unavailable(worker, remote, destination): + def unavailable(worker, remote, destination, **kwargs): if remote.endswith("-wal"): raise APIError("Synthetic Docker archive transport failure") - return original(worker, remote, destination) + return original(worker, remote, destination, **kwargs) monkeypatch.setattr(checkpoint_worker, "_archive", unavailable) else: replacement = "p.symlink_to('/phase')" if fault == "symlink" else "p.mkdir()" @@ -555,8 +555,8 @@ def test_capture_budget_is_shared_and_exhaustion_releases_writer(controllers, wr container, request = writer token = register(source, first) original = checkpoint_worker._archive_component - def slow_archive(*args): - copied = original(*args) + def slow_archive(*args, **kwargs): + copied = original(*args, **kwargs) time.sleep(2.1) return copied monkeypatch.setattr(checkpoint_worker, "_archive_component", slow_archive) @@ -607,3 +607,92 @@ def broken(*args): supervisor._run(container) assert len(reads) == 3 and isinstance(supervisor.failure, tarfile.ReadError) assert json.loads((tmp_path / "supervisor-failure.json").read_text()) == {"error_kind": "ReadError"} + + +@pytest.mark.parametrize("stage", ["download", "file_copy"]) +def test_one_progressing_archive_expires_without_stranding_writer( + controllers, writer, tmp_path, monkeypatch, stage): + from contextlib import contextmanager + import requests + from benchmark.edgebench import checkpoint_worker as worker + + source, _, first, _, files, grades = controllers + container, request = writer + token = register(source, first) + assert container.exec_run(["python", "-c", + "from pathlib import Path;Path('/data/payload').write_bytes(b'x'*(16*1024*1024))"]).exit_code == 0 + before = counter(container) + progress, responses = [], [] + original_get = type(container.client.api)._get + def observed_get(self, url, *args, **kwargs): + response = original_get(self, url, *args, **kwargs) + if "/archive" in url: + responses.append(response) + # The archive reader must not use Docker's process-stream helper, + # which deliberately changes this socket to an infinite timeout. + socket = self._get_raw_response_socket(response) + assert getattr(socket, "_sock", socket).gettimeout() is not None + return response + monkeypatch.setattr(type(container.client.api), "_get", observed_get) + if stage == "download": + original = requests.Response.iter_content + def progressing(response, *args, **kwargs): + if "/archive" not in response.url: + yield from original(response, *args, **kwargs) + return + for block in original(response, *args, **kwargs): + socket = container.client.api._get_raw_response_socket(response) + assert getattr(socket, "_sock", socket).gettimeout() is not None + progress.append(len(block)) + time.sleep(.04) + yield block + monkeypatch.setattr(requests.Response, "iter_content", progressing) + else: + original = worker.tempfile.TemporaryFile + @contextmanager + def slow_file(**kwargs): + with original(**kwargs) as raw: + class SlowFile: + def __getattr__(self, name): + return getattr(raw, name) + def read(self, size=-1): + block = raw.read(size) + if size > 512: + progress.append(len(block)) + time.sleep(.01) + return block + yield SlowFile() + monkeypatch.setattr(worker.tempfile, "TemporaryFile", slow_file) + responses.clear() + progress.clear() + started = time.monotonic() + unit = tmp_path / "expired" + try: + with pytest.raises(TimeoutError, match="capture window closed"): + capture_worker(files[0], token=token, capture_id="4" * 32, + destination=unit, request_file=request, timeout=3) + assert progress and sum(progress) < 16 * 1024 * 1024 + assert time.monotonic() - started < 10, "one archive must stop before consuming its slow remainder" + assert all(response.raw.closed for response in responses) + container.reload() + assert container.attrs["State"]["Running"] and not container.attrs["State"]["Paused"] + assert not unit.exists() and not grades + bank = Path(json.loads(files[0].read_text())["bank_root"]) + assert not (bank / ("4" * 32)).exists() + with pytest.raises(FileExistsError): + capture_worker(files[0], token=token, capture_id="4" * 32, + destination=tmp_path / "retry", request_file=request) + finally: + # Keep a failing characterization run safe too; no GC-based cleanup. + for response in responses: + response.close() + container.reload() + if container.attrs["State"]["Paused"]: + container.unpause() + # Observe the live writer only after the slowed archive reader is removed. + monkeypatch.setattr(type(container.client.api), "_get", original_get) + if stage == "download": + monkeypatch.setattr(requests.Response, "iter_content", original) + else: + monkeypatch.setattr(worker.tempfile, "TemporaryFile", original) + wait(lambda: counter(container) > before)