Distributed systems design

Four numbers, ninety seconds, out loud. Nobody is checking your arithmetic — they are checking that you reason from load to architecture instead of pattern-matching to a diagram.

// "Design a feed for 100M daily active users"

Reads:   100M DAU x 20 feed loads/day = 2B reads/day
         2B / 86,400s ~= 23,000 QPS average
         peak ~= 3x average ~= 70,000 QPS          // always state a peak factor

Writes:  100M x 0.1 posts/day = 10M writes/day ~= 120 QPS
         read:write ratio ~= 200:1                 // THIS drives the design

Storage: 10M posts/day x 1KB = 10GB/day ~= 3.6TB/year
         + media, which dominates: 1M images/day x 500KB = 500GB/day

Memory:  cache the hot 20% of the feed
         100M users x 1KB of feed ids x 0.2 = 20GB  // fits in Redis, comfortably

The numbers worth memorising because you will use them every time: 100,000 seconds in a day (86,400, round it); 1 million QPS is enormous, 10,000 QPS is a normal large service; a single Postgres box does roughly 5,000–20,000 simple QPS; a Redis node does 100,000+; disk seek about 10ms, SSD read about 100µs, memory read about 100ns, same-datacentre round trip about 0.5ms, cross-continent about 150ms.

Then say the sentence that turns numbers into a design: "a 200:1 read-to-write ratio means I should do the expensive work on write and keep reads dumb" — which is exactly the fan-out decision two questions down.

CAP, stated correctly: when a network partition occurs, you must choose between consistency and availability. That is all it says. It is not "pick two of three" — partitions are not optional, they happen, so you are really choosing CP or AP. When there is no partition you get both, which is why the theorem is less useful than people think.

PACELC is the more honest version and worth naming: if there is a Partition, choose Availability or Consistency; Else (normal operation), choose Latency or Consistency. That second half is the trade-off you actually make every day — synchronous replication costs latency, asynchronous replication costs consistency.

The practical answer that scores: "I would not pick one model for the whole system. The balance is linearizable, the like count is eventual, and the user's own profile needs read-your-writes. Consistency is chosen per data type, not per database."

And the concrete mechanism for read-your-writes, because they will ask: after a write, pin that user's reads to the primary for a few seconds, or carry the write's log position in a token and have the replica wait until it has caught up to it.

The naive shard key is hash(key) % N. Add or remove one node and N changes, so almost every key remaps — with 4 nodes going to 5, roughly 80% of your cache is invalidated at once, which usually means the database falls over.

Consistent hashing puts both nodes and keys on a ring of hash values. A key belongs to the first node clockwise from it. Adding a node only steals keys from its immediate neighbour, so on average only 1/N of keys move.

ring:   0 ─── A ───── B ─────── C ─── (wraps to 0)
        key k hashes here ─┘  → owned by B

// add node D between A and B:
        0 ─── A ── D ── B ─────── C ───
// only keys between A and D move, and only from B. Everything else stays put.

Virtual nodes are the part people forget and the part that makes it actually work: with only N points on the ring the distribution is lumpy and removing a node dumps its entire load onto one neighbour. So each physical node gets 100–200 virtual positions, which smooths distribution and spreads a failed node's load across all survivors.

Where you have met it without knowing: Cassandra and DynamoDB partitioning, Memcached client-side sharding, and load balancers doing sticky routing.

The questions that separate people who have done this from people who have read about it:

• Cross-shard joins. They effectively do not exist. You either denormalise, or you fetch and join in the application, or you keep related data on the same shard by choosing the key so it co-locates.
• Cross-shard transactions. Also effectively gone — which is what the saga question below is for.
• Resharding. Doubling shard count is far easier than going from 3 to 5, because each shard splits cleanly in two. Plan for powers of two, or use logical shards: create 1,024 logical shards up front and map many onto each physical node, so growing is a remapping rather than a rehash.
• The celebrity problem. One user with 50 million followers breaks any key-based scheme. It gets special-cased — and that is the correct answer, not a failure of the design.

Two-phase commit: a coordinator asks every participant to prepare, and if all say yes, tells them all to commit. It gives you real atomicity — and it is rarely used, for a good reason worth stating: it is a blocking protocol. If the coordinator dies after the prepare phase, every participant holds its locks indefinitely, waiting. In practice that means an outage.

