Open any social app today — Instagram, X (formerly Twitter), LinkedIn, Facebook — and the first thing you see is a scrolling list of posts from people and pages you follow. That scrolling list has a name: the news feed. It looks simple. You open the app, you scroll, new posts appear. But underneath that simple scroll is one of the hardest problems in distributed systems, and it is a favorite topic in system design interviews for a very good reason: it touches almost every important idea in backend engineering at once — storage, caching, queues, ranking, consistency, and scale.

Let us start with a bit of history, because it helps explain why the problem is shaped the way it is today. Before news feeds existed, social platforms worked more like a personal profile page. On early Friendster and MySpace, you visited someone’s page to see what they posted. There was no single place that aggregated everything happening across all your friends. You had to go looking for updates, one profile at a time.

That changed in September 2006, when Facebook launched a feature called News Feed. Instead of making you visit each friend’s page, Facebook pulled updates from everyone you were connected to into one continuously updating stream on your home page. It was controversial at first — many users felt it was an invasion of privacy, since actions that used to be quiet (like changing your relationship status) were now broadcast automatically to everyone who knew you. But within a few years, the feed model became the default way people consumed social content, and every major platform since — Twitter in 2006, Instagram in 2010, LinkedIn’s feed, TikTok’s “For You” page — has built some version of the same core idea: take content produced by a network of people, and assemble it into a personalized, ranked stream for every single viewer.

The interesting part, and the part interviewers care about most, is that this feature quietly forces you to solve a really hard distributed computing problem: if millions of people are producing content at the same time, and hundreds of millions of people are reading personalized combinations of that content at the same time, how do you make both directions fast?

Let us state the problem the way an interviewer would state it, and then break down why it is genuinely difficult.

On the surface, this sounds like a database query: “give me the latest posts from everyone I follow, sorted by time.” And for a small app with a few thousand users, that is genuinely all you would need — a simple SQL join between a follows table and a posts table would do the job in milliseconds.

The trouble starts when you scale two numbers at once: the number of people producing content, and the number of followers a single account can have. At 500 million daily active users, if even a modest fraction of them post regularly, you are looking at tens of thousands of new posts every second at peak times. Read that against a feed load: every one of those 500 million users might refresh their feed several times a day, which means the read traffic is enormously larger than the write traffic — often by a factor of 100 to 1 or more in social systems. A system that is naively built to “query on read” will get crushed by that read volume the moment it has any real number of users, because every single feed load would require scanning the posts of everyone that user follows, sorting them, and returning the top results — repeated hundreds of millions of times a day.

Now add the celebrity problem, which is the detail that makes this specific prompt interesting. Imagine a naive fix: whenever someone posts, immediately copy that post into the personal feed of every follower, so reads become instant (you just read your own pre-built feed). This is called fan-out-on-write, and we will cover it in depth soon. It works beautifully for a normal user with 300 followers — the system does 300 small writes, done in a fraction of a second. But if a celebrity with 100 million followers posts a single photo, the system suddenly has to perform 100 million individual writes, one for each follower’s feed, just because one person clicked “post.” Do that a few times a day, across a handful of extremely large accounts, and you can bring an entire fan-out pipeline to its knees — a phenomenon engineers at Twitter genuinely nicknamed the celebrity problem or the “Justin Bieber problem,” after episodes where a hugely followed account posting would spike system load dramatically.

So the real engineering problem being tested here is not “how do you build a feed” — it is “how do you build a feed that behaves gracefully at both extremes of the follower distribution, from an account with three followers to an account with a hundred million, without one extreme punishing the design decisions made for the other.”

Before drawing a single box in a diagram, a good system designer always writes down what the system actually needs to do, and roughly how big it needs to be. Skipping this step is one of the most common mistakes candidates make in interviews — they jump straight to drawing microservices without ever stating what “fast” or “scalable” actually means in numbers.

3.1 Functional requirements

3.2 Non-functional requirements

3.3 Back-of-the-envelope capacity estimation

Interviewers love this part because it separates candidates who can reason about scale from candidates who only know buzzwords. Let us work through realistic numbers for our 500 million DAU platform.

Now the read side. Assume each of the 500 million daily active users opens their feed an average of 10 times a day (a conservative number for a habit-forming social app).

That last row is the single most important number in this entire design. A 20:1 read-to-write ratio (in real systems like Twitter, the ratio has historically been closer to 100:1 or higher) tells you immediately: optimize aggressively for read performance, even if it costs you extra work at write time. This one number is the entire justification for fan-out-on-write, caching pre-built feeds, and every other read-optimization technique you will see in this tutorial. It is also why the celebrity problem is such a headache — it is the one case where a write briefly turns into millions of reads’ worth of work, right at the moment it happens.

With requirements and rough numbers in hand, we can now sketch the major pieces of the system. Think of this as the skeleton — later sections will go deep into how each piece actually works internally.

Let us walk through each box and what it is responsible for. In an interview, being able to explain each component’s single job clearly is worth more than drawing a lot of boxes.

4.1 API Gateway

The single entry point for every client request. It handles authentication (is this a real, logged-in user?), rate limiting (is this user or IP sending too many requests?), and routing (which internal service should handle this particular request?). Production example: Netflix’s Zuul gateway and Amazon’s API Gateway both perform this exact role — a single, hardened front door so that internal services never have to deal directly with untrusted public traffic.

4.2 Post Service

Owns the lifecycle of a post: validating content, storing text/metadata in the post database, kicking off media uploads to blob storage, and — critically — publishing an event (“a new post was created”) onto a message queue rather than doing any fan-out work itself. Keeping this service simple and fast is important, because it is on the direct path of the user hitting “post,” and we do not want that click to hang while 100 million fan-out writes happen synchronously.

4.3 Follow Graph Service

Owns the “who follows whom” relationship — arguably the most important piece of state in the entire system, since almost everything else depends on knowing a user’s follower list and following list quickly. This is often backed by a graph database or, for very high read speed, a Redis-based sorted-set structure per user.

4.4 Fanout Service

