V7 Event Store Enhancement Brainstorming -- Please join the discussion!

[jeremydmiller]

at @oskardudycz 's recommendation, this single markdown file is meant to capture comments and discussions about possible enhancements to the Marten event sourcing functionality in the V7 release timeframe. All feedback is welcome here!

[elexisvenator]

Had some thoughts about this, If the streams table was updated to include a "first_event_sequence_id" and "last_event_sequence_id" then these could be used with live projections to drastically cut back on the number of partitions scanned.

[jeremydmiller]

That's an interesting suggestion. I'll have to think on that one a bit. Might say that's an opt in behavior

[oskardudycz]

@elexisvenator could you expand on why would it impact the number of partition?

[jeremydmiller]

With @elexisvenator 's suggestion, doing a live aggregation of any kind could use the cached "lowest" & "highest" sequence numbers for the stream as part of the WHERE in the event fetching. That would let postgres limit the ranged partitions to look through.

[oskardudycz]

That's just decreasing the issue. For bigger databases when you need to query more than one partition then the performance will be highly degraded. So you should do all you can to just query a single partition. Scan between partitions is really costly in Postgres.

[elexisvenator]

it can get costly yes, but a low number of partitions wont make a huge impact. (eg <20, depending on each partitions size). The other challenges with this approach are:

• managing the provisioning of new partitions
• Queries based on timestamps - we can be smarter here with some form of prequery because we know that timestamps increase as sequence id increases but out of the box postgres will have a bad day.
• Unique constraints on event id, these simply cannot exist in any partitioned system, so will affect the partition-on-archived approach as well

[elexisvenator]

Another approach would be to slice on tenant, but this has similarly headache-inducing implications.
The only was this would work is to give every tenant their own highwatermark and treat them like they are in separate dbs even though they are conjoined.

This approach would benefit tasks like offboarding and even potential performance management (eg move hot tenants to their own hardware)

Overall, there is probably more to gain from partitioning on archived or eventsequence id

[jeremydmiller]

Slicing it on the tenant would be awful for the daemon, that's why I was pooh-poohing that before. Like you say, you'd need a separate sequence and high water mark for each. That's a bit more complexity

[elexisvenator]

I feel this would encourage the use of is_archived far more than it is currently being used by marten consumers. Might need to consider implications for things like projections.

For example, I currently only use is_archived as a soft-delete that will eventually be hard deleted (as part of restreaming usually). If it becomes a "move your old stuff here" feature then I will have challenges with projection consistency. In particular I would want to make sure that a new projection I create is backfilled with archived values as well, or at least configure my projection so that can be possible.

[jeremydmiller]

My thoughts as well. That's mostly inspired by Eventuous's "hot / cold storage" idea

[oskardudycz]

See my write-up on the log compaction process:

• https://event-driven.io/en/gdpr_in_event_driven_architecture/#log-compaction
• https://www.youtube.com/watch?v=gG6DGmYKk4I
• https://www.eventstore.com/blog/keep-your-streams-short-temporal-modelling-for-fast-reads-and-optimal-data-retention

And EventStoreDB scavenging process docs: https://developers.eventstore.com/server/v23.10/operations.html#scavenging-events

TLDR, we should have a formal archival process that works as a background process. It has to be either on-demand or semi-automatic. We should ensure that all archived events are published and processed via async daemon.

The detaching partition should be the last step, first people should be able to move the events somewhere else depending on the strategy.

I think that eventually we should support all scenarios I described. The basic building blocks should be part of Marten. So archival, also partial archival of stream, tombstoning, and hosted service giving option to plug in custom archival logic. The more advanced and precise scenarios should be part of CritterStackPro.

[ericgreenmix]

Big fan of this feature.

We should ensure that all archived events are published and processed via async daemon.

Yes please. This is currently an issue with my applications being able to use archival. The async daemon doesn't pick up any event that was just archived, so I can't move the async projection into a new state (or delete that document) on that initial archival event.

[elexisvenator]

Being able to apply this to inline projections as well would be invaluable (though that last step of consistency could be a challenge).

[jeremydmiller]

It's mostly for the inline projections. I think we'd fudge the lines between inline and async here. With a potential client that didn't pan out, I was calling this the "catch up mode". I think we have to encourage folks to use FetchForWriting() everywhere instead of LoadAsync() or AggregateStreamAsync() unless you're doing time travel.

[elexisvenator]

