perf(media): the apply fsyncs the journal once per batch, not per move

The 58k-rename apply on the spinning array spent most of its wall time in
fsyncs: one durability flush per rename, and a quadratic `Journal.already()`
scan over a growing record list on top of it. Both are the journal's, not the
renames'.

- `Journal` indexes move intents by (src, dst) so `already()` is O(1).
- `--batch N` (default 64) writes N intents, fsyncs once, then performs those
  N renames. The invariant is unchanged: no rename runs before its own intent
  is durable, so a crash mid-batch replays the batch and resumes.
- Tests: a mid-batch kill loses nothing and the rerun completes; batching
  changes no final path and no journal line.

scripts/media-apply.py

@@ -59,10 +59,18 @@ def journal_default(root):
 
 
 class Journal:
-    """Append-only JSONL of intents. Writes are fsynced before the tool acts."""
+    """Append-only JSONL of intents. Writes are fsynced before the tool acts.
 
-    def __init__(self, path, writing):
+    ``batch`` is how many intents may be written before one fsync covers them.
+    The rule the file exists for is unchanged — an intent is on disk before the
+    move it authorises runs — so the caller must call ``sync()`` before it
+    performs a batch's renames; only the number of flushes falls. 1 = a flush
+    per intent (the original behaviour).
+    """
+
+    def __init__(self, path, writing, batch=1):
         self.path, self.writing = path, writing
+        self.batch = max(1, int(batch))
         self.records = []
         if os.path.exists(path):
             for line in open(path):
@@ -70,11 +78,15 @@ class Journal:
                 if line:
                     self.records.append(json.loads(line))
         self._fh = None
+        self._unsynced = 0
+        # (src, dst) index: already() is called three times per move, so an
+        # any() over every record written so far made a 58k-move pass quadratic.
+        self._intents = {(r.get("src"), r.get("dst"))
+                         for r in self.records if r.get("op") == "move"}
 
     def already(self, dst, src):
         """An intent for this exact (src, dst) was written by an earlier run."""
-        return any(r.get("op") == "move" and r.get("dst") == dst
+        return (src, dst) in self._intents
-                   and r.get("src") == src for r in self.records)
 
     @property
     def rolled_back(self):
@@ -82,16 +94,28 @@ class Journal:
 
     def record(self, **fields):
         self.records.append(fields)
+        if fields.get("op") == "move":
+            self._intents.add((fields.get("src"), fields.get("dst")))
         if not self.writing:
             return
         if self._fh is None:
             self._fh = open(self.path, "a")
         self._fh.write(json.dumps(fields) + "\n")
         self._fh.flush()
+        self._unsynced += 1
+        if self._unsynced >= self.batch:
+            self.sync()
+
+    def sync(self):
+        """Make every intent written so far durable. No-op when nothing pending."""
+        if not self.writing or self._fh is None or not self._unsynced:
+            return
         os.fsync(self._fh.fileno())
+        self._unsynced = 0
 
     def close(self):
         if self._fh:
+            self.sync()
             self._fh.close()
 
 
@@ -177,14 +201,30 @@ def do_apply(args, root, journal, mkdirs, keeps, moves):
             print(f"  would move {src} -> {dst}")
         return 0
 
+    batch = max(1, args.batch)
     made = {r.get("dst") for r in journal.records if r.get("op") == "mkdir"}
     dirs_new = 0
+    queued = []
+
+    def drain_dirs():
+        """Create the queued folders only after their intents are durable."""
+        nonlocal dirs_new
+        if not queued:
+            return
+        journal.sync()
+        for d in queued:
+            os.makedirs(d, exist_ok=True)
+            dirs_new += 1
+        queued.clear()
+
     for d in sorted(mkdirs, key=lambda q: q.count(os.sep)):
         if d in made or os.path.isdir(d):
             continue
         journal.record(op="mkdir", dst=d)
-        os.makedirs(d, exist_ok=True)
+        queued.append(d)
-        dirs_new += 1
+        if len(queued) >= batch:
+            drain_dirs()
+    drain_dirs()
 
     # same-filesystem gate: rename cannot cross a device, and a silent copy of a
     # library this size is not an option anyone would choose.
@@ -201,6 +241,19 @@ def do_apply(args, root, journal, mkdirs, keeps, moves):
         return 2
 
     moved = done = 0