Listens to the queue for “new post” events and decides how to distribute that post to followers’ feeds. This is where the celebrity problem is actually solved, using the hybrid model we will cover in detail shortly.

4.5 Ranking Service

Takes a raw, chronological set of candidate posts and re-orders them based on a relevance score — factoring in recency, the viewer’s past engagement with the poster, predicted likelihood of like/comment/share, and content type. We look at this in depth in its own section.

4.6 Feed Service

Handles the read path: when a user opens the app, this service assembles their feed, pulling from a precomputed cache where possible and merging in a live query for accounts that were not fanned-out (this is the “hybrid” part again — more shortly).

4.7 Notification Service

A supporting service that tells users “so-and-so just posted” or “your comment got a reply” — decoupled from the core feed pipeline via the same queue, so a slow notification delivery never blocks a post from reaching feeds.

4.8 Message Queue (Kafka)

The backbone that decouples “a post was created” from “that post reached everyone who needs to see it.” Production example: LinkedIn originally built Kafka precisely to handle this kind of high-throughput, decoupled event pipeline internally, before open-sourcing it — it is now the industry standard for exactly this pattern at companies including Netflix, Uber, and Airbnb.

Now let us go deep on the two fundamental strategies for getting a post from its author to its readers, starting with the one that optimizes reads at the cost of writes.

Imagine you keep a small mailbox for every single user — call it their feed cache — that already contains, pre-sorted, the IDs of posts they should see next. When someone posts, instead of waiting for a reader to ask “what is new,” the system immediately walks through every one of that author’s followers and drops the new post’s ID into each of their mailboxes. By the time any follower opens the app, their feed is already sitting there, ready to be read with a single fast cache lookup.

This is fan-out-on-write, also called the push model, because the work of distributing the post is “pushed out” at write time, immediately after the post is created.

The feed cache itself is usually modeled as a simple, capped-length sorted list per user, most naturally implemented with a Redis sorted set, where the score is the post’s timestamp (or a ranking score) and the value is the post ID.

// Fanout worker: pushes a new post into every follower's feed cache
// Uses Redis sorted sets — score = timestamp, member = postId

public class FanoutWorker {

    private static final int MAX_FEED_LENGTH = 1000; // cap per user
    private final RedisClient redis;
    private final FollowGraphClient graphClient;

    public void handlePostCreated(PostCreatedEvent event) {
        String authorId = event.getAuthorId();
        String postId = event.getPostId();
        long timestamp = event.getTimestamp();

        // Step 1: fetch follower list (paginated for very large accounts)
        Iterator<String> followerIds = graphClient.getFollowersStream(authorId);

        // Step 2: push post into each follower's feed, in batches
        List<String> batch = new ArrayList<>();
        while (followerIds.hasNext()) {
            batch.add(followerIds.next());
            if (batch.size() == 500) {
                pushBatch(batch, postId, timestamp);
                batch.clear();
            }
        }
        if (!batch.isEmpty()) {
            pushBatch(batch, postId, timestamp);
        }
    }

    private void pushBatch(List<String> followerIds, String postId, long timestamp) {
        try (Pipeline pipeline = redis.openPipeline()) {
            for (String followerId : followerIds) {
                String feedKey = "feed:" + followerId;
                pipeline.zadd(feedKey, timestamp, postId);
                // trim the feed so it never grows unbounded
                pipeline.zremrangeByRank(feedKey, 0, -MAX_FEED_LENGTH - 1);
            }
            pipeline.sync(); // executes all commands in one network round trip
        }
    }
}

5.1 Why this works so well for typical users

For an average account with a few hundred followers, fan-out-on-write is close to ideal. The write cost of a single post — a few hundred small, fast Redis operations — is trivial for the system, and it buys every follower an instant, pre-built feed read for the rest of that post’s lifetime. Since our capacity estimate showed a 20:1 (often much higher) read-to-write ratio, paying a small extra cost at write time to make every subsequent read nearly free is exactly the right trade for the vast majority of accounts on the platform.

5.2 Why it breaks down at the extreme

The problem is entirely about the tail of the distribution, not the average. A celebrity with 100 million followers turns one post into 100 million Redis writes. Even at a very efficient 50,000 writes per second per fan-out worker, that is roughly 2,000 seconds — over 30 minutes — just to fan out one single post, and that assumes nothing else is competing for the same workers at the same time. In practice, platforms run many parallel fan-out workers, but the point stands: the cost of fan-out-on-write scales directly with follower count, and follower count for a small number of accounts is genuinely enormous.

The opposite strategy does no work at all when a post is created. The Post Service simply saves the post, and that is it — nothing gets copied anywhere. All the real work happens later, at the moment a follower actually opens their feed.

When that happens, the Feed Service looks up who the user follows, fetches the most recent posts from each of those accounts (typically from a cache, falling back to the database), merges them together, sorts by recency or relevance, and returns the result. This is fan-out-on-read, or the pull model, because the work of assembling the feed is “pulled” together on demand, at read time.

Notice the appeal immediately: posting is essentially free, no matter how many followers the author has. A celebrity with 100 million followers costs exactly the same at write time as a brand-new account with zero followers, because nothing gets copied to anyone. This completely sidesteps the celebrity problem — at write time.

But remember the 20:1 (or higher) read-to-write ratio we calculated earlier. The pull model takes a cost that used to happen once per post, and turns it into a cost that happens every single time any follower opens their feed. If a user follows 500 accounts, then every feed load means fetching recent posts from up to 500 different sources and merging them — repeated for every one of the hundreds of millions of daily feed opens. That is an enormous amount of repeated, redundant work, and it is precisely the wrong trade given how lopsided our read-write ratio is.

Since this is the specific twist in our prompt, it deserves its own dedicated section rather than being folded into the fan-out discussion. Let us define it precisely and look at exactly why it is dangerous.

