Change Streams: Watching a Collection in Real Time (Replica Set Required)

Change streams need something a single standalone server doesn’t have — and once that’s in place, watching a collection change in real time is only a few lines.

The bug: watch() on a standalone server

docker exec mongo-lesson mongosh --quiet --eval '
db = db.getSiblingDB("bookshop");
db.books.watch();
'

MongoServerError: The $changeStream stage is only supported on replica sets or mongos

Every earlier lesson in this track ran against a single standalone container. Change streams specifically refuse to work there — not a timeout, not a slow response, an outright rejection.

The fix: a real replica set

docker network create mongo-rs
docker run -d --name mongo-rs1 --network mongo-rs -p 27018:27017 mongo:8 --replSet rs0 --bind_ip_all
docker run -d --name mongo-rs2 --network mongo-rs -p 27019:27017 mongo:8 --replSet rs0 --bind_ip_all
docker run -d --name mongo-rs3 --network mongo-rs -p 27020:27017 mongo:8 --replSet rs0 --bind_ip_all
docker exec mongo-rs1 mongosh --quiet --eval '
rs.initiate({
  _id: "rs0",
  members: [
    { _id: 0, host: "mongo-rs1:27017" },
    { _id: 1, host: "mongo-rs2:27017" },
    { _id: 2, host: "mongo-rs3:27017" }
  ]
});
'
docker exec mongo-rs1 mongosh --quiet --eval 'rs.status().members.map(m => ({name: m.name, state: m.stateStr}))'

[
  { name: 'mongo-rs1:27017', state: 'PRIMARY' },
  { name: 'mongo-rs2:27017', state: 'SECONDARY' },
  { name: 'mongo-rs3:27017', state: 'SECONDARY' }
]

Three containers on their own Docker network, told about each other via rs.initiate(), and an election settles on one PRIMARY. This same three-node replica set is what the rest of this track’s remaining lessons use — a single-node replica set would also unlock change streams, but a real multi-node one is what later lessons on failover actually need.

Watching a collection

docker exec mongo-rs1 mongosh --quiet --eval '
db = db.getSiblingDB("bookshop");
const stream = db.books.watch();
db.books.insertOne({ title: "Dune", author: "Frank Herbert", price: 12.99 });
const change = stream.next();
printjson({ operationType: change.operationType, documentKey: change.documentKey, fullDocument: change.fullDocument });
stream.close();
'

{
  operationType: 'insert',
  documentKey: { _id: ObjectId('6aa6c6d99514865fec6d73c8') },
  fullDocument: {
    _id: ObjectId('6aa6c6d99514865fec6d73c8'),
    title: 'Dune',
    author: 'Frank Herbert',
    price: 12.99
  }
}

watch() returns a cursor immediately; the insert that follows is what stream.next() blocks on and returns. No polling, no re-querying the collection to spot what changed — the change itself arrives as a structured event.

docker exec mongo-rs1 mongosh --quiet --eval '
db = db.getSiblingDB("bookshop");
const stream = db.books.watch();
db.books.updateOne({ title: "Dune" }, { $set: { price: 14.99 } });
const change = stream.next();
printjson({ operationType: change.operationType, updatedFields: change.updateDescription.updatedFields });
stream.close();
'

{ operationType: 'update', updatedFields: { price: 14.99 } }

An update event reports exactly which fields changed (updateDescription.updatedFields), not the whole document — useful for reacting only to a change that actually matters (a price change, say) without diffing the full document yourself.

Resuming after a disconnect

docker exec mongo-rs1 mongosh --quiet --eval '
db = db.getSiblingDB("bookshop");
const stream1 = db.books.watch();
db.books.insertOne({ title: "Foundation", price: 9.99 });
const change1 = stream1.next();
const resumeToken = change1._id;
stream1.close();

db.books.insertOne({ title: "Neuromancer", price: 11.5 });

const stream2 = db.books.watch([], { resumeAfter: resumeToken });
const change2 = stream2.next();
printjson({ operationType: change2.operationType, title: change2.fullDocument.title });
stream2.close();
'

{ operationType: 'insert', title: 'Neuromancer' }

The first stream closed before Neuromancer was inserted — nothing was watching when that write happened. Resuming with the saved token still receives it: the oplog behind the replica set retained the change, and resumeAfter picks up exactly where the closed stream left off, not from whenever the new stream happens to start.

Practice

docker exec mongo-rs1 mongosh --quiet --eval '
db = db.getSiblingDB("bookshop");
const stream = db.books.watch([{ $match: { operationType: "delete" } }]);
db.books.insertOne({ title: "TempBook", price: 1 });
db.books.deleteOne({ title: "TempBook" });
const change = stream.next();
printjson({ operationType: change.operationType, documentKey: change.documentKey });
stream.close();
'

{ operationType: 'delete', documentKey: { _id: ObjectId('6aa6c6fe9161881927bf1d94') } }

docker exec mongo-rs1 mongosh --quiet --eval 'rs.status().members.find(m => m.stateStr === "PRIMARY").name'

mongo-rs1:27017

docker exec mongo-lesson mongosh --quiet --eval '
try {
  db.getSiblingDB("bookshop").books.watch();
} catch (e) {
  print("still fails:", e.codeName);
}
'

still fails: Location40573

Next: multi-document transactions — when they’re actually needed, and when they’re not.