Add stage-2 decomposition benchmark; fix run_with_timeout latency and enforceability

bin/bench.jl reports stage 2 as a single throughput number, which can't
distinguish slow extraction from a slow spawn — and those have opposite fixes.
bin/bench_stage2.jl times each component in isolation, then times the real
handle_known_job end to end. It draws its corpus from real files (default
data/done) because random bytes make exiftool bail out early and understate the
stage by ~10x, and it prices both fork-free alternatives (batched, -stay_open)
so the cost of one-fork-per-file is a measurement rather than a guess.

The benchmark found stage 2 to be ~98% exiftool, and found two problems in
run_with_timeout, which stages 2 and 4 share:

1. The watchdog polled with sleep(0.1) and then joined the polling task, so
   every call paid the remainder of an in-flight sleep after the child had
   already exited: ~25 ms per file, and a measured 101 ms on a process that
   exits instantly. Replaced with a one-shot Timer cancelled when the child
   exits. Stage 2 goes from 164.6 ms to 138.3 ms per file (6 -> 8 files/s on one
   worker); the wrapper is now within noise of a bare Base.run.

2. Writing the missing tests showed the timeout was never enforceable, in the
   old implementation as much as the new. wait(proc) returns only once the
   captured stdout pipe closes, and grandchildren inherit that pipe, so
   signalling the child alone left the worker blocked until the whole process
   tree finished on its own — a `sh -c "trap '' TERM; sleep 30"` child ran the
   full 30 s against a 1 s timeout. The child now runs in its own process group
   and the timeout signals the group. The trade is that a hard crash of the
   server orphans an in-flight child rather than taking it down with it.

Three new tests cover the fast path, the timeout, and the SIGTERM-ignoring
escalation; the second was previously unexercised, which is why the bug stood.

Not addressed here, but measured and documented in the README: Perl interpreter
startup is 76.7 ms of the remaining 135.9 ms call, so a persistent exiftool
(-stay_open, 40.1 ms/file) would cut the stage by roughly another 70%. And
fsync_dir measures 1.75 us, too fast to be a real flush — commit_enriched!'s
durability may not hold on this filesystem, which is a correctness question
left open.

Claude-Session: https://claude.ai/code/session_01Xy9At7HNLHWNmfh1yw71Uy

README.md

@@ -510,11 +510,13 @@ the moment it is wired up, and never on the read path.
 
 ## Benchmarking (throughput + memory)
 
-There are three harnesses. Only the first needs a running server:
+There are five harnesses. Only the first needs a running server:
 
 | script | measures | server? |
 |---|---|---|
 | `bin/bench.jl` (below) | intake, end-to-end and per-stage throughput; server RSS | **yes** |
