Scaling Writes

Why This Matters

Read scaling is well-understood: add caches, add replicas, add CDN. Write scaling is harder because every write must eventually land on a single authoritative copy of the data. You cannot cache a mutation. You cannot serve a stale write. The moment your system exceeds what a single database node can absorb, you are forced into architectural decisions — sharding, buffering, event sourcing — that constrain every query pattern, migration, and feature for years.

This is why write scaling questions separate Staff from senior in system design interviews. Seniors propose "shard the database" as a performance optimization. Staff engineers understand that sharding is a schema commitment: the shard key determines which queries are cheap (shard-local), which are expensive (scatter-gather), and which become impossible. Getting the shard key wrong costs a quarter of migration work. Getting it right requires understanding the access patterns deeply enough to predict what the system needs two years from now.

If you can articulate why you chose a specific write-scaling strategy, name the queries it makes expensive, and explain what vertical scaling would have given you for free — you demonstrate the systems reasoning that Staff interviews demand.

The 60-Second Version

• Write scaling is a schema commitment. Sharding, partitioning, and event sourcing constrain every query you write for years. This is not a performance tuning exercise — it is an architectural decision you will live with long after the throughput crisis passes.
• Exhaust vertical scaling first. A bigger machine preserves all existing query patterns and costs nothing in engineering complexity. PostgreSQL handles ~10K TPS on a single node. If you are at 2K TPS, you have 5x headroom — do not shard prematurely.
• The shard key is a multi-year decision. Choose based on: high cardinality (even distribution), present in all queries (no scatter-gather), natural business boundary (data locality), and immutable (no row migration).
• Not all writes need the same latency. Payment confirmations require synchronous acknowledgment. Analytics events can tolerate 500ms of buffering. Design two write paths — sync for critical, async for everything else.
• Cross-shard queries are the ongoing tax. Accept this for rare operations (analytics, batch reports), but never let user-facing features depend on scatter-gather. If a feature needs all-shard data, build a pre-computed view.
• Hot shards are inevitable. The question is not how to prevent them but how fast you can detect and isolate them. Per-shard dashboards from day one — aggregate metrics hide shard-level problems.

The Problem

Write throughput hits a ceiling when a single node owns all mutations. Scaling writes means distributing mutation responsibility across nodes, partitions, or time windows. Every distribution strategy is a schema commitment that affects query patterns, consistency model, and operational complexity for years. You are not optimizing throughput — you are choosing a data model you will live with long after the performance crisis passes.

Playbooks That Use This Pattern

• Database Sharding — Shard key selection, rebalancing, cross-shard query cost
• Stream Processing — Partitioned event ingestion, ordering guarantees per partition
• Payment Processing — Ledger write distribution, double-entry consistency
• Order Matching — Matching engine write throughput, contention on order books
• Leaderboard & Counting — High-frequency counter updates, approximate vs. exact counts

The Core Tradeoff

Staff Default Position

Exhaust vertical scaling first — it preserves every existing query pattern and costs nothing in engineering complexity. When vertical scaling is insufficient, shard by natural business key (user ID, tenant ID, account ID). Event sourcing only when the audit trail is a stated business requirement, not a performance optimization. Batching is a latency-for-throughput trade that should be measured, not assumed.

When to Deviate

• Multi-tenant SaaS with regulatory isolation — shard by tenant even before hitting write limits; compliance forces your hand.
• Financial systems requiring reconstruction — event sourcing earns its complexity when regulators demand point-in-time replay.
• Extreme burst patterns (flash sales, market opens) — write-behind buffering is justified when the alternative is dropping writes entirely.
• Sub-millisecond latency requirements — skip batching; the latency cost is the system's core constraint, not throughput.

Common Interview Mistakes

Quick Reference

Staff Sentence Templates

Implementation Deep Dive

Sharded Writes with PostgreSQL — The Staff Default

When vertical scaling is exhausted, the next step is horizontal sharding by a natural business key. Here is the full implementation path.

