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)