From d90cf1a0b6e557d3a181a5fb68f63fbb0f747b80 Mon Sep 17 00:00:00 2001 From: Brandon Haney <121782102+Brandon-Haney@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:49:00 -0500 Subject: [PATCH] Publish cache copies atomically so Plex never sees a partial file 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 was copied straight to its final name on the pool, so that name existed at zero bytes the moment the copy opened it, and stayed partial for the whole transfer. Plex reads the share through that same union. Measured on Unraid 7.3.1 / Plex 1.42 with a shadowed probe, twice: pool state what Plex recorded ---------------- --------------------------------------------- whole correct size, item healthy truncated ingested the truncated size zero bytes item removed from the library restored still gone; a scan brought it back as a new ratingKey Copy to a dot-prefixed .pc-part name, verify size against the source, then rename into place. The rename is same-directory and therefore atomic, so the real name goes from absent to complete with no observable middle state. Refs #207. Also sweeps .pc-part files left by an interrupted run, since a crash skips _cleanup_failed_cache_copy and nothing else references them. --- core/file_operations.py | 113 +++++++++++- tests/test_atomic_cache_publish.py | 257 +++++++++++++++++++++++++++ tests/test_file_operations_safety.py | 27 ++- 3 files changed, 389 insertions(+), 8 deletions(-) create mode 100644 tests/test_atomic_cache_publish.py diff --git a/core/file_operations.py b/core/file_operations.py index 71bdc4ff..80bec938 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 00000000..37da42f9 --- /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 8c492254..d779dee1 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()