🌐 Distributed Systems Interview Questions

1. Primary Node

• Receives all write operations
• Records all changes to oplog (operations log)
• By default, handles read operations
• Only one primary per replica set

2. Secondary Nodes

• Replicate data from primary's oplog
• Can serve read operations (with read preference)
• Can become primary during election
• Typically have 2+ secondaries for redundancy

3. Arbiter (Optional)

• Participates in elections but holds NO data
• Used for odd number of voting members (avoid ties)
• Lightweight, minimal resources
• Cannot become primary

Step 1: Failure Detection (10 seconds default)

• Secondary nodes send heartbeats every 2 seconds
• If primary doesn't respond within 10 seconds → failure detected
• Secondaries initiate election process

Step 2: Election Initiation

• Eligible secondary calls for an election
• Must have priority > 0 and not be hidden/delayed
• Must be up-to-date (within 10 seconds of most recent oplog entry)

Step 3: Voting

• Each voting member gets one vote
• Requires majority (n/2 + 1) to win
• Higher priority nodes more likely to win
• More up-to-date nodes preferred

Step 4: New Primary Elected

• Winning secondary becomes primary
• Starts accepting writes immediately
• Other secondaries start replicating from new primary
• Typical election time: 12-30 seconds

Majority Requirement

Why majority matters: Prevents split-brain scenarios

• 3-member set: needs 2 votes (can tolerate 1 failure)
• 5-member set: needs 3 votes (can tolerate 2 failures)
• 7-member set: needs 4 votes (can tolerate 3 failures)

Write Concern Levels:

• w: 0 - No acknowledgment (fire and forget)
• w: 1 - Acknowledged by primary only (default)
• w: "majority" - Acknowledged by majority of replica set
• w: N - Acknowledged by N nodes

Read Preference Modes:

• primary - All reads from primary (default, strongest consistency)
• primaryPreferred - Primary if available, else secondary
• secondary - All reads from secondaries (reduces primary load)
• secondaryPreferred - Secondary if available, else primary
• nearest - Lowest network latency (any node)

1. Mongos (Query Router)

• Application connects to mongos (not directly to shards)
• Routes queries to appropriate shard(s)
• Merges results from multiple shards
• Stateless - can run multiple instances

2. Config Servers

• Store cluster metadata (which chunks are on which shards)
• Must be deployed as replica set (3 members minimum)
• Critical - if unavailable, sharding stops working
• Small data volume but high importance

3. Shards

• Each shard is a replica set holding subset of data
• Minimum 2 shards required for sharded cluster
• Each shard independent - can scale independently
• Should be geographically distributed for disaster recovery

Chunks

Data divided into contiguous ranges called "chunks" (default 128MB)

• Each chunk contains documents with shard key in specific range
• Example: userId 1-10000 in chunk 1, 10001-20000 in chunk 2
• Chunks automatically split when they exceed size limit
• Balancer migrates chunks to maintain even distribution

Targeted Query (Best Case)

Query includes shard key → mongos routes to specific shard

db.users.find({ userId: 12345 })  // Goes to 1 shard only

Broadcast Query (Worst Case)

Query doesn't include shard key → mongos queries all shards

db.users.find({ email: "[email protected]" })  // Queries all shards

1. High Cardinality

Many unique values allow data to be distributed across many chunks

• Good: userId (millions of unique values)
• Bad: country (only ~200 values) - creates large undivisible chunks

2. Even Distribution

Values should distribute writes evenly across shards (avoid hotspots)

• Good: hashed userId
• Bad: timestamp (all new writes go to same shard holding latest range)

3. Query Isolation

Shard key should appear in most queries for targeted routing

• Good: userId for user-centric queries
• Bad: Random field not used in queries (forces broadcast)

• C (Consistency): All nodes see the same data at the same time
• A (Availability): Every request gets a response (success or failure)
• P (Partition Tolerance): System continues despite network failures

Default Behavior (CP)

