Port definitions for background job processing.
This module defines protocols for asynchronous job processing, similar to Sidekiq (Ruby) or Celery (Python). Jobs can be queued, scheduled, and processed asynchronously with retry logic.
Key Features:
Port definitions for background job processing. This module defines protocols for asynchronous job processing, similar to Sidekiq (Ruby) or Celery (Python). Jobs can be queued, scheduled, and processed asynchronously with retry logic. Key Features: - Job enqueueing and scheduling - Priority queues - Retry logic with exponential backoff - Dead letter queue for failed jobs - Job monitoring and statistics
The dead-letter queue's retention policy. Each job store implements it.
The dead-letter queue's retention policy. Each job store implements it.
(purge-dead-letters-before! this cutoff)Delete every dead-lettered job that finally failed before cutoff, an
Instant. Returns the count.
Delete every dead-lettered job that finally failed before `cutoff`, an Instant. Returns the count.
Remembers which (scope, key) pairs have been seen, so a retried or repeated
operation can tell it already ran. Scope and key are strings; jobs uses the
job type as scope. Any module may take :wagoe/idempotency-store.
Remembers which (scope, key) pairs have been seen, so a retried or repeated operation can tell it already ran. Scope and key are strings; jobs uses the job type as scope. Any module may take `:wagoe/idempotency-store`.
(claim-key! this scope key ttl-ms)True when this call is the first to claim (scope, key) within ttl-ms;
false when it is already claimed.
True when this call is the first to claim (scope, key) within `ttl-ms`; false when it is already claimed.
(purge-expired-keys! this)Delete expired claims. Returns the count, or nil where the backend expires them itself.
Delete expired claims. Returns the count, or nil where the backend expires them itself.
(release-key! this scope key)Forget (scope, key), so the next claim succeeds. For an operation that claimed and then failed.
Forget (scope, key), so the next claim succeeds. For an operation that claimed and then failed.
Optional capability of a job store: how each job type last fared, so an application can tell a job that stopped succeeding without keeping a table of its own (BOU-832). The worker records every run when its store has it.
Optional capability of a job store: how each job type last fared, so an application can tell a job that stopped succeeding without keeping a table of its own (BOU-832). The worker records every run when its store has it.
(job-outcome this job-type)job-type's record, or nil before its first run:
{:job-type :last-outcome :last-started-at :last-duration-ms
:last-succeeded-at :last-failed-at :last-error :failures-since-success}
`job-type`'s record, or nil before its first run:
{:job-type :last-outcome :last-started-at :last-duration-ms
:last-succeeded-at :last-failed-at :last-error :failures-since-success}(job-outcomes this)Every job type's record, by job type.
Every job type's record, by job type.
(record-outcome! this outcome)Fold one run's outcome (see wagoe.jobs.core.outcome/run-outcome) into
its job type's record. One write, no read.
Fold one run's outcome (see `wagoe.jobs.core.outcome/run-outcome`) into its job type's record. One write, no read.
Protocol for job queue operations.
Protocol for job queue operations.
(enqueue-job! this queue-name job)Enqueue a job for immediate processing.
Args: queue-name - Queue name (e.g., :default, :critical, :low-priority) job - Job map with {:job-type :job-id :args :metadata}, and optionally :unique-key
Returns: Job ID (UUID), or nil when the job's :unique-key is held by another job still pending or running. The key is freed when that job is acked without being put back (done, or failed for good); a retry keeps it.
Enqueue a job for immediate processing.
Args:
queue-name - Queue name (e.g., :default, :critical, :low-priority)
job - Job map with {:job-type :job-id :args :metadata}, and optionally
:unique-key
Returns:
Job ID (UUID), or nil when the job's :unique-key is held by another job
still pending or running. The key is freed when that job is acked
without being put back (done, or failed for good); a retry keeps it.(schedule-job! this queue-name job execute-at)Schedule a job for future execution.
Args: queue-name - Queue name job - Job map execute-at - Instant when job should execute
Returns: Job ID (UUID), or nil as for enqueue-job!
Schedule a job for future execution. Args: queue-name - Queue name job - Job map execute-at - Instant when job should execute Returns: Job ID (UUID), or nil as for enqueue-job!
(ack-job! this queue-name worker-id job-id)Acknowledge that a worker has finished with a dequeued job (completed, re-enqueued for retry, or dead-lettered), removing it from the worker's in-flight processing list. Must be called for every dequeued job.
Returns: true
Acknowledge that a worker has finished with a dequeued job (completed, re-enqueued for retry, or dead-lettered), removing it from the worker's in-flight processing list. Must be called for every dequeued job. Returns: true
(reclaim-abandoned-jobs! this queue-name)Return jobs stranded in the processing lists of dead workers back to the ready queue so they are re-executed (at-least-once). A worker is considered dead when its heartbeat has expired. Safe and cheap to call periodically.
Returns: {:reclaimed <count>}
Return jobs stranded in the processing lists of dead workers back to the
ready queue so they are re-executed (at-least-once). A worker is considered
dead when its heartbeat has expired. Safe and cheap to call periodically.
Returns:
{:reclaimed <count>}(queue-size this queue-name)Get number of jobs in queue.
Args: queue-name - Queue name
Returns: Integer count
Get number of jobs in queue. Args: queue-name - Queue name Returns: Integer count
(dequeue-job! this queue-name worker-id)Reliably dequeue the next job from queue for a worker.
Atomically moves the job into a per-worker in-flight (processing) list so a worker that crashes mid-job does not lose it — the job stays in the processing list until acked (see ack-job!) or reclaimed (see reclaim-abandoned-jobs!).
Args: queue-name - Queue name worker-id - Id of the worker taking the job (owns the processing list)
Returns: Job map or nil if queue is empty
Reliably dequeue the next job from queue for a worker. Atomically moves the job into a per-worker in-flight (processing) list so a worker that crashes mid-job does not lose it — the job stays in the processing list until acked (see ack-job!) or reclaimed (see reclaim-abandoned-jobs!). Args: queue-name - Queue name worker-id - Id of the worker taking the job (owns the processing list) Returns: Job map or nil if queue is empty
(delete-job! this job-id)Delete a job from queue.
Args: job-id - Job ID
Returns: true if deleted, false if not found
Delete a job from queue. Args: job-id - Job ID Returns: true if deleted, false if not found
(process-scheduled-jobs! this)Process scheduled jobs that are due for execution. Moves jobs from scheduled queue to execution queues.
Returns: Number of jobs moved
Process scheduled jobs that are due for execution. Moves jobs from scheduled queue to execution queues. Returns: Number of jobs moved
(list-queues this)List all queue names.
Returns: Vector of queue names
List all queue names. Returns: Vector of queue names
(peek-job this queue-name)Peek at next job without removing it.
Args: queue-name - Queue name
Returns: Job map or nil
Peek at next job without removing it. Args: queue-name - Queue name Returns: Job map or nil
Protocol for job handler registry.
Protocol for job handler registry.
(get-handler this job-type)Get handler function for job type.
Args: job-type - Job type keyword
Returns: Handler function or nil
Get handler function for job type. Args: job-type - Job type keyword Returns: Handler function or nil
(list-handlers this)List all registered job types.
Returns: Vector of job type keywords
List all registered job types. Returns: Vector of job type keywords
(register-handler! this job-type handler-fn)Register a handler function for job type.
Args: job-type - Keyword identifying job type (e.g., :send-email) handler-fn - Function that processes job: (fn [job-args] ...)
Returns: job-type
Register a handler function for job type. Args: job-type - Keyword identifying job type (e.g., :send-email) handler-fn - Function that processes job: (fn [job-args] ...) Returns: job-type
(unregister-handler! this job-type)Unregister a handler.
Args: job-type - Job type keyword
Returns: true if unregistered
Unregister a handler. Args: job-type - Job type keyword Returns: true if unregistered
Protocol for job statistics and monitoring.
Protocol for job statistics and monitoring.
(job-history this job-type limit)Get recent job history for job type.
Args: job-type - Job type keyword limit - Maximum number of jobs
Returns: Vector of recent jobs with results
Get recent job history for job type. Args: job-type - Job type keyword limit - Maximum number of jobs Returns: Vector of recent jobs with results
(job-stats this)Get overall job statistics.
Returns: Map with: :total-processed - Total jobs processed :total-failed - Total jobs failed :total-succeeded - Total jobs succeeded :queues - Map of queue-name -> {:size :processed :failed} :workers - Map of worker-id -> status
Get overall job statistics.
Returns:
Map with:
:total-processed - Total jobs processed
:total-failed - Total jobs failed
:total-succeeded - Total jobs succeeded
:queues - Map of queue-name -> {:size :processed :failed}
:workers - Map of worker-id -> status(queue-stats this queue-name)Get statistics for specific queue.
Args: queue-name - Queue name
Returns: Stats map {:size :processed :failed :avg-duration}
Get statistics for specific queue.
Args:
queue-name - Queue name
Returns:
Stats map {:size :processed :failed :avg-duration}Protocol for job persistence and history.
Protocol for job persistence and history.
(failed-jobs this limit)Get failed jobs from dead letter queue.
Args: limit - Maximum number of jobs to return
Returns: Vector of failed jobs
Get failed jobs from dead letter queue. Args: limit - Maximum number of jobs to return Returns: Vector of failed jobs
(find-job this job-id)Find job by ID.
Args: job-id - Job ID
Returns: Job map or nil
Find job by ID. Args: job-id - Job ID Returns: Job map or nil
(find-jobs this filters)Find jobs matching filters.
Args: filters - Map with optional keys: :status - Job status :job-type - Job type :queue - Queue name :from-date - Start date :to-date - End date
Returns: Vector of jobs
Find jobs matching filters.
Args:
filters - Map with optional keys:
:status - Job status
:job-type - Job type
:queue - Queue name
:from-date - Start date
:to-date - End date
Returns:
Vector of jobs(retry-job! this job-id)Move failed job back to queue for retry.
Args: job-id - Job ID
Returns: Updated job
Move failed job back to queue for retry. Args: job-id - Job ID Returns: Updated job
(save-job! this job)Persist job to store.
Args: job - Job map
Returns: Saved job with metadata
Persist job to store. Args: job - Job map Returns: Saved job with metadata
(update-job-status! this job-id status result)Update job status and result.
Args: job-id - Job ID status - Status keyword (:pending :running :completed :failed :retrying) result - Result data or error
Returns: Updated job
Update job status and result. Args: job-id - Job ID status - Status keyword (:pending :running :completed :failed :retrying) result - Result data or error Returns: Updated job
Protocol for job worker operations.
Protocol for job worker operations.
(process-job! this job)Process a single job.
Args: job - Job map
Returns: Result map with {:success? :result :error}
Process a single job.
Args:
job - Job map
Returns:
Result map with {:success? :result :error}(start-worker! this)Start worker to process jobs from queue.
Returns: Worker ID
Start worker to process jobs from queue. Returns: Worker ID
(stop-worker! this worker-id)Stop worker by ID.
Args: worker-id - Worker ID
Returns: true if stopped
Stop worker by ID. Args: worker-id - Worker ID Returns: true if stopped
(worker-status this worker-id)Get worker status.
Args: worker-id - Worker ID
Returns: Status map {:status :current-job :processed-count}
Get worker status.
Args:
worker-id - Worker ID
Returns:
Status map {:status :current-job :processed-count}Optional capability: a queue whose claim on a dequeued job expires. The worker renews it while the handler runs, so a live job is never reclaimed and run a second time (BOU-823).
Optional capability: a queue whose claim on a dequeued job expires. The worker renews it while the handler runs, so a live job is never reclaimed and run a second time (BOU-823).
(renew-interval-ms this)How often a running job's claim must be renewed, in milliseconds.
How often a running job's claim must be renewed, in milliseconds.
(renew-lease! this queue-name worker-id job-id)Extend worker-id's claim on job-id. False when the worker no longer
holds it.
Extend `worker-id`'s claim on `job-id`. False when the worker no longer holds it.
What recurring-job tickers on several replicas share: a lease, so one enqueues at a time, and the last tick each schedule fired. Schedule names are strings; jobs uses the job type.
What recurring-job tickers on several replicas share: a lease, so one enqueues at a time, and the last tick each schedule fired. Schedule names are strings; jobs uses the job type.
(acquire-lease! this owner lease-ms)Take the ticker lease for owner, or renew it. True when owner holds it.
Take the ticker lease for `owner`, or renew it. True when `owner` holds it.
(last-tick this schedule-name)The instant schedule-name last fired for, or nil.
The instant `schedule-name` last fired for, or nil.
(record-tick! this schedule-name tick)Record that schedule-name fired for instant tick.
Record that `schedule-name` fired for instant `tick`.
(release-lease! this owner)Give up the lease if owner holds it.
Give up the lease if `owner` holds it.
Optional capability: enqueue a job inside a caller-managed transaction.
Implemented only by queues backed by the same database as the business data.
Use transactional-queue? to check before calling.
Optional capability: enqueue a job inside a caller-managed transaction. Implemented only by queues backed by the same database as the business data. Use `transactional-queue?` to check before calling.
(enqueue-in-tx! this tx queue-name job)Enqueue a job within the caller's open transaction, so the job commits atomically with the business change (outbox semantics: rollback leaves no orphan job; commit means a worker will pick it up).
Args: tx - an OPEN next.jdbc transaction, not a datasource queue-name - Queue name job - Job map
Returns: Job ID (UUID), or nil as for enqueue-job!
Throws:
ex-info {:type :validation-error} when tx is not an open transaction —
passing a datasource would autocommit and silently reintroduce the
dual-write window this exists to close.
Enqueue a job within the caller's open transaction, so the job commits
atomically with the business change (outbox semantics: rollback leaves no
orphan job; commit means a worker will pick it up).
Args:
tx - an OPEN next.jdbc transaction, not a datasource
queue-name - Queue name
job - Job map
Returns:
Job ID (UUID), or nil as for enqueue-job!
Throws:
ex-info {:type :validation-error} when `tx` is not an open transaction —
passing a datasource would autocommit and silently reintroduce the
dual-write window this exists to close.(transactional-queue? q)True when q supports enqueue-in-tx! (i.e. the queue is co-located with the
business database). Lets callers branch instead of discovering the limitation
as a runtime error.
True when `q` supports `enqueue-in-tx!` (i.e. the queue is co-located with the business database). Lets callers branch instead of discovering the limitation as a runtime error.
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 |