A standalone, agent-native durable job engine on Postgres (or in-memory for tests). No Oban, no Ecto — Postgrex, Jason, and telemetry only.
Start an instance
Configure in application env (keyed by the instance name, like Ecto.Repo
/ Phoenix.Endpoint) and pull it in with otp_app:
# config/config.exs
config :my_app, MyBelay,
queues: [default: 10, ai: [limit: 5, global_limit: 2, rate: [allowed: 60, period: 60]]],
crons: [[name: "digest", expr: "0 8 * * *", worker: MyApp.Digest]]
# config/runtime.exs
config :my_app, MyBelay,
storage: [adapter: :postgres, url: System.fetch_env!("DATABASE_URL")]
# application.ex — inline opts override the app-env base
children = [{Belay, otp_app: :my_app, name: MyBelay}]Or pass every option inline on the child spec and skip otp_app — both
forms take the same keys.
Define work
defmodule MyApp.Agent do
use Belay.Worker, queue: :ai, max_attempts: 10
@impl Belay.Worker
def run(ctx) do
text = Belay.step(ctx, :fetch, fn -> fetch!(ctx.job.input["url"]) end)
summary =
Belay.step(ctx, :summarize, fn -> llm!(text) end,
cost: [usd: 0.02, tokens: 1200]
)
%{"approved" => true} = Belay.await(ctx, :approval)
{:ok, summary}
end
end
Belay.insert(MyBelay, MyApp.Agent.new(%{"url" => url}, budget: [usd: 1.0]))Committed step results are memoized per job — retries replay past journaled
work. Step bodies remain at-least-once until the journal write commits.
Budgets fail the job with :budget_exceeded after a step's declared cost
crosses the limit. Signals
(Belay.signal_job/4) wake awaiting jobs instantly; steer_job/3 injects
guidance readable via steering/1 at step boundaries.
Instance options
| Option | Default | Purpose |
|---|---|---|
:name | Belay | instance name (atom); first argument to every API call |
:otp_app | nil | read the options below from config :otp_app, name, ...; inline opts override |
:storage | required | [adapter: :postgres, url: ...] or [adapter: :memory] |
:queues | [] | queue: limit or queue: [limit:, global_limit:, rate:, partition:, manual:] |
:crons | [] | [[name:, expr:, worker:, input:, opts:]]; merged with Belay.Crons rows |
:notifiers | [:local] | add :postgres for cross-fleet pg_notify wake-ups |
:poll_interval | 500 | idle polling ceiling (ms) |
:busy_poll | 25 | hot polling cadence (ms) |
:lease_ttl | 30_000 | running-job lease (ms); crash-orphan recovery window |
:sweep_interval | 5_000 | reclaim/retention cadence (ms) |
:cron_interval | 20_000 | cron tick (ms; slots dedup regardless) |
:dynamic_sync | 5_000 | runtime-queue reconciliation cadence (ms) |
:shutdown_grace | 15_000 | time running jobs get on shutdown (ms) |
:retention | 1d/7d/7d | per-terminal-state pruning: [succeeded:, failed:, cancelled:] seconds or :infinity |
:signal_ttl | 604_800 | seconds before undelivered signals are pruned |
:max_result_bytes | nil | reject job results larger than this (unlimited by default; oversized results are warned about either way) |
:encryption | nil | [key: {mod, fun, args}] returning a 32-byte key |
:clock | system | Belay.Clock implementation (tests use Belay.Clock.Sim) |
:node_id | derived | stable identity for leases |
Summary
Functions
Wait for a signal. Returns the payload, or {:error, :timeout} after
:timeout seconds. Parks the job (no process held) until signalled.
Park (at zero cost) until every spawned child reaches a terminal state. Returns the children ordered by id.
Await a job's terminal result. Returns {:ok, value} for success,
{:error, {:job, state}} for failure/cancellation, {:error, :timeout}.
Cancel a job: immediate for parked states, cooperative for running.
Delete a previously delivered signal from a scope.
Record actual resource usage against the queue's rate resource bucket.
Append to the job's durable event stream (progress, tokens, partial output).
Subscribers receive {:belay_event, job_id, seq, payload} live;
events/3 replays from any offset.
Replay a job's event stream from an offset (0 for everything).
Fetch a job by id.
Insert a job built with WorkerModule.new(input, opts).
Insert many jobs at once. Rows deduped by uniqueness or cron slots are silently skipped; only inserted jobs are returned.
List jobs newest-first. Filters: :queue, :state, :worker,
:workflow_id, :parent_id, :before_id (pagination cursor), :limit
(default 50).
Fan out one child per input and wait for all of them — the map phase of a map-reduce inside a single durable job. Returns child jobs in input order.
Stop a queue's local producer from claiming (running jobs finish).
Resume a paused queue.
Resurrect a failed or cancelled job (grants one more attempt if exhausted).
Deliver a durable signal to a scope, waking any awaiting jobs.
Deliver a durable signal scoped to one job.
Durably sleep: parks the job (freeing its slot) and resumes after the
target. The wake time is memoized under name, so replays skip past it.
Spawn a child job from inside a running job. Memoized under name, so a
crash after spawning cannot duplicate the child. Returns the child job id.
Spawn many children as one memoized step. Returns their ids in order.
Per-queue, per-state job counts.
Inject steering guidance readable by the running job via steering/1.
Read the latest steering payload, or nil.
Run and memoize fun under a per-job step name, with optional
cost: [usd:, tokens:].
List a job's recorded steps with costs.
Subscribe the calling process to a job's event stream.
Unsubscribe the calling process from a job's live event stream.
Types
Functions
@spec await(Belay.Ctx.t(), String.t() | atom(), keyword()) :: map() | {:error, :timeout}
Wait for a signal. Returns the payload, or {:error, :timeout} after
:timeout seconds. Parks the job (no process held) until signalled.
@spec await_children(Belay.Ctx.t()) :: [Belay.Job.t()]
Park (at zero cost) until every spawned child reaches a terminal state. Returns the children ordered by id.
@spec await_result(instance(), integer(), timeout()) :: {:ok, term()} | {:error, {:job, :failed | :cancelled} | :not_found | :timeout}
Await a job's terminal result. Returns {:ok, value} for success,
{:error, {:job, state}} for failure/cancellation, {:error, :timeout}.
Cancel a job: immediate for parked states, cooperative for running.
Delete a previously delivered signal from a scope.
@spec debit(Belay.Ctx.t(), String.t(), integer()) :: :ok
Record actual resource usage against the queue's rate resource bucket.
@spec emit(Belay.Ctx.t(), map()) :: {:ok, integer() | :replayed | :no_registry}
Append to the job's durable event stream (progress, tokens, partial output).
Subscribers receive {:belay_event, job_id, seq, payload} live;
events/3 replays from any offset.
@spec events(instance(), integer(), non_neg_integer()) :: [map()]
Replay a job's event stream from an offset (0 for everything).
@spec get_job(instance(), integer()) :: {:ok, Belay.Job.t()} | {:error, :not_found}
Fetch a job by id.
@spec insert(instance(), buildable()) :: {:ok, Belay.Job.t()}
Insert a job built with WorkerModule.new(input, opts).
Options on new/2: :queue, :priority, :max_attempts, :schedule_in,
:partition_key, :meta, :budget ([usd: 5.0, tokens: 100_000]), and
:unique — either "key" (dedupe while incomplete) or
[key: k, within: seconds] (dedupe per time window). A deduped insert
returns the existing job flagged duplicate?: true.
@spec insert_all(instance(), [buildable()]) :: [Belay.Job.t()]
Insert many jobs at once. Rows deduped by uniqueness or cron slots are silently skipped; only inserted jobs are returned.
@spec list_jobs(instance(), Enumerable.t()) :: [Belay.Job.t()]
List jobs newest-first. Filters: :queue, :state, :worker,
:workflow_id, :parent_id, :before_id (pagination cursor), :limit
(default 50).
@spec map_children(Belay.Ctx.t(), String.t() | atom(), module(), [map()], keyword()) :: [Belay.Job.t()]
Fan out one child per input and wait for all of them — the map phase of a map-reduce inside a single durable job. Returns child jobs in input order.
results =
Belay.map_children(ctx, :chunks, MyApp.Summarize, chunk_inputs)
|> Enum.map(&Belay.Job.result/1)Stop a queue's local producer from claiming (running jobs finish).
Resume a paused queue.
@spec retry_job(instance(), integer()) :: {:ok, Belay.Job.t()} | {:error, :not_retryable | :not_found}
Resurrect a failed or cancelled job (grants one more attempt if exhausted).
Deliver a durable signal to a scope, waking any awaiting jobs.
Deliver a durable signal scoped to one job.
@spec sleep(Belay.Ctx.t(), String.t() | atom(), non_neg_integer()) :: :ok
Durably sleep: parks the job (freeing its slot) and resumes after the
target. The wake time is memoized under name, so replays skip past it.
@spec spawn(Belay.Ctx.t(), String.t() | atom(), buildable()) :: integer()
Spawn a child job from inside a running job. Memoized under name, so a
crash after spawning cannot duplicate the child. Returns the child job id.
@spec spawn_many(Belay.Ctx.t(), String.t() | atom(), [buildable()]) :: [integer()]
Spawn many children as one memoized step. Returns their ids in order.
@spec stats(instance()) :: %{ optional(String.t()) => %{optional(String.t()) => non_neg_integer()} }
Per-queue, per-state job counts.
Inject steering guidance readable by the running job via steering/1.
@spec steering(Belay.Ctx.t()) :: map() | nil
Read the latest steering payload, or nil.
Run and memoize fun under a per-job step name, with optional
cost: [usd:, tokens:].
A committed result is replayed without re-running fun. The body is
at-least-once until that journal write commits, so external effects should
be idempotent across a crash-before-journal window.
List a job's recorded steps with costs.
Subscribe the calling process to a job's event stream.
Unsubscribe the calling process from a job's live event stream.