MongoDB

@ocoda/event-sourcing-mongodb provides the MongoDBEventStore and the MongoDBSnapshotStore. It needs the mongodb driver (^6.10 or ^7), which is a peer dependency: see Installation and Database drivers are peer dependencies. It is tested on MongoDB 7, 8 and 9.0, each as a standalone server and as a replica set. Databases that speak MongoDB’s protocol but aren’t MongoDB are not tested: see MongoDB-compatible engines.

MongoDB 8.3 and 9.0 need driver 7.3 or later

Use mongodb 7.3 or later with MongoDB 8.3 or 9.0: it’s the first driver release whose supported servers go beyond 8.2, up to 9.0. The last 6.x release, 6.21, and 7.0 to 7.2 support MongoDB up to 8.2. An older driver still connects to a newer server, because a driver only refuses a server that requires a newer wire protocol than it speaks, but MongoDB doesn’t support that combination, and the tests on 8.3 and 9.0 use a 7.x driver that does. The peer range stays ^6.10 or ^7, for the servers up to 8.2.

Coming from 3.x?

4.0 stores events and snapshots in a new schema (schema v2). Collections that 3.x created must be migrated once, offline: see Migrating from 3.x. The design is in ADR 0002.

Configuration

EventSourcingModule.forRootAsync({
  useFactory: () => ({
    events: [...],
    eventStore: { driver: MongoDBEventStore, url: 'mongodb://db-1,db-2,db-3/app?replicaSet=rs0' },
    snapshotStore: { driver: MongoDBSnapshotStore, url: 'mongodb://db-1,db-2,db-3/app?replicaSet=rs0' },
  }),
});

• url is the connection string. Its database holds the collections of every pool and their catalog.
• Every other option is passed to the MongoClient, such as serverSelectionTimeoutMS, which bounds how long connect() waits for a server before the bootstrap fails.
• ddl: 'auto' (the default) lets ensureCollection create missing collections, indexes and catalog entries. With 'none', the store only checks and registers them, and a missing collection throws an error that lists the mongosh statements that create it; see Privileges.

Topologies and capabilities

connect() asks the server what it is (hello) and sets the event store’s capabilities accordingly. On MongoDB:

Deployment	atomicAppend	globalOrder	headers
Replica set (also a single-node one)	true	'gap-safe'	true
Sharded cluster (mongos)	true	'best-effort'	true
Standalone server	false	'best-effort'	true

Run MongoDB as a replica set. A single-node replica set is enough. On a replica set an append is one transaction, and readAll never skips an event that commits late, so a projection that tails the store with readAll({ fromPosition: checkpoint + 1n }) sees every event exactly once.

A standalone server has no transactions. The store warns once when it connects to one, and:

• appends are not atomic: when an insert fails halfway, the store removes the events it inserted; if that fails too, the append throws an EventStorePersistenceException with outcome: 'unknown';
• until that removal, the events of the failed append are visible: another append can build on them (ExpectedVersion.Any reads them as the version of the stream), and once they are removed that stream has a gap in its versions. A reader that tails the store can also process events that are removed afterwards;
• a failed or conflicting append burns the positions it reserved, which leaves holes that never fill;
• a reader that tails the store can miss events for good, because an append can commit below a position the reader already passed. readAll over events that are no longer being written returns every event, in order;
• a snapshot append unflags the previous latest snapshot before it inserts the new one. A crash in between, or a failure to flag the previous one again, leaves the stream without a latest flag: getLastSnapshot still reads the highest version, but getLastEnvelopesForAggregate (and SnapshotRepository.loadAll) skip the stream until its next snapshot.

On a sharded cluster, appends are transactions, but the visibility of positions across shards is unproven, so the global order is 'best-effort'.

MongoDB-compatible engines

Some databases speak MongoDB’s protocol without running MongoDB. 'gap-safe' rests on MongoDB’s transactions and majority reads, which these engines implement differently or not at all, and none of them is tested. So connect() also asks which engine it talks to, and gives each of the engines below globalOrder: 'best-effort', whatever topology it reports:

Engine	What tells it apart	atomicAppend
FerretDB	buildInfo has a ferretdb document (FerretDB 2) or a ferretdbVersion (FerretDB 1).	false
DocumentDB (the open-source gateway) and Azure DocumentDB (formerly Azure Cosmos DB for MongoDB vCore)	hello has an internal document.	true
Amazon DocumentDB	getEngineVersion answers an engineVersion.	true on an instance-based cluster (from 4.0 on; on 3.6, which has no transactions, appends fail), false on an elastic cluster
Azure Cosmos DB for MongoDB (request units)	getLastRequestStatistics answers a RequestCharge.	false

