JoobQ

Appearance

Unique jobs ​

A job enqueued with a unique_key takes a lock on (queue, unique_key). While the lock is held, an identical enqueue answers status: "deduplicated" with the jid of the job that holds it, and the duplicate is dropped. unique_mode says when the lock frees.

ProThe release modes until_executing and until_executed needs a JoobQ Pro key. Without one every Core feature keeps working: the Pro routes of the SDK gateway answer 402 pro_required and the console's Pro routes are not served. Get a licence · Core and Pro · Pro features index

Modes ​

unique_modeEditionThe lock frees
while_locked (default)Corewhen unique_ttl_ms passes (default 1 hour, clamped to 1 s .. 7 days)
until_executingProthe moment a worker claims the job, so an identical job may be enqueued while this one runs
until_executedProwhen the job completes successfully. A job that fails for good (dead letter) does not free it: the lock then lives until its TTL

Every mode keeps the TTL as an upper bound. Only the job that holds the lock frees it (the release checks its jid), so a later job that took the lock after the TTL is not affected.

Core: while_locked ​

The second enqueue below reuses the key within its TTL and is deduplicated:

python
@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"))
ts
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)
} }),
ruby
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
end
go
joobq.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)
}),
rust
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!({}))
}),
java
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());
}),
php
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);
    }
}
csharp
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 @expires and JoobQ Pro's unique jobs itself.

Tested against a real server in every language shown.

Pro: until_executed ​

Both enqueues happen while the first job is still pending or running, so the second is deduplicated; once the first has completed, the same key is accepted again.

python
@job("conf.produce_unique", queue=QUEUE)
def produce_unique_job(payload, ctx):
    probe = payload["probe"]
    key = "um-" + probe
    # n=1 acquires the until_executed lock and runs ~2s; n=2 reuses the key while n=1 is pending/running → deduplicated.
    ctx.enqueue_job(QUEUE, "conf.uniqueslow", {"probe": probe, "n": 1}, unique_key=key, unique_mode="until_executed")
    ctx.enqueue_job(QUEUE, "conf.uniqueslow", {"probe": probe, "n": 2}, unique_key=key, unique_mode="until_executed")
    report(ctx, "produce_unique", payload)
ts
defineJob<App>({ name: 'conf.produce_unique', queue: QUEUE, perform: async (p, ctx) => {
  const key = `um-${p.probe}` // same key + until_executed: the second enqueue is deduplicated while the first runs
  await ctx.enqueue(uniqueslow, { probe: p.probe, n: 1 }, { uniqueKey: key, uniqueMode: 'until_executed' })
  await ctx.enqueue(uniqueslow, { probe: p.probe, n: 2 }, { uniqueKey: key, uniqueMode: 'until_executed' })
  return report(ctx, 'produce_unique', p)
} }),
ruby
class ProduceUniqueJob
  include JoobQ::Job
  job_name "conf.produce_unique"
  queue QUEUE
  def self.perform(p, ctx)
    key = "um-#{p['probe']}"
    ctx.enqueue_job(queue: QUEUE, job_type: "conf.uniqueslow", payload: { probe: p["probe"], n: 1 },
                    unique_key: key, unique_mode: "until_executed")
    ctx.enqueue_job(queue: QUEUE, job_type: "conf.uniqueslow", payload: { probe: p["probe"], n: 2 },
                    unique_key: key, unique_mode: "until_executed")
    report(ctx, "produce_unique", p)
  end
