Change streams

So far we have changed data and read it back ourselves. But many applications need to react to changes that other parts of the system make — send an email when a member is added, refresh a cache when a record changes, push an update to a dashboard the moment a fine is paid. Polling the collection over and over is wasteful and slow. A change stream solves this directly: you open a stream with watch() and the database hands you each insert, update, and delete as it happens, in order.

Change streams are built on the oplog, the same internal log a replica set uses to copy writes from the primary to its secondaries, which is why — like a transaction — a change stream needs a replica set (or sharded cluster). On a standalone server watch() fails with MongoServerError 40573: The $changeStream stage is only supported on replica sets or mongos (real output). Outputs below come from mongo 8.3.11 (one-node replica set), mongosh 2.11.1 and Node driver 7.7.0 on this Mac; ids and tokens differ on your machine. Python/Go/Rust tabs mirror the verified shell/Node call and are not executed by the course’s automated checker; their loops block by design, so a service runs them in a worker — the runnable examples here read with a bounded tryNext() poll instead.

Opening a stream on a collection

Each event is a document with an operationType (insert, update, delete, replace, drop, invalidate, …), the documentKey, the namespace ns, and a resume token in _id. The infinite while (stream.hasNext()) loop you often see blocks the shell forever waiting for the next event, so the examples poll with tryNext() (returns null when nothing is waiting) and stop:

mongosh/cs-seed.js

db.createCollection("members");
assert.strictEqual(db.members.countDocuments(), 0);

mongosh

mongosh/cs-watch.js

const stream = db.members.watch();
db.members.insertOne({ name: "Katherine", fines: 0 });

let event = null;
const end = Date.now() + 3000;
while (!event && Date.now() < end) event = stream.tryNext();
console.log(event.operationType, event.documentKey, event.fullDocument.name);
stream.close();
assert.strictEqual(event.operationType, "insert");

Node.js

app/cs-watch.mjs

const members = db.collection("members");
await db.createCollection("members");
const next = async (cs, ms = 3000) => { const end = Date.now() + ms; while (Date.now() < end) { const e = await cs.tryNext(); if (e) return e; } return null; };

const stream = members.watch();
await stream.tryNext();                       // opens the stream on the server (see the trap below)
await members.insertOne({ name: "Katherine", fines: 0 });

const event = await next(stream);
console.log(event.operationType, event.documentKey, event.fullDocument.name);
await stream.close();
if (event.operationType !== "insert") throw new Error("expected the insert event");

Python

with db.members.watch() as stream:
    for event in stream:
        print(event["operationType"], event["documentKey"])

Go

stream, err := coll.Watch(ctx, mongo.Pipeline{})
if err != nil {
    return err
}
defer stream.Close(ctx)

for stream.Next(ctx) {
    var event bson.M
    if err := stream.Decode(&event); err != nil {
        return err
    }
    fmt.Println(event["operationType"], event["documentKey"])
}

Rust

let mut stream = coll.watch().await?;
while let Some(event) = stream.next().await.transpose()? {
    println!("{:?} {:?}", event.operation_type, event.document_key);
}

insert { _id: ObjectId('6abccd0f5465aaa53d0dda5c') } Katherine

insert { _id: new ObjectId('6abccd113997625b550fdab1') } Katherine

How the feed flows

A write goes to the primary, is recorded in the oplog, and the change stream reads it from there and delivers it to every consumer that is watching. One write fans out to many independent listeners:

flowchart LR
  W["A write on the primary"] --> O["Oplog records the change in order"]
  O --> CS["Change stream reads the oplog"]
  CS --> C1["Consumer: send a welcome email"]
  CS --> C2["Consumer: refresh a cache"]
  CS --> C3["Consumer: update a dashboard"]

A write lands in the oplog, the change stream reads it, and fans the event out to every consumer that is watching

What an event contains

A real insert event (Node), a real update event without and with fullDocument: "updateLookup", and the resume token:

app/cs-events.mjs

const members = db.collection("events_members");
await db.createCollection("events_members");
const next = async (cs, ms = 3000) => { const end = Date.now() + ms; while (Date.now() < end) { const e = await cs.tryNext(); if (e) return e; } return null; };

const plain = members.watch();
await plain.tryNext();
await members.insertOne({ name: "Grace", fines: 4 });
const ins = await next(plain);
console.log("insert keys:", Object.keys(ins));
console.log("resume token: length", ins._id._data.length, "starts", ins._id._data.slice(0, 24) + "...");

await members.updateOne({ name: "Grace" }, { $set: { fines: 3 } });
const upd = await next(plain);
console.log("update:", upd.updateDescription, "| has fullDocument:", "fullDocument" in upd);
await plain.close();