+| [`bin/bench_stage1.jl`](#stage-1-component-benchmark-binbench_stage1jl) | stage 1 taken apart: classify vs. rename vs. enqueue vs. logging | no |
+| [`bin/bench_stage2.jl`](#stage-2-component-benchmark-binbench_stage2jl) | stage 2 taken apart: exiftool spawn vs. extraction vs. commit | no |
 | [`bin/bench_model.jl`](#model-microbenchmark-binbench_modeljl) | the classifier alone: inference, feature reads, thread scaling | no |
 | [`bin/cluster_calibrate.jl`](#unknown-format-discovery-stage-5-offline) | stage-5 clustering quality vs. an NCD baseline | no |
 
@@ -526,6 +528,12 @@ julia --project=. -e 'using Pkg; Pkg.instantiate()'   # once
 # 1. the model, on its own — no server involved
 julia --project=. -t auto bin/bench_model.jl
 
+# 1b. stage 1 taken apart — also no server
+julia --project=. -t auto bin/bench_stage1.jl
+
+# 1c. stage 2 taken apart — needs a directory of real files, not generated ones
+julia --project=. -t auto bin/bench_stage2.jl
+
 # 2. the pipeline. Start the server in one terminal…
 julia --project=. -t auto bin/server.jl
 
@@ -631,12 +639,160 @@ resets the kernel's peak-RSS counter (`/proc/<pid>/clear_refs`) per run and flag
 a drifted baseline, but for a clean growth figure restart the server between
 memory runs.
 
+### Stage-1 component benchmark (`bin/bench_stage1.jl`)
+
+`bin/bench.jl` reports stage 1 as one number and `bin/bench_model.jl` takes the
+*classifier* apart — but stage 1 is more than the model. Per file it also
+renames the file into its stage directory, pushes a reference onto the
+downstream queue, and logs. `bin/bench_stage1.jl` times each of those in
+isolation, then times the real `handle_classify_job` end to end so the parts can
+be checked against the whole:
+
+```bash
+julia --project=. -t auto bin/bench_stage1.jl
+```
+
+Measured on this machine (Ryzen 7 2700X, 8 cores/16 threads, Julia 1.12; 2000 ×
+64 KiB files, minimum of 5 trials):
+
+| component | per file | share of the handler |
+|---|---|---|
+| `classify()` | 10.7 µs | 38% |
+| ↳ `read_features` | 7.9 µs | 28% |
+| ↳ `Lux.apply` | 2.3 µs | 8% |
+| `move_to` (rename) | 11.7 µs | 41% |
+| `enqueue_blocking!` | 0.12 µs | 0.4% |
+| per-file logging (disabled `@debug`) | 0.29 µs | 1% |
+| **`handle_classify_job`** | **28.3 µs** | 100% |
+
+**This benchmark is why stage 1's per-file log lines are `@debug` rather than
+`@info`.** As `@info` they cost ~71 µs of the handler's ~118 µs — about 6× the
+classifier and 6× the rename — and nearly all of it was `ConsoleLogger`
+*formatting* (~64 µs), not the `FlushLogger`'s per-message flush (~8 µs on top).
+Demoting them took stage 1 from 8.5k files/s to 35.3k files/s on a single worker,
+a 4.2× speedup for no algorithmic change. The script still prices a formatted
+line, so the cost of turning them back on is visible: running the handler under
+`JULIA_DEBUG=FileServer` measures 133 µs per file, a 4.7× slowdown. That is the
+trade — per-file tracing is available when you want it, and off by default,
+with `GET /stats` giving per-file observability that is counted rather than
+formatted.
+
+What's left is evenly split between the rename and the classifier, and neither
+has an easy 2×. Two things worth knowing:
+
+- **The rename, not the model, is the single largest component** (11.7 µs), and
+  it's a plain `mv` within one filesystem. Inside `classify`, the same pattern
+  holds: 7.9 µs of the 10.7 µs is `read_features` — the `open`, the two reads
+  and the `seek` — against 2.3 µs of actual inference. Stage 1 is now a
+  filesystem-bound stage with a neural network attached, not the reverse.
+- **Stage 1 now peaks at ~4 workers.** With the logger removed from the hot path
+  the sweep reads 35.0k/s at 1 worker, 66.6k/s at 2, **73.3k/s at 4**, then
+  *falls back* to 60.1k/s at 8 and 53.3k/s at 16 — every worker renaming into the
+  same two directories contends on the same directory inode. That ceiling
+  coincides with the one `bin/bench_model.jl` finds for inference, so ~4 is the
+  number from both directions: raising `FS_WORKERS` past it costs throughput.
+
+Reported times are the **minimum** over trials. Flags: `--files`, `--reps`,
+`--trials`, `--size`, `--dir`, `--model`, `--threads`, `--no-threads`,
+`--json PATH`.
+
+### Stage-2 component benchmark (`bin/bench_stage2.jl`)
+
+Stage 2 is the one stage whose cost is dominated by something outside Julia
+entirely: it forks `exiftool`, a Perl program, once per file. `bin/bench.jl`
+reports the stage as a single throughput number, which can't distinguish "the
+extraction is slow" from "the *spawn* is slow" — and those have opposite fixes.
+`bin/bench_stage2.jl` times each piece in isolation, then times the real
+`handle_known_job` end to end:
+
+```bash
+julia --project=. -t auto bin/bench_stage2.jl
+```
+
+Two things make this benchmark different from the stage-1 one:
+
+- **The corpus must be real files.** exiftool's cost depends on what it finds; a
+  file of random bytes bails out early and understates the stage by ~10×. The
+  default corpus is `data/done` — files that already went through stage 2 on this
+  machine. `--corpus PATH` points it elsewhere.
+- **It prices the alternatives to one-fork-per-file**, because if the fork
+  dominates then the only fixes are to stop paying it per file. `exiftool
+  (batched Nx)` runs the whole corpus through one process; `exiftool
+  (-stay_open)` keeps one process alive and feeds it one file at a time over a
+  pipe — the shape a streaming pipeline could actually adopt. Both are measured,
+  not assumed.
+
+Measured on this machine (Ryzen 7 2700X, 8 cores/16 threads, Julia 1.12,
+exiftool 12.40; 150 real files / 102 MiB, minimum of 2 trials):
+
+| component | per file | share of the handler |
+|---|---|---|
+| `run_exiftool()` | 135.9 ms | 98% |
+| ↳ bare fork + Perl boot (`exiftool -ver`) | 76.7 ms | 55% |
+| ↳ `JSON3.read` + tag map | 10 µs | 0.0% |
+| `normalize_metadata` | 1.7 µs | 0.0% |
+| `commit_enriched!` (sidecar + fsyncs + rename) | 2.0 ms | 1.5% |
+| per-file logging (`@info`, flush→file) | 47 µs | 0.0% |
+| **`handle_known_job`** | **138.3 ms** | 100% |
+| *alt:* `exiftool -stay_open` | 40.1 ms | 29% |
+| *alt:* `exiftool` batched 150× | 37.3 ms | 27% |
+
+**Stage 2 is exiftool and nothing else.** Everything the Julia code does —
+parsing, normalizing, the durable sidecar-first commit, the log line — sums to
+about 1.5% of the stage. There is no point optimizing any of it.
+
+**More than half the stage is interpreter startup, not metadata extraction.**
+The bare `exiftool -ver` (fork, Perl boot, module loads, read no file) costs
+76.7 ms against a 135.9 ms full call. Both fork-free alternatives agree on what's
+left: ~37–40 ms of actual work per file. So a persistent exiftool would cut the
+stage by ~70%, and `-stay_open` gets there without giving up the one-file-in,
+one-result-out shape the pipeline needs. That remains the single biggest
+available win in this stage; it is measured here but not yet implemented.
+
+**This benchmark is also why `run_with_timeout` no longer polls.** The original
+watchdog polled with `sleep(0.1)` and then joined the polling task, so every call
+paid the remainder of an in-flight sleep *after* the child had already exited —
+~25 ms per file here, and a measured 101 ms on a process that exits instantly.
+Replacing it with a one-shot `Timer` took the stage from 164.6 ms to 138.3 ms per
+file (6→8 files/s on one worker) and cost nothing in behavior. Stage 4 shares the
+wrapper and got the same fix for free.
+
+Writing the missing tests for that wrapper turned up a second, worse problem:
+**the timeout was never enforceable.** `wait(proc)` returns only once the
+captured stdout pipe closes, and any grandchild inherits that pipe — so
+signalling the child alone left the worker blocked until the whole process tree
+finished on its own (a `sh -c "trap '' TERM; sleep 30"` child ran the full 30 s
+against a 1 s timeout, under both the old and new watchdog). The child now runs
+in its own process group and the timeout signals the group. The trade is that a
+hard crash of the server orphans an in-flight child rather than taking it down;
+these children are short-lived and timeout-bounded, which is the cheaper side of
+it.
+
+**Stage 2 scales to ~8 workers, then flattens**: 8 files/s at 1 worker, 14 at 2,
+28 at 4, **49 at 8**, and 49 at 16 — the machine runs out of cores to run Perl
+on, which is exactly what you'd expect of a stage that is ~100% subprocess. Note
+that the sweep pulls from a shared counter rather than splitting the corpus into
+contiguous slices: per-file exiftool time spans two orders of magnitude on a real
+corpus (one 2.1 s archive among 48 files), and a static split reports a scaling
+ceiling that is really just load imbalance.
+
+One caveat the numbers raise but don't answer: **`fsync_dir` measures 1.75 µs**,
+which is far too fast to be a real disk flush. The durability that
+`commit_enriched!` is written for may not survive power loss on this filesystem,
+even though the code is correct. That's a correctness question, not a speed one,
+and it is not yet resolved.
+
+Reported times are the **minimum** over trials. Flags: `--files`, `--reps`,
+`--trials`, `--corpus`, `--dir`, `--timeout`, `--threads`, `--no-threads`,
+`--no-stay-open`, `--json PATH`.
+
 ### Model microbenchmark (`bin/bench_model.jl`)
 
 `bin/bench.jl` reports stage 1 as a single number — the wall time of
-`handle_classify_job`, which is a feature read, an inference, a rename, a log
-line, and whatever contention the other three pools create. That's the right
-number for capacity planning and the wrong one for "is the model slow?".
+`handle_classify_job`, which is a feature read, an inference, a rename, a
+(disabled) debug line, and whatever contention the other three pools create.
+That's the right number for capacity planning and the wrong one for "is the
+model slow?".
 `bin/bench_model.jl` answers that separately, with no server, queue, or HTTP
 involved:
 
@@ -654,7 +810,12 @@ Measured on this machine (Ryzen 7 2700X, 8 cores/16 threads, Julia 1.12):
 
 So the model is **not** the pipeline's problem, by three orders of magnitude: the
 same run measured stage 1 at 38.7 ms per file, ~4,500× the 8.4 µs `classify()`
-costs. Whatever stage 1 spends its time on, it isn't the network.
+costs. Whatever stage 1 spends its time on, it isn't the network. (That 38.7 ms
+predates the `@debug` demotion above and is a whole-pipeline figure — it includes
+time the stage-1 worker spends *blocked* on a full downstream queue, which is why
+it is three orders of magnitude above the 28.3 µs the handler costs in
+isolation. For the uncontended split, see
+[the stage-1 decomposition](#stage-1-component-benchmark-binbench_stage1jl).)
 
 Two findings worth acting on if stage 1 ever *does* become the constraint:
 
@@ -702,6 +863,7 @@ bin/
   server.jl           entry point
   bench.jl            throughput + memory harness against a running server
   bench_model.jl      classifier microbenchmark (inference, feature reads, scaling)
+  bench_stage1.jl     stage-1 decomposition (classify vs. rename vs. enqueue vs. logging)
   train.jl            offline training script → model/classifier.jld2
   cluster_calibrate.jl offline stage-5 hyperparameter calibration + NCD baseline
   cluster_sweep.jl    stage-5 phase-B runner: sweep binary/, update catalog, write nominations

bin/bench_stage1.jl

@@ -0,0 +1,556 @@
+#!/usr/bin/env julia
+#
+# bench_stage1.jl — take stage 1 apart and find the slowest component.
+#
+# bin/bench.jl reports stage 1 as a single number (the wall time of
+# `handle_classify_job` under whole-pipeline contention) and bin/bench_model.jl
+# takes the *classifier* apart. Neither answers "which part of stage 1 costs the
+# most?", because stage 1 is more than the model. Per file it does:
+#
+#   classify        filesize + two 16-byte reads + a 32x1 forward pass
+#     read_features   open, read head, seek, read tail, scale to Float32
+#     Lux.apply       the network on a feature vector already in memory
+#   move_to         rename spool/<f> -> known/<f> or unknown/<f>
+#   enqueue         push a Job reference onto the downstream bounded queue
+#   logging         two @info lines ("classified file", "routed to ...")
+#
+# This script times each of those in isolation, then times the real
+# `handle_classify_job` end to end so the parts can be checked against the whole.
+# The two knobs that most change the answer get their own sweeps:
+#
+#   * Logger. The server runs `FlushLogger(ConsoleLogger(stderr))` — it formats
+#     and flushes every message. Under redirect (a log file, journald) that is a
+#     syscall per line, two lines per file, on the hot path. We time the handler
+#     under a null logger, a formatting-but-discarding logger, and the real
+#     flushing-to-file logger, so the cost of logging is a subtraction, not a
+#     guess. This sweep is what demoted stage 1's per-file lines to `@debug`
+#     (see the note in src/worker.jl); the standalone `logging (...)` rows below
+#     still price a *formatted* line, i.e. what those lines cost when switched
+#     back on with `JULIA_DEBUG=FileServer`, while the handler rows show what the
+#     stage pays with them off.
+#   * Concurrency. Components that own a lock (the queue's condition, the
+#     logger's stream) don't scale, and the ranking at one worker need not be the
+#     ranking at sixteen. The `--threads` sweep runs the full handler across
+#     worker counts.
+#
+# Usage:
+#   julia --project=. -t auto bin/bench_stage1.jl [options]
+#
+#   --files N        files per timed pass for consuming benchmarks (default: 2000)
+#   --reps N         calls per timed pass for non-consuming benchmarks (default: 20000)
+#   --trials N       timed passes; the minimum is reported (default: 5)
+#   --size SPEC      corpus file size (default: 64k)
+#   --dir PATH       working directory for the corpus (default: a temp dir under data/)
+#   --model PATH     classifier artifact (default: $FS_MODEL_PATH or model/classifier.jld2)
+#   --threads LIST   worker counts for the concurrency sweep (default: 1,2,4,8,nthreads)
+#   --no-threads     skip the concurrency sweep
+#   --json PATH      also write the results as JSON
+#
+# Reported times are the *minimum* over trials: the floor is the signal and
+# everything above it is scheduler, page-cache and GC noise.
+
+using FileServer
+using Lux
+using JSON3
+using Logging
+using Printf
+using Random
+
+const FS = FileServer
+
+# ---------------------------------------------------------------- option parsing
+
+const DEFAULTS = Dict{String,Any}(
+    "files"      => 2000,
+    "reps"       => 20_000,
+    "trials"     => 5,
+    "size"       => "64k",
+    "dir"        => nothing,
+    "model"      => get(ENV, "FS_MODEL_PATH", "model/classifier.jld2"),
+    "threads"    => nothing,
+    "no-threads" => false,
+    "json"       => nothing,
+)
+
+const FLAGS = ("no-threads",)
+const INTS  = ("files", "reps", "trials")
+
+function parse_size(s::AbstractString)::Int
+    m = match(r"^(\d+(?:\.\d+)?)\s*([kKmMgG]?)[bB]?$", strip(s))
+    m === nothing && error("bad size: $s (expected e.g. 512, 64k, 8m, 1g)")
+    mult = Dict('k' => 1024, 'm' => 1024^2, 'g' => 1024^3)
+    scale = isempty(m[2]) ? 1 : mult[lowercase(m[2])[1]]
+    return round(Int, parse(Float64, m[1]) * scale)
+end
+
+function parse_args(argv)
+    opts = copy(DEFAULTS)
+    i = 1
+    while i <= length(argv)
+        a = argv[i]
+        startswith(a, "--") || error("unexpected argument: $a")
+        key = a[3:end]
+        haskey(opts, key) || error("unknown option: $a")
+        if key in FLAGS
+            opts[key] = true; i += 1; continue
+        end
+        i + 1 <= length(argv) || error("option --$key needs a value")
+        opts[key] = key in INTS ? parse(Int, argv[i+1]) : argv[i+1]
+        i += 2
+    end
+    return opts
+end
+
+# ------------------------------------------------------------------- measurement
+
+# Every timed loop stores its result here. Without a visible side effect the
+# compiler is free to hoist a pure call out of the loop and we would be timing an
+# empty `for`.
+const SINK = Ref{Any}(nothing)
+
+"""
+    best_of(pass, prepare; trials) -> ns_per_op
+
+Run `pass()` `trials` times and report the fastest, in nanoseconds per operation
+(`pass` returns the number of operations it performed). `prepare()` runs before
+each pass and is *not* timed — that is where a consuming benchmark puts the file
+back where it started. `pass` comes first so callers can pass it as a `do` block.
+
+The first pass is thrown away: it pays Julia's JIT compilation, which on calls
+this small is orders of magnitude more than the thing being measured.
+"""
+function best_of(pass, prepare; trials::Int)
+    best = Inf
+    for t in 0:trials
+        prepare()
+        GC.gc()
+        t0 = time_ns()
+        n = pass()
+        dt = Float64(time_ns() - t0)
+        t == 0 && continue          # warm-up: compiled, not measured
+        best = min(best, dt / n)
+    end
+    return best
+end
+
+noop() = nothing
+
+# ------------------------------------------------------------------- formatting
+
+function human_time(ns::Real)
+    ns < 1_000      && return @sprintf("%.0f ns", ns)
+    ns < 1_000_000  && return @sprintf("%.2f µs", ns / 1e3)
+    ns < 1e9        && return @sprintf("%.2f ms", ns / 1e6)
+    return @sprintf("%.2f s", ns / 1e9)
+end
+
+function human_rate(r::Real)
+    r >= 1e6 && return @sprintf("%.2fM/s", r / 1e6)
+    r >= 1e3 && return @sprintf("%.1fk/s", r / 1e3)
+    return @sprintf("%.0f/s", r)
+end
+
+rate(ns::Real) = 1e9 / max(ns, 1e-9)
+
+rule(n = 78) = println("-" ^ n)
+
+function header(title)
+    println()
+    println(title)
+    rule()
+end
+
+# ------------------------------------------------------------------------- corpus
+
+"""
+Write a file of exactly `size` random bytes, in bounded chunks.
+
+Content is random rather than zeros so the classifier sees a realistic input —
+and so the filesystem can't cheat with a sparse file, which would make the tail
+`seek` unrepresentatively fast.
+"""
+function write_file(path::AbstractString, size::Int, rng)
+    chunk = 1024 * 1024
+    open(path, "w") do io
+        remaining = size
+        while remaining > 0
+            n = min(chunk, remaining)
+            write(io, rand(rng, UInt8, n))
+            remaining -= n
+        end
+    end
+    return path
+end
+
+"""
+    make_corpus(cfg, n, size, rng) -> Vector{Job}
+
+Create `n` spooled files and the `Job` references a stage-1 worker would dequeue
+for them — the exact input `handle_classify_job` sees.
+"""
+function make_corpus(cfg::FS.Config, n::Int, size::Int, rng)
+    jobs = FS.Job[]
+    for i in 1:n
+        id, path = FS.spool_path(cfg, @sprintf("bench-%06d.bin", i))
+        write_file(path, size, rng)
+        push!(jobs, FS.Job(id, basename(path), path, size, time()))
+    end
+    return jobs
+end
+
+"""
+    respool!(cfg, jobs)
+
+Put every corpus file back in `spool/`, wherever the last pass left it (known/,
+unknown/, or already home). This is the untimed `prepare` step for benchmarks
+that consume their input by moving it.
+"""
+function respool!(cfg::FS.Config, jobs::Vector{FS.Job})
+    for job in jobs
+        isfile(job.path) && continue
+        for dir in (cfg.known_dir, cfg.unknown_dir, cfg.failed_dir)
+            candidate = joinpath(dir, basename(job.path))
+            if isfile(candidate)
+                mv(candidate, job.path; force = true)
+                break
+            end
+        end
+    end
+    return nothing
+end
+
+"Drain a queue without blocking, so the next pass starts from empty."
+function drain!(q::FS.ChannelQueue)
+    while length(q) > 0
+        FS.dequeue!(q)
+    end
+    return nothing
+end
+
+# --------------------------------------------------------------------- loggers
+
+"""
+    with_logger_named(name, path, f)
+
+Run `f` under one of the three loggers the cost of logging is bracketed by:
+
+  * `:null`    — `NullLogger`: the `@info` macro's own overhead, nothing else.
+  * `:format`  — `ConsoleLogger` to `devnull`: message formatting and key/value
+                 interpolation, but no I/O.
+  * `:flush`   — `FlushLogger(ConsoleLogger(io))` to a real file: what
+                 `FileServer.run` installs, under the redirect it was written for.
+  * `:debug`   — the same, at `Debug` level: the stage's per-file lines are
+                 `@debug`, so this is the equivalent of running the server with
+                 `JULIA_DEBUG=FileServer` and the only setting under which they
+                 are emitted at all.
+"""
+function with_logger_named(f, which::Symbol, path::AbstractString)
+    if which === :null
+        return with_logger(f, NullLogger())
+    elseif which === :format
+        return with_logger(f, ConsoleLogger(devnull))
+    elseif which === :flush || which === :debug
+        level = which === :debug ? Logging.Debug : Logging.Info
+        return open(path, "w") do io
+            with_logger(f, FS.FlushLogger(ConsoleLogger(io, level)))
+        end
+    end
+    error("unknown logger: $which")
+end
+
+# ------------------------------------------------------------------ components
+
+"""
+    component_rows(cfg, clf, jobs, opts) -> Vector
+
+Time each piece of stage 1 on its own. Non-consuming pieces (`filesize`,
+`read_features`, `Lux.apply`, `classify`, the log lines) run `reps` times over
+the corpus; consuming pieces (`move_to`, the full handler) run once per corpus
+file with an untimed reset between passes.
+"""
+function component_rows(cfg::FS.Config, clf::FS.Classifier, jobs::Vector{FS.Job}, opts)
+    reps, trials = opts["reps"], opts["trials"]
+    nfiles = length(jobs)
+    paths = [j.path for j in jobs]
+    rows = []
+
+    # A feature vector already in memory, so the inference row measures the net
+    # and not the disk read in front of it.
+    feats = FS.read_features(paths[1])
+    x = reshape(feats, FS.FEATURE_DIM, 1)
+
+    logger = ConsoleLogger(devnull)   # components other than the log rows: quiet
+
+    # --- filesize: the stat() read_features does before touching the bytes
+    push!(rows, (; name = "filesize (stat)", part = "classify",
+        ns = best_of(noop; trials) do
+            @inbounds for i in 1:reps
+                SINK[] = filesize(paths[(i - 1) % nfiles + 1])
+            end
+            reps
+        end))
+
+    # --- read_features: open + head read + seek + tail read + scale
+    push!(rows, (; name = "read_features", part = "classify",
+        ns = best_of(noop; trials) do
+            @inbounds for i in 1:reps
+                SINK[] = FS.read_features(paths[(i - 1) % nfiles + 1])
+            end
+            reps
+        end))
+
+    # --- Lux.apply: the network alone, features already in memory
+    push!(rows, (; name = "Lux.apply (1x32)", part = "classify",
+        ns = best_of(noop; trials) do
+            for _ in 1:reps
+                SINK[] = Lux.apply(clf.model, x, clf.ps, clf.st)
+            end
+            reps
+        end))
+
+    # --- classify: read_features + apply + argmax, what the handler calls
+    push!(rows, (; name = "classify (total)", part = "classify",
+        ns = best_of(noop; trials) do
+            @inbounds for i in 1:reps
+                SINK[] = FS.classify(clf, paths[(i - 1) % nfiles + 1])
+            end
+            reps
+        end))
+
+    # --- move_to: the rename out of spool/. Consuming: reset before each pass.
+    push!(rows, (; name = "move_to (rename)", part = "route",
+        ns = best_of(() -> respool!(cfg, jobs); trials) do
+            @inbounds for job in jobs
+                SINK[] = FS.move_to(cfg.unknown_dir, job)
+            end
+            nfiles
+        end))
+
+    # --- enqueue: lock, push, notify on an uncontended, non-full queue
+    q = FS.ChannelQueue(nfiles + 1)
+    stats = FS.StageStats()
+    push!(rows, (; name = "enqueue_blocking!", part = "route",
+        ns = best_of(() -> drain!(q); trials) do
+            @inbounds for job in jobs
+                SINK[] = FS.enqueue_blocking!(q, job, stats;
+                                              retry_seconds = FS.ROUTE_ENQUEUE_RETRY_SECONDS)
+            end
+            nfiles
+        end))
+    drain!(q)
+
+    # --- the two @info lines, under each of the three loggers
+    job1 = jobs[1]
+    logfile = joinpath(cfg.spool_dir, "..", "bench_stage1.log")
+    for (which, label) in ((:null, "logging (NullLogger)"),
+                           (:format, "logging (format only)"),
+                           (:flush, "logging (flush→file)"))
+        ns = with_logger_named(which, logfile) do
+            best_of(noop; trials) do
+                for _ in 1:reps
+                    @info "classified file" worker=1 id=job1.id name=job1.original_name size=job1.size classification=:unknown
+                    @info "routed to content triage" worker=1 id=job1.id dest=job1.path
+                end
+                reps
+            end
+        end
+        push!(rows, (; name = label, part = "log", ns))
+    end
+    rm(logfile; force = true)
+
+    return rows, logger
+end
+
+"""
+    handler_rows(cfg, jobs, opts) -> Vector
+
+Time the real `handle_classify_job` end to end under each logger. The difference
+between the rows is the cost logging adds to a file; the `:flush` row is what the
+running server actually pays.
+"""
+function handler_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts)
+    trials = opts["trials"]
+    nfiles = length(jobs)
+    logfile = joinpath(dirname(cfg.spool_dir), "bench_stage1.log")
+    known   = FS.ChannelQueue(nfiles + 1)
+    unknown = FS.ChannelQueue(nfiles + 1)
+    stats   = FS.StageStats()
+    rows = []
+    for (which, label) in ((:null, "handle_classify_job (NullLogger)"),
+                           (:format, "handle_classify_job (format only)"),
+                           (:flush, "handle_classify_job (flush→file)"),
+                           (:debug, "handle_classify_job (JULIA_DEBUG)"))
+        ns = with_logger_named(which, logfile) do
+            best_of(() -> (respool!(cfg, jobs); drain!(known); drain!(unknown)); trials) do
+                @inbounds for job in jobs
+                    FS.handle_classify_job(job, cfg, 1, known, unknown, stats)
+                end
+                nfiles
+            end
+        end
+        push!(rows, (; name = label, part = "total", ns))
+    end
+    respool!(cfg, jobs); drain!(known); drain!(unknown)
+    rm(logfile; force = true)
+    return rows
+end
+
+"""
+    thread_rows(cfg, jobs, opts) -> Vector
+
+Run the full handler across worker counts, under the server's real logger. A
+component that owns a lock — the queue's condition variable, the logger's
+stream — stops scaling here even though it looked cheap single-threaded, so this
+is where the single-thread ranking gets checked against the deployed one.
+"""
+function thread_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts)
+    trials = opts["trials"]
+    nfiles = length(jobs)
+    counts = opts["threads"] === nothing ?
+        unique([1; 2; 4; 8; Threads.nthreads()]) :
+        [parse(Int, s) for s in split(String(opts["threads"]), ",")]
+    counts = sort(unique(filter(k -> 1 <= k <= Threads.nthreads(), counts)))
+
+    logfile = joinpath(dirname(cfg.spool_dir), "bench_stage1.log")
+    known   = FS.ChannelQueue(nfiles + 1)
+    unknown = FS.ChannelQueue(nfiles + 1)
+    stats   = FS.StageStats()
+    rows = []
+    base = 0.0
+    for k in counts
+        ns = with_logger_named(:flush, logfile) do
+            best_of(() -> (respool!(cfg, jobs); drain!(known); drain!(unknown)); trials) do
+                # Static split: each task takes a contiguous slice, so the only
+                # sharing between workers is the state the server also shares —
+                # the classifier, the queues, the logger, the filesystem.
+                chunk = cld(nfiles, k)
+                @sync for t in 1:k
+                    lo = (t - 1) * chunk + 1
+                    hi = min(t * chunk, nfiles)
+                    lo > hi && continue
+                    Threads.@spawn begin
+                        @inbounds for i in lo:hi
+                            FS.handle_classify_job(jobs[i], cfg, t, known, unknown, stats)
+                        end
+                    end
+                end
+                nfiles
+            end
+        end
+        r = rate(ns) # files/sec aggregate (ns is already per file, wall-clock)
+        k == counts[1] && (base = r)
+        push!(rows, (; workers = k, ns, files_per_sec = r, speedup = r / base))
+    end
+    respool!(cfg, jobs); drain!(known); drain!(unknown)
+    rm(logfile; force = true)
+    return rows
+end
+
+# ------------------------------------------------------------------- reporting
+
+function print_components(rows, total_ns)
+    @printf("%-34s %-9s %12s %10s %9s\n", "component", "part", "per file", "rate", "% total")
+    rule()
+    for r in rows
+        @printf("%-34s %-9s %12s %10s %8.1f%%\n", r.name, r.part, human_time(r.ns),
+                human_rate(rate(r.ns)), 100 * r.ns / total_ns)
+    end
+end
+
+function print_threads(rows)
+    @printf("%-9s %12s %12s %9s\n", "workers", "per file", "throughput", "speedup")
+    rule()
+    for r in rows
+        @printf("%-9d %12s %12s %8.2fx\n", r.workers, human_time(r.ns),
+                human_rate(r.files_per_sec), r.speedup)
+    end
+end
+
+# ------------------------------------------------------------------------- main
+
+function main(argv)
+    opts = parse_args(argv)
+    modelpath = String(opts["model"])
+    isfile(modelpath) || (println(stderr, "model artifact not found: $modelpath"); return 1)
+
+    size = parse_size(String(opts["size"]))
+    nfiles, trials = opts["files"], opts["trials"]
+
+    root = opts["dir"] === nothing ?
+        mktempdir(pwd(); prefix = "bench_stage1_") : String(opts["dir"])
+    owned = opts["dir"] === nothing
+    cfg = FS.Config(
+        spool_dir   = joinpath(root, "spool"),
+        known_dir   = joinpath(root, "known"),
+        unknown_dir = joinpath(root, "unknown"),
+        failed_dir  = joinpath(root, "failed"),
+        model_path  = modelpath,
+    )
+    for d in (cfg.spool_dir, cfg.known_dir, cfg.unknown_dir, cfg.failed_dir)
+        mkpath(d)
+    end
+
+    clf = FS.load_classifier(modelpath)
+    FS.CLASSIFIER[] = clf     # handle_classify_job reads the global, as in the server
+
+    println("stage-1 component benchmark")
+    rule()
+    @printf("%-22s %s\n", "julia threads",  Threads.nthreads())
+    @printf("%-22s %s\n", "model",          modelpath)
+    @printf("%-22s %s\n", "corpus",         "$(nfiles) files x $(size) B in $(root)")
+    @printf("%-22s %s\n", "reps / trials",  "$(opts["reps"]) / $(trials)")
+
+    rng = MersenneTwister(0x5741524d)
+    jobs = make_corpus(cfg, nfiles, size, rng)
+
+    try
+        comps, _ = component_rows(cfg, clf, jobs, opts)
+        handlers = handler_rows(cfg, jobs, opts)
+
+        # The denominator is the handler as the server actually runs it: the real
+        # flushing logger at Info level, one worker. Percentages are shares of
+        # that, so they are directly comparable and the parts can be checked
+        # against the whole. The JULIA_DEBUG row is deliberately *not* the
+        # baseline — it is the opt-in configuration, and letting it set the scale
+        # would make every other component look free.
+        total = only(r.ns for r in handlers if r.name == "handle_classify_job (flush→file)")
+
+        header("Components (single worker)")
+        print_components(comps, total)
+
+        header("Whole handler, by logger")
+        print_components(handlers, total)
+
+        # Stage 1's own log lines are `@debug`, so what the deployed handler pays
+        # for them is the disabled-macro cost, not a formatted line.
+        accounted = sum(r.ns for r in comps if r.name in
+            ("classify (total)", "move_to (rename)", "enqueue_blocking!", "logging (NullLogger)"))
+        println()
+        @printf("accounted: %s of %s (%.0f%%); unaccounted overhead %s\n",
+                human_time(accounted), human_time(total), 100 * accounted / total,
+                human_time(max(total - accounted, 0)))
+
+        threads = nothing
+        if !opts["no-threads"] && Threads.nthreads() > 1
+            threads = thread_rows(cfg, jobs, opts)
+            header("Full handler across workers (server logger)")
+            print_threads(threads)
+        end
+
+        if opts["json"] !== nothing
+            open(String(opts["json"]), "w") do io
+                JSON3.write(io, (;
+                    julia_threads = Threads.nthreads(),
+                    file_size = size, files = nfiles, reps = opts["reps"], trials,
+                    components = comps, handlers, threads,
+                ))
+            end
+            println("\nwrote ", opts["json"])
+        end
+    finally
+        owned && rm(root; recursive = true, force = true)
+    end
+    return 0
+end
+
+exit(main(ARGS))

bin/bench_stage2.jl

@@ -0,0 +1,699 @@
+#!/usr/bin/env julia
+#
+# bench_stage2.jl — take stage 2 apart and find the slowest component.
+#
+# bin/bench.jl reports stage 2 as a single number (its throughput and worker
+# utilization under whole-pipeline contention). It doesn't say *which part* of
+# the stage costs the most, and stage 2 is the one stage whose cost is dominated
+# by something outside Julia entirely: it forks `exiftool`, a Perl program, once
+# per file. Per file the stage does:
+#
+#   build_metadata
+#     run_exiftool        fork/exec exiftool -json -G -n, capture stdout
+#       run_with_timeout    the watchdog wrapper around the subprocess
+#     JSON3.read          parse the dump
+#     normalize_metadata  coalesce ~20 tag names into the sidecar schema
+#   finalize_known!  (commit_enriched!)
+#     JSON3.write         serialize the sidecar payload
+#     write + fsync       durably persist the sidecar bytes to a temp name
+#     mv + fsync_dir      commit the sidecar, then persist the rename itself
+#     move_to             rename known/<f> -> done/<f>, the commit point
+#   logging             one @info line ("enriched")
+#
+# This script times each of those in isolation, then times the real
+# `handle_known_job` end to end so the parts can be checked against the whole.
+#
+# Three things here that the stage-1 benchmark has no equivalent of:
+#
+#   * The corpus must be real files. exiftool's cost depends on what it finds;
+#     random bytes exit early and would understate the stage by a lot. The
+#     default corpus is `data/done` — files that already went through stage 2 on
+#     this machine — copied back into a scratch known/ dir.
+#   * Two rows price the *alternatives* to one-fork-per-file, because if the
+#     fork dominates then the only fixes are to stop paying it per file:
+#     `exiftool (batched Nx)` runs the whole corpus through one process, and
+#     `exiftool (-stay_open)` keeps a single process alive and feeds it one file
+#     at a time over a pipe — the shape a streaming pipeline could actually use.
+#     Both are measured, not assumed.
+#   * `run_with_timeout` gets its own row *next to* a bare `Base.run` of the same
+#     command. The difference is what the watchdog costs, and its polling loop
+#     (`sleep(0.1)`) is a suspicious enough design to want measured rather than
+#     reasoned about.
+#
+# The `--threads` sweep runs the full handler across worker counts: subprocess
+# spawning contends on things (the kernel's fork path, page cache, the logger's
+# stream) that a single-threaded ranking can't reveal.
+#
+# Usage:
+#   julia --project=. -t auto bin/bench_stage2.jl [options]
+#
+#   --files N        corpus files per timed pass (default: 48). The concurrency
+#                    sweep wants more than the component rows do — with a single
+#                    2 s file in the corpus, 48 files can't show more than ~4x no
+#                    matter how many workers run, so pass --files 150 when the
+#                    question is scaling.
+#   --reps N         calls per timed pass for cheap, non-consuming benchmarks (default: 2000)
+#   --trials N       timed passes; the minimum is reported (default: 3)
+#   --corpus PATH    directory of real files to draw the corpus from (default: data/done)
+#   --dir PATH       working directory for the corpus (default: a temp dir under data/)
+#   --timeout SEC    exiftool timeout, as Config.exiftool_timeout (default: 30)
+#   --threads LIST   worker counts for the concurrency sweep (default: 1,2,4,8,nthreads)
+#   --no-threads     skip the concurrency sweep
+#   --no-stay-open   skip the persistent-exiftool probe
+#   --json PATH      also write the results as JSON
+#
+# Reported times are the *minimum* over trials: the floor is the signal and
+# everything above it is scheduler, page-cache and GC noise.
+
+using FileServer
+using JSON3
+using Logging
+using Printf
+using Random
+
+const FS = FileServer
+
+# ---------------------------------------------------------------- option parsing
+
+const DEFAULTS = Dict{String,Any}(
+    "files"         => 48,
+    "reps"          => 2000,
+    "trials"        => 3,
+    "corpus"        => "data/done",
+    "dir"           => nothing,
+    "timeout"       => 30,
+    "threads"       => nothing,
+    "no-threads"    => false,
+    "no-stay-open"  => false,
+    "json"          => nothing,
+)
+
+const FLAGS = ("no-threads", "no-stay-open")
+const INTS  = ("files", "reps", "trials", "timeout")
+
+function parse_args(argv)
+    opts = copy(DEFAULTS)
+    i = 1
+    while i <= length(argv)
+        a = argv[i]
+        startswith(a, "--") || error("unexpected argument: $a")
+        key = a[3:end]
+        haskey(opts, key) || error("unknown option: $a")
+        if key in FLAGS
+            opts[key] = true; i += 1; continue
+        end
+        i + 1 <= length(argv) || error("option --$key needs a value")
+        opts[key] = key in INTS ? parse(Int, argv[i+1]) : argv[i+1]
+        i += 2
+    end
+    return opts
+end
+
+# ------------------------------------------------------------------- measurement
+
+# Every timed loop stores its result here. Without a visible side effect the
+# compiler is free to hoist a pure call out of the loop and we would be timing an
+# empty `for`.
+const SINK = Ref{Any}(nothing)
+
+"""
+    best_of(pass, prepare; trials) -> ns_per_op
+
+Run `pass()` `trials` times and report the fastest, in nanoseconds per operation
+(`pass` returns the number of operations it performed). `prepare()` runs before
+each pass and is *not* timed — that is where a consuming benchmark puts the file
+back where it started. `pass` comes first so callers can pass it as a `do` block.
+
+The first pass is thrown away: it pays Julia's JIT compilation, which on calls
+this small is orders of magnitude more than the thing being measured.
+"""
+function best_of(pass, prepare; trials::Int)
+    best = Inf
+    for t in 0:trials
+        prepare()
+        GC.gc()
+        t0 = time_ns()
+        n = pass()
+        dt = Float64(time_ns() - t0)
+        t == 0 && continue          # warm-up: compiled, not measured
+        best = min(best, dt / n)
+    end
+    return best
+end
+
+noop() = nothing
+
+# ------------------------------------------------------------------- formatting
+
+function human_time(ns::Real)
+    ns < 1_000      && return @sprintf("%.0f ns", ns)
+    ns < 1_000_000  && return @sprintf("%.2f µs", ns / 1e3)
+    ns < 1e9        && return @sprintf("%.2f ms", ns / 1e6)
+    return @sprintf("%.2f s", ns / 1e9)
+end
+
+function human_rate(r::Real)
+    r >= 1e6 && return @sprintf("%.2fM/s", r / 1e6)
+    r >= 1e3 && return @sprintf("%.1fk/s", r / 1e3)
+    return @sprintf("%.0f/s", r)
+end
+
+rate(ns::Real) = 1e9 / max(ns, 1e-9)
+
+rule(n = 84) = println("-" ^ n)
+
+function header(title)
+    println()
+    println(title)
+    rule()
+end
+
+# ------------------------------------------------------------------------- corpus
+
+"""
+    make_corpus(cfg, corpus_dir, n) -> Vector{Job}
+
+Copy up to `n` real files from `corpus_dir` into `known/` and build the `Job`
+references a stage-2 worker would dequeue for them — the exact input
+`handle_known_job` sees.
+
+Real files, not generated ones: exiftool's cost is a function of what it can
+parse, and a file of random bytes bails out early enough to understate the stage
+by an order of magnitude. `.meta.json` sidecars are skipped — they are stage-2
+*output*, and enriching them would measure the wrong population.
+"""
+function make_corpus(cfg::FS.Config, corpus_dir::AbstractString, n::Int)
+    isdir(corpus_dir) || error("corpus dir not found: $corpus_dir")
+    names = filter(readdir(corpus_dir)) do f
+        !endswith(f, ".meta.json") && isfile(joinpath(corpus_dir, f))
+    end
+    isempty(names) && error("no usable files in corpus dir: $corpus_dir")
+    sort!(names)                       # deterministic selection across runs
+    length(names) > n && (names = names[1:n])
+
+    jobs = FS.Job[]
+    for (i, name) in enumerate(names)
+        src = joinpath(corpus_dir, name)
+        # Give it a fresh id/spool-style filename so nothing collides with the
+        # corpus the file came from.
+        id, spooled = FS.spool_path(cfg, @sprintf("s2-%04d-%s", i, basename(name)))
+        cp(src, spooled; force = true)
+        dest = joinpath(cfg.known_dir, basename(spooled))
+        mv(spooled, dest; force = true)
+        push!(jobs, FS.Job(id, basename(name), dest, filesize(dest), time()))
+    end
+    return jobs
+end
+
+"""
+    reknown!(cfg, jobs)
+
+Put every corpus file back in `known/`, wherever the last pass left it (done/ or
+already home), and delete any sidecar it produced. This is the untimed `prepare`
+step for benchmarks that consume their input by committing it.
+"""
+function reknown!(cfg::FS.Config, jobs::Vector{FS.Job})
+    for job in jobs
+        base = basename(job.path)
+        for dir in (cfg.done_dir, cfg.failed_dir)
+            sidecar = joinpath(dir, string(base, ".meta.json"))
+            rm(sidecar; force = true)
+            rm(string(sidecar, ".tmp"); force = true)
+        end
+        isfile(job.path) && continue
+        for dir in (cfg.done_dir, cfg.failed_dir)
+            candidate = joinpath(dir, base)
+            if isfile(candidate)
+                mv(candidate, job.path; force = true)
+                break
+            end
+        end
+    end
+    return nothing
+end
+
+# --------------------------------------------------------------------- loggers
+
+"""
+    with_logger_named(f, which, path)
+
+Run `f` under one of the loggers the cost of logging is bracketed by:
+
+  * `:null`    — `NullLogger`: the `@info` macro's own overhead, nothing else.
+  * `:format`  — `ConsoleLogger` to `devnull`: message formatting and key/value
+                 interpolation, but no I/O.
+  * `:flush`   — `FlushLogger(ConsoleLogger(io))` to a real file: what
+                 `FileServer.run` installs, under the redirect it was written
+                 for. Stage 2's per-file line is `@info`, not `@debug`, so this
+                 row is what the deployed server actually pays.
+"""
+function with_logger_named(f, which::Symbol, path::AbstractString)
+    if which === :null
+        return with_logger(f, NullLogger())
+    elseif which === :format
+        return with_logger(f, ConsoleLogger(devnull))
+    elseif which === :flush
+        return open(path, "w") do io
+            with_logger(f, FS.FlushLogger(ConsoleLogger(io, Logging.Info)))
+        end
+    end
+    error("unknown logger: $which")
+end
+
+# --------------------------------------------------- exiftool spawn alternatives
+
+"""
+    capture(cmd) -> Vector{UInt8}
+
+Run `cmd` and return its stdout, tolerating a non-zero exit the way
+`run_with_timeout` does. `read(cmd, String)` would throw instead, and a real
+corpus makes that a question of when, not whether: exiftool exits 1 on a file
+whose type it can't recognize, which in this pipeline is a routine outcome (it
+yields a degraded sidecar, not a failure). This is `run_with_timeout` minus the
+watchdog, so the gap between the two rows prices the watchdog exactly.
+"""
+function capture(cmd::Cmd)
+    out = IOBuffer()
+    proc = Base.run(pipeline(cmd; stdout = out, stderr = devnull); wait = false)
+    wait(proc)
+    return take!(out)
+end
+
+"""
+    batched_ns(paths, trials) -> ns_per_file
+
+Run the whole corpus through *one* `exiftool` process and divide by the file
+count. This is the floor for "what does exiftool cost if you stop paying the
+interpreter startup per file" — the fork, the Perl boot and the module loads are
+paid once for the batch instead of once per file.
+"""
+function batched_ns(paths::Vector{String}, trials::Int)
+    return best_of(noop; trials) do
+        SINK[] = capture(`exiftool -json -G -n $paths`)
+        length(paths)
+    end
+end
+
+"""
+    stay_open_ns(paths, trials) -> ns_per_file (or nothing if unsupported)
+
+Feed files one at a time to a single long-lived `exiftool -stay_open True -@ -`
+process over a pipe, reading its `{ready}` sentinel after each. Unlike the
+batched row this preserves the pipeline's actual shape — one file in, one result
+out, arriving whenever it arrives — so it prices the realistic fix rather than
+an unrealistic one.
+"""
+function stay_open_ns(paths::Vector{String}, trials::Int)
+    inp, outp = Pipe(), Pipe()
+    proc = Base.run(pipeline(`exiftool -stay_open True -@ -`;
+                             stdin = inp, stdout = outp, stderr = devnull); wait = false)
+    close(inp.out); close(outp.in)
+
+    ask(path) = begin
+        write(inp, "-json\n-G\n-n\n", path, "\n-execute\n")
+        flush(inp)
+        readuntil(outp, "{ready}")
+    end
+    try
+        ask(paths[1])               # pay the one-time process startup untimed
+        return best_of(noop; trials) do
+            for p in paths
+                SINK[] = ask(p)
+            end
+            length(paths)
+        end
+    finally
+        try
+            write(inp, "-stay_open\nFalse\n"); flush(inp); close(inp)
+            wait(proc)
+        catch
+            kill(proc, Base.SIGKILL)
+        end
+    end
+end
+
+# ------------------------------------------------------------------ components
+
+"""
+    component_rows(cfg, jobs, opts) -> Vector
+
+Time each piece of stage 2 on its own. The subprocess rows run once per corpus
+file (they cost milliseconds and don't need repetition); the in-memory and
+filesystem rows run `reps` times; the committing rows run once per corpus file
+with an untimed reset between passes.
+"""
+function component_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts)
+    reps, trials = opts["reps"], opts["trials"]
+    timeout = opts["timeout"]
+    nfiles = length(jobs)
+    paths = String[j.path for j in jobs]
+    rows = []
+    add!(name, part, ns) = push!(rows, (; name, part, ns))
+
+    # --- the bare interpreter: fork/exec + Perl boot, reading no file at all.
+    # Everything the real call does beyond this is actual work.
+    add!("exiftool -ver (spawn)", "extract", best_of(noop; trials) do
+        for _ in 1:nfiles
+            SINK[] = capture(`exiftool -ver`)
+        end
+        nfiles
+    end)
+
+    # --- the real command, run directly: spawn + parse the file, no watchdog.
+    add!("exiftool -json (raw run)", "extract", best_of(noop; trials) do
+        @inbounds for p in paths
+            SINK[] = capture(`exiftool -json -G -n $p`)
+        end
+        nfiles
+    end)
+
+    # --- the same command through the watchdog wrapper the stage actually uses.
+    # The gap to the row above is what the timeout costs.
+    add!("run_with_timeout", "extract", best_of(noop; trials) do
+        @inbounds for p in paths
+            SINK[] = FS.run_with_timeout(`exiftool -json -G -n $p`, timeout)
+        end
+        nfiles
+    end)
+
+    # --- run_exiftool: the wrapper plus JSON3.read plus the group-stripped Dict.
+    add!("run_exiftool (total)", "extract", best_of(noop; trials) do
+        @inbounds for p in paths
+            SINK[] = FS.run_exiftool(p, timeout)
+        end
+        nfiles
+    end)
+
+    # --- what a fork-free exiftool would cost, two ways (see the docstrings).
+    add!("exiftool (batched $(nfiles)x)", "alt", batched_ns(paths, trials))
+    if !opts["no-stay-open"]
+        try
+            add!("exiftool (-stay_open)", "alt", stay_open_ns(paths, trials))
+        catch e
+            @warn "persistent-exiftool probe failed; skipping" exception = e
+        end
+    end
+
+    # --- parsing alone, from bytes already captured: isolates JSON3 from the fork.
+    raw = [FS.run_with_timeout(`exiftool -json -G -n $p`, timeout) for p in paths]
+    valid = [b for b in raw if b !== nothing]
+    if !isempty(valid)
+        add!("JSON3.read (parse dump)", "extract", best_of(noop; trials) do
+            @inbounds for i in 1:reps
+                SINK[] = JSON3.read(String(copy(valid[(i - 1) % length(valid) + 1])))
+            end
+            reps
+        end)
+    end
+
+    # --- normalize_metadata: the ~20 tag coalesces, on a tag map already in memory.
+    bytags = [FS.run_exiftool(p, timeout) for p in paths]
+    good = [(j, b) for (j, b) in zip(jobs, bytags) if b !== nothing]
+    isempty(good) && error("exiftool produced no parseable output for any corpus file")
+    add!("normalize_metadata", "extract", best_of(noop; trials) do
+        @inbounds for i in 1:reps
+            j, b = good[(i - 1) % length(good) + 1]
+            SINK[] = FS.normalize_metadata(j, b)
+        end
+        reps
+    end)
+
+    # The sidecar payloads, built once, untimed: the commit rows below measure
+    # committing, not extracting.
+    metas = [FS.normalize_metadata(j, b) for (j, b) in good]
+
+    # --- serializing the sidecar (the raw dump makes this bigger than it looks).
+    add!("JSON3.write (sidecar)", "commit", best_of(noop; trials) do
+        @inbounds for i in 1:reps
+            SINK[] = JSON3.write(metas[(i - 1) % length(metas) + 1])
+        end
+        reps
+    end)
+
+    # --- sidecar bytes: open + write + flush + fsync, to a temp name.
+    # Non-consuming: same path rewritten each rep, as commit_enriched! does.
+    tmp = joinpath(cfg.done_dir, "bench_stage2_sidecar.tmp")
+    blobs = [JSON3.write(m) for m in metas]
+    # One pass over the real sidecar population, not `reps` of them. Two reasons,
+    # and the first is a correctness trap: thousands of back-to-back fsyncs
+    # saturate the device's write cache and each one starts waiting on the
+    # queue, which reported this row at 16 ms/file — eight times the whole
+    # `commit_enriched!` that contains it. The real stage fsyncs once per file
+    # with ~160 ms of exiftool between, and never queues that way. Second, real
+    # sidecars vary hugely in size (a zip's raw dump dwarfs a jpeg's), so the
+    # honest per-file number is one pass over all of them, not a cycle.
+    nio = length(blobs)
+    add!("write + fsync (sidecar)", "commit", best_of(noop; trials) do
+        @inbounds for i in 1:nio
+            open(tmp, "w") do io
+                write(io, blobs[(i - 1) % length(blobs) + 1])
+                flush(io)
+                FS.fsync_fd(fd(io))
+            end
+        end
+        nio
+    end)
+    rm(tmp; force = true)
+
+    # --- fsync_dir: persisting the rename itself, once per file in the real path.
+    add!("fsync_dir (done/)", "commit", best_of(noop; trials) do
+        for _ in 1:nio
+            SINK[] = FS.fsync_dir(cfg.done_dir)
+        end
+        nio
+    end)
+
+    # --- move_to: the rename known/<f> -> done/<f>. Consuming: reset each pass.
+    add!("move_to (rename)", "commit", best_of(() -> reknown!(cfg, jobs); trials) do
+        @inbounds for job in jobs
+            SINK[] = FS.move_to(cfg.done_dir, job)
+        end
+        nfiles
+    end)
+
+    # --- commit_enriched!: the whole sidecar-first commit, extraction excluded.
+    committable = [j for (j, _) in good]
+    add!("commit_enriched! (total)", "commit",
+        best_of(() -> reknown!(cfg, jobs); trials) do
+            @inbounds for (k, job) in enumerate(committable)
+                SINK[] = FS.commit_enriched!(cfg.done_dir, job, metas[k])
+            end
+            length(committable)
+        end)
+    reknown!(cfg, jobs)
+
+    # --- the one @info line, under each logger.
+    job1, meta1 = good[1][1], metas[1]
+    logfile = joinpath(dirname(cfg.known_dir), "bench_stage2.log")
+    for (which, label) in ((:null, "logging (NullLogger)"),
+                           (:format, "logging (format only)"),
+                           (:flush, "logging (flush→file)"))
+        ns = with_logger_named(which, logfile) do
+            best_of(noop; trials) do
+                for _ in 1:reps
+                    @info "enriched" worker=1 id=job1.id dest=job1.path sidecar="x.meta.json" file_type=meta1.file_type created_by=meta1.created_by degraded=(meta1.error !== nothing)
+                end
+                reps
+            end
+        end
+        add!(label, "log", ns)
+    end
+    rm(logfile; force = true)
+
+    return rows
+end
+
+"""
+    handler_rows(cfg, jobs, opts) -> Vector
+
+Time the real `handle_known_job` end to end under each logger. The difference
+between the rows is the cost logging adds to a file; the `:flush` row is what the
+running server actually pays.
+"""
+function handler_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts)
+    trials = opts["trials"]
+    nfiles = length(jobs)
+    logfile = joinpath(dirname(cfg.known_dir), "bench_stage2.log")
+    rows = []
+    for (which, label) in ((:null, "handle_known_job (NullLogger)"),
+                           (:format, "handle_known_job (format only)"),
+                           (:flush, "handle_known_job (flush→file)"))
+        ns = with_logger_named(which, logfile) do
+            best_of(() -> reknown!(cfg, jobs); trials) do
+                @inbounds for job in jobs
+                    FS.handle_known_job(job, cfg, 1)
+                end
+                nfiles
+            end
+        end
+        push!(rows, (; name = label, part = "total", ns))
+    end
+    reknown!(cfg, jobs)
+    rm(logfile; force = true)
+    return rows
+end
+
+"""
+    thread_rows(cfg, jobs, opts) -> Vector
+
+Run the full handler across worker counts, under the server's real logger. Stage
+2 spends most of its time in a child process, so this is the sweep that matters
+most: whether the stage scales is a question about the kernel's fork path and the
+machine's cores, not about Julia.
+
+Workers pull from a shared atomic counter rather than taking a contiguous slice.
+That matches the server (its pool pulls from one queue), and it matters here in a
+way it doesn't for stage 1: per-file exiftool time spans two orders of magnitude
+on a real corpus — a single 2 s archive among 48 files — so a static split leaves
+whichever worker drew it running alone while the rest idle, and the sweep would
+report a scaling ceiling that is really just load imbalance.
+"""
+function thread_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts)
+    trials = opts["trials"]
+    nfiles = length(jobs)
+    counts = opts["threads"] === nothing ?
+        unique([1; 2; 4; 8; Threads.nthreads()]) :
+        [parse(Int, s) for s in split(String(opts["threads"]), ",")]
+    counts = sort(unique(filter(k -> 1 <= k <= Threads.nthreads(), counts)))
+
+    logfile = joinpath(dirname(cfg.known_dir), "bench_stage2.log")
+    rows = []
+    base = 0.0
+    for k in counts
+        ns = with_logger_named(:flush, logfile) do
+            next = Threads.Atomic{Int}(1)
+            best_of(() -> (reknown!(cfg, jobs); next[] = 1); trials) do
+                # Shared counter, not a contiguous slice: every worker takes the
+                # next unclaimed file the moment it frees up, exactly as the
+                # server's pool takes the next job off the known queue.
+                @sync for t in 1:k
+                    Threads.@spawn begin
+                        while true
+                            i = Threads.atomic_add!(next, 1)
+                            i > nfiles && break
+                            @inbounds FS.handle_known_job(jobs[i], cfg, t)
+                        end
+                    end
+                end
+                nfiles
+            end
+        end
+        r = rate(ns)   # files/sec aggregate (ns is already per file, wall-clock)
+        k == counts[1] && (base = r)
+        push!(rows, (; workers = k, ns, files_per_sec = r, speedup = r / base))
+    end
+    reknown!(cfg, jobs)
+    rm(logfile; force = true)
+    return rows
+end
+
+# ------------------------------------------------------------------- reporting
+
+function print_components(rows, total_ns)
+    @printf("%-34s %-9s %12s %10s %9s\n", "component", "part", "per file", "rate", "% total")
+    rule()
+    for r in rows
+        @printf("%-34s %-9s %12s %10s %8.1f%%\n", r.name, r.part, human_time(r.ns),
+                human_rate(rate(r.ns)), 100 * r.ns / total_ns)
+    end
+end
+
+function print_threads(rows)
+    @printf("%-9s %12s %12s %9s\n", "workers", "per file", "throughput", "speedup")
+    rule()
+    for r in rows
+        @printf("%-9d %12s %12s %8.2fx\n", r.workers, human_time(r.ns),
+                human_rate(r.files_per_sec), r.speedup)
+    end
+end
+
+# ------------------------------------------------------------------------- main
+
+function main(argv)
+    opts = parse_args(argv)
+    try
+        FS.assert_exiftool()
+    catch e
+        println(stderr, sprint(showerror, e)); return 1
+    end
+
+    root = opts["dir"] === nothing ?
+        mktempdir(pwd(); prefix = "bench_stage2_") : String(opts["dir"])
+    owned = opts["dir"] === nothing
+    cfg = FS.Config(
+        spool_dir        = joinpath(root, "spool"),
+        known_dir        = joinpath(root, "known"),
+        done_dir         = joinpath(root, "done"),
+        failed_dir       = joinpath(root, "failed"),
+        exiftool_timeout = opts["timeout"],
+    )
+    for d in (cfg.spool_dir, cfg.known_dir, cfg.done_dir, cfg.failed_dir)
+        mkpath(d)
+    end
+
+    jobs = try
+        make_corpus(cfg, String(opts["corpus"]), opts["files"])
+    catch e
+        owned && rm(root; recursive = true, force = true)
+        println(stderr, sprint(showerror, e)); return 1
+    end
+    bytes = sum(j.size for j in jobs)
+
+    println("stage-2 component benchmark")
+    rule()
+    @printf("%-22s %s\n", "julia threads",    Threads.nthreads())
+    @printf("%-22s %s\n", "exiftool",         strip(read(`exiftool -ver`, String)))
+    @printf("%-22s %s\n", "corpus",           "$(length(jobs)) files ($(round(bytes / 1024^2; digits=1)) MiB) from $(opts["corpus"])")
+    @printf("%-22s %s\n", "scratch",          root)
+    @printf("%-22s %s\n", "reps / trials",    "$(opts["reps"]) / $(opts["trials"])")
+    @printf("%-22s %s\n", "exiftool timeout", "$(opts["timeout"]) s")
+
+    try
+        comps    = component_rows(cfg, jobs, opts)
+        handlers = handler_rows(cfg, jobs, opts)
+
+        # The denominator is the handler as the server actually runs it: the real
+        # flushing logger, one worker. Percentages are shares of that, so they are
+        # directly comparable and the parts can be checked against the whole.
+        total = only(r.ns for r in handlers if r.name == "handle_known_job (flush→file)")
+
+        header("Components (single worker)")
+        print_components(comps, total)
+
+        header("Whole handler, by logger")
+        print_components(handlers, total)
+
+        pick(name) = only(r.ns for r in comps if r.name == name)
+        accounted = pick("run_exiftool (total)") + pick("commit_enriched! (total)") +
+                    pick("logging (flush→file)")
+        println()
+        @printf("accounted: %s of %s (%.0f%%); unaccounted overhead %s\n",
+                human_time(accounted), human_time(total), 100 * accounted / total,
+                human_time(max(total - accounted, 0)))
+
+        threads = nothing
+        if !opts["no-threads"] && Threads.nthreads() > 1
+            threads = thread_rows(cfg, jobs, opts)
+            header("Full handler across workers (server logger)")
+            print_threads(threads)
+        end
+
+        if opts["json"] !== nothing
+            open(String(opts["json"]), "w") do io
+                JSON3.write(io, (;
+                    julia_threads = Threads.nthreads(),
+                    files = length(jobs), corpus_bytes = bytes,
+                    reps = opts["reps"], trials = opts["trials"],
+                    exiftool_timeout = opts["timeout"],
+                    components = comps, handlers, threads,
+                ))
+            end
+            println("\nwrote ", opts["json"])
+        end
+    finally
+        owned && rm(root; recursive = true, force = true)
+    end
+    return 0
+end
+
+exit(main(ARGS))

