jobs

Background job processing for Kit — PostgreSQL-backed queues, workers, retries, and scheduling

Files

Dependencies

Native Dependencies

Requires a running PostgreSQL server and the libpq client library (see kit-postgres for install instructions).

Installation

kit add gitlab.com/kit-lang/packages/kit-jobs.git

Usage

Define workers and enqueue jobs

import Map

import Kit.Jobs as Jobs
import Kit.Postgres as Postgres

email-worker = Jobs.define-worker "send-email" (fn(args) =>
  to = Map.get "to" args ?? ""
  if String.is-blank? to then
    Err (Jobs.DiscardJob "no recipient")          # permanent failure
  else if send to then
    Ok no-op                                      # success
  else
    Err (Jobs.RetryIn 60 "delivery failed")       # retry in 60s
)
  |> Jobs.worker-queue "email"
  |> Jobs.worker-max-attempts 5

main = fn(-env: Env) =>
  db = Postgres.connect "postgresql://localhost/myapp"
  Jobs.install-schema db

  config = Jobs.new-config db
    |> Jobs.with-workers [email-worker]
    |> Jobs.with-queues ["email"]

  spec = Jobs.job-for email-worker (Map.from-list [("to", "user@example.com")])
    |> Jobs.with-tags ["welcome"]
  Jobs.enqueue-job config spec

  Jobs.run-once config    # one synchronous poll cycle
  Postgres.close db

main

Enqueue options

Jobs.new-job "report" args
  |> Jobs.in-queue "reports"        # queue routing (default: "default")
  |> Jobs.with-priority 1           # 0 highest .. 9 lowest (default: 5)
  |> Jobs.with-max-attempts 7       # retry budget (default: 3)
  |> Jobs.with-tags ["user-42"]     # tags for querying and bulk cancel
  |> Jobs.schedule-in 3600          # run in one hour

Use Jobs.schedule-at "2026-01-15T09:00:00Z" for absolute timestamps.

Run a supervised worker pool

import Auth.Concurrency.{actor-auth, concurrency-auth}
import Auth.Process.{process-auth}

main = fn(env: Env) =>
  db = Postgres.connect "postgresql://localhost/myapp"
  config = Jobs.new-config db
    |> Jobs.with-workers [email-worker]
    |> Jobs.with-queues ["email"]
    |> Jobs.with-pool-size 4
    |> Jobs.with-poll-interval 500

  actors = actor-auth (concurrency-auth env.root)
  process = process-auth env.root
  pool = Jobs.start-pool config actors
  Jobs.run-pool pool 1000000 actors process   # poll loop with sleeps
  Jobs.stop-pool pool actors

main

Each pool actor runs a full poll cycle per :poll message: rescue stuck jobs, stage due jobs, schedule recurring occurrences, claim a batch with FOR UPDATE SKIP LOCKED, and execute the claimed jobs.

Failure outcomes

A worker's perform function returns Result Unit JobFailure:

The default backoff is 2^attempt * 10 seconds (20s, 40s, 80s, ...) capped at one day. Override per worker with Jobs.worker-backoff (fn(attempt) => ...). When the attempt budget is exhausted the job is discarded, keeping every error in the row's history.

Recurring (cron) jobs

entries = match Jobs.new-recurring "nightly-cleanup" cleanup-worker "0 0 * * *" Map.empty
  | Ok entry -> [entry]
  | Err -e -> []
config = Jobs.new-config db |> Jobs.with-recurring entries

Cron expressions use the standard five fields (minute, hour, day-of-month, month, day-of-week) with *, */n, ranges, and lists. Each poll cycle inserts the next occurrence unless one is already pending for that entry name, so recurring configurations are safe to run on multiple nodes. The cron engine is exposed directly as Jobs.parse-cron, Jobs.cron-matches?, Jobs.next-run, and Jobs.iso-from-epoch.

Querying and control

# Query builder
stmt = Jobs.new-query
  |> Jobs.where-state Jobs.Retryable
  |> Jobs.where-queue "email"
  |> Jobs.where-tag "user-42"
  |> Jobs.with-limit 50
  |> Jobs.to-statement
jobs = Jobs.fetch-jobs db stmt

