Tapping “Like” is the smallest interaction Facebook has. It is one bit: you either like a post or you do not. It looks like a toggle on a boolean. It is not. Behind that single tap sits a counter that must be accurate (a user liking twice counts once, an unlike must decrement, the number cannot drift over a billion taps), durable (unlike a live viewer count, a like is a real fact the user expects to persist forever), and live (the number should tick up on every viewer’s screen as others tap, without a refresh). And then there is the part that actually breaks systems: a celebrity or a viral post can take 100,000 likes per second, all landing on the count of a single post. One row. One key. A firehose.

That last fact reframes the whole problem. This is not a “store a number” problem, it is a hot-key write-contention problem wrapped around an idempotent, durable counter, with a real-time fanout tail. The naive design - UPDATE posts SET likes = likes + 1 WHERE id = ? - is correct and completely useless: every one of those 100K taps per second serializes on a single row lock, and the row melts. The interview is really asking three questions at once: how do you take 100K writes/sec onto one logical counter without contention, how do you keep that counter exactly right under toggles and retries, and how do you push the changing number to millions of viewers live. Let me build all three.

Functional Requirements (FR)

In scope:

  • Like a post. A user taps like on a post. It is recorded, and the post’s total like count reflects it.
  • Unlike a post. A user removes their like. The count decrements. Like/unlike is a toggle per (user, post) - a user contributes at most one like to a post, ever, no matter how many times they tap.
  • Idempotency. Tapping like twice (double-tap, retry, network replay) counts once. Tapping unlike when not liked is a no-op. The count must never double-count or go negative.
  • Read the count. Anyone viewing a post sees its total like count. This is read enormously more often than it is written.
  • “Did I like this?” For the current viewer, show whether they personally liked the post (the button state), which is a per-user membership question, not the aggregate.
  • Live updates. While a user is looking at a post, the count ticks up (and down) in near real time as other people like and unlike it, without a page reload.
  • Who liked it (sample). Show a small sample or a “liked by Alice, Bob and 4,201 others” line. We serve a sample of likers plus the total, not the full million-name list.

Explicitly out of scope (own the scope):

  • Reactions (love, haha, wow). The real system generalizes a like to N reaction types. That is one extra dimension (a small enum per row) and does not change the architecture; I will note where it slots in but design for a single like to keep the core sharp.
  • The post itself, feed ranking, notifications. Creating posts, the news feed, and “your friend liked this” notifications are separate systems that consume like events. We produce the count and the events; we do not build the feed.
  • Comments and shares. Separate counters with the same shape - solved identically.
  • Spam / like-fraud detection. We note where a fraud filter hooks in but do not design the abuse pipeline.
  • Full liker list pagination for a viral post. We serve total + a sample + a “did you like it.” Enumerating all 3M likers is a separate cold analytical read.

The one decision that drives everything: a like is an idempotent, durable membership fact - (user_id, post_id) exists or it does not - and the count is a derived aggregate over that membership. Unlike a live viewer count (approximate, ephemeral, rebuildable), this count must be exactly right and persistent. That single difference - durability plus accuracy under a 100K/sec hot key - is the entire challenge, and it forbids the cheap tricks (pure approximation, TTL decay) that make ephemeral counters easy.

Non-Functional Requirements (NFR)

  • Scale: ~2B users, ~500M posts created/day. Global like writes on the order of ~1M/sec average, with hot posts spiking to 100K likes/sec on a single post. Reads dwarf writes: like counts are rendered on essentially every feed impression, on the order of ~10M reads/sec globally.
  • Latency: the like tap must feel instant - p99 write-ack under ~100ms (the client optimistically flips the button immediately regardless). Count reads served from cache in single-digit ms. Live update propagation to viewers within ~1-2s is fine.
  • Availability: 99.99% on the read path (a feed that cannot show like counts is broken). Writes must be highly available too, but a like can be accepted and processed asynchronously - we ack fast and settle the count behind it.
  • Consistency: read-your-own-writes for the actor (when I like a post, my button must show liked and the count must include me immediately on my next read), but eventual consistency for everyone else’s view of the aggregate (other viewers can lag the true total by a second or two). The count converges to exactly correct.
  • Durability: strong. A like is a persisted user action. Once acked, it must survive node loss. This is the hard contrast with the ephemeral-counter problems - we cannot keep this only in RAM and shrug off a crash.
  • Accuracy: the count must be exactly correct at rest - the sum of distinct current likers, never double-counting a retry, never negative, no permanent drift. Transient display lag is fine; permanent error is not.

