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
-                   and r.get("src") == src for r in self.records)
+        return (src, dst) in self._intents
 
     @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)
-        dirs_new += 1
+        queued.append(d)
+        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)
-        moved += 1
+        pending.append((src, dst))
+        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}")