distributed-systems

Design Philosophy

Design for failure, scale, and simplicity — in that order.

Fundamental Principles

1. Understand the hardware — Know your latencies, bandwidth, failure rates
2. Simple abstractions that compose — MapReduce: just map() and reduce()
3. Design for 10x, rewrite at 100x — Don't over-engineer, but know your limits
4. Tail latency dominates — p99 matters more than average
5. Data locality is king — Move compute to data, not data to compute
6. Failures are routine, not exceptional — GFS assumes disks fail daily
7. Back-of-envelope first — If the math doesn't work, the system won't

Core Truths

• Networks are unreliable (packets drop, delay, duplicate, reorder)
• Clocks are unreliable (drift, skew, NTP jumps)
• Processes are unreliable (crash, pause, slow)
• Disks are unreliable (corruption, failure, latency spikes)
• You cannot distinguish slow from dead

Design Principles

• Prefer crash-recovery over complex state management
• Make operations idempotent by default
• Use leases, not locks (distributed locks are a lie)
• Checksum everything — silent corruption is worse than failure
• Batch for throughput, stream for latency — know which you need
• Compress aggressively — network and storage are expensive
• Design for debuggability — you will debug this at 3 AM

Numbers Everyone Should Know

Operation                              Time
─────────────────────────────────────────────────
L1 cache reference                     1 ns
L2 cache reference                     4 ns
Branch mispredict                      5 ns
L3 cache reference                    20 ns
Mutex lock/unlock                     25 ns
Main memory reference                100 ns
Compress 1KB with Snappy           3,000 ns    (3 μs)
Send 1KB over 10 Gbps network     10,000 ns   (10 μs)
SSD random read                   16,000 ns   (16 μs)
Read 1 MB sequentially from SSD  100,000 ns  (100 μs)
Round trip within datacenter     500,000 ns  (500 μs)
Read 1 MB sequentially from disk   2 ms
Disk seek                          4 ms
Round trip US West → Netherlands  75 ms
Round trip US West → Singapore   180 ms

Implications

Quick Math

1 million requests/day   ≈ 12/second
1 billion requests/day   ≈ 12,000/second
86,400 seconds/day       ≈ 100K (use 10^5)
1 GB/sec for 1 day       ≈ 86 TB

Back-of-Envelope Estimation

If you can't estimate it, you can't build it.

The Process

1. Clarify requirements — QPS, data size, latency, durability
2. Estimate data volume — Users × actions × size × retention
3. Estimate throughput — Peak QPS = 3-5x average
4. Estimate storage — Raw + replication + indexes + headroom
5. Check against limits — Network, disk, memory, CPU

Example: URL shortener (100M URLs/month, 10:1 reads, 5y retention, <100ms p99)

Writes: 100M / (30 × 24 × 3600) ≈ 40/sec → reads 400/sec → peak 2000/sec (5x avg)
Storage: ~1KB/URL × 6B URLs over 5y = 6 TB → 18 TB with 3x replication
Single machine? Storage yes (18TB SSD); throughput 2000/s × 1KB = 2 MB/sec trivial;
latency SSD 16μs + network 500μs ≪ 100ms  →  start simple, shard later

Scaling Laws

Amdahl:  speedup = 1 / (S + (1-S)/N)   — 5% serial caps speedup at 20x, regardless of N
Little:  L = λ × W                      — 100 req/s × 1.0s latency = 100 in-flight (10x the threads of 0.1s)
USL:     C(N) = N / (1 + α(N-1) + βN(N-1)) — α contention, β coherence;
         β > 0 → throughput DECREASES past some N (retrograde scaling)

Red Flags