const lookup = members.watch([], { fullDocument: "updateLookup" });
await lookup.tryNext();
await members.updateOne({ name: "Grace" }, { $set: { fines: 2 } });
const upd2 = await next(lookup);
console.log("updateLookup fullDocument:", upd2.fullDocument);
await lookup.close();
if (ins._id._data.length !== 188 || "fullDocument" in upd || upd2.fullDocument.fines !== 2) throw new Error("unexpected event shape");

insert keys: [
  '_id',
  'operationType',
  'clusterTime',
  'wallTime',
  'fullDocument',
  'ns',
  'documentKey'
]
resume token: length 188 starts 826ABCCD12000000012B042C...
update: { updatedFields: { fines: 3 }, removedFields: [], truncatedArrays: [] } | has fullDocument: false
updateLookup fullDocument: {
  _id: new ObjectId('6abccd12c596e5cafe63b469'),
  name: 'Grace',
  fines: 2
}

Three facts from that run. A real resume token is a 188-character hex string (_data), not a short toy value. A plain update event carries only the updateDescription delta — no fullDocument unless you ask for updateLookup, which fetches the current document at delivery time (so it can already contain later changes, or be null if the document was deleted meanwhile; see the manual, consulted 2026-09-30). And each stream is only as fresh as the moment it was opened, as the next section shows.

For the exact before/after state, enable pre- and post-images on the collection, then request them. Without enabling, requesting required makes the stream fail with a real MongoServerError 47: ... the pre-image was not found for event:

app/cs-images.mjs

const next = async (cs, ms = 3000) => { const end = Date.now() + ms; while (Date.now() < end) { const e = await cs.tryNext(); if (e) return e; } return null; };
await db.createCollection("images_members", { changeStreamPreAndPostImages: { enabled: true } });
const members = db.collection("images_members");
await members.insertOne({ _id: 1, name: "Ada", fines: 4 });

const stream = members.watch([], { fullDocument: "required", fullDocumentBeforeChange: "required" });
await stream.tryNext();
await members.updateOne({ _id: 1 }, { $inc: { fines: -2 } });
const e = await next(stream);
console.log("before:", e.fullDocumentBeforeChange, "| after:", e.fullDocument);
await stream.close();
if (e.fullDocumentBeforeChange.fines !== 4 || e.fullDocument.fines !== 2) throw new Error("expected the images");

before: { _id: 1, name: 'Ada', fines: 4 } | after: { _id: 1, name: 'Ada', fines: 2 }

Pre-images are stored (in config.system.preimages) and cost space; enable them per collection only when a consumer needs them (manual).

Resuming where you left off

Networks drop and processes restart. Every event carries a resume token in _id; store the token of the last event you handled and reopen the stream from there to receive every change since. Below, a stream reads one event, closes, two more writes happen while nobody listens, and a new stream resumes from the saved token. Two failure modes follow: a made-up token, and resuming after an invalidate (the collection was dropped) — where resumeAfter is refused and startAfter is the tool:

mongosh

mongosh/cs-resume.js

db.createCollection("resume_members");
const first = db.resume_members.watch();
db.resume_members.insertOne({ n: 1 });
let e1 = null;
for (const end = Date.now() + 3000; !e1 && Date.now() < end;) e1 = first.tryNext();
const savedToken = e1._id;                          // persist this
first.close();

db.resume_members.insertMany([{ n: 2 }, { n: 3 }]);   // nobody is listening
const again = db.resume_members.watch([], { resumeAfter: savedToken });
const got = [];
for (const end = Date.now() + 3000; got.length < 2 && Date.now() < end;) {
  const e = again.tryNext();
  if (e) got.push(e.fullDocument.n);
}
again.close();
console.log("resumed events:", got);
assert.deepStrictEqual(got, [2, 3]);

Node.js

app/cs-resume.mjs

const members = db.collection("resume_members");
await db.createCollection("resume_members");
const next = async (cs, ms = 3000) => { const end = Date.now() + ms; while (Date.now() < end) { const e = await cs.tryNext(); if (e) return e; } return null; };

const first = members.watch();
await first.tryNext();
await members.insertOne({ n: 1 });
const savedToken = (await next(first))._id;          // persist this
await first.close();

await members.insertMany([{ n: 2 }, { n: 3 }]);      // nobody is listening
const again = members.watch([], { resumeAfter: savedToken });
const got = [];
for (let i = 0; i < 2; i++) got.push((await next(again)).fullDocument.n);
await again.close();
console.log("resumed events:", got);

