QuickCart Stage 4: Scaling Writes With Replication and Sharding

Splitting the data: sharding

Sharding and partitioning

Partitioning divides a data set into non-overlapping pieces. Sharding is partitioning where the pieces live on different machines, each serving reads and writes for its subset; a shard is usually a replicated group. Sharding goes deeper. Three ways to decide where a row goes:

QuickCart hashes the shopper id to a point on a ring, and a small ring map says which shard owns each arc: hash plus directory.

Choosing the shard key for orders

The shard key decides which queries stay on one shard. Choose it from the queries you must serve cheaply, not from the column that looks unique.

QuickCart shards orders by shopper id. The price is that a store asking "my open orders" would hit every shard. Stores and couriers instead read a read model fed from order events (see the order pipeline and CQRS).

CREATE TABLE orders (
  order_id    bigint PRIMARY KEY,   -- unique across all shards (see the ids section)
  shopper_id  bigint      NOT NULL, -- the shard key
  store_id    bigint      NOT NULL,
  status      text        NOT NULL,
  placed_at   timestamptz NOT NULL
);

CREATE INDEX orders_by_shopper ON orders (shopper_id, placed_at DESC);

"My 20 latest orders" is WHERE shopper_id = $1 ORDER BY placed_at DESC LIMIT 20, answered by one shard from one index.

Hot shards, resharding and cross-shard queries

A hot shard gets far more traffic than its siblings. Hashing spreads ordinary shoppers evenly, so the risks are special cases: a business account ordering at a rate no household would, or a bot. Responses, in order: rate-limit abusive keys, give a heavy account its own shard (a directory entry makes this cheap), and only then split one key across shards.

Resharding changes the layout while the system runs. Consistent hashing limits what moves, but moving an arc still means copying rows, catching up with changes made meanwhile, and switching ownership at one instant (see the rollout below).

Cross-shard queries run on every shard and merge results, so latency is the slowest shard's, and a cross-shard transaction cannot use a plain local one. Checkout meets this at once: it touches the order (shopper's shard) and stock (the inventory side). QuickCart avoids two-phase commit by splitting the work into steps with compensations, a saga (Stage 5; see distributed transactions).

Figure 4. Routing one order write. How to read this: the router hashes the shopper id, finds the owning arc in the cached ring map, and sends the write to that shard only. The others see nothing, which is the point of matching the shard key to the query.

Consistent hashing

Consistent hashing assigns keys to nodes so adding or removing a node moves only a small share of keys. With hash(key) mod N, going from 3 shards to 4 keeps a key in place only when its hash gives the same remainder for both, which happens for 3 of every 12 hash values, so 75 percent of keys move, at the moment you needed capacity.

1. Arrange hash values on a circle, the ring. Here positions run 0 to 99; a real ring uses something like 64 bits.
2. Hash each node (shard) to one or more points on the ring.
3. Hash each key, here shopper_id, to a point.
4. The key belongs to the first node point at or after it, walking clockwise and wrapping at the end.

Worked example with shards at A = 20, B = 50, C = 80:

A owns 81 to 99 and 0 to 20 (40 positions), B owns 21 to 50 (30), C owns 51 to 80 (30). Add shard D at 65. D takes only the arc 51 to 65, which C owned. Shopper 1003 moves from C to D and everyone else stays: 15 of 100 positions moved, not 75.

Figure 5. The tiny ring before and after adding a node. How to read this: arrows go from a shopper to the owning shard. Compare the two groups: exactly one arrow changed, the key inside D's new arc.

Removal is the mirror. If B leaves, its arc 21 to 50 goes to the next point clockwise, C, which then owns 21 to 80 (60 percent of the ring). One neighbour doubles, which virtual nodes (vnodes) fix: give each physical shard many points on the ring. Arcs become small, shares even out, a departing shard's load scatters over many neighbours, and an arriving shard takes a little from many. Cassandra's documentation describes the same idea (tokens) and notes more tokens slow cluster-wide maintenance. A lookup is a binary search:

type Point = { position: bigint; shard: string };

