Opt-in second-level aggregate cache for FetchForWriting()

[erdtsieck]

The goal

Make FetchForWriting() cheap for the very common shape where a chain of commands walks over one
aggregate stream, without forcing the aggregate onto an Inline or Async snapshot.

I'd like to add an opt-in, second-level aggregate cache behind FetchForWriting(), in the shape of an
additional IFetchPlanner. This issue is the design + the safety argument; I'm preparing the proposal
and the PR myself, and would like a sanity check on the direction (and on the two small enabling PRs
below) before I go further.

The problem

In a CQRS system built on Wolverine's aggregate handler workflow, one business operation is often a chain
of cascading commands over a single stream. A real example from our claims system — six handlers, one
ProvidedCare stream:

DeterminePrestatie → DetermineWlzDebtor → DetermineDebtor
  → ValidateNationality → DetermineTariff → PrepareForInvoicing

Every hop is its own message, its own session and its own transaction — by design, because each hop has
to be independently retryable. So each hop calls FetchForWriting() again on the same stream.

For a Live-registered aggregate that means every hop reads the entire stream back and folds it from
scratch, even though the previous hop, milliseconds earlier and usually in the same process, had that
exact aggregate in memory and appended two events to it.

The existing options don't cover this:

• Events.UseIdentityMapForAggregates is exactly the right idea, but it caches on the session, and
  each cascading message gets a fresh session. It never spans hops.
• AsyncOptions.CacheLimitPerTenant gives the async daemon a bounded MRU cache of aggregates
  (IAggregateCache<TKey, TItem> / RecentlyUsedCache<TKey, TItem> in JasperFx.Core). Command handling
  has no equivalent.
• Switching the aggregate to Inline does fix the read cost, but it is a much bigger change: a
  snapshot write joins every append transaction, the projection has to be rebuild-safe and serialization-safe,
  and for a large aggregate that is a real trade rather than a free win. It shouldn't be the only answer.

The proposal

A FetchCachedPlan<TDoc, TId> that keeps (tenantId, streamId) -> (aggregate, version) in a bounded
RecentlyUsedCache, and on a hit issues the same two-statement batch the Live plan issues today, with
one predicate added:

statement 1:  select version from mt_streams where id = ?            -- unchanged
statement 2:  select ... from mt_events where stream_id = ? and version > @cachedVersion

then folds those events onto the cached aggregate.

This is not a new mechanism — it is FetchAsyncPlan's existing "start from a snapshot at version N, read
only the events after it, fold" strategy, with the snapshot coming from memory instead of from the
document table. IEventIdentityStrategy<TId>.BuildEventQueryHandler() already takes an optional
ISqlFragment filter, so statement 2 needs no new SQL machinery.

Why this is concurrency-safe

This is the part I most want scrutinised, so spelling it out in full.

The cached version is never trusted. Every fetch still reads the true stream version from
mt_streams and still reads every event after the cached version, in the same batch. Therefore:

• Another node appended between caching and this fetch → those events come back in statement 2 and are
  folded. The result is identical to a cold fetch.
• The cache entry is stale by any amount, or absent, or from a stream that has since been rewritten →
  worst case the delta is the whole stream and you are exactly back to today's Live behavior.
• The optimistic concurrency check on SaveChangesAsync() is unaffected, because
  AppendToStream(..., version, ...) still gets the version that came from the database, never the
  cached one.

A stale cache entry can therefore only cost a normal read. It cannot produce a wrong aggregate and it
cannot weaken the concurrency check. That is the property that makes this safe to do in Marten and not
safe to do a layer up in the message framework, which has no way to verify the version.

The read race is unchanged. The window between statement 1 and statement 2 already exists in
FetchLivePlan today — same two statements, same order, only the event predicate narrows. If anything a
cached plan can improve on it by reusing the begin transaction isolation level repeatable read read only
wrapper FetchAsyncPlan already applies to its non-forUpdate path, giving both statements one snapshot.

The cache is only ever written from a successful read, never from a write. After folding, the plan
stores the result at the version it just verified. Nothing hooks into commit, so there is no "did
SaveChangesAsync succeed" question, a rolled-back transaction leaves nothing behind, and the next hop
simply reads a slightly larger delta. This also keeps the whole thing out of the write path.

