Jeffrey Ward
e55129e3a4
Each uploaded file is scored by a fixed-structure neural net that labels it known (resembling the training set) or unknown — novelty detection over the first 16 + last 16 bytes (scaled to [0,1]), Dense(32->64->16->2), argmax. - src/model.jl: shared architecture + byte->feature mapping (trainer + server) - src/classify.jl: load committed artifact, classify a file at inference - bin/train.jl: offline trainer, 1:1 blended negatives (random + grab-bag), seeded 80/20 split, writes model/classifier.jld2 - worker: classify (annotate-only) and log classification=known|unknown - config: FS_MODEL_PATH; server fails fast if the artifact is missing - deps: Lux, JLD2, Optimisers, Zygote
49 lines
1.9 KiB
Julia
49 lines
1.9 KiB
Julia
# 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.
|
|
|
|
"""
|
|
handle_job(job, cfg, worker_id)
|
|
|
|
Do the work for a single job, then move the file to `done/`.
|
|
|
|
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.
|
|
"""
|
|
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.
|
|
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
|
|
return nothing
|
|
end
|
|
|
|
"""
|
|
worker_loop(worker_id, cfg, queue)
|
|
|
|
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.
|
|
"""
|
|
function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue)
|
|
@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)
|
|
catch e
|
|
@error "processing failed" worker=worker_id id=job.id name=job.original_name exception=(e, catch_backtrace())
|
|
try
|
|
move_to(cfg.failed_dir, job)
|
|
catch e2
|
|
@error "could not quarantine failed file" worker=worker_id id=job.id path=job.path exception=(e2, catch_backtrace())
|
|
end
|
|
end
|
|
end
|
|
@info "worker stopped" worker=worker_id
|
|
return nothing
|
|
end
|