The celebrity problem (sometimes called the “hotspot” or “hot key” problem in distributed systems more broadly) happens whenever the amount of work a system must do for one entity is wildly disproportionate to the amount of work it does for a typical entity. In our case, the “entity” is a user account, and the “work” is fan-out writes. A normal account might trigger a few hundred writes per post. A celebrity account triggers tens of millions. That is not just “a bit more load” — it is several orders of magnitude more load, concentrated on one operation, at one moment in time.

This creates three concrete failure risks if left unhandled:

This is the section that answers our exact interview prompt. The industry-proven answer to “how do you handle celebrities with 100 million followers” is: do not use one strategy for everyone — pick the strategy per account, based on follower count, and merge the results at read time.

Here is the rule in plain language: pick a threshold — a common real-world number is somewhere between 10,000 and 1,000,000 followers, tuned based on actual system load testing. Any account below that threshold gets fan-out-on-write, because the write cost is cheap and it makes every future read instant. Any account above that threshold — the celebrities, the huge media pages — is treated with fan-out-on-read: their posts are never pushed anywhere, and instead get pulled in and merged live whenever a follower loads their feed.

This design gives every account the cheapest possible path. A normal user’s post costs a few hundred writes and gives every follower an instant read. A celebrity’s post costs almost nothing at write time — it is simply saved — and the small extra cost of fetching celebrity content is paid only by the users who actually follow celebrities, and only at the moment they actually check their feed, not by the celebrity’s followers who never open the app that day.

// Deciding fanout strategy at write time
public class FanoutRouter {

    private static final long CELEBRITY_THRESHOLD = 1_000_000L;
    private final FollowGraphClient graphClient;
    private final FanoutWorker fanoutWorker;
    private final PostStore postStore;

    public void handleNewPost(Post post) {
        long followerCount = graphClient.getFollowerCount(post.getAuthorId());

        if (followerCount > CELEBRITY_THRESHOLD) {
            // Celebrity path: no fanout, just mark and store.
            // We tag the author as a "celebrity" so the read path
            // knows to fetch their posts live instead of expecting
            // them in the follower's precomputed feed cache.
            postStore.save(post);
            postStore.markCelebrityPost(post.getAuthorId(), post.getPostId());
        } else {
            // Normal path: eagerly push to every follower's feed.
            postStore.save(post);
            fanoutWorker.handlePostCreated(
                new PostCreatedEvent(post.getAuthorId(), post.getPostId(), post.getTimestamp())
            );
        }
    }
}

// Assembling a feed at read time — the merge step
public class FeedAssembler {

    private final RedisClient feedCache;
    private final FollowGraphClient graphClient;
    private final PostStore postStore;

    public List<Post> getFeed(String userId, int limit) {
        // Step 1: get the precomputed feed (built by fanout-on-write)
        List<String> cachedPostIds = feedCache.zrevrange("feed:" + userId, 0, limit);

        // Step 2: find celebrities this user follows
        List<String> celebrityIds = graphClient.getFollowedCelebrities(userId);

        // Step 3: pull recent posts from those celebrities live
        List<Post> celebrityPosts = new ArrayList<>();
        for (String celebId : celebrityIds) {
            celebrityPosts.addAll(postStore.getRecentPosts(celebId, 20));
        }

        // Step 4: merge cached feed with live celebrity posts, then rank
        List<Post> cachedPosts = postStore.getPostsByIds(cachedPostIds);
        List<Post> merged = mergeAndDedupe(cachedPosts, celebrityPosts);
        return RankingEngine.rank(userId, merged, limit);
    }

    private List<Post> mergeAndDedupe(List<Post> a, List<Post> b) {
        Map<String, Post> byId = new LinkedHashMap<>();
        for (Post p : a) byId.put(p.getPostId(), p);
        for (Post p : b) byId.put(p.getPostId(), p);
        return new ArrayList<>(byId.values());
    }
}

8.1 Handling the boundary: what if a normal account suddenly goes viral?

A subtle real-world wrinkle: follower counts change, sometimes very quickly. An account with 50,000 followers today could gain a million overnight after going viral. Production systems handle this with a periodic or event-triggered re-evaluation: whenever a follower count crosses the threshold, a background job flips that account’s fanout mode, and — importantly — this switch does not require any migration of existing feed data, because the switch only changes how future posts are distributed. Old, already-fanned-out posts remain in followers’ cached feeds; the account simply starts being treated as a celebrity going forward.

8.2 Handling the boundary: what if a celebrity has 100 million followers but a user follows only 3 celebrities?

This is exactly why the merge step in the read path is cheap. The Feed Service does not fetch anything from all 100 million followers — it only does a small amount of extra work, per reader, proportional to how many celebrities that specific reader follows (usually a handful). The expensive part of the celebrity’s fan-out — the “100 million” — never actually happens anywhere in this design. It is replaced by many small, cheap lookups, each done only when a real reader actually needs it.

It helps to trace a single post from the moment it is created to the moment it disappears from active circulation, because this end-to-end view is often exactly what interviewers want you to narrate.

Let us define the core entities and how they relate to each other, then look at concrete schema choices.

A few schema design decisions are worth calling out explicitly, because they are exactly the kind of thing interviewers probe on.

10.1 Why the Post table is partitioned by author_id, not post_id

It is tempting to shard posts by a random or hash-based post ID for even distribution, but the most common access pattern is “give me this author’s recent posts” (needed constantly by the pull-model celebrity path, and by profile pages). Partitioning primarily by author_id keeps one author’s posts physically close together, making that very common query cheap — a single shard lookup instead of a scatter-gather across many shards.

10.2 Why post IDs should be time-sortable, not purely random

Using a scheme like Snowflake IDs (popularized by Twitter, later adopted in similar forms by Instagram and Discord) bakes a timestamp into the most significant bits of the ID itself. This means sorting posts by ID is the same as sorting them by creation time, without needing a separate index or an extra sort step — a small trick that saves enormous amounts of computation at this scale.

// Simplified Snowflake-style unique ID generator
// 41 bits timestamp | 10 bits machine/shard id | 12 bits sequence number
public class SnowflakeIdGenerator {

