Building a Job Board API, Part 12: One Email, Even When the Message Arrives Twice

The problem I left open

At the end of Part 11, I wrote down a risk. My worker sends the email first, and deletes the SQS message after. If it crashes in between, SQS delivers the message again after 30 seconds, and the candidate gets two emails.

This isn’t a rare edge case. In Part 10 I explained that SQS uses at-least-once delivery: every message arrives at least once, but sometimes more than once. A worker that restarts, a network blip, a deployment at the wrong moment: any of these can cause a redelivery. A system built on a queue has to expect duplicates.

The answer is to make the worker idempotent. An idempotent operation has no extra effect when you repeat it. Pressing a lift’s call button five times still brings one lift. My worker must work the same way: the same message, delivered twice, must send one email.

This is the last step of Phase 3. And it’s the one I was preparing for when I added event_id to every message, back in Part 10.

Step 1: Remember what you’ve done

The idea is simple: before doing the work, write down that you are doing it. If the same work shows up again, you’ll find your note and skip it.

The note is a row in a new table:

class ProcessedEvent(models.Model):
    event_id = models.CharField(max_length=64, unique=True)
    event_type = models.CharField(max_length=100)
    processed_at = models.DateTimeField(auto_now_add=True)

unique=True is the whole mechanism

unique=True makes PostgreSQL create a unique constraint. The database itself refuses a second row with the same event_id, even if two transactions try at exactly the same moment. I checked it in the database, not just in Python:

"jobs_processedevent_event_id_key" UNIQUE CONSTRAINT, btree (event_id)

This is the same lesson as the duplicate applications back in Part 2: a Python check like “does it exist? If not, create it” has a race condition. A database constraint doesn’t.

The key is the event, not the message

Which ID should the table remember? SQS gives every message a MessageId. But if the same change were ever sent twice, the two messages would have different message IDs. The worker would think they were two different jobs.

That’s why I added event_id in Part 10. It identifies the change itself. Two messages about the same change share one event_id. That makes it the right idempotency key.

Small details

• CharField(max_length=64), not UUIDField. Real event IDs are 36-character UUIDs, but my hand-made test messages use IDs like test-missing. A UUIDField would reject those.
• processed_at records when each event was handled. I’ll need it to clean up old rows later.
• Django also created a second index, ending in _like. On PostgreSQL, Django adds one to every indexed text column, so LIKE ‘prefix%’ searches can be fast. I don’t need it here, but now I know why it’s there.

Step 2: Claiming an event

I claim an event before acting on it: write the event_id into the new table. If that write succeeds, the worker owns the event and can continue. If the row already exists, this event has already been handled, and the message is skipped. The claim and the work must happen together, so both live in the same atomic block:

def claim_event(event_id, event_type):
    try:
        with transaction.atomic():
            ProcessedEvent.objects.create(event_id=event_id, event_type=event_type)
    except IntegrityError:
        return False
    return True

It returns True if this call is the first to handle the event, and False if it’s a duplicate. I call it “claim”, because a successful insert claims the right to do the work. IntegrityError is Django’s exception for a broken database constraint, here the unique index.

Why the inner atomic()?

This function is always called inside a bigger transaction (you’ll see it in the next step). And PostgreSQL has a strict rule: when one statement fails inside a transaction, the whole transaction is marked as broken. Every query after it fails with “current transaction is aborted”, until a rollback.

The inner transaction.atomic() creates a savepoint: a small checkpoint inside the bigger transaction. If the insert fails, only that small part is rolled back. The outer transaction stays usable. It’s the same savepoint trick I used for duplicate applications back in Part 4.

Two workers at once

What if two workers receive the same event at the same moment? The database handles it for me. The second INSERT waits for the first transaction to finish:

• If the first one commits, the second gets IntegrityError, returns False, and sends no email.
• If the first one rolls back (because its email failed), the second insert succeeds, returns True, and sends the email.

No Python lock, no Redis lock. The unique index puts the two in order.

Step 3: The claim and the email in one transaction

The claim and the email have to be in the same transaction. The reason is simple: if they were separate, a crash between them would leave the system in a broken state. If the claim is saved first and the email fails, the event looks handled even though the candidate never got the email. If the email is sent first and the claim fails, the event is delivered again and the candidate gets a duplicate.

