Discussion: async policy handling

[TomKaltz]

Since policy events are handled one-by-one from the all stream, there is a possibility for a policy to severely lag behind if slow queries are used in the policy handler. My current solution is to have the policy add a job to a BullMQ queue using an external system artifact. This results in a lot of boilerplate code just to make policy handling performant.

It dawned on me that this sort of async parallel policy handling might be a candidate for a core feature or plugin for eventually. The general idea is that a policy artifact could have a configurable flag to allow queued/async handling instead of the outbox/serial/checkpoint style handling. The subscription store will mark what events were successfully put on the queue system and the queue system would handle redelivery, dead lettering, etc.

Does this type of feature sound wildly outside of the scope of a framework like eventually? Is there a huge gotcha I might be missing?

[Rotorsoft]

This is a topic I was planning to tackle next, but haven't found the time to focus on it. It's basically the concept of subscriptions by correlation instead of at the artifact level, so you can only serialize flows that are strongly correlated. This should solve the scalability issues you are describing, but it requires some work at the broker level to understand new metadata to auto configure the system.

[TomKaltz]

...continuing from here, this is what is roaming around in my head. It took a lot of iteration with Claude Sonnet to elucidate visually what I was thinking but here is how I envision the use of the new PostgresOrderedMessageQueue to allow scalable consumer processing.

Here are the important bits i'm trying to convey...

