Shopify Engineering

Ideas you will meet

Database Sharding

Partition data across nodes by key, range or directory.

Resilience

Timeouts, retries, circuit breakers, bulkheads and load shedding.

What we already know about this

Our own lessons on these ideas. Read them here before you start, so it makes sense.

• Advanced Distributed Systems · 13 minResilience: timeouts, retries, bulkheads, backpressure and circuit breakersHow to stop one slow or failing dependency from taking everything down: the patterns, how they fit together and where each one goes wrong.Read hereClose

  The core problem

  In a system of many services, slowness is more dangerous than failure. A dependency that fails fast is easy to handle. One that responds in 30 seconds makes callers hold threads, connections and memory until they too stop responding: a cascading failure. Resilience patterns limit how far a problem can spread.

  The toolbox

  100%

  Figure SDA-ADS-06-1 · Resilience layers around a call · © forcsguy.com

  Timeouts

  Every network call needs a timeout, including connect, read and overall. Pick from measured latency (for example slightly above the 99th percentile), not a guess, and make the caller's budget smaller than its own caller's so work is abandoned in an orderly way. Propagate deadlines across calls.

  Retries

  Retry only transient failures and only idempotent operations (or use idempotency keys). Use exponential backoff with jitter (random spread) so clients do not retry in lockstep, cap the attempts, and apply a retry budget (for example no more than 10 percent extra traffic). Retry at one layer, not at every layer, or attempts multiply.

  Circuit breaker

  A breaker watches failures or slow calls. Closed: normal. After too many failures it opens: calls fail immediately (or use the fallback) so the dependency can recover and callers stop wasting resources. After a cool-down it goes half-open: a few trial calls decide whether to close again.

  100%

  Figure SDA-ADS-06-2 · Circuit breaker states · © forcsguy.com

  Bulkheads

  Partition resources so one dependency cannot consume them all: separate thread pools or connection pools per dependency, separate queues per tenant or per priority, or separate instances for critical paths. A slow recommendation service then cannot starve checkout.

  Backpressure and load shedding

  If producers outpace consumers, work piles up. Without limits the buffer grows until memory is exhausted or latency becomes huge. Options:

  • Bounded queues that reject or block when full.
  • Signalling upstream to slow down (credit-based flow in streaming systems, TCP windows, reactive streams).
  • Load shedding: deliberately drop low-priority work early (return 429 or 503 quickly) to protect the rest.
  • Rate limiting at the edge (see the rate-limiter lesson).

  Failing fast and visibly under overload is better than accepting everything and failing slowly.

  Fallbacks and graceful degradation

  Return something useful when a dependency fails: cached data, a default, a simpler algorithm, or hide a non-essential section. Decide which features are essential before an incident, and test the degraded mode.

  How they interact

  • A timeout is the minimum; without it nothing else works.
  • Retries inside a breaker inside a bulkhead is the usual nesting.
  • Everything needs metrics: breaker state, rejected calls, queue depth, timeout counts. Alert on them.
  • Test it: inject latency and failures (chaos experiments) to confirm the patterns work before production does it for you.

  Common mistakes

  Retries with no backoff; timeouts longer than the caller's; unbounded queues; a shared thread pool for everything; breakers that never close; fallbacks that call the same failing service; and testing only the happy path.

  Real-world reading

  Related reading

  • Timeouts, retries, and backoff with jitterAmazon Builders' Library · Amazon Web ServicesBlogs
  • Using load shedding to avoid overloadAmazon Builders' Library · Amazon Web ServicesBlogs
  • Circuit BreakerMartin FowlerBlogs
  • Circuit Breaker patternMicrosoft · Microsoft AzureDocumentation

  Show 6 moreShow fewer

  • Google SRE Book: Addressing Cascading FailuresGoogle SRE team · GoogleBlogs
  • Chaos Monkey (source)Netflix · GitHubGitHub
  • Bulkheads: Partition and Tolerate Failure in One PartSkife engineers · SkifeCase Studies
  • Protector: Circuit Breaker for Time Series Databases at TrivagoTrivago engineers · TrivagoCase Studies
  • Autoscaling Jenkins at TrivagoTrivago engineers · TrivagoCase Studies
  • Troubleshooting a Connection Timeout Issue with tcp_tw_recycle Enabled at eBayeBay engineers · eBayCase Studies

  Open this lesson with its quiz and exercise