• MongoDB answers getEngineVersion and getLastRequestStatistics with CommandNotFound. The store sends each one only to a replica set with the vendor’s name (rs0, globaldb) or to a host of the vendor’s, and a command that fails counts as no signal.
• Without a signal, the host names of the managed services decide: *.docdb.amazonaws.com and *.docdb-elastic.amazonaws.com for Amazon DocumentDB, *.mongocluster.cosmos.azure.com for Azure DocumentDB, and *.mongo.cosmos.azure.com and *.documents.azure.com for Azure Cosmos DB (also under azure.cn).
• The store logs a warning once per process and engine.

The append path still follows the topology the engine reports, not the engine:

• FerretDB reports a standalone server, unless it runs with a development flag that makes it report a replica set. As a standalone server, appends run without a transaction, as on a standalone MongoDB server. As a replica set, every append runs as a transaction, which FerretDB doesn’t implement.
• Azure Cosmos DB for MongoDB reports a replica set (globaldb), so every append runs as a transaction that updates the catalog and inserts into the pool’s collection. Cosmos DB has no transaction that spans two collections.
• An elastic Amazon DocumentDB cluster presents itself as a sharded cluster (a mongos, as AWS’s documentation shows; not tested), so appends run as transactions, which elastic clusters don’t support.

On those three, appends fail or aren’t atomic. DocumentDB and Azure DocumentDB report a sharded cluster, and the instance-based clusters of Amazon DocumentDB a replica set (rs0). They support transactions (Amazon DocumentDB from 4.0 on), so appends run as transactions there, but the global order is still 'best-effort'. Other engines, such as Firestore with MongoDB compatibility, aren’t detected and aren’t supported either: see Other databases.

Use MongoDB, as a replica set, for atomic appends and a gap-safe order. A single-node replica set is enough.

Schema v2

Event collections. Each pool is a collection: events for the default pool, <pool>-events for the others. A document:

{
  _id: '01J9…',                    // the event id
  streamId: 'account-42',
  version: 3,
  event: 'account-credited',
  payload: { … },
  aggregateId: '42',
  occurredOn: ISODate('…'),
  correlationId: '…',              // optional
  causationId: '…',                // optional
  globalPosition: NumberLong('17'),
  headers: { tenant: 'acme' },     // optional
  eventVersion: 2,                 // optional, from pre-built envelopes
}

Absent fields are left out, never stored as null. The collection has a unique index on { streamId: 1, version: 1 }, a unique index on { globalPosition: 1 }, and a validator (validationLevel: 'strict') that requires a 64-bit globalPosition, a streamId and a version. The validator also fences off 3.x writers after a migration: their inserts fail with code 121.

Snapshot collections: snapshots and <pool>-snapshots. The latest snapshot of a stream carries latest: 'latest#<streamId>'; the others have no latest field. A unique partial index (latest_unique, on { aggregateName: 1, latest: 1 } where latest is a string) keeps a single latest snapshot per stream, also when appends race. getLastSnapshot and getLastEnvelope read the snapshot with the highest version. getLastEnvelopesForAggregate (and SnapshotRepository.loadAll) pages the streams in descending binary order of their aggregate ids, with aggregateId as an exclusive cursor.

The catalog, event_sourcing_collections, has one document per collection of the stores: { _id: 'events', kind: 'events', schemaVersion: 2, lastPosition: NumberLong('17') }. It is the position counter of each event pool, the registry of the schema versions, and the source of listCollections(), which therefore no longer lists collections that were never registered, such as 3.x collections that weren’t migrated. Running migrations hold leases in it (kind: 'lock'). Don’t drop it.

Dropping a pool. Delete the pool’s catalog document first, then drop its collection: appends to the pool then fail with outcome: 'not-persisted' instead of creating the collection again. An insert into a collection that is gone creates it implicitly, without the validator and the unique indexes; the next ensureCollection of the pool restores them (and fails if the events appended meanwhile break the unique indexes), or, with ddl: 'none', names the statements that restore them. A pool whose collection is dropped and created again by ensureCollection continues its counter.

Payloads are BSON documents: a Date in a payload comes back as a Date (the SQL stores return an ISO string), and a bigint is stored as an Int64 and comes back as a number, unless the client runs with useBigInt64: true. Global positions are always bigints, whatever the client options.

