Liking cljdoc? Tell your friends :D

wagoe.jobs.shell.adapters.db

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.
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.
raw docstring

wagoe.jobs.shell.adapters.in-memory

In-memory job queue implementation for development and testing.

Uses Clojure atoms and data structures for job queuing. This adapter is suitable for:

  • Local development without Redis
  • Fast unit testing
  • CI/CD pipelines
  • Learning and tutorials

NOT suitable for:

  • Production use (no persistence, single-process only)
  • Distributed systems (not shared across processes)
  • High-volume job processing (limited by memory)

State is stored in atoms with the following structure:

  • jobs: Map of job-id -> job
  • queues: Map of queue-name -> priority-map of [priority job-ids]
  • scheduled: Sorted set of [execute-at job-id] pairs
  • failed: Vector of failed job-ids
  • stats: Map of queue-name -> statistics
In-memory job queue implementation for development and testing.

Uses Clojure atoms and data structures for job queuing. This adapter is
suitable for:
- Local development without Redis
- Fast unit testing
- CI/CD pipelines
- Learning and tutorials

NOT suitable for:
- Production use (no persistence, single-process only)
- Distributed systems (not shared across processes)
- High-volume job processing (limited by memory)

State is stored in atoms with the following structure:
- jobs: Map of job-id -> job
- queues: Map of queue-name -> priority-map of [priority job-ids]
- scheduled: Sorted set of [execute-at job-id] pairs
- failed: Vector of failed job-ids
- stats: Map of queue-name -> statistics
raw docstring

wagoe.jobs.shell.adapters.redis

Redis-backed job queue implementation.

Uses Redis for distributed job queuing with the following Redis data structures:

  • Sorted Sets: For scheduled jobs (scored by execute-at timestamp)
  • Lists: For priority queues (critical, high, normal, low)
  • Hashes: For job data storage
  • Sets: For tracking workers

This adapter provides production-grade job queuing with:

  • Distributed queue across multiple workers
  • Priority-based job processing
  • Scheduled job execution
  • Job persistence
  • Atomic operations
Redis-backed job queue implementation.

Uses Redis for distributed job queuing with the following Redis data structures:
- Sorted Sets: For scheduled jobs (scored by execute-at timestamp)
- Lists: For priority queues (critical, high, normal, low)
- Hashes: For job data storage
- Sets: For tracking workers

This adapter provides production-grade job queuing with:
- Distributed queue across multiple workers
- Priority-based job processing
- Scheduled job execution
- Job persistence
- Atomic operations
raw docstring

wagoe.jobs.shell.module-wiring

Integrant wiring for the jobs module.

Config keys:

:wagoe/jobs The settings block that switches the module on: {:provider :memory | :db | :redis ; default :memory :lease-ms 60000 ; :db only, in-flight lease :redis {:host "localhost" :port 6379} :workers {:count 1 :queue-name :default}}

:wagoe/jobs-runtime Queue, store and stats from one adapter, so the three share a backend.

:wagoe/job-queue, :wagoe/job-store, :wagoe/job-stats What other modules and the devtools dashboard take a ref to — workflow and push already document :job-queue. Projections of the runtime rather than components of their own, because the in-memory adapter's queue and store must share state. :wagoe/job-stats is nil under :provider :db, which ships no IJobStats.

:wagoe/job-registry The handler map every module contributed. Modules contribute handlers the way they contribute routes: a :job-handlers vector in their ig-config, collected by wagoe.platform.shell.system.config (BOU-330 pattern).

:wagoe/job-workers The worker pool. :count 0 builds none — that is a web node in the web/worker split; the default of 1 is a single process that both enqueues and runs, which is what an application without a deployment topology has.

Until BOU-418 this namespace wired none of it: :wagoe/jobs was a settings passthrough, so java -jar wagoe.jar worker booted and processed nothing.

Integrant wiring for the jobs module.

Config keys:

