fix(benchmark): enforce shared deadlines within Docker archives

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

benchmark/edgebench/checkpoint_worker.py

@@ -64,21 +64,43 @@ 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.
     Keep only the declared object, not parent filesystem or process credentials.
     """
     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()

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()