Distributed Communication Patterns#

# Intolerant: strict mode turns any new producer field into a rejected message.
class Order(BaseModel):
    model_config = ConfigDict(extra="forbid")   # DON'T
    id: str
    amount_minor: int


# Tolerant: unknown fields are ignored, missing optionals get defaults.
class Order(BaseModel):
    model_config = ConfigDict(extra="ignore")   # DO
    id: str
    amount_minor: int
    currency: str = "GBP"                       # added in v2, defaulted here


# Tolerant AND lossless: if you re-emit the message, carry unknowns through,
# otherwise a middleman silently strips fields its neighbours depend on.
class Envelope(BaseModel):
    model_config = ConfigDict(extra="allow")
    event_id: str
    event_type: str

// Intolerant: .strict() turns any new producer field into a rejected message.
const OrderStrict = z.object({
  id: z.string(),
  amountMinor: z.number().int(),
}).strict();                                    // DON'T

// Tolerant: zod strips unknown keys by default, and optionals get defaults.
const Order = z.object({
  id: z.string(),
  amountMinor: z.number().int(),
  currency: z.string().default("GBP"),          // added in v2, defaulted here
});

// Tolerant AND lossless: passthrough keeps unknown fields so a middleman
// does not silently strip fields its neighbours depend on.
const Envelope = z.object({
  eventId: z.string(),
  eventType: z.string(),
}).passthrough();

# DON'T: offset pagination. Two problems that both get worse with scale.
#   1. OFFSET 100000 makes the database count and discard 100,000 rows.
#   2. If a row is inserted while the client pages, everything shifts by one
#      and the client sees an item twice - or never sees it at all.
"select * from orders order by created_at desc limit 20 offset 100000"


# DO: keyset (cursor) pagination. Constant cost, and stable under writes.
async def list_orders(cursor: str | None, limit: int = 20):
    if cursor:
        after_created, after_id = decode_cursor(cursor)
        rows = await db.fetch(
            """select * from orders
                where (created_at, id) < ($1, $2)   -- the index does the seek
                order by created_at desc, id desc
                limit $3""",
            after_created, after_id, limit,
        )
    else:
        rows = await db.fetch(
            "select * from orders order by created_at desc, id desc limit $1", limit
        )

    # Opaque cursor: encode it so clients cannot build one and depend on
    # its shape - that would make your index choice a public API.
    next_cursor = encode_cursor(rows[-1]) if len(rows) == limit else None
    return {"items": rows, "nextCursor": next_cursor}

// DON'T: offset pagination. Two problems that both get worse with scale.
//   1. OFFSET 100000 makes the database count and discard 100,000 rows.
//   2. If a row is inserted while the client pages, everything shifts by one
//      and the client sees an item twice - or never sees it at all.
"select * from orders order by created_at desc limit 20 offset 100000";

// DO: keyset (cursor) pagination. Constant cost, and stable under writes.
async function listOrders(cursor, limit = 20) {
  const rows = cursor
    ? (await db.query(
        `select * from orders
          where (created_at, id) < ($1, $2)      -- the index does the seek
          order by created_at desc, id desc
          limit $3`,
        [...decodeCursor(cursor), limit],
      )).rows
    : (await db.query(
        "select * from orders order by created_at desc, id desc limit $1", [limit],
      )).rows;

  // Opaque cursor: encode it so clients cannot build one and depend on
  // its shape - that would make your index choice a public API.
  return {
    items: rows,
    nextCursor: rows.length === limit ? encodeCursor(rows.at(-1)) : null,
  };
}

syntax = "proto3";
package shipping.v1;

import "google/protobuf/timestamp.proto";

service Shipping {
  rpc GetShipment(GetShipmentRequest) returns (Shipment);
  rpc WatchShipment(GetShipmentRequest) returns (stream ShipmentEvent);
  rpc UploadScans(stream Scan) returns (UploadSummary);
  rpc Sync(stream ClientMsg) returns (stream ServerMsg);
}

message Shipment {
  string id = 1;
  Status status = 2;
  google.protobuf.Timestamp updated_at = 3;

  enum Status {
    STATUS_UNSPECIFIED = 0;   // always reserve 0 for "unset"
    STATUS_PICKED      = 1;
    STATUS_IN_TRANSIT  = 2;
    STATUS_DELIVERED   = 3;
  }
}

# The deadline is part of the call, not a wrapper around it - and the server
# can read it, so it can abandon work it knows will be too late.
async def get_shipment(shipment_id: str):
    async with grpc.aio.insecure_channel(SHIPPING) as channel:
        stub = shipping_pb2_grpc.ShippingStub(channel)
        try:
            return await stub.GetShipment(
                shipping_pb2.GetShipmentRequest(id=shipment_id),
                timeout=0.3,                       # becomes grpc-timeout on the wire
                metadata=(("traceparent", current_traceparent()),),
            )
        except grpc.aio.AioRpcError as e:
            if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
                raise TooSlow() from e
            raise


# Server side: check the deadline before starting expensive work.
class ShippingServicer(shipping_pb2_grpc.ShippingServicer):
    async def GetShipment(self, request, context):
        if context.time_remaining() < 0.05:
            context.abort(grpc.StatusCode.DEADLINE_EXCEEDED, "not enough time left")
        return await load_shipment(request.id)

// A server-streaming call: one request, an async iterable of responses.
async function watchShipment(id, signal) {
  const call = client.watchShipment(
    { id },
    { deadline: Date.now() + 30_000, signal },     // travels on the wire
  );

  for await (const event of call) {
    handle(event);
    // Resume tokens matter: if this stream drops, you must restart from
    // where you got to, not from the beginning.
    lastSeen = event.sequence;
  }
}

// A DataLoader collects every key requested during one tick of the event
// loop, then issues a single batched query and scatters the results back.
const customerLoader = new DataLoader(async (ids) => {
  const { rows } = await db.query(
    "select * from customers where id = any($1)", [ids],
  );
  const byId = new Map(rows.map((r) => [r.id, r]));
  // The batch function MUST return results in the same order as the keys.
  return ids.map((id) => byId.get(id) ?? null);
});

const resolvers = {
  Order: {
    // Called 50 times; results in ONE query, not 50.
    customer: (order) => customerLoader.load(order.customerId),
  },
};

// Create loaders PER REQUEST. A process-wide loader caches across users,
// which is a data-leak bug, not a performance optimisation.
app.use((req, res, next) => {
  req.loaders = { customer: makeCustomerLoader() };
  next();
});

# strawberry / graphene: the same batching idea, per request.
class CustomerLoader(DataLoader):
    async def batch_load_fn(self, ids: list[str]):
        rows = await db.fetch("select * from customers where id = any($1)", ids)
        by_id = {r["id"]: r for r in rows}
        # The batch function MUST return results in the same order as the keys.
        return [by_id.get(i) for i in ids]


@strawberry.type
class Order:
    customer_id: strawberry.Private[str]

    @strawberry.field
    async def customer(self, info) -> Customer:
        # Called 50 times; results in ONE query, not 50.
        return await info.context["customer_loader"].load(self.customer_id)


# Build loaders PER REQUEST - a process-wide loader caches across users,
# which is a data-leak bug, not a performance optimisation.
async def context_getter() -> dict:
    return {"customer_loader": CustomerLoader()}

