diff --git a/README.md b/README.md index f5538fe..45a3a44 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,10 @@ 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): + ``` POST /upload (multipart) │ @@ -21,33 +25,92 @@ classifier that labels it **known** (a file type resembling the training set) or └────────┬─────────┘ enqueue reference (non-blocking) │ │ ▼ ▼ - 202 + job IDs ┌───────────────┐ - (503 if full) │ work queue │ bounded, thread-safe - │ (Channel-ish)│ - └───────┬───────┘ - │ dequeue - ┌───────────────┼───────────────┐ - ▼ ▼ ▼ - worker 1 worker 2 … worker N (Threads.@spawn) - │ - success ────┴──► data/done/- - failure ───────► data/failed/- + 202 + job IDs ┌────────────────────┐ + (503 if full) │ stage-1 queue │ classification + └─────────┬──────────┘ + │ dequeue + ┌───────────────────┼───────────────────┐ + ▼ ▼ ▼ + classify wkr 1 classify wkr 2 … classify wkr N + │ + ┌────────────┴────────────┐ + :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/- ``` Key properties: - **Fast intake:** the queue only ever carries small references; file bytes live on disk, so memory stays flat regardless of file size. -- **Backpressure:** the queue is bounded (default 1000). When full, uploads get - `503 Service Unavailable` instead of silently piling up. -- **Crash-resilient:** files survive on disk. On startup, anything left in - `data/spool/` is re-enqueued (`recovered = N` in the log). +- **Backpressure:** each queue is bounded (default 1000). When the *intake* queue + is full, uploads get `503 Service Unavailable`. When the *known* queue is full, + the stage-1 worker blocks and retries (a classified file is never dropped). +- **Crash-resilient:** files survive on disk. On startup, recovery is + stage-aware: leftovers in `data/spool/` re-enter classification 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. - **Graceful shutdown:** SIGINT (Ctrl-C) and SIGTERM (systemd/Docker/k8s `stop`) - both stop accepting uploads, drain the queue, wait for in-flight files to - finish, then exit. (See "Shutdown" below for one cosmetic caveat on SIGTERM.) + 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. + (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). +### Metadata enrichment (stage 2) + +Files the classifier labels **known** are handed to a second pool that extracts +metadata with [`exiftool`](https://exiftool.org/) (`exiftool -json -G -n`) — +chosen because no native Julia library comes close to its multi-format coverage. +The output is normalized into a small, stable, documented schema and written as a +JSON **sidecar** next to the file in `data/done/`, e.g. +`data/done/-.meta.json`. The original bytes are never modified. + +> **Prerequisite:** `exiftool` must be on `PATH` (e.g. `apt install +> libimage-exiftool-perl`). The server **fails fast at startup** if it's missing. + +Sidecar top-level fields (all nullable — present only when available), plus the +complete raw `exiftool` object under `raw`: + +| Field | Meaning | +|---|---| +| `id`, `original_name` | job id and client-supplied name | +| `file_type`, `mime_type` | e.g. `PDF` / `application/pdf` | +| `file_size` | bytes (authoritative, from intake — not exiftool) | +| `created_date`, `modified_date` | content timestamps | +| `author` | person (`Author`/`Artist`/`By-line`) | +| `created_by` | authoring app/tool (`Producer`/`CreatorTool`/`Creator`/`Software`/…) | +| `dimensions` | `{width, height}` for media | +| `duration` | seconds, for audio/video | +| `page_count` | for documents | +| `error` | set on a *degraded* sidecar (see below) | +| `raw` | full `exiftool` output | + +Each normalized field is a coalesce over a priority list of exiftool tags +(`src/metadata.jl`); extend a field by appending tag names. If extraction fails +or `exiftool` times out (`FS_EXIFTOOL_TIMEOUT`, default 30s), the file still +completes to `data/done/` with a **degraded sidecar** — `file_size`/`file_type` +plus an `error` note — rather than being quarantined, because it's still a wanted +known file. Only genuine I/O errors (can't write the sidecar or move the file) +send it to `data/failed/`. + +The sidecar is committed **before** the file is moved into `data/done/`, so a +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. + ## The queue seam (→ RabbitMQ later) The HTTP handler and workers only ever call `enqueue!`, `dequeue!`, and @@ -130,18 +193,23 @@ init, so the artifact is exactly regenerable from the same inputs. | Variable | Default | Meaning | |---------------------|----------------|------------------------------------------| -| `FS_HOST` | `127.0.0.1` | Bind address | -| `FS_PORT` | `8080` | Port | -| `FS_WORKERS` | `nthreads()` | Number of worker tasks | -| `FS_QUEUE_CAPACITY` | `1000` | Max pending jobs before `503` | -| `FS_SPOOL_DIR` | `data/spool` | Incoming files (pending) | -| `FS_DONE_DIR` | `data/done` | Files after successful processing | -| `FS_FAILED_DIR` | `data/failed` | Files whose processing threw | -| `FS_MODEL_PATH` | `model/classifier.jld2` | Classifier artifact loaded at startup | +| `FS_HOST` | `127.0.0.1` | Bind address | +| `FS_PORT` | `8080` | Port | +| `FS_WORKERS` | `nthreads()` | Stage-1 (classification) worker tasks | +| `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_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_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 match -> `FS_WORKERS`. If `FS_WORKERS` exceeds available threads you'll get a warning -> and workers will share threads. +> 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. ## Usage @@ -173,7 +241,8 @@ src/ spool.jl filename sanitizing, spool/move, startup recovery model.jl NN architecture + byte→feature mapping (shared with trainer) classify.jl load artifact + classify a file at inference time - worker.jl worker loop + per-job processing (classify + move) + metadata.jl exiftool extraction + normalized sidecar (stage 2) + worker.jl parametrized worker loop + classify/enrich handlers server.jl HTTP routes/handlers bin/ server.jl entry point diff --git a/src/FileServer.jl b/src/FileServer.jl index f544558..f8b5e9b 100644 --- a/src/FileServer.jl +++ b/src/FileServer.jl @@ -14,13 +14,15 @@ include("queue.jl") 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("worker.jl") # Globals the HTTP handlers read at request time. Set once in `run`, before the # server starts accepting connections. Declared after the includes above so the # `Config`/`ChannelQueue` types exist. const CONFIG = Ref{Config}() -const QUEUE = Ref{ChannelQueue}() +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 CLASSIFIER = Ref{Classifier}() # loaded once at startup, shared read-only across workers include("server.jl") # registers routes (references CONFIG/QUEUE at call time) @@ -68,23 +70,40 @@ function run(; overrides...) cfg = config_from_env(; overrides...) ensure_dirs(cfg) - if cfg.worker_count > Threads.nthreads() - @warn "worker_count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" worker_count=cfg.worker_count nthreads=Threads.nthreads() + # Both 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 + 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() end - queue = ChannelQueue(cfg.queue_capacity) - CONFIG[] = cfg - QUEUE[] = queue + # 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 # 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 - recovered = recover_spool!(cfg, queue) - @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count capacity=cfg.queue_capacity recovered=recovered + # 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 - workers = [Threads.@spawn worker_loop(i, cfg, queue) for i in 1:cfg.worker_count] + workers = [Threads.@spawn worker_loop(i, cfg, queue, + (job, c, wid) -> handle_classify_job(job, c, wid, known_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] register_routes() serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false) @@ -99,10 +118,13 @@ function run(; overrides...) drained = Threads.Atomic{Bool}(false) function drain() Threads.atomic_xchg!(drained, true) && return # run at most once - @info "draining queue and stopping workers" - terminate() # stop accepting new HTTP requests - close!(queue) # let workers drain buffered jobs, then exit - foreach(wait, workers) + @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 @info "shutdown complete" end atexit(drain) diff --git a/src/config.jl b/src/config.jl index 46e2bdc..979e360 100644 --- a/src/config.jl +++ b/src/config.jl @@ -7,10 +7,18 @@ Base.@kwdef struct Config port::Int = 8080 worker_count::Int = Threads.nthreads() queue_capacity::Int = 1000 - spool_dir::String = "data/spool" # files land here on intake (pending) - done_dir::String = "data/done" # files move here after successful processing + # Stage 2 (enrichment) has its own pool + queue: exiftool work is process-spawn + # and I/O bound, a different cost profile than the CPU-bound Lux classify, so + # the two pools are tuned independently. + known_worker_count::Int = Threads.nthreads() + known_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 + 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 + exiftool_timeout::Int = 30 # seconds before a stuck exiftool is killed → degraded sidecar end """ @@ -22,26 +30,35 @@ and for `FileServer.run(; port=...)`). Recognised variables: FS_HOST, FS_PORT, FS_WORKERS, FS_QUEUE_CAPACITY, - FS_SPOOL_DIR, FS_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH + 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 """ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, - queue_capacity=nothing, spool_dir=nothing, - done_dir=nothing, failed_dir=nothing, model_path=nothing) + queue_capacity=nothing, known_worker_count=nothing, + known_queue_capacity=nothing, spool_dir=nothing, + known_dir=nothing, unknown_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"))), - spool_dir = something(spool_dir, get(ENV, "FS_SPOOL_DIR", "data/spool")), - 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")), + 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"))), ) end -"Create the spool/done/failed directories if they don't already exist." +"Create all the pipeline-stage directories if they don't already exist." function ensure_dirs(cfg::Config) - for d in (cfg.spool_dir, cfg.done_dir, cfg.failed_dir) + for d in (cfg.spool_dir, cfg.known_dir, cfg.unknown_dir, cfg.done_dir, cfg.failed_dir) mkpath(d) end return nothing diff --git a/src/metadata.jl b/src/metadata.jl new file mode 100644 index 0000000..a9ad9e2 --- /dev/null +++ b/src/metadata.jl @@ -0,0 +1,177 @@ +# Stage-2 metadata extraction and enrichment. +# +# Known files are enriched by shelling out to `exiftool -json -G` (the only tool +# with broad, multi-format coverage — there is no comparable native Julia +# library), then normalizing its output into a small, stable, documented schema +# that downstream consumers can rely on, while preserving the full raw dump. +# +# exiftool being installed is a hard startup prerequisite (see `assert_exiftool`, +# called from `run`). A per-file extraction failure or hang does NOT quarantine +# the file — it produces a *degraded* sidecar recording what we know plus the +# error, because a file that passed classification is wanted regardless of +# whether we could read its metadata. + +"Throw at startup if the `exiftool` binary isn't on PATH — fail fast rather than discover it per file." +function assert_exiftool() + try + Base.run(pipeline(`exiftool -ver`; stdout=devnull, stderr=devnull)) + catch + error("exiftool not found on PATH (install libimage-exiftool-perl / exiftool). It is required for stage-2 metadata enrichment.") + end + return nothing +end + +# Each normalized field is a coalesce over exiftool tag names, tried in order; +# the first present, non-empty value wins. exiftool with `-G` prefixes tags by +# group (e.g. "EXIF:Software"), so we match on the bare tag name after the last +# colon. Extend a field simply by appending tag names here. +# +# Note the documented `Creator` ambiguity: in PDF it's the authoring app, but +# elsewhere it's the person. We take the simple route — `Creator` feeds +# `created_by` only, and `author` relies on the person-specific tags. +const CREATED_BY_TAGS = ["Producer", "CreatorTool", "Creator", "Software", "Application", "Encoder", "EncodingTool", "HostComputer"] +const AUTHOR_TAGS = ["Author", "Artist", "By-line", "Owner", "Artist"] +const CREATED_DATE_TAGS = ["DateTimeOriginal", "CreateDate", "MediaCreateDate", "CreationDate"] +const MODIFIED_DATE_TAGS = ["ModifyDate", "FileModifyDate"] + +"Strip exiftool's `-G` group prefix (`EXIF:Software` → `Software`) so lookups are group-agnostic." +strip_group(tag::AbstractString) = String(last(split(tag, ':'))) + +"Return the first present, non-empty value among `tags` in the group-stripped map, or `nothing`." +function coalesce_tag(bytag::Dict{String,Any}, tags) + for t in tags + v = get(bytag, t, nothing) + v === nothing && continue + s = string(v) + isempty(strip(s)) && continue + return v + end + return nothing +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. The subprocess is killed after +`timeout` seconds so one pathological file can't wedge a worker forever. +""" +function run_exiftool(path::AbstractString, 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) + + # 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. + killed = Ref(false) + t = Threads.@spawn begin + waited = 0.0 + while process_running(proc) && waited < timeout + sleep(0.1); waited += 0.1 + end + if process_running(proc) + killed[] = true + kill(proc) + end + end + wait(proc) + wait(t) + + (killed[] || !success(proc)) && return nothing + + parsed = try + JSON3.read(String(take!(out))) + catch + return nothing + end + # exiftool -json emits a one-element array of objects (one per input file). + (parsed isa AbstractVector && !isempty(parsed)) || return nothing + return Dict{String,Any}(String(strip_group(String(k))) => v for (k, v) in pairs(parsed[1])) +end + +""" + build_metadata(job, cfg) -> NamedTuple + +Extract and normalize metadata for a known file. Always returns a sidecar +payload: on extraction success, the normalized fields plus the full raw dump; on +failure/timeout, a *degraded* payload with what we know from the Job plus an +`error` note. `file_size` always comes from the Job (authoritative), never +exiftool. +""" +function build_metadata(job::Job, cfg::Config) + bytag = run_exiftool(job.path, cfg.exiftool_timeout) + + if bytag === nothing + return ( + id = job.id, + original_name = job.original_name, + file_type = nothing, + mime_type = nothing, + file_size = job.size, + created_date = nothing, + modified_date = nothing, + author = nothing, + created_by = nothing, + dimensions = nothing, + duration = nothing, + page_count = nothing, + error = "exiftool extraction failed or timed out", + raw = nothing, + ) + end + + return normalize_metadata(job, bytag) +end + +"Build the normalized sidecar payload from a successful exiftool tag map." +function normalize_metadata(job::Job, bytag::Dict{String,Any}) + w = get(bytag, "ImageWidth", nothing) + h = get(bytag, "ImageHeight", nothing) + dims = (w !== nothing && h !== nothing) ? (; width = w, height = h) : nothing + + return ( + id = job.id, + original_name = job.original_name, + file_type = get(bytag, "FileType", get(bytag, "FileTypeExtension", nothing)), + mime_type = get(bytag, "MIMEType", nothing), + file_size = job.size, # authoritative, from intake + created_date = coalesce_tag(bytag, CREATED_DATE_TAGS), + modified_date = coalesce_tag(bytag, MODIFIED_DATE_TAGS), + author = coalesce_tag(bytag, AUTHOR_TAGS), + created_by = coalesce_tag(bytag, CREATED_BY_TAGS), + dimensions = dims, + duration = get(bytag, "Duration", get(bytag, "MediaDuration", nothing)), + page_count = get(bytag, "PageCount", nothing), + error = nothing, + raw = bytag, + ) +end + +""" + finalize_known!(cfg, 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. + +Sequence: write `.meta.json` directly into `done/`, fsync-close it, 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. +""" +function finalize_known!(cfg::Config, job::Job, meta) + base = basename(job.path) + sidecar = joinpath(cfg.done_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. + open(tmp_sidecar, "w") do io + write(io, JSON3.write(meta)) + end + mv(tmp_sidecar, sidecar; force=true) # sidecar committed first + + file_dest = move_to(cfg.done_dir, job) # file arrival = commit point + return (file_dest, sidecar) +end + diff --git a/src/spool.jl b/src/spool.jl index 823ac61..ea8756d 100644 --- a/src/spool.jl +++ b/src/spool.jl @@ -42,17 +42,22 @@ end const UUID_LEN = 36 """ - recover_spool!(cfg, queue) -> Int + recover_dir!(dir, queue) -> Int -Re-enqueue any files already sitting in the spool directory (left by a crash, -a hard shutdown, or an intake that never got processed). This is the payoff of -spooling to disk: a restart resumes work instead of stranding it. Returns the -number of files recovered. +Re-enqueue any files sitting in `dir` (left by a crash, hard shutdown, or an +intake that never finished) onto `queue`. This is the payoff of spooling to +disk: a restart resumes work instead of stranding it. Stage-aware recovery uses +one call per stage — `spool/` → stage-1 queue, `known/` → known queue — so each +file re-enters at the correct stage rather than being reclassified from scratch. +Returns the number of files recovered. + +Skips `.meta.json` sidecars: those are stage-2 output, not work to redo. """ -function recover_spool!(cfg::Config, queue::JobQueue)::Int +function recover_dir!(dir::AbstractString, queue::JobQueue)::Int n = 0 - for path in sort(readdir(cfg.spool_dir; join=true)) + for path in sort(readdir(dir; join=true)) isfile(path) || continue + endswith(path, ".meta.json") && continue # sidecar, not a work item fname = basename(path) if length(fname) > UUID_LEN + 1 id = fname[1:UUID_LEN] diff --git a/src/worker.jl b/src/worker.jl index f39dc71..1d88d69 100644 --- a/src/worker.jl +++ b/src/worker.jl @@ -1,39 +1,82 @@ -# Worker task: pull jobs off the queue and process them. One of these runs per -# configured worker, each as its own `Threads.@spawn`'d task. +# 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: +# +# stage 1 handle_classify_job spool/ → classify → known/ (+known queue) | unknown/ +# stage 2 handle_known_job known/ → exiftool enrich → done/ (+ .meta.json) +# +# Adding a stage later (e.g. an unknown-file pool consuming unknown/) 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 """ - handle_job(job, cfg, worker_id) + handle_classify_job(job, cfg, worker_id, known_queue) -Do the work for a single job, then move the file to `done/`. +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. -For now the "work" is just logging the received filename to prove the flow — -this is the seam where real heavy-lifting will go later. +Sub-`MIN_FILE_BYTES` files short-circuit to `:unknown` inside `classify`. """ -function handle_job(job::Job, cfg::Config, worker_id::Int) - # Classify the spooled file (annotate-only for now: the result is logged but - # every file still moves to done/ regardless of known/unknown). Sub-32-byte - # files short-circuit to :unknown inside classify without touching the model. +function handle_classify_job(job::Job, cfg::Config, worker_id::Int, known_queue::JobQueue) classification = classify(CLASSIFIER[], job.path) - @info "received file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification - dest = move_to(cfg.done_dir, job) - @info "completed" worker=worker_id id=job.id dest=dest + @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 + 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 + end return nothing end """ - worker_loop(worker_id, cfg, queue) + handle_known_job(job, cfg, worker_id) -Consume jobs until the queue is closed and drained. A failure on one job is -logged and the file is quarantined in `failed/` — it must never kill the -worker, or the pool would silently shrink. +Stage 2. Extract metadata (exiftool, with timeout) and enrich: build the +normalized sidecar and commit both to `done/` sidecar-first. Extraction +failure/timeout yields a *degraded* sidecar (the file is still a wanted known +file), so the only way to land in `failed/` is a genuine I/O error writing the +sidecar or moving the file — handled by `worker_loop`'s quarantine. """ -function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue) +function handle_known_job(job::Job, cfg::Config, worker_id::Int) + meta = build_metadata(job, cfg) + file_dest, sidecar = finalize_known!(cfg, job, meta) + @info "enriched" worker=worker_id id=job.id dest=file_dest sidecar=basename(sidecar) file_type=meta.file_type created_by=meta.created_by degraded=(meta.error !== nothing) + return nothing +end + +""" + worker_loop(worker_id, cfg, queue, handler) + +Consume jobs from `queue` until it is closed and drained, running `handler` on +each. A failure on one job is logged and the file quarantined in `failed/` — it +must never kill the worker, or the pool would silently shrink. +""" +function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue, handler) @info "worker started" worker=worker_id while true job = dequeue!(queue) job === nothing && break # queue closed and drained → exit try - handle_job(job, cfg, worker_id) + handler(job, cfg, worker_id) catch e @error "processing failed" worker=worker_id id=job.id name=job.original_name exception=(e, catch_backtrace()) try