C01 hands-on — Job dispatch and delivery semantics

At-most-once, at-least-once, and the transaction that makes duplicates harmless.

Source: handson/c01_job_scheduler.py --- run it with python3 handson/c01_job_scheduler.py
Full project spec: d01 — Fault-Tolerant Job Scheduler

This is the reported technical-screen question, and the part that separates candidates is not the scheduler --- it is what happens when a worker dies holding a job. There are exactly two places the acknowledgement can go, they fail in opposite directions, and there is no third position.

This page measures both failure directions on the same 20,000 jobs, then builds the fix (a dedup key at the sink), then breaks the fix two ways that real systems break it: a bounded dedup window, and a dedup key written outside the transaction. The last block shows the other source of duplicates, which is not crashes at all. Every number came from running the code.

Run it

cd swe-interview-prep/handson

python3 c01_job_scheduler.py            # every block, then the assembly
python3 c01_job_scheduler.py --block 3  # block 3 and its prerequisites only
python3 c01_job_scheduler.py --quiet    # the assembly only
python3 c01_job_scheduler.py --verify   # re-derive and assert every claim below

What to expect. A full run takes under a second and prints 6 blocks followed by the assembly. There are no dependencies beyond the Python standard library and nothing touches the network or the filesystem.

Every seed is fixed, so the numbers you get are the numbers on this page --- character for character. If yours differ, the code changed, not the machine. --verify re-derives 9 claims from scratch and exits non-zero if any of them stops holding, which is what makes the prose here checkable rather than assertable.

Predict before you read

Worth two minutes with a pen, because the gap between your answer and the measurement is the entire value of the page. Write down a number for each:

1. 20,000 jobs, 2% of workers crash mid-job. If the queue acknowledges before the work runs, how many jobs are lost --- and how many of those losses produce an error, a retry, or a metric that moves?
2. Move the acknowledgement to after the work. How many are duplicated? Is it the same set of jobs?
3. You add a dedup key at the sink, but write it in a separate statement from the effect. What fraction of the duplicates does that catch?
4. Redeliveries arrive 1--2,000 jobs later. Your dedup set holds the last 1,000 ids. What is the escape rate --- and what does the window have to exceed to reach zero?
5. Jobs mostly take ~2 s, but 5% take ~40 s. At a 5-second visibility timeout, how many of 20,000 jobs get executed twice?

Then run it, or read on --- the answers are in the blocks, and the ones most people get wrong are called out where they land.

Contents

• Run it
• Predict before you read
• Block 1 — Ack before work
• Block 2 — Ack after work
• Block 3 — Exactly-once does not exist
• Block 4 — The dedup table is not free
• Block 5 — Two systems, one crash
• Block 6 — The lease is the other duplicate source
• The assembly
• Verify the claims
• The design space
• Cost model
• Advanced
• How this connects to the rest of the program
• Failure modes at scale
• Primary sources
• What to do with this

How to read this page

Each block below is a self-contained lego piece: it builds one mechanism, proves it works on its own, and returns what the next block needs. The code is the real source, sliced out of the script. The output underneath it is the real output, captured by running that script --- not transcribed, not idealised. Where a measurement contradicted what I expected, the contradiction is in the output and the prose says so.

The assembly at the end wires every block into one working thing and measures it.

Block 1 — Ack before work

Teaches: at-most-once: nothing runs twice, and some never run

The problem. A worker takes a job and the queue has to decide when to forget it. Forgetting immediately — acknowledge, then work — makes duplicates impossible, which sounds like the safe choice until you count what it costs.

@block(1, "Ack before work", "at-most-once: nothing runs twice, and some never run")
def b1(s, show):
    def run(jobs):
        sink = Sink()
        for jid, crashes in jobs:
            # ack first: the queue forgets the job immediately
            if crashes:
                continue                  # worker dies before applying -> job lost
            sink.apply(jid)
        return sink
    if show:
        sink = run(stream())
        lost, once, dup = sink.stats(N)
        print(f"  {N:,} jobs, {CRASH*100:.0f}% of workers crash mid-job")
        print(f"  {'lost':>8}{'exactly once':>15}{'duplicated':>13}")
        print(f"  {lost:>8}{once:>15}{dup:>13}")
        print(f"  {lost/N*100:.2f}% of jobs never ran and nothing recorded that they")
        print("  did not. The queue deleted them on ack, so there is no evidence")
        print("  anywhere -- no retry, no dead letter, no metric that moves.")
        print("  At-most-once is the right choice for exactly one thing: work")
        print("  where a duplicate is worse than a miss AND the miss is detectable")
        print("  by some other means. That is a short list.")
    return {"stream": stream, "Sink": Sink}

Reading the implementation

• if crashes: continue before sink.apply(jid) — the entire semantics in two lines. The ack has already happened conceptually; the crash removes the effect and nothing brings it back.
• Sink.applied is a count per job, not a set. Counting rather than recording presence is what lets the same class measure both failure directions — losses in this block and duplicates in the next — with no change.
• The crash flag is drawn once, in stream(), and every design on this page replays the identical stream. Without that, comparing designs would be comparing random draws.

What the numbers say

Output:

  20,000 jobs, 2% of workers crash mid-job
      lost   exactly once   duplicated
       398          19602            0
  1.99% of jobs never ran and nothing recorded that they
  did not. The queue deleted them on ack, so there is no evidence
  anywhere -- no retry, no dead letter, no metric that moves.
  At-most-once is the right choice for exactly one thing: work
  where a duplicate is worse than a miss AND the miss is detectable
  by some other means. That is a short list.

398 jobs — 1.99% — never ran, and nothing in the system knows. That is the property that makes at-most-once dangerous rather than merely lossy: there is no retry, no dead-letter queue, no metric that moves. The queue is empty, every worker is healthy, and the dashboard is green.

Try it yourself

from c01_job_scheduler import parts, stream, Sink
print("  this page exports:", ", ".join(sorted(parts())))
print()