The solution is to put them together. Both inside with transaction.atomic(): a successful claim and a sent email commit together. If the email raises, the claim rolls back too, so the event stays unclaimed and the message is retried.

I moved the claim and the email into one atomic block:

def handle_status_changed(body):
    with transaction.atomic():
        if not claim_event(body["event_id"], STATUS_CHANGED):
            logger.info("Event %s already processed; skipping duplicate", body["event_id"])
            return

        try:
            application = Application.objects.select_related("job__company", "candidate__user").get(
                id=body["application_id"]
            )
        except Application.DoesNotExist:
            logger.warning("Application %s no longer exists; skipping (event %s)", body["application_id"], body["event_id"])
            return

        send_status_changed_email(application, body["new_status"])

    logger.info("Emailed user %s about application %s: %s -> %s (event %s)", application.candidate.user_id, application.id, body["old_status"], body["new_status"], body["event_id"])

The outer transaction.atomic() puts the claim and the email in one group. Here is the rule that makes everything work, and the one most likely to confuse me later:

Inside with transaction.atomic():, a normal exit (including return) commits. Only an exception rolls back.

• First delivery: the claim succeeds, the email is sent, the block ends normally, and the claim is committed.
• A duplicate: the claim fails, so we return. That commits, but nothing was written. The main loop then deletes the message. I log this at INFO, not WARNING. With at-least-once delivery, duplicates are expected. Logging them as warnings would only hide the real warnings.
• A deleted application: we return, so the claim is committed. That’s fine: this event can never be emailed, so any later redelivery should be skipped quickly too.
• The email fails: the exception leaves the block, and the claim is rolled back. The message isn’t deleted, SQS retries it after 30 seconds, and the retry can claim it again.

Why the rollback matters so much

Imagine I claimed the event in its own transaction first, and sent the email afterwards. If the email failed, the claim would already be saved. Every retry would find it and skip the message as a “duplicate”. The candidate would never get the email.

That’s worse than a duplicate email. A duplicate is annoying. A missing email means the candidate never learns that they got an interview.

And I fixed the missing space from Part 11 along the way: it’s now skipping (event %s).

The trade-off

The transaction stays open while the email is being sent. With a slow mail server, that keeps a database connection busy for a few seconds. For one worker and a small volume, that’s fine.

Step 4: Testing it

Testing idempotency is harder than writing it, because I had to make rare failures happen on purpose.

1. The normal path still works

Acme changed sam’s application from “reviewing” to “interview”. One email, one Emailed user 5 … (event 23e48c51-…) line, and the newest row in the table had the same event ID. Nothing broke.

2. The same event, sent twice

I sent two separate SQS messages, with different message IDs but the same event_id:

body = {"type": "application.status_changed", "event_id": f"dup-test-{uuid.uuid4().hex[:8]}",
        "application_id": 3, "old_status": "interview", "new_status": "interview"}
q = get_queue()
q.send(body)
q.send(body)

The random suffix lets me run the test again without clashing with older runs. The worker log:

05:03:40,033  Emailed user 5 ... (event dup-test-b15e66d8)
05:03:40,047  Event dup-test-b15e66d8 already processed; skipping duplicate

One email. The duplicate was skipped 14 milliseconds later. Both messages were deleted, and the queue was empty.

3. The crash

This is the exact failure Part 11 left open. How do you crash a worker at precisely the wrong moment? I didn’t kill any process. I acted out the crash instead:

1. I stopped the worker, and changed a status, so the message waited in the queue.
2. In the Django shell, I received the message and called the worker’s own handle() function, but I did not delete the message. That’s exactly what a crash after sending looks like. The email was printed.
3. I started the real worker again and waited 35 seconds, a little longer than the 30-second visibility timeout.

The worker log showed only this:

Event 9c8dfc41-... already processed; skipping duplicate

No second email. Before this step, the same sequence would have emailed sam twice.

I could import handle() in the shell without starting the whole worker, because of the if __name__ == “__main__”: line from Part 10. It runs main() only when the file is executed, not when it’s imported.