This is great to see, and keen to try it though I dont know how to deal with the per-tenant db scaling issues.

Context:
In one aws region we have ~300 tenants. At this time our app uses conjoined tenancy on an "r6g.xl" instance size. To split them up would either mean creating ~300 rds instances. By default the instance limit is 40 per region, and even ignoring that the cost per instance would skyrocket as they would be forced to be massively over-provisioned instances (smallest viable size is "small", due to below issue).

The alternative is to group tenants together onto a smaller set of instances, say 50 per rds instance. This introduces a new problem where the memory overhead of connections starts to add up and consume all memory in the database. The memory consumed (by default) would be the mem allocated per connection (8mb) * max connections by app (100) * tenants in server (50) * nodes in app (2-4), resulting in >80gb memory used in this situation. This puts scaling limits on your application and forces the db to be over-provisioned.

A possible workaround is Amazon RDS Proxy. Previously this wasn't a good solution as the postgres support was always several major versions behind current, but now it seems to be in sync with postgres 16. It only really helps with the node scaling and not the tenant scaling as different dbs on the same server require a different connection. There are a bunch of other restrictions that will necessitate testing though.

This feels like a "postgres" or "aws" problem rather than a "marten" problem, but it would be interesting to see how other clients tackle it given this work is sponsored by clients.

[oskardudycz]

I think that one option is to also add an option per schema or mention in other places partitioning per tenant or some other way to allow other shared contexts (or mixed option).

I think that a database per tenant is useful when:

• we have a limited (and short) number of tenants,
• the legal requirement to prove data isolation,
• ops team to maintain it.

When databases number grows, as you described it gets extremely hard and costly from the DevOps perspective to maintain that. I wouldn't consider a database per tenant as a viable solution for a big SaaS platform.

[elexisvenator]

Ideally both?

We have a version of this heavily based on Oskar's example. By default the subscription is cold. If we want to rebuild or make a new one and find that a cold subscription is undesirable, we add a filter at the start of the subscription based on event timestamp and let it catch up skipping older events.

[jeremydmiller]

I think @oskardudycz is going to agree with that. He's brought that up before.

[oskardudycz]

Yup, see: https://github.com/JasperFx/marten/pull/2888/files#r1442806806

See also my write-up on how ESDB subscription works: https://event-driven.io/en/persistent_vs_catch_up_eventstoredb_subscriptions_in_action/. We should enable at least the same set of basic functionalities.

[elexisvenator]

In addition to my comments above:

IEventSession - love the idea, though personally wouldnt use it due to current patterns we have

Poor man's load balancing - Makes sense, especially the multitenant bit. Personally I'm waiting for the pro version ;)

FetchForReading - like the consistency of the apis for this, and that it can handle the future versioning/zero downtime magic.

Async FetchForWriting - 🥳 To me this feels critical for marten to remain performant in a large (volume) complex (variety of events/projections) system, as it allows for nearly all use cases for inline projections to be replaced with async.

[Hawxy]

I would exclude subscriptions by default and handle them in a different way. At least for us, the only scenario we'd need to "rebuild" a subscription is if there's a catastrophic failure of the downstream system (ie a third party provider that sends notification has an outage), in which case we'd like the ability to reprocess the events from the last x hours.

[oskardudycz]

We should be able to enable users to manage the sequence from which they subscribe. So they should be able to tell "sent me the events from this position". We used such procedures for, e.g. disaster recovery strategy when our messaging system died or the retention policy removed messages. It's also useful when we have a bug in the system, and we need to republish messages.

[oskardudycz]

I think that there are actually two things here. From what I see, async session listeners are usually used when people want to publish the changes to read models outside the world, so as I described in my blog article: https://event-driven.io/en/publishing_read_model_changes_from_marten/.

Some people are fine with just forwarding the change to the read model; this scenario is fine for building a local copy of data between modules.

The other scenario is enriching events. For cases like "shopping cart confirmation", external processes can be triggered (e.g. "issuing invoice"). Internal events should be granular so they won't contain all the information. Some people want to enrich it either based on the updated read model or with an aggregating stream and send new events (so as I explained in https://event-driven.io/en/internal_external_events/).

Those two scenarios don't necessarily have to be implemented the same way. For the read model propagation, we may use logical replication, so use it as I prototyped here: https://github.com/oskardudycz/PostgresOutboxPatternWithCDC.NET.