outlierDetection:
  consecutive5xxErrors: 5        # 5 in a row and you are out
  interval: 10s                  # evaluation window
  baseEjectionTime: 30s          # grows on repeat offences
  maxEjectionPercent: 50         # NEVER let this reach 100

# The cap is the important line. During a dependency-wide outage every
# backend looks unhealthy, and a balancer that ejects all of them turns a
# degraded service into a total one. Keeping half the pool means you fail
# some requests instead of all of them - and the same reasoning is why
# health checks should not cascade their own dependencies' failures.

Grey failure, and what catches it:

  symptom                        caught by
  -----------------------------  ------------------------------------------
  process dead                   TCP health check
  process wedged, port open      HTTP /healthz
  healthy but slow               least-connections / EWMA / outlier ejection
  healthy but wrong (bad deploy) canary analysis + error-rate alerting
  healthy for some keys only     per-route metrics; a global rate hides it

apiVersion: networking.istio.io/v1
kind: VirtualService
metadata:
  name: orders
spec:
  hosts: [orders.internal]
  http:
    - timeout: 2s                    # hard cap for the whole call
      retries:
        attempts: 2
        perTryTimeout: 700ms         # 2 x 700ms + overhead MUST fit in 2s
        retryOn: 5xx,reset,connect-failure
      route:
        - destination: { host: orders.internal, subset: v2 }
          weight: 5                  # canary: 5% to v2
        - destination: { host: orders.internal, subset: v1 }
          weight: 95
---
apiVersion: networking.istio.io/v1
kind: DestinationRule
metadata:
  name: orders
spec:
  host: orders.internal
  trafficPolicy:
    loadBalancer: { simple: LEAST_REQUEST }
    connectionPool:
      http: { maxRequestsPerConnection: 100 }   # forces periodic re-balance
    outlierDetection:
      consecutive5xxErrors: 5
      baseEjectionTime: 30s
      maxEjectionPercent: 50

What the sidecar does to one call, in order:

  app  --(plain HTTP to localhost, redirected by iptables)-->  sidecar out
       sidecar out: pick endpoint (LEAST_REQUEST, outliers ejected)
                    start timeout + retry budget
                    originate mTLS with a rotated workload identity
                    emit metrics + a client span
         |
         v  (encrypted, authenticated)
       sidecar in:  terminate mTLS, verify peer identity
                    check authorization policy (may checkout call orders?)
                    emit metrics + a server span
  app  <--(plain HTTP from localhost)--  sidecar in

Two extra hops. ~0.5-1 ms of added p50 latency, and roughly 0.5 vCPU
per 1000 rps of sidecar overhead. That is the price list.

DEADLINE_HEADER = "X-Request-Deadline-Unix-Ms"

async def handler(request):
    # Trust an inbound deadline, but never let it exceed our own maximum -
    # a caller can otherwise pin your threads for an hour with one header.
    inbound = request.headers.get(DEADLINE_HEADER)
    ours = time.time() * 1000 + 2000
    deadline_ms = min(float(inbound), ours) if inbound else ours

    async with deadline_scope(deadline_ms) as scope:
        # asyncio cancels the whole scope when the deadline passes, which
        # propagates into every await inside it, including open sockets.
        price, stock = await asyncio.gather(
            call("pricing", scope), call("stock", scope)
        )
    return render(price, stock)


async def call(service, scope):
    remaining = (scope.deadline_ms - time.time() * 1000) / 1000
    if remaining <= 0.05:
        raise DeadlineExceeded(service)      # do not even start
    return await client.get(
        f"http://{service}/x",
        timeout=remaining,
        headers={DEADLINE_HEADER: str(scope.deadline_ms)},
    )

const DEADLINE_HEADER = "x-request-deadline-unix-ms";

async function handler(req, res) {
  // Trust an inbound deadline, but never let it exceed our own maximum -
  // a caller can otherwise pin your event loop with one header.
  const inbound = Number(req.headers[DEADLINE_HEADER]);
  const deadline = Math.min(inbound || Infinity, Date.now() + 2000);

  const ac = new AbortController();
  const timer = setTimeout(() => ac.abort(), deadline - Date.now());
  // If the CLIENT hangs up, cancel everything downstream immediately.
  req.on("close", () => ac.abort());

  try {
    const [price, stock] = await Promise.all([
      call("pricing", deadline, ac.signal),
      call("stock", deadline, ac.signal),
    ]);
    res.json(render(price, stock));
  } finally {
    clearTimeout(timer);
  }
}

async function call(service, deadline, signal) {
  if (deadline - Date.now() <= 50) throw new DeadlineExceeded(service);
  return fetch(`http://${service}/x`, {
    signal,
    headers: { [DEADLINE_HEADER]: String(deadline) },
  });
}

class RetryBudget:
    """Token bucket over recent traffic. Retries spend; requests earn.

    The point is that during a widespread outage - exactly when naive
    retries would triple the load - the budget empties and retries stop.
    """

    def __init__(self, ratio: float = 0.1, min_per_sec: float = 10.0):
        self.ratio = ratio
        self.min_per_sec = min_per_sec
        self.tokens = 0.0

    def on_request(self) -> None:
        self.tokens = min(self.tokens + self.ratio, 100.0)

    def try_retry(self) -> bool:
        if self.tokens < 1.0:
            metrics.increment("retry.budget_exhausted")
            return False
        self.tokens -= 1.0
        return True


async def call(fn, budget: RetryBudget, attempts: int = 3):
    budget.on_request()
    for attempt in range(attempts):
        try:
            return await fn()
        except Retryable:
            if attempt == attempts - 1 or not budget.try_retry():
                raise
            await asyncio.sleep(random.uniform(0, min(10.0, 0.2 * 2 ** attempt)))

// Token bucket over recent traffic. Retries spend; requests earn.
// During a widespread outage - exactly when naive retries would triple
// the load - the budget empties and retries stop on their own.
class RetryBudget {
  #tokens = 0;
  constructor(ratio = 0.1, cap = 100) { this.ratio = ratio; this.cap = cap; }

