fix: keep the mirror stable across Lidarr upgrades and renames

Three defects in how the mirror tracked its source, all of which show up on a
library that something else reorganises.

Pruning probed the source tree for a mirror file's original name, trying each
known extension in turn. A source saved as .FLAC was never found, so its mirror
file was deleted as an orphan and re-encoded on the next pass, for ever.
Pruning now works from the set of paths the pass actually accounted for, which
cannot disagree with the walk over letter case or extension coverage.

Two sources could also claim one mirror path -- 01 Song.flac beside a leftover
01 Song.mp3, which is what an interrupted upgrade leaves behind. Both encoded
to the same destination, whichever finished last won the race, and every later
pass found the other one stale. The best-quality source now wins, ties break on
path, and the loser is logged.

The source mtime was read after encoding rather than before. A file still being
written when the pass reached it would be stamped with its final mtime while
holding truncated audio, and would never be revisited.

Adds regression tests for all three, plus the format-upgrade, album-rename and
whole-library-deletion cases, each verified to fail before the change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

music_mirror.py

@@ -50,6 +50,23 @@ SOURCE_EXTENSIONS = {
 # quality for nothing.
 COPY_EXTENSIONS = {".mp3"}
 
+# Best first. Used only to settle which source wins when two of them want the
+# same mirror path; see plan().
+SOURCE_PRIORITY = [
+    ".flac",
+    ".wav",
+    ".aif",
+    ".aiff",
+    ".ape",
+    ".wv",
+    ".alac",
+    ".m4a",
+    ".ogg",
+    ".opus",
+    ".wma",
+    ".mp3",
+]
+
 # Looked for in the source directory when a file has no embedded picture.
 COVER_NAMES = ("cover.jpg", "folder.jpg", "front.jpg", "cover.png", "folder.png")
 
@@ -186,6 +203,12 @@ def encode(source, mirror, quality_args, dry_run):
     mirror.parent.mkdir(parents=True, exist_ok=True)
     cover = None if has_embedded_picture(source) else find_cover(source.parent)
 
+    # Read the source's mtime before encoding, not after. If the file is still
+    # being written -- a Lidarr import landing mid-pass -- stamping the mirror
+    # with the later mtime would mark truncated output as current. Stamping the
+    # earlier one leaves the two mismatched, so the next pass re-encodes it.
+    stat = source.stat()
+
     # Encode to a temporary file in the destination directory and rename it
     # into place, so an interrupted run cannot leave a truncated MP3 that the
     # next run would treat as complete.
@@ -199,7 +222,6 @@ def encode(source, mirror, quality_args, dry_run):
         if completed.returncode != 0:
             lines = completed.stderr.strip().splitlines()
             return Result("failed", source, lines[-1] if lines else "ffmpeg failed")
-        stat = source.stat()
         os.utime(temporary, (stat.st_atime, stat.st_mtime))
         os.replace(temporary, mirror)
     except Exception as error:  # noqa: BLE001 - reported per file, run continues
@@ -227,9 +249,8 @@ def copy(source, mirror, dry_run):
     return Result("copied", mirror)
 
 
-def process(source, source_root, mirror_root, quality_args, dry_run):
+def process(source, mirror, quality_args, dry_run):
     """Bring one source file's mirror entry up to date."""
-    mirror = mirror_path_for(source, source_root, mirror_root)
     if is_current(source, mirror):
         return Result("skipped", mirror)
     if source.suffix.lower() in COPY_EXTENSIONS:
@@ -245,15 +266,47 @@ def find_sources(root):
             yield path
 
 
-def prune(source_root, mirror_root, dry_run):
-    """Delete mirror files whose source is gone, and any dirs left empty."""
-    extensions = SOURCE_EXTENSIONS | COPY_EXTENSIONS
+def plan(scan_root, source_root, mirror_root):
+    """Map each mirror path to the one source that should produce it.
+
+    Two sources can want the same mirror path -- `01 Song.flac` alongside a
+    leftover `01 Song.mp3`, which is what an interrupted Lidarr upgrade leaves
+    behind. Without a decision here both would encode to the same destination,
+    each pass would find the loser stale, and the mirror would be rewritten
+    forever. Preferring the highest-quality source, ties broken by path, makes
+    the outcome stable and predictable instead.
+    """
+    chosen = {}
+    for source in find_sources(scan_root):
+        mirror = mirror_path_for(source, source_root, mirror_root)
+        rival = chosen.get(mirror)
+        if rival is None:
+            chosen[mirror] = source
+            continue
+        winner, loser = sorted((source, rival), key=source_rank)
+        logger.warning("%s and %s both map to %s; using %s", rival, source, mirror, winner)
+        chosen[mirror] = winner
+    return chosen
+
+
+def source_rank(source):
+    """Sort key preferring better source formats, then a stable path order."""
+    suffix = source.suffix.lower()
+    position = SOURCE_PRIORITY.index(suffix) if suffix in SOURCE_PRIORITY else len(SOURCE_PRIORITY)
+    return (position, str(source))
+
+
+def prune(mirror_root, expected, dry_run):
+    """Delete mirror files this pass did not account for, and empty dirs.
+
+    Driven by the set of paths the pass expects to exist rather than by
+    probing the source tree for names, which would disagree with it over
+    letter case and over any extension the walker does not collect.
+    """
     removed = 0
 
     for mirror in sorted(mirror_root.rglob(f"*{MIRROR_SUFFIX}")):
-        relative = mirror.relative_to(mirror_root)
-        stem = source_root / relative
-        if any((stem.with_suffix(extension)).exists() for extension in extensions):
+        if mirror in expected:
             continue
         removed += 1
         if dry_run:
@@ -281,10 +334,12 @@ def run_once(scan_root, source_root, mirror_root, quality_args, jobs, dry_run, d
     counts = {"encoded": 0, "copied": 0, "skipped": 0, "failed": 0}
     failures = []
 
+    work = plan(scan_root, source_root, mirror_root)
+
     with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
         futures = [
-            pool.submit(process, source, source_root, mirror_root, quality_args, dry_run)
-            for source in find_sources(scan_root)
+            pool.submit(process, source, mirror, quality_args, dry_run)
+            for mirror, source in work.items()
         ]
         for future in concurrent.futures.as_completed(futures):
             result = future.result()
@@ -292,7 +347,7 @@ def run_once(scan_root, source_root, mirror_root, quality_args, jobs, dry_run, d
             if result.action == "failed":
                 failures.append(result)
 
-    removed = prune(source_root, mirror_root, dry_run) if do_prune else 0
+    removed = prune(mirror_root, set(work), dry_run) if do_prune else 0
 
     for failure in failures:
         logger.error("failed: %s: %s", failure.path, failure.error)