Shard Key Selection Process

The shard key determines write distribution, query routing, and rebalancing cost for the lifetime of the system. Choose wrong and every query pays the tax.

Selection criteria (in order of priority):

1. High cardinality — The key must distribute writes evenly. user_id (millions of values) is good. country_code (200 values) creates hot shards.
2. Present in all queries — Every query must include the shard key, or the system must scatter-gather across all shards.
3. Natural business boundary — The key should align with data locality. User data naturally groups by user_id; cross-user queries are rare.
4. Immutable — Changing a shard key requires moving rows between shards. Use values that never change (user_id, account_id), not values that might (region, plan_type).

Composite key example — E-commerce order system:

Shard key: merchant_id (not order_id)

Why merchant_id:
  - All order queries include merchant context (merchant dashboard, reports)
  - Merchants are the tenancy boundary — data isolation is natural
  - Cardinality: 500K merchants → even distribution across 16 shards
  - Immutable: orders never change merchants

Why NOT order_id:
  - Queries by merchant (dashboard, reports) would scatter-gather across all shards
  - Cross-merchant queries are rare but intra-merchant queries are every page load

🎯 Staff Move: State the shard key and immediately state what queries become expensive. "Sharding by merchant_id means any cross-merchant analytics query must scatter-gather. That is acceptable because cross-merchant queries are batch jobs, not user-facing."

Write-Behind Buffer Implementation

When write bursts exceed what shards can absorb synchronously, a write-behind buffer smooths the spikes:

# Write path with Redis-backed buffer
function createOrder(order):
    orderId = generateId()

    # Step 1 — Validate and accept
    validate(order)

    # Step 2 — Write to durable buffer (Redis Stream or Kafka)
    redis.XADD("orders:buffer", "*",
        "order_id", orderId,
        "merchant_id", order.merchantId,
        "payload", serialize(order))

    # Step 3 — Return accepted (not committed)
    return { id: orderId, status: "accepted" }

# Async consumer drains buffer to sharded DB
function bufferConsumer():
    while true:
        entries = redis.XREADGROUP("consumers", "worker-1",
            COUNT=100, BLOCK=1000, "orders:buffer", ">")

        for entry in entries:
            shard = shardRouter.getShard(entry.merchantId)
            shard.insert(deserialize(entry.payload))
            redis.XACK("orders:buffer", "consumers", entry.id)

Durability tradeoff: Between XADD and the consumer persisting to PostgreSQL, the order exists only in Redis. If Redis crashes, buffered orders are lost. Mitigation: use Redis with AOF persistence (appendfsync everysec) or replace with Kafka for stronger durability guarantees.

🎯 Staff Move: Always name the durability gap explicitly. "There is a window between accept and persist where data lives only in the buffer. For orders, that window is typically under 500ms. If that is unacceptable, we write to Kafka with acks=all instead of Redis."

Kafka-Based Write Distribution

For sustained high write throughput, Kafka provides durable, partitioned write distribution:

# Producer: partition by shard key
producer.send(
    topic = "orders",
    key = order.merchantId,     # Kafka partitions by this key
    value = serialize(order),
    acks = "all"                # Wait for all replicas
)

# Consumer: one consumer per shard
# Kafka partition N → PostgreSQL shard N
function shardConsumer(partitionId):
    consumer = KafkaConsumer(topic="orders", partition=partitionId)
    shard = postgresShards[partitionId]

    for batch in consumer.poll(batchSize=500, timeout=1000):
        shard.beginTransaction()
        for record in batch:
            shard.insert(deserialize(record.value))
        shard.commit()
        consumer.commitOffsets()

Why Kafka partitions map to DB shards: Each Kafka partition is consumed by exactly one consumer in a consumer group. This gives you ordered, single-writer access to each shard — no write contention between consumers.

Connection and Transaction Management

