Belay.Worker behaviour (Belay v2.0.0)

Define a durable worker.

defmodule MyApp.Summarize do
  use Belay.Worker, queue: :ai, max_attempts: 5

  @impl Belay.Worker
  def run(ctx) do
    text = Belay.step(ctx, :fetch, fn -> fetch!(ctx.job.input["url"]) end)
    {:ok, summarize(text)}
  end
end

Return values: `:ok`	`{:ok, result}`	`{:error, reason}` (retry with
backoff)	`{:cancel, reason}`	`{:snooze, seconds}`. Raised exceptions

retry. Belay.step/4, Belay.await/3, and Belay.sleep/3 manage control flow internally.

Chunk workers

Declare chunk: [size: n, gather_ms: t] and implement run_chunk/1 instead of run/1 to process many jobs in one execution — one bulk INSERT instead of hundreds, one batch-priced embeddings call instead of a hundred singles:

defmodule MyApp.EmbedChunk do
  use Belay.Worker, queue: :embeddings, chunk: [size: 100, gather_ms: 500]

  @impl Belay.Worker
  def run_chunk(ctxs) do
    texts = Enum.map(ctxs, & &1.job.input["text"])

    for {ctx, emb} <- Enum.zip(ctxs, MyApp.OpenAI.embed!(texts)),
        into: %{},
        do: {ctx.job.id, {:ok, emb}}
  end
end

The producer gathers claimed jobs of the same worker up to size, waiting at most gather_ms for stragglers (gather_ms: 0 dispatches each claim round as-is). Gathered jobs are already claimed and leased, so a crash mid-gather is reclaimed like any other crash. Give chunk workers their own queue — mixed queues chunk per worker, which fragments batches.

Return values: `:ok` (all succeed)	`{:ok, %{id => result}}` (per-job
results; missing ids succeed with `nil`)	`{:error, reason}` (every job
retries on its own backoff/max_attempts)	`%{id => outcome}` where each

outcome is any single-job return value — partial failure retries only the failed jobs. Raises retry the whole chunk. Prefer per-job outcomes over budgets/steps inside chunks: a control throw (budget kill, cancel honor) cannot be attributed to one job and retries the entire chunk.

Summary

Types

result()

Callbacks

backoff(attempt)

Per-attempt retry backoff in seconds. Overridable.

run(t)

run_chunk(list)

Process a gathered chunk of jobs in one execution (chunk workers).

Types

result()

@type result() ::
  :ok
  | {:ok, term()}
  | {:error, term()}
  | {:cancel, term()}
  | {:snooze, non_neg_integer()}

Callbacks

backoff(attempt)

(optional)

@callback backoff(attempt :: pos_integer()) :: non_neg_integer()

Per-attempt retry backoff in seconds. Overridable.

run(t)

(optional)

@callback run(Belay.Ctx.t()) :: result()

run_chunk(list)

(optional)

@callback run_chunk([Belay.Ctx.t()]) ::
  :ok
  | {:ok, %{required(integer()) => term()}}
  | {:error, term()}
  | %{required(integer()) => result()}

Process a gathered chunk of jobs in one execution (chunk workers).