# Control
Jobs.cancel db 123                 # cancel one job
Jobs.cancel-by-tag db "batch-7"    # cancel all pending jobs with a tag
Jobs.release db 123                # move a discarded job back to available
Jobs.prune db (86400 * 7)          # delete finished jobs older than 7 days

# Observability
counts = Jobs.count-by-state db
health = Jobs.health-check config  # counts plus active worker registrations

Telemetry

telemetry = Jobs.new-telemetry
  |> Jobs.on-event (fn(event) =>
    match event
      | Jobs.JobCompleted {job, duration-ms} -> println "job ${job.id} took ${duration-ms}ms"
      | Jobs.JobDiscarded {job, error} -> println "job ${job.id} gave up: ${error}"
      | _ -> no-op
  )
config = Jobs.new-config db |> Jobs.with-telemetry telemetry

Job Concepts

kit-jobs is a database-backed background job system inspired by Rails Solid Queue and Elixir Oban:

• Durable job state — every job lives in PostgreSQL, in one of seven states: scheduled, available, executing, retryable, completed, discarded, cancelled.
• Atomic claiming — workers claim jobs with FOR UPDATE SKIP LOCKED, so any number of processes can poll the same queues without blocking or double-processing.
• Kit-native workers — the worker pool is built on Kit's Actor and Supervisor primitives.
• Retries with backoff — failed jobs retry on an exponential schedule (or a custom per-worker one) until their attempt budget runs out, keeping the full error history in the row.
• Scheduling — one-time delayed jobs, absolute timestamps, and recurring cron entries with idempotent occurrence insertion.
• Telemetry — subscribe to lifecycle events (enqueued, started, completed, failed, discarded, cancelled, poll-failed) for metrics and logging.

Job Lifecycle

scheduled -> available -> executing -> completed
                 |            |
             cancelled    retryable -> discarded

Due scheduled and retryable jobs are staged back to available on every poll cycle. Jobs stuck in executing past the rescue timeout are retried or discarded automatically.

Database Schema

Jobs.install-schema creates two tables (idempotently): kit_jobs (job rows with state, queue, worker, JSONB args and error history, tags, priority, attempt counters, and timestamps) and kit_jobs_workers (worker process registrations with heartbeats for health monitoring). Jobs.drop-schema removes them.

Development

Running Examples

The examples need a running PostgreSQL server with a demo database:

createdb kit_jobs_demo

Run examples with the interpreter:

kit run --allow=ffi,net,actor,process examples/basic.kit
kit run --allow=ffi,net,actor,process examples/email-worker.kit
kit run --allow=ffi,net,actor,process examples/recurring.kit

Compile examples to a native binary:

kit build --allow=ffi,net,actor,process examples/basic.kit && ./basic

Running Tests

The unit tests are pure logic (state machine, backoff, cron, SQL construction, stubbed execution flows) and do not need a database. Run the test suite:

kit test

Run the test suite with coverage:

kit test --coverage

Running kit dev

Run the standard development workflow (format, check, test):

kit dev

This will:

1. Format and check source files in src/
2. Run tests in tests/ with coverage

Generating Documentation

Generate API documentation from doc comments:

kit doc