  onRequest() { this.#tokens = Math.min(this.#tokens + this.ratio, this.cap); }

  tryRetry() {
    if (this.#tokens < 1) { metrics.increment("retry.budget_exhausted"); return false; }
    this.#tokens -= 1;
    return true;
  }
}

async function call(fn, budget, attempts = 3) {
  budget.onRequest();
  for (let attempt = 0; attempt < attempts; attempt += 1) {
    try {
      return await fn();
    } catch (err) {
      if (attempt === attempts - 1 || !isRetryable(err) || !budget.tryRetry()) throw err;
      await sleep(Math.random() * Math.min(10_000, 200 * 2 ** attempt));
    }
  }
}

class CircuitBreaker:
    def __init__(self, *, failure_rate=0.5, min_calls=20, cooldown=30.0,
                 half_open_calls=1):
        self.failure_rate = failure_rate     # trip above this fraction
        self.min_calls = min_calls           # ...but only with enough samples
        self.cooldown = cooldown             # how long OPEN lasts
        self.half_open_calls = half_open_calls
        self.state, self.window, self.opened_at = "closed", deque(maxlen=100), 0.0

    def _allow(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.monotonic() - self.opened_at < self.cooldown:
                return False
            self.state = "half_open"          # let exactly one probe through
            self.probes = 0
        return self.probes < self.half_open_calls

    async def call(self, fn, fallback=None):
        if not self._allow():
            metrics.increment("breaker.rejected")
            if fallback is None:
                raise CircuitOpen()
            return await fallback()           # a breaker without a fallback
                                              # just moves the error around
        try:
            result = await fn()
        except Exception:
            self._record(False)
            raise
        self._record(True)
        return result

    def _record(self, ok: bool) -> None:
        if self.state == "half_open":
            # One bad probe reopens; one good probe closes. Never average here.
            self.state, self.opened_at = ("closed", 0.0) if ok else ("open", time.monotonic())
            self.window.clear()
            return
        self.window.append(ok)
        if len(self.window) >= self.min_calls:
            failures = self.window.count(False) / len(self.window)
            if failures >= self.failure_rate:
                self.state, self.opened_at = "open", time.monotonic()

The four numbers, and how to choose them:

  failure threshold   50% is a good start. Too low and a normal error
                      rate trips you; too high and you never trip.

  minimum calls       20-ish. Without it, 1 failure out of 1 call opens
                      the breaker on a quiet endpoint at 3 a.m.

  cool-down           30 s, DOUBLING on repeated failure. A fixed short
                      cool-down turns the breaker into a synchronised
                      traffic generator against a service trying to boot.

  half-open calls     1. The whole point is to probe with one request
                      rather than with the entire fleet at once.

Scope: ONE BREAKER PER DEPENDENCY (and per endpoint if their failure
modes differ). A global breaker means a sick recommendations service
cuts you off from a healthy payments service.

@app.get("/livez")
async def livez():
    # Deliberately trivial. If this handler runs, the event loop is alive.
    # NEVER touch a dependency here - see the warning above.
    return {"ok": True}


@app.get("/readyz")
async def readyz():
    # Is this instance able to serve? Check things WE own and that make
    # us useless if broken: migrations applied, caches warmed, pool open.
    if not app.state.migrations_done:
        raise HTTPException(503, "migrations pending")
    if app.state.shutting_down:
        raise HTTPException(503, "draining")      # graceful shutdown
    if db_pool.available() == 0 and db_pool.waiters() > 50:
        raise HTTPException(503, "pool saturated")
    return {"ok": True}


@app.get("/debug/deep")
async def deep():
    # Rich dependency status for humans and dashboards - NOT wired to any
    # probe, so a dependency blip can never restart or eject the fleet.
    return {
        "database": await ping(db, timeout=0.2),
        "cache": await ping(redis, timeout=0.1),
        "broker": await ping(kafka, timeout=0.3),
    }

livenessProbe:
  httpGet: { path: /livez, port: 8080 }
  periodSeconds: 10
  failureThreshold: 3            # 30 s of wedged before a restart
  timeoutSeconds: 2              # a probe with no timeout is not a probe

readinessProbe:
  httpGet: { path: /readyz, port: 8080 }
  periodSeconds: 5
  failureThreshold: 2            # come out of rotation faster than you restart

startupProbe:
  httpGet: { path: /readyz, port: 8080 }
  periodSeconds: 5
  failureThreshold: 60           # 5 minutes to boot, THEN liveness applies

lifecycle:
  preStop:
    # Fail readiness, wait for the LB to notice, then let SIGTERM drain.
    # Without this, in-flight requests are killed on every deploy.
    exec: { command: ["sh", "-c", "sleep 15"] }
terminationGracePeriodSeconds: 45

async def consume_forever():
    while True:
        messages = await sqs.receive_message(
            QueueUrl=QUEUE,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=20,           # long polling: fewer empty calls
            AttributeNames=["ApproximateReceiveCount"],
        )
        await asyncio.gather(*(handle_one(m) for m in messages.get("Messages", [])))


async def handle_one(msg):
    receive_count = int(msg["Attributes"]["ApproximateReceiveCount"])
    # Knowing the attempt number lets you log, alert, or degrade on the
    # nth try instead of discovering the loop from a DLQ alarm.
    if receive_count > 3:
        log.warning("message retried repeatedly", message_id=msg["MessageId"])

    keepalive = asyncio.create_task(extend_lease(msg))
    try:
        await process(json.loads(msg["Body"]))     # must be idempotent
        # Ack LAST. Acking first would be at-most-once - see section 23.
        await sqs.delete_message(QueueUrl=QUEUE, ReceiptHandle=msg["ReceiptHandle"])
    except TransientError:
        pass                                       # no ack: it will come back
    finally:
        keepalive.cancel()


async def extend_lease(msg):
    while True:
        await asyncio.sleep(20)
        await sqs.change_message_visibility(
            QueueUrl=QUEUE, ReceiptHandle=msg["ReceiptHandle"], VisibilityTimeout=60,
        )

async function handleOne(msg) {
  const attempt = Number(msg.Attributes.ApproximateReceiveCount);
  // Knowing the attempt number lets you log, alert, or degrade on the
  // nth try instead of discovering the loop from a DLQ alarm.
  if (attempt > 3) log.warn({ messageId: msg.MessageId }, "retried repeatedly");

  // Extend the lease while we work, rather than guessing one large timeout.
  const keepalive = setInterval(
    () => sqs.send(new ChangeMessageVisibilityCommand({
      QueueUrl: QUEUE, ReceiptHandle: msg.ReceiptHandle, VisibilityTimeout: 60,
    })),
    20_000,
  );

  try {
    await process(JSON.parse(msg.Body));          // must be idempotent
    // Ack LAST. Acking first would be at-most-once - see section 23.
    await sqs.send(new DeleteMessageCommand({
      QueueUrl: QUEUE, ReceiptHandle: msg.ReceiptHandle,
    }));
  } catch (err) {
    if (!isTransient(err)) throw err;             // no ack: it comes back
  } finally {
    clearInterval(keepalive);
  }
}

async def handle(message):
    event_id = message.headers["event-id"]

    async with db.transaction():
        try:
            # The dedupe row and the effect commit TOGETHER or not at all.
            await db.execute(
                "insert into processed_events (event_id, handler) values ($1, $2)",
                event_id, "billing",                # scope by handler: two
            )                                       # consumers legitimately
        except UniqueViolation:                     # each process the event
            log.info("duplicate ignored", event_id=event_id)
            return                                  # ack outside, message gone

        await apply_effect(message.payload)         # the real work

    # Ack only after the transaction committed. A crash before this line
    # redelivers the message, and the insert above makes that harmless.
    await message.ack()

async function handle(message) {
  const eventId = message.headers["event-id"];

  await db.transaction(async (tx) => {
    try {
      // The dedupe row and the effect commit TOGETHER or not at all.
      await tx.query(
        "insert into processed_events (event_id, handler) values ($1, $2)",
        [eventId, "billing"],       // scope by handler: two consumers may
      );                            // each legitimately process the event
    } catch (err) {
      if (err.code !== UNIQUE_VIOLATION) throw err;
      log.info({ eventId }, "duplicate ignored");
      return;
    }
    await applyEffect(message.payload);            // the real work
  });

  // Ack only after the transaction committed. A crash before this line
  // redelivers the message, and the insert above makes that harmless.
  await message.ack();
}

# The bound is chosen from a LATENCY target, not from available memory:
# 64 items x 10 ms each = ~640 ms of queueing, which is our budget.
queue: asyncio.Queue = asyncio.Queue(maxsize=64)


async def producer(source):
    async for item in source:
        # put() BLOCKS when the queue is full. That is the backpressure:
        # we stop reading from the source, TCP's window closes, and the
        # pressure travels all the way back to the sender.
        await queue.put(item)


async def producer_with_shedding(source):
    async for item in source:
        try:
            queue.put_nowait(item)
        except asyncio.QueueFull:
            # When you CANNOT push back (an HTTP handler, a fire-and-forget
            # producer), shed explicitly and count it. Silence here is how
            # overload becomes invisible.
            metrics.increment("pipeline.shed")
            raise HTTPException(503, headers={"Retry-After": "1"})


async def consumer():
    while True:
        item = await queue.get()
        try:
            if item.deadline < time.monotonic():
                metrics.increment("pipeline.expired")   # do not do dead work
                continue
            await process(item)
        finally:
            queue.task_done()

// Node streams implement backpressure for you - if you respect the
// return value of write() and use pipeline() rather than manual pumps.
import { pipeline } from "node:stream/promises";

await pipeline(
  source,                                  // readable
  new Transform({
    objectMode: true,
    highWaterMark: 64,                     // the bound, in items
    transform(item, _enc, cb) {
      // Skip work that is already too late rather than doing it slowly.
      if (item.deadline < Date.now()) { metrics.increment("expired"); return cb(); }
      process(item).then((out) => cb(null, out), cb);
    },
  }),
  sink,
);

// The bug this replaces: reading everything into an array first.
//   const all = [];
//   source.on("data", (d) => all.push(d));   // unbounded, OOM under load

CREATE_TABLE = """
create table outbox (
  id            bigserial primary key,        -- ordering for the relay
  event_id      uuid not null unique,         -- consumers dedupe on this
  topic         text not null,
  partition_key text not null,                -- preserves per-entity order
  payload       jsonb not null,
  created_at    timestamptz not null default now(),
  published_at  timestamptz                   -- null = not yet published
);
create index on outbox (id) where published_at is null;   -- partial index:
"""                                                       -- stays tiny


async def relay_batch():
    async with db.transaction():
        rows = await db.fetch(
            """select * from outbox
                where published_at is null
                order by id
                limit 100
                for update skip locked"""      # two relay replicas can run
        )                                      # safely: each takes different
        for row in rows:                       # rows, neither blocks
            await broker.publish(
                row["topic"],
                key=row["partition_key"],      # same key -> same partition
                value=row["payload"],
                headers={"event-id": str(row["event_id"])},
            )
            await db.execute(
                "update outbox set published_at = now() where id = $1", row["id"]
            )
    # A crash between publish and update republishes on restart. That is
    # at-least-once, which is why event-id is on the message.


# Prune, or the table becomes your largest one within a month.
async def prune():
    await db.execute(
        "delete from outbox where published_at < now() - interval '7 days'"
    )

Outbox checklist - the five things people miss:

  1. event_id on the row AND on the message. Without it, consumers
     cannot deduplicate the at-least-once republishing.

  2. partition_key on the row. Publishing with a random key destroys
     the ordering you carefully preserved in the database.

  3. FOR UPDATE SKIP LOCKED (or a single leader relay). Two relay
     replicas polling the same rows will double-publish constantly.

  4. A partial index on unpublished rows. Without it the relay's query
     degrades as the table grows, which is the moment it matters most.

  5. A prune job. An unpruned outbox in a busy service outgrows the
     business tables and slows down the transactions that write to it.

Polling interval: 100-500 ms is usually fine, and adds that much
latency to every event. If you need lower, use CDC (section 28)
to tail the WAL instead of polling.

{
  "specversion": "1.0",
  "id": "9f21b4c8-...",
  "type": "com.acme.orders.placed.v1",
  "source": "/services/orders",
  "subject": "A-4471",
  "time": "2026-09-16T10:31:02.441Z",
  "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",

  "data": {
    "orderId": "A-4471",
    "customerId": "C-1182",
    "totalMinor": 8210,
    "currency": "GBP",
    "items": [{ "sku": "TT-01", "qty": 2 }],
    "occurredAt": "2026-09-16T10:31:02.180Z",
    "version": 3
  }
}

Why each field earns its place:

  id           dedupe key. Non-negotiable: delivery is at-least-once.
  type         includes a VERSION. "orders.placed.v1" lets v2 coexist.
  subject      the entity id - also your partition key (section 24).
  time         when the broker got it. Different from occurredAt!
  occurredAt   when the FACT happened in the domain. This is the one
               business logic should use; the other is for operations.
  version      monotonic per entity -> the guard in section 24.
  traceparent  carries the trace across the broker. Without it, your
               distributed trace stops dead at every queue.

Naming:  PAST TENSE, always. "OrderPlaced", never "PlaceOrder".
         An event is a fact that already happened and cannot be
         rejected. If the publisher expects something specific to
         happen next, it is issuing a command - use a queue and be
         honest about the coupling.

Size:    keep under ~100 KB. For anything bigger use a claim check
         (section 33): put the blob in object storage, send the URL.

STEPS = [
    Step("capture_payment", do=payments.capture,  undo=payments.refund),
    Step("reserve_stock",   do=inventory.reserve, undo=inventory.release),
    Step("schedule_ship",   do=shipping.schedule, undo=None),   # the PIVOT
    Step("send_receipt",    do=email.send,        undo=None),   # retryable
]


async def run(saga_id: str):
    # State lives in the DATABASE, not in memory. The orchestrator can be
    # killed between any two lines and resume exactly where it left off.
    state = await load_or_create(saga_id)

    while state.index < len(STEPS):
        step = STEPS[state.index]
        try:
            # Idempotency key is derived, not random: a retry of this step
            # after a crash reuses the same key. See section 16.
            await step.do(saga_id, idempotency_key=f"{saga_id}:{step.name}")
        except Compensatable as e:
            await compensate(saga_id, state, upto=state.index)
            await mark_failed(saga_id, reason=str(e))
            return
        except Exception:
            # Past the pivot there is no way back: retry forever with
            # backoff, and page a human if it stays stuck.
            await schedule_retry(saga_id, backoff_for(state.attempts))
            return

        state = await advance(saga_id)          # committed after EVERY step

    await mark_complete(saga_id)


async def compensate(saga_id, state, upto):
    # Reverse order, and compensations must be idempotent too - this loop
    # can itself be interrupted and resumed.
    for i in range(upto - 1, -1, -1):
        if STEPS[i].undo:
            await STEPS[i].undo(saga_id, idempotency_key=f"{saga_id}:undo:{i}")
        await record_compensated(saga_id, i)

Orchestration vs choreography - choosing:

  ORCHESTRATION when:            CHOREOGRAPHY when:
    more than 3 steps              2-3 stable steps
    the flow is business-critical  participants are owned by one team
    you need timeouts per step     you add participants often
    a human may intervene          nobody needs a single view of progress
    you need a progress view

The decisive question is not coupling, it is: WHO DO YOU ASK
"why is order A-4471 stuck?" With an orchestrator, one query answers
it. With choreography, you correlate logs across five services and
hope everyone propagated the trace id.

Either way you need:
  - a state machine persisted after every step (survives restarts)
  - idempotent steps AND idempotent compensations
  - a timeout per step, with an explicit "stuck" state
  - a dashboard of in-flight sagas by state and age
  - a manual retry/cancel control for support staff

async def search(query: str, deadline_ms: float):
    budget = (deadline_ms - now_ms()) / 1000
    tasks = {name: asyncio.create_task(shard.search(query, timeout=budget))
             for name, shard in SHARDS.items()}

    done, pending = await asyncio.wait(tasks.values(), timeout=budget)
    for task in pending:
        task.cancel()                          # stop work nobody will read

    results, missing = [], []
    for name, task in tasks.items():
        if task in done and not task.exception():
            results.extend(task.result())
        else:
            missing.append(name)               # degrade, do not fail

    # Tell the caller the answer is incomplete instead of pretending.
    return {"results": rank(results), "partial": bool(missing), "missing": missing}


# Claim check: never put a 40 MB document on the broker. Put a pointer.
async def publish_document(doc_bytes: bytes, doc_id: str):
    key = f"documents/{doc_id}"
    await s3.put_object(Bucket=BUCKET, Key=key, Body=doc_bytes)
    await broker.publish("documents", key=doc_id, value={
        "documentId": doc_id,
        "location": f"s3://{BUCKET}/{key}",    # the claim check
        "sizeBytes": len(doc_bytes),
        "sha256": hashlib.sha256(doc_bytes).hexdigest(),
    })
    # Keeps messages small and fast, keeps sensitive bytes out of broker
    # logs and replicas, and lets access control live with the object.
    # The cost: object lifetime must outlive the topic's retention.

async function search(query, deadlineMs) {
  const budget = deadlineMs - Date.now();
  const ac = new AbortController();
  const timer = setTimeout(() => ac.abort(), budget);

  // allSettled, not all: one rejected shard must not fail the request.
  const entries = Object.entries(SHARDS);
  const settled = await Promise.allSettled(
    entries.map(([, shard]) => shard.search(query, { signal: ac.signal })),
  );
  clearTimeout(timer);

  const results = [];
  const missing = [];
  settled.forEach((r, i) => {
    if (r.status === "fulfilled") results.push(...r.value);
    else missing.push(entries[i][0]);          // degrade, do not fail
  });

  // Tell the caller the answer is incomplete instead of pretending.
  return { results: rank(results), partial: missing.length > 0, missing };
}

POLL_TIMEOUT = 25        # below every proxy's idle timeout (commonly 30-60 s)

@app.get("/updates")
async def updates(after: int, user=Depends(current_user)):
    # Subscribe FIRST, then check. Checking first leaves a window where an
    # event is published after the check but before the wait - and the
    # request then sleeps for 25 s with news sitting in the log.
    waiter = notifier.subscribe(user.id)
    try:
        events = await log.read_after(user.id, after, limit=100)
        if not events:
            try:
                async with asyncio.timeout(POLL_TIMEOUT):
                    await waiter.wait()           # a parked coroutine, not a thread
            except TimeoutError:
                return Response(status_code=204) # empty: the client re-polls
            events = await log.read_after(user.id, after, limit=100)
        return {"events": events, "cursor": events[-1].seq if events else after}
    finally:
        waiter.close()

const POLL_TIMEOUT_MS = 25_000;  // below every proxy's idle timeout

app.get("/updates", requireUser, async (req, res) => {
  const after = Number(req.query.after ?? 0);
  // Subscribe FIRST, then check, so no publish can slip between the two.
  const waiter = notifier.subscribe(req.user.id);
  try {
    let events = await log.readAfter(req.user.id, after, 100);
    if (events.length === 0) {
      const woke = await waiter.wait(POLL_TIMEOUT_MS);   // resolves false on timeout
      if (!woke) return res.sendStatus(204);             // empty: the client re-polls
      events = await log.readAfter(req.user.id, after, 100);
    }
    res.json({ events, cursor: events.at(-1)?.seq ?? after });
  } finally {
    waiter.close();
  }
});

async def poll_forever(session: aiohttp.ClientSession, cursor: int):
    failures = 0
    while True:
        try:
            async with session.get(f"{BASE}/updates", params={"after": cursor},
                                   timeout=aiohttp.ClientTimeout(total=35)) as res:
                if res.status == 200:
                    body = await res.json()
                    for ev in body["events"]:
                        await handle(ev)              # must be idempotent
                    cursor = body["cursor"]
                    await save_cursor(cursor)
                failures = 0                          # a 204 is success too
        except (aiohttp.ClientError, asyncio.TimeoutError):
            failures += 1
            # Full jitter (section 15): a server restart must not bring
            # every client back in the same 100 ms.
            await asyncio.sleep(random.uniform(0, min(30, 0.5 * 2 ** failures)))

async function pollForever(signal) {
  let cursor = Number(localStorage.getItem("cursor") ?? 0);
  let failures = 0;
  while (!signal.aborted) {
    try {
      const res = await fetch(`/updates?after=${cursor}`, { signal });
      if (res.status === 200) {
        const { events, cursor: next } = await res.json();
        events.forEach(render);                  // must be idempotent
        cursor = next;
        localStorage.setItem("cursor", String(cursor));
      }
      failures = 0;                              // a 204 is success too
    } catch (err) {
      if (signal.aborted) return;
      failures += 1;
      // Full jitter (section 15): a server restart must not bring
      // every client back in the same 100 ms.
      await sleep(Math.random() * Math.min(30_000, 500 * 2 ** failures));
    }
  }
}

KEEPALIVE = 15   # seconds; comfortably inside every idle timeout on the path

@app.get("/stream")
async def stream(request: Request, user=Depends(current_user),
                 last_event_id: str | None = Header(None)):
    async def gen():
        last_seq = int(last_event_id or 0)
        # Subscribe BEFORE reading the backlog; the subscription buffers
        # anything published while the replay is running.
        live = bus.subscribe(f"user:{user.id}")
        try:
            # Resumption: the browser sent Last-Event-ID automatically.
            for ev in await backlog.since(user.id, last_seq):
                yield sse(ev)
                last_seq = ev.seq

            while not await request.is_disconnected():
                try:
                    ev = await asyncio.wait_for(live.next(), timeout=KEEPALIVE)
                except asyncio.TimeoutError:
                    yield ": keepalive\n\n"           # keeps NATs and LBs from reaping us
                    continue
                if ev.seq <= last_seq:                # already sent during replay
                    continue
                yield sse(ev)
                last_seq = ev.seq
        finally:
            await live.close()

    return StreamingResponse(gen(), media_type="text/event-stream", headers={
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",     # tell nginx not to buffer the stream
    })

def sse(ev) -> str:
    # id: enables resumption. retry: sets the client's reconnect delay -
    # jittered per connection so a deploy does not synchronise everyone.
    retry_ms = random.randint(2000, 8000)
    return f"id: {ev.seq}\nevent: {ev.type}\nretry: {retry_ms}\ndata: {json.dumps(ev.data)}\n\n"

const KEEPALIVE_MS = 15_000;
const MAX_BUFFER = 1 << 20;      // 1 MiB queued for one client is "not keeping up"

app.get("/stream", requireUser, async (req, res) => {
  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    "X-Accel-Buffering": "no",   // tell nginx not to buffer the stream
  });

  let lastSeq = Number(req.get("Last-Event-ID") ?? 0);
  const send = (ev) => {
    if (ev.seq <= lastSeq) return;                 // already sent during replay
    lastSeq = ev.seq;
    const retry = 2000 + Math.floor(Math.random() * 6000);   // jittered per connection
    res.write(`id: ${ev.seq}\nevent: ${ev.type}\nretry: ${retry}\ndata: ${JSON.stringify(ev.data)}\n\n`);
    // A slow client must not grow an unbounded buffer (section 26). Cut it
    // loose; it reconnects with Last-Event-ID and loses nothing.
    if (res.writableLength > MAX_BUFFER) res.end();
  };

  // Subscribe BEFORE replaying, and hold live events until the replay ends.
  const pending = [];
  let replaying = true;
  const live = await bus.subscribe(`user:${req.user.id}`, (ev) =>
    replaying ? pending.push(ev) : send(ev));
  for (const ev of await backlog.since(req.user.id, lastSeq)) send(ev);
  replaying = false;
  pending.forEach(send);

  const keepalive = setInterval(() => res.write(": keepalive\n\n"), KEEPALIVE_MS);
  req.on("close", () => {
    clearInterval(keepalive);
    live.unsubscribe();
  });
});

const es = new EventSource("/stream", { withCredentials: true });

es.addEventListener("price", (e) => render(JSON.parse(e.data)));

es.onerror = () => {
  // CONNECTING: the browser is already retrying and will send Last-Event-ID.
  // CLOSED: it gave up - a non-200 status or the wrong Content-Type stops
  // EventSource for good, and reconnecting is now your job.
  if (es.readyState === EventSource.CLOSED) scheduleReconnect();
};

@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
    # Browsers send cookies on cross-site WebSocket handshakes and CORS does
    # not apply, so an unchecked Origin is a cross-site hijack (CSWSH).
    if ws.headers.get("origin") not in ALLOWED_ORIGINS:
        await ws.close(code=1008)                     # policy violation
        return
    user = await authenticate(ws.cookies.get("session"))
    if user is None:
        await ws.close(code=1008)
        return
    await ws.accept(subprotocol="chat.v2")