    private static final long EPOCH = 1700000000000L; // custom epoch
    private final long machineId;
    private long lastTimestamp = -1L;
    private long sequence = 0L;

    public SnowflakeIdGenerator(long machineId) {
        this.machineId = machineId;
    }

    public synchronized long nextId() {
        long timestamp = System.currentTimeMillis() - EPOCH;

        if (timestamp == lastTimestamp) {
            sequence = (sequence + 1) & 0xFFF; // 12 bits, wraps at 4096
            if (sequence == 0) {
                // sequence exhausted for this millisecond, wait for next one
                while (timestamp == lastTimestamp) {
                    timestamp = System.currentTimeMillis() - EPOCH;
                }
            }
        } else {
            sequence = 0L;
        }
        lastTimestamp = timestamp;

        return (timestamp << 22) | (machineId << 12) | sequence;
    }
}

10.3 Why the Follow relationship needs two access patterns, not one

Almost every operation in this system needs to answer one of two very different questions: “who does this user follow?” (needed to build a feed) and “who follows this user?” (needed to fan out a post). Storing the relationship once and querying it two directions is inefficient at scale, so production systems typically maintain two denormalized copies of the same relationship — a “following list” keyed by follower, and a “followers list” keyed by followee — accepting the small extra write cost and storage duplication in exchange for both reads being equally fast.

A common interview trap is treating “the database” as one monolithic decision. A real system like this uses several different storage technologies, each chosen for a specific access pattern.

11.1 SQL vs NoSQL for post storage — the real reasoning

A relational database gives you strong consistency and rich query flexibility (joins, transactions), which is valuable for data like account settings or payments where correctness matters more than raw throughput. But a single relational database, even a powerful one, struggles to sustain tens of thousands of writes per second while also serving low-latency reads, because relational engines pay overhead for maintaining strict consistency guarantees and complex indexes.

A wide-column store like Cassandra takes a different trade: it is built from the ground up for horizontal scale and high write throughput, using a partition-and-replicate model where you deliberately give up some consistency guarantees (you typically read “eventually consistent” data unless you explicitly ask for stronger consistency at extra cost) in exchange for near-linear scalability — write throughput that grows almost proportionally as you add machines. For post metadata, where a few seconds of eventual consistency is completely acceptable (as we established in the requirements), this trade is clearly worth it.

Given our 20:1-or-higher read-to-write ratio, caching is not an optimization we bolt on later — it is central to the whole design. Let us cover the layers of caching this system needs, and a genuinely tricky problem specific to celebrity accounts: the hot key problem.

12.1 Layer 1 — Precomputed feed cache

Already covered: a Redis sorted set per user, holding the last ~1,000 post IDs for their feed, populated by fan-out-on-write. This is the single most important cache in the system, since it turns the most frequent operation (loading a feed) into a fast in-memory lookup.

12.2 Layer 2 — Post content cache

Feed caches store only post IDs, not full content (to keep them small and cheap to update). A separate cache — typically also Redis, or a dedicated in-memory content store — maps post IDs to their full hydrated content: text, media URLs, current like/comment counts. This is populated with a standard cache-aside pattern: check the cache first, and on a miss, fetch from the Post DB and populate the cache for next time.

12.3 Layer 3 — Follow graph cache

Follower and following lists are read extremely frequently (every fan-out, every feed merge) and change relatively rarely compared to how often they are read, making them an excellent caching candidate — often held almost entirely in Redis rather than falling back to a slower backing store except for cold, rarely-accessed accounts.

12.4 The hot key problem — when one cache key gets too much traffic

Here is a genuinely subtle issue that trips up a lot of otherwise strong designs. A celebrity’s post content, once hydrated, might be requested millions of times in a short window — every follower who happens to check their feed around the same time asks the content cache for the same post ID. In a typical distributed cache, that key lives on one specific cache node (determined by its hash). If millions of requests hit that same node for that same key at once, that single node can become overwhelmed, even though the overall cache cluster has plenty of spare capacity elsewhere. This is exactly what people mean by a “hot key” or “hotspot” — the problem is not total load, it is load concentrated unevenly on one small part of the system.

// A simple local "L1" cache in front of the distributed cache,
// specifically to absorb hot-key traffic for viral / celebrity posts.
public class HotKeyAwareCache {

    private final Cache<String, Post> localCache; // small, in-process, short TTL
    private final RedisClient distributedCache;    // the normal shared L2 cache

    public HotKeyAwareCache() {
        this.localCache = Caffeine.newBuilder()
            .maximumSize(5_000)
            .expireAfterWrite(Duration.ofSeconds(5)) // short TTL is enough to smooth spikes
            .build();
    }

    public Post getPost(String postId) {
        // Check the tiny local cache first — this alone can absorb
        // a huge share of traffic for a viral post, because thousands
        // of requests on the same application server all hit the same
        // in-process cache instead of all hammering Redis.
        Post cached = localCache.getIfPresent(postId);
        if (cached != null) {
            return cached;
        }

        Post post = distributedCache.get("post:" + postId);
        if (post == null) {
            post = postStore.getById(postId); // fall back to DB on full miss
            distributedCache.set("post:" + postId, post, Duration.ofMinutes(10));
        }
        localCache.put(postId, post);
        return post;
    }
}

The trick above — a small, short-lived local cache sitting in front of the shared distributed cache — is a genuinely common production pattern, because it converts a single hot key hitting one Redis node millions of times into that same key being served locally, on every individual application server, most of the time. It is cheap to implement and dramatically reduces the blast radius of any one piece of content going viral.

12.5 Cache invalidation for changing data

Like counts and comment counts change constantly on a popular post, which conflicts with wanting to cache that same post’s content for a while. The common resolution is to split the data: cache the mostly-static content (text, media URLs) with a longer TTL, and serve frequently-changing counters from a separate, very fast counter store (often Redis INCR-based counters) that is read alongside the cached content and merged at render time, rather than invalidating the whole cached post every time someone likes it.

