Database-backed job queue (IJobQueue) on next.jdbc.
Lets an app run durable background jobs on its existing SQL database (H2 or PostgreSQL) without standing up Redis, and survive a Redis outage. Implements the same reliable-dequeue contract as the Redis adapter:
dequeue-job! atomically claims the next ready row (sets status=processing,
locked_by/locked_at) so a worker that crashes mid-job does not lose it.ack-job! removes the row once the worker is done with it.reclaim-abandoned-jobs! returns rows whose lease has expired
(locked_at older than lease-ms) back to ready, i.e. at-least-once.Portable SQL only (no SELECT ... FOR UPDATE SKIP LOCKED): the claim is a
SELECT of the best candidate followed by a conditional UPDATE ... WHERE status='ready'. Concurrent workers that pick the same row have their UPDATEs
serialized by the row lock — exactly one wins (update count 1); the losers
retry the next candidate. This runs identically on H2 and PostgreSQL, so the
reliability tests run on the default in-memory H2 test database.
The authoritative job value is the JSON payload column (same wire format as
the Redis adapter — keyword fields restored, instants as epoch-millis). The
indexed columns (queue, priority_rank, status, execute_at, locked_at) exist
only for ordering, claiming, and reclaim.
Caveats:
lease-ms is reclaimed and may run again concurrently. Set :lease-ms
comfortably above your longest job, and/or keep handlers idempotent.
(Lease renewal is a possible future addition.)payload is VARCHAR(1000000); keep serialized jobs under ~1 MB.Database-backed job queue (IJobQueue) on next.jdbc. Lets an app run durable background jobs on its existing SQL database (H2 or PostgreSQL) without standing up Redis, and survive a Redis outage. Implements the same reliable-dequeue contract as the Redis adapter: - `dequeue-job!` atomically claims the next ready row (sets status=processing, locked_by/locked_at) so a worker that crashes mid-job does not lose it. - `ack-job!` removes the row once the worker is done with it. - `reclaim-abandoned-jobs!` returns rows whose lease has expired (locked_at older than `lease-ms`) back to `ready`, i.e. at-least-once. Portable SQL only (no `SELECT ... FOR UPDATE SKIP LOCKED`): the claim is a SELECT of the best candidate followed by a conditional `UPDATE ... WHERE status='ready'`. Concurrent workers that pick the same row have their UPDATEs serialized by the row lock — exactly one wins (update count 1); the losers retry the next candidate. This runs identically on H2 and PostgreSQL, so the reliability tests run on the default in-memory H2 test database. The authoritative job value is the JSON `payload` column (same wire format as the Redis adapter — keyword fields restored, instants as epoch-millis). The indexed columns (queue, priority_rank, status, execute_at, locked_at) exist only for ordering, claiming, and reclaim. Caveats: - The lease is fixed, not renewed (no heartbeat): a job still running after `lease-ms` is reclaimed and may run again concurrently. Set `:lease-ms` comfortably above your longest job, and/or keep handlers idempotent. (Lease renewal is a possible future addition.) - `payload` is `VARCHAR(1000000)`; keep serialized jobs under ~1 MB.
(create-db-job-queue ds & {:keys [lease-ms] :or {lease-ms default-lease-ms}})Create a DB-backed job queue over the given next.jdbc datasource ds.
:lease-ms (default 60s) is the in-flight lease after which a claimed job is
reclaimable. Call create-jobs-table! once before use (or run the migration).
Create a DB-backed job queue over the given `next.jdbc` datasource `ds`. `:lease-ms` (default 60s) is the in-flight lease after which a claimed job is reclaimable. Call `create-jobs-table!` once before use (or run the migration).
(create-jobs-table! ds)Create the job_queue table + ready-scan index if absent. Portable DDL
(H2 + PostgreSQL). Call once at startup, or ship as a migration.
Create the `job_queue` table + ready-scan index if absent. Portable DDL (H2 + PostgreSQL). Call once at startup, or ship as a migration.
How long a claimed job may stay in-flight before reclaim-abandoned-jobs!
considers its worker dead and returns it to the ready queue.
How long a claimed job may stay in-flight before `reclaim-abandoned-jobs!` considers its worker dead and returns it to the ready queue.
(enqueue-in-tx! tx queue-name job)Transactional (outbox) enqueue: insert the job into the queue within the caller's open transaction, so the job commits atomically with the business change. If the business transaction rolls back, no job is left behind; if it commits, a worker picks the job up. This is the durable alternative to a dual-write (enqueueing to an external queue separately from the DB commit, where a crash between the two loses the job or runs it for a rolled-back change).
tx is a next.jdbc transactable already inside a transaction, e.g.:
(jdbc/with-transaction [tx ds] (orders/create! tx order) (db/enqueue-in-tx! tx :emails receipt-job))
Because the queue is a table in the same database, no separate outbox table
or relay is needed — the queue row is the outbox row. tx MUST be a
transaction the caller manages (do not pass a bare datasource, or the enqueue
would autocommit and defeat the point). Returns the job id.
Transactional (outbox) enqueue: insert the job into the queue **within the
caller's open transaction**, so the job commits atomically with the business
change. If the business transaction rolls back, no job is left behind; if it
commits, a worker picks the job up. This is the durable alternative to a
dual-write (enqueueing to an external queue separately from the DB commit,
where a crash between the two loses the job or runs it for a rolled-back
change).
`tx` is a next.jdbc transactable already inside a transaction, e.g.:
(jdbc/with-transaction [tx ds]
(orders/create! tx order)
(db/enqueue-in-tx! tx :emails receipt-job))
Because the queue is a table in the same database, no separate outbox table
or relay is needed — the queue row *is* the outbox row. `tx` MUST be a
transaction the caller manages (do not pass a bare datasource, or the enqueue
would autocommit and defeat the point). Returns the job id.(insert-job! connectable queue-name job)INSERT a job row using the given connectable — a datasource, a connection,
or (the point of this fn) an open next.jdbc transaction. Returns the job id.
Used by enqueue-job! (autocommit via the adapter's datasource) and by
enqueue-in-tx! (the caller's transaction, for a transactional enqueue).
INSERT a job row using the given `connectable` — a datasource, a connection, or (the point of this fn) an open `next.jdbc` transaction. Returns the job id. Used by `enqueue-job!` (autocommit via the adapter's datasource) and by `enqueue-in-tx!` (the caller's transaction, for a transactional enqueue).
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 |