    # Backplane: the user may be messaged from any node in the fleet.
    sub = bus.subscribe(f"user:{user.id}", max_pending=1000)   # bounded buffer

    async def pump_out():
        async for ev in sub:
            await ws.send_json({"type": ev.type, "seq": ev.seq, "data": ev.data})

    out = asyncio.create_task(pump_out())
    try:
        async for msg in ws.iter_json():
            if msg["type"] == "send":
                # Same rules as any write API: validate, authorise, dedupe.
                await chat.post(user, msg["room"], msg["text"], key=msg["id"])
                await ws.send_json({"type": "ack", "id": msg["id"]})
            elif msg["type"] == "resume":
                for ev in await backlog.since(user.id, msg["after"]):
                    await ws.send_json({"type": ev.type, "seq": ev.seq, "data": ev.data})
    except WebSocketDisconnect:
        pass
    finally:
        out.cancel()
        await sub.close()

# Heartbeats: run uvicorn with --ws-ping-interval 20 --ws-ping-timeout 20,
# so a silently dropped connection is detected in under a minute.

import { WebSocketServer } from "ws";

const wss = new WebSocketServer({ noServer: true });
const MAX_BUFFER = 1 << 20;

server.on("upgrade", async (req, socket, head) => {
  // The handshake is the only HTTP moment you get: check Origin and auth here.
  // Browsers send cookies cross-site and CORS does not apply (CSWSH).
  const user = ALLOWED_ORIGINS.has(req.headers.origin) ? await authenticate(req) : null;
  if (!user) {
    socket.end("HTTP/1.1 401 Unauthorized\r\n\r\n");
    return;
  }
  wss.handleUpgrade(req, socket, head, (ws) => onConnection(ws, user));
});

function onConnection(ws, user) {
  ws.isAlive = true;
  ws.on("pong", () => { ws.isAlive = true; });

  // Backplane: the user may be messaged from any node in the fleet.
  const sub = bus.subscribe(`user:${user.id}`, (ev) => {
    // bufferedAmount is bytes queued for a client that is not reading.
    if (ws.bufferedAmount > MAX_BUFFER) return ws.close(1013, "too slow");
    ws.send(JSON.stringify({ type: ev.type, seq: ev.seq, data: ev.data }));
  });

  ws.on("message", async (raw) => {
    const msg = JSON.parse(raw);
    if (msg.type === "send") {
      // Same rules as any write API: validate, authorise, dedupe.
      await chat.post(user, msg.room, msg.text, { key: msg.id });
      ws.send(JSON.stringify({ type: "ack", id: msg.id }));
    } else if (msg.type === "resume") {
      for (const ev of await backlog.since(user.id, msg.after)) {
        ws.send(JSON.stringify({ type: ev.type, seq: ev.seq, data: ev.data }));
      }
    }
  });
  ws.on("close", () => sub.unsubscribe());
}

// Heartbeat: a NAT or LB drops idle sockets silently, so prove liveness.
setInterval(() => {
  for (const ws of wss.clients) {
    if (!ws.isAlive) { ws.terminate(); continue; }   // missed the last ping
    ws.isAlive = false;
    ws.ping();
  }
}, 30_000);

function connect(state) {
  const ws = new WebSocket("wss://chat.example.com/ws", "chat.v2");

  ws.onopen = () => {
    state.attempt = 0;
    ws.send(JSON.stringify({ type: "resume", after: state.lastSeq }));
    // A message sent but never acked may or may not have arrived (section 1).
    // Resend it with the SAME id and let the server deduplicate.
    for (const m of state.unacked.values()) ws.send(JSON.stringify(m));
  };

  ws.onmessage = (e) => {
    const msg = JSON.parse(e.data);
    if (msg.type === "ack") state.unacked.delete(msg.id);
    else if (msg.seq > state.lastSeq) {
      state.lastSeq = msg.seq;
      render(msg);
    }
  };

  ws.onclose = (e) => {
    if (e.code === 1008) return showLogin();             // policy: do not retry
    const cap = Math.min(30_000, 1000 * 2 ** state.attempt++);
    setTimeout(() => connect(state), Math.random() * cap); // full jitter
  };

  state.ws = ws;
}

function sendChat(state, room, text) {
  const msg = { type: "send", id: crypto.randomUUID(), room, text };
  state.unacked.set(msg.id, msg);                       // until the server acks
  if (state.ws.readyState === WebSocket.OPEN) state.ws.send(JSON.stringify(msg));
}

# --- Sender -------------------------------------------------------------
def sign(secret: str, body: bytes, timestamp: int) -> str:
    # Sign the TIMESTAMP + RAW BODY. Signing only the body lets anyone who
    # captures one delivery replay it forever.
    mac = hmac.new(secret.encode(), f"{timestamp}.".encode() + body, hashlib.sha256)
    return f"t={timestamp},v1={mac.hexdigest()}"


# --- Receiver -----------------------------------------------------------
@app.post("/webhooks/acme")
async def receive(request: Request):
    raw = await request.body()            # RAW bytes - re-serialising the
                                          # parsed JSON changes the bytes
                                          # and the signature will not match
    header = request.headers["X-Acme-Signature"]
    ts, sig = parse(header)

