MongoDB Change Streams let an application subscribe to committed changes across a collection, database, or deployment, but receiving events is not the same as recovering without loss or duplication. A production consumer should store the complete _id as an opaque resume token, advance its checkpoint only after the business effect succeeds, reopen with the original pipeline and options, and define explicit paths for expired history, invalidate events, backpressure, and duplicate delivery.

When are Change Streams the right primitive?

The MongoDB Change Streams documentation describes a database change subscription that avoids manually tailing the oplog. A consumer can watch one collection, a database, or an entire deployment and use an aggregation pipeline to filter events. Common applications include search-index projection, cache invalidation, audit views, notifications, and derived data.

A change stream is not automatically a job queue. It tells you that a database change committed; it does not decide whether an email was sent, a search document was indexed, or a third-party API accepted a command. The consumer still owns process crashes, retries, duplicate reads, downstream throttling, and poison events. If the work also needs schedules, priorities, human retries, or compensation, first use the FastAPI background-task guide to decide whether the handler should enqueue a durable job or workflow.

Change Streams require a replica set or sharded cluster. The event _id is the resume token. A pipeline must not remove or modify it; MongoDB rejects such change stream pipelines because they make reliable resumption impossible.

Which event fields should a consumer preserve?

Do not reduce every event to fullDocument. Preserve enough context to route, deduplicate, investigate, and recover:

Field Purpose Frequent mistake
_id Complete resume token Parsing it into a timestamp or saving only part
operationType insert, update, replace, delete, invalidate, and others Assuming every event has a full document
ns Database and collection Losing source identity in a shared consumer
documentKey Identity of the affected document Depending on fullDocument for deletes
updateDescription Delta for an update Treating the delta as final current state
clusterTime Server-side ordering context Using it as a universal business version

Treat the resume token as opaque BSON. Store the entire _id using a representation that round-trips exactly, such as BSON or correctly handled Extended JSON. Do not decompose and reconstruct it, and do not substitute clusterTime. Token shape can vary by server version and event type.

For update operations, the default event emphasizes the delta. fullDocument="updateLookup" asks MongoDB to fetch the current majority-committed document, but the lookup occurs when the event is processed. A later update or deletion can make that document differ from the exact state at the original event. If the requirement is historical auditing, evaluate pre- and post-images rather than treating update lookup as a version archive.

When should the checkpoint move?

The safe default is simple: persist an event's resume token only after its business effect has succeeded.

Suppose the consumer receives event E, saves its checkpoint, and then calls a search index. If that call fails and the process exits, restart begins after E and the index update can be permanently missing. Reverse the order—apply the index update, then checkpoint—and a crash between the two causes E to be read again. The downstream action therefore needs to be idempotent.

This is an at-least-once processing design: duplicates are acceptable and silent gaps are not. The following skeleton expresses ordering, not a complete runnable service:

with collection.watch(
    pipeline,
    full_document="updateLookup",
    resume_after=checkpoint.resume_token,
) as stream:
    for event in stream:
        operation_id = encode_token(event["_id"])
        apply_idempotently(operation_id, event)
        checkpoint_store.compare_and_set(
            expected=checkpoint.version,
            token=event["_id"],
        )

apply_idempotently must be a real contract. A search projection can upsert by the source document key and an authoritative version. A notification worker can put the operation ID behind a unique constraint in an outbox or deduplication table. Billing and inventory changes need business-level versions and transactional boundaries; merely retrying the same HTTP call is not proof of safety.

The checkpoint store also needs compare-and-set semantics. If two consumers accidentally share a stream identity, a slower instance must not overwrite a newer token and trigger a large replay. A practical record contains the stream identity, pipeline digest, options digest, token, monotonic revision, and lease or owner information.

The existing atomic FastAPI, MongoDB, and Redis publishing guide separates the committed database revision, cache invalidation, and public verification. Apply the same discipline here: database event committed, external effect complete, checkpoint advanced, and reconciliation passed are distinct states.

How do resumeAfter, startAfter, and startAtOperationTime differ?

