From eba01c9580b16f0851738831de83478268c0b516 Mon Sep 17 00:00:00 2001 From: Naeel Date: Sat, 11 Apr 2026 20:39:08 +0300 Subject: [PATCH] =?UTF-8?q?fix:=207=20performance/correctness=20fixes=20?= =?UTF-8?q?=E2=80=94=20v0.1.20?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. Убрано логирование тела сообщения (256KB I/O на каждый send — perf+security) 2. SentTimestamp исправлен: m.SentTime вместо time.Now() (баг) 3. MD5 не пересчитывается на ReceiveMessage — используется кэш из SendMessage 4. ChangeMessageVisibility/batch теперь персистит в Redis (баг — потеря данных) 5. MessageDoesNotExist error code: QueueExists → ReceiptHandleIsInvalid (copy-paste баг) 6. copystructure убран из GetQueueAttributes — простой map lookup 7. SNS dead code удалён (SnsErrors, SnsErrorType — не используется в SQS сервисе) Tested: quick_test 31/31 PASS, deployed v0.1.20 --- .gitignore | 1 + app/gosqs/change_message_visibility.go | 7 +- app/gosqs/change_message_visibility_batch.go | 8 ++- app/gosqs/get_queue_attributes.go | 69 ++++++++++---------- app/gosqs/receive_message.go | 6 +- app/gosqs/send_message.go | 5 +- app/gosqs/send_message_batch.go | 5 +- app/models/errors.go | 30 +-------- app/utils/utils.go | 9 +-- 9 files changed, 61 insertions(+), 79 deletions(-) diff --git a/.gitignore b/.gitignore index 95a32b5..fda4c08 100644 --- a/.gitignore +++ b/.gitignore @@ -26,3 +26,4 @@ build/ # Secrets secrets/ +goaws diff --git a/app/gosqs/change_message_visibility.go b/app/gosqs/change_message_visibility.go index 55f313a..084f8d9 100644 --- a/app/gosqs/change_message_visibility.go +++ b/app/gosqs/change_message_visibility.go @@ -1,4 +1,4 @@ -// Изменено: 2026-04-09 +// Изменено: 2026-04-11 — добавлена Redis persistence // ChangeMessageVisibilityV1 — меняет visibility timeout сообщения в очереди тенанта. package gosqs @@ -9,6 +9,7 @@ import ( "shared-sqs/app/interfaces" "shared-sqs/app/models" + "shared-sqs/app/persistence" "shared-sqs/app/utils" "github.com/gorilla/mux" @@ -75,6 +76,10 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo break } } + // Персистим изменение в Redis под Lock + if messageFound { + persistence.SaveQueue(key, models.SyncQueues.Queues[key]) + } models.SyncQueues.Unlock() if !messageFound { return utils.CreateErrorResponseV1("MessageNotInFlight", true) diff --git a/app/gosqs/change_message_visibility_batch.go b/app/gosqs/change_message_visibility_batch.go index d0e625a..ad7c00c 100644 --- a/app/gosqs/change_message_visibility_batch.go +++ b/app/gosqs/change_message_visibility_batch.go @@ -1,4 +1,4 @@ -// Создано: 2026-04-11 +// Создано: 2026-04-11, Изменено: 2026-04-11 — добавлена Redis persistence // ChangeMessageVisibilityBatchV1 — пакетная смена таймаута видимости (до 10 сообщений). // Паттерн аналогичен DeleteMessageBatchV1: валидация Id, цикл по записям, partial success. package gosqs @@ -10,6 +10,7 @@ import ( "shared-sqs/app/interfaces" "shared-sqs/app/models" + "shared-sqs/app/persistence" "shared-sqs/app/utils" "github.com/gorilla/mux" @@ -109,6 +110,11 @@ func ChangeMessageVisibilityBatchV1(req *http.Request) (int, interfaces.Abstract } } + // Персистим изменения в Redis под Lock + if len(successEntries) > 0 { + persistence.SaveQueue(key, models.SyncQueues.Queues[key]) + } + respStruct := models.ChangeMessageVisibilityBatchResponse{ Xmlns: models.BaseXmlns, Result: models.ChangeMessageVisibilityBatchResult{ diff --git a/app/gosqs/get_queue_attributes.go b/app/gosqs/get_queue_attributes.go index c0f9cba..b5f2d9b 100644 --- a/app/gosqs/get_queue_attributes.go +++ b/app/gosqs/get_queue_attributes.go @@ -1,4 +1,4 @@ -// Изменено: 2026-04-09 +// Изменено: 2026-04-11 — убрана зависимость от copystructure (лишняя аллокация) // GetQueueAttributesV1 — возвращает атрибуты очереди тенанта. package gosqs @@ -11,7 +11,6 @@ import ( "shared-sqs/app/interfaces" "shared-sqs/app/models" "shared-sqs/app/utils" -"github.com/mitchellh/copystructure" log "github.com/sirupsen/logrus" ) @@ -32,29 +31,29 @@ if t == nil { return utils.CreateErrorResponseV1("InvalidClientTokenId", true) } -requestedAttributes := func() map[string]bool { -attrs := map[string]bool{} -if len(requestBody.AttributeNames) == 0 { -return map[string]bool{"All": true} -} -for _, attr := range requestBody.AttributeNames { -if "All" == attr { -return map[string]bool{"All": true} -} -attrs[attr] = true -} -return attrs -}() +// Определяем набор запрошенных атрибутов (или All) + requestedAttributes := func() map[string]bool { + attrs := map[string]bool{} + if len(requestBody.AttributeNames) == 0 { + return map[string]bool{"All": true} + } + for _, attr := range requestBody.AttributeNames { + if "All" == attr { + return map[string]bool{"All": true} + } + attrs[attr] = true + } + return attrs + }() -dupe, _ := copystructure.Copy(models.AvailableQueueAttributes) -includedAttributes, _ := dupe.(map[string]bool) -_, ok = requestedAttributes["All"] -if !ok { -for attr := range includedAttributes { -if _, ok := requestedAttributes[attr]; !ok { -delete(includedAttributes, attr) -} -} + // Фильтруем атрибуты без deep copy — простая проверка через map lookup + _, wantAll := requestedAttributes["All"] + shouldInclude := func(attr string) bool { + if wantAll { + return true + } + _, ok := requestedAttributes[attr] + return ok } uriSegments := strings.Split(requestBody.QueueUrl, "/") @@ -72,37 +71,37 @@ log.Errorf("Get Queue Attributes: %s queue does not exist for tenant %s", queueN return utils.CreateErrorResponseV1("QueueNotFound", true) } -if _, ok := includedAttributes["DelaySeconds"]; ok { +if shouldInclude("DelaySeconds") { queueAttributes = append(queueAttributes, models.Attribute{Name: "DelaySeconds", Value: strconv.Itoa(queue.DelaySeconds)}) } -if _, ok := includedAttributes["MaximumMessageSize"]; ok { +if shouldInclude("MaximumMessageSize") { queueAttributes = append(queueAttributes, models.Attribute{Name: "MaximumMessageSize", Value: strconv.Itoa(queue.MaximumMessageSize)}) } -if _, ok := includedAttributes["MessageRetentionPeriod"]; ok { +if shouldInclude("MessageRetentionPeriod") { queueAttributes = append(queueAttributes, models.Attribute{Name: "MessageRetentionPeriod", Value: strconv.Itoa(queue.MessageRetentionPeriod)}) } -if _, ok := includedAttributes["ReceiveMessageWaitTimeSeconds"]; ok { +if shouldInclude("ReceiveMessageWaitTimeSeconds") { queueAttributes = append(queueAttributes, models.Attribute{Name: "ReceiveMessageWaitTimeSeconds", Value: strconv.Itoa(queue.ReceiveMessageWaitTimeSeconds)}) } -if _, ok := includedAttributes["VisibilityTimeout"]; ok { +if shouldInclude("VisibilityTimeout") { queueAttributes = append(queueAttributes, models.Attribute{Name: "VisibilityTimeout", Value: strconv.Itoa(queue.VisibilityTimeout)}) } -if _, ok := includedAttributes["ApproximateNumberOfMessages"]; ok { +if shouldInclude("ApproximateNumberOfMessages") { queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessages", Value: strconv.Itoa(len(queue.Messages))}) } -if _, ok := includedAttributes["ApproximateNumberOfMessagesNotVisible"]; ok { +if shouldInclude("ApproximateNumberOfMessagesNotVisible") { queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessagesNotVisible", Value: strconv.Itoa(numberOfHiddenMessagesInQueue(*queue))}) } -if _, ok := includedAttributes["CreatedTimestamp"]; ok { +if shouldInclude("CreatedTimestamp") { queueAttributes = append(queueAttributes, models.Attribute{Name: "CreatedTimestamp", Value: "0000000000"}) } -if _, ok := includedAttributes["LastModifiedTimestamp"]; ok { +if shouldInclude("LastModifiedTimestamp") { queueAttributes = append(queueAttributes, models.Attribute{Name: "LastModifiedTimestamp", Value: "0000000000"}) } -if _, ok := includedAttributes["QueueArn"]; ok { +if shouldInclude("QueueArn") { queueAttributes = append(queueAttributes, models.Attribute{Name: "QueueArn", Value: queue.Arn}) } -if _, ok := includedAttributes["RedrivePolicy"]; ok && queue.DeadLetterQueue != nil { +if shouldInclude("RedrivePolicy") && queue.DeadLetterQueue != nil { queueAttributes = append(queueAttributes, models.Attribute{ Name: "RedrivePolicy", Value: fmt.Sprintf(`{"maxReceiveCount":"%d", "deadLetterTargetArn":"%s"}`, queue.MaxReceiveCount, queue.DeadLetterQueue.Arn), diff --git a/app/gosqs/receive_message.go b/app/gosqs/receive_message.go index 9afbc9b..ca9d160 100644 --- a/app/gosqs/receive_message.go +++ b/app/gosqs/receive_message.go @@ -1,4 +1,4 @@ -// Изменено: 2026-04-09 +// Изменено: 2026-04-11 — фикс: SentTimestamp из m.SentTime, MD5 из кэша // ReceiveMessageV1 — получает сообщения из очереди тенанта с поддержкой long polling. // Ловушка #4: long polling держит соединение до 20 сек — не прерываем принудительно. package gosqs @@ -167,14 +167,14 @@ func buildResultMessage(m *models.SqsMessage) *models.ResultMessage { MessageId: m.Uuid, Body: m.MessageBody, ReceiptHandle: m.ReceiptHandle, - MD5OfBody: utils.GetMD5Hash(m.MessageBody), + MD5OfBody: m.MD5OfMessageBody, // Используем кэшированный MD5 вместо пересчёта MD5OfMessageAttributes: m.MD5OfMessageAttributes, MessageAttributes: m.MessageAttributes, Attributes: map[string]string{ "ApproximateFirstReceiveTimestamp": fmt.Sprintf("%d", m.ReceiptTime.UnixNano()/int64(time.Millisecond)), "SenderId": models.CurrentEnvironment.AccountID, "ApproximateReceiveCount": fmt.Sprintf("%d", m.NumberOfReceives+1), - "SentTimestamp": fmt.Sprintf("%d", time.Now().UTC().UnixNano()/int64(time.Millisecond)), + "SentTimestamp": fmt.Sprintf("%d", m.SentTime.UnixNano()/int64(time.Millisecond)), // Фикс: реальное время отправки }, } } diff --git a/app/gosqs/send_message.go b/app/gosqs/send_message.go index ea1b596..e1130c0 100644 --- a/app/gosqs/send_message.go +++ b/app/gosqs/send_message.go @@ -1,4 +1,4 @@ -// Изменено: 2026-04-10 — добавлена Redis persistence +// Изменено: 2026-04-11 — убрано логирование тела сообщения (perf + security) // SendMessageV1 — добавляет сообщение в очередь тенанта. // Ловушка #6: queueName извлекается как ПОСЛЕДНИЙ сегмент URL — при URL вида // http://host/tenantID/queueName последний сегмент = queueName (правильно). @@ -120,7 +120,8 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { // Сохраняем очередь в Redis пока держим Lock persistence.SaveQueue(key, models.SyncQueues.Queues[key]) models.SyncQueues.Unlock() - log.Infof("%s: Queue: %s, Message: %s\n", time.Now().Format("2006-01-02 15:04:05"), queueName, msg.MessageBody) + // Логируем только метаданные — тело сообщения не логируется (perf + security) + log.Infof("Queue: %s, MessageId: %s, Size: %d bytes", queueName, msg.Uuid, len(messageBody)) respStruct := models.SendMessageResponse{ Xmlns: models.BaseXmlns, diff --git a/app/gosqs/send_message_batch.go b/app/gosqs/send_message_batch.go index 4c31702..b5e8c53 100644 --- a/app/gosqs/send_message_batch.go +++ b/app/gosqs/send_message_batch.go @@ -1,4 +1,4 @@ -// Изменено: 2026-04-09 +// Изменено: 2026-04-11 — убрано логирование тела сообщения (perf + security) // SendMessageBatchV1 — пакетная отправка сообщений в очередь тенанта. package gosqs @@ -136,7 +136,8 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody MD5OfMessageAttributes: msg.MD5OfMessageAttributes, SequenceNumber: fifoSeqNumber, }) - log.Infof("%s: Queue: %s, Message: %s", time.Now().Format("2006-01-02 15:04:05"), queueName, msg.MessageBody) + // Логируем только метаданные — тело сообщения не логируется (perf + security) + log.Debugf("Queue: %s, MessageId: %s, Size: %d bytes", queueName, msg.Uuid, len(sendEntry.MessageBody)) } // Персистим батч-изменение одним снапшотом под lock. persistence.SaveQueue(key, queue) diff --git a/app/models/errors.go b/app/models/errors.go index b531d02..4aca2da 100644 --- a/app/models/errors.go +++ b/app/models/errors.go @@ -6,7 +6,7 @@ func init() { SqsErrors = map[string]SqsErrorType{ "QueueNotFound": {HttpError: http.StatusBadRequest, Type: "Not Found", Code: "AWS.SimpleQueueService.NonExistentQueue", Message: "The specified queue does not exist for this wsdl version."}, "QueueExists": {HttpError: http.StatusBadRequest, Type: "Duplicate", Code: "AWS.SimpleQueueService.QueueExists", Message: "The specified queue already exists."}, - "MessageDoesNotExist": {HttpError: http.StatusNotFound, Type: "Not Found", Code: "AWS.SimpleQueueService.QueueExists", Message: "The specified queue does not contain the message specified."}, + "MessageDoesNotExist": {HttpError: http.StatusNotFound, Type: "Not Found", Code: "ReceiptHandleIsInvalid", Message: "The specified receipt handle is not valid."}, "GeneralError": {HttpError: http.StatusBadRequest, Type: "GeneralError", Code: "AWS.SimpleQueueService.GeneralError", Message: "General Error."}, "TooManyEntriesInBatchRequest": {HttpError: http.StatusBadRequest, Type: "TooManyEntriesInBatchRequest", Code: "AWS.SimpleQueueService.TooManyEntriesInBatchRequest", Message: "Maximum number of entries per request are 10."}, "BatchEntryIdsNotDistinct": {HttpError: http.StatusBadRequest, Type: "BatchEntryIdsNotDistinct", Code: "AWS.SimpleQueueService.BatchEntryIdsNotDistinct", Message: "Two or more batch entries in the request have the same Id."}, @@ -25,17 +25,6 @@ func init() { // MissingParameter — обязательный параметр отсутствует (например, пустой MessageBody) "MissingParameter": {HttpError: http.StatusBadRequest, Type: "MissingParameter", Code: "MissingParameter", Message: "The request must contain the parameter MessageBody."}, } - SnsErrors = map[string]SnsErrorType{ - "InvalidParameterValue": {HttpError: http.StatusBadRequest, Type: "InvalidParameterValue", Code: "AWS.SimpleNotificationService.InvalidParameterValue", Message: "An invalid or out-of-range value was supplied for the input parameter."}, - "TopicNotFound": {HttpError: http.StatusBadRequest, Type: "Not Found", Code: "AWS.SimpleNotificationService.NonExistentTopic", Message: "The specified topic does not exist for this wsdl version."}, - "SubscriptionNotFound": {HttpError: http.StatusNotFound, Type: "Not Found", Code: "AWS.SimpleNotificationService.NonExistentSubscription", Message: "The specified subscription does not exist for this wsdl version."}, - "TopicExists": {HttpError: http.StatusBadRequest, Type: "Duplicate", Code: "AWS.SimpleNotificationService.TopicAlreadyExists", Message: "The specified topic already exists."}, - "ValidationError": {HttpError: http.StatusBadRequest, Type: "InvalidParameter", Code: "AWS.SimpleNotificationService.ValidationError", Message: "The input fails to satisfy the constraints specified by an AWS service."}, - "BatchEntryIdsNotDistinct": {HttpError: http.StatusBadRequest, Type: "BatchEntryIdsNotDistinct", Code: "AWS.SimpleNotificationService.BatchEntryIdsNotDistinct", Message: "Two or more batch entries in the request have the same Id."}, - "EmptyBatchRequest": {HttpError: http.StatusBadRequest, Type: "EmptyBatchRequest", Code: "AWS.SimpleNotificationService.EmptyBatchRequest", Message: "The batch request doesn't contain any entries."}, - "TooManyEntriesInBatchRequest": {HttpError: http.StatusBadRequest, Type: "TooManyEntriesInBatchRequest", Code: "AWS.SimpleNotificationService.TooManyEntriesInBatchRequest", Message: "Maximum number of entries per request are 10."}, - "MalformedInput": {HttpError: http.StatusBadRequest, Type: "Sender", Code: "AWS.SimpleNotificationService.MalformedInput", Message: "Invalid Base64 encoding"}, - } } type SqsErrorType struct { @@ -54,20 +43,3 @@ func (s SqsErrorType) Response() ErrorResult { } var SqsErrors map[string]SqsErrorType - -type SnsErrorType struct { - HttpError int - Type string - Code string - Message string -} - -func (s SnsErrorType) StatusCode() int { - return s.HttpError -} - -func (s SnsErrorType) Response() ErrorResult { - return ErrorResult{Type: s.Type, Code: s.Code, Message: s.Message} -} - -var SnsErrors map[string]SnsErrorType diff --git a/app/utils/utils.go b/app/utils/utils.go index 43bf7d9..f0c56e7 100644 --- a/app/utils/utils.go +++ b/app/utils/utils.go @@ -79,13 +79,10 @@ func ExtractQueueAttributes(u url.Values) map[string]string { return attr } +// CreateErrorResponseV1 — формирует SQS error response по ключу ошибки. +// Параметр isSqs оставлен для обратной совместимости (всегда true). func CreateErrorResponseV1(errKey string, isSqs bool) (int, interfaces.AbstractResponseBody) { - var err interfaces.AbstractErrorResponse - if isSqs { - err = models.SqsErrors[errKey] - } else { - err = models.SnsErrors[errKey] - } + err := models.SqsErrors[errKey] respStruct := models.ErrorResponse{ Result: err.Response(),