With 500 million users and 250 million daily posts, no single machine — no matter how powerful — could hold all the data or handle all the traffic. The system has to be split across many machines, a process called sharding or partitioning, and traffic has to be distributed evenly across those machines, which is load balancing.

13.1 Choosing a partition key

The most natural partition key for both the post store and the feed cache is user ID (author_id for posts, follower’s user_id for feed caches) — because almost every query in this system is scoped to a specific user, either “posts by this author” or “feed for this reader.” Partitioning by user ID keeps those queries confined to a single shard, avoiding expensive scatter-gather queries across the whole cluster.

13.2 Consistent hashing — why it beats simple modulo sharding

A naive approach — shard = hash(userId) % numberOfShards — works until you need to add or remove a shard, at which point nearly every key remaps to a different shard, forcing a massive, disruptive data migration. Consistent hashing solves this by arranging shards on a conceptual ring, so that adding or removing one shard only reshuffles the small slice of keys immediately next to it on the ring, leaving the vast majority of keys undisturbed.

// Simplified consistent hashing ring using virtual nodes
// Virtual nodes smooth out uneven load distribution across physical shards
public class ConsistentHashRing {

    private final TreeMap<Long, String> ring = new TreeMap<>();
    private static final int VIRTUAL_NODES_PER_SHARD = 100;

    public void addShard(String shardId) {
        for (int i = 0; i < VIRTUAL_NODES_PER_SHARD; i++) {
            long hash = hash(shardId + "#" + i);
            ring.put(hash, shardId);
        }
    }

    public void removeShard(String shardId) {
        for (int i = 0; i < VIRTUAL_NODES_PER_SHARD; i++) {
            ring.remove(hash(shardId + "#" + i));
        }
    }

    public String getShardFor(String key) {
        long hash = hash(key);
        Map.Entry<Long, String> entry = ring.ceilingEntry(hash);
        if (entry == null) {
            entry = ring.firstEntry(); // wrap around the ring
        }
        return entry.getValue();
    }

    private long hash(String input) {
        return Hashing.murmur3_128().hashString(input, StandardCharsets.UTF_8).asLong();
    }
}

Virtual nodes, shown in the code above, solve a secondary problem: with only a handful of real shards placed on the ring, load can still end up quite uneven if two shards happen to land close together. Giving each physical shard 100 (or more) virtual positions on the ring spreads its share of the keyspace out evenly, so real-world load balances much more smoothly across physical machines.

13.3 Load balancing at the service layer

Beyond data sharding, stateless services (API Gateway, Feed Service, Fanout Service) sit behind standard load balancers that distribute incoming requests across many identical service instances, usually with health checks so traffic automatically avoids instances that are unhealthy or overloaded. Production example: Netflix’s internal services sit behind their own load balancing layer, Eureka-based service discovery historically, now often paired with more modern service mesh technology, so that any instance can be added or removed without manual reconfiguration.

13.4 Rebalancing without downtime

Adding a new shard to a running cluster is a routine, ongoing operational task at this scale, not a rare event — user growth and data growth both push clusters toward needing more capacity over time. Consistent hashing already limits how much data has to move when a shard is added, but that data still has to move somewhere, and it has to do so without interrupting live traffic. Production systems typically handle this with a background migration process: the new shard is added to the ring, and a controlled, throttled process copies the specific slice of keys that now belong to it from their old shard, while both the old and new locations remain readable during the transition. Only once the copy is verified complete does the system fully cut over reads and writes for that slice of keys to the new shard, and the old copy is cleaned up afterward. Throttling this migration matters just as much as doing it correctly — an unthrottled bulk copy competing with live production traffic for the same disk and network resources can itself become a self-inflicted outage, which is a subtlety many first-draft designs miss entirely.

13.5 Read replicas versus shards — a distinction worth keeping straight

It is easy to conflate two related but different scaling techniques. Sharding splits the entire dataset into disjoint pieces, each living on a different set of machines, so that no single machine needs to hold all the data. Read replication, by contrast, copies the same piece of data onto multiple machines specifically so that read traffic against that data can be spread across more than one machine. A well-scaled system typically uses both together: a partition key decides which shard owns a given piece of data, and then that shard itself has several read replicas behind it, so that both the total dataset size and the read throughput against any individual slice of that dataset can each grow independently, along their own dimension, as the platform’s usage patterns demand.

Everything so far has assembled a candidate set of posts — the raw material for a feed. Ranking decides the order those posts actually appear in, and it is arguably the single biggest lever platforms have over how engaging a feed feels.

14.1 Why not just sort by time?

A purely chronological feed is simple, predictable, and easy to reason about — and for years it is exactly what most platforms used. But it has an obvious flaw: if you follow 500 accounts and only check your feed twice a day, a strictly time-sorted feed means the handful of accounts that happen to post most frequently completely dominate what you see, burying content from accounts you care about more but who post less often. Purely chronological ordering treats “most recent” as identical to “most relevant,” and those two things are often quite different.

14.2 Moving to relevance-based ranking

Modern feeds score each candidate post with a model that combines several signals, then sort by that score instead of raw timestamp. A simplified version of the kinds of features involved:

// Simplified relevance scoring — real systems replace the
// hand-tuned weights below with a trained ML model, but the
// feature set and overall shape stays conceptually similar.
public class RankingEngine {

    public static List<Post> rank(String viewerId, List<Post> candidates, int limit) {
        PriorityQueue<ScoredPost> heap = new PriorityQueue<>(
            Comparator.comparingDouble(ScoredPost::score).reversed()
        );

        for (Post post : candidates) {
            double score = computeScore(viewerId, post);
            heap.add(new ScoredPost(post, score));
        }

        List<Post> result = new ArrayList<>();
        while (!heap.isEmpty() && result.size() < limit) {
            result.add(heap.poll().post());
        }
        return result;
    }