Appends and global positions

On a replica set, an append is one transaction (readConcern: 'snapshot', writeConcern: 'majority') whose first write increments the pool’s counter in the catalog; then it inserts the events. Two appends to the same pool conflict on the counter, so they commit in the order of their positions, and a rolled-back append also rolls back its positions: no holes. The store retries a transaction that fails with a transient error (a write conflict, a failover) with a short random backoff for up to 30 seconds, then throws an EventStorePersistenceException with outcome: 'not-persisted'. The commit waits for a majority of the replica set at most until those 30 seconds are over (at least one second, maxCommitTimeMS), so an append doesn’t hang when the replica set has lost its majority, for instance a primary, a secondary and an arbiter with the secondary down. A commit whose result is unknown (a network error, a failover, that timeout) is retried up to three times, then reported as outcome: 'unknown'. The version check before an append reads the primary, whatever the read preference of the connection string.

readAll reads with readConcern: 'majority' on replica sets and sharded clusters, so a failover can’t take back a position a reader already saw. Restoring a backup still can: see Failover and replicas.

Throughput. The appends of one pool serialize on its counter: each waits for the previous commit. On a single-node MongoDB 8.2 replica set in Docker on a laptop, one pool took about 300 appends per second with 8 concurrent writers (266 with one writer, 195 with 16), with about 1.8 retried write conflicts per append and a worst case of 1.7 seconds. Four pools took about 510 appends per second together. Pools are the unit of write scale: spread tenants or bounded contexts over pools. See the evidence.

ExpectedVersion.Any makes up to 16 attempts when concurrent appends take the versions of a stream, and can still end in a conflict under sustained contention on one stream: retry such a conflict in your application.

Privileges

• Running the stores with ddl: 'auto': the readWrite role on the database.
• Running them with ddl: 'none': find, insert, update, remove, listCollections and listIndexes on the database. Create the collections, indexes and validators from the statements that ensureCollection names, or from a dry run of migrate().
• migrate(): also collMod for the fence, as in the dbAdmin role. A dry run reports the privileges that the steps still to run need and the user lacks as blocking, before any write; a later run that only removes eventDate needs update and dropIndex, not collMod.
• On a sharded cluster, migrate() reads config.collections to refuse sharded collections: also grant find on it (the clusterMonitor role has it), or the dry run reports the collection as blocked.

Migrating from 3.x

The 3.x collections have no global positions, and 4.0 refuses them: ensureCollection throws an EventStoreSchemaException with found: 'v1' for a 3.x event collection, so the bootstrap fails on a 3.x default pool. 3.x snapshot collections keep working with a warning until you migrate them. The migration is offline: stop every 3.x instance first.

1. Take a backup, and check the room it needs (below).

2. Dry run, without deploying anything. It inspects and plans, and writes nothing:

   import { MongoDBEventStore, MongoDBSnapshotStore } from '@ocoda/event-sourcing-mongodb';
   
   const config = { url: process.env.MONGODB_URL };
   const events = await MongoDBEventStore.migrate(config, { dryRun: true });
   const snapshots = await MongoDBSnapshotStore.migrate(config, { dryRun: true });
   console.dir({ events, snapshots }, { depth: null });

   Review, per collection, blocking (the migration leaves such a collection untouched), gappedStreams, nonCrockfordEventIds (event ids that are not canonical ULIDs, which choose the slower numbering key), warnings and steps, which hold the exact mongosh statements and the locks they take.

3. Stop every 3.x instance.

4. Migrate the events, then the snapshots: await MongoDBEventStore.migrate(config), then await MongoDBSnapshotStore.migrate(config). Both need no Nest application; a connected store also has an instance migrate().

5. Deploy 4.0.

By default migrate() finds every collection named events or <pool>-events (snapshots, <pool>-snapshots) with the unique { streamId, version } index; pools: [undefined, 'tenant-a'] limits it to those pools. A second run reports skip. onProgress reports each step.

What happens to an event collection, in order, under a lease in the catalog:

