The Cost of Polling: Why Traditional Sync Breaks at Scale
Last year, our product catalog hit 1.8 million documents, and our synchronization pipeline hit a wall. Every 10 seconds, a Celery worker scanned the collection looking for new writes:
# The naive polling approach that saturated our database
last_sync = datetime.utcnow() - timedelta(seconds=10)
updates = db.products.find({"updated_at": {"$gte": last_sync}})At 2,000 queries a minute, this pattern broke down fast:
- Wasted CPU and Disk I/O: Even during dead hours with zero writes, index scans ran constantly. Primary node CPU usage hovered stubbornly around 45% just from sync checks alone.
- Lag Gaps: Search indices trailed our transactional store by up to 10 seconds. Customers regularly added items to carts that had already gone out of stock.
- Silent Misses: Hard deletes left no trace. A deleted document vanished completely from
$gtequeries, forcing us to run daily 40-minute reconciliation batch jobs to catch orphaned records.
Relational databases solve this with Write-Ahead Log decoders like Debezium or PostgreSQL replication slots. MongoDB takes a cleaner route: Change Streams, an event capture mechanism built right into the replication engine.
How Change Streams Capture Events from the Oplog
Change Streams let applications track document mutations across a collection, a database, or an entire cluster without issuing manual read queries.
The Mechanics Under the Hood
Every replica set maintains an internal capped collection named oplog.rs. When a client performs a write, the primary node records the operation in this log and replicates it to secondaries. Change Streams attach a tailable cursor to the oplog, decoding raw database operations into clean, structured JSON documents.
You open a stream. Then, MongoDB pushes notifications across the wire the instant a write commits to a majority of nodes.
Why This Beats Dual-Writing
- True Atomicity: Dual-writing to both MongoDB and an external broker like Kafka or RabbitMQ creates split-brain risks. If your network hiccups between the database commit and the broker publish, your data desynchronizes. Change Streams guarantee you only process changes that genuinely landed in storage.
- Resilient Recovery: Every emitted event carries an opaque binary resume token in its
_idfield. If your consumer restarts after an unexpected crash, pass that token on reconnect. MongoDB picks up right where the worker left off. - Zero-Cost Network Filtering: You can run aggregation stages—such as
$matchand$project—directly inside the database engine. The cluster drops irrelevant updates before sending bytes across the network.
Hands-On Implementation: Building a Resilient Sync Worker
Here is a battle-tested Python worker that monitors inventory levels and synchronizes downstream services.
Prerequisites
Change Streams require an active oplog, which means you need a replica set. If you are developing against a local standalone instance, convert it to a single-node replica set:
# Start mongod with replica set naming enabled
mongod --port 27017 --dbpath /var/lib/mongodb --replSet rs0
# Initialize the replica set inside mongosh
mongosh --eval 'rs.initiate()'Next, install PyMongo:
pip install pymongo1. Initializing and Filtering the Stream
Do not stream all collection activity to application memory. Instead, tell MongoDB to evaluate an aggregation pipeline server-side. The filter below matches inserts, updates, and replacements where the stock field actually changed:
import time
import json
from pymongo import MongoClient
from pymongo.errors import PyMongoError
MONGO_URI = "mongodb://localhost:27017/?replicaSet=rs0"
client = MongoClient(MONGO_URI)
db = client["store"]
collection = db["inventory"]
# Filter events on the database node
pipeline = [
{
"$match": {
"operationType": {"$in": ["insert", "update", "replace"]},
"updateDescription.updatedFields.stock": {"$exists": True}
}
}
]2. Tracking Resumption Tokens
To survive restarts, write the latest resume token to disk or Redis after successfully handling each event:
TOKEN_FILE = "last_resume_token.json"
def save_resume_token(token):
if not token:
return
with open(TOKEN_FILE, "w") as f:
json.dump(token, f)
def load_resume_token():
try:
with open(TOKEN_FILE, "r") as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return None3. Writing the Reactive Consumer Loop
Combine token management with a reconnect loop to keep your listener resilient against network blips and cluster step-downs:
def process_change_event(event):
op_type = event["operationType"]
doc_id = event["documentKey"]["_id"]
if op_type == "insert":
product = event.get("fullDocument", {})
print(f"[SYNC] Item created: {product.get('name')} (ID: {doc_id})")
elif op_type in ("update", "replace"):
updated = event["updateDescription"]["updatedFields"]
print(f"[SYNC] Stock updated for {doc_id}: {updated}")
def run_stream():
resume_token = load_resume_token()
while True:
try:
print("Opening change stream cursor...")
with collection.watch(
pipeline=pipeline,
resume_after=resume_token,
full_document="updateLookup"
) as stream:
for change in stream:
process_change_event(change)
# Commit token only after handling succeeds
resume_token = change["_id"]
save_resume_token(resume_token)
except PyMongoError as err:
print(f"Stream interrupted: {err}. Retrying in 5 seconds...")
time.sleep(5)
if __name__ == "__main__":
run_stream()Testing the Pipeline
Start your Python script in one terminal. In another, connect via mongosh and insert a new test product:
use store
db.inventory.insertOne({ name: "Mechanical Keyboard", stock: 45 })Within 15 to 25 milliseconds, your worker outputs the processed event:
[SYNC] Item created: Mechanical Keyboard (ID: 651fa4bc88a7c1e39...)Production Gotchas to Watch For
Lookups Add Server Load
By default, an update event only returns changed keys in updateDescription. Setting full_document="updateLookup" forces MongoDB to query the live document’s post-image. At 5,000 updates per second, this can double your read volume. Whenever possible, rely on delta payloads instead of full document lookups.
Oplog Rollover Triggers Fatal Errors
The oplog operates as a fixed-size ring buffer. If a consumer stays offline for 6 hours on an oplog configured with a 4-hour retention window, its resume token expires. When that happens, the server raises a fatal ChangeStreamHistoryLost error. Size your oplog for at least 24 to 48 hours of write volume, and configure Prometheus alerts on consumer replication lag.
Wrapping Up
Polling trades simplicity for latency, high CPU load, and messy edge cases. MongoDB Change Streams eliminate that compromise. By streaming oplog events directly through an aggregation pipeline, you get sub-second synchronization and robust crash recovery with minimal boilerplate.
