Prevent missing updates using a message broker after creating a task

Question

I'm trying to design a data updates mechanism in my micro-services architecture. For the sake of simplicity, let's assume we have two micro-services A and B, B exposes an API for creating some tasks, using simple REST, POST /tasks, which creates a task and returns a unique task identifier to query on - task_id. Then any created task can be queried on status using another API endpoint: GET /tasks/{task_id}. Now A can create tasks and use polling mechanism to track progress. The next improvement, we would like to add is "push API" - progress updates asynchronously using a message broker (e.g., RabbitMQ). Now, whenever the status has changed, B will publish a data update using a message broker and A will get this update instead of polling.

This is the expected flow:

1. A requests B to create a task synchronously
2. A subscribes to changes of tasks.{task_id}
3. B publishes a change of task_id

Steps 2 + 3 can be re-ordered causing A to miss updates or even never get any at all (if the task was completed before).

The only way to handle this race condition I can think of is to change step 2:

1. A subscribes to changes of tasks.{task_id}
2. A queries for current status GET /tasks/{task_id}
3. For any received notification we need to check that it is a newer version than the state received in the manual querying (and vice-versa).

Is there another approach or a better practice for this problem?

• marstato: Why is a constant subscription to a topic tasks not possible?
• Sawel: 1. You may need to process many tasks that are not relevant to you. 2. Even if you subscribe constantly to all tasks, you still need to sync the interested task ids and the notifications received from the queue somehow.
• amon: If the task ID can be assigned decentrally (e.g. as an UUID), then the client can select the ID and can subscribe to the topic before the task with that ID is created (swapping step 1 & 2). Some message brokers like Kafka also allow clients to read past messages, but I don't think AMQP-based brokers support this (not sure though).
• marstato: @Sawel Re. 1. So? Just filter them out. Just try it out, if the performance is not up to scratch you can still improve, but as of now you don't know that you gave a performance issue with that approach. Re. 2.: That syncing is not necessary if you have a queue per task? Again, I think the syncing is simple enough. Try it out and see how far it gets you.
• Sawel: @marstato Filtering millions of tasks doesn't make sense, this load is redundant. Even so, imagine that there is a thread that consumes the queue, while another creates tasks, so between the time the task thread informs the consumer thread about the new id, it might receive a notification and drop it

Answer 1

For terminology: For the task-completed events, B is the producer and A is the consumer.

Create a queue per consumer

When creating the task, identify the creator/consumer. Send the task-completed event to a queue specific to that consumer. This way, the consumer doesn't need to process any messages from the queue that are not relevant to it and you avoid race conditions.

While thinking about it, this may even be a good idea: Allow the task-creator/consumer to specify the queue to which the completion message should be sent to when the task is created.

• Sawel: I think it couples the producers and consumers. The producer should not know its subscribers...
• marstato: Then have the queue name be Parameter to create the task

Answer 2

It sounds like you're overcomplicating this with subscriptions to specific (and short lived) queues for individual messages. You're effectively creating a Request/Response within Queues - possible, occasionally useful, but rarely required.

I think you should set it up like so:

• Setup permanent topics/subscriptions/queues for "RequestReceived" and "ThingProcessed". A subscribes to "ThingProcessed", B subscribes to "RequestReceived".
• A receives a request
• If the request is valid, A publishes a message to a topic/queue "RequestReceived", and responds to requster with a tracking ID.
• B receives the event from it's subscription. It processes, then publishes to outcome "ThingProcessed"
• A receives B's message via it's subscription.

You can't have race conditions here, as there's no temporary queues to receive specific updates. It has the added advantages of decoupling A and B (they just talk to a Queue, they have no idea who or what is sending/receiving their messages), and also allowing you to subscribe other systems in due course to those events, if required.

If the original requester to A wants an update, they can call A's API and see the status. Alternatively, A could be setup to make a call to the original requester when receiving it's update from B (via the message queue).

Requester -> A -> Publish "RequestReceived" -> B -> Publish "ThingProcessed" -> A -> Requester