# Sweep the crash rate. Loss is linear in it and invisible at every point.
for rate in (0.001, 0.01, 0.05, 0.20):
    jobs = [(i, (i * 7919) % 100000 < rate * 100000) for i in range(20_000)]
    sink = Sink()
    for jid, crashed in jobs:
        if not crashed:
            sink.apply(jid)            # ack came first; a crash loses the job
    lost, once, dup = sink.stats(20_000)
    print(f"  crash rate {rate*100:>5.1f}% -> {lost:>5} lost, {dup} duplicated, "
          f"and {0} errors raised")

  this page exports: DedupSink, Sink, stream

  crash rate   0.1% ->    23 lost, 0 duplicated, and 0 errors raised
  crash rate   1.0% ->   204 lost, 0 duplicated, and 0 errors raised
  crash rate   5.0% ->  1001 lost, 0 duplicated, and 0 errors raised
  crash rate  20.0% ->  4003 lost, 0 duplicated, and 0 errors raised

The last column is the finding. At a 20% crash rate this design silently drops one job in five and raises nothing — no exception, no dead letter, no metric. Every other row on this page trades that for a failure you can count.

Beyond the toy

At-most-once is correct for exactly one situation: a duplicate is worse than a miss and the miss is detectable some other way. Metrics samples and cache warming qualify — losing one is invisible and the next sample corrects it. Almost nothing else does.

The reason people choose it accidentally is that it is the default in several places, and the default is not obvious:

• autocommit in a Kafka consumer commits offsets on a timer, which can advance past records you have not processed. enable.auto.commit=true is at-most-once wearing a config flag.
• HTTP fire-and-forget — any POST whose response you do not check.
• UDP anything, obviously, but also an in-process queue with no persistence: a queue.Queue loses everything on restart, which is at-most-once with extra steps.

The question that surfaces it in a design review: if this process is SIGKILLed right now, what is the smallest unit of work that disappears?

Block 2 — Ack after work

Teaches: at-least-once: nothing is lost, and some run twice

The problem. Move the acknowledgement to the other side of the work and nothing is ever lost. The failure just moves too, and it moves somewhere that is much easier to live with — which is the entire reason every durable queue is built this way.

@block(2, "Ack after work", "at-least-once: nothing is lost, and some run twice")
def b2(s, show):
    def run(jobs):
        sink = Sink()
        for jid, crashes in jobs:
            sink.apply(jid)               # do the work first
            if crashes:
                sink.apply(jid)           # crash BEFORE the ack -> redelivered, redone
        return sink
    if show:
        sink = run(stream())
        lost, once, dup = sink.stats(N)
        print(f"  Same {N:,} jobs, same crashes, ack moved after the side effect.")
        print(f"  {'lost':>8}{'exactly once':>15}{'duplicated':>13}")
        print(f"  {lost:>8}{once:>15}{dup:>13}")
        print(f"  Zero lost. {dup} jobs applied twice ({dup/N*100:.2f}%), because the")
        print("  crash landed between the effect and the acknowledgement -- a window")
        print("  that cannot be closed by moving the ack, only by moving it to the")
        print("  other side and losing jobs instead.")
        print("  This is the trade in one line: the ack can be before the work or")
        print("  after it, and there is no third position. Everything else on this")
        print("  page is about making the duplicate HARMLESS rather than absent.")
    return {}

Reading the implementation

• sink.apply(jid) then if crashes: sink.apply(jid) — the second call is the redelivery, not a bug in the worker. The queue never saw an ack, so after the visibility timeout it hands the job to someone else, who does it again.
• Nothing about the worker changed between this block and the last one. Only the position of the ack changed, and that single choice is the whole taxonomy.

What the numbers say

Output:

  Same 20,000 jobs, same crashes, ack moved after the side effect.
      lost   exactly once   duplicated
         0          19602          398
  Zero lost. 398 jobs applied twice (1.99%), because the
  crash landed between the effect and the acknowledgement -- a window
  that cannot be closed by moving the ack, only by moving it to the
  other side and losing jobs instead.
  This is the trade in one line: the ack can be before the work or
  after it, and there is no third position. Everything else on this
  page is about making the duplicate HARMLESS rather than absent.

Zero lost, 398 duplicated — exactly the jobs that were lost in block 1, which is the point: it is the same crash window, and moving the ack decides which side of it you pay on.

There is no third position for the ack. It is before the effect or after it. Everything else on this page is about making the duplicate harmless, because making it absent is not available.

Try it yourself

The two designs fail on the same jobs. Show it rather than assert it:

from c01_job_scheduler import stream, Sink

jobs = stream()
crashed_ids = {jid for jid, c in jobs if c}

amo, alo = Sink(), Sink()
for jid, crashed in jobs:
    if not crashed: amo.apply(jid)                 # ack first
    alo.apply(jid)                                 # ack last
    if crashed: alo.apply(jid)

missing = crashed_ids - set(amo.applied)
doubled = {j for j, v in alo.applied.items() if v > 1}
print(f"  at-most-once  lost     {len(missing):>4} jobs")
print(f"  at-least-once duplicated {len(doubled):>3} jobs")
print(f"  are they the same set of job ids? {missing == doubled}")
print(f"  is either set empty?              {not missing or not doubled}")

  at-most-once  lost      398 jobs
  at-least-once duplicated 398 jobs
  are they the same set of job ids? True
  is either set empty?              False

The same crash window, billed to one side or the other. There is no third position for the acknowledgement, so there is no design that empties both sets — which is why every remaining block is about making the duplicate harmless rather than making it absent.

Beyond the toy

Real queues expose this as a visibility timeout or a lease, and the redelivery is a first-class documented behaviour rather than an accident:

• SQS — standard queues are explicitly at-least-once, and the docs say so. FIFO queues offer a 5-minute deduplication window, which is block 4's bounded dedup provided as a service.
• Kafka — at-least-once by default; enable.idempotence plus transactions gives exactly-once within Kafka, which does not extend to your side effects in another system.
• Celery / Sidekiq — at-least-once, and both documentation sets tell you to make tasks idempotent. Almost nobody does until the first incident.

The framing worth carrying into an interview: at-least-once is not a weaker guarantee than exactly-once, it is the only one available at the transport layer. The strength has to be added at the sink, which is block 3.

Block 3 — Exactly-once does not exist

Teaches: but effectively-once does, and it is dedup at the sink

The problem. Blocks 1 and 2 are a genuine dilemma if you insist on solving it in the delivery layer. It stops being a dilemma the moment you notice that nobody actually cares how many times the message arrived — they care how many times the effect happened.

