Replication

# Replica set architecture — connecting and inspecting member roles

from pymongo import MongoClient

# Connect to a replica set — provide all members in the connection string
# MongoDB automatically discovers all members from any one of them
client = MongoClient(
    "mongodb://localhost:27017,localhost:27018,localhost:27019/",
    replicaSet="rs0"    # replica set name must match the configured name
)

db = client["dataplexa"]

# Check which member is currently the primary
hello = db.command("hello")   # modern replacement for isMaster
print("Replica set status from hello command:")
print(f"  isWritablePrimary: {hello.get('isWritablePrimary', False)}")
print(f"  primary:           {hello.get('primary', 'unknown')}")
print(f"  hosts:             {hello.get('hosts', [])}")
print(f"  setName:           {hello.get('setName', 'unknown')}")
print(f"  me:                {hello.get('me', 'unknown')}")

# Full replica set status — all member details
rs_status = client.admin.command("replSetGetStatus")
print(f"\nReplica set: {rs_status['set']}")
print(f"Members ({len(rs_status['members'])}):")
for member in rs_status["members"]:
    state_str = member.get("stateStr", "UNKNOWN")
    name      = member.get("name", "")
    health    = member.get("health", 0)
    lag_secs  = member.get("optimeDate", None)
    print(f"  {name:30}  state: {state_str:10}  health: {health}")

# Replica set roles
print("\nReplica set member roles:")
roles = [
    ("Primary",   "Accepts all writes. One at a time — elected by majority vote"),
    ("Secondary", "Replicates from primary oplog. Can serve reads if configured"),
    ("Arbiter",   "Votes in elections but holds no data — lightweight tie-breaker"),
    ("Hidden",    "Replicates data but invisible to clients — for backups/reporting"),
    ("Delayed",   "Replicates with a time delay — protection against accidental deletes"),
]
for role, desc in roles:
    print(f"  {role:10}  {desc}")

# Reading the oplog — understanding what replication tracks

from pymongo import MongoClient, DESCENDING

client = MongoClient(
    "mongodb://localhost:27017,localhost:27018,localhost:27019/",
    replicaSet="rs0"
)

local_db = client["local"]   # oplog lives in the local database

# Check oplog size and current usage
oplog_stats = local_db.command("collStats", "oplog.rs")
oplog_size_mb  = oplog_stats.get("maxSize",  0) / (1024 * 1024)
oplog_used_mb  = oplog_stats.get("size",     0) / (1024 * 1024)
oplog_count    = oplog_stats.get("count",    0)
print(f"Oplog statistics:")
print(f"  Max size:   {oplog_size_mb:.0f} MB")
print(f"  Used:       {oplog_used_mb:.1f} MB")
print(f"  Operations: {oplog_count:,}")

# Read the most recent oplog entries
print("\nMost recent oplog entries:")
entries = local_db["oplog.rs"].find(
    {},
    {"op": 1, "ns": 1, "o": 1, "ts": 1, "wall": 1}
).sort("$natural", DESCENDING).limit(5)

op_names = {
    "i": "INSERT",
    "u": "UPDATE",
    "d": "DELETE",
    "c": "COMMAND",
    "n": "NOOP"
}
for entry in entries:
    op   = op_names.get(entry.get("op", "?"), entry.get("op", "?"))
    ns   = entry.get("ns", "")
    wall = entry.get("wall", "")
    print(f"  {op:8}  {ns:35}  {str(wall)[:19]}")

# Replication lag — how far behind each secondary is
print("\nReplication lag per member:")
rs_status = client.admin.command("replSetGetStatus")
primary_optime = None
for m in rs_status["members"]:
    if m.get("stateStr") == "PRIMARY":
        primary_optime = m.get("optimeDate")
        break

for member in rs_status["members"]:
    state     = member.get("stateStr", "")
    name      = member.get("name", "")
    optime    = member.get("optimeDate")
    if state == "SECONDARY" and primary_optime and optime:
        lag_secs = (primary_optime - optime).total_seconds()
        print(f"  {name:30}  lag: {lag_secs:.1f} seconds")
    else:
        print(f"  {name:30}  {state}")

# Elections and failover — understanding the process and handling it in code

from pymongo import MongoClient
from pymongo.errors import (
    NotPrimaryError,
    ConnectionFailure,
    ServerSelectionTimeoutError
)
import time

client = MongoClient(
    "mongodb://localhost:27017,localhost:27018,localhost:27019/",
    replicaSet="rs0",
    # How long the driver waits for a primary before raising an error
    serverSelectionTimeoutMS=5000,
    # Retry writes automatically once after a network error or failover
    retryWrites=True
)

db = client["dataplexa"]

def write_with_failover_handling(document: dict, max_attempts: int = 3) -> bool:
    """
    Write a document with proper failover handling.
    retryWrites=True handles most cases automatically —
    this shows explicit handling for remaining edge cases.
    """
    for attempt in range(1, max_attempts + 1):
        try:
            db.orders.insert_one(document)
            return True

        except NotPrimaryError:
            # Current connection target is no longer the primary
            # Driver will automatically discover the new primary
            print(f"  Attempt {attempt}: not primary — waiting for election...")
            time.sleep(2)

        except ServerSelectionTimeoutError:
            # No primary available within serverSelectionTimeoutMS
            print(f"  Attempt {attempt}: no primary available yet — retrying...")
            time.sleep(3)

        except ConnectionFailure as e:
            print(f"  Attempt {attempt}: connection failure — {e}")
            time.sleep(2)

    return False