let fake = null;
try { await next(members.watch([], { resumeAfter: { _data: "8265F0B3D4000000012B0229" } })); } catch (e) { fake = e; }
console.log("made-up token:", fake.name, fake.code, fake.message);
if (got.join() !== "2,3" || fake.code !== 50793) throw new Error("resume did not behave as expected");

Python

stream = db.members.watch(resume_after=saved_token)

Go

opts := options.ChangeStream().SetResumeAfter(savedToken)
stream, err := coll.Watch(ctx, mongo.Pipeline{}, opts)
if err != nil {
    return err
}
defer stream.Close(ctx)

Rust

let stream = coll
    .watch()
    .resume_after(saved_token)
    .await?;

resumed events: [ 2, 3 ]

resumed events: [ 2, 3 ]
made-up token: MongoServerError 50793 Resume Token does not contain txnOpIndex

Dropping the watched collection ends the stream with a drop event followed by invalidate. You cannot resumeAfter an invalidate; startAfter starts a fresh stream from it:

app/cs-startafter.mjs

const next = async (cs, ms = 3000) => { const end = Date.now() + ms; while (Date.now() < end) { const e = await cs.tryNext(); if (e) return e; } return null; };
await db.createCollection("tmp_members");
const c = db.collection("tmp_members");

const s = c.watch();
await s.tryNext();
await c.drop();
const kinds = [(await next(s)).operationType, null];
const inv = await next(s);
kinds[1] = inv.operationType;
await s.close();
console.log("events after drop:", kinds);

await db.createCollection("tmp_members");
let refused = null;
try { await next(c.watch([], { resumeAfter: inv._id })); } catch (e) { refused = e; }
console.log("resumeAfter invalidate:", refused.code, refused.message);

const again = c.watch([], { startAfter: inv._id });
await again.tryNext();
await c.insertOne({ k: 2 });
const e = await next(again);
console.log("startAfter invalidate ->", e.operationType, e.fullDocument.k);
await again.close();
if (refused.code !== 260 || e.operationType !== "insert") throw new Error("expected resumeAfter refused, startAfter working");

events after drop: [ 'drop', 'invalidate' ]
resumeAfter invalidate: 260 Attempting to resume a change stream using 'resumeAfter' is not allowed from an invalidate notification
startAfter invalidate -> insert 2

Filter at the source

Pass an aggregation pipeline to watch() so the server only sends what you care about. This stream wants updates that drop fines below 3:

app/cs-filter.mjs

const next = async (cs, ms = 3000) => { const end = Date.now() + ms; while (Date.now() < end) { const e = await cs.tryNext(); if (e) return e; } return null; };
await db.createCollection("filter_members");
const members = db.collection("filter_members");
await members.insertMany([{ _id: 1, fines: 4 }, { _id: 2, fines: 9 }]);

const stream = members.watch([
  { $match: { operationType: "update", "updateDescription.updatedFields.fines": { $lt: 3 } } },
]);
await stream.tryNext();
await members.updateOne({ _id: 2 }, { $set: { fines: 8 } });   // filtered out
await members.insertOne({ _id: 3, fines: 0 });                  // filtered out
await members.updateOne({ _id: 1 }, { $set: { fines: 1 } });    // delivered
const e = await next(stream);
console.log(e.operationType, e.documentKey, e.updateDescription.updatedFields);
await stream.close();
if (e.documentKey._id !== 1) throw new Error("the filter should deliver only member 1");

update { _id: 1 } { fines: 1 }

How far back can a token go?

A token can only be resumed while the oplog still holds the operation (“The oplog must have enough history to locate the operation associated with the token”, manual); once it has rolled off, the stream fails. So size the oplog for your longest expected consumer outage. The window is one line to check:

mongosh/cs-oplog.js

const info = db.getReplicationInfo();
console.log({ configuredLogSizeMB: Math.round(info.logSizeMB), usedMB: Math.round(info.usedMB), windowHours: Number((info.timeDiffHours).toFixed(2)) });
assert.ok(info.logSizeMB > 0);

{ configuredLogSizeMB: 1292, usedMB: 1284, windowHours: 0.42 }

That harness server is shared and busy, so its window is only a fraction of an hour — a good reminder that the window shrinks as write volume grows.

In Compass: there is no dedicated change-stream view, but you can run the watch() code in the embedded mongosh tab and make an edit in the Documents tab in another window.

Tips and gotchas

• A change stream needs a replica set or sharded cluster, for the same reason a transaction does.
• Persist the resume token after each handled event; it is what makes a consumer crash-safe. Handlers must be idempotent, because a crash between handling and saving replays the event.
• Tokens expire with the oplog. If the consumer is offline longer than the oplog window, the stream cannot resume and you must reconcile another way.
• Events are capped at 16 MiB; with both pre- and post-images on big documents that matters (manual).