src/metadata.jl

@@ -67,6 +67,19 @@ function coalesce_tag(bytag::Dict{String,Any}, tags)
     return nothing
 end
 
+"How long a child gets to honor SIGTERM before `run_with_timeout` escalates to SIGKILL."
+const KILL_GRACE_SECONDS = 2.0
+
+"""
+Send `signum` to the whole process group `pgid` (a negative pid means "the group"
+to `kill(2)`). Julia's `kill(::Process, sig)` signals only the child itself,
+which is not enough to enforce a timeout — see `run_with_timeout`.
+"""
+function signal_group(pgid::Integer, signum::Integer)
+    ccall(:kill, Cint, (Cint, Cint), -pgid, signum)
+    return nothing
+end
+
 """
     run_with_timeout(cmd, timeout) -> Union{Vector{UInt8},Nothing}
 
@@ -75,33 +88,56 @@ Run `cmd`, capturing stdout, and return the captured bytes on clean exit, or
 SIGKILL after a grace period) once it overruns `timeout` seconds, so one
 pathological input can't wedge a worker forever. Shared by the exiftool (stage 2)
 and github-linguist (stage 4) shells.
+
+The child runs in its own process group and the timeout signals the *group*, not
+just the child. This is what makes the timeout enforceable: `wait` below returns
+only once the captured stdout pipe closes, and any grandchild inherits that pipe,
+so signalling the child alone leaves a `sh -c "...; sleep 30"`-shaped process
+tree running to completion with the worker still blocked on it. The cost of the
+process group is that a hard crash of the server orphans an in-flight child
+rather than taking it down; these children are short-lived and timeout-bounded,
+which is the cheaper side of that trade.
 """
 function run_with_timeout(cmd::Cmd, timeout::Integer)
     out = IOBuffer()