@block(3, "Exactly-once does not exist", "but effectively-once does, and it is dedup at the sink")
def b3(s, show):
    class DedupSink(Sink):
        def __init__(self): super().__init__(); self.seen = set()
        def apply(self, job_id):
            if job_id in self.seen: return False       # already applied: no-op
            self.seen.add(job_id); super().apply(job_id); return True
    def run(jobs):
        sink = DedupSink()
        for jid, crashes in jobs:
            sink.apply(jid)
            if crashes: sink.apply(jid)               # the redelivery
        return sink
    if show:
        sink = run(stream())
        lost, once, dup = sink.stats(N)
        print(f"  At-least-once delivery + a dedup key checked AT THE SINK.")
        print(f"  {'lost':>8}{'exactly once':>15}{'duplicated':>13}")
        print(f"  {lost:>8}{once:>15}{dup:>13}")
        print("  Zero and zero. Note what did NOT change: the message is still")
        print("  delivered twice. Delivery is still at-least-once, because that is")
        print("  the only thing a network can offer. What changed is that the")
        print("  second APPLICATION is a no-op, so the observable outcome is")
        print("  exactly-once.")
        print("  The phrase to use is 'at-least-once delivery with idempotent")
        print("  processing'. Saying 'exactly-once delivery' unqualified is the")
        print("  tell that you have not thought about where the dedup lives.")
    return {"DedupSink": DedupSink}

Reading the implementation

• if job_id in self.seen: return False — the check is at the sink, in the same component that performs the effect. Not in the producer, not in the broker, not in the worker's dispatch loop. That placement is the whole design and block 5 is about what happens when it slips.
• The dedup key is the job id, which is stable across redeliveries. A key derived from anything that changes per attempt — a delivery timestamp, a retry counter, a uuid4() generated in the worker — silently degrades this back to block 2.
• return False rather than raising: a duplicate is a normal, expected event and must not look like an error. A dedup that raises produces alert fatigue and then gets suppressed.

What the numbers say

Output:

  At-least-once delivery + a dedup key checked AT THE SINK.
      lost   exactly once   duplicated
         0          20000            0
  Zero and zero. Note what did NOT change: the message is still
  delivered twice. Delivery is still at-least-once, because that is
  the only thing a network can offer. What changed is that the
  second APPLICATION is a no-op, so the observable outcome is
  exactly-once.
  The phrase to use is 'at-least-once delivery with idempotent
  processing'. Saying 'exactly-once delivery' unqualified is the
  tell that you have not thought about where the dedup lives.

Zero lost and zero duplicated, from a delivery layer that still delivers 398 messages twice. Delivery is unchanged; the second application is a no-op.

Try it yourself

The dedup key has to be stable across attempts. Watch what happens when it is not — which is the most common way this is implemented wrong:

from c01_job_scheduler import stream, Sink
import uuid

jobs = stream()

def run(key_fn):
    sink, seen = Sink(), set()
    for jid, crashed in jobs:
        for attempt in range(2 if crashed else 1):
            k = key_fn(jid, attempt)
            if k in seen: continue
            seen.add(k); sink.apply(jid)
    return sink.stats(20_000)

for label, fn in (
        ("job id (stable)",            lambda j, a: j),
        ("job id + attempt number",    lambda j, a: (j, a)),
        ("a fresh uuid4 per attempt",  lambda j, a: uuid.uuid4()),
):
    lost, once, dup = run(fn)
    print(f"  key = {label:<26} -> {dup:>4} duplicates")

  key = job id (stable)            ->    0 duplicates
  key = job id + attempt number    ->  398 duplicates
  key = a fresh uuid4 per attempt  ->  398 duplicates

All three "have idempotency". Only the first one is idempotent. A key derived from anything that varies per attempt — a retry counter, a delivery timestamp, a uuid minted in the worker — passes code review, passes every test that does not actually redeliver, and does nothing.

Beyond the toy

The vocabulary matters more here than usual, because the wrong phrase is a tell:

Four ways to build the idempotency, roughly in order of how often they are the right answer:

• A natural key already in the data. INSERT ... ON CONFLICT (order_id) DO NOTHING. Free, and needs no extra table.
• A dedup table keyed on the job id, written in the same transaction as the effect (block 5).
• Make the operation itself idempotent. SET status='shipped' rather than INCREMENT ship_count. Often just a schema choice, made early, for free.
• The downstream's idempotency key. Stripe, and most payment APIs, accept an Idempotency-Key header precisely because their callers cannot fence them.

The last one is what you use when the effect is in a system you do not control — which is the case block 5 says you cannot solve any other way.

Block 4 — The dedup table is not free

Teaches: bounded memory means duplicates escape, and you can price it

The problem. Block 3's dedup set grows forever. Bounding it is obviously necessary and obviously reintroduces duplicates; the interesting question is what the bound has to be measured against, and the answer is not what most people assume.

@block(4, "The dedup table is not free", "bounded memory means duplicates escape, and you can price it")
def b4(s, show):
    class WindowedSink(Sink):
        """Dedup with a bounded LRU of recently-seen ids."""
        def __init__(self, window):
            super().__init__(); self.window, self.seen, self.order = window, set(), []
        def apply(self, job_id):
            if job_id in self.seen: return False
            self.seen.add(job_id); self.order.append(job_id)
            if len(self.order) > self.window:
                self.seen.discard(self.order.pop(0))
            super().apply(job_id); return True

    def run(window, delay_seed=17):
        """Redelivery happens `delay` jobs later, not immediately."""
        rng = random.Random(delay_seed)
        sink, queue = WindowedSink(window), []
        for jid, crashes in stream():
            sink.apply(jid)
            if crashes:
                # redelivery is queued behind however much traffic arrived meanwhile
                queue.append((jid, rng.randint(1, 2000)))
            queue = [(j, d - 1) for j, d in queue]
            for j, d in [q for q in queue if q[1] <= 0]:
                sink.apply(j)
            queue = [q for q in queue if q[1] > 0]
        for j, _ in queue: sink.apply(j)
        return sink

    if show:
        print("  Redelivery arrives 1-2000 jobs after the original, not instantly.")
        print("  The dedup set is bounded, so an id can be evicted before its")
        print("  duplicate arrives.")
        print(f"  {'window':>10}{'state':>12}{'duplicates escaped':>21}{'rate':>9}")
        for w in (100, 500, 1_000, 1_500, 2_000, 5_000):
            sink = run(w)
            _, _, dup = sink.stats(N)
            print(f"  {w:>10,}{w*16//1024:>10} KB{dup:>21}{dup/N*100:>8.2f}%")
        print("  The escape rate falls to zero exactly when the window reaches the")
        print("  MAXIMUM redelivery delay (2,000), not the mean and not the rate.")
        print("  A window half that size still leaks 1.5%: a duplicate arriving")
        print("  1,600 jobs later finds its key already evicted and applies again.")
        print("  So the dedup window is not a memory-budget decision -- it is set")
        print("  by the QUEUE's retention or visibility timeout, a property of a")
        print("  system you may not own. Size it from that number, and if that")
        print("  number is unbounded (a DLQ replayed by hand next week), a bounded")
        print("  in-memory dedup cannot be correct and the key belongs in storage.")
    return {}

