diff --git a/README.md b/README.md index 98f191a..917a6ef 100644 --- a/README.md +++ b/README.md @@ -22,12 +22,13 @@ enrichment mixes CPU with a subprocess): ▼ ┌─────────────────┐ stream bytes to disk (never buffered) │ HTTP handler │────────────────────────► data/spool/- - │ (streaming) │ - └────────┬─────────┘ enqueue reference (non-blocking) - │ │ - ▼ ▼ - 202 + job IDs ┌────────────────────┐ - (503 if full) │ stage-1 queue │ classification + │ (streaming) │ the file's home for its + └────────┬─────────┘ enqueue reference whole time in flight + │ (non-blocking) + ▼ │ + 202 + job IDs ▼ + (503 if full) ┌────────────────────┐ + │ stage-1 queue │ classification └─────────┬──────────┘ │ dequeue ┌───────────────────┼───────────────────┐ @@ -35,9 +36,10 @@ enrichment mixes CPU with a subprocess): classify wkr 1 classify wkr 2 … classify wkr N │ ┌────────────┴────────────┐ - :unknown :known - │ move to data/unknown/, │ move to data/known/, then - ▼ then enqueue (blocking) ▼ enqueue (blocking backpressure) + :unknown :known the file itself never moves — + │ │ routing is the enqueue alone + │ enqueue (blocking) │ enqueue (blocking backpressure) + ▼ ▼ ┌────────────────────┐ ┌────────────────────┐ │ unknown queue │ │ known queue │ enrichment └─────────┬──────────┘ └─────────┬──────────┘ @@ -47,10 +49,10 @@ enrichment mixes CPU with a subprocess): unk 1 unk 2 … unk K known wkr 1 known wkr 2 … known wkr M │ binary-vs-text sniff │ exiftool → normalized sidecar ├─► data/binary/- success ──┴──► data/done/- - │ (terminal) data/done/-.meta.json - │ (sidecar-first commit) - │ :text move to data/text/, failure ───────► data/failed/- - ▼ then enqueue (blocking backpressure) + │ (terminal — moved) data/done/-.meta.json + │ (sidecar-first commit) + │ :text enqueue (blocking failure ───────► data/failed/- + ▼ backpressure) ┌────────────────────┐ │ text queue │ language enrichment └─────────┬──────────┘ @@ -63,6 +65,16 @@ enrichment mixes CPU with a subprocess): (sidecar-first commit) ``` +A file is written **once**, into `data/spool/`, and stays there for its entire +time in the pipeline. Stages hand it on by enqueueing its small `Job` reference, +never by moving bytes — the queue holding the reference *is* the record of which +stage the file has reached. The only move is the last one: into a terminal sink +(`data/done/`, `data/text_done/`, `data/binary/`) or into `data/failed/` if a +worker throws. Stage 1 used to rename each file into `data/known/` or +`data/unknown/` first, and stage 3 into `data/text/`; those three directories are +gone, and with them 11.6 µs per file — the equal of the classifier itself, which +is why stage 1 now runs at ~85k files/s on one worker instead of ~35k. + Stages 2 (known-file enrichment) and 3 (content triage) run in parallel: stage 1 feeds both the known and unknown queues. Stage 3 in turn feeds stage 4 (language enrichment) for every file it sorts as text. @@ -79,12 +91,24 @@ Key properties: - **Backpressure:** each queue is bounded (default 1000). When the *intake* queue is full, uploads get `503 Service Unavailable`. When the *known* queue is full, the stage-1 worker blocks and retries (a classified file is never dropped). -- **Crash-resilient:** files survive on disk. On startup, recovery is - stage-aware: leftovers in `data/spool/` re-enter classification, `data/known/` - re-enter enrichment, `data/unknown/` re-enter content triage, and `data/text/` - re-enter language enrichment (`recovered` / `recovered_known` / - `recovered_unknown` / `recovered_text` in the log), so a file resumes at its - correct stage instead of restarting from scratch. +- **Crash-resilient:** files survive on disk. On startup everything left in + `data/spool/` is re-enqueued (`recovered` in the log) and **replays from stage + 1**. That is safe rather than merely tolerable: classification and the + binary/text sniff are pure functions of the file's bytes, and the terminal + commits rename with `force=true`, so a replayed file lands where it would have + landed and overwrites its own sidecar. Recovery *blocks* on a full queue rather + than dropping the excess, and runs with the worker pools already live, so a + backlog larger than one queue's capacity takes longer to re-drive but none of + it is abandoned. + + The cost of replay is redoing stages a file had already cleared. That is a + property of the **queue**, not of the directory layout: the queues are + in-process (`src/queue.jl`), so a crash destroys the only record of how far + each file got. Per-stage directories used to stand in for that record, at the + price of a rename per file per stage on the hot path — paying a permanent cost + on every file to buy a cheaper restart. `src/queue.jl` already defines the seam + for the real fix: swap `ChannelQueue` for a broker-backed `JobQueue` and exact + resume comes back durably, rather than being inferred from a pathname. - **Graceful shutdown:** SIGINT (Ctrl-C) and SIGTERM (systemd/Docker/k8s `stop`) both stop accepting uploads, then drain the stages *in order* — close the stage-1 queue and wait out the classify workers (the only producer of the known @@ -182,21 +206,26 @@ idempotently. Files the classifier labels **unknown** are handed to a third pool that sorts them into two coarse buckets so downstream tooling can treat them differently: -- **`data/binary/`** — the file looks like binary data. -- **`data/text/`** — the file looks like text. +- **binary** — the file looks like binary data. This is the end of the live path, + so the file is committed to `data/binary/` (which is also the corpus the + offline stage-5 discovery sweep reads). +- **text** — the file looks like text. Stage 4 is still to come, so nothing + moves; the file stays in `data/spool/` and its reference goes onto the + stage-4 queue, ending up in `data/text_done/` once enriched. The test is a **UTF-8 sniff**: read the first 8000 bytes and call the file text when that window is valid UTF-8 and holds no control bytes outside the text-safe set (tab, newline, CR, and friends, plus ESC for ANSI-colored logs); otherwise binary. It's cheap (no full read) and Unicode-aware — unlike the older NUL-byte or printable-ASCII heuristics, it keeps non-ASCII text (accents, CJK, emoji) in -`text/` instead of misfiling it, while binary formats — which rarely form valid -UTF-8 near their start — still land in `binary/`. A NUL byte is valid UTF-8 but +the text bucket instead of misfiling it, while binary formats — which rarely +form valid UTF-8 near their start — still read as binary. A NUL byte is valid UTF-8 but not a text control byte, so it still reads as binary. A multi-byte character split by the 8000-byte boundary is trimmed before the check so it isn't mistaken -for malformed bytes. An empty file is treated as text. `binary/` is terminal on -the live path (but is the input the offline **stage-5 discovery** sweeps — see -below); `text/` is handed to stage 4 (`src/content.jl`). +for malformed bytes. An empty file is treated as text. Binary is terminal on the +live path and is committed to `data/binary/` (which the offline **stage-5 +discovery** sweep reads as its corpus — see below); text is handed to stage 4 +without moving (`src/content.jl`). ### Language enrichment (stage 4) @@ -367,10 +396,10 @@ tell you "PDF", just "this looks like something I was trained on, or not". fast rather than run without classification. - **Effect today:** *active routing*. The class is logged (`classification=known|unknown`) and drives the pipeline split: `:known` files - go to `known/` for metadata enrichment (stage 2), `:unknown` files go to - `unknown/` for content triage (stage 3). The class chooses the downstream - stage; what's still unproven is the model's *accuracy*, not whether the routing - path runs. + go onto the stage-2 queue for metadata enrichment, `:unknown` files onto the + stage-3 queue for content triage. The class chooses the downstream stage (the + file itself stays in `data/spool/` either way); what's still unproven is the + model's *accuracy*, not whether the routing path runs. The architecture and byte→feature mapping are defined once in `src/model.jl` and shared by the trainer and the server, so they can't drift apart. @@ -410,11 +439,8 @@ init, so the artifact is exactly regenerable from the same inputs. | `FS_UNKNOWN_QUEUE_CAPACITY` | `1000` | Max pending triage jobs (then backpressure) | | `FS_TEXT_WORKERS` | `nthreads()` | Stage-4 (language enrichment) worker tasks | | `FS_TEXT_QUEUE_CAPACITY` | `1000` | Max pending language jobs (then backpressure) | -| `FS_SPOOL_DIR` | `data/spool` | Incoming files (pending classification) | -| `FS_KNOWN_DIR` | `data/known` | Classified-known, awaiting enrichment | -| `FS_UNKNOWN_DIR` | `data/unknown` | Classified-unknown, awaiting content triage | +| `FS_SPOOL_DIR` | `data/spool` | Every in-flight file, at every stage | | `FS_BINARY_DIR` | `data/binary` | Stage-3 sink: unknown files that look binary | -| `FS_TEXT_DIR` | `data/text` | Classified-text, awaiting language enrichment | | `FS_DONE_DIR` | `data/done` | Enriched known files (+ `.meta.json`) | | `FS_TEXT_DONE_DIR` | `data/text_done` | Enriched text files (+ `.meta.json`) | | `FS_FAILED_DIR` | `data/failed` | Files whose processing threw | @@ -462,9 +488,9 @@ Each file in a request becomes its own job. Responses: ### `GET /stats` — per-stage counters The pipeline counts its own work (`src/stats.jl`), because nothing outside it -can: `known/`, `unknown/` and `text/` are *transient*, so a file can cross one -between two directory polls and an external sampler will miss exactly the stages -you most want to measure. +can. Every in-flight file sits in `data/spool/` no matter which stage it has +reached — the stage is a property of the queue holding its reference, and only +the pipeline can see that. There is no directory to poll. ```jsonc { @@ -560,9 +586,9 @@ Three properties of this design dictate how it measures: so `ab`/`hey`/`wrk` would only ever measure intake. The bench instead uploads a corpus and polls the terminal sinks (`done/`, `text_done/`, `binary/`, `failed/`) until the count stops moving, and reports both numbers separately: - intake rate *and* end-to-end completion rate. It also samples the intermediate - stage dirs, so the peak depth of `spool/`/`known/`/`unknown/`/`text/` shows - where work piles up. + intake rate *and* end-to-end completion rate. It also samples `spool/`, whose + peak depth is the high-water mark of files in flight — but *which* stage they + are waiting on comes from `/stats`, not from disk. - **End-to-end throughput doesn't name the slow stage.** The four stages run concurrently behind their own queues, so the pipeline's rate *is* the slowest stage's rate and the others are invisible in it. The bench scrapes diff --git a/bin/bench.jl b/bin/bench.jl index 4974d63..e0aa460 100755 --- a/bin/bench.jl +++ b/bin/bench.jl @@ -14,8 +14,9 @@ # 2. End-to-end throughput doesn't name the slow stage. The four stages run # concurrently behind their own queues, so the pipeline's rate is the # slowest stage's rate and the others are invisible. Directory polling can't -# recover them either — known/, unknown/ and text/ are transient, and a file -# can cross one between two samples. So the server keeps per-stage counters +# recover them either — every in-flight file sits in spool/ whatever stage it +# is at, since stages route by enqueueing rather than by moving bytes. So the +# server keeps per-stage counters # (src/stats.jl) and we scrape GET /stats before and after: the deltas give # each stage's throughput, mean service time, and worker utilization, and # utilization is what actually names the bottleneck (see `stage_report`). @@ -151,12 +152,12 @@ sinkdirs() = ( failed = get(ENV, "FS_FAILED_DIR", "data/failed"), ) -stagedirs() = ( - spool = get(ENV, "FS_SPOOL_DIR", "data/spool"), - known = get(ENV, "FS_KNOWN_DIR", "data/known"), - unknown = get(ENV, "FS_UNKNOWN_DIR", "data/unknown"), - text = get(ENV, "FS_TEXT_DIR", "data/text"), -) +# One directory, not four. Files no longer move between stages — spool/ holds +# every in-flight file at every stage, and which stage it has reached lives in +# the queue holding its reference (src/worker.jl header). So this depth is +# "files in flight", full stop; per-stage depth comes from /stats, which is the +# only place that can see it at all. +stagedirs() = (spool = get(ENV, "FS_SPOOL_DIR", "data/spool"),) "Count work items in `dir`, ignoring the .meta.json sidecars stages 2/4 write." function count_files(dir::AbstractString)::Int @@ -513,7 +514,7 @@ function main(argv) # Leftovers mid-pipeline would land in the sinks during our window and be # counted as our throughput, so say so up front rather than quietly skewing. pending = total(counts(stages)) - pending > 0 && @warn "pipeline is not idle: $pending file(s) in the stage dirs; " * + pending > 0 && @warn "pipeline is not idle: $pending file(s) still in spool/; " * "throughput will include their completions" # --- corpus @@ -570,7 +571,7 @@ function main(argv) "(the server predates src/stats.jl)" end - # --- sampler: RSS curve, stage dir depths, and queue depths. + # --- sampler: RSS curve, spool depth, and queue depths. stop = Threads.Atomic{Bool}(false) rss_samples = Float64[] depth_max = Dict(k => 0 for k in keys(stages)) @@ -731,7 +732,7 @@ function main(argv) println("\nnote: $(sink_delta.failed) file(s) landed in $(sinks.failed) — check the server log.") timed_out && println("\nnote: drain stalled with $(accepted - completed) file(s) outstanding. " * - "Check the server log and the stage dirs; raise --timeout if the pipeline is just slow.") + "Check the server log and spool/; raise --timeout if the pipeline is just slow.") if opts["json"] !== nothing result = ( diff --git a/bin/bench_stage1.jl b/bin/bench_stage1.jl index c3011e3..3f9b974 100644 --- a/bin/bench_stage1.jl +++ b/bin/bench_stage1.jl @@ -10,7 +10,6 @@ # 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/ -> known/ or unknown/ # enqueue push a Job reference onto the downstream bounded queue # logging two @info lines ("classified file", "routed to ...") # @@ -36,8 +35,8 @@ # 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) +# --files N files per timed pass for whole-corpus benchmarks (default: 2000) +# --reps N calls per timed pass for per-call 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/) @@ -113,8 +112,8 @@ const SINK = Ref{Any}(nothing) 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. +each pass and is *not* timed — that is where a benchmark resets whatever its +last pass consumed (today: the queues). `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. @@ -198,26 +197,11 @@ function make_corpus(cfg::FS.Config, n::Int, size::Int, rng) 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 +# There is no `respool!` step any more. Stage 1 used to consume its input by +# renaming each file into known/ or unknown/, so every repeated pass had to put +# the corpus back first. It now routes by enqueueing alone and leaves the file in +# spool/, so the corpus is reusable as-is and the only per-pass reset is draining +# the queues. "Drain a queue without blocking, so the next pass starts from empty." function drain!(q::FS.ChannelQueue) @@ -263,10 +247,10 @@ end """ component_rows(cfg, clf, jobs, opts) -> Vector -Time each piece of stage 1 on its own. Non-consuming pieces (`filesize`, +Time each piece of stage 1 on its own. Per-call 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. +the corpus; whole-corpus pieces (`enqueue_blocking!`, the full handler) run once +per corpus file, with an untimed queue drain between passes. """ function component_rows(cfg::FS.Config, clf::FS.Classifier, jobs::Vector{FS.Job}, opts) reps, trials = opts["reps"], opts["trials"] @@ -317,14 +301,11 @@ function component_rows(cfg::FS.Config, clf::FS.Classifier, jobs::Vector{FS.Job} 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)) + # Stage 1 used to rename each file into known/ or unknown/ before enqueueing + # it, and that rename was measured here as its own line item. It is gone: + # routing is the enqueue alone, and a file does not move until it is + # committed to a terminal sink (src/worker.jl header). So `route` below is + # now just the queue handoff. # --- enqueue: lock, push, notify on an uncontended, non-full queue q = FS.ChannelQueue(nfiles + 1) @@ -381,7 +362,7 @@ function handler_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) (: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 + best_of(() -> (drain!(known); drain!(unknown)); trials) do @inbounds for job in jobs FS.handle_classify_job(job, cfg, 1, known, unknown, stats) end @@ -390,7 +371,7 @@ function handler_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) end push!(rows, (; name = label, part = "total", ns)) end - respool!(cfg, jobs); drain!(known); drain!(unknown) + drain!(known); drain!(unknown) rm(logfile; force = true) return rows end @@ -419,7 +400,7 @@ function thread_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) 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 + best_of(() -> (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. @@ -441,7 +422,7 @@ function thread_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) 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) + drain!(known); drain!(unknown) rm(logfile; force = true) return rows end @@ -481,12 +462,10 @@ function main(argv) 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) + for d in (cfg.spool_dir, cfg.failed_dir) mkpath(d) end @@ -524,7 +503,7 @@ function main(argv) # 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)")) + ("classify (total)", "enqueue_blocking!", "logging (NullLogger)")) println() @printf("accounted: %s of %s (%.0f%%); unaccounted overhead %s\n", human_time(accounted), human_time(total), 100 * accounted / total, diff --git a/bin/bench_stage2.jl b/bin/bench_stage2.jl index 5a4022b..a7b57ae 100644 --- a/bin/bench_stage2.jl +++ b/bin/bench_stage2.jl @@ -17,7 +17,7 @@ # 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/ -> done/, the commit point +# move_to rename spool/ -> done/, the commit point # logging one @info line ("enriched") # # This script times each of those in isolation, then times the real @@ -28,7 +28,7 @@ # * 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. +# 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 @@ -173,7 +173,7 @@ end """ make_corpus(cfg, corpus_dir, n) -> Vector{Job} -Copy up to `n` real files from `corpus_dir` into `known/` and build the `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. @@ -198,21 +198,23 @@ function make_corpus(cfg::FS.Config, corpus_dir::AbstractString, n::Int) # 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())) + # 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 """ - reknown!(cfg, jobs) + respool!(cfg, jobs) -Put every corpus file back in `known/`, wherever the last pass left it (done/ or +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. +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 reknown!(cfg::FS.Config, jobs::Vector{FS.Job}) +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) @@ -463,8 +465,8 @@ function component_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) nio end) - # --- move_to: the rename known/ -> done/. Consuming: reset each pass. - add!("move_to (rename)", "commit", best_of(() -> reknown!(cfg, jobs); trials) do + # --- move_to: the rename spool/ -> done/. 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 @@ -474,17 +476,17 @@ function component_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) # --- 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 + 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) - reknown!(cfg, jobs) + respool!(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") + logfile = joinpath(dirname(cfg.spool_dir), "bench_stage2.log") for (which, label) in ((:null, "logging (NullLogger)"), (:format, "logging (format only)"), (:flush, "logging (flush→file)")) @@ -513,13 +515,13 @@ 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") + 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(() -> reknown!(cfg, jobs); trials) do + best_of(() -> respool!(cfg, jobs); trials) do @inbounds for job in jobs FS.handle_known_job(job, cfg, 1) end @@ -528,7 +530,7 @@ function handler_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) end push!(rows, (; name = label, part = "total", ns)) end - reknown!(cfg, jobs) + respool!(cfg, jobs) rm(logfile; force = true) return rows end @@ -556,13 +558,13 @@ function thread_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) [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") + 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(() -> (reknown!(cfg, jobs); next[] = 1); trials) do + 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. @@ -582,7 +584,7 @@ function thread_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) k == counts[1] && (base = r) push!(rows, (; workers = k, ns, files_per_sec = r, speedup = r / base)) end - reknown!(cfg, jobs) + respool!(cfg, jobs) rm(logfile; force = true) return rows end @@ -622,12 +624,11 @@ function main(argv) 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) + for d in (cfg.spool_dir, cfg.done_dir, cfg.failed_dir) mkpath(d) end diff --git a/src/FileServer.jl b/src/FileServer.jl index 4ba6f4a..ee56d17 100644 --- a/src/FileServer.jl +++ b/src/FileServer.jl @@ -117,18 +117,13 @@ function run(; overrides...) DETECTOR[] = LanguageDetector() @info "loaded language detector" - # Stage-aware recovery: re-drive each stage's leftovers onto its own queue so - # files resume where they were, not from scratch. spool/ → stage-1, - # known/ → stage-2, unknown/ → stage-3. - recovered = recover_dir!(cfg.spool_dir, queue) - recovered_known = recover_dir!(cfg.known_dir, known_queue) - recovered_unknown = recover_dir!(cfg.unknown_dir, unknown_queue) - recovered_text = recover_dir!(cfg.text_dir, text_queue) - @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count text_workers=cfg.text_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity text_capacity=cfg.text_queue_capacity recovered=recovered recovered_known=recovered_known recovered_unknown=recovered_unknown recovered_text=recovered_text + @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count text_workers=cfg.text_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity text_capacity=cfg.text_queue_capacity # Zero the counters here, not at module load: `since` should mean "serving # since", so a scrape's totals cover the run, not the minutes spent loading - # the classifier. Nothing has been processed yet — recovery only enqueues. + # the classifier. It must also come *before* the pools start, since recovery + # below runs against live workers — the replayed files are real work and + # belong in the totals, but only from this point on. reset_metrics!() # Each pool gets its stage's counters (src/stats.jl); `worker_loop` records @@ -151,6 +146,20 @@ function run(; overrides...) st.language) for i in 1:cfg.text_worker_count] + # Recovery runs *after* the pools are live, and blocks rather than dropping + # when the queue is full. `spool/` holds every in-flight file now, so a crash + # under load can leave far more leftovers than `queue_capacity`; with the + # consumers already running the queue drains as we fill it, so a big backlog + # takes longer to re-drive but none of it is abandoned. Serving hasn't + # started yet, so intake cannot race us for the directory. + # + # Everything re-enters at stage 1 and replays: the queues are in-process, so + # a crash loses the only record of how far each file had got. Safe, because + # every stage is a pure function of the file's bytes and every commit is + # idempotent — see the header of src/worker.jl for the trade and the fix. + recovered = recover_dir!(cfg.spool_dir, queue) + recovered > 0 && @info "recovered leftover files; replaying from stage 1" recovered=recovered + register_routes() # `handler` replaces Oxygen's root stream handler so POST /upload can read its # body incrementally instead of having it buffered into memory first; every diff --git a/src/config.jl b/src/config.jl index d1fd1a9..9841d82 100644 --- a/src/config.jl +++ b/src/config.jl @@ -13,7 +13,7 @@ Base.@kwdef struct Config known_worker_count::Int = Threads.nthreads() known_queue_capacity::Int = 1000 # Stage 3 (content triage) also gets its own pool + queue: sorting an - # unrecognized file into binary/ vs text/ is cheap I/O, tuned independently + # unrecognized file into binary vs text is cheap I/O, tuned independently # of the classify and enrich pools. unknown_worker_count::Int = Threads.nthreads() unknown_queue_capacity::Int = 1000 @@ -22,11 +22,12 @@ Base.@kwdef struct Config # github-linguist) is a mix of CPU and process-spawn work, tuned independently. text_worker_count::Int = Threads.nthreads() text_queue_capacity::Int = 1000 - spool_dir::String = "data/spool" # files land here on intake (pending classification) - known_dir::String = "data/known" # classified-known, awaiting enrichment (stage 2) - unknown_dir::String = "data/unknown" # classified-unknown, awaiting content triage (stage 3) + # Every in-flight file lives here, at every stage, from intake until it is + # committed to a terminal sink below. Stages route by enqueueing a reference, + # not by moving bytes (src/worker.jl header), so there are no per-stage + # directories — the queues are what know how far a file has got. + spool_dir::String = "data/spool" binary_dir::String = "data/binary" # stage-3 sink: unknown files that look like binary data - text_dir::String = "data/text" # classified-text, awaiting language enrichment (stage 4) done_dir::String = "data/done" # fully enriched known files (+ .meta.json sidecars) text_done_dir::String = "data/text_done" # fully enriched text files (+ .meta.json sidecars) failed_dir::String = "data/failed" # files move here if a worker throws @@ -67,7 +68,7 @@ Recognised variables: FS_KNOWN_WORKERS, FS_KNOWN_QUEUE_CAPACITY, FS_UNKNOWN_WORKERS, FS_UNKNOWN_QUEUE_CAPACITY, FS_TEXT_WORKERS, FS_TEXT_QUEUE_CAPACITY, - FS_SPOOL_DIR, FS_KNOWN_DIR, FS_UNKNOWN_DIR, FS_BINARY_DIR, FS_TEXT_DIR, + FS_SPOOL_DIR, FS_BINARY_DIR, FS_DONE_DIR, FS_TEXT_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH, FS_UPLOAD_CHUNK_BYTES, FS_EXIFTOOL_TIMEOUT, FS_LINGUIST_TIMEOUT, FS_CLUSTER_DIR, FS_CLUSTER_N, FS_CLUSTER_ALPHA, FS_CLUSTER_PSEUDOCOUNT, @@ -79,8 +80,7 @@ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, known_queue_capacity=nothing, unknown_worker_count=nothing, unknown_queue_capacity=nothing, text_worker_count=nothing, text_queue_capacity=nothing, spool_dir=nothing, - known_dir=nothing, unknown_dir=nothing, binary_dir=nothing, - text_dir=nothing, done_dir=nothing, text_done_dir=nothing, + binary_dir=nothing, done_dir=nothing, text_done_dir=nothing, failed_dir=nothing, model_path=nothing, upload_chunk_bytes=nothing, exiftool_timeout=nothing, linguist_timeout=nothing, cluster_dir=nothing, cluster_n=nothing, @@ -100,10 +100,7 @@ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, text_worker_count = something(text_worker_count, parse(Int, get(ENV, "FS_TEXT_WORKERS", string(Threads.nthreads())))), text_queue_capacity = something(text_queue_capacity, parse(Int, get(ENV, "FS_TEXT_QUEUE_CAPACITY", "1000"))), spool_dir = something(spool_dir, get(ENV, "FS_SPOOL_DIR", "data/spool")), - known_dir = something(known_dir, get(ENV, "FS_KNOWN_DIR", "data/known")), - unknown_dir = something(unknown_dir, get(ENV, "FS_UNKNOWN_DIR", "data/unknown")), binary_dir = something(binary_dir, get(ENV, "FS_BINARY_DIR", "data/binary")), - text_dir = something(text_dir, get(ENV, "FS_TEXT_DIR", "data/text")), done_dir = something(done_dir, get(ENV, "FS_DONE_DIR", "data/done")), text_done_dir = something(text_done_dir, get(ENV, "FS_TEXT_DONE_DIR", "data/text_done")), failed_dir = something(failed_dir, get(ENV, "FS_FAILED_DIR", "data/failed")), @@ -125,9 +122,8 @@ end "Create all the pipeline-stage directories if they don't already exist." function ensure_dirs(cfg::Config) - for d in (cfg.spool_dir, cfg.known_dir, cfg.unknown_dir, cfg.binary_dir, - cfg.text_dir, cfg.done_dir, cfg.text_done_dir, cfg.failed_dir, - cfg.nominated_dir) + for d in (cfg.spool_dir, cfg.binary_dir, cfg.done_dir, cfg.text_done_dir, + cfg.failed_dir, cfg.nominated_dir) mkpath(d) end return nothing diff --git a/src/content.jl b/src/content.jl index 2f0f331..287d45f 100644 --- a/src/content.jl +++ b/src/content.jl @@ -1,8 +1,10 @@ # Stage-3 content triage for unknown files. # # A file that stage-1 couldn't recognize is still sorted into one of two coarse -# buckets so downstream tooling can treat them differently: `text/` for -# human-readable content, `binary/` for everything else. We sniff only the first +# buckets so downstream tooling can treat them differently: text (human-readable) +# and binary (everything else). Only binary is a directory — it is terminal, so +# the file is committed to `binary/`; a text file has stage 4 still to come, so it +# stays in `spool/` and only its queue reference moves on. We sniff only the first # `CONTENT_SNIFF_BYTES` (no full read) and ask two questions: does the window # decode as valid UTF-8, and are any of its control bytes ones that don't belong # in text? This is the Unicode-aware successor to the classic "NUL byte" test — diff --git a/src/language.jl b/src/language.jl index 8272416..6749036 100644 --- a/src/language.jl +++ b/src/language.jl @@ -1,7 +1,7 @@ # Stage-4 language enrichment for text files. # -# A file that stage-3 sorted into `text/` is human-readable, but we don't yet -# know *what* it is. This stage answers two questions and records them in a +# A file that stage-3 sorted as text is human-readable, but we don't yet know +# *what* it is. This stage answers two questions and records them in a # `.meta.json` sidecar, exactly like the stage-2 known-file enrichment: # # * natural language — via Languages.jl's `LanguageDetector` (a Julia port of diff --git a/src/queue.jl b/src/queue.jl index 44da94d..b4a688b 100644 --- a/src/queue.jl +++ b/src/queue.jl @@ -8,6 +8,15 @@ abstract type JobQueue end +# How long a producer backs off before retrying an enqueue onto a full queue. +# Blocking backpressure: a file that is already on disk is never dropped, so the +# producer parks until the consumer makes room. Used by the stage-1 and stage-3 +# routing handoffs and by startup recovery (see `enqueue_blocking!`). Intake is +# deliberately *not* a user: the HTTP path's `enqueue!` stays non-blocking and +# turns a full queue into a 503, so a saturated pipeline slows uploads down +# rather than holding request threads hostage. +const ROUTE_ENQUEUE_RETRY_SECONDS = 0.05 + """ enqueue!(q, job) -> Bool diff --git a/src/spool.jl b/src/spool.jl index dcb9a79..efc20cc 100644 --- a/src/spool.jl +++ b/src/spool.jl @@ -48,7 +48,15 @@ function spool_stream(write_body!, cfg::Config, original_name::AbstractString):: return Job(id, String(original_name), path, nbytes, time()) end -"Move a spooled file into `dir` (done/ or failed/), returning the destination." +""" + move_to(dir, job) -> String + +Move a spooled file into `dir`, returning the destination. + +This runs once per file, at the end: into a terminal sink (`done/`, `text_done/`, +`binary/`) or into `failed/` when a handler throws. Files are *not* moved between +stages — see the header of src/worker.jl. +""" function move_to(dir::AbstractString, job::Job)::String dest = joinpath(dir, basename(job.path)) mv(job.path, dest; force=true) @@ -64,12 +72,24 @@ const UUID_LEN = 36 Re-enqueue any files sitting in `dir` (left by a crash, hard shutdown, or an intake that never finished) onto `queue`. This is the payoff of spooling to -disk: a restart resumes work instead of stranding it. Stage-aware recovery uses -one call per stage — `spool/` → stage-1 queue, `known/` → known queue — so each -file re-enters at the correct stage rather than being reclassified from scratch. -Returns the number of files recovered. +disk: a restart resumes work instead of stranding it. Returns the number of +files recovered. -Skips `.meta.json` sidecars: those are stage-2 output, not work to redo. +A file's stage is not recorded on disk — it lives in whichever in-process queue +holds its reference, and a crash loses that (src/worker.jl header). So everything +still in `spool/` re-enters at stage 1 and replays. Replay is safe rather than +merely tolerable: classification and the binary/text sniff are pure functions of +the file's bytes, and the terminal commits rename with `force=true`, so a +replayed file lands where it would have landed and overwrites its own sidecar. + +**This blocks on a full queue**, and that is the whole point: `spool/` now holds +every in-flight file, so a crash under load can easily leave more leftovers than +one queue's capacity. Dropping the excess would mean recovery quietly losing the +work it exists to save. Because the caller spawns the worker pools *before* +recovering, the consumers are live and the queue drains as we fill it — parking +here costs latency, never progress. + +Skips `.meta.json` sidecars: those are enrichment output, not work to redo. """ function recover_dir!(dir::AbstractString, queue::JobQueue)::Int n = 0 @@ -85,7 +105,10 @@ function recover_dir!(dir::AbstractString, queue::JobQueue)::Int name = fname end job = Job(id, name, path, filesize(path), time()) - enqueue!(queue, job) || @warn "queue full during recovery; leaving file for next start" path + # `nothing` stats: recovery is not a stage, so its wait must not be + # charged to any stage's blocked_ns (see `enqueue_blocking!`). + enqueue_blocking!(queue, job, nothing; + retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS) n += 1 end return n diff --git a/src/stats.jl b/src/stats.jl index 542a51f..bd64bfd 100644 --- a/src/stats.jl +++ b/src/stats.jl @@ -1,10 +1,12 @@ # Per-stage pipeline metrics. # # End-to-end throughput says how fast the pipeline is; it does not say which -# stage is the reason. The external harness can't answer that either: `known/`, -# `unknown/` and `text/` are *transient* — a file can pass through one between -# two directory polls — so sampling directories from outside undercounts exactly -# the stages we most want to measure. Only the pipeline itself sees every job. +# stage is the reason. The external harness can't answer that either, and now +# cannot even approximate it: every in-flight file sits in `spool/` regardless of +# how far it has got, because stages route by enqueueing a reference rather than +# by moving bytes (src/worker.jl header). A file's stage is a property of the +# queue holding it, so only the pipeline itself can see it — there is no +# directory to poll. # # So each stage keeps four counters, all incremented in `worker_loop` (the one # place every stage's work passes through) and read by `GET /stats`: @@ -154,15 +156,22 @@ so a full downstream queue means waiting, not failing. Wrapping the retry loop here — rather than repeating `while !enqueue! sleep end` at each call site — is what makes that wait measurable at all, and keeps the three routing paths from drifting into three different backoff behaviours. + +`stats` is `nothing` for callers that are not a pipeline stage — startup +recovery (`recover_dir!`) uses the same never-drop retry, but it runs outside any +handler, so it has no `busy_ns` to charge against. Billing its wait to a stage's +`blocked_ns` would make `/stats` utilization — `(Δbusy_ns - Δblocked_ns) / …` — +go *negative* after a long recovery. Skipping the charge keeps `blocked_ns` +meaning exactly what its docs say: time a handler spent parked on its successor. """ -function enqueue_blocking!(queue::JobQueue, job::Job, stats::StageStats; +function enqueue_blocking!(queue::JobQueue, job::Job, stats::Union{StageStats,Nothing}; retry_seconds::Real) enqueue!(queue, job) && return nothing t0 = time_ns() while !enqueue!(queue, job) sleep(retry_seconds) end - Threads.atomic_add!(stats.blocked_ns, Int(time_ns() - t0)) + stats === nothing || Threads.atomic_add!(stats.blocked_ns, Int(time_ns() - t0)) return nothing end diff --git a/src/worker.jl b/src/worker.jl index fa3e48f..db56ef1 100644 --- a/src/worker.jl +++ b/src/worker.jl @@ -3,28 +3,48 @@ # stage, so `worker_loop` is parametrized with a `handler` and reused. Today # there are four stages: # -# stage 1 handle_classify_job spool/ → classify → known/ (+known queue) | unknown/ (+unknown queue) -# stage 2 handle_known_job known/ → exiftool enrich → done/ (+ .meta.json) -# stage 3 handle_unknown_job unknown/ → binary-vs-text sniff → binary/ | text/ (+text queue) -# stage 4 handle_text_job text/ → language enrich → text_done/ (+ .meta.json) +# 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. - -# How long a stage-1 worker backs off before retrying an enqueue onto a full -# downstream queue (known or unknown). Blocking backpressure: a classified file -# is never dropped, so the stage-1 worker parks until the next stage makes room. -# Keeps intake decoupled — the HTTP path's `enqueue!` stays non-blocking; only -# this worker-to-worker handoff blocks. -const ROUTE_ENQUEUE_RETRY_SECONDS = 0.05 +# +# 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): 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. +# 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 @@ -38,11 +58,13 @@ const ROUTE_ENQUEUE_RETRY_SECONDS = 0.05 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` → move to `known/`, enqueue onto the known queue for stage 2. - * `:unknown` → move to `unknown/`, enqueue onto the unknown queue for stage 3. + * `:known` → enqueue onto the known queue for stage 2. + * `:unknown` → enqueue onto the unknown queue for stage 3. -In both cases move first so the file physically lives in its stage dir before the -reference is visible downstream; the moved path becomes the routed job's location. +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`. """ @@ -53,18 +75,14 @@ function handle_classify_job(job::Job, cfg::Config, worker_id::Int, @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) - routed = Job(job.id, job.original_name, dest, job.size, job.received_at) # known queue full → park and retry, don't drop (time charged to blocked_ns) - enqueue_blocking!(known_queue, routed, stats; + enqueue_blocking!(known_queue, job, stats; retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS) - @debug "routed to enrichment" worker=worker_id id=job.id dest=dest + @debug "routed to enrichment" worker=worker_id id=job.id path=job.path 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; + enqueue_blocking!(unknown_queue, job, stats; retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS) - @debug "routed to content triage" worker=worker_id id=job.id dest=dest + @debug "routed to content triage" worker=worker_id id=job.id path=job.path end return nothing end @@ -73,7 +91,8 @@ 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. Extraction +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. @@ -89,10 +108,14 @@ 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: `binary/` (terminal — no further stage) if it looks like binary -data, `text/` otherwise. A text file is then routed onward to the stage-4 -language-enrichment queue, retrying on a full queue rather than dropping the file -(the same blocking backpressure stage 1 uses for its downstream queues). +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) @@ -100,12 +123,10 @@ function handle_unknown_job(job::Job, cfg::Config, worker_id::Int, 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 - dest = move_to(cfg.text_dir, job) - routed = Job(job.id, job.original_name, dest, job.size, job.received_at) # text queue full → park and retry, don't drop (time charged to blocked_ns) - enqueue_blocking!(text_queue, routed, stats; + enqueue_blocking!(text_queue, job, stats; retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS) - @info "routed to language enrichment" worker=worker_id id=job.id dest=dest + @info "routed to language enrichment" worker=worker_id id=job.id path=job.path end return nothing end @@ -115,7 +136,8 @@ end 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. Detection failure yields a *degraded* sidecar (the +`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. """ diff --git a/test/runtests.jl b/test/runtests.jl index 0f0892a..a3853c8 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -8,7 +8,8 @@ using JSON3 using FileServer: Job, Config, ChannelQueue, enqueue!, dequeue!, close!, length, sanitize_filename, recover_dir!, normalize_metadata, build_metadata, finalize_known!, run_exiftool, - is_binary, handle_unknown_job, worker_loop, + is_binary, handle_unknown_job, handle_classify_job, worker_loop, + load_classifier, CLASSIFIER, capacity, StageStats, IntakeStats, Metrics, METRICS, reset_metrics!, record_job!, enqueue_blocking!, stats_snapshot, STAGE_KEYS, detect_natural_language, run_linguist, detect_programming_language, @@ -70,10 +71,7 @@ end function tmp_config(root; kwargs...) cfg = Config(; spool_dir = joinpath(root, "spool"), - known_dir = joinpath(root, "known"), - unknown_dir = joinpath(root, "unknown"), binary_dir = joinpath(root, "binary"), - text_dir = joinpath(root, "text"), done_dir = joinpath(root, "done"), text_done_dir = joinpath(root, "text_done"), failed_dir = joinpath(root, "failed"), @@ -279,7 +277,7 @@ end mktempdir() do root cfg = tmp_config(root; exiftool_timeout=5) # Point at a nonexistent file → exiftool exits non-zero → degraded. - job = Job("id-3", "gone.dat", joinpath(cfg.known_dir, "id-3-gone.dat"), 99, 0.0) + job = Job("id-3", "gone.dat", joinpath(cfg.spool_dir, "id-3-gone.dat"), 99, 0.0) m = build_metadata(job, cfg) @test m.error !== nothing @test m.file_type === nothing @@ -341,7 +339,7 @@ end mktempdir() do root cfg = tmp_config(root) # A real known-stage file to enrich. - src = joinpath(cfg.known_dir, "id-9-pixel.png") + src = joinpath(cfg.spool_dir, "id-9-pixel.png") write(src, PNG_1x1) job = Job("id-9", "pixel.png", src, Base.length(PNG_1x1), 0.0) @@ -425,7 +423,7 @@ end stats = StageStats() # A binary file (embedded NUL) lands in binary/ and is NOT enqueued. - bpath = joinpath(cfg.unknown_dir, "id-b-blob.dat") + bpath = joinpath(cfg.spool_dir, "id-b-blob.dat") write(bpath, UInt8[0x00, 0xFF, 0x10]) bjob = Job("id-b", "blob.dat", bpath, filesize(bpath), 0.0) handle_unknown_job(bjob, cfg, 1, text_queue, stats) @@ -433,19 +431,49 @@ end @test !isfile(bpath) @test length(text_queue) == 0 - # A text file lands in text/ AND is routed onto the stage-4 queue, - # with its path updated to the new text/ location. - tpath = joinpath(cfg.unknown_dir, "id-t-notes.log") + # A text file is routed onto the stage-4 queue and *does not move*: + # stage 4 is still to come, so the bytes stay in spool/ and the same + # reference is handed on. Regression guard against re-introducing an + # intermediate hop — the queue, not a directory, is what records that + # this file has cleared triage. + tpath = joinpath(cfg.spool_dir, "id-t-notes.log") write(tpath, "just some log text\n") tjob = Job("id-t", "notes.log", tpath, filesize(tpath), 0.0) handle_unknown_job(tjob, cfg, 1, text_queue, stats) - moved = joinpath(cfg.text_dir, "id-t-notes.log") - @test isfile(moved) - @test !isfile(tpath) + @test isfile(tpath) # stayed put @test length(text_queue) == 1 routed = dequeue!(text_queue) @test routed.id == "id-t" - @test routed.path == moved + @test routed.path == tpath # reference unchanged + end + end + + @testset "handle_classify_job: routes by enqueue only, never moves the file" begin + # Stage 1's entire job is picking a queue. Whichever way the classifier + # votes, the bytes must stay where intake wrote them and the *same* Job + # must be handed on — the queue records the phase, not a directory. + # Asserted against both queues at once so the test doesn't depend on how + # the committed model happens to label these bytes. + CLASSIFIER[] = load_classifier(joinpath(@__DIR__, "..", "model", "classifier.jld2")) + mktempdir() do root + cfg = tmp_config(root) + known_queue, unknown_queue = ChannelQueue(10), ChannelQueue(10) + stats = StageStats() + + path = joinpath(cfg.spool_dir, "id-c-pixel.png") + write(path, PNG_1x1) + job = Job("id-c", "pixel.png", path, Base.length(PNG_1x1), 0.0) + + handle_classify_job(job, cfg, 1, known_queue, unknown_queue, stats) + + @test isfile(path) # stayed put + @test length(known_queue) + length(unknown_queue) == 1 # routed exactly once + routed = length(known_queue) == 1 ? dequeue!(known_queue) : dequeue!(unknown_queue) + @test routed.path == path # reference unchanged + @test routed.id == "id-c" + # No intermediate directory was invented alongside spool/. + @test !isdir(joinpath(root, "known")) + @test !isdir(joinpath(root, "unknown")) end end @@ -503,7 +531,7 @@ end cfg = tmp_config(root) d = LanguageDetector() - src = joinpath(cfg.text_dir, "id-x-script.py") + src = joinpath(cfg.spool_dir, "id-x-script.py") write(src, join(["# a short program in English prose comment", "import sys", "def greet(name):", @@ -542,7 +570,7 @@ end cfg = tmp_config(root) d = LanguageDetector() - src = joinpath(cfg.text_dir, "id-h-readme.md") + src = joinpath(cfg.spool_dir, "id-h-readme.md") write(src, "# Project\n\nThis project does something useful and interesting for everyone.\n") job = Job("id-h", "readme.md", src, filesize(src), 0.0) @@ -739,6 +767,39 @@ end end end + @testset "recover_dir!: blocks on a full queue, recovers every file" begin + # The regression this guards is data loss, not slowness. `spool/` now + # holds every in-flight file at every stage, so a crash under load leaves + # far more leftovers than one queue's capacity — and the old non-blocking + # recovery warned and abandoned the excess, which is exactly the work + # recovery exists to save. With the pool live, recovery must park until + # the consumer makes room and come back with all of it. + mktempdir() do root + dir = joinpath(root, "spool"); mkpath(dir) + n_files = 25 + for i in 1:n_files + write(joinpath(dir, string("id-", lpad(i, 3, '0'), "-f.dat")), "x") + end + + q = ChannelQueue(4) # deliberately far smaller than n_files + drained = Job[] + consumer = Threads.@spawn begin + while Base.length(drained) < n_files + job = dequeue!(q) + job === nothing && break + push!(drained, job) + end + end + + n = recover_dir!(dir, q) # would strand 21 files if it didn't block + wait(consumer) + + @test n == n_files + @test Base.length(drained) == n_files + @test Base.length(Set(j.path for j in drained)) == n_files # no duplicates + end + end + # Helper: write a "file" of raw bytes into a dir with a UUID-ish unique name, # returning its path. Mirrors what stage-3 deposits into binary/. function drop_binary(dir, bytes; name=string(rand(UInt128))) @@ -1012,7 +1073,7 @@ end text_queue = ChannelQueue(1) @test enqueue!(text_queue, Job("filler", "f", "/tmp/f", 1, 0.0)) - p = joinpath(cfg.unknown_dir, "id-t-notes.log") + p = joinpath(cfg.spool_dir, "id-t-notes.log") write(p, "plain text\n") job = Job("id-t", "notes.log", p, filesize(p), 0.0) @@ -1024,7 +1085,7 @@ end wait(drainer) @test stats.blocked_ns[] > 50_000_000 @test length(text_queue) == 1 # the file did get through - @test isfile(joinpath(cfg.text_dir, "id-t-notes.log")) + @test isfile(p) # and stayed in spool/ end end