    # Reject old deliveries so a captured request cannot be replayed later.
    if abs(time.time() - ts) > 300:
        raise HTTPException(400, "timestamp outside tolerance")

    expected = hmac.new(SECRET.encode(), f"{ts}.".encode() + raw, hashlib.sha256)
    # constant-time compare: == leaks the signature one byte at a time
    if not hmac.compare_digest(expected.hexdigest(), sig):
        raise HTTPException(401, "bad signature")

    # Return 2xx FAST. Do the work asynchronously, or the sender's timeout
    # becomes your retry storm.
    await inbox.enqueue(json.loads(raw))
    return {"received": True}

Sending webhooks - the production checklist:

  [ ] never send inline from the request that caused the event; a
      customer's slow endpoint would be inside your own latency
  [ ] sign timestamp + raw body with HMAC-SHA256, per-endpoint secret
  [ ] support TWO active secrets so customers can rotate without downtime
  [ ] include an event id; delivery is at-least-once and they must dedupe
  [ ] include a sequence number or occurredAt; retries mean a refund can
      arrive BEFORE the payment it refunds
  [ ] retry with exponential backoff and jitter over hours or days -
      receivers are down for deploys, not for 200 ms
  [ ] cap the per-attempt timeout hard (5-10 s) and the payload size
  [ ] after the window: disable the endpoint, email a human, offer replay
  [ ] expose a delivery log with request, response and manual re-send
  [ ] "fetch, don't trust" for sensitive data: send an id and let the
      receiver call your authenticated API for the payload
  [ ] SSRF: validate customer URLs - block private ranges, link-local
      169.254.169.254 (cloud metadata!), and re-validate after redirects

Receiving webhooks:
  [ ] verify the signature on the RAW body, in constant time
  [ ] reject stale timestamps
  [ ] deduplicate on the event id
  [ ] return 2xx immediately, process asynchronously
  [ ] tolerate out-of-order arrival; use occurredAt, not receipt order

@app.post("/reports", status_code=202)
async def start_report(req: ReportRequest, response: Response,
                       idempotency_key: str = Header(...)):
    # Starting a job is a write like any other: a retried POST must return
    # the SAME job, not start a second one. A unique constraint on the key
    # settles two concurrent retries (section 16).
    job = await jobs.get_by_key(idempotency_key)
    if job is None:
        async with db.transaction():
            job = await jobs.create(key=idempotency_key, kind="report",
                                    params=req.model_dump(), status="queued")
            # The job row and its event commit together (section 27).
            await outbox.append("report.requested", {"jobId": job.id})
    response.headers["Location"] = f"/jobs/{job.id}"
    response.headers["Retry-After"] = "5"
    return {"jobId": job.id, "status": job.status}


@app.get("/jobs/{job_id}")
async def job_status(job_id: str, response: Response):
    job = await jobs.get(job_id)
    if job is None:
        raise HTTPException(404)
    if job.expired:
        raise HTTPException(410, "job expired")
    if job.status == "succeeded":
        # 303: the answer lives elsewhere; the client follows with a GET.
        return RedirectResponse(f"/reports/{job.result_id}", status_code=303)
    if job.status in ("queued", "running"):
        response.headers["Retry-After"] = "10"
    # A FAILED job is still a successful status request: 200, error in the
    # body, so clients and retry layers never confuse the two.
    return {"jobId": job.id, "status": job.status,
            "progress": job.progress, "error": job.error}

app.post("/reports", async (req, res) => {
  const key = req.get("Idempotency-Key");
  if (!key) return res.status(400).json({ error: "Idempotency-Key required" });

  // Starting a job is a write like any other: a retried POST must return
  // the SAME job. A unique constraint on the key settles concurrent retries.
  let job = await jobs.getByKey(key);
  if (!job) {
    job = await db.transaction(async (tx) => {
      const created = await jobs.create(tx, { key, kind: "report", params: req.body, status: "queued" });
      await outbox.append(tx, "report.requested", { jobId: created.id });   // section 27
      return created;
    });
  }
  res.status(202).location(`/jobs/${job.id}`).set("Retry-After", "5")
    .json({ jobId: job.id, status: job.status });
});

app.get("/jobs/:id", async (req, res) => {
  const job = await jobs.get(req.params.id);
  if (!job) return res.sendStatus(404);
  if (job.expired) return res.status(410).json({ error: "job expired" });
  if (job.status === "succeeded") return res.redirect(303, `/reports/${job.resultId}`);
  if (job.status === "queued" || job.status === "running") res.set("Retry-After", "10");
  // A FAILED job is still a successful status request: 200, error in the body.
  res.json({ jobId: job.id, status: job.status, progress: job.progress, error: job.error });
});

async function runReport(params) {
  const key = crypto.randomUUID();                 // one key per INTENT, reused on retry
  const start = await fetch("/reports", {
    method: "POST",
    headers: { "Content-Type": "application/json", "Idempotency-Key": key },
    body: JSON.stringify(params),
  });
  const statusUrl = start.headers.get("Location");
  let wait = Number(start.headers.get("Retry-After") ?? 5);
  const giveUpAt = Date.now() + 15 * 60_000;

  while (Date.now() < giveUpAt) {
    await sleep(wait * 1000);
    const res = await fetch(statusUrl);            // fetch follows the 303 itself
    if (res.redirected) return res.json();         // landed on /reports/7: done
    const job = await res.json();
    if (job.status === "failed") throw new JobFailed(job.error);
    wait = Number(res.headers.get("Retry-After") ?? Math.min(wait * 2, 60));
  }
  // Giving up waiting is not cancelling: the job is still running.
  throw new Error(`still running - check ${statusUrl} later`);
}

class RpcClient:
    def __init__(self, channel):
        self.channel = channel
        self.pending: dict[str, asyncio.Future] = {}