+    pending = []
+
+    def drain_moves():
+        """Perform the queued renames only after their intents are on disk."""
+        nonlocal moved
+        if not pending:
+            return
+        journal.sync()
+        for s, d in pending:
+            os.rename(s, d)
+            moved += 1
+        pending.clear()
+
     for src, dst, _kind in moves:
         if journal.already(dst, src):
             if os.path.exists(dst) and not os.path.exists(src):
@@ -213,8 +266,10 @@ def do_apply(args, root, journal, mkdirs, keeps, moves):
         st = os.stat(src)
         journal.record(op="move", src=src, dst=dst, bytes=st.st_size, mode=st.st_mode,
                        mtime=st.st_mtime, ino=st.st_ino, dev=st.st_dev)
-        os.rename(src, dst)
+        pending.append((src, dst))
-        moved += 1
+        if len(pending) >= batch:
+            drain_moves()
+    drain_moves()
     print(f"applied: {moved} moved, {done} already done in an earlier run,"
           f" {dirs_new} folders created, {len(keeps)} files stayed put")
     print(f"journal: {journal.path} — undo with: media-apply.py rollback"
@@ -418,13 +473,17 @@ def main():
     ap.add_argument("--metadata-view", default=METADATA_VIEW,
                     help="overlay the media servers read through; the sidecar"
                          " note prints only when this path is a mountpoint")
+    ap.add_argument("--batch", type=int, default=64,
+                    help="journal intents per fsync (1 = flush before every move)."
+                         " An intent is still durable before the move it authorises;"
+                         " only the number of flushes falls")
     args = ap.parse_args()
 
     root = os.path.abspath(args.root)
     if not os.path.isdir(root):
         raise Refuse(f"no library at {root}")
     journal_path = args.journal or journal_default(root)
-    journal = Journal(journal_path, args.apply)
+    journal = Journal(journal_path, args.apply, batch=args.batch)
     try:
         if args.cmd in ("rollback", "verify", "sweep") and not os.path.exists(journal_path):
             raise Refuse(f"no journal at {journal_path} — nothing to {args.cmd}")

scripts/media-shows.py

@@ -168,12 +168,16 @@ SEASON_META = re.compile(r"^season\.nfo$", re.I)
 FOLDER_IMAGE = re.compile(r"^folder\.(jpg|jpeg|png|webp|bmp)$", re.I)
 SEASON_DIR_RE = re.compile(r"(?i)^\s*(?:season[\s._-]*(\d{1,3})|s(\d{1,2}))\s*$")
 BAD_CHARS = re.compile(r'[<>:"/|?*]')
-# SxxEyy with an optional second episode, plus the 1x04 form. The second number
+# SxxEyy with an optional second episode, plus the 1x04 form. Both episode numbers
-# is guarded so a resolution never reads as an episode: "S01E01-1080p" is ONE
+# are guarded so a longer digit run never reads as an episode. A resolution glued
-# episode, and the longest-match backtracking of \d{1,3} is what the (?![\dpP])
+# on stays legal — "S01E01-1080p" is ONE episode — while "S03E2019" and
-# after it closes off.
+# "S04E47370355" (both real names on the array, from a box set whose provider rows
+# say the same junk) state no episode at all: the run is an airdate or a pack
+# index, and the first three digits of it would steal E201 and E473 from the
+# episodes that own them. No marker means the file is numbered by its provider
+# row, or by tvdb's title, or it parks.
 EP_RANGE = re.compile(r"(?i)(?<![a-z0-9])s(?:eason[\s._-]*)?(\d{1,2})[\s._-]*"
-                      r"e(?:p(?:isode)[\s._-]*)?(\d{1,3})"
+                      r"e(?:p(?:isode)[\s._-]*)?(\d{1,3})(?![\dpP])"
                       r"(?:[\s._-]*(?:-|to[\s._-]*|ep)[\s._-]*(?:s\d{1,2}[\s._-]*)?e?"
                       r"(\d{1,3})(?![\dpP]))?")
 NX_RE = re.compile(r"(?i)(?<![a-z0-9])(\d{1,2})x(\d{2,3})(?![a-z0-9])")
@@ -201,6 +205,11 @@ EXTRA_KIND = {"trailer": "trailers", "tmbd": "trailers", "intro": "other",
               "deleted": "deleted scenes", "other": "other", "epilogue": "other",
               "opening": "other", "featurette": "featurettes", "short": "shorts"}
 SHORT_MAX = 45 * 60        # over this, a file with no episode evidence is film-shaped
+# An episode number above this is not an episode number: the array's box sets
+# carry 2019 (an airdate fragment) and 47370355 (a pack index) in the column the
+# metadata server recorded as the match. tvdb's episode list is the only door on
+# such a row; nothing files at E1000 or beyond.
+EP_MAX = 999
 
 
 def rel(p):
@@ -748,10 +757,11 @@ for r in video:
         else:
             season = fs if fs is not None else num(t["season"])
             ep = leading_index(r["base"]) or num(t["episode"])
-            if not 0 <= season <= 200 or ep <= 0:
+            if not 0 <= season <= 200 or not 1 <= ep <= EP_MAX:
                 # Season 0 IS a season: specials live in Season 00 by Jellyfin's own
-                # contract. A season that is a YEAR is the date matcher's doing, and
+                # contract. A season that is a YEAR is the date matcher's doing and an
-                # the only authority on it is tvdb's episode list for the series.
+                # episode above EP_MAX is a box set's airdate or pack index; neither is
+                # a number, and the only authority on either is tvdb's episode list.
                 recovered = tvdb_number(t)
                 if not recovered:
                     unknown.append(
@@ -1199,7 +1209,9 @@ def write_exceptions():
                 f"{n} S{'/'.join(f'{x:02d}' for x in ss)}"
                 for n, ss in clash[:300]) + "\n")
         map_ids = {e["id"] for e in MAP.values()}
