diff --git a/Manifest.toml b/Manifest.toml index 93ebca6..4f585ac 100644 --- a/Manifest.toml +++ b/Manifest.toml @@ -2,7 +2,7 @@ julia_version = "1.12.6" manifest_format = "2.0" -project_hash = "1d7ce552eaac13c97732dccac45c97d40527ec58" +project_hash = "a623ff56053e3a56c1799a1cb2080ec48d933b73" [[deps.ADTypes]] git-tree-sha1 = "d9aaef7c63466eee4de23b4d9dad03629df54bea" @@ -309,7 +309,7 @@ weakdeps = ["HTTP"] HTTPExt = "HTTP" [[deps.FileServer]] -deps = ["HTTP", "JSON3", "Logging", "Oxygen", "UUIDs"] +deps = ["HTTP", "JLD2", "JSON3", "Logging", "Lux", "Optimisers", "Oxygen", "UUIDs", "Zygote"] path = "." uuid = "b3f1c2d4-5e6a-4b7c-8d9e-0f1a2b3c4d5e" version = "0.1.0" @@ -468,6 +468,12 @@ weakdeps = ["Serialization"] [deps.LRUCache.extensions] SerializationExt = ["Serialization"] +[[deps.Languages]] +deps = ["InteractiveUtils", "JSON", "RelocatableFolders"] +git-tree-sha1 = "023ac3b12f82da68ed2556c71a134a03e1a11343" +uuid = "8ef0a80b-9436-5d2c-a485-80b904378c43" +version = "0.4.7" + [[deps.LibCURL]] deps = ["LibCURL_jll", "MozillaCACerts_jll"] uuid = "b27032c2-a3e7-50c8-80cd-2d36dbcbfd21" diff --git a/Project.toml b/Project.toml index f392f64..3b788a5 100644 --- a/Project.toml +++ b/Project.toml @@ -7,6 +7,7 @@ authors = ["wardjm@gmail.com"] HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3" JLD2 = "033835bb-8acc-5ee8-8aae-3f567f8a3819" JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1" +Languages = "8ef0a80b-9436-5d2c-a485-80b904378c43" Logging = "56ddb016-857b-54e1-b83d-db4d58db5568" Lux = "b2108857-7c20-44ae-9111-449ecde12c47" Optimisers = "3bd65402-5787-11e9-1adc-39752487f4e2" @@ -14,20 +15,21 @@ Oxygen = "df9a0d86-3283-4920-82dc-4555fc0d1d8b" UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4" Zygote = "e88e6eb3-aa80-5325-afca-941959d7151f" -[extras] -JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1" -Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40" - -[targets] -test = ["Test", "JSON3"] - [compat] HTTP = "1.11.0" JLD2 = "0.6.4" JSON3 = "1.14.3" +Languages = "0.4.7" Logging = "1.11.0" Lux = "1.31.4" Optimisers = "0.4.7" Oxygen = "1.10.2" UUIDs = "1.11.0" Zygote = "0.7.11" + +[extras] +JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1" +Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40" + +[targets] +test = ["Test", "JSON3"] diff --git a/README.md b/README.md index b5b2602..e007e07 100644 --- a/README.md +++ b/README.md @@ -11,9 +11,10 @@ classifier that labels it **known** (a file type resembling the training set) or ## Architecture -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): +The pipeline is four stages, each with its own bounded queue and its own worker +pool (tuned independently, since classification is CPU-bound, known-file +enrichment is process-/IO-bound, content triage is cheap IO, and language +enrichment mixes CPU with a subprocess): ``` POST /upload (multipart) @@ -46,13 +47,25 @@ process-/IO-bound, and content triage is cheap IO): 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/- + │ (terminal) data/done/-.meta.json + │ (sidecar-first commit) + │ :text move to data/text/, failure ───────► data/failed/- + ▼ then enqueue (blocking backpressure) + ┌────────────────────┐ + │ text queue │ language enrichment + └─────────┬──────────┘ + │ dequeue + ┌────────┼────────┐ + ▼ ▼ ▼ + txt 1 txt 2 … txt P + │ Languages.jl (natural language) + github-linguist (programming language) + └─► data/text_done/- + data/text_done/-.meta.json + (sidecar-first commit) ``` -Stages 2 (enrichment) and 3 (content triage) run in parallel: stage 1 feeds both -the known and unknown queues. +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. Key properties: @@ -63,14 +76,16 @@ Key properties: 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, and `data/unknown/` re-enter content triage (`recovered` / - `recovered_known` / `recovered_unknown` in the log), so a file resumes at its + 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. - **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 - *and* unknown queues) before closing those two queues and waiting out the - enrich and content-triage workers. + *and* unknown queues), then close those queues and wait out the enrich and + content-triage workers (content triage being the only producer of the text + queue), then close the text queue and wait out the language-enrichment 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). @@ -134,8 +149,55 @@ or printable-ASCII heuristics, it keeps non-ASCII text (accents, CJK, emoji) in UTF-8 near their start — still land in `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. Richer handling can hang -off either bucket later (`src/content.jl`). +for malformed bytes. An empty file is treated as text. `binary/` is terminal; +`text/` is handed to stage 4 (`src/content.jl`). + +### Language enrichment (stage 4) + +Files that stage 3 sorts as **text** are handed to a fourth pool that identifies +their language and writes a `.meta.json` sidecar, mirroring the stage-2 +known-file enrichment. Two detectors run per file: + +- **natural language** — [`Languages.jl`](https://github.com/JuliaText/Languages.jl)'s + `LanguageDetector` (a Julia port of the `whatlang` n-gram model) reads a bounded + prefix (up to `LANG_SAMPLE_BYTES`, 64 KiB) and reports the language's English + name, ISO 639-3 code, and a confidence in `[0,1]`. Pure Julia, no subprocess. + The detector is built once at startup and shared read-only across the pool. +- **programming language** — the [`github-linguist`](https://github.com/github-linguist/linguist) + CLI recognizes source and markup by extension + content heuristics (e.g. + `Python`, `Markdown`). Plain prose reports as `Text` and unrecognized content as + `null`; both collapse to *no programming language*. + +The sidecar schema: + +| field | meaning | +|---|---| +| `id`, `original_name` | from intake | +| `file_size` | bytes (authoritative, from intake) | +| `content_type` | always `"text"` | +| `language` | natural-language English name (e.g. `English`), or `null` | +| `language_code` | ISO 639-3 code (e.g. `eng`), or `null` | +| `language_confidence` | detector confidence in `[0,1]`, or `null` | +| `programming_language` | e.g. `Python`, `Markdown`, or `null` | +| `error` | set if natural-language detection produced nothing | + +> **`github-linguist` and the git-repo quirk:** run against a path *inside* a git +> repository, linguist reads the file's committed git blob, not the on-disk bytes +> — and an untracked file (which everything under `data/` is) has no blob, so it +> crashes. Stage 4 sidesteps this by copying each file to a fresh temp dir under +> `/tmp` (outside any repo, preserving the name so extension heuristics still +> fire) and pointing linguist there. +> +> Programming-language detection is **best-effort**: if `github-linguist` is +> missing (a startup warning, not a fatal error, unlike `exiftool`), fails, or +> times out (`FS_LINGUIST_TIMEOUT`, default 30s), `programming_language` is simply +> `null` and the file still completes. Natural-language detection failing produces +> a **degraded sidecar** (with an `error` note) rather than a quarantine, because +> the file is still wanted text. + +Like stage 2, the sidecar is committed **before** the file is moved into +`data/text_done/`, so the file's presence there always implies its sidecar is +present; recovery re-enriches idempotently (`src/language.jl`). ## The queue seam (→ RabbitMQ later) @@ -151,6 +213,11 @@ or worker code changes. # install deps (first time) julia --project=. -e 'using Pkg; Pkg.instantiate()' +# external tools: exiftool (stage 2, required) and github-linguist (stage 4, +# optional — programming-language detection). e.g. on Debian/Ubuntu: +# apt install libimage-exiftool-perl +# gem install github-linguist + # start the server; -t sets the number of OS threads available to workers julia --project=. -t auto bin/server.jl ``` @@ -227,19 +294,24 @@ init, so the artifact is exactly regenerable from the same inputs. | `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_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_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_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 | | `FS_MODEL_PATH` | `model/classifier.jld2` | Classifier artifact loaded at startup | | `FS_EXIFTOOL_TIMEOUT` | `30` | Seconds before a stuck exiftool is killed | +| `FS_LINGUIST_TIMEOUT` | `30` | Seconds before a stuck github-linguist is killed | > 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. +> pools. If `FS_WORKERS + FS_KNOWN_WORKERS + FS_UNKNOWN_WORKERS + FS_TEXT_WORKERS` +> exceeds available threads you'll get a warning (non-fatal) and workers will +> share threads. ## Usage @@ -273,7 +345,8 @@ src/ classify.jl load artifact + classify a file at inference time metadata.jl exiftool extraction + normalized sidecar (stage 2) content.jl binary-vs-text sniff for unknown files (stage 3) - worker.jl parametrized worker loop + classify/enrich/triage handlers + language.jl natural + programming language enrichment for text (stage 4) + worker.jl parametrized worker loop + classify/enrich/triage/language handlers server.jl HTTP routes/handlers bin/ server.jl entry point diff --git a/src/FileServer.jl b/src/FileServer.jl index d3468e0..fc232ce 100644 --- a/src/FileServer.jl +++ b/src/FileServer.jl @@ -7,6 +7,7 @@ using JSON3 using Oxygen using Lux using JLD2 +using Languages include("config.jl") include("job.jl") @@ -16,6 +17,7 @@ include("model.jl") # build_model() + read_features(); shared with bin/train 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("language.jl") # natural + programming language enrichment for text (stage 4) include("worker.jl") # Globals the HTTP handlers read at request time. Set once in `run`, before the @@ -25,7 +27,9 @@ 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 TEXT_QUEUE = Ref{ChannelQueue}() # stage-4 (language enrichment) queue; stage-3 workers enqueue here const CLASSIFIER = Ref{Classifier}() # loaded once at startup, shared read-only across workers +const DETECTOR = Ref{LanguageDetector}() # natural-language detector; built once at startup, shared read-only include("server.jl") # registers routes (references CONFIG/QUEUE at call time) @@ -74,43 +78,60 @@ function run(; overrides...) # 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 + cfg.unknown_worker_count + total_workers = cfg.worker_count + cfg.known_worker_count + cfg.unknown_worker_count + cfg.text_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 unknown_workers=cfg.unknown_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 text_workers=cfg.text_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() + # github-linguist powers stage-4 *programming*-language detection, but it's + # best-effort (natural-language enrichment stands on its own), so a missing + # binary is a warning, not a fatal error — per-file lookups degrade to none. + linguist_available() || @warn "github-linguist not found on PATH; stage-4 text files will have no programming language (install it to enable)" + queue = ChannelQueue(cfg.queue_capacity) known_queue = ChannelQueue(cfg.known_queue_capacity) unknown_queue = ChannelQueue(cfg.unknown_queue_capacity) + text_queue = ChannelQueue(cfg.text_queue_capacity) CONFIG[] = cfg QUEUE[] = queue KNOWN_QUEUE[] = known_queue UNKNOWN_QUEUE[] = unknown_queue + TEXT_QUEUE[] = text_queue # Load the classifier before serving. Fail fast: a server that silently # doesn't classify is a worse surprise than a clear startup error. CLASSIFIER[] = load_classifier(cfg.model_path) @info "loaded classifier" path=cfg.model_path + # Build the natural-language detector once (it loads the whatlang n-gram + # model) and share it read-only across the stage-4 pool, like the classifier. + 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) - @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 + 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 workers = [Threads.@spawn worker_loop(i, cfg, 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) + unknown_workers = [Threads.@spawn worker_loop(i, cfg, unknown_queue, + (job, c, wid) -> handle_unknown_job(job, c, wid, text_queue)) for i in 1:cfg.unknown_worker_count] + text_workers = [Threads.@spawn worker_loop(i, cfg, text_queue, + (job, c, wid) -> handle_text_job(job, c, wid, DETECTOR[])) + for i in 1:cfg.text_worker_count] register_routes() serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false) @@ -130,10 +151,12 @@ function run(; overrides...) close!(queue) # 2. no new classify jobs; stage-1 drains buffered 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!(known_queue) # 4. now safe to close the queues stage-1 fed close!(unknown_queue) - foreach(wait, known_workers) # 5. wait out stage-2 and stage-3 - foreach(wait, unknown_workers) + foreach(wait, known_workers) # 5. wait out stage-2 (terminal) and stage-3 — stage-3 is + foreach(wait, unknown_workers) # the ONLY producer of the text queue + close!(text_queue) # 6. now safe to close the queue stage-3 fed + foreach(wait, text_workers) # 7. wait out stage-4 @info "shutdown complete" end atexit(drain) diff --git a/src/config.jl b/src/config.jl index 88c37ca..6a7bd25 100644 --- a/src/config.jl +++ b/src/config.jl @@ -17,15 +17,22 @@ Base.@kwdef struct Config # of the classify and enrich pools. unknown_worker_count::Int = Threads.nthreads() unknown_queue_capacity::Int = 1000 + # Stage 4 (language enrichment) has its own pool + queue too: detecting a text + # file's natural language (Languages.jl) and programming language (shelling to + # 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) 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 + 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 model_path::String = "model/classifier.jld2" # committed classifier artifact, loaded at startup 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 end """ @@ -39,16 +46,20 @@ Recognised variables: FS_HOST, FS_PORT, FS_WORKERS, FS_QUEUE_CAPACITY, 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_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH, FS_EXIFTOOL_TIMEOUT + FS_DONE_DIR, FS_TEXT_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH, + FS_EXIFTOOL_TIMEOUT, FS_LINGUIST_TIMEOUT """ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, queue_capacity=nothing, known_worker_count=nothing, known_queue_capacity=nothing, unknown_worker_count=nothing, - unknown_queue_capacity=nothing, spool_dir=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, - failed_dir=nothing, model_path=nothing, exiftool_timeout=nothing) + text_dir=nothing, done_dir=nothing, text_done_dir=nothing, + failed_dir=nothing, model_path=nothing, exiftool_timeout=nothing, + linguist_timeout=nothing) Config( host = something(host, get(ENV, "FS_HOST", "127.0.0.1")), port = something(port, parse(Int, get(ENV, "FS_PORT", "8080"))), @@ -58,22 +69,26 @@ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, 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"))), + 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")), 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"))), + linguist_timeout = something(linguist_timeout, parse(Int, get(ENV, "FS_LINGUIST_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.binary_dir, - cfg.text_dir, cfg.done_dir, cfg.failed_dir) + cfg.text_dir, cfg.done_dir, cfg.text_done_dir, cfg.failed_dir) mkpath(d) end return nothing diff --git a/src/language.jl b/src/language.jl new file mode 100644 index 0000000..8272416 --- /dev/null +++ b/src/language.jl @@ -0,0 +1,155 @@ +# 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 +# `.meta.json` sidecar, exactly like the stage-2 known-file enrichment: +# +# * natural language — via Languages.jl's `LanguageDetector` (a Julia port of +# the `whatlang` n-gram model): English vs. French vs. Japanese, plus a +# confidence score. Pure Julia, no subprocess. +# * programming language — via the `github-linguist` CLI, which recognizes +# source and markup by extension + content heuristics. There is no +# comparable native Julia library, so we shell out (mirroring stage-2's +# exiftool dependency). +# +# Neither detector failing quarantines the file: a text file is wanted whether +# or not we can name its language, so a failure yields a *degraded* sidecar +# (what we know plus an `error` note), just like stage 2. +# +# The github-linguist quirk that shapes this code: run against a path *inside* a +# git repository, linguist reads the file's committed git blob, not the bytes on +# disk — and an untracked file (which every file under `data/` is) has no blob, +# so it crashes. We sidestep this by copying the file to a fresh temp dir outside +# any repo (preserving its name so linguist's extension heuristics still fire) +# and pointing linguist there. + +# How much of a text file to feed the natural-language detector. The whatlang +# model saturates quickly, so a bounded prefix keeps memory flat on huge logs +# while still giving the detector plenty of signal. +const LANG_SAMPLE_BYTES = 65_536 + +"Return true if the `github-linguist` binary is on PATH." +function linguist_available() + try + Base.run(pipeline(`github-linguist --version`; stdout=devnull, stderr=devnull)) + return true + catch + return false + end +end + +""" + read_text_sample(path) -> String + +Read up to `LANG_SAMPLE_BYTES` of `path` as UTF-8 text, trimming a multi-byte +character the window may have cut in half (reusing stage-3's `trim_truncated_utf8`) +so the tail isn't misread as garbage. +""" +function read_text_sample(path::AbstractString)::String + open(path, "r") do io + chunk = read(io, LANG_SAMPLE_BYTES) + return String(copy(trim_truncated_utf8(chunk))) + end +end + +""" + detect_natural_language(detector, text) -> (name, code, confidence) + +Run the `LanguageDetector` on `text`, returning the language's English name +(e.g. `"English"`), its ISO 639-3 code (e.g. `"eng"`), and the model's +confidence in `[0,1]`. Returns `(nothing, nothing, nothing)` when there is no +usable text (empty/whitespace) or the detector errors — the caller records that +as a degraded result rather than failing the file. +""" +function detect_natural_language(detector, text::AbstractString) + isempty(strip(text)) && return (nothing, nothing, nothing) + try + lang, _script, confidence = detector(text) + return (Languages.english_name(lang), Languages.isocode(lang), confidence) + catch + return (nothing, nothing, nothing) + end +end + +""" + run_linguist(path, timeout) -> Union{String,Nothing} + +Ask `github-linguist --json` for the programming/markup language of the file at +`path`, returning the language name (e.g. `"Python"`, `"Markdown"`) or `nothing` +when linguist can't identify one. Plain prose reports as `"Text"` and +unrecognized content as JSON `null`; both collapse to `nothing` here, since only +a real programming/markup language is worth recording. + +`path` MUST be outside any git repository — see the module header for why. +""" +function run_linguist(path::AbstractString, timeout::Integer) + bytes = run_with_timeout(`github-linguist --json $path`, timeout) + bytes === nothing && return nothing + + parsed = try + JSON3.read(String(bytes)) + catch + return nothing + end + # linguist --json emits a single object keyed by the file path; pull the one + # entry rather than depend on the exact key spelling. + isempty(parsed) && return nothing + entry = first(values(parsed)) + lang = get(entry, :language, nothing) + (lang === nothing || lang == "Text") && return nothing + return String(lang) +end + +""" + detect_programming_language(job, cfg) -> Union{String,Nothing} + +Programming/markup language of a text file, or `nothing`. Copies the file to a +throwaway temp dir *outside* the git repo — under its sanitized original name so +linguist's extension heuristics still apply — runs linguist there, and cleans up. +""" +function detect_programming_language(job::Job, cfg::Config) + lang = nothing + mktempdir() do dir # tempdir() → /tmp, outside the repo + safe = sanitize_filename(job.original_name) + tmp = joinpath(dir, safe) + cp(job.path, tmp; force=true) + lang = run_linguist(tmp, cfg.linguist_timeout) + end + return lang +end + +""" + build_text_metadata(detector, job, cfg) -> NamedTuple + +Build the stage-4 sidecar payload for a text file: its natural language (name + +ISO code + confidence) and programming/markup language, plus the Job's +authoritative id/name/size. `error` is set only when natural-language detection +produced nothing usable (the file is still enriched and committed); programming +language is best-effort and its absence is normal, not an error. +""" +function build_text_metadata(detector, job::Job, cfg::Config) + text = read_text_sample(job.path) + name, code, confidence = detect_natural_language(detector, text) + programming_language = detect_programming_language(job, cfg) + + return ( + id = job.id, + original_name = job.original_name, + file_size = job.size, # authoritative, from intake + content_type = "text", + language = name, + language_code = code, + language_confidence = confidence, + programming_language = programming_language, + error = name === nothing ? "language detection produced no result" : nothing, + ) +end + +""" + finalize_text!(cfg, job, meta) -> (file_dest, sidecar_dest) + +Commit an enriched text file (stage 4) to `text_done/` via the shared +sidecar-first `commit_enriched!`, giving text files the same crash-safe +"file implies sidecar" guarantee as stage-2 known files. +""" +finalize_text!(cfg::Config, job::Job, meta) = commit_enriched!(cfg.text_done_dir, job, meta) diff --git a/src/metadata.jl b/src/metadata.jl index 007077a..b188348 100644 --- a/src/metadata.jl +++ b/src/metadata.jl @@ -68,20 +68,20 @@ function coalesce_tag(bytag::Dict{String,Any}, tags) end """ - run_exiftool(path, timeout) -> Union{Dict{String,Any},Nothing} + run_with_timeout(cmd, timeout) -> Union{Vector{UInt8},Nothing} -Run `exiftool -json -G` on `path`, returning the parsed tag object, or `nothing` -on non-zero exit, unparseable output, or timeout. The subprocess is killed after -`timeout` seconds so one pathological file can't wedge a worker forever. +Run `cmd`, capturing stdout, and return the captured bytes on clean exit, or +`nothing` on non-zero exit or timeout. The subprocess is killed (SIGTERM, then +SIGKILL after a grace period) once it overruns `timeout` seconds, so one +pathological input can't wedge a worker forever. Shared by the exiftool (stage 2) +and github-linguist (stage 4) shells. """ -function run_exiftool(path::AbstractString, timeout::Integer) +function run_with_timeout(cmd::Cmd, timeout::Integer) out = IOBuffer() - # -json: machine output; -G: group-prefixed tags; -n: numeric (unformatted) - # values so sizes/durations are numbers, not display strings. - proc = Base.run(pipeline(`exiftool -json -G -n $path`; stdout=out, stderr=devnull); wait=false) + proc = Base.run(pipeline(cmd; stdout=out, stderr=devnull); wait=false) # Kill the process if it overruns the timeout. `t` polls rather than blocking - # so we can `kill` a hung exiftool; the poll interval bounds shutdown latency. + # so we can `kill` a hung child; the poll interval bounds shutdown latency. killed = Ref(false) t = Threads.@spawn begin waited = 0.0 @@ -104,9 +104,23 @@ function run_exiftool(path::AbstractString, timeout::Integer) wait(t) (killed[] || !success(proc)) && return nothing + return take!(out) +end + +""" + run_exiftool(path, timeout) -> Union{Dict{String,Any},Nothing} + +Run `exiftool -json -G` on `path`, returning the parsed tag object, or `nothing` +on non-zero exit, unparseable output, or timeout. +""" +function run_exiftool(path::AbstractString, timeout::Integer) + # -json: machine output; -G: group-prefixed tags; -n: numeric (unformatted) + # values so sizes/durations are numbers, not display strings. + bytes = run_with_timeout(`exiftool -json -G -n $path`, timeout) + bytes === nothing && return nothing parsed = try - JSON3.read(String(take!(out))) + JSON3.read(String(bytes)) catch return nothing end @@ -174,34 +188,44 @@ function normalize_metadata(job::Job, bytag::Dict{String,Any}) end """ - finalize_known!(cfg, job, meta) -> (file_dest, sidecar_dest) + commit_enriched!(dest_dir, job, meta) -> (file_dest, sidecar_dest) -Commit an enriched known file to `done/` with the sidecar-first ordering so the -invariant *"a file in done/ implies its sidecar is already there"* always holds. +Commit an enriched file to `dest_dir` with the sidecar-first ordering so the +invariant *"a file in dest_dir implies its sidecar is already there"* always +holds. Shared by every enrichment stage that emits a `.meta.json` sidecar +(stage-2 known files → `done/`, stage-4 text files → `text_done/`). Sequence: write `.meta.json` to a temp name, fsync its bytes, rename it -into place, fsync `done/` so the rename itself is durable, THEN move the file -into `done/`. A crash between the two leaves only a harmless orphan sidecar in -`done/` while the file stays in `known/`, so stage-aware recovery re-drives it -and overwrites the sidecar — idempotent. The fsyncs make the ordering hold -across power loss, not just process crashes. +into place, fsync `dest_dir` so the rename itself is durable, THEN move the file +into `dest_dir`. A crash between the two leaves only a harmless orphan sidecar in +`dest_dir` while the file stays in its stage dir, so stage-aware recovery +re-drives it and overwrites the sidecar — idempotent. The fsyncs make the +ordering hold across power loss, not just process crashes. """ -function finalize_known!(cfg::Config, job::Job, meta) +function commit_enriched!(dest_dir::AbstractString, job::Job, meta) base = basename(job.path) - sidecar = joinpath(cfg.done_dir, string(base, ".meta.json")) + sidecar = joinpath(dest_dir, string(base, ".meta.json")) tmp_sidecar = string(sidecar, ".tmp") - # Write to a temp name then rename, so a reader in done/ never sees a partial - # sidecar and a crash mid-write can't masquerade as a committed one. + # Write to a temp name then rename, so a reader in dest_dir never sees a + # partial sidecar and a crash mid-write can't masquerade as a committed one. open(tmp_sidecar, "w") do io write(io, JSON3.write(meta)) flush(io) fsync_fd(fd(io)) # durably persist bytes before the rename end mv(tmp_sidecar, sidecar; force=true) # sidecar committed first - fsync_dir(cfg.done_dir) # persist the rename itself, not just the bytes + fsync_dir(dest_dir) # persist the rename itself, not just the bytes - file_dest = move_to(cfg.done_dir, job) # file arrival = commit point + file_dest = move_to(dest_dir, job) # file arrival = commit point return (file_dest, sidecar) end +""" + finalize_known!(cfg, job, meta) -> (file_dest, sidecar_dest) + +Commit an enriched known file (stage 2) to `done/` via the shared sidecar-first +`commit_enriched!`. +""" +finalize_known!(cfg::Config, job::Job, meta) = commit_enriched!(cfg.done_dir, job, meta) + diff --git a/src/worker.jl b/src/worker.jl index 904f19d..563aec8 100644 --- a/src/worker.jl +++ b/src/worker.jl @@ -1,11 +1,12 @@ # 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 three stages: +# 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/ +# 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) # # Adding a stage later is just another queue + pool + handler; the loop below # doesn't change. @@ -70,17 +71,42 @@ function handle_known_job(job::Job, cfg::Config, worker_id::Int) end """ - handle_unknown_job(job, cfg, worker_id) + 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/` 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. +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). """ -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 +function handle_unknown_job(job::Job, cfg::Config, worker_id::Int, text_queue::JobQueue) + if is_binary(job.path) + dest = move_to(cfg.binary_dir, job) + @info "sorted unknown" worker=worker_id id=job.id name=job.original_name kind=:binary dest=dest + else + dest = move_to(cfg.text_dir, job) + routed = Job(job.id, job.original_name, dest, job.size, job.received_at) + while !enqueue!(text_queue, routed) + sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # text queue full → back off, don't drop + end + @info "routed to language enrichment" worker=worker_id id=job.id dest=dest + end + return nothing +end + +""" + handle_text_job(job, cfg, worker_id, detector) + +Stage 4. Enrich a text file with its natural language (via `detector`) and +programming language (via github-linguist): build the sidecar and commit both to +`text_done/` sidecar-first. Detection failure yields a *degraded* sidecar (the +file is still wanted text), so the only way to land in `failed/` is a genuine I/O +error committing — handled by `worker_loop`'s quarantine. +""" +function handle_text_job(job::Job, cfg::Config, worker_id::Int, detector) + meta = build_text_metadata(detector, job, cfg) + file_dest, sidecar = finalize_text!(cfg, job, meta) + @info "enriched text" worker=worker_id id=job.id dest=file_dest sidecar=basename(sidecar) language=meta.language confidence=meta.language_confidence programming_language=meta.programming_language degraded=(meta.error !== nothing) return nothing end diff --git a/test/runtests.jl b/test/runtests.jl index fcdad5c..59f657e 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -8,7 +8,11 @@ using JSON3 using FileServer: Job, Config, ChannelQueue, enqueue!, dequeue!, length, sanitize_filename, recover_dir!, normalize_metadata, build_metadata, finalize_known!, run_exiftool, - is_binary, handle_unknown_job + is_binary, handle_unknown_job, + detect_natural_language, run_linguist, detect_programming_language, + read_text_sample, build_text_metadata, finalize_text!, handle_text_job, + linguist_available +using Languages: LanguageDetector # 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. @@ -25,6 +29,7 @@ function tmp_config(root; kwargs...) 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"), kwargs..., ) @@ -195,25 +200,137 @@ end end end - @testset "handle_unknown_job: routes to binary/ and text/" begin + @testset "handle_unknown_job: binary terminal, text routed to stage 4" begin mktempdir() do root cfg = tmp_config(root) + text_queue = ChannelQueue(10) - # A binary file (embedded NUL) lands in binary/. + # A binary file (embedded NUL) lands in binary/ and is NOT enqueued. 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) + handle_unknown_job(bjob, cfg, 1, text_queue) @test isfile(joinpath(cfg.binary_dir, "id-b-blob.dat")) @test !isfile(bpath) + @test length(text_queue) == 0 - # A text file lands in text/. + # 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") 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")) + handle_unknown_job(tjob, cfg, 1, text_queue) + moved = joinpath(cfg.text_dir, "id-t-notes.log") + @test isfile(moved) @test !isfile(tpath) + @test length(text_queue) == 1 + routed = dequeue!(text_queue) + @test routed.id == "id-t" + @test routed.path == moved + end + end + + @testset "detect_natural_language" begin + d = LanguageDetector() + name, code, conf = detect_natural_language(d, + "The quick brown fox jumps over the lazy dog and then runs away quickly today.") + @test name == "English" + @test code == "eng" + @test conf isa Real && 0.0 <= conf <= 1.0 + + # Empty / whitespace-only text yields no result rather than throwing + # (the detector itself errors on empty input). + @test detect_natural_language(d, "") == (nothing, nothing, nothing) + @test detect_natural_language(d, " \n\t ") == (nothing, nothing, nothing) + end + + @testset "read_text_sample: bounded, UTF-8 safe" begin + mktempdir() do root + p = joinpath(root, "notes.txt") + write(p, "café — 日本語 — hello\n") + @test read_text_sample(p) == "café — 日本語 — hello\n" + + # Reads at most LANG_SAMPLE_BYTES, and doesn't choke on a multi-byte + # char straddling that boundary (trailing 'é' half-in the window). + big = joinpath(root, "big.txt") + write(big, vcat(fill(UInt8('a'), FileServer.LANG_SAMPLE_BYTES - 1), + UInt8[0xc3, 0xa9])) # 'é' split by the edge + s = read_text_sample(big) + @test Base.length(s) == FileServer.LANG_SAMPLE_BYTES - 1 # trailing half-char trimmed + @test all(==('a'), s) + end + end + + @testset "run_linguist: real detection on source vs. prose" begin + if !linguist_available() + @info "github-linguist not on PATH; skipping run_linguist tests" + else + mktempdir() do root + # A Python source file → linguist names the language. + py = joinpath(root, "script.py") + write(py, "import sys\ndef main():\n print('hi')\nmain()\n") + @test run_linguist(py, 30) == "Python" + + # Plain prose reports as "Text", which collapses to nothing. + prose = joinpath(root, "notes.txt") + write(prose, "The quarterly report shows steady growth this year.\n") + @test run_linguist(prose, 30) === nothing + end + end + end + + @testset "build_text_metadata + finalize_text!: end to end" begin + mktempdir() do root + cfg = tmp_config(root) + d = LanguageDetector() + + src = joinpath(cfg.text_dir, "id-x-script.py") + write(src, join(["# a short program in English prose comment", + "import sys", + "def greet(name):", + " print('hello ' + name + ' welcome to the show today')", + "greet('world')", ""], "\n")) + job = Job("id-x", "script.py", src, filesize(src), 0.0) + + meta = build_text_metadata(d, job, cfg) + @test meta.id == "id-x" + @test meta.content_type == "text" + @test meta.file_size == filesize(src) + @test meta.language !== nothing # some natural language detected + @test meta.error === nothing + # programming_language is best-effort; present only when linguist is. + if linguist_available() + @test meta.programming_language == "Python" + end + + file_dest, sidecar = finalize_text!(cfg, job, meta) + # File moved into text_done/, original gone from text/. + @test isfile(file_dest) + @test dirname(file_dest) == cfg.text_done_dir + @test !isfile(src) + # Sidecar committed alongside it, valid JSON, no leftover .tmp. + @test isfile(sidecar) + @test endswith(sidecar, ".meta.json") + @test !isfile(string(sidecar, ".tmp")) + parsed = JSON3.read(read(sidecar, String)) + @test parsed.content_type == "text" + @test parsed.file_size == filesize(file_dest) + end + end + + @testset "handle_text_job: enriches and commits to text_done/" begin + mktempdir() do root + cfg = tmp_config(root) + d = LanguageDetector() + + src = joinpath(cfg.text_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) + + handle_text_job(job, cfg, 1, d) + @test isfile(joinpath(cfg.text_done_dir, "id-h-readme.md")) + @test isfile(joinpath(cfg.text_done_dir, "id-h-readme.md.meta.json")) + @test !isfile(src) end end