From 1f32e1ac22fc01690d473a7e2b3a2415a31f962a Mon Sep 17 00:00:00 2001 From: AJ Slater Date: Mon, 24 Aug 2026 11:34:12 -0700 Subject: [PATCH] fix(tagging): hold a tag write's paths against a scan that lands mid-batch A tag write that converts archives (CBR->CBZ with delete_original) records the conversion as one targeted move ImportTask enqueued when the whole batch finishes. A batch long enough to force a mid-batch watcher flush (60s of continuous activity) or to catch a poll gets that scan's task enqueued first, and ScribeThread's PriorityQueue breaks ties between equal priority ImportTasks by enqueue time. The scan ran first, deleted the comic rows by their now-dead paths -- cascading bookmarks away -- and left the move with no source row. Short writes were safe; only long ones lost data, which the conversion fix documented as best-effort. Register every path a pending move passes through (the DB source, the interim converted archive, the destination) in a process-local registry, and drop registered paths from a task's created/modified/deleted sets in init_apply, the importer's first phase. A task that carries the registered move reconciles it, so it releases the guard and is exempt from it; the exemption is computed from the task rather than from the release so an unappliable move cannot cost a task its own paths. Ordering stops mattering: the scan becomes a no-op for those paths whenever it runs, and this covers the poller as well as the watcher. Guarding creates matters as much as guarding deletes. Without a rename, the move's destination is the interim CBZ, and a scan that imported it first would leave the move to be dropped as a destination collision -- stranding the original row, bookmarks and all, on the dead path. Port the poller's _is_move_compatible file-type and size check into the watcher's inode move detection. A bulk conversion mass-frees CBR inodes while mass-creating CBZ files, so on an inode-reusing filesystem a new CBZ can be handed the inode a different comic's CBR just released, re-pathing one comic's row onto another comic's file. The size check is waived when the same batch also reports the source as written, which is the in-place write-then-rename flow whose stored size is legitimately stale. Since #822 codex states its own renames rather than leaving them to be paired, so the check now only gates moves inferred for third-party changes. Co-Authored-By: Claude Fable 5 --- NEWS.md | 6 + codex/librarian/fs/watcher/move.py | 124 ++++++++--- codex/librarian/scribe/importer/init.py | 40 ++++ codex/librarian/scribe/tag_writer.py | 33 ++- codex/librarian/scribe/tagwrite_moves.py | 105 +++++++++ tests/importer/test_tag_write_move_guard.py | 224 ++++++++++++++++++++ tests/test_tag_writer_rename.py | 27 ++- tests/test_watcher_move_detect.py | 165 ++++++++++++++ 8 files changed, 690 insertions(+), 34 deletions(-) create mode 100644 codex/librarian/scribe/tagwrite_moves.py create mode 100644 tests/importer/test_tag_write_move_guard.py create mode 100644 tests/test_watcher_move_detect.py diff --git a/NEWS.md b/NEWS.md index 66a197def..a3c6a8ffa 100644 --- a/NEWS.md +++ b/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 diff --git a/codex/librarian/fs/watcher/move.py b/codex/librarian/fs/watcher/move.py index 30e0b3db2..1070e36d7 100644 --- a/codex/librarian/fs/watcher/move.py +++ b/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,8 +41,8 @@ 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) @@ -32,19 +50,57 @@ def _get_db_inode(event: FSEvent, library_pk: int) -> int | None: .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 ] diff --git a/codex/librarian/scribe/importer/init.py b/codex/librarian/scribe/importer/init.py index eae9c875b..44495d70d 100644 --- a/codex/librarian/scribe/importer/init.py +++ b/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: diff --git a/codex/librarian/scribe/tag_writer.py b/codex/librarian/scribe/tag_writer.py index 442161e80..982e58d7e 100644 --- a/codex/librarian/scribe/tag_writer.py +++ b/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: diff --git a/codex/librarian/scribe/tagwrite_moves.py b/codex/librarian/scribe/tagwrite_moves.py new file mode 100644 index 000000000..6315a52e8 --- /dev/null +++ b/codex/librarian/scribe/tagwrite_moves.py @@ -0,0 +1,105 @@ +""" +Paths a tag-write batch has moved but has not yet reconciled in the database. + +Writing tags can move a comic's archive out from under its database row. +Comicbox repacks an unwritable CBR as a CBZ — a new inode at a new path — +and the rename pass then gives the result its scheme name. ``TagWriter`` +records those moves as one targeted ``ImportTask`` enqueued when the whole +batch finishes, but the batch can run for minutes and both scanners report +the same filesystem churn in the meantime. + +A watcher batch force-yielded mid-write (60s of continuous activity, see +``codex.librarian.fs.watcher.watcher``) or a poll that lands during the +write carries a conversion as an unrelated delete plus create: the new +archive is a new inode, so neither scanner can pair it into a move. That +task is enqueued *before* the tag writer's, and ``ScribeThread``'s +``PriorityQueue`` breaks ties between equal-priority import tasks by +enqueue time, so it runs first — deleting the comic row by its now-dead +path, cascading its bookmarks away, and leaving the tag writer's move with +no source row to find. + +This registry lets the importer recognize those paths as codex's own +in-flight work and leave them to it. ``TagWriter`` registers every path a +move it is about to enqueue passes through; ``init_apply`` drops +registered paths from a task's created/modified/deleted sets and releases +a move's whole group when a task actually carries that move — which the +tag writer's own task does, in the phase that runs before its own reads +and deletes. + +The store is process-local on purpose. Only ``ScribeThread`` writes and +reads it, and its lifetime should match the librarian queue's: a librarian +restart loses the pending ``ImportTask`` along with the queue, so a guard +that outlived it would strand those paths instead. The TTL is a backstop +for a batch whose move task never arrives at all; expiry simply restores +the unguarded behavior, which the next scan reconciles. +""" + +from __future__ import annotations + +from threading import Lock +from time import monotonic +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping + +# Long enough to outlast a queue backed up behind a large import, short +# enough that a leaked entry heals the same day. +_TTL = 60 * 60 * 6 + +_LOCK = Lock() +# Guarded path -> (move source, move destination, expiry). Absolute paths +# are unique across libraries — the admin serializer rejects a library +# path that nests inside another — so no library scoping is needed here. +_PENDING: dict[str, tuple[str, str, float]] = {} + + +def _prune(now: float) -> None: + """Drop expired entries. Call with the lock held.""" + expired = [path for path, entry in _PENDING.items() if entry[-1] <= now] + for path in expired: + del _PENDING[path] + + +def register_tag_write_move( + src_path: str, dest_path: str, waypoints: Iterable[str] = () +) -> None: + """Guard every path one pending tag-write move passes through.""" + now = monotonic() + expiry = now + _TTL + with _LOCK: + _prune(now) + for path in (src_path, dest_path, *waypoints): + _PENDING[path] = (src_path, dest_path, expiry) + + +def get_pending_tag_write_paths() -> frozenset[str]: + """Return every path guarded by an unreconciled tag-write move.""" + now = monotonic() + with _LOCK: + _prune(now) + return frozenset(_PENDING) + + +def release_tag_write_moves(moves: Mapping[str, str]) -> None: + """ + Release the guarded group of every registered move a task carries. + + Matches on the whole move, not just its source: a scanner that infers + some *other* destination for a guarded source has not reconciled this + move and must not lift its guard. + """ + if not moves: + return + with _LOCK: + released = [ + path for path, (src, dest, _) in _PENDING.items() if moves.get(src) == dest + ] + for path in released: + del _PENDING[path] + + +def clear_tag_write_moves() -> None: + """Drop every guard. For test isolation; the process owns the lifetime.""" + with _LOCK: + _PENDING.clear() diff --git a/tests/importer/test_tag_write_move_guard.py b/tests/importer/test_tag_write_move_guard.py new file mode 100644 index 000000000..766b297f7 --- /dev/null +++ b/tests/importer/test_tag_write_move_guard.py @@ -0,0 +1,224 @@ +""" +A scan that lands mid tag-write must not reconcile the paths it is moving. + +Writing tags to a CBR converts it to a CBZ at a new inode, so neither the +watcher nor the poller can pair the change into a move. A watcher batch +force-yielded during a long write (or a poll landing in it) therefore +carries the conversion as an unrelated delete plus create, and it reaches +the importer *before* the tag writer's end-of-batch move task. Deleting +the row by its now-dead path would cascade the comic's bookmarks away and +leave the move with no source. ``_defer_pending_tag_write_moves`` holds +those paths for the task that owns them. +""" + +import shutil +from pathlib import Path +from threading import Event, Lock +from typing import override + +from loguru import logger + +from codex.librarian.mp_queue import LIBRARIAN_QUEUE +from codex.librarian.scribe.importer.importer import ComicImporter +from codex.librarian.scribe.importer.tasks import ImportTask +from codex.librarian.scribe.tagwrite_moves import ( + clear_tag_write_moves, + get_pending_tag_write_paths, + register_tag_write_move, +) +from codex.models import ( + Comic, + Folder, + Imprint, + Library, + Publisher, + Series, + Volume, +) +from tests.importer.test_basic import ( + COMIC_PATH, + LIBRARY_PATH, + BaseTestImporter, +) + +# The DB path, the interim archive the conversion wrote, and the final +# name the rename pass gave it. +_CBR_PATH = str(LIBRARY_PATH / "converted.cbr") +_CBZ_PATH = str(LIBRARY_PATH / "converted.cbz") +_RENAMED_PATH = str(LIBRARY_PATH / "Renamed #001.cbz") +_UNRELATED_PATH = str(LIBRARY_PATH / "unrelated.cbz") +_UNRELATED_PATH_ALT = str(LIBRARY_PATH / "unrelated-alt.cbz") + + +class TestImporterTagWriteMoveGuard(BaseTestImporter): + """Paths an in-flight tag write owns are deferred to its own task.""" + + @override + def setUp(self) -> None: + super().setUp() + clear_tag_write_moves() + self.library = Library.objects.get(pk=self.task.library_id) + self.folder = Folder.objects.create( + library=self.library, + path=str(LIBRARY_PATH), + name=LIBRARY_PATH.name, + ) + pub = Publisher.objects.create(name="Guard Pub") + imp = Imprint.objects.create(name="Guard Imprint", publisher=pub) + ser = Series.objects.create(name="Guard Series", imprint=imp, publisher=pub) + self.tags = { + "publisher": pub, + "imprint": imp, + "series": ser, + "volume": Volume.objects.create( + name="1", series=ser, imprint=imp, publisher=pub + ), + } + self.issue_number = 0 + + @override + def tearDown(self) -> None: + clear_tag_write_moves() + super().tearDown() + + def _create_comic(self, path: str) -> Comic: + """Create a comic with its file present, as presave stats disk.""" + shutil.copy(COMIC_PATH, path) + self.issue_number += 1 + return Comic.objects.create( + library=self.library, + path=path, + parent_folder=self.folder, + issue_number=self.issue_number, + name=Path(path).stem, + size=1, + page_count=1, + **self.tags, + ) + + def _importer(self, **task_kwargs) -> ComicImporter: + task = ImportTask(library_id=self.library.pk, **task_kwargs) + return ComicImporter(task, logger, LIBRARIAN_QUEUE, Lock(), Event()) + + @staticmethod + def _register_conversion() -> None: + """Register the move a tag write is about to enqueue.""" + register_tag_write_move(_CBR_PATH, _RENAMED_PATH, (_CBZ_PATH,)) + + def test_delete_of_a_pending_move_source_is_deferred(self) -> None: + """The row the pending move needs survives the scan's delete.""" + comic = self._create_comic(_CBR_PATH) + self._register_conversion() + importer = self._importer(files_deleted=frozenset({_CBR_PATH})) + + importer._defer_pending_tag_write_moves() # noqa: SLF001 + importer.delete() + + assert not importer.task.files_deleted + assert importer.counts.comics_deleted == 0 + assert Comic.objects.filter(pk=comic.pk).exists() + + def test_create_of_a_pending_move_path_is_deferred(self) -> None: + """The interim and final archives are not imported as new comics.""" + self._create_comic(_CBR_PATH) + self._register_conversion() + importer = self._importer( + files_created=frozenset({_CBZ_PATH, _RENAMED_PATH}), + ) + + importer._defer_pending_tag_write_moves() # noqa: SLF001 + + assert not importer.task.files_created + + def test_unregistered_paths_are_untouched(self) -> None: + """A guard for one comic never defers another comic's delete.""" + comic = self._create_comic(_UNRELATED_PATH) + self._register_conversion() + importer = self._importer(files_deleted=frozenset({_UNRELATED_PATH})) + + importer._defer_pending_tag_write_moves() # noqa: SLF001 + importer.delete() + + assert importer.counts.comics_deleted == 1 + assert not Comic.objects.filter(pk=comic.pk).exists() + + def test_carrying_the_move_releases_its_own_guard(self) -> None: + """The tag writer's own task keeps the move and the re-read it asked for.""" + self._create_comic(_CBR_PATH) + self._register_conversion() + importer = self._importer( + files_moved={_CBR_PATH: _RENAMED_PATH}, + files_modified=frozenset({_RENAMED_PATH}), + ) + + importer._defer_pending_tag_write_moves() # noqa: SLF001 + + # Its own move and re-read survive... + assert importer.task.files_moved == {_CBR_PATH: _RENAMED_PATH} + assert importer.task.files_modified == frozenset({_RENAMED_PATH}) + # ...and the guard is gone, so a later scan reconciles normally. + later = self._importer(files_deleted=frozenset({_CBZ_PATH})) + later._defer_pending_tag_write_moves() # noqa: SLF001 + assert later.task.files_deleted == frozenset({_CBZ_PATH}) + + def test_a_different_destination_does_not_release_the_guard(self) -> None: + """Only the registered move reconciles it; a scanner's guess must not.""" + comic = self._create_comic(_CBR_PATH) + self._register_conversion() + # A scanner inferred some other destination for the same source. + scan = self._importer( + files_moved={_CBR_PATH: _UNRELATED_PATH}, + files_deleted=frozenset({_CBZ_PATH}), + ) + + scan._defer_pending_tag_write_moves() # noqa: SLF001 + + # The guard held, so the interim archive was not reaped... + assert not scan.task.files_deleted + # ...and the real move still finds its source row. + move = self._importer(files_moved={_CBR_PATH: _RENAMED_PATH}) + move._defer_pending_tag_write_moves() # noqa: SLF001 + assert not get_pending_tag_write_paths() + assert Comic.objects.filter(pk=comic.pk).exists() + + def test_a_tasks_own_move_paths_are_never_deferred(self) -> None: + """A task's own move survives even when its guard is still registered.""" + self._create_comic(_CBR_PATH) + # Registered against a different destination, so the release + # below does not fire and the guard stays live. + register_tag_write_move(_CBR_PATH, _UNRELATED_PATH, (_CBZ_PATH,)) + importer = self._importer( + files_moved={_CBR_PATH: _UNRELATED_PATH_ALT}, + files_modified=frozenset({_UNRELATED_PATH_ALT}), + ) + + importer._defer_pending_tag_write_moves() # noqa: SLF001 + + assert importer.task.files_moved == {_CBR_PATH: _UNRELATED_PATH_ALT} + assert importer.task.files_modified == frozenset({_UNRELATED_PATH_ALT}) + + def test_scan_delete_then_move_keeps_the_original_row(self) -> None: + """End to end: the scan is a no-op and the move lands on the same row.""" + comic = self._create_comic(_CBR_PATH) + # The conversion happened on disk: the cbr is gone, the renamed + # cbz is in its place. + shutil.copy(COMIC_PATH, _RENAMED_PATH) + Path(_CBR_PATH).unlink() + self._register_conversion() + + # The scan that force-flushed mid-write runs first. + scan = self._importer( + files_deleted=frozenset({_CBR_PATH}), + files_created=frozenset({_CBZ_PATH}), + ) + scan._defer_pending_tag_write_moves() # noqa: SLF001 + scan.delete() + + # Then the tag writer's move task. + move = self._importer(files_moved={_CBR_PATH: _RENAMED_PATH}) + move._defer_pending_tag_write_moves() # noqa: SLF001 + move.move_and_modify_dirs() + + assert move.counts.comics_moved == 1 + comic.refresh_from_db() + assert comic.path == _RENAMED_PATH diff --git a/tests/test_tag_writer_rename.py b/tests/test_tag_writer_rename.py index 5760605d1..7a28b1b5a 100644 --- a/tests/test_tag_writer_rename.py +++ b/tests/test_tag_writer_rename.py @@ -4,7 +4,8 @@ Covers the rename pass and its watcher-aware DB sync: rename-only (no tag patch) and tag-write-plus-rename, both enqueueing a targeted move ``ImportTask`` whether or not the library is watched, the in-place write a -watched library is left to notice for itself, and the skip-and-report +watched library is left to notice for itself, the paths a recorded move +holds against a scan that lands mid-batch, and the skip-and-report collision guard. """ @@ -23,6 +24,10 @@ from codex.librarian.scribe.importer.tasks import ImportTask from codex.librarian.scribe.tag_writer import TagWriter from codex.librarian.scribe.tagwrite_errors import get_tag_write_errors +from codex.librarian.scribe.tagwrite_moves import ( + clear_tag_write_moves, + get_pending_tag_write_paths, +) from codex.librarian.scribe.tasks import BulkTagWriteTask from codex.models import ( Comic, @@ -127,9 +132,11 @@ class TagWriterRenameTests(TestCase): def setUp(self) -> None: caches["default"].clear() caches["tagging"].clear() + clear_tag_write_moves() @override def tearDown(self) -> None: + clear_tag_write_moves() shutil.rmtree(_TMP_DIR, ignore_errors=True) def test_rename_only_unwatched_enqueues_move(self) -> None: @@ -191,6 +198,11 @@ def test_rename_only_watched_enqueues_move(self) -> None: assert imports[0].files_moved == {str(old_path): str(new_path)} # Rename-only: metadata unchanged, so no re-read is requested. assert imports[0].files_modified == frozenset() + # Both ends are held until that move is applied, so a scan landing + # first can't reconcile them out from under it. + assert get_pending_tag_write_paths() == frozenset( + {str(old_path), str(new_path)} + ) def test_tag_write_and_rename_watched_enqueues_move_and_reread(self) -> None: """ @@ -225,6 +237,9 @@ def test_tag_write_and_rename_watched_enqueues_move_and_reread(self) -> None: assert len(imports) == 1 assert imports[0].files_moved == {str(old_path): str(new_path)} assert imports[0].files_modified == frozenset({str(new_path)}) + assert get_pending_tag_write_paths() == frozenset( + {str(old_path), str(new_path)} + ) def test_write_only_watched_enqueues_nothing(self) -> None: """An in-place write with no rename is still left to the watcher.""" @@ -246,6 +261,7 @@ def test_write_only_watched_enqueues_nothing(self) -> None: # The path never changed, so there is no move to state; the # watcher's modify event carries the re-read. assert not [i for i in queue.items if isinstance(i, ImportTask)] + assert not get_pending_tag_write_paths() def test_collision_skips_and_reports(self) -> None: """A target collision skips the rename and records a tag-write error.""" @@ -327,9 +343,11 @@ class TagWriterConversionTests(TestCase): def setUp(self) -> None: caches["default"].clear() caches["tagging"].clear() + clear_tag_write_moves() @override def tearDown(self) -> None: + clear_tag_write_moves() shutil.rmtree(_TMP_DIR, ignore_errors=True) @staticmethod @@ -375,6 +393,11 @@ def test_converted_write_renames_the_cbz_and_enqueues_move(self) -> None: # The move source is the DB's path (the dead .cbr), not the interim cbz. assert imports[0].files_moved == {str(old_path): str(renamed_path)} assert imports[0].files_modified == frozenset({str(renamed_path)}) + # Every path the move passes through is held against a scan that + # lands before the move task does. + assert get_pending_tag_write_paths() == frozenset( + {str(old_path), str(cbz_path), str(renamed_path)} + ) def test_converted_write_without_rename_enqueues_move(self) -> None: """Conversion alone moves the DB row onto the new cbz.""" @@ -449,3 +472,5 @@ def test_converted_write_keeping_original_creates_not_moves(self) -> None: assert not imports[0].files_moved assert imports[0].files_created == frozenset({str(cbz_path)}) assert imports[0].files_modified == frozenset() + # Nothing moved, so nothing needs holding back from a scan. + assert not get_pending_tag_write_paths() diff --git a/tests/test_watcher_move_detect.py b/tests/test_watcher_move_detect.py new file mode 100644 index 000000000..5754107c0 --- /dev/null +++ b/tests/test_watcher_move_detect.py @@ -0,0 +1,165 @@ +""" +The watcher only pairs a delete+add into a move when it could be a rename. + +Stored inodes carry no device, and a bulk CBR->CBZ conversion frees many +inodes while creating many files, so an inode match alone is not proof of +a rename: on an inode-reusing filesystem a new CBZ can be handed the inode +a different comic's CBR just released. ``_is_move_compatible`` (ported +from the poller) makes the match load-bearing only when the file type and +size agree, while still pairing the write-then-rename flow whose stored +size is legitimately stale. +""" + +from __future__ import annotations + +import shutil +from pathlib import Path +from typing import Final, override + +from django.test import TestCase + +from codex.librarian.fs.events import FSChange, FSEvent +from codex.librarian.fs.watcher.data import ChangeBatch +from codex.librarian.fs.watcher.move import detect_moves +from codex.models import ( + Comic, + Imprint, + Library, + Publisher, + Series, + Volume, +) + +_TMP_DIR: Final = Path("/tmp/codex.tests.watchermove") # noqa: S108 +_OLD_NAME: Final = "old.cbz" +_NEW_NAME: Final = "new.cbz" + + +class WatcherMoveDetectTests(TestCase): + """Inode pairs are only trusted when they could be a rename.""" + + @override + def setUp(self) -> None: + _TMP_DIR.mkdir(exist_ok=True, parents=True) + self.library = Library.objects.create( # pyright: ignore[reportUninitializedInstanceVariable] + path=str(_TMP_DIR), events=True + ) + publisher = Publisher.objects.create(name="P") + imprint = Imprint.objects.create(name="I", publisher=publisher) + series = Series.objects.create(name="S", publisher=publisher, imprint=imprint) + self.tags = { # pyright: ignore[reportUninitializedInstanceVariable] + "publisher": publisher, + "imprint": imprint, + "series": series, + "volume": Volume.objects.create( + name="1", publisher=publisher, imprint=imprint, series=series + ), + } + + @override + def tearDown(self) -> None: + shutil.rmtree(_TMP_DIR, ignore_errors=True) + + def _make_comic(self, name: str, contents: str) -> Comic: + """Create a comic whose stored stat reflects its own file.""" + path = _TMP_DIR / name + path.write_text(contents) + return Comic.objects.create( + library=self.library, + path=str(path), + issue_number=1, + name=name, + size=len(contents), + file_type="CBZ", + **self.tags, + ) + + @staticmethod + def _adopt_inode_of(comic: Comic, path: Path) -> None: + """Point the comic's stored inode at another file, as reuse would.""" + assert comic.stat is not None + stat = list(comic.stat) + stat[1] = path.stat().st_ino + Comic.objects.filter(pk=comic.pk).update(stat=stat) + + def _batch(self, deleted: str, added: str, *, modified: str = "") -> ChangeBatch: + batch = ChangeBatch() + batch.deleted.append( + (self.library.pk, FSEvent(src_path=deleted, change=FSChange.deleted)) + ) + batch.added.append( + (self.library.pk, FSEvent(src_path=added, change=FSChange.added)) + ) + if modified: + batch.modified.append( + (self.library.pk, FSEvent(src_path=modified, change=FSChange.modified)) + ) + return batch + + def test_real_rename_is_paired(self) -> None: + """A rename keeps the inode, type and size, so it pairs.""" + comic = self._make_comic(_OLD_NAME, "comic") + old_path = Path(comic.path) + new_path = _TMP_DIR / _NEW_NAME + old_path.rename(new_path) + batch = self._batch(str(old_path), str(new_path)) + + moves = detect_moves(batch) + + assert len(moves) == 1 + _, event = moves[0] + assert event.src_path == str(old_path) + assert event.dest_path == str(new_path) + # Matched events are consumed out of the batch. + assert not batch.added + assert not batch.deleted + + def test_reused_inode_with_different_size_is_rejected(self) -> None: + """A recycled inode on an unrelated file is not a move.""" + comic = self._make_comic(_OLD_NAME, "comic") + old_path = Path(comic.path) + # A conversion freed this comic's inode; an unrelated new archive + # of a different size was handed it. + other_path = _TMP_DIR / _NEW_NAME + other_path.write_text("a completely different archive") + old_path.unlink() + self._adopt_inode_of(comic, other_path) + batch = self._batch(str(old_path), str(other_path)) + + moves = detect_moves(batch) + + # No pair, and both events survive to be handled as delete + add. + assert not moves + assert len(batch.added) == 1 + assert len(batch.deleted) == 1 + + def test_written_source_waives_the_size_check(self) -> None: + """An in-place write before a rename still pairs despite the size.""" + comic = self._make_comic(_OLD_NAME, "comic") + old_path = Path(comic.path) + # The tagger rewrote the archive in place — same inode, new size — + # and then renamed it, all inside one watch batch. + old_path.write_text("comic with many more tags than before") + new_path = _TMP_DIR / _NEW_NAME + old_path.rename(new_path) + batch = self._batch(str(old_path), str(new_path), modified=str(old_path)) + + moves = detect_moves(batch) + + assert len(moves) == 1 + assert moves[0][1].dest_path == str(new_path) + + def test_file_type_mismatch_is_rejected(self) -> None: + """A comic's inode colliding with a directory is never a move.""" + comic = self._make_comic(_OLD_NAME, "comic") + old_path = Path(comic.path) + other_dir = _TMP_DIR / "somedir" + other_dir.mkdir() + old_path.unlink() + self._adopt_inode_of(comic, other_dir) + batch = self._batch(str(old_path), str(other_dir)) + + moves = detect_moves(batch) + + assert not moves + assert len(batch.deleted) == 1