MySQL Projections skipping events (SingleStreamStrategy)

[fritz-gerneth]

Issue: under unknown circumstances the PdoEventStoreReadModelProjector skips applying events to registered handlers.

I have seen this issue happen a few times over the past year only. Yet since missing events in the projections causes quite a few problems I'm trying to figure out the issue since then. Creating this issue for tracking & by any chance input from others.

Setup: Mysql EventStore & Projections with SingleStreamStrategy running in their own process. DB Server and process server running on different VM hosts.

Projection options:

PdoEventStoreReadModelProjector::OPTION_PERSIST_BLOCK_SIZE => 1,
PdoEventStoreReadModelProjector::OPTION_PCNTL_DISPATCH => true,
PdoEventStoreReadModelProjector::OPTION_LOCK_TIMEOUT_MS => 20000,
PdoEventStoreReadModelProjector::OPTION_UPDATE_LOCK_THRESHOLD => 15000,

Symptoms: take a simple set of IssueCreated, IssueUpdated, IssueDeleted of any examplary Issue AR. Events occuring in the typical order of IssueCreated -> IssueUpdated (n times) -> IssueDeleted. The projection simply maintains a list of all (undeleted) issues and their latest values. Each event has a handler registered to either insert a row, update a row or delete the row respectively.
For 99.9% of the time this works perfectly fine as expected. Yet for unknown reasons very rarely some event is not handled. The effect on the projection differes depending on the event (e.g. IssueUpdate -> update lost) but is particular bad for IssueCreated as rows are missing at all (subsequent events might rely on the presence).
Resetting the projection solves this issue and all events are applied as expected. But this is a pretty unpractical to do in a production environment when this takes days each time.