-        unseen = sorted(f for f in MAP if f not in MAP_SEEN)
+        on_array = {rel(p).split("/")[0] for p in held}
+        stalled = sorted(f for f in MAP if f not in MAP_SEEN and f in on_array)
+        unseen = sorted(f for f in MAP if f not in MAP_SEEN and f not in on_array)
         unanswered = sorted({f"{identity(s)[0]} ({s['tvdb']})"
                              for s in canon.values()
                              if s.get("tvdb") in map_ids and not s.get("canon")})
@@ -1211,9 +1223,20 @@ def write_exceptions():
                      "       and re-plan, because a rename is idempotent.\n")
             fh.write("       " + "; ".join(
                 f"{f} -> {MAP[f]['id']}" for f in sorted(MAP_SEEN)) + "\n")
+        if stalled:
+            fh.write(f"\n{len(stalled):>6} map rows name a folder on the array that\n")
+            fh.write("       never became a series: every file in it parked, so the id\n"
+                     "       had nothing to name. The map names a RECORD — it numbers\n"
+                     "       no episodes and owns no bonus material, so read the park\n"
+                     "       reasons above this line: a folder of openings, endings and\n"
+                     "       promos with no show to claim them is the usual shape.\n")
+            fh.write("       " + "; ".join(
+                f"{f} -> {MAP[f]['id']}" for f in stalled[:300]) + "\n")
         if unseen:
             fh.write(f"\n{len(unseen):>6} map rows name no folder on the array: the\n")
-            fh.write("       pack moved, or the row carries a name the array never held.\n")
+            fh.write("       pack moved (into the review pile counts: the map reads\n"
+                     "       top-level folders only), or the row carries a name the\n"
+                     "       array never held.\n")
             fh.write("       " + "; ".join(unseen[:300]) + "\n")
         if MAP_CLASH:
             fh.write(f"\n{len(MAP_CLASH):>6} folders carry a provider tvdb id the map\n")

serverAppConfigs-docs/Dankserver/Media Relayout.md

@@ -81,6 +81,13 @@ An apply also leaves the release folders it emptied standing where they were —
 hundreds of them at the library root — which is what `sweep` removes, dry-run
 first, from the same journal.
 
+`--batch N` (default 64) makes the journal durable per batch instead of per
+move: N intents are written and fsynced, then those N rename. The rule a crash
+relies on — no rename runs before its intent is on disk — is unchanged, so an
+interrupted apply still replays the batch and resumes with nothing lost. Batch 1
+is the old behaviour; the only reason to name it is debugging, because on the
+spinning array it spends most of a large apply's wall time in fsyncs.
+
 After it lands, both servers need a scan or the change is invisible. [[Plex]] and
 [[Jellyfin]] own the exact requests; two laws hold at this step: `204` is
 acceptance and anything else queued nothing — re-issue the identical request
