Как организовать дедупликацию сообщений?

Question

[alexg-nn]

Привет!

На уровне сервиса хочу организовать дедупликацию входящих от брокера сообщений.

Вводные:
1. У сообщения есть уникальный ID
2. У сообщения есть дата отправки продюсером.

Задачи:
1. Отклонить сообщение, если ID уже зарегистрирован.
2. Отклонить сообщение, если его дата старее, чем последняя дата принятого сообщения (этим добиваемся исключения ошибок в очерёдности доставки).

Решения:
Табличка Mysql видится вроде бы достаточно простой:
| ID | Date | Message data |

Первая задача решается просто из природы primary key во время операции INSERT.
Как решить вторую, чтобы избежать race condition? Я так понимаю, без каких-то локов тут не обойтись? В идеале хотелось бы видеть какую то атомарную операцию "вставил новую строку или отклонил". Смотрел на вариант INSERT...SELECT, но я так понимаю, между селектом и вставкой может вклиниться другая операция вставки и всё пойдёт насмарку.

Comments

[alexg-nn]

Например, при использовании отложенных сообщений. Если что-то пойдёт не так в брокере, например, и он не успеет втиснуть в очередь сообщение в правильном порядке дат.

Answer 1

[Rsa97]

Можно повесить триггер BEFORE INSERT, проверять в нём дату и выбрасывать при необходимости ошибку.
Можно так:

INSERT INTO `table` (`id`, `date`, `message`)
  (
    SELECT `t`.`id`, `t`.`date`, `t`.`message`
      FROM (SELECT :id AS `id`, :date AS `date`, :message AS `message`) AS `t`
      JOIN (SELECT MAX(`date`) AS `date` FROM `table`) AS `m`
      WHERE `t`.`date` > `m`.`date`
  )

Comments

[alexg-nn]

Я так понимаю, это будет правильно работать только если обернуть всё в транзакцию с уровнем изоляции SERIALIZABLE? Иначе как избежать фантомного чтения во всех этих вложенных селектах при параллельном инсерте.

Answer 2

[d'Ivan]

Можно вот так производить вставки:

INSERT INTO `table1` (`date`, `message`) VALUES (?, ?)
ON DUPLICATE KEY UPDATE
    message = IF(VALUES(`date`) > `date`, VALUES(message), message),
    `date`  = IF(VALUES(`date`) > `date`, VALUES(`date`), `date`)

Это позволяет иметь много консьюмеров, обновляющих данные в БД.
IF можно заменить на конструкцию CASE

Comments

[alexg-nn]

Это решает проблему вставки сообщений с одним и тем же ID, но разной датой. У меня же задача в принципе не вставлять сообщение, если в таблице уже есть сообщение (с любым ID) новее прилетевшего. То есть по сути, не пускать сообщения из прошлого относительно последнего зарегистрированного сообщения.

Answer 3

[alexg-nn]

Пока не придумал ничего лучше, чем блокирующую транзакцию

SET TRANSACTION ISOLATION LEVEL SERIALIZABLE
BEGIN TRAN
INSERT INTO messages (id,date,msg) VALUES(?,?,?)
SELECT id FROM messages ORDER BY date DESC LIMIT 1
COMMIT TRAN

И потом на уровне приложения проверить, что ID из последнего селекта равен ID переданому в инсерт.

Comments

[d'Ivan]

Пока не придумал ничего лучше, чем блокирующую транзакцию

Лучше не торопиться с блокировками. Они еще дадут о себе знать позже.

[alexg-nn]

Роман Мирр, где-то ведь эта блокировка должна случиться. Или на уровне БД, или на уровне приложения. Мне кажется, у сервера Mysql это лучше выйдет сделать, нежели мне в php использовать какие то семафоры.