end
go
joobq.DefineJob[*App]("conf.produce_unique", opts, func(p map[string]any, ec *EC) (any, error) {
	key := fmt.Sprintf("um-%v", p["probe"])
	if _, err := uniqueSlow.EnqueueFrom(ec, map[string]any{"probe": p["probe"], "n": 1},
		joobq.EnqueueOptions{UniqueKey: key, UniqueMode: "until_executed"}); err != nil {
		return nil, err
	}
	// same key + mode while the first is still unexecuted: the gateway holds the lock
	if _, err := uniqueSlow.EnqueueFrom(ec, map[string]any{"probe": p["probe"], "n": 2},
		joobq.EnqueueOptions{UniqueKey: key, UniqueMode: "until_executed"}); err != nil {
		return nil, err
	}
	return report(ec, "produce_unique", p, nil)
}),
rust
JobDef::new("conf.produce_unique", queue.clone(), move |p: Value, ctx: &Ctx| {
    let key = format!("um-{}", p["probe"].as_str().unwrap_or(""));
    ctx.enqueue(&producer_unique, json!({"probe": p["probe"], "n": 1}), EnqueueOptions { unique_key: Some(key.clone()), unique_mode: Some("until_executed".into()), ..Default::default() })?;
    ctx.enqueue(&producer_unique, json!({"probe": p["probe"], "n": 2}), EnqueueOptions { unique_key: Some(key.clone()), unique_mode: Some("until_executed".into()), ..Default::default() })?;
    report(ctx, "produce_unique", &p, json!({}))
}),
java
JobDefinition.<App>of("conf.produce_unique", QUEUE, 3, (p, ctx) -> {
    String key = "um-" + m(p).get("probe");
    ctx.enqueue(uniqueslow, map("probe", m(p).get("probe"), "n", 1), new EnqueueOptions().uniqueKey(key).uniqueMode("until_executed"));
    ctx.enqueue(uniqueslow, map("probe", m(p).get("probe"), "n", 2), new EnqueueOptions().uniqueKey(key).uniqueMode("until_executed"));
    return report(ctx, "produce_unique", p, Map.of());
}),
php
final class ProduceUniqueJob extends ConfJob
{
    public static function jobName(): string { return 'conf.produce_unique'; }
    public static function perform(array $p, ExecutionContext $ctx): mixed
    {
        $key = 'um-' . $p['probe'];
        $ctx->client->enqueueJob(QUEUE, 'conf.uniqueslow', ['probe' => $p['probe'], 'n' => 1], ['unique_key' => $key, 'unique_mode' => 'until_executed']);
        $ctx->client->enqueueJob(QUEUE, 'conf.uniqueslow', ['probe' => $p['probe'], 'n' => 2], ['unique_key' => $key, 'unique_mode' => 'until_executed']);
        return report($ctx, 'produce_unique', $p);
    }
}
csharp
JobDefinition<App>.Define("conf.produce_unique", Queue, 3, async (p, ctx) =>
{
    var probe = p?["probe"]?.GetValue<string>();
    var key = "um-" + probe;
    await ctx.EnqueueAsync(uniqueSlowJob, new { probe, n = 1 }, new EnqueueOptions(UniqueKey: key, UniqueMode: "until_executed"));
    await ctx.EnqueueAsync(uniqueSlowJob, new { probe, n = 2 }, new EnqueueOptions(UniqueKey: key, UniqueMode: "until_executed"));
    return await Report(ctx, "produce_unique", p);
}),
Not in every language
  • Crystal: Gateway-mediated Pro unique_mode: Crystal uses JoobQ Pro's UniqueJobs library directly.

Tested against a real server in every language shown.

Without a Pro key ​

An enqueue with a unique_key and unique_mode until_executing or until_executed answers 402 pro_required and stores nothing; in a batch (POST /sdk/jobs/batch) one such job refuses the whole batch. Any other unique_mode value is 422 invalid_request. A unique_mode without a unique_key does nothing and is not gated.

Gotchas ​

  • The key is per queue: the same unique_key on two queues is two locks.
  • deduplicated is not an error: check status if your producer must know.
  • Delivery is at least once. A unique key stops a duplicate enqueue; it does not stop a job from running twice after a worker crash.
  • Crystal: JoobQ Pro's own unique jobs (JoobQPro::UniquenessMode UntilExecuting, UntilExecuted, WhileLocked) are a library feature with their own API, not this gateway option.

JoobQ Core is MIT licensed. JoobQ Pro needs a licence key.