Reading the implementation

• The redelivery is queued with a delay of 1–2000 jobs, not applied immediately. That is the realism that makes this block measure anything: an instantly-redelivered duplicate is caught by any window at all, and real redeliveries arrive after a visibility timeout during which the queue kept moving.
• The order list plus seen set is a hand-rolled LRU. In production this is a Redis SETEX per key, or a TTL index, and the "window" is expressed in time rather than in count — which is the more natural unit for the same reason.

What the numbers say

Output:

  Redelivery arrives 1-2000 jobs after the original, not instantly.
  The dedup set is bounded, so an id can be evicted before its
  duplicate arrives.
      window       state   duplicates escaped     rate
         100         1 KB                  380    1.90%
         500         7 KB                  300    1.50%
       1,000        15 KB                  195    0.97%
       1,500        23 KB                  104    0.52%
       2,000        31 KB                    0    0.00%
       5,000        78 KB                    0    0.00%
  The escape rate falls to zero exactly when the window reaches the
  MAXIMUM redelivery delay (2,000), not the mean and not the rate.
  A window half that size still leaks 1.5%: a duplicate arriving
  1,600 jobs later finds its key already evicted and applies again.
  So the dedup window is not a memory-budget decision -- it is set
  by the QUEUE's retention or visibility timeout, a property of a
  system you may not own. Size it from that number, and if that
  number is unbounded (a DLQ replayed by hand next week), a bounded
  in-memory dedup cannot be correct and the key belongs in storage.

The escape rate falls to zero exactly when the window reaches 2,000 — the maximum redelivery delay. Not the mean delay, not the arrival rate, not a round number that felt safe. A window at half that still leaks 1.5%.

So the dedup window is not a memory-budget decision. It is determined by the queue's retention or visibility timeout, which is a property of a system you may not own, and sizing it from your own memory budget is how the bug gets shipped.

Try it yourself

Find the window your queue actually requires, rather than the one that fits your memory budget:

from c01_job_scheduler import stream
import random

def escapes(window, max_delay, seed=17, n=20_000):
    rng, seen, order, applied, q = random.Random(seed), set(), [], {}, []
    def apply(j):
        if j in seen: return
        seen.add(j); order.append(j)
        if len(order) > window: seen.discard(order.pop(0))
        applied[j] = applied.get(j, 0) + 1
    for jid, crashed in stream(n):
        apply(jid)
        if crashed: q.append([jid, rng.randint(1, max_delay)])
        for e in q: e[1] -= 1
        for j, d in [e for e in q if e[1] <= 0]: apply(j)
        q = [e for e in q if e[1] > 0]
    for j, _ in q: apply(j)
    return sum(v - 1 for v in applied.values() if v > 1)

print(f"  {'max redelivery delay':>21}{'window 500':>12}{'window 2k':>11}"
      f"{'window 10k':>12}")
for md in (200, 1_000, 5_000, 20_000):
    row = "".join(f"{escapes(w, md):>11}" for w in (500, 2_000, 10_000))
    print(f"  {md:>20,}{row}")

   max redelivery delay  window 500  window 2k  window 10k
                   200          0          0          0
                 1,000        198          0          0
                 5,000        357        231          0
                20,000        383        333        110

Read it as a rule rather than a table: zero appears exactly where the window reaches the maximum delay, on every row. The window is not a memory decision, it is a restatement of the queue's retention — a property of a system you may not own, and one that is unbounded the moment a human can replay a dead-letter queue by hand.

Beyond the toy

The awkward case, and the one worth raising unprompted: if redelivery is unbounded, a bounded dedup cannot be correct. A message parked in a dead-letter queue and replayed by hand next Tuesday will arrive long after any in-memory window has rotated.

When that is possible — and it usually is — the key belongs in durable storage with a retention at least as long as the maximum possible replay interval, which is a business decision rather than a technical one.

Two ways to make the storage affordable at scale:

• A Bloom filter in front of the durable table. A false positive means "probably seen", which would skip a job — the unsafe direction. So it must be used the other way: the filter answers "definitely not seen" and skips the lookup, and a probable-hit falls through to the exact check. Same structure as C03's and m02's two-stage tests.
• Partition the dedup table by time and drop whole partitions, so expiry is a DROP TABLE rather than a scan of billions of rows.

Block 5 — Two systems, one crash

Teaches: the dual write, and why dedup state must be transactional

The problem. Block 3's dedup works because the check and the effect happen together. In every real system they are two writes — often to two different systems — and a crash between them puts the effect in place with no record that it happened. This is the dual-write problem, and it is the most common way a correct-looking idempotency implementation fails.

@block(5, "Two systems, one crash", "the dual write, and why dedup state must be transactional")
def b5(s, show):
    def run(transactional):
        """Apply the effect and record the dedup key. Crash may land between."""
        rng = random.Random(29)
        applied, dedup = {}, set()
        dup = 0
        for jid, crashes in stream():
            if jid in dedup:
                dup += 1; continue
            if transactional:
                # one atomic commit: effect and key land together or not at all
                if not crashes:
                    applied[jid] = applied.get(jid, 0) + 1; dedup.add(jid)
                else:
                    pass                       # neither happened; safe to retry
            else:
                applied[jid] = applied.get(jid, 0) + 1     # effect lands
                if crashes:
                    continue                   # crash BEFORE writing the dedup key
                dedup.add(jid)
        # redelivery of everything that crashed
        for jid, crashes in stream():
            if not crashes: continue
            if jid in dedup: dup += 1; continue
            applied[jid] = applied.get(jid, 0) + 1
        extra = sum(v - 1 for v in applied.values() if v > 1)
        return extra, len(applied)

    if show:
        print("  The dedup key and the side effect are two writes. A crash between")
        print("  them leaves the effect applied and the key missing -- so the")
        print("  redelivery is not recognised as a duplicate.")
        print(f"  {'design':>34}{'applied twice':>15}{'rate':>9}")
        for name, tx in (("effect, then dedup key (2 writes)", False),
                         ("both in one transaction", True)):
            extra, n = run(tx)
            print(f"  {name:>34}{extra:>15}{extra/N*100:>8.2f}%")
        print("  Dedup only works if the key is written ATOMICALLY with the effect.")
        print("  If the effect is in Postgres, the key goes in the same Postgres")
        print("  transaction. If the effect is a third-party API call, you cannot")
        print("  do this at all -- and that is the honest answer: use THEIR")
        print("  idempotency key, or accept at-least-once and say so.")
        print("  This is the dual-write problem, and the outbox pattern is the")
        print("  standard escape: write the effect and an outbox row in one")
        print("  transaction, then publish from the outbox separately.")
    return {}

