file-server/src/worker.jl

# Worker tasks: pull jobs off a queue and process them. The loop scaffolding
# (dequeue-until-drained, try/catch, quarantine-on-throw) is identical for every
# stage, so `worker_loop` is parametrized with a `handler` and reused. Today
# there are four stages:
#
#   stage 1  handle_classify_job  classify → known queue | unknown queue
#   stage 2  handle_known_job     exiftool enrich → done/ (+ .meta.json)
#   stage 3  handle_unknown_job   binary-vs-text sniff → binary/ | text queue
#   stage 4  handle_text_job      language enrich → text_done/ (+ .meta.json)
#
# Adding a stage later is just another queue + pool + handler; the loop below
# doesn't change.
#
# A file does not move between stages. It is written once into `spool/` at
# intake and stays there for its whole in-flight life; only the small `Job`
# reference travels, and `job.path` is therefore constant from intake until the
# file is committed. The stage a file has reached lives in the queue holding its
# reference, not in which directory the bytes sit — so the intermediate hops
# (spool→known, spool→unknown, unknown→text) are three renames per file that buy
# nothing on the live path. What remains is one move at the end: into a terminal
# sink (`done/`, `text_done/`, `binary/`) or into `failed/` on a throw.
#
# The cost is on restart. The queues are in-process (src/queue.jl), so a crash
# loses them, and recovery can only re-drive everything in `spool/` from stage 1.
# That is safe — classification and the UTF-8 sniff are pure functions of the
# file's bytes and every commit is idempotent — but it redoes work the old
# directory-per-stage layout could skip. The durable fix is the queue seam, not
# the directories: a broker-backed `JobQueue` restores exact resume for free.
#
# (`ROUTE_ENQUEUE_RETRY_SECONDS`, the backoff the routing handoffs below use,
# now lives in src/queue.jl, since startup recovery shares it.)

# 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): a formatted pair of log lines costs 71.2 µs per file, against
# the 11.7 µs the whole handler takes with them switched off — six times the rest
# of the stage put together. Nearly all of it is `ConsoleLogger` formatting
# (64.2 µs); the FlushLogger's per-message flush is only ~7 µs on top. Switching
# them back on with `JULIA_DEBUG=FileServer` takes the handler to 106.3 µs, i.e.
# from 85.4k files/s down to 9.4k on one worker.
#
# With logging off the stage is the classifier and nothing else: classify 10.30 µs
# (of which read_features is 7.82 µs), the queue handoff 0.12 µs, the disabled
# `@debug` lines 0.29 µs — 10.7 µs of the 11.7 µs total. Removing the inter-stage
# rename is what left it that way: that rename was 11.6 µs per file, so it was
# the equal of the classifier, and dropping it took one worker from ~35k to
# ~85k files/s (16 workers: ~333k files/s).
#
# `@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)

Stage 1. Classify the spooled file and route it to the next stage's queue,
retrying on a full queue rather than dropping the file:
  * `:known`   → enqueue onto the known queue for stage 2.
  * `:unknown` → enqueue onto the unknown queue for stage 3.

Routing is the enqueue and nothing else: the file stays where intake wrote it and
the *same* `Job` is handed on, so `job.path` still points at it. Reaching stage 2
is a fact about which queue holds the reference, not about which directory holds
the bytes.

Sub-`MIN_FILE_BYTES` files short-circuit to `:unknown` inside `classify`.
"""
function handle_classify_job(job::Job, cfg::Config, worker_id::Int,
                             known_queue::JobQueue, unknown_queue::JobQueue,
                             stats::StageStats)
    classification = classify(CLASSIFIER[], job.path)
    @debug "classified file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification

    if classification === :known
        # known queue full → park and retry, don't drop (time charged to blocked_ns)
        enqueue_blocking!(known_queue, job, stats;
                          retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS)
        @debug "routed to enrichment" worker=worker_id id=job.id path=job.path
    else
        enqueue_blocking!(unknown_queue, job, stats;
                          retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS)
        @debug "routed to content triage" worker=worker_id id=job.id path=job.path
    end
    return nothing
end

"""
    handle_known_job(job, cfg, worker_id)

