At-Least-Once and Idempotent Jobs

The stream delivers every job at least once, which means sometimes twice. The worker sent the email, was killed before it acknowledged, and 60 seconds later the reclaim loop handed the same entry to another worker, which sent the email again. No queue fixes this. The handler does, by being idempotent: running it twice has the effect of running it once. Chapter 1 said the sentence and Chapter 7 wrote it into the outbox on purpose; this topic is where Stagedoor's handlers are written to honour it.

There are four techniques, and they are not alternatives. A handler that reads the current state and does only what is missing is idempotent for the state it can read. A unique constraint on the effect's row makes two concurrent runs agree on which one owns it. An idempotency key passed to the provider covers the send the service does not remember making. A job id recorded in the same transaction as the job's effects covers the effects that have no state to read. Stagedoor's ticket jobs use all four, one per failure they close, and the last section is the one technique that looks like these and is not.

Why Twice Happens

Four paths lead to a second run, and all of them are the fourth outcome of Chapter 1 wearing the worker's clothes. The worker crashes between the work and the acknowledgement, and the entry is reclaimed. A worker that is alive but slow, held up 70 seconds behind organizer reports on the replica, has its entry reclaimed at the 60-second threshold and finishes it anyway while the second worker starts. The relay of Chapter 7 crashed between publishing an outbox row and marking it, so two entries with the same outbox id are on the stream. And a retry after a timeout in Topic 46 re-runs a handler whose first attempt did the work and then failed to report it.

The timeline is the argument against wanting exactly-once from the queue. To deliver once and only once across that crash, the acknowledgement in Redis and the send at the mail provider would have to be one atomic operation, and they are two systems on two networks with no transaction between them. Any product that promises exactly-once end to end is promising at-least-once plus something on the receiving side, and this topic is that something.

Check State First

The first technique rewrites the job's meaning. "Render the tickets for order 4471" is an instruction to do an action, and an action done twice is done twice. "Make sure order 4471's tickets are rendered" is a request to reach a state, and a handler that reads the current state and does only what is missing reaches it once however many times it runs. The state is a column: tickets.pdf_key, null until the PDF is stored, added by this chapter's migration. A second run finds it set and returns.

async def render_tickets(job):                       # "order 4471's tickets are rendered and queued to send"
    order = await load_order(conn, job.order_id)
    if order.pdf_key is not None:                    # the state is already reached: nothing to do
        return
    pdf = await asyncio.to_thread(render_pdf, order)      # 4 s in the thread pool, no transaction open
    key = await store_pdf(pdf, f"tickets/{order.public_id}.pdf")   # same key every run; a second write is harmless
    async with conn.transaction():
        won = await conn.execute(
            "UPDATE tickets SET pdf_key = %s WHERE order_id = %s AND pdf_key IS NULL RETURNING id",
            (key, job.order_id))                            # the conditional write of Chapter 6
        if await won.fetchone() is not None:               # this run reached the state: it queues the email
            await outbox.add(conn, "send_tickets", order_id=job.order_id,
                             request_id=job.request_id, trace=job.trace)

The handler loads the order and, if the PDF key is already set, returns without doing anything. Otherwise it renders in a thread with no transaction open, which is the fix for the worker's version of the pinned connection in Topic 31 of Chapter 6, stores the file under a key derived from the order so that a second store overwrites the first with identical bytes, and then in one transaction writes the key on the order's ticket rows with the conditional write of Topic 34 of Chapter 6, and queues the email job through the outbox only if that write changed something. Two workers that both pass the null check render twice and waste 4 seconds of CPU; they cannot store two different files, and only the one whose UPDATE found the key still null queues the email, because the other's update matches no rows. Reading state first turns "did I already run" into "is the world already the way this job wants it," and the second question has an answer in the database.

Unique Constraint on the Effect

The email has no column to read: it left the process, and the mail provider has it. So the handler gives the effect a row. email_sends has order_id, kind, claimed_at and sent_at, with a unique constraint on (order_id, kind), and the handler inserts the row before it sends. Two runs that arrive at the same moment both try the insert, the constraint lets one through, and the other finds the row and reads it. It is the shape of the idempotency table in Topic 39 of Chapter 7 and of the seat constraint in Topic 34 of Chapter 6, applied to a side effect instead of a resource, and it decides the race the same way, in the database, where two concurrent writers cannot both win.

