3

I'm trying to figure out the best solution for the below. Any help would be great.

So basically I have a service (that can be scaled horizontally), which listens on a queue. Every message received will be dispatched into a job and processed concurrently.

The job will (in same order):

  1. Generate some data based on the message payload
  2. Cache the data on Redis
  3. Send the data to another service

My issue is when another message is received for the same logical record (same table record but with altered data).
I need to make sure that the latest version of the data is cached on Redis and sent to the next service in the scenario where 2 or more messages with the same record id are being processed. Hence avoiding that a job with an old version of the payload is overwriting the latest one.

I thinking about using some distributed locking mechanism, not sure if that's efficient, especially when I want the latest version to be sent to the next service quickly as possible.
Maybe someway to cancel a job for an outdated payload instead of locking the whole job? Using Redis pubsub to communicate between the service (When scaling) or have a better way?

  • What makes a record the "old version" versus the "latest one"? Creation time? Submission time? Something else? – JimmyJames Aug 30 '18 at 13:53
  • @JimmyJames The updated time – Alan Aug 30 '18 at 13:58
  • 1
    OK perfect but what determines that? There are many times you could use: when you read it off of the queue to process it, when it was placed on a queue, when it was created in before being put on the queue (this could have any number of times to choose from) – JimmyJames Aug 30 '18 at 14:05
  • 1
    I'm sure it's not as simple as this but based on what you've stated so far checking that that timestamp of the message you are processing is not before the timestamp of the current version in the DB should suffice. In order to come up with a really robust solution, it's important to completely define the problem you are trying to solve. I think you might be trying to skip that part of the process. My intention with these questions is to help you get to that point. – JimmyJames Sep 5 '18 at 13:43
  • 1
    @Alan You keep mentioning the 'same time'. Not to get too philosophical on you, but Einstein showed that there is no such thing in general. I know it sounds silly but these ideas were formulated in part because he was evaluating patents on clock synchronization systems for railroads which isn't that different from the kind of issue you are working on here. Anyway, you say you need to avoid sending an old request overwriting a new request and then mention they could happen at the same time. If they are 'at the same time' there is no new or old. So you need to figure out what that means to you. – JimmyJames Sep 6 '18 at 14:46
1

If I were you, I would create a meta record per message containing the following components:

  • Key of the logical record
  • Timestamp of this message
  • Payload value

let each process create its record in an interim data store, and invalidates the permanent data store. then allow a housekeeping process (call it the referee) to decide on the latest version, and promotes it to be visible in the final data store.

This setup optimizes for writes.

of course this technique has scale-ability limitations, below a certain threshold it should reach eventual consistency.

when retrieving your data, you pull it from a priority retrieving system such that;

  • If permanent data store is not invalidated, select it
  • If data store is invalidated, select data from your temporary data store, sorted by timestamp descending
0

Correct me if im wrong but it sounds like you have the following data flow

Message Queue: InBox
Service1 : read message from InBox
           process InBox.Id
           save result to Redis
           send message to S1Out with S1Out.Id = Start.Id

Message Queue: S1Out
Service2 : read message from S1Out
           read data from Redis with S1Out.Id
           save data to DB

The problem being that you can get 2 InBox messages for the same Id and the redis data will be overwritten by the second request processed. Causing Service 2 to behave oddly.

Really I would like to send all the data required by Service 2 in the S1Out Message rather than have S2 lookup the data in the redis cache. (Big Messages)

This eliminates the problem and a central point of failure/bottleneck of the cache.

However. An alternate approach would be to generate a new Id for each item saved in the cache and to add that Id to the message. This allows Service 2 to pick up the correct data for the job.

The second problem of having a redundant job flowing through the queue is harder to deal with until you hit a shared resource.

In my example Service 2 saves the data to a DB. At this point it is able to check to see if its job was redundant. It knows the timestamp from the original message and the database knows the timestamp of processed messages, so the service can compare and skip or drop out of date work.

Alternatively, you could add an extra 'Router Worker' whose only job is to read from S1Out, keep a buffer of messages and remove or reorder redundant work and post to S1Out2. But this becomes another central point of failure; as if you have two of them they can't see each others buffer.

My preferred approach is to process all messages and to drop redundant effects only when they would cause a real issue. (ie, not just be more optimal of cpu usage)

  • Thanks for the answer, I'll check it and reply you asap, won't be available for the next few hours. – Alan Aug 30 '18 at 14:00
  • am actually sending the generated data payload to Service 2 (not using redis). Also, the generated data is not being save in a DB. It is just being pushed to clients through websocket by Service 2. I want to prevent Service 1 from sending an outdated payload (if let say it was slow to process) to Service 2. – Alan Aug 30 '18 at 19:25
  • its the client that can check the timestamp then – Ewan Aug 30 '18 at 21:36
  • what's best way to make sure Service 1 is always caching the latest record, and not having a slow process with an outdated payload overwrite the cache, or possibly 2 process running at the same time. Use Redis distributed locking when caching the data? – Alan Aug 31 '18 at 10:13
  • @Alan if you are sending the data in the payload then i don't understand where you cache fits. can you clarify your question? But i think my 'Router Worker' is the solution you want – Ewan Aug 31 '18 at 10:24

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.