Subscriptions sometimes skips events

[fbjerggaard]

I have a funny problem with my setup that is running with the Postgresql EventStore (using npgsql 7) and MongoDB EventHandlers (using MongoDB.Entities) that are really simple - basically only upserting data without any weird data modification along the way.

However, sometimes when loading a bunch of data (fx from importing data from a legacy application) the checkpoint store has passed through a specific event but the EventHandler has never been run. If I then reset the checkpoint to a lower value and restart the application the EventHandlers run fine and everything is up to date.

Any ideas on how to debug this further is much appreciated - I am working on a PoC sample to see if it can be replicated outside our environment

[alexeyzimarev]

Have you confirmed that the handler hasn't fired, like with logs or something?

[fbjerggaard]

Yes (sorta), I am logging every MongoDB call and the one to upsert the offending documents was never called

However, digging deeper into the logs I noticed my container was getting OOMKilled which might explain it, however that possibly raises another issue where the checkpoint is updated before the handlers have run?

[alexeyzimarev]

No it can't be. The checkpoint commit is downstream from the projector. Are you sure you are doing SaveAsync or ExecuteAsync and it's not being delayed in any way by the library you use?

[fbjerggaard]

This is a part of the handler that doesn't always run:

public class RegistrationProjections : EventHandler
{
    public RegistrationProjections()
    {
        On<V1.RegistrationReceived>(
            async ctx =>
                await new Registration
                {
                    ID = ctx.Message.Id.ToString(),
                    [... Other properties ...]
                }.SaveExceptAsync(x => new {x.Barcodes})
        );
    }
}

Barcodes are being updated in another event, which is why they are ignored here.

The handler is being registered like this:

services.AddSubscription<PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions>(
    "RegistrationProjections",
    builder => builder.Configure(x => x.MaxPageSize = 256).AddEventHandler<RegistrationProjections>()
);

I tried lowering the MaxPageSize to 256 see if it had any effect, but that doesn't seem to have done anything

[alexeyzimarev]

What if you try using Eventuous MongoDB tools for that upsert? Or the MongoDB driver native API? I just want to remove the possibility that it's a third-party dependency causing the issue.

[fbjerggaard]

I understand, however that would require quite the refactor in my handlers.

I will try to add some more debug logging and try to dig deeper into it, and maybe create a simple handler using the Eventuous MongoDB tools to see if I can replicate it there.

[fbjerggaard]

I tried putting some more logging into the handler:

public class RegistrationProjections : EventHandler
{
    public RegistrationProjections(ILogger<RegistrationProjections> logger)
    {
        On<V1.RegistrationReceived>(
            async ctx =>
                {
                    logger.LogDebug(
                                    "Running projection for {EventType} with ID {RegistrationID}",
                                    typeof(V1.RegistrationReceived).FullName,
                                    ctx.Message.Id
                                );

                    await new Registration
                    {
                        ID = ctx.Message.Id.ToString(),
                        [... Other properties ...]
                    }.SaveExceptAsync(x => new {x.Barcodes});
                }
        );
    }
}

Most of the events log it fine, but some are skipped. The event streams does exist in the database and this time there were no OOM kills of the container

[alexeyzimarev]

Can you add global position to the logs? Also, how do you produce these migration events? Do you commit from multiple processes/threads, or is it a linear fetch-produce?

[fbjerggaard]

Here is an excerpt from the Eventuous debug logs - the Position property is the one highlighted in blue:

- Dont mind that the event handler names etc are different from above - the above is merely a sample whereas the below is from real code

Note that positions 246197, 246201 & 246202 are missing even though they exist in the database:

(The ID's are off by 1 because of #163)

I have 1 service that discovers what data should be migrated which pushes a bunch of messages to another service using MassTransit that then uses the command service to apply the events and then a third service that generates the projections - the logs are from the third service. So the commits are multi-threaded from multiple containers (running the same code) but the projections are guaranteed to only be run by 1 instance at a time

[alexeyzimarev]

I see that the global position appears out of order in the logs, do you partition the subscription?

For missing events, my suspicion is that it's the sequence issue. In case of frequent concurrent writes, in once service the sequence might get allocated before and commit later than in the other service. As the result, you can events with the lower sequence number committed after events with the higher sequence number. It results in the subscription receiving the events with higher number first and the next call will say "more than the one I have", and you have skipped events.

I know that only in theory as I am not an expert in Postgres, but I have read about it somewhere. Will try to find more about it, you can spend some time googling too...

[alexeyzimarev]

Unexpected results might be obtained if a cache setting greater than one is used for a sequence object that will be used concurrently by multiple sessions. Each session will allocate and cache successive sequence values during one access to the sequence object and increase the sequence object's last_value accordingly. Then, the next cache-1 uses of nextval within that session simply return the preallocated values without touching the sequence object. So, any numbers allocated but not used within a session will be lost when that session ends, resulting in "holes" in the sequence.

Furthermore, although multiple sessions are guaranteed to allocate distinct sequence values, the values might be generated out of sequence when all the sessions are considered. For example, with a cache setting of 10, session A might reserve values 1..10 and return nextval=1, then session B might reserve values 11..20 and return nextval=11 before session A has generated nextval=2. Thus, with a cache setting of one it is safe to assume that nextval values are generated sequentially; with a cache setting greater than one you should only assume that the nextval values are all distinct, not that they are generated purely sequentially. Also, last_value will reflect the latest value reserved by any session, whether or not it has yet been returned by nextval.

[alexeyzimarev]

Basically, what the docs say is to use cache setting of one to guarantee order in the sequence.

[alexeyzimarev]

As we don't use sequences explicitly, I am wondering what are the sequence settings for the unique auto incremented id...

[fbjerggaard]

do you partition the subscription?

Partitioning is not enabled

--

A quick search around suggests that the cache values are only really a thing for a sequence, not for the identity. I will try searching deeper

[alexeyzimarev]

Ok, looks like I am spamming here, but still.

Basically, what I am trying to say is that the issue is not that the subscription skips events. It's most probably caused by events with higher global position being committed to the database before events with lower values in global position. I am not exactly sure how to solve it.

I thought of the following:

• Don't use autogenerated id
• Query the max id value after the transaction is opened
• Assign incrementing values to global position when inserting
• Commit the transaction
• It might fail in case of optimistic concurrency, then retry

It will probably slow down the appends, but should solve the issue.