file-server/bin/bench_stage2.jl

#!/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 spool/<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 spool/ 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 `spool/` 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)
        # Staged in spool/, not a known/ dir: stage 1 routes by enqueueing and
        # leaves the bytes where intake put them, so this is the exact on-disk
        # state a stage-2 worker dequeues into (src/worker.jl header).
        push!(jobs, FS.Job(id, basename(name), spooled, filesize(spooled), time()))
    end
    return jobs
end

"""
    respool!(cfg, jobs)

Put every corpus file back in `spool/`, 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 — stage 2 still
moves, because its move is the terminal commit, not an inter-stage hop.
"""
function respool!(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 spool/<f> -> done/<f>. Consuming: reset each pass.
    add!("move_to (rename)", "commit", best_of(() -> respool!(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(() -> respool!(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)
    respool!(cfg, jobs)

    # --- the one @info line, under each logger.
    job1, meta1 = good[1][1], metas[1]
    logfile = joinpath(dirname(cfg.spool_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.spool_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(() -> respool!(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
    respool!(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.spool_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(() -> (respool!(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
    respool!(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"),
        done_dir         = joinpath(root, "done"),
        failed_dir       = joinpath(root, "failed"),
        exiftool_timeout = opts["timeout"],
    )
    for d in (cfg.spool_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))