How to test duplicate job execution

Count the side effect where it lands, not the lines the worker logged. Tag every execution with a run id and every delivery with an attempt number, then read them back: 2 effect(s) from 2 run(s) is a job that executed twice, and 2 effect(s) from 1 run(s) is one execution sending twice.

Why check this

A queue that promises at-least-once delivery hands the same job to a worker again after a lost acknowledgement, an expired lease or a deploy that replaces a consumer. Run this check when a customer reports two receipts for one order, and on staging sign-off after a change to worker count or lease duration.

It prevents this: a deploy stops a worker between the payment call and the acknowledgement, the lease expires, a second worker charges the customer again, and the job row reads status=done attempts=2 and looks healthy.

Three questions follow. Did the work happen twice. Which layer produced the second one. Was that delivery a defect at all.

Prerequisites

• Node 22 or later, for the bundled node:sqlite. It warns on start, so every command below is wrapped in ( ... ) 2>/dev/null.
• A free local port. This run used 8933. Confirm it with netstat -ano | grep 8933, and stop the service afterwards by the process id that prints.
• receipts.mjs, the service the job calls. It records each request on arrival, so its list counts work performed, not work reported.

// receipts.mjs: the service the job calls, and the only honest witness to the work.
import { createServer } from 'node:http';

const effects = [];
createServer((req, res) => {
  const url = new URL(req.url, 'http://127.0.0.1');
  if (req.method === 'DELETE') { effects.length = 0; res.end('cleared'); return; }
  if (req.method === 'POST') {
    let body = '';
    req.on('data', (d) => { body += d; }).on('end', () => {
      effects.push({ at: Date.now(), ...JSON.parse(body) });   // recorded on arrival
      setTimeout(() => res.end('ok'), Number(url.searchParams.get('delay') ?? 0));
    });
    return;
  }
  res.setHeader('content-type', 'application/json');
  res.end(JSON.stringify(effects));
}).listen(8933, '127.0.0.1', () => console.log('receipts service on 127.0.0.1:8933'));

• queue.mjs, a leased queue and its worker. The worker claims a job row, sends the receipt, then acknowledges, and the flags fail it in one place at a time.

// queue.mjs: a leased queue, a worker, and the two places a duplicate comes from.
import { DatabaseSync } from 'node:sqlite';
import { spawn } from 'node:child_process';
import { randomUUID } from 'node:crypto';
import { appendFileSync, writeFileSync } from 'node:fs';

const db = new DatabaseSync('queue.db', { timeout: 5000 });
db.exec('PRAGMA journal_mode = WAL');
db.exec(`CREATE TABLE IF NOT EXISTS jobs (
  id INTEGER PRIMARY KEY, order_id TEXT, status TEXT DEFAULT 'queued',
  attempts INTEGER DEFAULT 0, run TEXT, lease INTEGER DEFAULT 0)`);

const rest = process.argv.slice(3);
const arg = (n, d = null) => { const i = rest.indexOf(n); return i < 0 ? d : rest[i + 1]; };
const has = (n) => rest.includes(n);
const svc = 'http://127.0.0.1:8933';
const cmd = process.argv[2];

if (cmd === 'add') db.prepare('INSERT INTO jobs (order_id) VALUES (?)').run(rest[0]);

if (cmd === 'reset') {
  db.exec('DELETE FROM jobs');
  writeFileSync('worker.log', '');
  await fetch(svc, { method: 'DELETE' });
}

if (cmd === 'jobs') for (const j of db.prepare('SELECT id, order_id, status, attempts, run FROM jobs').all())
  console.log(`job ${j.id} ${j.order_id} status=${j.status} attempts=${j.attempts} run=${j.run ?? '-'}`);

if (cmd === 'effects') {                       // count the work, not the log lines
  const rows = await (await fetch(svc)).json();
  const by = {};
  for (const e of rows) (by[e.order] ??= []).push(e);
  for (const [order, list] of Object.entries(by)) {
    const tag = list.map((e) => `${e.run} a${e.attempt} s${e.send}`);
    const span = list[list.length - 1].at - list[0].at;   // a race lands in ms, a lease in seconds
    console.log(`${order}: ${list.length} effect(s) from ${new Set(list.map((e) => e.run)).size} run(s) over ${span} ms [${tag.join(', ')}]`);
  }
  if (!rows.length) console.log('no effects recorded');
}

