Communication & RPC

Before building abstractions, you must internalize why distributed communication resists perfection. The Two-Generals Problem (Akkoyunlu et al., 1975) captures the core impossibility.

Two armies, led by General Alice and General Bob, must coordinate an attack on a city between them. They can only communicate by sending messengers through enemy territory, where any message may be captured (lost). If both attack simultaneously, they win. If only one attacks, that army is destroyed.

Alice sends: "Attack at dawn." Bob receives it — but does Alice know Bob received it? Bob sends an acknowledgment. Alice receives it — but does Bob know Alice received his ACK? Each confirmation requires another message, which might also be lost, creating an infinite regress. No finite number of messages guarantees mutual knowledge of a shared decision over an unreliable channel.

This is not just a puzzle — it is a formal theorem. Any protocol over a lossy network has an execution where the last message might be lost, so the sender can never be certain the receiver got it. The TCP three-way handshake does not solve this (the final ACK can be lost); it merely reduces the window of uncertainty to be practically acceptable.

The engineering implication: distributed systems must be designed to tolerate uncertainty rather than eliminate it. You will see this theme recur in every failure-semantics discussion.

A function call in a single process passes arguments as in-memory data structures: a pointer to a list, a struct on the stack, a float in a register. Across a network, you must convert these into a flat sequence of bytes, transmit them, and reconstruct the original values on the remote side. This is marshalling (or serialization; the terms are nearly synonymous in modern usage, though marshalling historically implies also handling remote references).

Why it is hard:

1. Endianness. Intel x86 is little-endian (least-significant byte first). Network byte order (and many older systems) is big-endian. The 32-bit integer 0x01020304 is stored as 01 02 03 04 on a big-endian machine but 04 03 02 01 on little-endian. If you transmit raw bytes without conversion, the receiver reconstructs the wrong number.
2. Pointer invalidity. A pointer like 0x7fff5c3a is meaningless in another process's address space. You must serialize the data the pointer refers to, not the pointer itself. This extends recursively to linked lists, trees, and graphs — requiring a traversal to produce a flat encoding.
3. Type representation differences. int might be 32 bits on one machine and 64 on another. Floating-point representation (IEEE 754 vs. older formats) varies. Struct padding and alignment differ across compilers and architectures.
4. Schema evolution. If the server is updated to add a field to a struct, older clients must still work. Serialization formats differ greatly in how gracefully they handle this.

Common serialization formats:

• JSON: human-readable, self-describing, universally supported. Verbose; no binary types; floating-point round-trip issues.
• Protocol Buffers (protobuf): compact binary format, schema-defined (.proto files), excellent versioning (field numbers, not names). Used by gRPC.
• Apache Avro: schema stored alongside data or in a registry; excellent for streaming/Kafka.
• MessagePack: binary JSON; smaller than JSON, no schema needed.
• XDR (External Data Representation): the original Sun RPC format, fixed big-endian encoding.
• Cap'n Proto / FlatBuffers: zero-copy formats where the serialized bytes are the in-memory representation (no parsing step).

The fundamental operation is: marshal(args) → bytes on the client, unmarshal(bytes) → args on the server, then marshal(return_value) → bytes on the server, unmarshal(bytes) → return_value on the client.

# Python: manual marshalling vs. protobuf-style
import struct

# Naive manual marshalling of (int32, float32, string)
def marshal(x: int, y: float, s: str) -> bytes:
    encoded = s.encode('utf-8')
    # Network byte order (big-endian): !i = int32, !f = float32
    return struct.pack(f'!if{len(encoded)}s', x, y, encoded)

def unmarshal(data: bytes) -> tuple:
    x, y = struct.unpack_from('!if', data)
    s = data[8:].decode('utf-8')  # 4 bytes int + 4 bytes float
    return x, y, s

bytes_on_wire = marshal(42, 3.14, "hello")
print(marshal(42, 3.14, "hello").hex())  # deterministic bytes

# --- With protobuf (schema-driven, handles versioning) ---
# In search.proto:
# message SearchRequest {
#   string query = 1;
#   int32  page  = 2;
# }
# Generated Python:
# req = SearchRequest(query="kernel", page=1)
# wire = req.SerializeToString()   # compact binary
# req2 = SearchRequest.FromString(wire)  # reconstruct
# assert req2.query == "kernel"

The core insight of RPC, articulated in Birrell and Nelson's landmark 1984 paper "Implementing Remote Procedure Calls," is that the complexity of network communication can be hidden inside auto-generated stub code, leaving application code that looks like ordinary local function calls.

The five-layer architecture:

1. Client application code calls what looks like a local function: result = server.Add(3, 4).
2. Client stub (also called proxy): auto-generated from an interface definition. It marshals the arguments into bytes, sends them to the server over the network, waits for a response, unmarshals the result, and returns it to the caller. To the caller, it looks like a normal function.
3. Network transport: the bytes travel over TCP/UDP/HTTP2/etc.
4. Server stub (also called skeleton or dispatcher): receives the bytes, unmarshals the arguments, calls the actual server-side implementation function, marshals the return value, and sends it back.
5. Server implementation: the real function, written by the server developer, runs locally and knows nothing about the network.

Interface Definition Language (IDL): Both client and server stubs are generated from a common interface specification. In gRPC this is a .proto file; in Java RMI it is a Java interface; in the original Sun RPC it was XDR/rpcgen. The IDL is the contract between client and server.

Synchronous vs. asynchronous stubs: A blocking stub halts the calling thread until the response arrives. An async stub returns a future/promise immediately, and the result is delivered via callback or await. Async is essential for high-throughput services where blocking a thread per outstanding call would exhaust thread pools.

The abstraction leaks when: the network is slow (latency you cannot see locally), the network drops packets (you get an exception, not a return value), the server crashes mid-execution (did it run once? Zero times?), or arguments are too large to serialize in one TCP segment (backpressure/streaming).

# Pseudocode illustrating what a generated client stub does
class AddServiceStub:
    def __init__(self, channel):   # channel = TCP connection to server
        self.channel = channel

    def Add(self, a: int, b: int) -> int:
        # 1. Marshal arguments
        request_bytes = marshal_AddRequest(a=a, b=b)

        # 2. Send over the wire with a unique call ID
        call_id = generate_unique_id()
        self.channel.send(call_id, request_bytes)

        # 3. Block waiting for reply with matching call_id
        reply_bytes = self.channel.recv_for(call_id, timeout=5.0)

        # 4. Unmarshal result
        return unmarshal_AddResponse(reply_bytes).result

# Server-side skeleton (also auto-generated)
def handle_request(call_id, request_bytes):
    req = unmarshal_AddRequest(request_bytes)
    result = AddServiceImpl().Add(req.a, req.b)   # actual work
    reply_bytes = marshal_AddResponse(result=result)
    channel.send(call_id, reply_bytes)

This is the intellectual heart of RPC. When a client sends a request and waits for a reply, five things can go wrong before the reply arrives:

1. The request is lost in the network.
2. The server receives the request but crashes before executing.
3. The server executes the request but crashes before sending the reply.
4. The server sends the reply but it is lost in the network.
5. The reply arrives but is delayed so long the client has already retried.

In cases 1–2 the operation did not execute. In cases 3–5 it did. From the client's perspective, cases 1–5 all look the same: timeout with no reply. This is the core difficulty.

At-least-once semantics: The client retries on timeout until it gets a reply. This guarantees the operation eventually executes — but it may execute multiple times if the original request or reply was lost after execution (cases 3–5). This is acceptable only for idempotent operations: operations where applying them multiple times has the same effect as applying them once. GET /users/42 is idempotent. POST /payments ("charge $100") is not.

At-most-once semantics: The client sends the request once and either gets a result or an error. If no reply arrives within timeout, it reports failure without retrying. The operation executed zero or one time. This is safer for non-idempotent operations, but it sacrifices availability (a transient network hiccup means the call fails even if the server is healthy).

Exactly-once semantics (the holy grail): The operation executes precisely once, even in the presence of retries. Achieving this requires cooperation from both client and server:

• The client attaches a unique request ID (idempotency key) to each call.
• The server maintains a deduplication log (sometimes called an idempotency table): before executing, check if this request ID has been processed before. If yes, return the stored result without re-executing. If no, execute, store the result, then reply.
• The client may retry freely; duplicate requests are silently deduplicated by the server.