// points must be sorted by position, ascending
function ownerOf(points: Point[], keyHash: bigint): string {
  let lo = 0;
  let hi = points.length; // find the first point with position >= keyHash
  while (lo < hi) {
    const mid = (lo + hi) >> 1;
    if (points[mid].position < keyHash) lo = mid + 1;
    else hi = mid;
  }
  return points[lo === points.length ? 0 : lo].shard; // wrap around the ring
}

For hash 91 on the ring above, no point is at or after 91, so the lookup wraps to the first point, A. Use a hash that is stable across processes and languages (Python's built-in string hash is randomised per process), and version the ring map so every router can tell whether it is behind. QuickCart's choice: 4 shards with 64 virtual points each (a teaching example). A simpler alternative is fixed partitions: many more logical partitions than nodes, moved whole between nodes.

What you promise: CAP, PACELC and consistency models

CAP and PACELC

The CAP theorem is a precise result, not a menu. Gilbert and Lynch (2002) proved Brewer's conjecture for an asynchronous network: a read/write object cannot be guaranteed both consistent (atomic, that is linearizable) and available (every request to a non-failing node gets a response) in all executions where messages can be lost, that is, when the network may partition.

What it does not say:

• It does not say "pick any two of three". Partitions are a fact of networks, so the real choice is C versus A during a partition.
• "Consistency" means linearizability, not ACID consistency. "Availability" means every non-failing node responds, not "99.9 percent uptime".
• It says nothing about normal operation or latency, which is what you feel daily.

PACELC (Daniel Abadi, "Consistency Tradeoffs in Modern Distributed Database System Design", IEEE Computer, 2012) adds the other half: if there is a Partition, choose A or C; Else, choose Latency or Consistency. Waiting for a synchronous standby buys C at the cost of L; answering from an asynchronous replica buys L at the cost of C.

Figure 6. CAP is decided per operation, not per system. How to read this: one partition, three answers. Money and identity choose consistency and accept waits or failures; a status read and a browse page choose availability and accept slightly old data.

In PACELC terms QuickCart's orders are PC/EC; the catalogue and browse pages are PA/EL. More in CAP and consistency.

Consistency models

A consistency model is a promise about what a read may return after writes. The strongest is only the most expensive, so use it where wrong data costs real money:

• Linearizable ("strong"): each operation appears to happen atomically at one instant between its start and end, respecting real time; after a write completes, every later read sees it. A single primary gives this.
• Sequential: everyone sees one agreed order that respects each client's own order, but not necessarily real time.
• Causal: causally related operations (a reply, a write that depended on a read) are seen in that order by everyone; unrelated ones may differ.
• Eventual: if writes stop, copies converge. Meanwhile reads can be old and readers can disagree.
• Read-your-writes (a session guarantee): a client always sees its own earlier writes. Monotonic reads adds that it never sees data go backwards.

Read-your-writes for "my order just placed"

This is the bug from Figure 2, and it needs an explicit decision.

QuickCart combines the last two. POST /orders returns the full order, so the confirmation page needs no read. The response also carries an opaque consistency token (QuickCart's own design, not a standard) holding the primary's WAL position just after the commit. A later read passes it back, and the router uses a replica only if that replica has replayed at least that position; otherwise it reads the primary.

-- on the shard primary, right after COMMIT: the token to hand to the client
SELECT pg_current_wal_lsn();            -- for example '0/3000148'

-- on a replica: may this replica serve the read?
SELECT pg_last_wal_replay_lsn() >= '0/3000148'::pg_lsn;

The router can keep each replica's last reported replay position from health checks. WAL positions only advance, so a stale value is conservative: extra primary reads, never a stale answer. Each shopper is also pinned to one replica (by hash of the shopper id) for monotonic reads.

Figure 7. The read-your-writes decision. How to read this: most reads end at a replica, sparing the primary. Only a read that could expose lag goes to the primary.

Names that never collide: unique ids

Unique ID generation

With one database, an auto-increment column is perfect: compact, ordered, one source. With shards it breaks: shard 1 and shard 3 each count 1, 2, 3, so "order 1,000,017" can exist twice. Order ids travel far beyond the shard, into event streams, the courier app and support tools, where all orders mix. Ids must be unique across all shards without a central call per order.

Time order matters because B-tree indexes like inserts near the end; v7 and ULID give it, v4 does not.

A Snowflake-style id is a design, not a library. The original Snowflake service split 64 bits like this; variants change the widths:

| 1 bit unused | 41 bits: milliseconds since a custom epoch | 10 bits: worker id | 12 bits: sequence |

2^41 milliseconds is about 69.7 years from the epoch; 10 worker bits allow 1,024 generators; 12 sequence bits allow 4,096 ids per millisecond per worker. QuickCart needs a sliver of that (58 checkouts per second at peak). Decoding one id:

QuickCart's epoch is 2026-01-01T00:00:00Z (1,767,225,600,000 Unix milliseconds). The generator refuses to issue an id when the clock moves backwards, because that is how duplicates are born.

const EPOCH_MS = 1_767_225_600_000n; // 2026-01-01T00:00:00Z; fixed forever
const WORKER_BITS = 10n;
const SEQ_BITS = 12n;
const MAX_SEQ = (1n << SEQ_BITS) - 1n; // 4095

export class OrderIdGenerator {
  private lastMs = -1n;
  private seq = 0n;

  constructor(private readonly workerId: bigint) {
    if (workerId < 0n || workerId >= 1n << WORKER_BITS) throw new Error("worker id out of range");
  }

  next(): bigint {
    let ms = BigInt(Date.now());
    if (ms < this.lastMs) throw new Error("clock moved backwards; refusing to issue ids");
    if (ms === this.lastMs) {
      this.seq = (this.seq + 1n) & MAX_SEQ;
      if (this.seq === 0n) {
        // sequence used up in this millisecond: wait for the next one
        while (ms <= this.lastMs) ms = BigInt(Date.now());
      }
    } else {
      this.seq = 0n;
    }
    this.lastMs = ms;
    return ((ms - EPOCH_MS) << (WORKER_BITS + SEQ_BITS)) | (this.workerId << SEQ_BITS) | this.seq;
  }
}

Two duties follow. Each running generator needs a worker id nobody else holds: assign it from a coordination service, a stable instance number, or an expiring lease so a crashed worker's id is not reused while it might still be alive. And the clock must be disciplined (IDs, clocks and ordering covers both). Ids are ordered only to about a millisecond, so they are not proof of event order.

QuickCart's choice is Snowflake-style 64-bit ids from the order service: half a UUID's size in every key and index, time-ordered, unique across shards because the worker field differs, no network call. UUID v7 would also be sound for a team that prefers to run no worker leases. Shoppers see a separate random order code, since these ids are guessable in principle.

Decisions and trade-offs

In practice: QuickCart at this stage

This is a realistic scenario, not a case history. It is a composite built from common production patterns, it describes no real company, and every number in it is an illustrative assumption.

The situation

QuickCart is heading from 5,000,000 toward 10,000,000 monthly active users. Orders sit in one PostgreSQL primary with a read replica. The orders table is by far the largest, the restore test takes longer each month, and the forecast shows several terabytes inside the planning horizon. Even a clean failover stops checkout for everyone.

Starting architecture

Figure 1, with an auto-increment order_id and an index on shopper_id. No code is shard-aware.

Options considered

The decision

Rollout and verification

1. Ids first. New orders get Snowflake-style ids; old small ids stay valid and cannot overlap them.
2. Router with a one-shard ring. The ring map assigns every arc to the existing database, so behaviour is unchanged; only the code path is new. Run it behind a flag.
3. Provision shards and copy one arc at a time. Bulk copy, then catch up with changes (logical replication or change data capture). Verify row counts and checksums per arc.
4. Cut over an arc. Pause that arc's writes for seconds, apply the last changes, publish the next ring-map version, resume. Every shard rejects a write for an arc it does not own, so a router with an old map cannot misplace data.
5. Verify. Shadow-read old and new owners and compare; watch per-shard write rate, skew, replica lag, rejected writes and primary read share.

Figure 8. One arc moving during the rollout. How to read this: data flows from the old owner to the new one (dashed) while writes still go to the old owner. At cutover the map version changes and writes go to the new shard.

The 2 a.m. failure

At 02:00, during an arc cutover, one order-service pod misses the ring-map update because its watch connection dropped. For nine minutes (illustrative) it keeps sending writes for shoppers in the moved arc to the old shard. Next morning support sees "my order vanished" tickets: the orders exist, but on a shard that no longer owns the arc, so "my orders" on the new owner does not show them.

The alert says "rejected writes by non-owner shard above 0" or "ring-map versions differ across pods". First three checks: the map version each pod reports; rejected-write counts on the old shard (the step 4 ownership check makes this loud; without it orders land silently in the wrong place); orders on the old shard with placed_at after the cutover. Mitigation: force-refresh the stale pod, then replay the misplaced rows to their owner. Because order ids are unique across shards, they copy over with no collision. Follow-ups: pods refuse to serve with a map older than the control plane's, and the runbook resumes writes only after every pod confirms the new version.

Architecture after this stage

Figure 9. QuickCart after Stage 4. How to read this: highlighted boxes are new in this stage; muted ones are unchanged. Order traffic passes through the router to one of four shards; everything else is as in Stage 3.

What QuickCart looks like now

Architect's notes

• Shard last, but design for it early. Make every order query name its shard key long before you split; adding it later touches every query.
• Version everything that routes. The ring map and each shard's ownership check make resharding safe.
• Name the id owner. Worker leases and clock policy need an owner.

Common mistakes

• Sharding for the write rate alone. At these numbers one primary may cope.
• Choosing the key from the table, not the queries.
• hash(key) mod N. It works until you add a shard, then moves most of the data.
• Trusting W + R > N in every case. Sloppy quorums, concurrent writes and clock-based last-write-wins break it.
• Per-shard auto-increment ids. They collide once data from two shards is combined.
• Trusting the clock in an id generator. Refuse on a backwards step.

Key takeaways

• Replication copies data for availability and read scale; sharding splits it for write scale and size, and each shard is a replicated group.
• Synchronous commit trades latency and availability for durability; asynchronous replication trades possible loss and stale reads for speed. Multi-leader and leaderless designs buy write availability at the cost of conflicts.
• Choose the shard key from the dominant query: shopper_id for QuickCart orders. Consistent hashing with virtual nodes moves a small share of keys when a shard joins.
• CAP concerns linearizability and availability during a partition; PACELC adds latency versus consistency in normal operation.
• Assign a consistency model per data type: money and orders strong, browsing eventual, "my order" read-your-writes.
• Order ids must be unique across shards; Snowflake-style or time-ordered UUID ids do that without a central call.

Practice exercises

Exercise 1 - Starter (30 min): On the tiny ring (A = 20, B = 50, C = 80), place shoppers at hash positions 5, 22, 50, 51, 80 and 99. Add shard D at 35 and list who moves and what share of the ring moved.

Exercise 2 - Intermediate (2 hours): Add three QuickCart data types to the consistency table (for example delivery slots or coupon redemptions). Choose a model for each and say what breaks under a weaker one.

Exercise 3 - Intermediate (1 hour): Decode the id 362,387,865,620,483 into its three fields, then compute the id 100 ms later from worker 5 with sequence 0.

Exercise 4 - Advanced (half day): Extend ownerOf to return the owner plus the next two distinct shards clockwise (a replica set for N = 3), skipping repeated physical shards when virtual nodes are used.

Related reading

• QuickCart Stage 2: Scaling Reads - read replicas and caching.
• QuickCart Stage 3: Choosing the Data Model - the stores this stage builds on.
• QuickCart Stage 5: The Order Pipeline - the saga that crosses shards.
• System Design: Replication - leaders, followers, lag.
• System Design: Sharding - shard keys and resharding.
• System Design: CAP and Consistency - the consistency vocabulary used here.
• System Design: IDs, Clocks and Ordering - clocks, worker ids and ordering.

Official documentation

• PostgreSQL: log-shipping standby servers and synchronous replication - streaming and synchronous replication.
• PostgreSQL: write-ahead log settings, including synchronous_commit - what each commit level waits for.
• Apache Cassandra: Dynamo-style architecture - tokens, vnodes, consistency levels.
• Dynamo: Amazon's Highly Available Key-value Store (SOSP 2007) - sloppy quorum, hinted handoff.
• RFC 9562: Universally Unique IDentifiers (UUIDs) - UUID v4 and v7 layouts.
• Gilbert and Lynch: Brewer's conjecture and the feasibility of consistent, available, partition-tolerant web services - the formal CAP result.