Queues and Retries

A queue separates accepting work from doing it. A request can record an instruction to generate a report, then return while a worker performs the expensive operation. Workers can scale independently, and a temporary burst can wait in a buffer. The system now has another responsibility: deciding which work is unfinished when a worker disappears.

A retry is another attempt at the same intended operation. It can recover from a temporary failure, but it can also repeat a payment, multiply overload or complete work after its result has become useless. Reliable asynchronous work needs both a delivery protocol and an application contract for repeated execution.

The downloadable fixture uses Python 3.11+ and its standard-library SQLite module. Extract it, change into the extracted directory, and run:

python3 lab.py
python3 -m unittest discover -s . -v
# The demo asserts its observations; the tests include negative cases.

No broker or cloud account is required. SQLite stores real transactions; queue leases and stream offsets are deliberately small teaching models. The script terminates worker subprocesses at specific failure points and removes its temporary databases afterward.

Work ownership and replay

In a work queue, consumers compete to handle jobs. Receiving a job normally makes it unavailable to other consumers while it is in flight. An acknowledgement marks that delivery complete. If completion is uncertain, the queue may deliver it again. A publisher confirmation answers a different question: whether the broker accepted a publication, rather than whether a worker finished its business operation.

Some queues use a visibility timeout or lease; others detect a closed consumer connection. RabbitMQ manual acknowledgements, for example, allow unacknowledged deliveries to be requeued after connection loss. The fixture uses a five-second lease and a changing delivery token. With a fresh queue.db, these operations expose ownership separately from completion:

from lab import initialize, enqueue, claim, ack

initialize("queue.db")
enqueue("queue.db", "credit-17", 10, capacity=1)
first = claim("queue.db", now=0)
assert claim("queue.db", now=4) is None
second = claim("queue.db", now=5)
assert not ack("queue.db", first, now=5)   # Old delivery token.
assert ack("queue.db", second, now=5)

Lease expiry does not stop the old worker. Two executions can overlap, so rejecting stale acknowledgements alone cannot protect a business effect.

A retained stream separates records from reading progress. Completing a record advances a consumer group's position; it does not delete that record for everyone else. A billing group and an analytics group can progress independently and replay available history:

records = ("created", "paid", "shipped")
next_offset = {"billing": 2, "analytics": 0}
assert records[next_offset["billing"]:] == ("shipped",)
assert records[next_offset["analytics"]:] == records

This tuple illustrates offsets, not Kafka's implementation. Actual replay is bounded by retention and compaction; ordering is generally a partition-level contract. Choose a work queue for distributing tasks and a retained stream when independent readers need history. Products can support both patterns.

Buffering does not create capacity. If arrivals average 120 jobs/s and workers finish 100 jobs/s, unfinished work grows by 20 jobs/s. The fixture rejects admission above its configured number of ready plus in-flight jobs. A real system also needs a byte bound, an overload response and an age limit. Consumer concurrency and prefetch bound work already handed to workers; they do not bound the broker's entire backlog. Watch oldest-job age and completion rate alongside queue depth.

The crash between effect and acknowledgement

Consider a job that credits an account by 10 units. The worker commits the credit, then dies before acknowledging. On redelivery, an unguarded increment credits another 10. Acknowledging first reverses the failure: a crash can permanently lose the credit. Neither ordering makes two independent operations atomic.

The demo runs both variants. Its unsafe subprocess exits immediately after commit; the parent advances the lease clock and starts another worker. Two deliveries leave the account at 20. The safe variant leaves it at 10 because the domain update and a durable idempotency record share one transaction.

Here is the safe transaction's central logic, with job carrying a stable business identifier and canonical JSON payload. The complete apply function in lab.py also validates input, rolls back on error and closes the connection:

db.execute("BEGIN IMMEDIATE")
prior = db.execute(
    "SELECT payload FROM processed WHERE id=?", (job["id"],)
).fetchone()
if prior:
    if prior["payload"] != job["payload"]:
        raise ValueError("idempotency key reused with a different payload")
else:
    db.execute("INSERT INTO processed VALUES (?,?)",
               (job["id"], job["payload"]))
    db.execute("UPDATE account SET total=total+? WHERE id=1", (amount,))
    db.execute("INSERT INTO outbox(id,payload) VALUES (?,?)",
               (job["id"], job["payload"]))
db.commit()
# Acknowledge only after this commit succeeds.

processed.id is a primary key, and BEGIN IMMEDIATE acquires SQLite's write transaction before checking it. A crash before commit leaves the credit, deduplication record and outbox row absent. A crash after commit leaves all three present. Redelivery sees the record and skips the mutation. The tests kill actual processes at both boundaries and reopen the database to check these facts.

