diff --git a/README.md b/README.md index 91a674b..ea2c3bb 100644 --- a/README.md +++ b/README.md @@ -20,9 +20,9 @@ enrichment mixes CPU with a subprocess): POST /upload (multipart) │ ▼ - ┌─────────────────┐ spool bytes to disk + ┌─────────────────┐ stream bytes to disk (never buffered) │ HTTP handler │────────────────────────► data/spool/- - │ (Oxygen.jl) │ + │ (streaming) │ └────────┬─────────┘ enqueue reference (non-blocking) │ │ ▼ ▼ @@ -70,7 +70,12 @@ enrichment) for every file it sorts as text. Key properties: - **Fast intake:** the queue only ever carries small references; file bytes live - on disk, so memory stays flat regardless of file size. + on disk, so memory stays flat regardless of file size. This holds end to end: + intake **streams** each upload from the socket to the spool file a chunk at a + time (`FS_UPLOAD_CHUNK_BYTES`, default 64 KiB) rather than buffering the body, + and every worker reads only a bounded prefix. Measured: uploads of 256 MiB, + 1 GiB and 2 GiB each grow resident memory by ~20 MiB — a flat line in file + size. See "Streaming intake" and "Benchmarking" below. - **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). @@ -90,6 +95,46 @@ Key properties: - **Safe filenames:** client-supplied names are sanitized and prefixed with a server-minted UUID before touching the filesystem (no path traversal). +### Streaming intake + +The upload endpoint never holds a file in memory. Bytes go socket → spool file in +`FS_UPLOAD_CHUNK_BYTES` chunks, so resident memory per in-flight upload is set by +the chunk size, not the file size — a 2 GiB upload costs about what a 2 KiB one +does. Two pieces make that work, and both are deliberate: + +- **`src/multipart.jl` — an incremental multipart parser.** HTTP.jl's + `parse_multipart_form` takes the *complete* body as a byte vector, so using it + means every file in the request is in memory at once (and copied again per + part). The reader here pulls fixed-size chunks and hands each part's bytes + straight to its spool file. Its interface is two calls in a loop — + `next_part!` then `write_part_body!` (or `skip_part_body!`) — so the handler + keeps ordinary control flow instead of inverting into callbacks. The subtle + part is that a boundary delimiter can straddle two chunks, so the buffer always + retains the last `length(delimiter)-1` bytes; the test suite parses the same + body at chunk sizes from 1 byte upward to put that split at every offset. +- **`/upload` bypasses Oxygen's router.** Oxygen's root handler wraps + `HTTP.streamhandler`, which does `request.body = read(stream)` *before* + dispatching — even for an Oxygen `@stream` route, so no route can stream an + upload. `run` therefore passes its own `handler` to `serve` + (`root_stream_handler`), which intercepts `POST /upload` at the stream level and + delegates everything else to Oxygen unchanged. The trade-off: `/upload` is + absent from Oxygen's built-in metrics and docs. + +Streaming also changes what the endpoint can promise. A buffered handler knows up +front how many files a request holds; this one discovers them as they arrive. So +when the intake queue fills mid-request it does not abandon the connection: it +stops spooling (discarding the remaining parts rather than writing files it can't +queue), drains the body, and answers `503` with the `accepted` list of whatever +got in first. Files already queued stay queued, and the client can retry the rest. + +A client that hangs up mid-upload is treated as routine: the partial spool file is +removed (so restart recovery can never pick up a truncated upload as if it were +complete) and the event is logged `upload aborted by client`. One cosmetic caveat, +like the SIGTERM one below: when a request body is cut short, HTTP.jl's own +`closeread` logs an `EOFError` after the handler returns, because the connection +promised more bytes via `Content-Length` than arrived. It's harmless noise from +inside HTTP.jl — the partial file is already cleaned up and the connection closed. + ### Metadata enrichment (stage 2) Files the classifier labels **known** are handed to a second pool that extracts @@ -374,6 +419,7 @@ init, so the artifact is exactly regenerable from the same inputs. | `FS_TEXT_DONE_DIR` | `data/text_done` | Enriched text files (+ `.meta.json`) | | `FS_FAILED_DIR` | `data/failed` | Files whose processing threw | | `FS_MODEL_PATH` | `model/classifier.jld2` | Classifier artifact loaded at startup | +| `FS_UPLOAD_CHUNK_BYTES` | `65536` | Socket read size at intake; bounds intake memory per in-flight upload | | `FS_EXIFTOOL_TIMEOUT` | `30` | Seconds before a stuck exiftool is killed | | `FS_LINGUIST_TIMEOUT` | `30` | Seconds before a stuck github-linguist is killed | | `FS_CLUSTER_DIR` | `data/binary` | Stage-5 input: the unknown/binary pile to sweep | @@ -410,6 +456,61 @@ Each file in a request becomes its own job. Responses: - `503 Service Unavailable` — queue full, retry later - `500 Internal Server Error` — failed to write a file to disk +## Benchmarking (throughput + memory) + +`bin/bench.jl` measures the pipeline against a **running** server. It must run on +the same machine (it reads the sink dirs and `/proc`), and it changes nothing in +`src/` — it only speaks HTTP and counts files. + +```bash +julia --project=. -t auto bin/bench.jl --files 2000 --size 8k --concurrency 32 +julia --project=. -t auto bin/bench.jl --files 2 --size 1g --concurrency 1 # memory +julia --project=. -t auto bin/bench.jl --corpus ../training_set --concurrency 16 +``` + +Two properties of this design dictate how it measures: + +- **HTTP latency is not throughput.** `/upload` returns `202` once the bytes are + spooled and a reference is enqueued — all four stages run *after* the response, + 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/` names the + bottleneck stage directly. +- **Memory should be flat in file size, and the sweep is what proves it.** Both + halves of the pipeline are bounded: the workers read bounded prefixes (16+16 + bytes to classify, 8 KB to sniff, 64 KB to language-detect), and intake streams + each upload to disk a chunk at a time. So peak RSS should track *concurrency*, + not size. Measured on this machine (16 threads, 64 KiB chunk): + + | upload size | concurrency | RSS growth | + |---|---|---| + | 256 MiB × 4 | 1 | 21.8 MiB | + | 1 GiB × 2 | 1 | 20.8 MiB | + | 2 GiB × 1 | 1 | 17.0 MiB | + | 256 MiB × 4 | 4 | 86.9 MiB (21.7 MiB per in-flight upload) | + + Flat across a 8× range of file sizes, and linear in concurrency — which is the + shape to expect. (The residual ~20 MiB per in-flight upload is GC churn from the + chunk reads, not retained buffers; it does not grow with the file.) Before intake + was streamed, the same 256 MiB upload grew RSS by ~700 MiB and 4 concurrent ones + pushed a 950 MiB baseline past 2 GiB, so **a `--size` sweep that slopes upward is + the regression signal** — it means something has started buffering bodies again. + +Flags: `--files`, `--size` (`8k`/`64m`/`1g`), `--concurrency`, `--kind` +(`binary`/`text`/`mixed` — chooses which stages get loaded), `--corpus DIR` (real +files, the only way to exercise stage 2's exiftool path), `--pid`, `--no-mem`, +`--sample-ms`, `--timeout`, `--json PATH`. Full list in the script header. + +Two caveats the script reports rather than hides: it counts sink *deltas*, so it +warns if the pipeline isn't idle at the start (in-flight leftovers would be +counted as its own throughput); and because Julia's GC returns memory to the OS +lazily, a second run in the same process starts from an inflated baseline — it +resets the kernel's peak-RSS counter (`/proc//clear_refs`) per run and flags +a drifted baseline, but for a clean growth figure restart the server between +memory runs. + ## Layout ``` @@ -418,7 +519,8 @@ src/ config.jl Config struct + env parsing job.jl Job (the queue reference) queue.jl JobQueue seam + in-process ChannelQueue - spool.jl filename sanitizing, spool/move, startup recovery + multipart.jl streaming multipart/form-data reader (intake never buffers a file) + spool.jl filename sanitizing, streaming spool/move, startup recovery model.jl NN architecture + byte→feature mapping (shared with trainer) classify.jl load artifact + classify a file at inference time metadata.jl exiftool extraction + normalized sidecar (stage 2) @@ -427,9 +529,10 @@ src/ cluster.jl header-byte clustering model + Gibbs + scoring core (stage 5, science) catalog.jl durable single-owner format catalog + sweep + nominations (stage 5, phase B) worker.jl parametrized worker loop + classify/enrich/triage/language handlers - server.jl HTTP routes/handlers + server.jl HTTP routes + the streaming /upload handler bin/ server.jl entry point + bench.jl throughput + memory harness against a running server train.jl offline training script → model/classifier.jld2 cluster_calibrate.jl offline stage-5 hyperparameter calibration + NCD baseline cluster_sweep.jl stage-5 phase-B runner: sweep binary/, update catalog, write nominations diff --git a/bin/bench.jl b/bin/bench.jl new file mode 100755 index 0000000..a4709ca --- /dev/null +++ b/bin/bench.jl @@ -0,0 +1,563 @@ +#!/usr/bin/env julia +# +# bench.jl — measure end-to-end throughput and server memory for a running FileServer. +# +# Two things make this pipeline awkward to benchmark with off-the-shelf tools +# (ab/hey/wrk), and both shape what this script does: +# +# 1. HTTP latency is not throughput. /upload returns 202 as soon as the bytes +# are spooled and a reference is enqueued — all four stages run afterwards. +# So real throughput is the *arrival rate at the terminal sinks* +# (done/, text_done/, binary/, failed/), not the response rate. We upload a +# corpus, then poll the sinks until the file count stops moving. +# +# 2. Memory should be flat in file size, and that claim needs checking on two +# axes. The workers read bounded prefixes (16+16 bytes to classify, 8 KB to +# sniff, 64 KB to language-detect), and intake streams each upload from the +# socket to the spool file a chunk at a time (FS_UPLOAD_CHUNK_BYTES, see +# src/multipart.jl). So peak RSS should track *concurrency*, not file size: +# sweeping --size at fixed --concurrency should be a flat line, and that is +# the regression this measures. We sample the server's RSS throughout and +# report the high-water mark. +# +# (Before intake was streamed it buffered each upload whole, several times +# over, and a 256 MiB upload grew RSS by ~700 MiB. If a --size sweep ever +# slopes upward again, something has started buffering.) +# +# This is an external harness: it makes no assumptions about the server beyond +# the HTTP contract and the on-disk sink layout, and requires no changes to src/. +# It must run on the same machine as the server (it reads sink dirs and /proc). +# +# Usage: +# julia --project=. -t auto bin/bench.jl [options] +# +# --url URL server base URL (default: $FS_URL or http://127.0.0.1:8080) +# --files N number of files to upload (default: 200) +# --size S size of each generated file, e.g. 4k, 512k, 8m, 1g (default: 64k) +# --concurrency J uploads in flight at once (default: 8) +# --kind K binary | text | mixed — what to generate (default: binary) +# --corpus DIR upload an existing directory instead of generating +# (the only way to exercise stage 2: point it at real known files) +# --keep-corpus don't delete the generated corpus on exit +# --pid PID server pid for memory sampling (default: autodetect) +# --no-mem skip memory sampling entirely +# --sample-ms MS sink/RSS sampling interval (default: 200) +# --timeout SEC give up after this long with no drain progress (default: 120) +# --json PATH also write the results as JSON +# --force skip the corpus-size safety check +# +# Examples: +# # throughput: many small files, high concurrency +# julia --project=. -t auto bin/bench.jl --files 2000 --size 8k --concurrency 32 +# +# # memory: flat in file size? sweep --size with concurrency pinned +# julia --project=. -t auto bin/bench.jl --files 4 --size 256m --concurrency 1 +# julia --project=. -t auto bin/bench.jl --files 2 --size 1g --concurrency 1 +# julia --project=. -t auto bin/bench.jl --files 8 --size 1g --concurrency 8 +# +# # stage 2 (exiftool) with real known files +# julia --project=. -t auto bin/bench.jl --corpus ../training_set --concurrency 16 + +using HTTP +using JSON3 +using Random + +# ---------------------------------------------------------------- option parsing + +const DEFAULTS = Dict{String,Any}( + "url" => get(ENV, "FS_URL", "http://127.0.0.1:8080"), + "files" => 200, + "size" => 64 * 1024, + "concurrency" => 8, + "kind" => "binary", + "corpus" => nothing, + "keep-corpus" => false, + "pid" => nothing, + "no-mem" => false, + "sample-ms" => 200, + "timeout" => 120, + "json" => nothing, + "force" => false, +) + +const FLAGS = ("keep-corpus", "no-mem", "force") + +# Approximate RSS of a freshly started server (Lux + the loaded classifier + the +# language detector, measured on Julia 1.12 / -t auto). Only used to notice that +# a baseline is inflated by a previous run's un-returned GC memory, so a rough +# figure is enough. +const FRESH_RSS_HINT = 950 * 1024^2 + +"Parse `4k`/`8M`/`1g`/`4096` into a byte count." +function parse_size(s::AbstractString)::Int + m = match(r"^(\d+(?:\.\d+)?)\s*([kKmMgG]?)[bB]?$", strip(s)) + m === nothing && error("bad --size: $s (expected e.g. 512, 64k, 8m, 1g)") + mult = Dict('k' => 1024, 'm' => 1024^2, 'g' => 1024^3) + scale = isempty(m[2]) ? 1 : mult[lowercase(m[2])[1]] + return round(Int, parse(Float64, m[1]) * scale) +end + +function parse_args(argv)::Dict{String,Any} + opts = copy(DEFAULTS) + i = 1 + while i <= length(argv) + a = argv[i] + startswith(a, "--") || error("unexpected argument: $a (see the header of $(PROGRAM_FILE))") + 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") + val = argv[i+1] + opts[key] = key == "size" ? parse_size(val) : + key in ("files", "concurrency", "sample-ms") ? parse(Int, val) : + key == "timeout" ? parse(Float64, val) : + key == "pid" ? parse(Int, val) : + val + i += 2 + end + opts["kind"] in ("binary", "text", "mixed") || + error("--kind must be binary, text or mixed (got $(opts["kind"]))") + opts["files"] >= 1 || error("--files must be >= 1") + opts["concurrency"] >= 1 || error("--concurrency must be >= 1") + return opts +end + +# ------------------------------------------------------------------------- dirs +# +# Resolved from the same environment variables src/config.jl reads, so a server +# started with custom dirs is benchmarked correctly. Kept as a standalone table +# rather than `using FileServer` so the harness doesn't pay to load Lux. + +sinkdirs() = ( + done = get(ENV, "FS_DONE_DIR", "data/done"), + text_done = get(ENV, "FS_TEXT_DONE_DIR", "data/text_done"), + binary = get(ENV, "FS_BINARY_DIR", "data/binary"), + 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"), +) + +"Count work items in `dir`, ignoring the .meta.json sidecars stages 2/4 write." +function count_files(dir::AbstractString)::Int + isdir(dir) || return 0 + n = 0 + for name in readdir(dir) + endswith(name, ".meta.json") && continue + isfile(joinpath(dir, name)) && (n += 1) + end + return n +end + +counts(dirs) = NamedTuple{keys(dirs)}(map(count_files, values(dirs))) +total(c) = sum(values(c)) +deltas(now_, base) = NamedTuple{keys(now_)}(map(-, values(now_), values(base))) + +# ----------------------------------------------------------------------- memory + +"Read (VmRSS, VmHWM) in bytes for `pid`, or `nothing` if unreadable." +function read_rss(pid::Int) + rss = hwm = nothing + try + for line in eachline("/proc/$pid/status") + if startswith(line, "VmRSS:") + rss = parse(Int, split(line)[2]) * 1024 + elseif startswith(line, "VmHWM:") + hwm = parse(Int, split(line)[2]) * 1024 + end + end + catch + return nothing + end + return (rss === nothing || hwm === nothing) ? nothing : (rss, hwm) +end + +"Find the running server process, or `nothing`." +function detect_pid() + out = try + readchomp(`pgrep -f "bin/server.jl"`) + catch + return nothing + end + pids = parse.(Int, split(out)) + isempty(pids) && return nothing + length(pids) > 1 && @warn "multiple server processes matched; sampling the first" pids + return first(pids) +end + +""" +Reset the kernel's peak-RSS counter so VmHWM reflects only this run. + +Without it, VmHWM carries the high-water mark from startup (model load) or from +an earlier benchmark, which would silently dominate a small run's result. +Requires the server to run as the same user; on failure we say so and fall back +to sampled VmRSS, which can miss a spike between samples. +""" +function reset_peak_rss(pid::Int)::Bool + try + write("/proc/$pid/clear_refs", "5") + return true + catch + return false + end +end + +# ------------------------------------------------------------------------ corpus + +const WORDS = split("the quick brown fox jumps over a lazy dog while parsing " * + "headers and spooling bytes onto disk for later enrichment " * + "because throughput matters more than latency here") + +"Write one file of exactly `size` bytes, in bounded chunks so the generator +itself never holds a whole 1 GB file in memory." +function write_file(path::AbstractString, size::Int, kind::Symbol, rng) + chunk = 1024 * 1024 + open(path, "w") do io + remaining = size + while remaining > 0 + n = min(chunk, remaining) + if kind === :binary + write(io, rand(rng, UInt8, n)) + else + buf = IOBuffer() + while buf.size < n + print(buf, rand(rng, WORDS), rand(rng) < 0.06 ? ".\n" : " ") + end + write(io, take!(buf)[1:n]) + end + remaining -= n + end + end + return nothing +end + +"Generate the corpus and return (dir, paths, total_bytes)." +function make_corpus(opts) + n, size, kind = opts["files"], opts["size"], opts["kind"] + totalbytes = n * size + if totalbytes > 16 * 1024^3 && !opts["force"] + error("corpus would be $(human(totalbytes)) on disk; pass --force if that's intended") + end + dir = mktempdir(; prefix = "fsbench-") + rng = MersenneTwister(1234) + paths = String[] + for i in 1:n + k = kind == "mixed" ? (isodd(i) ? :binary : :text) : + kind == "text" ? :text : :binary + ext = k === :text ? "txt" : "bin" + path = joinpath(dir, "bench-$(lpad(i, 6, '0')).$ext") + write_file(path, size, k, rng) + push!(paths, path) + end + return dir, paths, totalbytes +end + +function existing_corpus(dir::AbstractString) + isdir(dir) || error("--corpus is not a directory: $dir") + paths = sort(filter(isfile, readdir(dir; join = true))) + isempty(paths) && error("--corpus directory is empty: $dir") + return paths, sum(filesize, paths) +end + +# ----------------------------------------------------------------------- upload + +struct Upload + status::Int # HTTP status, or 0 if the request threw + accepted::Int # jobs the server actually queued (from the 202/503 body) + seconds::Float64 +end + +"POST one file as multipart/form-data and report what the server accepted." +function upload_one(url::String, path::String)::Upload + t0 = time() + try + form = HTTP.Form(["file" => HTTP.Multipart(basename(path), open(path, "r"), + "application/octet-stream")]) + resp = HTTP.post(url, [], form; status_exception = false, retry = false) + # 202 and 503 both carry an `accepted` array: a partially-accepted batch + # still queued those jobs, and they will show up in the sinks. + acc = try + length(JSON3.read(String(resp.body)).accepted) + catch + resp.status == 202 ? 1 : 0 + end + return Upload(resp.status, acc, time() - t0) + catch e + e isa InterruptException && rethrow() + @warn "upload failed" file = basename(path) exception = e + return Upload(0, 0, time() - t0) + end +end + +""" +Upload every path, at most `concurrency` in flight. + +A bounded set of worker tasks pulling from a shared index keeps exactly +`concurrency` requests in flight for the whole run — unlike batching, where each +batch stalls on its slowest (largest) file and the real concurrency sags. +""" +function upload_all(url::String, paths::Vector{String}, concurrency::Int) + endpoint = string(rstrip(url, '/'), "/upload") + results = Vector{Upload}(undef, length(paths)) + next = Threads.Atomic{Int}(1) + @sync for _ in 1:min(concurrency, length(paths)) + Threads.@spawn while true + i = Threads.atomic_add!(next, 1) + i > length(paths) && break + results[i] = upload_one(endpoint, paths[i]) + end + end + return results +end + +# ------------------------------------------------------------------- formatting + +function human(bytes::Real) + b = Float64(bytes) + for unit in ("B", "KiB", "MiB", "GiB", "TiB") + (abs(b) < 1024 || unit == "TiB") && return "$(round(b; digits = 2)) $unit" + b /= 1024 + end +end + +fmt(x::Real, digits::Int = 2) = string(round(Float64(x); digits = digits)) + +function percentile(sorted::Vector{Float64}, p::Float64) + isempty(sorted) && return NaN + idx = clamp(ceil(Int, p * length(sorted)), 1, length(sorted)) + return sorted[idx] +end + +# ------------------------------------------------------------------------- main + +function main(argv) + opts = parse_args(argv) + url = string(opts["url"]) + + # Fail fast and clearly if there's no server, rather than reporting a run of zeros. + try + HTTP.get(string(rstrip(url, '/'), "/health"); retry = false, readtimeout = 5) + catch e + println(stderr, "cannot reach $url/health — is the server running?") + println(stderr, " start it with: julia --project=. -t auto bin/server.jl") + return 1 + end + + sinks, stages = sinkdirs(), stagedirs() + + # 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; " * + "throughput will include their completions" + + # --- corpus + generated = opts["corpus"] === nothing + corpusdir, paths, corpusbytes = if generated + print("generating corpus: $(opts["files"]) × $(human(opts["size"])) ($(opts["kind"]))… ") + t = time() + d, p, b = make_corpus(opts) + println("done in $(fmt(time() - t))s → $d") + d, p, b + else + p, b = existing_corpus(String(opts["corpus"])) + println("using corpus: $(length(p)) file(s), $(human(b)) from $(opts["corpus"])") + String(opts["corpus"]), p, b + end + + try + pid = opts["no-mem"] ? nothing : something(opts["pid"], detect_pid(), Some(nothing)) + if pid === nothing && !opts["no-mem"] + @warn "could not find the server process; skipping memory (pass --pid PID)" + end + + baseline_rss = nothing + peak_reset = false + if pid !== nothing + r = read_rss(pid) + r === nothing && (@warn "cannot read /proc/$pid/status; skipping memory"; pid = nothing) + if pid !== nothing + baseline_rss = r[1] + peak_reset = reset_peak_rss(pid) + peak_reset || @warn "could not reset the peak-RSS counter (/proc/$pid/clear_refs); " * + "reporting sampled RSS only" + end + end + + if Threads.nthreads() == 1 && opts["size"] > 64 * 1024^2 + @warn "running with 1 thread and large files: the sampler shares a thread with " * + "blocking file reads, so the RSS curve will be coarse. Prefer -t auto." + end + + base_sinks = counts(sinks) + interval = opts["sample-ms"] / 1000 + + # --- sampler: RSS curve + stage depths, for bottleneck attribution. + stop = Threads.Atomic{Bool}(false) + rss_samples = Float64[] + depth_max = Dict(k => 0 for k in keys(stages)) + sampler = Threads.@spawn begin + while !stop[] + if pid !== nothing + r = read_rss(pid) + r !== nothing && push!(rss_samples, Float64(r[1])) + end + d = counts(stages) + for k in keys(d) + depth_max[k] = max(depth_max[k], getfield(d, k)) + end + sleep(interval) + end + end + + # --- intake + println("uploading $(length(paths)) file(s) at concurrency $(opts["concurrency"])…") + t_start = time() + ups = upload_all(url, paths, opts["concurrency"]) + t_intake_end = time() + + accepted = sum(u.accepted for u in ups) + n_202 = count(u -> u.status == 202, ups) + n_503 = count(u -> u.status == 503, ups) + n_err = count(u -> !(u.status in (202, 503)), ups) + intake_secs = t_intake_end - t_start + lat = sort([u.seconds for u in ups]) + + println(" intake: $accepted job(s) accepted in $(fmt(intake_secs))s " * + "($(fmt(accepted / max(intake_secs, 1e-9))) files/s, " * + "$(fmt(corpusbytes / max(intake_secs, 1e-9) / 1024^2)) MiB/s)") + n_503 > 0 && println(" backpressure: $n_503 request(s) got 503 (intake queue full)") + n_err > 0 && println(" errors: $n_err request(s) failed or returned an unexpected status") + + # --- drain: poll the terminal sinks until they stop moving. + println("draining (polling sinks every $(opts["sample-ms"])ms)…") + completed = 0 + t_last_progress = time() + t_last_completion = t_intake_end + timed_out = false + while completed < accepted + sleep(interval) + c = total(deltas(counts(sinks), base_sinks)) + if c > completed + completed = c + t_last_completion = time() + t_last_progress = t_last_completion + elseif time() - t_last_progress > opts["timeout"] + timed_out = true + break + end + end + stop[] = true + wait(sampler) + + sink_delta = deltas(counts(sinks), base_sinks) + e2e_secs = t_last_completion - t_start + + final_rss = pid === nothing ? nothing : read_rss(pid) + peak_rss = if final_rss !== nothing && peak_reset + final_rss[2] # kernel VmHWM: catches spikes between samples + elseif !isempty(rss_samples) + maximum(rss_samples) + else + nothing + end + + # --- report + println() + println("=" ^ 68) + println("corpus $(length(paths)) file(s), $(human(corpusbytes)) total, " * + "$(human(corpusbytes / length(paths))) avg") + println("concurrency $(opts["concurrency"])") + println() + println("INTAKE (HTTP 202 — bytes spooled, not processed)") + println(" accepted $accepted of $(length(paths)) [202: $n_202, 503: $n_503, error: $n_err]") + println(" wall $(fmt(intake_secs))s") + println(" rate $(fmt(accepted / max(intake_secs, 1e-9))) files/s, " * + "$(fmt(corpusbytes / max(intake_secs, 1e-9) / 1024^2)) MiB/s") + println(" latency p50 $(fmt(percentile(lat, 0.5) * 1000, 1))ms " * + "p95 $(fmt(percentile(lat, 0.95) * 1000, 1))ms " * + "max $(fmt(percentile(lat, 1.0) * 1000, 1))ms") + println() + println("END-TO-END (files reaching a terminal sink)") + println(" completed $completed of $accepted accepted" * (timed_out ? " ** TIMED OUT **" : "")) + println(" wall $(fmt(e2e_secs))s (first upload → last completion)") + println(" throughput $(fmt(completed / max(e2e_secs, 1e-9))) files/s, " * + "$(fmt(corpusbytes / max(e2e_secs, 1e-9) / 1024^2)) MiB/s") + println(" sinks done $(sink_delta.done) text_done $(sink_delta.text_done) " * + "binary $(sink_delta.binary) failed $(sink_delta.failed)") + println(" peak depth spool $(depth_max[:spool]) known $(depth_max[:known]) " * + "unknown $(depth_max[:unknown]) text $(depth_max[:text])") + println(" (the stage that backs up is the bottleneck)") + println() + if peak_rss !== nothing + println("SERVER MEMORY (pid $pid)") + println(" baseline RSS $(human(baseline_rss))") + println(" peak RSS $(human(peak_rss))" * + (peak_reset ? " (kernel VmHWM, reset at start)" : " (sampled — may miss spikes)")) + println(" growth $(human(peak_rss - baseline_rss))") + println(" per in-flight $(human((peak_rss - baseline_rss) / opts["concurrency"])) " * + "at $(human(corpusbytes / length(paths))) avg file size") + println(" (should not grow with file size — intake streams to disk)") + # Julia's GC returns memory to the OS lazily, so a second run on the + # same process starts from an inflated baseline and under-reports + # growth. Absolute peak is the number to trust across runs. + baseline_rss > 1.3 * FRESH_RSS_HINT && + println(" ! baseline is well above a fresh start ($(human(FRESH_RSS_HINT))): the GC " * + "has not\n returned memory from earlier work. Compare " * + "absolute peak, or restart\n the server for a clean growth figure.") + else + println("SERVER MEMORY not sampled") + end + println("=" ^ 68) + + sink_delta.failed > 0 && + 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.") + + if opts["json"] !== nothing + result = ( + url, concurrency = opts["concurrency"], kind = opts["kind"], + files = length(paths), corpus_bytes = corpusbytes, + avg_file_bytes = corpusbytes / length(paths), + intake = (; accepted, n_202, n_503, n_err, seconds = intake_secs, + files_per_sec = accepted / max(intake_secs, 1e-9), + p50_ms = percentile(lat, 0.5) * 1000, + p95_ms = percentile(lat, 0.95) * 1000, + max_ms = percentile(lat, 1.0) * 1000), + end_to_end = (; completed, seconds = e2e_secs, timed_out, + files_per_sec = completed / max(e2e_secs, 1e-9), + mib_per_sec = corpusbytes / max(e2e_secs, 1e-9) / 1024^2, + sinks = sink_delta, peak_stage_depth = depth_max), + memory = (; pid, baseline_rss, peak_rss, peak_is_kernel_hwm = peak_reset, + growth = peak_rss === nothing ? nothing : peak_rss - baseline_rss, + samples = rss_samples), + ) + open(String(opts["json"]), "w") do io + JSON3.write(io, result) + end + println("\nwrote $(opts["json"])") + end + + return timed_out ? 1 : 0 + finally + if generated && !opts["keep-corpus"] + rm(corpusdir; recursive = true, force = true) + elseif generated + println("\nkept corpus: $corpusdir") + end + end +end + +if abspath(PROGRAM_FILE) == @__FILE__ + exit(main(ARGS)) +end diff --git a/src/FileServer.jl b/src/FileServer.jl index 438c965..75a4dd8 100644 --- a/src/FileServer.jl +++ b/src/FileServer.jl @@ -10,6 +10,7 @@ using Lux using JLD2 using Languages +include("multipart.jl") # streaming multipart reader (defines UPLOAD_CHUNK_BYTES, used by config.jl) include("config.jl") include("job.jl") include("queue.jl") @@ -137,7 +138,11 @@ function run(; overrides...) for i in 1:cfg.text_worker_count] register_routes() - serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false) + # `handler` replaces Oxygen's root stream handler so POST /upload can read its + # body incrementally instead of having it buffered into memory first; every + # other route still goes through Oxygen (see `root_stream_handler`). + serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false, + handler = root_stream_handler) # Idempotent graceful drain: stop accepting uploads, let workers finish the # buffered jobs, then exit. Called from two places: diff --git a/src/config.jl b/src/config.jl index 1c971ca..d1fd1a9 100644 --- a/src/config.jl +++ b/src/config.jl @@ -31,6 +31,10 @@ Base.@kwdef struct Config 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 model_path::String = "model/classifier.jld2" # committed classifier artifact, loaded at startup + # Intake reads each upload off the socket in chunks of this size and streams + # them straight to the spool file, so this — not the file size — is what + # bounds intake memory per in-flight upload (see src/multipart.jl). + upload_chunk_bytes::Int = UPLOAD_CHUNK_BYTES exiftool_timeout::Int = 30 # seconds before a stuck exiftool is killed → degraded sidecar linguist_timeout::Int = 30 # seconds before a stuck github-linguist is killed → no programming language # Stage 5 (unknown-format discovery). A separate single-owner batch process @@ -65,7 +69,7 @@ Recognised variables: FS_TEXT_WORKERS, FS_TEXT_QUEUE_CAPACITY, FS_SPOOL_DIR, FS_KNOWN_DIR, FS_UNKNOWN_DIR, FS_BINARY_DIR, FS_TEXT_DIR, FS_DONE_DIR, FS_TEXT_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH, - FS_EXIFTOOL_TIMEOUT, FS_LINGUIST_TIMEOUT, + FS_UPLOAD_CHUNK_BYTES, FS_EXIFTOOL_TIMEOUT, FS_LINGUIST_TIMEOUT, FS_CLUSTER_DIR, FS_CLUSTER_N, FS_CLUSTER_ALPHA, FS_CLUSTER_PSEUDOCOUNT, FS_CLUSTER_BG_MASS, FS_PROMOTE_MIN_MEMBERS, FS_PROMOTE_MIN_MAGIC, FS_CLUSTER_CATALOG, FS_NOMINATED_DIR @@ -77,7 +81,8 @@ function config_from_env(; host=nothing, port=nothing, 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, - failed_dir=nothing, model_path=nothing, exiftool_timeout=nothing, + failed_dir=nothing, model_path=nothing, + upload_chunk_bytes=nothing, exiftool_timeout=nothing, linguist_timeout=nothing, cluster_dir=nothing, cluster_n=nothing, cluster_alpha=nothing, cluster_pseudocount=nothing, cluster_bg_mass=nothing, promote_min_members=nothing, @@ -103,6 +108,7 @@ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, 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")), model_path = something(model_path, get(ENV, "FS_MODEL_PATH", "model/classifier.jld2")), + upload_chunk_bytes = something(upload_chunk_bytes, parse(Int, get(ENV, "FS_UPLOAD_CHUNK_BYTES", string(UPLOAD_CHUNK_BYTES)))), exiftool_timeout = something(exiftool_timeout, parse(Int, get(ENV, "FS_EXIFTOOL_TIMEOUT", "30"))), linguist_timeout = something(linguist_timeout, parse(Int, get(ENV, "FS_LINGUIST_TIMEOUT", "30"))), cluster_dir = something(cluster_dir, get(ENV, "FS_CLUSTER_DIR", "data/binary")), diff --git a/src/multipart.jl b/src/multipart.jl new file mode 100644 index 0000000..285f730 --- /dev/null +++ b/src/multipart.jl @@ -0,0 +1,296 @@ +# Streaming multipart/form-data reader. +# +# Why this exists: HTTP.jl's `parse_multipart_form` takes the *complete* request +# body as a byte vector, so using it means every file in the request sits in +# memory at once — and is then copied again per part. That contradicts the whole +# point of this service: file bytes belong on disk, and only a small reference +# travels through the queue. So intake needs a parser that never holds a file. +# +# This reader walks the body incrementally: it pulls fixed-size chunks off the +# socket and hands each part's bytes straight to a sink (the spool file). Peak +# memory per connection is `chunk_bytes` + the boundary length, regardless of how +# large — or how many — the uploaded files are. +# +# Interface: two calls in a loop, so the caller keeps ordinary control flow +# rather than inverting into callbacks. +# +# r = MultipartReader(io, boundary) +# while (part = next_part!(r)) !== nothing +# part.filename === nothing ? skip_part_body!(r) : write_part_body!(sink, r) +# end +# +# The grammar it implements (RFC 2046 §5.1, RFC 7578): +# +# [preamble] "--" boundary CRLF +# part-headers CRLF CRLF part-body +# CRLF "--" boundary CRLF ... another part ... +# CRLF "--" boundary "--" CRLF ... end of form, [epilogue] +# +# So the delimiter that *closes* a body is CRLF + "--" + boundary, and the two +# bytes after it say whether another part follows (CRLF) or the form is over +# ("--"). Every read is bounded, and the buffer retains only the last +# `length(delimiter)-1` bytes when no delimiter is found — that tail is what +# makes a delimiter split across two chunks parse correctly. + +"Default socket read size, and therefore the memory bound per in-flight upload." +const UPLOAD_CHUNK_BYTES = 64 * 1024 + +""" +A part header block bigger than this is abuse, not a filename. Bounding it keeps +the one genuinely unbounded-looking read (headers, which must be buffered whole +to be parsed) from being a memory hole. +""" +const MAX_PART_HEADER_BYTES = 16 * 1024 + +const CRLF = UInt8[0x0d, 0x0a] +const CRLFCRLF = UInt8[0x0d, 0x0a, 0x0d, 0x0a] +const DASHDASH = UInt8[0x2d, 0x2d] + +"A malformed (or truncated) multipart body. Callers turn this into a 400." +struct MultipartError <: Exception + msg::String +end + +Base.showerror(io::IO, e::MultipartError) = print(io, "MultipartError: ", e.msg) + +"What a part's headers said about it. `filename === nothing` means a plain form field, not a file." +struct MultipartPart + name::Union{String,Nothing} + filename::Union{String,Nothing} + content_type::Union{String,Nothing} +end + +""" + MultipartReader(io, boundary; chunk_bytes = UPLOAD_CHUNK_BYTES) + +An incremental reader over the multipart body arriving on `io`. `boundary` is the +value from the request's `Content-Type` header (see [`multipart_boundary`](@ref)). +""" +mutable struct MultipartReader{I<:IO} + io::I + dash_boundary::Vector{UInt8} # "--" boundary: opens the first part + delimiter::Vector{UInt8} # CRLF "--" boundary: closes every part + buf::Vector{UInt8} # rolling window; bounded by chunk_bytes + delimiter + pos::Int # next unconsumed index in buf + scratch::Vector{UInt8} # reused socket read target, so chunks don't churn the GC + chunk_bytes::Int + state::Symbol # :preamble | :at_delimiter | :body | :done +end + +function MultipartReader(io::IO, boundary::AbstractString; + chunk_bytes::Int = UPLOAD_CHUNK_BYTES) + chunk_bytes > 0 || throw(ArgumentError("chunk_bytes must be positive")) + isempty(boundary) && throw(MultipartError("empty multipart boundary")) + dash_boundary = Vector{UInt8}(codeunits(string("--", boundary))) + return MultipartReader(io, dash_boundary, vcat(CRLF, dash_boundary), + UInt8[], 1, Vector{UInt8}(undef, chunk_bytes), + chunk_bytes, :preamble) +end + +""" + multipart_boundary(content_type) -> String | nothing + +Pull the boundary out of a `multipart/form-data` Content-Type header. Returns +`nothing` if the header is missing, is some other media type, or has no boundary +— all of which are the same 400 to a caller. +""" +function multipart_boundary(content_type::Union{AbstractString,Nothing}) + content_type === nothing && return nothing + occursin(r"^\s*multipart/form-data"i, content_type) || return nothing + m = match(r"(?i:\bboundary)=(?:\"([^\"]+)\"|([^\s;]+))", content_type) + m === nothing && return nothing + return String(something(m[1], m[2])) +end + +# ------------------------------------------------------------------ buffer plumbing + +"Unconsumed bytes currently buffered." +navail(r::MultipartReader) = length(r.buf) - r.pos + 1 + +"Drop already-consumed bytes so the buffer stays bounded across a long body." +function compact!(r::MultipartReader) + r.pos == 1 && return nothing + n = navail(r) + n > 0 && copyto!(r.buf, 1, r.buf, r.pos, n) + resize!(r.buf, max(n, 0)) + r.pos = 1 + return nothing +end + +""" +Pull one more chunk off the wire, returning `false` at end of body. + +`readbytes!` on an `HTTP.Stream` returns at most what remains of the current +content-length or chunk, so this is bounded by `chunk_bytes`; `eof` is what +advances a chunked-encoded body to its next chunk, hence the guard. +""" +function fill_more!(r::MultipartReader) + eof(r.io) && return false + n = readbytes!(r.io, r.scratch, r.chunk_bytes) + n == 0 && return false + append!(r.buf, view(r.scratch, 1:n)) + return true +end + +"Buffer until `needle` is found, returning its range, or `nothing` at end of body." +function seek_needle!(r::MultipartReader, needle::Vector{UInt8}; limit::Int = 0) + while true + idx = findnext(needle, r.buf, r.pos) + idx === nothing || return idx + # Only the last length(needle)-1 bytes can still be part of a match, but + # the caller may need the skipped bytes (a part body), so trimming is the + # caller's job — we only enforce the optional limit. + limit > 0 && navail(r) > limit && + throw(MultipartError("no delimiter within $limit bytes")) + compact!(r) + fill_more!(r) || return nothing + end +end + +"Ensure at least `n` bytes are buffered; `false` if the body ended first." +function ensure!(r::MultipartReader, n::Int) + while navail(r) < n + compact!(r) + fill_more!(r) || return false + end + return true +end + +# Write `r.buf[range]` to `sink`. Goes through `unsafe_write` because +# `write(io, ::SubArray{UInt8})` falls back to a byte-at-a-time loop in Base, +# which would dominate the cost of a large upload. +function emit!(sink::IO, r::MultipartReader, from::Int, to::Int) + n = to - from + 1 + n <= 0 && return 0 + buf = r.buf # GC.@preserve needs a plain symbol, not a field access + GC.@preserve buf unsafe_write(sink, pointer(buf, from), UInt(n)) + return n +end + +# ------------------------------------------------------------------- parts + +""" + next_part!(r) -> MultipartPart | nothing + +Advance to the next part and return its headers, or `nothing` at the end of the +form. The previous part's body must have been consumed first (with +[`write_part_body!`](@ref) or [`skip_part_body!`](@ref)) — the reader cannot skip +a body it hasn't been told to, because the body is only bounded by finding the +next delimiter. +""" +function next_part!(r::MultipartReader) + r.state === :done && return nothing + r.state === :body && + throw(MultipartError("the current part's body must be consumed before the next part")) + + if r.state === :preamble + # Discard the preamble (RFC says ignore it) and consume the opening + # delimiter. Bounded: real clients send no preamble at all, and an + # unbounded scan here would be a way to make us buffer a whole body. + idx = seek_needle!(r, r.dash_boundary; limit = MAX_PART_HEADER_BYTES) + idx === nothing && throw(MultipartError("no multipart boundary found in body")) + r.pos = last(idx) + 1 + r.state = :at_delimiter + end + + # Just after a delimiter: "--" ends the form, CRLF introduces another part. + ensure!(r, 2) || throw(MultipartError("truncated body after a boundary delimiter")) + if view(r.buf, r.pos:r.pos+1) == DASHDASH + r.pos += 2 + r.state = :done + return nothing + end + skip_linear_whitespace!(r) + ensure!(r, 2) || throw(MultipartError("truncated body after a boundary delimiter")) + view(r.buf, r.pos:r.pos+1) == CRLF || + throw(MultipartError("boundary delimiter is not followed by a line ending")) + r.pos += 2 + + part = read_part_headers!(r) + r.state = :body + return part +end + +"RFC 2046 allows spaces/tabs between the delimiter and its line ending." +function skip_linear_whitespace!(r::MultipartReader) + while ensure!(r, 1) && (r.buf[r.pos] == 0x20 || r.buf[r.pos] == 0x09) + r.pos += 1 + end + return nothing +end + +function read_part_headers!(r::MultipartReader) + # A part with no headers at all is `CRLF CRLF body`: the empty line comes + # immediately, so searching for CRLFCRLF would run past it into the body. + if ensure!(r, 2) && view(r.buf, r.pos:r.pos+1) == CRLF + r.pos += 2 + return MultipartPart(nothing, nothing, nothing) + end + # The `limit` here bounds *buffering* — it only fires when the headers span + # chunks. The explicit length check below is the actual policy, so the rule + # doesn't depend on how the body happened to be chunked on the wire. + idx = seek_needle!(r, CRLFCRLF; limit = MAX_PART_HEADER_BYTES) + idx === nothing && throw(MultipartError("truncated body inside a part's headers")) + first(idx) - r.pos > MAX_PART_HEADER_BYTES && + throw(MultipartError("part headers exceed $MAX_PART_HEADER_BYTES bytes")) + # Copying is fine: the check above bounds this block. + block = String(r.buf[r.pos:first(idx)-1]) + r.pos = last(idx) + 1 + return parse_part_headers(block) +end + +"Unescape the backslash escapes RFC 2045 allows inside a quoted-string." +unquote(s::AbstractString) = replace(s, r"\\(.)" => s"\1") + +function parse_part_headers(block::AbstractString) + name = filename = content_type = nothing + for line in eachsplit(block, "\r\n") + colon = findfirst(':', line) + colon === nothing && continue + key = lowercase(strip(line[1:colon-1])) + value = strip(line[colon+1:end]) + if key == "content-disposition" + # `\b` matters: it keeps the `name=` pattern from matching inside `filename=`. + m = match(r"(?i:\bname)=(?:\"((?:[^\"\\]|\\.)*)\"|([^\s;]+))", value) + m === nothing || (name = unquote(String(something(m[1], m[2])))) + m = match(r"(?i:\bfilename)=(?:\"((?:[^\"\\]|\\.)*)\"|([^\s;]+))", value) + m === nothing || (filename = unquote(String(something(m[1], m[2])))) + elseif key == "content-type" + content_type = String(value) + end + end + return MultipartPart(name, filename, content_type) +end + +""" + write_part_body!(sink, r) -> Int + +Stream the current part's body into `sink`, returning the number of bytes +written. Nothing larger than a chunk is ever held in memory. +""" +function write_part_body!(sink::IO, r::MultipartReader) + r.state === :body || throw(MultipartError("no part body is open")) + total = 0 + keep = length(r.delimiter) - 1 # a delimiter may straddle two chunks + while true + idx = findnext(r.delimiter, r.buf, r.pos) + if idx !== nothing + total += emit!(sink, r, r.pos, first(idx) - 1) + r.pos = last(idx) + 1 + r.state = :at_delimiter + return total + end + # Emit only what cannot be the start of a straddling delimiter, then + # keep that tail and read more. + emit_to = length(r.buf) - keep + if emit_to >= r.pos + total += emit!(sink, r, r.pos, emit_to) + r.pos = emit_to + 1 + end + compact!(r) + fill_more!(r) || throw(MultipartError("truncated body inside a part")) + end +end + +"Consume and discard the current part's body (a form field, or a file we can't take)." +skip_part_body!(r::MultipartReader) = write_part_body!(devnull, r) diff --git a/src/server.jl b/src/server.jl index d00b7ac..3df308b 100644 --- a/src/server.jl +++ b/src/server.jl @@ -1,13 +1,21 @@ # HTTP layer: a single multipart upload endpoint plus a health check. # # The handler's whole job is to get files onto the queue fast and get out of the -# way: spool each uploaded file to disk, enqueue a reference, respond 202. It +# way: stream each uploaded file to disk, enqueue a reference, respond 202. It # never does real processing — that's the workers' job. # +# Intake is *streamed*, not buffered (see src/multipart.jl for the reader). This +# is what makes the service's memory story hold end to end: bytes go from the +# socket to the spool file a chunk at a time, so a 4 GB upload costs the same +# resident memory as a 4 KB one. It is also why /upload is served by its own +# stream handler rather than an Oxygen route — see `root_stream_handler`. +# # NOTE: routes are registered at runtime via `register_routes()` (called from # `run`), NOT with top-level macros. In a precompiled package, top-level # `@get`/`@post` would execute during precompilation and be lost before serving. +const UPLOAD_PATH = "/upload" + jsonresp(status::Int, data) = HTTP.Response(status, ["Content-Type" => "application/json"], JSON3.write(data)) @@ -15,48 +23,190 @@ function health_handler(_::HTTP.Request) return jsonresp(200, (; status = "ok")) end -function upload_handler(req::HTTP.Request) - cfg = CONFIG[] - queue = QUEUE[] - - parts = try - HTTP.parse_multipart_form(req) - catch - nothing - end - parts === nothing && - return jsonresp(400, (; error = "expected multipart/form-data")) - - files = filter(p -> p.filename !== nothing && !isempty(p.filename), parts) - isempty(files) && - return jsonresp(400, (; error = "no files found in request")) - - accepted = NamedTuple{(:id, :name),Tuple{String,String}}[] - for p in files - bytes = read(p.data) - - job = try - spool_file(cfg, p.filename, bytes) - catch e - @error "spool failed" name=p.filename exception=(e, catch_backtrace()) - return jsonresp(500, (; error = "failed to store file", accepted)) - end - - if !enqueue!(queue, job) - rm(job.path; force = true) # never queued → don't leave it in spool - return jsonresp(503, (; error = "queue full, retry later", accepted)) - end - - @info "accepted" id=job.id name=job.original_name size=job.size - push!(accepted, (; id = job.id, name = job.original_name)) - end - - return jsonresp(202, (; accepted)) -end - -"Register HTTP routes on the Oxygen instance. Must run at runtime, before serve." -function register_routes() - @get("/health", health_handler) - @post("/upload", upload_handler) +"Write a JSON response onto a raw stream (the streaming handler's `jsonresp`)." +function stream_jsonresp(stream::HTTP.Stream, status::Int, data) + body = JSON3.write(data) + HTTP.setstatus(stream, status) + HTTP.setheader(stream, "Content-Type" => "application/json") + HTTP.setheader(stream, "Content-Length" => string(sizeof(body))) + HTTP.startwrite(stream) + write(stream, body) + return nothing +end + +""" +Read and discard whatever is left of the request body. + +HTTP.jl's server calls `closeread` after the handler and *errors* if the body was +only partly consumed — a half-read body can't be followed by another request on a +keep-alive connection. So every exit path drains first. Discarding is bounded in +memory (one chunk) and costs nothing in the normal case, where the body is +already fully consumed and this returns immediately. +""" +function discard_body!(stream::HTTP.Stream, chunk_bytes::Int) + scratch = Vector{UInt8}(undef, chunk_bytes) + while !eof(stream) + readbytes!(stream, scratch, chunk_bytes) == 0 && break + end + return nothing +end + +""" +Did this exception mean the client went away, rather than something being wrong +on our side? + +A client that hangs up mid-upload (user cancels, network drops) is routine, and +must not be logged as a server error or reported as a failed write — but it looks +like an I/O failure from inside the parser, so the distinction has to be made +explicitly. `EOFError` is what `HTTP.Stream` raises when a connection dies with +bytes still promised by `Content-Length`. +""" +is_client_gone(e) = + e isa EOFError || + (e isa Base.IOError && e.code in (Base.UV_EPIPE, Base.UV_ECONNRESET, Base.UV_ECONNABORTED)) + +""" +Stream a multipart upload to disk, one part at a time. + +Each file part is spooled straight from the socket and its reference enqueued. +The response reports what was actually queued: + + * `202` — every file in the request was spooled and queued + * `400` — not multipart/form-data, no files present, or a malformed body + * `503` — the intake queue filled up; `accepted` lists what got in first + * `500` — a file could not be written to disk + +Unlike a buffered handler, this one cannot know up front how many files a request +holds or whether they will fit. So when the queue fills mid-request it does not +abandon the connection: it stops spooling (discarding the remaining parts rather +than writing files it can't queue), drains the body, and answers `503` with the +`accepted` list. Files already queued stay queued — a client can retry the rest. +""" +function upload_stream_handler(stream::HTTP.Stream) + try + return serve_upload!(stream) + catch e + # The client vanished — while we were reading its body, draining it, or + # answering. Nothing is wrong on our side, so log it as the routine event + # it is instead of a server error. + is_client_gone(e) || rethrow() + @info "upload aborted by client" + # HTTP.jl insists a handler write *some* response before returning. The + # client is probably already gone, so this write is best-effort: attempt + # it only if we haven't started a response, and let it fail silently. + if isopen(stream) && !iswritable(stream) + try + stream_jsonresp(stream, 400, (; error = "upload truncated")) + catch e2 + is_client_gone(e2) || rethrow() + end + end + return nothing + end +end + +function serve_upload!(stream::HTTP.Stream) + cfg = CONFIG[] + queue = QUEUE[] + chunk = cfg.upload_chunk_bytes + + boundary = multipart_boundary(HTTP.header(stream.message, "Content-Type", nothing)) + if boundary === nothing + discard_body!(stream, chunk) + return stream_jsonresp(stream, 400, (; error = "expected multipart/form-data")) + end + + reader = MultipartReader(stream, boundary; chunk_bytes = chunk) + accepted = NamedTuple{(:id, :name),Tuple{String,String}}[] + n_files = 0 + queue_full = false + failure = nothing # (status, message) from a fatal error mid-body + + try + while (part = next_part!(reader)) !== nothing + # A part with no filename is an ordinary form field, not a file. + if part.filename === nothing || isempty(part.filename) + skip_part_body!(reader) + continue + end + n_files += 1 + + # Already backpressured: consume the part, but don't write a file we + # know we cannot enqueue. + if queue_full + skip_part_body!(reader) + continue + end + + job = try + spool_stream(cfg, part.filename) do io + write_part_body!(io, reader) + end + catch e + (e isa MultipartError || is_client_gone(e)) && rethrow() + # A disk error leaves the reader mid-part, so the rest of the body + # can no longer be parsed — stop and report. + @error "spool failed" name=part.filename exception=(e, catch_backtrace()) + failure = (500, "failed to store file") + break + end + + if enqueue!(queue, job) + @info "accepted" id=job.id name=job.original_name size=job.size + push!(accepted, (; id = job.id, name = job.original_name)) + else + rm(job.path; force = true) # never queued → don't leave it in spool + queue_full = true + end + end + catch e + e isa MultipartError || rethrow() + @warn "malformed multipart upload" reason=e.msg accepted=length(accepted) + failure = (400, "malformed multipart body") + end + + discard_body!(stream, chunk) + + failure !== nothing && + return stream_jsonresp(stream, failure[1], (; error = failure[2], accepted)) + n_files == 0 && + return stream_jsonresp(stream, 400, (; error = "no files found in request")) + queue_full && + return stream_jsonresp(stream, 503, (; error = "queue full, retry later", accepted)) + return stream_jsonresp(stream, 202, (; accepted)) +end + +""" + root_stream_handler(middleware) -> (stream -> nothing) + +Oxygen's root handler wraps `HTTP.streamhandler`, which does +`request.body = read(stream)` — the entire body into memory — *before* any route +is dispatched. That happens even for an Oxygen `@stream` route, so no route can +stream an upload. We therefore intercept `POST /upload` at the stream level and +hand everything else to Oxygen unchanged. + +Trade-off: `/upload` bypasses Oxygen's middleware chain, so it is absent from +Oxygen's built-in metrics and docs. Deliberate — flat intake memory is the point +of this service, and the pipeline's own `@info` records cover intake anyway. +""" +function root_stream_handler(middleware::Function) + oxygen_handler = Oxygen.Core.stream_handler(middleware) + return function (stream::HTTP.Stream) + req = stream.message + if req.method == "POST" && HTTP.URI(req.target).path == UPLOAD_PATH + return upload_stream_handler(stream) + end + return oxygen_handler(stream) + end +end + +""" +Register HTTP routes on the Oxygen instance. Must run at runtime, before serve. + +`POST /upload` is deliberately absent: it is served by `upload_stream_handler` +via `root_stream_handler`, ahead of Oxygen's router. +""" +function register_routes() + @get("/health", health_handler) return nothing end diff --git a/src/spool.jl b/src/spool.jl index ea8756d..dcb9a79 100644 --- a/src/spool.jl +++ b/src/spool.jl @@ -19,15 +19,33 @@ function sanitize_filename(name::AbstractString)::String return first(base, MAX_NAME_LEN) end -"Write `bytes` to the spool dir under `-` and return the Job." -function spool_file(cfg::Config, original_name::AbstractString, bytes::Vector{UInt8})::Job - id = string(uuid4()) - safe = sanitize_filename(original_name) - path = joinpath(cfg.spool_dir, string(id, "-", safe)) - open(path, "w") do io - write(io, bytes) +"Build the spool path for a client-supplied name: `-`." +function spool_path(cfg::Config, original_name::AbstractString) + id = string(uuid4()) + return id, joinpath(cfg.spool_dir, string(id, "-", sanitize_filename(original_name))) +end + +""" + spool_stream(write_body!, cfg, original_name) -> Job + +Create the spool file for `original_name`, hand the open `IO` to `write_body!`, +and build the `Job` from however many bytes it reports writing. + +This is the streaming counterpart to `spool_file`: the caller pumps bytes in from +the network as they arrive, so a file never exists in memory in one piece. A +partial file left by a failed or abandoned write is removed — intake either +produces a complete spooled file or nothing at all, so recovery on restart never +picks up a truncated upload. +""" +function spool_stream(write_body!, cfg::Config, original_name::AbstractString)::Job + id, path = spool_path(cfg, original_name) + nbytes = try + open(write_body!, path, "w") + catch + rm(path; force = true) + rethrow() end - return Job(id, String(original_name), path, length(bytes), time()) + return Job(id, String(original_name), path, nbytes, time()) end "Move a spooled file into `dir` (done/ or failed/), returning the destination." diff --git a/test/runtests.jl b/test/runtests.jl index 31c2735..a55da9a 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -18,7 +18,10 @@ using FileServer: Job, Config, ChannelQueue, enqueue!, dequeue!, length, adjusted_rand_index, v_measure, HEADER_N, ALPHABET, PAST_EOF, Catalog, load_catalog, save_catalog!, catalog_sweep!, compact!, write_nominations!, run_cluster_sweep, binary_files, record_example!, - signature_hex, ensure_dirs + signature_hex, ensure_dirs, + MultipartReader, MultipartError, MultipartPart, next_part!, + write_part_body!, skip_part_body!, multipart_boundary, + parse_part_headers, spool_stream, UPLOAD_CHUNK_BYTES using Random: MersenneTwister using Languages: LanguageDetector @@ -28,6 +31,39 @@ const PNG_1x1 = UInt8[137,80,78,71,13,10,26,10,0,0,0,13,73,72,68,82,0,0,0,1,0, 0,0,1,8,6,0,0,0,31,21,196,137,0,0,0,11,73,68,65,84,120,218,99,100,248,255, 191,30,0,5,132,2,127,194,91,30,42,0,0,0,0,73,69,78,68,174,66,96,130] +""" +Assemble a multipart/form-data body. `parts` are `(name, filename, content_type, +data)` tuples; a `nothing` filename makes a plain form field rather than a file. +""" +function multipart_body(boundary, parts; preamble = "", terminate = true) + io = IOBuffer() + write(io, preamble) + for (name, filename, content_type, data) in parts + write(io, "--$boundary\r\n") + write(io, "Content-Disposition: form-data; name=\"$name\"") + filename === nothing || write(io, "; filename=\"$filename\"") + write(io, "\r\n") + content_type === nothing || write(io, "Content-Type: $content_type\r\n") + write(io, "\r\n") + write(io, data) + write(io, "\r\n") + end + write(io, terminate ? "--$boundary--\r\n" : "--$boundary\r\n") + return take!(io) +end + +"Read every part out of `bytes`, returning `(part, body, nbytes)` triples." +function read_all_parts(bytes, boundary; chunk_bytes = UPLOAD_CHUNK_BYTES) + r = MultipartReader(IOBuffer(bytes), boundary; chunk_bytes = chunk_bytes) + out = Tuple{MultipartPart,String,Int}[] + while (part = next_part!(r)) !== nothing + sink = IOBuffer() + n = write_part_body!(sink, r) + push!(out, (part, String(take!(sink)), n)) + end + return out +end + "Build a Config whose data dirs all live under a fresh temp directory." function tmp_config(root; kwargs...) cfg = Config(; @@ -65,6 +101,139 @@ end @test Base.length(sanitize_filename("a"^500)) == FileServer.MAX_NAME_LEN end + @testset "multipart_boundary: extraction from Content-Type" begin + @test multipart_boundary("multipart/form-data; boundary=abc") == "abc" + @test multipart_boundary("multipart/form-data; boundary=\"a b;c\"") == "a b;c" + @test multipart_boundary("MULTIPART/FORM-DATA; BOUNDARY=xyz") == "xyz" + @test multipart_boundary("multipart/form-data; charset=utf-8; boundary=q1") == "q1" + # Anything that isn't a usable multipart header is the same 400 to a caller. + @test multipart_boundary("multipart/form-data") === nothing + @test multipart_boundary("application/json") === nothing + @test multipart_boundary(nothing) === nothing + end + + @testset "parse_part_headers" begin + p = parse_part_headers("Content-Disposition: form-data; name=\"f\"; filename=\"a b.txt\"\r\n" * + "Content-Type: text/plain") + @test p.name == "f" + @test p.filename == "a b.txt" + @test p.content_type == "text/plain" + + # `name=` must not match inside `filename=` — that would label every + # file part with a bogus name and (worse) hide a missing real name. + p = parse_part_headers("Content-Disposition: form-data; filename=\"only.txt\"") + @test p.name === nothing + @test p.filename == "only.txt" + + # No filename means a plain form field, which intake must not spool. + p = parse_part_headers("Content-Disposition: form-data; name=\"note\"") + @test p.filename === nothing + end + + @testset "MultipartReader: parts, fields, and bodies" begin + B = "----testboundary" + body = multipart_body(B, [("f0", "a.txt", "text/plain", "hello world"), + ("note", nothing, nothing, "just-a-field"), + ("f1", "b.bin", nothing, "\x00\x01\x02")]) + got = read_all_parts(body, B) + @test length(got) == 3 + @test got[1][1].filename == "a.txt" + @test got[1][2] == "hello world" + @test got[1][3] == 11 # reported byte count + @test got[1][1].content_type == "text/plain" + @test got[2][1].filename === nothing # the form field + @test got[2][2] == "just-a-field" + @test codeunits(got[3][2]) == UInt8[0x00, 0x01, 0x02] + + # A zero-byte file is legal and must survive as zero bytes. + empty_got = read_all_parts(multipart_body(B, [("f", "empty.bin", nothing, "")]), B) + @test empty_got[1][3] == 0 + @test empty_got[1][2] == "" + end + + @testset "MultipartReader: delimiter straddling every chunk offset" begin + # The one thing a chunked parser can get catastrophically wrong is a + # delimiter split across two reads. Parsing the same body at many chunk + # sizes puts the split at every offset. The payload deliberately contains + # CR, LF and '-' bytes, so a sloppy scan finds false delimiters. + B = "----testboundary" + rng = MersenneTwister(7) + payload = String(rand(rng, UInt8[0x41:0x5a; 0x0d; 0x0a; 0x2d], 5000)) + body = multipart_body(B, [("f", "big.bin", nothing, payload)]) + for chunk in (1, 2, 3, 5, 7, 13, 16, 17, 64, 255, 4096, 10_000) + got = read_all_parts(body, B; chunk_bytes = chunk) + @test length(got) == 1 + @test got[1][2] == payload + end + + # A payload containing a *prefix* of the real delimiter must not end the part. + tricky = "aaa\r\n--" * "----testboundar" * "bbb\r\n--x\r\nccc" + body2 = multipart_body(B, [("f", "t.bin", nothing, tricky)]) + for chunk in (1, 4, 9, 64, 4096) + got = read_all_parts(body2, B; chunk_bytes = chunk) + @test length(got) == 1 + @test got[1][2] == tricky + end + end + + @testset "MultipartReader: memory stays bounded, not proportional to the part" begin + # The whole point of the streaming reader. A 16 MiB part read with a + # 64 KiB chunk must allocate on the order of the chunk, not the part. + B = "----testboundary" + payload = String(rand(MersenneTwister(11), UInt8, 16 * 1024 * 1024)) + body = multipart_body(B, [("f", "huge.bin", nothing, payload)]) + r = MultipartReader(IOBuffer(body), B; chunk_bytes = 64 * 1024) + next_part!(r) + GC.gc() + allocated = @allocated write_part_body!(devnull, r) + @test allocated < 4 * 1024 * 1024 + end + + @testset "MultipartReader: malformed bodies raise MultipartError" begin + B = "----testboundary" + valid = multipart_body(B, [("f", "a.bin", nothing, "hello")]) + @test_throws MultipartError read_all_parts(Vector{UInt8}("no delimiter here"), B) + @test_throws MultipartError read_all_parts(valid[1:end-20], B) # truncated mid-part + @test_throws MultipartError read_all_parts( + multipart_body(B, [("f", "a.bin", nothing, "x")]; terminate = false), B) + + # Part headers must be bounded regardless of how the body was chunked, + # since they are the one thing that has to be buffered whole to parse. + oversized = Vector{UInt8}("--$B\r\nContent-Disposition: form-data; name=\"" * + "x"^30_000 * "\"\r\n\r\ndata\r\n--$B--\r\n") + @test_throws MultipartError read_all_parts(oversized, B) + + # A part's body must be consumed before advancing: the reader cannot skip + # a body on its own, because a body only ends at the next delimiter. + r = MultipartReader(IOBuffer(valid), B) + next_part!(r) + @test_throws MultipartError next_part!(r) + end + + @testset "spool_stream: streams to disk, cleans up a failed write" begin + mktempdir() do root + cfg = tmp_config(root) + job = spool_stream(cfg, "report v2.pdf") do io + write(io, "abc") + write(io, "de") + end + @test isfile(job.path) + @test read(job.path, String) == "abcde" + @test job.size == 5 # from the bytes actually written + @test job.original_name == "report v2.pdf" + @test basename(job.path) == "$(job.id)-report_v2.pdf" # sanitized, uuid-prefixed + + # A write that throws must leave nothing behind: recovery on restart + # re-enqueues whatever is in spool/, and a truncated upload there + # would be silently processed as if it were complete. + before = length(readdir(cfg.spool_dir)) + @test_throws ErrorException spool_stream(cfg, "bad.bin") do io + write(io, "partial") + error("disk went away") + end + @test length(readdir(cfg.spool_dir)) == before + end + end + @testset "normalize_metadata" begin job = Job("id-1", "photo.jpg", "/data/known/id-1-photo.jpg", 4242, 0.0) # Group-prefixed tags as exiftool -G emits them are already group-stripped