• Can't fit working set in memory → random I/O death
• QPS × latency > available threads → queuing collapse (Little's Law)
• Replication lag > SLA → consistency model mismatch
• Peak > 10x average → need queue/buffer/shed
• β > 0 in USL → adding nodes makes things SLOWER

System Design Process

Structured thinking prevents expensive rewrites.

1. Functional: core operations, data flow (sources → transformations → sinks)
2. Non-functional: throughput and latency at p50/p99, availability (99.9% = 8.7h downtime/year), durability (no loss on N failures), consistency model (below)
3. Constraints: budget, team, timeline, existing infrastructure, regulatory (data residency, retention, encryption)

Write an ADR for every key architectural decision: context (volume, ordering, constraints) → decision → consequences (good / bad / neutral) → alternatives considered with reasons rejected.

Fundamental Trade-offs

CAP — partitions WILL happen; during one, choose:

• CP: refuse requests, stay consistent (banking, locks, leader election)
• AP: accept requests, reconcile later (shopping cart, DNS, caches)

PACELC — if Partition → choose A or C; Else → choose Latency or Consistency. Even a healthy system pays the consistency-latency tax.

Consistency Models

Weaker consistency = better availability/latency. Choose the weakest that's correct.

The Spectrum

graph TD
    L["Linearizable\nSingle copy, real-time order\nUse: locks, leader election, CAS"]
    S["Sequential\nAll see same order, not real-time\nUse: log replication, total order broadcast"]
    CA["Causal\nCausally related ops in order\nUse: social feeds, collaborative editing"]
    RYW["Read-Your-Writes\nSee own writes immediately\nUse: profile updates"]
    E["Eventual\nWill converge, no order guarantee\nUse: DNS, caches, metrics"]

    L -->|stronger| S --> CA --> RYW -->|weaker| E

    style L fill:#fee,stroke:#c33
    style E fill:#efe,stroke:#3c3

Choosing a Model

Implementation mechanics per model (consensus for linearizable, vector clocks for causal, LWW/CRDTs for eventual): references/consensus-and-time.md

Consensus & Coordination

Consensus is expensive. Avoid it when possible. Use it correctly when necessary.

• Raft: randomized election timeouts (150-300ms) prevent split votes; a write commits when a majority ACKs. Linearizable = 2 RTT minimum — if a lease or single leader suffices, skip consensus.
• Election and log-replication sequence diagrams: references/consensus-and-time.md

Fencing Tokens

Every write to shared storage carries the holder's token (monotonically increasing term); storage rejects any token lower than the highest seen. This is what makes leases and leader election safe — without it, a paused old leader corrupts state after the partition heals. Sequence diagram: references/consensus-and-time.md

Leases, Not Locks

A lock is held until released — if the holder dies, it is held forever. A lease is held for a fixed window (say 30s) and must be renewed: holder dies → lease expires → others acquire. Renewal discipline: renew at T/3 intervals (10s for a 30s lease); two failed renewals = assume lost; stop doing protected work BEFORE expiry.

Leader Election

// O(1) compute and memory per iteration — coordination glue; the real cost is the lease-store RTT
sealed interface LeaseResult permits LeaseResult.Acquired, LeaseResult.NotAcquired {
    record Acquired(String token, Instant expiresAt) implements LeaseResult {}
    record NotAcquired(String reason) implements LeaseResult {}
}

while (running) {
    switch (acquireLease("leader-lock", Duration.ofSeconds(30))) {
        case Acquired(var token, var expiry) -> {
            while (renewLease("leader-lock", token)) {
                doLeaderWork(token);   // every write carries the fencing token
            }
            releaseLease("leader-lock", token);
        }
        case NotAcquired(var reason) -> waitForLeaseExpiry();
    }
}

Comment doctrine (the big-O comments above are the convention, not decoration):

• Comments are for the person reading and reviewing the code: the public API, anything not obvious, and the reason behind the code as a plain sentence. No WHY: label.
• Keep the work plan out of comments: no phases, plan steps, ticket numbers, or words like "for now", "temporary", "later". Those go stale and someone has to find and delete them. Unfinished work belongs in plan.md.
• Any function that does real work gets a header comment above it: one line on what it is for, plus its big-O for compute and memory (// O(1) compute and memory ... as shown). Tiny obvious helpers are exempt. The big-O lives in the header, not on a line inside the body.
• Editing counts too: when you change a function, keep the comments already there, fix any your change made wrong, and add the missing header, big-O, or reason. Do this only for code you add or change, not for nearby code you did not touch.

Time & Ordering

Physical clocks lie — never order events by wall clock. NTP gives ~1ms skew within a datacenter, ~100ms across regions. Lamport timestamps, vector clocks, hybrid logical clocks, happens-before: references/consensus-and-time.md

Storage Engines

LSM trees (RocksDB, Cassandra) for write-heavy sequential workloads; B-trees (PostgreSQL, InnoDB) for read-heavy point lookups. WAL-then-apply is the durability pattern beneath every database. Write paths, compaction strategies (size-tiered vs leveled), checksums/scrubbing/end-to-end integrity: references/storage-engines.md

Concurrency & Parallelism

Shared mutable state is the root of all concurrency bugs.

Fastest → slowest: immutable → thread-local → lock-free (CAS) → lock-based (mutex/rwlock) → distributed lock (consensus + network RTT). Choose the highest level that works.

Memory: 1 OS thread ≈ 1MB stack → 10K threads = 10GB before any work happens.

Distributed Transactions

Avoid if possible. Use Saga or Outbox when necessary. 2PC as last resort.

Why avoid: coordinator is a single point of failure; locks held across network boundaries; latency = sum of all participants; failure handling is complex.

• Saga — sequence of local transactions, each with a compensation. Choreography (services react to events) for simple flows; orchestration (central coordinator) when the flow needs visibility and ownership.
• Outbox — business write + event row in ONE local transaction; a relay publishes. Prefer log-based CDC of the outbox table (Debezium's outbox event router) over polling; a poller is the fallback when CDC infra is unavailable.
• 2PC — blocks on coordinator failure while participants hold locks. Non-blocking commit means replicating the coordinator's decision via consensus (Paxos/Raft-backed 2PC, e.g. Spanner). 3PC assumes no partitions — not practical.

Compensating Transactions

Forward actions:         Compensating actions:
1. Create order         ← Cancel order
2. Reserve inventory    ← Release inventory
3. Charge payment       ← Refund payment
4. Ship order           ← (can't undo — manual process)

Design compensations at the same time as forward actions!

Sequence diagrams (saga choreography/orchestration, outbox, 2PC) and the outbox implementation: references/distributed-transactions.md

Failure Modes & Detection

Understand how things fail. Design for recovery, not prevention.

Types of Failures

Most systems assume crash-recovery. Byzantine tolerance is expensive.

Two Generals: no protocol guarantees agreement over an unreliable network. Implication: at-least-once or at-most-once delivery — exactly-once only via a higher-level protocol (dedup, idempotency).

Failure Detection: φ Accrual

Perfect failure detection is impossible (you can't distinguish slow from dead). Track heartbeat inter-arrival times; φ = -log10(P(alive)); φ > 8 → probably dead. Aggressive (φ=5) detects fast with more false positives; conservative (φ=12) the reverse.

Split-Brain

A partition leaves both sides believing they are the leader → divergent writes → corruption. Prevention:

1. Quorum — need majority to operate (3 of 5 nodes)
2. Fencing — storage rejects the old leader's writes (see Consensus & Coordination)
3. STONITH — "Shoot The Other Node In The Head" (force restart)

Gray Failures

Not dead, not healthy — the worst kind: wrong data behind a 200 OK, 0.1% packet loss, occasional disk latency spikes (10ms → 10s), random bit flips. Detection: end-to-end health checks (not just ping), checksums on data, anomaly detection on latency distributions, cross-node verification of state.

Kafka

Kafka 4.x is KRaft-only — ZooKeeper support removed in 4.0; never emit zookeeper.connect. Safe by default since 3.0/4.0: acks=all, enable.idempotence=true, linger.ms=5 — don't present these as opt-in tuning, and don't disable them.

Topic design, exactly-once (Kafka→Kafka vs external-sink use different mechanisms), compacted topics, producer tuning, consumer groups + KIP-848 rebalance protocol, share groups (KIP-932), schema registry, BC/DR with MirrorMaker 2: references/kafka.md

Replication & Partitioning

Single-leader + read replicas handles most cases. Avoid multi-leader unless cross-region write latency forces it — then you own conflict resolution (LWW silently loses data; merge functions and CRDTs are real work). Leader-follower mechanics, conflict strategies, consistent hashing, gossip, anti-entropy/read-repair/hinted-handoff, zero-downtime migration: references/replication-and-partitioning.md

Data Pipelines & Derived Data

Single writer per dataset; every derived view (search index, cache, OLAP, materialized view) must be rebuildable from the source of truth. Stream over batch when latency matters; kappa over lambda (one codebase, replay to reprocess). Schemas: Protobuf/Avro, BACKWARD-compatible by default, registry-enforced. Batch-vs-stream trade-offs, schema-evolution field rules, CDC approaches (log-based beats polling), event sourcing: references/data-pipelines.md

Resilience Patterns

Circuit Breaker

stateDiagram-v2
    [*] --> Closed
    Closed --> Open : failure_rate > threshold
    Open --> HalfOpen : timeout elapsed
    HalfOpen --> Closed : probe succeeds
    HalfOpen --> Open : probe fails

    state Closed {
        [*] : Normal operation, count failures
    }
    state Open {
        [*] : Fail fast, don't call downstream
    }
    state HalfOpen {
        [*] : Allow single probe request
    }

Implementation: Failsafe's CircuitBreaker — java-skill owns the code; don't hand-roll one (the state machine races under concurrency are subtle). This skill owns the policy: what failure rate trips it, per-dependency breakers, half-open probe budget.

Retry policy: exponential backoff + jitter, capped (e.g. 100ms base, ×2 per attempt, 30s cap); bounded attempts; retry budget ≤ 10% of total requests; honor server Retry-After over the client schedule; retry only idempotent operations.

Bulkhead: isolate dependencies into separate pools (threads, connections) — one exhausted pool must not starve the others.

Idempotency

The claim must be ATOMIC — a unique constraint + insert, never find-then-save (two concurrent requests both miss the find, both charge).

public record IdempotencyRecord(
    String key,
    Status status,            // enum — never stringly typed
    String result,
    Instant createdAt,
    Instant leaseExpiresAt    // IN_PROGRESS past this = crashed holder, re-claimable
) {
    public enum Status { IN_PROGRESS, COMPLETED }
}

// O(1) compute and memory per call (single-row insert + single-row update)
public Result<PaymentResult, Failure> processPayment(
        String idempotencyKey, PaymentRequest request) {

    // atomic claim: insert ... on conflict (key) do nothing returning * —
    // empty means we own the key; a row back means someone got there first
    var existing = store.tryClaim(idempotencyKey, Instant.now().plus(LEASE));

    if (existing.isPresent()) {
        var rec = existing.get();
        return switch (rec.status()) {
            case COMPLETED -> Result.ok(deserialize(rec.result()));
            case IN_PROGRESS -> rec.leaseExpiresAt().isBefore(Instant.now())
                ? reclaimAndProcess(idempotencyKey, request)  // CAS on lease_expires_at
                : Result.err(Failure.of("IN_PROGRESS", "request already in flight"));
        };
    }

    var result = chargePayment(request);
    store.markCompleted(idempotencyKey, serialize(result));
    return Result.ok(result);
}

Backpressure & Load Management

Graceful degradation > catastrophic failure.

The Cascade Problem

One slow dependency kills the chain: C slows to 5s → B's threads block waiting on C → B's pool exhausts → A's calls to B time out → A's pool exhausts → users get 503 from a system where only C was sick. Fix: timeouts + circuit breakers + bulkheads, at every hop.

Rate Limiting

Token bucket at ingress: a bucket of N tokens refilled at R/sec gives R/sec sustained with bursts up to N. Reject with 429 + Retry-After — tell clients when to come back, never silently queue.

Load Shedding

When overloaded, reject strategically:

• Priority-based: health checks always; paid tier < 90% capacity; free tier < 70%; batch jobs < 50%
• LIFO: shed oldest requests first (the client has likely already timed out)
• Selective: shed expensive queries and reads (retryable) first; protect writes

Admission Control

Little's Law feedback loop: latency up → queue grows → latency up → collapse. Reject early instead of queueing: if in_flight > avg_latency × max_throughput × safety_margin → 503 fast-fail.

Backpressure Propagation

Don't hide overload — propagate it upstream. Sync: slow responses block the caller. Async: bounded queues block the producer. Reactive: demand signaling (consumer requests N items). Surface it as HTTP 429 + Retry-After, gRPC RESOURCE_EXHAUSTED, Kafka producer blocking when buffer.memory is full.

ML, IoT & Edge

Designing ML serving or feature pipelines (training-serving skew, point-in-time correctness, serving patterns), or device/edge ingestion (MQTT/CoAP, offline-first sync, time-series downsampling): references/ml-iot-edge.md

Chaos Engineering & Testing

1. Define steady state — what "healthy" looks like (latency, error rate, throughput)
2. Hypothesize — "system stays healthy when X fails"
3. Inject real failures — don't simulate, actually break things
4. Minimize blast radius — start small, expand gradually
5. Automate — run chaos continuously, not just game days

Distributed logic is still TDD: red → green → vacuous mutation check; reproduce the failure deterministically (fault injection in tests) before the fix lands.

Injection catalog — network (latency, packet loss, partition, DNS failure), process (SIGKILL, SIGSTOP, CPU hog, OOM), dependency (error/slow/malformed responses, timeout), infra (kill node, fill disk, clock skew, revoke credentials).

Game day: hypothesis + blast radius + kill switch before; measure time-to-detect and time-to-mitigate during; file tickets, update runbooks, schedule the next one after.

Anti-Patterns

Before Deploying (Self-Review)

Design

• Back-of-envelope math works at 10x current scale
• Consistency model explicitly chosen and documented
• Failure modes identified and handled
• No single points of failure in critical path
• ADR written for key architectural decisions

Data Integrity

• Checksums on data at rest and in flight
• Idempotency keys for all mutating operations
• No data loss on any single component failure
• Schema evolution strategy defined (forward + backward compatible)

Resilience

• Timeouts on ALL external calls
• Circuit breakers on critical dependencies
• Graceful degradation (fallbacks) defined
• Bounded queues, bounded retries, bounded everything

Consistency

• Distributed transactions avoided (Saga/Outbox if needed)
• Fencing tokens used for leader-based operations
• Clock skew handled (logical time where needed)

Failure Handling

• Split-brain prevented (quorum, fencing)
• Crash-recovery tested (kill process, verify recovery)
• Network partition behavior defined
• Gray failure detection in place

Observability

• RED metrics per service (Rate, Errors, Duration); USE per resource (Utilization, Saturation, Errors)
• Distributed tracing — trace_id/correlation id propagated through every service hop
• Alerting on SLO violations
• Dashboards for capacity planning

Load Management

• Rate limiting at ingress
• Load shedding strategy defined
• Backpressure propagates upstream
• Capacity headroom: running < 70% at peak

Chaos Readiness

• Hypothesis for major failure scenarios
• Kill switch / rollback plan
• Game day scheduled