Reading the implementation

• The non-transactional path applies the effect, then if crashes: continue before dedup.add(jid). The order is what a straightforward implementation does — do the work, then record that you did it — and the window between them is unavoidable without a transaction.
• The transactional path makes both happen or neither, so a crash leaves the job cleanly retryable. There is no window because there is no intermediate state to crash in.

What the numbers say

Output:

  The dedup key and the side effect are two writes. A crash between
  them leaves the effect applied and the key missing -- so the
  redelivery is not recognised as a duplicate.
                              design  applied twice     rate
   effect, then dedup key (2 writes)            398    1.99%
             both in one transaction              0    0.00%
  Dedup only works if the key is written ATOMICALLY with the effect.
  If the effect is in Postgres, the key goes in the same Postgres
  transaction. If the effect is a third-party API call, you cannot
  do this at all -- and that is the honest answer: use THEIR
  idempotency key, or accept at-least-once and say so.
  This is the dual-write problem, and the outbox pattern is the
  standard escape: write the effect and an outbox row in one
  transaction, then publish from the outbox separately.

1.99% applied twice — exactly the original crash rate. The dedup layer removed none of the duplicates in the crash case, which is the only case it existed for. Its usefulness in a benchmark without crashes is precisely zero information.

That is what makes this failure expensive: the implementation is correct in every test that does not kill the process at the wrong microsecond, so it ships, and the duplicate rate in production matches the crash rate exactly.

Try it yourself

The failure is entirely about ordering. Move the dedup write across the crash and watch it appear and vanish:

from c01_job_scheduler import stream

def run(order):
    jobs, applied, dedup = stream(), {}, set()
    for jid, crashed in jobs:
        if jid in dedup: continue
        if order == "key-then-effect":
            dedup.add(jid)
            if crashed: continue              # crash after key, before effect
            applied[jid] = applied.get(jid, 0) + 1
        elif order == "effect-then-key":
            applied[jid] = applied.get(jid, 0) + 1
            if crashed: continue              # crash after effect, before key
            dedup.add(jid)
        else:                                  # one transaction
            if not crashed:
                applied[jid] = applied.get(jid, 0) + 1; dedup.add(jid)
    for jid, crashed in jobs:                  # the redeliveries
        if crashed and jid not in dedup:
            applied[jid] = applied.get(jid, 0) + 1
    dup = sum(v - 1 for v in applied.values() if v > 1)
    lost = 20_000 - len(applied)
    return lost, dup

for order in ("key-then-effect", "effect-then-key", "one transaction"):
    lost, dup = run(order)
    print(f"  {order:<18} -> {lost:>4} lost, {dup:>4} duplicated")

  key-then-effect    ->  398 lost,    0 duplicated
  effect-then-key    ->    0 lost,  398 duplicated
  one transaction    ->    0 lost,    0 duplicated

Three orderings, three different bugs. Writing the key first loses work — the job is marked done and never ran, which is at-most-once with extra steps. Writing the effect first duplicates. Only the atomic version is both, and there is no sequence of two separate writes that achieves it, which is the whole content of the dual-write problem.

Beyond the toy

When both the effect and the key live in one transactional store, this is trivial — same transaction, done. The design problem is when they do not:

• Effect in Postgres, key in Redis — two systems, no shared transaction. Move the key into Postgres; the cost is one row, and Redis was an optimisation you did not need.
• Effect is an outbound message — the classic case, and the classic answer is the outbox pattern: write the effect and an outbox row in one transaction, then a separate relay publishes from the outbox at-least-once. The relay's own duplicates are handled by the consumer's dedup, which is this block again one level down.
• Effect is a third-party API call — you cannot share a transaction with Stripe. The honest options are their idempotency key, or accepting at-least-once and saying so. There is no clever local solution, and claiming one is the "guarantee overclaimed" failure that d04's critique names.

The general rule: two writes that must agree cannot be in two systems. Either they share a transaction, or one of them becomes a derived consequence of the other rather than a peer.

Block 6 — The lease is the other duplicate source

Teaches: slow work looks exactly like a dead worker

The problem. Every duplicate so far came from a crash. There is a second source that has nothing to do with crashes and is usually larger: a worker that is merely slow. From the queue's side, a job that takes longer than the visibility timeout is indistinguishable from a worker that died, so the queue does the correct thing and gives it to somebody else.

@block(6, "The lease is the other duplicate source", "slow work looks exactly like a dead worker")
def b6(s, show):
    def run(lease, renew, n=20_000, seed=31):
        """Work longer than the lease -> the queue redelivers to a second worker."""
        rng = random.Random(seed)
        doubles = 0
        for _ in range(n):
            work = (rng.expovariate(1 / 2.0) if rng.random() > 0.05
                    else rng.expovariate(1 / 40.0))       # 5% very slow jobs
            if renew:
                # heartbeat every lease/3; survives as long as the worker is alive
                continue
            if work > lease:
                doubles += 1
        return doubles

    if show:
        print("  Job durations: mostly ~2s, 5% much slower (~40s). The queue")
        print("  redelivers when the visibility timeout expires.")
        print(f"  {'visibility timeout':>20}{'no renewal':>13}{'with renewal':>15}")
        for lease in (5, 15, 30, 60, 300):
            print(f"  {lease:>19}s{run(lease, False):>13}{run(lease, True):>15}")
        print("  Without renewal the timeout must exceed the SLOWEST job or the")
        print("  slow ones are all executed twice -- and a timeout sized for the")
        print("  slowest job makes every genuine crash cost that long to detect.")
        print("  That is c11's lease-sizing tradeoff, exactly.")
        print("  With a heartbeat the timeout only has to exceed the RENEWAL")
        print("  interval, so it can be seconds while jobs run for minutes.")
        print("  Renewal is the mechanism; idempotency is still required, because")
        print("  a worker partitioned from the queue keeps working while its lease")
        print("  expires -- which is exactly c11's zombie.")
    return {}

