Idempotent Jobs

The double-execution problem

A worker picks up a job, processes it (sends the email), and then crashes before calling completeJob. The job is still in "processing" status. After a timeout, the lock expires and another worker picks it up. The email is sent a second time.

This is not a bug in the queue — it is a fundamental property of distributed systems. Any job might run more than once. Your job handlers must be safe to repeat.

What idempotent means

An operation is idempotent if running it multiple times produces the same result as running it once. Sending the same email twice is not idempotent (the user gets two emails). Updating a status to “shipped” is idempotent (the status is “shipped” regardless of how many times you set it).

[!NOTE] The REST API Design course’s idempotency lesson explained this concept for HTTP methods: PUT and DELETE are idempotent, POST is not. The same principle applies to jobs.

Making jobs idempotent

Pattern 1: Idempotency keys. Track which jobs have been processed:

// Before processing, check if already done
async function handleSendEmail(payload: SendEmailPayload, jobId: string): Promise<void> {
  const already = db.prepare("SELECT 1 FROM processed_jobs WHERE job_id = ?").get(jobId);

  if (already) {
    console.log(`Job ${jobId} already processed. Skipping.`);
    return;
  }

  await sendEmail(payload.to, payload.subject, payload.body);

  db.prepare("INSERT INTO processed_jobs (job_id, processed_at) VALUES (?, datetime('now'))").run(
    jobId,
  );
}

The processed_jobs table records which jobs have been completed. If the same job runs again, the handler skips it.

Pattern 2: Naturally idempotent operations. Some operations are idempotent by nature:

// SET is idempotent — running it twice has the same result
db.prepare("UPDATE orders SET status = 'shipped' WHERE id = ?").run(orderId);

// INSERT OR IGNORE is idempotent — the second insert does nothing
db.prepare("INSERT OR IGNORE INTO notifications (order_id, type) VALUES (?, 'shipped')").run(
  orderId,
);

Pattern 3: Check-then-act. Check the current state before acting:

async function handleDeductStock(payload: {
  productId: string;
  orderId: string;
  quantity: number;
}): Promise<void> {
  // Check if stock was already deducted for this order
  const deduction = db
    .prepare("SELECT 1 FROM stock_deductions WHERE order_id = ? AND product_id = ?")
    .get(payload.orderId, payload.productId);

  if (deduction) return; // Already deducted

  db.transaction(() => {
    db.prepare("UPDATE products SET stock = stock - ? WHERE id = ?").run(
      payload.quantity,
      payload.productId,
    );
    db.prepare(
      "INSERT INTO stock_deductions (order_id, product_id, quantity) VALUES (?, ?, ?)",
    ).run(payload.orderId, payload.productId, payload.quantity);
  })();
}

The stock_deductions table records that stock was deducted for this order. If the job runs again, it sees the existing record and skips. The transaction ensures the deduction and the record are created together.

Stale lock cleanup

When a worker crashes, its jobs are left in "processing" with a stale lock. Add a cleanup function that releases locks older than a threshold:

export function releaseStaleJobs(staleLockMinutes: number = 10): number {
  const result = db
    .prepare(
      `
    UPDATE jobs
    SET status = 'pending', locked_by = NULL, locked_at = NULL,
        updated_at = datetime('now')
    WHERE status = 'processing'
      AND locked_at < datetime('now', '-' || ? || ' minutes')
  `,
    )
    .run(staleLockMinutes);

  return result.changes;
}

Run this periodically (every minute) or at worker startup. Jobs that have been processing for more than 10 minutes are assumed to be from a crashed worker and returned to "pending" for reprocessing.

This is exactly the scenario where idempotency matters — the job might have already been processed before the crash.

Exercises

Exercise 1: Simulate a crash: process a job, then do NOT call completeJob. Run releaseStaleJobs. Verify the job is picked up again.

Exercise 2: Make the email handler idempotent using the processed_jobs table. Process a job twice. Verify the email is sent only once.

Exercise 3: Write an idempotent stock deduction handler using a deductions tracking table. Deduct stock, then run the job again. Verify stock is deducted only once.