Demote stage-1 per-file logging to @debug; add stage-1 decomposition benchmark

bin/bench_stage1.jl takes handle_classify_job apart — filesize, read_features,
Lux.apply, classify, move_to, enqueue_blocking!, and the log lines — times each
in isolation, then times the real handler end to end under four loggers so the
parts can be checked against the whole.

It found that logging was stage 1's dominant cost: as @info the two per-file
lines cost ~71 us of the handler's ~118 us, roughly 6x the classifier (10.6 us)
and 6x the rename (11.6 us). Nearly all of it is ConsoleLogger formatting
(~64 us), not the FlushLogger's per-message flush (~8 us).

Demoting them to @debug takes stage 1 from 8.5k files/s to 35.3k files/s on one
worker (4.2x). The messages are still available with JULIA_DEBUG=FileServer,
which the benchmark also prices (133 us/file). What remains splits evenly
between the rename (11.7 us) and classify (10.7 us, itself 74% feature read),
so stage 1 is now filesystem-bound; its thread sweep peaks at ~4 workers.

README.md

@@ -510,11 +510,12 @@ 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 four 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_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 +527,9 @@ 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
+
 # 2. the pipeline. Start the server in one terminal…
 julia --project=. -t auto bin/server.jl
 
@@ -631,12 +635,70 @@ 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`.
+
 ### 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
+`handle_classify_job`, which is a feature read, an inference, a rename, a
-line, and whatever contention the other three pools create. That's the right
+(disabled) debug line, and whatever contention the other three pools create.
-number for capacity planning and the wrong one for "is the model slow?".
+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 +716,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 +769,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))

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