Background Jobs — Wagoe Docs

The jobs library provides background job processing with priority queues, scheduled execution, automatic retries with exponential backoff, and multi-tenant support. :provider picks the backend: :memory (the default), :db or :redis.

Defining a job handler

;; Handlers receive an args map, return a result map
(defn send-welcome-email [args]
  (let [{:keys [user-id email]} args]
    (email/send! {:to email :subject "Welcome!"})
    {:success? true :result {:sent-to email}}))

;; A failing handler
(defn process-payment [args]
  {:success? false
   :error {:message "Payment gateway unavailable" :type :gateway-error}})

Registering handlers

A module contributes its handlers from its ig-config, as a :job-handlers vector of maps from job type to handler, or refs to components returning one. :wagoe/job-registry registers them all.

(defn ig-config [_settings _ctx]
  {:job-handlers [{:send-welcome-email send-welcome-email
                   :process-payment    process-payment}]})

Outside a system, register them yourself:

(require '[wagoe.jobs.shell.worker :as worker]
         '[wagoe.jobs.ports :as ports])

(def registry (worker/create-job-registry))
(ports/register-handler! registry :send-welcome-email send-welcome-email)

Enqueueing jobs

(require '[wagoe.jobs.core.job :as job])

(let [new-job (job/create-job {:job-type  :send-welcome-email
                                :args      {:user-id  (random-uuid)
                                            :email    "alice@example.com"}
                                :priority  :high
                                :max-retries 3}
                               (random-uuid)
                               (java.time.Instant/now))]
  (ports/enqueue-job! job-queue :default new-job))

job-queue is the :wagoe/job-queue component. Priority levels: :critical, :high, :normal (the default), :low.

Starting workers

;; 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/stop-worker-pool! pool)

In a running system :wagoe/job-workers does this from the config below.

Integrant configuration

;; config.edn
:wagoe/jobs {:provider :redis                       ; :memory (default) | :db | :redis
             :redis    {:host "localhost" :port 6379}
             :workers  {:count 5 :queue-name :default}}

:count 0 starts no workers: a node that only enqueues. :workers {:queues [:default :push]} runs a pool per queue.

Retry logic

Failed jobs are retried with exponential backoff: the first retry waits two seconds, each next one twice as long, capped at a minute, plus up to 100 ms of jitter. :workers {:retry-config …} changes it. After max-retries attempts (default 3) the job fails for good and lands in the dead letter queue.

Multi-tenant jobs

;; With the tenant module on, the job runs in that tenant's schema
(job/create-job {:job-type :sync-inventory
                 :args     {:product-id sku}
                 :metadata {:tenant-id "acme-corp"}}
                (random-uuid)
                (java.time.Instant/now))

Without the tenant module such a job fails.

See jobs library for the full reference.