Background job processing with priority queues, scheduled execution, automatic retries with exponential backoff, and multi-tenant support.
| Namespace | Purpose |
|---|---|
| Pure functions: job creation, state transitions, retry logic |
| Protocols: |
| Pure cron parsing and next-run calculation |
| Recurring-job ticker and |
| Malli schemas: job states, priorities, retry config |
| In-memory adapter (atoms, for dev/test) |
| Database adapter ( |
| Redis adapter (sorted sets, lists, hashes) |
| Worker implementation: polling, processing, pool management |
| Multi-tenant job execution with schema switching |
;; Handlers receive args map, return result map
(defn send-email-handler [args]
{:success? true
:result {:sent-to (:email args)}})
;; Failing handler
(defn failing-handler [args]
{:success? false
:error {:message "Processing failed" :type :processing-error}})(require '[wagoe.jobs.core.job :as job]
'[wagoe.jobs.ports :as ports])
(let [new-job (job/create-job {:job-type :send-email
:args {:to "user@example.com"}
:priority :high
:max-retries 3}
(random-uuid)
(java.time.Instant/now))]
(ports/enqueue-job! job-queue :default new-job))Priority levels: :critical, :high, :normal (the default), :low.
Give a job a :unique-key and the queue refuses another job with that key
while the first is queued, scheduled or running: enqueue-job! returns nil
instead of the job id. A retry keeps the key; it is free once the job succeeds
or fails for good.
(if (ports/enqueue-job! job-queue :default
(job/create-job {:job-type :geo/refresh :unique-key "geo/refresh"}
(random-uuid) (java.time.Instant/now)))
:started
:already-running)(require '[wagoe.jobs.shell.worker :as worker])
;; Creates and starts the workers
(def pool (worker/create-worker-pool
{:queue-name :default :worker-count 5 :poll-interval-ms 500}
queue store registry))
(worker/worker-pool-status pool)
(worker/stop-worker-pool! pool)Declare cron schedules on :wagoe/jobs. Each tick enqueues a job of that
type; the handler is the one registered for it.
:wagoe/jobs {:provider :db
:recurring {:nightly-report {:cron "0 2 * * *"
:zone "Europe/Amsterdam"
:args {:kind :sales}}}
:scheduler {:tick-ms 5000 :lease-ms 30000 :zone "UTC"}}Or at runtime, on the :wagoe/job-scheduler component:
(require '[wagoe.jobs.shell.recurring :as recurring])
(recurring/schedule-recurring! scheduler :nightly-report "0 2 * * *"
{:zone "Europe/Amsterdam" :queue :reports})Expressions have five fields (minute, hour, day of month, month, day of
week) with *, lists, ranges, steps and names (MON, JAN), plus
@hourly, @daily, @weekly, @monthly and @yearly. A malformed
expression, an unknown zone, or one that never fires fails the boot. A job
type with no handler on the node is logged as a warning by the lease holder,
on its first tick, with the types that have one: it does not fail the boot, because a handler may be
registered after boot or run only on worker nodes.
How it behaves:
One replica enqueues. Every replica runs a ticker; the one holding a
lease enqueues (a database row timed by the database clock under :db,
SET NX PX under :redis). Each tick is also claimed as an idempotency key,
job type plus scheduled instant, so two tickers overlapping after a lease
handover still enqueue it once.
Zones are wall-clock. The zone defaults to UTC. A time skipped by a DST jump runs at the shifted instant; a time that occurs twice runs once.
Missed ticks run at most once. After downtime a schedule runs its latest missed tick once and skips the earlier ones. A new schedule does not run for ticks before it was registered.
A tick is enqueued at most once. A process that dies between claiming a tick and enqueueing it loses that tick.
:provider :memory keeps all of this in one process.
enqueue-once! enqueues unless the job type and key were claimed before, so a
retried request does not queue the same work twice. It returns the job id, or
nil for a duplicate. Claims last a week unless :ttl-ms says otherwise.
(recurring/enqueue-once! job-queue idempotency-store :default receipt-job
(str "receipt/" order-id))The store behind it, :wagoe/idempotency-store, implements
wagoe.jobs.ports/IIdempotencyStore (claim-key!, release-key!) on the
jobs provider’s backend, for any module to use.
The worker reports how every run of a handler ended, for every job type, Wagoe’s own included. Nothing to wrap.
:wagoe/jobs {:provider :db
:on-outcome my.app.jobs/on-outcome ; optional, (fn [outcome])
:report-failures-every-ms 3600000 ; default: one hour
:expect-success {:nightly-report 26}} ; job type -> hoursEach outcome goes to:
The job store (IJobOutcomes, on memory, db and Redis): one record per job
type, so a watcher needs no table of its own.
(ports/job-outcome (:wagoe/job-store system) :reports/generate)
;; => {:job-type :reports/generate :last-outcome :failure
;; :last-started-at #inst ... :last-duration-ms 840
;; :last-succeeded-at #inst ... :last-failed-at #inst ...
;; :last-error "Connection refused" :failures-since-success 3}:wagoe/metrics: counter jobs.runs (tags job-type, outcome),
histogram jobs.run.duration in seconds, and gauge jobs.last-success, the
epoch second of the last success. Tagged job-type.
:wagoe/error-reporting: the first failure after a success, then one per
:report-failures-every-ms while the type keeps failing. Across nodes: the
throttle lives in :wagoe/idempotency-store.
:on-outcome: {:job-id :job-type :queue :attempt :outcome :started-at
:finished-at :duration-ms}, and on a failure :error (redacted, as stored)
and :retrying?.
With no metrics or error tracker configured both are no-ops. A hook or store
that throws is logged; the job’s own result stands. A handler that throws an
Error, such as an AssertionError, is a failed run like any other.
:expect-success names job types that must succeed within so many hours. The
scheduler checks every minute and reports a late one to the error tracker,
through the same throttle as its failures. A type that never ran, or never
succeeded, is not late: its failures are reported already. Nor is a type this
node has no handler for, such as one switched off by config. A report the
tracker refuses is logged and tried again on the next check.
To list only the job types this node has handlers for:
(outcomes/registered-outcomes job-store job-registry) ; wagoe.jobs.shell.outcomesFailed jobs retry with exponential backoff: the first retry waits two seconds
and each next one twice as long, capped at a minute, plus up to 100 ms of jitter. The worker’s
:retry-config (:backoff-strategy, :initial-delay-ms, :max-delay-ms,
:jitter) changes it. After max-retries attempts (default 3) the job is failed for
good and lands in the dead letter queue.
Stats are stored as Redis keys: jobs:stats:global and jobs:stats:queue:{name}.
clojure -M:test :jobsCan you improve this documentation? These fine people already did:
thijscreemers & Thijs CreemersEdit on GitHub
cljdoc builds & hosts documentation for Clojure/Script libraries
| Ctrl+k | Jump to recent docs |
| ← | Move to previous article |
| → | Move to next article |
| Ctrl+/ | Jump to the search field |