diff --git a/app/gosqs/delete_message.go b/app/gosqs/delete_message.go index 2163369..a5c0ae0 100644 --- a/app/gosqs/delete_message.go +++ b/app/gosqs/delete_message.go @@ -56,7 +56,9 @@ func DeleteMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { queue.ClearInFlight(i, receiptHandle) queue.UnlockGroup(msg.GroupID) queue.RemoveIndex(i) // tombstone за O(1) + подрезка хвоста/амортизированная компакция - delete(queue.Duplicates, msg.DeduplicationID) + // ВАЖНО: запись в Duplicates НЕ удаляем — AWS держит dedup-окно 5 минут + // независимо от удаления сообщения (иначе повторный send с тем же + // dedupId доставлялся бы повторно). Чистит только истечение окна (PeriodicTasks). // Удаляем одно сообщение из Redis — O(1) persistence.DeleteMessagePersist(key, msg.Uuid) // Метаданные очереди (duplicates, FIFO state) diff --git a/app/gosqs/delete_message_batch.go b/app/gosqs/delete_message_batch.go index 02de28b..9c0a223 100644 --- a/app/gosqs/delete_message_batch.go +++ b/app/gosqs/delete_message_batch.go @@ -93,10 +93,10 @@ func DeleteMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBo } log.Debugf("FIFO Queue %s unlocking group %s:", queueName, message.GroupID) // Читаем поля ДО RemoveIndex: компакция может перезаписать слот. - groupID, dedupID, uuid := message.GroupID, message.DeduplicationID, message.Uuid + groupID, uuid := message.GroupID, message.Uuid queue.ClearInFlight(i, message.ReceiptHandle) queue.UnlockGroup(groupID) - delete(queue.Duplicates, dedupID) + // Duplicates НЕ чистим — dedup-окно 5 минут живёт независимо от удаления (AWS). de.Deleted = true deletedEntries = append(deletedEntries, models.DeleteMessageBatchResultEntry{Id: de.Id}) deletedUuids = append(deletedUuids, uuid) diff --git a/app/gosqs/send_message.go b/app/gosqs/send_message.go index f7b3bba..d9fb63b 100644 --- a/app/gosqs/send_message.go +++ b/app/gosqs/send_message.go @@ -92,6 +92,7 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { maxMessageSize := queue.MaximumMessageSize currentMsgCount := queue.LiveCount() queueIsFIFO := queue.IsFIFO + contentBasedDedup := queue.ContentBasedDeduplication delaySecs := queue.DelaySeconds // FIFO-очередь БЕЗ MessageGroupId — ошибка: AWS требует GroupId для каждой @@ -100,6 +101,12 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { if queueIsFIFO && messageGroupID == "" { return utils.CreateErrorResponseV1("MissingParameter", true) } + // ContentBasedDeduplication: если MessageDeduplicationId не передан, AWS сам + // вычисляет его из тела сообщения (SHA-256). Без этого включённый флаг + // ничего не делал бы и FIFO-дедупликация не работала. + if queueIsFIFO && contentBasedDedup && messageDeduplicationID == "" { + messageDeduplicationID = utils.GetSHA256Hash(messageBody) + } if maxMessageSize > 0 && len(messageBody) > maxMessageSize { return utils.CreateErrorResponseV1("MessageTooBig", true) diff --git a/app/gosqs/send_message_batch.go b/app/gosqs/send_message_batch.go index 81207ca..72ec79a 100644 --- a/app/gosqs/send_message_batch.go +++ b/app/gosqs/send_message_batch.go @@ -127,7 +127,13 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody } msg.MD5OfMessageBody = utils.GetMD5Hash(sendEntry.MessageBody) msg.GroupID = sendEntry.MessageGroupId - msg.DeduplicationID = sendEntry.MessageDeduplicationId + // ContentBasedDeduplication: dedupId вычисляется из тела (SHA-256), + // если не передан явно (AWS-семантика FIFO). + dedupID := sendEntry.MessageDeduplicationId + if queue.IsFIFO && queue.ContentBasedDeduplication && dedupID == "" { + dedupID = utils.GetSHA256Hash(sendEntry.MessageBody) + } + msg.DeduplicationID = dedupID msg.Uuid = uuid.NewString() msg.SentTime = time.Now() @@ -135,12 +141,12 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody if queue.IsFIFO { fifoSeqNumber = queue.NextSequenceNumber(sendEntry.MessageGroupId) } - if !queue.IsDuplicate(sendEntry.MessageDeduplicationId) { + if !queue.IsDuplicate(dedupID) { appendedIdx = append(appendedIdx, queue.AppendMessage(msg)) newMsgs = append(newMsgs, msg) - appendedDedupIDs = append(appendedDedupIDs, sendEntry.MessageDeduplicationId) + appendedDedupIDs = append(appendedDedupIDs, dedupID) } else { - log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", sendEntry.MessageDeduplicationId, queueName) + log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", dedupID, queueName) } sentEntries = append(sentEntries, models.SendMessageBatchResultEntry{ diff --git a/app/utils/utils.go b/app/utils/utils.go index f0c56e7..a4bb288 100644 --- a/app/utils/utils.go +++ b/app/utils/utils.go @@ -2,6 +2,7 @@ package utils import ( "crypto/md5" + "crypto/sha256" "encoding/base64" "encoding/binary" "encoding/hex" @@ -97,6 +98,13 @@ func GetMD5Hash(text string) string { return hex.EncodeToString(hasher.Sum(nil)) } +// GetSHA256Hash — SHA-256 (hex) от значения. Используется для +// ContentBasedDeduplication: AWS сам вычисляет MessageDeduplicationId из тела. +func GetSHA256Hash(text string) string { + sum := sha256.Sum256([]byte(text)) + return hex.EncodeToString(sum[:]) +} + func HashAttributes(attributes map[string]models.MessageAttribute) string { hasher := md5.New()