Cookie preferences

Essential cookies keep this site working. Optional Google Analytics cookies help us understand page visits and improve the site. You can change your choice anytime. Cookie policy

Skip to content
ForCSGuy System Design Atlas home

Library / Case Studies

High-performance Replicated Log Service at Twitter

Building DistributedLog: High-performance replicated log service

IntermediateReplicationDistributed SystemsObservability

By Twitter engineers · Twitter · Added 2026-10-12

Open original source

All content belongs to Twitter engineers / Twitter. We only link to it and give credit. Report a problem

Learn this case study

About 37 min left. Your work is saved as you go.

What you are about to study

Twitter explains a real system. You should know the basics: caching, queues and replication.

Ideas you will meet

Replication
Leader/follower, multi-leader and leaderless replication.
Distributed Systems
Consensus, time, failure and the fundamentals of scale-out.
Observability
Metrics, logs, traces, SLOs and alerting.

What we already know about this

Our own lessons on these ideas. Read them here before you start, so it makes sense.

  • Databases Deep Dive · 12 minReplication topologies and multi-region databasesSingle-leader, multi-leader and leaderless replication, how they handle conflicts and lag, and what changes across regions.Read here

    Why replicate

    Replication keeps copies of data on several machines for three reasons: availability (survive a node failure), read scaling (serve reads from replicas) and latency (put data near users). The hard part is keeping copies consistent while writes continue. (Basics are in the replication and sharding lesson of the foundations course; this lesson compares the topologies.)

    Single-leader (primary and replicas)

    One node, the leader (primary or master), accepts writes and ships its changes to followers (replicas), which serve reads.

    Figure SDA-DDV-04-1 · Single-leader replication · © forcsguy.com

    • Synchronous replication waits for a follower before acknowledging: safer, slower, and a stalled follower can block writes. Asynchronous is faster but a failover can lose the latest writes. A common compromise: one synchronous follower, others asynchronous.
    • Failover must detect failure, choose the most up-to-date replica, redirect clients and ensure the old leader does not come back believing it still leads (split brain).
    • Replication lag causes stale reads. Techniques: read-your-writes (route a user's reads to the leader shortly after they write), monotonic reads (keep a user on one replica), and checking replica position.

    Multi-leader

    Several nodes accept writes (often one per region) and replicate to each other. Benefits: local write latency and continued operation if a region is cut off. Cost: write conflicts.

    Conflict handling options:

    • Avoid them: route all writes for one record to one leader (for example by user or region).
    • Last write wins by timestamp: simple, but silently drops one of two concurrent updates and depends on clocks.
    • Merge values (union of sets, counters) using mergeable data types (CRDTs; see the collaborative editing lesson).
    • Custom resolution in application code, or ask the user.

    Multi-leader also appears in offline-capable apps (each device is a leader) and collaborative editing.

    Before: single leader

    After: multi-leader across regions

    What changed

    • Users in each region write to a nearby database
    • Concurrent edits to the same row need conflict handling
    • More moving parts to run and monitor

    Leaderless (Dynamo-style)

    Any replica accepts writes; clients write to several replicas and read from several, using quorums (W + R > N) to get fresh data (see the key-value store lesson). Background anti-entropy and read repair fix stale copies; versions or vector clocks detect conflicts. It favours availability and is typical of Cassandra and DynamoDB-style stores.

    Figure SDA-DDV-04-2 · Choosing a replication model · © forcsguy.com

    Multi-region designs

    • Active-passive: one region serves writes; another is a warm standby. Simple; failover loses a little data (the replication lag) and takes minutes.
    • Active-active: several regions serve traffic. Requires either partitioning data by home region (each record has one home, cross-region reads are replicas) or true multi-leader conflict handling.
    • Global databases use consensus across regions (see the consensus lesson) for strong consistency at the cost of cross-region write latency; some offer tunable consistency per query.
    • Latency numbers matter: a cross-ocean round trip is around 100 ms or more, so synchronous cross-region replication adds that to every write.
    • Data residency laws may require keeping certain users' data in a region.

    Partitioning and replication together

    Real systems shard data into partitions and replicate each partition (a leader per partition spreads write load across nodes). Rebalancing, hot partitions and cross-partition transactions are the new problems (see sharding and the consistent-hashing lesson).

    Master/slave naming

    You will see "master/slave", "primary/replica" and "leader/follower" for the same single-leader idea; most current documentation prefers the last two.

    Checklist

    • What happens to writes if the leader or a region is lost, and how long?
    • How much data can we lose (RPO) and how long can we be down (RTO)?
    • How do we handle stale reads from replicas?
    • Who resolves conflicts, and how?
    • Have we practised failover?

    Real-world reading

    Related reading

    Show 6 more

    Open this lesson with its quiz and exercise

  • System Design Foundations · 11 minReplication and shardingTwo ways to spread data over many machines: copy it for safety and read speed, or split it for capacity.Read here

    Replication: keep copies

    Replication keeps the same data on more than one machine. Reasons:

    • Availability: if one machine dies, another has the data.
    • Read capacity: reads can go to any copy.
    • Locality: place a copy near users in another region.

    The most common arrangement is leader and followers. All writes go to the leader, which sends the changes to followers. Reads can use followers.

    Figure SDA-FND-05-1 · Leader and follower replication · © forcsguy.com

    Many teams hide the split between the primary and the replicas behind a small router, so the application does not need to know:

    Figure SDA-FND-05-2 · Read and write splitting through a router · © forcsguy.com

    Synchronous or asynchronous

    • Synchronous: the leader waits for followers to confirm before saying "done". Safer, but slower, and a down follower can block writes.
    • Asynchronous: the leader confirms immediately and followers catch up. Fast, but a follower can be behind, and a leader crash can lose the most recent writes.

    Most systems use asynchronous replication for most followers, sometimes with one synchronous follower as a safety net.

    Replication lag

    Because followers can lag, a user may write something and then read an older value from a follower. Fixes include reading the user's own recent writes from the leader, or tracking how fresh each replica is. This is the first taste of a bigger idea, consistency, covered in lesson 7.

    When the leader fails

    A follower must be promoted (failover). That needs failure detection and agreement on who the new leader is, so that two machines do not both believe they are the leader (a split brain). This is why teams rely on tested tooling or managed databases for failover rather than hand-written scripts.

    Read-after-write consistency

    With asynchronous replicas, a user can save something and then not see it: the write went to the primary, the next read hit a replica that is a few seconds behind. Three common fixes:

    • Send reads that must be fresh (the user's own profile, an order just placed) to the primary.
    • For a short window after a user writes, route that user's reads to the primary.
    • Let the router check whether a replica has caught up to the user's last write; if not, read from the primary or wait.

    A database proxy in front of the cluster can do this routing so application code stays simple, at the cost of one more hop and one more component that must itself be highly available.

    Before: one database

    After: primary with read replicas

    What changed

    • Reads are spread across copies
    • A copy can be promoted if the primary fails
    • Replicas may lag slightly behind the primary

    Sharding: split the data

    Replication does not help when the data or the writes no longer fit on one machine. Sharding (partitioning) divides the data so each shard holds a part. Each shard is usually replicated too.

    Figure SDA-FND-05-3 · Sharding by key range · © forcsguy.com

    Choosing a shard key

    The shard key decides which shard owns a row. A good key:

    • Spreads data evenly, so no shard is much bigger than the others.
    • Spreads traffic evenly, so no shard is a hot spot.
    • Matches your queries, so most queries touch one shard.

    Common schemes:

    SchemeHowWatch out for
    Rangekey ranges per shardhot spots, e.g. newest dates
    Hashhash(key) mod Nadding shards remaps most keys
    Directorya lookup table says where each key livesthe directory is another thing to run and scale
    Consistent hashingkeys and nodes placed on a ringneeds virtual nodes to balance well

    Consistent hashing in one picture

    Place both servers and keys on a ring. A key belongs to the first server clockwise from it. Adding a server only takes keys from its neighbour; the rest do not move.

    Figure SDA-FND-05-4 · Consistent hashing ring · © forcsguy.com

    Horizontal and vertical partitioning

    Sharding is horizontal partitioning: the rows of a table are split across machines, each shard holding the same columns for a different set of rows. Vertical partitioning splits by columns or by feature instead: rarely used or large columns (a profile photo, a long description) move to a separate table or store, or whole groups of tables go to different databases (users on one, orders on another). It is simpler than sharding and keeps frequently read data small and fast, but it does not help when a single table's rows outgrow one machine.

    Figure SDA-FND-05-5 · Splitting a table by rows and by columns · © forcsguy.com

    What sharding costs you

    • Cross-shard queries and joins become slow or need extra work.
    • Transactions across shards are hard (lesson 8 touches on how systems cope).
    • Rebalancing when a shard gets too big is operational work.
    • More moving parts to monitor.

    A worked example: sizing replicas and shards

    Replicas for reads. A social app has 20,000 reads and 1,000 writes per second. One database node can do about 5,000 simple operations per second. A primary alone is hopeless. With writes of 1,000 per second on the primary, you have capacity left for about 4,000 reads, so you need at least four replicas to serve 20,000 reads (16,000 on replicas plus 4,000 on the primary) with no headroom, so six in practice. Adding replicas scales reads almost linearly. It does nothing for the 1,000 writes, which still all go to one primary.

    Shards for writes. If writes grow to 12,000 per second, no single primary can take them. Split the data into shards, each with its own primary: with 4 shards each takes about 3,000 writes per second. Each shard then has its own replicas for reads. Your final count is shards x (1 primary + replicas), for example 4 x (1 + 2) = 12 database nodes.

    Figure SDA-FND-05-6 · Replicas scale reads, shards scale writes · © forcsguy.com

    Walking through a failover

    Replication is only useful if you know what happens when a primary dies. With asynchronous replication:

    1. The primary acknowledges writes 100 to 140, but the follower has only received up to 135 (replication lag).
    2. The primary crashes. Writes 136 to 140 exist only on the dead machine.
    3. A monitor notices missing heartbeats, usually after several seconds, and promotes the most up-to-date follower to primary.
    4. Clients are redirected. Writes 136 to 140 are lost (or recoverable only when the old primary returns).
    5. When the old primary comes back it must not accept writes as if nothing happened (that would create two primaries, "split brain"); it rejoins as a follower and discards or reconciles what the new primary does not have.

    With synchronous replication to at least one follower, step 2 loses nothing, but every write waits for that follower, adding latency and making writes fail if the follower is unavailable. Many systems choose a compromise: wait for one follower to confirm (semi-synchronous), accepting a little latency to avoid losing acknowledged writes.

    Figure SDA-FND-05-7 · What an asynchronous failover can lose · © forcsguy.com

    Choosing a shard key: three examples

    • User id for a messaging app. Almost every query is "this user's messages", so a hash of the user id keeps each user's data together and spreads users evenly. The pain: a query across all users (an admin search) must ask every shard.
    • Order date for an orders table. Range sharding by date makes "last month's orders" cheap, but all new writes land on the newest shard, a hot spot, while old shards sit idle. A better key adds a hash of the customer id.
    • Country for a multi-region product. Good for keeping data near users and meeting data-residency rules, but shards are uneven: one country may hold half your users.

    A good key has high cardinality (many distinct values), even load and matches your most common query. Changing it later means moving most of your data, so test it against a sample of real traffic first.

    Mistakes people make

    • Reading from a replica right after writing. The replica may not have the write yet, so users do not see their own change. Read your own writes from the primary for a short time, or route that user to the primary.
    • Sharding too early. Sharding multiplies operational work: backups, schema changes, cross-shard queries and transactions all get harder. First try indexes, caching, replicas and a bigger machine.
    • Counting on replicas as backups. A mistaken delete is replicated within milliseconds. Keep real backups and test restoring them.
    • Forgetting to rebalance. Shards grow unevenly. Plan how to split a shard, ideally with consistent hashing or many small virtual shards that you can move between machines.

    Try it yourself

    In the System Builder, open "The database goes down". Run it with no replicas and see reads and writes fail, then add replicas and a write queue and compare. For hashing itself, build the consistent hash ring lab and measure how many keys move when you add a node. To place replicas and a queue yourself, open the database-outage design on the free canvas and kill the database.

    More on this site

    Related reading

    Show 6 more

    Open this lesson with its quiz and exercise

  • System Design Foundations · 9 minObservability and running a systemMetrics, logs and traces, how to define 'healthy' with SLOs, and how to ship changes without breaking things.Read here

    You cannot fix what you cannot see

    A system you cannot observe is a system you cannot trust. Observability is the ability to understand what is happening inside from the outside. Three kinds of signal work together.

    SignalWhat it isBest for
    MetricsNumbers over time (requests per second, error rate, CPU)Dashboards, alerts, trends
    LogsTimestamped records of eventsDetails of what happened
    TracesThe path of one request through many servicesFinding where time is spent

    Figure SDA-FND-11-1 · From alert to root cause · © forcsguy.com

    What to measure

    A simple checklist, the four golden signals:

    1. Latency: how long requests take (use percentiles).
    2. Traffic: how much demand there is.
    3. Errors: how many requests fail.
    4. Saturation: how full the busiest resource is (CPU, memory, connections, disk, queue depth).

    Averages hide pain. If the average is 100 ms but p99 is 4 seconds, one user in a hundred is waiting four seconds, and with many requests per page that is most of your users sometimes.

    Defining healthy: SLIs, SLOs, error budgets

    • An SLI (indicator) is a measurement, like "fraction of requests answered successfully in under 300 ms".
    • An SLO (objective) is the target, like "99.9% over 30 days".
    • The error budget is what is left over: 0.1% of requests may fail. While there is budget, ship features; when it runs out, focus on reliability.

    Alert on symptoms users feel (error rate, latency) more than on causes (one machine's CPU). Page a human only for things that need action now.

    Ship changes safely

    Most outages come from changes. Reduce the risk:

    • Small, frequent releases are easier to understand and undo than big ones.
    • Canary release: send a few percent of traffic to the new version and watch the metrics before going wider.
    • Feature flags: deploy code turned off, then enable it gradually, and switch it off instantly if needed.
    • Blue-green: keep the old version running so you can switch back in seconds.
    • Backwards-compatible changes: database and API changes should work with both old and new code during rollout.
    • Fast rollback must be a rehearsed, boring button.

    When things go wrong

    Have an incident process: someone leads, someone communicates, mitigate first and investigate later. Afterwards write a blameless postmortem that asks "how did our system and process allow this?" rather than "who did this?". That is how reliability actually improves.

    A worked example: error budgets and burn rate

    Your service has an SLO of 99.9% of requests succeed over 30 days, and it serves 10 million requests a month.

    • Error budget: 0.1% of 10,000,000 = 10,000 failed requests may happen in the month before you break the promise.
    • Burn rate is how fast you spend it. A burn rate of 1 uses exactly the budget over 30 days. If 5% of requests are failing now, the error rate is 50 times the allowed 0.1%, a burn rate of 50: the whole month's budget is gone in 30 / 50 = 0.6 days, about 14 hours.
    • Alert on burn rate, not on a single bad minute. A common pattern watches two windows together: page someone if the burn rate over the last hour is above about 14 and the last 5 minutes agree (a fast burn: 2% of the monthly budget gone in an hour). Open a ticket for a slower burn, say rate above 6 over 6 hours. Two windows avoid paging for blips and avoid missing slow leaks.

    Why this beats "alert when error rate is above X": it ties alerts to what users were promised, it pages only when the budget is in danger, and it gives the team a rule for trade-offs: when the budget is healthy, ship faster; when it is nearly spent, slow down and fix reliability.

    Figure SDA-FND-11-2 · Alerting on how fast the error budget burns · © forcsguy.com

    Finding the cause with all three signals

    A worked investigation shows why you want metrics, traces and logs together.

    1. A metric alerts: the checkout success rate dropped and p99 latency rose. Metrics tell you that something is wrong and how bad, cheaply and over time.
    2. Narrow it down by dimension: break the metric down by region, version and endpoint. Only the new version in one region is affected, a strong hint it is a deploy.
    3. A trace shows where the time goes: one slow checkout request shows 4.8 of its 5 seconds in a call to the payment provider's tokenisation endpoint.
    4. Logs give the detail: filter logs by the trace id from that request and read the exact error: a timeout talking to a new tokenisation URL.
    5. Fix and verify: roll back the version, and watch the metric recover.

    Figure SDA-FND-11-3 · From an alert to a root cause · © forcsguy.com

    The glue is a correlation id (trace id) carried on every request and written into every log line, so you can jump from one signal to the others.

    Mistakes people make

    • Alerting on causes instead of symptoms. "CPU is 85%" may be fine; "users are seeing errors" never is. Page on symptoms users feel, and use cause metrics for diagnosis.
    • Averaging percentiles. The mean of two hosts' p99 is meaningless; merge distributions instead (see the t-digest lesson).
    • High-cardinality labels. A label such as user_id on a metric creates a new time series per user, which can multiply storage and cost by millions. Put such detail in logs or traces.
    • Logs without structure or ids. Free text is hard to query. Log JSON with a trace id, service and version.
    • Dashboards no one owns. A wall of graphs nobody reads is noise. Build a small dashboard per service around the golden signals: latency, traffic, errors and saturation.
    • No runbook. An alert that does not say what to check first wastes the first ten minutes of every incident.

    Try it yourself

    Run any test in the System Builder and read its charts as if they were production graphs: which signal moves first, and which part would you look at next? Build the search metrics lab to practise percentile maths, and read the t-digest lesson for how real systems summarise latency at scale.

    Real-world reading

    Related reading

    Show 6 more

    Open this lesson with its quiz and exercise

Continue to Twitter

Go deeper on this

A path through videos, articles, other case studies and code on the same ideas as High-performance Replicated Log Service at Twitter.

  1. Warm upGet the vocabulary first.

  2. WatchHow others solve the same problem, on video.

    Videos belong to their creators and play from YouTube. More on YouTube (opens in a new tab)

  3. ReadEngineers explain it in their own words.

  4. See it in another companyThe same ideas, a different system.

  5. Read the codeSee how it looks when it is real.

Keep learning

Loading community…