Note: Kit sources with doc comments (##) will generate HTML documents in docs/*.html

Cleaning Build Artifacts

Remove generated files, caches, and build artifacts:

kit task clean

Note: Defined in kit.toml.

Local Installation

To install this package locally for development:

kit install

This installs the package to ~/.kit/packages/@kit/jobs/, making it available for import as Kit.Jobs in other projects.

License

This package is released under the MIT License - see LICENSE for details.

Exported Functions & Types

define-worker

Define a worker with a name and a perform function.

Defaults: queue "default", priority 5, max-attempts 3, exponential backoff.

String -> (Map String String -> Result Unit JobFailure) -> WorkerDef

worker-queue

Set the default queue for jobs enqueued through this worker.

WorkerDef -> String -> WorkerDef

worker-priority

Set the default priority for jobs enqueued through this worker.

WorkerDef -> Int -> WorkerDef

worker-max-attempts

Set the default retry budget for jobs enqueued through this worker.

WorkerDef -> Int -> WorkerDef

worker-backoff

Set a custom backoff function (attempt number -> delay seconds).

WorkerDef -> (Int -> Int) -> WorkerDef

find-worker

Find a worker definition by name in a registry list.

String -> [WorkerDef] -> Option WorkerDef

job-for

Build an enqueue spec from a worker definition, inheriting its defaults.

WorkerDef -> Map String String -> Spec

jobs-table

Name of the jobs table.

workers-table

Name of the worker registration table.

schema-statements

Idempotent DDL statements that create the job system schema.

The job state is stored as TEXT with a CHECK constraint instead of a PostgreSQL ENUM so the DDL stays idempotent and re-runnable.

[String]

drop-statements

DDL statements that drop the job system schema (for tests and teardown).

[String]

job-columns

Columns selected whenever full job rows are returned. Timestamps and JSONB are cast to text for uniform decoding.

to-pg-text-array

Render a list of strings as a PostgreSQL text[] array literal.

Elements are double-quoted with embedded backslashes and quotes escaped, so the result is safe to pass as a bind parameter cast with ::text[].

[String] -> String

insert-job

Build the INSERT statement for an enqueue spec.

The spec record needs: worker, queue, priority, max-attempts, tags, args-json, and schedule (RunNow | RunIn seconds | RunAt iso-timestamp). Jobs scheduled for later start in the 'scheduled' state; immediate jobs start 'available'.

Spec -> {text: String, binds: [String]}

insert-recurring

Build the INSERT statement for a recurring (cron) job occurrence.

Inserts a scheduled job tagged with the cron entry name in meta, unless an identical pending occurrence already exists. This keeps recurring enqueues idempotent across pollers and nodes.

{name: String, worker: String, queue: String, priority: Int, max-attempts: Int, args-json: String, cron-text: String, ...} -> String -> {text: String, binds: [String]}

stage-jobs

Promote due scheduled and retryable jobs to 'available'.

{text: String, binds: [String]}

claim-jobs

Atomically claim up to limit available jobs from a queue.

Uses FOR UPDATE SKIP LOCKED so concurrent workers never block each other or double-claim a job. Claimed jobs move to 'executing' with attempt counters and attribution updated, and the full job rows are returned.

String -> Int -> String -> {text: String, binds: [String]}

complete-job

Mark an executing job completed.

Int -> {text: String, binds: [String]}

retry-job

Mark an executing job retryable, scheduling the retry after a delay and appending the failure to the error history.

Int -> Int -> String -> {text: String, binds: [String]}

discard-job

Mark an executing job discarded (no attempts remaining or permanent failure), appending the failure to the error history.

Int -> String -> {text: String, binds: [String]}

cancel-job

Cancel a pending or executing job by id.

Int -> {text: String, binds: [String]}

cancel-jobs-by-tag

Cancel all pending jobs carrying a tag.

String -> {text: String, binds: [String]}

release-job

Manually move a discarded, cancelled, or retryable job back to 'available'.

Int -> {text: String, binds: [String]}

rescue-stuck-jobs

Rescue jobs stuck in 'executing' longer than the timeout: discard the ones out of attempts, make the rest retryable immediately.

Int -> {text: String, binds: [String]}

prune-jobs

Delete finished jobs older than the retention window.

Int -> {text: String, binds: [String]}

get-job

Fetch a single job by id.

Int -> {text: String, binds: [String]}

count-by-state

Count jobs grouped by state.

{text: String, binds: [String]}

count-active-workers

Count worker registrations with a recent heartbeat.

Int -> {text: String, binds: [String]}

register-worker

Register (or refresh) a worker process for health monitoring.

{node-name: String, queues: [String], pool-size: Int, ...} -> {text: String, binds: [String]}

heartbeat-worker

Refresh a worker heartbeat.

String -> {text: String, binds: [String]}

deregister-worker

Remove a worker registration on shutdown.

String -> {text: String, binds: [String]}

default-query-limit

Default maximum number of rows returned by a query.

new-query

Create an empty job query (no filters, limit 100).

JobQuery

where-state

Filter by job state. Repeated calls accumulate (OR semantics).

JobQuery -> JobState -> JobQuery

where-queue

Filter by queue name. Repeated calls accumulate (OR semantics).

JobQuery -> String -> JobQuery

where-worker

Filter by worker name. Repeated calls accumulate (OR semantics).

JobQuery -> String -> JobQuery

where-tag

Filter by tag. Repeated calls accumulate (job matches any listed tag).

JobQuery -> String -> JobQuery

with-limit

Limit the number of rows returned.

JobQuery -> Int -> JobQuery

to-statement

Render the query as a parameterized SELECT statement.

JobQuery -> {text: String, binds: [String]}

Scheduled

Scheduled job state (waiting for its scheduled-at time).

JobState

Available

Available job state (ready to be claimed).

JobState

Executing

Executing job state (claimed by a worker).

JobState

Retryable

Retryable job state (failed with attempts remaining).

JobState

Completed

Completed job state (terminal).

JobState

Discarded

Discarded job state (terminal, attempts exhausted or permanent failure).

JobState

Cancelled

Cancelled job state (terminal).

JobState

state-to-string

Convert a JobState to its database string.

state-from-string

Parse a database string into a JobState.

is-terminal-state?

Check whether a state is terminal.

is-valid-transition?

Check whether a state transition is valid in the job state machine.

has-attempts-remaining?

Check whether a job has attempts remaining.

RunNow

Run immediately.

Schedule

RunIn

Run after a delay in seconds.

Int -> Schedule

RunAt

Run at an absolute ISO-8601 timestamp.

String -> Schedule

RetryIn

Retry after an explicit delay in seconds: Err (Jobs.RetryIn 30 "reason").

Int -> String -> JobFailure

RetryDefault

Retry using the worker's backoff schedule: Err (Jobs.RetryDefault "reason").

String -> JobFailure

DiscardJob

Permanent failure, never retried: Err (Jobs.DiscardJob "reason").

String -> JobFailure

failure-message

Extract the message from a JobFailure.

DbError

Database error constructor.

{message: String} -> JobsError

DecodeError

Row decode error constructor.

{message: String} -> JobsError

ValidationError

Spec validation error constructor.

{message: String} -> JobsError

UnknownWorker

Unknown worker error constructor.

{message: String} -> JobsError

install-schema

Create the job tables and indexes (idempotent).

drop-schema

Drop the job tables (tests and teardown).

new-job

Create an enqueue spec for a worker name with arguments.

in-queue

Route the spec to a named queue.

with-priority

Set the spec priority (0 highest .. 9 lowest).

with-max-attempts

Set the spec retry budget.

with-tags

Attach tags for querying and bulk cancellation.

schedule-in

Delay the first run by a number of seconds.

schedule-at

Schedule the first run at an absolute ISO-8601 timestamp.

validate-spec

Validate a spec without enqueueing it.

enqueue

Insert a validated spec directly against a database connection.

enqueue-job

Insert a spec through a config, emitting a JobEnqueued telemetry event.

define-worker

Define a worker: a name plus a perform function (Map String String -> Result Unit JobFailure).

worker-queue

Set the worker's default queue.

worker-priority

Set the worker's default priority.

worker-max-attempts

Set the worker's default retry budget.

worker-backoff

Set the worker's backoff function (attempt -> delay seconds).

find-worker

Find a worker definition by name.

job-for

Build an enqueue spec from a worker definition, inheriting its defaults.

default-backoff

Default exponential backoff: 2^attempt * 10s, capped at one day.

delay-for

Compute the retry delay for a failure outcome.

stage

Promote due scheduled and retryable jobs to available.

claim

Atomically claim jobs (FOR UPDATE SKIP LOCKED).

complete

Mark an executing job completed.

retry-after

Mark an executing job retryable after a delay.

discard

Mark an executing job discarded.

cancel

Cancel a pending or executing job by id.

cancel-by-tag

Cancel all pending jobs carrying a tag.

release

Release a retryable, discarded, or cancelled job back to available.

rescue-stuck

Rescue jobs stuck in executing past the timeout.

prune

Delete finished jobs older than the retention window.

get-job

Fetch a job by id.

count-by-state

Count jobs grouped by state.

fetch-jobs

Run a query built with the query builder and decode job rows.

decode-job

Decode a kit-postgres row into a Job record.

new-query

An empty job query (no filters, limit 100).

where-state

Filter by job state.

where-queue

Filter by queue name.

where-worker

Filter by worker name.

where-tag

Filter by tag.

with-limit

Limit the number of rows returned.

to-statement

Render the query as a parameterized SELECT statement.

parse-cron

Parse a five-field cron expression.

cron-matches?

Check whether a cron schedule matches a UTC epoch-seconds timestamp.

next-run

Compute the next cron occurrence strictly after a timestamp.

iso-from-epoch

Format an epoch-seconds timestamp as ISO-8601 UTC.

JobEnqueued

Enqueued event constructor.

{id: Int, queue: String, worker: String} -> JobEvent

JobStarted

Started event constructor.

{job: Job} -> JobEvent

JobCompleted

Completed event constructor.

{job: Job, duration-ms: Int} -> JobEvent

JobFailed

Failed (will retry) event constructor.

{job: Job, error: String, retry-in-seconds: Int} -> JobEvent

JobDiscarded

Discarded event constructor.

{job: Job, error: String} -> JobEvent

JobCancelled

Cancelled event constructor.

{id: Int} -> JobEvent

PollFailed

Poll failure event constructor.

{message: String} -> JobEvent

new-telemetry

An empty telemetry configuration.

on-event

Subscribe a handler to all job events.

emit

Emit an event to all handlers (returns handler count).

execute-job

Execute one claimed job and persist its outcome.

failure-transition

Pure failure transition: Retryable or Discarded for a job and failure.

new-config

Create a job system configuration for a database connection.

with-workers

Register worker definitions on a config.

with-queues

Set the queues the pool polls.

with-pool-size

Set the worker actor pool size.

with-poll-interval

Set the poll interval in milliseconds.

with-batch-size

Set the per-queue claim batch size.

with-node-name

Set the node name recorded on claimed jobs.

with-rescue-after

Set the stuck-job rescue timeout in seconds.

with-recurring

Set the recurring (cron) entries.

with-telemetry

Attach telemetry to a config.

new-recurring

Define a recurring (cron) entry for a worker.

run-once

Run one poll cycle (rescue, stage, recurring, claim, execute).

start-pool

Start a supervised pool of polling worker actors (requires ActorAuth).

poll-pool

Trigger one poll on every pool actor (requires ActorAuth).

run-pool

Drive a pool for N cycles with sleeps (requires ActorAuth + ProcessAuth).

stop-pool

Deregister the node and stop the pool (requires ActorAuth).

health-check

Report queue depth by state and active worker count.

register-worker

Register this worker process in the workers table.

heartbeat

Refresh this worker's heartbeat.

count-active-workers

Count workers with a recent heartbeat.

default-queue

Default queue name for new specs.

default-priority

Default priority (0 is highest, larger numbers run later).

default-max-attempts

Default maximum number of attempts.

new-job

Create an enqueue spec for a worker with arguments.

Defaults: queue "default", priority 5, max-attempts 3, no tags, run now.

String -> Map String String -> Spec

in-queue

Route the job to a named queue.

Spec -> String -> Spec

with-priority

Set the job priority (0 is highest; jobs order by priority then id).

Spec -> Int -> Spec

with-max-attempts

Set the maximum number of attempts before the job is discarded.

Spec -> Int -> Spec

with-tags

Attach tags for querying and bulk cancellation.

Spec -> [String] -> Spec

schedule-in

Schedule the job to run after a delay in seconds.

Spec -> Int -> Spec

schedule-at

Schedule the job to run at an absolute ISO-8601 timestamp.

Spec -> String -> Spec

validate-spec

Validate an enqueue spec.

Checks: non-empty worker and queue names, priority in 0..9, positive max-attempts, non-negative schedule delay, and non-empty tags.

Spec -> Result Spec String

parse

Parse a five-field cron expression.

Returns a record with the expanded value sets for each field plus restriction flags used for the day-of-month/day-of-week rule.

String -> Result Cron String

parse-field

Parse one cron field into a sorted list of allowed values.

String -> Int -> Int -> Result [Int] String

days-from-civil

Convert a civil date to days since the Unix epoch. Uses Howard Hinnant's days-from-civil algorithm.

Int -> Int -> Int -> Int

civil-from-days

Convert days since the Unix epoch to a civil date {year, month, day}. Uses Howard Hinnant's civil-from-days algorithm.

Int -> {year: Int, month: Int, day: Int}

weekday-of-days

Day of week for days since the Unix epoch (0 = Sunday).

Int -> Int

iso-from-epoch

Format an epoch-seconds timestamp as an ISO-8601 UTC string.

Int -> String

matches?

Check whether a cron schedule matches an epoch-seconds timestamp (UTC).

Seconds within the minute are ignored, matching cron's minute resolution.

Cron -> Int -> Bool

next-run

Compute the next occurrence strictly after an epoch-seconds timestamp.

Searches forward by skipping whole months, days, and hours that cannot match, so it stays fast even for sparse schedules. Returns an error for unreachable schedules (for example, February 30th).

Cron -> Int -> Result Int String

execute-job

Execute one claimed (executing) job and persist its outcome.

Returns the job's final state, or a JobsError when the database rejects the state transition. Jobs whose worker name has no registered handler are discarded with an explanatory error.

Db -> [WorkerDef] -> Telemetry -> Job -> Result JobState JobsError

failure-transition

Decide the failure transition without touching the database.

Pure helper used by execute-job and unit tests: returns Discarded for permanent failures or exhausted attempts, otherwise Retryable.

Job -> JobFailure -> JobState

fail-with

Convenience: wrap a plain error message as a default-backoff retry failure.

String -> JobFailure

install-schema

Create the job system tables and indexes (idempotent).

Db -> Result Unit JobsError

drop-schema

Drop the job system tables (for tests and teardown).

Db -> Result Unit JobsError

enqueue

Validate a spec and insert it as a new job. Returns the new job id.

Db -> Spec -> Result Int JobsError

enqueue-recurring

Insert the next occurrence of a recurring entry unless one is pending. Returns Some id when a new occurrence was inserted.

Db -> RecurringEntry -> String -> Result (Option Int) JobsError

stage

Promote due scheduled and retryable jobs to 'available'.

Db -> Result Unit JobsError

claim

Atomically claim up to limit jobs from a queue using FOR UPDATE SKIP LOCKED. Returns the claimed jobs, now 'executing'.

Db -> String -> Int -> String -> Result [Job] JobsError

complete

Mark an executing job completed.

Db -> Int -> Result Unit JobsError

retry-after

Mark an executing job retryable after a delay, recording the error.

Db -> Int -> Int -> String -> Result Unit JobsError

discard

Mark an executing job discarded, recording the error.

Db -> Int -> String -> Result Unit JobsError

cancel

Cancel a pending or executing job.

Db -> Int -> Result Unit JobsError

cancel-by-tag

Cancel all pending jobs carrying a tag.

Db -> String -> Result Unit JobsError

release

Manually release a retryable, discarded, or cancelled job back to 'available' for another run.

Db -> Int -> Result Unit JobsError

rescue-stuck

Move jobs stuck in 'executing' past the timeout back to retryable (or discarded when out of attempts).

Db -> Int -> Result Unit JobsError

prune

Delete completed, discarded, and cancelled jobs older than the window.

Db -> Int -> Result Unit JobsError

get-job

Fetch a job by id.

Db -> Int -> Result (Option Job) JobsError

count-by-state

Count jobs grouped by state string.

Db -> Result (Map String Int) JobsError

fetch-jobs

Run a select built by the query module and decode the job rows.

Db -> {text: String, binds: [String]} -> Result [Job] JobsError

register-worker

Register (or refresh) this worker process for health monitoring.

Db -> {node-name: String, queues: [String], pool-size: Int, ...} -> Result Unit JobsError

count-active-workers

Count worker processes whose heartbeat is fresher than the window.

Db -> Int -> Result Int JobsError

heartbeat

Refresh this worker's heartbeat.

Db -> String -> Result Unit JobsError

deregister-worker

Remove this worker's registration on shutdown.

Db -> String -> Result Unit JobsError

decode-job

Decode a kit-postgres row into a Job record.

Map String (Option DbValue) -> Result Job JobsError

new-config

Create a job system configuration for a database connection.

Defaults: queue "default", pool of 2 workers, 1s poll interval, batches of 5 jobs, node name "worker-1", 1h stuck-job rescue, no recurring entries.

Db -> Config

with-workers

Register the worker definitions the pool can execute.

Config -> [WorkerDef] -> Config

with-queues

Set the queues polled by the pool.

Config -> [String] -> Config

with-pool-size

Set the number of worker actors in the pool.

Config -> Int -> Config

with-poll-interval

Set the poll interval in milliseconds.

Config -> Int -> Config

with-batch-size

Set how many jobs one poll claims per queue.

Config -> Int -> Config

with-node-name

Set the node name recorded on claimed jobs and worker registrations.

Config -> String -> Config

with-rescue-after

Set how long a job may stay 'executing' before the rescuer intervenes.

Config -> Int -> Config

with-recurring

Set the recurring (cron) entries scheduled by every poll cycle.

Config -> [RecurringEntry] -> Config

with-telemetry

Attach a telemetry configuration.

Config -> Telemetry -> Config

new-recurring

Define a recurring (cron) entry for a worker.

The name uniquely identifies the entry across nodes; only one pending occurrence per name exists at a time. Fails when the cron expression is invalid.

String -> WorkerDef -> String -> Map String String -> Result RecurringEntry String

enqueue-job

Enqueue a spec and emit a JobEnqueued telemetry event.

Config -> Spec -> Result Int JobsError

run-once

Run one poll cycle: rescue stuck jobs, stage due jobs, schedule recurring occurrences, then claim and execute up to batch-size jobs per queue.

Returns the number of jobs executed.

Config -> Result Int JobsError

start-pool

Start a supervised pool of polling worker actors.

Each actor runs a full poll cycle when it receives a :poll message. Poll errors are reported through telemetry as PollFailed events and the actor keeps running. The pool is registered in the workers table for health monitoring. Requires ActorAuth (derive with actor-auth (concurrency-auth env.root)).

Config -> ActorAuth -> Pool

poll-pool

Trigger one poll on every pool actor and refresh the heartbeat.

Ticks each actor so the interpreter processes mailboxes; compiled worker threads treat the tick as a no-op.

Pool -> ActorAuth -> Unit

run-pool

Drive a pool for a number of poll cycles, sleeping the configured interval between cycles. Use a large count for long-running worker processes.

Requires ProcessAuth (derive with process-auth env.root at program entry) because it sleeps between cycles.

Pool -> Int -> ActorAuth -> ProcessAuth -> Unit

stop-pool

Deregister the node and stop the supervisor and its actors.

Pool -> ActorAuth -> Unit

health-check

Report queue depth by state plus recently active worker count.

Config -> Result {counts: Map String Int, active-workers: Int} JobsError

max-backoff-seconds

Maximum backoff delay in seconds (one day).

base-backoff-seconds

Base multiplier for the default exponential backoff, in seconds.

default-backoff

Default exponential backoff: 2^attempt * 10 seconds, capped at one day.

Non-positive attempts are treated as attempt 1.

Int -> Int

delay-for

Compute the retry delay for a failure outcome.

- RetryIn uses the explicit delay from the handler - RetryDefault and DiscardJob fall back to the supplied backoff function

(Int -> Int) -> Int -> JobFailure -> Int

JobState

The lifecycle state of a job.

Jobs transition through these states:

scheduled -> available -> executing -> completed | | cancelled retryable -> discarded

- Scheduled: waiting for its scheduled-at time - Available: ready to be claimed and executed - Executing: currently being processed by a worker - Retryable: failed but has attempts remaining - Completed: successfully finished - Discarded: failed after max attempts exhausted - Cancelled: manually cancelled before completion

Scheduled

Scheduled state constructor.

JobState

Available

Available state constructor.

JobState

Executing

Executing state constructor.

JobState

Retryable

Retryable state constructor.

JobState

Completed

Completed state constructor.

JobState

Discarded

Discarded state constructor.

JobState

Cancelled

Cancelled state constructor.

JobState

state-to-string

Convert a JobState to its database string representation.

JobState -> String

state-from-string

Parse a database string into a JobState.

String -> Option JobState

is-terminal-state?

Check whether a state is terminal (no further transitions allowed).

JobState -> Bool

is-valid-transition?

Check whether a state transition is valid in the job state machine.

Valid transitions: - Scheduled -> Available | Cancelled - Available -> Executing | Cancelled - Executing -> Completed | Retryable | Discarded | Cancelled - Retryable -> Available | Discarded | Cancelled - Completed, Discarded, Cancelled are terminal

JobState -> JobState -> Bool

Schedule

When an enqueued job should first become runnable.

- RunNow: immediately available - RunIn seconds: available after a relative delay - RunAt iso-timestamp: available at an absolute ISO-8601 time

RunNow

Run immediately.

Schedule

RunIn

Run after a delay in seconds.

Int -> Schedule

RunAt

Run at an absolute ISO-8601 timestamp.

String -> Schedule

JobFailure

The failure outcome a job handler returns via Err to control retry behavior.

- RetryIn seconds message: retry after an explicit delay - RetryDefault message: retry using the worker's backoff function - DiscardJob message: permanent failure, do not retry

RetryIn

Retry after an explicit delay in seconds.

Int -> String -> JobFailure

RetryDefault

Retry using the worker's backoff schedule.

String -> JobFailure

DiscardJob

Permanent failure — mark the job discarded without further attempts.

String -> JobFailure

failure-message

Extract the human-readable message from a JobFailure.

JobFailure -> String

JobsError

Error type for queue and worker operations.

- DbError: the database rejected a statement - DecodeError: a database row could not be decoded into a Job - ValidationError: an enqueue spec failed validation - UnknownWorker: a claimed job references a worker name with no registered handler

DbError

Database error constructor.

{message: String} -> JobsError

DecodeError

Row decode error constructor.

{message: String} -> JobsError

ValidationError

Spec validation error constructor.

{message: String} -> JobsError

UnknownWorker

Unknown worker error constructor.

{message: String} -> JobsError

make-job

Construct a Job record from decoded database fields.

Timestamps are ISO-8601 strings as returned by PostgreSQL.

Int -> JobState -> String -> String -> Map String String -> [String] -> Int -> Int -> Int -> String -> String -> Job

has-attempts-remaining?

Check whether a job has attempts remaining after the current one.

{attempt: Int, max-attempts: Int, ...} -> Bool

JobEvent

Lifecycle events emitted by the job system.

- JobEnqueued: a job was inserted (id + spec queue/worker) - JobStarted: a claimed job began executing - JobCompleted: a job finished successfully, with its duration - JobFailed: an attempt failed and the job will be retried - JobDiscarded: a job failed permanently (no attempts left or DiscardJob) - JobCancelled: a job was cancelled - PollFailed: a poll cycle hit an error (for example, a lost connection)

JobEnqueued

Enqueued event constructor.

{id: Int, queue: String, worker: String} -> JobEvent

JobStarted

Started event constructor.

{job: Job} -> JobEvent

JobCompleted

Completed event constructor.

{job: Job, duration-ms: Int} -> JobEvent

JobFailed

Failed (will retry) event constructor.

{job: Job, error: String, retry-in-seconds: Int} -> JobEvent

JobDiscarded

Discarded event constructor.

{job: Job, error: String} -> JobEvent

JobCancelled

Cancelled event constructor.

{id: Int} -> JobEvent

PollFailed

Poll failure event constructor.

{message: String} -> JobEvent

new-telemetry

Create an empty telemetry configuration with no handlers.

Telemetry

on-event

Add an event handler. Handlers run in subscription order on every event.

Telemetry -> (JobEvent -> Unit) -> Telemetry

emit

Emit an event to every subscribed handler. Returns the number of handlers invoked.

Telemetry -> JobEvent -> Int

args-to-json

Encode job arguments as a JSON object string.

Keys are sorted by Map ordering; values are encoded as JSON strings.

Map String String -> String

args-from-json

Decode a JSON object string into job arguments.

Non-string JSON values are re-encoded to their JSON text so nothing is lost.

String -> Result (Map String String) String

json-value-to-text

Render a JSONValue as the string form used for job args.

Strings are unwrapped; every other value keeps its JSON text form.

JSONValue -> String