diff --git a/Manifest.toml b/Manifest.toml index 5c24b4b..ec39448 100644 --- a/Manifest.toml +++ b/Manifest.toml @@ -2,7 +2,7 @@ julia_version = "1.12.6" manifest_format = "2.0" -project_hash = "ed6bd1b772452682c906ce1236b89ccb1b0876fc" +project_hash = "d7c84379a20829ec867546b4a887bc3c10df8311" [[deps.ADTypes]] git-tree-sha1 = "d9aaef7c63466eee4de23b4d9dad03629df54bea" @@ -15,6 +15,12 @@ weakdeps = ["ChainRulesCore", "ConstructionBase", "EnzymeCore"] ADTypesConstructionBaseExt = "ConstructionBase" ADTypesEnzymeCoreExt = "EnzymeCore" +[[deps.AMQPClient]] +deps = ["Logging", "MbedTLS", "Sockets"] +git-tree-sha1 = "508457ed7a2afb432590247dc363fffc51f242fc" +uuid = "79c8b4cd-a41a-55fa-907c-fab5288e1383" +version = "0.5.1" + [[deps.AbstractFFTs]] deps = ["LinearAlgebra"] git-tree-sha1 = "d92ad398961a3ed262d8bf04a1a2b8340f915fef" diff --git a/Project.toml b/Project.toml index 5881e81..2525511 100644 --- a/Project.toml +++ b/Project.toml @@ -4,6 +4,7 @@ version = "0.1.0" authors = ["wardjm@gmail.com"] [deps] +AMQPClient = "79c8b4cd-a41a-55fa-907c-fab5288e1383" HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3" JLD2 = "033835bb-8acc-5ee8-8aae-3f567f8a3819" JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1" @@ -17,6 +18,7 @@ UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4" Zygote = "e88e6eb3-aa80-5325-afca-941959d7151f" [compat] +AMQPClient = "0.5.1" HTTP = "1.11.0" JLD2 = "0.6.4" JSON3 = "1.14.3" diff --git a/README.md b/README.md index dbc2c3b..b7908f8 100644 --- a/README.md +++ b/README.md @@ -102,13 +102,14 @@ Key properties: it is abandoned. The cost of replay is redoing stages a file had already cleared. That is a - property of the *queue*, not of the directory layout: the queues are + property of the *queue*, not of the directory layout: the default queues are in-process (`src/queue.jl`), so a crash destroys the only record of how far each file got. Per-stage directories used to stand in for that record, at the price of a rename per file per stage on the hot path: a permanent cost on - every file, to buy a cheaper restart. `src/queue.jl` already defines the seam - for the real fix: swap `ChannelQueue` for a broker-backed `JobQueue` and exact - resume comes back durably, rather than being inferred from a pathname. + every file, to buy a cheaper restart. The seam in `src/queue.jl` is where that + is actually fixed: run with `FS_QUEUE_BACKEND=rabbitmq` and a job stays unacked + until its handler commits, so a restart resumes each file at the stage it had + reached instead of re-driving `spool/` from stage 1. See "Queue backends". - **Graceful shutdown:** SIGINT (Ctrl-C) and SIGTERM (systemd/Docker/k8s `stop`) 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 @@ -336,13 +337,94 @@ baseline agrees. See `DESIGN_clustering.md` §11 for the full results, including one known limitation: ELF and these tarballs share a long run of header zero-padding and merge. The v2 fix is inverse-entropy position weighting. -## The queue seam (→ RabbitMQ later) +## Queue backends -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. +The HTTP handler and the workers only ever call `enqueue!`, `dequeue!`, `ack!`, +`nack!` and `close!` on a `JobQueue` (`src/queue.jl`). Two implementations sit +behind those five methods, chosen at startup by `FS_QUEUE_BACKEND`: + +| | `channel` (default) | `rabbitmq` | +|---|---|---| +| Where jobs live | in-process, bounded buffer | durable broker queues, persistent messages | +| Crash recovery | everything in `spool/` replays from stage 1 | each file resumes at the stage it had reached | +| External dependency | none | a RabbitMQ broker | +| Capacity | hard limit, enforced per enqueue | advisory, checked against a polled depth | +| Delivery | exactly once (nothing to redeliver) | at least once | + +`ack!` is the whole difference. A job is settled only after its handler commits, +so a crash mid-enrichment leaves that job on the enrich queue and the restart +picks it up there — not at stage 1, and not lost. `ChannelQueue` implements +`ack!`/`nack!` as no-ops, which is honest rather than lazy: an in-process queue +has no delivery to settle, and its recovery story is `recover_dir!`. + +### Running it + +```bash +# broker + server, both in compose +docker compose -f docker-compose.yml -f docker-compose.rabbitmq.yml up + +# just the broker, with the server on the host (5672 and the management UI on +# 15672 are published for exactly this) +docker compose -f docker-compose.yml -f docker-compose.rabbitmq.yml up -d rabbitmq +FS_QUEUE_BACKEND=rabbitmq \ +FS_AMQP_URL=amqp://fileserver:fileserver@localhost:5672/ \ + julia --project=. -t auto bin/server.jl +``` + +Queues are named `.` — `fileserver.classify`, +`.enrich`, `.triage`, `.language` — so the broker's queue list reads like the +pipeline, and `/stats` reports each one's depth from a once-a-second poll. + +### What it guarantees, and what it deliberately doesn't + +- **At least once, not exactly once.** Stages 1 and 3 publish downstream *then* + ack upstream, so a crash in that window redelivers a job that was already + routed. Safe for the same reason replay is: every stage is a pure function of + the file's bytes and every commit is idempotent. The one new case is that a + duplicate can run *concurrently* with the original and find the file already + committed; `worker_loop` recognises a vanished `job.path` and settles it + quietly, rather than quarantining a file that in fact succeeded. +- **Publisher confirms on intake only.** A `202` means the broker has the file, + because a client may delete its copy on the strength of it. That costs a round + trip per file and serializes intake publishes; `FS_AMQP_CONFIRMS=false` turns + it off. Inter-stage publishes are fire-and-forget by design — a lost one leaves + its source job unacked, and redelivery repairs it for free, on a handoff that + otherwise costs 0.12 µs. +- **Advisory capacity.** `enqueue!` compares `FS_QUEUE_CAPACITY` against a depth + polled once a second (and adjusted locally in between), so the existing `503` + and `blocked_ns` behaviour still works, but a burst can overshoot by up to a + poll interval. Making the limit real would mean `x-max-length` with + `overflow: reject-publish`, which needs a confirm per message to detect. +- **No reconnect.** A dropped connection invalidates every in-flight delivery + tag, so reconnecting would silently reprocess whatever the workers were + holding. Instead the server drains and exits non-zero, and the restart policy + brings it back to redelivered messages and an intact `spool/`. +- **No dead-letter queue.** A failed job is quarantined to `failed/` and then + acked: the file's story and the message's story end in the same place. A DLQ + would hold messages pointing at files that had already moved. +- **One consumer process.** `Job` is a claim check carrying a *local* path, so a + second server on another host would be handed jobs whose files it cannot see. + The broker buys durable resume across restarts, not horizontal scale; scaling + out would additionally need `spool/` on shared storage and an aggregated + `/stats`. +- **Restart recovery skips `spool/`.** The broker is the record of what is in + flight, so re-driving `spool/` would duplicate the whole backlog on every + restart. `FS_RECOVER_SPOOL=true` forces it, for the one case the broker cannot + cover: it was purged or recreated and the spooled files are all that is left. + +### Tests + +The pure parts (URL parsing, the wire format, backend selection, the duplicate +branch) run in the normal suite. The round trip against a real broker is gated: + +```bash +docker compose -f docker-compose.yml -f docker-compose.rabbitmq.yml up -d rabbitmq +FS_TEST_AMQP_URL=amqp://fileserver:fileserver@localhost:5672/ \ + julia --project=. -e 'using Pkg; Pkg.test()' +``` + +Without `FS_TEST_AMQP_URL` those tests are skipped with a notice, so the suite +still passes on a machine with no Docker. ## Running @@ -359,6 +441,10 @@ julia --project=. -e 'using Pkg; Pkg.instantiate()' julia --project=. -t auto bin/server.jl ``` +By default the pipeline's queues are in-process, and nothing external is needed. +For durable queues that survive a crash, run against RabbitMQ instead — see +"Queue backends" above. + ## Shutdown Both SIGINT and SIGTERM trigger the same idempotent graceful drain @@ -366,6 +452,19 @@ Both SIGINT and SIGTERM trigger the same idempotent graceful drain - **SIGINT** is caught as an `InterruptException` (we call `Base.exit_on_sigint(false)`), so shutdown is clean and quiet. + + One caveat specific to the RabbitMQ backend: Julia delivers SIGINT to whichever + task happens to be running on thread 1, which is usually the main loop but is + not guaranteed to be — AMQPClient runs receiver tasks of its own, and an + interrupt that lands in one of those kills the broker connection instead of + reaching the main loop. That is not a stuck server: the depth poller notices + the closed connection within a second and asks for the same drain, so the + process still stops (within ~4s, exiting non-zero, and logging it as a lost + connection rather than an interrupt). **Under a process manager, prefer SIGTERM + for the broker backend** — it goes through `atexit`, which has no such lottery. + Shutdown on either signal also prints a few `Consumer ... task exiting` warnings + from AMQPClient, which are cosmetic: they are its consumer tasks noticing that + we cancelled them. - **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 @@ -454,6 +553,12 @@ init, so the artifact is exactly regenerable from the same inputs. | `FS_PROMOTE_MIN_MAGIC` | `3` | Required fixed signature positions to nominate | | `FS_CLUSTER_CATALOG` | `data/catalog.json` | Durable stage-5 catalog file (single-owner) | | `FS_NOMINATED_DIR` | `data/nominated` | One JSON per self-nominated cluster (human promote gate) | +| `FS_QUEUE_BACKEND` | `channel` | `channel` (in-process) or `rabbitmq` (durable); see "Queue backends" | +| `FS_AMQP_URL` | `amqp://guest:guest@localhost:5672/` | Broker connection, credentials included | +| `FS_AMQP_PREFIX` | `fileserver` | Queue names are `.` | +| `FS_AMQP_PREFETCH` | worker count | Unacked messages the broker hands one stage at a time | +| `FS_AMQP_CONFIRMS` | `true` | Wait for a publisher confirm before a `202` (intake only) | +| `FS_RECOVER_SPOOL` | `false` | Re-drive `spool/` at startup even on the broker backend (use after a purged broker) | > To get real parallelism, start Julia with enough threads (`-t N`) to cover all > pools. If `FS_WORKERS + FS_KNOWN_WORKERS + FS_UNKNOWN_WORKERS + FS_TEXT_WORKERS` @@ -869,6 +974,7 @@ src/ config.jl Config struct + env parsing job.jl Job (the queue reference) queue.jl JobQueue seam + in-process ChannelQueue + rabbit.jl RabbitMQ-backed JobQueue: durable queues, acks, confirms stats.jl per-stage counters behind GET /stats (throughput, utilization) multipart.jl streaming multipart/form-data reader (intake never buffers a file) spool.jl filename sanitizing, streaming spool/move, startup recovery @@ -881,6 +987,8 @@ src/ catalog.jl durable single-owner format catalog + sweep + nominations (stage 5, phase B) worker.jl parametrized worker loop + classify/enrich/triage/language handlers server.jl HTTP routes + the streaming /upload handler +docker-compose.yml the server, in-process queues +docker-compose.rabbitmq.yml overlay: adds the broker and switches the backend bin/ server.jl entry point bench.jl throughput + memory harness against a running server diff --git a/docker-compose.rabbitmq.yml b/docker-compose.rabbitmq.yml new file mode 100644 index 0000000..157601f --- /dev/null +++ b/docker-compose.rabbitmq.yml @@ -0,0 +1,55 @@ +# Overlay that swaps the in-process queues for durable RabbitMQ ones. +# +# docker compose up # channels +# docker compose -f docker-compose.yml -f docker-compose.rabbitmq.yml up +# +# The base file is left alone deliberately: `docker compose up` still brings up +# exactly what it always did, so neither backend is the awkward one to run. +# +# What the broker changes: a job stays unacked until its handler commits, so a +# restart resumes each file at the stage it had reached instead of re-driving +# everything in spool/ from stage 1. See "Queue backends" in the README for the +# guarantees, and for the deliberate holes. + +services: + rabbitmq: + image: rabbitmq:4-management + environment: + # The built-in guest/guest only authenticates over loopback, so it cannot + # work across the compose network. Override these for anything real; they + # are here so the stack comes up with one command, not as a credential. + RABBITMQ_DEFAULT_USER: "fileserver" + RABBITMQ_DEFAULT_PASS: "fileserver" + ports: + # 5672 is published so you can run the server on the host + # (`julia --project=. -t auto bin/server.jl`) against this broker; 15672 is + # the management UI, which is most of the reason to run a broker you can + # look at. Neither needs publishing for the composed server to work. + - "5672:5672" + - "15672:15672" + volumes: + # Durable queues and persistent messages are only as durable as what they + # are written to: without this, `docker compose down` discards the backlog + # the acks exist to protect. + - rabbitmq-data:/var/lib/rabbitmq + healthcheck: + test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"] + interval: 10s + timeout: 5s + retries: 12 + start_period: 30s + restart: unless-stopped + + file-server: + environment: + FS_QUEUE_BACKEND: "rabbitmq" + FS_AMQP_URL: "amqp://fileserver:fileserver@rabbitmq:5672/" + depends_on: + rabbitmq: + # The server retries a broker that isn't up yet (AMQP_CONNECT_RETRY_SECONDS), + # so this gate is belt-and-braces rather than load-bearing — but it keeps + # a cold start quiet instead of a screenful of retry lines. + condition: service_healthy + +volumes: + rabbitmq-data: diff --git a/src/FileServer.jl b/src/FileServer.jl index 30f381f..0f3f765 100644 --- a/src/FileServer.jl +++ b/src/FileServer.jl @@ -15,6 +15,7 @@ include("config.jl") include("job.jl") include("queue.jl") include("stats.jl") # per-stage counters behind GET /stats (needs Config/Job/JobQueue) +include("rabbit.jl") # RabbitMQ-backed JobQueue (needs Config/Job/JobQueue/STAGE_KEYS) 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) @@ -27,12 +28,17 @@ 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. +# `Config`/`JobQueue` types exist. +# +# Typed as the abstract `JobQueue` because which implementation these hold is a +# runtime choice (`FS_QUEUE_BACKEND`). That makes `enqueue!` a dynamic dispatch +# on the intake path — nanoseconds against a handler that writes a file to disk, +# and the price of the backend being swappable at all. 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 TEXT_QUEUE = Ref{ChannelQueue}() # stage-4 (language enrichment) queue; stage-3 workers enqueue here +const QUEUE = Ref{JobQueue}() # stage-1 (classification) queue; HTTP intake enqueues here +const KNOWN_QUEUE = Ref{JobQueue}() # stage-2 (enrichment) queue; stage-1 workers enqueue here +const UNKNOWN_QUEUE = Ref{JobQueue}() # stage-3 (content triage) queue; stage-1 workers enqueue here +const TEXT_QUEUE = Ref{JobQueue}() # stage-4 (language enrichment) queue; stage-3 workers enqueue here const CLASSIFIER = Ref{Classifier}() # loaded once at startup, shared read-only across workers const DETECTOR = Ref{LanguageDetector}() # natural-language detector; built once at startup, shared read-only @@ -97,10 +103,21 @@ function run(; overrides...) # binary is a warning, not a fatal error; per-file lookups degrade to none. linguist_available() || @warn "github-linguist not found on PATH; stage-4 text files will have no programming language (install it to enable)" - queue = ChannelQueue(cfg.queue_capacity) - known_queue = ChannelQueue(cfg.known_queue_capacity) - unknown_queue = ChannelQueue(cfg.unknown_queue_capacity) - text_queue = ChannelQueue(cfg.text_queue_capacity) + # The one place the backend choice is made. Everything downstream — handlers, + # workers, /stats — sees only `JobQueue` (src/queue.jl). + # Set if the broker connection dies under us. This backend does not + # reconnect — see `poll_loop!` — so the honest response is to drain and let + # the restart policy bring us back to redelivered messages. + connection_lost = Threads.Atomic{Bool}(false) + backend = cfg.queue_backend === :rabbitmq ? connect_backend(cfg) : nothing + queues = backend === nothing ? + (classify = ChannelQueue(cfg.queue_capacity), + enrich = ChannelQueue(cfg.known_queue_capacity), + triage = ChannelQueue(cfg.unknown_queue_capacity), + language = ChannelQueue(cfg.text_queue_capacity)) : + open_queues(backend, cfg; on_lost = () -> (connection_lost[] = true)) + queue, known_queue = queues.classify, queues.enrich + unknown_queue, text_queue = queues.triage, queues.language CONFIG[] = cfg QUEUE[] = queue KNOWN_QUEUE[] = known_queue @@ -117,7 +134,7 @@ function run(; overrides...) DETECTOR[] = LanguageDetector() @info "loaded language detector" - @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 text_workers=cfg.text_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity text_capacity=cfg.text_queue_capacity + @info "starting file-server" backend=cfg.queue_backend host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count text_workers=cfg.text_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity text_capacity=cfg.text_queue_capacity # Zero the counters here, not at module load: `since` should mean "serving # since", so a scrape's totals cover the run, not the minutes spent loading @@ -153,12 +170,25 @@ function run(; overrides...) # takes longer to re-drive but none of it is abandoned. Serving hasn't # started yet, so intake cannot race us for the directory. # - # Everything re-enters at stage 1 and replays: the queues are in-process, so - # a crash loses the only record of how far each file had got. Safe, because - # every stage is a pure function of the file's bytes and every commit is - # idempotent; see the header of src/worker.jl for the trade and the fix. - recovered = recover_dir!(cfg.spool_dir, queue) - recovered > 0 && @info "recovered leftover files; replaying from stage 1" recovered=recovered + # Everything re-enters at stage 1 and replays: in-process queues die with the + # process, so a crash loses the only record of how far each file had got. + # Safe, because every stage is a pure function of the file's bytes and every + # commit is idempotent; see the header of src/worker.jl. + # + # Broker-backed, this is exactly the wrong thing to do: the broker still + # holds every unacked job at the stage it reached, so re-driving `spool/` + # would duplicate the entire in-flight backlog on every restart and drag + # finished stages back to the start. `FS_RECOVER_SPOOL` forces it anyway, for + # the one case the broker cannot cover: it was purged or recreated, and the + # spooled files are the only surviving record of the work. + if cfg.queue_backend === :channel || cfg.recover_spool + recovered = recover_dir!(cfg.spool_dir, queue) + recovered > 0 && @info "recovered leftover files; replaying from stage 1" recovered=recovered + else + in_flight = count(p -> isfile(p) && !endswith(p, ".meta.json"), + readdir(cfg.spool_dir; join = true)) + @info "broker holds the in-flight record; not re-driving spool/ (set FS_RECOVER_SPOOL=true if the broker was purged)" spooled=in_flight + end register_routes() # `handler` replaces Oxygen's root stream handler so POST /upload can read its @@ -188,14 +218,26 @@ function run(; overrides...) foreach(wait, unknown_workers) # the ONLY producer of the text queue close!(text_queue) # 6. now safe to close the queue stage-3 fed foreach(wait, text_workers) # 7. wait out stage-4 + backend === nothing || close_backend!(backend) # 8. release the broker connection @info "shutdown complete" end atexit(drain) + # Two ways out of here: an interrupt, or the broker connection dying. + # + # SIGINT is delivered to whichever task happens to be running on thread 1, + # which is the main task in the common case but is not guaranteed to be: + # AMQPClient runs its own receiver tasks, and an interrupt that lands in one + # of those kills the connection instead of reaching this loop. That is not a + # hole so much as the same door: the poller sees the dead connection within + # a second and sets `connection_lost`, so the drain happens either way. Under + # a process manager, prefer SIGTERM for the broker backend — it goes through + # Julia's `atexit`, which has no such lottery (see "Shutdown" in the README). try - while true + while !connection_lost[] sleep(0.5) # interruptible; SIGINT throws in here end + @error "stopping: broker connection closed" catch e e isa InterruptException || rethrow(e) @info "shutdown requested (SIGINT)" @@ -203,6 +245,9 @@ function run(; overrides...) drain() end + # A lost broker is a failure, and saying so in the exit status is what lets a + # supervisor tell "asked to stop" from "fell over". + connection_lost[] && exit(1) return nothing end diff --git a/src/config.jl b/src/config.jl index 67a4c45..449a721 100644 --- a/src/config.jl +++ b/src/config.jl @@ -22,6 +22,30 @@ Base.@kwdef struct Config # github-linguist) is a mix of CPU and process-spawn work, tuned independently. text_worker_count::Int = Threads.nthreads() text_queue_capacity::Int = 1000 + # Which `JobQueue` implementation backs the four stage queues (src/queue.jl): + # `:channel` (in-process, the default, nothing external to run) or + # `:rabbitmq` (durable broker queues, src/rabbit.jl). The queue *capacities* + # above apply to both, but mean different things: a hard limit in-process, an + # advisory one against a polled depth on the broker (see `RabbitQueue`). + queue_backend::Symbol = :channel + amqp_url::String = "amqp://guest:guest@localhost:5672/" + amqp_prefix::String = "fileserver" # queue names are "." + # Unacked messages the broker will hand one stage at a time. Defaults to that + # stage's worker count (0 below means "use it"), so a worker holds at most the + # one it is working on and a restart redelivers the minimum. + amqp_prefetch::Int = 0 + # Wait for a publisher confirm before intake reports a file accepted, so a + # 202 means the broker has it rather than only that our socket took it. Costs + # a round trip per file on the intake path *only*: inter-stage publishes stay + # fire-and-forget, because a lost one leaves its source message unacked and + # redelivery repairs it. See "Queue backends" in the README. + amqp_confirms::Bool = true + # Broker-backed startup does not re-drive `spool/`: the broker is the record + # of what is in flight, and re-driving would duplicate the whole backlog at + # stage 1 on every restart. Set this when the broker was purged or recreated + # and the spooled files are the only surviving record. Ignored (always on) + # for the in-process backend, whose queues never survive a restart. + recover_spool::Bool = false # Every in-flight file lives here, at every stage, from intake until it is # committed to a terminal sink below. Stages route by enqueueing a reference, # not by moving bytes (src/worker.jl header), so there are no per-stage @@ -56,6 +80,29 @@ Base.@kwdef struct Config nominated_dir::String = "data/nominated" # one JSON per self-nominated cluster, awaiting a human promote end +""" + parse_backend(s) -> Symbol + +Turn `FS_QUEUE_BACKEND` into a `Config.queue_backend`. Unknown values are a +startup error rather than a silent fallback: a typo that quietly left you on the +in-process queue would look exactly like durability working right up until a +crash proved it wasn't. +""" +function parse_backend(s::AbstractString) + b = Symbol(lowercase(strip(s))) + b in (:channel, :rabbitmq) || + throw(ArgumentError("FS_QUEUE_BACKEND must be \"channel\" or \"rabbitmq\", got \"$s\"")) + return b +end + +"Parse a boolean env var. Accepts the usual spellings; anything else is an error." +function parse_bool(s::AbstractString) + v = lowercase(strip(s)) + v in ("1", "true", "yes", "on") && return true + v in ("0", "false", "no", "off") && return false + throw(ArgumentError("expected a boolean (true/false), got \"$s\"")) +end + """ config_from_env(; overrides...) @@ -73,7 +120,9 @@ Recognised variables: FS_UPLOAD_CHUNK_BYTES, FS_EXIFTOOL_TIMEOUT, FS_LINGUIST_TIMEOUT, FS_CLUSTER_DIR, FS_CLUSTER_N, FS_CLUSTER_ALPHA, FS_CLUSTER_PSEUDOCOUNT, FS_CLUSTER_BG_MASS, FS_PROMOTE_MIN_MEMBERS, FS_PROMOTE_MIN_MAGIC, - FS_CLUSTER_CATALOG, FS_NOMINATED_DIR + FS_CLUSTER_CATALOG, FS_NOMINATED_DIR, + FS_QUEUE_BACKEND, FS_AMQP_URL, FS_AMQP_PREFIX, FS_AMQP_PREFETCH, + FS_AMQP_CONFIRMS, FS_RECOVER_SPOOL """ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, queue_capacity=nothing, known_worker_count=nothing, @@ -87,7 +136,10 @@ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, cluster_alpha=nothing, cluster_pseudocount=nothing, cluster_bg_mass=nothing, promote_min_members=nothing, promote_min_magic=nothing, cluster_catalog_path=nothing, - nominated_dir=nothing) + nominated_dir=nothing, queue_backend=nothing, + amqp_url=nothing, amqp_prefix=nothing, + amqp_prefetch=nothing, amqp_confirms=nothing, + recover_spool=nothing) Config( host = something(host, get(ENV, "FS_HOST", "127.0.0.1")), port = something(port, parse(Int, get(ENV, "FS_PORT", "8080"))), @@ -117,6 +169,12 @@ function config_from_env(; host=nothing, port=nothing, worker_count=nothing, promote_min_magic = something(promote_min_magic, parse(Int, get(ENV, "FS_PROMOTE_MIN_MAGIC", "3"))), cluster_catalog_path = something(cluster_catalog_path, get(ENV, "FS_CLUSTER_CATALOG", "data/catalog.json")), nominated_dir = something(nominated_dir, get(ENV, "FS_NOMINATED_DIR", "data/nominated")), + queue_backend = something(queue_backend, parse_backend(get(ENV, "FS_QUEUE_BACKEND", "channel"))), + amqp_url = something(amqp_url, get(ENV, "FS_AMQP_URL", "amqp://guest:guest@localhost:5672/")), + amqp_prefix = something(amqp_prefix, get(ENV, "FS_AMQP_PREFIX", "fileserver")), + amqp_prefetch = something(amqp_prefetch, parse(Int, get(ENV, "FS_AMQP_PREFETCH", "0"))), + amqp_confirms = something(amqp_confirms, parse_bool(get(ENV, "FS_AMQP_CONFIRMS", "true"))), + recover_spool = something(recover_spool, parse_bool(get(ENV, "FS_RECOVER_SPOOL", "false"))), ) end diff --git a/src/job.jl b/src/job.jl index 1569829..71b9f53 100644 --- a/src/job.jl +++ b/src/job.jl @@ -1,8 +1,11 @@ # 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). +# size, and it's the shape published to RabbitMQ (the "claim check" pattern: +# enqueue a reference, not the payload). +# +# `path` is a *local* filesystem path, which is why a broker-backed pipeline is +# still single-consumer-process: see "Queue backends" in the README. struct Job id::String # server-minted UUID; also the on-disk filename prefix @@ -10,4 +13,18 @@ struct Job path::String # absolute-ish path to the spooled file size::Int # bytes received_at::Float64 # time() at intake + # Broker delivery tag, and the one field that is *not* part of the job: it + # identifies this delivery on the channel that delivered it, so it is set on + # receipt and never serialized into the message body (a tag from a previous + # connection is meaningless, and a redelivery gets a fresh one). 0 means "no + # broker" — every job under `ChannelQueue`, and every job before publishing. + # Int64 rather than UInt64 to match AMQPClient's `TAMQPDeliveryTag`. + delivery_tag::Int64 end + +Job(id, original_name, path, size, received_at) = + Job(id, original_name, path, size, received_at, Int64(0)) + +"Copy of `job` carrying the delivery tag it arrived with." +with_delivery_tag(job::Job, tag::Integer) = + Job(job.id, job.original_name, job.path, job.size, job.received_at, Int64(tag)) diff --git a/src/queue.jl b/src/queue.jl index b238b83..863938d 100644 --- a/src/queue.jl +++ b/src/queue.jl @@ -1,10 +1,24 @@ # 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. +# The rest of the app only ever calls `enqueue!`, `dequeue!`, `ack!`, `nack!` +# and `close!`. Two implementations sit behind those five methods: +# +# ChannelQueue in-process, bounded, thread-safe (the Go-channel / Julia- +# `Channel` model). Lost on crash: everything still in `spool/` +# is re-driven from stage 1 by `recover_dir!`. +# RabbitQueue durable RabbitMQ queues (src/rabbit.jl). Survives a crash: a +# job stays unacked until its handler commits, so a restart +# resumes each file at the stage it had actually reached. +# +# `FS_QUEUE_BACKEND` picks one; `run` constructs accordingly. No HTTP handler or +# worker code knows which it got. +# +# The ack pair is what makes the broker worth having. Without it a delivery is +# settled the moment it is handed to a worker, so a crash mid-handler loses the +# job exactly as the in-process queue does, and the network hop buys nothing. +# `ChannelQueue` implements both as no-ops, which is honest rather than lazy: +# an in-process queue has no delivery to settle, and its recovery story is +# `recover_dir!`. abstract type JobQueue end @@ -34,6 +48,31 @@ queue has been closed *and* fully drained, which is a worker's signal to exit. """ function dequeue! end +""" + ack!(q, job) + +Settle `job` as done: the broker may forget it. Called by `worker_loop` after +the handler returns, and after a quarantine, so the message outlives the process +for exactly as long as the work is unfinished. + +No-op for `ChannelQueue`. +""" +function ack! end + +""" + nack!(q, job) + +Settle `job` as rejected, without requeueing. Requeueing is deliberately not +offered: a job that failed on its bytes will fail again, and redelivering it is +a poison-message loop. With no dead-letter exchange configured the broker simply +discards it, so this differs from `ack!` only in what it says, not in what +happens — which is the point at the one call site that uses it (see +`worker_loop`). + +No-op for `ChannelQueue`. +""" +function nack! end + """ close!(q) @@ -79,6 +118,10 @@ function dequeue!(q::ChannelQueue)::Union{Job,Nothing} end end +# Nothing to settle: an in-process job was never a delivery. +ack!(::ChannelQueue, ::Job) = nothing +nack!(::ChannelQueue, ::Job) = nothing + function close!(q::ChannelQueue) lock(q.cond) try diff --git a/src/rabbit.jl b/src/rabbit.jl new file mode 100644 index 0000000..cd15be5 --- /dev/null +++ b/src/rabbit.jl @@ -0,0 +1,512 @@ +# RabbitMQ-backed `JobQueue` (src/queue.jl defines the seam this implements). +# +# What it buys, and what it doesn't +# --------------------------------- +# The in-process `ChannelQueue` dies with the process, so a crash leaves no +# record of how far each file had got and `recover_dir!` can only re-drive +# everything in `spool/` from stage 1. Durable queues plus per-job acks replace +# that guess with a fact: a job stays unacked until its handler commits, so a +# restart resumes each file at the stage it had actually reached, and no file is +# processed from scratch just because the process died. +# +# It is *at-least-once*, not exactly-once. Stages 1 and 3 publish downstream and +# then ack upstream (`worker_loop`), so a crash in that window redelivers a job +# that was already routed, and the file gets processed twice. That is safe here +# for the same reason replay was: classification and the UTF-8 sniff are pure +# functions of the file's bytes, and every terminal commit renames with +# `force=true`. The one new wrinkle is that a duplicate can now run *concurrently* +# with the original, and the loser finds the file already moved; `worker_loop` +# recognises that (a vanished `job.path`) and settles it quietly instead of +# quarantining a file that in fact succeeded. +# +# It also does not buy horizontal scale. `Job.path` is a local filesystem path, +# so a second consumer process on another host would be handed jobs whose files +# it cannot see. See "Queue backends" in the README. +# +# Shape +# ----- +# One connection for the whole process, and per stage queue: +# +# con_chan consumes, with `basic_qos` prefetch bounding unacked messages, +# and carries this stage's acks (a delivery tag is only meaningful +# on the channel that delivered it) +# pub_chan publishes *into* this queue, from whichever stage feeds it +# buffer a bounded `Channel{Job}` bridging AMQPClient's push consumer to +# the blocking `dequeue!` that `worker_loop` expects +# +# plus one shared admin channel for the depth polls. Publishing and acking from +# many worker tasks at once is safe: AMQPClient serializes multi-frame publishes +# under a per-channel send lock and queues frames onto a thread-safe channel. + +using AMQPClient + +# The stage queues, in `STAGE_KEYS` order. The suffix is what appears after +# `FS_AMQP_PREFIX`, so a broker's queue list reads like the pipeline. +const AMQP_QUEUE_SUFFIXES = (classify = "classify", enrich = "enrich", + triage = "triage", language = "language") + +# Startup connect: retry a broker that isn't up yet (compose starts the app long +# before RabbitMQ is ready) but give up rather than hang, so a genuinely +# misconfigured URL is a startup failure and not a silent hang. +const AMQP_CONNECT_RETRY_SECONDS = 30.0 +const AMQP_CONNECT_RETRY_INTERVAL = 2.0 + +# How often the depth poller asks the broker for each queue's ready-message +# count. This is what `/stats` reports and what the advisory capacity check +# compares against, so it trades staleness for one round trip per queue per +# second instead of one per file. +const AMQP_DEPTH_POLL_SECONDS = 1.0 + +# A publisher confirm that never arrives must not park an HTTP request thread +# forever. The poller doubles as the watchdog (see `poll_loop!`). +const AMQP_CONFIRM_TIMEOUT_SECONDS = 10.0 + +"Connection parameters parsed out of an `amqp://` URL." +struct AMQPTarget + host::String + port::Int + virtualhost::String + login::String + password::String +end + +""" + parse_amqp_url(url) -> AMQPTarget + +Parse `amqp://user:pass@host:port/vhost`. Every component is optional and falls +back to the AMQP defaults, except that we do *not* default the credentials to +`guest`/`guest`: those only authenticate over loopback in a stock RabbitMQ, so +inheriting them across a container network would fail at connect time with an +authentication error that says nothing about the real mistake. + +The vhost is the path with its leading `/` removed and percent-escapes decoded, +so the conventional default vhost `/` can be written either as a bare trailing +slash or as the `%2F` form. +""" +function parse_amqp_url(url::AbstractString) + uri = HTTP.URI(url) + scheme = lowercase(uri.scheme) + scheme in ("amqp", "amqps") || + throw(ArgumentError("FS_AMQP_URL must be an amqp:// or amqps:// URL, got \"$url\"")) + scheme == "amqps" && + throw(ArgumentError("amqps:// (TLS) is not supported yet; use amqp:// on a trusted network")) + + userinfo = split(uri.userinfo, ':'; limit = 2) + login = isempty(userinfo[1]) ? "guest" : HTTP.unescapeuri(userinfo[1]) + password = length(userinfo) < 2 ? "guest" : HTTP.unescapeuri(userinfo[2]) + + path = lstrip(uri.path, '/') + vhost = isempty(path) ? "/" : HTTP.unescapeuri(path) + + return AMQPTarget(isempty(uri.host) ? "localhost" : uri.host, + isempty(uri.port) ? AMQPClient.AMQP_DEFAULT_PORT : parse(Int, uri.port), + vhost, login, password) +end + +"The process's single broker connection, plus the shared channel depth polls use." +mutable struct RabbitBackend + conn::AMQPClient.Connection + admin_chan::AMQPClient.MessageChannel + const admin_lock::ReentrantLock # one synchronous RPC on admin_chan at a time + const prefix::String +end + +""" + RabbitQueue(backend, name, capacity, prefetch; confirms = false) + +One durable queue. `capacity` is *advisory*: `enqueue!` compares it against a +depth that is polled once a second and adjusted locally in between, so a burst +can overshoot by up to a poll interval before the queue starts refusing. The +alternative — `x-max-length` with `overflow: reject-publish` — needs a publisher +confirm per message to learn about the rejection, which would put a round trip +on an inter-stage handoff that currently costs 0.12 µs. + +`confirms` puts this queue's *publish* channel into confirm mode, so `enqueue!` +returns only once the broker has accepted the message. It is set on the stage-1 +queue alone, because intake is the only publisher that answers to someone: a +`202` tells a client its file is safe, and that client may delete its copy. +Inter-stage publishes stay fire-and-forget, where a lost message leaves its +source job unacked and redelivery repairs it for free. +""" +mutable struct RabbitQueue <: JobQueue + const backend::RabbitBackend + const name::String + const capacity::Int + const buffer::Channel{Job} # consumer callback → dequeue! + const pub_chan::AMQPClient.MessageChannel + const con_chan::AMQPClient.MessageChannel + const depth::Threads.Atomic{Int} # last polled ready count, locally adjusted + const closed::Threads.Atomic{Bool} + const confirms::Bool + const pub_lock::ReentrantLock # serializes publish+confirm on pub_chan + const confirm_ch::Channel{Bool} # broker's verdict on the outstanding publish + const pending_since::Threads.Atomic{Float64} # when it was published (0.0 = nothing pending) + const on_lost::Function # called once if the connection dies + consumer_tag::String + poller::Union{Task,Nothing} +end + +capacity(q::RabbitQueue) = q.capacity + +""" +Depth as of the last poll, adjusted by our own publishes and takes since. + +The broker's number counts *ready* messages, so it excludes the ones already +handed to workers and not yet acked. That is the right number for `/stats`: +depth is meant to say how much work is piled up in front of a stage, and the +messages in its workers' hands are not piled up, they are being worked. +""" +Base.length(q::RabbitQueue) = max(0, q.depth[]) + +# --- wire format ------------------------------------------------------------ + +""" + job_json(job) -> String + +Serialize a `Job` for the wire. `delivery_tag` is deliberately absent: it names a +delivery on one channel of one connection, not the job, and a redelivery gets a +fresh one. +""" +job_json(job::Job) = JSON3.write((; job.id, job.original_name, job.path, + job.size, job.received_at)) + +"Rebuild a `Job` from a message body, carrying the tag it was delivered under." +function job_from_message(msg::AMQPClient.Message) + o = JSON3.read(String(copy(msg.data))) + return Job(String(o.id), String(o.original_name), String(o.path), + Int(o.size), Float64(o.received_at), Int64(msg.delivery_tag)) +end + +# --- connect / declare ------------------------------------------------------ + +""" + connect_backend(cfg) -> RabbitBackend + +Open the process's connection, retrying a broker that is not up yet for up to +`AMQP_CONNECT_RETRY_SECONDS`. Compose starts this container long before RabbitMQ +finishes booting even with a healthcheck gate, so a first refusal is routine; a +persistent one is a startup error, because falling back to the in-process queue +would silently deliver none of the durability that was asked for. +""" +function connect_backend(cfg::Config) + t = parse_amqp_url(cfg.amqp_url) + auth = Dict{String,Any}("MECHANISM" => "AMQPLAIN", + "LOGIN" => t.login, "PASSWORD" => t.password) + deadline = time() + AMQP_CONNECT_RETRY_SECONDS + attempt = 0 + while true + attempt += 1 + try + conn = AMQPClient.connection(; host = t.host, port = t.port, + virtualhost = t.virtualhost, + auth_params = auth) + admin = AMQPClient.channel(conn, AMQPClient.UNUSED_CHANNEL, true) + @info "connected to broker" host=t.host port=t.port vhost=t.virtualhost attempts=attempt + return RabbitBackend(conn, admin, ReentrantLock(), cfg.amqp_prefix) + catch e + time() < deadline || rethrow() + @info "broker not ready, retrying" host=t.host port=t.port attempt=attempt + sleep(AMQP_CONNECT_RETRY_INTERVAL) + end + end +end + +""" + open_queue(backend, cfg, key, capacity, workers) -> RabbitQueue + +Declare one stage's durable queue and start consuming from it. + +Prefetch defaults to the stage's worker count: enough that every worker can hold +a job, and no more, so a crash redelivers the smallest possible set. The bridge +`Channel` is sized to match, since it is exactly the set of messages the broker +has handed us and we have not yet acked. +""" +function open_queue(backend::RabbitBackend, cfg::Config, key::Symbol, + capacity::Int, workers::Int; on_lost::Function = () -> nothing) + name = string(backend.prefix, ".", AMQP_QUEUE_SUFFIXES[key]) + prefetch = cfg.amqp_prefetch > 0 ? cfg.amqp_prefetch : max(1, workers) + + con_chan = AMQPClient.channel(backend.conn, AMQPClient.UNUSED_CHANNEL, true) + pub_chan = AMQPClient.channel(backend.conn, AMQPClient.UNUSED_CHANNEL, true) + + # durable: the queue definition survives a broker restart. Combined with + # PERSISTENT messages below, so do the jobs in it. + AMQPClient.queue_declare(con_chan, name; durable = true) + AMQPClient.basic_qos(con_chan, 0, prefetch, false) + + confirms = cfg.amqp_confirms && key === :classify + q = RabbitQueue(backend, name, capacity, Channel{Job}(prefetch), + pub_chan, con_chan, Threads.Atomic{Int}(0), + Threads.Atomic{Bool}(false), confirms, ReentrantLock(), + Channel{Bool}(1), Threads.Atomic{Float64}(0.0), on_lost, "", nothing) + + if confirms + confirm_select!(pub_chan) + # The broker reports each publish's fate asynchronously by sequence + # number. We hold `pub_lock` across publish-and-wait, so at most one + # publish is ever outstanding and any verdict that arrives is this one's. + AMQPClient.handle(pub_chan, :Basic, :Ack, (_c, _m, _x) -> settle_confirm!(q, true)) + AMQPClient.handle(pub_chan, :Basic, :Nack, (_c, _m, _x) -> settle_confirm!(q, false)) + end + + ok, tag = AMQPClient.basic_consume(con_chan, name, msg -> deliver!(q, msg)) + ok || error("could not start consuming from queue $name") + q.consumer_tag = tag + q.poller = Threads.@spawn poll_loop!(q) + + @info "queue ready" queue=name prefetch=prefetch capacity=capacity confirms=confirms + return q +end + +""" + open_queues(backend, cfg) -> NamedTuple + +The four stage queues, keyed by `STAGE_KEYS` so they line up with the per-stage +worker counts, capacities and metrics without a second mapping to keep in sync. +""" +function open_queues(backend::RabbitBackend, cfg::Config; on_lost::Function = () -> nothing) + caps = (classify = cfg.queue_capacity, enrich = cfg.known_queue_capacity, + triage = cfg.unknown_queue_capacity, language = cfg.text_queue_capacity) + workers = (classify = cfg.worker_count, enrich = cfg.known_worker_count, + triage = cfg.unknown_worker_count, language = cfg.text_worker_count) + return NamedTuple{STAGE_KEYS}(map(STAGE_KEYS) do key + open_queue(backend, cfg, key, caps[key], workers[key]; on_lost) + end) +end + +"Close the shared connection. Safe to call on a connection that is already gone." +function close_backend!(backend::RabbitBackend) + try + close(backend.conn) + catch e + @debug "error closing broker connection" exception=e + end + return nothing +end + +""" + confirm_select!(chan) + +Put `chan` into publisher-confirm mode. + +Not `AMQPClient.confirm_select`, which is broken in AMQPClient 0.5.1: it builds +the `Confirm.Select` payload with no fields, while the spec (and the library's +own `CLASS_MAP`) declares a `Nowait` bit, so it throws a `BoundsError` before +anything reaches the wire. This is the same call with that one field supplied. +Drop it for the library's version once the fix is released. +""" +function confirm_select!(chan::AMQPClient.MessageChannel) + AMQPClient._wait_resp(chan, true, false, AMQPClient.on_confirm_select_ok, + :Confirm, :SelectOk, false, AMQPClient.DEFAULT_TIMEOUT) do + AMQPClient.send(chan, AMQPClient.TAMQPMethodPayload(:Confirm, :Select, (false,))) + end +end + +# --- consume ---------------------------------------------------------------- + +""" +Consumer callback: hand a delivery to the bridge channel. + +Blocking here is the mechanism, not a problem. The bridge holds `prefetch` jobs, +so when workers fall behind this parks, the broker's prefetch window fills, and +it stops sending — backpressure all the way to the queue, with no unbounded +local buffer. +""" +function deliver!(q::RabbitQueue, msg::AMQPClient.Message) + try + put!(q.buffer, job_from_message(msg)) + Threads.atomic_sub!(q.depth, 1) + catch e + # Either we are shutting down (buffer closed after `basic_cancel`, and a + # delivery raced it) or the body was unparseable. Neither may kill the + # consumer task. The message stays unacked, so the broker redelivers it. + q.closed[] || @error "could not accept delivery" queue=q.name exception=(e, catch_backtrace()) + end + return nothing +end + +function dequeue!(q::RabbitQueue)::Union{Job,Nothing} + try + return take!(q.buffer) + catch e + # Closed and drained: the shutdown signal `worker_loop` waits for. + e isa InvalidStateException && return nothing + rethrow() + end +end + +""" +Acks and rejects go out on the channel that made the delivery, because a +delivery tag is scoped to its channel. + +A failure here is logged, not thrown: an unacked job is redelivered, which is +the correct outcome, and killing the worker over it would shrink the pool. +""" +function ack!(q::RabbitQueue, job::Job) + job.delivery_tag == 0 && return nothing + try + AMQPClient.basic_ack(q.con_chan, job.delivery_tag) + catch e + @error "ack failed; job will be redelivered" queue=q.name id=job.id exception=e + end + return nothing +end + +function nack!(q::RabbitQueue, job::Job) + job.delivery_tag == 0 && return nothing + try + AMQPClient.basic_reject(q.con_chan, job.delivery_tag; requeue = false) + catch e + @error "reject failed; job will be redelivered" queue=q.name id=job.id exception=e + end + return nothing +end + +# --- publish ---------------------------------------------------------------- + +""" + enqueue!(q, job) -> Bool + +Publish `job` as a persistent message. Returns `false` — which intake turns into +a `503` and the routing stages turn into park-and-retry — when the queue is +closed, when the polled depth says it is at capacity, or when a confirm-mode +publish is not confirmed. + +`delivery_mode = PERSISTENT` against a durable queue is what makes a broker +restart survivable; without it the queue would come back empty and the whole +point of the backend would be gone. +""" +function enqueue!(q::RabbitQueue, job::Job)::Bool + q.closed[] && return false + q.depth[] >= q.capacity && return false # advisory: see the `RabbitQueue` docstring + + msg = AMQPClient.Message(Vector{UInt8}(job_json(job)); + content_type = "application/json", + delivery_mode = AMQPClient.PERSISTENT) + ok = q.confirms ? publish_confirmed!(q, msg) : publish!(q, msg) + ok && Threads.atomic_add!(q.depth, 1) + return ok +end + +function publish!(q::RabbitQueue, msg::AMQPClient.Message) + try + AMQPClient.basic_publish(q.pub_chan, msg; exchange = "", routing_key = q.name) + return true + catch e + @error "publish failed" queue=q.name exception=(e, catch_backtrace()) + return false + end +end + +""" +Publish and wait for the broker's confirm. + +Serialized on `pub_lock`, so exactly one publish is outstanding per queue and the +next verdict to arrive belongs to it. That costs intake a round trip per file and +caps its rate at roughly one file per broker round trip; it is why confirms are +scoped to intake and configurable (`FS_AMQP_CONFIRMS`). +""" +function publish_confirmed!(q::RabbitQueue, msg::AMQPClient.Message) + lock(q.pub_lock) do + while isready(q.confirm_ch) # discard a verdict left by a timed-out publish + take!(q.confirm_ch) + end + q.pending_since[] = time() + try + publish!(q, msg) || return false + return take!(q.confirm_ch) # poll_loop! guarantees this is answered + finally + q.pending_since[] = 0.0 + end + end +end + +"Record the broker's verdict for the one outstanding publish." +function settle_confirm!(q::RabbitQueue, ok::Bool) + q.pending_since[] == 0.0 && return nothing # nothing waiting; a late/duplicate frame + isready(q.confirm_ch) || put!(q.confirm_ch, ok) + return nothing +end + +# --- poll ------------------------------------------------------------------- + +""" + poll_loop!(q) + +One task per queue, doing the two things that must happen on a timer: + + * refresh `depth` from the broker (a passive `queue.declare` returns the ready + count), which feeds `/stats` and the advisory capacity check, and + * answer a publisher confirm that has gone missing, so a broker that accepted + a publish and then went quiet can't park an HTTP request thread forever, and + * notice that the connection has died and call `on_lost`, which asks `run` to + shut down. This backend does not reconnect (every in-flight delivery tag + would be invalid, and the jobs holding them would be processed twice), so a + lost connection means the process is finished: it drains, exits, and the + restart policy brings it back to redelivered messages and an intact + `spool/`. Without this the pipeline would sit there looking healthy with + four queues that can never receive anything again. + +Folding the watchdog in here keeps it free: it is a comparison on a task that was +going to wake up anyway. +""" +function poll_loop!(q::RabbitQueue) + while !q.closed[] + sleep(AMQP_DEPTH_POLL_SECONDS) + q.closed[] && break + + pending = q.pending_since[] + if pending != 0.0 && time() - pending > AMQP_CONFIRM_TIMEOUT_SECONDS + @warn "publisher confirm timed out; reporting the file as not accepted" queue=q.name + settle_confirm!(q, false) + end + + try + lock(q.backend.admin_lock) do + ok, _, count, _ = AMQPClient.queue_declare(q.backend.admin_chan, q.name; + passive = true) + ok && (q.depth[] = Int(count)) + end + catch e + q.closed[] && break + # A dead connection is not something this backend tries to ride out: + # every in-flight delivery tag becomes invalid on reconnect, so the + # honest recovery is a restart, where unacked messages redeliver and + # `spool/` is intact. Report it and let the supervisor act. + if !isopen(q.backend.conn) + # Two causes, indistinguishable from here: the broker or the + # network went away, or a SIGINT was delivered to one of + # AMQPClient's tasks instead of the main loop (see `run`). + # The response is the same either way — drain and exit. + @error "broker connection closed (lost broker, or an interrupt delivered to a broker task); shutting down (unacked jobs will be redelivered)" queue=q.name + q.on_lost() + break + end + @warn "queue depth poll failed" queue=q.name exception=e + end + end + return nothing +end + +# --- shutdown --------------------------------------------------------------- + +""" + close!(q) + +Stop consuming, then close the bridge so workers drain what is already in hand +and exit — the same drain-then-exit contract `ChannelQueue` offers. + +Jobs the broker has delivered but that we never got to stay unacked and are +redelivered on the next start. That is the backend working as intended, and the +reason shutdown does not need to be careful here. +""" +function close!(q::RabbitQueue) + q.closed[] || try + AMQPClient.basic_cancel(q.con_chan, q.consumer_tag) + catch e + @debug "error cancelling consumer" queue=q.name exception=e + end + q.closed[] = true + isopen(q.buffer) && close(q.buffer) + return nothing +end diff --git a/src/worker.jl b/src/worker.jl index 423440b..c5c833e 100644 --- a/src/worker.jl +++ b/src/worker.jl @@ -20,12 +20,14 @@ # nothing on the live path. What remains is one move at the end: into a terminal # sink (`done/`, `text_done/`, `binary/`) or into `failed/` on a throw. # -# The cost is on restart. The queues are in-process (src/queue.jl), so a crash -# loses them, and recovery can only re-drive everything in `spool/` from stage 1. -# That is safe, since classification and the UTF-8 sniff are pure functions of -# the file's bytes and every commit is idempotent, but it redoes work the old -# directory-per-stage layout could skip. The durable fix is the queue seam, not -# the directories: a broker-backed `JobQueue` restores exact resume for free. +# The cost is on restart, and which cost depends on the backend. Under the +# in-process `ChannelQueue` a crash loses the queues, and recovery can only +# re-drive everything in `spool/` from stage 1. That is safe, since +# classification and the UTF-8 sniff are pure functions of the file's bytes and +# every commit is idempotent, but it redoes work the old directory-per-stage +# layout could skip. Under `RabbitQueue` (src/rabbit.jl) it doesn't arise: a job +# stays unacked until its handler commits, so a restart resumes each file at the +# stage it had reached. The fix was the queue seam, not the directories. # # (`ROUTE_ENQUEUE_RETRY_SECONDS`, the backoff the routing handoffs below use, # now lives in src/queue.jl, since startup recovery shares it.) @@ -173,21 +175,46 @@ function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue, handler, job === nothing && break # queue closed and drained → exit Threads.atomic_add!(stats.in_flight, 1) t0 = time_ns() - ok = true + outcome = :ok try handler(job, cfg, worker_id) + ack!(queue, job) catch e - ok = false - @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()) + if !isfile(job.path) + # The file is gone and we did not move it: another worker + # committed it, i.e. this is the duplicate half of an + # at-least-once redelivery (src/rabbit.jl). The work is done, so + # settle the message and say nothing louder than @debug. It is + # emphatically not a failure: quarantining here would file a + # `failed/` entry against a file that succeeded, and counting it + # either way would double-count one file's work. + outcome = :duplicate + @debug "duplicate delivery; already committed by another worker" worker=worker_id id=job.id name=job.original_name + ack!(queue, job) + else + outcome = :failed + @error "processing failed" worker=worker_id id=job.id name=job.original_name exception=(e, catch_backtrace()) + try + move_to(cfg.failed_dir, job) + ack!(queue, job) # quarantined: the file's story ends in failed/ + catch e2 + @error "could not quarantine failed file" worker=worker_id id=job.id path=job.path exception=(e2, catch_backtrace()) + # Failed *and* unquarantinable: the file is still in spool/. + # Reject rather than ack, because we are not claiming this + # one is done. With no dead-letter exchange the broker + # discards it either way — `nack!` never requeues, so this + # cannot become a poison-message loop — but the distinction + # is free and the message is the honest one. + nack!(queue, job) + end end finally # In a `finally` so an InterruptException during shutdown can't leave # in_flight permanently above zero, which would read as a stuck job. - record_job!(stats, ok, job.size, Int(time_ns() - t0)) + # A duplicate records nothing: its work was already counted by the + # delivery that actually did it. + outcome === :duplicate || + record_job!(stats, outcome === :ok, job.size, Int(time_ns() - t0)) Threads.atomic_sub!(stats.in_flight, 1) end end diff --git a/test/runtests.jl b/test/runtests.jl index 65d8f1b..bc8d40d 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -22,6 +22,9 @@ using FileServer: Job, Config, ChannelQueue, enqueue!, dequeue!, close!, length, Catalog, load_catalog, save_catalog!, catalog_sweep!, compact!, write_nominations!, run_cluster_sweep, binary_files, record_example!, signature_hex, ensure_dirs, + ack!, nack!, with_delivery_tag, parse_backend, parse_bool, config_from_env, + parse_amqp_url, job_json, job_from_message, AMQPTarget, + RabbitQueue, connect_backend, open_queues, close_backend!, MultipartReader, MultipartError, MultipartPart, next_part!, write_part_body!, skip_part_body!, multipart_boundary, parse_part_headers, spool_stream, UPLOAD_CHUNK_BYTES @@ -1132,4 +1135,180 @@ end end end + @testset "queue backends" begin + @testset "backend selection is explicit or an error" begin + @test parse_backend("channel") === :channel + @test parse_backend(" RabbitMQ ") === :rabbitmq + # A typo must not quietly leave you on the in-process queue: that + # looks exactly like durability working, until a crash proves it isn't. + @test_throws ArgumentError parse_backend("rabbit") + @test_throws ArgumentError parse_backend("") + + @test parse_bool("true") && parse_bool("1") && parse_bool("ON") + @test !parse_bool("false") && !parse_bool("0") && !parse_bool("off") + @test_throws ArgumentError parse_bool("maybe") + + @test config_from_env().queue_backend === :channel # default is unchanged + cfg = config_from_env(; queue_backend = :rabbitmq, amqp_prefetch = 4, + amqp_confirms = false, recover_spool = true) + @test cfg.queue_backend === :rabbitmq + @test cfg.amqp_prefetch == 4 && !cfg.amqp_confirms && cfg.recover_spool + end + + @testset "amqp url parsing" begin + t = parse_amqp_url("amqp://user:pw@broker.internal:5673/prod") + @test t == AMQPTarget("broker.internal", 5673, "prod", "user", "pw") + + # Bare form: default port, default vhost, and the guest credentials + # that only work over loopback anyway. + t2 = parse_amqp_url("amqp://localhost/") + @test t2.port == 5672 && t2.virtualhost == "/" && t2.login == "guest" + @test parse_amqp_url("amqp://localhost").virtualhost == "/" + # The conventional default vhost is often written percent-escaped. + @test parse_amqp_url("amqp://h:1/%2F").virtualhost == "/" + # Credentials may contain characters that need escaping in a URL. + @test parse_amqp_url("amqp://u%40b:p%2Fw@h/").password == "p/w" + + @test_throws ArgumentError parse_amqp_url("http://localhost:5672/") + @test_throws ArgumentError parse_amqp_url("amqps://localhost:5671/") + end + + @testset "the delivery tag never goes on the wire" begin + job = with_delivery_tag(Job("id-9", "a b.txt", "/tmp/spool/id-9-a_b.txt", + 4096, 1234.5), 77) + @test job.delivery_tag == 77 + + # A tag names a delivery on one channel of one connection, not the + # job, so publishing it would be meaningless at best and would + # survive a restart as a lie at worst. + wire = JSON3.read(job_json(job)) + @test !haskey(wire, :delivery_tag) + @test wire.id == "id-9" && wire.size == 4096 && wire.received_at == 1234.5 + + # Everything else must survive the round trip untouched, including a + # name with a space in it. + @test wire.original_name == "a b.txt" + @test wire.path == "/tmp/spool/id-9-a_b.txt" + end + + @testset "acks are a no-op on the in-process queue" begin + # ChannelQueue has no delivery to settle, so `worker_loop` calling + # these on every job must cost nothing and change nothing. + q = ChannelQueue(2) + job = Job("id-1", "a.bin", "/tmp/a.bin", 1, 0.0) + @test enqueue!(q, job) + @test ack!(q, job) === nothing + @test nack!(q, job) === nothing + @test length(q) == 1 + end + + @testset "a duplicate delivery is settled, not quarantined" begin + # At-least-once means the same file can be handed to two workers, and + # the loser finds it already committed. That is a success it arrived + # too late for, not a failure: quarantining would file a `failed/` + # entry against a file that worked, and counting it either way would + # double-count one file. + mktempdir() do root + cfg = tmp_config(root) + q = ChannelQueue(10) + stats = StageStats() + + dup = joinpath(cfg.spool_dir, "id-1-gone.bin") + write(dup, "x" ^ 10) + @test enqueue!(q, Job("id-1", "gone.bin", dup, 10, 0.0)) + + real = joinpath(cfg.spool_dir, "id-2-here.bin") + write(real, "y" ^ 10) + @test enqueue!(q, Job("id-2", "here.bin", real, 10, 0.0)) + close!(q) + + worker_loop(1, cfg, q, (job, _, _) -> begin + # The duplicate's file was committed by the winner before the + # handler ran; the other job fails with its file still there. + job.id == "id-1" && rm(job.path) + error("handler blew up") + end, stats) + + @test stats.completed[] == 0 + @test stats.failed[] == 1 # only the genuine failure + @test stats.bytes[] == 10 # the duplicate records nothing + @test stats.in_flight[] == 0 + @test !isfile(joinpath(cfg.failed_dir, "id-1-gone.bin")) + @test isfile(joinpath(cfg.failed_dir, "id-2-here.bin")) + end + end + + # The rest needs a live broker. Point FS_TEST_AMQP_URL at one to run it: + # docker compose -f docker-compose.yml -f docker-compose.rabbitmq.yml up -d rabbitmq + # FS_TEST_AMQP_URL=amqp://fileserver:fileserver@localhost:5672/ julia --project=. -e 'using Pkg; Pkg.test()' + amqp_url = get(ENV, "FS_TEST_AMQP_URL", "") + if isempty(amqp_url) + @info "skipping RabbitMQ integration tests (set FS_TEST_AMQP_URL to run them)" + else + @testset "rabbitmq round trip, redelivery and depth" begin + # A prefix per run, so a leftover queue from a previous run can + # never make this pass (or fail) for the wrong reason. + prefix = "fstest-" * string(rand(UInt32); base = 16) + cfg = config_from_env(; queue_backend = :rabbitmq, amqp_url = amqp_url, + amqp_prefix = prefix, queue_capacity = 5, + worker_count = 2, known_worker_count = 2, + unknown_worker_count = 2, text_worker_count = 2) + backend = connect_backend(cfg) + queues = open_queues(backend, cfg) + q = queues.classify + @test q isa RabbitQueue + @test capacity(q) == 5 + + @testset "a job survives the wire intact" begin + @test enqueue!(q, Job("id-1", "hello.txt", "/tmp/hello.txt", 123, 1.5)) + got = dequeue!(q) + @test got.id == "id-1" + @test got.original_name == "hello.txt" + @test got.path == "/tmp/hello.txt" + @test got.size == 123 + @test got.received_at == 1.5 + @test got.delivery_tag != 0 # a real delivery, ackable + ack!(q, got) + end + + @testset "an unacked job comes back after a crash" begin + # The whole point of the backend: take a job, never ack it, + # lose the connection the way a SIGKILL would, and find it + # waiting on restart. + @test enqueue!(q, Job("id-2", "again.txt", "/tmp/again.txt", 7, 2.5)) + taken = dequeue!(q) + @test taken.id == "id-2" + + foreach(close!, values(queues)) + close_backend!(backend) + sleep(1.0) + + backend2 = connect_backend(cfg) + queues2 = open_queues(backend2, cfg) + q2 = queues2.classify + redelivered = dequeue!(q2) + @test redelivered.id == "id-2" + ack!(q2, redelivered) + + sleep(2 * 1.0 + 0.5) # let the depth poller catch up + @test length(q2) == 0 + + # Advisory capacity: publishing without draining must start + # refusing rather than let the queue grow without bound. + refused = 0 + for i in 1:40 + enqueue!(q2, Job("f$i", "f$i", "/tmp/f$i", 1, 0.0)) && continue + refused = i + break + end + @test refused > 0 + @test length(q2) >= capacity(q2) + + foreach(close!, values(queues2)) + close_backend!(backend2) + end + end + end + end + end