Appending Events

Starting a stream

// Fisher assigns the id. StartStream returns a StreamAction — its Id is the stream's identity.
var stream = session.Events.StartStream<Order>(new OrderPlaced(…), new OrderLineAdded(…));
var streamId = stream.Id;

// Or you name it
session.Events.StartStream<Order>(orderId, new OrderPlaced(…));

// Without an aggregate type
session.Events.StartStream(streamId, new SomethingHappened(…));

Appending to an existing stream

session.Events.Append(streamId, new OrderShipped(DateTimeOffset.UtcNow));
session.Events.Append(streamId, evt1, evt2, evt3);

await session.SaveChangesAsync();

Nothing is written until SaveChangesAsync. Events commit in the same transaction as documents, patches, raw SQL commands and inline projection writes.

How the append works

Fisher uses QuickAppend — direct INSERT statements, no stored procedures. Version numbers are assigned client-side from the stream's current version, read inside the write transaction under BEGIN IMMEDIATE.

The sequence numbers SQLite assigns come back via a trailing SELECT by stream and version range, where Marten uses a bulk function and Polecat OUTPUT … INTO. That read-back is what supplies the seq_id a tag row is keyed by.

Optimistic concurrency

session.Events.Append(streamId, expectedVersion: 5, new OrderShipped(…));

A losing write throws EventStreamUnexpectedMaxEventIdException at SaveChangesAsync.

session.Events.AppendOptimistic(streamId, new OrderShipped(…));   // reads the version for you

The exclusive methods fail where the siblings wait

WARNING

AppendExclusive, FetchForExclusiveWriting and WriteExclusivelyToAggregate are the optimistic methods on Fisher.

Marten takes an advisory lock and Polecat a row lock, so a competing session waits its turn. SQLite has no row locks and one writer per file, so the equivalent would mean holding a BEGIN IMMEDIATE open from the fetch until SaveChangesAsync — blocking every other writer in the process for as long as the caller holds the session.

The safety property is unchanged: the version guard still runs inside the write transaction, so there is no lost update. What differs is that a loser gets EventStreamUnexpectedMaxEventIdException instead of waiting.

FetchForWriting

The command-handling shape: fetch, decide, append, commit.

var stream = await session.Events.FetchForWriting<Order>(orderId);

if (stream.Aggregate is { Shipped: false })
{
    stream.AppendOne(new OrderShipped(DateTimeOffset.UtcNow));
}

await session.SaveChangesAsync();

Or with the callback form:

await session.Events.WriteToAggregate<Order>(orderId, stream =>
{
    stream.AppendOne(new OrderShipped(DateTimeOffset.UtcNow));
});

FetchForWriting folds the stream on every call, whatever the aggregate's projection lifecycle. That is deliberate and differs from FetchLatest, which reads an Inline aggregate's projected document: the two ask different questions. FetchLatest reports current state, while this is the read half of a read-modify-write whose guard is the stream's version — so folding the stream is what the version it hands back has to agree with.

WARNING

Fisher tracks pending streams in a dictionary keyed by identity, where Polecat uses a list. So FetchForWriting reuses an already-tracked StreamAction rather than constructing a fresh one — replacing the dictionary entry would silently drop events an earlier Append had queued for the same stream in the same session.

Caching the aggregate between fetches

A hot stream re-folds its whole history on every command. CacheAggregatesForWriting<T>() keeps recently fetched snapshots in a node-local cache, so a later fetch folds only the events after the cached one:

opts.Events.CacheAggregatesForWriting<Order>();          // off for every type by default
opts.Events.CacheAggregatesForWriting<Order>(sizeLimit: 5000);

The cached snapshot is only ever a baseline. The stream's version and every event after the cached one are still read on every call, and the optimistic concurrency assertion on append is untouched — so a stale entry costs a larger fold, never a wrong aggregate and never a suppressed concurrency failure. Turning it on is unobservable except in latency.

TIP

This is worth more on Fisher than on Marten or Polecat. There the cache removes a snapshot load; here it removes the fold itself, because FetchForWriting folds by design. It is opt-in per aggregate type because the win is proportional to how often one stream is fetched for writing — real on a hot aggregate under load, and only overhead on one written once.

The contract is JasperFx.Events.Fetching.IAggregateWriteCache, shared with Marten and Polecat, so a consumer targeting more than one store configures caching once. Supply your own implementation — an adapter over a shared IMemoryCache, say — through opts.Events.AggregateWriteCaching.Cache.

WARNING

A cached baseline is derived state, and event rewriting does not reach derived state. Masking or overwriting an event body below a cached baseline leaves that baseline holding what the old body produced, exactly as it leaves an already-written snapshot, document or flat table holding it. The cache is node-local by design — masking on one node could not evict another's — so this is the same caveat masking already carries, one place further along. Leave an aggregate whose history you rewrite unenrolled, or treat a rewrite as requiring a restart of the processes holding it.

By natural key or strong-typed id

var stream = await session.Events.FetchForWriting<Order, OrderId>(orderId);
var stream = await session.Events.FetchForWritingByNaturalKey<Order>("INV-2026-0042");

TIP

Where the two readings coincide — a string id on a string-identity store — the stream identity type wins, and the string is read as the stream key. Which reading applies must not depend on whichever aggregate types happen to declare a natural key. FetchForWritingByNaturalKey is the unambiguous spelling.

See Natural Keys.

By tags

var boundary = await session.Events.FetchForWritingByTags<Basket>(query);

See DCB.

FetchLatest and ProjectLatest

var order = await session.Events.FetchLatest<Order>(orderId);

ProjectLatest folds the session's pending events on top of the committed state — see ProjectLatest.

Event metadata

The session's correlation id, causation id, user name and headers are copied onto each event that does not already carry its own, each gated on its Enable* option. See Event Metadata.

Cross-tenant appends

session.ForTenant("globex").Events.StartStream<Order>(id, new OrderPlaced(…));

The append path needed nothing for this: the planner already writes the stream action's tenant rather than the session's. See Writing across tenants.

An append observer

opts.Events.AppendObserver = events => { … };

Fires after commit, so "everyone can see this now" is true when it runs — which is why it does not fire for an enlisted session, where Fisher is not told when you commit.

Versions and inline projections

An inline projection needs its events to already know their versions, but Fisher normally assigns those inside the write transaction. So the version is read early, outside the lock, for the projection to read.

TIP

That is not a weakened guard. The same versions are re-derived inside the transaction and the optimistic concurrency check still runs there, so a racing writer still fails the commit. The early pass exists only to give projections something to read.