4. A failed email

The console backend never fails, so I had to force a failure. I used unittest.mock:

with mock.patch.object(run_worker, "send_status_changed_email", side_effect=RuntimeError("mail server down")):
    try:
        run_worker.handle_status_changed(body)
    except RuntimeError as e:
        print("1st attempt failed:", e)
print("claimed after failure:", ProcessedEvent.objects.filter(event_id=event_id).exists())

run_worker.handle_status_changed(body)
print("claimed after retry:", ProcessedEvent.objects.filter(event_id=event_id).exists())

patch.object replaces the email function for the length of the with block, and side_effect makes it raise an error every time it’s called. The results:

1st attempt failed: mail server down
claimed after failure: False
claimed after retry: True

The failure rolled the claim back, and the retry sent the email. Both halves matter. Without the rollback, the claim would survive, and the candidate would never get the email.

One detail in this test took me a moment to understand: patch the name where it is used, not where it is defined. My worker did from jobs.emails import send_status_changed_email, which copied the name into the worker module. If I had patched jobs.emails.send_status_changed_email, the worker would still hold its own copy of the real function. The test would have sent a real email instead of failing.

5. Nothing else broke

The worker stayed Up, there were 0 “Failed to handle message” lines in five minutes, and the DLQ was empty.

Effectively once, not exactly once

I want to be precise here, because “exactly once” is a phrase that gets used too easily.

At-least-once delivery plus an idempotent worker gives you effectively-once processing. That’s the standard pattern with SQS, Azure Service Bus and most other queues. But it is not a perfect guarantee.

There is still one tiny gap: the moment after the email has left, but before the database commit. If the worker crashed in exactly that instant, the claim would roll back, and the retry would send a second email. Nothing can fully close that gap for an outside side effect like email. It would need the mail provider itself to remove duplicates by key, and Amazon SES doesn’t offer that for normal sends. What I can say honestly: the window went from “any crash between sending and deleting” to “a crash in one exact instant”.

Other limits

• The table grows forever. But a row is only useful while its message could still be redelivered. SQS keeps messages for 4 days by default, and 14 days at most. A scheduled cleanup that deletes rows older than about 14 days would be enough. It isn’t built yet.
• The transaction is held open during the send. At high volume, the better design splits it into three steps: claim, send, then mark as sent.

Why not Redis for the claim?

Redis has SET NX, which is very fast and could also “claim” a key. But it is not part of the database transaction. If the email failed, the Redis key wouldn’t roll back with it, and the email would never be retried. And as I learned in Part 9, Redis data is lost when the container is recreated. PostgreSQL was already there, and the claim and the rollback come together for free.

What I learned

• A queue will deliver duplicates. Design for them from day one, not after the first double email.
• Remember what you’ve done, and let the database enforce it. A unique index is a race-proof memory.
• The idempotency key identifies the change, not the delivery. event_id, not MessageId.
• Put the claim and the side effect in one transaction. A failed send must release the claim. A missing email is worse than a duplicate.
• return commits, an exception rolls back. Inside transaction.atomic(), that one rule decides every outcome.
• Use a savepoint to catch a database error without breaking the transaction around it.
• Expected events aren’t warnings. A skipped duplicate is INFO.
• You can test rare failures on purpose. Act out the crash, and use mock to make a dependency fail.
• Patch the name where it’s used.
• Be honest about the guarantee. This is effectively once, not exactly once.

Phase 3 is complete

That was the last item on Phase 3’s checklist. Over four parts:

• Part 9: a Redis cache for job search, with versioned keys.
• Part 10: an SQS queue and a separate background worker.
• Part 11: the worker sends the status emails.
• Part 12: the worker is idempotent, so a message delivered twice sends one email.

The goal of the phase was: “a status change queues a job, and the worker processes it.” It does, safely.

What comes next

Phase 4 is where my DevOps work really starts: making the app operations-ready. That means /healthz and /readyz endpoints for Kubernetes probes, Prometheus metrics, structured JSON logs (which will also finally fix the missing INFO logs from Part 10), and a multi-stage Dockerfile with a non-root user. The goal: the container runs with only environment variables, and exposes health and metrics.