System Design interview questions: Sharding

Shard on a high-cardinality key, such as a hash of the tenant or entity id, so writes spread evenly across partitions. A monotonic timestamp routes every current write to whichever shard owns the latest range, creating a hot partition no matter how many shards exist. Adding replicas or vertically scaling the busy shard only raises one box's ceiling; neither balances the write distribution.

Use Snowflake-style ids that pack a timestamp, a node id, and a per-node sequence counter — coordination-free, unique, and naturally time-ordered. Every node mints ids locally with no round-trip, and the timestamp prefix gives the index locality you want. A central sequence reintroduces the coordination and single point of failure; UUIDv4 is coordination-free but fully random, so it scatters B-tree inserts; and hashing row contents lets identical rows collide.

The approaches that genuinely help are splitting the hot key range into smaller sub-ranges and migrating some while double-writing during cutover, using consistent hashing with virtual nodes so adding capacity remaps only a fraction of keys, and introducing a finer-grained or composite shard key so the concentrated key spreads across partitions. Taking the shard offline to dump and restore is downtime, and vertically scaling only the hot node never rebalances anything.

A good shard key has high cardinality and a roughly uniform access distribution so reads and writes spread evenly across shards, and it ideally matches your most common query so requests hit a single shard instead of scattering. A bad shard key creates a hotspot: a low-cardinality or monotonically increasing key (like an auto-increment id or a timestamp) sends most traffic to one shard, producing uneven load and a hot partition while others sit idle. It also forces expensive scatter-gather queries and painful resharding when one shard outgrows its capacity.

Requirements. Two operations dominate: create (write a mapping, optionally with a custom alias and expiry) and redirect (look up a code and 30x to the long URL). The system is overwhelmingly read-heavy (100:1), so the redirect path is what we optimise. I'd confirm: are custom aliases required (yes), do links expire (optional TTL), and how precise must click counts be (approximate/eventually-consistent is fine). I'd use a 302 (temporary) redirect so we keep serving redirects through our system and can still count clicks; a 301 would let browsers cache and bypass us.

Capacity. ~40 writes/sec average, ~200/sec peak; ~4K reads/sec average, ~20K peak. Storage: ~6B links over 5 years; at ~500 bytes/row (code, URL, metadata) that's ~3 TB — comfortably shardable. Keyspace: base62 with 7 characters gives 62⁷ ≈ 3.5×10¹² codes, far more than 6B, so 7 chars (often padded to a fixed length) is plenty. The hot working set (recently/ popularly accessed links) is a small fraction of 6B, so a cache of tens of GB covers the bulk of redirect traffic.

Data model. A single mapping table/collection keyed by the short code: code (PK) → long_url, created_at, expires_at, owner_id. Point lookups by primary key suit a key-value store (DynamoDB/Redis-backed) or a sharded relational table sharded by code. Click counts live separately — incrementing a counter on every redirect would put write load on the read path — so clicks are emitted as events and aggregated asynchronously.

Short-code generation. Generate a unique 64-bit id (a distributed counter handed out in ranges per server, or a Snowflake-style id) and base62-encode it to get the short code; this guarantees uniqueness with no collision checks and no hotspot from a single shared counter. Custom aliases are written directly with a uniqueness check and stored in the same table, reserving that code. (A hash-of-URL scheme is the alternative but needs collision handling and breaks idempotency for duplicate URLs.)

Read/write paths. Write: allocate id → base62 → insert mapping → return the short URL. Redirect: a load balancer / CDN fronts stateless app servers; the server does a cache-aside lookup (code → long_url) in Redis, falling back to the store on a miss and populating the cache, then returns a 302. This keeps p99 well under 100 ms for cache hits. Expiry is enforced by TTL on both the row and the cache entry. Each redirect fires a lightweight click event onto a queue (Kafka) that a consumer aggregates into per-link counts.

Bottleneck & scaling. The bottleneck is the redirect read path at peak. We scale it with cache layers (most reads never touch the store), read replicas, and sharding the mapping store by code so lookups stay single-shard. Id generation scales by handing each server an independent id range (or using Snowflake), avoiding a single global counter as a SPOF. At 10× traffic, the cache absorbs most of it; we add cache nodes and replicas and, if needed, push redirects further to the edge.

False. Monotonically increasing keys concentrate all writes on the last shard — the one owning the current maximum range — the classic hotspot-shard problem, while other shards sit idle for writes. Even though such keys are high-cardinality, they are temporally correlated. Even write distribution needs a hash-based or consistent-hash key, or random keys like UUIDv4; timestamp-ordered ULIDs recreate the same hotspot.

Naive sharding with hash(key) % N remaps almost every key when N changes, forcing a massive reshuffle whenever you add or remove a node. Consistent hashing places nodes and keys on a hash ring so that adding/removing a node only moves the keys in one adjacent arc — roughly 1/N of the data — rather than nearly all of it. Virtual nodes (many ring positions per physical node) smooth out the uneven key distribution and let you weight heterogeneous machines, avoiding hot shards.

Requirements. Two delivery modes dominate: online (push immediately over a live connection) and offline (store and deliver on reconnect). Groups are bounded (~256), so a message fans out to at most a few hundred recipients. I'd confirm ordering is per-conversation (not global), and that 'exactly once' really means at-least-once delivery plus client dedup by message id. Presence and receipts are separate, lighter sub-systems.