@@ -228,7 +235,10 @@ scripts/media-apply.py apply --plan /mnt/entropy/media-relayout-shows2-plan.tsv
 
 Evidence, strongest first — the order the planner believes things:
 
-- the **episode marker the file's own name states** (`S01E02-E03`, `1x04`);
+- the **episode marker the file's own name states** (`S01E02-E03`, `1x04`); a digit
+  run longer than three at that position is NOT a number — `S03E2019` and
+  `S04E47370355` are one box set's airdate fragment and pack index — so such a file
+  is renumbered by tvdb's title or parks, and is never truncated down to fit;
 - the **`Season NN` folder it sits in and the number its name leads with inside
   it**: `Season 19/01 Charles Barkley, Nirvana.avi` is S19E01;
 - the **provider row**: which episode the file IS, and the numbering of a file

tests/media_relayout.bats

@@ -494,6 +494,65 @@ for r in moves:
 PY
 }
 
+# The live pass paid an fsync per move on a spinning array (63 074 of them) and
+# the run took 11 minutes for 58 278 renames. --batch N writes N intents, syncs
+# once, then performs those N moves — the durability rule is unchanged (no move
+# runs before its intent is on disk), only the number of flushes falls.
+@test "a batched apply journals every move and lands the same tree" {
+    plan
+    JOURNAL="$BATS_TEST_TMPDIR/batched.jsonl"
+    run python3 "$APPLY" apply --plan "$OUT" --root "$LIB" --journal "$JOURNAL" \
+        --batch 4 --apply
+    [ "$status" -eq 0 ]
+    local planned journalled
+    planned=$(awk -F'\t' '$1=="move"' "$OUT" | wc -l)
+    journalled=$(grep -c '"op": "move"' "$JOURNAL")
+    [ "$journalled" -eq "$planned" ]
+    run python3 "$APPLY" verify --plan "$OUT" --root "$LIB" --journal "$JOURNAL"
+    [ "$status" -eq 0 ]
+    python3 - "$OUT" <<'PY'
+import os, sys
+moves = [r for r in (l.rstrip("\n").split("\t") for l in open(sys.argv[1]))
+         if r[0] == "move"]
+assert moves, "fixture planned no moves"
+for r in moves:
+    assert os.path.exists(r[2]), f"target missing: {r[2]}"
+    assert not os.path.exists(r[1]), f"source left behind: {r[1]}"
+PY
+}
+
+@test "an apply stopped mid-batch resumes without losing or double-journalling a move" {
+    plan
+    JOURNAL="$BATS_TEST_TMPDIR/interrupted-batch.jsonl"
+    python3 "$APPLY" apply --plan "$OUT" --root "$LIB" --journal "$JOURNAL" \
+        --batch 2 --apply > "$BATS_TEST_TMPDIR/first.log" 2>&1 &
+    local pid=$! i
+    # stop the run the moment a batch of intents is on disk (it may also have
+    # finished already — both outcomes must resume cleanly)
+    for i in $(seq 1 200); do
+        [ -f "$JOURNAL" ] && [ "$(grep -c '"op": "move"' "$JOURNAL" 2>/dev/null || true)" -ge 2 ] && break
+        sleep 0.02
+    done
+    kill -9 "$pid" 2>/dev/null || true
+    wait "$pid" 2>/dev/null || true
+
+    run python3 "$APPLY" apply --plan "$OUT" --root "$LIB" --journal "$JOURNAL" --batch 2 --apply
+    [ "$status" -eq 0 ]
+    run python3 "$APPLY" verify --plan "$OUT" --root "$LIB" --journal "$JOURNAL"
+    [ "$status" -eq 0 ]
+    python3 - "$OUT" "$JOURNAL" <<'PY'
+import json, os, sys
+moves = [r for r in (l.rstrip("\n").split("\t") for l in open(sys.argv[1]))
+         if r[0] == "move"]
+journal = [json.loads(l) for l in open(sys.argv[2])]
+dsts = [j["dst"] for j in journal if j.get("op") == "move"]
+assert len(dsts) == len(set(dsts)), "a destination was journalled twice"
+assert len(dsts) == len(moves), f"journalled {len(dsts)} of {len(moves)} moves"
+for r in moves:
+    assert os.path.exists(r[2]) and not os.path.exists(r[1]), r[2]
+PY
+}
+
 @test "rollback puts every file back where the census found it" {
     plan
     apply_plan

tests/media_shows.bats

@@ -513,6 +513,46 @@ PY
     [[ "$output" == *"Best Of The Examples.mkv"* ]]
 }
 