MongoDB exposes three relevant starting mechanisms:

  • resumeAfter continues after a stored event and is the ordinary recovery choice.
  • startAfter also begins after a token and can create a new stream after an invalidate event.
  • startAtOperationTime starts after a server operation time and is useful for a controlled bootstrap when no token exists; it is not a drop-in substitute for a durable checkpoint.

MongoDB explicitly warns that a resume token must be used with the same pipeline and options that produced it. Changing the filter, fullDocument behavior, collation, or watch scope while reusing the token can cause unpredictable behavior, data-consistency problems, or a failed resume.

Compute a stable digest for the stream configuration:

stream_id = orders-search-v3
pipeline_digest = sha256(canonical_pipeline)
options_digest = sha256(canonical_options)

Automatically resume only when the digests match. A material configuration change gets a new stream identity, a declared backfill or cutover point, and its own checkpoint. Never silently attach an old token to a new logical stream.

Why can a valid token become unusable?

Resumption depends on enough oplog history still being available. High write volume, a small oplog, or a consumer that has been down too long can move the required event outside the retained window. Drivers can recover from some transient network and cursor errors, but they cannot reconstruct history that no longer exists.

Monitor at least:

  • wall-clock lag since the most recently processed event;
  • checkpoint age and revision;
  • input and completion rate;
  • handler failures and retry-queue depth;
  • resume attempts by error class;
  • downstream deduplication hits;
  • missing or divergent records found by reconciliation.

If a token cannot be resumed, silently starting “now” creates an invisible data gap. A rebuildable search index or cache should stop incremental consumption, perform a full or partitioned backfill from authoritative MongoDB state, and then establish a new checkpoint. A non-rebuildable audit record or irreversible external effect needs a transactional outbox, separate event archive, or human incident path; it cannot honestly claim continuity.

For a search projection, pair this design with the MongoDB $rankFusion hybrid-search guide. The stream discovers source changes, while the indexing path uses stable document keys and reconcilable versions. Ranking logic does not belong inside the checkpoint transaction.

What should happen after an invalidate event?

Dropping or renaming a collection can emit invalidate and close the cursor. Ordinary resumeAfter cannot be assumed to cross that boundary; startAfter exists to start a new stream after the invalidate token.

Technical resumability is not the same as business validity. A renamed collection can have a new namespace, indexes, schema, ownership, and retention policy. Treat invalidate as a control-plane event: stop effects, retain the token and DDL context, validate the new target, and decide whether to use startAfter, create a new stream, or rebuild.

An infinite retry loop should not swallow invalidate. Classify transient network failures, resumable cursor errors, expired history, authentication failures, and DDL termination separately. Each class needs a bounded action and an alert owner.

How should a PyMongo worker shut down and restart?

The PyMongo Change Streams guide documents pipeline, full_document, resume_after, start_after, start_at_operation_time, and max_await_time_ms. Blocking iteration is appropriate for a dedicated worker. On shutdown, stop fetching, allow in-flight work to finish within a deadline, persist the resulting checkpoint, and close the cursor.

Reading and processing can be decoupled, but the queue between them must be bounded. An unlimited memory queue converts downstream latency into an out-of-memory failure. When a high-water mark is reached, pause or slow reading, reduce concurrency if the target is throttling, and continue reporting lag.

A useful internal state machine is:

READ -> VALIDATE -> APPLY -> CHECKPOINT -> ACK_LOCAL
                  |          |
                  v          v
                RETRY      RECONCILE

ACK_LOCAL means the consumer completed its local contract. It does not by itself mean that a search cluster, email provider, or other third party is globally consistent. Operational reports should distinguish source committed, effect accepted, checkpoint saved, and reconciliation complete.

How do you backfill without racing the live stream?

A naive backfill reads all current documents and starts a stream afterward. Writes that occur between those two operations can be lost. Starting the stream first and buffering every event indefinitely can exhaust memory or disk.

Choose a documented bootstrap strategy. One practical pattern is:

  1. Open the stream and record a starting token or operation time.
  2. Begin a bounded, durable event buffer.
  3. Scan the authoritative collection in partitions and upsert the projection.
  4. Drain buffered events idempotently in stream order.
  5. Switch to steady-state consumption and advance the normal checkpoint.