-    proc = Base.run(pipeline(cmd; stdout=out, stderr=devnull); wait=false)
+    # `detach` puts the child in a fresh process group (it becomes the group
+    # leader, so the group id is its pid); see the docstring for why the group,
+    # and not the child, is what the timeout has to signal.
+    proc = Base.run(pipeline(detach(cmd); stdout=out, stderr=devnull); wait=false)
+    pgid = Base.getpid(proc)
 
-    # Kill the process if it overruns the timeout. `t` polls rather than blocking
-    # so we can `kill` a hung child; the poll interval bounds shutdown latency.
+    # A one-shot timer, cancelled the moment the child exits, rather than a
+    # polling loop the caller has to join. The polling version charged every
+    # call the remainder of its in-flight `sleep(0.1)` *after* the child had
+    # already exited — ~50 ms on average, and a measured 101 ms on a process
+    # that exits instantly. That is pure latency on the hot path of two stages
+    # (exiftool here, github-linguist in stage 4), and it dwarfed the work on
+    # anything but a slow file. Waiting on the process directly costs nothing
+    # when the child exits normally, which is the overwhelmingly common case.
     killed = Ref(false)
-    t = Threads.@spawn begin
-        waited = 0.0
-        while process_running(proc) && waited < timeout
-            sleep(0.1); waited += 0.1
-        end
-        if process_running(proc)
-            killed[] = true
-            kill(proc, Base.SIGTERM)
-            # Escalate: a process that ignores/defers SIGTERM would otherwise pin
-            # the worker forever on the wait(proc) below, defeating the timeout.
-            grace = 0.0
-            while process_running(proc) && grace < 2.0
-                sleep(0.1); grace += 0.1
+    timer = Timer(timeout) do _
+        process_running(proc) || return
+        killed[] = true
+        signal_group(pgid, Base.SIGTERM)
+        # Escalate: a process that ignores/defers SIGTERM would otherwise pin the
+        # worker forever on the wait(proc) below, defeating the timeout. This
+        # runs off the timer's task so the event loop isn't held during the
+        # grace period, and it is not joined — by the time it wakes, `wait(proc)`
+        # has long since returned and `process_running` settles it.
+        Threads.@spawn begin
+            deadline = time() + KILL_GRACE_SECONDS
+            while process_running(proc) && time() < deadline
+                sleep(0.05)
             end