Debugging:
To help me make this problem visible in the first place I slightly modified the [PdoEventStoreReadModelProjector](https://github.com/prooph/pdo-event-store/blob/master/src/Projection/PdoEventStoreReadModelProjector.php#L575) to simply log a line about which event it is now dispatching:

foreach ($events as $key => $event) {
	$this->logger->info(sprintf(
		'Dispatching event %s[%s]',
		$streamName,
		$key
	));

	if ($this->triggerPcntlSignalDispatch) {
		\pcntl_signal_dispatch();
	}

In normal operations I get sequential messages in this format ( gcl_collection is my stream's name):

2019-01-29T12:51:11+00:00 INFO (6): Dispatching event gcl_collection[1]
2019-01-29T12:51:12+00:00 INFO (6): Dispatching event gcl_collection[2]
2019-01-29T12:51:13+00:00 INFO (6): Dispatching event gcl_collection[3]
2019-01-29T12:51:14+00:00 INFO (6): Dispatching event gcl_collection[4]

Now there obvisouly might be gaps in the event no due to ConcurrencyExceptions when inserting many events in parallel. But this still gives me an idea if an event has been handled at all. Any no in the stream should also show up in my log as being dispatched at least once.

Having this deployed and running for a few weeks now I finally coud capture this event again (on our low-volumne testing environment this time). My stream reported a created event which had been skipped in the projection:

2019-01-29T12:51:11+00:00 INFO (6): Dispatching event gcl_collection[111]
2019-01-29T12:51:31+00:00 INFO (6): Dispatching event gcl_collection[112]
2019-01-29T12:53:11+00:00 INFO (6): Dispatching event gcl_collection[113]
// Missing event 114 here
2019-01-30T11:58:12+00:00 INFO (6): Dispatching event gcl_collection[115]
2019-01-30T11:58:12+00:00 INFO (6): Dispatching event gcl_collection[116]
2019-01-30T11:58:12+00:00 INFO (6): Dispatching event gcl_collection[117]
2019-01-30T11:58:29+00:00 INFO (6): Dispatching event gcl_collection[118]

Looking at the event-stream itself this event is clearly here though:

|| *event_id*                           || *created_at*               || *aggregate_id*                               || *aggregate_version*  ||
|| 72e6eebc-723c-4a7e-a5e7-ced418c3d08d || 2019-01-29 12:51:10.970799 || GCL:COL-e5baa4c0-23c3-11e9-a495-e7066c3b34e9 || 27                   ||
|| d02d3f6b-6fc5-4592-b88f-1a8e735b2ec4 || 2019-01-29 12:51:31.127142 || GCL:COL-e5baa4c0-23c3-11e9-a495-e7066c3b34e9 || 28                   ||
|| fd2c0841-3198-4d60-8204-ce9e44d489eb || 2019-01-29 12:53:11.159632 || GCL:COL-e5baa4c0-23c3-11e9-a495-e7066c3b34e9 || 29                   ||
|| 5dd81617-2cd7-4539-ac65-5eb9609481da || 2019-01-30 11:58:12.775746 || GCL:COL-50aa81c1-2486-11e9-a9f8-e1ff3db07a2f || 1                    ||
|| df5f4332-9537-4c2e-85be-e5e28519b763 || 2019-01-30 11:58:12.801373 || GCL:COL-50aa81c1-2486-11e9-a9f8-e1ff3db07a2f || 2                    ||
|| e415335f-6889-4652-8d86-d89e3f45166b || 2019-01-30 11:58:12.801517 || GCL:COL-50aa81c1-2486-11e9-a9f8-e1ff3db07a2f || 3                    ||
|| 686f835a-de20-48e8-8dfd-792e7614dcac || 2019-01-30 11:58:12.811411 || GCL:COL-50aa81c1-2486-11e9-a9f8-e1ff3db07a2f || 4                    ||

This is not limited to specific aggregate roots but can happen for any event, unrelated to the aggregate version or such. I have not seen this to happen during replays.

This only seems to happen when I insert events at a very high rate. In fact I think this has only happened so far if a single process rapdily creates many events (not necessarily on the same AR though). My current working theory is that while ordered in that way rows become visible to selects in a different order (e.g. the projection selects and gets event 115 while 114 becomes visible to the select a millisecond later or so. But then I;m not that deep into MySQL internals if this is posibble.

Any pointers on how to continue investigating this issue are welcome.

[codeliner]

that's a serious problem ...

We had a similar issue with our mongoDB v6 event-store. Same symptoms:

• only seems to happen when I insert events at a very high rate
• while ordered in that way rows become visible to selects in a different order

Some background. We have an event sourced service that processes SAP changesets (IDOC XML files). A Java Middleware sits between our service and the SAP ERP. When there is a mass update in SAP (during data migration for example) the Java Middleware takes all updates and forwards them to us at once. This results in thousands of events within a couple of seconds. Under that load we experienced the same issue.

Back in the days we used MongoDB 3.6. No auto increment fields and no transactions available. Our custom made projections used the aggregate version + timestamp to remember stream positions until we saw that events are being skipped if the write model is on fire.

Replays worked just fine. Which made the problem hard to debug.

We did exactly the same to investigate the issue: Log processing of events in the projections and compare the log with the event streams. We identified the problem:

Events with higher versions became slightly faster visible for the projection.

To be honest: We also discussed (me and @sandrokeil ) if this could happen with prooph v7 projections, too. But since they rely on transactions and a stream sequence we thought the problem is only related to our custom implementation for MongoDB .... Looks like we were wrong :(

How did we solve the Problem?

One option was to let projections detect "gaps" and if a gap is detected wait a moment and reread the stream. That's not a 100% guarantee but could work if the sleep time is high enough.

Finally we went for a 100% save solution. It looks like this:
We have an Event-Metadata-Enricher that adds a "projection_xyz_handled" flag to each event metadata with an initial value of false. When a projection has processed the event it sets the flag to true.
When we want to replay, we simple reset the flags.

First we thought that this will produce a lot of load on our MongoDB-Cluster but Mongo can handle this just fine. Only real bottleneck for us is when we want to deploy a new projection. In this case we need to add a new flag to each existing event in the stream (several millions) and mongo needs to update the index. But we don't deploy new projections that often ....

Our final solution can't be implemented in prooph v7 projections, but maybe the gap detection is an acceptable workaround. It could be activated with an additional projection option.

I'm 99% sure that your working theory is correct @fritz-gerneth

My current working theory is that while ordered in that way rows become visible to selects in a different order (e.g. the projection selects and gets event 115 while 114 becomes visible to the select a millisecond later or so. But then I;m not that deep into MySQL internals if this is posibble.

@prolic What do you think?

Not sure if this is a MySql issue or could happen with Postgres, too. I don't want to bet against it ....
Does AxonFramework support projections on RDBMS backends? /cc @mrook
If so, how do they avoid the problem? A projection could read the WAL of the database directly. Kafka Connect does that for example.

[mrook]

@fritz-gerneth Axon supports any projection, as long as you can write the event handler for it. With the default SequencingPolicy (which is SequentialPerAggregatePolicy), events are guaranteed to arrive at the projection event handler(s) in the order they were published on the event bus - per aggregate.

[prolic]

I read a little about auto increments in MySQL. The problem is, that the auto increment is done on write, not on commit.

F.e.
Start two transactions in two terminals.
Insert into ... (First in terminal one, then terminal two)
Commit in terminal two.

Select * from ...

You see a gap.

Commit terminal one.

Select * from ...

Gap is closed.

Probably the same is true for postgres as well. So in case there is a gap, we need to detect, back off and read again. Really painful, but I don't see a better solution.

[prolic]

@codeliner your approach with using metadata for this should also work, but I prefer to keep the event immutable.

[fritz-gerneth]

Appreciate all your input! :)

@mrook : from what I understand this solves a similiar but different problem: making sure all selected events arrive in sequence (which they do for projections here). The issue in here is more about making sure we don't skip events when querying the event bus. I suppose this guarantee is implemented quite differently in Axon. But then my knowledge about Axon is old and very basic :)

@prolic : Do you have any resource detailing this behavior (for personal education :))?

I managed to reproduce this scenario in a slightly different setup as well: a AR has to append 5 events in one call where the first event has a significantly larger payload then the others (e.g. large complex JSON object vs two small string keys). All events were published in the same appendTo call. But again this is going into MySQL internals I have no experience with :)

Modifying events is something I neither want or can do. We are resetting / changing our projections quite rapidly, often for testing purposes only and many streams are used by many projections concurrently.
Since we cannot avoid gaps in the first place I agree some gap detection / handling is pretty much the only way arround.
I think the best way would be to extract loading of streams into a strategy.

foreach ($this->streamPositions as $streamName => $position) {
	try {
		$streamEvents = $this->eventLoadStrategy->load($streamName, $position, $this->metadataMatcher);
	} catch (Exception\StreamNotFound $e) {
		// ignore
		continue;
	} 

   // .....
}

The default one doing exactly the same as now.

class DefaultEventLoadStrategy 
{
    private $eventStore;

    public function load(string $streamName, $position, $metadataMatcher)
    {
        return $this->eventStore->load(new StreamName($streamName), $position + 1, null, $metadataMatcher);
    }
}

We then could add a second strategy supporting GAP detection and some (configureable) delay to retry. If the retry fails we assume a gap.

class GapDetectingEventLoadStrategy
{
    private $eventStore;
    
    private $sleepTime;
    
    public function load(string $streamName, $position, $metadataMatcher)
    {
        $events = $this->loadEvents($streamName, $position, $metadataMatcher);
        $nextEventNumber = $events->key();
        
        if ($nextEventNumber === $position + 1) {
            return $events;
        }
        
        sleep($this->sleepTime);
        
        return $this->loadEvents($streamName, $position, $metadataMatcher);
    }
    
    private function loadEvents(string $streamName, $position, $metadataMatcher)
    {
        return $this->eventStore->load(new StreamName($streamName), $position + 1, null, $metadataMatcher);
    }
}

That way I'd be backwards compatible and you can configure performance vs safety you won't miss an event. Am I thinking too naive here or missing any other vital parts or better approaches?

A more definitive approach might be to look at aggregate versions. But this has its own downsides as well:

• Events are not necessarily aggregate events
• Still possible to miss the first event of an aggregate
• Need to keep track of the version of each Aggregate (like to avoid this as we have hundrets of thousands of ARs in some streams, updating this JSON object every time puts quite a load on the DB server)

[prolic]

@fritz-gerneth researching for this problem I found this article https://www.matthewbilyeu.com/blog/auto-incremented-values-can-appear-out-of-order-if-inserted-in-transactions/ - I hope it helps.

About adding an eventLoadStrategy: I think we should not do that, as it further complicates the usage of the event-store (we have already lots of configurable options, persistence strategies, ...). It's better to have a useful default value for sleep timeout and give a hint in the docs on why it was added and what impact you have changing the default.

Another possible solution is a complete different way of inserting rows, namely without any autoincrement. But I didn't work this out yet, so I don't know if this is possible.

[mrook]

@fritz-gerneth I see, and yes it's implemented differently. Projections (event handlers) in Axon typically do not directly query the event bus (they can, but there's really no point), but instead are fed events through a processor, which either follows the live stream or replays from a certain position.

[sandrokeil]

Another possible solution is a complete different way of inserting rows, namely without any autoincrement. But I didn't work this out yet, so I don't know if this is possible.

I would really appreciate that. We only need the autoincrement value to sort the events and read them forward and backward. If we use a stream/watch mechanism like in MongoDB or Postgres, events occur in the order they are saved to DB. As we can see, the autoincrement value is not really the order of events. I have discussed this a little with @codeliner and we have no really solution for the sort thing. Microseconds or Nanoseconds are not enough.

The question is, how can we preserve the order of events when they are saved at the same time and how is it sortable ascending and descending?

[prolic]

EventStore (http://eventstore.org/) also has incrementing event positions, starting from 0. So all its events are positioned as 0, 1, 2, 3, 4, ... and there are no gaps, never ever.

I looked into this on how they solved this on their side (although they don't use SQL) and their solution (I don't know if that's the only reason why they do this, and I suppose that it's not the main reason) is to use a single writer thread only. All write operations will get sent to that writer queue so there is no possibility of any race conditions.

As we don't have a server implementation here but merely a thin layer on top of a PDO connection, the closest we can do to get to the same behaviour is creating a lock for each write, so only one process can ever write and all others have to wait until it's finished. This will definitely reduce the write performance, but maybe that's even acceptable and definitely worth a try for the prooph/event-store v8 sql implementation.

To reduce the time the lock is acquired, maybe the best way to do this is with a stored procedure, but this would be a BC break in v7.

[fritz-gerneth]

@mrook : I think Axon's processors are then pretty much the equivalent to projections here then (the TrackingEventProcessor to be precise). Think the GapAwareTrackingToken [source] is your answer to this issue :)

Implementation of a TrackingToken that uses the global insertion sequence number of the event to determine tracking order and additionally stores a set of possible gaps that have been detected while tracking the event store.

By storing the sequence numbers of gaps, i.e. sequence numbers of events that may have been inserted but have not been committed to the store, consumers are able to track the event store uninterruptedly even when there are gaps in the sequence numbers of events. If a gap is detected the event store can check if meanwhile this gap has been filled each time a new batch of events is fetched.

Digging through the source code on GitHub my understanding on how Axon does this (with this token) is a more aggressive at-least-once approach:

• A token does not only represent a single position but a set of possible gaps in addition
• When querying for new events not only the next position but all gaps are queried for as well
• Gaps that have fallen too far behind the index are assumed to be actual gaps at some point and will be actually skipped at some point

Doing similiar on our end certainly would work as well but require a bit of a change in the event store api to allow multiple range-queries.

As a side-note: dismissing my former approach. Giving it a second thought made it clear that this does not reliably solve the issue but only solve this for gaps starting at the start of my iteration but not in between.

[fritz-gerneth]

As we don't have a server implementation here but merely a thin layer on top of a PDO connection, the closest we can do to get to the same behaviour is creating a lock for each write, so only one process can ever write and all others have to wait until it's finished. This will definitely reduce the write performance, but maybe that's even acceptable and definitely worth a try for the prooph/event-store v8 sql implementation.

Performance impact of doing so would greatly depend on the stream strategy. Per aggregate should not be any major impact since Aggregates already kind of enforce this through event versions. But for for the others this would be a table lock affecting writes to either all aggregates of the same type or all aggregates on the system.
Even not allowing concurrent writes to the different ARs of the same type already has severe impact in scenarios with high write-volumes - which is exactly the scenario this would apply to to solve this issue here. Not saying this is a particularly bad or something though, just that the impact might be huge.
Would there be some BC way using GET_LOCK to enforce this on a per-stream basis?

Besides this I'd follow up on the gap-tracking stream positions. This might require a bit of work to actually track & store the gaps alongside the current position but would mitigate this issue too (the longer we allow gaps to be tracked the more safely it becomes). There're two major challenges I'd see for this though:

• Current event-stream interface only allows to load a single range of events. When querying with gaps we'd have at least two ranges: for the current position + for any gaps we still track. This could be handled with multiple calls to load for each range too. But doing this in a single go certainly would be more performant.
• For each stream we'd have to store the gaps along the current position, requiring a new format to store stream positions. Think this could be opt-in behavior with a big warning to 'do not change this for running projections' or a on-the-fly migration mechanism.

(To be clear: I'd be willing to implement both changes, for me that's merely a matter on if this is fixed in here or with a custom projection in our code-base)

[prolic]

@fritz-gerneth if you have high write throughput right now, maybe you can test using "GET_LOCK" for a few hours in production and see if the impact is really that huge? Maybe it's still acceptable. The main problem that I see is the additional roundtrip which leads to higher locking times. That's why I think having stored procedures in place might be a better solution then sending multiple queries from PHP.

[fritz-gerneth]

Before testing this in production I wanted to see how this behaves and if it can actually solve the problem. I created a simple set of scripts to illustrate & test this issue:

• A generator script that inserts a fixed amount of events into a table, optioinally with using GET_LOCK
• A wrapper to invoke n instances of the generator in parallel (simulate n parllel producers
• A consumer counting gaps while selecting the latest set of events (e.g. like our projections)

The generator can be invoked like this:

./event-gen-bulk.sh 2 50000 1

Arguments are:

• Number of producers to start
• Number of events to insert (each producer)
• 0/1 to use GET_LOCK or not

Be sure to invoke the consumer before starting the generators though

 php event-consumer.php

Interesstingly enough, this issue seems to decline as the amount of parallel producers increases (no locking):

Producers | Events Per Producer | Gaps  | Events per gap
-------------------------------------------------------
        2 | 50005               |  350  | 29
        4 | 50005               | 1916  | 104
        6 | 50005               | 1153  | 260
        8 | 50005               |   23  | 17393
       10 | 50005               |    2  | 250025

So my next test was wrapping the insert in a GET_LOCK("test-ar", -1);:

Producers | Events Per Producer | Events/sec / producer | Events/sec total | Lock 
---------------------------------------------------------------------------------
        2 | 50005               | 3300                  | 6600             | no
        2 | 50005               | 1620                  | 3240 (-49%)      | yes
        4 | 50005               | 1541                  | 6164             | no
        4 | 50005               |  707                  | 2828 (-45%)      | yes
        6 | 50005               | 1230                  | 7380             | no
        6 | 50005               |  409                  | 2454 (-33%)      | yes
        8 | 50005               |  962                  | 7696             | no
        8 | 50005               |  279                  | 2232 (-29%)      | yes
       10 | 50005               |  830                  | 8300             | no
       10 | 50005               |  250                  | 2500 (-30%)      | yes

Now those values are to be taken with a (big) grain of salt as numbers have a great variance but they give at least some indication. The good news is: this does take care of any gaps in the consumer.

Two things to note:

• the GET_LOCK and RELEASE_LOCK statements need to be outside of the transaction. When inside the transaction I did not see any effect / see gaps on the consumer.
• This does not include network latency. Scripts & DB were on the same server

Pushing such a number of events on a single table concurrently is certainly not daily operations business so performance impact in any daily situation is proably (much) less if measurable at all.
For any situation where this might happen it pushes resource consumption from the consumer (having to deal with gaps) and database (reduced select load) to the producer (having to wait until the lock is available). Again not sure if this is really a bad thing or just something to be aware of.
Ultimateively I think this is something to preference. Using locks introduces a potential bottleneck for writes. How much of a bottleneck this would be in worst-case is hard to say as I don't have a append-only optimize database instance running on some decicated hardware. But then if I would have and were to require that kind of performance I certainly could roll with a custom implementation and disable locking the event store for inserts.

Summary: using GET_LOCK would eliminate the problem of gaps, pushing the responsbility on the write-side. As with any single write approach, performance is worse than concurrent writes but probably neglegtable for usual use-cases. That'd then be an option for the event-store maybe?

[prolic]

As I thought already. Can you try the same with a stored procedure? So a) getlock, b) start transaction, c) write data, d) commit transaction and e) release lock is only one db roundtrip?

[fritz-gerneth]

Since the amount of rows we want to insert in one transaction varies I'm not entirely sure what would be the best way for stored procedures.. the only solutions I can find for this rely in passing in the data as a string and do some string-spliting afterwards.. which is a solution I hardy would want to start with even. but then i'm no expert with stored procedures and maybe there's a better solution?