if (cmd === 'work') {
  const worker = arg('--worker', 'W1');
  const run = randomUUID().slice(0, 8);         // one id per execution, not per job
  const lease = Number(arg('--lease', 30)) * 1000;
  while (Date.now() < Number(arg('--at', 0))) { /* both workers leave the gate together */ }
  const now = Date.now();
  const job = db.prepare('SELECT * FROM jobs WHERE id = ?').get(Number(arg('--job')));

  let claimed;
  if (arg('--claim', 'read') === 'read') {      // read the row, then write it
    claimed = job.status === 'queued' || job.lease < now;
    if (claimed) db.prepare('UPDATE jobs SET status = ?, run = ?, attempts = attempts + 1, lease = ? WHERE id = ?')
      .run('running', run, now + lease, job.id);
  } else {                                      // one conditional UPDATE decides it
    claimed = db.prepare(`UPDATE jobs SET status = 'running', run = ?, attempts = attempts + 1, lease = ?
      WHERE id = ? AND (status = 'queued' OR lease < ?)`).run(run, now + lease, job.id, now).changes === 1;
  }
  if (!claimed) { console.log(`${worker} run ${run}: job ${job.id} already claimed`); process.exit(0); }

  const attempt = db.prepare('SELECT attempts FROM jobs WHERE id = ?').get(job.id).attempts;
  const tag = has('--trace') ? `[run ${run} attempt ${attempt}] ` : '';
  const delay = arg('--delay', '0');
  if (has('--crash-before-effect')) { console.log(`${worker} run ${run}: killed before the receipt`); process.exit(1); }

  for (let send = 1; send <= 2; send += 1) {    // one client-side retry on a timeout
    appendFileSync('worker.log', `${tag}sending receipt for ${job.order_id}\n`);
    try {
      await fetch(`${svc}/?delay=${send === 1 ? delay : 0}`, {
        method: 'POST', signal: AbortSignal.timeout(Number(arg('--timeout', 5000))),
        body: JSON.stringify({ order: job.order_id, run, attempt, worker, send }),
      });
      break;
    } catch (e) {
      console.log(`${worker} run ${run}: POST ${e.name}, retrying`);
      if (send === 2) throw e;
    }
  }
  console.log(`${worker} run ${run}: receipt sent for ${job.order_id}`);

  if (has('--crash-before-ack')) { console.log(`${worker} run ${run}: killed before the ack`); process.exit(1); }
  db.prepare("UPDATE jobs SET status = 'done' WHERE id = ?").run(job.id);
}

if (cmd === 'race') {                           // two workers, one start instant, N times
  const claim = rest[0], reps = Number(rest[1] ?? 20);
  let both = 0;
  for (let i = 0; i < reps; i += 1) {
    db.exec('DELETE FROM jobs');
    await fetch(svc, { method: 'DELETE' });
    db.prepare('INSERT INTO jobs (order_id) VALUES (?)').run('order-77');
    const id = db.prepare('SELECT max(id) AS id FROM jobs').get().id;
    const at = Date.now() + 400;
    const one = (w) => new Promise((done) => spawn(process.execPath,
      [import.meta.filename, 'work', '--job', String(id), '--worker', w, '--claim', claim, '--at', String(at)],
      { stdio: i === 0 ? 'inherit' : 'ignore' }).on('exit', done));
    await Promise.all([one('W1'), one('W2')]);
    if ((await (await fetch(svc)).json()).length > 1) both += 1;
  }
  console.log(`claim=${claim}: two effects in ${both} of ${reps} races`);
}

• One SQLite file on loopback is not a production queue: the counts carry over, the milliseconds do not. Delete queue.db and worker.log afterwards.

Steps

1. 1

   Step 1.

   Start the service with node receipts.mjs in its own shell, then enqueue one job and work it.

   ( node queue.mjs reset && node queue.mjs add order-77 && node queue.mjs work --job 1 --worker W1 && cat worker.log && node queue.mjs jobs && node queue.mjs effects ) 2>/dev/null
   

   W1 run 98f36948: receipt sent for order-77
   sending receipt for order-77
   job 1 order-77 status=done attempts=1 run=98f36948
   order-77: 1 effect(s) from 1 run(s) over 0 ms [98f36948 a1 s1]

   One log line, one attempt, one effect. The line names the order and nothing else, which is where most handlers are.

2. 2

   Step 2.

   Shorten the lease to one second, stop the worker between the receipt and the acknowledgement, then wait for the lease to expire.

   ( node queue.mjs reset && node queue.mjs add order-77 && node queue.mjs work --job 1 --worker W1 --lease 1 --crash-before-ack; sleep 2; node queue.mjs work --job 1 --worker W2 && cat worker.log && node queue.mjs jobs && node queue.mjs effects ) 2>/dev/null
   

   W1 run 69a76c08: receipt sent for order-77
   W1 run 69a76c08: killed before the ack
   W2 run 5f0fb134: receipt sent for order-77
   sending receipt for order-77
   sending receipt for order-77
   job 1 order-77 status=done attempts=2 run=5f0fb134
   order-77: 2 effect(s) from 2 run(s) over 2159 ms [69a76c08 a1 s1, 5f0fb134 a2 s1]

   The two log lines are identical, so the log cannot say whether the work happened twice. The effect list can: two run ids, 2159 ms apart. Nothing malfunctioned. The first execution succeeded and its acknowledgement was lost.

