MongoDB Academy · Aula

Retomando fluxos de alterações após uma interrupção

Você persistirá o token de retomada e reiniciará um fluxo de alterações a partir do último evento processado para garantir a entrega pelo menos uma vez.

Aula 4 de 413 etapas

Retomando fluxos de alterações após uma interrupção é uma aula grátis de MongoDB Academy no CoddyKit. Esta é a aula 4 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de MongoDB Academy, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de MongoDB Academy inclui 4 aulas no total.

Partes desta aula ainda não foram traduzidas e aparecem em inglês.

The Problem: What Happens After a Crash?

A change stream consumer process can be interrupted by network failures, deployments, crashes, or intentional restarts. Without a mechanism to resume from the last processed event, your application would restart the stream from the current moment and miss all events that occurred during the downtime. MongoDB's resume tokens solve this problem by letting you reopen a stream from an exact position in the oplog history.

What Is a Resume Token?

Every change event document has an _id field that serves as its resume token—a binary opaque value that uniquely identifies the event's position in the oplog. It looks like { _data: '...' } where the data is a hex-encoded internal identifier. You do not need to understand its contents; you only need to save it and pass it back to MongoDB to resume from that exact position.

// A resume token is the _id field of a change event
// Example structure (your values will differ):
// {
//   _id: {
//     _data: '8264CE3B7200000002463C6F5A100...'
//   }
// }

// Store it as-is — do not parse or modify the _data value

Persisting the Resume Token

After processing each event, save the resume token to a persistent store (database, file, Redis) before acknowledging that the event was processed. This 'checkpoint' pattern ensures you can pick up from the last successfully processed event after a restart. Save after processing, not before, to avoid skipping events if your processor crashes during handling.

const CHECKPOINT_COLLECTION = 'changeStreamCheckpoints';
const STREAM_ID = 'orderProcessorV1';

for await (const change of changeStream) {
  // Process the event
  await processOrderChange(change);

  // Save token AFTER successful processing
  await db.collection(CHECKPOINT_COLLECTION).updateOne(
    { streamId: STREAM_ID },
    { $set: { resumeToken: change._id, processedAt: new Date() } },
    { upsert: true }
  );
}

Resuming With resumeAfter

To resume from a saved token, pass it to watch() via the resumeAfter option. MongoDB will replay events from after the event identified by the token—your last processed event is not replayed. If the token refers to an event still in the oplog, MongoDB starts delivering from the next event. This provides exactly-once delivery when combined with idempotent event processing.

async function startOrResumeStream() {
  // Load the last saved resume token
  const checkpoint = await db.collection('changeStreamCheckpoints')
    .findOne({ streamId: 'orderProcessorV1' });

  const watchOptions = checkpoint
    ? { resumeAfter: checkpoint.resumeToken }
    : {}; // Start from current position if no token

  const changeStream = db.collection('orders').watch([], watchOptions);

  for await (const change of changeStream) {
    await processOrderChange(change);
    await saveResumeToken(change._id);
  }
}

startAfter vs resumeAfter

MongoDB provides two similar options: resumeAfter and startAfter. resumeAfter resumes from a normal operation token but throws an error if given an 'invalidate' event token. startAfter works the same way but can also resume after an invalidate event—useful when you want to reopen a stream on a collection after it was dropped and recreated. For most use cases, resumeAfter is the correct choice.

// resumeAfter — standard use case, does not work after invalidate tokens
db.collection('orders').watch([], { resumeAfter: savedToken });

// startAfter — can also start after an invalidate event
// (use when the collection may have been dropped and recreated)
db.collection('orders').watch([], { startAfter: savedToken });

startAtOperationTime for Time-Based Resumption

If you do not have a resume token but know the timestamp from which you want to start, you can use startAtOperationTime with a MongoDB Timestamp object. This opens the stream at a specific cluster time rather than a specific event. It is useful for replay scenarios—for example, reprocessing all events since a deployment started—but requires that the oplog still contains history back to that time.

const { Timestamp } = require('mongodb');

// Replay events since a specific time
const startTime = new Timestamp({ t: Math.floor(Date.now() / 1000) - 3600, i: 1 }); // 1 hour ago

const changeStream = db.collection('orders').watch([], {
  startAtOperationTime: startTime
});

// Events from 1 hour ago will be delivered

Oplog Retention and the Horizon

Change stream resumption only works if the token's corresponding event is still in the oplog. MongoDB's oplog is a capped collection with finite size—older entries are overwritten as new ones arrive. On Atlas, you can configure oplog retention (typically 24–72 hours). If your application was down longer than the oplog retention window, the token points to an expired position and MongoDB will return an ChangeStreamHistoryLost error (code 286). Your application must handle this by starting fresh.

async function startOrResumeWithFallback() {
  const checkpoint = await loadResumeToken();
  try {
    const options = checkpoint ? { resumeAfter: checkpoint } : {};
    const stream = db.collection('orders').watch([], options);
    for await (const change of stream) {
      await processChange(change);
      await saveResumeToken(change._id);
    }
  } catch (error) {
    if (error.code === 286) { // ChangeStreamHistoryLost
      console.warn('Resume token expired — starting from now and running catch-up scan');
      await deleteResumeToken();
      await catchUpScan(); // full collection scan to catch missed changes
      await startOrResumeWithFallback();
    } else {
      throw error;
    }
  }
}

