diff --git a/core/file_operations.py b/core/file_operations.py index 71bdc4f..80bec93 100644 --- a/core/file_operations.py +++ b/core/file_operations.py @@ -24,6 +24,30 @@ # Extension used to mark array files that have been cached PLEXCACHED_EXTENSION = ".plexcached" +# Suffix for a cache copy that is still being written. Media is copied to this +# name and only renamed to its real name once complete, so the real name never +# exists empty or half-written. +# +# On Unraid the same share-relative path can exist on both the pool and the +# array, and shfs answers /mnt/user/... with the POOL copy. Writing straight to +# the final name therefore publishes a 0-byte file to Plex the moment the copy +# opens, and a growing partial file for its whole duration. Measured against +# Plex 1.42 with a deliberately shadowed probe: a truncated copy makes Plex +# record the wrong size, and a 0-byte copy makes it drop the item entirely and +# not restore it until the next scan (StudioNirin/PlexCache-D#207). +# +# Dot-prefixed so Plex's scanner ignores it if it surfaces through the share. +PARTIAL_EXTENSION = ".pc-part" + + +def get_partial_cache_path(cache_file_name: str) -> str: + """Temporary name used while writing `cache_file_name`. + + Same directory, so the publish rename stays on one filesystem and is atomic. + """ + directory, basename = os.path.split(cache_file_name) + return os.path.join(directory, f".{basename}{PARTIAL_EXTENSION}") + # Minimum free space (in bytes) required for metadata operations during rename MINIMUM_SPACE_FOR_RENAME = 100 * 1024 * 1024 # 100 MB @@ -4476,6 +4500,12 @@ def move_media_files(self, files: List[str], destination: str, if destination == 'cache': self.last_cache_moves_count = len(move_commands) self.last_cache_moves_bytes = total_bytes + # Sweep the folders we are about to write into. A crash between the + # copy and the publish rename leaves a .pc-part behind, and nothing + # else ever references it — not the exclude file, not the trackers. + self.cleanup_stale_partials( + {os.path.dirname(cache_name) for _, cache_name, _, _ in move_commands} + ) # Execute the move commands self._execute_move_commands(move_commands, max_concurrent_moves_array, @@ -5105,15 +5135,36 @@ def combined_stop_check(): return True return False + # Write to a partial name, then publish with an atomic rename. Plex + # reads this share through shfs, which prefers the pool copy, so the + # real name must never be visible empty or half-written (#207). + partial_file = get_partial_cache_path(cache_file_name) + if os.path.exists(partial_file): + os.remove(partial_file) + logging.debug(f"Removed stale partial file: {partial_file}") + self.file_utils.copy_file_with_permissions( - array_file, cache_file_name, verbose=True, display_dest=display_dest, + array_file, partial_file, verbose=True, display_dest=display_dest, stop_check=combined_stop_check, progress_callback=byte_callback ) logging.debug(f"Copy complete: {os.path.basename(array_file)}") - # Validate copy succeeded + # Validate the copy before it becomes visible under the real name. + if not os.path.isfile(partial_file): + raise IOError(f"Copy verification failed: cache file not created at {partial_file}") + source_size = os.path.getsize(array_file) + partial_size = os.path.getsize(partial_file) + if partial_size != source_size: + raise IOError( + f"Copy verification failed: {partial_file} is {partial_size} bytes, " + f"expected {source_size}" + ) + + # Publish. Same directory, so this is atomic — the real name goes + # straight from absent to complete, with no observable middle state. + os.rename(partial_file, cache_file_name) if not os.path.isfile(cache_file_name): - raise IOError(f"Copy verification failed: cache file not created at {cache_file_name}") + raise IOError(f"Publish failed: cache file not present at {cache_file_name}") # Step 2: Handle array file based on backup setting and hard-link status # Hard-linked files must be deleted (not renamed) to avoid FUSE issues @@ -5770,6 +5821,52 @@ def _remove_symlink(self, path: str) -> bool: return False return False + def cleanup_stale_partials(self, directories, min_age_seconds: int = 3600) -> int: + """Remove .pc-part files left behind by an interrupted run. + + `_cleanup_failed_cache_copy` handles anything that raises, but a crash, + an OOM kill or a power cut skips it entirely. The leftover is invisible + to every other cleanup path — it is dot-prefixed, absent from the + exclude file and unknown to the trackers — so it would sit on the pool + consuming space indefinitely. + + Args: + directories: Cache directories to sweep (not recursive). + min_age_seconds: Leave anything newer alone, so a copy in flight + elsewhere is never pulled out from under it. + + Returns: + Number of files removed. + """ + removed = 0 + now = time.time() + for directory in directories: + if not directory or not os.path.isdir(directory): + continue + try: + entries = os.listdir(directory) + except OSError as e: + logging.debug(f"Could not scan {directory} for stale partials: {e}") + continue + for name in entries: + if not (name.startswith('.') and name.endswith(PARTIAL_EXTENSION)): + continue + path = os.path.join(directory, name) + try: + if now - os.path.getmtime(path) < min_age_seconds: + logging.debug(f"Leaving recent partial alone: {path}") + continue + size = os.path.getsize(path) + os.remove(path) + removed += 1 + logging.info( + f"[CACHE] Removed stale partial copy from an interrupted " + f"run: {name} ({format_bytes(size)})" + ) + except OSError as e: + logging.warning(f"Could not remove stale partial {path}: {e}") + return removed + def _cleanup_failed_cache_copy(self, array_file: str, cache_file_name: str, original_path: str = None) -> None: """Clean up after a failed cache copy operation.""" @@ -5786,10 +5883,16 @@ def _cleanup_failed_cache_copy(self, array_file: str, cache_file_name: str, if os.path.isfile(plexcached_file) and not os.path.isfile(array_file): os.rename(plexcached_file, array_file) logging.info(f"Cleanup: Restored array file after failed copy") - # Remove partial cache file if it exists + # Remove the in-progress copy. Almost every failure leaves the file + # under its partial name, since it is only renamed once complete. + partial_file = get_partial_cache_path(cache_file_name) + if os.path.isfile(partial_file): + os.remove(partial_file) + logging.info(f"Cleanup: Removed partial cache file") + # And the real name, for a failure after the publish rename. if os.path.isfile(cache_file_name): os.remove(cache_file_name) - logging.info(f"Cleanup: Removed partial cache file") + logging.info(f"Cleanup: Removed incomplete cache file") except Exception as e: logging.error(f"Error during cleanup: {type(e).__name__}: {e}") diff --git a/tests/test_atomic_cache_publish.py b/tests/test_atomic_cache_publish.py new file mode 100644 index 0000000..37da42f --- /dev/null +++ b/tests/test_atomic_cache_publish.py @@ -0,0 +1,257 @@ +"""A cache copy is never visible under its real name until it is complete. + +Unraid's shfs answers /mnt/user//... with the POOL copy whenever the +same share-relative path exists on both the pool and the array. Media copied +straight to its final name on the pool therefore publishes itself to Plex the +instant the copy opens the file — at zero bytes — and stays partial for the +whole transfer. + +Measured on Unraid 7.3.1 / Plex 1.42 with a deliberately shadowed probe +(StudioNirin/PlexCache-D#207), reproduced twice: + + state on pool what Plex recorded + --------------- ------------------------------------------ + 82202531 (whole) size 82202531, item healthy + 5242880 (part) size 5242880 <- ingested the truncated size + 0 (just opened) item GONE from the library entirely + 82202531 restored still gone; only a scan brought it back, as a + NEW ratingKey + +The 0-byte state is the damaging one, and `open(dest, 'wb')` created it for +every single file PlexCache cached. Writing to a dot-prefixed partial name and +publishing with an atomic same-directory rename removes both windows. +""" + +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + +sys.modules.setdefault('fcntl', MagicMock()) +for _mod in ['plexapi', 'plexapi.server', 'plexapi.video', 'plexapi.myplex', + 'plexapi.library', 'requests', 'apscheduler', + 'apscheduler.schedulers', 'apscheduler.schedulers.background', + 'apscheduler.triggers', 'apscheduler.triggers.cron', + 'apscheduler.triggers.interval']: + sys.modules.setdefault(_mod, MagicMock()) + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from core.file_operations import ( + FileMover, PARTIAL_EXTENSION, get_partial_cache_path, +) + + +def _mover(tmp_path, **kw): + exclude_file = os.path.join(str(tmp_path), "exclude.txt") + open(exclude_file, "w").close() + + file_utils = MagicMock() + file_utils.is_docker = False + file_utils.is_linux = True + file_utils.create_directory_with_permissions = MagicMock() + + return FileMover( + real_source="/mnt/user/media", + cache_dir=os.path.join(str(tmp_path), "cache"), + is_unraid=False, + file_utils=file_utils, + debug=False, + mover_cache_exclude_file=exclude_file, + create_plexcached_backups=kw.get("create_backups", True), + ) + + +def _source(tmp_path, size=4096): + array_dir = tmp_path / "array" + array_dir.mkdir(exist_ok=True) + src = array_dir / "Show - S01E01.mkv" + src.write_bytes(b"\xab" * size) + return str(src) + + +class TestThePartialName: + + def test_is_dot_prefixed_so_plex_ignores_it(self): + """It surfaces through the user share, where Plex would otherwise scan it.""" + p = get_partial_cache_path("/mnt/cache/TV/Show - S01E01.mkv") + assert os.path.basename(p).startswith(".") + assert p.endswith(PARTIAL_EXTENSION) + + def test_stays_in_the_same_directory(self): + """Cross-filesystem renames are not atomic, which is the whole point.""" + final = "/mnt/cache/TV/Season 01/Show - S01E01.mkv" + assert os.path.dirname(get_partial_cache_path(final)) == os.path.dirname(final) + + def test_does_not_collide_between_files(self): + a = get_partial_cache_path("/mnt/cache/TV/A.mkv") + b = get_partial_cache_path("/mnt/cache/TV/B.mkv") + assert a != b + + +class TestTheRealNameIsNeverIncomplete: + """The regression this whole change exists to prevent.""" + + def test_final_path_does_not_exist_while_copying(self, tmp_path): + src = _source(tmp_path) + cache_dir = tmp_path / "cache" / "TV" + cache_dir.mkdir(parents=True) + final = str(cache_dir / "Show - S01E01.mkv") + mover = _mover(tmp_path) + + observed = [] + + def fake_copy(s, d, **kw): + # Mirror the real copy loop: open(dest,'wb') creates the file at + # zero bytes, then chunks are written. The observation has to + # happen AFTER creation — that empty moment is what deleted the + # item from Plex, and checking before it would pass either way. + with open(d, "wb") as f: + f.flush() + observed.append({ + "final_exists": os.path.exists(final), + "dest_is_partial": d == get_partial_cache_path(final), + }) + f.write(open(s, "rb").read()) + + mover.file_utils.copy_file_with_permissions = fake_copy + + with patch("core.file_operations.get_console_lock"), \ + patch("tqdm.tqdm.write"), \ + patch("core.logging_config.mark_file_activity"): + rc = mover._move_to_cache(src, str(cache_dir), final) + + assert rc == 0 + assert observed[0]["dest_is_partial"], "copy must target the partial name" + assert not observed[0]["final_exists"], ( + "the real name existed during the copy — Plex can read it there" + ) + assert os.path.getsize(final) == 4096 + + def test_partial_is_gone_after_a_successful_publish(self, tmp_path): + src = _source(tmp_path) + cache_dir = tmp_path / "cache" / "TV" + cache_dir.mkdir(parents=True) + final = str(cache_dir / "Show - S01E01.mkv") + mover = _mover(tmp_path) + mover.file_utils.copy_file_with_permissions = \ + lambda s, d, **kw: open(d, "wb").write(open(s, "rb").read()) + + with patch("core.file_operations.get_console_lock"), \ + patch("tqdm.tqdm.write"), \ + patch("core.logging_config.mark_file_activity"): + mover._move_to_cache(src, str(cache_dir), final) + + assert not os.path.exists(get_partial_cache_path(final)) + assert os.path.isfile(final) + + +class TestFailureLeavesNothingBehind: + + def test_a_short_copy_is_rejected_before_publishing(self, tmp_path): + """A truncated copy must never reach the real name.""" + src = _source(tmp_path, size=8192) + cache_dir = tmp_path / "cache" / "TV" + cache_dir.mkdir(parents=True) + final = str(cache_dir / "Show - S01E01.mkv") + mover = _mover(tmp_path) + # Write only half the bytes, as a truncated transfer would. + mover.file_utils.copy_file_with_permissions = \ + lambda s, d, **kw: open(d, "wb").write(open(s, "rb").read()[:4096]) + + with patch("core.file_operations.get_console_lock"), \ + patch("tqdm.tqdm.write"), \ + patch("core.logging_config.mark_file_activity"): + rc = mover._move_to_cache(src, str(cache_dir), final) + + assert rc == 1, "a size mismatch must fail the move" + assert not os.path.exists(final), "the short copy reached the real name" + assert not os.path.exists(get_partial_cache_path(final)) + + def test_partial_removed_when_the_copy_raises(self, tmp_path): + src = _source(tmp_path) + cache_dir = tmp_path / "cache" / "TV" + cache_dir.mkdir(parents=True) + final = str(cache_dir / "Show - S01E01.mkv") + mover = _mover(tmp_path) + + def exploding_copy(s, d, **kw): + open(d, "wb").write(b"partial") + raise RuntimeError("disk fell over") + + mover.file_utils.copy_file_with_permissions = exploding_copy + + with patch("core.file_operations.get_console_lock"), \ + patch("tqdm.tqdm.write"), \ + patch("core.logging_config.mark_file_activity"): + rc = mover._move_to_cache(src, str(cache_dir), final) + + assert rc == 1 + assert not os.path.exists(get_partial_cache_path(final)) + assert not os.path.exists(final) + + def test_a_stale_partial_does_not_block_a_retry(self, tmp_path): + src = _source(tmp_path) + cache_dir = tmp_path / "cache" / "TV" + cache_dir.mkdir(parents=True) + final = str(cache_dir / "Show - S01E01.mkv") + open(get_partial_cache_path(final), "wb").write(b"junk from last time") + + mover = _mover(tmp_path) + mover.file_utils.copy_file_with_permissions = \ + lambda s, d, **kw: open(d, "wb").write(open(s, "rb").read()) + + with patch("core.file_operations.get_console_lock"), \ + patch("tqdm.tqdm.write"), \ + patch("core.logging_config.mark_file_activity"): + rc = mover._move_to_cache(src, str(cache_dir), final) + + assert rc == 0 + assert os.path.getsize(final) == 4096 + + +class TestTheStaleSweep: + """A crash or power cut skips _cleanup_failed_cache_copy entirely.""" + + def _partial(self, d, name, age_seconds): + p = d / name + p.write_bytes(b"x" * 128) + past = os.path.getmtime(p) - age_seconds + os.utime(p, (past, past)) + return p + + def test_removes_old_partials(self, tmp_path): + d = tmp_path / "cache"; d.mkdir() + old = self._partial(d, ".Show - S01E01.mkv" + PARTIAL_EXTENSION, 7200) + mover = _mover(tmp_path) + + assert mover.cleanup_stale_partials([str(d)]) == 1 + assert not old.exists() + + def test_leaves_recent_partials_alone(self, tmp_path): + """Something else may be mid-copy; do not pull it out from under it.""" + d = tmp_path / "cache"; d.mkdir() + fresh = self._partial(d, ".Show - S01E02.mkv" + PARTIAL_EXTENSION, 60) + mover = _mover(tmp_path) + + assert mover.cleanup_stale_partials([str(d)]) == 0 + assert fresh.exists() + + def test_never_touches_real_media(self, tmp_path): + d = tmp_path / "cache"; d.mkdir() + keep = [d / "Show - S01E01.mkv", d / "Show - S01E01.mkv.plexcached", + d / ".hidden.mkv", d / "poster.jpg"] + for k in keep: + k.write_bytes(b"real") + past = os.path.getmtime(k) - 99999 + os.utime(k, (past, past)) + mover = _mover(tmp_path) + + assert mover.cleanup_stale_partials([str(d)]) == 0 + assert all(k.exists() for k in keep) + + def test_missing_directories_are_not_an_error(self, tmp_path): + mover = _mover(tmp_path) + assert mover.cleanup_stale_partials( + [str(tmp_path / "nope"), "", None]) == 0 diff --git a/tests/test_file_operations_safety.py b/tests/test_file_operations_safety.py index 8c49225..d779dee 100644 --- a/tests/test_file_operations_safety.py +++ b/tests/test_file_operations_safety.py @@ -394,7 +394,14 @@ def test_converts_user_to_user0_before_rename(self, tmp_path): cache_file = "/mnt/cache/media/Movies/Movie.mkv" mover = _make_file_mover(tmp_path, is_unraid=True, create_backups=True) - mover.file_utils.copy_file_with_permissions = lambda src, dest, **kw: None + # Model what a copy actually does — register the destination in the + # virtual filesystem — rather than returning None. The destination is + # the partial name now (published by rename), and these tests are about + # the array-side path conversion, not the copy mechanics. + def _record_copy(src, dest, **kw): + known_files[dest] = True + known_sizes[dest] = known_sizes.get(src, 1000) + mover.file_utils.copy_file_with_permissions = _record_copy rename_calls = [] renamed_files = set() @@ -543,7 +550,14 @@ def test_converts_user_to_user0_even_with_zfs_detection(self, tmp_path): cache_file = "/mnt/cache/media/Movies/Movie.mkv" mover = _make_file_mover(tmp_path, is_unraid=True, create_backups=True) - mover.file_utils.copy_file_with_permissions = lambda src, dest, **kw: None + # Model what a copy actually does — register the destination in the + # virtual filesystem — rather than returning None. The destination is + # the partial name now (published by rename), and these tests are about + # the array-side path conversion, not the copy mechanics. + def _record_copy(src, dest, **kw): + known_files[dest] = True + known_sizes[dest] = known_sizes.get(src, 1000) + mover.file_utils.copy_file_with_permissions = _record_copy # Simulate incorrect ZFS detection: share marked as pool-only old_prefixes = _zfs_user_prefixes.copy() @@ -614,7 +628,14 @@ def test_zfs_pool_only_uses_fuse_path(self, tmp_path): cache_file = "/mnt/cache/media/Movies/Movie.mkv" mover = _make_file_mover(tmp_path, is_unraid=True, create_backups=True) - mover.file_utils.copy_file_with_permissions = lambda src, dest, **kw: None + # Model what a copy actually does — register the destination in the + # virtual filesystem — rather than returning None. The destination is + # the partial name now (published by rename), and these tests are about + # the array-side path conversion, not the copy mechanics. + def _record_copy(src, dest, **kw): + known_files[dest] = True + known_sizes[dest] = known_sizes.get(src, 1000) + mover.file_utils.copy_file_with_permissions = _record_copy rename_calls = [] renamed_files = set()