    async def start(self):
        # One exclusive, auto-deleted reply queue PER INSTANCE, so a reply
        # always reaches the process that holds the waiting future.
        self.reply_queue = await self.channel.declare_queue(exclusive=True, auto_delete=True)
        await self.reply_queue.consume(self.on_reply, no_ack=True)

    async def on_reply(self, msg):
        fut = self.pending.pop(msg.correlation_id, None)
        if fut is None:
            # A late reply for a request we already gave up on. The work may
            # still have happened - which is why requests must be idempotent.
            log.info("late reply discarded", correlation_id=msg.correlation_id)
            return
        fut.set_result(json.loads(msg.body))

    async def call(self, queue: str, payload: dict, timeout: float = 2.0):
        corr_id = str(uuid.uuid4())
        fut = asyncio.get_running_loop().create_future()
        self.pending[corr_id] = fut
        await self.channel.default_exchange.publish(
            aio_pika.Message(
                body=json.dumps(payload).encode(),
                correlation_id=corr_id,
                reply_to=self.reply_queue.name,
                expiration=timeout,          # a stale request is not worth doing
            ),
            routing_key=queue,
        )
        try:
            return await asyncio.wait_for(fut, timeout)
        finally:
            self.pending.pop(corr_id, None)  # never leak an entry on timeout

export async function createRpcClient(channel) {
  const pending = new Map();
  // One exclusive reply queue per instance: replies return to the process
  // that holds the waiting promise.
  const { queue: replyTo } = await channel.assertQueue("", { exclusive: true, autoDelete: true });

  await channel.consume(replyTo, (msg) => {
    const id = msg.properties.correlationId;
    const entry = pending.get(id);
    if (!entry) return log.info({ id }, "late reply discarded");   // we timed out
    pending.delete(id);
    clearTimeout(entry.timer);
    entry.resolve(JSON.parse(msg.content.toString()));
  }, { noAck: true });

  return function call(queue, payload, timeoutMs = 2000) {
    const correlationId = randomUUID();
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        pending.delete(correlationId);             // never leak an entry
        reject(new Error(`rpc to ${queue} timed out`));
      }, timeoutMs);
      pending.set(correlationId, { resolve, timer });
      channel.sendToQueue(queue, Buffer.from(JSON.stringify(payload)), {
        correlationId,
        replyTo,
        expiration: String(timeoutMs),            // a stale request is not worth doing
      });
    });
  };
}

