From 66b2728e14b361c72a50295f66013bf5b235a543 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 22 Sep 2026 14:36:07 +0000 Subject: [PATCH] fix(store): packed cache hides own writes; EOFError escapes; no fsync Post-merge review of #83. - A writer whose disk-clear was skipped by the _packed_stale guard loaded a packed set another process published before its latest writes, so it did not see records it had just written. _load_packed now treats the disk cache as a miss while this process has unpublished writes; the rebuild republishes, which also heals the cache for other readers. - np.load of an empty/truncated .npy raises EOFError, which escaped _load_packed and made every matrix() call raise while that sig stayed. - Data files and the sig tmp are fsynced before os.replace, and the directory after (POSIX), so a power loss cannot leave a durable sig naming data blocks that never reached disk. Co-Authored-By: Claude Opus 5 --- ir/store.py | 72 +++++++++++++++++++++++++++++--------- tests/test_store_packed.py | 61 ++++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+), 17 deletions(-) diff --git a/ir/store.py b/ir/store.py index 480debb..2c22da4 100644 --- a/ir/store.py +++ b/ir/store.py @@ -47,6 +47,34 @@ _PACKED_FORMAT = 1 +def _write_durably(path: Path, write) -> None: + """Create ``path``, let ``write(f)`` fill it, then flush and fsync it.""" + with open(path, "wb") as f: + write(f) + f.flush() + os.fsync(f.fileno()) + + +def _fsync_dir(path: Path) -> None: + """Fsync a directory so a rename in it is durable (no-op where unsupported). + + POSIX needs this for ``os.replace`` to survive a power loss; Windows cannot + open a directory this way (and NTFS journals the rename), so it is skipped. + """ + if os.name == "nt": + return + try: + fd = os.open(path, os.O_RDONLY) + except OSError: + return + try: + os.fsync(fd) + except OSError: + pass + finally: + os.close(fd) + + def _packed_content_sig(ids_json: bytes, metas_json: bytes) -> str: """Signature binding a packed matrix to the exact ids/metas written with it. @@ -456,6 +484,12 @@ def _load_packed(self): """Load ``(ids, mmap_matrix, metas)`` from the packed cache, or ``None``.""" if self._packed_dir is None: return None + if self._packed_stale: + # This process wrote records since it last published a matrix. A + # packed set on disk now is either absent (we cleared it) or was + # published by another process that may have built before our + # writes, so it cannot be trusted to include them: rebuild. + return None sig_path = self._packed_paths()["sig"] if not sig_path.exists(): return None @@ -474,7 +508,9 @@ def _load_packed(self): metas_json = paths["metas"].read_bytes() ids = json.loads(ids_json) metas = json.loads(metas_json) - except (OSError, ValueError): + except (OSError, ValueError, EOFError): + # EOFError: np.load of an empty/truncated ``.npy`` (e.g. a crash + # before the data reached disk). Any unreadable set is a miss. return None if len(ids) != mat.shape[0] or len(metas) != len(ids): return None @@ -501,7 +537,10 @@ def _save_packed(self, result: tuple[list[str], np.ndarray, list[dict]]) -> None crash mid-write leaves the previous sig (or none) in charge, and two concurrent writers can never leave one writer's matrix beside the other's ids: a reader follows ``sig.json`` to exactly one writer's - complete set. Older generations are swept after publishing. + complete set. Older generations are swept after publishing. Every data + file is fsynced before the sig that names it is published (and the + directory after), so a power loss cannot leave a durable ``sig.json`` + pointing at data blocks that never reached the disk. """ if self._packed_dir is None: return @@ -515,23 +554,22 @@ def _save_packed(self, result: tuple[list[str], np.ndarray, list[dict]]) -> None arr = np.asarray(mat, dtype=np.float32) ids_json = json.dumps(ids).encode("utf-8") metas_json = json.dumps(metas).encode("utf-8") - np.save(paths["matrix"], arr) - paths["ids"].write_bytes(ids_json) - paths["metas"].write_bytes(metas_json) + _write_durably(paths["matrix"], lambda f: np.save(f, arr)) + _write_durably(paths["ids"], lambda f: f.write(ids_json)) + _write_durably(paths["metas"], lambda f: f.write(metas_json)) + sig_json = json.dumps( + { + "format": _PACKED_FORMAT, + "count": len(ids), + "generation": generation, + "content_sig": _packed_content_sig(ids_json, metas_json), + "shape": list(arr.shape), + } + ).encode("utf-8") sig_tmp = self._packed_dir / f"sig-{generation}.tmp" - sig_tmp.write_text( - json.dumps( - { - "format": _PACKED_FORMAT, - "count": len(ids), - "generation": generation, - "content_sig": _packed_content_sig(ids_json, metas_json), - "shape": list(arr.shape), - } - ), - encoding="utf-8", - ) + _write_durably(sig_tmp, lambda f: f.write(sig_json)) os.replace(sig_tmp, paths["sig"]) + _fsync_dir(self._packed_dir) self._packed_stale = False except OSError: # A read cache that can't be written is non-fatal: fall back to the diff --git a/tests/test_store_packed.py b/tests/test_store_packed.py index 8e7efec..aedf15e 100644 --- a/tests/test_store_packed.py +++ b/tests/test_store_packed.py @@ -7,6 +7,7 @@ """ import json +import os import numpy as np @@ -264,3 +265,63 @@ def test_republishing_sweeps_older_generations(tmp_path): store.put_record(_rec("r2")) # a write clears the cache entirely assert list((root / "matrix").iterdir()) == [] + + +def test_writer_sees_its_own_write_despite_another_process_publishing(tmp_path): + """A packed set published by another process must not hide our own writes. + + Writer P1 clears the cache on its first write (and then skips the clear on + later writes, which is what keeps bulk builds cheap). If reader P2 publishes + a matrix in between, P1's next write leaves that set on disk, and P1 then + loaded it -- missing the record it had just written. + """ + p1, _root = _file_store(tmp_path) + p1.put_record(_rec("r1", vec=(1.0, 0.0, 0.0))) + p2, _ = _file_store(tmp_path) + assert p2.matrix()[0] == ["r1"] # P2 builds and publishes {r1} + p1.put_record(_rec("r2", vec=(0.0, 1.0, 0.0))) + assert sorted(p1.matrix()[0]) == ["r1", "r2"] + assert sorted(p1.metas()[0]) == ["r1", "r2"] + # ...and P1's rebuild republished, so a fresh reader now sees both too. + p3, _ = _file_store(tmp_path) + assert sorted(p3.matrix()[0]) == ["r1", "r2"] + + +def test_empty_or_truncated_packed_matrix_is_a_miss_not_an_error(tmp_path): + """A sig naming a zero-length/truncated ``.npy`` (crash before the data hit + disk) must rebuild, not raise ``EOFError`` from every ``matrix()`` call.""" + store, root = _file_store(tmp_path) + store.put_record(_rec("r1", vec=(3.0, 0.0, 0.0))) + store.put_record(_rec("r2", vec=(0.0, 4.0, 0.0))) + store.matrix() + matrix_file = _packed_file(root, "matrix") + full = matrix_file.read_bytes() + for corrupt in (b"", full[:20], full[:-4]): + matrix_file.write_bytes(corrupt) + store2, _ = _file_store(tmp_path) + store2._save_packed = lambda result: None # keep the corrupt set on disk + ids2, mat2, _metas2 = store2.matrix() + assert sorted(ids2) == ["r1", "r2"] + assert np.asarray(mat2).shape == (2, 3) + + +def test_packed_files_are_fsynced_before_sig_is_published(tmp_path, monkeypatch): + """Each data file is fsynced before ``sig.json`` names it.""" + import ir.store as ir_store + + store, root = _file_store(tmp_path) + store.put_record(_rec("r1")) + events = [] + real_fsync, real_replace = os.fsync, os.replace + monkeypatch.setattr( + ir_store.os, "fsync", lambda fd: (events.append("fsync"), real_fsync(fd)) + ) + monkeypatch.setattr( + ir_store.os, + "replace", + lambda a, b: (events.append("replace"), real_replace(a, b)), + ) + store.matrix() + assert "replace" in events + assert events.index("replace") >= 4 # matrix, ids, metas, sig tmp first + assert _packed_file(root, "matrix").exists()