For events enrichment, I think that we allow appending new events in the projections/subscriptions, but then defer calling external systems with projections based on the enriched events.

I also believe that having support for that will enable lightweight saga processing.

The biggest question here is what will be part of Wolverine and what will be part of Marten. I'm voting that the stuff I mentioned should be part of vanilla Marten. The logical replication part could be a plugin (could be part of CritterStackPro).

[oskardudycz]

Sidenote, it'd be easier to comment if the auto-formatting in Rider would be disabled as it's randomly breaking lines by the characters limit.

[oskardudycz]

I think that we should also give the possibility to scan per tenant and stream type. This could speed up a lot of processing, but of course, it would mean that we don't have global ordering between partitions. Still, in many cases this should be good enough to speed up performance and allow easier database size management.

[oskardudycz]

Personally, I don't see that as super useful. Imho, it adds another layer of indirection.

[ericgreenmix]

Agree on this

[oskardudycz]

I think that maybe it'd be better to treat that as an internal option for projection usage. Making that public would make the number of permutations even higher.

[jeremydmiller]

We've just got to have the capability is all. It's actually not going to be that big of a deal mechanically. Really just an enabler step

[oskardudycz]

I see some value in that, but I see that as an edge case. In general, I'm against promoting the snapshot usage, so I'm not super positive about that. I'm not sure how many people really need such performance optimisation. It creates a pit of further issues like migrations, upcasting inline projections, and managing stuff being out of sync.

[jeremydmiller]

I've read everything you've had to say about this, but don't reach the same conclusions. I think we're just about to invest in a whole lot of functionality in v7 that I think will address the migration problems with at least single stream aggregations. And besides, we've already got requests for this on the Wolverine + Marten side of things.

[oskardudycz]

Sure, I understand why people want it, but we should start limiting the use cases we don't want to support. This is one of such features that, IMHO, will create more issues than benefits in the long term, both for users and us. I think that we should encourage people to solve that stuff differently.

Could you expand on what migration problems it will solve and how?

[oskardudycz]

Would it still be based on the async daemon id? So deployment would need:

• new document class,
• new projection,
• new versioned API?

What if one would like to reuse the same document when:

• the structure didn't change, but there was a bug in apply method,
• Did they provide a new version of the document that's backwards compatible? Would they need to type document alias?

I think that we should give also an option to run it in the same node. So deploying a new version of the application that can handle both the old projection and the new one. That's needed for monoliths.

[elexisvenator]

As far as I can see, versioning only makes sense if:

• The document remains the same shape
• The changes to the apply methods would result different content in the document (eg adding an apply for an event that is already in use)
• OR: the devs have decided that a projection needs to be rebuilt for some undetectable reason (maybe they did a db restore or some old bug caused consistency problems)

