Event Store

The event store keeps the events of every aggregate in a stream, EventStream.for(Account, id), whose events are numbered 1, 2, 3, … by their version. Streams live in a pool: the default pool, or a named one (ensureCollection('tenant-1')), for instance per tenant. Each pool is a collection (table) of its own.

Appending events

const envelopes = await eventStore.appendEvents(stream, events, {
  expectedVersion: account.committedVersion, // the version of the stream before the append
  pool: tenantId, // optional, the default pool otherwise
  metadata: { correlationId, causationId, headers }, // optional, see Envelopes
  publish: true, // default; false for imports and migrations
});

An append stores all of its events or none of them, and returns one envelope per event, with its version and its globalPosition.

expectedVersion is the version the stream must have before the append: ExpectedVersion.NoStream (0) for a new stream, or the version of the last event you read. If the stream is at another version, the append throws an EventStoreVersionConflictException and writes nothing. That includes a stream that is behind the expected version: an append can’t leave a gap in the versions.
ExpectedVersion.Any appends after whatever the stream holds. When concurrent appends keep taking the versions, it makes up to ANY_MAX_ATTEMPTS (16) attempts with a short random backoff, with the same event ids. Under sustained contention on one stream it can still end in a conflict, so retry such a conflict in your application.
An empty append returns [] without touching the database or the publishers.
Invalid arguments are rejected before any I/O: an InvalidAppendOptionsException for the options, an InvalidEventMetadataException for the metadata, an InvalidEventEnvelopeException for pre-built envelopes that don’t continue the stream or values over the limits.
Other failures throw an EventStorePersistenceException. Its outcome is 'not-persisted' when nothing was written, and 'unknown' when the events may have been stored. Retrying with the same numeric expected version is safe in both cases: if the first attempt was stored, the retry conflicts.

Match these errors on their code and fields, with isEventSourcingError(error, EventSourcingErrorCode.EventStoreVersionConflict), not on their message.

The 3.x form, appendEvents(stream, aggregateVersion, events, pool?), passes the version of the aggregate after the append: the expected version is aggregateVersion - events.length. It is deprecated, emits a DeprecationWarning (OCODA_ES_POSITIONAL_APPEND) once per process, and will be removed in 5.0.

Pre-built envelopes

An append can store EventEnvelopes instead of events, for instance to import events from another store. They keep their eventId, occurredOn, correlation id, causation id, headers and event version; the store assigns their global position. They need a numeric expected version, the aggregate id of the stream, and the versions that continue it: the item at index i has the version expectedVersion + 1 + i. Their eventId must be an EventId and their occurredOn a valid Date, so revive envelopes parsed from JSON first. Events and envelopes can be mixed in one append.

await eventStore.appendEvents(stream, importedEnvelopes, { expectedVersion: ExpectedVersion.NoStream, publish: false });

Publishing

Once the events are stored, the store hands their envelopes to the EventBus, which passes them to the event publishers and subscribers. See Event Pub/Sub. Publishing never makes the append fail: the events are stored at that point. publish: false skips it.

Reading a stream

getEvents(stream, filter?) yields the events of a stream in batches, getEvent(stream, version, pool?) returns one. Both deserialize the events with their serializer.
getEnvelopes(stream, filter?) and getEnvelope(stream, version, pool?) return the envelopes, with the metadata and the global position, without deserializing.

The filter takes fromVersion, direction (StreamReadingDirection.FORWARD or BACKWARD), limit, batch and pool. Every read of a pool whose collection doesn’t exist throws an EventCollectionNotFoundException.

Reading all events

Every event gets a global position when it is stored: a bigint, per pool, starting at 1n. The events of one append have consecutive positions, and a position is never reused. readAll reads a pool across streams, in the order of these positions:

let checkpoint = await loadCheckpoint(); // 0n the first time

for await (const envelopes of eventStore.readAll({ fromPosition: checkpoint + 1n, batch: 100, pool })) {
  for (const envelope of envelopes) {
    await project(envelope);
    checkpoint = envelope.metadata.globalPosition;
  }
  await saveCheckpoint(checkpoint);
}

fromPosition is inclusive, like fromVersion: resume after a checkpoint with checkpoint + 1n. batch is a positive integer (100 by default); anything else throws a RangeError. Store the checkpoint as a string (checkpoint.toString()) where JSON can’t hold a bigint.

