Added RabbitMQ

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"

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"

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
+  every file, to buy a cheaper restart. The seam in `src/queue.jl` is where that
-  for the real fix: swap `ChannelQueue` for a broker-backed `JobQueue` and exact
+  is actually fixed: run with `FS_QUEUE_BACKEND=rabbitmq` and a job stays unacked
-  resume comes back durably, rather than being inferred from a pathname.
+  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
+The HTTP handler and the workers only ever call `enqueue!`, `dequeue!`, `ack!`,
-`close!` on a `JobQueue` (see `src/queue.jl`). Today that's an in-process
+`nack!` and `close!` on a `JobQueue` (`src/queue.jl`). Two implementations sit
-`ChannelQueue`. To move to RabbitMQ (or any broker), implement a new `JobQueue`
+behind those five methods, chosen at startup by `FS_QUEUE_BACKEND`:
-subtype with those three methods and swap the construction in `run`. No handler
+
-or worker code changes.
+| | `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 `<FS_AMQP_PREFIX>.<stage>` — `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 `<prefix>.<stage>`           |
+| `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

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:

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 QUEUE  = Ref{JobQueue}()          # stage-1 (classification) queue; HTTP intake enqueues here
-const KNOWN_QUEUE = Ref{ChannelQueue}()     # stage-2 (enrichment) queue; stage-1 workers enqueue here
+const KNOWN_QUEUE = Ref{JobQueue}()     # 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 UNKNOWN_QUEUE = Ref{JobQueue}()   # 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 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)
+    # The one place the backend choice is made. Everything downstream — handlers,
-    known_queue   = ChannelQueue(cfg.known_queue_capacity)
+    # workers, /stats — sees only `JobQueue` (src/queue.jl).
-    unknown_queue = ChannelQueue(cfg.unknown_queue_capacity)
+    # Set if the broker connection dies under us. This backend does not
-    text_queue    = ChannelQueue(cfg.text_queue_capacity)
+    # 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
+    # Everything re-enters at stage 1 and replays: in-process queues die with the
-    # a crash loses the only record of how far each file had got. Safe, because
+    # process, so a crash loses the only record of how far each file had got.
-    # every stage is a pure function of the file's bytes and every commit is
+    # Safe, because every stage is a pure function of the file's bytes and every
-    # idempotent; see the header of src/worker.jl for the trade and the fix.
+    # 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
 

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 "<prefix>.<stage>"
+    # 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
 

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"
+# size, and it's the shape published to RabbitMQ (the "claim check" pattern:
-# pattern: enqueue a reference, not the payload).
+# 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))

src/queue.jl

@@ -1,10 +1,24 @@
 # The queue seam.
 #
-# The rest of the app only ever calls `enqueue!`, `dequeue!`, and `close!`.
+# The rest of the app only ever calls `enqueue!`, `dequeue!`, `ack!`, `nack!`
-# Today those are backed by an in-process, bounded, thread-safe buffer
+# and `close!`. Two implementations sit behind those five methods:
-# (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
+#   ChannelQueue  in-process, bounded, thread-safe (the Go-channel / Julia-
-# the construction in `run`. No HTTP handler or worker code needs to change.
+#                 `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

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

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
+# The cost is on restart, and which cost depends on the backend. Under the
-# loses them, and recovery can only re-drive everything in `spool/` from stage 1.
+# in-process `ChannelQueue` a crash loses the queues, and recovery can only
-# That is safe, since classification and the UTF-8 sniff are pure functions of
+# re-drive everything in `spool/` from stage 1. That is safe, since
-# the file's bytes and every commit is idempotent, but it redoes work the old
+# classification and the UTF-8 sniff are pure functions of the file's bytes and
-# directory-per-stage layout could skip. The durable fix is the queue seam, not
+# every commit is idempotent, but it redoes work the old directory-per-stage
-# the directories: a broker-backed `JobQueue` restores exact resume for free.
+# 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
+            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

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