MongoDB prioritizes Consistency and Partition tolerance:

• During network partition, minority side becomes read-only
• Primary steps down if it can't reach majority
• Sacrifices availability to maintain consistency

Strong Consistency Configuration

YES - MongoDB can provide strong consistency:

// Strong consistency: Read your own writes guaranteed
db.collection.insertOne(
  { data: "important" },
  { writeConcern: { w: "majority" } }  // Wait for majority ack
)

db.collection.findOne(
  { data: "important" },
  { readPreference: "primary" }  // Read from primary only
)

Result: If write succeeds, subsequent read WILL see it

Eventual Consistency Configuration

Can tune toward availability (AP):

// Eventual consistency: Prioritize availability
db.collection.insertOne(
  { data: "log_entry" },
  { writeConcern: { w: 1 } }  // Primary only
)

db.collection.findOne(
  { data: "log_entry" },
  { readPreference: "secondaryPreferred" }  // Can read from secondary
)

Result: May read stale data, but always available

Key Characteristics:

• Special capped collection: local.oplog.rs
• Fixed size (configurable, default ~5% of disk)
• FIFO - oldest entries automatically removed when full
• Idempotent operations - can be applied multiple times safely

Why Oplog Size Matters:

• Too Small: Secondary might not catch up before old entries are overwritten
• Replication Window: Time it takes to fill the oplog
• Recovery Time: Determines how long a secondary can be offline

When Rollback Happens:

