diff --git a/README.md b/README.md index 45a3a44..6037fe0 100644 --- a/README.md +++ b/README.md @@ -11,9 +11,9 @@ classifier that labels it **known** (a file type resembling the training set) or ## Architecture -The pipeline is two stages, each with its own bounded queue and its own worker -pool (tuned independently, since classification is CPU-bound and enrichment is -process-/IO-bound): +The pipeline is three stages, each with its own bounded queue and its own worker +pool (tuned independently, since classification is CPU-bound, enrichment is +process-/IO-bound, and content triage is cheap IO): ``` POST /upload (multipart) @@ -35,21 +35,25 @@ process-/IO-bound): │ ┌────────────┴────────────┐ :unknown :known - │ │ move to data/known/, then - ▼ ▼ enqueue (blocking backpressure) - data/unknown/- ┌────────────────────┐ - (parked; future pipeline) │ known queue │ enrichment - └─────────┬──────────┘ - │ dequeue - ┌───────────────────┼───────────────────┐ - ▼ ▼ ▼ - known wkr 1 known wkr 2 … known wkr M - │ exiftool → normalized sidecar - success ────┴──► data/done/- - data/done/-.meta.json (sidecar-first commit) - failure ───────► data/failed/- + │ move to data/unknown/, │ move to data/known/, then + ▼ then enqueue (blocking) ▼ enqueue (blocking backpressure) + ┌────────────────────┐ ┌────────────────────┐ + │ unknown queue │ │ known queue │ enrichment + └─────────┬──────────┘ └─────────┬──────────┘ + │ dequeue │ dequeue + ┌────────┼────────┐ ┌───────────┼───────────┐ + ▼ ▼ ▼ ▼ ▼ ▼ + 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/- + └─► data/text/- data/done/-.meta.json + (sidecar-first commit) + failure ───────► data/failed/- ``` +Stages 2 (enrichment) and 3 (content triage) run in parallel: stage 1 feeds both +the known and unknown queues. + Key properties: - **Fast intake:** the queue only ever carries small references; file bytes live @@ -58,13 +62,15 @@ Key properties: 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 and leftovers - in `data/known/` re-enter enrichment (`recovered` / `recovered_known` in the - log), so a file resumes at its correct stage instead of restarting from scratch. + stage-aware: leftovers in `data/spool/` re-enter classification, `data/known/` + re-enter enrichment, and `data/unknown/` re-enter content triage (`recovered` / + `recovered_known` / `recovered_unknown` in the log), so a file resumes at its + correct stage instead of restarting from scratch. - **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 - queue) before closing the known queue and waiting out the enrich workers. + *and* unknown queues) before closing those two queues and waiting out the + enrich and content-triage workers. (See "Shutdown" below for one cosmetic caveat on SIGTERM.) - **Safe filenames:** client-supplied names are sanitized and prefixed with a server-minted UUID before touching the filesystem (no path traversal). @@ -111,6 +117,21 @@ file's presence there always implies its sidecar is already present; a crash in between leaves only a harmless orphan sidecar, and recovery re-enriches idempotently. +### Content triage (stage 3) + +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. + +The test is the classic **NUL-byte sniff** (the same heuristic `git` and +`file(1)` use): read the first 8000 bytes and, if any is NUL, call it binary, +else text. It's cheap (no full read) and reliable in practice — text encodings +don't embed NUL bytes, while binary formats almost always do near the start. An +empty file has no NUL, so it's treated as text. This is deliberately simple for +now; richer handling can hang off either bucket later (`src/content.jl`). + ## The queue seam (→ RabbitMQ later) The HTTP handler and workers only ever call `enqueue!`, `dequeue!`, and @@ -199,17 +220,21 @@ init, so the artifact is exactly regenerable from the same inputs. | `FS_QUEUE_CAPACITY` | `1000` | Max pending intake jobs before `503` | | `FS_KNOWN_WORKERS` | `nthreads()` | Stage-2 (enrichment) worker tasks | | `FS_KNOWN_QUEUE_CAPACITY` | `1000` | Max pending enrichment jobs (then backpressure) | +| `FS_UNKNOWN_WORKERS` | `nthreads()` | Stage-3 (content triage) worker tasks | +| `FS_UNKNOWN_QUEUE_CAPACITY` | `1000` | Max pending triage 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, parked for a future pool | +| `FS_UNKNOWN_DIR` | `data/unknown` | Classified-unknown, awaiting content triage | +| `FS_BINARY_DIR` | `data/binary` | Stage-3 sink: unknown files that look binary | +| `FS_TEXT_DIR` | `data/text` | Stage-3 sink: unknown files that look like text | | `FS_DONE_DIR` | `data/done` | Enriched known files (+ `.meta.json`) | | `FS_FAILED_DIR` | `data/failed` | Files whose processing threw | | `FS_MODEL_PATH` | `model/classifier.jld2` | Classifier artifact loaded at startup | | `FS_EXIFTOOL_TIMEOUT` | `30` | Seconds before a stuck exiftool is killed | -> To get real parallelism, start Julia with enough threads (`-t N`) to cover both -> pools. If `FS_WORKERS + FS_KNOWN_WORKERS` exceeds available threads you'll get a -> warning (non-fatal) and workers will share threads. +> To get real parallelism, start Julia with enough threads (`-t N`) to cover all +> pools. If `FS_WORKERS + FS_KNOWN_WORKERS + FS_UNKNOWN_WORKERS` exceeds available +> threads you'll get a warning (non-fatal) and workers will share threads. ## Usage @@ -242,7 +267,8 @@ src/ 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) - worker.jl parametrized worker loop + classify/enrich handlers + content.jl binary-vs-text sniff for unknown files (stage 3) + worker.jl parametrized worker loop + classify/enrich/triage handlers server.jl HTTP routes/handlers bin/ server.jl entry point diff --git a/src/FileServer.jl b/src/FileServer.jl index f8b5e9b..d3468e0 100644 --- a/src/FileServer.jl +++ b/src/FileServer.jl @@ -15,6 +15,7 @@ include("spool.jl") include("model.jl") # build_model() + read_features(); shared with bin/train.jl include("classify.jl") # Classifier + load_classifier/classify (needs model.jl) include("metadata.jl") # exiftool extraction + sidecar enrichment (stage 2) +include("content.jl") # binary-vs-text triage for unknown files (stage 3) include("worker.jl") # Globals the HTTP handlers read at request time. Set once in `run`, before the @@ -23,6 +24,7 @@ include("worker.jl") const CONFIG = Ref{Config}() const QUEUE = Ref{ChannelQueue}() # stage-1 (classification) queue; HTTP intake enqueues here const KNOWN_QUEUE = Ref{ChannelQueue}() # stage-2 (enrichment) queue; stage-1 workers enqueue here +const UNKNOWN_QUEUE = Ref{ChannelQueue}() # stage-3 (content triage) queue; stage-1 workers enqueue here const CLASSIFIER = Ref{Classifier}() # loaded once at startup, shared read-only across workers include("server.jl") # registers routes (references CONFIG/QUEUE at call time) @@ -70,22 +72,24 @@ function run(; overrides...) cfg = config_from_env(; overrides...) ensure_dirs(cfg) - # Both pools draw from the same OS threads. Warn on the *combined* size (still + # All pools draw from the same OS threads. Warn on the *combined* size (still # allowed): oversubscription just means tasks share threads, not a failure. - total_workers = cfg.worker_count + cfg.known_worker_count + total_workers = cfg.worker_count + cfg.known_worker_count + cfg.unknown_worker_count if total_workers > Threads.nthreads() - @warn "combined worker count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" classify_workers=cfg.worker_count known_workers=cfg.known_worker_count total=total_workers nthreads=Threads.nthreads() + @warn "combined worker count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" classify_workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count total=total_workers nthreads=Threads.nthreads() end # exiftool is a hard prerequisite for stage-2 enrichment. Fail fast at # startup rather than discover it missing on the first known file. assert_exiftool() - queue = ChannelQueue(cfg.queue_capacity) - known_queue = ChannelQueue(cfg.known_queue_capacity) - CONFIG[] = cfg - QUEUE[] = queue - KNOWN_QUEUE[] = known_queue + queue = ChannelQueue(cfg.queue_capacity) + known_queue = ChannelQueue(cfg.known_queue_capacity) + unknown_queue = ChannelQueue(cfg.unknown_queue_capacity) + CONFIG[] = cfg + QUEUE[] = queue + KNOWN_QUEUE[] = known_queue + UNKNOWN_QUEUE[] = unknown_queue # Load the classifier before serving. Fail fast: a server that silently # doesn't classify is a worse surprise than a clear startup error. @@ -93,17 +97,20 @@ function run(; overrides...) @info "loaded classifier" path=cfg.model_path # Stage-aware recovery: re-drive each stage's leftovers onto its own queue so - # files resume where they were, not from scratch. (unknown/ has no consumer - # yet, so it isn't recovered — it just accumulates for a future pool.) - recovered = recover_dir!(cfg.spool_dir, queue) - recovered_known = recover_dir!(cfg.known_dir, known_queue) - @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity recovered=recovered recovered_known=recovered_known + # 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) + @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 capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity recovered=recovered recovered_known=recovered_known recovered_unknown=recovered_unknown workers = [Threads.@spawn worker_loop(i, cfg, queue, - (job, c, wid) -> handle_classify_job(job, c, wid, known_queue)) + (job, c, wid) -> handle_classify_job(job, c, wid, known_queue, unknown_queue)) for i in 1:cfg.worker_count] known_workers = [Threads.@spawn worker_loop(i, cfg, known_queue, handle_known_job) for i in 1:cfg.known_worker_count] + unknown_workers = [Threads.@spawn worker_loop(i, cfg, unknown_queue, handle_unknown_job) + for i in 1:cfg.unknown_worker_count] register_routes() serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false) @@ -121,10 +128,12 @@ function run(; overrides...) @info "draining queues and stopping workers" terminate() # 1. stop accepting new HTTP requests close!(queue) # 2. no new classify jobs; stage-1 drains buffered - foreach(wait, workers) # 3. wait out stage-1 — the ONLY producer of the - # known queue — so nothing else will be enqueued - close!(known_queue) # 4. now safe to close stage-2's queue - foreach(wait, known_workers) # 5. wait out stage-2 + foreach(wait, workers) # 3. wait out stage-1 — the ONLY producer of BOTH the + # known and unknown queues — so nothing else enqueues + close!(known_queue) # 4. now safe to close the downstream queues + close!(unknown_queue) + foreach(wait, known_workers) # 5. wait out stage-2 and stage-3 + foreach(wait, unknown_workers) @info "shutdown complete" end atexit(drain) diff --git a/src/config.jl b/src/config.jl index 979e360..88c37ca 100644 --- a/src/config.jl +++ b/src/config.jl @@ -12,9 +12,16 @@ Base.@kwdef struct Config # the two pools are tuned independently. 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 + # of the classify and enrich pools. + unknown_worker_count::Int = Threads.nthreads() + unknown_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, parked for a future pipeline + unknown_dir::String = "data/unknown" # classified-unknown, awaiting content triage (stage 3) + binary_dir::String = "data/binary" # stage-3 sink: unknown files that look like binary data + text_dir::String = "data/text" # stage-3 sink: unknown files that look like text done_dir::String = "data/done" # fully enriched known 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 @@ -31,34 +38,42 @@ and for `FileServer.run(; port=...)`). Recognised variables: FS_HOST, FS_PORT, FS_WORKERS, FS_QUEUE_CAPACITY, FS_KNOWN_WORKERS, FS_KNOWN_QUEUE_CAPACITY, - FS_SPOOL_DIR, FS_KNOWN_DIR, FS_UNKNOWN_DIR, FS_DONE_DIR, FS_FAILED_DIR, - FS_MODEL_PATH, FS_EXIFTOOL_TIMEOUT + FS_UNKNOWN_WORKERS, FS_UNKNOWN_QUEUE_CAPACITY, + FS_SPOOL_DIR, FS_KNOWN_DIR, FS_UNKNOWN_DIR, FS_BINARY_DIR, FS_TEXT_DIR, + FS_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH, FS_EXIFTOOL_TIMEOUT """ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, queue_capacity=nothing, known_worker_count=nothing, - known_queue_capacity=nothing, spool_dir=nothing, - known_dir=nothing, unknown_dir=nothing, done_dir=nothing, + known_queue_capacity=nothing, unknown_worker_count=nothing, + unknown_queue_capacity=nothing, spool_dir=nothing, + known_dir=nothing, unknown_dir=nothing, binary_dir=nothing, + text_dir=nothing, done_dir=nothing, failed_dir=nothing, model_path=nothing, exiftool_timeout=nothing) Config( - host = something(host, get(ENV, "FS_HOST", "127.0.0.1")), - port = something(port, parse(Int, get(ENV, "FS_PORT", "8080"))), - worker_count = something(worker_count, parse(Int, get(ENV, "FS_WORKERS", string(Threads.nthreads())))), - queue_capacity = something(queue_capacity, parse(Int, get(ENV, "FS_QUEUE_CAPACITY", "1000"))), - known_worker_count = something(known_worker_count, parse(Int, get(ENV, "FS_KNOWN_WORKERS", string(Threads.nthreads())))), - known_queue_capacity = something(known_queue_capacity, parse(Int, get(ENV, "FS_KNOWN_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")), - done_dir = something(done_dir, get(ENV, "FS_DONE_DIR", "data/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")), - exiftool_timeout = something(exiftool_timeout, parse(Int, get(ENV, "FS_EXIFTOOL_TIMEOUT", "30"))), + host = something(host, get(ENV, "FS_HOST", "127.0.0.1")), + port = something(port, parse(Int, get(ENV, "FS_PORT", "8080"))), + worker_count = something(worker_count, parse(Int, get(ENV, "FS_WORKERS", string(Threads.nthreads())))), + queue_capacity = something(queue_capacity, parse(Int, get(ENV, "FS_QUEUE_CAPACITY", "1000"))), + known_worker_count = something(known_worker_count, parse(Int, get(ENV, "FS_KNOWN_WORKERS", string(Threads.nthreads())))), + known_queue_capacity = something(known_queue_capacity, parse(Int, get(ENV, "FS_KNOWN_QUEUE_CAPACITY", "1000"))), + unknown_worker_count = something(unknown_worker_count, parse(Int, get(ENV, "FS_UNKNOWN_WORKERS", string(Threads.nthreads())))), + unknown_queue_capacity = something(unknown_queue_capacity, parse(Int, get(ENV, "FS_UNKNOWN_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")), + 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")), + exiftool_timeout = something(exiftool_timeout, parse(Int, get(ENV, "FS_EXIFTOOL_TIMEOUT", "30"))), ) 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.done_dir, cfg.failed_dir) + for d in (cfg.spool_dir, cfg.known_dir, cfg.unknown_dir, cfg.binary_dir, + cfg.text_dir, cfg.done_dir, cfg.failed_dir) mkpath(d) end return nothing diff --git a/src/content.jl b/src/content.jl new file mode 100644 index 0000000..75353e0 --- /dev/null +++ b/src/content.jl @@ -0,0 +1,24 @@ +# 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. The test is the classic +# "NUL byte in the first sniff window" heuristic that git and file(1) use — cheap +# (no full read), and reliable in practice: text encodings don't embed NUL bytes, +# while binary formats almost always do near the start. + +const CONTENT_SNIFF_BYTES = 8000 + +""" + is_binary(path) -> Bool + +Classify a file as binary (`true`) or text (`false`) by sniffing its first +`CONTENT_SNIFF_BYTES` bytes for a NUL byte. An empty file has no NUL, so it is +treated as text. +""" +function is_binary(path::AbstractString)::Bool + open(path, "r") do io + chunk = read(io, CONTENT_SNIFF_BYTES) + return any(==(0x00), chunk) + end +end diff --git a/src/worker.jl b/src/worker.jl index 1d88d69..904f19d 100644 --- a/src/worker.jl +++ b/src/worker.jl @@ -1,48 +1,54 @@ # Worker tasks: pull jobs off a queue and process them. The loop scaffolding # (dequeue-until-drained, try/catch, quarantine-on-throw) is identical for every # stage, so `worker_loop` is parametrized with a `handler` and reused. Today -# there are two stages: +# there are three stages: # -# stage 1 handle_classify_job spool/ → classify → known/ (+known queue) | unknown/ +# 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/ # -# Adding a stage later (e.g. an unknown-file pool consuming unknown/) is just -# another queue + pool + handler; the loop below doesn't change. +# 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 -# known queue. Blocking backpressure: a classified file is never dropped, so the -# stage-1 worker parks until stage 2 makes room. Keeps intake decoupled — the -# HTTP path's `enqueue!` stays non-blocking; only this worker-to-worker handoff -# blocks. -const KNOWN_ENQUEUE_RETRY_SECONDS = 0.05 +# 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 """ - handle_classify_job(job, cfg, worker_id, known_queue) + handle_classify_job(job, cfg, worker_id, known_queue, unknown_queue) -Stage 1. Classify the spooled file and route it: - * `:unknown` → move to `unknown/` (parked for a future pipeline; terminal here). - * `:known` → move to `known/`, then enqueue onto the known queue for stage 2, - retrying on a full queue rather than dropping the file. +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. + +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. Sub-`MIN_FILE_BYTES` files short-circuit to `:unknown` inside `classify`. """ -function handle_classify_job(job::Job, cfg::Config, worker_id::Int, known_queue::JobQueue) +function handle_classify_job(job::Job, cfg::Config, worker_id::Int, + known_queue::JobQueue, unknown_queue::JobQueue) classification = classify(CLASSIFIER[], job.path) @info "classified file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification if classification === :known - # Move first so the file physically lives in known/ before the reference - # is visible to stage 2; then hand off. The moved path becomes the job's - # new location for the known queue. dest = move_to(cfg.known_dir, job) routed = Job(job.id, job.original_name, dest, job.size, job.received_at) while !enqueue!(known_queue, routed) - sleep(KNOWN_ENQUEUE_RETRY_SECONDS) # known queue full → back off, don't drop + sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # known queue full → back off, don't drop end @info "routed to enrichment" worker=worker_id id=job.id dest=dest else dest = move_to(cfg.unknown_dir, job) - @info "parked unknown" worker=worker_id id=job.id dest=dest + routed = Job(job.id, job.original_name, dest, job.size, job.received_at) + while !enqueue!(unknown_queue, routed) + sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # unknown queue full → back off, don't drop + end + @info "routed to content triage" worker=worker_id id=job.id dest=dest end return nothing end @@ -63,6 +69,21 @@ function handle_known_job(job::Job, cfg::Config, worker_id::Int) return nothing end +""" + handle_unknown_job(job, cfg, worker_id) + +Stage 3. Sort an unrecognized file into a coarse content bucket by sniffing its +first bytes: `binary/` if it looks like binary data, `text/` otherwise. Terminal +— there is no further stage. Simple by design for now; richer handling can hang +off either bucket later. +""" +function handle_unknown_job(job::Job, cfg::Config, worker_id::Int) + binary = is_binary(job.path) + dest = move_to(binary ? cfg.binary_dir : cfg.text_dir, job) + @info "sorted unknown" worker=worker_id id=job.id name=job.original_name kind=(binary ? :binary : :text) dest=dest + return nothing +end + """ worker_loop(worker_id, cfg, queue, handler) diff --git a/test/runtests.jl b/test/runtests.jl index 1ab4d0b..a543de9 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -7,7 +7,8 @@ using JSON3 # directly rather than only through the HTTP surface. using FileServer: Job, Config, ChannelQueue, enqueue!, dequeue!, length, sanitize_filename, recover_dir!, normalize_metadata, - build_metadata, finalize_known!, run_exiftool + build_metadata, finalize_known!, run_exiftool, + is_binary, handle_unknown_job # A minimal, valid 1×1 PNG. Lets the real-exiftool tests assert stable facts # (FileType == "PNG", 1×1 dimensions) that don't drift across exiftool versions. @@ -21,6 +22,8 @@ function tmp_config(root; kwargs...) 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"), failed_dir = joinpath(root, "failed"), kwargs..., @@ -138,6 +141,52 @@ end end end + @testset "is_binary: NUL-byte sniff" begin + mktempdir() do root + # Plain text → text. + txt = joinpath(root, "notes.txt") + write(txt, "hello, world\nsecond line\n") + @test is_binary(txt) == false + + # A NUL byte anywhere in the sniff window → binary. + bin = joinpath(root, "blob.dat") + write(bin, UInt8[0x01, 0x02, 0x00, 0x03]) + @test is_binary(bin) == true + + # Empty file has no NUL → treated as text. + empty = joinpath(root, "empty") + write(empty, UInt8[]) + @test is_binary(empty) == false + + # A NUL past the sniff window is not seen → still text. + far = joinpath(root, "far.txt") + write(far, vcat(fill(UInt8('a'), FileServer.CONTENT_SNIFF_BYTES), UInt8[0x00])) + @test is_binary(far) == false + end + end + + @testset "handle_unknown_job: routes to binary/ and text/" begin + mktempdir() do root + cfg = tmp_config(root) + + # A binary file (embedded NUL) lands in binary/. + bpath = joinpath(cfg.unknown_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) + @test isfile(joinpath(cfg.binary_dir, "id-b-blob.dat")) + @test !isfile(bpath) + + # A text file lands in text/. + tpath = joinpath(cfg.unknown_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) + @test isfile(joinpath(cfg.text_dir, "id-t-notes.log")) + @test !isfile(tpath) + end + end + @testset "recover_dir!: re-enqueues work, skips sidecars" begin mktempdir() do root dir = joinpath(root, "known"); mkpath(dir)