DocumentDB Change Streams for MCP Servers — Resume Tokens, Real-time Sync, and Lambda Triggers

Enabling change streams

Change streams are not enabled by default. You must set the cluster parameter change_stream_log_retention_duration to a non-zero value:

# AWS CLI — set change stream log retention on the cluster parameter group
aws docdb modify-db-cluster-parameter-group \
  --db-cluster-parameter-group-name my-docdb-params \
  --parameters \
    ParameterName=change_stream_log_retention_duration,\
    ParameterValue=86400,\
    ApplyMethod=immediate

# 86400 seconds = 24 hours
# Valid range: 1 hour (3600) to 7 days (604800)
# Default: 10800 (3 hours) — not set by default, change streams are disabled

# Apply to cluster — requires a cluster reboot if the parameter group is not already attached
aws docdb modify-db-cluster \
  --db-cluster-identifier my-cluster \
  --db-cluster-parameter-group-name my-docdb-params \
  --apply-immediately

What the retention duration means: DocumentDB retains the oplog (operation log) for the specified duration. A resume token only remains valid if it references an oplog entry within the retention window. If your consumer crashes and is offline for longer than the retention period, the resume token becomes stale and you receive MongoServerError: Resume point too far back in past. For production, set retention to at least 24 hours; for critical pipelines, consider 48–72 hours.

CDK configuration:

import { CfnDBClusterParameterGroup, CfnDBCluster } from 'aws-cdk-lib/aws-docdb';

const params = new CfnDBClusterParameterGroup(this, 'DocdbParams', {
  family: 'docdb5.0',
  description: 'DocumentDB cluster parameter group with change streams enabled',
  parameters: {
    change_stream_log_retention_duration: '86400',  // 24 hours
    tls: 'enabled',
  },
});

const cluster = new CfnDBCluster(this, 'DocdbCluster', {
  dbClusterParameterGroupName: params.ref,
  // ... other cluster properties
});

collection.watch() — basic change stream consumer

import { MongoClient, ChangeStreamDocument } from 'mongodb';

async function consumeChangeStream(client: MongoClient): Promise {
  const coll = client.db('mydb').collection('orders');

  // Pipeline filter — only receive insert and update events, not deletes
  const pipeline = [
    { $match: {
        operationType: { $in: ['insert', 'update', 'replace'] },
        // Filter to specific field changes — reduces noise
        // 'updateDescription.updatedFields.status' means only updates that changed 'status'
    }},
  ];

  const changeStream = coll.watch(pipeline, {
    fullDocument: 'updateLookup',
    // 'updateLookup' = on update events, fetch and include the full post-update document
    // Without this, update events only include the delta (updatedFields/removedFields)
    // Note: fullDocumentBeforeChange is NOT supported in DocumentDB
    batchSize: 100,
  });

  changeStream.on('change', async (event: ChangeStreamDocument) => {
    console.log('operationType:', event.operationType);
    console.log('documentKey:', event.documentKey);  // { _id: ObjectId(...) }

    switch (event.operationType) {
      case 'insert':
        await handleInsert(event.fullDocument!);
        break;
      case 'update':
        // event.updateDescription.updatedFields — only the changed fields
        // event.fullDocument — full document (available because of updateLookup)
        await handleUpdate(event.documentKey._id, event.fullDocument!);
        break;
      case 'replace':
        // Replace events have fullDocument set to the new complete document
        await handleReplace(event.fullDocument!);
        break;
      case 'delete':
        await handleDelete(event.documentKey._id);
        break;
    }
  });

  changeStream.on('error', (error) => {
    console.error('Change stream error:', error);
    // Error handler is called on network errors — stream is now closed
    // Reconnect logic must be in the outer loop (see below)
  });

  changeStream.on('close', () => {
    console.log('Change stream closed');
  });
}

// IMPORTANT: change stream cursor closes on network errors, idle timeouts, and
// cluster failovers. Never treat the cursor as permanent — always use a reconnect loop.

Resume tokens — reliable at-least-once processing

A resume token is the _id field of each change event. It encodes the oplog position and allows the stream to restart from exactly that position after a crash. Store it in a durable external store after successfully processing each event.

import { DynamoDBClient, PutItemCommand, GetItemCommand } from '@aws-sdk/client-dynamodb';

const ddb = new DynamoDBClient({ region: process.env.AWS_REGION });
const RESUME_TOKEN_TABLE = 'docdb-resume-tokens';
const STREAM_ID = 'orders-stream-v1';  // key in DynamoDB

async function saveResumeToken(token: object): Promise {
  await ddb.send(new PutItemCommand({
    TableName: RESUME_TOKEN_TABLE,
    Item: {
      streamId: { S: STREAM_ID },
      token:    { S: JSON.stringify(token) },
      updatedAt: { S: new Date().toISOString() },
    },
  }));
}

async function loadResumeToken(): Promise