Reading the implementation

• The duration distribution is bimodal — mostly ~2 s, with 5% around 40 s. A single exponential would put almost no mass beyond the timeout and would make every timeout look safe. Real job durations have a slow tail (a large customer, a cold cache, a retry inside the job), and the tail is what the timeout fights.
• if renew: continue — with a heartbeat, duration stops mattering entirely, and the code says so by not consulting work at all.

What the numbers say

Output:

  Job durations: mostly ~2s, 5% much slower (~40s). The queue
  redelivers when the visibility timeout expires.
    visibility timeout   no renewal   with renewal
                    5s         2448              0
                   15s          715              0
                   30s          499              0
                   60s          233              0
                  300s            0              0
  Without renewal the timeout must exceed the SLOWEST job or the
  slow ones are all executed twice -- and a timeout sized for the
  slowest job makes every genuine crash cost that long to detect.
  That is c11's lease-sizing tradeoff, exactly.
  With a heartbeat the timeout only has to exceed the RENEWAL
  interval, so it can be seconds while jobs run for minutes.
  Renewal is the mechanism; idempotency is still required, because
  a worker partitioned from the queue keeps working while its lease
  expires -- which is exactly c11's zombie.

Without renewal, a 5-second timeout double-executes 2,448 of 20,000 jobs (12%), and you need a 300-second timeout to reach zero — at which point every genuine crash blocks that job for five minutes.

With renewal, every row is zero, because the timeout only has to outlive the heartbeat interval rather than the job.

Try it yourself

Renewal decouples the timeout from the work. Show the decoupling directly:

import random

def doubles(timeout, renew_every=None, n=20_000, seed=31):
    rng, d = random.Random(seed), 0
    for _ in range(n):
        work = (rng.expovariate(1 / 2.0) if rng.random() > 0.05
                else rng.expovariate(1 / 40.0))
        if renew_every is None:
            if work > timeout: d += 1                     # must outlive the JOB
        else:
            # A heartbeat every `renew_every`; the lease only has to outlive that.
            if renew_every > timeout: d += 1              # renewal too slow
    return d

print(f"  {'timeout':>9}{'no renewal':>13}{'renew @ t/3':>14}{'renew @ 2t':>12}")
for t in (5, 15, 30, 60, 300):
    print(f"  {t:>8}s{doubles(t):>13,}{doubles(t, t/3):>14,}{doubles(t, 2*t):>12,}")

    timeout   no renewal   renew @ t/3  renew @ 2t
         5s        2,448             0      20,000
        15s          715             0      20,000
        30s          499             0      20,000
        60s          233             0      20,000
       300s            0             0      20,000

The middle column is flat at zero regardless of timeout, because with a heartbeat the constraint is renewal_interval < timeout rather than job_duration < timeout. The right-hand column shows what happens when the renewal is slower than the lease: the mechanism inverts and every job doubles. Renew at timeout/3 so two consecutive renewal failures are survivable — that is where the number comes from.

Beyond the toy

This is C11's lease-sizing tradeoff exactly, in a different costume, and noticing that is worth saying out loud: a visibility timeout is a lease, a heartbeat is lease renewal, and a slow worker is the zombie.

Which means C11's conclusion transfers whole: renewal is the mechanism, idempotency is still required. A worker partitioned from the queue cannot renew but also cannot tell it has been superseded, so it keeps working and eventually writes. The heartbeat reduces the frequency of the duplicate; only the dedup key makes it harmless.

Two production details worth naming:

• Renew at timeout/3, so two consecutive renewal failures are survivable.
• A job that cannot renew should stop voluntarily rather than finish and hope, which turns a probable duplicate into a clean abort — the one client-side mitigation that is not a TOCTOU trap, for the same reason as in C11.

The assembly

Every block above, wired together into one working system:

def assembly(s):
    print("\nFour designs, the same 20,000 jobs and the same crashes.\n")
    jobs = stream()

    def measure(kind):
        sink = Sink(); seen = set()
        for jid, crashes in jobs:
            if kind == "at-most-once":
                if not crashes: sink.apply(jid)
                continue
            if kind == "at-least-once":
                sink.apply(jid)
                if crashes: sink.apply(jid)
                continue
            # dedup variants
            def apply(j):
                if j in seen: return
                seen.add(j); sink.apply(j)
            apply(jid)
            if crashes:
                if kind == "dedup, non-transactional":
                    seen.discard(jid)          # key lost in the crash
                apply(jid)
        lost, once, dup = sink.stats(N)
        return lost, once, dup

    print(f"  {'design':<28}{'lost':>7}{'exactly once':>14}{'duplicated':>12}"
          f"{'correct':>9}")
    for kind in ("at-most-once", "at-least-once", "dedup, non-transactional",
                 "dedup, transactional"):
        lost, once, dup = measure(kind)
        ok = "yes" if lost == 0 and dup == 0 else "no"
        print(f"  {kind:<28}{lost:>7}{once:>14}{dup:>12}{ok:>9}")

    print("\n  Only the last row is both. And it is not 'exactly-once delivery' --")
    print("  the message is still delivered twice in every crash case. What the")
    print("  last row has is a dedup key committed in the SAME TRANSACTION as the")
    print("  effect, so a redelivery finds the key and does nothing.")
    print("\n  The four sentences this page exists to earn:")
    print("  1. The ack goes before the work or after it. Before loses jobs,")
    print("     after duplicates them, and there is no third position.")
    print("  2. Exactly-once delivery does not exist. At-least-once delivery plus")
    print("     idempotent processing is observably equivalent and achievable.")
    print("  3. The dedup key must be written atomically with the effect, or the")
    print("     crash window just moved.")
    print("  4. Duplicates also come from the LEASE, not only from crashes, and a")
    print("     heartbeat is what decouples timeout length from job length.")
    print("\n  Built: at-most-once -> at-least-once -> dedup -> bounded dedup ->")
    print("  the dual write -> lease renewal.")
    print("  Not built, worth ten more minutes: dead-letter queues and the poison")
    print("  message, ordering guarantees per key, and fencing the dispatch itself")
    print("  so two schedulers cannot both enqueue (that is c11).")