The tension to name out loud: accuracy demands durable, deduplicated, per-user state; the hot key demands we never contend on one row; and freshness demands we push changes live. These pull in different directions. A design that nails accuracy naively (one row, one lock) dies on the hot key. A design that nails the hot key naively (fire-and-forget increments) loses accuracy under retries and unlikes. The whole art is satisfying all three at once.

Back-of-the-Envelope Estimation (BoE)

Real numbers, because they dictate the architecture.

Write load:

Total likes/sec (global average):        ~1,000,000 writes/sec
Peak single-post hot key (celebrity):    ~100,000 likes/sec onto ONE post_id
Ratio of unlikes to likes:               ~5% (people undo, but mostly like)
Idempotent retries (client/network):     assume ~10-20% of taps arrive >1x

The killer is not the 1M/sec aggregate - that shards cleanly across posts. The killer is 100K/sec onto a single post_id. INCR on one Redis key tops out well below that on a single shard, and a SQL row with row-level locking serializes far lower. One key cannot take it. This one number forces sharded counters.

Read load:

Feed impressions/sec (global):           ~10,000,000/sec
Each impression renders like counts for  ~10 posts on screen
Naive count-reads/sec:                    ~100,000,000/sec of "get count"
Plus "did I like this?" per (viewer, post) membership checks at the same rate.

100M count-reads/sec is the read wall. It is only survivable because a count is a tiny, slowly-changing integer that is cacheable to the extreme - CDN/edge for the hottest posts, in-memory replicas for warm ones.

Storage of the membership (the durable truth):

Total likes ever (lifetime):    say ~100 trillion is unrealistic; take active window.
Likes stored per row: (user_id 8B, post_id 8B, ts 8B, +overhead) ~ 40B
Daily new likes: ~1M/sec * 86,400 ~ 86B likes/day
  = 86B * 40B ~ 3.4 TB/day of raw like edges
Per year: ~1.2 PB/year of like membership edges (before compression).

That is a lot, but it shards flat by key and compresses well. The point: the membership table is huge and must be on disk (sharded persistent store), not RAM. Only the counts and hot membership live in cache.

Count storage:

500M posts/day created, keep counts for active posts (say ~50B active posts)
Per post: post_id + count(s) ~ 20B -> 50B * 20B = ~1 TB of counts
Hot working set (posts being actively viewed): far smaller, fits in a Redis fleet.

Live fanout bandwidth:

A viral post with 3M concurrent viewers, count changing ~100K/sec.
Naive: push every increment to every viewer = 3M * 100K = 3e11 msgs/sec. Absurd.
Coalesced: push the current count at most ~1/sec per viewer
  = 3M viewers * 1 update/sec * ~30B = ~90 MB/sec for that one post. Manageable.

That naive-vs-coalesced gap (300 billion msgs/sec vs a few per second per viewer) is the whole fanout design in one line: never push per-increment; push a coalesced current value on a tick.

Takeaways: the aggregate write rate shards fine; the single-post 100K/sec hot key is the write wall; 100M/sec count-reads is the read wall, beaten by caching a tiny integer; the membership table is PB-scale and durable on disk; and live fanout is only viable coalesced. Every technique below attacks one of those four.

High-Level Design (HLD)

