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 explores what happens when the JVM receives a shutdown signal (Ctrl-C) while a job is actively running. You will enqueue three different kinds of long-running jobs — a Thread/sleep blocker, a core.async/<!! blocker, and a CPU-bound tight loop — and observe that only blocking operations can be interrupted immediately. It also introduces Proletarian’s advanced handler mode, in which the handler function receives the full job map rather than separate job-type and payload arguments.

​
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 other examples, start fresh for accurate summary counts: 
    make examples.db.recreate
This example uses advanced handler mode (:proletarian/handler-fn-mode :advanced). In this mode the handler function receives a single map containing all job details (:proletarian.job/job-type, :proletarian.job/payload, etc.) rather than separate job-type and payload positional arguments. See the worker and handler source below for how this looks in practice.

​
Walkthrough

1

Run the queue worker

In terminal window 1, start the queue worker:
clj -X:examples example-c.worker/run
The worker polls the default queue every 1 second:
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 worker for :proletarian/default queue with polling interval 1 s
: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 on, until you press Ctrl-C]
Leave this running while you work through the steps below.
2

Enqueue a blocking job (Thread/sleep)

In terminal window 2, enqueue a job that sleeps for 10 seconds using Thread/sleep:
clj -X:examples example-c.enqueue-jobs/run-1
Output:
Adding new blocking job to :proletarian/default queue:
{:job-id #uuid "...",
 :job-type :example-c.enqueue-jobs/blocking-job,
 :payload {:sleep-ms 10000, :timestamp #inst "..."}}
Switch to terminal window 1. Within a second the worker picks up the job and starts sleeping:
:proletarian.worker/handling-job {:job-type :example-c.enqueue-jobs/blocking-job, ...}
Running job :example-c.enqueue-jobs/blocking-job. Payload:
{:sleep-ms 10000, :timestamp #inst "..."}
Sleeping 10000...
If you press Ctrl-C now, you should observe the following events:
  :proletarian.executor/shutting-down
  :proletarian.worker/job-interrupted
  :proletarian.executor/completed-shutdown

The current job (timestamp: ...) should then be picked up
again when restarting the worker process.
If you don't interrupt, then the job will finish and won't be run again.
3

Interrupt the blocking job with Ctrl-C

While the worker is sleeping, press Ctrl-C in terminal window 1. Proletarian interrupts the sleeping thread and shuts down gracefully:
:proletarian.executor/shutting-down {:proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
:proletarian.worker/job-interrupted {:job-type :example-c.enqueue-jobs/blocking-job, ...}
:proletarian.executor/completed-shutdown {:proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
The job was not marked complete, so it remains in the queue and will be retried on the next worker startup.
4

Restart the worker and verify the job re-runs

Start the worker again in terminal window 1:
clj -X:examples example-c.worker/run
You should see the same job (identified by its original timestamp) picked up and this time run to completion — assuming you do not interrupt it again.
5

Enqueue a core.async blocking job

In terminal window 2, enqueue a job that blocks on core.async/<!! for 10 seconds:
clj -X:examples example-c.enqueue-jobs/run-2
Output:
Adding new core.async blocking job to :proletarian/default queue:
{:job-id #uuid "...",
 :job-type :example-c.enqueue-jobs/async-blocking-job,
 :payload {:timestamp #inst "...", :wait-ms 10000}}
Switch to terminal window 1 and watch the worker pick it up:
:proletarian.worker/handling-job {:job-type :example-c.enqueue-jobs/async-blocking-job, ...}
Running job :example-c.enqueue-jobs/async-blocking-job. Payload:
{:timestamp #inst "...", :wait-ms 10000}
Waiting on core.async/<!! for 10000 ms...
If you press Ctrl-C now, you should observe the following events:
  :proletarian.executor/shutting-down
  :proletarian.worker/job-interrupted
  :proletarian.executor/completed-shutdown

The current job (timestamp: ...) should then be picked up
again when restarting the worker process.
If you don't interrupt, then the job will finish and won't be run again.
6

Interrupt the core.async job with Ctrl-C

Press Ctrl-C while the job is waiting. Just like Thread/sleep, core.async/<!! responds to thread interrupts — the job is cancelled immediately:
:proletarian.executor/shutting-down {:proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
:proletarian.worker/job-interrupted {...}
:proletarian.executor/completed-shutdown {:proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
Restart the worker and the job will be picked up and completed as before.
7

Enqueue a CPU-bound job

In terminal window 2, enqueue a job that busy-loops for 10 seconds — no blocking calls, no I/O:
clj -X:examples example-c.enqueue-jobs/run-3
Output:
Adding new time-consuming job to :proletarian/default queue:
{:job-id #uuid "...",
 :job-type :example-c.enqueue-jobs/cpu-bound-job,
 :payload {:run-ms 10000, :timestamp #inst "..."}}
Switch to terminal window 1:
:proletarian.worker/handling-job {:job-type :example-c.enqueue-jobs/cpu-bound-job, ...}
Running job :example-c.enqueue-jobs/cpu-bound-job. Payload:
{:run-ms 10000, :timestamp #inst "..."}
This job is CPU-bound. We cannot interrupt/stop such a job. It will run until it
is finished, but the workers will not pick up any more jobs while it is shutting down.
Running for 10000 ms...
If you press Ctrl-C now, you should observe the following events:
  :proletarian.executor/shutting-down
  (Pause until job finishes)
  :proletarian.worker/job-finished
  :proletarian.executor/completed-shutdown
8

Try to interrupt the CPU-bound job with Ctrl-C

Press Ctrl-C in terminal window 1 while the tight loop is running. This time the shutdown is not instant:
:proletarian.executor/shutting-down {:proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
Done.
:proletarian.worker/job-finished {:job-type :example-c.enqueue-jobs/cpu-bound-job, ...}
:proletarian.executor/completed-shutdown {:proletarian.worker/queue-worker-id proletarian[:proletarian/default]}
Notice the pause between :proletarian.executor/shutting-down and :proletarian.worker/job-finished. Proletarian signals the thread to stop, but because the job never calls a blocking operation the interrupt flag is never checked. The worker waits for the job to finish naturally before completing shutdown. No new jobs are picked up in the meantime.
CPU-bound jobs that do not yield to the JVM thread interrupt mechanism will delay a graceful shutdown for as long as they take to complete. Keep this in mind when designing long-running handlers — prefer blocking I/O over tight loops wherever possible.

​
Source code

​
Worker (example_c/worker.clj)

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

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

(defn run
  [_]
  (let [ds (jdbc/get-datasource (:jdbc-url common/config))]
    (common/preamble ds)
    (println "Starting worker for :proletarian/default queue with polling interval 1 s")
    (let [worker (worker/create-queue-worker ds
                                             enqueue-jobs/handle-job!
                                             {:proletarian/polling-interval-ms        1000
                                              :proletarian/on-shutdown                (partial on-shutdown ds)
                                              :proletarian/install-jvm-shutdown-hook? true
                                              :proletarian/handler-fn-mode            :advanced})]
      (worker/start! worker)
      worker)))

​
Enqueue jobs and handlers (example_c/enqueue_jobs.clj)

(ns example-c.enqueue-jobs
  (:require [clojure.core.async :as async]
            [examples.common :as common]
            [next.jdbc :as jdbc]
            [proletarian.job :as job]
            [puget.printer :as puget])
  (:import (java.time Instant)))

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

(defn run-1
  "Enqueue a blocking job that can be interrupted."
  [_]
  (let [ds (jdbc/get-datasource (:jdbc-url common/config))]
    (common/preamble ds)
    (println "Adding new blocking job to :proletarian/default queue:")
    (let [conn (jdbc/get-connection ds)
          job-type ::blocking-job
          payload {:sleep-ms 10000
                   :timestamp (Instant/now)}
          job-id (job/enqueue! conn job-type payload)]
      (puget/cprint {:job-id job-id
                     :job-type ::blocking-job
                     :payload payload}))))

(defn run-2
  "Enqueue a blocking job that uses core.async/<!! and can be interrupted."
  [_]
  (let [ds (jdbc/get-datasource (:jdbc-url common/config))]
    (common/preamble ds)
    (println "Adding new core.async blocking job to :proletarian/default queue:")
    (let [conn (jdbc/get-connection ds)
          job-type ::async-blocking-job
          payload {:wait-ms 10000
                   :timestamp (Instant/now)}
          job-id (job/enqueue! conn job-type payload)]
      (puget/cprint {:job-id job-id
                     :job-type job-type
                     :payload payload}))))

(defn run-3
  "Enqueue a CPU-bound job that can not be interrupted."
  [_]
  (let [ds (jdbc/get-datasource (:jdbc-url common/config))]
    (common/preamble ds)
    (println "Adding new time-consuming job to :proletarian/default queue:")
    (let [conn (jdbc/get-connection ds)
          job-type ::cpu-bound-job
          payload {:run-ms 10000
                   :timestamp (Instant/now)}
          job-id (job/enqueue! conn job-type payload)]
      (puget/cprint {:job-id job-id
                     :job-type ::cpu-bound-job
                     :payload payload}))))

(defmulti handle-job!
  "This multimethod is passed as an argument to create-queue-worker as the job handler function.
   Note that the worker has been configured with :proletarian/handler-fn-mode set to :advanced,
   so this multimethod accepts only one argument: a map with all the job details."
  (fn [job] (:proletarian.job/job-type job)))

(defmethod handle-job! :default
  [{:proletarian.job/keys [job-type payload]}]
  (throw (ex-info (format "handle-job! multimethod not implemented for job-type '%s'" job-type)
                  {:job-type job-type
                   :payload payload})))

(defmethod handle-job! ::blocking-job
  [{{:keys [sleep-ms timestamp] :as payload} :proletarian.job/payload}]
  (println (str "Running job " ::blocking-job ". Payload:"))
  (puget/cprint payload)
  (println (str "Sleeping " sleep-ms "..."))
  (println "If you press Ctrl-C now, you should observe the following events:")
  (println " " :proletarian.executor/shutting-down)
  (println " " :proletarian.worker/job-interrupted)
  (println " " :proletarian.executor/completed-shutdown)
  (println)
  (println (format "The current job (timestamp: %s) should then be picked up\nagain when restarting the worker process."
                   timestamp))
  (println "If you don't interrupt, then the job will finish and won't be run again.")
  (println)
  (Thread/sleep ^long sleep-ms)
  (println "Done."))

(defmethod handle-job! ::async-blocking-job
  [{{:keys [wait-ms timestamp] :as payload} :proletarian.job/payload}]
  (println (str "Running job " ::async-blocking-job ". Payload:"))
  (puget/cprint payload)
  (let [ch (async/chan)]
    (println (str "Waiting on core.async/<!! for " wait-ms " ms..."))
    (println "If you press Ctrl-C now, you should observe the following events:")
    (println " " :proletarian.executor/shutting-down)
    (println " " :proletarian.worker/job-interrupted)
    (println " " :proletarian.executor/completed-shutdown)
    (println)
    (println (format "The current job (timestamp: %s) should then be picked up\nagain when restarting the worker process."
                     timestamp))
    (println "If you don't interrupt, then the job will finish and won't be run again.")
    (println)
    (async/<!! (async/timeout wait-ms))
    (async/close! ch))
  (println "Done."))

(defmethod handle-job! ::cpu-bound-job
  [{{:keys [run-ms _timestamp] :as payload} :proletarian.job/payload}]
  (println (str "Running job " ::cpu-bound-job ". Payload:"))
  (puget/cprint payload)
  (println "This job is CPU-bound. We cannot interrupt/stop such a job. It will run until it\nis finished, but the workers will not pick up any more jobs while it is shutting down.")
  (println (str "Running for " run-ms " ms..."))
  (println "If you press Ctrl-C now, you should observe the following events:")
  (println " " :proletarian.executor/shutting-down)
  (println "  (Pause until job finishes)")
  (println " " :proletarian.worker/job-finished)
  (println " " :proletarian.executor/completed-shutdown)
  (println)
  (let [start-time (System/currentTimeMillis)]
    (loop [t (System/currentTimeMillis)]
      (when (< t (+ start-time run-ms))
        (recur (System/currentTimeMillis)))))
  (println "Done."))

​
How it works

The key difference between the three job types is how Java’s thread interrupt mechanism interacts with each kind of work.
Job typeOperationInterruptible?Outcome on Ctrl-C
::blocking-jobThread/sleep✅ YesThrows InterruptedException → job left in queue
::async-blocking-jobcore.async/<!!✅ YesThrows InterruptedException → job left in queue
::cpu-bound-jobtight loop❌ NoWorker waits for natural completion before shutting down
Blocking operations — Thread/sleep, I/O reads, JDBC queries, core.async/<!!, and any other call that parks the calling thread — respond to thread interrupts by throwing java.lang.InterruptedException. Proletarian catches this, marks the job as interrupted, leaves it in the queue for reprocessing on the next startup, and completes the shutdown immediately. CPU-bound operations — tight loops, heavy computation — never call a blocking primitive, so they never encounter the interrupt flag. Java’s Thread.interrupt() sets the flag but it is not checked, and the job runs to completion regardless. Proletarian waits for it before calling :proletarian/on-shutdown.
Most real-world jobs — HTTP calls, database writes, file I/O — are naturally blocking. This means Proletarian can shut them down quickly and cleanly in the vast majority of cases. Make your handlers idempotent so that a job that is interrupted and later re-run produces the same final result.

Example A: Basics

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

Example B: Retries

Retry strategies, failure callbacks, and graceful shutdown with job summaries.

Build docs developers (and LLMs) love

Get started for freeTalk to us