Move shards online: Stripe
Stripe moved hundreds of millions of objects without downtime, then built a Data Movement Platform that copies chunks between MongoDB shards while the app keeps writing.
In 2017 Stripe migrated hundreds of millions of Subscriptions in four phases, with dual writes and a MapReduce backfill, and kept serving throughout. Later it built DocDB on MongoDB, with a Data Movement Platform that moves chunks between shards online.
The system#
DocDB is a self-managed MongoDB fleet. The Data Movement Platform reads each shard's oplog, sends it to Kafka, archives it to object storage, and replays mutations on the target shard. The cutover is kept shorter than a planned primary failover, because that is the retry budget the product applications already have. Stripe reports 99.999% uptime across the migrations.
The board#
Drag to pan. Hold Ctrl or Cmd and scroll to zoom.
| Where | Setting | Value | Why |
|---|---|---|---|
| App Server | Instance size | large | Large, because the app keeps writing while a shard moves. |
| App Server | Copies | 3 | Three copies stand for the API tier. |
| Database | Shards | 4 | DocDB splits its data across shards. Four is the board’s widest, and it stands for the fleet. |
| Database | Shard key | hash | A hash spreads objects across shards, so no shard owns one big customer. |
| Database | Instance size | medium | Medium for each shard, so a busy shard shows. |
| Event Stream | Partitions | 12 | Many partitions let the Warehouse read each shard’s changes in order. |
| Event Stream | Acks | all | A change is not acknowledged until every replica has it. |
| Event Stream | Keyed by entity | On | One object’s changes stay in order. |
| Event Stream | Retention | 3600 | An hour of retention lets a Warehouse replay its work. |
Research this Design needs: Change data capture, Event streaming, Data warehousing, Sharding, Change data capture, Event streaming, Background jobs.
The Database has four shards on a hash key, standing for DocDB's shards. The CDC Connector is the oplog reader, the Event Stream is Kafka, and the Data Warehouse stands in for the target shard that replays the changes. The app keeps writing to the old shard while the change stream catches the new one up. The chunk copy and the dual-write cutover are outside the board.
Why it works#
A shard move is a copy plus a change stream, not a dump and restore. The app writes once, and the change stream carries the same write to the target. The cutover only has to cover the last few seconds, so the read-only window fits the retry budget. Online migrations and change data capture are the two mechanisms.
Where the game stops#
The game's Database is not MongoDB or DocDB, and the board has no chunk copy or dual-write cutover. It shows the sharded store and the change stream, which are the parts a Design can carry. The caption is Quorum's result, not Stripe's uptime.
Sources#
How Stripe's document databases supported 99.999% uptime with zero-downtime data migrations, Stripe: DocDB, the Data Movement Platform, the oplog to Kafka and S3, and the failover-sized cutover.
Online migrations at scale, Stripe: the four-phase Subscriptions migration and the MapReduce backfill.
Sharding, MongoDB documentation: the shard, chunk, and move operations DocDB builds on.