Four planes. A write plane accepts a like tap, makes it idempotent and durable, and updates a sharded counter so no single row is contended. A count-read plane serves the aggregate from cache at massive read rates. A membership plane answers “did user X like post Y” and stores the durable truth. A live plane pushes coalesced count changes to viewers. The spine connecting them is an event log: a like is written once as a durable event, and counts, membership, and live updates are all derived from it.

   Client tap (Like/Unlike)                         Feed render (read count)
        │ POST /like  {post_id, user_id, action}         │ GET count / did-I-like
        │ (optimistic UI flips instantly)                ▼
        ▼                                          ┌───────────────────┐
  ┌──────────────┐                                 │  Edge / CDN cache  │ hot posts
  │ Like Write   │                                 │  (count TTL ~1s)   │
  │ Service (API)│                                 └─────────┬──────────┘
  │ - idempotency│                                           │ miss
  │   key check  │                                 ┌─────────▼──────────┐
  │ - append     │                                 │  Count Read API    │
  │   like event │                                 │  (in-mem replicas) │
  └──────┬───────┘                                 └─────────┬──────────┘
         │ append (durable, ordered)                         │ read
         ▼                                                   ▼
  ┌──────────────────────┐                          ┌────────────────────┐
  │  Like Event Log       │  consumers              │  Count Store        │
  │  (Kafka, partitioned  │────────┬──────────────▶ │  (Redis, SHARDED    │
  │   by post_id)         │        │                │   counters per post)│
  └──────────────────────┘        │                └─────────┬──────────┘
         │                        │                          │ count changed
         ▼                        ▼                          ▼
  ┌──────────────────┐   ┌──────────────────┐        ┌────────────────────┐
  │ Membership Store  │   │ Counter Updater  │        │  Live Fanout Tier   │
  │ (sharded KV/SQL:  │   │ - dedup by idem  │        │  (WebSocket/SSE,    │
  │  (user,post)->1)  │   │ - +/- sharded    │        │   coalesced ticks)  │
  │  durable truth    │   │   counter        │        └─────────┬──────────┘
  └──────────────────┘   └──────────────────┘                  │ push count
                                                        viewers' open sockets

Write flow (a user taps like):

  1. The client optimistically flips the button to “liked” and increments the number it shows locally - the tap must feel instant. It fires POST /like with {post_id, user_id, action: like|unlike, idempotency_key}.
  2. The Like Write Service validates and checks the idempotency key (derived from (user_id, post_id, action) plus a client nonce). If this exact action was already recorded, it returns the prior result - no double count. Otherwise it appends a like event to the durable, ordered Like Event Log (Kafka), partitioned by post_id, and acks the client. The ack is fast because durability is the log append, not a synchronous counter update.
  3. Consumers of the log do the real work asynchronously: the Membership Store consumer upserts the durable (user_id, post_id) edge (and detects like-after-like / unlike-when-not-liked as no-ops, the second line of idempotency defense). The Counter Updater consumer applies +1 / -1 to the sharded counter for the post.

Count flow (deriving the number):

  1. The post’s like count is stored as N sub-counters (shards) in the Count Store, not one. The Counter Updater increments a random or worker-pinned shard, so 100K writes/sec spread across N keys, each taking 100K/N. The logical count is the sum of the N shards.
  2. When the summed count changes, it is written to a read-optimized display count and published so caches and the live tier refresh.

Read flow (feed shows the number):

  1. A feed render calls GET /posts/{id}/likes. Hot posts are served from the edge/CDN cache (TTL ~1s); warm posts from an in-memory replica on read nodes; only cold misses touch the Count Store (where they read the pre-summed display value, not the N shards). The “did I like this?” state is a separate per-user membership lookup, cached per viewer.

Live flow (the number ticks):

  1. A viewer looking at a post holds a connection to the Live Fanout Tier (WebSocket/SSE), subscribed to that post’s count channel. When the count changes, the tier pushes the new coalesced value at most ~1/sec per post, not per increment. The viewer’s number updates live.

The core insight: write once to a durable, ordered log, then derive counts, membership, and live updates asynchronously from it - and shard the one contended counter. The log gives durability and accuracy (a replayable source of truth with idempotency); sharded counters beat the hot key; caching beats the read wall; coalesced push beats the fanout wall. Each plane attacks exactly one of the four BoE walls.

Component Deep Dive

The hard parts, naive-first then evolved: (1) the 100K/sec hot-key counter, (2) idempotency and accuracy under retries and toggles, (3) the read wall and “did I like this,” (4) live fanout without a push storm.

1. The hot-key counter - 100K likes/sec on one post

Naive approach: one counter, incremented per like. UPDATE posts SET likes = likes + 1 WHERE id = ?, or in Redis INCR likes:{post_id}.

on like(post_id):   INCR likes:{post_id}      # ONE key
on unlike(post_id): DECR likes:{post_id}
read:               GET  likes:{post_id}