The numbers that matter. PostgreSQL single node handles approximately 10K TPS for write-heavy workloads (INSERT/UPDATE with indexes). With 10 shards, expect approximately 80K TPS aggregate — not 100K, because shard routing overhead, connection management, and occasional cross-shard coordination consume 15–20% of theoretical throughput. With 100 shards, the overhead grows to 25–30% due to increased routing complexity and metadata management.

🎯 Staff Move: Quote the overhead honestly. Candidates who say "10 shards = 10x throughput" reveal they have not operated a sharded system. The real multiplier is 7–8x, and it decreases as shard count increases.

Architecture Diagram

Sync path: For writes requiring immediate acknowledgment (payment confirmations, inventory decrements), the app server routes through the shard router directly to the correct PostgreSQL shard.

Async path: For writes that tolerate buffering (analytics events, activity logs, non-critical updates), the app server publishes to Kafka. Each Kafka partition maps to one DB shard. Consumers drain partitions to their corresponding shards in batches.

🎯 Staff Move: In an interview, present both paths and let the interviewer choose which to explore. This demonstrates that you understand not all writes have the same latency requirement.

Failure Scenarios

1. Hot Shard from Skewed Partition Key

Timeline: A viral merchant generates 100x normal order volume. All orders hash to shard 7. Shard 7 hits 100% CPU and I/O saturation. Write latency on shard 7 goes from 5ms to 2,000ms. Other merchants on shard 7 experience collateral damage.

Blast radius: All tenants on the hot shard. Other shards are unaffected, but if the application retries failed writes, retry storms can cascade to the app tier.

Detection: Per-shard write latency monitoring. CPU and I/O utilization per shard instance. Query queue depth exceeding threshold.

Recovery:

1. Immediate: rate-limit the hot tenant at the application tier to cap their write throughput
2. Short-term: split the hot shard — re-hash the tenant to a dedicated shard
3. Long-term: use consistent hashing with virtual nodes so that a single tenant's traffic spreads across multiple physical shards

🎯 Staff Insight: Hot shards are inevitable. The question is not "how do we prevent them" but "how fast can we detect and isolate them." Build per-shard dashboards from day one — aggregate metrics hide shard-level problems.

2. Write Buffer Overflow During Traffic Spike

Timeline: Flash sale drives 20x normal write volume. Kafka consumers cannot keep up — consumer lag grows from 0 to 500K messages. Buffer retention period (24 hours) approaches. If consumers do not catch up, oldest messages will be dropped.

Blast radius: Delayed writes for all tenants, not just the spike source. If messages are dropped, writes are permanently lost. Downstream systems reading from the database see incomplete data.

Detection: Kafka consumer group lag (messages behind). Consumer throughput (messages/sec processed). Buffer disk utilization approaching retention limit.

Recovery:

1. Scale consumers horizontally — add more consumer instances (up to the number of partitions)
2. Increase consumer batch size to improve throughput (1,000 → 5,000 per commit)
3. If critical, temporarily bypass the buffer and write directly to shards (sync path)
4. Increase retention period to buy time for consumers to catch up

🎯 Staff Insight: Size the buffer for 10x normal load, but have a runbook for 100x. The buffer exists to absorb spikes, but if the spike exceeds buffer capacity, you need a fallback — and that fallback should be tested, not theoretical.

3. Cross-Shard Query Timeout During Scatter-Gather

Timeline: An analytics dashboard queries "total orders across all merchants for the last 24 hours." The query fans out to all 16 shards. 15 shards respond in 50ms. Shard 7 (the hot shard) responds in 8 seconds. The aggregate query times out at the 5-second threshold.

Blast radius: Analytics dashboards and reporting features that require cross-shard aggregation. Transactional user-facing queries (which are shard-local) are unaffected.

Detection: Scatter-gather query p99 latency. Per-shard response time within scatter-gather operations. Timeout rate for aggregate queries.

Recovery:

1. Set per-shard timeouts within the scatter-gather — return partial results from responsive shards with a warning
2. Pre-compute cross-shard aggregations asynchronously (materialized views rebuilt on a schedule)
3. Use a dedicated analytics replica per shard that handles aggregate queries without competing with transactional writes
4. Move cross-shard analytics to a separate OLAP system (ClickHouse, BigQuery) fed by CDC