The responder's half - three rules:

  1. publish the reply to msg.reply_to with msg.correlation_id copied over
  2. ack the REQUEST only after the reply is published; acking first means
     a crash in between loses the reply and the requester waits for nothing
  3. be idempotent on the request id - a requester that timed out will
     retry, and the first attempt may already have done the work

# WRONG - the classic broken distributed lock.
if await redis.set("lock:invoice", me, nx=True, ex=30):
    await charge_customer()          # a 35 s GC pause here means the lease
    await redis.delete("lock:invoice")  # expired, someone else acquired it,
                                        # and we BOTH charge the customer


# RIGHT - a monotonically increasing fencing token, checked at the resource.
token = await lock_service.acquire("invoice")      # returns 41, then 42, ...

# The RESOURCE enforces the ordering. It does not matter how long we paused
# or what we believe: a write carrying an older token is rejected.
await db.execute(
    """update invoices
          set status = 'charged', fence = $1
        where id = $2 and fence < $1""",
    token, invoice_id,
)
# A stale holder (token 41) arriving after the new holder (42) has written
# matches zero rows. No double charge, no reliance on wall-clock timing.

What to use, and what it gives you:

  etcd / ZooKeeper / Consul   leader election, config, service registry.
                              Linearizable, and they hand you fencing
                              tokens (etcd revision, ZK zxid) for free.

  Kafka consumer groups       partition assignment and leadership, done
                              for you. Most teams never need more.

  A database row              "select ... for update" or an advisory
                              lock is a perfectly good lock when all
                              contenders already share that database.

  Redis SETNX                 a LEASE, not a lock. Fine for "probably
                              only one cron runs this" - never for
                              correctness without a fencing token.

