Distributed behaviour

Queue vs log

Delivery semantics

Ordering

Global ordering across a distributed log is a throughput ceiling — it means one partition, one consumer, no parallelism. What you actually want is ordering within a key:

partition = hash(ordering_key) % num_partitions

ordering_key = conversation_id   # messages in a chat stay ordered
ordering_key = account_id        # ledger entries per account stay ordered
ordering_key = order_id          # created → paid → shipped, in order

# Consequences to state out loud:
#  · One key's throughput is capped by one partition.
#  · Increasing partition count re-maps keys — ordering breaks across the change.
#  · A slow message blocks its partition (head-of-line blocking).

The dual-write problem and the outbox pattern

The single most valuable messaging pattern to know, because the naive version is broken and everyone writes the naive version first.

# BROKEN — two systems, no shared transaction
db.save(order)             # succeeds
kafka.publish(OrderPlaced) # crashes here → order exists, nobody knows

# FIXED — transactional outbox
BEGIN
  INSERT INTO orders  (...)
  INSERT INTO outbox  (topic, payload, created_at)   # same transaction
COMMIT
# A separate relay tails the outbox (poll, or change-data-capture via
# Debezium) and publishes. It may publish twice after a crash — which is
# fine, because consumers are idempotent. Atomicity is preserved because
# both rows commit together or neither does.

Consumer groups and rebalancing

• Each partition is owned by exactly one consumer in a group. More consumers than partitions means idle consumers — partition count is your parallelism ceiling, so over-provision partitions early (they're cheap; adding them later breaks key→partition mapping).
• A consumer joining, leaving or timing out triggers a rebalance, which stops consumption briefly. Frequent rebalances usually mean processing takes longer than the poll interval — fix by processing faster, batching smaller, or raising max.poll.interval.ms.
• Consumer lag (latest offset − committed offset) is the single most important metric on any queue. Alert on it. Rising lag means you're falling behind permanently, not temporarily.

What belongs async

Designing a job

• Idempotent. It will run twice. Key on a stable job ID.
• Small. One job = one unit. "Send 10,000 emails" should be a fan-out job that enqueues 10,000 jobs — otherwise a failure at email 9,999 replays all of them.
• Reference, don't embed. Put an ID in the payload and let the worker fetch. Embedding a 5 MB blob bloats the broker and the payload goes stale.
• Versioned. Jobs enqueued before a deploy are consumed after it. Old payloads must still parse.
• Checkpointed, if long-running. A 4-hour job that restarts from zero on a deploy will never finish.
• Time-bounded. A visibility timeout / lease longer than the worst-case runtime, or another worker picks it up while the first is still going.

Backpressure

Producers are usually faster than consumers. Without a response to that, queues grow without bound until the broker fills, memory dies, or latency becomes meaningless. Pick one deliberately:

Retries and the dead-letter queue

attempt 1  → fail → wait  1s ± jitter
attempt 2  → fail → wait  2s ± jitter
attempt 3  → fail → wait  4s ± jitter
attempt 4  → fail → wait  8s ± jitter
attempt 5  → fail → DEAD LETTER QUEUE

# Retry only what can succeed later:
#   retry:     timeouts, 429, 503, connection reset, deadlock
#   never:     400, 401, 403, 422, malformed payload — these fail forever

The DLQ is not a graveyard, it's an inbox. It needs: an alert when anything lands in it, the original message plus the error and stack trace, and a documented, tested replay path for after you fix the bug. A DLQ nobody looks at is just a slower way to lose data.

Watch for the poison message: one malformed message that fails, gets redelivered, fails again, and blocks its partition forever while consuming all your retry capacity. A max-attempts count that routes to the DLQ is what stops it.

Scheduling and priority

• Delayed jobs — a delay parameter (SQS), a Redis sorted set keyed on run-at, or a scheduler topic.
• Cron across N instances — needs a leader or a lock, or every instance runs the job. See topic 15.
• Priority — use separate queues per priority with dedicated worker pools, not a priority field. A single priority queue starves the low tier completely; separate pools guarantee low-priority work still moves.
• Fairness — one tenant enqueueing a million jobs must not delay everyone. Per-tenant queues, or weighted round-robin across tenants.

The algorithms

Distributed rate limiting

With 50 servers, a per-server limit of limit/50 is wrong the moment traffic is uneven, and a shared counter adds a network hop to every request. The standard answer is Redis with an atomic script:

-- Token bucket in one atomic Redis Lua call (no read-modify-write race)
local tokens, last = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local now, rate, cap, cost = tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3]), 1
tokens = math.min(cap, (tokens or cap) + (now - (last or now)) * rate)
if tokens < cost then return {0, tokens} end
redis.call('HMSET', KEYS[1], 'tokens', tokens - cost, 'ts', now)
redis.call('EXPIRE', KEYS[1], 3600)
return {1, tokens - cost}

To avoid the hop on every request, use local buckets with periodic reconciliation: each server holds a slice of the budget locally and syncs with Redis every second. Slightly over-permissive during the sync interval, which is almost always an acceptable trade for removing a network call from the hot path.

What to limit, and where

Choose the key deliberately. Per-IP alone punishes corporate NAT and mobile carriers, and is trivially bypassed with a proxy pool; per-user requires authentication, so unauthenticated endpoints still need an IP or device dimension. Most real systems layer several: per-IP and per-user and per-tenant, with different windows. Weight by cost too — a single expensive query can be worth 100 tokens.

Always return the metadata clients need to behave:

HTTP/1.1 429 Too Many Requests
Retry-After: 30
RateLimit-Limit: 1000
RateLimit-Remaining: 0
RateLimit-Reset: 1756483200

Timeouts — non-negotiable

