fix(store): packed cache hid a writer's own records; EOFError escaped; add fsync (review of #83)

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

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