Building Real-Time Event-Driven Systems with MongoDB Change Streams

The Problem: Reacting to Data Changes Without Polling

Applications that need to trigger workflows, update caches, or push notifications the moment a document is inserted, updated, or deleted often fall back to periodic polling. Polling adds latency, wastes CPU cycles, and makes it hard to guarantee exactly‑once processing. MongoDB Change Streams replace that pattern with a server‑side, tailable cursor on the oplog (or its sharded equivalent) that streams document‑level events to any driver that can consume a cursor.

Core Mechanics: Oplog Cursor + Resume Tokens

A change stream opens a cursor that reads the replica set’s oplog in real time. Each event contains the operation type (insert, update, delete, replace), the namespace, the document _id, and—when requested—the full pre‑ or post‑image of the document. The cursor returns a resume token (an opaque BSON value derived from the oplog entry’s _id or the cluster time). Storing that token lets a consumer restart exactly where it left off after a crash, a network partition, or a planned rolling upgrade—provided the oplog window hasn’t been exceeded.

Key configuration options:

• fullDocument: 'updateLookup' – fetches the current version of the document after an update (extra round‑trip to the primary).
• fullDocument: 'whenAvailable' – returns the post‑image when the storage engine can provide it without an extra read.
• fullDocumentBeforeChange: 'required' – includes the pre‑image for updates and deletes (MongoDB 6.0+).
• maxAwaitTimeMS – bounds the time the server waits before returning an empty batch, controlling client‑side latency.

Deployment Topologies: Single Replica Set vs. Sharded Cluster

Replica Set

On a standalone replica set the change stream reads directly from the primary’s oplog. Ordering is strict per‑document because there is a single oplog. Latency is typically sub‑millisecond for local clients.

Sharded Cluster

When the target collection is sharded, the change stream is opened through a mongos. The router merges events from all shards using the global clusterTime field, delivering a single globally ordered stream. The trade‑offs:

• Requires readConcern: 'majority' on the cursor, adding a small coordination cost.
• Global ordering is only by clusterTime; unrelated document changes on different shards can interleave.
• Latency is higher than a single‑node stream because each shard must forward its oplog entries to the mongos.

Operational Trade‑offs and Limits

• Connection pressure: Each open change stream holds a cursor that counts against the server’s max connections. Thousands of concurrent streams can degrade primary performance.
• Oplog retention: A slow consumer prevents oplog truncation. If the oplog window (default ~5 % of disk or a configured hour limit) is exceeded, resume tokens become invalid and a full resync is required.
• Large document changes: MongoDB 6.0 introduced the $changeStreamSplitLargeEvent pipeline stage, which breaks a single massive array update into multiple smaller events, reducing memory pressure on the consumer.
• Security: The authenticated user must have find and changeStream privileges on the target collection or database. TLS and authentication are mandatory for production.

Worked Example: Node.js Consumer with Resume Token Persistence

The following snippet shows a minimal, production‑ready pattern. It assumes a replica set connection string stored in process.env.MONGO_URI and a collection named orders in database shop.

// file: change-stream-consumer.js
const { MongoClient } = require('mongodb');
const fs = require('fs');
const TOKEN_FILE = './resume-token.json';

async function main() {
  const client = new MongoClient(process.env.MONGO_URI, {
    // driver‑side retryable reads are enabled by default
  });
  await client.connect();
  const coll = client.db('shop').collection('orders');

  // Load previous token if present
  let resumeToken = null;
  if (fs.existsSync(TOKEN_FILE)) {
    resumeToken = JSON.parse(fs.readFileSync(TOKEN_FILE, 'utf8'));
  }

  const pipeline = [
    { $match: { 'operationType': { $in: ['insert', 'update', 'replace'] } } },
    { $project: { _id: 1, operationType: 1, fullDocument: 1, ns: 1, documentKey: 1 } }
  ];

  const options = { resumeAfter: resumeToken, fullDocument: 'updateLookup' };
  const stream = coll.watch(pipeline, options);

  stream.on('change', (event) => {
    // Persist token *before* processing to guarantee at‑least‑once semantics
    fs.writeFileSync(TOKEN_FILE, JSON.stringify(event._id));
    console.log('Received:', event.operationType, event.documentKey._id);
    // TODO: invoke business logic (e.g., publish to message bus)
  });

  stream.on('error', (err) => {
    console.error('Change stream error:', err);
    // Driver will attempt automatic resume if token is still valid
  });

  // Graceful shutdown
  process.on('SIGINT', async () => {
    await stream.close();
    await client.close();
    process.exit(0);
  });
}

main().catch(console.error);

Where to run: Any host with network access to the replica set (or mongos for sharded clusters). The process needs read/write access to the token file directory.

Required permissions: The MongoDB user authenticating the client must have find and changeStream on shop.orders (or the whole shop database). Example grant via mongosh:

use admin
db.createUser({
  user: "streamUser",
  pwd: "",
  roles: [
    { role: "read", db: "shop" },
    { role: "changeStream", db: "shop" }
  ]
})

Expected checks: After starting the consumer, insert a document in mongosh (db.orders.insertOne({item: "widget", qty: 5})) and verify the consumer logs the event. Kill the process, wait a few seconds, restart it, and confirm the same event is not duplicated (token persisted before processing).

Risks: If the consumer crashes after processing but before writing the token, the next run will re‑process the last event (at‑least‑once). Design downstream handlers to be idempotent or use a deduplication store.

Practical Verification Checklist

1. Oplog window: Run db.printReplicationInfo() on the primary to see the current oplog time range. Ensure your longest expected downtime is shorter than this window.
2. Resume token validity: After a planned maintenance window, restart the consumer with the stored token and confirm no ResumeTokenInvalid error appears.
3. Sharded ordering test: Write to two shards concurrently, capture clusterTime from events, and verify global ordering matches the timestamps.
4. Connection count: Monitor db.serverStatus().connections while scaling the number of change streams; keep well below maxIncomingConnections.

Closing Action Items

• Prototype a single change stream on a development replica set using the example above.
• Measure oplog growth under a deliberately slow consumer (e.g., await new Promise(r => setTimeout(r, 5000)) per event) and verify the oplog window stays comfortable.
• If you run a sharded cluster, open the stream through mongos and add a $match filter to limit the event volume per consumer.
• Document the resume‑token storage strategy (file, KV store, or database) and include it in your disaster‑recovery runbook.

Change Streams turn MongoDB’s replication log into a first‑class event source. By understanding the cursor lifecycle, the impact of fullDocument options, and the operational limits of oplog retention, you can build resilient, real‑time pipelines without the complexity of external CDC tools.