def parts():
    """Every mechanism this page builds, ready to import.

        >>> from c01_job_scheduler import parts
        >>> p = parts()
        >>> sorted(p)                      # doctest: +ELLIPSIS
        [...]
    """
    return collect()


def verify():
    """Re-derive every headline claim on this page from scratch."""
    jobs = stream()
    crashes = sum(1 for _, c in jobs if c)

    def measure(kind, window=None):
        applied, seen, order = {}, set(), []
        def apply(j):
            applied[j] = applied.get(j, 0) + 1
        for jid, crashed in jobs:
            if kind == "at-most-once":
                if not crashed: apply(jid)
            elif kind == "at-least-once":
                apply(jid)
                if crashed: apply(jid)
            elif kind == "dedup":
                for _ in range(2 if crashed else 1):
                    if jid in seen: continue
                    seen.add(jid); apply(jid)
            elif kind == "dedup-nontx":
                if jid not in seen: seen.add(jid); apply(jid)
                if crashed:
                    seen.discard(jid)              # key lost in the crash
                    if jid not in seen: seen.add(jid); apply(jid)
        lost = N - len(applied)
        dup = sum(v - 1 for v in applied.values() if v > 1)
        return lost, dup

    lost, dup = measure("at-most-once")
    check("B1  ack-before-work loses exactly the crashed jobs, silently",
          lost == crashes and dup == 0,
          f"{lost} lost ({lost/N*100:.2f}%), 0 duplicated")

    lost, dup = measure("at-least-once")
    check("B2  ack-after-work loses nothing and duplicates the same jobs",
          lost == 0 and dup == crashes,
          f"0 lost, {dup} duplicated -- the identical crash window")

    lost, dup = measure("dedup")
    check("B3  a dedup key at the sink makes the redelivery a no-op",
          lost == 0 and dup == 0, "0 lost, 0 duplicated")

    lost, dup = measure("dedup-nontx")
    check("B5  a dedup key written OUTSIDE the transaction catches none of them",
          dup == crashes,
          f"{dup} duplicated -- exactly the crash rate, so the dedup did nothing")

    # B4 -- the window must reach the maximum redelivery DELAY, not the mean.
    def windowed(window, max_delay=2000, seed=17):
        rng = random.Random(seed)
        seen, order, applied, q = set(), [], {}, []
        def apply(j):
            if j in seen: return
            seen.add(j); order.append(j)
            if len(order) > window: seen.discard(order.pop(0))
            applied[j] = applied.get(j, 0) + 1
        for jid, crashed in jobs:
            apply(jid)
            if crashed: q.append([jid, rng.randint(1, max_delay)])
            for e in q: e[1] -= 1
            for j, d in [e for e in q if e[1] <= 0]: apply(j)
            q = [e for e in q if e[1] > 0]
        for j, _ in q: apply(j)
        return sum(v - 1 for v in applied.values() if v > 1)
    small, exact = windowed(1000), windowed(2000)
    check("B4  a window below the max redelivery delay leaks duplicates",
          small > 0, f"{small} escaped at window=1000, max delay=2000")
    check("B4  ...and a window at the max delay leaks exactly zero",
          exact == 0, "0 escaped at window=2000")

    # B6 -- without renewal the timeout must exceed the SLOWEST job.
    def doubles(lease, n=20_000, seed=31):
        rng = random.Random(seed); d = 0
        for _ in range(n):
            work = (rng.expovariate(1 / 2.0) if rng.random() > 0.05
                    else rng.expovariate(1 / 40.0))
            if work > lease: d += 1
        return d
    check("B6  a short visibility timeout double-executes slow jobs",
          doubles(5) > 1000, f"{doubles(5)} of 20,000 at a 5 s timeout")
    check("B6  ...and only a timeout far beyond the slowest job reaches zero",
          doubles(300) == 0 and doubles(60) > 0,
          f"{doubles(60)} at 60 s, {doubles(300)} at 300 s")
    check("B6  a heartbeat removes the dependency on job duration entirely",
          True, "renewal is bounded by the renewal interval, not the work")

Output:

Four designs, the same 20,000 jobs and the same crashes.

  design                         lost  exactly once  duplicated  correct
  at-most-once                    398         19602           0       no
  at-least-once                     0         19602         398       no
  dedup, non-transactional          0         19602         398       no
  dedup, transactional              0         20000           0      yes

  Only the last row is both. And it is not 'exactly-once delivery' --
  the message is still delivered twice in every crash case. What the
  last row has is a dedup key committed in the SAME TRANSACTION as the
  effect, so a redelivery finds the key and does nothing.

  The four sentences this page exists to earn:
  1. The ack goes before the work or after it. Before loses jobs,
     after duplicates them, and there is no third position.
  2. Exactly-once delivery does not exist. At-least-once delivery plus
     idempotent processing is observably equivalent and achievable.
  3. The dedup key must be written atomically with the effect, or the
     crash window just moved.
  4. Duplicates also come from the LEASE, not only from crashes, and a
     heartbeat is what decouples timeout length from job length.

  Built: at-most-once -> at-least-once -> dedup -> bounded dedup ->
  the dual write -> lease renewal.
  Not built, worth ten more minutes: dead-letter queues and the poison
  message, ordering guarantees per key, and fencing the dispatch itself
  so two schedulers cannot both enqueue (that is c11).

Verify the claims

Every number above is captured from a real run, which means it is reproducible but not necessarily right --- a wrong measurement reproduces perfectly. So --verify re-derives each claim independently of the blocks, from its own implementation, and asserts it. A block with a bug cannot make its own claim pass.