🍃 Watch, do not poll

When your application needs to react to data changes, open a change stream instead of polling. Prime it (first read) before you write, store the resume token of every handled event, and size the oplog for your longest outage.

Reading AI-generated code

An assistant writes a test for “the listener receives the insert” — open the stream, then write:

🐛 Spot the bug

// AI comment: "watch() subscribes immediately, so the insert below will be delivered"
const stream = members.watch();
await members.insertOne({ name: "Katherine" });
for await (const event of stream) { console.log(event.operationType); break; }

Show the answer

In mongosh watch() opens the server cursor immediately, but the Node driver opens the change stream lazily, on the first read. The insert happens before the stream exists on the server, so it is never delivered, and the for await waits forever. Below the missed run (bounded to 2 s instead of hanging) and the fix — do a first read (tryNext()) so the stream is open before writing:

app/cs-lazy-open.mjs

const members = db.collection("lazy_members");
await db.createCollection("lazy_members");
const next = async (cs, ms = 2000) => { const end = Date.now() + ms; while (Date.now() < end) { const e = await cs.tryNext(); if (e) return e; } return null; };

const lazy = members.watch();
await members.insertOne({ name: "Missed" });
const missed = await next(lazy);
await lazy.close();

const primed = members.watch();
await primed.tryNext();                         // FIX: opens the stream on the server
await members.insertOne({ name: "Seen" });
const seen = await next(primed);
await primed.close();

console.log({ lazy: missed?.operationType ?? "MISSED", primed: seen?.fullDocument.name });
if (missed !== null || seen?.fullDocument.name !== "Seen") throw new Error("expected lazy=missed, primed=seen");

{ lazy: 'MISSED', primed: 'Seen' }

If you cannot prime, capture startAtOperationTime before writing, or resume from a saved token.

Mental Model

A change stream is a tail of the oplog with a bookmark. Each event is one entry, in commit order; the resume token is the bookmark. The stream only sees entries from the moment it is opened (or from a bookmark you give it), so the questions are always “where does my stream start?” (prime it, or pass a token) and “is the bookmark still in the log?” (oplog window). A stream tells you what happened, never what the state is now.

Trade-offs

Choice	Benefit	Cost
Change stream (watch())	Real-time events in commit order, resumable from a token, filterable at the source	Requires a replica set; you store the token; the token expires with the oplog; Node opens lazily
fullDocument: "updateLookup"	The whole document on update events with no schema knowledge	An extra lookup per event; returns the current document (possibly later than the event, or null if deleted)
Pre/post images (required)	Exact before/after state	Stored per collection (space) and the stream errors if an image is missing
Polling	Simple, no oplog dependency	Latency = the interval, wasted queries, easy to miss updates

Common Misconceptions

• “watch() is subscribed the moment it returns.” — not in the Node driver (lazy open); the insert made right after watch() was missed, and a first tryNext() fixed it.
• “A while (hasNext()) loop is how you consume a stream.” — it blocks forever waiting for events; use tryNext() with a deadline in scripts, an async iterator in a service.
• “Update events contain the document.” — only the delta unless updateLookup or a post-image is requested.
• “resumeAfter works after any event.” — not after invalidate (code 260); use startAfter.
• “A resume token is a short opaque string.” — real ones are 188 hex characters, and a made-up one fails with code 50793.
• “A change stream tells you the current state.” — it tells you what happened; bootstrap state with a query.

Python/Go/Rust tabs mirror the verified shell/Node behaviour; they are not executed by the course’s automated checker.

What internal structure are change streams built on?

The index B-tree The oplog that records every write in order A separate change collection you create The query plan cache

In the Node driver you call const s = coll.watch(), then insertOne, then read from s. What arrives?

The insert event Nothing: the driver opens the stream lazily on the first read, after the insert An error about the missing token Every event since the collection was created

A plain update event has no fullDocument. How do you get the whole document?

You cannot Pass fullDocument: "updateLookup" (current document) or enable pre/post images and request "required"/"whenAvailable" Set a read concern of linearizable Use a $match stage

You store the token of an invalidate event and call watch([], { resumeAfter: token }). What happens?

It resumes normally MongoServerError 260: resumeAfter is not allowed from an invalidate notification; use startAfter It replays the whole oplog It returns the dropped collection

What limits how long a consumer can be offline before its resume token becomes useless?

The size of the token The oplog window: history must still contain the token's operation transactionLifetimeLimitSeconds The number of collections

Submit

Mark as complete