jeff/file-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add stage-3 content triage: sort unknown files into binary/ and text/

Browse Source 
Unknown files are no longer terminal. Stage 1 now routes :unknown onto a
dedicated queue (with the same blocking backpressure as the known queue),
and a third worker pool sorts each file into data/binary/ or data/text/
using a NUL-byte sniff of the first 8000 bytes.

- content.jl: is_binary content sniff (stage 3)
- worker.jl: handle_unknown_job; stage-1 routes unknown with backpressure;
  KNOWN_ENQUEUE_RETRY_SECONDS -> ROUTE_ENQUEUE_RETRY_SECONDS (serves both)
- config.jl: unknown_worker_count/queue_capacity, binary_dir, text_dir + env
- FileServer.jl: unknown queue, pool, stage-aware recovery, drain ordering
- tests for is_binary and handle_unknown_job; tmp_config isolates new dirs
- README: three-stage pipeline
This commit is contained in:
Jeffrey Ward
2026-07-02 16:54:17 -04:00
parent 2a46f5021a
commit e42e8ef8af
6 changed files with 227 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/FileServer.jl
View File

@@ -15,6 +15,7 @@ include("spool.jl")
include("model.jl") # build_model() + read_features(); shared with bin/train.jl
include("classify.jl") # Classifier + load_classifier/classify (needs model.jl)
include("metadata.jl") # exiftool extraction + sidecar enrichment (stage 2)
include("content.jl") # binary-vs-text triage for unknown files (stage 3)
include("worker.jl")
# Globals the HTTP handlers read at request time. Set once in `run`, before the
@@ -23,6 +24,7 @@ include("worker.jl")
const CONFIG = Ref{Config}()
const QUEUE = Ref{ChannelQueue}() # stage-1 (classification) queue; HTTP intake enqueues here
const KNOWN_QUEUE = Ref{ChannelQueue}() # stage-2 (enrichment) queue; stage-1 workers enqueue here
const UNKNOWN_QUEUE = Ref{ChannelQueue}() # stage-3 (content triage) queue; stage-1 workers enqueue here
const CLASSIFIER = Ref{Classifier}() # loaded once at startup, shared read-only across workers
include("server.jl") # registers routes (references CONFIG/QUEUE at call time)
@@ -70,22 +72,24 @@ function run(; overrides...)
cfg = config_from_env(; overrides...)
ensure_dirs(cfg)
# Both pools draw from the same OS threads. Warn on the *combined* size (still
# All pools draw from the same OS threads. Warn on the *combined* size (still
# allowed): oversubscription just means tasks share threads, not a failure.
total_workers = cfg.worker_count + cfg.known_worker_count
total_workers = cfg.worker_count + cfg.known_worker_count + cfg.unknown_worker_count
if total_workers > Threads.nthreads()
@warn "combined worker count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" classify_workers=cfg.worker_count known_workers=cfg.known_worker_count total=total_workers nthreads=Threads.nthreads()
@warn "combined worker count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" classify_workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count total=total_workers nthreads=Threads.nthreads()
end
# exiftool is a hard prerequisite for stage-2 enrichment. Fail fast at
# startup rather than discover it missing on the first known file.
assert_exiftool()
queue = ChannelQueue(cfg.queue_capacity)
known_queue = ChannelQueue(cfg.known_queue_capacity)
CONFIG[] = cfg
QUEUE[] = queue
KNOWN_QUEUE[] = known_queue
queue = ChannelQueue(cfg.queue_capacity)
known_queue = ChannelQueue(cfg.known_queue_capacity)
unknown_queue = ChannelQueue(cfg.unknown_queue_capacity)
CONFIG[] = cfg
QUEUE[] = queue
KNOWN_QUEUE[] = known_queue
UNKNOWN_QUEUE[] = unknown_queue
# Load the classifier before serving. Fail fast: a server that silently
# doesn't classify is a worse surprise than a clear startup error.
@@ -93,17 +97,20 @@ function run(; overrides...)
@info "loaded classifier" path=cfg.model_path
# Stage-aware recovery: re-drive each stage's leftovers onto its own queue so
# files resume where they were, not from scratch. (unknown/ has no consumer
# yet, so it isn't recovered — it just accumulates for a future pool.)
recovered = recover_dir!(cfg.spool_dir, queue)
recovered_known = recover_dir!(cfg.known_dir, known_queue)
@info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity recovered=recovered recovered_known=recovered_known
# files resume where they were, not from scratch. spool/ → stage-1,
# known/ → stage-2, unknown/ → stage-3.
recovered = recover_dir!(cfg.spool_dir, queue)
recovered_known = recover_dir!(cfg.known_dir, known_queue)
recovered_unknown = recover_dir!(cfg.unknown_dir, unknown_queue)
@info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity recovered=recovered recovered_known=recovered_known recovered_unknown=recovered_unknown
workers = [Threads.@spawn worker_loop(i, cfg, queue,
(job, c, wid) -> handle_classify_job(job, c, wid, known_queue))
(job, c, wid) -> handle_classify_job(job, c, wid, known_queue, unknown_queue))
for i in 1:cfg.worker_count]
known_workers = [Threads.@spawn worker_loop(i, cfg, known_queue, handle_known_job)
for i in 1:cfg.known_worker_count]
unknown_workers = [Threads.@spawn worker_loop(i, cfg, unknown_queue, handle_unknown_job)
for i in 1:cfg.unknown_worker_count]
register_routes()
serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false)
@@ -121,10 +128,12 @@ function run(; overrides...)
@info "draining queues and stopping workers"
terminate() # 1. stop accepting new HTTP requests
close!(queue) # 2. no new classify jobs; stage-1 drains buffered
foreach(wait, workers) # 3. wait out stage-1 — the ONLY producer of the
# known queue — so nothing else will be enqueued
close!(known_queue) # 4. now safe to close stage-2's queue
foreach(wait, known_workers) # 5. wait out stage-2
foreach(wait, workers) # 3. wait out stage-1 — the ONLY producer of BOTH the
# known and unknown queues — so nothing else enqueues
close!(known_queue) # 4. now safe to close the downstream queues
close!(unknown_queue)
foreach(wait, known_workers) # 5. wait out stage-2 and stage-3
foreach(wait, unknown_workers)
@info "shutdown complete"
end
atexit(drain)

