1

I am trying to improve an event-driven processing system, which is having a few problems because the events are not guaranteed to arrive in the correct chronological sequence. This is due to batching and caching upstream which is currently out of my control.

The sequence errors are not a complete disaster for my processor, mainly because it's FSM oriented and so copes with "weird" transitions and mostly ignores them. Also the events are timestamped and expected to be delayed so we are used to reconstructing history. However there are a few cases which cause unwanted behaviour or unnecessary double-processing.

I'm looking for a technique which can either somehow queue up, group the events and sort them before processing, or perhaps identify "transactions" in the stream.

An example correct event stream would be like this, noting that all events are timestamped:

10:01 User Session Start (user A)
10:02 Device event 1
10:03 Device event 2
10:10 User Session End (user A)
10:15 User Session Start (user A)
10:16 User Session End (user A)
10:32 User Session Start (user B)
10:34 Device event 3
10:35 Device event 4
10:50 User Session End (user B)

My downstream processor keeps track of who is using a device, but also needs to correlate the other device events with the users. It does that by keeping state of the sessions, while receiving the other events.

Each event is in practise processed by different message queue workers, with a central database. So there are potential race hazards, but that's not the focus of this question.

The problems arise when the same stream arrives like this, where the ... indicate gaps between the three "batches" as they are received much later.

10:10 User Session End (user A)
10:01 User Session Start (user A)
10:02 Device event 1
10:03 Device event 2
...
10:16 User Session End (user A)
10:15 User Session Start (user A)
10:34 Device event 3
10:35 Device event 4
...
10:50 User Session End (user B)
10:32 User Session Start (user B)

I am particularly interested in "the final device event in a session". So here I need the 10:10 session and + the 10:03 Device event 2, to complete the picture. I know that any device event timeboxed between 10:01 and 10:10 is "owned" by user A, so when I receive device event 2, I can correlate it - OK. When I receive the 10:01 start event I can ignore it, as I already saw the corresponding end (just annoying). When I receive device event 1, I can't tell if it's the final one or not, so I process it. Then I receive device event 2 immediately after and re-do the same work, update the state to presume this is the last one. I cannot predict if there will be any more device events coming, so the FSM has to just remain with that assumption - which in this case is correct.

The next batch is harder to deal with. I get a second "empty" session from user A - not a problem in itself. Then I get some device events out of sequence which are for the user B session which I've not received yet. This isn't a critical problem, I can update the associated device model with this information, but cannot complete the processing yet.

Eventually the user B events arrive and I can correlate back with the device events, again ignoring "older" events where possible.

You can hopefully see this adds a lot of difficulty to the processing and is probably leading to some missing cases.

What can I do to massage this event stream to make it more processable?