• broker enqueues events in strict order to each consumer's queue table based on it's event name interest
• each consumer subscription is acked as batches are enqueued
  • policy consumer queue is "streamed" by the original event stream id
  • process manager consumer queue is "streamed" by the actor name/hash (calculated by the broker using the artifact at time of enqueuing to the consumer's queue)
• Each consumer queue processor will dequeue and fill concurrent processing slots as messages are available. (2 in our examples but can be configured for more)
  • When a slot is available, the next ordered event(from an unlocked stream) is dequeued and begins processing
  • The entire stream/actor hash is locked during processing, this ensures ordered processing of each stream/actor hash
  • Events from the same stream/actor hash will never process at the same time...they will be processed one-by-one in strict order. This is guaranteed by the dequeue implementation.
  • Streams processing does not necessarily stick to a certain slot, events are distributed as slots and new events are available (from unlocked streams).

I'm working on a POC broker right now that would handle all this. So far I've only implemented policy event enqueueing. I will build the processing side next and let you know if I can get something working.

Here are the diagrams that Claude Sonnet and I came up with. I know the styling is messy, I apologize.

Lets start with how a policy would be handled...

%%{init: {'theme': 'base', 'themeVariables': { 'lineColor': '#666666' }}}%%
graph TB
    subgraph EventStore["Event Store (Single Ordered Log)"]
        direction LR
        E1["1: Registered<br/>Stream1"] --> 
        E2["2: Arrived<br/>Stream2"] --> 
        E3["3: Registered<br/>Stream1"] --> 
        E4["4: Departed<br/>Stream3"] --> 
        E5["5: Registered<br/>Stream2"] --> 
        E6["6: Arrived<br/>Stream1"] -->
        E7["7: Departed<br/>Stream3"] -->
        E8["8: Arrived<br/>Stream2"] -->
        E9["9: Departed<br/>Stream1"] -->
        E10["10: Registered<br/>Stream3"]
    end

    subgraph Subscriptions["Consumer Subscriptions"]
        S1["Consumer1<br/>Checkpoint: 1<br/>Interested in: Arrived, Departed Events"]
    end

    subgraph Broker["Policy Broker"]
        B1["Batch Query Handler<br/>Fetches events after checkpoint<br/>Filters by event names"]
    end
    
    subgraph Queue1["Consumer1 Queue (Events in Event Store Order)"]
        direction LR
        Q1E1["2: Arrived<br/>Stream2"] --> 
        Q1E2["4: Departed<br/>Stream3"] --> 
        Q1E3["6: Arrived<br/>Stream1"] --> 
        Q1E4["7: Departed<br/>Stream3"] --> 
        Q1E5["8: Arrived<br/>Stream2"] --> 
        Q1E6["9: Departed<br/>Stream1"]
    end

    subgraph Processor["Parallel Event Processor<br/>(See Timeline Below)"]
        P1["Slot1"]
        P2["Slot2"]
    end

    %% Main flow connections
    EventStore --> |"Query events after checkpoint<br/>filtered by event names"| Broker
    Broker --> Queue1
    Queue1 --> Processor

    %% Subscription influence
    Subscriptions --> |"Checkpoint info"| Broker

    classDef store fill:#f9f,stroke:#333,stroke-width:2px,color:#000
    classDef sub fill:#bbf,stroke:#333,stroke-width:2px,color:#000
    classDef broker fill:#bfb,stroke:#333,stroke-width:2px,color:#000
    classDef queue fill:#ffb,stroke:#333,stroke-width:2px,color:#000
    classDef processor fill:#ffd,stroke:#333,stroke-width:2px,color:#000

    class EventStore store
    class Subscriptions sub
    class Broker broker
    class Queue1 queue
    class Processor processor

Loading

gantt
    title Slot-Based View: Processing Timeline
    dateFormat x
    axisFormat %L ms
    
    section Consumer1 Slot1
    E2 S2    :crit, 0, 30
    E6 S1    :done, 30, 40
    E7 S3    :active, 40, 50
    E9 S1    :done, 50, 80
    
    section Consumer1 Slot2
    E4 S3    :active, 0, 40
    E8 S2    :crit, 40, 70

Loading

Now here is a diagram showing how eventually ProcessManagers would be handled...

graph TB
    subgraph EventStore["Event Store (Single Ordered Log)"]
        direction LR
        E1["1: Registered"] --> 
        E2["2: Arrived"] --> 
        E3["3: Registered"] --> 
        E4["4: Departed"] --> 
        E5["5: Registered"] --> 
        E6["6: Arrived"] -->
        E7["7: Departed"] -->
        E8["8: Arrived"] -->
        E9["9: Departed"] -->
        E10["10: Registered"]
    end

    subgraph Subscriptions["Process Manager Subscriptions"]
        S1["ProcessManager1<br/>Checkpoint: 1<br/>Interested in: Arrived, Departed Events<br/>Correlate by: actorHash"]
    end

    subgraph Broker["Process Manager Broker"]
        B1["Batch Query Handler<br/>Fetches events after checkpoint<br/>Filters by event names<br/>Groups by actorHash"]
    end
    
    subgraph Queue1["ProcessManager1 Queue (Events in actorHash Order)"]
        direction LR
        Q1E1["2: Arrived<br/>Stream: xyz"] --> 
        Q1E2["4: Departed<br/>Stream: xyz"] --> 
        Q1E3["6: Arrived<br/>Stream: abc"] --> 
        Q1E4["7: Departed<br/>Stream: def"] --> 
        Q1E5["8: Arrived<br/>Stream: ghi"] --> 
        Q1E6["9: Departed<br/>Stream: abc"]
    end

    subgraph Processor["Parallel Event Processor<br/>(See Timeline Below)"]
        P1["Slot1"]
        P2["Slot2"]
    end

    %% Main flow connections
    EventStore --> |"Query events after checkpoint<br/>filtered by event names"| Broker
    Broker --> Queue1
    Queue1 --> Processor

    %% Subscription influence
    Subscriptions --> |"Checkpoint info"| Broker

    classDef store fill:#f9f,stroke:#333,stroke-width:2px
    classDef sub fill:#bbf,stroke:#333,stroke-width:2px
    classDef broker fill:#bfb,stroke:#333,stroke-width:2px
    classDef queue fill:#ffb,stroke:#333,stroke-width:2px
    classDef processor fill:#ffd,stroke:#333,stroke-width:2px

    class EventStore store
    class Subscriptions sub
    class Broker broker
    class Queue1 queue
    class Processor processor

Loading

gantt
    title Slot-Based View: Processing Timeline
    dateFormat x
    axisFormat %L ms
    
    section ProcessManager1 Slot1
    E2 xyz    :crit, 0, 30
    E4 xyz    :crit, 30, 40
    E8 ghi    :active, 40, 60
    E9 abc    :done, 60, 80
    
    section ProcessManager1 Slot2
    E6 abc    :done, 0, 25
    E7 def    :active, 25, 100

Loading

[Rotorsoft]

This aligns closely with what I have in mind, but the main difference is that you introduce slots to support concurrent processing. In contrast, I’m envisioning this being resolved at the enqueueing level. Essentially, queues in the pool would behave like your slots, eliminating the need for explicit slot management.

Here’s how I’m visualizing the broker’s role:

1. Scheduling Events

The broker would schedule new events into “correlated queues” within a pool. Queues that fail to empty could be moved to a DLQ (Dead Letter Queue) facility and "released" back to the pool.

There will be basically two correlation types:

• Type 1: Aggregate Streams
  For business flows fully encapsulated within an aggregate, where multiple aggregate instances are independent of each other, we can schedule their lifecycles in separate concurrent queues. For example, consider independent aggregates like user accounts that don’t interact among them. In this case, consumers (e.g., policies or projections) are correlated to individual aggregate instances, allowing parallel processing.

• Type 2: Process Managers, Sagas, and High-Level Projections
  For business flows involving the coordination of multiple aggregate instances, where event order matters, correlation occurs at a higher level. For instance, if user 1 reserves a resource, user 2 shouldn’t consume it before the reservation completes. Here, the correlation key should be derived from "shared" attributes of the business flow (e.g., a resource ID as the unique identifier for the process instance). Events from multiple streams/aggregates are placed in a correlated queue for this ID and delivered sequentially to consumer policies. Concurrent processing would still be possible across different flow IDs.

2. Dispatching Events

Once queues are properly constructed, the broker’s role becomes simpler: dispatching events to consumers sequentially, handling retries, and managing DLQs (e.g., stopping a flow instance when necessary). This dispatching process should remain generic and straightforward.

Basic Principles

The complexity of the broker lies primarily in the generalization of scheduling “correlated queues” from different streams based on computable correlation keys. This is the "business knowledge" that should be coded in the "subscription" schemas. Beyond that, the system should handle event dispatching with standard mechanisms like sequential delivery, retries, and queue management.

Of course, there is more to this, and additional features would include robust logging and tooling for visualizing the state of the queues, offering clear insights into system behavior.

Hope this makes sense.

[TomKaltz]

Where in your design concept are these queues processed? I envisioned the consumers queues(one queue/table for each consumer artifact) being processed outside of the broker, by consumer queue processors that can be spun up as needed for parallel processing. Again, the PostgresOrderedMessageQueue implementation allows for parallel dequeueing and will ensure serialized message processing within a stream("correlation key", I like this nomenclature!!) no matter how many consumer queue processor instances are competing for message.

I guess I ask the above question being I don't think I understand the concept or need for queue pools. What immediately comes to my mind is that this would introduce a sort of sharding responsibility at the broker level that might not need to exist? In my mind the broker should just enqueue each message to the appropriate consumer queue tables with either it's aggregate stream id or it's calculated correlation key. The correlation key is calculated by calling artifact.actoreventName.

This reminds me, we should probably change the nomenclature for the MessageQueue interface/column to "correlationKey" and not "stream" since we probably want to leave all the message data (like stream id) intact in the data column?

[Rotorsoft]

You seem to be placing the complexity in the queue processing side, where concurrency is determined by creating slots. I’m thinking that it can be placed in the scheduling side, when queues are already created with concurrency in mind, one queue per process instance, that’s why you need a pool to handle a large number of instances similar to web requests handled by a thread pool. We need to think about the trade offs of each approach, but the results are the same.

…

On Sun, Dec 22, 2024 at 12:19 PM Thomas Kaltz III ***@***.***> wrote: Where in your design concept are these queues processed? I envisioned the consumers queues(one queue/table for each consumer artifact) being processed outside of the broker, by consumer queue processors that can be spun up as needed for parallel processing. Again, the PostgresOrderedMessageQueue implementation allows for parallel dequeueing and will ensure serialized message processing within a stream("correlation key", I like this nomenclature!!) no matter how many consumer queue processor instances are competing for message. I guess I ask the above question being I don't think I understand the concept or need for queue pools. What immediately comes to my mind is that this would introduce a sort of sharding responsibility at the broker level that might not need to exist? In my mind the broker should just enqueue each message to the appropriate consumer queue tables with either it's aggregate stream id or it's calculated correlation key. The correlation key is calculated by calling artifact.actoreventName <http://event>. This reminds me, we should probably change the nomenclature for the MessageQueue interface/column to "correlationKey" and not "stream" since we probably want to leave all the message data (like stream id) intact in the data column? — Reply to this email directly, view it on GitHub <#78 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AJ6GL73E3E6BJ734J5FEXZ32G3YC7AVCNFSM6AAAAABMWMXLXSVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDKNJYGUZDKOJTGE> . You are receiving this because you commented.Message ID: ***@***.***>

[TomKaltz]

You could remove the concept of slots inside a queue's processor and just make the processor simply dequeue and process one message at a time in a loop. Then you can spin up as many instances as you need based on demand and it accomplishes the same exact thing. The PostgresOrderedMessageQueue implementation guards for correlationKey concurrency by not allowing any of that queue's multiple producer processor instances to dequeue/process a given correlationKey message at the same time. Lets say processor1 is processing a message with correllationKey1 and the next message in the queue also has correllationKey1, when another processor instance calls dequeue while processor1 is processing correllationKey1, the dequeue implementation dequeues the next available message that doesn't have that correllationKey (since that stream is locked in the database).

To me this reduces complexity because the broker isn't concerned with sharding into a queue pool. The messages get queued in each consumer's queue table, which provides logical organization and containment for those messages. The user can choose how many processor instances to spin up to process that queue table and the PostgresOrderedMessageQueue dequeue implementation already guarantees that correlated messages will be processed in strict order...even when multiple processes are dequeueing from that queue.

[Rotorsoft]

I need to think more about this with a clear mind. 1. We have business flows implemented via "eventually" artifacts akin to classes in OOP. 2. Flows are implemented using a combination of aggregates and policies and have a finite lifetime (ideally)… policies are the "automation" and can span multiple aggregates (sagas, etc) 3. Each flow can have many (potentially concurrent) uncorrelated instances (recorded as correlated streams of events) akin to object instances in OOP 4. Instances have a correlation key and can be reduced to states (projected), either as a contained aggregate or using projections for viewing or reference. Basically, you are proposing one queue per consumer artifact (kind of the existing broker) that gets resolved later, and I’m thinking about one frontend on-demand queue per flow instance. The limitation of the existing broker is that it serializes at the consumer artifact level. One flow can be composed of many consumer artifacts. (This is important when evaluating a queue per consumer approach), while a flow instance is unique. Queues are ephemeral memory that gets allocated on demand when new events arrive (in real-time or when replaying), and their main responsibility is to serialize correlated events to a flow instance. We need to keep a "watermark" on streams processed by instance to support parallel processing. This can be a large table of watermarks identifying all instances in the system (need more t thinking). Since we can have a large amount of active flows in the system, we need a pool of resources when processing. The first question is: Is it better to schedule (enqueue) instances upfront or schedule consumers classes and delegate concurrent instances to the dispatcher? It comes down to how we define subscriptions and manage watermarks. The internal implementation can vary. I’m leaning toward a queue per instance to keep the queue adapters very simple, otherwise, we have to deal with correlations inside the queue.

…

On Sun, Dec 22, 2024 at 1:47 PM Thomas Kaltz III ***@***.***> wrote: You could remove the concept of slots inside a queue's processor and just make the processor simply dequeue and process one message at a time in a loop. Then you can spin up as many instances as you need based on demand and it accomplishes the same exact thing. The PostgresOrderedMessageQueue implementation guards for correlationKey concurrency by not allowing any of that queue's multiple producer processor instances to dequeue/process a given correlationKey message at the same time. Lets say processor1 is processing a message with correllationKey1 and the next message in the queue also has correllationKey1, when another processor instance calls dequeue while processor1 is processing correllationKey1, the dequeue implementation dequeues the next available message that doesn't have that correllationKey (since that stream is locked in the database). To me this reduces complexity because the broker isn't concerned with sharding into a queue pool. The messages get queued in each consumer's queue table, which provides logical organization and containment for those messages. The user can choose how many processor instances to spin up to process that queue table and the PostgresOrderedMessageQueue dequeue implementation already guarantees that correlated messages will be processed in strict order...even when multiple processes are dequeueing from that queue. — Reply to this email directly, view it on GitHub <#78 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AJ6GL73ZL4CMOG5PVSXBGCD2G4CK5AVCNFSM6AAAAABMWMXLXSVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDKNJYGU2TIOBSGY> . You are receiving this because you commented.Message ID: ***@***.***>

[TomKaltz]

However this gets implemented, it's clear to me that each event handling artifact will need to be provided functions that calculate it's correllationKey based on the event it's processing, exactly like how "actor" is specified in an eventually ProcessManager, right?

[Rotorsoft]

Not necessarily if the queue is already correlated and it’s guaranteed to deliver the right events all the time. The key question is where to place the responsibility to serialize correlated events

…

On Sun, Dec 22, 2024 at 3:37 PM Thomas Kaltz III ***@***.***> wrote: However this gets implemented, it's clear to me that each event handling artifact will need to be provided functions that calculate it's correllationKey based on the event it's processing, exactly like how "actor" is specified in an eventually ProcessManager, right? — Reply to this email directly, view it on GitHub <#78 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AJ6GL72USZYKKQKJXGUBOCL2G4PI3AVCNFSM6AAAAABMWMXLXSVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDKNJYGU4TAOJXG4> . You are receiving this because you commented.Message ID: ***@***.***>

[Rotorsoft]

I'll wait for your broker implementation and will provide feedback, but I think you are on the right track.

If the broker you have in mind takes care of enqueuing events by correlation key in your ordered msg queue, that will represent the pool of queues. The other side is already taking care of delivering concurrently uncorrelated flows, and just needs some basic reliability guarantees and flagging the ones that are blocked by errors independently of the rest (DLQ like)

Looking forward to seeing that work 👍

[TomKaltz]

@Rotorsoft give this a whirl https://github.com/TomKaltz/ConsumerQueueExp

[Rotorsoft]

It works on my end and looks fine at first glance 👍

The end goal I had in mind was to refactor the broker interface and hide the scheduling/dispatching (including subscriptions) behind the bootstrapping code, so no more public subscription stores or queue stores... we can probably make it work with just a single subscriptions table and a single message queue per service/process.

The other thing I noticed is that you are still using the old subscriptions interface and some of the existing broker logic, but with the introduction of correlated queues, the broker implementation should be much simpler.

Just high level without thinking much about the details:

• bootstrap broker with one queue adapter per service
• listening/polling for new events to be enqueued by correlation key (put on top of dispatcher priority queue)
• should probably store watermarks by correlation key too
• the generic broker loop can build a small priority queue in memory (pool) to keep track of active correlation queues and push until empty ... feels like this should be a lot simpler than the current implementation... probably needs some sketching too for clarity, but in any case, I was never happy with the existing broker.

[TomKaltz]

Yeah I was only handling policies...this was just hacked together to get it to work and to elucidate what I was thinking. There are still a ton of concepts and bugs that need to be worked out.

In my design idea/concept, the current subscription model seems to still work great. The event store is written to atomically then it kicks off the broker to dispatch events to each individual consumer queue....the subscription watermark tell us which messages were successfully dispatched to their appropriate consumer queues from the all stream. The consumer queue tables shows which correllationKeys are unprocessed (inverse of a correllationKey watermark table, which could grow forever). We could query with a partial index, probably) to get a list of in-flight/unprocessed keys. In my mind the above promotes the best durability without having to store correllationKey watermarks in perpetuuty.

[Rotorsoft]

That's true, but imagine the amount of value you can extract from that system level projection. The stats feature is another incomplete area that would benefit from a table with all instances (aggregates and PMs) including their class, watermark, etc

[TomKaltz]

Ah yes, Marten does this. I like the idea.