53
src/config.jl
View File

@@ -12,9 +12,16 @@ Base.@kwdef struct Config
# the two pools are tuned independently.
known_worker_count::Int = Threads.nthreads()
known_queue_capacity::Int = 1000
# Stage 3 (content triage) also gets its own pool + queue: sorting an
# unrecognized file into binary/ vs text/ is cheap I/O, tuned independently
# of the classify and enrich pools.
unknown_worker_count::Int = Threads.nthreads()
unknown_queue_capacity::Int = 1000
spool_dir::String = "data/spool" # files land here on intake (pending classification)
known_dir::String = "data/known" # classified-known, awaiting enrichment (stage 2)
unknown_dir::String = "data/unknown" # classified-unknown, parked for a future pipeline
unknown_dir::String = "data/unknown" # classified-unknown, awaiting content triage (stage 3)
binary_dir::String = "data/binary" # stage-3 sink: unknown files that look like binary data
text_dir::String = "data/text" # stage-3 sink: unknown files that look like text
done_dir::String = "data/done" # fully enriched known files (+ .meta.json sidecars)
failed_dir::String = "data/failed" # files move here if a worker throws
model_path::String = "model/classifier.jld2" # committed classifier artifact, loaded at startup
@@ -31,34 +38,42 @@ and for `FileServer.run(; port=...)`).
Recognised variables:
FS_HOST, FS_PORT, FS_WORKERS, FS_QUEUE_CAPACITY,
FS_KNOWN_WORKERS, FS_KNOWN_QUEUE_CAPACITY,
FS_SPOOL_DIR, FS_KNOWN_DIR, FS_UNKNOWN_DIR, FS_DONE_DIR, FS_FAILED_DIR,
FS_MODEL_PATH, FS_EXIFTOOL_TIMEOUT
FS_UNKNOWN_WORKERS, FS_UNKNOWN_QUEUE_CAPACITY,
FS_SPOOL_DIR, FS_KNOWN_DIR, FS_UNKNOWN_DIR, FS_BINARY_DIR, FS_TEXT_DIR,
FS_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH, FS_EXIFTOOL_TIMEOUT
"""
function config_from_env(; host=nothing, port=nothing, worker_count=nothing,
queue_capacity=nothing, known_worker_count=nothing,
known_queue_capacity=nothing, spool_dir=nothing,
known_dir=nothing, unknown_dir=nothing, done_dir=nothing,
known_queue_capacity=nothing, unknown_worker_count=nothing,
unknown_queue_capacity=nothing, spool_dir=nothing,
known_dir=nothing, unknown_dir=nothing, binary_dir=nothing,
text_dir=nothing, done_dir=nothing,
failed_dir=nothing, model_path=nothing, exiftool_timeout=nothing)
Config(
host = something(host, get(ENV, "FS_HOST", "127.0.0.1")),
port = something(port, parse(Int, get(ENV, "FS_PORT", "8080"))),
worker_count = something(worker_count, parse(Int, get(ENV, "FS_WORKERS", string(Threads.nthreads())))),
queue_capacity = something(queue_capacity, parse(Int, get(ENV, "FS_QUEUE_CAPACITY", "1000"))),
known_worker_count = something(known_worker_count, parse(Int, get(ENV, "FS_KNOWN_WORKERS", string(Threads.nthreads())))),
known_queue_capacity = something(known_queue_capacity, parse(Int, get(ENV, "FS_KNOWN_QUEUE_CAPACITY", "1000"))),
spool_dir = something(spool_dir, get(ENV, "FS_SPOOL_DIR", "data/spool")),
known_dir = something(known_dir, get(ENV, "FS_KNOWN_DIR", "data/known")),
unknown_dir = something(unknown_dir, get(ENV, "FS_UNKNOWN_DIR", "data/unknown")),
done_dir = something(done_dir, get(ENV, "FS_DONE_DIR", "data/done")),
failed_dir = something(failed_dir, get(ENV, "FS_FAILED_DIR", "data/failed")),
model_path = something(model_path, get(ENV, "FS_MODEL_PATH", "model/classifier.jld2")),
exiftool_timeout = something(exiftool_timeout, parse(Int, get(ENV, "FS_EXIFTOOL_TIMEOUT", "30"))),
host = something(host, get(ENV, "FS_HOST", "127.0.0.1")),
port = something(port, parse(Int, get(ENV, "FS_PORT", "8080"))),
worker_count = something(worker_count, parse(Int, get(ENV, "FS_WORKERS", string(Threads.nthreads())))),
queue_capacity = something(queue_capacity, parse(Int, get(ENV, "FS_QUEUE_CAPACITY", "1000"))),
known_worker_count = something(known_worker_count, parse(Int, get(ENV, "FS_KNOWN_WORKERS", string(Threads.nthreads())))),
known_queue_capacity = something(known_queue_capacity, parse(Int, get(ENV, "FS_KNOWN_QUEUE_CAPACITY", "1000"))),
unknown_worker_count = something(unknown_worker_count, parse(Int, get(ENV, "FS_UNKNOWN_WORKERS", string(Threads.nthreads())))),
unknown_queue_capacity = something(unknown_queue_capacity, parse(Int, get(ENV, "FS_UNKNOWN_QUEUE_CAPACITY", "1000"))),
spool_dir = something(spool_dir, get(ENV, "FS_SPOOL_DIR", "data/spool")),
known_dir = something(known_dir, get(ENV, "FS_KNOWN_DIR", "data/known")),
unknown_dir = something(unknown_dir, get(ENV, "FS_UNKNOWN_DIR", "data/unknown")),
binary_dir = something(binary_dir, get(ENV, "FS_BINARY_DIR", "data/binary")),
text_dir = something(text_dir, get(ENV, "FS_TEXT_DIR", "data/text")),
done_dir = something(done_dir, get(ENV, "FS_DONE_DIR", "data/done")),
failed_dir = something(failed_dir, get(ENV, "FS_FAILED_DIR", "data/failed")),
model_path = something(model_path, get(ENV, "FS_MODEL_PATH", "model/classifier.jld2")),
exiftool_timeout = something(exiftool_timeout, parse(Int, get(ENV, "FS_EXIFTOOL_TIMEOUT", "30"))),
)
end
"Create all the pipeline-stage directories if they don't already exist."
function ensure_dirs(cfg::Config)
for d in (cfg.spool_dir, cfg.known_dir, cfg.unknown_dir, cfg.done_dir, cfg.failed_dir)
for d in (cfg.spool_dir, cfg.known_dir, cfg.unknown_dir, cfg.binary_dir,
cfg.text_dir, cfg.done_dir, cfg.failed_dir)
mkpath(d)
end
return nothing

24
src/content.jl Normal file
View File

@@ -0,0 +1,24 @@
# Stage-3 content triage for unknown files.
#
# A file that stage-1 couldn't recognize is still sorted into one of two coarse
# buckets so downstream tooling can treat them differently: `text/` for
# human-readable content, `binary/` for everything else. The test is the classic
# "NUL byte in the first sniff window" heuristic that git and file(1) use — cheap
# (no full read), and reliable in practice: text encodings don't embed NUL bytes,
# while binary formats almost always do near the start.
const CONTENT_SNIFF_BYTES = 8000
"""
is_binary(path) -> Bool
Classify a file as binary (`true`) or text (`false`) by sniffing its first
`CONTENT_SNIFF_BYTES` bytes for a NUL byte. An empty file has no NUL, so it is
treated as text.
"""
function is_binary(path::AbstractString)::Bool
open(path, "r") do io
chunk = read(io, CONTENT_SNIFF_BYTES)
return any(==(0x00), chunk)
end
end

61
src/worker.jl
View File

@@ -1,48 +1,54 @@
# Worker tasks: pull jobs off a queue and process them. The loop scaffolding
# (dequeue-until-drained, try/catch, quarantine-on-throw) is identical for every
# stage, so `worker_loop` is parametrized with a `handler` and reused. Today
# there are two stages:
# there are three stages:
#
# stage 1 handle_classify_job spool/ → classify → known/ (+known queue) | unknown/
# stage 1 handle_classify_job spool/ → classify → known/ (+known queue) | unknown/ (+unknown queue)
# stage 2 handle_known_job known/ → exiftool enrich → done/ (+ .meta.json)
# stage 3 handle_unknown_job unknown/ → binary-vs-text sniff → binary/ | text/
#
# Adding a stage later (e.g. an unknown-file pool consuming unknown/) is just
# another queue + pool + handler; the loop below doesn't change.
# Adding a stage later is just another queue + pool + handler; the loop below
# doesn't change.
# How long a stage-1 worker backs off before retrying an enqueue onto a full
# known queue. Blocking backpressure: a classified file is never dropped, so the
# stage-1 worker parks until stage 2 makes room. Keeps intake decoupled — the
# HTTP path's `enqueue!` stays non-blocking; only this worker-to-worker handoff
# blocks.
const KNOWN_ENQUEUE_RETRY_SECONDS = 0.05
# downstream queue (known or unknown). Blocking backpressure: a classified file
# is never dropped, so the stage-1 worker parks until the next stage makes room.
# Keeps intake decoupled — the HTTP path's `enqueue!` stays non-blocking; only
# this worker-to-worker handoff blocks.
const ROUTE_ENQUEUE_RETRY_SECONDS = 0.05
"""
handle_classify_job(job, cfg, worker_id, known_queue)
handle_classify_job(job, cfg, worker_id, known_queue, unknown_queue)
Stage 1. Classify the spooled file and route it:
* `:unknown` → move to `unknown/` (parked for a future pipeline; terminal here).
* `:known` → move to `known/`, then enqueue onto the known queue for stage 2,
retrying on a full queue rather than dropping the file.
Stage 1. Classify the spooled file and route it to the next stage's queue,
retrying on a full queue rather than dropping the file:
* `:known` → move to `known/`, enqueue onto the known queue for stage 2.
* `:unknown` → move to `unknown/`, enqueue onto the unknown queue for stage 3.
In both cases move first so the file physically lives in its stage dir before the
reference is visible downstream; the moved path becomes the routed job's location.
Sub-`MIN_FILE_BYTES` files short-circuit to `:unknown` inside `classify`.
"""
function handle_classify_job(job::Job, cfg::Config, worker_id::Int, known_queue::JobQueue)
function handle_classify_job(job::Job, cfg::Config, worker_id::Int,
known_queue::JobQueue, unknown_queue::JobQueue)
classification = classify(CLASSIFIER[], job.path)
@info "classified file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification
if classification === :known
# Move first so the file physically lives in known/ before the reference
# is visible to stage 2; then hand off. The moved path becomes the job's
# new location for the known queue.
dest = move_to(cfg.known_dir, job)
routed = Job(job.id, job.original_name, dest, job.size, job.received_at)
while !enqueue!(known_queue, routed)
sleep(KNOWN_ENQUEUE_RETRY_SECONDS) # known queue full → back off, don't drop
sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # known queue full → back off, don't drop
end
@info "routed to enrichment" worker=worker_id id=job.id dest=dest
else
dest = move_to(cfg.unknown_dir, job)
@info "parked unknown" worker=worker_id id=job.id dest=dest
routed = Job(job.id, job.original_name, dest, job.size, job.received_at)
while !enqueue!(unknown_queue, routed)
sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # unknown queue full → back off, don't drop
end
@info "routed to content triage" worker=worker_id id=job.id dest=dest
end
return nothing
end
@@ -63,6 +69,21 @@ function handle_known_job(job::Job, cfg::Config, worker_id::Int)
return nothing
end
"""
handle_unknown_job(job, cfg, worker_id)
Stage 3. Sort an unrecognized file into a coarse content bucket by sniffing its
first bytes: `binary/` if it looks like binary data, `text/` otherwise. Terminal
— there is no further stage. Simple by design for now; richer handling can hang
off either bucket later.
"""
function handle_unknown_job(job::Job, cfg::Config, worker_id::Int)
binary = is_binary(job.path)
dest = move_to(binary ? cfg.binary_dir : cfg.text_dir, job)
@info "sorted unknown" worker=worker_id id=job.id name=job.original_name kind=(binary ? :binary : :text) dest=dest
return nothing
end
"""
worker_loop(worker_id, cfg, queue, handler)