Make operation to deduplicate messages atomic #14
Conversation
for job_exeuctions, utilizing unique index on message_id.
|
Looking for voice from other people 🔈 |
| job_retry = Barbeque::JobRetry.find_or_initialize_by(message_id: @message.id) | ||
| job_retry.update!(job_execution: job_execution) | ||
| begin | ||
| job_retry = Barbeque::JobRetry.create(message_id: @message.id, job_execution: job_execution) |
There was a problem hiding this comment.
I'm not sure, but can we retry the job more than twice?
There was a problem hiding this comment.
"Retry" in barbeque means re-enqueuing a message having type: "JobRetry" to SQS. Thus it can be possible.
If we don't dedup this, there may be the situation that barbeque-console says it is retried once but actually it was retried twice. In that situation, one of their logs will be lost.
There was a problem hiding this comment.
In addition, only when deduping both JobExecution and JobRetry, we can assume that the same message isn't consumed by multiple workers. People who write code of a job won't be necessary to care about exclusive control.
There was a problem hiding this comment.
"Retry" in barbeque means re-enqueuing a message having type: "JobRetry" to SQS.
Aha, now I can know how this change will work. Looks good.
|
@k0kubun Can you release this? If not, I'll take. |
|
Done. Released as v0.2.3. |
Background
In
Barbeque::MessageHandler::JobExecution, we execute SELECT and INSERT queries for job_execution without a transaction. We didn't introduce a transaction to avoid performance regression. So it is possible that a single job is executed twice.But I noticed that we can create a record and detect duplication in a single INSERT query because we have message_id unique index.
Changes
So I changed
Barbeque::MessageHandler::JobExecutionto use such a query. I also applied it toBarbeque::MessageHandler::JobExecution.I believe this PR resolves message duplication by SQS's at-least-once delivery.