Every network call gets a timeout. A call without one waits forever, holding a thread and a connection; enough of them and your service is down because someone else's service is slow. Set the timeout from the dependency's p99, not its average (say p99 × 1.5), and make sure timeouts decrease as you go deeper: if the client waits 3 s, the API can wait 2 s and the database 1 s. The other way round means you keep working on requests nobody is listening for any more.

Retries — and how they cause outages

delay = min(cap, base × 2^attempt) × random(0.5, 1.5)   # full jitter

# Without jitter, every client that failed at T retries at T+1s together —
# a synchronised thundering herd on a service that is already struggling.

# Retry amplification: 3 retries per layer over 4 layers = 3^4 = 81×
# load on the bottom service. THIS is how a slow database becomes an outage.
# Fixes:
#   · Retry at ONE layer (usually the outermost), not every layer.
#   · Retry budgets: cap retries at ~10% of total requests; when the budget
#     is spent, fail fast instead.
#   · Never retry a non-idempotent call without an idempotency key.

Circuit breakers

Bulkheads

Named after ship compartments: partition your resources so one flooded section doesn't sink the vessel. Concretely — separate connection pools and thread pools per dependency, so that when the recommendations service hangs, it exhausts only its own 20 connections and checkout still has its own pool. Without bulkheads, one slow dependency consumes every thread in a shared pool and every endpoint goes down with it.

Bulkheads also apply at coarser grains: separate fleets per tenant tier, separate clusters per region, separate queues per priority. Shuffle sharding is the elegant version — give each tenant a random pair of shards from a pool, and one abusive tenant degrades only the small fraction of tenants sharing both of its shards.

Graceful degradation

Decide in advance what to drop. A product page that loses recommendations is a product page; a product page that returns 500 is lost revenue.

Availability arithmetic

99%     "two nines"    3d 15h down/year   ≈ 7.2h/month
99.9%   "three nines"  8h 46m/year        ≈ 43m/month
99.99%  "four nines"   52m/year           ≈ 4.3m/month
99.999% "five nines"   5m 15s/year        ≈ 26s/month

# Dependencies in SERIES multiply — this is the number that surprises people
5 services at 99.9% each  →  0.999^5  =  99.5%   (≈ 3.6 hours/month)

# Redundant components in PARALLEL: failure probabilities multiply
2 replicas at 99% each    →  1 − 0.01²  =  99.99%

# Which is the whole argument for redundancy AND for having fewer
# hard dependencies on the critical path. Making a dependency optional
# (degrade instead of fail) removes it from the series product entirely.

Three ways to get it

A correct idempotent handler

def charge(idem_key, amount, card):
    # 1. Claim the key. The UNIQUE constraint IS the lock — this is why
    #    it works under concurrency, not just under sequential retries.
    try:
        db.insert(idempotency, key=idem_key, state='IN_PROGRESS',
                  request_hash=sha256(amount, card))
    except UniqueViolation:
        row = db.get(idempotency, idem_key)
        # Same key, different body = a client bug. Never silently accept it.
        if row.request_hash != sha256(amount, card):
            raise Conflict("idempotency key reused with different payload")
        if row.state == 'IN_PROGRESS':
            raise Conflict("in flight, retry shortly")   # 409
        return row.response                              # replay, do nothing

    # 2. Do the work, passing the key downstream so the PSP dedups too.
    result = psp.charge(amount, card, idempotency_key=idem_key)

    # 3. Store the response WITH the key, atomically with any state change.
    with db.transaction():
        db.insert(payments, id=result.id, amount=amount)
        db.update(idempotency, key=idem_key, state='DONE', response=result)
    return result

Exactly-once, honestly

exactly-once DELIVERY  =  impossible across a network
                          (the ack can always be lost)

exactly-once EFFECTS   =  at-least-once delivery
                        + idempotent processing
                        =  achievable, and what everyone actually means

# Kafka's "exactly-once semantics" is real but scoped: it covers
# read → process → write when both ends are Kafka, via transactional
# offset commits. The moment you call an external API inside that
# loop, you are back to at-least-once + dedup.

The basic lease

SET lock:job123 <random-uuid> NX PX 30000
#  NX = only if absent (atomic acquire)
#  PX = auto-expire in 30s, so a crashed holder does not deadlock forever
#  the UUID is the ownership proof — release must be conditional:

if redis.call("get", KEYS[1]) == ARGV[1] then
  return redis.call("del", KEYS[1])       # atomic check-and-delete
else return 0 end
# A plain DEL would let you delete a lock that has already expired and
# been acquired by someone else.

Fencing tokens — the actual fix

The critical implication: fencing requires cooperation from the resource you're protecting. If the database or object store won't check a token, you cannot make the lock safe — you can only make it usually work. That is the honest framing, and stating it is a senior-level signal.

Where to keep the lock

Leader election

Often what you actually want. One node holds a lease and renews it; if it dies, the lease expires and another takes over. Raft/ZAB implementations (etcd, Zookeeper, Consul) give you this with a session, automatic renewal, and a revision number you can use directly as a fencing token. Kubernetes does exactly this for controller leases. Two things to handle: the leader must stop acting the moment it loses the lease (not when it notices), and every follower must be ready to take over cold.

Designs that avoid the lock entirely

• Partition ownership. Assign key ranges to workers (Kafka consumer groups do this). Only one worker owns a key, so no lock is needed for it.
• Optimistic concurrency. UPDATE … WHERE version = 7; zero rows updated means someone beat you — reload and retry. Excellent under low contention and requires no lock service at all.
• Database constraints. A unique index does the mutual exclusion for you, transactionally.
• Single-writer per entity. Route all commands for an entity to one process (actor model, per-key queue). Serialisation by routing instead of by locking.
• Idempotent workers. If running twice is harmless, you never needed exclusion (topic 14).