The recurring lesson: you cannot guarantee mutual exclusion with
timeouts alone, because you cannot distinguish a crashed holder from
a paused one (section 19). You CAN make the resource reject writes
from anyone but the current holder. Push the check to the resource.

from opentelemetry import trace, propagate

tracer = trace.get_tracer(__name__)

# HTTP is the easy case - auto-instrumentation handles traceparent.
# The broker is where traces usually break, because nothing injects
# the context into message headers unless you do it explicitly.
async def publish(topic: str, key: str, payload: dict):
    with tracer.start_as_current_span(f"publish {topic}", kind=SpanKind.PRODUCER) as span:
        span.set_attribute("messaging.destination.name", topic)
        headers: dict[str, str] = {}
        propagate.inject(headers)                # writes traceparent
        await broker.send(topic, key=key, value=payload, headers=headers)


async def consume(message):
    # Link, do not nest: a consumer may process a batch from many traces,
    # and the consumer span can outlive the producer's request by hours.
    ctx = propagate.extract(message.headers)
    with tracer.start_as_current_span(
        "process order-events", context=ctx, kind=SpanKind.CONSUMER
    ) as span:
        span.set_attribute("messaging.message.id", message.headers["event-id"])
        # Put the trace id in every log line: one id, five services.
        with log.contextualize(trace_id=format(span.get_span_context().trace_id, "032x")):
            await handle(message)

import { context, propagation, trace, SpanKind } from "@opentelemetry/api";

const tracer = trace.getTracer("orders");

// HTTP is the easy case. The broker is where traces usually break,
// because nothing injects the context into message headers for you.
async function publish(topic, key, payload) {
  await tracer.startActiveSpan(`publish ${topic}`, { kind: SpanKind.PRODUCER },
    async (span) => {
      const headers = {};
      propagation.inject(context.active(), headers);   // writes traceparent
      await broker.send({ topic, messages: [{ key, value: payload, headers }] });
      span.end();
    });
}

async function consume(message) {
  // Link, do not nest: a consumer may process a batch from many traces.
  const ctx = propagation.extract(context.active(), message.headers);
  await context.with(ctx, async () => {
    await tracer.startActiveSpan("process order-events",
      { kind: SpanKind.CONSUMER }, async (span) => {
        // Put the trace id in every log line: one id, five services.
        await logger.child({ traceId: span.spanContext().traceId })
          .runWith(() => handle(message));
        span.end();
      });
  });
}

# 1. Unit: assert the RESILIENCE behaviour, not just the happy path.
async def test_breaker_opens_and_serves_fallback():
    dependency = FlakyStub(fail_times=25)
    client = Resilient(dependency, breaker=CircuitBreaker(min_calls=20))

    for _ in range(25):
        with contextlib.suppress(Exception):
            await client.get_price("SKU-1")

    assert client.breaker.state == "open"
    # The point of the breaker is the fallback, so assert on that.
    assert await client.get_price("SKU-1", fallback=cached) == cached_price
    # And assert we stopped CALLING - that is what protects the dependency.
    assert dependency.calls == 25


# 2. Integration: prove duplicates are harmless, because they WILL happen.
async def test_handler_is_idempotent():
    event = make_event(id="evt-1", amount=8210)
    await handle(event)
    await handle(event)                       # redelivery
    await handle(event)                       # rebalance
    assert await ledger.total() == 8210       # charged ONCE


# 3. Integration: prove the outbox survives a crash between the two steps.
async def test_outbox_republishes_after_relay_crash():
    await place_order(order)
    broker.fail_next_publish()                # crash after publish, before
    await relay_batch()                       # marking published
    await relay_batch()
    assert broker.published_count(order.id) == 2      # at-least-once, and
    assert consumer_applied_count(order.id) == 1      # deduped downstream

# 4. Mesh-level fault injection: real failures, in a real environment,
# scoped to traffic carrying one header - so only your test client sees it.
apiVersion: networking.istio.io/v1
kind: VirtualService
metadata:
  name: pricing-chaos
spec:
  hosts: [pricing.internal]
  http:
    - match:
        - headers:
            x-chaos-experiment: { exact: "pricing-latency" }
      fault:
        delay:
          percentage: { value: 100 }
          fixedDelay: 5s              # does the caller's timeout fire?
        abort:
          percentage: { value: 10 }
          httpStatus: 503             # does the breaker open? fallback serve?
      route:
        - destination: { host: pricing.internal }
    - route:
        - destination: { host: pricing.internal }   # everyone else: normal