    private static double computeScore(String viewerId, Post post) {
        double recencyScore = decay(post.getTimestamp());
        double affinityScore = AffinityStore.get(viewerId, post.getAuthorId());
        double engagementScore = EngagementPredictor.predict(viewerId, post);
        double velocityScore = Math.log(1 + post.getRecentLikeVelocity());

        // Weighted combination — weights are typically learned,
        // not hand-picked, in a real production ranking model.
        return (0.25 * recencyScore)
             + (0.30 * affinityScore)
             + (0.35 * engagementScore)
             + (0.10 * velocityScore);
    }

    private static double decay(long timestamp) {
        long ageSeconds = (System.currentTimeMillis() / 1000) - timestamp;
        return Math.exp(-ageSeconds / 43200.0); // half-life-style decay, ~12 hours
    }
}

14.3 The exploration vs exploitation trade-off

A purely engagement-maximizing ranker risks over-fitting to a viewer’s past behavior, showing them an increasingly narrow slice of content and starving new or different creators of visibility (a real concern often described as a filter bubble effect). Production ranking systems typically inject some deliberate exploration — occasionally surfacing content slightly outside a viewer’s established pattern — both to keep the feed feeling fresh and to keep gathering the training signal needed to keep the underlying prediction models accurate over time.

Let us define the actual contracts a client application would use, and talk about why this system is naturally suited to a microservices architecture rather than one large application.

15.1 Core API endpoints

GET /v1/feed?cursor=eyJ0cyI6MTcwMDAwMDAwMH0&limit=20 HTTP/1.1
Authorization: Bearer <token>

Response 200 OK:
{
  "posts": [
    {
      "postId": "1823456789012345",
      "authorId": "u_9182",
      "text": "Excited to share our new feature launch!",
      "mediaUrls": ["https://cdn.example.com/media/abcd.jpg"],
      "likeCount": 4210,
      "commentCount": 312,
      "createdAt": "2026-07-24T09:14:00Z"
    }
  ],
  "nextCursor": "eyJ0cyI6MTY5OTk5OTk4MH0"
}

Note the use of a cursor rather than a page number for pagination. Feeds are constantly changing — new posts arrive continuously — so offset-based pagination (“page 2, page 3”) tends to produce duplicate or skipped posts as the underlying data shifts between requests. A cursor, typically encoding the timestamp or rank of the last item seen, gives a stable, consistent scroll experience regardless of what is being written concurrently.

15.2 Why microservices fit this problem well

Post creation, fan-out, ranking, and feed assembly have genuinely different scaling characteristics and failure profiles. The Fanout Service needs to burst-scale during celebrity posting events. The Ranking Service needs specialized compute and frequent independent deployment. The Feed Service needs to stay responsive above almost everything else, since it is the most latency-sensitive read path in the whole product. Splitting these into independently deployable, independently scalable services means a spike or an incident in one does not automatically take down the others — a monolith handling all of this in one process would couple failure domains together unnecessarily.

16.1 Patterns worth naming explicitly

16.2 Anti-patterns to avoid — and explicitly call out in an interview

No design in distributed systems is free of trade-offs — the mark of a mature engineer is being able to state them plainly instead of presenting a design as if it has no downsides.

The overall system-level trade-off worth stating explicitly in an interview: this design deliberately favors availability and low latency over strict consistency almost everywhere. A follower briefly not seeing a new post, a like count that is a few seconds stale, a feed that is slightly less personalized during a partial outage — all of these are accepted costs, in exchange for a system that stays fast and up even under heavy, uneven load. That is exactly the right trade for a social feed, where a few seconds of staleness is invisible to users but even a brief full outage is highly visible and damaging.

Let us connect the earlier capacity numbers to concrete scaling techniques.

18.1 Horizontal scaling as the primary lever

Every stateful component in this design — post storage, follow graph, feed cache — is partitioned so that adding more machines directly increases capacity, rather than relying on making any single machine bigger (vertical scaling), which always hits a hard ceiling. This is what makes it possible to grow from 500 million to, hypothetically, a billion daily active users largely by adding shards and cache nodes rather than redesigning the system.

18.2 Read replicas for further read scaling

Beyond sharding, each shard typically has multiple read replicas — copies of the data that can serve read traffic while a single primary handles writes. Since our read-to-write ratio is so heavily skewed toward reads, having several read replicas per shard lets read capacity scale independently of write capacity, which is exactly where the imbalance in our capacity estimation needs the most help.

18.3 Batching and pipelining

The fan-out code shown earlier deliberately batches Redis writes and uses pipelining (sending many commands in one network round trip rather than one at a time). At the scale of millions of writes for a single celebrity post, this single technique can be the difference between a fan-out finishing in minutes versus hours, because network round-trip latency, not actual processing time, dominates cost when operations are small and numerous.

18.4 Backpressure and queue-based load shedding

During an extreme spike — say, several large accounts posting within the same minute — the Fanout Service’s consumers might not keep up in real time with the incoming event rate. Because the Post Service publishes events to a durable queue rather than calling Fanout Service directly, this is safe: events simply queue up and get processed as capacity allows, rather than being lost or causing cascading failure. This buffering behavior — sometimes called backpressure absorption — is one of the most valuable properties a message queue provides in bursty systems like this one.

18.5 Precomputing versus computing on demand — a recurring theme

Across nearly every part of this design — feed caches, denormalized follow lists, cached post content — the same underlying idea repeats: shift work from the frequent operation (read) to the infrequent one (write), whenever the read-write ratio justifies it. Recognizing this as a repeating pattern, rather than a series of unrelated tricks, is exactly the kind of systems-level thinking interviewers are trying to draw out.

19.1 A quick, precise refresher on the CAP theorem

The CAP theorem states that a distributed system can only fully guarantee two of these three properties at once, during a network partition (when some machines cannot communicate with others):

In practice, partition tolerance is not really optional for any real distributed system spread across multiple machines or data centers — networks will experience partitions eventually, full stop. So the real, practical choice most systems actually make is between CP (favor consistency, sacrifice availability during a partition) and AP (favor availability, sacrifice strict consistency during a partition).

As established throughout this tutorial, a news feed system should lean heavily AP. Users tolerate a few seconds of staleness far better than they tolerate an error page. Cassandra (used for our post store) and Redis clusters (used for our caches) are both commonly configured to favor availability, with tunable consistency that we deliberately dial toward “fast and eventually correct” rather than “always perfectly correct, sometimes unavailable.”