If the document changes shape, then versioning would either require returning a different document type based on whether the new version has caught up (not a thing that C# supports) or migrating the old document shape to the new one which is worse because the resulting document is at best inaccurate at worst invalid until the rebuild is complete anyway. It would be better to just have both deployed so that developers could land a change that switches their business logic to the new one once it is ready.

Finally, there are some situations where you would want to change the projection, but not rebuild/version the projection:

• Adding an apply for a new event that is not yet in your system.
• Removing an apply for an event that has been removed from your system (eg through restreaming, or versioning to another supported event)

[oskardudycz]

I disagree that using the same document is the wrong way of doing migration.

The biggest benefit of having a weak schema (as we have with JSON) is having non-breaking migration. Some changes require a big revamp of the schema, and for those, it's better to have new documents, but for smaller ones like new fields, it's perfectly fine to add a new field as nullable keeping it backwards and forward compatible.

[elexisvenator]

I agree that adding a new field with a sensible default value can be done with the same document, though only when that value doesnt need to be backfilled to be accurate... unless you are ok with it being inaccurate until the backing projection table is rebuilt... so many caveats

[oskardudycz]

Do you think about keeping it as sharding, as I described or adding also something more to that? The presented approach assumed that each async daemon instance has a different set of projections. It could be useful to think if we could give option to specify the projection set of active projections through the environment variables or config, then one can use the same artefact (e.g. docker image) but provide different settings to decide on what to handle.

[oskardudycz]

Would new daemon be a new process or new hosted service sharing the host?

[oskardudycz]

Could you expand on the use case? I'm having a hard time understanding the need for that.

[elexisvenator]

Example wolverine code I ended up writing yesterday:

public record GetGroupResult(GroupId Id, string Name, IReadOnlyList<string> Rules);
public static class GetGroupEndpoint
{
    public static async Task<Group?> LoadAsync(
        string groupId,
        IDocumentSession session,
        ILookupService lookupService,
        CancellationToken token)
    {
        return await lookupService.LookupGroupIdForRead<Group>(session, groupId, ProjectionLifecycle.Live, token);
    }

    [Tags("Groups")]
    [WolverineGet("group/{groupId}")]
    public static GetGroupResult Get([Required] Group group)
    {
        return new GetGroupResult(group.GroupId, group.Name, group.Rules);
    }
}

Where the lookupService is implemented as:

public async Task<TAggregate?> LookupGroupIdForRead<TAggregate>(
    IQuerySession session, 
    string group, 
    ProjectionLifecycle aggregateLifecycle,
    CancellationToken token = default) where TAggregate : class
{
    var lookup = await LookupGroupId(session, group, token);
    if (lookup is null)
    {
        return null;
    }

    if (aggregateLifecycle == ProjectionLifecycle.Live)
    {
        return await session.Events.AggregateStreamAsync<TAggregate>(lookup.Value.streamKey, token: token);
    }

    return await session.LoadAsync<TAggregate>(lookup.Value.streamKey, token);
}

public async Task<(GroupId logicalId, string streamKey)?> LookupGroupId(
    IQuerySession session, 
    string group, 
    CancellationToken token = default)
{
    if (!GroupId.TryBuild(session.GetTenant(), group, out var logicalId))
    {
        return null;
    }

    // TODO: Cache this as needed
    var streamKey = await session.Query<IdLookupProjection>()
        .Where(l => l.LogicalId.Equals(logicalId.Value))
        .Select(l => l.StreamKey)
        .FirstOrDefaultAsync(token);

    if (streamKey is null)
    {
        return null;
    }

    return (logicalId, streamKey);
}

Other ways I could have done this:

• Use FetchForWriting and be scared of optimistic concurrency messing with a get that should have no side effects
• Use reflection into StoreOptions to do the same checks that FetchForWriting does
• Create multiple service methods for each aggregate type.

This way is just neater, and removes the need for users to know the difference like they dont need to know when using [Aggregate]

[oskardudycz]

Same comment as above: https://github.com/JasperFx/marten/pull/2888/files#r1442796105

[oskardudycz]

Other ideas that we should include in considerations:

• give users possibility to include their own event type mapping policy (e.g. by strict namespace). That would require changes potentially breaking, as we have event type mapping distributed around the whole codebase, but I think that this is must-have for monoliths,
• potentially adding upcasters,
• open telemetry.

[ericgreenmix]

This would be an awesome feature. This would make async projections even more usable in Marten, as having a clean way to hook into async projection updates is something I have wanted to be able to do a number of times.

[elexisvenator]

(Tagging a random line near the bottom so that discussion threads are possible)

Additional possible idea: Remove the need for multi-stream projections to use the same ID type as is configured for the event store.

Reasoning here is that if you use guids for stream keys and you have to combine streams in some way for a multi-stream projection, there is a good chance that the resulting grain of the projection is not going to be a guid.
The only reason I am not using guid stream keys for events is so that multi-stream projections can support string ids. The limit feels arbitrary and I suspect is not their as intentional design.

[jeremydmiller]

I don't think that is a limitation, or I don't know why you think that exists. You're limited by the types that can be used as an identity.

Or did you mean the single stream projections?

[AlexZeitler]

Would be great to have it as I was building workarounds for this several times.

[AlexZeitler]

Having support for this in the free version would be great.

[jeremydmiller]

That's what I meant by the "poor man's load balancing". It will be in the free core, but it'll be less capable than the paid model. We do need to make a living here man!

[ElanHasson]

Excuse my lack of understanding here, I am new to the platform.

I imagine async processing, refers to inline vs async projections processing.

Does the daemon transaction you're referring to mean the upsert+event publishing?

Is aync processing doing 1 txn per identity/grouping?

[ElanHasson]

This raises the question of if the high water mark and the progression checkpoint should be dimensioned by $group.

Is it possible to have a distinct sequence per $group?

$group here refers to what you're partitioning on. As sharding is typically highly dependent on your use case and how it's accessed.

[jeremydmiller]

We'd have to do that, yes. That's also why the flexible sharding strategies won't happen in the 7.0 release though too:(