MongoDB — Replica Sets and Sharding

Replication for HA + read scaling; sharding for write scaling + data volume. They compose.

Replica sets

A replica set is a group of 3+ MongoDB instances replicating the same data. One is primary (accepts writes); the rest are secondaries (apply oplog asynchronously).

primary
  |
  | oplog replication (async)
  |
  +→ secondary 1
  +→ secondary 2

Election

When the primary becomes unavailable, secondaries hold an election (Raft-like) and one is promoted. Failover takes ~10-30s normally — clients see brief unavailability, then reconnect.

Each member has a priority (default 1) and votes (default 1). Priority 0 = “eligible only if no others available”; commonly used for cross-region DR replicas.

Oplog

The operation log — a capped collection on the primary. Each write produces an oplog entry that secondaries replay. Sized roughly 5% of disk by default; tune for high write rate (need enough history that a recovering replica can catch up without doing a full resync).

// Check oplog status
rs.printReplicationInfo()

If a secondary falls so far behind it can’t catch up via oplog, it must do a full initial sync (re-copy the whole dataset).

Read preferences (recap)

db.read_preference = ReadPreference.SECONDARY_PREFERRED

Scales reads but reads from secondaries are eventually consistent. See 04_transactions_concerns.md for readConcern interactions.

Hidden / delayed members

• Hidden — secondary that doesn’t accept reads, doesn’t participate in elections (priority 0). Used for backups (run mongodump against it without affecting traffic).
• Delayed — secondary that replays oplog with intentional delay (e.g., 1 hour). Used for “oops” recovery — recover from accidental DROP TABLE within the delay window.

Sharding

Horizontal partitioning across multiple replica sets (“shards”). Each shard owns a range of data based on the shard key.

mongos (router)
  |
  +→ shard 1 (replica set)  : data range A
  +→ shard 2 (replica set)  : data range B
  +→ shard 3 (replica set)  : data range C
  |
  +← config servers (metadata)

Components:

• shard — a replica set holding part of the data.
• mongos — router that knows which shard owns each query. Clients connect to mongos, never directly to shards.
• config servers — replica set storing cluster metadata (shard ranges, chunk locations).

Shard keys — the critical decision

The shard key determines:

• Write distribution — high-cardinality, well-distributed keys spread writes.
• Query routing — queries that include the shard key are “targeted” (one shard); without, “scattered” (broadcast to all shards).
• Chunk migration — MongoDB rebalances 64MB chunks across shards to keep them balanced.

Range vs hashed sharding

Range-based (default for compound shard keys):

db.command({"shardCollection": "myapp.orders", "key": {"user_id": 1}})

Data is split into ranges of the shard key. Range queries are efficient (one shard). Monotonically increasing keys (timestamps, ObjectIds) create a hot shard — new writes all hit the last shard.

Hashed:

db.command({"shardCollection": "myapp.orders", "key": {"_id": "hashed"}})

Hash of the shard key distributes writes evenly. Range queries fan out to all shards.

Compound shard keys

db.command({
    "shardCollection": "myapp.events",
    "key": {"user_id": 1, "timestamp": 1}
})

Combines high cardinality (user_id) with range queryability (timestamp). Writes spread; per-user queries targeted; “all events in 2024” still scatters.

Bad shard key choices

• Low cardinality (status with 3 values) — only 3 chunks possible, can’t distribute.
• Monotonically increasing (created_at, default _id ObjectId) — hot last shard.
• Query keys not included — every query scatters; sharding adds latency without parallelism.
• _id alone if _id is sequential ObjectIds — see monotonic above. Hash the _id instead, or pick a domain key.

Reshardable shard keys (5.0+)

Older MongoDB: shard key is immutable. 5.0+ supports resharding — change the shard key at the cost of significant resource use during the resharding window. Massive improvement; before this, a bad shard key meant rebuilding the cluster.

When to shard

Most apps don’t need sharding. Replica set + good indexes scales to hundreds of GB and tens of thousands of ops/sec.

Shard when:

• Working set > available RAM across the cluster.
• Write throughput exceeds one primary’s capacity.
• Storage size exceeds what one replica set can hold.
• Geographic distribution (zone sharding — keep EU data in EU shards).

