Jobs
A job is a named handler on a queue. Define it once with its name, queue, retries and sealed fields; enqueueing it sends all of them. A returned value completes the job; an exception fails it.
Define a job
The handler receives the payload and a context: ctx.app is your application's container, ctx.context the request context the producer attached (see Middleware and app context).
struct ConfJob
include JoobQ::Job
@queue = QUEUE
@retries = 3
@max_retries = 3
@expires = 1.day.from_now.to_unix_ms@job("conf.context", queue=QUEUE)
def context_job(payload, ctx):
report(ctx, "context", payload, app_marker=ctx.app.marker, tenant=ctx.context.get("tenant_id"), x=payload.get("x"))defineJob<App>({ name: 'conf.context', queue: QUEUE, perform: async (p, ctx) =>
report(ctx, 'context', p, { app_marker: ctx.app.marker, tenant: ctx.context.tenant_id, x: p.x }) }),class ContextJob
include JoobQ::Job
job_name "conf.context"
queue QUEUE
def self.perform(p, ctx) = report(ctx, "context", p, app_marker: ctx.app.marker, tenant: ctx.context["tenant_id"], x: p["x"])
endjoobq.DefineJob[*App]("conf.context", opts, func(p map[string]any, ec *EC) (any, error) {
return report(ec, "context", p, map[string]any{"app_marker": ec.App.Marker, "tenant": ec.Meta["tenant_id"], "x": p["x"]})
}),JobDef::new("conf.context", queue.clone(), |p: Value, ctx: &Ctx| {
report(ctx, "context", &p, json!({"app_marker": ctx.app.marker, "tenant": ctx.context.get("tenant_id"), "x": p["x"]}))
}),JobDefinition.<App>of("conf.context", QUEUE, 3, (p, ctx) ->
report(ctx, "context", p, map("app_marker", ctx.app().marker, "tenant", ctx.context().get("tenant_id"), "x", m(p).get("x")))),final class ContextJob extends ConfJob
{
public static function jobName(): string { return 'conf.context'; }
public static function perform(array $p, ExecutionContext $ctx): mixed
{
return report($ctx, 'context', $p, ['app_marker' => $ctx->app->marker, 'tenant' => $ctx->context['tenant_id'] ?? null, 'x' => $p['x'] ?? null]);
}
}JobDefinition<App>.Define("conf.context", Queue, 3, (p, ctx) =>
Report(ctx, "context", p, new() { ["app_marker"] = ctx.App.Marker, ["tenant"] = ctx.Context.GetValueOrDefault("tenant_id")?.GetValue<string>(), ["x"] = Num(p?["x"]) })),Tested against a real server in every language shown.
Retries and the dead letter queue
A failed job is retried while it has retries left (max_retries, 3 by default, 0 to 100), with JoobQ's exponential backoff (2^attempt seconds, capped at an hour); then it goes to the dead letter queue. A non-retryable error goes to the dead letter queue at once: NonRetryableError (Python, TypeScript), JoobQ::NonRetryableError (Ruby, Crystal), joobq.NewNonRetryableError (Go), Error::non_retryable (Rust), NonRetryableException (Java, PHP, .NET).
when "conf.retry"
raise "transient failure on attempt #{attempt}" if attempt < 3
Conf.report("retry", probe, {"attempt" => Conf.any(attempt)})
when "conf.dead"
Conf.report("dead_attempt", probe)
raise JoobQ::NonRetryableError.new("permanent failure #{probe}")@job("conf.retry", queue=QUEUE, retries=3)
def retry_job(payload, ctx):
if ctx.attempt < 3:
raise RuntimeError(f"transient failure on attempt {ctx.attempt}")
report(ctx, "retry", payload, attempt=ctx.attempt)
@job("conf.dead", queue=QUEUE)
def dead_job(payload, ctx):
report(ctx, "dead_attempt", payload)
raise NonRetryableError(f"permanent failure {payload['probe']}")defineJob<App>({ name: 'conf.retry', queue: QUEUE, retries: 3, perform: async (p, ctx) => {
if (ctx.attempt < 3) throw new Error(`transient failure on attempt ${ctx.attempt}`)
return report(ctx, 'retry', p, { attempt: ctx.attempt })
} }),
defineJob<App>({ name: 'conf.dead', queue: QUEUE, perform: async (p, ctx) => {
await report(ctx, 'dead_attempt', p)
throw new NonRetryableError(`permanent failure ${p.probe}`)
} }),class RetryJob
include JoobQ::Job
job_name "conf.retry"
queue QUEUE
retries 3
def self.perform(p, ctx)
raise "transient failure on attempt #{ctx.attempt}" if ctx.attempt < 3
report(ctx, "retry", p, attempt: ctx.attempt)
end
end
class DeadJob
include JoobQ::Job
job_name "conf.dead"
queue QUEUE
def self.perform(p, ctx)
report(ctx, "dead_attempt", p)
raise JoobQ::NonRetryableError, "permanent failure #{p['probe']}"
end
endjoobq.DefineJob[*App]("conf.retry", joobq.JobOptions{Queue: queue, Retries: 3}, func(p map[string]any, ec *EC) (any, error) {
if ec.Task.Attempt < 3 {
return nil, joobq.Errorf("transient failure on attempt %d", ec.Task.Attempt)
}
return report(ec, "retry", p, map[string]any{"attempt": ec.Task.Attempt})
}),
joobq.DefineJob[*App]("conf.dead", opts, func(p map[string]any, ec *EC) (any, error) {
_, _ = report(ec, "dead_attempt", p, nil)
return nil, joobq.NewNonRetryableError(fmt.Sprintf("permanent failure %v", p["probe"]))
}),JobDef::new("conf.retry", queue.clone(), |p: Value, ctx: &Ctx| {
if ctx.attempt() < 3 {
return Err(Error::new(format!("transient failure on attempt {}", ctx.attempt())));
}
report(ctx, "retry", &p, json!({"attempt": ctx.attempt()}))
}).retries(3),
JobDef::new("conf.dead", queue.clone(), |p: Value, ctx: &Ctx| report(ctx, "dead_attempt", &p, json!({})).and(Err(Error::non_retryable(format!("permanent failure {}", p["probe"].as_str().unwrap_or("")))))),JobDefinition.<App>of("conf.retry", QUEUE, 3, (p, ctx) -> {
if (ctx.attempt() < 3) throw new IllegalStateException("transient failure on attempt " + ctx.attempt());
return report(ctx, "retry", p, map("attempt", ctx.attempt()));
}),
JobDefinition.<App>of("conf.dead", QUEUE, 3, (p, ctx) -> {
report(ctx, "dead_attempt", p, Map.of());
throw new NonRetryableException("permanent failure " + m(p).get("probe"));
}),final class RetryJob extends ConfJob
{
public static int $retries = 3;
public static function jobName(): string { return 'conf.retry'; }
public static function perform(array $p, ExecutionContext $ctx): mixed
{
$attempt = (int)$ctx->task['attempt'];
if ($attempt < 3) {
throw new RuntimeException("transient failure on attempt {$attempt}");
}
return report($ctx, 'retry', $p, ['attempt' => $attempt]);
}
}
final class DeadJob extends ConfJob
{
public static function jobName(): string { return 'conf.dead'; }
public static function perform(array $p, ExecutionContext $ctx): mixed
{
report($ctx, 'dead_attempt', $p);
throw new NonRetryableException("permanent failure {$p['probe']}");
}
}JobDefinition<App>.Define("conf.retry", Queue, 3, (p, ctx) =>
{
if (ctx.Attempt < 3) throw new InvalidOperationException($"transient failure on attempt {ctx.Attempt}");
return Report(ctx, "retry", p, new() { ["attempt"] = ctx.Attempt });
}),
JobDefinition<App>.Define("conf.dead", Queue, 3, async (p, ctx) =>
{
await Report(ctx, "dead_attempt", p);
throw new NonRetryableException($"permanent failure {p?["probe"]}");
}),Tested against a real server in every language shown.
The life of an SDK job on the gateway:
Queued is the queue's Redis list, Delayed JoobQ's delayed set, Dead the dead letter queue. A lease is renewed by the worker's heartbeats; delivery is at least once.
Enqueue later
delay_ms (0 to 86,400,000, a day) parks the job in JoobQ's delayed set. The gateway promotes due jobs itself, so a deployment without a Crystal --worker process still runs them.
when "conf.produce"
before = Conf.now_ms # before the call: the delay counts from when the job is stored
jid = SealedJob.delay(for: 1.5.seconds, probe: probe, card_number: card_number, enqueued_before_ms: before)
Conf.report("produce", probe, {"jid" => Conf.any(jid.to_s)})@job("conf.produce", queue=QUEUE)
def produce_job(payload, ctx):
before = now_ms() # before the call: the gateway counts the delay from when it receives it
res = ctx.enqueue(sealed_job, {"probe": payload["probe"], "card_number": payload["card_number"], "enqueued_before_ms": before},
delay_ms=1500)
report(ctx, "produce", payload, jid=res["jid"])defineJob<App>({ name: 'conf.produce', queue: QUEUE, perform: async (p, ctx) => {
const before = Date.now() // before the call: the gateway counts the delay from when it receives it
const res = await ctx.enqueue(sealed, { probe: p.probe, card_number: p.card_number, enqueued_before_ms: before }, { delayMs: 1500 })
return report(ctx, 'produce', p, { jid: res.jid })
} }),class ProduceJob
include JoobQ::Job
job_name "conf.produce"
queue QUEUE
def self.perform(p, ctx)
before = now_ms # before the call: the gateway counts the delay from when it receives it
res = ctx.enqueue(SealedJob, { probe: p["probe"], card_number: p["card_number"], enqueued_before_ms: before }, delay_ms: 1500)
report(ctx, "produce", p, jid: res["jid"])
end
endjoobq.DefineJob[*App]("conf.produce", opts, func(p map[string]any, ec *EC) (any, error) {
before := nowMs() // before the call: the gateway counts the delay from when it receives it
jid, err := sealed.EnqueueFrom(ec, map[string]any{"probe": p["probe"], "card_number": p["card_number"], "enqueued_before_ms": before}, joobq.EnqueueOptions{DelayMs: 1500})
if err != nil {
return nil, err
}
return report(ec, "produce", p, map[string]any{"jid": jid})
}),JobDef::new("conf.produce", queue.clone(), move |p: Value, ctx: &Ctx| {
let before = now_ms(); // before the call: the gateway counts the delay from when it receives it
let jid = ctx.enqueue(&producer_sealed, json!({"probe": p["probe"], "card_number": p["card_number"], "enqueued_before_ms": before}), EnqueueOptions { delay_ms: 1500, ..Default::default() })?;
report(ctx, "produce", &p, json!({"jid": jid}))
}),JobDefinition.<App>of("conf.produce", QUEUE, 3, (p, ctx) -> {
long before = System.currentTimeMillis(); // before the call: the gateway counts the delay from when it receives it
String jid = ctx.enqueue(sealed, map("probe", m(p).get("probe"), "card_number", m(p).get("card_number"), "enqueued_before_ms", before), EnqueueOptions.delay(1500));
return report(ctx, "produce", p, map("jid", jid));
}),final class ProduceJob extends ConfJob
{
public static function jobName(): string { return 'conf.produce'; }
public static function perform(array $p, ExecutionContext $ctx): mixed
{
$before = now_ms(); // before the call: the gateway counts the delay from when it receives it
$res = $ctx->enqueue(SealedJob::class, ['probe' => $p['probe'], 'card_number' => $p['card_number'], 'enqueued_before_ms' => $before], ['delay_ms' => 1500]);
return report($ctx, 'produce', $p, ['jid' => $res['jid']]);
}
}JobDefinition<App>.Define("conf.produce", Queue, 3, async (p, ctx) =>
{
var before = NowMs(); // before the call: the gateway counts the delay from when it receives it
var jid = await ctx.EnqueueAsync(sealedJob, new { probe = p?["probe"]?.GetValue<string>(), card_number = p?["card_number"]?.GetValue<string>(), enqueued_before_ms = before }, new EnqueueOptions(DelayMs: 1500));
return await Report(ctx, "produce", p, new() { ["jid"] = jid });
}),Tested against a real server in every language shown.
run_at_ms schedules the job for an absolute time (Unix epoch milliseconds). It wins over delay_ms; a time at or before now runs at once, and it is clamped to at most a year out.
@job("conf.produce_runat", queue=QUEUE)
def produce_runat_job(payload, ctx):
before = now_ms() # before the call: the gateway schedules at the absolute run_at_ms
res = ctx.enqueue(runat_job, {"probe": payload["probe"], "enqueued_before_ms": before}, run_at_ms=before + 1500)
report(ctx, "produce_runat", payload, jid=res["jid"])defineJob<App>({ name: 'conf.produce_runat', queue: QUEUE, perform: async (p, ctx) => {
const before = Date.now() // before the call: the gateway computes run_at relative to when it receives it
const res = await ctx.enqueue(runat, { probe: p.probe, enqueued_before_ms: before }, { runAtMs: before + 1500 })
return report(ctx, 'produce_runat', p, { jid: res.jid })
} }),class ProduceRunatJob
include JoobQ::Job
job_name "conf.produce_runat"
queue QUEUE
def self.perform(p, ctx)
before = now_ms # before the call: the gateway schedules the job to run at run_at_ms
res = ctx.enqueue(RunatJob, { probe: p["probe"], enqueued_before_ms: before }, run_at_ms: before + 1500)
report(ctx, "produce_runat", p, jid: res["jid"])
end
endjoobq.DefineJob[*App]("conf.produce_runat", opts, func(p map[string]any, ec *EC) (any, error) {
before := nowMs() // before the call: run_at_ms is an absolute instant the gateway honours
jid, err := runat.EnqueueFrom(ec, map[string]any{"probe": p["probe"], "enqueued_before_ms": before}, joobq.EnqueueOptions{RunAtMs: before + 1500})
if err != nil {
return nil, err
}
return report(ec, "produce_runat", p, map[string]any{"jid": jid})
}),JobDef::new("conf.produce_runat", queue.clone(), move |p: Value, ctx: &Ctx| {
let before = now_ms(); // before the call: the run_at is an absolute time the gateway schedules to
let jid = ctx.enqueue(&producer_runat, json!({"probe": p["probe"], "enqueued_before_ms": before}), EnqueueOptions { run_at_ms: Some(before + 1500), ..Default::default() })?;
report(ctx, "produce_runat", &p, json!({"jid": jid}))
}),JobDefinition.<App>of("conf.produce_runat", QUEUE, 3, (p, ctx) -> {
long before = System.currentTimeMillis(); // before the call: the gateway schedules at the absolute run_at_ms
String jid = ctx.enqueue(runat, map("probe", m(p).get("probe"), "enqueued_before_ms", before), new EnqueueOptions().runAtMs(before + 1500));
return report(ctx, "produce_runat", p, map("jid", jid));
}),final class ProduceRunAtJob extends ConfJob
{
public static function jobName(): string { return 'conf.produce_runat'; }
public static function perform(array $p, ExecutionContext $ctx): mixed
{
$before = now_ms(); // absolute schedule: the gateway honours run_at_ms as a wall-clock instant
$res = $ctx->enqueue(RunAtJob::class, ['probe' => $p['probe'], 'enqueued_before_ms' => $before], ['run_at_ms' => $before + 1500]);
return report($ctx, 'produce_runat', $p, ['jid' => $res['jid']]);
}
}JobDefinition<App>.Define("conf.produce_runat", Queue, 3, async (p, ctx) =>
{
var before = NowMs(); // read before the enqueue call; the job runs at this absolute time + 1500ms
var jid = await ctx.EnqueueAsync(runatJob, new { probe = p?["probe"]?.GetValue<string>(), enqueued_before_ms = before }, new EnqueueOptions(RunAtMs: before + 1500));
return await Report(ctx, "produce_runat", p, new() { ["jid"] = jid });
}),Not in every language
- Crystal: JoobQ Core schedules a Crystal job with
delay(for:), which serves for a run-at too.
Tested against a real server in every language shown.