:wagoe/jobs
  The settings block that switches the module on:
    {:provider :memory | :db | :redis     ; default :memory
     :lease-ms 60000                      ; :db only, in-flight lease
     :redis    {:host "localhost" :port 6379}
     :workers  {:count 1 :queue-name :default}}

:wagoe/jobs-runtime
  Queue, store and stats from one adapter, so the three share a backend.

:wagoe/job-queue, :wagoe/job-store, :wagoe/job-stats
  What other modules and the devtools dashboard take a ref to — `workflow`
  and `push` already document `:job-queue`. Projections of the runtime rather
  than components of their own, because the in-memory adapter's queue and
  store must share state. `:wagoe/job-stats` is nil under `:provider :db`,
  which ships no IJobStats.

:wagoe/job-registry
  The handler map every module contributed. Modules contribute handlers the
  way they contribute routes: a `:job-handlers` vector in their `ig-config`,
  collected by `wagoe.platform.shell.system.config` (BOU-330 pattern).

:wagoe/job-workers
  The worker pool. `:count 0` builds none — that is a web node in the
  web/worker split; the default of 1 is a single process that both enqueues
  and runs, which is what an application without a deployment topology has.

Until BOU-418 this namespace wired none of it: `:wagoe/jobs` was a settings
passthrough, so `java -jar wagoe.jar worker` booted and processed nothing.
raw docstring

wagoe.jobs.shell.tenant-context

Tenant context support for background job processing.

This namespace provides utilities for executing background jobs within tenant-specific database schema contexts. It integrates with the multi-tenant architecture by extracting tenant-id from job metadata and setting the appropriate PostgreSQL search_path before job execution.

Key Features:

  • Extract tenant-id from job metadata
  • Set database search_path to tenant schema
  • Execute job handlers in tenant context
  • Automatic fallback to public schema if no tenant

Usage: (require '[wagoe.jobs.shell.tenant-context :as tenant-jobs])

;; Enqueue job with tenant context (tenant-jobs/enqueue-tenant-job! job-queue tenant-id :send-email {:to "user@example.com"})

;; Process job with tenant context (tenant-jobs/process-tenant-job! job handler-fn db-ctx tenant-service)

See ADR-004 for architecture details (lines 525-554).

Tenant context support for background job processing.

This namespace provides utilities for executing background jobs within
tenant-specific database schema contexts. It integrates with the multi-tenant
architecture by extracting tenant-id from job metadata and setting the
appropriate PostgreSQL search_path before job execution.

Key Features:
- Extract tenant-id from job metadata
- Set database search_path to tenant schema
- Execute job handlers in tenant context
- Automatic fallback to public schema if no tenant

Usage:
  (require '[wagoe.jobs.shell.tenant-context :as tenant-jobs])
  
  ;; Enqueue job with tenant context
  (tenant-jobs/enqueue-tenant-job! job-queue tenant-id :send-email
                                   {:to "user@example.com"})
  
  ;; Process job with tenant context
  (tenant-jobs/process-tenant-job! job handler-fn db-ctx tenant-service)

See ADR-004 for architecture details (lines 525-554).
raw docstring

wagoe.jobs.shell.worker

Background job worker implementation.

Workers poll job queues, execute jobs, and handle retries. This module provides a production-ready worker with:

  • Graceful shutdown
  • Configurable polling intervals
  • Automatic retry handling
  • Scheduled job processing
  • Job handler registry
  • Comprehensive error handling
Background job worker implementation.

Workers poll job queues, execute jobs, and handle retries. This module
provides a production-ready worker with:
- Graceful shutdown
- Configurable polling intervals
- Automatic retry handling
- Scheduled job processing
- Job handler registry
- Comprehensive error handling
raw docstring

cljdoc builds & hosts documentation for Clojure/Script libraries

Keyboard shortcuts
Ctrl+kJump to recent docs
Move to previous article
Move to next article
Ctrl+/Jump to the search field
× close