What a position guarantees depends on the store’s capabilities.globalOrder:

'gap-safe': once readAll yielded a position, no event at or below it commits later. A reader that resumes after its last position never misses an event, even while other appends are running. The in-memory store, PostgreSQL, MariaDB and MySQL on InnoDB, and MongoDB replica sets are gap-safe, on the server the appends run on: see Failover and replicas.
'best-effort': positions strictly increase, but an event can still commit below a position that was already read, while appends run. Read again from an older position, or wait until the writers are done, to be sure to see every event.

Every built-in store implements readAll. It replaces getAllEnvelopes({ since, until }), which is removed: to read a period, filter the envelopes on metadata.occurredOn.

Failover and replicas

A position is never handed out twice by the server the appends run on, the primary. A database that loses its newest commits can hand it out again:

An asynchronous failover. The replica that becomes the primary can lack the last appends of the old one, which readers may already have read. Its counter is behind their positions, so it hands those positions out again, to other events, and a reader whose checkpoint is past them never reads those events. The events the reader did read are gone. PostgreSQL streaming replication and MariaDB and MySQL replication are asynchronous unless you configure them otherwise, and the stores can’t tell that a failover happened.
A restore of a backup, or to a point in time, takes the counters back the same way, on every database.

On a MongoDB replica set, a failover can’t take back a position that a reader has seen: appends commit with w: 'majority' and readAll reads with readConcern: 'majority', so a reader only sees appends that a majority of the members have written, and an election keeps those.

To fail over without losing commits that readers may have seen, replicate synchronously:

PostgreSQL: synchronous replication (synchronous_standby_names, with synchronous_commit = on), and promote only a standby that has confirmed every commit.
MySQL: semi-synchronous replication with rpl_semi_sync_source_wait_point = AFTER_SYNC, the default, so a commit becomes visible on the primary only once a replica has acknowledged it.
MariaDB: semi-synchronous replication with rpl_semi_sync_master_wait_point = AFTER_SYNC. MariaDB’s default, AFTER_COMMIT, lets other sessions see a commit before a replica has it.

Semi-synchronous replication falls back to asynchronous when no replica acknowledges in time (rpl_semi_sync_source_timeout or rpl_semi_sync_master_timeout), so a failover after such a fallback can still lose commits.

After an unplanned failover without synchronous replication, and after any restore, check the readers. Save the eventId of the last envelope with each checkpoint, then read readAll({ fromPosition: checkpoint, batch: 1 }): if its first envelope isn’t that event at that position, the reader has read events that the database lost, and their positions may now belong to other events. Rebuild what the reader built, for instance by replaying its read model from the first position.

Read from the primary. The 'gap-safe' claims and their tests are about reads from the server the appends run on. Don’t send the event store’s reads to replicas through a proxy that splits reads from writes: on MariaDB and MySQL, a lagging replica makes readAll skip events (see MariaDB and MySQL). On MongoDB, a readPreference other than primary sends readAll to secondaries, which isn’t tested.

Capabilities

eventStore.capabilities lists the optional guarantees of a store, final once it is connected:

Capability	Meaning
`atomicAppend`	An append stores all of its events or none.
`headers`	The store persists headers; without it, an append with headers is rejected.
`globalOrder`	`'gap-safe'` or `'best-effort'`, see Reading all events.

The built-in stores declare:

Store, on	`atomicAppend`	`headers`	`globalOrder`
In-memory	`true`	`true`	`'gap-safe'`
PostgreSQL	`true`	`true`	`'gap-safe'`
MariaDB, on MariaDB or MySQL (InnoDB)	`true`	`true`	`'gap-safe'`
MariaDB, on Galera (MariaDB Galera Cluster, Percona XtraDB Cluster), MySQL Group Replication in multi-primary mode, or TiDB	`true`	`true`	`'best-effort'`
MariaDB, on Vitess (PlanetScale)	`true` with `transaction_mode` `SINGLE` or `TWOPC`, `false` otherwise	`true`	`'best-effort'`
MongoDB, on a replica set	`true`	`true`	`'gap-safe'`
MongoDB, on a sharded cluster	`true`	`true`	`'best-effort'`
MongoDB, on a standalone server	`false`	`true`	`'best-effort'`
MongoDB, on DocumentDB, Azure DocumentDB or an instance-based Amazon DocumentDB cluster (4.0 and later)	`true`	`true`	`'best-effort'`
MongoDB, on FerretDB, Azure Cosmos DB for MongoDB or an elastic Amazon DocumentDB cluster	`false`	`true`	`'best-effort'`