Enqueue options
| Field (wire name) | Default | What it does |
|---|---|---|
queue, job_type | (required) | Where the job goes and which handler runs it (1 to 128 and 1 to 200 characters) |
payload | {} | The job's input |
context | {} | Request context handed to the handler (ctx.context) |
jid | a random UUID | The job's id; the answer carries it |
max_retries | 3 | 0 to 100 |
delay_ms | 0 | 0 to 86,400,000 |
run_at_ms | none | Absolute time; wins over delay_ms |
expires_ms | JoobQ's configured expires | The job's lifetime from when it is due, after its delay or at run_at_ms (1 s to a year), stored on the job as expires |
timeout_ms | none | A per-run deadline (1 ms to a day), from the moment a worker claims the job |
encrypt_fields | [] | Payload fields sealed with JOOBQUI_PAYLOAD_KEYS (Payload encryption) |
unique_key, unique_ttl_ms, unique_mode | none, 1 h, while_locked | Unique jobs |
A job can also be enqueued by its queue and type without its definition: enqueue_job(queue, job_type, payload, ...) in Python (enqueueJob, EnqueueJob, ... in the others), from a client or from a handler's context, as some of the unique-jobs samples do.
The SDKs name the options after their language: run_at_ms (Python, Ruby, Rust, PHP), runAtMs (TypeScript, Java), RunAtMs (Go, .NET), and so on. The answer is 201 {jid, queue, status} with status enqueued, scheduled (delayed) or deduplicated.
@job("conf.produce_options", queue=QUEUE)
def produce_options_job(payload, ctx):
probe = payload["probe"]
key = "optkey-" + probe
r1 = ctx.enqueue_job(QUEUE, "conf.opt", {"probe": probe, "tag": "first"},
expires_ms=3_600_000, timeout_ms=60_000, unique_key=key, unique_ttl_ms=60_000)
r2 = ctx.enqueue_job(QUEUE, "conf.opt", {"probe": probe, "tag": "second"},
unique_key=key, unique_ttl_ms=60_000)
report(ctx, "produce_options", payload, status1=r1.get("status"), status2=r2.get("status"))defineJob<App>({ name: 'conf.produce_options', queue: QUEUE, perform: async (p, ctx) => {
const key = `optkey-${p.probe}` // same unique key on both enqueues: the second is deduplicated
await ctx.enqueue(opt, { probe: p.probe, tag: 'first' }, { expiresMs: 3600000, timeoutMs: 60000, uniqueKey: key, uniqueTtlMs: 60000 })
await ctx.enqueue(opt, { probe: p.probe, tag: 'second' }, { uniqueKey: key, uniqueTtlMs: 60000 })
return report(ctx, 'produce_options', p)
} }),class ProduceOptionsJob
include JoobQ::Job
job_name "conf.produce_options"
queue QUEUE
def self.perform(p, ctx)
key = "optkey-#{p['probe']}"
ctx.enqueue(OptJob, { probe: p["probe"], tag: "first" },
expires_ms: 3_600_000, timeout_ms: 60_000, unique_key: key, unique_ttl_ms: 60_000)
ctx.enqueue(OptJob, { probe: p["probe"], tag: "second" }, unique_key: key, unique_ttl_ms: 60_000)
report(ctx, "produce_options", p)
end
endjoobq.DefineJob[*App]("conf.produce_options", opts, func(p map[string]any, ec *EC) (any, error) {
key := fmt.Sprintf("optkey-%v", p["probe"])
if _, err := opt.EnqueueFrom(ec, map[string]any{"probe": p["probe"], "tag": "first"},
joobq.EnqueueOptions{ExpiresMs: 3600000, TimeoutMs: 60000, UniqueKey: key}); err != nil {
return nil, err
}
// same unique key within its TTL: the gateway deduplicates this one
if _, err := opt.EnqueueFrom(ec, map[string]any{"probe": p["probe"], "tag": "second"},
joobq.EnqueueOptions{UniqueKey: key}); err != nil {
return nil, err
}
return report(ec, "produce_options", p, nil)
}),JobDef::new("conf.produce_options", queue.clone(), move |p: Value, ctx: &Ctx| {
let key = format!("optkey-{}", p["probe"].as_str().unwrap_or(""));
ctx.enqueue(&producer_opt, json!({"probe": p["probe"], "tag": "first"}), EnqueueOptions { expires_ms: Some(3600000), timeout_ms: Some(60000), unique_key: Some(key.clone()), ..Default::default() })?;
ctx.enqueue(&producer_opt, json!({"probe": p["probe"], "tag": "second"}), EnqueueOptions { unique_key: Some(key.clone()), ..Default::default() })?;
report(ctx, "produce_options", &p, json!({}))
}),JobDefinition.<App>of("conf.produce_options", QUEUE, 3, (p, ctx) -> {
String key = "optkey-" + m(p).get("probe");
ctx.enqueue(opt, map("probe", m(p).get("probe"), "tag", "first"), new EnqueueOptions().expiresMs(3600000).timeoutMs(60000).uniqueKey(key).uniqueTtlMs(60000));
ctx.enqueue(opt, map("probe", m(p).get("probe"), "tag", "second"), new EnqueueOptions().uniqueKey(key).uniqueTtlMs(60000));
return report(ctx, "produce_options", p, Map.of());
}),final class ProduceOptionsJob extends ConfJob
{
public static function jobName(): string { return 'conf.produce_options'; }
public static function perform(array $p, ExecutionContext $ctx): mixed
{
$key = 'optkey-' . $p['probe'];
$ctx->enqueue(OptJob::class, ['probe' => $p['probe'], 'tag' => 'first'], ['expires_ms' => 3600000, 'timeout_ms' => 60000, 'unique_key' => $key, 'unique_ttl_ms' => 60000]);
$ctx->enqueue(OptJob::class, ['probe' => $p['probe'], 'tag' => 'second'], ['unique_key' => $key, 'unique_ttl_ms' => 60000]); // same key -> deduplicated
return report($ctx, 'produce_options', $p);
}
}JobDefinition<App>.Define("conf.produce_options", Queue, 3, async (p, ctx) =>
{
var probe = p?["probe"]?.GetValue<string>();
var key = "optkey-" + probe;
await ctx.EnqueueAsync(optJob, new { probe, tag = "first" }, new EnqueueOptions(UniqueKey: key, ExpiresMs: 3600000, TimeoutMs: 60000));
await ctx.EnqueueAsync(optJob, new { probe, tag = "second" }, new EnqueueOptions(UniqueKey: key)); // same key -> deduplicated
return await Report(ctx, "produce_options", p);
}),Not in every language
- Crystal: Gateway-mediated enqueue options (expires/timeout/unique): a Crystal job sets
@expiresand JoobQ Pro's unique jobs itself.
Tested against a real server in every language shown.
What expires_ms and timeout_ms do
The gateway enforces both, as JoobQ's Timeout middleware does for a Crystal worker:
expires_ms: counted from when the job is due, so a delayed or scheduled job is not expired before it could run. A job still waiting when it expires is not handed to a worker. It goes to the dead letter queue with the error typeJobExpired(its uniqueness lock, if any, is freed), and the poll takes the next job.timeout_ms: counted from the moment a worker claims the job. The lease ends at that deadline at the latest: renewals do not extend it further. Within a second or so of the deadline (the gateway's sweep) the run fails with the error typeJobTimeout. That spends a retry like any other failure: the job runs again while retries remain, else it is dead-lettered. Acompletethat arrives later answers 404.
A handler is not interrupted: it runs on, but its result is no longer accepted. A renewal made after the deadline, if the lease still exists, answers cancel_requested: true.
Batch enqueue and enqueueing without waiting
enqueue_many (enqueueMany, EnqueueMany, EnqueueManyAsync) sends up to 500 jobs in one POST /sdk/jobs/batch: see Batch enqueue and batches. enqueue answers once the job is stored; enqueue_async (enqueueAsync, EnqueueAsync) does not make the caller wait (a future, a channel or a thread) in Python, Ruby, Go, Rust and Java. TypeScript's and .NET's enqueue are asynchronous already; PHP has none.
Core or Pro
Everything on this page is Core. The release modes of unique jobs (until_executing, until_executed) are Pro, and so is rate limiting.