I have been thinking about:

  • event sourcing (but it requires correct sequence)
  • re-buffering the queue for X minutes (but I still can't be sure how long)
  • implementing something like the Nagle Algorithm for chunking and pause/gap detection
  • Combine all the workers into one, with an FSM (mirroring the session-boxing) which then outputs the events once they've satisfied the inter-dependency sequence checks
  • Don't fix the queue and implement a random-order resilient processor

Because I can make some assumptions about the likely contents of the stream, I can considering a "transaction detector" or making no assumptions just a more generalised "stream re-order" approach.

I know sequence numbers would solve it easily, but as mentioned I cannot presently modify the upstream publisher.

I am not looking for an entire solution - just pointers to algorithms or techniques used for this class of problem, so I can do further research.

Update Clarification on conditions:

  • upper bound on the event delivery delay: No. Most devices report their data at the end of a user session, and the events are delivered relatively quickly. However some go offline for a while and then sync up later. In some cases the data arrives so late it's beyond the useful processing time period and we have to discard it. Often it arrives a bit late, and we have to provide live updates to administrators during time-critical and realtime workflows, to correct the processing outcomes.
  • if events are guaranteed to have correct timestamps : Yes
  • will be delivered eventually: Yes
  • can processing output be corrected: Yes this happens via the FSMs and is a great feature when it makes sense, but looks strange when it "goes backwards"
  • should output be produced as soon as you can - Sometimes. There are some time constraints where the processing needs to happen before a fixed future moment. A long time before that moment we're in no rush and could queue up the data. But when that moment of decision looms, and we cross timing thresholds, we now have to make choices and trigger workflows etc. After that time being realtime is important (in minutes not seconds).
  • Is associating events with the wrong sessions acceptable - No, that would be a major functional failure. However the sequence problems don't cause this, due to various logic and validation in the processor.
  • 3
    The first thing to find out is whether there is an upper bound on the event delivery delay, and if events are guaranteed to have correct timestamps and will be delivered eventually. If at least one of these conditions are false, you cannot possibly guarantee correct output, and you should settle for a best-effort solution. If all conditions are met, re-buffering and sorting the events until the event delivery timeout should be enough to get correct output. – Hans-Martin Mosner Mar 25 at 6:49
  • 1
    For a best-effort solution, you need to know which requirements should be met, and which compromises can be made. For example, should output be produced as soon as you can, and corrected later when additional events are processed? Is associating events with the wrong sessions acceptable? And so on... – Hans-Martin Mosner Mar 25 at 6:50
  • I would argue that the proper answer depends ultimately on what you need this information for. If the purpose is to simply log events sequentially, then by keeping items in memory, you risk that should a system failure occur, you don't log any of it, in which case the best solution is to log as you receive the events and if anything perform post-processing later. Even if this is not your case, putting a complex system into place should be weighed against simply leaving it out of order and having a simpler system. It may be better to simply process it at a later period. – Neil Mar 25 at 8:15
  • @Hans-MartinMosner good points - I have added some clarification. – scipilot Mar 25 at 8:20
  • @Neil The stream is used to provide realtime intel to administrators who need to respond to certain user behaviour. If we could look back a day later yes it could all be sorted out, but it needs adjustment on the fly. For simplicity assume it needs to be realtime to the nearest minute or two. – scipilot Mar 25 at 8:23
2

SLAs

Service Level Agreements are perhaps not the most technically interesting approach, but nail down how long the data will be tolerated to be incomplete for. Your data will likely be feeding downstream report/decision making systems and you need to be sure it has the correct picture from upstream.

As a consequence this gives a definitive end-state for your processing. Either:

  • an error has been detected and reporting/fudging will take place,
  • the data is correct and can now be ignored for general processing, or
  • the upstream system owners will come to you with the rest of the data, after having had a bad hair day.

Buckets

Those session events can be used to form buckets into which the other events fall. As you only care about the most recent, the bucket need only hold the most recent message.

Track the last time the bucket was modified/checked. When it becomes sufficiently old, its presumably stable, and can be evaluated.

Incomplete buckets shouldn't be processed. (Though after sometime they should flag an alert.)

If a message doesn't fall into a bucket then it needs to be placed in a sorted list. When a new bucket is created the list can be intersected to fill the bucket.

Markov Chain Analysis

The data feed looks like a Markov Chain: Several underlying models being spliced together. If you can safely assume that each model delivers its contents serially then you can break the feed back into its constituent models and make an educated guess as to whether or not all data has been flushed or not.

Presumably when a cache flushes the contents were chronologically ordered. Hence whenever time jumps, particularly backwards, then its a different data source (aka underlying model).

Also presumably each cache services different devices, so when a device appears it indicates one data source. If several devices always appear in chronological order then presumably they are from the same data source. By extension if other details about the events are considered, some subset of them will always appear chronologically hence they represent a single data-source.

You can heuristically stipulate that all data has been received from all data-sources up to a given time if the chronology of the guessed data-sources have each an event at or past the given time.

  • This could work, but it's lacking error handling. Considering that the worst case scenario is "processing the same request twice", an argument could be said for simply leaving it as is. – Neil Mar 25 at 8:18
  • By SLA you mean Service Level Agreement? Or is this another term? – scipilot Mar 25 at 8:31
  • Markov Chain looks interesting (although slightly overwhelming). I am trying to avoid knowing anything about the upstream provider and specifically "fixing" it, as it's likely to change which would then break me. But I do in fact know more about why some of the data is actually reversed and mixed with other types - that's mainly due to processing delays in different event types, rather than different devices, and due to to parallel event message queues. I like the probabilistic approach as I think there are some unknowables in there. – scipilot Mar 25 at 8:40
  • @scipilot good point, I deed mean Service Level Agreement. I've updated my answer to detail that. – Kain0_0 Mar 25 at 22:47
  • @Neil Yes, you would need to handle errors. The problem with not trying to rationalise the data and reprocess it instead comes down to what is meant by "processing a request". If its an idempotent and/or reversible action then your fine. But when its an irreversible action or it is reasonably expensive to undo a previous decision then you need to be careful, or atleast need to minimise the opportunities for mistakes. – Kain0_0 Mar 25 at 22:52

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.