Saga: a sequence of local transactions, each with a compensating action that semantically undoes it. No distributed locks, no global atomicity — you get eventual consistency and you must design the undo path.

reserve seat    →  charge card    →  send confirmation
     |                  |
     |                  └─ fails → compensate: release seat
     └─ fails → nothing to undo, return 409

// compensation is NOT a rollback — it is a new forward action.
// You cannot un-charge a card; you issue a refund, which is
// visible to the user and appears on their statement.
// That business consequence is the real cost of a saga.

Two flavours worth naming: choreography (each service listens for events and reacts — no coordinator, but the flow is scattered across services and hard to follow) and orchestration (a single saga coordinator drives the steps — easier to reason about, monitor and debug, at the cost of a component that knows the whole flow). At six services, orchestration is almost always the right call, and saying so with that reasoning is a strong answer.

And the related pattern they may fish for: the outbox. Writing to your database and publishing to Kafka are two systems and cannot be atomic — so you write the event into an outbox table in the same transaction as the business change, and a separate relay publishes from that table. That is how you avoid the "committed the order but never published the event" bug.

Exactly-once delivery is impossible over an unreliable network. The sender cannot distinguish "message lost" from "acknowledgement lost", so it either retries (risking duplicates) or does not (risking loss). Every system that advertises exactly-once is really doing at-least-once delivery plus idempotent processing, which produces exactly-once effects. That distinction is the answer.

At-most-once: fire and forget. Fine for metrics, never for money.
At-least-once: retry until acknowledged. The default, and it means your consumer will see duplicates.
Effectively-once: at-least-once plus deduplication on a stable key.

async function handle(event) {
  // the dedupe key must come from the PRODUCER and be stable across retries
  const first = await db.processed.insert({ id: event.id })
    .catch(e => e.code === 'UNIQUE_VIOLATION' ? null : Promise.reject(e))
  if (!first) return                       // already handled, ack and move on

  await doTheWork(event)                   // same transaction, ideally
}

Three details that show real experience: the dedupe key must be generated by the producer, not the broker, or a producer retry creates a new id and defeats it; the dedupe table needs a retention policy or it grows forever; and if the work and the dedupe insert are not in one transaction there is a window where you can crash between them — which is why the natural business key (an order id, an idempotency key) is better than a separate table when you can use it.

A topic is split into partitions; each partition is an ordered, append-only log. Ordering is guaranteed within a partition and nowhere else — that single sentence is most of the marks, and it drives everything else.

• The partition key decides ordering. Key by userId and all events for one user land on one partition and stay ordered. Key randomly and you get even distribution and no ordering at all. This is the design decision.
• A consumer group gets each message once; within a group, one partition is owned by exactly one consumer. So partition count is your maximum parallelism — 10 partitions means at most 10 useful consumers, and the eleventh sits idle.
• Offsets are the consumer's bookmark. Commit after processing for at-least-once; commit before for at-most-once. There is no third option.
• Rebalance is the operational pain: when a consumer joins, leaves or times out, partitions are reassigned and the whole group pauses. A slow consumer that exceeds max.poll.interval.ms gets kicked out, triggering a rebalance, which slows everyone — a classic cascading failure.
• Retention is time or size based, not consumption based. Messages stay after being read, which is what makes replay possible and is the real difference from a queue.

Queue vs log, stated crisply: a queue (SQS, RabbitMQ, BullMQ) distributes work — a message goes to one consumer and is then gone. A log (Kafka) retains an ordered history that many independent consumer groups read at their own offsets. Choose Kafka when you need replay, ordering, or several unrelated systems consuming the same stream. Choose a queue for background jobs. Using Kafka as a job queue is the most common over-engineering at this level, and saying that out loud is a point in your favour.

Scope first: one-to-one and group messages, delivery and read receipts, online presence, message history, push when offline. Exclude voice, video and end-to-end encryption unless asked.

Connections. WebSockets, one persistent connection per device. At 10 million concurrent connections and roughly 50–100k connections per gateway node, that is 100–200 gateway nodes. The gateway is stateful — it knows which sockets it holds — so you need a session registry in Redis mapping userId → gatewayNodeId, so a message for user B can be routed to the node holding B's socket.

