MongoDB – Replication and Sharding

Two different words for two different problems, and they get confused constantly. Replication is copies of the same data for durability and availability. Sharding is splitting different data across machines for capacity. You will almost certainly need the first. You very likely do not need the second.

Replica sets

A replica set is a group of mongod processes holding the same data. One is the primary and takes every write; the others are secondaries that replicate from it. If the primary becomes unreachable, the remaining members hold an election and one is promoted, usually within a few seconds.

Three members is the standard minimum, and the reason is arithmetic: an election needs a majority. With two members, losing one leaves a single node that cannot form a majority of two, so it steps down to a secondary and the set stops accepting writes. Two nodes are less available than one.

rs.status()      // members, states, election history
rs.conf()        // the configuration, including votes and priorities
rs.hello()       // which member am I talking to, and who is primary

A member can be given priority: 0 so it never becomes primary — useful for a node in another region kept for disaster recovery, or an analytics replica you do not want serving production traffic.

Configuring one

A set is defined once, with rs.initiate(), and adjusted afterwards with rs.add() and rs.reconfig():

rs.initiate({
  _id: "rs0",
  members: [
    { _id: 0, host: "mongo-a:27017", priority: 2 },   // preferred primary
    { _id: 1, host: "mongo-b:27017", priority: 1 },
    { _id: 2, host: "mongo-c:27017", priority: 0, hidden: true }   // backups only
  ]
})

priority influences elections — higher wins, and 0 means never primary. hidden additionally keeps a member out of client read preference, which is how you dedicate a node to backups or analytics without production traffic finding it.

You can watch a failover rather than waiting for one. rs.stepDown() asks the current primary to become a secondary and triggers an election:

rs.stepDown(60)     // step down, and do not seek re-election for 60 seconds
rs.status().members.map(m => ({ name: m.name, state: m.stateStr }))

Doing that once on a staging cluster, with your application running against it, is the cheapest way to find out whether your driver retries properly. Modern drivers have retryable writes on by default and handle it; application code that opens its own sessions often does not.

The oplog

The oplog is a capped collection in the local database recording every write in order. Secondaries tail it and apply the same operations, which is the whole replication mechanism.

It is worth understanding because three other features are built on it: change streams, transactions, and resumable reads after a failover. A standalone server has no oplog, which is why none of those work without a replica set — and why this track has used one since lesson 2.

rs.printReplicationInfo()      // oplog size, and how many hours it covers

That second number is the one to watch. The oplog is capped, so it holds a rolling window of history. A secondary offline longer than the window cannot catch up incrementally and needs a full resync. If your oplog covers four hours, a two-hour maintenance job on a secondary is fine and a six-hour one is a rebuild.

Reading from secondaries

By default every read goes to the primary, which is the safe choice. Read preference can send reads elsewhere:

db.reels.find().readPref("primary")             // default — always current
db.reels.find().readPref("secondaryPreferred")  // offload reads
db.reels.find().readPref("nearest")             // lowest latency

The catch is that replication is asynchronous. A secondary is behind the primary by some small amount, so a read from one can miss a write you just made. That is the source of “I saved it and it is not there” bugs, and it is not a MongoDB quirk — it is what asynchronous replication means anywhere.

Use secondary reads for work that tolerates staleness: analytics, exports, reports. Keep anything a user just wrote on the primary. Reaching for secondary reads to fix a write-throughput problem does not work either, because every write still goes to the primary regardless.

Watching replication lag

Lag is the number that tells you whether secondary reads are safe and whether a failover would lose anything. Two commands report it:

rs.printSecondaryReplicationInfo()    // per member: how far behind, in seconds

db.adminCommand({ replSetGetStatus: 1 }).members.map(m => ({
  name: m.name, state: m.stateStr, optime: m.optimeDate
}))

A healthy set on a quiet network sits at zero to a couple of seconds. Sustained lag that grows means the secondary cannot keep up — usually slower disks than the primary, or an expensive index build running on it.

Write concern is how you convert that from a monitoring problem into a guarantee. Asking for majority means a write is not acknowledged until enough members hold it, so it cannot be lost to a failover:

db.reels.insertOne(doc, { writeConcern: { w: "majority", wtimeout: 5000 } })

Always set wtimeout with majority. Without it, a write issued while too few members are reachable waits indefinitely rather than failing — and an indefinite wait inside a request handler is how one degraded node takes an application down. Note that a timeout does not roll the write back; it only stops waiting for acknowledgement.

Sharding, and whether you need it

Sharding partitions a collection across several replica sets by a shard key. A router process (mongos) sends each query to the shards that can answer it.

You need it when a single machine genuinely cannot hold your working set or absorb your write rate. That threshold is much higher than most people assume — a well-indexed replica set on modern hardware handles a very large application. Before sharding, exhaust the cheaper options: better indexes, a larger machine, archiving old data, secondary reads for reporting.

The reason for that caution is the shard key.

The shard key decides everything

The shard key determines which shard a document lives on, and it is very hard to change later — resharding exists in recent versions but is a major operation. A bad key produces one of two failures:

• A hotspot. A monotonically increasing key — a timestamp, or an ObjectId — sends every new document to the same shard. You have bought several machines and are writing to one of them.
• Scatter-gather everywhere. If your common queries do not include the shard key, every query goes to every shard and the router merges the results. That is slower than one machine would have been, and it gets slower as you add shards.

So the shard key must appear in your frequent queries and must distribute writes evenly. Those two requirements often conflict, which is the real difficulty. Hashed sharding distributes perfectly and destroys range queries; a compound key is usually the compromise.

For ReelCMS, sharding reels on creator._id would keep a creator’s reels together and route their profile page to one shard — but the public feed, which filters only on status, would hit every shard. That tension is normal, and resolving it takes knowing which query you care about most.

The short version

• Replication is copies for availability; sharding is partitions for capacity. Different problems.
• Three members minimum — elections need a majority, so two is worse than one.
• The oplog drives replication, and change streams and transactions are built on it.
• Check how many hours your oplog covers before taking a secondary offline.
• Secondary reads are stale by design. Never for read-your-own-write.
• Shard last, after indexes, hardware and archiving.
• The shard key must be in your hot queries and spread writes. Getting it wrong is expensive to undo.