async def send_tickets(job):                         # "order 4471's ticket email has been sent"
    claimed = await conn.execute(
        "INSERT INTO email_sends (order_id, kind, claimed_at) VALUES (%s, 'tickets', now()) "
        "ON CONFLICT (order_id, kind) DO UPDATE SET claimed_at = now() "
        "WHERE email_sends.sent_at IS NULL AND email_sends.claimed_at < now() - interval '2 minutes' "
        "RETURNING order_id", (job.order_id,))            # one statement decides who owns the send
    if await claimed.fetchone() is None:
        send = await fetch_send(conn, job.order_id, "tickets")
        if send.sent_at is not None:
            return                                   # the state is reached
        raise RetryLater(seconds=120)                # another run holds a fresh claim; look again when it is stale
    order = await load_order(conn, job.order_id)
    await mailer.send(order, order.pdf_key,
                      idempotency_key=f"tickets-{job.order_id}")   # the provider deduplicates what we cannot
    await conn.execute("UPDATE email_sends SET sent_at = now() WHERE order_id = %s AND kind = 'tickets'",
                       (job.order_id,))

The handler's first statement is the claim. If no row exists, the insert creates one with the claim time and returns it; if a row exists, the conflict clause re-claims it only when the email is not yet sent and the previous claim is older than 2 minutes, and otherwise returns nothing. A run that gets nothing back reads the row: sent means the state is reached and it returns; unsent with a fresh claim means another run is sending at this moment, and it asks Topic 46 to try again in 2 minutes, by which time the row will say sent or the claim will be stale. The run that won sends, with an idempotency key derived from the order, and then records the time. Send, then mark, is the relay's choice again: a crash between the two produces a second send rather than a row that says sent above an email that never left, and the second send is what the next section is for.

The Provider's Idempotency

A crash between the send and the UPDATE leaves a row with a claim, a null sent_at, and an email already in the buyer's inbox. Two minutes later the claim is stale, the retry re-claims the row and sends again. The row cannot know the first send happened; it happened on another network. The mail provider can, if the request carried an idempotency key, and every provider Stagedoor talks to takes one: the mail API and Payrail both keep their own table of keys, the way Topic 39 of Chapter 7 built one for Stagedoor's clients. The job passes tickets-4471 on every attempt, so the retry's send with the same key returns the provider's first answer and no second email leaves. It is the last line of defence and the one that covers the case nothing inside the service can see, which is why the key is derived from the order and not generated per attempt.

Topic 53 of Chapter 10 builds the provider client and the rule that its key is the intent translated into the provider's vocabulary. Here the intent is "the ticket email for order 4471," and there is exactly one of those however many times the job runs.

Job Id as the Key

Some effects have no state to read and no natural row. The organizer's sales counter is one: update_sales_counter adds the order's seat count to the event's total, and a counter that was incremented twice looks exactly like a counter that was incremented once by a larger order. The fourth technique gives every job an identity and records it. Each entry carries an identity: the outbox_id field for jobs born in the outbox, and a job_id the scheduler mints for the ones in Topic 46. The handler inserts whichever the entry carries into processed_jobs in the same transaction as the increment, with ON CONFLICT DO NOTHING RETURNING, and if the insert returns nothing the effects are already committed and the handler returns. Insert and increment commit together or not at all, so there is no moment at which the counter has moved and the id is not recorded, or the reverse.

The outbox id is the right identity and the stream's entry id is not, because the relay's duplicate is two entries with two ids and one outbox row. A processed_jobs keyed on the entry id would let both through. The limit of the technique is its scope: it makes any handler idempotent for effects inside the database, and says nothing about the email, because the email is outside the transaction that records the id. That is why the three earlier techniques exist, and why send_tickets uses them and not this one.

What Does Not Work

"It will rarely happen" is not a technique. The reclaim fires on every deploy that kills a worker mid-job, and the relay's duplicate arrives on every night the relay is restarted, which is 3 or 4 times a month between them; a mechanism that fails 4 times a month is a bug with a schedule. The second thing that does not work is a check that is in the right place and the wrong transaction. A handler that reads sent_at IS NULL, sends, and then writes sent_at has the check and the effect 300 milliseconds apart, and two workers running the reclaimed and the original entry at once both read null, both send, and both write. It is the read-then-write race of Topic 34 of Chapter 6 across a mail provider instead of a seat, and it is why the constraint decides who owns the row and the provider's key catches the send that slipped past it.