Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/msolli/proletarian/llms.txt

Use this file to discover all available pages before exploring further.

This example demonstrates how Proletarian handles jobs that fail. You will enqueue a batch of jobs that fail roughly half the time, watch the queue worker retry them according to a configurable retry strategy, and see what happens when a job finally exhausts all its attempts. It also introduces the failed-job-fn callback and the JVM shutdown hook that prints a summary of successful and failed jobs on exit.

​
Prerequisites

  • PostgreSQL 9.5+ installed and running locally (or via Docker — see Example A for the Docker setup).
  • Clojure CLI (clj) version 1.10.1.697 or later, with -X option support. (Installation instructions)
  • Two terminal windows — one for the worker, one for enqueueing jobs.
  • The example database created with make examples.db.install (see Example A). If you have already run Example A, start fresh so the summary counts are accurate: 
    make examples.db.recreate

​
Walkthrough

1

Run the queue worker

In terminal window 1, start the queue worker:
clj -X:examples example-b.worker/run
The worker starts with a 5-second polling interval and 1 worker thread:
Number of jobs in :proletarian/default queue: 0
Number of jobs in proletarian.jobs table: 0
Number of jobs in proletarian.archived_jobs table: 0

Starting workers for :proletarian/default queue
Polling interval: 5 seconds. Worker threads: 1
:proletarian.worker/polling-for-jobs {:worker-thread-id 1, :proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
:proletarian.worker/polling-for-jobs {:worker-thread-id 1, :proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
:proletarian.worker/polling-for-jobs {:worker-thread-id 1, :proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
[...and so forth, until you press Ctrl-C]
Leave this running and continue to the next step.
2

(Optional) Run with multiple threads and a faster polling interval

You can pass keyword arguments on the command line to override the defaults. For example, to use 2 worker threads and poll every second:
clj -X:examples example-b.worker/run :worker-threads 2 :polling-interval 1
Multiple worker threads let you process several jobs in parallel. Combine this with a short polling interval when you want to drain a queue quickly during testing.
3

Enqueue a batch of jobs

In terminal window 2, enqueue 10 jobs at once:
clj -X:examples example-b.enqueue-jobs/run
You should see each job printed as it is inserted:
Adding 10 new jobs to :proletarian/default queue:
{:job-id #uuid "...",
 :job-type :example-b.enqueue-jobs/sometimes-failing,
 :payload {:batch-no 1, :counter 0}}
{:job-id #uuid "...",
 :job-type :example-b.enqueue-jobs/sometimes-failing,
 :payload {:batch-no 1, :counter 1}}
[...8 more jobs...]

Press Enter to enqueue more jobs (Ctrl-C to exit)
Press Enter to enqueue another batch of 10, or Ctrl-C to exit. The enqueue process stays running so you can trigger multiple batches without restarting.
4

Watch retries in the worker terminal

Switch back to terminal window 1. Each job sleeps 0.5–1.5 seconds and then fails approximately 50% of the time.A successful run looks like this:
:proletarian.worker/handling-job {:job-type :example-b.enqueue-jobs/sometimes-failing, :attempt 1, ...}
[  1/0 "Running job :example-b.enqueue-jobs/sometimes-failing. Payload:"]
[  1/0 {:batch-no 1, :counter 0}]
[  1/0 "This will fail 50 % of the time and take on average a second:"]
[  1/0 "Phew, it didn't fail. Done."]
:proletarian.worker/job-finished {:job-type :example-b.enqueue-jobs/sometimes-failing, :attempt 1, ...}
When a job fails, Proletarian logs the exception and schedules a retry based on the retry strategy:
:proletarian.worker/handling-job {..., :attempt 1, ...}
[  1/1 "Running job :example-b.enqueue-jobs/sometimes-failing. Payload:"]
[  1/1 {:batch-no 1, :counter 1}]
[  1/1 "This will fail 50 % of the time and take on average a second:"]
:proletarian.worker/handle-job-exception {:exception #error {
 :cause This operation failed for some reason
 :data {:retry-after 726}
 ...}, ...}
:proletarian.retry/retrying {:retry-at #object[java.time.Instant ...], :retries-left 2, :attempt 1, ...}
If a job exhausts all retries (1 initial attempt + 2 retries = 3 attempts total), the failed-job-fn is called and the job is archived with a :failure status:
:proletarian.retry/not-retrying {:retry-spec {:retries-left 0}, :attempt 3, ...}
[  1/2 "Job failed after 3 attempts (exception message: 'This operation failed for some reason')"]
5

Shut down and view the summary

When you are done observing retries, press Ctrl-C in the worker terminal. The JVM shutdown hook stops the worker gracefully and prints a summary drawn from the proletarian.archived_job table:
:proletarian.executor/completed-shutdown {:proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
Number of successful jobs: 9
Number of failed jobs: 1
Your numbers will vary depending on how the random failures fell for your run.

​
Source code

​
Worker (example_b/worker.clj)

(ns example-b.worker
  (:require [examples.common :as common]
            [example-b.enqueue-jobs :as enqueue-jobs]
            [next.jdbc :as jdbc]
            [proletarian.worker :as worker]))

(defn on-shutdown
  [ds]
  (common/summary ds))

(defn on-polling-error
  [^Throwable t]
  (println (format "Polling error (will retry): [%s] %s" (class t) (.getMessage t)))
  false)

(defn run
  [{:keys [worker-threads polling-interval]
    :or {worker-threads 1
         polling-interval 5}}]
  (let [ds (jdbc/get-datasource (:jdbc-url common/config))]
    (common/preamble ds)
    (println "Starting workers for :proletarian/default queue")
    (println (format "Polling interval: %d seconds. Worker threads: %d" polling-interval worker-threads))
    (let [worker (worker/create-queue-worker ds
                                             enqueue-jobs/handle-job!
                                             #:proletarian{:retry-strategy-fn enqueue-jobs/retry-strategy
                                                           :failed-job-fn enqueue-jobs/handle-failed-job!
                                                           :polling-interval-ms (* 1000 polling-interval)
                                                           :worker-threads worker-threads
                                                           :on-polling-error on-polling-error
                                                           :on-shutdown (partial on-shutdown ds)
                                                           :install-jvm-shutdown-hook? true})]
      (worker/start! worker))))

​
Enqueue jobs (example_b/enqueue_jobs.clj)

(ns example-b.enqueue-jobs
  "In this example we're going to enqueue some jobs that will fail some of the
   time. The jobs will be retried by the workers according to their retry
   strategy."
  (:require [examples.common :as common]
            [next.jdbc :as jdbc]
            [proletarian.job :as job]
            [puget.printer :as puget]))

(set! *warn-on-reflection* true)

(defn run
  [_]
  (let [ds (jdbc/get-datasource (:jdbc-url common/config))]
    (common/preamble ds)
    (let [conn (jdbc/get-connection ds)
          job-type ::sometimes-failing]
      (loop [batch-no 1]
        (println "Adding 10 new jobs to :proletarian/default queue:")
        (dotimes [i 10]
          (let [payload {:batch-no batch-no
                         :counter i}
                job-id (job/enqueue! conn job-type payload)]
            (puget/cprint {:job-id job-id
                           :job-type ::sometimes-failing
                           :payload payload})))
        (println)
        (println "Press Enter to enqueue more jobs (Ctrl-C to exit)")
        (read-line)
        (recur (inc batch-no))))))

(defn do-possibly-failing-thing!
  "Function that sleeps between 500 and 1500 ms, and then throws an exception in about every other invocation.
   If it throws, the exception will have data with the key :retry-after, which is the number of milliseconds after which
   the operation could be retried. This mirrors a common backoff technique in web APIs."
  []
  (Thread/sleep ^long (+ 500 (rand-int 1000)))
  (when (zero? (rand-int 2))
    (throw (ex-info "This operation failed for some reason" {:retry-after (rand-int 1000)}))))

(defn handle-job!
  [_job-type payload]
  (let [{:keys [batch-no counter]} payload
        log #(puget/cprint [(symbol (format "%3d/%1d" batch-no counter)) %])]
    (log (str "Running job " ::sometimes-failing ". Payload:"))
    (log payload)
    (log "This will fail 50 % of the time and take on average a second:")
    (do-possibly-failing-thing!)
    (log "Phew, it didn't fail. Done.")))

(defn retry-strategy
  [_job exception]
  (let [retry-after (-> exception (ex-data) :retry-after)]
    {:retries 2
     :delays [retry-after]}))

(defn handle-failed-job!
  [{:proletarian.job/keys [payload attempts] :as _job} ^Exception exception]
  (let [{:keys [batch-no counter]} payload
        log #(puget/cprint [(symbol (format "%3d/%1d" batch-no counter)) %] {:width 120})]
    (log (str "Job failed after " attempts " attempts (exception message: '" (.getMessage exception) "')"))))

​
How the retry strategy works

The retry strategy function receives the job map and the exception that was thrown, and returns a map with two keys: 
(defn retry-strategy
  [_job exception]
  (let [retry-after (-> exception (ex-data) :retry-after)]
    {:retries 2
     :delays [retry-after]}))
KeyMeaning
:retriesMaximum number of retries, not counting the initial attempt.
:delaysSequence of delays in milliseconds between retries. If there are fewer delays than retries, the last delay is repeated for all remaining attempts.
In this example the delay is read directly from the exception’s :retry-after ex-data, mirroring a common pattern in web APIs that return backoff hints in their error responses. Because the retry strategy function receives the full exception object, you can inspect any aspect of the failure to make informed decisions about whether and when to retry.

​
How the failed job function works

When a job has exhausted all retries it is permanently failed. Proletarian calls the failed-job-fn with the job map and the final exception before archiving the job with a :failure status: 
(defn handle-failed-job!
  [{:proletarian.job/keys [payload attempts] :as _job} ^Exception exception]
  (let [{:keys [batch-no counter]} payload
        log #(puget/cprint [(symbol (format "%3d/%1d" batch-no counter)) %] {:width 120})]
    (log (str "Job failed after " attempts " attempts (exception message: '" (.getMessage exception) "')"))))
This is where you would typically log the failure, fire an alert, write to a dead-letter store, or trigger any other remediation needed when a job cannot be completed.
The failed-job-fn is called once — after all retries are spent. If you need to act on every failed attempt (not just the final one), use the :proletarian/on-polling-error callback instead.

Example A: Basics

Create a queue worker, enqueue a job, and watch it get processed.

Example C: Shutdown

Graceful shutdown, job interruption, and blocking vs. CPU-bound jobs.

Build docs developers (and LLMs) love

Get started for freeTalk to us