Don’t shard prematurely. It adds operational complexity, query latency (mongos hop), and constraints on operations (no findOneAndUpdate across shards without the shard key, etc.).

Zone sharding (geographic)

Tag shards with zones; tag chunk ranges with zones; the balancer keeps zone-tagged data on zone-tagged shards.

// Configure shards for EU users on EU shards
sh.addShardTag("eu-shard", "EU")
sh.addTagRange("myapp.users", {country: "DE"}, {country: "FR"}, "EU")

Use for: GDPR (EU data stays in EU), latency (US users on US shards).

Operational concerns

Balancer

Background process on the config servers. Moves 64MB chunks between shards to keep them balanced. Can run continuously or be scheduled (e.g., off-hours).

sh.disableBalancing("myapp.orders")   // pause for big bulk loads
sh.enableBalancing("myapp.orders")    // resume

Backup

• mongodump on a hidden secondary — for replica set without sharding.
• Cloud Manager / Atlas — managed backup with point-in-time recovery.
• Sharded backups — much harder; consistent backup across shards requires coordinated snapshots. Atlas handles this; DIY is painful.

Monitoring

• db.serverStatus() — connection counts, opcounters, network.
• rs.status() — replica set health, lag per member.
• sh.status() — sharded cluster overview.
• db.collection.stats() — collection size, indexes, sharding distribution.
• Atlas / Cloud Manager / Datadog MongoDB integration for production.

Common gotchas

• Initial sync triggered by oplog catch-up failure. Replica that’s been down for too long → must full-resync (hours to days for big DBs). Size oplog generously.
• mongos has no state. It caches cluster metadata; can be restarted freely. Don’t put data on mongos boxes.
• Cross-shard transactions are slow. Coordinated commit across shards. Restructure to single-shard if possible.
• Aggregation across all shards — $lookup to an unsharded collection works; to a sharded one is limited and expensive.
• countDocuments() vs estimatedDocumentCount() — first scans (accurate, slow); second uses metadata (fast, may be stale). Stale count is fine for “approximately how many users do we have?”

Atlas vs self-hosted

MongoDB Atlas (managed) handles:

• Replica set + sharding management.
• Backups, point-in-time recovery.
• Monitoring, alerting.
• Major version upgrades.
• TLS, IP allowlists, VPC peering, private endpoints.
• Atlas Search (managed Lucene-based search) and Atlas Vector Search.

For most teams: Atlas is the right call. Self-hosted MongoDB is non-trivial to operate (initial sync mechanics, backup coordination, version upgrades).

Interview angle

• “Why does shard key choice matter so much?” — determines write distribution (high cardinality = spread; low cardinality / monotonic = hot shard), query routing (targeted vs scattered), and chunk migration efficiency. Bad shard key means writes pile on one shard while others idle.
• “What’s wrong with sharding on a monotonically-increasing key like _id ObjectId?” — new writes all hit the shard owning the highest range. Hot shard. Solution: hashed sharding on _id, or pick a domain key (user_id) that distributes naturally.
• “Range vs hashed sharding?” — range: efficient range queries on the shard key but hot-shard risk on monotonic keys. Hashed: even write distribution, but range queries scatter to all shards. Compound shard keys can blend (user_id + timestamp).
• “When do you actually need sharding?” — when a replica set can’t hold the data or sustain the write rate. Most apps don’t need it; sharding adds operational complexity and query latency. Vertically scale + good indexes first.
• “Replica set election — what happens during failover?” — primary unreachable for ~10s, secondaries detect, hold election (Raft-like), promote highest-priority caught-up secondary. Clients reconnect via the driver’s retry logic. Total disruption ~10-30s.
• “Oplog size — why does it matter?” — capped collection of recent writes. If a secondary falls behind by more than the oplog’s history, it must full-resync (slow). Size oplog generously, especially on write-heavy systems.
• “What does Atlas give you over self-hosted?” — managed replica sets, automated backups + PITR, monitoring, major version upgrades, security perimeter (TLS / VPC), managed search and vector search. Operationally, MongoDB is significantly more complex to self-host than Postgres; Atlas removes most of that pain.