Bound background jobs and make retries safe

A service can return a timeout while a database query, conversion process or queued job continues consuming resources. Repeated client retries then multiply the work. Define the lifetime and cost of the operation independently from the lifetime of its first HTTP connection.

List the resources one accepted request can reserve: worker slots, database connections, temporary files, queue messages, storage and provider calls. Choose budgets from legitimate workloads and capacity measurements. Per-IP rate limiting at the API cannot protect a shared worker pool from one unusually expensive operation or a distributed set of valid accounts.

The queue in this example guarantees at-least-once delivery. That permits the same accepted operation to be delivered more than once. At 12:00 worker W charges an external provider; at 12:01 W crashes before acknowledging the message. Redelivery at 12:02 is expected queue behavior, not evidence that the queue is broken. If the second worker calls the provider with a fresh charge identity, the business operation can execute twice.

The repair must connect the logical operation to the external effect. Store operation O with an immutable digest of its material inputs and use a stable provider idempotency key derived from O within the provider’s documented scope and retention. A worker that cannot tell whether the first call completed must query or retry under that same identity, then reconcile the result. A local “already done” flag written after the provider call has the same crash window; a database transaction cannot atomically commit an unrelated remote provider without a supported protocol.

Reject reuse of O with different material inputs. Otherwise a key originally bound to tenant North, invoice 7, amount 5000 cents could silently authorize an altered payment or return a misleading old result. Define identity scope, input binding, retention and terminal states explicitly. When the provider’s idempotency window expires, automatic blind retry is unsafe; use a recorded provider reference and reconciliation path.

Cancellation is also a state question. A queued operation that has not begun may be cancelled without effect. A committed charge cannot be made nonexistent by disconnecting the client or setting cancelled=true. Report the completed result and, if the business policy permits, initiate a separate authorized refund operation. Bound the queue and active worker slots independently so a retry storm does not exhaust the service before its idempotency logic runs.

OPERATION O7
tenant=North, invoice=7, amount_cents=5000
Provider contract for this exercise: same key + same input returns same charge for 24 h; changed input is rejected
12:00 W1 sends key=O7; provider commits charge C9
12:01 W1 crashes before saving C9/acknowledging
12:02 queue redelivers O7
Queue: at least once; tenant max running=2, waiting=5

Set input and per-operation limits before expensive parsing or fan-out, as the API resource limits guide at /domains/runtime/api-security/resource-consumption-limits/ shows. Bound running work and queue depth, with a clear response when admission capacity is unavailable. Separate tenants or workload classes where one customer's bulk export could starve another customer's interactive task. Include an overall spending or call budget for paid downstream services where supported. Managed consumers have the same knobs: an SQS event source mapping can cap how many Lambda instances consume one queue with maximum concurrency, and its visibility timeout and redrive policy decide when a message is retried and when it is parked.

For asynchronous APIs, distinguish durable acceptance from completion. Return an operation identifier only when the system can track its state reliably. Define cancellation as a state transition with explicit limits: work may already have committed an irreversible effect. Let clients inspect status rather than repeatedly submitting the same job because the initial connection ended.

Propagate deadlines or remaining time to database, network and subprocess operations as timeouts, and implement cooperative cancellation where the library requires it. Clean up temporary artifacts and release slots on errors as well as success. Apply bounded retry counts and backoff with jitter only to failures whose operation semantics permit retry, and cap retries across the whole service with a retry budget, such as retries at most 10% of requests, so a dependency outage is not multiplied by every caller.

Assume at-least-once delivery where the queue contract allows duplicates. Bind a durable idempotency record to the logical operation and its material inputs, then coordinate it with business state; the webhook signature and replay lab at /labs/webhooks/ repairs a handler that pays twice. A worker crash after the effect but before acknowledgement must not cause a second charge or export grant. Poison messages need an observable, bounded failure path rather than endless immediate redelivery.

The worker below implements worker B in Python with SQLite as the durable store. It records the operation with a SHA-256 digest of its inputs before calling the provider, reuses the operation ID as the provider's idempotency key, and records the charge before the message is acknowledged. Retryable failures back off with full jitter up to five attempts; conflicts and exhausted retries go to a dead-letter queue. The tests in the verification section crash it between the provider call and the database update: the redelivery returned the same charge and the provider recorded one charge after two calls.

"""Handle an at-least-once charge job: one durable record per operation, the provider's
idempotency key, bounded retries with full-jitter backoff, then a dead-letter queue."""
import hashlib
import json
import random
import sqlite3
import time


class Conflict(Exception):
    """The operation ID was already used with different inputs."""


class Retryable(Exception):
    """A failure that may succeed later, such as a timeout or an HTTP 503."""


def open_db(path):
    db = sqlite3.connect(path, isolation_level=None)  # autocommit: each statement is durable
    db.execute("""CREATE TABLE IF NOT EXISTS operations (
        id TEXT PRIMARY KEY, input_digest TEXT NOT NULL,
        state TEXT NOT NULL CHECK (state IN ('started', 'charged')), charge_id TEXT)""")
    return db


def process(db, provider, op_id, inputs):
    """Safe to run again for the same op_id after a crash at any point."""
    digest = hashlib.sha256(json.dumps(inputs, sort_keys=True).encode()).hexdigest()
    db.execute("INSERT OR IGNORE INTO operations VALUES (?, ?, 'started', NULL)", (op_id, digest))
    stored, state, charge_id = db.execute(
        "SELECT input_digest, state, charge_id FROM operations WHERE id = ?", (op_id,)).fetchone()
    if stored != digest:
        raise Conflict(f"{op_id} was accepted with different inputs")
    if state == "charged":
        return charge_id  # a redelivery after success: nothing left to do
    charge_id = provider.charge(idempotency_key=op_id, **inputs)  # same key, same charge
    db.execute("UPDATE operations SET state = 'charged', charge_id = ? WHERE id = ?", (charge_id, op_id))
    return charge_id  # acknowledge the message only after this returns