This is called the idempotency key pattern and is used in production by Stripe, Twilio, AWS, and most financial systems. The caveat: the deduplication log must be durable (survives server crashes) and must be cleaned up eventually (log entries can be expired after the client's maximum retry window).

Practical note on "exactly-once" in message queues: Kafka's idempotent producer + transactional API provides exactly-once delivery within a single producer session to a single partition. Full end-to-end exactly-once (read-process-write) requires the consumer to participate in a transaction as well.

# At-most-once client (simple, no retry)
def call_once(stub, a, b, timeout=2.0):
    try:
        return stub.Add(a, b, timeout=timeout)
    except TimeoutError:
        raise RPCError("call may or may not have executed")

# At-least-once client (retry until success)
def call_at_least_once(stub, a, b, max_retries=5):
    for attempt in range(max_retries):
        try:
            return stub.Add(a, b, timeout=2.0)
        except TimeoutError:
            pass  # retry
    raise RPCError("gave up after retries")

# Exactly-once client (idempotency key)
import uuid
def call_exactly_once(stub, a, b, max_retries=5):
    idempotency_key = str(uuid.uuid4())  # stable across retries
    for attempt in range(max_retries):
        try:
            # server deduplicates by idempotency_key
            return stub.Add(a, b, idempotency_key=idempotency_key)
        except TimeoutError:
            pass
    raise RPCError("gave up")

# Server-side deduplication table
dedup_cache = {}   # key -> result  (persistent in production)
def handle_add(a, b, idempotency_key):
    if idempotency_key in dedup_cache:
        return dedup_cache[idempotency_key]   # replay stored result
    result = a + b
    dedup_cache[idempotency_key] = result     # persist before replying
    return result

gRPC (Google Remote Procedure Call, open-sourced 2015) is the dominant RPC framework for microservices. It combines three carefully chosen technologies:

1. HTTP/2 as the transport: multiplexes many logical streams over a single TCP connection (solving head-of-line blocking), supports bidirectional streaming, header compression (HPACK), and server push.
2. Protocol Buffers (protobuf) as the IDL and serialization format: compact binary encoding, strongly-typed schema with forward/backward compatibility via field numbers.
3. Code generation: the protoc compiler reads .proto files and emits client stubs and server interfaces in Go, Python, Java, C++, Rust, and many other languages.

Four communication patterns gRPC supports:

1. Unary RPC: one request, one response. Like a normal function call.
2. Server streaming RPC: client sends one request; server streams back a sequence of responses (e.g., a search returning results as they are found).
3. Client streaming RPC: client streams a sequence of requests (e.g., uploading a large file in chunks); server replies once.
4. Bidirectional streaming RPC: both sides stream. Used for real-time communication (e.g., live collaborative editing, telemetry).

Interceptors (middleware): gRPC has a clean interceptor/middleware chain on both client and server. This is where authentication (metadata headers carrying JWT tokens), logging, distributed tracing (injecting trace IDs), retries, and rate limiting are implemented — cleanly separated from business logic.

Status codes and error handling: gRPC defines a standard set of status codes (NOT_FOUND, UNAVAILABLE, DEADLINE_EXCEEDED, UNAUTHENTICATED, etc.) analogous to HTTP status codes but richer for RPC semantics. This standardization means cross-language clients all handle errors the same way.

Deadlines and cancellation: Every gRPC call can carry a deadline (absolute timestamp by which the call must complete). If the deadline passes, the client-side library cancels the call and signals the server via the HTTP/2 RST_STREAM frame. Servers should check context.is_active() in long-running handlers to abort early — this is critical to prevent cascading timeouts.

# 1) Define the contract (search.proto)
# syntax = "proto3";
# service SearchService {
#   rpc Search (SearchRequest) returns (SearchResponse);
#   rpc SearchStream (SearchRequest) returns (stream SearchResult);
# }
# message SearchRequest { string query = 1; int32 top_k = 2; }
# message SearchResponse { repeated SearchResult results = 1; }
# message SearchResult  { string title = 1; float score = 2; }

# 2) Server implementation (Python)
import grpc
from concurrent import futures
import search_pb2, search_pb2_grpc

class SearchServicer(search_pb2_grpc.SearchServiceServicer):
    def Search(self, request, context):
        # context.deadline() tells you when the client gives up
        results = do_search(request.query, request.top_k)
        return search_pb2.SearchResponse(
            results=[search_pb2.SearchResult(title=r.title,
                                             score=r.score)
                     for r in results])

    def SearchStream(self, request, context):
        for result in do_search_streaming(request.query):
            if not context.is_active():   # client cancelled?
                return
            yield search_pb2.SearchResult(title=result.title,
                                          score=result.score)

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
search_pb2_grpc.add_SearchServiceServicer_to_server(
    SearchServicer(), server)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()

# 3) Client
channel = grpc.insecure_channel('localhost:50051')
stub = search_pb2_grpc.SearchServiceStub(channel)
response = stub.Search(
    search_pb2.SearchRequest(query="kernel RPC", top_k=5),
    timeout=2.0)   # deadline: 2 seconds from now
for r in response.results:
    print(r.title, r.score)

RPC is synchronous coupling: the caller waits for the callee to finish. This is natural but brittle — if the payment service is slow, your checkout service stalls. Message queues (also called message brokers or event buses) decouple sender from receiver in time.

The model: A producer publishes a message to a named queue or topic at the broker. The broker durably stores the message. One or more consumers pull from or subscribe to the queue and process messages asynchronously. The producer never waits for the consumer to process the message.

Key properties:

• Durability: the broker writes messages to disk before acknowledging the producer. If the broker crashes and restarts, no messages are lost.
• Buffering / backpressure: if consumers are slow, messages accumulate in the queue. The producer can continue at its own rate. (Contrast with synchronous RPC: a slow callee immediately slows the caller.)
• Fan-out: a single message can be delivered to multiple consumers (pub/sub pattern). Each subscriber gets its own copy.
• Ordering: within a single queue/partition, most brokers preserve insertion order. Across partitions, ordering is not guaranteed.

Two fundamental delivery patterns:

1. Point-to-point (work queue): each message is processed by exactly one consumer. Multiple consumers compete to process messages from the same queue. Load is naturally distributed.
2. Publish-subscribe (pub/sub): messages are published to a topic. Every subscribed consumer group receives every message. Used for event broadcasting ("order placed" → billing service, warehouse service, analytics service all receive it).

Kafka's model takes this further: topics are divided into partitions (enabling parallel processing), and each message has an offset (an integer sequence number). Consumers track their offset explicitly. This means Kafka is a replay log, not a traditional queue — consumers can rewind and reprocess old messages. Retention can be hours or forever. This is architecturally significant: it separates consumption from deletion.

When to use queues vs. RPC: Use RPC when you need the result now (a user is waiting). Use queues when the work can be done later (sending an email, processing an uploaded video, updating analytics), when you need to buffer bursts of traffic, or when you want to decouple service availability (the producer succeeds even if the consumer is down for maintenance).

# Simplified RabbitMQ-style producer and consumer (using pika)
import pika, json

QUEUE = 'order_processing'

# --- Producer (in checkout service) ---
def publish_order(order_id: str, amount: float):
    conn = pika.BlockingConnection(
        pika.ConnectionParameters('localhost'))
    ch = conn.channel()
    ch.queue_declare(queue=QUEUE, durable=True)  # survive broker restart
    message = json.dumps({'order_id': order_id, 'amount': amount})
    ch.basic_publish(
        exchange='',
        routing_key=QUEUE,
        body=message,
        properties=pika.BasicProperties(
            delivery_mode=2))   # 2 = persist to disk
    conn.close()
    # Returns immediately — consumer may process in 1ms or 1 hour

# --- Consumer (in fulfillment service) ---
def on_message(ch, method, props, body):
    order = json.loads(body)
    try:
        process_order(order['order_id'], order['amount'])
        ch.basic_ack(delivery_tag=method.delivery_tag)  # success
    except Exception:
        ch.basic_nack(delivery_tag=method.delivery_tag,
                      requeue=True)   # retry later

conn = pika.BlockingConnection(
    pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue=QUEUE, durable=True)
ch.basic_qos(prefetch_count=1)   # process one at a time
ch.basic_consume(queue=QUEUE, on_message_callback=on_message)
ch.start_consuming()   # blocks, processing messages as they arrive

Idempotency is the property that applying an operation multiple times produces the same result as applying it once: f(f(x)) = f(x). In distributed systems, idempotency is not just a nice-to-have — it is the mechanism that enables safe retries and therefore reliable systems.

Why it matters: Because you can never be sure whether a remote call succeeded (due to the Two Generals problem), you will either:

• Never retry (at-most-once, risking lost operations), or
• Retry freely (at-least-once, risking duplicate operations), or
• Retry with idempotency keys (exactly-once effect, at-least-once delivery).

The third option is almost always the right choice for important operations.

Naturally idempotent operations:

• Setting a value: SET x = 5 (doing it twice is fine)
• Deleting a resource by ID (second delete on a non-existent resource is a no-op or a 404)
• HTTP GET, PUT (by definition in REST)

Non-idempotent operations (require idempotency keys):

• HTTP POST (creating a new resource, charging a card)
• Incrementing a counter: counter += 1 (double-execution = double-increment)
• Sending an email

Implementing idempotency keys correctly:

1. The client generates a UUID v4 and includes it in the request header or body.
2. The server stores (idempotency_key → response) in a durable store (database, Redis with persistence) before sending the response.
3. On duplicate request: return the stored response without re-executing.
4. The key must be stored atomically with the effect, or you risk:
   • Executing twice if you store the key after executing (crash between them)
   • Returning a cached response for an operation that never completed (if you store before executing and crash)

The correct order: BEGIN TRANSACTION → check for key → execute → store result → COMMIT → reply. This way, a crash before commit means the key is not stored, so a retry correctly re-executes.

Stripe's approach is the industry reference: every API call accepts an Idempotency-Key header. Keys expire after 24 hours. The response is stored in Redis. A second request with the same key within 24 hours replays the exact original response, including the exact status code and body.

# Correct server-side idempotency with atomic database transaction
import sqlite3, uuid

def charge_card(card_token: str, amount_cents: int,
                idempotency_key: str) -> dict:
    db = sqlite3.connect('payments.db')
    db.execute('''CREATE TABLE IF NOT EXISTS idempotency_log
                  (key TEXT PRIMARY KEY, response TEXT,
                   created_at REAL)''')
    # Check for prior execution
    row = db.execute('SELECT response FROM idempotency_log WHERE key=?',
                     (idempotency_key,)).fetchone()
    if row:
        import json
        return json.loads(row[0])   # replay — no charge

    # Execute the charge (calls payment processor API)
    charge_result = payment_processor.charge(card_token, amount_cents)
    response = {'charge_id': charge_result.id, 'status': 'succeeded'}

    # Atomically persist key + response
    import json, time
    db.execute('INSERT INTO idempotency_log VALUES (?,?,?)',
               (idempotency_key, json.dumps(response), time.time()))
    db.commit()   # if we crash here, key is stored and result is safe
    return response
    # Retries with same key → return stored response, no double-charge

Timeouts and retry policies are where theory meets operational reality. Getting them wrong is one of the most common causes of production outages.

Timeouts are mandatory. Without a timeout, a slow server can cause a thread to block forever, exhausting your thread pool and taking down the client service. Every network call must have a timeout.

Setting timeout values: A timeout that is too short causes spurious failures on a slow-but-healthy server. One that is too long delays failure detection and ties up resources. A useful heuristic: measure the P99 latency of the call under normal load and set the timeout at roughly 2–3× P99. Monitor and adjust.

Retry storms (the thundering herd problem): If a server becomes slow and thousands of clients time out simultaneously and all retry immediately, the burst of retries can overwhelm the already-struggling server, causing a positive feedback loop (a retry storm). This cascading failure is how a 5-second partial outage becomes a 30-minute full outage.

Exponential backoff with jitter: The solution is to space out retries and randomize them:

• After the 1st failure: wait base * 2^0 + jitter
• After the 2nd failure: wait base * 2^1 + jitter
• After the nth failure: wait min(cap, base * 2^n) + jitter

The jitter (random value in [0, current_backoff] or a fraction thereof) desynchronizes clients that all failed at the same moment. AWS explicitly recommends the full jitter approach.

Circuit breakers: An evolution of retry logic. A circuit breaker monitors the failure rate of a dependency. When failures exceed a threshold, the breaker opens — subsequent calls immediately return an error without even attempting the network call. After a timeout, the breaker enters half-open state and allows one probe request. If it succeeds, the breaker closes. This prevents retry storms and gives the failing dependency time to recover.

Deadline propagation: In a chain of services (A calls B calls C), if A has a 200ms deadline and B takes 150ms, C has only 50ms left. gRPC propagates deadlines as absolute timestamps across service boundaries. Services check remaining time before issuing further calls, aborting early if there is no time left. Without deadline propagation, a slow leaf service can cause a request to take much longer than the user's patience allows.

import time, random

def with_exponential_backoff(fn, max_retries=5,
                             base_ms=100, cap_ms=30_000):
    """
    Call fn() with exponential backoff + full jitter.
    Raises the last exception if all retries fail.
    """
    for attempt in range(max_retries + 1):
        try:
            return fn()
        except (TimeoutError, ServiceUnavailableError) as e:
            if attempt == max_retries:
                raise
            # Exponential backoff: 100ms, 200ms, 400ms, ..., capped
            ceiling = min(cap_ms, base_ms * (2 ** attempt))
            # Full jitter: random in [0, ceiling]
            sleep_ms = random.uniform(0, ceiling)
            time.sleep(sleep_ms / 1000)

# Simple circuit breaker
class CircuitBreaker:
    CLOSED, OPEN, HALF_OPEN = 'closed', 'open', 'half_open'

    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.state = self.CLOSED
        self.failures = 0
        self.threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.opened_at = None

    def call(self, fn):
        if self.state == self.OPEN:
            if time.time() - self.opened_at > self.recovery_timeout:
                self.state = self.HALF_OPEN
            else:
                raise CircuitOpenError('circuit open, fast-fail')
        try:
            result = fn()
            if self.state == self.HALF_OPEN:
                self.state = self.CLOSED
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            if self.failures >= self.threshold:
                self.state = self.OPEN
                self.opened_at = time.time()
            raise