The stores detect what they connect to. The MariaDB store tells MariaDB from MySQL, and recognizes Galera, Group Replication in multi-primary mode, TiDB and Vitess, where the lock on a pool’s counter doesn’t order the commits of the whole deployment. The MongoDB store detects its topology, and the engines that speak MongoDB’s protocol without being MongoDB. Both log a warning once per process and kind of deployment when they connect to one of these; see MariaDB and MySQL and MongoDB. Galera, Group Replication, TiDB, Vitess and the MongoDB-compatible engines aren’t tested. Amazon Aurora MySQL isn’t told apart from MySQL: it gets 'gap-safe', which is expected to hold but isn’t tested.

resolveCapabilities(store.capabilities) fills in the defaults of a store that leaves a flag out: atomicAppend: true, headers: false, globalOrder: 'best-effort'.

Limits

EVENT_STORE_LIMITS holds the largest values every store accepts, checked before any I/O: stream ids, aggregate ids, event names, correlation ids and causation ids of at most 255 characters, and headers of at most 8 KiB of JSON.

The base class and the driver methods

EventStore implements the store contract once, for every store. Its template methods, appendEvents, getEvent and getEvents, validate, serialize, check the version, retry ExpectedVersion.Any, stamp the global positions and publish. A store implements the driver methods, which do the I/O. They are abstract, so TypeScript reports the ones a store lacks:

Method	Does
`connect()`, `disconnect()`	Open and close the connection. The capabilities are final once `connect()` resolved.
`ensureCollection(pool?)`, `listCollections(filter?)`	Create the collection of a pool unless it exists; list the collections.
`getStreamVersion(stream, pool?)`	The version of the last event of a stream, `0` when it has none.
`getEnvelope(stream, version, pool?)`, `getEnvelopes(stream, filter?)`	Read a stream, with the global positions.
`readAll(filter?)`	Read a pool in the order of the global positions.
`persistEvents(envelopes, target)`	Store the envelopes of an append, all or nothing, and assign their positions.

The store gets a context when it’s constructed, new Store(context, options): the event map and the publisher, which is the EventBus. The driver methods don’t need the event map: the base class serializes and deserializes.

An append runs these steps:

Validate the options and the metadata. An empty append returns [] here, without any I/O.
Validate pre-built envelopes, serialize the events into envelopes (one monotonic id factory per store) and check the limits, still before any I/O.
Read the stream’s version with getStreamVersion. A numeric expectedVersion that doesn’t match throws a conflict with the actualVersion, and writes nothing. ExpectedVersion.Any numbers the envelopes after that version. A failure of this read is outcome: 'not-persisted'.
Call persistEvents, which returns one of:
- { status: 'committed', positions }: one bigint position per envelope, consecutive and in order. The base class stamps them on the envelopes it returns, and logs positions that break the contract instead of throwing, because the events are stored.
- { status: 'conflict', actualVersion?, cause? }: another append took one of the versions, as the unique (stream, version) key reported, and nothing was stored. ExpectedVersion.Any goes back to step 3, up to 16 attempts with a jittered backoff; otherwise the append throws an EventStoreVersionConflictException.
Any other failure is thrown. An EventStorePersistenceException passes through with the outcome the store gave it; any other error becomes one with outcome: 'unknown', because the commit may have happened.
Publish the envelopes with the context’s publisher, unless publish: false. A failure is logged, never thrown.

A store must not override the template methods: the module checks it when it creates the store, and an override (a method of a subclass, or a class field) fails the bootstrap with an InvalidEventStoreImplementationException that names the methods. This includes a 3.x store that overrides appendEvents: see Migrating a 3.x store. To decorate the appends of a store, for tracing for instance, extend it, override persistEvents and call super.

Custom stores

A store implements the driver methods, such as persistEvents and readAll, and the EventStore base class does the rest. See Custom Event and Snapshot Stores.