diff --git a/app/gosqs/create_queue.go b/app/gosqs/create_queue.go index 7ab8293..b3c29a5 100644 --- a/app/gosqs/create_queue.go +++ b/app/gosqs/create_queue.go @@ -71,7 +71,8 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { } models.SyncQueues.Lock() - if _, exists := models.SyncQueues.Queues[key]; !exists { + existing, exists := models.SyncQueues.Queues[key] + if !exists { // Проверка лимита очередей тенанта if t.MaxQueues > 0 && countTenantQueues(t.AccessKey) >= t.MaxQueues { models.SyncQueues.Unlock() @@ -79,10 +80,18 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { } log.Infof("Creating Queue: %s (tenant: %s)", queueName, t.ID) models.SyncQueues.AddQueue(key, queue) + existing = queue } - // Сохраняем очередь в Redis пока держим Lock — консистентный снапшот - persistence.SaveQueue(key, models.SyncQueues.Queues[key]) models.SyncQueues.Unlock() + + // SaveQueue маршалит map'ы очереди (Duplicates/FIFO*/Tags) в JSON. Это нужно + // делать под per-queue блокировкой, а НЕ под глобальным SyncQueues.Lock: + // мутаторы (send/tag/CMV) пишут в эти map под queue.mu, и идемпотентный + // CreateQueue существующей очереди без queue.mu даёт fatal + // "concurrent map iteration and map write" (краш всего процесса). + existing.LockMu() + persistence.SaveQueue(key, existing) + existing.UnlockMu() respStruct := models.CreateQueueResponse{ Xmlns: models.BaseXmlns, Result: models.CreateQueueResult{QueueUrl: queueUrl}, diff --git a/app/gosqs/delete_message_batch.go b/app/gosqs/delete_message_batch.go index 32ef866..02de28b 100644 --- a/app/gosqs/delete_message_batch.go +++ b/app/gosqs/delete_message_batch.go @@ -54,6 +54,12 @@ func DeleteMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBo return utils.CreateErrorResponseV1("BatchEntryIdsNotDistinct", true) } ids[v.Id] = true + // Пустой ReceiptHandle опасен: в JSON-протоколе (x-amz-json-1.0) он не + // отсекается парсером и совпадает с ReceiptHandle НЕвыданных сообщений (""), + // что удаляло бы все такие сообщения очереди одним запросом. + if v.ReceiptHandle == "" { + return utils.CreateErrorResponseV1("InvalidParameterValue", true) + } } queue := models.SyncQueues.GetLocked(key) diff --git a/app/gosqs/memory_guard.go b/app/gosqs/memory_guard.go new file mode 100644 index 0000000..1bd7dc2 --- /dev/null +++ b/app/gosqs/memory_guard.go @@ -0,0 +1,37 @@ +// app/gosqs/memory_guard.go +// Глобальный предохранитель памяти процесса (защита от OOM одним тенантом). +// +// Лимиты по числу сообщений (MaxMessagesForQueue) не ограничивают СУММАРНЫЙ +// объём: 120k × 256 КБ ≈ 30 ГБ на одну очередь. Этот предохранитель отказывает +// в SendMessage/SendMessageBatch по HeapAlloc выше порога — единственная защита +// процесса от суммы всех тенантов. +// +// Задаётся env `SHARED_SQS_MAX_HEAP_MB` (0 = выключен). По умолчанию выключен: +// конкретное число зависит от RAM ВМ и целевого числа тенантов (решение владельца). +// Рекомендация на старт: ~70% от RAM ВМ, напр. 1024 для 1.5 ГБ ВМ. +package gosqs + +import ( + "os" + "runtime" + "strconv" +) + +// maxHeapBytes — порог HeapAlloc в байтах; 0 = предохранитель выключен. +var maxHeapBytes = func() int64 { + mb, err := strconv.ParseInt(os.Getenv("SHARED_SQS_MAX_HEAP_MB"), 10, 64) + if err != nil || mb <= 0 { + return 0 + } + return mb * 1024 * 1024 +}() + +// heapOverLimit — true, если текущий HeapAlloc превышает порог. +func heapOverLimit() bool { + if maxHeapBytes <= 0 { + return false + } + var ms runtime.MemStats + runtime.ReadMemStats(&ms) + return int64(ms.HeapAlloc) > maxHeapBytes +} diff --git a/app/gosqs/send_message.go b/app/gosqs/send_message.go index 1b4a094..f7b3bba 100644 --- a/app/gosqs/send_message.go +++ b/app/gosqs/send_message.go @@ -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) } diff --git a/app/gosqs/send_message_batch.go b/app/gosqs/send_message_batch.go index 3fa6d20..81207ca 100644 --- a/app/gosqs/send_message_batch.go +++ b/app/gosqs/send_message_batch.go @@ -111,9 +111,14 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody if queue.LiveCount()+len(sendEntries) > MaxMessagesForQueue(queue.IsFIFO) { return utils.CreateErrorResponseV1("OverLimit", true) } + // Глобальный предохранитель памяти (HeapAlloc) — защита от OOM суммой тенантов. + if heapOverLimit() { + return utils.CreateErrorResponseV1("OverLimit", true) + } newMsgs := make([]models.SqsMessage, 0, len(sendEntries)) appendedIdx := make([]int, 0, len(sendEntries)) + appendedDedupIDs := make([]string, 0, len(sendEntries)) for _, sendEntry := range sendEntries { msg := models.SqsMessage{MessageBody: sendEntry.MessageBody} if len(sendEntry.MessageAttributes) > 0 { @@ -133,10 +138,10 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody if !queue.IsDuplicate(sendEntry.MessageDeduplicationId) { appendedIdx = append(appendedIdx, queue.AppendMessage(msg)) newMsgs = append(newMsgs, msg) + appendedDedupIDs = append(appendedDedupIDs, sendEntry.MessageDeduplicationId) } else { log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", sendEntry.MessageDeduplicationId, queueName) } - queue.InitDuplicatation(sendEntry.MessageDeduplicationId) sentEntries = append(sentEntries, models.SendMessageBatchResultEntry{ Id: sendEntry.Id, @@ -157,6 +162,11 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody } return utils.CreateErrorResponseV1("InternalError", true) } + // Регистрируем dedupId ТОЛЬКО после успешной записи — иначе ретрай с теми же + // dedupId получил бы 200 без сохранённых сообщений. + for _, id := range appendedDedupIDs { + queue.InitDuplicatation(id) + } // Метаданные очереди (FIFO state, duplicates) — без Messages persistence.SaveQueue(key, queue)