I have an interesting thing that I do not understand I am hoping someone can explain what I see. I will try to make simple my explanation.

1. There is a Phoenix PubSub topic for IDs.
2. There is a GenStage producer module that has a handle_info to listen for these messages.
3. Next, there is a GenStage producer consumer with a handle_events function that receives the messages (the IDs) and then it checks the mongo database to see if we have this ID. If we already have a record for this ID, we filter this message out like Enum.reject
4. Next there is another GenStage producer-consumer that will do an API lookup to external service to look up this ID. The message now holds the full response data from the API request.
5. Finally, there is a GenStage consumer with handle_events that receives the data and saves the API request data in mongo database.

I hope I have explained this data pipeline clearly!

The interesting problem I am seeing can sometimes happen when 2 messages contain the same ID, something fast like this:

iex> (
...> Phoenix.PubSub.broadcast(:my_pubsub, "ids", "ID123")
...> Phoenix.PubSub.broadcast(:my_pubsub, "ids", "ID123")
...> )

What I expect is that the second message checks the database and it sees that there is already a record for this ID. But actually what I see is that both first and second messages check the database and both times they do not see the record! So instead of 1 API lookups, I have 2, and 2 times the record attempts to write to the database.

Why is this? Are all things happening in parallel? Is there a good solution to this problem? Thank you for explanations!