Delivery Guarantees

FlexiQ gives exactly-once dispatch: an atomic claim in storage makes sure two workers never pick up the same job at once. But the system as a whole is at-least-once execution — if a worker crashes after it starts a task but before the result is recorded, the job is retried and the task body runs again. Design task code to tolerate that.

How a job is claimed

Before dispatching a job, the scheduler makes an atomic claim on it — a guarded update only one caller can win (SQLite: INSERT OR IGNORE; Postgres: INSERT ... ON CONFLICT DO NOTHING; Redis: SET NX). If another scheduler instance already claimed the job, the claim fails and that instance skips it.

• A job will not be lost — if a worker dies, the stale reaper detects the abandoned claim and retries the job.
• A job may run twice — if a worker crashes after starting the task but before its result is recorded.
• A job will not be dispatched concurrently — the atomic claim means only one worker is ever handed it at a time. The narrow exception: when an attempt times out, the reaper schedules a retry but cannot kill the still-running task body, so the new attempt can briefly overlap the old one. Idempotent task code covers this case too.

Crash recovery

The claim prevents duplicate dispatch — two workers picking up the same job at once. It does not prevent duplicate execution after a crash: the claim lapses once the stale reaper detects the timeout, and the job is dispatched again as a new attempt.

Why not exactly-once?

Exactly-once delivery is impossible in general: a worker can run a task's side effect and then crash before it acknowledges completion, so the job is retried and the side effect runs again. No protocol closes this gap — the effect and the acknowledgement can't be committed atomically across separate systems. What you can get is exactly-once effect, by making tasks idempotent (dedup keys, upserts) or routing writes through a transactional outbox/inbox. This matches Celery, SQS, and most production job systems: deliver at least once, design tasks to handle duplicates.

What's not guaranteed

• Exactly-once execution — not offered anywhere in the system (see above); pair at-least-once delivery with idempotent handlers instead.
• Strict ordering — within a queue, jobs dequeue by priority then age, but concurrent workers run them in parallel; don't rely on one finishing before another. Use a workflow when order matters.

Writing idempotent tasks

Because a job may run more than once, design task code to be safe on re-execution:

• Upsert or use a conditional write keyed by a stable id, instead of a blind insert.
• Dedupe external calls (payments, emails, webhooks) using the job id, or a stable business id, as an idempotency key with the downstream provider.
• Collapse duplicate enqueues into a single job with a unique key — see Deduplicating enqueues below.

@queue.task()
def create_user(email, name):
    # UPSERT — safe to run twice
    db.execute(
        "INSERT INTO users (email, name) VALUES (?, ?) "
        "ON CONFLICT (email) DO UPDATE SET name = ?",
        (email, name, name),
    )

@queue.task()
def charge_customer(order_id, amount):
    # Check if already charged
    if db.execute("SELECT 1 FROM charges WHERE order_id = ?", (order_id,)).fetchone():
        return  # Already processed

    payment_provider.charge(amount, idempotency_key=f"order-{order_id}")
    db.execute("INSERT INTO charges (order_id, amount) VALUES (?, ?)", (order_id, amount))

import { currentJob } from "@byteveda/flexiq";

queue.task("charge", async (orderId: string) => {
  const job = currentJob();
  await payments.charge(orderId, { idempotencyKey: job?.jobId });
});

// Java handlers receive only the payload, so idempotency comes from a
// natural key in it — here order.id().
Worker worker = queue.worker()
        .handle(CHARGE, order -> {
            if (charges.exists(order.id())) return;      // already processed on a prior attempt
            payments.charge(order.total(), order.id());   // key also dedupes at the provider
            charges.record(order.id(), order.total());
        })
        .start();

// A task body receives its decoded arguments and nothing else, so idempotency
// comes from a natural key in them — here `order_id`.
#[flexiq::task]
fn charge_customer(order_id: String, amount_cents: i64) -> flexiq::Outcome<()> {
    let retry = |e: std::io::Error| flexiq::TaskError::retryable(e.to_string());

    if charges::exists(&order_id).map_err(retry)? {
        return Ok(()); // already processed on a prior attempt
    }
    payments::charge(amount_cents, &order_id).map_err(retry)?; // also dedupes at the provider
    charges::record(&order_id, amount_cents).map_err(retry)?;
    Ok(())
}

# Bad — sends duplicate emails on retry
@queue.task()
def notify(user_id):
    send_email(user_id, "Your order shipped")

# Good — atomically claim the send before doing it
@queue.task()
def notify(user_id):
    claimed = db.execute(
        "UPDATE orders SET notified = 1 WHERE user_id = ? AND notified = 0",
        (user_id,),
    )
    if claimed.rowcount:
        send_email(user_id, "Your order shipped")

#[flexiq::task]
fn notify(user_id: String) -> flexiq::Outcome<()> {
    let retry = |e: std::io::Error| flexiq::TaskError::retryable(e.to_string());

    if orders::claim_notification(&user_id).map_err(retry)? {
        email::send(&user_id, "Your order shipped").map_err(retry)?;
    }
    Ok(())
}

Deduplicating enqueues

A unique key coalesces duplicate enqueues into the same job while the first job for that key is still pending or running — the second call returns the existing job's id instead of creating a new one. It's backed by a partial unique index in storage, so the dedup is atomic across producers and processes.

# Only one pending/running instance per key
job1 = send_report.apply_async(args=(user_id,), unique_key=f"report-{user_id}")
job2 = send_report.apply_async(args=(user_id,), unique_key=f"report-{user_id}")
assert job2.id == job1.id  # coalesced onto job1 — no duplicate created

queue.enqueue("welcome", [user.id], { uniqueKey: `welcome:${user.id}` });
queue.enqueue("welcome", [user.id], { uniqueKey: `welcome:${user.id}` }); // ignored

EnqueueOptions once = EnqueueOptions.builder()
        .uniqueKey("welcome:" + user.id())
        .build();

String first = queue.enqueue(WELCOME, user.id(), once);
String second = queue.enqueue(WELCOME, user.id(), once); // same id — no new job

let key = format!("welcome:{}", user.id);

let first = queue.enqueue(welcome::call(user.id.clone()).unique_key(key.clone()))?;
let second = queue.enqueue(welcome::call(user.id.clone()).unique_key(key))?;

assert_eq!(second.id, first.id); // coalesced onto the first — no new job

The key frees up once the job finishes — a duplicate enqueue after that creates a fresh job. Only a pending or running job holds the key; one that completed, was dead-lettered, or was cancelled no longer blocks reuse.

See Idempotency for more on designing dedup keys.

Framework vs. task responsibility

Comparison with other queues

Inspecting results from another process

A job's result is stored durably, so any process sharing the same storage can look it up by job id — not just the worker that produced it. Useful when a web backend enqueues a job and a different process needs to poll its outcome.

job = queue.get_job(job_id)
result = job.result(timeout=10)  # blocks until the job settles

while Instant::now() < deadline {
    if let Some(job) = queue.get_job(job_id)? {
        if matches!(job.status, JobStatus::Complete | JobStatus::Failed) {
            break;
        }
    }
    std::thread::sleep(Duration::from_millis(25));
}