Where it breaks:

  • Single-row / single-key contention. A SQL row-level write lock serializes every increment; at 100K/sec the lock queue is the bottleneck and latency explodes into seconds. Even Redis INCR, single-threaded per shard, tops out well under 100K/sec once you add network and the key lives on exactly one shard - you cannot spread one key across shards.
  • No idempotency. INCR on every request counts retries and double-taps. The number inflates.
  • Unlike accuracy. DECR on unlike is only correct if the user actually had a like; a spurious unlike (or a replayed one) drives the count wrong or negative.

The hot key alone kills this. One logical counter is physically one thing, and one thing cannot absorb 100K writes/sec.

First evolution: sharded counters (split one counter into N). Represent the post’s count as N sub-counters on N different shards. A write increments one of the N (chosen by worker id, or hash(user_id) % N, or random); a read sums all N. The write load on any single sub-counter is 100K / N - pick N = 100 and each shard takes 1K/sec, trivial.

Count Store, per post, N shards:
  INCRBY likes:{post_id}:{shard}  +1        # shard in [0, N)
  count(post) = SUM over s in [0,N) of GET likes:{post_id}:{s}
Choose N per post by heat:  cold post N=1, warm N=10, celebrity N=100+

The trade: reads now cost N lookups instead of 1. That is fine because we do not read the shards on the hot path - a background summer rolls the N shards into a single display:{post_id} value every ~200ms, and all the feed reads hit that one pre-summed value from cache. So writes spread across N keys, reads hit one cached key, and the N-way read cost is paid once per tick by a background job, not 100M times/sec by feeds.

Write path:   spread across N shard keys      -> 100K/N per key
Summer job:   every ~200ms, sum N -> display  -> N reads per post per tick
Read path:    GET display:{post_id} (cached)  -> one tiny integer

Second evolution: adaptive shard count + write batching. Two refinements make it production-grade:

  • Adaptive N. Sharding every post 100 ways wastes memory on the billions of cold posts (one like ever) and is not enough for a mega-viral post. So N is per-post and elastic: a post starts at N=1; when its write rate crosses a threshold (detected by the Counter Updater), it is promoted to more shards, and demoted back when it cools. The count is invariant to N changes because it is always the sum of whatever shards exist.
  • In-worker batching. The Counter Updater consumes the like log in batches. Instead of one INCRBY +1 per event, it aggregates a batch (e.g. 200ms or 10K events) per post and issues one INCRBY +delta. A celebrity post’s 100K likes/sec, batched at 200ms across, say, 20 consumer workers, becomes ~20 INCRBY operations of ~1000 each per 200ms - a rounding error of write ops. Batching is what truly tames the firehose; sharding then spreads even that residue.