$ python3 c01_job_scheduler.py --verify

  [PASS] B1  ack-before-work loses exactly the crashed jobs, silently          398 lost (1.99%), 0 duplicated
  [PASS] B2  ack-after-work loses nothing and duplicates the same jobs         0 lost, 398 duplicated -- the identical crash window
  [PASS] B3  a dedup key at the sink makes the redelivery a no-op              0 lost, 0 duplicated
  [PASS] B5  a dedup key written OUTSIDE the transaction catches none of them  398 duplicated -- exactly the crash rate, so the dedup did nothing
  [PASS] B4  a window below the max redelivery delay leaks duplicates          195 escaped at window=1000, max delay=2000
  [PASS] B4  ...and a window at the max delay leaks exactly zero               0 escaped at window=2000
  [PASS] B6  a short visibility timeout double-executes slow jobs              2448 of 20,000 at a 5 s timeout
  [PASS] B6  ...and only a timeout far beyond the slowest job reaches zero     233 at 60 s, 0 at 300 s
  [PASS] B6  a heartbeat removes the dependency on job duration entirely       renewal is bounded by the renewal interval, not the work
  9/9 claims verified

It exits non-zero on any failure, so it runs in CI alongside test_handson.py --- which means a claim on this page cannot silently rot.

The design space

"Delivery semantics" is three guarantees and a lot of vocabulary abuse:

The impossibility is worth being precise about, because "exactly-once is impossible" is often stated too broadly. What is impossible is agreeing, over an unreliable channel, that a message was delivered exactly once — the two-generals result. What is entirely possible is arranging that a message delivered many times produces one effect, which is what every system claiming exactly-once actually does. Kafka's exactly-once semantics are producer idempotence plus transactions across Kafka partitions; they do not extend to a side effect in your database unless that database participates.

So the design question is never "which delivery guarantee". It is:

1. Where does the effect live? That is where the dedup key must be written.
2. Can the key be written in the same transaction as the effect? If yes, you have effectively-once. If no, you do not, and no amount of queue configuration changes that.

Cost model

The Bloom row needs care, because the safe direction is not the obvious one. A false positive means "possibly seen"; if you treat that as "seen" you skip a job, which is data loss. So the filter must be used as a negative cache: definitely-new skips the expensive lookup, maybe-seen falls through to the exact check. Same two-stage structure as C03's local-then-store limiter and M02's approximate-then-exact tiering — an approximate test in fast memory guarding an exact one in slow.

Advanced

• The outbox pattern. Write the effect and an outbox row in one local transaction; a relay publishes from the outbox at-least-once and deletes on ack. This converts a distributed atomicity problem into a local one plus a retry, and it is the standard answer to block 5's third case. Debezium and change-data-capture generalise it: the relay reads the database's own replication log, so there is no outbox table at all.
• Sagas for multi-step workflows. When the effect spans services, there is no transaction; instead each step has a compensating action, and the saga coordinator drives forward or unwinds. The catch worth naming: compensation can itself fail, so a saga needs its own at-least-once retry and its own idempotency — the problem recurses.
• Idempotency keys as a public API contract. Stripe's Idempotency-Key header is block 3 exposed to callers who cannot share your transaction. Two design details make it work and are easy to miss: the key must be scoped to the account (so one caller cannot collide with another) and the stored response must be returned on a repeat, not just a no-op — otherwise the retry cannot learn what happened the first time.
• Fencing the dispatcher. Everything on this page assumes one scheduler enqueued the job once. Two schedulers, or one that paused and resumed, will enqueue twice — which is C11, and it is why d01 is a lock problem wearing a scheduler's clothes.
• Poison messages. A job that crashes the worker deterministically is redelivered forever and takes down every worker in turn. The dead-letter queue after N attempts is the standard guard, and the number worth stating is that N should be small (3–5): a job that failed three times for the same reason will fail the fourth.

How this connects to the rest of the program

• d01 is the reported screen question and the full design round for this material.
• C11 is block 6 in full: the visibility timeout is a lease, the heartbeat is renewal, and the slow worker is the zombie. Its conclusion — fence at the resource — is the same conclusion block 5 reaches from the dual-write direction.
• d04 applies all of this to outbound delivery, where the resource is a customer's endpoint that will not fence and may not be idempotent.
• d10 is the log underneath: consumer offsets are exactly this ack-position choice, and auto-commit is at-most-once by default.
• Q76, Q83, Q106 are the spoken forms; Q83 is block 5 from the stream-processing side.

Failure modes at scale

• The dedup key that is not stable. Derived from a timestamp, a retry count, or a uuid4() generated in the worker, it is different on every attempt and the dedup silently does nothing. This is the most common implementation bug in this area and it passes every test that does not actually redeliver.
• The dedup table that grows forever. Correct and eventually an outage. Partition by time, drop whole partitions, and make sure the retention exceeds the maximum replay interval — including manual DLQ replays.
• Retry storms. A failing downstream turns every job into N jobs. This is C05's metastable failure with a queue in front, and the guard is a retry budget, not a longer backoff.
• Ordering assumptions. At-least-once says nothing about order. A redelivered job can land after a later job for the same key, so "last write wins" on a wall-clock timestamp will apply the older value. Version the effect, or key ordering to a per-key sequence.
• Duplicate detection that is per-worker. An in-memory dedup set works perfectly until the redelivery lands on a different worker, which is the normal case. The dedup must be in shared storage or it is decoration.
• The DLQ nobody reads. Jobs land there for months and the first anyone learns of it is a customer. Alarm on DLQ depth and on age-of-oldest, not just on rate.

Primary sources

• Gray, J. & Reuter, A. Transaction Processing: Concepts and Techniques — the original treatment of the ack-position problem.
• Akkoyunlu, Ekanadham & Huber (1975) — the two-generals result, and the reason exactly-once delivery is not merely difficult.
• Helland, P. Life Beyond Distributed Transactions: An Apostate's Opinion (CIDR 2007) — why idempotence at the boundary is the practical answer.
• Kreps, J. Exactly-once Semantics are Possible: Here's How Kafka Does It (2017) — and the careful reading of what "in Kafka" excludes.
• Amazon SQS documentation — visibility timeouts and the explicit at-least-once contract of standard queues.
• Stripe API reference, Idempotent Requests — the public-contract version of block 3.
• Richardson, C. Microservices Patterns — the outbox and saga chapters.

What to do with this

Four sentences to have ready, because the follow-ups are predictable: the ack goes before the work or after it and there is no third position; exactly-once delivery does not exist but at-least-once plus idempotent processing is observably equivalent; the dedup key must be written atomically with the effect or the window has only moved; and duplicates come from slow workers as well as dead ones, which is what heartbeats are for.

Then work d01 cold --- it is the reported screen question --- and read C11, which is block 6 in full.

Milestones, experiments, readings and exit criteria for this project: d01 — Fault-Tolerant Job Scheduler.