From f7083f175dbac3688f98d5dd859ea06e1b4a5f8b Mon Sep 17 00:00:00 2001 From: dank Date: Sat, 31 Oct 2026 18:04:37 -0700 Subject: [PATCH] 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 | 77 ++++++++++++++++--- .../Dankserver/Media Relayout.md | 7 ++ tests/media_relayout.bats | 59 ++++++++++++++ 3 files changed, 134 insertions(+), 9 deletions(-) diff --git a/scripts/media-apply.py b/scripts/media-apply.py index 88fe766..f497cf0 100755 --- a/scripts/media-apply.py +++ b/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}") diff --git a/serverAppConfigs-docs/Dankserver/Media Relayout.md b/serverAppConfigs-docs/Dankserver/Media Relayout.md index d361960..b5508e3 100644 --- a/serverAppConfigs-docs/Dankserver/Media Relayout.md +++ b/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 diff --git a/tests/media_relayout.bats b/tests/media_relayout.bats index 3884a90..e3b2af4 100755 --- a/tests/media_relayout.bats +++ b/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 -- 2.54.0