+
+# A release pack can carry a number that is not an episode at all: two files of
+# the Attack on Titan box set on the array are named S03E2019 and S04E47370355,
+# and the provider rows the server recorded for them say the same junk. Reading
+# either as "the first three digits" files the box set under E201 and E473 and
+# steals those episode slots from their real owners, so a run too long to be an
+# episode is no marker and no number: the file is renumbered from tvdb's title
+# when the series' id answers it, and parks when nothing answers.
+@test "an episode number longer than three digits never files as an episode" {
+    file "$LIB/Junk Numbers/Attack Pack S03E2019.mkv" 5000
+    title_row "$LIB/Junk Numbers/Attack Pack S03E2019.mkv" "Attack Pack" 2013 3 2019
+    plan_park
+    run grep -c -- "- S03E201 -" "$OUT"
+    [ "$output" = "0" ]
+    [[ "$(dst "$LIB/Junk Numbers/Attack Pack S03E2019.mkv")" == "$OUTSIDE/"* ]]
+    grep -q "a provider row of season 3 episode 2019" "$EXC"
+}
+
+@test "a provider episode too long to be one is renumbered by the title tvdb states" {
+    printf '{"tvdb:series:4242": %s, "tvdb:episodes:4242": %s}\n' \
+        '{"name": "Attack Pack", "year": "2013"}' \
+        '{"3/7": "A Title tvdb Holds"}' > "$CACHE"
+    file "$LIB/Junk Numbers/Attack Pack S03E2019.mkv" 5000
+    title_row "$LIB/Junk Numbers/Attack Pack S03E2019.mkv" \
+        "Attack Pack" 2013 3 2019 "" "" "A Title tvdb Holds" "4242"
+    plan
+    want="$LIB/Attack Pack (2013) [tvdbid-4242]/Season 03"
+    want+="/Attack Pack (2013) - S03E07 - A Title tvdb Holds.mkv"
+    [ "$(dst "$LIB/Junk Numbers/Attack Pack S03E2019.mkv")" = "$want" ]
+}
+
+@test "a resolution glued to the marker still reads as one episode" {
+    file "$LIB/Junk Numbers/Glued Mark S01E02-1080p.mkv" 5000
+    title_row "$LIB/Junk Numbers/Glued Mark S01E02-1080p.mkv" \
+        "The Examples" 2010 1 2 "" "" "Second Look"
+    plan
+    [ "$(dst "$LIB/Junk Numbers/Glued Mark S01E02-1080p.mkv")" \
+        = "$LIB/The Examples (2010)/Season 01/The Examples (2010) - S01E02 - Second Look.mkv" ]
+}
+
 # Anime numbering is the one place a tvdb title can be confidently wrong: a pack
 # numbered by its release group still names S05E01, and tvdb answers that slot with
 # a different episode. A season the authority does not hold, or holds fewer of than
@@ -655,6 +695,20 @@ plan_map() {                          # the plan, with the operator's map in han
     [ "$status" -eq 0 ]
 }
 
+@test "a map row whose folder is on the array but never became a series says so" {
+    # The array holds three folders exactly like this one: a credits/extras pack
+    # the operator mapped to a record, whose every file is bonus material with no
+    # series to own it. The folder IS on the array, and telling the operator the
+    # pack moved sends them hunting for a folder that never left.
+    printf 'Credit Pack\t114801\tthe record tvdb names\n' > "$BATS_TEST_TMPDIR/ids.tsv"
+    file "$LIB/Credit Pack/01. Extras/NCOP 01.mkv" 1200
+    plan_map "$BATS_TEST_TMPDIR/ids.tsv"
+    grep -q "bonus material with no series to own it" "$EXC"
+    grep -q "never became a series" "$EXC"
+    run grep "no folder on the array" "$EXC"
+    [ "$status" -ne 0 ]
+}
+
 @test "a map line whose id is not a number stops the plan" {
     printf 'Black Clover 001\tnotanid\ttypo\n' > "$BATS_TEST_TMPDIR/ids.tsv"
     file "$LIB/Black Clover 001/Season 01/Black.Clover.001.mkv" 5000