Merge branch 'main' into fix/cascade-sync-single-writer

Author: gloryfromca

src/everos/memory/cascade/watcher.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Awaitable
 from pathlib import Path
 
 from watchdog.events import FileMovedEvent, FileSystemEvent, FileSystemEventHandler
@@ -78,6 +79,14 @@ def __init__(
     ) -> None:
         self._memory_root = memory_root
         self._loop = loop
+        # Upserts for one path must commit in delivery order. Each one awaits
+        # the database, so left concurrent the last committer wins and a stale
+        # duplicate overwrites a newer row. Windows synthesises a ``created``
+        # for every file under a freshly created parent directory, handing the
+        # same file to this handler four or five times; on the soak box one of
+        # those duplicates landed after an atomic save's ``added`` and put the
+        # first write's mtime back on the row (1 run in 4).
+        self._in_order = asyncio.Lock()
 
     def on_created(self, event: FileSystemEvent) -> None:
         self._enqueue(event.src_path, "added")
@@ -124,10 +133,14 @@ def _enqueue(self, raw_path: str, change_type: str) -> None:
             return
         mtime = _safe_mtime(raw_path)
         asyncio.run_coroutine_threadsafe(
-            _enqueue_async(spec, rel, change_type, mtime),
+            self._serialised(_enqueue_async(spec, rel, change_type, mtime)),
             self._loop,
         )
 