Stage 2. Extract metadata (exiftool, with timeout) and enrich: build the
normalized sidecar and commit both to `done/` sidecar-first — the file's one and
only move, straight out of `spool/`. Extraction
failure/timeout yields a *degraded* sidecar (the file is still a wanted known
file), so the only way to land in `failed/` is a genuine I/O error writing the
sidecar or moving the file — handled by `worker_loop`'s quarantine.
"""
function handle_known_job(job::Job, cfg::Config, worker_id::Int)
    meta = build_metadata(job, cfg)
    file_dest, sidecar = finalize_known!(cfg, job, meta)
    @info "enriched" worker=worker_id id=job.id dest=file_dest sidecar=basename(sidecar) file_type=meta.file_type created_by=meta.created_by degraded=(meta.error !== nothing)
    return nothing
end

"""
    handle_unknown_job(job, cfg, worker_id, text_queue)

Stage 3. Sort an unrecognized file into a coarse content bucket by sniffing its
first bytes.

The two outcomes are asymmetric, because one is terminal and one is not. Binary
is the end of the live path, so the file is committed to `binary/` — which is
also where the offline stage-5 discovery sweep reads its corpus, so the move is
load-bearing, not bookkeeping. Text has a stage 4 still to come, so nothing moves:
the same `Job` goes onto the language-enrichment queue, retrying on a full queue
rather than dropping the file (the blocking backpressure stage 1 also uses).
"""
function handle_unknown_job(job::Job, cfg::Config, worker_id::Int,
                            text_queue::JobQueue, stats::StageStats)
    if is_binary(job.path)
        dest = move_to(cfg.binary_dir, job)
        @info "sorted unknown" worker=worker_id id=job.id name=job.original_name kind=:binary dest=dest
    else
        # text queue full → park and retry, don't drop (time charged to blocked_ns)
        enqueue_blocking!(text_queue, job, stats;
                          retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS)
        @info "routed to language enrichment" worker=worker_id id=job.id path=job.path
    end
    return nothing
end

"""
    handle_text_job(job, cfg, worker_id, detector)

Stage 4. Enrich a text file with its natural language (via `detector`) and
programming language (via github-linguist): build the sidecar and commit both to
`text_done/` sidecar-first — the file's one and only move, straight out of
`spool/`. Detection failure yields a *degraded* sidecar (the
file is still wanted text), so the only way to land in `failed/` is a genuine I/O
error committing — handled by `worker_loop`'s quarantine.
"""
function handle_text_job(job::Job, cfg::Config, worker_id::Int, detector)
    meta = build_text_metadata(detector, job, cfg)
    file_dest, sidecar = finalize_text!(cfg, job, meta)
    @info "enriched text" worker=worker_id id=job.id dest=file_dest sidecar=basename(sidecar) language=meta.language confidence=meta.language_confidence programming_language=meta.programming_language degraded=(meta.error !== nothing)
    return nothing
end

"""
    worker_loop(worker_id, cfg, queue, handler, stats)

Consume jobs from `queue` until it is closed and drained, running `handler` on
each. A failure on one job is logged and the file quarantined in `failed/` — it
must never kill the worker, or the pool would silently shrink.

This loop is also where per-stage metrics are recorded (`stats`, see
src/stats.jl). Instrumenting here rather than in each handler means every stage
is measured the same way, by construction, and a new stage is measured the
moment it is wired up — there is no per-handler bookkeeping to forget.

The timed region is the handler alone, excluding the `dequeue!` above it: time
parked waiting for work is idleness, and counting it as service time would make
an idle stage look as busy as a saturated one. A quarantined job still counts
its time — the work was done, it just ended in `failed/`.
"""
function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue, handler,
                     stats::StageStats)
    @info "worker started" worker=worker_id
    while true
        job = dequeue!(queue)
        job === nothing && break   # queue closed and drained → exit
        Threads.atomic_add!(stats.in_flight, 1)
        t0 = time_ns()
        ok = true
        try
            handler(job, cfg, worker_id)
        catch e
            ok = false
            @error "processing failed" worker=worker_id id=job.id name=job.original_name exception=(e, catch_backtrace())
            try
                move_to(cfg.failed_dir, job)
            catch e2
                @error "could not quarantine failed file" worker=worker_id id=job.id path=job.path exception=(e2, catch_backtrace())
            end
        finally
            # In a `finally` so an InterruptException during shutdown can't leave
            # in_flight permanently above zero, which would read as a stuck job.
            record_job!(stats, ok, job.size, Int(time_ns() - t0))
            Threads.atomic_sub!(stats.in_flight, 1)
        end
    end
    @info "worker stopped" worker=worker_id
    return nothing
end