The one genuinely new hazard is instance sharing, and it needs an explicit answer: two threads in the
same process fetching the same stream would otherwise receive the same aggregate instance, and a handler
that mutates stream.Aggregate would poison the entry for the next command. So I'd propose:

• Default: cache the serialized snapshot and materialize a fresh instance per fetch. Every caller gets
  its own object, mutation cannot leak, and there is no data race — while still skipping the read and fold
  of N events in favour of one document deserialization.
• Opt-in: share the instance for aggregates that honour the decider contract. That is the same contract
  UseIdentityMapForAggregates already documents (FetchForWriting<T> + SnapshotLifecycle.Inline: in-memory aggregate mutations persisted alongside appended events (regression vs 8.x / between 9.0.0-alpha.1 and 9.0.0-alpha.2) #4439, FetchForWriting + inline snapshot persists in-memory aggregate mutations on top of appended events (snapshot diverges from event rebuild) #4509), so there's precedent and existing wording
  to reuse.

Other correctness details I think are handled, and would like checked:

• Archived streams — statement 1 already reads the stream row, so it can read is_archived in the same
  pass and evict rather than return a resurrected aggregate.
• Stream compacting — compaction only removes events at or below some version, so a version > N delta
  stays correct.
• Keying — (database, tenantId, aggregate type, stream id); the cache lives on the store, never static,
  mirroring AggregationCaching.FindCache<TId, TDoc>(tenantId).
• Bounding — RecentlyUsedCache with a per-tenant limit, same idiom and same name shape as
  AsyncOptions.CacheLimitPerTenant.

Opt-in and non-breaking

Nothing changes unless the flag is set:

opts.Events.CommandCacheLimitPerTenant = 256;   // 0 (default) = no cache, no plan, no behavior change

At zero, no cache plan is registered and FetchForWriting() resolves to exactly the plan it resolves to
today. The cache is a pure optimization — it changes latency, never results — which is why I think a flag
is the right shape here, as opposed to anything that would change failure or retry semantics.

On sizing: a useful cache here is small, not large. When the caller partitions its command processing by
stream id (Wolverine's inferred grouping for event streams does exactly this), the number of concurrently
live streams per node is bounded by the queue count times the per-queue parallelism — a couple of hundred
in our case. So a default in the low hundreds should hit essentially always within a chain.

Enabling PRs

Two small, independent PRs first, both useful on their own:

1. Make StoreOptions.Projections.FetchPlanners public. It already carries the doc comment "Any custom
   or extended IFetchPlanner strategies for customizing FetchForWriting() behavior", and IFetchPlanner
   and IAggregateFetchPlan<TDoc, TId> are already public — as is everything a plan needs. Only the list
   is internal, so today there is no way to supply one. Widening it lets this be prototyped against a real
   workload without a fork.

2. An opt-in marten.fetch_for_writing.events_replayed histogram, tagged by aggregate type and fetch
   plan. Right now there is no way to see what FetchForWriting() costs, which makes both "should I move
   this aggregate to Inline?" and "is the cache actually hitting?" unanswerable. Live records the full
   stream length every time, Async records only the daemon's outstanding lag, Inline records zero — so
   one histogram makes the lifecycles directly comparable.

I have both of these written and tested locally and will open them as separate PRs. Happy to drop or
rework either if you'd rather see this land differently — and equally happy to be told the whole direction
is wrong before I build more of it.

[erdtsieck]

Both enabling PRs are up:

• Make StoreOptions.Projections.FetchPlanners public #5226 — make StoreOptions.Projections.FetchPlanners public (internal -> public only; the test planner sits in Marten.Testing.OtherAssembly so the build breaks if it is ever narrowed again)
• Add an opt-in marten.fetch_for_writing.events_replayed histogram #5227 — the opt-in marten.fetch_for_writing.events_replayed histogram

They are independent of each other and of the cache proposal itself, so either can be taken, reworked or dropped on its own merit. I will not start on the cached fetch plan until there is a read on whether the direction above makes sense.

[jeremydmiller]

@erdtsieck I've actually done a lot of spiking around this in the past couple weeks

[erdtsieck]

@jeremydmiller isn't this fixed by CacheAggregatesForWriting<T>?