From 97a77700d758a0c757e5a0ee25d8d0e57af76ed2 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 30 Sep 2026 06:50:51 -1000 Subject: [PATCH 1/8] fix(store): the local object store writes atomically and keeps content types A write now lands whole or not at all. Bytes are staged in a reserved .agentenv-tmp directory inside the root and put in place with a rename (an overwrite) or a hard link (a write-once put), so a crash or a concurrent reader never sees a partial object, and a write-once put that loses a race raises ObjectAlreadyExistsError instead of replacing the winner. Filesystems without hard links fall back to an exclusive create. A write's content type is recorded in a reserved .agentenv-meta tree and returned by get_object_metadata. The default type, which a writer that names none gets, reads back as unknown, so readers keep guessing from the name as before. Both directories are refused as keys and left out of listings. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../store/object_store/local_object_store.py | 132 ++++++++++++++---- tst/unit/store/local_object_store_test.py | 90 +++++++++++- 2 files changed, 196 insertions(+), 26 deletions(-) diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local_object_store.py index 696726d6..5db87829 100644 --- a/src/agent_env/store/object_store/local_object_store.py +++ b/src/agent_env/store/object_store/local_object_store.py @@ -3,7 +3,12 @@ from __future__ import annotations import contextlib +import errno +import json +import logging +import os import shutil +import tempfile from collections.abc import Iterator from datetime import datetime, timezone from pathlib import Path @@ -13,11 +18,20 @@ from agent_env.store.local_state import ensure_state_dir from agent_env.store.object_store.object_store import DEFAULT_CONTENT_TYPE, ObjectMetadata, ObjectStore +logger = logging.getLogger(__name__) + +_META_DIR = ".agentenv-meta" # each object's content type, at the object's own key +_STAGING_DIR = ".agentenv-tmp" # files being written, inside the root so a rename into place is atomic +_RESERVED = (".gitignore", _META_DIR, _STAGING_DIR) +# A filesystem without hard links: a no-overwrite write reserves its key by exclusive create instead. +_NO_HARD_LINKS = frozenset({errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK}) + class LocalFilesystemObjectStore(ObjectStore): - """ObjectStore backed by a local directory tree; content types are not persisted. + """ObjectStore backed by a local directory tree. - The root-level ``.gitignore`` key is reserved for the ignore rule the store writes.""" + A write lands whole or not at all, and keeps any content type but the default. The root-level ``.gitignore``, + ``.agentenv-meta`` and ``.agentenv-tmp`` keys are reserved for the store's own files.""" def __init__(self, root: str) -> None: self._root = Path(root) @@ -36,18 +50,18 @@ def put( path = self._resolve(key) if path.exists() and not allow_overwrite: raise ObjectAlreadyExistsError(f"Object already exists at {path}.") - ensure_state_dir(self._root) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_bytes(data) + with self._staged() as staged: + staged.write_bytes(data) + self._commit(staged, path, content_type, allow_overwrite=allow_overwrite) return self._to_url(path) def put_file(self, key: str, file_path: str, content_type: str = DEFAULT_CONTENT_TYPE) -> str: path = self._resolve(key) if path.exists(): raise ObjectAlreadyExistsError(f"Object already exists at {path}.") - ensure_state_dir(self._root) - path.parent.mkdir(parents=True, exist_ok=True) - shutil.copyfile(file_path, path) + with self._staged() as staged: + shutil.copyfile(file_path, staged) + self._commit(staged, path, content_type, allow_overwrite=False) return self._to_url(path) def put_file_at(self, object_url: str, file_path: str, content_type: str = DEFAULT_CONTENT_TYPE) -> str: @@ -77,31 +91,25 @@ def get_object_metadata(self, key: str) -> ObjectMetadata | None: if not path.is_file(): return None st = path.stat() - return ObjectMetadata(content_type=None, size=st.st_size, last_modified=datetime.fromtimestamp(st.st_mtime, tz=timezone.utc)) + return ObjectMetadata( + content_type=self._read_content_type(path), + size=st.st_size, + last_modified=datetime.fromtimestamp(st.st_mtime, tz=timezone.utc), + ) def get_object_metadata_at(self, object_url: str) -> ObjectMetadata | None: return self.get_object_metadata(self.get_object_key(object_url)) def list(self, prefix: str) -> list[str]: - keys = [] - for path in self._root.rglob("*"): - if not path.is_file() or path == self._root / ".gitignore": - continue - key = path.relative_to(self._root).as_posix() - if key.startswith(prefix): - keys.append(key) - return keys + root = self._root.resolve() + return [key for key in (p.relative_to(root).as_posix() for p in self._objects_below(root)) if key.startswith(prefix)] def list_at(self, url_prefix: str) -> list[str]: base = url_prefix if url_prefix.endswith("/") else url_prefix + "/" dir_path = self._resolve(self.get_object_key(url_prefix)) if not dir_path.is_dir(): return [] - return [ - base + p.relative_to(dir_path).as_posix() - for p in sorted(dir_path.rglob("*")) - if p.is_file() and p != self._root.resolve() / ".gitignore" - ] + return [base + p.relative_to(dir_path).as_posix() for p in sorted(self._objects_below(dir_path))] def object_url(self, key: str) -> str: return self._to_url(self._resolve(key)) @@ -119,13 +127,87 @@ def get_object_key(self, object_url: str) -> str: return path.relative_to(root).as_posix() def _resolve(self, key: str) -> Path: + root = self._root.resolve() resolved = (self._root / key).resolve() - if not resolved.is_relative_to(self._root.resolve()): + if not resolved.is_relative_to(root): raise ValueError(f"Key {key!r} escapes the store root.") - if resolved == self._root.resolve() / ".gitignore": - raise ValueError(f"Key {key!r} is reserved: the store keeps its .gitignore there.") + parts = resolved.relative_to(root).parts + if parts and parts[0] in _RESERVED: + raise ValueError(f"Key {key!r} is reserved: the store keeps its own {parts[0]} there.") return resolved + def _objects_below(self, dir_path: Path) -> Iterator[Path]: + """Every object file below ``dir_path``, leaving out the store's own files at the root.""" + root = self._root.resolve() + if not dir_path.is_dir(): + return + for entry in dir_path.iterdir(): + if dir_path == root and entry.name in _RESERVED: + continue + if entry.is_file(): + yield entry + elif entry.is_dir(): + yield from (p for p in entry.rglob("*") if p.is_file()) + + @contextlib.contextmanager + def _staged(self) -> Iterator[Path]: + """An empty file to write an object into before it is committed; removed if it never is.""" + ensure_state_dir(self._root) + staging = self._root / _STAGING_DIR + staging.mkdir(exist_ok=True) + fd, name = tempfile.mkstemp(dir=staging) + os.close(fd) + staged = Path(name) + staged.chmod(0o644) + try: + yield staged + finally: + staged.unlink(missing_ok=True) + + def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrite: bool) -> None: + """Put a staged file in place at ``path`` and record its content type. An overwrite renames over + whatever is there; otherwise a hard link claims ``path`` only if no other writer got there first.""" + path.parent.mkdir(parents=True, exist_ok=True) + if allow_overwrite: + os.replace(staged, path) + else: + try: + os.link(staged, path) + except FileExistsError: + raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None + except OSError as e: + if e.errno not in _NO_HARD_LINKS: + raise + try: + os.close(os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)) + except FileExistsError: + raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None + os.replace(staged, path) + # The default type is what a writer that named none gets; it reads back as unknown, so readers + # still guess from the name as they did before types were kept. + try: + meta = self._meta_path(path) + if content_type == DEFAULT_CONTENT_TYPE: + meta.unlink(missing_ok=True) + return + meta.parent.mkdir(parents=True, exist_ok=True) + with self._staged() as staged_meta: + staged_meta.write_text(json.dumps({"content_type": content_type})) + os.replace(staged_meta, meta) + except OSError as e: # the object is in place; only its type goes unrecorded + logger.warning("Could not record the content type of %s: %s", path, e) + + def _read_content_type(self, path: Path) -> str | None: + """The content type recorded for the object at ``path``; None for one written before types were kept.""" + try: + return json.loads(self._meta_path(path).read_text()).get("content_type") + except (OSError, ValueError, AttributeError): + return None + + def _meta_path(self, path: Path) -> Path: + root = self._root.resolve() + return root / _META_DIR / path.relative_to(root) + @staticmethod def _to_url(path: Path) -> str: return f"file://{path.resolve()}" diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py index aec1fe12..3ac8e048 100644 --- a/tst/unit/store/local_object_store_test.py +++ b/tst/unit/store/local_object_store_test.py @@ -4,9 +4,13 @@ assertions that prove S3 parity also give quick backend-neutral coverage. """ +import errno +import os +from pathlib import Path + import pytest -from agent_env.store import LocalFilesystemObjectStore +from agent_env.store import LocalFilesystemObjectStore, ObjectAlreadyExistsError from tst.store import object_conformance @@ -105,3 +109,87 @@ def test_signed_get_url_is_none(store): def test_signed_put_url_is_none(store): """Local FS can't presign uploads either — VM-upload flows require a signable backend.""" assert store.signed_put_url("file:///whatever") is None + + +def test_a_write_keeps_its_content_type(store): + store.put("typed/a.json", b"{}", content_type="application/json") + store.put_file("typed/b.csv", __file__, content_type="text/csv") + assert store.get_object_metadata("typed/a.json").content_type == "application/json" + assert store.get_object_metadata("typed/b.csv").content_type == "text/csv" + + +def test_a_write_that_names_no_type_reads_back_as_unknown(store): + """Readers then guess from the name, as they did before types were kept.""" + store.put("untyped/a.zip", b"PK") + assert store.get_object_metadata("untyped/a.zip").content_type is None + + +def test_an_overwrite_replaces_the_recorded_type(store): + store.put("over/x", b"1", content_type="application/json") + store.put("over/x", b"2", content_type="text/plain", allow_overwrite=True) + assert store.get_object_metadata("over/x").content_type == "text/plain" + store.put("over/x", b"3", allow_overwrite=True) + assert store.get_object_metadata("over/x").content_type is None + + +def test_an_object_written_before_types_were_kept_has_none(store, tmp_path): + (tmp_path / "old").mkdir() + (tmp_path / "old" / "x.bin").write_bytes(b"legacy") + md = store.get_object_metadata("old/x.bin") + assert (md.content_type, md.size) == (None, 6) + + +def test_a_failed_write_leaves_no_object(store, monkeypatch): + def fail(self, data): + raise OSError("disk full") + + monkeypatch.setattr(Path, "write_bytes", fail) + with pytest.raises(OSError, match="disk full"): + store.put("partial/x.bin", b"payload") + monkeypatch.undo() + assert not store.exists("partial/x.bin") and store.list("") == [] + + +def test_a_no_overwrite_write_that_loses_a_race_keeps_the_winner(store, monkeypatch): + """Another writer lands the key between the existence check and the commit.""" + real_link = os.link + + def racing_link(src, dst): + Path(dst).write_bytes(b"winner") + return real_link(src, dst) + + monkeypatch.setattr(os, "link", racing_link) + with pytest.raises(ObjectAlreadyExistsError): + store.put("race/x", b"loser") + assert store.read("race/x") == b"winner" + + +def test_a_filesystem_without_hard_links_still_refuses_an_existing_key(store, monkeypatch): + def no_links(src, dst): + raise OSError(errno.EPERM, "hard links not supported") + + monkeypatch.setattr(os, "link", no_links) + url = store.put("nolink/x", b"first", content_type="text/plain") + assert store.get(url) == b"first" + with pytest.raises(ObjectAlreadyExistsError): + store.put("nolink/x", b"second") + assert store.read("nolink/x") == b"first" + + +@pytest.mark.parametrize("key", [".agentenv-meta/x", ".agentenv-tmp/x", "a/../.agentenv-meta"]) +def test_the_stores_own_directories_are_reserved(store, key): + with pytest.raises(ValueError, match="reserved"): + store.put(key, b"mine") + + +def test_listings_leave_out_the_stores_own_files(store, tmp_path): + store.put("a/x.json", b"{}", content_type="application/json") + (tmp_path / ".agentenv-tmp").mkdir(exist_ok=True) + (tmp_path / ".agentenv-tmp" / "crash-residue").write_bytes(b"half") + assert store.list("") == ["a/x.json"] + assert store.list_at(store.object_url("")) == [store.object_url("a/x.json")] + assert os.listdir(tmp_path / ".agentenv-tmp") == ["crash-residue"] # committed writes leave nothing staged + + +def test_list_of_an_unwritten_store_is_empty(tmp_path): + assert LocalFilesystemObjectStore(str(tmp_path / "never-written")).list("") == [] From e91eea98855b490b86cb3cac094b5bf4f7e32252 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 30 Sep 2026 07:51:22 -1000 Subject: [PATCH 2/8] fix(store): a recorded content type names the file it describes, and a write-once put never exposes an empty object A content type is recorded with the identity of the file it was written for (inode, size, mtime, which a rename keeps), so a type recorded for another write - a racing overwrite that renamed first, or an old record an overwrite could not remove - reads back as unknown, never as the wrong type. On a filesystem without hard links a write-once put now checks for the key and renames into place, instead of reserving the key with an empty file that readers could see and a crash could leave. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../store/object_store/local_object_store.py | 33 ++++++++++++------- tst/unit/store/local_object_store_test.py | 32 ++++++++++++++++++ 2 files changed, 54 insertions(+), 11 deletions(-) diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local_object_store.py index 5db87829..d6e10486 100644 --- a/src/agent_env/store/object_store/local_object_store.py +++ b/src/agent_env/store/object_store/local_object_store.py @@ -23,7 +23,7 @@ _META_DIR = ".agentenv-meta" # each object's content type, at the object's own key _STAGING_DIR = ".agentenv-tmp" # files being written, inside the root so a rename into place is atomic _RESERVED = (".gitignore", _META_DIR, _STAGING_DIR) -# A filesystem without hard links: a no-overwrite write reserves its key by exclusive create instead. +# A filesystem without hard links: a no-overwrite write checks, then renames, so two racing writers can both land. _NO_HARD_LINKS = frozenset({errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK}) @@ -92,7 +92,7 @@ def get_object_metadata(self, key: str) -> ObjectMetadata | None: return None st = path.stat() return ObjectMetadata( - content_type=self._read_content_type(path), + content_type=self._read_content_type(path, st), size=st.st_size, last_modified=datetime.fromtimestamp(st.st_mtime, tz=timezone.utc), ) @@ -166,7 +166,11 @@ def _staged(self) -> Iterator[Path]: def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrite: bool) -> None: """Put a staged file in place at ``path`` and record its content type. An overwrite renames over - whatever is there; otherwise a hard link claims ``path`` only if no other writer got there first.""" + whatever is there; otherwise a hard link claims ``path`` only if no other writer got there first. + + The type is recorded with the identity of the file it describes, so one recorded for another + write (a racing overwrite, or one whose type could not be recorded) reads back as unknown.""" + identity = _identity(staged.stat()) path.parent.mkdir(parents=True, exist_ok=True) if allow_overwrite: os.replace(staged, path) @@ -178,9 +182,7 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit except OSError as e: if e.errno not in _NO_HARD_LINKS: raise - try: - os.close(os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)) - except FileExistsError: + if path.exists(): raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None os.replace(staged, path) # The default type is what a writer that named none gets; it reads back as unknown, so readers @@ -192,15 +194,19 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit return meta.parent.mkdir(parents=True, exist_ok=True) with self._staged() as staged_meta: - staged_meta.write_text(json.dumps({"content_type": content_type})) + staged_meta.write_text(json.dumps({"content_type": content_type, "object": identity})) os.replace(staged_meta, meta) - except OSError as e: # the object is in place; only its type goes unrecorded + except OSError as e: # the object is in place and its type reads back as unknown logger.warning("Could not record the content type of %s: %s", path, e) - def _read_content_type(self, path: Path) -> str | None: - """The content type recorded for the object at ``path``; None for one written before types were kept.""" + def _read_content_type(self, path: Path, st: os.stat_result | None = None) -> str | None: + """The content type recorded for the object at ``path``, if it was recorded for this very file; None + otherwise, as for one written before types were kept.""" try: - return json.loads(self._meta_path(path).read_text()).get("content_type") + recorded = json.loads(self._meta_path(path).read_text()) + if recorded.get("object") != _identity(st or path.stat()): + return None + return recorded.get("content_type") except (OSError, ValueError, AttributeError): return None @@ -217,6 +223,11 @@ def _from_url(object_url: str) -> str: return object_url[len("file://"):] if object_url.startswith("file://") else object_url +def _identity(st: os.stat_result) -> list[int]: + """What tells one file apart from another written to the same key: a rename keeps all three.""" + return [st.st_ino, st.st_size, st.st_mtime_ns] + + @contextlib.contextmanager def _missing_as_not_found(object_url: str) -> Iterator[None]: try: diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py index 3ac8e048..e8295c63 100644 --- a/tst/unit/store/local_object_store_test.py +++ b/tst/unit/store/local_object_store_test.py @@ -176,6 +176,38 @@ def no_links(src, dst): assert store.read("nolink/x") == b"first" +def test_without_hard_links_a_failed_write_once_put_leaves_no_object(store, monkeypatch): + def no_links(src, dst): + raise OSError(errno.EPERM, "hard links not supported") + + def failing_replace(src, dst): + raise OSError(errno.EIO, "disk went away") + + monkeypatch.setattr(os, "link", no_links) + monkeypatch.setattr(os, "replace", failing_replace) + with pytest.raises(OSError, match="disk went away"): + store.put("nolink/y", b"never") + assert not store.exists("nolink/y") + + +def test_a_type_recorded_for_another_write_reads_back_as_unknown(store, tmp_path): + """Two overwrites of one key race: the one that renamed first records its type last. Its type must not + be read as the other's.""" + store.put("race/x", b"first", content_type="text/plain") + first_meta = (tmp_path / ".agentenv-meta" / "race" / "x").read_text() + store.put("race/x", b"second!", content_type="application/json", allow_overwrite=True) + (tmp_path / ".agentenv-meta" / "race" / "x").write_text(first_meta) + metadata = store.get_object_metadata("race/x") + assert (metadata.content_type, metadata.size) == (None, 7) + + +def test_an_overwrite_that_names_no_type_never_inherits_the_old_one(store, tmp_path, monkeypatch): + store.put("keep/x", b"typed", content_type="text/plain") + monkeypatch.setattr(Path, "unlink", lambda self, missing_ok=False: None) # the old record survives + store.put("keep/x", b"untyped!", allow_overwrite=True) + assert store.get_object_metadata("keep/x").content_type is None + + @pytest.mark.parametrize("key", [".agentenv-meta/x", ".agentenv-tmp/x", "a/../.agentenv-meta"]) def test_the_stores_own_directories_are_reserved(store, key): with pytest.raises(ValueError, match="reserved"): From 1271813d061ea456344b5926a90ebfff6f010109 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 30 Sep 2026 07:58:30 -1000 Subject: [PATCH 3/8] fix(store): a write-once put stays exclusive on a filesystem without hard links Where no hard link can claim the key, a write-once put now claims it with a marker created exclusively in the staging directory, held only from the existence check to the rename. A racing writer waits for the marker to go and then finds the winner's object; a marker a crash left behind is taken over once it is ten seconds old. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../store/object_store/local_object_store.py | 35 ++++++++++++++-- tst/unit/store/local_object_store_test.py | 42 +++++++++++++++++++ 2 files changed, 73 insertions(+), 4 deletions(-) diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local_object_store.py index d6e10486..a1608648 100644 --- a/src/agent_env/store/object_store/local_object_store.py +++ b/src/agent_env/store/object_store/local_object_store.py @@ -4,11 +4,13 @@ import contextlib import errno +import hashlib import json import logging import os import shutil import tempfile +import time from collections.abc import Iterator from datetime import datetime, timezone from pathlib import Path @@ -23,8 +25,9 @@ _META_DIR = ".agentenv-meta" # each object's content type, at the object's own key _STAGING_DIR = ".agentenv-tmp" # files being written, inside the root so a rename into place is atomic _RESERVED = (".gitignore", _META_DIR, _STAGING_DIR) -# A filesystem without hard links: a no-overwrite write checks, then renames, so two racing writers can both land. +# A filesystem without hard links: a no-overwrite write claims its key with a marker file instead. _NO_HARD_LINKS = frozenset({errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK}) +_STALE_CLAIM_SECONDS = 10 # a claim is held only from the check to the rename; one this old was left by a crash class LocalFilesystemObjectStore(ObjectStore): @@ -182,9 +185,10 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit except OSError as e: if e.errno not in _NO_HARD_LINKS: raise - if path.exists(): - raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None - os.replace(staged, path) + with self._claimed(path): + if path.exists(): + raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None + os.replace(staged, path) # The default type is what a writer that named none gets; it reads back as unknown, so readers # still guess from the name as they did before types were kept. try: @@ -199,6 +203,29 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit except OSError as e: # the object is in place and its type reads back as unknown logger.warning("Could not record the content type of %s: %s", path, e) + @contextlib.contextmanager + def _claimed(self, path: Path) -> Iterator[None]: + """The sole right to create ``path``, where no hard link can claim it: a marker in the staging directory, + created exclusively. A writer that finds one waits for it to go, taking it over once it is stale.""" + marker = self._root / _STAGING_DIR / f"{hashlib.sha256(str(path).encode()).hexdigest()}.claim" + while True: + try: + os.close(os.open(marker, os.O_CREAT | os.O_EXCL | os.O_WRONLY)) + break + except FileExistsError: + try: + stale = time.time() - marker.stat().st_mtime > _STALE_CLAIM_SECONDS + except FileNotFoundError: + continue + if stale: + marker.unlink(missing_ok=True) + else: + time.sleep(0.01) + try: + yield + finally: + marker.unlink(missing_ok=True) + def _read_content_type(self, path: Path, st: os.stat_result | None = None) -> str | None: """The content type recorded for the object at ``path``, if it was recorded for this very file; None otherwise, as for one written before types were kept.""" diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py index e8295c63..9002fdf9 100644 --- a/tst/unit/store/local_object_store_test.py +++ b/tst/unit/store/local_object_store_test.py @@ -5,7 +5,10 @@ """ import errno +import hashlib import os +import threading +import time from pathlib import Path import pytest @@ -190,6 +193,45 @@ def failing_replace(src, dst): assert not store.exists("nolink/y") +def test_without_hard_links_a_racing_write_once_put_keeps_the_winner(store, monkeypatch): + def no_links(src, dst): + raise OSError(errno.EPERM, "hard links not supported") + + real_replace = os.replace + first_holds_its_claim = threading.Event() + + def slow_first_replace(src, dst): + if threading.current_thread().name == "first" and str(dst).endswith("race/once"): + first_holds_its_claim.set() + time.sleep(0.3) + real_replace(src, dst) + + monkeypatch.setattr(os, "link", no_links) + monkeypatch.setattr(os, "replace", slow_first_replace) + first = threading.Thread(target=store.put, args=("race/once", b"first"), name="first") + first.start() + first_holds_its_claim.wait(5) + with pytest.raises(ObjectAlreadyExistsError): + store.put("race/once", b"second") + first.join() + assert store.read("race/once") == b"first" + + +def test_without_hard_links_a_claim_a_crash_left_is_taken_over(store, tmp_path, monkeypatch): + def no_links(src, dst): + raise OSError(errno.EPERM, "hard links not supported") + + monkeypatch.setattr(os, "link", no_links) + store.put("seed", b"s") # creates the staging directory + path = (tmp_path / "left/x").resolve() + marker = tmp_path / ".agentenv-tmp" / f"{hashlib.sha256(str(path).encode()).hexdigest()}.claim" + marker.write_bytes(b"") + os.utime(marker, (time.time() - 60, time.time() - 60)) + store.put("left/x", b"mine") + assert store.read("left/x") == b"mine" + assert not marker.exists() + + def test_a_type_recorded_for_another_write_reads_back_as_unknown(store, tmp_path): """Two overwrites of one key race: the one that renamed first records its type last. Its type must not be read as the other's.""" From 589642321a42cff253dfceb1ca6c31d8a712255e Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 30 Sep 2026 08:14:20 -1000 Subject: [PATCH 4/8] fix(store): a live writer keeps its claim, and recorded types survive a copy that keeps modification times Without hard links, a write-once put's claim marker now names its owner's pid and is taken over only once that process has exited, not after ten seconds, so a slow live writer never loses it. On Windows, where rename refuses an existing target, the rename itself is the claim. A recorded content type is now paired with the file's size and modification time, and each write stamps its staged file with a modification time no other write shares (now, to the millisecond, plus a random remainder). Racing writes still never pair bytes with another write's type, and a copy of the store that keeps modification times keeps its types, where the inode it no longer paired with would not survive one. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../store/object_store/local_object_store.py | 63 ++++++++++++++----- tst/unit/store/local_object_store_test.py | 41 ++++++++++-- 2 files changed, 84 insertions(+), 20 deletions(-) diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local_object_store.py index a1608648..77f1f6c7 100644 --- a/src/agent_env/store/object_store/local_object_store.py +++ b/src/agent_env/store/object_store/local_object_store.py @@ -8,6 +8,7 @@ import json import logging import os +import secrets import shutil import tempfile import time @@ -27,7 +28,7 @@ _RESERVED = (".gitignore", _META_DIR, _STAGING_DIR) # A filesystem without hard links: a no-overwrite write claims its key with a marker file instead. _NO_HARD_LINKS = frozenset({errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK}) -_STALE_CLAIM_SECONDS = 10 # a claim is held only from the check to the rename; one this old was left by a crash +_UNWRITTEN_CLAIM_SECONDS = 10 # a marker is given its owner's pid as it is created; one still empty this long never will be class LocalFilesystemObjectStore(ObjectStore): @@ -173,7 +174,7 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit The type is recorded with the identity of the file it describes, so one recorded for another write (a racing overwrite, or one whose type could not be recorded) reads back as unknown.""" - identity = _identity(staged.stat()) + identity = _stamp(staged) path.parent.mkdir(parents=True, exist_ok=True) if allow_overwrite: os.replace(staged, path) @@ -185,10 +186,16 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit except OSError as e: if e.errno not in _NO_HARD_LINKS: raise - with self._claimed(path): - if path.exists(): + if os.name == "nt": # rename there refuses an existing target, which is the claim + try: + os.rename(staged, path) + except FileExistsError: raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None - os.replace(staged, path) + else: + with self._claimed(path): + if path.exists(): + raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None + os.replace(staged, path) # The default type is what a writer that named none gets; it reads back as unknown, so readers # still guess from the name as they did before types were kept. try: @@ -206,21 +213,21 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit @contextlib.contextmanager def _claimed(self, path: Path) -> Iterator[None]: """The sole right to create ``path``, where no hard link can claim it: a marker in the staging directory, - created exclusively. A writer that finds one waits for it to go, taking it over once it is stale.""" + created exclusively and naming its owner's pid. A writer that finds one waits for it to go, and takes it + over only once its owner has exited.""" marker = self._root / _STAGING_DIR / f"{hashlib.sha256(str(path).encode()).hexdigest()}.claim" while True: try: - os.close(os.open(marker, os.O_CREAT | os.O_EXCL | os.O_WRONLY)) - break + fd = os.open(marker, os.O_CREAT | os.O_EXCL | os.O_WRONLY) except FileExistsError: - try: - stale = time.time() - marker.stat().st_mtime > _STALE_CLAIM_SECONDS - except FileNotFoundError: - continue - if stale: + if _abandoned(marker): marker.unlink(missing_ok=True) else: time.sleep(0.01) + continue + with os.fdopen(fd, "w") as f: + f.write(str(os.getpid())) + break try: yield finally: @@ -250,9 +257,35 @@ def _from_url(object_url: str) -> str: return object_url[len("file://"):] if object_url.startswith("file://") else object_url +def _stamp(staged: Path) -> list[int]: + """Give a staged file a modification time no other write shares (now, to the millisecond, plus a random + remainder) and return its identity. A rename keeps it, and so does a copy that keeps modification times.""" + now_ns = time.time_ns() + stamp = now_ns - now_ns % 1_000_000 + secrets.randbelow(1_000_000) + os.utime(staged, ns=(stamp, stamp)) + return _identity(staged.stat()) + + def _identity(st: os.stat_result) -> list[int]: - """What tells one file apart from another written to the same key: a rename keeps all three.""" - return [st.st_ino, st.st_size, st.st_mtime_ns] + return [st.st_size, st.st_mtime_ns] + + +def _abandoned(marker: Path) -> bool: + """Whether a claim marker's owner has exited, or never got to name itself.""" + try: + owner = marker.read_text() + age = time.time() - marker.stat().st_mtime + except FileNotFoundError: + return False + if not owner: + return age > _UNWRITTEN_CLAIM_SECONDS + try: + os.kill(int(owner), 0) + except ProcessLookupError: + return True + except (PermissionError, ValueError): + return False + return False @contextlib.contextmanager diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py index 9002fdf9..860e6f78 100644 --- a/tst/unit/store/local_object_store_test.py +++ b/tst/unit/store/local_object_store_test.py @@ -7,6 +7,9 @@ import errno import hashlib import os +import shutil +import subprocess +import sys import threading import time from pathlib import Path @@ -217,21 +220,49 @@ def slow_first_replace(src, dst): assert store.read("race/once") == b"first" -def test_without_hard_links_a_claim_a_crash_left_is_taken_over(store, tmp_path, monkeypatch): +def _claim_marker(tmp_path, key): + path = (tmp_path / key).resolve() + return tmp_path / ".agentenv-tmp" / f"{hashlib.sha256(str(path).encode()).hexdigest()}.claim" + + +@pytest.fixture +def no_hard_links(store, monkeypatch): def no_links(src, dst): raise OSError(errno.EPERM, "hard links not supported") monkeypatch.setattr(os, "link", no_links) store.put("seed", b"s") # creates the staging directory - path = (tmp_path / "left/x").resolve() - marker = tmp_path / ".agentenv-tmp" / f"{hashlib.sha256(str(path).encode()).hexdigest()}.claim" - marker.write_bytes(b"") - os.utime(marker, (time.time() - 60, time.time() - 60)) + + +def test_without_hard_links_a_claim_whose_owner_exited_is_taken_over(store, tmp_path, no_hard_links): + exited = subprocess.run([sys.executable, "-c", "import os; print(os.getpid())"], capture_output=True, text=True) + marker = _claim_marker(tmp_path, "left/x") + marker.write_text(exited.stdout.strip()) store.put("left/x", b"mine") assert store.read("left/x") == b"mine" assert not marker.exists() +def test_without_hard_links_a_live_writers_claim_is_never_taken_over(store, tmp_path, no_hard_links): + marker = _claim_marker(tmp_path, "held/x") + marker.write_text(str(os.getpid())) + os.utime(marker, (time.time() - 3600, time.time() - 3600)) + writer = threading.Thread(target=store.put, args=("held/x", b"late")) + writer.start() + writer.join(0.3) + assert writer.is_alive() and not store.exists("held/x") + marker.unlink() + writer.join(5) + assert store.read("held/x") == b"late" + + +def test_types_survive_a_copy_that_keeps_modification_times(store, tmp_path): + store.put("copied/x", b"typed", content_type="text/plain") + copy = tmp_path.parent / f"{tmp_path.name}-copy" + shutil.copytree(tmp_path, copy) # copies modification times, not inodes + assert LocalFilesystemObjectStore(str(copy)).get_object_metadata("copied/x").content_type == "text/plain" + + def test_a_type_recorded_for_another_write_reads_back_as_unknown(store, tmp_path): """Two overwrites of one key race: the one that renamed first records its type last. Its type must not be read as the other's.""" From a2d2793bac46039015dc4945953fcf7d04f83312 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 30 Sep 2026 08:24:04 -1000 Subject: [PATCH 5/8] fix(store): claim a key without hard links with a lock that dies with its holder A write-once put on a filesystem without hard links now claims its key with an exclusive flock on a file in the staging directory rather than a marker naming a pid. The kernel drops the lock when its holder exits, so a crashed writer never blocks the key and a reused pid can never pass for a live one; a live holder is waited for however long it takes. The file stays in place so every writer locks the same one. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../store/object_store/local_object_store.py | 49 +++++-------------- tst/unit/store/local_object_store_test.py | 38 +++++++------- 2 files changed, 32 insertions(+), 55 deletions(-) diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local_object_store.py index 77f1f6c7..a60cf3c7 100644 --- a/src/agent_env/store/object_store/local_object_store.py +++ b/src/agent_env/store/object_store/local_object_store.py @@ -18,6 +18,9 @@ from typing import BinaryIO from agent_env.store.base import ObjectAlreadyExistsError, ObjectNotFoundError + +if os.name != "nt": + import fcntl from agent_env.store.local_state import ensure_state_dir from agent_env.store.object_store.object_store import DEFAULT_CONTENT_TYPE, ObjectMetadata, ObjectStore @@ -26,9 +29,8 @@ _META_DIR = ".agentenv-meta" # each object's content type, at the object's own key _STAGING_DIR = ".agentenv-tmp" # files being written, inside the root so a rename into place is atomic _RESERVED = (".gitignore", _META_DIR, _STAGING_DIR) -# A filesystem without hard links: a no-overwrite write claims its key with a marker file instead. +# A filesystem without hard links: a no-overwrite write claims its key with a lock instead. _NO_HARD_LINKS = frozenset({errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK}) -_UNWRITTEN_CLAIM_SECONDS = 10 # a marker is given its owner's pid as it is created; one still empty this long never will be class LocalFilesystemObjectStore(ObjectStore): @@ -212,26 +214,16 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit @contextlib.contextmanager def _claimed(self, path: Path) -> Iterator[None]: - """The sole right to create ``path``, where no hard link can claim it: a marker in the staging directory, - created exclusively and naming its owner's pid. A writer that finds one waits for it to go, and takes it - over only once its owner has exited.""" + """The sole right to create ``path``, where no hard link can claim it: an exclusive lock on a file in the + staging directory, which the kernel drops if its holder dies. The file stays, so every writer locks the + same one.""" marker = self._root / _STAGING_DIR / f"{hashlib.sha256(str(path).encode()).hexdigest()}.claim" - while True: + with marker.open("a") as f: + fcntl.flock(f, fcntl.LOCK_EX) try: - fd = os.open(marker, os.O_CREAT | os.O_EXCL | os.O_WRONLY) - except FileExistsError: - if _abandoned(marker): - marker.unlink(missing_ok=True) - else: - time.sleep(0.01) - continue - with os.fdopen(fd, "w") as f: - f.write(str(os.getpid())) - break - try: - yield - finally: - marker.unlink(missing_ok=True) + yield + finally: + fcntl.flock(f, fcntl.LOCK_UN) def _read_content_type(self, path: Path, st: os.stat_result | None = None) -> str | None: """The content type recorded for the object at ``path``, if it was recorded for this very file; None @@ -270,23 +262,6 @@ def _identity(st: os.stat_result) -> list[int]: return [st.st_size, st.st_mtime_ns] -def _abandoned(marker: Path) -> bool: - """Whether a claim marker's owner has exited, or never got to name itself.""" - try: - owner = marker.read_text() - age = time.time() - marker.stat().st_mtime - except FileNotFoundError: - return False - if not owner: - return age > _UNWRITTEN_CLAIM_SECONDS - try: - os.kill(int(owner), 0) - except ProcessLookupError: - return True - except (PermissionError, ValueError): - return False - return False - @contextlib.contextmanager def _missing_as_not_found(object_url: str) -> Iterator[None]: diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py index 860e6f78..cdbe38e5 100644 --- a/tst/unit/store/local_object_store_test.py +++ b/tst/unit/store/local_object_store_test.py @@ -234,26 +234,28 @@ def no_links(src, dst): store.put("seed", b"s") # creates the staging directory -def test_without_hard_links_a_claim_whose_owner_exited_is_taken_over(store, tmp_path, no_hard_links): - exited = subprocess.run([sys.executable, "-c", "import os; print(os.getpid())"], capture_output=True, text=True) +def test_without_hard_links_a_claim_dies_with_its_holder(store, tmp_path, no_hard_links): + """The claim is a lock the kernel drops when its holder exits, so a crashed writer never blocks the key.""" marker = _claim_marker(tmp_path, "left/x") - marker.write_text(exited.stdout.strip()) - store.put("left/x", b"mine") + holder = subprocess.Popen( + [sys.executable, "-c", ( + "import fcntl, sys, time; f = open(sys.argv[1], 'a'); fcntl.flock(f, fcntl.LOCK_EX); " + "print('locked', flush=True); time.sleep(60)" + ), str(marker)], + stdout=subprocess.PIPE, text=True, + ) + try: + assert holder.stdout.readline().strip() == "locked" + writer = threading.Thread(target=store.put, args=("left/x", b"mine")) + writer.start() + writer.join(0.3) + assert writer.is_alive() and not store.exists("left/x") # a live holder, however old, is waited for + holder.kill() + holder.wait() + writer.join(5) + finally: + holder.kill() assert store.read("left/x") == b"mine" - assert not marker.exists() - - -def test_without_hard_links_a_live_writers_claim_is_never_taken_over(store, tmp_path, no_hard_links): - marker = _claim_marker(tmp_path, "held/x") - marker.write_text(str(os.getpid())) - os.utime(marker, (time.time() - 3600, time.time() - 3600)) - writer = threading.Thread(target=store.put, args=("held/x", b"late")) - writer.start() - writer.join(0.3) - assert writer.is_alive() and not store.exists("held/x") - marker.unlink() - writer.join(5) - assert store.read("held/x") == b"late" def test_types_survive_a_copy_that_keeps_modification_times(store, tmp_path): From d46296db1578b40c40be6eb43bd2ef683914123d Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 30 Sep 2026 09:50:18 -1000 Subject: [PATCH 6/8] fix(store): a key's bytes and type are written and read under its lock, and staged files of dead writes are swept Chaos testing on exFAT, which keeps modification times to 10 ms, showed two same-size overwrites of one key could leave one's type recorded against the other's bytes (5 in 4,129 races): the per-write time stamp that tells writes apart is rounded away there. A write now holds the key's lock (an exclusive flock) across placing its bytes and recording its type, and a metadata read holds it shared, so no write lands between another's two steps and no reader sees one write's type with another's bytes, whatever the filesystem's time resolution. An overwrite drops the old type first, so one cut short leaves the type unknown, never wrong. Keys share 64 lock files, which also replace the per-key claim files the no-hard-link path used; Windows, which has no flock, keeps relying on the stamp, which NTFS's 100 ns times preserve. A store's first write also removes files staged by writes that died (untouched for an hour), which crashes left behind in .agentenv-tmp. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../store/object_store/local_object_store.py | 132 +++++++++++------- tst/unit/store/local_object_store_test.py | 76 ++++++++-- 2 files changed, 143 insertions(+), 65 deletions(-) diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local_object_store.py index a60cf3c7..d15aab29 100644 --- a/src/agent_env/store/object_store/local_object_store.py +++ b/src/agent_env/store/object_store/local_object_store.py @@ -29,8 +29,11 @@ _META_DIR = ".agentenv-meta" # each object's content type, at the object's own key _STAGING_DIR = ".agentenv-tmp" # files being written, inside the root so a rename into place is atomic _RESERVED = (".gitignore", _META_DIR, _STAGING_DIR) -# A filesystem without hard links: a no-overwrite write claims its key with a lock instead. +# A filesystem without hard links: a no-overwrite write relies on the key's lock instead. _NO_HARD_LINKS = frozenset({errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK}) +_LOCK_STRIPES = 64 # keys share this many lock files, so locking leaves no file behind per key +_STAGED_PREFIX = "staged-" +_STALE_STAGED_SECONDS = 3600 # a staged file untouched this long belongs to a write that died class LocalFilesystemObjectStore(ObjectStore): @@ -41,6 +44,7 @@ class LocalFilesystemObjectStore(ObjectStore): def __init__(self, root: str) -> None: self._root = Path(root) + self._swept = False @property def root(self) -> Path: @@ -94,11 +98,13 @@ def open(self, object_url: str) -> BinaryIO: def get_object_metadata(self, key: str) -> ObjectMetadata | None: path = self._resolve(key) - if not path.is_file(): - return None - st = path.stat() + with self._locked(path, shared=True): # never between another write's bytes and its type + if not path.is_file(): + return None + st = path.stat() + content_type = self._read_content_type(path, st) return ObjectMetadata( - content_type=self._read_content_type(path, st), + content_type=content_type, size=st.st_size, last_modified=datetime.fromtimestamp(st.st_mtime, tz=timezone.utc), ) @@ -161,7 +167,10 @@ def _staged(self) -> Iterator[Path]: ensure_state_dir(self._root) staging = self._root / _STAGING_DIR staging.mkdir(exist_ok=True) - fd, name = tempfile.mkstemp(dir=staging) + if not self._swept: + self._swept = True + _sweep(staging) + fd, name = tempfile.mkstemp(dir=staging, prefix=_STAGED_PREFIX) os.close(fd) staged = Path(name) staged.chmod(0o644) @@ -171,59 +180,70 @@ def _staged(self) -> Iterator[Path]: staged.unlink(missing_ok=True) def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrite: bool) -> None: - """Put a staged file in place at ``path`` and record its content type. An overwrite renames over - whatever is there; otherwise a hard link claims ``path`` only if no other writer got there first. + """Put a staged file in place at ``path`` and record its content type, both under the key's lock, so no + other write to the key lands in between and no reader sees one write's type with another's bytes. An + overwrite drops the old type first, so one cut short leaves the type unknown, never wrong. - The type is recorded with the identity of the file it describes, so one recorded for another - write (a racing overwrite, or one whose type could not be recorded) reads back as unknown.""" + The type is also recorded with the identity of the file it describes, which is what keeps them paired + where there is no lock (Windows, whose NTFS keeps times fine enough to tell writes apart).""" identity = _stamp(staged) + meta = self._meta_path(path) path.parent.mkdir(parents=True, exist_ok=True) - if allow_overwrite: - os.replace(staged, path) - else: - try: - os.link(staged, path) - except FileExistsError: - raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None - except OSError as e: - if e.errno not in _NO_HARD_LINKS: - raise - if os.name == "nt": # rename there refuses an existing target, which is the claim + with self._locked(path): + if allow_overwrite: + meta.unlink(missing_ok=True) + os.replace(staged, path) + else: + try: + os.link(staged, path) # exclusive even against a writer that takes no lock + except FileExistsError: + raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None + except OSError as e: + if e.errno not in _NO_HARD_LINKS: + raise + # The lock makes the check and the rename one step; on Windows, rename refuses an existing target. + if path.exists(): + raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None try: - os.rename(staged, path) + (os.rename if os.name == "nt" else os.replace)(staged, path) except FileExistsError: raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None - else: - with self._claimed(path): - if path.exists(): - raise ObjectAlreadyExistsError(f"Object already exists at {path}.") from None - os.replace(staged, path) - # The default type is what a writer that named none gets; it reads back as unknown, so readers - # still guess from the name as they did before types were kept. - try: - meta = self._meta_path(path) - if content_type == DEFAULT_CONTENT_TYPE: - meta.unlink(missing_ok=True) - return - meta.parent.mkdir(parents=True, exist_ok=True) - with self._staged() as staged_meta: - staged_meta.write_text(json.dumps({"content_type": content_type, "object": identity})) - os.replace(staged_meta, meta) - except OSError as e: # the object is in place and its type reads back as unknown - logger.warning("Could not record the content type of %s: %s", path, e) + # The default type is what a writer that named none gets; it reads back as unknown, so readers + # still guess from the name as they did before types were kept. + try: + if content_type == DEFAULT_CONTENT_TYPE: + meta.unlink(missing_ok=True) + return + meta.parent.mkdir(parents=True, exist_ok=True) + with self._staged() as staged_meta: + staged_meta.write_text(json.dumps({"content_type": content_type, "object": identity})) + os.replace(staged_meta, meta) + except OSError as e: # the object is in place and its type reads back as unknown + logger.warning("Could not record the content type of %s: %s", path, e) @contextlib.contextmanager - def _claimed(self, path: Path) -> Iterator[None]: - """The sole right to create ``path``, where no hard link can claim it: an exclusive lock on a file in the - staging directory, which the kernel drops if its holder dies. The file stays, so every writer locks the - same one.""" - marker = self._root / _STAGING_DIR / f"{hashlib.sha256(str(path).encode()).hexdigest()}.claim" - with marker.open("a") as f: - fcntl.flock(f, fcntl.LOCK_EX) - try: - yield - finally: - fcntl.flock(f, fcntl.LOCK_UN) + def _locked(self, path: Path, *, shared: bool = False) -> Iterator[None]: + """The key's lock, exclusive to write it and shared to read its type: a flock on one of a fixed set of + files in the staging directory, which the kernel drops if its holder dies. None on Windows.""" + if os.name == "nt": + yield + return + try: + f = self._lock_path(path).open("r" if shared else "a") + except OSError: + if not shared: + raise + f = None # no write has taken this lock yet, or the store is read-only: nothing to wait for + if f is None: + yield + return + with f: + fcntl.flock(f, fcntl.LOCK_SH if shared else fcntl.LOCK_EX) + yield + + def _lock_path(self, path: Path) -> Path: + stripe = int(hashlib.sha256(str(path).encode()).hexdigest()[:8], 16) % _LOCK_STRIPES + return self._root / _STAGING_DIR / f"lock-{stripe:02x}" def _read_content_type(self, path: Path, st: os.stat_result | None = None) -> str | None: """The content type recorded for the object at ``path``, if it was recorded for this very file; None @@ -258,6 +278,16 @@ def _stamp(staged: Path) -> list[int]: return _identity(staged.stat()) +def _sweep(staging: Path) -> None: + """Remove files staged by writes that died. A write in progress keeps touching its file, and a finished one + renames it away, so one untouched for an hour will never be committed.""" + cutoff = time.time() - _STALE_STAGED_SECONDS + for staged in staging.glob(f"{_STAGED_PREFIX}*"): + with contextlib.suppress(OSError): + if staged.stat().st_mtime < cutoff: + staged.unlink() + + def _identity(st: os.stat_result) -> list[int]: return [st.st_size, st.st_mtime_ns] diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py index cdbe38e5..72213a13 100644 --- a/tst/unit/store/local_object_store_test.py +++ b/tst/unit/store/local_object_store_test.py @@ -220,9 +220,17 @@ def slow_first_replace(src, dst): assert store.read("race/once") == b"first" -def _claim_marker(tmp_path, key): - path = (tmp_path / key).resolve() - return tmp_path / ".agentenv-tmp" / f"{hashlib.sha256(str(path).encode()).hexdigest()}.claim" +def _hold_lock(lock_file): + """A process holding ``lock_file`` exclusively until killed, as a writer mid-commit would.""" + holder = subprocess.Popen( + [sys.executable, "-c", ( + "import fcntl, sys, time; f = open(sys.argv[1], 'a'); fcntl.flock(f, fcntl.LOCK_EX); " + "print('locked', flush=True); time.sleep(60)" + ), str(lock_file)], + stdout=subprocess.PIPE, text=True, + ) + assert holder.stdout.readline().strip() == "locked" + return holder @pytest.fixture @@ -236,20 +244,12 @@ def no_links(src, dst): def test_without_hard_links_a_claim_dies_with_its_holder(store, tmp_path, no_hard_links): """The claim is a lock the kernel drops when its holder exits, so a crashed writer never blocks the key.""" - marker = _claim_marker(tmp_path, "left/x") - holder = subprocess.Popen( - [sys.executable, "-c", ( - "import fcntl, sys, time; f = open(sys.argv[1], 'a'); fcntl.flock(f, fcntl.LOCK_EX); " - "print('locked', flush=True); time.sleep(60)" - ), str(marker)], - stdout=subprocess.PIPE, text=True, - ) + holder = _hold_lock(store._lock_path((tmp_path / "left/x").resolve())) try: - assert holder.stdout.readline().strip() == "locked" writer = threading.Thread(target=store.put, args=("left/x", b"mine")) writer.start() writer.join(0.3) - assert writer.is_alive() and not store.exists("left/x") # a live holder, however old, is waited for + assert writer.is_alive() and not (tmp_path / "left/x").exists() # a live holder is waited for holder.kill() holder.wait() writer.join(5) @@ -295,7 +295,55 @@ def test_listings_leave_out_the_stores_own_files(store, tmp_path): (tmp_path / ".agentenv-tmp" / "crash-residue").write_bytes(b"half") assert store.list("") == ["a/x.json"] assert store.list_at(store.object_url("")) == [store.object_url("a/x.json")] - assert os.listdir(tmp_path / ".agentenv-tmp") == ["crash-residue"] # committed writes leave nothing staged + staging = [name for name in os.listdir(tmp_path / ".agentenv-tmp") if not name.startswith("lock-")] + assert staging == ["crash-residue"] # committed writes leave nothing staged + + +def test_a_type_is_never_read_while_a_write_to_the_key_is_in_progress(store, tmp_path): + store.put("busy/x", b"typed", content_type="text/plain") + holder = _hold_lock(store._lock_path((tmp_path / "busy/x").resolve())) + try: + read = [] + reader = threading.Thread(target=lambda: read.append(store.get_object_metadata("busy/x"))) + reader.start() + reader.join(0.3) + assert reader.is_alive() # waits for the write holding the key + holder.kill() + holder.wait() + reader.join(5) + finally: + holder.kill() + assert read[0].content_type == "text/plain" + + +def test_an_overwrite_cut_short_leaves_the_type_unknown_not_the_old_one(store, tmp_path, monkeypatch): + store.put("cut/x", b"first", content_type="text/plain") + real_replace = os.replace + + def replace(src, dst): + if str(dst).endswith("cut/x") and ".agentenv-meta" not in str(dst): + raise OSError(errno.EIO, "disk went away") + real_replace(src, dst) + + monkeypatch.setattr(os, "replace", replace) + with pytest.raises(OSError, match="disk went away"): + store.put("cut/x", b"second", content_type="application/json", allow_overwrite=True) + assert (store.read("cut/x"), store.get_object_metadata("cut/x").content_type) == (b"first", None) + + +def test_a_store_sweeps_files_staged_by_writes_that_died(tmp_path): + staging = tmp_path / ".agentenv-tmp" + staging.mkdir() + dead, live = staging / "staged-dead", staging / "staged-live" + dead.write_bytes(b"half") + live.write_bytes(b"in progress") + os.utime(dead, (time.time() - 7200, time.time() - 7200)) + (staging / "lock-00").write_bytes(b"") + os.utime(staging / "lock-00", (time.time() - 7200, time.time() - 7200)) + + LocalFilesystemObjectStore(str(tmp_path)).put("k", b"v") + + assert not dead.exists() and live.exists() and (staging / "lock-00").exists() def test_list_of_an_unwritten_store_is_empty(tmp_path): From 1b8b548d60936a8ba296155a61c86ddedc170f17 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 30 Sep 2026 10:26:37 -1000 Subject: [PATCH 7/8] fix(store): a failed overwrite keeps the old type, and the sweep spares any file a live write holds An overwrite now sets the old type record aside while its bytes land and puts it back if they do not, so a rename that fails keeps the old object's type; one cut short after its bytes land still reads as unknown, never as the old type. Each staged file is held under an exclusive flock by its writer for as long as the write lives, and the sweep removes only staged files no process holds (and none younger than a minute, which may not be held yet), so a write paused for any length of time never loses its file. Windows, which has no flock, is not swept. Keys now share 256 lock files rather than 64, so fewer unrelated keys wait on one another, and the tests that hold a lock with flock skip on Windows. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../store/object_store/local_object_store.py | 53 +++++++++++++++---- tst/unit/store/local_object_store_test.py | 50 +++++++++++------ 2 files changed, 77 insertions(+), 26 deletions(-) diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local_object_store.py index d15aab29..27dbf90c 100644 --- a/src/agent_env/store/object_store/local_object_store.py +++ b/src/agent_env/store/object_store/local_object_store.py @@ -31,9 +31,9 @@ _RESERVED = (".gitignore", _META_DIR, _STAGING_DIR) # A filesystem without hard links: a no-overwrite write relies on the key's lock instead. _NO_HARD_LINKS = frozenset({errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK}) -_LOCK_STRIPES = 64 # keys share this many lock files, so locking leaves no file behind per key +_LOCK_STRIPES = 256 # keys share this many lock files, so locking leaves no file behind per key _STAGED_PREFIX = "staged-" -_STALE_STAGED_SECONDS = 3600 # a staged file untouched this long belongs to a write that died +_STAGED_GRACE_SECONDS = 60 # a staged file younger than this may not be locked by its writer yet class LocalFilesystemObjectStore(ObjectStore): @@ -171,12 +171,17 @@ def _staged(self) -> Iterator[Path]: self._swept = True _sweep(staging) fd, name = tempfile.mkstemp(dir=staging, prefix=_STAGED_PREFIX) - os.close(fd) staged = Path(name) - staged.chmod(0o644) + if os.name == "nt": # an open file cannot be renamed there, and there is no flock to hold + os.close(fd) + else: + fcntl.flock(fd, fcntl.LOCK_EX) # held while the write lives, so a sweep never takes its file try: + staged.chmod(0o644) yield staged finally: + if os.name != "nt": + os.close(fd) staged.unlink(missing_ok=True) def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrite: bool) -> None: @@ -191,8 +196,18 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit path.parent.mkdir(parents=True, exist_ok=True) with self._locked(path): if allow_overwrite: - meta.unlink(missing_ok=True) - os.replace(staged, path) + # The old type is set aside while the bytes land: a write cut short reads as unknown, never as the + # old type, and one that fails before its bytes land puts the old type back. + aside = self._set_aside(meta) + try: + os.replace(staged, path) + except BaseException: + if aside is not None: + with contextlib.suppress(FileNotFoundError): # swept meanwhile: the type reads as unknown + os.replace(aside, meta) + raise + if aside is not None: + aside.unlink(missing_ok=True) else: try: os.link(staged, path) # exclusive even against a writer that takes no lock @@ -241,6 +256,17 @@ def _locked(self, path: Path, *, shared: bool = False) -> Iterator[None]: fcntl.flock(f, fcntl.LOCK_SH if shared else fcntl.LOCK_EX) yield + def _set_aside(self, meta: Path) -> Path | None: + """Move a type record into the staging directory, fresh enough that no sweep takes it; None when there is + none.""" + aside = self._root / _STAGING_DIR / f"{_STAGED_PREFIX}aside-{secrets.token_hex(8)}" + try: + os.replace(meta, aside) + except FileNotFoundError: + return None + os.utime(aside) + return aside + def _lock_path(self, path: Path) -> Path: stripe = int(hashlib.sha256(str(path).encode()).hexdigest()[:8], 16) % _LOCK_STRIPES return self._root / _STAGING_DIR / f"lock-{stripe:02x}" @@ -279,12 +305,17 @@ def _stamp(staged: Path) -> list[int]: def _sweep(staging: Path) -> None: - """Remove files staged by writes that died. A write in progress keeps touching its file, and a finished one - renames it away, so one untouched for an hour will never be committed.""" - cutoff = time.time() - _STALE_STAGED_SECONDS + """Remove files staged by writes that died: each live write holds a lock on its file, which the kernel drops + when the writer exits, so a file no one holds will never be committed. Not on Windows, which has no flock.""" + if os.name == "nt": + return + cutoff = time.time() - _STAGED_GRACE_SECONDS for staged in staging.glob(f"{_STAGED_PREFIX}*"): - with contextlib.suppress(OSError): - if staged.stat().st_mtime < cutoff: + with contextlib.suppress(OSError): # BlockingIOError, the one that matters: a live write holds it + if staged.stat().st_mtime > cutoff: + continue + with staged.open("rb") as f: + fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB) staged.unlink() diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py index 72213a13..549c00fa 100644 --- a/tst/unit/store/local_object_store_test.py +++ b/tst/unit/store/local_object_store_test.py @@ -220,6 +220,9 @@ def slow_first_replace(src, dst): assert store.read("race/once") == b"first" +_needs_flock = pytest.mark.skipif(os.name == "nt", reason="holds a lock with flock, which Windows lacks") + + def _hold_lock(lock_file): """A process holding ``lock_file`` exclusively until killed, as a writer mid-commit would.""" holder = subprocess.Popen( @@ -242,6 +245,7 @@ def no_links(src, dst): store.put("seed", b"s") # creates the staging directory +@_needs_flock def test_without_hard_links_a_claim_dies_with_its_holder(store, tmp_path, no_hard_links): """The claim is a lock the kernel drops when its holder exits, so a crashed writer never blocks the key.""" holder = _hold_lock(store._lock_path((tmp_path / "left/x").resolve())) @@ -299,6 +303,7 @@ def test_listings_leave_out_the_stores_own_files(store, tmp_path): assert staging == ["crash-residue"] # committed writes leave nothing staged +@_needs_flock def test_a_type_is_never_read_while_a_write_to_the_key_is_in_progress(store, tmp_path): store.put("busy/x", b"typed", content_type="text/plain") holder = _hold_lock(store._lock_path((tmp_path / "busy/x").resolve())) @@ -316,34 +321,49 @@ def test_a_type_is_never_read_while_a_write_to_the_key_is_in_progress(store, tmp assert read[0].content_type == "text/plain" -def test_an_overwrite_cut_short_leaves_the_type_unknown_not_the_old_one(store, tmp_path, monkeypatch): - store.put("cut/x", b"first", content_type="text/plain") +def _failing_replace(monkeypatch, suffix): real_replace = os.replace def replace(src, dst): - if str(dst).endswith("cut/x") and ".agentenv-meta" not in str(dst): + if str(dst).endswith(suffix): raise OSError(errno.EIO, "disk went away") real_replace(src, dst) monkeypatch.setattr(os, "replace", replace) + + +def test_an_overwrite_whose_bytes_do_not_land_keeps_the_old_type(store, tmp_path, monkeypatch): + store.put("cut/x", b"first", content_type="text/plain") + _failing_replace(monkeypatch, str(tmp_path / "cut/x")) with pytest.raises(OSError, match="disk went away"): store.put("cut/x", b"second", content_type="application/json", allow_overwrite=True) - assert (store.read("cut/x"), store.get_object_metadata("cut/x").content_type) == (b"first", None) + assert (store.read("cut/x"), store.get_object_metadata("cut/x").content_type) == (b"first", "text/plain") -def test_a_store_sweeps_files_staged_by_writes_that_died(tmp_path): - staging = tmp_path / ".agentenv-tmp" - staging.mkdir() - dead, live = staging / "staged-dead", staging / "staged-live" - dead.write_bytes(b"half") - live.write_bytes(b"in progress") - os.utime(dead, (time.time() - 7200, time.time() - 7200)) - (staging / "lock-00").write_bytes(b"") - os.utime(staging / "lock-00", (time.time() - 7200, time.time() - 7200)) +def test_an_overwrite_whose_type_cannot_be_recorded_reads_unknown_not_the_old_type(store, tmp_path, monkeypatch): + store.put("cut/y", b"first", content_type="text/plain") + _failing_replace(monkeypatch, str(tmp_path / ".agentenv-meta" / "cut/y")) + store.put("cut/y", b"second!", content_type="application/json", allow_overwrite=True) + assert (store.read("cut/y"), store.get_object_metadata("cut/y").content_type) == (b"second!", None) - LocalFilesystemObjectStore(str(tmp_path)).put("k", b"v") - assert not dead.exists() and live.exists() and (staging / "lock-00").exists() +@_needs_flock +def test_a_store_sweeps_only_files_no_live_write_holds(tmp_path): + staging = tmp_path / ".agentenv-tmp" + staging.mkdir() + hour_ago = (time.time() - 3600, time.time() - 3600) + dead, paused, fresh = staging / "staged-dead", staging / "staged-paused", staging / "staged-fresh" + for f in (dead, paused, fresh, staging / "lock-00"): + f.write_bytes(b"half") + for f in (dead, paused, staging / "lock-00"): + os.utime(f, hour_ago) + holder = _hold_lock(paused) # a live write paused for an hour + try: + LocalFilesystemObjectStore(str(tmp_path)).put("k", b"v") + finally: + holder.kill() + assert not dead.exists() + assert paused.exists() and fresh.exists() and (staging / "lock-00").exists() def test_list_of_an_unwritten_store_is_empty(tmp_path): From 35ea9bc9f60013f9fbb3bc5a43829a06f3ad134b Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 30 Sep 2026 10:36:41 -1000 Subject: [PATCH 8/8] fix(store): an overwrite that lands is never reported as failed, and a staged file that cannot be locked is cleaned up Removing the set-aside type record after an overwrite's bytes land, and refreshing its time when it is set aside, are now best effort: a failure there leaves a file the sweep removes, rather than reporting a completed write as failed. A staged file whose lock cannot be taken is closed and removed like any other. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../store/object_store/local_object_store.py | 17 +++++++----- tst/unit/store/local_object_store_test.py | 26 +++++++++++++++++++ 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local_object_store.py index 27dbf90c..58038aa1 100644 --- a/src/agent_env/store/object_store/local_object_store.py +++ b/src/agent_env/store/object_store/local_object_store.py @@ -172,15 +172,16 @@ def _staged(self) -> Iterator[Path]: _sweep(staging) fd, name = tempfile.mkstemp(dir=staging, prefix=_STAGED_PREFIX) staged = Path(name) - if os.name == "nt": # an open file cannot be renamed there, and there is no flock to hold - os.close(fd) - else: - fcntl.flock(fd, fcntl.LOCK_EX) # held while the write lives, so a sweep never takes its file try: + if os.name == "nt": # an open file cannot be renamed there, and there is no flock to hold + os.close(fd) + fd = None + else: + fcntl.flock(fd, fcntl.LOCK_EX) # held while the write lives, so a sweep never takes its file staged.chmod(0o644) yield staged finally: - if os.name != "nt": + if fd is not None: os.close(fd) staged.unlink(missing_ok=True) @@ -207,7 +208,8 @@ def _commit(self, staged: Path, path: Path, content_type: str, *, allow_overwrit os.replace(aside, meta) raise if aside is not None: - aside.unlink(missing_ok=True) + with contextlib.suppress(OSError): # the bytes have landed; a sweep takes what is left + aside.unlink(missing_ok=True) else: try: os.link(staged, path) # exclusive even against a writer that takes no lock @@ -264,7 +266,8 @@ def _set_aside(self, meta: Path) -> Path | None: os.replace(meta, aside) except FileNotFoundError: return None - os.utime(aside) + with contextlib.suppress(OSError): + os.utime(aside) return aside def _lock_path(self, path: Path) -> Path: diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py index 549c00fa..20387986 100644 --- a/tst/unit/store/local_object_store_test.py +++ b/tst/unit/store/local_object_store_test.py @@ -17,6 +17,7 @@ import pytest from agent_env.store import LocalFilesystemObjectStore, ObjectAlreadyExistsError +from agent_env.store.object_store import local_object_store from tst.store import object_conformance @@ -340,6 +341,31 @@ def test_an_overwrite_whose_bytes_do_not_land_keeps_the_old_type(store, tmp_path assert (store.read("cut/x"), store.get_object_metadata("cut/x").content_type) == (b"first", "text/plain") +def test_an_overwrite_whose_cleanup_fails_still_succeeds(store, tmp_path, monkeypatch): + store.put("clean/x", b"first", content_type="text/plain") + real_unlink = Path.unlink + + def unlink(self, missing_ok=False): + if "aside-" in self.name: + raise OSError(errno.EIO, "disk went away") + real_unlink(self, missing_ok=missing_ok) + + monkeypatch.setattr(Path, "unlink", unlink) + store.put("clean/x", b"second!", content_type="application/json", allow_overwrite=True) + assert (store.read("clean/x"), store.get_object_metadata("clean/x").content_type) == (b"second!", "application/json") + + +@_needs_flock +def test_a_write_that_cannot_lock_its_staged_file_leaves_nothing_staged(store, tmp_path, monkeypatch): + def no_lock(fd, op): + raise OSError(errno.ENOLCK, "no locks available") + + monkeypatch.setattr(local_object_store.fcntl, "flock", no_lock) + with pytest.raises(OSError, match="no locks"): + store.put("nolock/x", b"v") + assert [n for n in os.listdir(tmp_path / ".agentenv-tmp") if n.startswith("staged-")] == [] + + def test_an_overwrite_whose_type_cannot_be_recorded_reads_unknown_not_the_old_type(store, tmp_path, monkeypatch): store.put("cut/y", b"first", content_type="text/plain") _failing_replace(monkeypatch, str(tmp_path / ".agentenv-meta" / "cut/y"))