fix(concurrency,durability): address Opus review critical findings
- create_queue: SaveQueue under per-queue lock, not global SyncQueues.Lock (fixes fatal concurrent map iteration/write on idempotent CreateQueue of existing FIFO queue) - delete_message_batch: reject empty ReceiptHandle (JSON protocol bypass deleted all unreceived msgs) - send_message/send_message_batch: register dedupId only after successful persist (retry with same dedupId no longer returns 200 without persisting) - memory_guard: global HeapAlloc guard (SHARED_SQS_MAX_HEAP_MB, 0=off) in send paths
This commit is contained in:
@@ -132,12 +132,12 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||||
if queue.LiveCount() >= MaxMessagesForQueue(queue.IsFIFO) {
|
||||
return utils.CreateErrorResponseV1("OverLimit", true)
|
||||
}
|
||||
|
||||
fifoSeqNumber := ""
|
||||
if queue.IsFIFO {
|
||||
fifoSeqNumber = queue.NextSequenceNumber(messageGroupID)
|
||||
// Глобальный предохранитель памяти (HeapAlloc) — защита от OOM суммой тенантов.
|
||||
if heapOverLimit() {
|
||||
return utils.CreateErrorResponseV1("OverLimit", true)
|
||||
}
|
||||
|
||||
fifoSeqNumber := ""
|
||||
persistErr := false
|
||||
if !queue.IsDuplicate(messageDeduplicationID) {
|
||||
idx := queue.AppendMessage(msg)
|
||||
@@ -149,15 +149,21 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||||
log.Errorf("SendMessage: не удалось персистить сообщение %s в очередь %s: %v", msg.Uuid, key, err)
|
||||
queue.RemoveIndex(idx)
|
||||
persistErr = true
|
||||
} else {
|
||||
// Регистрируем dedupId и порядковый номер ТОЛЬКО при успешной записи.
|
||||
// Иначе ретрай с тем же dedupId получил бы 200 без сохранённого сообщения
|
||||
// (InitDuplicatation после отката append помечал бы id как обработанный).
|
||||
queue.InitDuplicatation(messageDeduplicationID)
|
||||
if queue.IsFIFO {
|
||||
fifoSeqNumber = queue.NextSequenceNumber(messageGroupID)
|
||||
}
|
||||
// Метаданные очереди (FIFO state, duplicates) — без Messages
|
||||
persistence.SaveQueue(key, queue)
|
||||
}
|
||||
} else {
|
||||
log.Debugf("Duplicate message deduplicationId [%s] in queue [%s]", messageDeduplicationID, queueName)
|
||||
}
|
||||
|
||||
queue.InitDuplicatation(messageDeduplicationID)
|
||||
// Сохраняем метаданные очереди (FIFO state, duplicates) — без Messages
|
||||
persistence.SaveQueue(key, queue)
|
||||
|
||||
if persistErr {
|
||||
return utils.CreateErrorResponseV1("InternalError", true)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user