Writing a deduplication marker in an earlier transaction would allow a crash to suppress work that never happened. Writing it afterward would leave the duplicate window open. Keep the key stable across retries, scope it to the caller and operation, reject changed parameters, and retain it through the supported replay window. Deleting old records changes the idempotency contract.

The outbox stores an intent to notify another system in that same local transaction. A dispatcher sends committed rows and records successful delivery afterward. Sending is still a separate boundary:

receive(receiver_db, row["id"], row["payload"], dedupe=True)
# Injected crash here: receiver committed; sender still sees sent=0.
db.execute("UPDATE outbox SET sent=1 WHERE id=?", (row["id"],))

The fixture's receiver is an independent SQLite database with its own transaction. With receiver deduplication disabled, restarting the dispatcher produces an external total of 20; enabled, it stays at 10. The receiver atomically records the event key with its own mutation. An email or payment API needs an equivalent idempotency contract, or reconciliation when its outcome is ambiguous.

The outbox prevents a local commit from forgetting its notification intent. It does not make arbitrary external effects exactly once. Broker transactions and delivery guarantees have boundaries too. Production dispatchers also need bounded batches, retry scheduling and ordering rules where order matters. Database isolation itself is developed in Indexes and MVCC.

Retry only while work is useful

Retry a failure only when it may recover and repeating the operation is safe. A temporary unavailable dependency differs from an invalid payload. A timeout reports uncertainty, not rollback; use the same operation identity for the next attempt.

Bound attempts per operation, elapsed time across all attempts, and extra attempts across operations. Otherwise many individually small retry loops can exhaust a struggling dependency. If three layers each allow three total attempts, one request can cause 27 calls at the bottom. Assign retry ownership deliberately and account for SDK retries.

The fixture's retry helper passes a timeout clipped to the remaining deadline. Before every additional call it spends a token from a shared finite retry budget. Only Retryable exceptions enter the retry path; permanent errors propagate immediately. The backoff decision is:

remaining = deadline - clock.now
if remaining <= 0:
    raise Stopped("deadline")
delay = rng.uniform(0.0, min(cap, base * 2 ** attempt))
if delay >= remaining:
    raise Stopped("deadline")
clock.sleep(delay)

This is capped exponential backoff with full jitter: after failure number one the delay lies between zero and base seconds, then between zero and twice base, until cap. Randomness spreads coordinated retries; an attempt limit still matters because a random delay may be near zero.

In the demo, two retry tokens, a one-second deadline, a 0.2-second attempt timeout and seed 7 allow a third attempt to succeed after 0.213 simulated seconds. This measures a policy, not network performance. Tests separately force budget exhaustion across two operations, a permanent error, the attempt cap, an already-expired deadline and a backoff that would consume the remaining deadline.

The finite budget does not replenish. A production budget needs a deliberate refill policy, synchronization across concurrent callers and appropriate scope. Initial requests still need admission control. The helper also assumes its client honors the supplied timeout; it cannot interrupt arbitrary work. Use a monotonic clock for local elapsed time. Carrying deadlines between machines needs a defined timestamp or remaining-budget protocol that accounts for clock differences and time already spent queued.

Poison messages need a stopping point

A poison message consistently fails because its content or required state cannot be processed. Immediately requeueing it forever spends capacity without making progress. Park permanent failures at once; cap transient redeliveries so an unavailable dependency also has a stopping point.

The fixture preserves parked jobs in state='dead', retaining their identifier, payload, delivery count and failure reason. After claiming a negative credit amount, its worker path is:

try:
    apply(path, job)
except ValueError as error:
    fail(path, job, now=0, reason=str(error), permanent=True)
# state=dead; account unchanged; payload and reason remain inspectable.

The tests also reject an acknowledgement from an expired delivery, park a transient failure after three deliveries, and park a job whose business deadline expired before execution. A visibility timeout and a business deadline are different: one makes ownership recoverable, while the other decides whether the result is still wanted. Long tasks may need lease renewal, but renewal does not remove the need for idempotency.

A dead-letter destination is an operational workflow. Alert on its growth, protect sensitive payloads, preserve enough context to diagnose the cause and decide who repairs it. Broker dead lettering requires its own configured routing and delivery guarantees. Before replay, check whether an earlier attempt already applied effects. Repair the cause, preserve the business identity when repeating the same operation, and replay at a bounded rate. Otherwise recovery itself can become the next burst of overload.