wagoe
Docs Why Wagoe Blog Home
Get Started

Background job processing with priority queues, scheduled execution, automatic retries with exponential backoff, and multi-tenant support.

Key namespaces

Namespace Purpose

wagoe.jobs.core.job

Pure functions: job creation, state transitions, retry logic

wagoe.jobs.ports

Protocols: IJobQueue, IJobStore, IJobStats, IJobOutcomes, IJobWorker, IJobRegistry, IIdempotencyStore, IRecurringState, ITransactionalJobQueue, ILeasedJobQueue, IDeadLetterRetention

wagoe.jobs.core.cron

Pure cron parsing and next-run calculation

wagoe.jobs.shell.recurring

Recurring-job ticker and enqueue-once!

wagoe.jobs.schema

Malli schemas: job states, priorities, retry config

wagoe.jobs.shell.adapters.in-memory

In-memory adapter (atoms, for dev/test)

wagoe.jobs.shell.adapters.db

Database adapter (:provider :db)

wagoe.jobs.shell.adapters.redis

Redis adapter (sorted sets, lists, hashes)

wagoe.jobs.shell.worker

Worker implementation: polling, processing, pool management

wagoe.jobs.shell.tenant-context

Multi-tenant job execution with schema switching

Handler signature

;; 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}})

Enqueueing jobs

(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.

One job at a time

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)

Worker pool

(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)

Recurring jobs

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.

Idempotency keys

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.

Job outcomes

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 -> hours

Each 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.outcomes

Retry behaviour

Failed 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.

Redis counters

Stats are stored as Redis keys: jobs:stats:global and jobs:stats:queue:{name}.

Testing

clojure -M:test :jobs