The buffer needs capacity planning and a failure path. If it reaches its limit, abort the bootstrap and retry with smaller partitions or more throughput; do not discard old events. Another option is an application-maintained version field that allows reconciliation to pick the newest source state, but it must be monotonic for the relevant business object.

Always run a post-backfill reconciliation. Compare counts only as a coarse signal. Sample or hash stable identifiers, versions, and required fields so equal counts cannot conceal different records.

What should production verification cover?

Functional behavior

  1. Insert, update, replace, and delete events follow the intended path.
  2. Update deltas, fullDocument, and missing fields on deletes are handled deliberately.
  3. The pipeline preserves event _id unchanged.
  4. Identical document keys from different namespaces do not collide.

Failure behavior

  1. Kill the worker after the external effect but before checkpointing; replay is harmless.
  2. Kill it before the effect; the event appears again.
  3. Disconnect and restore the network; driver and application retry responsibilities remain bounded.
  4. Supply an unusable token; the system stops for rebuild or reconciliation instead of skipping forward.
  5. In an isolated environment, rename or drop the watched collection and verify the invalidate path.
  6. Throttle the downstream target and confirm bounded queues and rising lag alerts.

Observability

Emit structured fields such as stream ID, operation type, namespace, duration, result, and a privacy-safe operation ID. Do not log whole business documents or credentials. The FastAPI OpenTelemetry tracing guide explains context propagation across requests and background execution. Use it to connect an originating write with a derived update, but never make a high-cardinality resume token a metric label.

Production checklist

  • [ ] The full opaque resume token round-trips without transformation.
  • [ ] Checkpoints advance only after successful idempotent effects.
  • [ ] The checkpoint store uses CAS and a lease or single-consumer guarantee.
  • [ ] Pipeline and option digests are bound to the checkpoint.
  • [ ] Queue capacity and downstream backpressure are bounded.
  • [ ] Transient, resumable, expired-history, auth, and invalidate errors are distinct.
  • [ ] A token-loss rebuild path and reconciliation procedure have been tested.
  • [ ] Shutdown waits for in-flight work and persists progress within a deadline.
  • [ ] Logs exclude full documents, secrets, and raw tokens where unnecessary.

Common mistakes

  • Checkpointing before the external effect and skipping a failed event.
  • Checkpointing afterward without downstream idempotency and duplicating effects.
  • Storing only clusterTime or part of the token.
  • Reusing a token after changing the pipeline.
  • Treating updateLookup as an event-time historical snapshot.
  • Starting at the current time after token expiry without declaring a gap.
  • Sharing a checkpoint across workers without a lease and CAS.
  • Using an unbounded queue as a backpressure strategy.
  • Interpreting cursor auto-resume as end-to-end exactly once.

FAQ

Do Change Streams provide exactly-once processing?

They provide a resumable stream of database changes. They do not create one transaction across MongoDB, your checkpoint store, and an arbitrary external system. Most production consumers use at-least-once delivery plus idempotent effects, unique constraints, business versions, and reconciliation.

Is checkpointing every event too expensive?

Small batch checkpoints are possible, but a crash replays the entire uncheckpointed batch. Batch size trades checkpoint writes against recovery time and duplicate work. Establish idempotency before optimizing the frequency.

Can the driver handle all recovery automatically?

No. A driver can resume selected errors, but it does not know whether your external effect completed and cannot resolve expired oplog history, changed configuration, or business divergence. The application still needs a durable checkpoint and recovery state machine.

When is an outbox better?

If the requirement is a rebuildable projection of authoritative MongoDB state, Change Streams are often a good fit. If every semantic business event must commit atomically with the business write and remain independently auditable, a transactional outbox is usually clearer. Change Streams can then consume the outbox collection.

A reliable Change Streams service is not merely for event in stream. It is a replication pipeline with explicit checkpoints, idempotency, bounded backpressure, recovery, and reconciliation. Once those states are visible, a network interruption or process restart can be handled honestly instead of being mistaken for a consistency guarantee.