Idempotent Event Processing

Because change streams provide at-least-once delivery (the same event might be delivered more than once after a resume), your event handlers must be idempotent—processing the same event twice must produce the same result as processing it once. Techniques include: checking if the document already reflects the change before applying it, using upsert instead of insert, or storing processed event IDs in a 'processed events' set and skipping duplicates.

async function processOrderChange(change) {
  if (change.operationType === 'insert') {
    const { fullDocument } = change;
    // Idempotent upsert — safe to replay:
    await db.collection('orderSummaries').updateOne(
      { _id: fullDocument._id },       // match by _id
      { $set: { ...summarize(fullDocument) } }, // idempotent set
      { upsert: true }                 // create if not exists
    );
  }
}

Automatic Driver-Level Resume

The MongoDB Node.js driver (and other official drivers) includes automatic resume for transient network errors. When the connection to the MongoDB server is temporarily lost, the driver transparently reopens the change stream from the last received event token without your code needing to handle it. This automatic resume handles many common interruption scenarios—you only need to implement manual resume logic for application-level restarts (process crashes, deployments).

// The driver automatically resumes after network blips
// No code needed in your loop for this case:
for await (const change of changeStream) {
  // If the network drops and reconnects, the driver resumes automatically
  // and continues delivering events from where it left off.
  await processChange(change);
  await saveCheckpoint(change._id); // Still save tokens for process restarts
}

Production-Ready Change Stream Handler

A production change stream consumer should combine: resume token persistence (to survive process restarts), oplog expiry handling (fallback to a full scan), idempotent processing (safe to replay events), and graceful shutdown (close streams on SIGTERM). These four elements together achieve reliable event processing even in the face of network failures, deployments, and extended downtime.

class ReliableChangeStreamConsumer {
  constructor(collection, handler) {
    this.collection = collection;
    this.handler = handler;
    this.running = false;
  }

  async start() {
    this.running = true;
    while (this.running) {
      const token = await loadToken();
      const stream = this.collection.watch([], token ? { resumeAfter: token } : {});
      try {
        for await (const change of stream) {
          await this.handler(change);
          await saveToken(change._id);
        }
      } catch (e) {
        if (e.code === 286) { await deleteToken(); continue; } // expired, restart fresh
        if (!this.running) break; // shutting down
        throw e;
      } finally {
        await stream.close();
      }
    }
  }

  stop() { this.running = false; }
}

Graceful Shutdown and Token Flush

On a planned shutdown (deployment, restart), always flush the latest resume token before closing the change stream. If your application processes events in batches (buffering for performance), ensure the token is saved after every batch—not after every individual event. Track the token of the last successfully processed event, not the last received one. This distinction matters: if you receive an event, save its token, but then crash before processing it, you will miss that event when you resume.

// Correct: save token AFTER processing, not before
for await (const change of changeStream) {
  await processChange(change);        // process first
  await saveToken(change._id);        // token saved only after success
  // If crash happens here, the event was already processed — safe
}

// Wrong: saving token before processing
for await (const change of changeStream) {
  await saveToken(change._id);        // token saved
  await processChange(change);        // if crash here, event is skipped on resume!
}

Quick Check

Test your understanding of MongoDB & NoSQL Databases concepts from this lesson.

Responda na aula interativa

Lesson Recap

In this lesson you learned: the resume token is stored in change._id and must be persisted after processing each event, pass the token to watch() as resumeAfter to restart the stream from the last processed event, and handle ChangeStreamHistoryLost (code 286) when the token has expired from the oplog by restarting from current time and running a catch-up scan. Next up we explore creating Atlas Search indexes for full-text search.

Grátis para começar

Aprenda JavaScript com um tutor de IA — grátis

Escreva e execute código real no seu navegador, obtenha ajuda instantânea de um tutor de IA 24/7 e continue de onde parou na web ou no app.

Cursos: 30
Aulas: 120

Começar a aprender grátis Ver o curso completo

Perguntas Frequentes

A aula “Retomando fluxos de alterações após uma interrupção” é grátis?

Sim — o texto completo de “Retomando fluxos de alterações após uma interrupção” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de MongoDB Academy, atualize para CoddyKit PRO. O curso de MongoDB Academy inclui 4 aulas no total.

O que vou aprender em “Retomando fluxos de alterações após uma interrupção”?

Você persistirá o token de retomada e reiniciará um fluxo de alterações a partir do último evento processado para garantir a entrega pelo menos uma vez. Você pratica MongoDB Academy com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar MongoDB Academy?

Nenhuma experiência prévia é necessária. MongoDB Academy no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 4 de 4.

Quanto tempo leva a aula “Retomando fluxos de alterações após uma interrupção”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de MongoDB Academy?

Sim. Cada aula de MongoDB Academy inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

Abrindo um fluxo de alterações em uma coleção
Estrutura do documento de evento de alteração
Filtrando eventos com um pipeline de agregação
Retomando fluxos de alterações após uma interrupção

← Voltar para MongoDB Academy