-            process_running(proc) && kill(proc, Base.SIGKILL)
+            process_running(proc) && signal_group(pgid, Base.SIGKILL)
         end
     end
-    wait(proc)
-    wait(t)
+
+    try
+        wait(proc)
+    finally
+        close(timer)      # cancel the pending kill; a no-op if it already fired
+    end
 
     (killed[] || !success(proc)) && return nothing
     return take!(out)

src/worker.jl

@@ -18,6 +18,21 @@
 # this worker-to-worker handoff blocks.
 const ROUTE_ENQUEUE_RETRY_SECONDS = 0.05
 
+# Per-file logging in stage 1 is `@debug`, not `@info`, because it is the
+# stage's dominant cost. Measured by bin/bench_stage1.jl (2000 x 64 KiB files,
+# min of 5 trials): the two log lines cost ~71 µs of the ~118 µs
+# `handle_classify_job` spent per file — roughly 6x the classifier (10.6 µs) and
+# 6x the rename (11.6 µs). Nearly all of it is `ConsoleLogger` formatting
+# (~64 µs); the FlushLogger's per-message flush is only ~8 µs on top. Demoting
+# them takes stage 1 from ~8.5k files/s to ~35k files/s on one worker.
+#
+# `@debug` is compiled to a min-level check that doesn't evaluate its arguments,
+# so a disabled line costs ~0.15 µs rather than ~36 µs. The messages are still
+# there when wanted: run with `JULIA_DEBUG=FileServer` to get them back. Errors,
+# quarantines and lifecycle events stay at `@error`/`@info` — they are rare and
+# their cost doesn't scale with throughput. `GET /stats` (src/stats.jl) is the
+# per-file observability that survives, and it is counted, not formatted.
+
 """
     handle_classify_job(job, cfg, worker_id, known_queue, unknown_queue)
 
@@ -35,7 +50,7 @@ function handle_classify_job(job::Job, cfg::Config, worker_id::Int,
                              known_queue::JobQueue, unknown_queue::JobQueue,
                              stats::StageStats)
     classification = classify(CLASSIFIER[], job.path)
-    @info "classified file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification
+    @debug "classified file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification
 
     if classification === :known
         dest = move_to(cfg.known_dir, job)
@@ -43,13 +58,13 @@ function handle_classify_job(job::Job, cfg::Config, worker_id::Int,
         # known queue full → park and retry, don't drop (time charged to blocked_ns)
         enqueue_blocking!(known_queue, routed, stats;
                           retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS)
-        @info "routed to enrichment" worker=worker_id id=job.id dest=dest
+        @debug "routed to enrichment" worker=worker_id id=job.id dest=dest
     else
         dest = move_to(cfg.unknown_dir, job)
         routed = Job(job.id, job.original_name, dest, job.size, job.received_at)
         enqueue_blocking!(unknown_queue, routed, stats;
                           retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS)
-        @info "routed to content triage" worker=worker_id id=job.id dest=dest
+        @debug "routed to content triage" worker=worker_id id=job.id dest=dest
     end
     return nothing
 end

test/runtests.jl

@@ -289,6 +289,42 @@ end
         end
     end
 
+    @testset "run_with_timeout: returns as soon as the child exits" begin
+        # Regression guard. The original implementation polled with sleep(0.1)
+        # and joined the polling task, so every call paid the remainder of an
+        # in-flight sleep after the child had already exited — ~101 ms on a
+        # process that exits instantly, on the hot path of stages 2 and 4. The
+        # bound here is deliberately loose (a loaded CI box is slow) but far
+        # under the 100 ms floor the polling version could not beat.
+        FileServer.run_with_timeout(`true`, 30)                       # warm up / compile
+        t0 = time()
+        out = FileServer.run_with_timeout(`echo hi`, 30)
+        elapsed = time() - t0
+        @test out !== nothing
+        @test strip(String(out)) == "hi"
+        @test elapsed < 0.05
+    end
+
+    @testset "run_with_timeout: kills an overrunning child and reports failure" begin
+        t0 = time()
+        out = FileServer.run_with_timeout(`sleep 30`, 1)
+        elapsed = time() - t0
+        @test out === nothing            # timed out → no output, caller degrades
+        @test elapsed < 5                # killed near the timeout, not after 30 s
+    end
+
+    @testset "run_with_timeout: escalates to SIGKILL when SIGTERM is ignored" begin
+        # A child that traps SIGTERM. Without the escalation the worker would
+        # block on wait(proc) forever and the timeout would be unenforceable.
+        cmd = `sh -c "trap '' TERM; sleep 30"`
+        t0 = time()
+        out = FileServer.run_with_timeout(cmd, 1)
+        elapsed = time() - t0
+        @test out === nothing
+        # 1 s timeout + up to KILL_GRACE_SECONDS before SIGKILL lands.
+        @test elapsed < 1 + FileServer.KILL_GRACE_SECONDS + 3
+    end
+
     @testset "run_exiftool: real extraction on a PNG" begin
         mktempdir() do root
             p = joinpath(root, "pixel.png")