🎯 Staff Insight: Cross-shard queries are the ongoing tax of sharding. Accept this tax for rare operations, but never let user-facing features depend on scatter-gather. If a feature requires data from all shards, build a pre-computed view.

In the Wild

Abstract write-scaling patterns are easier to internalize when you see how they were applied under real production constraints. These are public, documented examples — not speculation.

Instagram: Sharding PostgreSQL by User ID

Instagram ran on a single PostgreSQL instance until their growth forced a decision: re-architect from scratch or shard the existing system. They chose to shard PostgreSQL by user_id using a logical sharding layer (PL/Proxy routing to multiple PostgreSQL backends). Every user's photos, likes, and comments live on the same shard — making the most common queries (a user's feed, their profile, their notifications) shard-local and fast.

The Staff-level insight: Instagram chose user_id over photo_id despite photos being the dominant write entity. Why? Because 90% of queries start from a user context (a user's feed, a user's profile). Sharding by photo_id would have made user-centric queries scatter-gather across all shards — the most common operation would have been the most expensive. The shard key decision was driven by query patterns, not by write distribution.

Uber: Schemaless — Write-Optimized Sharding

Uber built Schemaless, a sharded MySQL-backed storage layer, to handle tens of millions of trip writes per day. Each row is identified by a (row_key, column_name, ref_key) triple, and the row_key determines the shard. Writes are append-only — updates create new versions rather than mutating in place. This eliminates write contention (no row-level locks) and makes the system naturally audit-friendly.

The Staff-level insight: By making writes append-only, Uber turned a write-scaling problem into a storage problem. Append-only writes never contend with each other — two writes to the same logical entity land in different physical rows. The tradeoff is that reads must merge multiple versions, but that merge is simple and cacheable. This is a deliberate choice: they paid read complexity to buy write throughput.

LinkedIn: Kafka as the Write-Distribution Layer

LinkedIn built Apache Kafka specifically to solve their write-scaling problem: hundreds of microservices each generating thousands of events per second (profile views, connection requests, message sends), with downstream systems unable to keep up with the write volume. Kafka acts as a durable, partitioned write buffer — producers write at full speed, and consumers drain at their own pace.

The Staff-level insight: LinkedIn's key insight was separating write acceptance from write processing. The producer's write latency is just the Kafka append (5–50ms). The consumer's processing latency is decoupled — it can batch, retry, and backfill without affecting the producer. This separation is the fundamental write-scaling pattern: accept fast, process later. The partition key determines write distribution, and LinkedIn uses entity-specific keys (member_id for profile events, connection_id for connection events) to maintain per-entity ordering while distributing load across partitions.

Practice Drill

Staff Interview Application

How to Introduce This Pattern

State the concrete throughput requirement, the single-node ceiling, and the chosen shard key. Then immediately name the tradeoff. This tells the interviewer you understand that sharding is a commitment, not a free lunch.

When NOT to Use This Pattern

• Below the single-node ceiling: If you are at 2K TPS and the ceiling is 10K, you have 5x headroom. Do not shard prematurely — optimize indexes, connection pooling, and query patterns first.
• No natural shard key exists: If every query touches every partition, sharding adds overhead without reducing per-node load. Consider a different data model or an OLAP system instead.
• Write contention on a single entity: If the bottleneck is concurrent updates to one row (counter, balance, inventory), sharding does not help — you need optimistic concurrency, CAS operations, or partitioned counters.
• Temporary burst, not sustained load: If writes spike for minutes during flash sales but are low otherwise, use a write-behind buffer to absorb the burst. Do not architect permanent sharding for temporary load.

Follow-Up Questions to Anticipate

Capacity Planning Quick Reference

Sizing a Sharded Write System

Write Throughput by Storage Engine

Key Numbers Worth Memorizing

Common Pitfalls Checklist