def deliver(db, provider, message, dead_letters, attempts=5, base=0.2, cap=10.0, sleep=time.sleep):
    for attempt in range(attempts):
        try:
            return process(db, provider, message["id"], message["inputs"])
        except Retryable:
            sleep(random.uniform(0, min(cap, base * 2 ** attempt)))  # full jitter
        except Conflict:
            break  # retrying cannot fix a conflict
    dead_letters.append(message)  # bounded: park it for a person to reconcile
    return None

Run a small controlled workload with an expensive job, duplicate deliveries, a full queue, a stalled dependency and a cancelled client. Observe actual worker count, connection use, temporary storage and provider stub calls. Verify that another tenant's legitimate request still receives the promised service and the system returns to its baseline after pressure stops. Alerts alone do not stop work or spending; test the enforcing mechanism.

The tests below cover the crash window, the conflicting redelivery, transient failures and the dead-letter path. They pass on Python 3.14.7 with only the standard library.

import unittest

import charge_worker as w


class Provider:
    """Stub with the exercise's contract: same key and inputs return the same charge."""

    def __init__(self, failures=0):
        self.charges, self.calls, self.failures = {}, 0, failures

    def charge(self, idempotency_key, **inputs):
        self.calls += 1
        if self.failures:
            self.failures -= 1
            raise w.Retryable("503 from provider")
        if idempotency_key in self.charges:
            if self.charges[idempotency_key][0] != inputs:
                raise ValueError("key reused with different parameters")
            return self.charges[idempotency_key][1]
        self.charges[idempotency_key] = (inputs, f"ch_{len(self.charges) + 1}")
        return self.charges[idempotency_key][1]


O7 = {"tenant": "North", "invoice": 7, "amount_cents": 5000}


class CrashBeforeSave(Exception):
    pass


class Tests(unittest.TestCase):
    def setUp(self):
        self.db, self.provider = w.open_db(":memory:"), Provider()

    def test_crash_after_charge_then_redelivery_charges_once(self):
        real = self.db

        class Crashing:  # fails the UPDATE that would record the charge
            def execute(self, sql, params=()):
                if sql.startswith("UPDATE"):
                    raise CrashBeforeSave()
                return real.execute(sql, params)

        with self.assertRaises(CrashBeforeSave):
            w.process(Crashing(), self.provider, "O7", O7)
        self.assertEqual(w.process(self.db, self.provider, "O7", O7), "ch_1")  # redelivery
        self.assertEqual((self.provider.calls, len(self.provider.charges)), (2, 1))

    def test_changed_amount_under_the_same_id_is_rejected(self):
        w.process(self.db, self.provider, "O7", O7)
        with self.assertRaises(w.Conflict):
            w.process(self.db, self.provider, "O7", {**O7, "amount_cents": 9000})

    def test_transient_failures_retry_with_backoff_then_succeed(self):
        provider, delays = Provider(failures=2), []
        result = w.deliver(self.db, provider, {"id": "O8", "inputs": O7}, [], sleep=delays.append)
        self.assertEqual(result, "ch_1")
        self.assertEqual(len(delays), 2)
        self.assertTrue(all(0 <= d <= 10 for d in delays))

    def test_persistent_failure_and_conflict_go_to_the_dead_letter_queue(self):
        dead = []
        w.deliver(self.db, Provider(failures=99), {"id": "O9", "inputs": O7}, dead, sleep=lambda s: None)
        w.process(self.db, self.provider, "O7", O7)
        w.deliver(self.db, self.provider, {"id": "O7", "inputs": {**O7, "amount_cents": 1}}, dead)
        self.assertEqual([m["id"] for m in dead], ["O9", "O7"])


if __name__ == "__main__":
    unittest.main()

An in-memory set of completed IDs disappears on restart and usually does not coordinate multiple workers. A durable unique constraint can arbitrate ownership, but the worker still needs a crash-recovery state machine and external-effect identity. Test the crash before the call, after the provider commit and after local persistence separately.

Retry policies must classify outcomes. A validation rejection is not transient. An ambiguous network timeout is not proof that the provider did nothing. Keep bounded attempts with backoff and a terminal/manual-reconciliation state; repeated immediate redelivery can exhaust capacity while making the uncertainty worse.

These outcomes belong to the worked example; the tests above check the first two rows against the reference worker.

Sources & further reading

The standards, specifications and documentation behind the guidance above. Each is the evidence for a claim on this page; consult it for detail beyond the defaults and versions listed here.

1. Amazon SQS at-least-once delivery
2. gRPC deadlines and cancellation
3. AWS Lambda best practices
4. Amazon SQS: visibility timeout
5. Amazon SQS: dead-letter queues
6. AWS Lambda: configuring an SQS event source
7. AWS Architecture Blog: exponential backoff and jitter. 4 March 2015.
8. Stripe: idempotent requests
9. Google SRE book, chapter 21: handling overload (retry budgets)

Reference information

Tags

Business logic and race conditionsDenial of service and resource limits

SDLC phase

DesignDevelopmentTestingMaintenance

Role

Developers

ASVS 5.0.0 requirements

v5.0.0-15.2.2 · v5.0.0-16.5.2

Download this guide as Markdown

Continue exploring