Add stage-2 metadata enrichment pipeline for known files
Known-classified files now flow to a second queue with its own worker pool that extracts metadata via exiftool and writes a normalized JSON sidecar next to the file in done/, leaving the original bytes untouched. - Two-stage pipeline: spool/ → classify → known/ → enrich → done/; unknowns park in unknown/ as a seam for a future pool - src/metadata.jl: exiftool -json -G -n with timeout, normalized schema (file_type, mime_type, author, created_by, dimensions, ...) + raw dump; degraded sidecar on extraction failure rather than quarantine - Sidecar-first commit so a file in done/ always has its sidecar - Parametrized worker_loop with classify/enrich handlers; blocking backpressure on a full known queue (never drop a classified file) - Stage-aware recovery: spool/ and known/ resume at their correct stage - Ordered drain: close stage-1 and wait its workers (the known queue's only producer) before closing the known queue - exiftool required at startup (fail-fast); new FS_KNOWN_*/FS_UNKNOWN_DIR/ FS_EXIFTOOL_TIMEOUT config knobs; combined-pool thread warning
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
177
src/metadata.jl
Normal file
177
src/metadata.jl
Normal file
@@ -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 `<name>.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
|
||||
|
||||
19
src/spool.jl
19
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]
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user