Ordering guarantees per entity

[arnarg]

I need to process events for an "entity", and these events are just metadata for said entity. There are "entity creation", "entity modification" and "entity deletion" event types, all going on the same subject (per entity ID). The processing involves creating some resources in a cloud environment so can take up to 10-20 seconds but the destruction can take up to a minute.

In order to add some scalability but still strict ordering per entity ID I have constructed the following idea:

• A stream capturing the subject APPLICATION.entity.* and has a re-publish rule with source APPLICATION.entity.* and destination APPLICATION.q.entity.{{partition(100,1)}}.{{wildcard(1)}} to enable partitioning by the entity ID.
• Another stream with retention of WorkQueuePolicy captures subject APPLICATION.q.entity.*.*.
• In that second stream 100 consumers are created, each only specifying a certain number from 0-99 in the partition position and having MaxAckPending of 1.
• Then each instance of the application, written in go, creates 100 goroutines, 1 for each consumer.

Bonus idea: Adding AllowRollup on the second work queue stream and publish the deletion events with Nats-Rollup header so that any unprocessed events for the entity gets removed. But I haven't verified that this would work.

This gives me the ordering guarantees per entity but still some parallel processing happening. I however saw some problems with this approach.

• Each instance of the application can be quite overloaded as I would like to run fewer parallel goroutines in the application itself.
• We can only process 100 entities in parallel. The number of partitions can of course just be increased but that only makes the previous issue worse and there is still a set limit, although higher.

Is there any way of achieving my goals without these issues. I do not want to have to run many instances of the application, 1 instance should be able to process all of them but scaling it to more instances should be faster. I don't want keep a huge amount of state myself, ideally.

Thanks in advance!

[bruth]

Hi @arnarg, great question. One baseline assertion I want to make is that given a stream and a consumer with a single subscription, you have guaranteed ordered processing per entity (barring redeliveries). Of course the events per entity are interleaved, but relative order per entity is maintained.

How I am reading your question is more like you want to achieve some degree of concurrent processing of entity events? Have you benchmarked how fast a single stream and consumer can process these events to get a baseline? The one issue with the 100 goroutines within the application is that the degree of parallelism you will achieve is dependent on how many CPU cores you have. With the overhead of context switching, it will likely be more performant and use less memory if you have fewer. But again, I think getting a baseline for one should be a starting point.

If you determine you need to scale out, I would recommend to define partitioning as a subject mapping and have N corresponding streams for each partition. Publishers can still publish to APPLICATION.entity.<id>, but the server will automatically map the subjects to their equivalent partitioned subject, e.g. APPLICATION.entity.0.<id> and be received by a stream APP_ENTITY_0 for example.

Then you would create one consumer per stream for totally ordered processed of all entities in that "partition". You could then choose to run N goroutines for the consumers in a single application or deploy an application per consumer so they get their own resources, etc.

If you want to run multiple instances for HA, then you can still apply the MaxAckPending setting, but of course that will be significantly decrease your consumption throughput. There is a different strategy for active-failover that can be achieved by using a KV bucket to grab a "lease" which provides exclusivity of one instance in an HA setup.. let me know if you want to know more about that 😉

Hopefully this helps!

[arnarg]

Thanks for you response @bruth.

What I want is basically automatic partition rebalancing which is available in kafka and pulsar.

But given that this is not possible out of the box in NATS Jetstream and I don't really want to implement the rebalancing myself, I was wondering if some stream and subject design is possible to achieve my needs.

I want to be able to have high parallel processing if I scale the application up to many instances, but low parallelism in the application itself. I don't want to have to run more than 1 instance to cover all partitions and I don't want to have to statically configure my application to choose which partition subjects to subscribe to.

My application is running an external process (terraform) per entity which is why I don't want to have to have high parallel processing in a single instance as it will require each instance to need a lot of resources, which would often be idle. But if there are a lot of pending messages on the stream the system could automatically scale the application to run more instances and then scale back to 1 once the queue is empty. But when the load is low a single instance can process every entity message.

I hope any of this makes sense.

This is all solved with automatic partition rebalancing in other systems and is in my opinion a short-coming in NATS Jetstream (but in every other aspect it is better).

[ripienaar]

You can control ordering by forcing only max 1 outstanding ack. A Nak is considered outstanding

[Zetanova]

Then it's not a WorkQueue anymore.
The above issue was that @arnarg whated to use it for both event-sourcing and signaling

The easier approach would be to write all events into its entity event subject entites.id483483
and then publish an work item msg entity-demand to entity.jobs

I controller could monitor entity.jobs for unprocessed work items/latency and scale the deployment as needed.

[ripienaar]

I guess that’s a matter of interpretation. It 100% is still a work queue. Just one capable of doing one job per consumer with partitioning you can scale that out to more consumers/workers.

But yes what you describe might be a better fit for this use case.

[derekcollison]

If order is important the system can of course provide that. If high speed and order is important, you might consider an ordered consumer, which is high speed, lightweight and is a 1:1 consumer.

[bruth]

☝️ Seconded. For reference here is a per-entity Load method on the EventStore abstraction in my Rita project (which I need to revisit).

[theofanisM]

@bruth could you explain more about the different strategies for active-failover using a KV bucket for HA setup? thanks!