リトライで二重に処理される。「ちょうど1回」は作れない

// Three clients, one flaky network, one side effect that must not happen twice.
// Requires Node 18+ (global fetch). Run: node once.mjs   (counts vary slightly between runs: real timers)
import http from "node:http";

const INTENTS = 200;                 // distinct things the "user" wants to happen exactly once
const CONCURRENCY = 10;
const applied = new Map();           // intent id -> how many times the side effect really ran
const seen = new Map();              // idempotency key -> { fingerprint, promise }  (the dedupe store)

const server = http.createServer(async (req, res) => {
  let body = ""; for await (const c of req) body += c;
  const key = req.headers["idempotency-key"];
  const send = (code, obj) => { res.writeHead(code, { "content-type": "application/json" }); res.end(JSON.stringify(obj)); };

  const run = async () => {                                    // the side effect
    await new Promise((r) => setTimeout(r, Math.random() * 60)); // slow enough that clients sometimes give up first
    const { intent } = JSON.parse(body);
    applied.set(intent, (applied.get(intent) ?? 0) + 1);
    return { status: 201, intent };
  };

  if (!key) return send(201, await run());                     // no key: every request is a new request
  const fingerprint = body;
  const prev = seen.get(key);
  if (prev && prev.fingerprint !== fingerprint) return send(422, { error: "key reused with a different request" });
  if (prev) { const r = await prev.promise; return send(r.status, r); }   // duplicate or concurrent duplicate: share the one result
  const promise = run();
  seen.set(key, { fingerprint, promise });                     // record BEFORE awaiting, so concurrent duplicates find it
  const r = await promise;
  send(r.status, r);
});

// A flaky network: drops 15% of requests before they arrive; the client gives up after 30 ms.
const call = async (intent, key) => {
  if (Math.random() < 0.15) throw new Error("request lost");
  const headers = { "content-type": "application/json", ...(key && { "idempotency-key": key }) };
  const res = await fetch(url, { method: "POST", headers, body: JSON.stringify({ intent }), signal: AbortSignal.timeout(30) });
  if (res.status >= 500) throw new Error("server error");
  return res.status;
};

const jitter = (n, base = 5, cap = 100) => Math.random() * Math.min(cap, base * 2 ** n); // full jitter
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

const clients = {
  "never retry": async (i) => { try { await call(i); } catch {} },
  "retry, no key": async (i) => { for (let n = 0; n < 6; n++) { try { return await call(i); } catch { await sleep(jitter(n)); } } },
  "retry, same key": async (i) => { const key = `intent-${i}`; for (let n = 0; n < 6; n++) { try { return await call(i, key); } catch { await sleep(jitter(n)); } } },
};

await new Promise((r) => server.listen(0, "127.0.0.1", r));
const url = `http://127.0.0.1:${server.address().port}/`;

for (const [name, client] of Object.entries(clients)) {
  applied.clear(); seen.clear();
  let next = 0;                                                // a small worker pool, so that we measure the protocol and not connection storms
  await Promise.all(Array.from({ length: CONCURRENCY }, async () => { while (next < INTENTS) await client(`${name}-${next++}`); }));
  await sleep(200);                                            // let slow in-flight requests finish
  const counts = [...applied.values()];
  console.log(`${name.padEnd(16)} lost=${INTENTS - counts.length}  applied once=${counts.filter((c) => c === 1).length}  duplicated=${counts.filter((c) => c > 1).length}`);
}
server.close();

# The dedupe record and the side effect must commit together. Run: python3 atomic_dedupe.py
import sqlite3

db = sqlite3.connect(":memory:", isolation_level=None)   # we manage transactions ourselves
db.executescript("""
  CREATE TABLE balance (id INTEGER PRIMARY KEY, amount INTEGER);
  INSERT INTO balance VALUES (1, 0);
  CREATE TABLE idempotency (key TEXT PRIMARY KEY, response TEXT);
""")

def charge(key, amount, crash_before_commit=False):
    db.execute("BEGIN IMMEDIATE")
    row = db.execute("SELECT response FROM idempotency WHERE key = ?", (key,)).fetchone()
    if row:                                   # duplicate: replay the stored answer, do nothing else
        db.execute("COMMIT")
        return row[0]
    db.execute("UPDATE balance SET amount = amount + ? WHERE id = 1", (amount,))   # the side effect
    if crash_before_commit:
        db.execute("ROLLBACK")                # the process dies here: neither the effect nor the key survives
        raise RuntimeError("crashed before commit")
    db.execute("INSERT INTO idempotency VALUES (?, ?)", (key, f"charged {amount}"))
    db.execute("COMMIT")
    return f"charged {amount}"

def balance():
    return db.execute("SELECT amount FROM balance").fetchone()[0]

try:
    charge("k1", 100, crash_before_commit=True)
except RuntimeError as e:
    print("first attempt:", e, "| balance =", balance())
print("retry         :", charge("k1", 100), "| balance =", balance())
print("duplicate     :", charge("k1", 100), "| balance =", balance())

// 1000 clients fail at the same instant (say, the server restarts). When do their retries arrive?
// Deterministic: seeded PRNG. Run: node jitter.mjs
const rng = (a) => () => { a = (a + 0x6d2b79f5) | 0; let t = Math.imul(a ^ (a >>> 15), 1 | a); t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; return ((t ^ (t >>> 14)) >>> 0) / 2 ** 32; };
const rand = rng(42);
const BASE = 100, CAP = 10_000, CLIENTS = 1000, ATTEMPTS = 5; // ms

const strategies = {
  "no jitter":     (n) => Math.min(CAP, BASE * 2 ** n),
  "full jitter":   (n) => rand() * Math.min(CAP, BASE * 2 ** n),
  "equal jitter":  (n) => { const d = Math.min(CAP, BASE * 2 ** n); return d / 2 + rand() * (d / 2); },
  "decorrelated":  (n, prev) => Math.min(CAP, BASE + rand() * ((prev ?? BASE) * 3 - BASE)),
};

for (const [name, delay] of Object.entries(strategies)) {
  const buckets = new Map();                                // 10 ms bucket -> number of retry requests
  for (let c = 0; c < CLIENTS; c++) {
    let t = 0, prev;
    for (let n = 0; n < ATTEMPTS; n++) {
      prev = delay(n, prev); t += prev;                      // every retry fails again, so we measure pure scheduling
      const b = Math.floor(t / 10); buckets.set(b, (buckets.get(b) ?? 0) + 1);
    }
  }
  const peak = Math.max(...buckets.values());
  console.log(`${name.padEnd(13)} peak retries in one 10 ms bucket: ${String(peak).padStart(4)}   (total ${CLIENTS * ATTEMPTS})`);
}