MongoDB
Explica cómo observar cambios en colecciones, databases o clusters con Change Streams, resume tokens, filtros y manejo de reconexión.
Change Streams permiten observar cambios confirmados en una colección, database o deployment sin leer directamente el oplog. El servidor produce eventos con un que permite continuar después de desconexiones mientras la historia necesaria siga disponible.
const stream = orders.watch([
{
$match: {
operationType: { $in: ['insert', 'update'] },
'fullDocument.businessId': businessId
}
}
], {
fullDocument: 'updateLookup'
});
for await (const event of stream) {
await handleEvent(event);
}Change Streams necesitan una topología con oplog, como replica set o sharded cluster. Un standalone normal no ofrece la misma capacidad.
Un evento puede incluir:
operationType;No todos los eventos contienen automáticamente el documento completo.
Después de procesar y confirmar un evento, guarda el token:
leer evento
→ aplicar efecto idempotente
→ confirmar destino
→ guardar resume tokenGuardar el token antes del efecto puede perder eventos. Guardarlo después puede repetir el último evento si el proceso falla; por eso el consumidor debe ser idempotente.
Según el caso se utilizan opciones como resumeAfter, startAfter o un tiempo de inicio compatible. El token depende de la historia disponible. Si el oplog ya no contiene el punto requerido, no puede reanudarse desde allí y se necesita reconciliación completa.
El tamaño y velocidad del oplog determinan cuánto tiempo puede permanecer desconectado un consumidor. Monitorea:
Change Streams no constituyen retención infinita.
En updates, el evento básico puede incluir solo campos modificados. updateLookup solicita el documento actual después del cambio.
Esto añade una lectura y el documento puede haber cambiado otra vez antes del lookup. No lo interpretes necesariamente como una snapshot exacta del instante del evento.
Cuando están habilitadas, ciertas configuraciones permiten acceder a la imagen anterior o posterior. Tienen costes de almacenamiento y retención y dependen de versión. Verifica disponibilidad antes de diseñar auditoría sobre ellas.
Un consumidor puede recibir de nuevo un evento tras reanudar. Utiliza una identidad:
{
consumer: 'search-indexer',
eventTokenHash,
processedAt
}Un índice unique o una operación convergente evita duplicar efectos.
Los eventos tienen orden dentro del stream observado, pero particionar consumo, combinar streams o enviar a otros sistemas puede cambiarlo. Conserva versión o cluster time cuando el destino necesita rechazar eventos antiguos.
Si el consumidor procesa más lento que las escrituras:
No leas sin límite hacia una cola en memoria. Controla concurrencia, batches y capacidad del destino.
Son útiles para:
No sustituyen automáticamente Kafka, RabbitMQ u otro broker cuando necesitas retención larga, replay arbitrario, múltiples políticas de entrega o desacoplamiento operacional independiente.
Si el evento debe representar una intención de negocio exacta, no siempre basta observar cambios genéricos. Una transacción puede guardar:
{
type: 'OrderConfirmed',
aggregateId: orderId,
payload,
createdAt,
publishedAt: null
}Un worker observa o consulta el outbox y publica de forma idempotente. Esto conserva semántica de dominio.
En un cluster sharded, mongos combina eventos de shards. Cambios de topología, resharding y transacciones distribuidas añaden complejidad. Usa drivers compatibles y prueba failovers.
El usuario necesita privilegios de lectura y change stream apropiados. La pipeline debe filtrar tenants sin exponer datos. No registres documentos completos si contienen secretos.
updateLookup devuelve una versión posterior.Change Streams ofrecen un stream reanudable sobre cambios confirmados, no una cola infinita ni exactly-once. La confiabilidad depende de resume tokens, oplog window, consumidores idempotentes, backpressure y reconciliación. Para semántica de negocio fuerte, combina con outbox.
Replica sets.