Connection layer. Clients hold a WebSocket to a stateless gateway server (one of thousands; at ~500K connections each, ~1,000 servers cover the peak). The hard part is routing: when A sends to B, A's gateway must find B's gateway. A connection registry (user_id → gateway_id, in Redis) updated on connect/disconnect, or a pub/sub channel per user that B's gateway subscribes to, does this. The sender's gateway looks up B, forwards the message to B's gateway, which pushes it down B's socket.

Data model. Messages are stored keyed by conversation_id and ordered by a time-sortable id, in a write-optimised, horizontally sharded store (Cassandra/HBase-style), sharded by conversation. For offline delivery each recipient has a mailbox/inbox (a queue of undelivered message ids); on reconnect the client drains it. Recent history is hot (cached / on fast storage); old history ages into cold archive.

Send → deliver flow. Send → server assigns a message id and persists to the conversation log → for each recipient, if online, push to their gateway; if offline, enqueue in their mailbox. Delivery is at-least-once; the client deduplicates by message id, giving exactly-once feel. Delivery and read receipts are just small messages flowing the other way, updating per-message state. The sender never blocks on fan-out — it returns once the message is durably persisted.

Storage & fan-out scaling. ~15 TB/day shards cleanly by conversation; group fan-out (1 send → up to 256 deliveries) is done asynchronously by workers reading the conversation log, so a large group never stalls the sender. Hot recent messages live in cache; archives compress to cheap storage.

Presence & failure. Presence is a heartbeat with a short TTL in an in-memory store — the client refreshes every ~30 s, and absence of a refresh means offline; we never write every status flip to a database. If a gateway crashes, its ~500K connections drop, clients reconnect to another gateway (the registry updates), and any messages sent meanwhile are already durably in each recipient's mailbox, so nothing is lost — they arrive on reconnect.

Requirements. The workload is ~20:1 read-heavy, so the read (feed assembly) path is what we optimise. Ordering can be reverse-chronological to start (ranking is an orthogonal layer). The defining edge case is the follower skew: a normal post reaches ~200 timelines, a celebrity post reaches up to 100M — those need different handling.

Fan-out strategy. The core decision. Fan-out-on-write (push each post id into every follower's precomputed timeline) makes reads cheap but explodes on celebrities — one post = 100M writes. Fan-out-on-read (assemble at query time by pulling recent posts from everyone you follow) makes celebrity posts free but makes every feed read expensive. I'd use a hybrid: push on write for ordinary accounts, and for celebrities skip the push and pull their recent posts at read time, merging them into the timeline. This caps write amplification while keeping normal reads cheap.

Feed storage. Each user has a precomputed timeline: a capped list of recent post ids (say the latest few hundred) in Redis. Post bodies live once in a separate post store (sharded by post id). Storing ids — not full copies — per follower keeps the timeline small and means an edited/deleted post is resolved at hydration time.

Read path. On refresh: read the user's precomputed timeline (ids from pushed accounts) → pull recent post ids from the handful of celebrities they follow → merge and sort by time → hydrate the top N bodies from the post cache → return. Almost everything is served from cache, so p99 stays under 200 ms. The celebrity pull is bounded because a user follows only a few of them.

Bottleneck & scaling. The bottleneck is fan-out write amplification: 100M posts/day × ~200 avg followers ≈ tens of billions of timeline writes/day, and the tail (celebrities) is what kills pure push. The hybrid split removes the worst offenders; the remaining fan-out runs on async workers off a queue so posting never blocks; timelines and post bodies are cached and sharded. At higher load we add fan-out workers and cache capacity, and tune the push/pull follower threshold.

Requirements. Beyond upload/download, the product is sync: a change on one device must propagate to the others, efficiently. 'Efficiently' means delta sync — never re-send an unchanged file. We must handle conflicts (two offline edits) without data loss and keep version history. Metadata ops (browse folders) are latency-sensitive; bulk byte transfer is throughput-bound.

Chunking & dedup. Files are split into content-hashed chunks (~4 MB; a chunk's id is the hash of its bytes). Editing one paragraph of a 1 GB file changes only the chunk(s) covering it, so the client uploads just those — a few MB, not a GB. Because chunks are content-addressed, identical chunks (same file shared by many users, or unchanged blocks across versions) are stored once — large dedup savings.

Block vs metadata split. Chunk bytes go to an object store (S3-style): content-addressed, effectively infinite, optimised for large immutable blobs — this holds the exabytes. The metadata DB holds the file→chunk-list mapping, the folder tree, ownership, and version history — small records with fast point/range reads to meet the <100 ms browse budget. The two layers have opposite access patterns, so they're separate systems.

Sync protocol & conflicts. Each device keeps a local view and learns of remote changes via a notification service (long-poll/push): 'your namespace changed past version V'. The client then pulls the metadata diff since its last-known version and downloads only the new chunks. Concurrent offline edits are detected by version: the second writer's base version is stale, so rather than overwrite, the system keeps both as a conflict copy (or merges where it safely can) and surfaces it — never a silent loss.

Consistency. Per file/namespace there's a monotonic version; metadata operations are ordered by it so all devices converge on the same state. Upload is commit-after-chunks: the client first writes all chunks to the object store, then flips the metadata pointer to the new chunk list. A half-uploaded file's pointer never goes live, so a reader never sees a file referencing chunks that aren't fully stored.

Scaling. The metadata DB shards by user_id (a user's files and folder tree stay together). The object store scales horizontally on its own. Sync notifications fan out to each of a user's devices. Dedup + delta sync are the multipliers — they cut both stored bytes and transferred bytes by large factors, which is what makes exabyte-scale storage and 5,800 ops/sec affordable.