commit a6dbcaef8ba9b1cf7bfbc2609b88b6509e8dbd85 Author: Jeffrey Ward Date: Thu Jul 2 10:53:39 2026 -0400 Initial file-ingestion service REST endpoint (Oxygen.jl POST /upload, multipart) that spools uploaded files to disk, enqueues lightweight references onto a bounded thread-safe work queue, and hands off immediately (202 + job IDs; 503 when full). A configurable pool of worker threads pulls jobs off the queue, logs the received filename (placeholder for real processing), and moves files to done/ on success or failed/ on error. - Queue behind an enqueue!/dequeue!/close! seam for a future RabbitMQ swap - Startup recovery: re-enqueues leftover files in spool/ - Graceful drain on SIGINT and SIGTERM (via atexit) - Env-var config; filenames sanitized + UUID-prefixed on disk Co-Authored-By: Claude Opus 4.8 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8300306 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +/data/ +Manifest.toml diff --git a/Project.toml b/Project.toml new file mode 100644 index 0000000..eea2254 --- /dev/null +++ b/Project.toml @@ -0,0 +1,18 @@ +name = "FileServer" +uuid = "b3f1c2d4-5e6a-4b7c-8d9e-0f1a2b3c4d5e" +version = "0.1.0" +authors = ["wardjm@gmail.com"] + +[deps] +HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3" +JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1" +Logging = "56ddb016-857b-54e1-b83d-db4d58db5568" +Oxygen = "df9a0d86-3283-4920-82dc-4555fc0d1d8b" +UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4" + +[compat] +HTTP = "1.11.0" +JSON3 = "1.14.3" +Logging = "1.11.0" +Oxygen = "1.10.2" +UUIDs = "1.11.0" diff --git a/README.md b/README.md new file mode 100644 index 0000000..a3ade6b --- /dev/null +++ b/README.md @@ -0,0 +1,133 @@ +# FileServer + +A minimal Julia service that receives files over HTTP and hands them off to a +pool of worker threads for processing. The HTTP endpoint does no real work: it +spools each uploaded file to disk, pushes a lightweight reference onto a work +queue, and responds immediately — staying free to accept the next upload. + +Right now the "processing" is just logging the received filename, to prove the +flow. That's the seam where real heavy-lifting goes later. + +## Architecture + +``` + POST /upload (multipart) + │ + ▼ + ┌─────────────────┐ spool bytes to disk + │ HTTP handler │────────────────────────► data/spool/- + │ (Oxygen.jl) │ + └────────┬─────────┘ 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/- +``` + +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). +- **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.) +- **Safe filenames:** client-supplied names are sanitized and prefixed with a + server-minted UUID before touching the filesystem (no path traversal). + +## The queue seam (→ RabbitMQ later) + +The HTTP handler and workers only ever call `enqueue!`, `dequeue!`, and +`close!` on a `JobQueue` (see `src/queue.jl`). Today that's an in-process +`ChannelQueue`. To move to RabbitMQ (or any broker), implement a new `JobQueue` +subtype with those three methods and swap the construction in `run` — no handler +or worker code changes. + +## Running + +```bash +# install deps (first time) +julia --project=. -e 'using Pkg; Pkg.instantiate()' + +# start the server; -t sets the number of OS threads available to workers +julia --project=. -t auto bin/server.jl +``` + +## Shutdown + +Both SIGINT and SIGTERM trigger the same idempotent graceful drain +(stop serving → close queue → wait for workers → exit): + +- **SIGINT** is caught as an `InterruptException` (we call + `Base.exit_on_sigint(false)`), so shutdown is clean and quiet. +- **SIGTERM** can't be intercepted directly — Julia blocks it on worker threads + and handles it in its own runtime, so a user `signal()` handler never fires. + Instead we hook the drain into an `atexit` handler, which Julia's SIGTERM path + does run. Caveat: Julia prints its own `signal 15: Terminated` backtrace + *before* `atexit` runs. It's harmless noise — the drain still completes right + after it — but if you want a fully quiet stop under a process manager, + configure it to send SIGINT instead (systemd: `KillSignal=SIGINT`; Docker: + `STOPSIGNAL SIGINT`). Give the stop timeout enough headroom to drain + in-flight work (systemd: `TimeoutStopSec`). + +## Configuration (environment variables) + +| 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 | + +> 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. + +## Usage + +```bash +# health check +curl http://127.0.0.1:8080/health +# {"status":"ok"} + +# upload one or more files (multipart/form-data) +curl -F "a=@report.pdf" -F "b=@data.csv" http://127.0.0.1:8080/upload +# 202 {"accepted":[{"id":"","name":"report.pdf"}, ...]} +``` + +Each file in a request becomes its own job. Responses: + +- `202 Accepted` — all files spooled and queued (with per-file job IDs) +- `400 Bad Request` — not multipart, or no files present +- `503 Service Unavailable` — queue full, retry later +- `500 Internal Server Error` — failed to write a file to disk + +## Layout + +``` +src/ + FileServer.jl module + run() (startup, recovery, workers, serve, shutdown) + config.jl Config struct + env parsing + job.jl Job (the queue reference) + queue.jl JobQueue seam + in-process ChannelQueue + spool.jl filename sanitizing, spool/move, startup recovery + worker.jl worker loop + per-job processing (placeholder) + server.jl HTTP routes/handlers +bin/ + server.jl entry point +``` diff --git a/bin/server.jl b/bin/server.jl new file mode 100644 index 0000000..2332796 --- /dev/null +++ b/bin/server.jl @@ -0,0 +1,4 @@ +#!/usr/bin/env julia +# Entry point. Run with: julia --project -t auto bin/server.jl +using FileServer +FileServer.run() diff --git a/src/FileServer.jl b/src/FileServer.jl new file mode 100644 index 0000000..49010c5 --- /dev/null +++ b/src/FileServer.jl @@ -0,0 +1,92 @@ +module FileServer + +using Logging +using UUIDs +using HTTP +using JSON3 +using Oxygen + +include("config.jl") +include("job.jl") +include("queue.jl") +include("spool.jl") +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}() + +include("server.jl") # registers routes (references CONFIG/QUEUE at call time) + +export run + +""" + run(; overrides...) + +Start the file server: build config, ensure directories, recover any leftover +spooled files, spawn the worker pool, then serve HTTP until interrupted +(Ctrl-C / SIGINT or SIGTERM). On shutdown it stops accepting uploads, drains the +queue, waits for workers to finish in-flight files, and exits cleanly. + +Keyword `overrides` (e.g. `port=9000`) take precedence over environment vars. +""" +function run(; overrides...) + # By default a Julia script exits immediately on SIGINT (Ctrl-C). Disable + # that so the interrupt surfaces as a catchable InterruptException, letting + # us drain the queue gracefully below. + Base.exit_on_sigint(false) + + 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() + end + + queue = ChannelQueue(cfg.queue_capacity) + CONFIG[] = cfg + QUEUE[] = queue + + 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 + + workers = [Threads.@spawn worker_loop(i, cfg, queue) for i in 1:cfg.worker_count] + + register_routes() + serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false) + + # Idempotent graceful drain: stop accepting uploads, let workers finish the + # buffered jobs, then exit. Called from two places: + # * the `finally` below, for SIGINT (Ctrl-C) and normal return, and + # * an `atexit` hook, for SIGTERM (systemd/Docker/k8s `stop`). + # We can't intercept SIGTERM directly — Julia blocks it on worker threads and + # handles it in its own runtime, so a user signal() handler never fires. But + # Julia's SIGTERM path runs `atexit` hooks, which gives us a reliable seam. + 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 "shutdown complete" + end + atexit(drain) + + try + while true + sleep(0.5) # interruptible; SIGINT throws in here + end + catch e + e isa InterruptException || rethrow(e) + @info "shutdown requested (SIGINT)" + finally + drain() + end + + return nothing +end + +end # module diff --git a/src/config.jl b/src/config.jl new file mode 100644 index 0000000..6e4a237 --- /dev/null +++ b/src/config.jl @@ -0,0 +1,46 @@ +# Runtime configuration. Built once at startup, then treated as immutable. +# Every knob has a default so the service runs with zero configuration, and +# every knob can be overridden by an environment variable (see `config_from_env`). + +Base.@kwdef struct Config + host::String = "127.0.0.1" + 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 + failed_dir::String = "data/failed" # files move here if a worker throws +end + +""" + config_from_env(; overrides...) + +Build a `Config` from environment variables, falling back to the struct +defaults. Any keyword `overrides` win over the environment (useful for tests +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 +""" +function config_from_env(; host=nothing, port=nothing, worker_count=nothing, + queue_capacity=nothing, spool_dir=nothing, + done_dir=nothing, failed_dir=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")), + ) +end + +"Create the spool/done/failed directories if they don't already exist." +function ensure_dirs(cfg::Config) + for d in (cfg.spool_dir, cfg.done_dir, cfg.failed_dir) + mkpath(d) + end + return nothing +end diff --git a/src/job.jl b/src/job.jl new file mode 100644 index 0000000..9b519e5 --- /dev/null +++ b/src/job.jl @@ -0,0 +1,13 @@ +# A unit of work on the queue. Deliberately lightweight: the file *bytes* live +# on disk in the spool directory, and only this small reference travels through +# the queue. This is what keeps intake fast and memory flat regardless of file +# size, and it's the shape you'd publish to RabbitMQ later (the "claim check" +# pattern — enqueue a reference, not the payload). + +struct Job + id::String # server-minted UUID; also the on-disk filename prefix + original_name::String # client-supplied name, kept for display/logging only + path::String # absolute-ish path to the spooled file + size::Int # bytes + received_at::Float64 # time() at intake +end diff --git a/src/queue.jl b/src/queue.jl new file mode 100644 index 0000000..d816c88 --- /dev/null +++ b/src/queue.jl @@ -0,0 +1,92 @@ +# The queue seam. +# +# The rest of the app only ever calls `enqueue!`, `dequeue!`, and `close!`. +# Today those are backed by an in-process, bounded, thread-safe buffer +# (the Go-channel / Julia-`Channel` model). To move to RabbitMQ (or any broker) +# later, implement a new `JobQueue` subtype with these three methods and swap +# the construction in `run` — no HTTP handler or worker code needs to change. + +abstract type JobQueue end + +""" + enqueue!(q, job) -> Bool + +Non-blocking. Returns `true` if the job was accepted, `false` if the queue is +full (the HTTP layer turns `false` into a 503) or closed. Never blocks the +calling request thread. +""" +function enqueue! end + +""" + dequeue!(q) -> Union{Job,Nothing} + +Blocks until a job is available and returns it. Returns `nothing` only when the +queue has been closed *and* fully drained — the signal for a worker to exit. +""" +function dequeue! end + +""" + close!(q) + +Mark the queue closed and wake all waiting workers. Buffered jobs are still +handed out (drain-then-exit); no new jobs are accepted. +""" +function close! end + +# --- In-process bounded implementation -------------------------------------- + +mutable struct ChannelQueue <: JobQueue + const buffer::Vector{Job} + const capacity::Int + const cond::Threads.Condition # its internal lock guards `buffer` + `closed` + closed::Bool +end + +ChannelQueue(capacity::Integer) = + ChannelQueue(Job[], Int(capacity), Threads.Condition(), false) + +function enqueue!(q::ChannelQueue, job::Job)::Bool + lock(q.cond) + try + (q.closed || length(q.buffer) >= q.capacity) && return false + push!(q.buffer, job) + notify(q.cond) # wake a waiting worker (notify wakes all by default) + return true + finally + unlock(q.cond) + end +end + +function dequeue!(q::ChannelQueue)::Union{Job,Nothing} + lock(q.cond) + try + while isempty(q.buffer) + q.closed && return nothing # closed and drained → tell worker to stop + wait(q.cond) # releases lock while parked + end + return popfirst!(q.buffer) + finally + unlock(q.cond) + end +end + +function close!(q::ChannelQueue) + lock(q.cond) + try + q.closed = true + notify(q.cond) # wake every parked worker so they can drain/exit + finally + unlock(q.cond) + end + return nothing +end + +"Number of jobs currently buffered (for logging/introspection)." +function Base.length(q::ChannelQueue) + lock(q.cond) + try + return length(q.buffer) + finally + unlock(q.cond) + end +end diff --git a/src/server.jl b/src/server.jl new file mode 100644 index 0000000..d00b7ac --- /dev/null +++ b/src/server.jl @@ -0,0 +1,62 @@ +# HTTP layer: a single multipart upload endpoint plus a health check. +# +# The handler's whole job is to get files onto the queue fast and get out of the +# way: spool each uploaded file to disk, enqueue a reference, respond 202. It +# never does real processing — that's the workers' job. +# +# NOTE: routes are registered at runtime via `register_routes()` (called from +# `run`), NOT with top-level macros. In a precompiled package, top-level +# `@get`/`@post` would execute during precompilation and be lost before serving. + +jsonresp(status::Int, data) = + HTTP.Response(status, ["Content-Type" => "application/json"], JSON3.write(data)) + +function health_handler(_::HTTP.Request) + return jsonresp(200, (; status = "ok")) +end + +function upload_handler(req::HTTP.Request) + cfg = CONFIG[] + queue = QUEUE[] + + parts = try + HTTP.parse_multipart_form(req) + catch + nothing + end + parts === nothing && + return jsonresp(400, (; error = "expected multipart/form-data")) + + files = filter(p -> p.filename !== nothing && !isempty(p.filename), parts) + isempty(files) && + return jsonresp(400, (; error = "no files found in request")) + + accepted = NamedTuple{(:id, :name),Tuple{String,String}}[] + for p in files + bytes = read(p.data) + + job = try + spool_file(cfg, p.filename, bytes) + catch e + @error "spool failed" name=p.filename exception=(e, catch_backtrace()) + return jsonresp(500, (; error = "failed to store file", accepted)) + end + + if !enqueue!(queue, job) + rm(job.path; force = true) # never queued → don't leave it in spool + return jsonresp(503, (; error = "queue full, retry later", accepted)) + end + + @info "accepted" id=job.id name=job.original_name size=job.size + push!(accepted, (; id = job.id, name = job.original_name)) + end + + return jsonresp(202, (; accepted)) +end + +"Register HTTP routes on the Oxygen instance. Must run at runtime, before serve." +function register_routes() + @get("/health", health_handler) + @post("/upload", upload_handler) + return nothing +end diff --git a/src/spool.jl b/src/spool.jl new file mode 100644 index 0000000..823ac61 --- /dev/null +++ b/src/spool.jl @@ -0,0 +1,69 @@ +# Disk lifecycle: sanitize names, spool bytes to disk, move processed files, +# and recover leftover files on startup. + +const MAX_NAME_LEN = 100 + +""" + sanitize_filename(name) -> String + +Turn an untrusted, client-supplied filename into something safe to place in a +path. Strips directory components, replaces anything outside a conservative +charset, removes leading dots (so `..` and dotfiles can't sneak through), caps +the length, and falls back to `"unnamed"` if nothing usable remains. +""" +function sanitize_filename(name::AbstractString)::String + base = basename(String(name)) # drop any path components + base = replace(base, r"[^A-Za-z0-9._-]" => "_") # ASCII-only safe charset + base = lstrip(base, '.') # kill "..", ".hidden", etc. + isempty(base) && (base = "unnamed") + return first(base, MAX_NAME_LEN) +end + +"Write `bytes` to the spool dir under `-` and return the Job." +function spool_file(cfg::Config, original_name::AbstractString, bytes::Vector{UInt8})::Job + id = string(uuid4()) + safe = sanitize_filename(original_name) + path = joinpath(cfg.spool_dir, string(id, "-", safe)) + open(path, "w") do io + write(io, bytes) + end + return Job(id, String(original_name), path, length(bytes), time()) +end + +"Move a spooled file into `dir` (done/ or failed/), returning the destination." +function move_to(dir::AbstractString, job::Job)::String + dest = joinpath(dir, basename(job.path)) + mv(job.path, dest; force=true) + return dest +end + +# The UUID string produced by `uuid4()` is always 36 chars, followed by '-', +# then the sanitized name. That fixed width lets us split reliably on recovery. +const UUID_LEN = 36 + +""" + recover_spool!(cfg, 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. +""" +function recover_spool!(cfg::Config, queue::JobQueue)::Int + n = 0 + for path in sort(readdir(cfg.spool_dir; join=true)) + isfile(path) || continue + fname = basename(path) + if length(fname) > UUID_LEN + 1 + id = fname[1:UUID_LEN] + name = fname[(UUID_LEN + 2):end] # skip the '-' + else + id = string(uuid4()) # unexpected name; give it an id + name = fname + end + job = Job(id, name, path, filesize(path), time()) + enqueue!(queue, job) || @warn "queue full during recovery; leaving file for next start" path + n += 1 + end + return n +end diff --git a/src/worker.jl b/src/worker.jl new file mode 100644 index 0000000..289b930 --- /dev/null +++ b/src/worker.jl @@ -0,0 +1,46 @@ +# 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) + # --- placeholder for real heavy-lifting work ------------------------- + @info "received file" worker=worker_id id=job.id name=job.original_name size=job.size + # -------------------------------------------------------------------- + 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