Scaling MongoDB at Mailbox

Why Mailbox split its hot MongoDB collection

Mailbox’s rapid growth surfaced a painful bottleneck: MongoDB’s database-level write lock. The backend waited on that lock often enough to add visible latency for users. Sharding the cluster was one option, but the team preferred to isolate the email-related collection — among the most frequently updated — onto its own cluster. That move promised at least a 50% reduction in write-lock contention, and it opened the door to tuning and administering the two data workloads separately.

The migration path wasn’t obvious. MongoDB’s cloneCollection command cannot run through a mongos, so it’s unusable for a sharded collection. renameCollection is equally off-limits for sharded data. Other workarounds either impaired performance for live users or broke at Mailbox’s scale. The result was a pair of custom Python tools for copying and verifying collections, later open-sourced as Hydra.

Building a fast snapshot copier

The first naive implementation — essentially a loop over find() results — proved far too slow. Micro-optimizations like adjusting batch sizes helped only marginally. A quick benchmark with a C++ implementation using the MongoDB C++ SDK showed that for simple cases, C++ outperformed Python by 5–10x. But the team wanted code that Python-proficient engineers could easily adapt, so the search continued within Python itself.

Initial attempts with worker threads and worker processes were disappointing. The process pool, in particular, lost its potential gains to the overhead of IPC through Python Queue objects and pipes. The breakthrough came with gevent, which patches standard Python modules like socket to run asynchronously. With gevent, the copy code reads sequentially, free of callbacks:

import gevent
gevent.monkey.patch_all()
 
def copy_documents(source_collection, destination_collection, _ids):
    """
    Given a list of _id's (MongoDB's unique identifier field for each document),
    copies the corresponding documents from the source collection to the destination
    collection
    """
 
    # copies each document using a separate greenlet; optimizations are certainly
    # possible but omitted in this example
    for _id in _ids:
        gevent.spawn(copy_document, source_collection, destination_collection, _id)
 
def copy_document(source_collection, destination_collection, _id):
    """
    Copies document corresponding to the given _id from the source to the
    destination.
    """
    # both of the following function calls block without gevent; with gevent they
    # simply cede control to another greenlet while waiting for Mongo to respond
    source_doc = source_collection.find_one({'$id': _id})
    destination_collection.insert(source_doc) # another MongoDB operation

The greenlets in copy_documents delegate each document’s copy to copy_document(). Whenever a greenlet blocks on a MongoDB request, it yields to other ready greenlets — but because all greenlets share one thread and process, there’s no need for inter-greenlet locking. The performance difference was dramatic:

Combining gevent with worker processes — one per shard — gave a linear performance increase, provided IPC overhead was minimized. A single gevent process could copy a full collection in just under half the time of mongodump, despite mongodump being written in C++. The catch was that mongodump queries synchronously, single-process and single-threaded.

Keeping the copy current with the oplog

MongoDB’s lack of transactions means a large find() against a hot collection returns a result set spanning different points in time. To minimize downtime at cutover, the team needed to stream changes from the source cluster to the destination with minimal latency.

MongoDB’s oplog is the natural mechanism: it records every insert, update, and remove on a mongod for replication to secondaries. Given a snapshot, replaying oplog entries since the snapshot’s start brings the destination up to date. Inserts and removes serialize straightforwardly; updates did not. In MongoDB 2.2.x, update oplog entries contain duplicate keys that neither the Mongo shell nor most drivers can display. The workaround: use the _id embedded in the update to trigger a fresh copy of that document from the source. This doesn’t apply the exact update, but it guarantees the destination is at least as recent as the op received. Intermediate versions (like a v2) may be skipped, but source and destination end up eventually consistent:

Replaying ops serially per shard was too slow to keep pace with Mailbox’s traffic. Applying ops concurrently introduced a correctness risk: two ops on the same _id must execute in order. The fix was a Python set tracking _ids with in-progress operations. When an update arrives for an _id already being processed, that update — and everything queued behind it — waits until the older operation finishes:

Verifying data while both sides move

Comparing copied data against the source is easy when both are static; it’s harder when both are receiving writes. A first approach logged warnings for discrepancies in recently updated documents, then re-verified those later. That fails for deleted documents, which leave no timestamp to check.

The team leaned on eventual consistency: given enough retries, the source and copy converge. The comparison tool was extended with retries on an increasing backoff. Certain pathological patterns — such as data oscillating between values — could evade matching, but the migrated dataset had none. Before the production cutover, a command-line option was added to compare the documents touched by the most recent N ops. Running that over hundreds of thousands of ops per shard takes only minutes and provides strong assurance that pending changes were applied correctly during the brief downtime.

Recovering from the unexpected

Final test runs leading to production migration surfaced a steady stream of issues: sporadic network failures, one document set that consistently made mongos drop the connection, and occasional resets from mongod. Rather than enumerate every possible failure mode, the team shifted to fast recovery. The comparison tool logged the _ids of any inconsistent documents, and a dedicated tool re-copied exactly those documents from the source.

Executing the Migration

With the tooling validated, we moved into production. copy_collection.py began by taking an initial snapshot of hundreds of millions of emails, then proceeded to replay more than 100 million MongoDB operations. The complete process—initial snapshot, index builds, and replication catch-up—finished in roughly nine hours. That was comfortably inside our 24-hour target.

For an extra margin of safety, I let copy_collection.py continue replaying operations from the source cluster's oplogs for another full day. During that window, I ran compare_collections.py three separate times to verify that all data had been copied correctly.

The final cutover to the new cluster was recent and remarkably brief. In a short maintenance window, we ran compare_collections.py against the documents touched by the last 500,000 operations in each shard. With no inconsistencies found in that recent data, we executed smoke tests, repointed the Mailbox backend at the new cluster, and restored public service. Users have reported no issues stemming from the switch—the outcome we consider the true mark of success for a backend migration.

Our monitoring told a more dramatic story than the user experience did:

Releasing Hydra

We are open-sourcing Hydra, the full suite of tools we built to handle this live MongoDB collection migration. The code is available for anyone facing the challenge of re-partitioning MongoDB data without taking their service offline.