# Election timeline explanation
print("Automatic failover timeline:\n")
timeline = [
    ("0s",    "Primary stops responding (crash / network / maintenance)"),
    ("0–2s",  "Secondaries detect heartbeat timeout (default: 10s)"),
    ("2–10s", "Secondaries declare primary unreachable after missed heartbeats"),
    ("10s",   "Election triggered — candidates request votes from all members"),
    ("10–30s","New primary elected by majority vote — begins accepting writes"),
    ("30s+",  "Old primary (if it recovers) steps down and rejoins as secondary"),
]
for t, event in timeline:
    print(f"  {t:8}  {event}")

# retryWrites behaviour
print("\nretryWrites=True (default in Atlas connection strings):")
print("  ✓ Automatically retries once after a network error or primary failover")
print("  ✓ Safe for idempotent operations: insert, update, replace, delete")
print("  ✗ Does NOT retry multi-document transactions automatically")
print("  ✗ Does NOT retry non-idempotent operations like $inc mid-transaction")

# Read preferences — routing reads to secondaries

from pymongo import MongoClient, ReadPreference

client = MongoClient(
    "mongodb://localhost:27017,localhost:27018,localhost:27019/",
    replicaSet="rs0"
)

db = client["dataplexa"]

# READ PREFERENCE MODES
print("Read preference modes:\n")
modes = {
    "primary":            "Always read from primary (default). Strongly consistent.",
    "primaryPreferred":   "Read from primary if available, fall back to secondary.",
    "secondary":          "Always read from a secondary. May return stale data.",
    "secondaryPreferred": "Read from secondary if available, fall back to primary.",
    "nearest":            "Read from the member with lowest network latency.",
}
for mode, desc in modes.items():
    print(f"  {mode:22}  {desc}")

# Apply read preference at the collection level
# Primary — for financial data that must be current
orders_primary = db.get_collection(
    "orders",
    read_preference=ReadPreference.PRIMARY
)

# Secondary — for reporting queries (can tolerate slight staleness)
orders_secondary = db.get_collection(
    "orders",
    read_preference=ReadPreference.SECONDARY
)

# secondaryPreferred — for product catalogue (high read volume, tolerate lag)
products_sec_pref = db.get_collection(
    "products",
    read_preference=ReadPreference.SECONDARY_PREFERRED
)

# Nearest — for globally distributed apps (lowest latency regardless of role)
reviews_nearest = db.get_collection(
    "reviews",
    read_preference=ReadPreference.NEAREST
)

# Demonstrate reading from secondary with maxStalenessSeconds
# Reject secondaries that are more than 90 seconds behind the primary
from pymongo.read_preferences import Secondary

orders_bounded = db.get_collection(
    "orders",
    read_preference=Secondary(max_staleness=90)   # seconds
)

print("\nRead preference applied per collection:")
print(f"  orders (financial):  PRIMARY           — always current")
print(f"  orders (reporting):  SECONDARY         — tolerate slight lag")
print(f"  products (catalogue):SECONDARY_PREFERRED — high read volume")
print(f"  reviews (global):    NEAREST           — lowest latency")
print(f"  orders (bounded):    SECONDARY ≤90s lag — staleness bounded")

# Run a reporting query on secondary — total revenue by month
results = list(orders_secondary.aggregate([
    {"$group": {
        "_id":     {"$month": "$date"},
        "revenue": {"$sum": "$total"},
        "count":   {"$sum": 1}
    }},
    {"$sort": {"_id": 1}},
    {"$project": {"month": "$_id", "revenue": {"$round": ["$revenue",2]},
                  "count": 1, "_id": 0}}
]))
print("\nMonthly revenue (read from secondary):")
for r in results:
    print(f"  Month {r['month']:2}  orders: {r['count']}  revenue: ${r['revenue']:.2f}")

# Write concern — controlling replication acknowledgement level

from pymongo import MongoClient
from pymongo.write_concern import WriteConcern

client = MongoClient(
    "mongodb://localhost:27017,localhost:27018,localhost:27019/",
    replicaSet="rs0"
)

db = client["dataplexa"]

# Write concern levels
print("Write concern options:\n")
wc_levels = [
    ("w=0",          "Fire and forget — no acknowledgement. Fastest, no durability."),
    ("w=1",          "Primary acknowledges (default). Fast, survives primary restart."),
    ("w=2",          "Primary + 1 secondary acknowledge. Survives one member failure."),
    ("w='majority'", "Majority of voting members acknowledge. Strongest guarantee."),
    ("j=True",       "Journal flush required before acknowledgement. Crash-safe."),
    ("wtimeout",     "Milliseconds to wait for acknowledgement before error."),
]
for level, desc in wc_levels:
    print(f"  {level:16}  {desc}")

# Apply write concern at the collection level
# Critical financial operation — majority write concern with journal
orders_safe = db.get_collection(
    "orders",
    write_concern=WriteConcern(w="majority", j=True, wtimeout=5000)
)

# High-throughput logging — w=1, no journal (speed over durability)
from pymongo.write_concern import WriteConcern as WC
events_fast = db.get_collection(
    "events",
    write_concern=WC(w=1, j=False)
)

# Insert an order with majority write concern
import datetime
result = orders_safe.insert_one({
    "_id":     "o_wc_demo",
    "user_id": "u001",
    "status":  "processing",
    "total":   59.99,
    "date":    "2024-04-05"
})
print(f"\nOrder inserted with w=majority acknowledged: {result.acknowledged}")

# Read concern — consistency level for reads
print("\nRead concern levels:")
rc_levels = [
    ("local",    "Returns most recent data on the queried member (default)"),
    ("majority", "Returns data acknowledged by majority — no rollback risk"),
    ("snapshot", "Reads a consistent snapshot — used inside transactions"),
    ("available","Fastest — may return data that has been rolled back (shards only)"),
]
for level, desc in rc_levels:
    print(f"  {level:12}  {desc}")

# Clean up demo document
db.orders.delete_one({"_id": "o_wc_demo"})