fix(tagging): hold a tag write's paths against a scan that lands mid-batch

NEWS.md

@@ -16,6 +16,12 @@ border-radius: 128px;
       being dropped and re-created as a new comic on the next scan.
     - Renaming comics in a watched library keeps their bookmarks and read
       progress. PDFs lost them every time, other formats occasionally.
+    - Long tag write batches keep their bookmarks too. A library scan landing
+      partway through a big conversion run no longer deletes the comics it is
+      still converting.
+    - A watched library no longer mistakes an unrelated new file for a renamed
+      comic when a bulk conversion recycles a deleted file's identity, which
+      pointed one comic's row at another comic's file.
 
 ## v2.2.10
 

codex/librarian/fs/watcher/move.py

@@ -1,6 +1,9 @@
 """Watchfiles Move detection."""
 
+import os
+from dataclasses import dataclass
 from pathlib import Path
+from stat import S_IFMT, S_ISDIR
 
 from loguru import logger
 
@@ -10,8 +13,23 @@
 from codex.models.comic import Comic
 from codex.models.paths import CustomCover
 
-# stat field index for inode
+# stat field indexes, as stored by WatchedPath.set_stat
+_MODE_INDEX = 0
 _INODE_INDEX = 1
+_SIZE_INDEX = 6
+
+
+@dataclass(frozen=True, slots=True)
+class _DeletedEntry:
+    """A deleted path that could be the source of a move."""
+
+    index: int
+    library_pk: int
+    event: FSEvent
+    stat: list
+    # The same batch also reports this path as written, so its stored
+    # size predates that write. See ``_is_move_compatible``.
+    written: bool
 
 
 def _model_for_event(event: FSEvent):
@@ -23,28 +41,66 @@ def _model_for_event(event: FSEvent):
     return Comic
 
 
-def _get_db_inode(event: FSEvent, library_pk: int) -> int | None:
-    """Look up the inode for a path from the database stat field."""
+def _get_db_stat(event: FSEvent, library_pk: int) -> list | None:
+    """Return the stored stat for a path, when it carries a usable inode."""
     model = _model_for_event(event)
     stat = (
         model.objects.filter(library_id=library_pk, path=event.src_path)
         .values_list("stat", flat=True)
         .first()
     )
     if stat and len(stat) > _INODE_INDEX and stat[_INODE_INDEX]:
-        return stat[_INODE_INDEX]
+        return stat
     return None
 
 
-def _get_disk_inode(path: str) -> int | None:
-    """Stat a path on disk and return its inode, or None."""
+def _get_disk_stat(path: str) -> os.stat_result | None:
+    """Stat a path on disk, or None when it can't be read."""
     try:
-        p = Path(path)
-        return p.stat().st_ino
+        return Path(path).stat()
     except OSError:
         return None
 
 