19.2 Replication for durability and availability

Every piece of data in this system is replicated across multiple nodes — commonly three or more copies, often spread across different physical availability zones or data centers. If one node or even an entire data center fails, replicas elsewhere can continue serving traffic. Cassandra achieves this with a configurable replication factor and a “quorum” style read/write model: writes and reads can require acknowledgment from a subset of replicas (say 2 out of 3), balancing consistency and availability per operation.

19.3 Consensus — where it is actually needed, and where it is not

It is worth being precise here, because a common interview mistake is reaching for heavyweight consensus algorithms (like Raft or Paxos) everywhere, when most of this system does not need them. Consensus protocols exist to make multiple nodes agree on a single, strictly ordered sequence of operations — essential for things like leader election in a database cluster, or maintaining strong consistency for critical configuration data. But the bulk of this design — feed caches, post content, engagement counters — deliberately uses weaker, eventually-consistent replication specifically because we do not need strict ordering agreement across the whole system for those pieces of data. Consensus shows up narrowly: for example, within Cassandra’s own internal cluster coordination, or in a configuration service (like ZooKeeper or etcd) that the platform might use to track which shard owns which key range.

19.4 Failure recovery patterns

20.1 Authentication and authorization

Every request passes through the API Gateway, which validates a signed token (commonly a JWT or an equivalent opaque token backed by a session store) before any request reaches internal services. Internal service-to-service calls typically use mutual TLS or short-lived internal tokens, so that even inside the private network, one compromised service cannot freely impersonate another.

20.2 Privacy controls in the feed itself

A feed system has to respect account privacy settings at every layer that touches content — a private account’s posts should never be fanned out to, cached for, or returned to anyone who is not an approved follower. This means the Fanout Service and Feed Service both need to check privacy state, not just the Post Service at creation time, since privacy settings can change after a post already exists (a user can go from public to private, and previously-visible content should stop being served to non-followers going forward).

20.3 Rate limiting and abuse prevention

The API Gateway enforces rate limits per user and per IP to prevent both accidental abuse (a buggy client hammering the feed endpoint) and deliberate abuse (scraping, spam posting, automated fake engagement). A common implementation is the token bucket algorithm: each user has a bucket that refills at a steady rate, and each request consumes a token — once the bucket is empty, further requests are rejected or delayed until it refills.

// Simplified token bucket rate limiter
public class TokenBucketRateLimiter {

    private final int capacity;
    private final double refillRatePerSecond;
    private double tokens;
    private long lastRefillTimestamp;

    public TokenBucketRateLimiter(int capacity, double refillRatePerSecond) {
        this.capacity = capacity;
        this.refillRatePerSecond = refillRatePerSecond;
        this.tokens = capacity;
        this.lastRefillTimestamp = System.nanoTime();
    }

    public synchronized boolean allowRequest() {
        refill();
        if (tokens >= 1) {
            tokens -= 1;
            return true;
        }
        return false;
    }

    private void refill() {
        long now = System.nanoTime();
        double secondsElapsed = (now - lastRefillTimestamp) / 1_000_000_000.0;
        tokens = Math.min(capacity, tokens + secondsElapsed * refillRatePerSecond);
        lastRefillTimestamp = now;
    }
}

20.4 Content moderation and spam

Newly created posts commonly pass through an asynchronous moderation pipeline — automated classifiers looking for spam, harassment, or policy-violating content, sometimes backed by human review for borderline cases — running in parallel with (not blocking) the normal fan-out pipeline. A post can be live and fanning out to followers while moderation runs in the background, with the ability to retract or hide a post shortly after if it is flagged, rather than delaying every single post while every single one waits for a moderation verdict.

20.5 Protecting the follow graph from scraping

Follower and following lists are sensitive from an abuse perspective — they are exactly the data set that malicious actors want for large-scale scraping or harassment campaigns. Beyond basic rate limiting, production systems often add anomaly detection specifically on graph-read patterns (a single account rapidly querying millions of different users’ follower lists is a strong scraping signal) and may require additional authentication or CAPTCHA-style challenges when that pattern is detected.

A system this large fails in ways that are only visible if you are specifically watching for them. Good observability is not an afterthought bolted on at the end — it is what makes it possible to run this system confidently in production at all.

21.1 The three pillars, applied specifically to this system

21.2 Alerting on the right signals

The most useful alerts for this system are ones tied directly to user-facing symptoms rather than internal implementation details: feed read latency crossing a threshold, fan-out queue depth growing faster than it is draining (a leading indicator of an unfolding celebrity-post overload), and cache hit ratio dropping sharply (which often signals a cache eviction storm or a newly-viral piece of content overwhelming existing capacity).

21.3 Dashboards built around the celebrity scenario specifically

Given how central the celebrity problem is to this design, a dedicated dashboard tracking active large-scale fan-out jobs, their progress, and their resource consumption is genuinely valuable operationally — it turns “is a celebrity post currently stressing the system” from a question engineers discover the hard way into something they can see clearly and act on proactively.

22.1 Containerization and orchestration

Each service — Post Service, Fanout Service, Feed Service, Ranking Service — is packaged as a container and run under an orchestrator like Kubernetes, which handles scheduling, automatic restarts on failure, and horizontal auto-scaling based on load metrics (like CPU usage or queue depth). This is what lets the Fanout Service, specifically, scale out additional worker instances automatically during a celebrity-post-triggered traffic spike, and scale back down afterward to control cost.

22.2 Multi-region deployment

With 500 million daily active users spread globally, running everything from a single data center would mean users far from that location experience high network latency on every request. Production systems deploy across multiple geographic regions, routing users to their nearest region via DNS-based or Anycast-based routing, with data replicated across regions for both latency and disaster-recovery purposes.

22.3 CI/CD and safe rollouts