+    async def _serialised(self, upsert: Awaitable[None]) -> None:
+        async with self._in_order:
+            await upsert
+
 
 async def _enqueue_async(
     spec: KindSpec, rel: str, change_type: str, mtime: float
@@ -152,15 +165,30 @@ async def _enqueue_async(
 def _relative_to_root(root: Path, raw: str) -> str | None:
     """Return ``raw`` relative to ``root`` using POSIX separators.
 
-    ``None`` when the path is outside the memory root (defensive — the
-    watcher only watches inside ``root``, but external symlinks could
-    surface).
+    watchdog reports absolute paths inside the watched root (``root`` is
+    already resolved by ``MemoryRoot``), so the common case is a pure string
+    operation and matches the scanner's key byte for byte. ``resolve()`` on
+    Windows is two ``GetFinalPathNameByHandle`` opens plus a ``stat()`` per
+    event, and a per-component walk when the path is already gone — about
+    40 % of the worker threads' CPU under write load (py-spy on the 2 h
+    soak). It is kept for the defensive case only: a textual path that is
+    not under ``root`` (an external symlink surfacing inside the tree) or
+    one carrying ``..`` segments. Known residue: if ``ReadDirectoryChangesW``
+    reports a file by its 8.3 short name, the key differs from the
+    scanner's long-name key until the next sweep reconciles it.
+
+    ``None`` when the path is outside the memory root even after resolving.
     """
+    path = Path(raw)
+    if ".." not in path.parts:
+        try:
+            return path.relative_to(root).as_posix()
+        except ValueError:
+            pass
     try:
-        rel = Path(raw).resolve().relative_to(root)
+        return path.resolve().relative_to(root).as_posix()
     except ValueError:
         return None
-    return rel.as_posix()
 
 
 def _safe_mtime(raw: str) -> float:

tests/unit/test_memory/test_cascade/test_watcher_helpers.py

@@ -7,8 +7,11 @@
 
 from __future__ import annotations
 
+import sys
 from pathlib import Path
 
+import pytest
+
 from everos.memory.cascade.watcher import _relative_to_root, _safe_mtime
 
 
@@ -34,3 +37,43 @@ def test_safe_mtime_existing_path_returns_positive(tmp_path: Path) -> None:
     f = tmp_path / "f.md"
     f.write_text("ok")
     assert _safe_mtime(str(f)) > 0
+
+
+def test_relative_to_root_within_does_not_touch_the_filesystem(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An in-root path is relativised textually — no ``resolve()``.
+
+    ``resolve()`` is a filesystem round trip per component (through
+    Defender on Windows) and the watcher calls this on every event.
+    """
+
+    def _boom(self: Path, strict: bool = False) -> Path:
+        raise AssertionError("resolve() must not run for an in-root path")
+
+    monkeypatch.setattr(Path, "resolve", _boom)
+    target = tmp_path / "users" / "u1" / "x.md"  # need not exist
+    assert _relative_to_root(tmp_path, str(target)) == "users/u1/x.md"
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need a privilege")
+def test_relative_to_root_via_symlink_still_resolves(
+    tmp_path: Path, tmp_path_factory: pytest.TempPathFactory
+) -> None:
+    """A path that reaches the root through an external symlink still maps."""
+    root = tmp_path.resolve()
+    (root / "users" / "u1").mkdir(parents=True)
+    (root / "users" / "u1" / "x.md").write_text("x", encoding="utf-8")
+    link = tmp_path_factory.mktemp("elsewhere") / "link_to_root"
+    link.symlink_to(root, target_is_directory=True)
+    assert (
+        _relative_to_root(root, str(link / "users" / "u1" / "x.md")) == "users/u1/x.md"
+    )
+
+
+def test_relative_to_root_normalises_dotdot_segments(tmp_path: Path) -> None:
+    """A ``..`` segment takes the resolving path so the key stays canonical."""
+    root = tmp_path.resolve()
+    (root / "users" / "u1").mkdir(parents=True)
+    raw = str(root / "users" / "u2" / ".." / "u1" / "x.md")
+    assert _relative_to_root(root, raw) == "users/u1/x.md"

tests/unit/test_memory/test_cascade/test_watcher_upsert_order.py

@@ -0,0 +1,70 @@
+"""Upserts for one path commit in delivery order.
+
+Windows synthesises a ``created`` for every file under a freshly created
+parent directory, so one ``mkdir -p`` + write hands the handler the same path
+four or five times. Each upsert awaits the database; run concurrently, the
+last committer wins, and on the Windows soak box a stale duplicate carrying
+the first write's mtime overwrote the row of an atomic save (1 run in 4).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from pathlib import Path
+
+import pytest
+from watchdog.events import FileCreatedEvent
+
+from everos.core.persistence import MemoryRoot
+from everos.memory.cascade import watcher as watcher_mod
+from everos.memory.cascade.watcher import _Handler
+
+_M1 = 1_700_000_000
+_M2 = 1_700_000_060
+
+
+class _SlowFirstRepo:
+    """The first upsert commits 50 ms late; the row is whatever committed last."""
+
+    def __init__(self) -> None:
+        self.calls = 0
+        self.committed: list[float] = []
+
+    async def upsert(
+        self, md_path: str, *, kind: str, change_type: str, mtime: float
+    ) -> int:
+        self.calls += 1
+        if self.calls == 1:
+            await asyncio.sleep(0.05)
+        self.committed.append(mtime)
+        return self.calls
+
+
+async def test_duplicate_events_commit_in_delivery_order(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    repo = _SlowFirstRepo()
+    monkeypatch.setattr(watcher_mod, "md_change_state_repo", repo)
+    md = tmp_path.joinpath(
+        "default_app",
+        "default_project",
+        "users",
+        "u1",
+        "episodes",
+        "episode-2026-01-01.md",
+    )
+    md.parent.mkdir(parents=True)
+    md.write_text("v1", encoding="utf-8")
+    handler = _Handler(MemoryRoot(tmp_path), asyncio.get_running_loop())
+
+    os.utime(md, (_M1, _M1))
+    handler.on_created(FileCreatedEvent(str(md)))  # the synthetic duplicate
+    os.utime(md, (_M2, _M2))
+    handler.on_created(FileCreatedEvent(str(md)))  # the save that must win
+    await asyncio.sleep(0.2)
+
+    assert repo.committed == [_M1, _M2], (
+        "the stale duplicate committed after the newer event; the row now "
+        "carries the old mtime"
+    )