1. Fence: collMod adds the 4.0 validator. From here on, a 3.x insert fails with code 121.
2. Number: one aggregation numbers the events 1…N in 3.x’s order, (eventDate, _id), and writes the positions with $merge. Where two versions of a stream sort the other way round in that order (two appends in one millisecond, or non-canonical event ids), the stream keeps its version order (ADR 0001 D33). When every event id is a canonical ULID, the aggregation ranks the events along the _id index, which is the same order; otherwise it sorts them by a concatenated key. Either way, two more sorts over the whole collection keep every stream in version order, and above roughly 100,000 events these sorts spill to disk. The step reports its progress only when it starts and when it ends.
3. Index: the unique { globalPosition: 1 } index.
4. Register: the catalog document, with the counter at the highest position. This is the commit point: 4.0 can run from here, and appends continue at N + 1.
5. Clean up: drop the 3.x indexes on eventDate, and remove the eventDate field in batches. 4.0 ignores the field, so unsetEventDate: false defers this, for instance to spread the oplog volume; a later migrate() finishes it.

A snapshot collection loses its latest: null fields, gets exactly one latest flag per stream, on its highest version, then the unique latest_unique index, which replaces the 3.x index on the flag, and is registered.

Interruptions. Every step checks the state first, so a run that stopped (a crash, a lost connection) resumes where it stopped: run migrate() again. The lease of a crashed run expires after 10 minutes; force: true takes it over sooner, when you are sure the other run is gone. The server may still be running the last operation of a run whose client gave up, such as the numbering: check that db.currentOp({ 'command.aggregate': '<collection>' }) shows nothing before you force. A run whose lease another run took over stops before its next step and reports the collection as blocked. Until the commit point, ensureCollection refuses a partly migrated collection (found: 'v1-partial').

Client timeouts. The static migrate(config) leaves socketTimeoutMS and timeoutMS of the config and of the connection string out of its client, because the numbering and the index build are single operations that run for minutes on a large collection. The instance migrate() of a connected store uses the application’s client, with its timeouts: prefer the static form for large collections. lockTimeoutMs (default 10 seconds) bounds how long the fence and the index drops wait for their exclusive collection lock; a step that doesn’t get it reports the collection as blocked, and a later run resumes it. It doesn’t apply to a lease, which blocks at once, nor to the index build.

What blocks a collection: a sharded collection (or one the user can’t tell is sharded), a MongoDB server before 5.0, event ids that aren’t strings, a validator of your own, a missing unique { streamId, version } index, missing privileges, another migration’s lease, or an exclusive lock that isn’t free within lockTimeoutMs.

Snapshots under 4.0. 3.x snapshot collections keep working under 4.0, so the snapshot migration can run while 4.0 appends snapshots: it reads the highest version of each stream it repairs right before it flags it. A snapshot appended in the instant between that read and the flag leaves two flags, on which the build of the unique index fails; run migrate() again.

Streams with gaps in their versions (a stream at versions 1, 2, 3 and 5) keep them. Their next append in 4.0 conflicts with the actual version: load such an aggregate from its envelopes, or append with the version of its last event as the expected version.

Room it needs.

• Time: about one minute per million events on a single-node replica set, of which the numbering takes about 0.6 and the eventDate clean-up 0.25; the dry run takes about 4 seconds per million. Measured with 100,000 events on MongoDB 8.2 in Docker on a busy laptop, a size at which the sorts of the numbering still fit in memory. Collections of a million events and more weren’t measured with this pipeline: their sorts spill to disk, so plan for more, also when the data and indexes exceed the WiredTiger cache.
• Oplog: each event is updated twice, the numbering and the clean-up, which writes about 0.45 to 0.6 GB of oplog per million events. Change streams see both updates of every event. Size the oplog window for it (replSetResizeOplog), or a secondary that falls behind needs a full resync.
• Disk: at least 2.5 times the collection’s storage size, plus the oplog, plus the sort spill of the numbering in the dbPath: about 100 to 150 MB per million events with canonical ids, and 165 to 250 MB when the ids aren’t canonical. Keep at least 500 MB free there on top of that: MongoDB 8.2 refuses to spill a sort, and 7.1 and later refuse to build an index, below that.
• Index build: on a replica set, the unique { globalPosition: 1 } index is committed once every voting member has built it (the default commitQuorum), so a lagging member slows the step down.

Without the Node.js API, for instance for a DBA who runs the stores with ddl: 'none': the package ships migrations/4.0.mongosh.js, which migrates the default pools the same way (mongosh "mongodb://host/app" node_modules/@ocoda/event-sourcing-mongodb/migrations/4.0.mongosh.js). Before it writes, it checks what blocks migrate() (except sharding and privileges) and stops without changing anything. It doesn’t check the state before each step, so run it once, on 3.x collections. migrate() resumes a run of it that stopped once the script’s lease expires, 10 minutes after the script took it, or at once with force: true.