Sending a message. Client → gateway → message service → persist → route to recipient's gateway → push down the socket. If the recipient is offline, hand off to the push notification service. Persist before acknowledging, or a crash loses a message the sender believes was sent.

Ordering is the subtle part. Wall-clock timestamps from clients are unreliable — clocks skew. Use a per-conversation monotonic sequence number assigned server-side, or a Snowflake-style id that is time-sortable and globally unique. Then the client sorts by that, not by Date.now().

Storage. Extremely write-heavy, always read by conversation, almost never updated — that shape points at Cassandra or DynamoDB with a partition key of conversationId and a clustering key of the sequence number, so "the last 50 messages in this conversation" is one sequential read.

Group messages are the fan-out decision: for a 10-person group, write to all 10 inboxes. For a 100,000-member channel, that is 100,000 writes per message — so large groups switch to fan-out-on-read, where members pull from a shared conversation log. Naming that threshold explicitly is exactly the kind of trade-off this round rewards.

Presence is the sneaky scale problem: naive presence means every status change is broadcast to every contact, which is O(users × contacts) and will dominate your traffic. The real answers are a heartbeat with a TTL in Redis, and only pushing presence for conversations the user currently has open.

Everything follows from the 200:1 read-to-write ratio you computed at the start.

The real answer is hybrid, and this is what they are waiting for: fan-out on write for normal accounts, and for accounts above a follower threshold, do not fan out — merge their posts in at read time. Twitter's actual design, and it is the correct answer because it puts each strategy where its cost is lowest.

Store the feed as a capped list of post ids in Redis (say the newest 800), not the post bodies — hydrate the bodies from a cache or the database at read time, so an edited or deleted post does not need rewriting across millions of feed lists.

The framing that sets the tone: "this is the one system where I will trade availability and latency for consistency without hesitating."

• Double-entry ledger, append-only. Every transaction is two entries that sum to zero — debit one account, credit another. Balances are derived from the ledger, never stored as a mutable field you increment. This makes every balance auditable and makes a lost update impossible by construction.
• Idempotency everywhere. An idempotency key on every payment request, unique-indexed. A retry returns the original result rather than charging again.
• State machine, explicitly. initiated → authorised → captured → settled, with failed and refunded as terminal states. Only legal transitions allowed, enforced in the database rather than in the application.
• Webhooks plus reconciliation. Confirm on the provider's webhook, verify its signature, and dedupe on the provider's event id. Then run a scheduled job that queries the provider for anything still pending past its window — because webhooks are lost, and reconciliation is how real payment systems stay correct.
• Money as integer minor units. Never a float. Store the currency alongside every amount.
• Exactly-once effects via the saga above, with refund as the compensating action, and the human consequence acknowledged.

If they push on scale: payments are usually low-QPS and high-stakes, so the interesting scaling problem is not throughput but the ledger growing forever — which is solved with periodic balance snapshots so you never replay the whole history, plus partitioning by account and archiving cold periods.

The three pillars, and what each is actually for: metrics tell you something is wrong (cheap, aggregate, alertable); traces tell you where (per-request, sampled, expensive); logs tell you why (detailed, most expensive to store). Alert on metrics, diagnose with traces, confirm with logs — in that order.

SLI, SLO, error budget — the vocabulary of this round:

• SLI — the measurement. "Proportion of requests served under 300ms."
• SLO — the target. "99.9% over 30 days."
• Error budget — what 99.9% permits: about 43 minutes of failure per month. The budget is the point. It converts reliability from an argument into arithmetic: budget remaining means you can ship risky changes, budget spent means you stop feature work and fix reliability. Being able to explain that trade-off is a staff-level signal.

Alert on symptoms, not causes. Page on "checkout error rate above 1%" — a user-visible symptom — not on "CPU above 80%", which may be entirely fine. Every alert that does not require a human to act on it immediately should be a dashboard instead, because alert fatigue is how real outages get missed.

Also know: p99 over average, always — an average hides the tail, and the tail is what users actually complain about. Percentiles do not average across services, so you aggregate histograms, not percentiles. And the four golden signals — latency, traffic, errors, saturation — which is the shortest correct answer to "what would you monitor?"