+def _is_move_compatible(entry: _DeletedEntry, disk_stat: os.stat_result) -> bool:
+    """
+    Reject inode-match pairs that can't be a real rename.
+
+    Ported from the poller's identically-named check (see
+    ``codex.librarian.fs.poller.snapshot_diff``), which the watcher needs
+    for the same reason and for one of its own. Stored inodes carry no
+    device, so a deleted path's inode can collide with an added path from
+    another mount; and a bulk CBR->CBZ conversion frees many inodes while
+    creating many files, so on an inode-reusing filesystem a new CBZ can
+    be handed the inode a *different* comic's CBR just released. Pairing
+    either re-paths one comic's row onto another comic's file.
+
+    Two cheap sanity checks make the inode match load-bearing only when
+    it's plausibly a rename:
+
+    - File type must match. A real rename never crosses ``stat()``
+      file-type bits, so a mode mismatch is always a collision.
+    - For files, size must match too. Renames preserve size, and two
+      unrelated archives are vanishingly unlikely to share a byte count.
+      Directory ``st_size`` varies with entry count, so it is exempt.
+
+    The size check compares against the size stored at import, so it only
+    holds while that is still current. A tagger that writes tags in place
+    and then renames — the flow ``build_import_task`` remaps modify
+    events for — changes the size before the rename, so a batch that also
+    reports the source as written waives the size check rather than
+    dropping a real pair.
+    """
+    db_stat = entry.stat
+    db_mode = db_stat[_MODE_INDEX] if len(db_stat) > _MODE_INDEX else 0
+    if db_mode and S_IFMT(db_mode) != S_IFMT(disk_stat.st_mode):
+        return False
+    if entry.written or S_ISDIR(disk_stat.st_mode):
+        return True
+    db_size = db_stat[_SIZE_INDEX] if len(db_stat) > _SIZE_INDEX else None
+    return db_size is None or db_size == disk_stat.st_size
+
+
 def _detect_one_move(
     add_idx: int,
     add_value: tuple[int, FSEvent],
@@ -54,27 +110,29 @@ def _detect_one_move(
     matched_deleted,
 ) -> None:
     add_lib_pk, add_event = add_value
-    disk_inode = _get_disk_inode(add_event.src_path)
-    if not disk_inode:
+    disk_stat = _get_disk_stat(add_event.src_path)
+    if not disk_stat or not disk_stat.st_ino:
         return
 
-    match = deleted_by_inode.get(disk_inode)
-    if not match:
+    entry = deleted_by_inode.get(disk_stat.st_ino)
+    if not entry:
         return
 
-    del_idx, del_lib_pk, del_event = match
     # Only match within the same library
-    if add_lib_pk != del_lib_pk:
+    if add_lib_pk != entry.library_pk:
         return
 
-    is_dir = Path(add_event.src_path).is_dir()
-    is_cover = add_event.is_cover or del_event.is_cover
+    if not _is_move_compatible(entry, disk_stat):
+        return
+
+    is_dir = S_ISDIR(disk_stat.st_mode)
+    is_cover = add_event.is_cover or entry.event.is_cover
 
     move_events.append(
         (
             add_lib_pk,
             FSEvent(
-                src_path=del_event.src_path,
+                src_path=entry.event.src_path,
                 change=FSChange.moved,
                 dest_path=add_event.src_path,
                 is_directory=is_dir,
@@ -83,8 +141,26 @@ def _detect_one_move(
         )
     )
     matched_added.add(add_idx)
-    matched_deleted.add(del_idx)
-    del deleted_by_inode[disk_inode]
+    matched_deleted.add(entry.index)
+    del deleted_by_inode[disk_stat.st_ino]
+
+
+def _index_deleted(batch: ChangeBatch) -> dict[int, _DeletedEntry]:
+    """Build inode -> deleted entry from the batch's deleted list."""
+    written_paths = frozenset(event.src_path for _, event in batch.modified)
+    deleted_by_inode: dict[int, _DeletedEntry] = {}
+    for idx, (lib_pk, event) in enumerate(batch.deleted):
+        stat = _get_db_stat(event, lib_pk)
+        if not stat:
+            continue
+        deleted_by_inode[stat[_INODE_INDEX]] = _DeletedEntry(
+            index=idx,
+            library_pk=lib_pk,
+            event=event,
+            stat=stat,
+            written=event.src_path in written_paths,
+        )
+    return deleted_by_inode
 
 
 def detect_moves(batch: ChangeBatch) -> list[tuple[int, FSEvent]]:
@@ -94,13 +170,7 @@ def detect_moves(batch: ChangeBatch) -> list[tuple[int, FSEvent]]:
     Returns move events. Matched FSEvents are removed from batch.added
     and batch.deleted in place.
     """
-    # Build inode -> (index, library_pk, event) from deleted list
-    deleted_by_inode: dict[int, tuple[int, int, FSEvent]] = {}
-    for idx, (lib_pk, event) in enumerate(batch.deleted):
-        inode = _get_db_inode(event, lib_pk)
-        if inode:
-            deleted_by_inode[inode] = (idx, lib_pk, event)
-
+    deleted_by_inode = _index_deleted(batch)
     if not deleted_by_inode:
         return []
 
@@ -117,7 +187,7 @@ def detect_moves(batch: ChangeBatch) -> list[tuple[int, FSEvent]]:
             matched_added,
             matched_deleted,
         )
-    # Remove matched entries from added and deleted (reverse order to keep indices valid)
+    # Drop the matched entries; the move events carry them now.
     batch.added = [
         pair for idx, pair in enumerate(batch.added) if idx not in matched_added
     ]

codex/librarian/scribe/importer/init.py

@@ -50,6 +50,10 @@
 from codex.librarian.scribe.importer.tasks import ImportTask
 from codex.librarian.scribe.search.status import SearchIndexCleanStatus
 from codex.librarian.scribe.status import UpdateCollectionTimestampsStatus
+from codex.librarian.scribe.tagwrite_moves import (
+    get_pending_tag_write_paths,
+    release_tag_write_moves,
+)
 from codex.librarian.worker import WorkerStatusBase
 from codex.models import Library
 from codex.settings import LOGLEVEL
@@ -158,6 +162,41 @@ def timed_step(self, name: str, method: Callable[[], Any]) -> Any:
         self.phase_times[name] = self.phase_times.get(name, 0.0) + elapsed
         return result
 
+    def _defer_pending_tag_write_moves(self) -> None:
+        """
+        Leave paths a tag-write batch is still moving to that batch.
+
+        A scan that lands during a long tag write reports the conversion
+        it is watching as an unrelated delete plus create, and its task
+        outranks the tag writer's end-of-batch move by enqueue time.
+        Dropping those paths here makes the scan a no-op for them, so the
+        move still finds its source row — and its bookmarks — in place.
+
+        A task that carries a registered move reconciles it, so it both
+        releases that guard and is exempt from it — the tag writer's own
+        task keeps the re-read it asked for. The exemption is computed
+        from the task rather than from the release so a move that later
+        turns out to be unappliable can't cost the task its own paths.
+        Runs before the write wait and the status init so neither counts
+        a deferred path.
+        """
+        release_tag_write_moves(self.task.files_moved)
+        own = frozenset(self.task.files_moved) | frozenset(
+            self.task.files_moved.values()
+        )
+        pending = get_pending_tag_write_paths() - own
+        if not pending:
+            return
+        deferred = pending & (
+            self.task.files_deleted | self.task.files_created | self.task.files_modified
+        )
+        if not deferred:
+            return
+        self.task.files_deleted -= pending
+        self.task.files_created -= pending
+        self.task.files_modified -= pending
+        self.log.info(f"Deferred {len(deferred)} path(s) to an in-flight tag write.")
+
     def _wait_for_filesystem_ops_to_finish(self) -> bool:
         """Watcher sends events before filesystem events finish, so wait for them."""
         started_checking = time()
@@ -384,6 +423,7 @@ def _init_librarian_status(self, path) -> None:
     def init_apply(self) -> None:
         """Initialize the library and status flags."""
         self.start_time = now()
+        self._defer_pending_tag_write_moves()
         self.library.start_update()
         too_long = self._wait_for_filesystem_ops_to_finish()
         if too_long:

codex/librarian/scribe/tag_writer.py

@@ -23,6 +23,7 @@
 from codex.librarian.scribe.importer.tasks import ImportTask
 from codex.librarian.scribe.status import TagWriteStatus
 from codex.librarian.scribe.tagwrite_errors import add_tag_write_error
+from codex.librarian.scribe.tagwrite_moves import register_tag_write_move
 from codex.librarian.worker import WorkerStatusAbortableBase
 from codex.models.comic import Comic
 from codex.settings import COMICBOX_CONFIG
@@ -295,6 +296,22 @@ def _sync_ops_for_comic(
             return None, None, None
         return None, end_path, None
 
+    @staticmethod
+    def _guard_move_paths(src: str, written_path: Path | None, move_to: str) -> None:
+        """
+        Hold every path this move passes through until the importer applies it.
+
+        A scan that lands mid-batch reports the same conversion as an
+        unrelated delete plus create and, being enqueued first, reaches
+        the importer first. Registering the DB's now-dead source, the
+        interim archive the write produced, and the final destination
+        makes that scan a no-op for them, so the move below still finds
+        its source row — and its bookmarks — in place. See
+        ``codex.librarian.scribe.tagwrite_moves``.
+        """
+        waypoints = (str(written_path),) if written_path else ()
+        register_tag_write_move(src, move_to, waypoints)
+
     def _sync_db(
         self,
         task: BulkTagWriteTask,
@@ -314,10 +331,11 @@ def _sync_db(
         nor the poller can pair into a move — left alone, the row would be
         deleted and recreated, losing bookmarks. Codex must record the move
         itself, for watched libraries too; the watcher's later add/delete
-        events reconcile as no-ops against the already-moved row. Best-effort:
-        a write batch long enough to force a mid-batch watcher flush can land
-        the watcher's delete first, degrading to the old delete+recreate —
-        never worse. When the
+        events reconcile as no-ops against the already-moved row. A batch
+        long enough to force a mid-batch watcher flush (or a poll that lands
+        during it) would otherwise get that scan's delete in first, so every
+        path a move passes through is registered in ``tagwrite_moves`` and
+        the importer holds it for this task. When the
         original is kept (``delete_original`` off), the DB comic is untouched
         and the converted CBZ is simply a new file: watched libraries see its
         create event, unwatched ones are told here.
@@ -336,7 +354,8 @@ def _sync_db(
         occupied destination, or matches no source row in
         ``_bulk_comics_move_prepare``. The move is targeted, so
         ``move_and_modify_dirs`` runs before the per-comic ``read`` phase,
-        and the same mid-batch-flush caveat as a conversion applies.
+        and its paths are held against a mid-batch scan exactly as a
+        conversion's are.
 
         In-place write (no conversion, no rename): watched libraries re-read
         via the watcher's modify event; unwatched ones are told here.
@@ -354,7 +373,9 @@ def _sync_db(
                 delete_original=task.delete_original,
             )
             if move_to:
-                moved[library_id][str(db_path)] = move_to
+                src = str(db_path)
+                moved[library_id][src] = move_to
+                self._guard_move_paths(src, written_paths.get(pk), move_to)
             if modify:
                 modified[library_id].add(modify)
             if create: