Communication: RPC, messaging and failure

Machines in a distributed system cooperate only by sending messages. Everything else, from replication to consensus, is built on top of that. CMPUT-313 explained how messages travel: packets, routing, TCP connections and sockets (its lessons 5 to 7). This lesson moves up a level, to the abstractions programmers actually use to talk to remote services, and to the failure behaviour those abstractions inherit from the network. The two dominant styles are request–response, usually as remote procedure calls, and asynchronous messaging through queues and logs. Both must deal with the same hard question raised in lesson 1: when a reply does not come, what should the sender do?

Remote procedure calls

A remote procedure call (RPC) makes a request to another process look like a function call. The idea was published by Birrell and Nelson in 1984, based on work at Xerox PARC, and it has dominated service-to-service communication ever since. The programmer writes something like balance = ledger.get_balance(account). Behind that line, a client stub marshals the arguments into bytes, sends them over the network, waits for a reply, and unmarshals the result. On the server, a matching stub receives the bytes, decodes them, calls the real procedure and sends back its result. Stubs are generated from an interface definition, so client and server agree on message formats even when they are written in different languages.

Modern RPC frameworks include gRPC, which Google released as open source in 2015 and which uses Protocol Buffers for messages and HTTP/2 for transport; Apache Thrift, from Facebook; and many in-house systems. A Protocol Buffers interface for Ledgerline's ledger might look like this.

syntax = "proto3";
package ledgerline.v1;

// A payment is recorded at most once per idempotency key.
message RecordPaymentRequest {
  string idempotency_key = 1;   // chosen by the client, reused on every retry
  string account_id      = 2;
  int64  amount_cents    = 3;
  string currency        = 4;   // "CAD" or "USD"
}

message RecordPaymentReply {
  string payment_id = 1;
  int64  balance_cents = 2;
}

service Ledger {
  rpc RecordPayment (RecordPaymentRequest) returns (RecordPaymentReply);
  rpc GetBalance    (GetBalanceRequest)    returns (GetBalanceReply);
}

message GetBalanceRequest { string account_id = 1; }
message GetBalanceReply   { int64 balance_cents = 1; }

The field numbers in the message definitions matter for evolution. Protocol Buffers identify fields on the wire by number, not name, so a newer server can add field 5 and older clients will simply skip it, while a newer client can omit a field that an older server treats as having its default value. This backward and forward compatibility is essential in a distributed system, because you can never upgrade every machine at once: during a rolling upgrade, old and new versions of a service talk to each other for hours. Changing the meaning of an existing field number, or reusing a deleted one, breaks that compatibility and is a classic cause of outages.

RPC is convenient, but lesson 1 warned that remote calls are not local calls. Five differences matter. Latency is thousands of times larger. Arguments must be copied, so passing a pointer or a reference to a large shared structure makes no sense. The call can fail in ways a local call cannot, and the failure may leave the outcome unknown. The server may be running a different version of the code. And the server is shared, so it may be overloaded by other clients. A good RPC interface acknowledges all five: it sets deadlines, it designs operations to be safe to retry, and it versions its messages.

Call semantics: at most once, at least once, exactly once

When a client receives no reply, it can give up or try again. These choices lead to three possible guarantees, usually called invocation semantics. With at-most-once semantics the client never retries, or the server detects and discards duplicates, so the operation runs zero or one times. With at-least-once semantics the client retries until it gets a reply, so the operation runs one or more times. Exactly-once semantics, running the operation precisely once despite failures, is what programmers usually want, but it cannot be achieved by the network layer alone: if the server may crash after executing a request and before replying, no protocol between client and server can make the client certain whether to retry.

What practical systems achieve is effectively-once execution: at-least-once delivery combined with operations whose repeated execution has the same effect as a single execution. An operation with that property is idempotent. Setting a value is idempotent: setting a balance to 100 twice leaves it at 100. Adding to a value is not: adding 40 twice adds 80. Deleting a record by key is idempotent; appending a line to a file is not. When an operation is not naturally idempotent, it can be made so with an idempotency key: the client attaches a unique identifier to each logical request and reuses it on every retry, and the server records which identifiers it has already processed, together with the reply it sent, so that a retry is answered from that record instead of being executed again. Payment processors such as Stripe expose exactly this mechanism to their clients.

# Idempotency keys: a server that remembers request identifiers can absorb retries safely.
import random
random.seed(7)

class Ledger:
    def __init__(self, dedupe):
        self.balance, self.seen, self.dedupe = 0, {}, dedupe
    def record_payment(self, request_id, amount):
        if self.dedupe and request_id in self.seen:
            return self.seen[request_id]          # replay the original reply
        self.balance += amount
        reply = ("ok", self.balance)
        self.seen[request_id] = reply
        return reply

def client(ledger, payments, p_lost_reply=0.3, max_tries=5):
    for rid, amount in payments:
        for _ in range(max_tries):
            reply = ledger.record_payment(rid, amount)   # the server always applies it...
            if random.random() >= p_lost_reply:          # ...but the reply may be lost
                break                                     # reply arrived: stop retrying

payments = [(f"inv-{i}", 40) for i in range(100)]          # 100 payments of 40 each
for dedupe in (False, True):
    random.seed(7)
    led = Ledger(dedupe)
    client(led, payments)
    print(f"deduplication {'on ' if dedupe else 'off'}: balance = {led.balance} (correct: {100 * 40})")

deduplication off: balance = 6360 (correct: 4000)
deduplication on : balance = 4000 (correct: 4000)

Deduplication has costs and subtleties. The server must store the keys durably, in the same transaction as the effect, otherwise a crash between applying the payment and recording the key reopens the window for duplicates. It must decide how long to keep keys, since storing them forever is expensive; Stripe documents keeping them for at least 24 hours. And if the server is replicated, every replica must see the same set of processed keys, which is a replication problem of the kind lessons 6 to 9 solve.

Timeouts, deadlines and retries

Every remote call needs a limit on how long to wait. A per-call timeout is the simplest form. A better form, which gRPC builds in, is a deadline: an absolute time by which the whole user request must finish, propagated from service to service. If a front end gives a request 500 ms, and has used 120 ms before calling the ledger, the ledger call carries a deadline 380 ms away, and the ledger in turn passes the remainder to anything it calls. Deadlines stop work on requests that nobody is waiting for any more, which matters greatly under overload.

Retrying is the natural response to a failed call, but naive retries cause some of the worst outages in practice. Imagine a server that restarts after a crash. Every client whose call failed retries at once, and the server, still warming up, is overwhelmed and fails again; the clients retry again, in step, and the cycle repeats. This is the thundering herd, and a related problem, the retry storm, arises when each layer of a call chain retries independently: if three layers each make up to three attempts, a single failing back end can receive 27 times the normal traffic. The remedies are exponential backoff, which doubles the wait after each failure; jitter, which randomizes the wait so clients spread out; limits on the number of attempts; and retry budgets, which cap retries at a small fraction of normal traffic.

# Retries after an outage: synchronized retries versus exponential backoff with jitter.
# 1000 clients all fail at t = 0 when a server restarts. Each 10 ms tick the server has
# 100 units of work. Serving a request costs 1 unit; even rejecting one costs 0.2 units
# (accepting the connection, parsing, replying "busy"), so a big enough crowd can leave
# no capacity for real work at all.
import random
random.seed(450)
CLIENTS, CAPACITY, REJECT_COST, TICK, LIMIT = 1000, 100, 0.2, 10, 10_000

def simulate(strategy):
    next_try = {c: 0.0 for c in range(CLIENTS)}
    attempt = {c: 0 for c in range(CLIENTS)}
    done, t, wasted, peak = set(), 0, 0, 0
    while len(done) < CLIENTS and t < LIMIT:
        arrivals = [c for c in next_try if c not in done and t <= next_try[c] < t + TICK]
        n = len(arrivals); peak = max(peak, n)
        # serve s requests so that s + REJECT_COST * (n - s) <= CAPACITY
        s = max(0, min(n, int((CAPACITY - REJECT_COST * n) / (1 - REJECT_COST))))
        random.shuffle(arrivals)
        done.update(arrivals[:s])
        for c in arrivals[s:]:
            wasted += 1
            attempt[c] += 1
            base = 10 * 2 ** min(attempt[c], 8)        # 20, 40, 80 ... ms, capped at 2560
            delay = {"fixed": 10, "backoff": base}.get(strategy) or random.uniform(0, base)
            next_try[c] = t + TICK + delay
        t += TICK
    status = f"all served after {t:5d} ms" if len(done) == CLIENTS else f"only {len(done):4d} served after {t} ms"
    return status, peak, wasted

for s in ("fixed", "backoff", "jitter"):
    status, peak, wasted = simulate(s)
    print(f"{s:8s} {status:32s} peak arrivals/tick {peak:5d}   rejected attempts {wasted:6d}")

fixed    only    0 served after 10000 ms  peak arrivals/tick  1000   rejected attempts 500000
backoff  only    0 served after 10000 ms  peak arrivals/tick  1000   rejected attempts  10000
jitter   all served after   390 ms        peak arrivals/tick  1000   rejected attempts   3091

The result is starker than a slowdown. Because rejecting a request also costs the server something, a crowd of a thousand simultaneous attempts leaves no capacity for real work. With a fixed retry delay the whole crowd returns together every 10 ms, and the server never serves anyone: this is congestion collapse, a metastable failure that persists although the server is healthy. Exponential backoff without randomness does not help, because clients that failed together compute the same delays and keep returning together. Full jitter, where each client waits a random time between zero and its backoff ceiling, breaks up the crowd; within a few rounds the arrival rate falls below capacity and everyone is served. Marc Brooker's analysis of these strategies for Amazon Web Services reached the same conclusion, and most cloud SDKs now use backoff with jitter by default.

Asynchronous messaging

RPC couples the caller and the callee in time: both must be running and reachable during the call. Messaging decouples them. A producer puts a message into a queue or log managed by a broker, and consumers read it whenever they are ready. If the consumer is down, messages wait. If producers briefly outpace consumers, the queue absorbs the burst. Messaging also decouples them in space: the producer does not need to know who consumes the message, or how many consumers there are.

Apache Kafka, created at LinkedIn around 2010, popularized the log abstraction. A Kafka topic is divided into partitions, each an append-only, ordered sequence of records stored durably and replicated across brokers. Producers append records; consumers read sequentially and record their position, called an offset. Because the log is retained, a new consumer can replay history from the beginning, and a consumer that crashes resumes from its last committed offset. Order is guaranteed within a partition but not across partitions, so records that must stay in order, such as all events for one account, are given the same partition key. Kafka originally relied on ZooKeeper to store its metadata; it later built its own Raft-based metadata quorum, called KRaft, and Kafka 4.0, released in March 2025, removed ZooKeeper support entirely. We return to Raft in lesson 9.

Messaging systems face the same delivery-semantics question as RPC. A consumer that processes a message and then crashes before committing its offset will process the message again after restarting, so the usual guarantee is at-least-once, and consumers must again be idempotent. Kafka offers transactions that tie consuming, producing and committing offsets together, giving exactly-once processing within Kafka itself; lesson 12 discusses what that does and does not guarantee once side effects leave the system.

Backpressure and overload

A system that accepts work faster than it can process it builds up queues, and queues increase latency without increasing throughput. Eventually requests time out while still queued, clients retry, and the extra load makes the queues longer: a metastable failure, in which the system stays broken even after the original trigger is gone. Bronson and colleagues described such failures at Facebook and elsewhere in a 2021 paper. The defence is backpressure, signalling upstream that a component is full: bounded queues that reject work when full, explicit flow control as in TCP and HTTP/2, rate limiting, and load shedding, dropping low-priority work early so that high-priority work completes. Lesson 11 returns to these patterns, together with circuit breakers.

Serialization and the network's cost

Every message must be encoded as bytes. Text formats such as JSON are easy to read and debug but larger and slower to parse; binary formats such as Protocol Buffers, Avro and Cap'n Proto are compact and fast and support schema evolution, at the cost of needing a schema to read them. Whichever is chosen, two rules prevent many outages: make every change to a message format both backward compatible, so new code reads old messages, and forward compatible, so old code reads new messages; and never deploy a change that requires every machine to switch at once. Finally, remember the seventh fallacy: transport is not free. Traffic between cloud regions is billed, and serialization, compression and encryption consume CPU on both ends. Chatty protocols that make many small calls are both slow and expensive; batching and caching are the usual remedies.

Network partitions and the limits of the transport

A network partition, introduced in lesson 1, looks to each side exactly like a set of crashed peers. TCP cannot help: it will report a broken connection after its own timeouts, but it cannot say whether the peer processed the last messages, and it cannot distinguish a crashed peer from an unreachable one. Higher layers therefore need their own mechanisms. Heartbeats, small periodic messages, let nodes notice silence quickly. Sequence numbers and acknowledgements at the application level let a node resend and deduplicate. And protocols that must remain correct during partitions, such as the replication and consensus protocols of lessons 6 to 9, are designed so that at most one side of a partition can make progress on any given piece of state.

Worked examples