Given how central the Ranking Service and Fanout Service are, changes to either need careful, gradual rollout rather than an all-at-once deployment. Common patterns include canary deployments (rolling a new version out to a small percentage of traffic first, watching key metrics, then expanding) and feature flags (letting a new ranking model or fan-out threshold be toggled independently of a full code deployment, and rolled back instantly if something looks wrong).

23.1 Disaster recovery

The multi-region replication described above doubles as disaster recovery: if an entire region becomes unavailable (a data center outage, a natural disaster, a major cloud provider incident), traffic can be rerouted to healthy regions, since data was already being replicated there continuously rather than only backed up periodically. This is meaningfully better than a traditional “restore from last night’s backup” recovery model, because it minimizes both downtime (RTO — recovery time objective) and data loss (RPO — recovery point objective).

23.2 Backup strategy for the source-of-truth stores

Even with multi-region replication handling most failure scenarios, regular backups of the primary post and graph stores protect against a different class of problem: logical corruption or accidental deletion, which replication would faithfully copy everywhere just as fast as it copies legitimate data. Point-in-time snapshots, retained on a rolling schedule (e.g., daily for 30 days, weekly for a year), give a way to recover from “we accidentally deleted the wrong data” in a way that live replication cannot.

23.3 Cost optimization

At this scale, infrastructure cost becomes a genuine engineering concern, not just a finance question. A few concrete levers specific to this design:

A few classic data structure and algorithm ideas show up naturally in this system, and interviewers often ask candidates to actually implement one of them on the spot.

24.1 Merging K sorted lists — the heart of feed assembly

When the pull-model path fetches recent posts from several followed accounts (each individually already sorted by time), combining them into one overall time-sorted list is exactly the classic “merge K sorted lists” problem. The efficient solution uses a min-heap (priority queue): keep one pointer per source list, always pull the globally smallest (or, for our purposes, most recent) next item from the heap, and push that source’s next item back onto the heap.

// Merge K sorted post lists (by timestamp, descending) into one feed
public class FeedMerger {

    public static List<Post> mergeKSortedFeeds(List<List<Post>> sources, int limit) {
        // Max-heap ordered by timestamp so the most recent post is always on top
        PriorityQueue<PostCursor> heap = new PriorityQueue<>(
            Comparator.comparingLong((PostCursor c) -> c.post.getTimestamp()).reversed()
        );

        // Seed the heap with the first (most recent) post from each source
        for (int i = 0; i < sources.size(); i++) {
            List<Post> source = sources.get(i);
            if (!source.isEmpty()) {
                heap.add(new PostCursor(source.get(0), i, 0));
            }
        }

        List<Post> result = new ArrayList<>();
        while (!heap.isEmpty() && result.size() < limit) {
            PostCursor cursor = heap.poll();
            result.add(cursor.post);

            List<Post> source = sources.get(cursor.sourceIndex);
            int nextIndex = cursor.indexInSource + 1;
            if (nextIndex < source.size()) {
                heap.add(new PostCursor(source.get(nextIndex), cursor.sourceIndex, nextIndex));
            }
        }
        return result;
    }

    private record PostCursor(Post post, int sourceIndex, int indexInSource) {}
}

Time complexity: with K sources and N total posts across all of them, this runs in O(N log K) — each of the N items is pushed and popped from a heap of size at most K, and each heap operation costs O(log K). This beats the naive approach of concatenating everything and sorting from scratch, which would cost O(N log N) — meaningfully worse when K (the number of followed accounts contributing) is much smaller than N (the total candidate posts).

24.2 LRU cache — the eviction policy behind most caching layers

When a cache reaches its capacity limit, it needs a policy for deciding what to remove to make room for new entries. Least Recently Used (LRU) is the most common choice for feed and content caches, because social content access naturally clusters around recency — old, unused entries really are the ones least likely to be needed again soon.

// LRU cache using a HashMap + doubly linked list, O(1) get and put
public class LRUCache<K, V> {

    private final int capacity;
    private final Map<K, Node<K, V>> map = new HashMap<>();
    private final Node<K, V> head = new Node<>(null, null); // most recently used end
    private final Node<K, V> tail = new Node<>(null, null); // least recently used end

    public LRUCache(int capacity) {
        this.capacity = capacity;
        head.next = tail;
        tail.prev = head;
    }

    public V get(K key) {
        Node<K, V> node = map.get(key);
        if (node == null) return null;
        moveToFront(node);
        return node.value;
    }

    public void put(K key, V value) {
        if (map.containsKey(key)) {
            Node<K, V> node = map.get(key);
            node.value = value;
            moveToFront(node);
            return;
        }
        if (map.size() >= capacity) {
            Node<K, V> lru = tail.prev;
            remove(lru);
            map.remove(lru.key);
        }
        Node<K, V> node = new Node<>(key, value);
        map.put(key, node);
        addToFront(node);
    }

    private void moveToFront(Node<K, V> node) { remove(node); addToFront(node); }

    private void addToFront(Node<K, V> node) {
        node.next = head.next;
        node.prev = head;
        head.next.prev = node;
        head.next = node;
    }

    private void remove(Node<K, V> node) {
        node.prev.next = node.next;
        node.next.prev = node.prev;
    }

    private static class Node<K, V> {
        K key; V value;
        Node<K, V> prev, next;
        Node(K key, V value) { this.key = key; this.value = value; }
    }
}

24.3 Concurrency considerations in the fan-out worker pool

Fan-out workers process events concurrently across many threads or processes for throughput, which introduces classic concurrency concerns: two workers must never double-process the same event (handled via the idempotency approach discussed earlier), and shared resources like a connection pool to Redis must be sized and managed carefully — too few connections and workers block waiting for one; too many and you overwhelm the downstream Redis cluster itself. A bounded thread pool combined with a bounded connection pool, sized based on load testing rather than guesswork, is the standard, practical answer.

25.1 Best practices worth internalizing

25.2 Common mistakes candidates make in interviews (and engineers make in real systems)

Let us bring everything together into the core narrative you would want to walk an interviewer through, start to finish, if asked to design this system live.