Counter Updater (per consumer, per 200ms window):
  batch = drain like events for partitions I own
  per post_id: net = (#likes - #unlikes) in batch, deduped by idempotency
  INCRBY likes:{post_id}:{my_shard}  net     # one op per (post, worker) per window

Batching + sharding together turn 100K raw taps/sec on one post into a handful of coalesced INCRBYs across a handful of keys. The firehose becomes a trickle of large, cheap, spread-out writes.

ApproachWrite cost on hot postRead costAccuracyVerdict
Single row/key INCR100K/sec on 1 lock/key -> melts1 readno idempotencydies
Sharded counters (N)100K/N per keyN reads (summed offline)needs dedupsurvives
Sharded + batched + adaptive Na few INCRBYs/sec per post1 cached readdedup in batchproduction

2. Idempotency and accuracy - the count must be exactly right

Naive approach: trust the request; INCR on like, DECR on unlike. Every POST /like bumps the counter.

Where it breaks:

  • Retries double count. The client retries on a timeout (it got no ack but the write went through). Two INCRs, one like.
  • Toggle races. A user rapidly taps like/unlike/like. Out-of-order processing can leave the counter and the membership disagreeing (count says liked, membership says not).
  • Unlike-when-not-liked. A replayed or spurious unlike DECRs a like that was never there - count drifts down, can go negative.
  • No source of truth. With only a counter, there is no way to recompute or audit the number. Any drift is permanent.

Accuracy under retries and toggles is where the durable-counter problem is genuinely harder than the ephemeral one. We cannot approximate our way out; the number is a real fact.

The fix: membership is the truth, the counter is derived, and both are made idempotent.

  • Membership store is authoritative. The durable truth is the set of (user_id, post_id) edges. Liking is PUT (user, post); unliking is DELETE (user, post). These are idempotent by construction: PUT an existing edge is a no-op, DELETE a missing edge is a no-op. The count is defined as the cardinality of that set - so it can always be recomputed and audited, and it can never be wrong for long.
  • The counter only moves on a real membership transition. The Counter Updater applies +1 only when a like actually created a new edge, and -1 only when an unlike actually removed an existing edge. It learns this from the membership upsert result (did the row change?), not from the raw request. A retry that hits an already-liked edge produces no counter change. This binds counter deltas to genuine state transitions.
process like(user, post):
  changed = membership.put(user, post)     # returns true only if newly added
  if changed: enqueue counter_delta(post, +1)
process unlike(user, post):
  changed = membership.delete(user, post)   # returns true only if it existed
  if changed: enqueue counter_delta(post, -1)
  • Idempotency key on the write API. Each tap carries idempotency_key = hash(user_id, post_id, action, client_nonce). The Like Write Service dedups on it (a short-TTL seen-keys cache plus the log’s own key), so a network-level replay is dropped before it becomes an event. Two layers - the idempotency key at ingress and the membership transition at apply - mean neither retries nor reorders can double count.
  • Ordering for toggle races. The event log is partitioned by post_id, and within a partition events for a given (user, post) are processed in order by a single consumer, so like-then-unlike for one user resolves deterministically to the final state. The membership store’s compare-and-set on the edge is the tiebreaker if a stale event arrives late (each edge carries a last-write timestamp; older writes lose).
  • Periodic reconciliation. Because membership is the source of truth, a background job periodically recomputes count = COUNT(*) of edges for post for hot/audited posts and corrects any counter drift from lost or double-applied deltas. The sharded counter is a fast cache of a number that can always be re-derived exactly - so even a rare bug self-heals.

The model: the counter is a fast, sharded, eventually-consistent cache of an exact quantity (the size of a durable, idempotent membership set). Deltas are bound to real state transitions, retries are dropped at two layers, toggles are ordered per post, and the true value is always recomputable. That is how you get 100K/sec throughput and an exactly-correct number.

3. The read wall and “did I like this?”

Naive approach: read the count (and membership) from the store on every feed render. Each of 10 posts on screen does GET count + GET (viewer, post) membership.

Where it breaks:

  • 100M reads/sec at origin. Feed impressions render like counts constantly; reading through to the Count Store for each melts it, and summing N shards per read multiplies the cost.
  • Membership fan-out. “Did I like this?” is a per-(viewer, post) point read; a feed of 10 posts is 10 membership reads, times 10M feeds/sec.

The fix: cache the count hard, and batch/cache membership per viewer.

  • Count is a tiny, slow, cacheable integer. Serve display:{post_id} from an edge/CDN cache for hot posts (a viral post read millions of times/sec is served from the edge, TTL ~1s) and from in-memory replicas on read nodes for warm posts. Origin sees ~1 read per post per TTL. The N-shard sum is never on this path - it was pre-summed by the background summer.
  • Approximate display for huge numbers, exact for small. Reuse the same accuracy insight as any counter: a post at 12 likes must show exactly 12; a post at 2.3M can show “2.3M” and nobody notices +/-1000. So small counts are served exact, huge counts are rounded for display (and updated less aggressively), which relaxes the freshness pressure on exactly the hottest posts.
  • Membership: batch and cache per viewer. The “did I like this?” check for a feed is one batched membership read (did user U like posts [p1..p10]), served from a per-user liked-set cache (a compact bitmap/Bloom-backed set of the user’s recent likes in Redis). A Bloom filter answers “definitely not liked” instantly for the overwhelming common case; only possible-hits fall through to an exact membership read. This collapses 10 point reads into one cache hit.
GET /posts/{id}/likes         -> edge cache (TTL 1s) -> replica -> origin display
GET /me/liked?posts=p1,..,p10 -> per-user liked-set cache (Bloom + exact fallback)
Response: { "p1": {count: 2300000, liked: true, sample:[...] }, "p2": {...} }
  • Read-your-own-writes for the actor. When I like a post, my client already flipped optimistically, and my next read is served a value that includes my like: the Like Write Service writes my membership to my per-user liked-set cache synchronously on ack, and bumps a local session view of the count, so I never see my own like “disappear” even while the global aggregate is still settling. Everyone else gets eventual consistency, which is fine.

The read mantra: the count is one tiny cached integer summed offline; membership is a batched, Bloom-fronted per-user set; and the actor’s own like is reflected immediately while the global number settles behind it.

4. Live fanout without a push storm

Naive approach: push every increment to every viewer. When the count changes, notify all connected viewers of the post.

Where it breaks:

  • Quadratic blowup. A viral post with 3M concurrent viewers whose count changes 100K times/sec would push 3M * 100K = 3e11 messages/sec. Impossible, and pointless - no human perceives 100K updates/sec.
  • Connection state. Tracking which of hundreds of millions of open sockets cares about which post, and delivering per-increment, is untenable.

The fix: coalesced, tick-based push over a pub/sub fanout tree.

  • Coalesce to a tick. The count for a post is pushed at most ~1/sec (or when it changes by a meaningful bucket for huge numbers), carrying the current value, not the deltas. 100K increments/sec collapse into one message/sec per post carrying the latest total. Humans cannot tell the difference, and the payload is one integer.
  • Fanout tree, not point delivery. Viewers subscribe to a post’s channel on the Live Fanout Tier (a WebSocket/SSE gateway fleet). The Count Store publishes a post’s coalesced value to a pub/sub topic keyed by post_id; edge gateway nodes each hold many viewer sockets and subscribe once per post they have viewers for, then locally fan the single message out to their sockets. So the origin publishes one message per post per tick; each of, say, 1000 gateway nodes receives it once and fans to its local viewers. Delivery cost is posts * ticks * gateways, not posts * ticks * viewers.
Count changes -> publish coalesced value to topic post:{post_id}  (<=1/sec)
Gateway node (holds 100K sockets): subscribes once per post it serves
  on message: fan to local sockets viewing that post
3M viewers over 1000 gateways = one message to each gateway, local fanout after.
  • Fall back to poll for the long tail. Most posts have few concurrent viewers and change slowly; a plain poll of the cached count every few seconds is cheaper than holding a socket. Live push (WebSocket) is reserved for posts a user is actively focused on (the one on screen, an open post detail). This keeps hundreds of millions of idle feed items on cheap cached polls and spends sockets only where liveness is watched.
  • Client-side smoothing. To avoid a jittery number, the client animates from its current value to the pushed value and, for huge counts, shows rounded buckets (“2.3M”) that change on threshold crossings, hiding both jitter and the last bit of aggregate lag.

The fanout mantra: push a coalesced current value on a slow tick over a subscribe-once fanout tree, poll for the cold tail, and animate on the client - one integer per post per second reaches millions of screens without a storm.

API Design & Data Schema

Writes are idempotent and async-durable; count reads are cached; membership is per-user batched; live updates are coalesced push.

Endpoints

POST /api/v1/posts/{post_id}/like
  headers: Idempotency-Key: <hash(user,post,action,nonce)>
  body: { "user_id": "u_123", "action": "like" | "unlike" }
  -> dedup on idempotency key; append like event; ack fast
  Response 202: { "post_id": "p_88", "your_state": "liked", "count_hint": 2300001 }
     (your_state is authoritative for the actor; count_hint is optimistic)

GET /api/v1/posts/{post_id}/likes
  -> served from edge cache (TTL ~1s) -> replica -> origin display value
  Response 200: { "post_id": "p_88", "count": 2300000, "approx": true,
                  "sample_likers": ["u_9","u_4"], "ts": 1721... }

GET /api/v1/me/liked?posts=p_88,p_91,p_02        (batched "did I like these")
  -> per-user liked-set cache (Bloom + exact fallback)
  Response 200: { "p_88": true, "p_91": false, "p_02": true }

GET /api/v1/posts/{post_id}/likes/stream          (WebSocket/SSE live count)
  -> subscribe to post:{post_id}; server pushes coalesced current count <=1/sec
     event: count
     data: { "count": 2300420, "approx": true }

GET /api/v1/posts/{post_id}/likers?cursor=...      (paginated sample, cold read)
  -> reads from membership store, paginated; not used on the hot feed path
  Response 200: { "likers": [...], "next_cursor": "..." }

The like write is 202 Accepted, not 200 OK: we durably logged it and will settle the count asynchronously, and the client already flipped optimistically. your_state is the one field that is authoritative-for-the-actor immediately (read-your-own-writes); the aggregate count everywhere else is eventually consistent.

Data stores

1. Like Event Log - Kafka, partitioned by post_id. The durable, ordered spine. Every accepted like/unlike is one event. Partitioning by post_id keeps a post’s events ordered on one partition (so toggles resolve deterministically) and lets consumers batch per post. Replayable, so counts and membership can be rebuilt or audited.

Event: { idempotency_key, post_id, user_id, action, ts }
Partition = hash(post_id) % P     # per-post ordering; hot posts get more partitions
Retention: days (source for rebuild/reconciliation), then compacted.

2. Membership Store - sharded, durable KV / wide-column (the source of truth). The authoritative (user_id, post_id) edge set. Needs cheap idempotent upsert/delete, point membership reads, and per-user + per-post scans. A wide-column store (Cassandra-style) fits: shard by key, dual-indexed. NoSQL, not SQL: the access is key-value membership at PB scale with no cross-row transactions - exactly the wide-column sweet spot, and a single SQL table cannot hold or shard 1.2 PB/year of edges without heavy manual partitioning.

Table likes_by_post   (for count recompute + liker sample):
  partition key: post_id
  clustering key: user_id          # scan likers of a post, COUNT for reconcile
  value: { ts }
Table likes_by_user   (for "did I like this" + a user's likes):
  partition key: user_id
  clustering key: post_id          # point read: did user like post
  value: { ts }
Both idempotent: PUT existing = no-op, DELETE missing = no-op.

3. Count Store - Redis, sharded counters per post. The fast, derived aggregate. Per post, N sub-counters plus a pre-summed display value. In-memory because counts are hot and tiny; the durable truth lives in the membership store, so a Redis loss is recoverable by recompute.

Redis, sharded by post_id:
  INCRBY likes:{post_id}:{shard}  delta      # N shards, adaptive N by heat
  SET    display:{post_id}  {summed_count}   # background summer, every ~200ms
  Per-user liked set (read cache):
    Bloom/bitmap of user's recent likes for fast "did I like" negatives
Reads hit display:{post_id} only; shards are summed offline.

4. Live Fanout Tier - stateful WebSocket/SSE gateways. Hold viewer sockets, subscribe once per post to the coalesced count topic, fan out locally. State is just “which sockets are watching which post,” rebuildable on reconnect.

5. Edge / CDN cache - count read path. GET .../likes cached by post_id with a ~1s TTL; hot posts served entirely at the edge.

Why this split of stores: the workload has two different shapes at once - a durable, PB-scale, exactly-correct membership set (wide-column NoSQL on disk) and a tiny, blazing-hot, approximate-until-settled count (sharded Redis in RAM), joined by a durable log. Using one database for both fails: a durable store cannot take 100K writes/sec on one key, and a pure in-memory counter cannot be the durable truth. The log lets each store do only what it is good at, and makes the count always re-derivable from the truth.

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each:

1. Hot-key write storm on a celebrity post (breaks first, hardest). 100K likes/sec onto one post_id. Fix: sharded counters (N sub-counters, adaptive by heat) + in-worker batching. Writes spread across N keys and are coalesced into a few large INCRBY per window, so no single key sees more than a trickle. The logical count is the sum of shards, computed offline by a summer, never on the read path.

2. Read wall on the count. ~100M count-reads/sec across feeds. Fix: cache a tiny pre-summed integer at the edge/CDN (TTL ~1s) and in-memory replicas. Origin sees ~1 read per post per TTL. Huge counts are rounded for display, relaxing freshness on the hottest posts. Shards are never read on this path.

3. “Did I like this?” fan-out. Per-(viewer, post) membership reads at feed scale. Fix: batched, Bloom-fronted per-user liked-set cache. One batched read per feed, with a Bloom filter answering “definitely not liked” for the common case; only possible hits fall to an exact membership point read.

4. Live fanout blowup. Millions of viewers, count changing 100K/sec. Fix: coalesce to <=1/sec current-value pushes over a subscribe-once fanout tree. Origin publishes one message per post per tick; gateways each receive it once and fan locally to their sockets. Cold posts fall back to cheap cached polling.

5. Accuracy drift under retries, toggles, lost deltas. Fix: membership is the idempotent source of truth; counter deltas fire only on real state transitions; idempotency keys drop replays at ingress; per-post ordering resolves toggles; periodic reconciliation recomputes COUNT(*) from edges and corrects any counter drift. The counter is a cache of a number that is always recomputable exactly.

6. Sharding, stated plainly. Like Event Log -> partition by post_id (per-post ordering + per-post batching). Membership Store -> two tables, sharded by post_id (likers/reconcile) and by user_id (did-I-like/user’s likes). Count Store -> keyed by post_id, with N sub-keys per post for the hot ones. Live Fanout -> topic per post_id. The classic mistake would be to shard the counter only by post and stop there - that leaves the single hot post on one shard. The extra :{shard} fan within a post is what actually beats the hot key; post_id alone is not enough for the viral tail.

7. Hot partition on the log for a viral post. A single post’s events all land on one Kafka partition. Fix: for detected mega-posts, spread a post’s events across a small set of partitions (sub-partitioning by hash(user_id) % k within the post) and preserve per-(user, post) ordering by keeping each user’s events on a stable sub-partition. Order that matters (a given user’s like/unlike) is preserved; order that does not (across different users) is relaxed for throughput.

8. Membership store write load. 1M+ durable upserts/sec. Fix: it shards flat by key across the wide-column ring; writes are single-partition idempotent upserts (no cross-row transactions), which scale horizontally by adding nodes. Batched, log-driven consumers smooth spikes.

9. Single points of failure. The Like Write Service and gateways are stateless and horizontally scaled behind LBs. The log is replicated (multi-broker, acks=all). The Count Store is replicated Redis; on loss, counts are recomputed from the durable membership store - the aggregate is never the source of truth, so losing it costs recompute time, not data. The membership store is replicated (quorum) and is the only thing that must not lose data - and it is built for exactly that.

10. Multi-region. Likes come from everywhere; a post is global. Fix: accept likes in the local region (fast ack, local log + local membership + local sharded counter), then asynchronously replicate like events cross-region and sum regional sub-counts into a global count. Because a like is an idempotent edge keyed by (user, post), cross-region replication is conflict-free (a user’s like is the same fact everywhere; last-write-wins on the edge timestamp settles a like/unlike race across regions). The global count is the sum of per-region counts, exchanged async - no global coordination on the hot path, and the number converges to exactly the set of distinct likers worldwide.

Wrap-Up

The trade-offs that define this design:

  • Durable idempotent membership as the source of truth, a sharded counter as its fast cache. Unlike an ephemeral viewer count, a like is a persisted fact, so the truth is the (user, post) edge set - and the count is derived from it, always recomputable, never permanently wrong. This is what buys exact accuracy under a billion taps.
  • Shard the one contended counter, and batch before you shard. The whole game against 100K likes/sec on one post is: coalesce a firehose of taps into a few large INCRBY per window, spread those across N adaptive sub-counters, and sum them offline. One logical counter, physically many, contended by none.
  • Bind counter deltas to real state transitions, and dedup at two layers. Idempotency keys drop replays at ingress; the counter moves only when a like actually creates or removes an edge; per-post ordering resolves toggles; reconciliation recomputes from the truth. Retries, reorders, and lost messages cannot make the number drift for long.
  • Cache a tiny integer to beat the read wall. 100M count-reads/sec survive only because a count is one small, slowly-changing, pre-summed value served from the edge, with huge numbers rounded and the actor’s own like reflected immediately for read-your-own-writes.
  • Coalesce and fan out for live. Never push per-increment; push a current value on a <=1/sec tick over a subscribe-once fanout tree, poll for the cold tail, animate on the client. Millions of screens tick live on one integer per post per second.

One-line summary: model a like as a durable, idempotent (user, post) membership edge written once to a per-post-ordered log, derive the count as an adaptively sharded, batched, eventually-consistent counter that beats a 100K/sec hot key while staying exactly recomputable from the truth, serve that tiny number to 100M reads/sec from an aggressively cached edge, and push it live to millions of viewers as one coalesced value per post per second - accurate and durable where it must be, approximate and fast where it can be.