From fa47026400ffc7d57f2633483509b600e04f882c Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sun, 11 Oct 2026 08:58:30 +0300 Subject: [PATCH] =?UTF-8?q?SQS:=20dedup=20=D0=B2=D0=BD=D1=83=D1=82=D1=80?= =?UTF-8?q?=D0=B8=20=D1=86=D0=B8=D0=BA=D0=BB=D0=B0=20SendMessageBatch=20(N?= =?UTF-8?q?extSequenceNumber=20=D1=82=D0=BE=D0=BB=D1=8C=D0=BA=D0=BE=20?= =?UTF-8?q?=D0=B4=D0=BB=D1=8F=20=D0=BD=D0=B5-=D0=B4=D1=83=D0=B1=D0=BB?= =?UTF-8?q?=D0=B8=D0=BA=D0=B0=D1=82=D0=B0,=20=D0=BE=D1=82=D0=BA=D0=B0?= =?UTF-8?q?=D1=82=20dedup=20=D0=BF=D1=80=D0=B8=20=D0=BE=D1=88=D0=B8=D0=B1?= =?UTF-8?q?=D0=BA=D0=B5=20=D0=BF=D0=B5=D1=80=D1=81=D0=B8=D1=81=D1=82=D0=B0?= =?UTF-8?q?,=20per-entry=20DelaySeconds)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/gosqs/send_message_batch.go | 38 ++++++++++++++++++++++----------- 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/app/gosqs/send_message_batch.go b/app/gosqs/send_message_batch.go index 72ec79a..4fc549d 100644 --- a/app/gosqs/send_message_batch.go +++ b/app/gosqs/send_message_batch.go @@ -127,6 +127,7 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody } msg.MD5OfMessageBody = utils.GetMD5Hash(sendEntry.MessageBody) msg.GroupID = sendEntry.MessageGroupId + msg.DelaySecs = sendEntry.DelaySeconds // ContentBasedDeduplication: dedupId вычисляется из тела (SHA-256), // если не передан явно (AWS-семантика FIFO). dedupID := sendEntry.MessageDeduplicationId @@ -137,17 +138,30 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody msg.Uuid = uuid.NewString() msg.SentTime = time.Now() + // Дубликат проверяем ДО NextSequenceNumber/Append: иначе seq инкрементировался + // бы и для отброшенного дубликата, а два entry с одним dedupID в одном батче + // оба добавились бы (InitDuplicatation был после цикла — ревью 2026-10-11). + if queue.IsDuplicate(dedupID) { + log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", dedupID, queueName) + sentEntries = append(sentEntries, models.SendMessageBatchResultEntry{ + Id: sendEntry.Id, + MessageId: msg.Uuid, + MD5OfMessageBody: msg.MD5OfMessageBody, + MD5OfMessageAttributes: msg.MD5OfMessageAttributes, + SequenceNumber: "", + }) + continue + } + fifoSeqNumber := "" if queue.IsFIFO { fifoSeqNumber = queue.NextSequenceNumber(sendEntry.MessageGroupId) } - if !queue.IsDuplicate(dedupID) { - appendedIdx = append(appendedIdx, queue.AppendMessage(msg)) - newMsgs = append(newMsgs, msg) - appendedDedupIDs = append(appendedDedupIDs, dedupID) - } else { - log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", dedupID, queueName) - } + // Регистрируем dedupId сразу — второй entry с тем же id увидит его. + queue.InitDuplicatation(dedupID) + appendedIdx = append(appendedIdx, queue.AppendMessage(msg)) + newMsgs = append(newMsgs, msg) + appendedDedupIDs = append(appendedDedupIDs, dedupID) sentEntries = append(sentEntries, models.SendMessageBatchResultEntry{ Id: sendEntry.Id, @@ -166,13 +180,13 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody for _, idx := range appendedIdx { queue.RemoveIndex(idx) } + // Откат dedup-регистраций: иначе retry с тем же dedupId получил бы + // «duplicate» без сохранённых сообщений. + for _, id := range appendedDedupIDs { + delete(queue.Duplicates, id) + } return utils.CreateErrorResponseV1("InternalError", true) } - // Регистрируем dedupId ТОЛЬКО после успешной записи — иначе ретрай с теми же - // dedupId получил бы 200 без сохранённых сообщений. - for _, id := range appendedDedupIDs { - queue.InitDuplicatation(id) - } // Метаданные очереди (FIFO state, duplicates) — без Messages persistence.SaveQueue(key, queue)