3. 3

   Step 3.

   Deliver the job once, and time out the first receipt call after the service records it. --trace puts the run id in the log.

   ( node queue.mjs reset && node queue.mjs add order-77 && node queue.mjs work --job 1 --worker W1 --trace --delay 400 --timeout 300 && cat worker.log && node queue.mjs jobs && node queue.mjs effects ) 2>/dev/null
   

   W1 run bc05071b: POST TimeoutError, retrying
   W1 run bc05071b: receipt sent for order-77
   [run bc05071b attempt 1] sending receipt for order-77
   [run bc05071b attempt 1] sending receipt for order-77
   job 1 order-77 status=done attempts=1 run=bc05071b
   order-77: 2 effect(s) from 1 run(s) over 282 ms [bc05071b a1 s1, bc05071b a1 s2]

   Two effects, and two identical log lines again. One run id, attempts 1, two sends: the job was delivered once and executed once. The duplicate came from the HTTP client inside that execution.

4. 4

   Step 4.

   Hand one job to two workers released on a shared timestamp, 20 times, with the claim that reads the row then writes it.

   ( node queue.mjs race read 20 && node queue.mjs effects ) 2>/dev/null
   

   W2 run 36b623c4: receipt sent for order-77
   W1 run 184ff4ee: receipt sent for order-77
   claim=read: two effects in 19 of 20 races
   order-77: 2 effect(s) from 2 run(s) over 2 ms [3b8c3810 a1 s1, fdf530f2 a2 s1]

   Both workers read status=queued before either wrote running, so both owned the job. Only the first race prints its workers; the effect list is the last race. The run ids look like step 2 at 2 ms instead of 2159 ms, and that gap separates a claim race from an expired lease. An earlier run duplicated 20 of 20.

5. 5

   Step 5.

   Repeat the races with the claim that decides ownership in one conditional UPDATE.

   ( node queue.mjs race lock 20 && node queue.mjs effects ) 2>/dev/null
   

   W2 run 7ac8164e: job 1 already claimed
   W1 run 831228d8: receipt sent for order-77
   claim=lock: two effects in 0 of 20 races
   order-77: 1 effect(s) from 1 run(s) over 0 ms [142ffce1 a1 s1]

   Zero of 20. One statement is the only difference from step 4, which pins the duplicate on the claim rather than on the lease or the client.

6. 6

   Step 6.

   Fail the worker one line earlier, before the receipt goes out, and let the lease expiry redeliver.

   ( node queue.mjs reset && node queue.mjs add order-77 && node queue.mjs work --job 1 --worker W1 --lease 1 --trace --crash-before-effect; sleep 2; node queue.mjs work --job 1 --worker W2 --trace && cat worker.log && node queue.mjs jobs && node queue.mjs effects ) 2>/dev/null
   

   W1 run ac6a7144: killed before the receipt
   W2 run 4c95c329: receipt sent for order-77
   [run 4c95c329 attempt 2] sending receipt for order-77
   job 1 order-77 status=done attempts=2 run=4c95c329
   order-77: 1 effect(s) from 1 run(s) over 0 ms [4c95c329 a2 s1]

   Two deliveries, one effect, and the attempt counter reads 2 exactly as in step 2. A duplicate delivery is the guarantee working, a duplicate effect is the defect.

How to read the result

| What you see | What it means | What to do | | --- | --- | --- | | 2 effect(s) from 2 run(s), gap in seconds | Two executions, the second after a lease expired | Find the execution that did the work and never acknowledged it. | | 2 effect(s) from 2 run(s), gap in milliseconds | Two workers claimed one delivery | Compare against step 5. A claim that wins in one conditional UPDATE ends it. | | 2 effect(s) from 1 run(s), one attempt | One execution sent the effect twice | Look at the HTTP client's retry, not at the queue. | | attempts 2 with one effect | The queue redelivered and no work repeated | Nothing. This is at-least-once delivery behaving. | | Identical log lines and no run id | The evidence cannot answer the question | Put a run id and an attempt number on the effect before testing further. |

Common mistakes

What to check next

• How to check a background job is idempotent: the fix, under the race in step 4.
• How to test duplicate payment prevention: the same count when money moves.
• How to check correlation id in logs: carrying the run id between services.
• How to test a stuck background job: the opposite failure, a lease never released.
• How to check if a cron job ran: the prior question, whether it ran at all.

FAQ

What is at-least-once delivery?

A guarantee that a job reaches a worker, with no promise about how often. A lost acknowledgement, an expired lease or a restarted consumer each produce a second delivery, as step 6 does.

Why does my queue process the same job twice?

Three causes, and steps 2, 3 and 4 tell them apart. A lease expired under the first execution, two workers claimed one row, or the effect call was retried inside one execution.

How do I tell a duplicate delivery from a duplicate effect?

Count them in different stores. The job row counts deliveries in its attempt column, the service receiving the work counts effects. Step 6 ends at two deliveries and one effect, step 2 at two of each.

How many repeats does a race need?

More than one. The 20 races in step 4 duplicated 19 times and an earlier 20 duplicated every time, so one run reports either as the rule. Publish the count, not one observation.

Verified