• System Design Foundations · 11 minReplication and shardingTwo ways to spread data over many machines: copy it for safety and read speed, or split it for capacity.Read hereClose

  Replication: keep copies

  Replication keeps the same data on more than one machine. Reasons:

  • Availability: if one machine dies, another has the data.
  • Read capacity: reads can go to any copy.
  • Locality: place a copy near users in another region.

  The most common arrangement is leader and followers. All writes go to the leader, which sends the changes to followers. Reads can use followers.

  100%

  Figure SDA-FND-05-1 · Leader and follower replication · © forcsguy.com

  Many teams hide the split between the primary and the replicas behind a small router, so the application does not need to know:

  100%

  Figure SDA-FND-05-2 · Read and write splitting through a router · © forcsguy.com

  Synchronous or asynchronous

  • Synchronous: the leader waits for followers to confirm before saying "done". Safer, but slower, and a down follower can block writes.
  • Asynchronous: the leader confirms immediately and followers catch up. Fast, but a follower can be behind, and a leader crash can lose the most recent writes.

  Most systems use asynchronous replication for most followers, sometimes with one synchronous follower as a safety net.

  Replication lag

  Because followers can lag, a user may write something and then read an older value from a follower. Fixes include reading the user's own recent writes from the leader, or tracking how fresh each replica is. This is the first taste of a bigger idea, consistency, covered in lesson 7.

  When the leader fails

  A follower must be promoted (failover). That needs failure detection and agreement on who the new leader is, so that two machines do not both believe they are the leader (a split brain). This is why teams rely on tested tooling or managed databases for failover rather than hand-written scripts.

  Read-after-write consistency

  With asynchronous replicas, a user can save something and then not see it: the write went to the primary, the next read hit a replica that is a few seconds behind. Three common fixes:

  • Send reads that must be fresh (the user's own profile, an order just placed) to the primary.
  • For a short window after a user writes, route that user's reads to the primary.
  • Let the router check whether a replica has caught up to the user's last write; if not, read from the primary or wait.

  A database proxy in front of the cluster can do this routing so application code stays simple, at the cost of one more hop and one more component that must itself be highly available.

  Before: one database

  100%

  After: primary with read replicas

  100%

  What changed

  • Reads are spread across copies
  • A copy can be promoted if the primary fails
  • Replicas may lag slightly behind the primary

  Sharding: split the data

  Replication does not help when the data or the writes no longer fit on one machine. Sharding (partitioning) divides the data so each shard holds a part. Each shard is usually replicated too.

  100%

  Figure SDA-FND-05-3 · Sharding by key range · © forcsguy.com

  Choosing a shard key

  The shard key decides which shard owns a row. A good key:

  • Spreads data evenly, so no shard is much bigger than the others.
  • Spreads traffic evenly, so no shard is a hot spot.
  • Matches your queries, so most queries touch one shard.

  Common schemes:

  Scheme	How	Watch out for
  Range	key ranges per shard	hot spots, e.g. newest dates
  Hash	hash(key) mod N	adding shards remaps most keys
  Directory	a lookup table says where each key lives	the directory is another thing to run and scale
  Consistent hashing	keys and nodes placed on a ring	needs virtual nodes to balance well

  Consistent hashing in one picture

  Place both servers and keys on a ring. A key belongs to the first server clockwise from it. Adding a server only takes keys from its neighbour; the rest do not move.

  100%

  Figure SDA-FND-05-4 · Consistent hashing ring · © forcsguy.com

  Horizontal and vertical partitioning

  Sharding is horizontal partitioning: the rows of a table are split across machines, each shard holding the same columns for a different set of rows. Vertical partitioning splits by columns or by feature instead: rarely used or large columns (a profile photo, a long description) move to a separate table or store, or whole groups of tables go to different databases (users on one, orders on another). It is simpler than sharding and keeps frequently read data small and fast, but it does not help when a single table's rows outgrow one machine.

  100%

  Figure SDA-FND-05-5 · Splitting a table by rows and by columns · © forcsguy.com

  What sharding costs you

  • Cross-shard queries and joins become slow or need extra work.
  • Transactions across shards are hard (lesson 8 touches on how systems cope).
  • Rebalancing when a shard gets too big is operational work.
  • More moving parts to monitor.

  A worked example: sizing replicas and shards

  Replicas for reads. A social app has 20,000 reads and 1,000 writes per second. One database node can do about 5,000 simple operations per second. A primary alone is hopeless. With writes of 1,000 per second on the primary, you have capacity left for about 4,000 reads, so you need at least four replicas to serve 20,000 reads (16,000 on replicas plus 4,000 on the primary) with no headroom, so six in practice. Adding replicas scales reads almost linearly. It does nothing for the 1,000 writes, which still all go to one primary.

  Shards for writes. If writes grow to 12,000 per second, no single primary can take them. Split the data into shards, each with its own primary: with 4 shards each takes about 3,000 writes per second. Each shard then has its own replicas for reads. Your final count is shards x (1 primary + replicas), for example 4 x (1 + 2) = 12 database nodes.

  100%

  Figure SDA-FND-05-6 · Replicas scale reads, shards scale writes · © forcsguy.com

  Walking through a failover

  Replication is only useful if you know what happens when a primary dies. With asynchronous replication:

  1. The primary acknowledges writes 100 to 140, but the follower has only received up to 135 (replication lag).
  2. The primary crashes. Writes 136 to 140 exist only on the dead machine.
  3. A monitor notices missing heartbeats, usually after several seconds, and promotes the most up-to-date follower to primary.
  4. Clients are redirected. Writes 136 to 140 are lost (or recoverable only when the old primary returns).
  5. When the old primary comes back it must not accept writes as if nothing happened (that would create two primaries, "split brain"); it rejoins as a follower and discards or reconciles what the new primary does not have.

  With synchronous replication to at least one follower, step 2 loses nothing, but every write waits for that follower, adding latency and making writes fail if the follower is unavailable. Many systems choose a compromise: wait for one follower to confirm (semi-synchronous), accepting a little latency to avoid losing acknowledged writes.

  100%

  Figure SDA-FND-05-7 · What an asynchronous failover can lose · © forcsguy.com

  Choosing a shard key: three examples

  • User id for a messaging app. Almost every query is "this user's messages", so a hash of the user id keeps each user's data together and spreads users evenly. The pain: a query across all users (an admin search) must ask every shard.
  • Order date for an orders table. Range sharding by date makes "last month's orders" cheap, but all new writes land on the newest shard, a hot spot, while old shards sit idle. A better key adds a hash of the customer id.
  • Country for a multi-region product. Good for keeping data near users and meeting data-residency rules, but shards are uneven: one country may hold half your users.

  A good key has high cardinality (many distinct values), even load and matches your most common query. Changing it later means moving most of your data, so test it against a sample of real traffic first.

  Mistakes people make

  • Reading from a replica right after writing. The replica may not have the write yet, so users do not see their own change. Read your own writes from the primary for a short time, or route that user to the primary.
  • Sharding too early. Sharding multiplies operational work: backups, schema changes, cross-shard queries and transactions all get harder. First try indexes, caching, replicas and a bigger machine.
  • Counting on replicas as backups. A mistaken delete is replicated within milliseconds. Keep real backups and test restoring them.
  • Forgetting to rebalance. Shards grow unevenly. Plan how to split a shard, ideally with consistent hashing or many small virtual shards that you can move between machines.

  Try it yourself

  In the System Builder, open "The database goes down". Run it with no replicas and see reads and writes fail, then add replicas and a write queue and compare. For hashing itself, build the consistent hash ring lab and measure how many keys move when you add a node. To place replicas and a queue yourself, open the database-outage design on the free canvas and kill the database.

  More on this site

  Related reading

  • How Discord Stores Trillions of MessagesBo Ingram · DiscordBlogs
  • MongoDB: Shard KeysMongoDB · MongoDB Inc.Documentation
  • Uber Schemaless: A Scalable MySQL DatastoreUber Engineering · UberCase Studies
  • How Discord Stores Billions of MessagesDiscordCase Studies

  Show 6 moreShow fewer

  • Consistent Hashing and Random Trees (paper)Karger et al. · MITPapers
  • Amazon Dynamo: Highly Available Key-Value StoreWerner Vogels · AmazonCase Studies
  • Mitigating MySQL Replication Lag and Reducing Read Load at GithubGithub engineers · GithubCase Studies
  • MongoDB at Baidu: Multi-Tenant Cluster Storing 200+ Billion Documents across 160 ShardsMongodb engineers · MongodbCase Studies
  • Shopify EngineeringShopify engineers · ShopifyBlogs
  • Sharding MySQL at QuoraQuora engineers · QuoraCase Studies

  Open this lesson with its quiz and exercise