• Primary gets isolated (can't reach majority)
• Primary continues accepting writes (from clients before timeout)
• New primary elected in majority partition
• Old primary rejoins with divergent operations

Rollback Process:

1. Old primary detects it's behind new primary
2. Finds common point in oplog
3. Saves rolled-back operations to rollback files
4. Reverts to common point
5. Syncs from new primary
6. Becomes secondary in current state

1. Read from Secondaries

Distribute read load across replica set members

// Scale reads across secondaries
db.users.find({ city: "NYC" })
  .readPref("secondary")

// Distribute to nearest node
db.users.find({ status: "active" })
  .readPref("nearest")

Benefits:
✓ Offloads read traffic from primary
✓ Reduces primary CPU/memory pressure
✓ Can handle 5x-10x more reads

Trade-off:
✗ Eventual consistency (may read stale data)
✗ Replication lag (typically 0-500ms)

2. Add More Secondaries

Horizontal scaling of read capacity

3-node replica set:  1 primary + 2 secondaries
                     → 2x read capacity

7-node replica set:  1 primary + 6 secondaries  
                     → 6x read capacity

Note: Up to 7 voting members max
      Can have up to 50 total members (43 non-voting)

3. Create Read-Only Secondaries

Dedicated secondaries for analytics/reporting

// Configure priority 0 secondary (won't become primary)
rs.reconfig({
  members: [
    { _id: 0, host: "primary:27017", priority: 2 },
    { _id: 1, host: "secondary1:27017", priority: 1 },
    { _id: 2, host: "analytics:27017", priority: 0, hidden: true }
  ]
})

// Analytics queries go here
// Doesn't affect production traffic

4. Shard the Collection

Distribute data across multiple shards for massive scale

No sharding: 1 replica set = 6 secondaries = 6x read capacity

Sharding:    4 shards × 6 secondaries each = 24x read capacity
             
Each shard handles subset of data
Queries can be parallelized across shards

5. Proper Indexing

Most impactful optimization - essential before scaling

// Without index: Collection scan (slow)
db.users.find({ email: "user@example.com" })  // 1000ms

// With index: Index scan (fast)
db.users.createIndex({ email: 1 })
db.users.find({ email: "user@example.com" })  // 5ms

// 200x faster!
// Always optimize queries before adding hardware

Write Scaling Strategies:

1. Vertical Scaling (Short-term)

Upgrade primary to more powerful hardware

• More CPU cores for parallel processing
• More RAM for larger working set
• Faster SSDs for disk I/O
• Limitation: Eventually hit hardware ceiling

2. Optimize Write Operations

// Batch inserts (much faster than individual)
db.logs.insertMany([...1000 documents...])  // 1 network round trip

// vs individual inserts
for (doc of docs) {
  db.logs.insertOne(doc)  // 1000 network round trips!
}

// Use bulk operations
var bulk = db.logs.initializeUnorderedBulkOp()
bulk.insert({ ... })
bulk.insert({ ... })
bulk.execute()  // Single batch

// Reduce index overhead
// Each index adds write cost
// Keep only necessary indexes

3. Lower Write Concern (Trade-off)

// w:1 - Acknowledge from primary only (fastest)
db.logs.insertOne(
  { event: "user_click" },
  { writeConcern: { w: 1 } }
)

// w:"majority" - Wait for majority (slower but safer)
db.orders.insertOne(
  { total: 1000 },
  { writeConcern: { w: "majority" } }
)

// Trade-off: speed vs durability

4. Sharding - The Real Solution

The ONLY way to truly scale writes horizontally

// Without sharding: 1 primary = 10K writes/sec max

// With 4 shards: 4 primaries = 40K writes/sec
Shard 1 Primary: 10K writes/sec (userId 1-250K)
Shard 2 Primary: 10K writes/sec (userId 250K-500K)
Shard 3 Primary: 10K writes/sec (userId 500K-750K)
Shard 4 Primary: 10K writes/sec (userId 750K-1M)

Each shard handles subset of writes based on shard key

5. Application-Level Partitioning

Multiple independent MongoDB clusters

// Route writes to different clusters by geography
if (user.region === "US") {
  writeToCluster("us-mongo-cluster")
} else if (user.region === "EU") {
  writeToCluster("eu-mongo-cluster")
}

// Or by tenant
writeToCluster(`tenant-${tenantId}-cluster`)

How the Balancer Works:

1. Balancer runs on primary config server
2. Checks chunk distribution every 60 seconds
3. If imbalance detected (threshold: 8+ chunk difference)
4. Selects chunks to migrate
5. Coordinates chunk migration between shards
6. Updates metadata in config servers

Chunk Migration Process:

Step 1: Balancer identifies imbalance
   Shard A: 50 chunks
   Shard B: 30 chunks
   Difference: 20 > threshold (8)

Step 2: Select chunk to migrate from Shard A

Step 3: Copy chunk data to Shard B
   - Data copied in background
   - Both shards can serve reads/writes during copy

Step 4: Brief lock for final sync
   - Transfer any writes that occurred during copy
   - Takes milliseconds

Step 5: Update metadata
   - Config servers update: chunk now on Shard B
   - mongos starts routing queries to Shard B

Step 6: Delete old data from Shard A

Balancer Configuration:

// Check balancer status
sh.getBalancerState()

// Enable/disable balancer
sh.startBalancer()
sh.stopBalancer()

// Schedule balancer window (e.g., only run at night)
db.settings.update(
  { _id: "balancer" },
  {
    $set: {
      activeWindow: {
        start: "23:00",  // 11 PM
        stop: "06:00"    // 6 AM
      }
    }
  },
  { upsert: true }
)

// Disable balancing for specific collection during migration
sh.disableBalancing("mydb.users")
sh.enableBalancing("mydb.users")

Timeline of Events:

t=0: DC1 Fails

• Primary and Secondary3 become unreachable
• Remaining nodes: Secondary1, Secondary2, Secondary4
• 3 nodes available out of 5 = majority ✓

t=10s: Failure Detected

• Heartbeat timeout (default 10 seconds)
• Surviving secondaries detect primary is down
• Election process initiated

t=12s: New Primary Elected

• Secondary1 or Secondary4 becomes primary (higher priority wins)
• Let's say Secondary1 in DC2 becomes new primary
• Total downtime: ~12 seconds

t=12s+: System Operational

• ✓ Writes accepted by new primary in DC2
• ✓ Reads served by all 3 surviving members
• ✓ Replication continues DC2 → DC3
• ✗ Lost 2 nodes but still have majority
• ⚠️ Now vulnerable (losing 1 more node = no majority)

When DC1 Recovers:

1. Nodes in DC1 come back online
2. They discover they're behind new primary
3. Automatic sync from current primary
4. May involve rollback if they had un-replicated writes
5. Rejoin replica set as secondaries
6. System back to full 5-node redundancy

How MongoDB Prevents Split-Brain:

1. Majority Requirement: Primary must see majority of nodes
2. Heartbeat Monitoring: Primary checks it can reach majority
3. Step Down: If primary loses majority, it IMMEDIATELY steps down
4. No Writes: Stepped-down node becomes read-only
5. Single Primary: Only partition with majority can elect primary

Step-by-Step Prevention:

5-node replica set: Primary + 4 Secondaries

Network partition splits them:
- Partition A: 2 nodes
- Partition B: 3 nodes

Partition A (has old primary):
1. Primary detects it can only reach 1 other node (2 total)
2. 2 < 3 (majority of 5) → NO MAJORITY
3. Primary STEPS DOWN immediately
4. Becomes SECONDARY (read-only)
5. Cannot elect new primary (need 3 votes, only have 2)

Partition B:
1. Detects primary is gone
2. Has 3 nodes = MAJORITY ✓
3. Initiates election
4. Elects NEW primary
5. Accepts writes

Result: Only partition B has writable primary
        Partition A is read-only
        No conflicting writes = No split-brain!

Scale Characteristics:

• 100M users, ~10M DAU (daily active)
• Heavy read workload (90% reads, 10% writes)
• User-generated content: posts, comments, likes
• Social graph: followers/following
• Feed generation (personalized, real-time)
• Global user base (latency-sensitive)

Collection Design:

Sharding Strategy:

Users Collection:

sh.shardCollection("social.users", { userId: "hashed" })

Why hashed?
✓ Even distribution of users across shards
✓ Prevents hotspots from sequential user IDs
✓ Queries by userId are targeted (single shard)

Replica Set Configuration:
- 3 nodes per shard (1 Primary + 2 Secondaries)
- Geographic distribution across 3 datacenters
- Read preference: primaryPreferred for user profiles

Posts Collection:

sh.shardCollection("social.posts", { userId: 1, timestamp: 1 })

Why compound key?
✓ userId ensures user's posts stay together (feed queries)
✓ timestamp prevents hotspots on recent posts
✓ Range queries on timestamp within user work well

Replica Set Configuration:
- 5 nodes per shard (1 Primary + 4 Secondaries)
- More secondaries for high read load (feed generation)
- Read preference: secondary for feed reads
- Write concern: w:1 for posts (eventually consistent OK)

Read Optimization:

• Caching Layer: Redis for hot user data, feed cache (TTL 5 min)
• Read from Secondaries: Offload feed generation to secondaries
• Materialized Views: Pre-compute popular user feeds
• CDN: Media files served from CDN, not MongoDB

Write Optimization:

• Write Concern w:1: Fast acknowledgment for posts/likes
• Bulk Operations: Batch notifications, likes
• Async Processing: Feed updates via message queue

High Availability:

Geographic Distribution:
- Primary DC: 40% of nodes
- Secondary DC 1: 30% of nodes  
- Secondary DC 2: 30% of nodes

Failure Handling:
✓ Single shard failure: Automatic failover (12s downtime)
✓ Datacenter failure: Majority survives, operations continue
✓ Config server: 3-node replica set for metadata
✓ Mongos: Stateless routers, run multiple instances

Monitoring & Alerts:

• Replication lag > 1 second
• Chunk migration activity
• Shard imbalance > 10 chunks
• Write hotspots (single shard > 50% writes)
• Connection pool exhaustion