diff --git a/app/gosqs/send_message.go b/app/gosqs/send_message.go index f764845..97bcf7a 100644 --- a/app/gosqs/send_message.go +++ b/app/gosqs/send_message.go @@ -148,10 +148,18 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { fifoSeqNumber = queue.NextSequenceNumber(messageGroupID) } + persistErr := false if !queue.IsDuplicate(messageDeduplicationID) { queue.Messages = append(queue.Messages, msg) - // Персистим одно сообщение отдельно — O(msg_size) вместо O(N*msg_size) - persistence.SaveMessage(key, &msg) + // Персистим сообщение СИНХРОННО до ответа клиенту: краш пода в окне + // между 200 OK и async-записью приводил бы к потере подтверждённого + // сообщения. При ошибке — откатываем append (не оставляем в памяти + // сообщение, которого нет в Redis). + if err := persistence.SaveMessageSync(key, &msg); err != nil { + log.Errorf("SendMessage: не удалось персистить сообщение %s в очередь %s: %v", msg.Uuid, key, err) + queue.Messages = queue.Messages[:len(queue.Messages)-1] + persistErr = true + } } else { log.Debugf("Duplicate message deduplicationId [%s] in queue [%s]", messageDeduplicationID, queueName) } @@ -160,6 +168,10 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { // Сохраняем метаданные очереди (FIFO state, duplicates) — без Messages persistence.SaveQueue(key, queue) models.SyncQueues.Unlock() + + if persistErr { + return utils.CreateErrorResponseV1("InternalError", true) + } // Логируем только метаданные — тело сообщения не логируется (perf + security) log.Infof("Queue: %s, MessageId: %s, Size: %d bytes", queueName, msg.Uuid, len(messageBody)) diff --git a/app/models/errors.go b/app/models/errors.go index d52b344..e3c4dc2 100644 --- a/app/models/errors.go +++ b/app/models/errors.go @@ -30,6 +30,8 @@ func init() { "PurgeQueueInProgress": {HttpError: http.StatusForbidden, Type: "Sender", Code: "AWS.SimpleQueueService.PurgeQueueInProgress", Message: "Only one PurgeQueue operation on a queue is allowed every 60 seconds."}, // OverLimit — превышен лимит сообщений в очереди (OOM-защита) "OverLimit": {HttpError: http.StatusBadRequest, Type: "OverLimit", Code: "OverLimit", Message: "The queue contains the maximum number of messages."}, + // InternalError — внутренняя ошибка сервера (например, не удалось персистить сообщение) + "InternalError": {HttpError: http.StatusInternalServerError, Type: "InternalError", Code: "AWS.SimpleQueueService.InternalError", Message: "We encountered an internal error. Please try again."}, } } diff --git a/app/persistence/redis.go b/app/persistence/redis.go index c94edf2..3db85ed 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -150,6 +150,29 @@ func SaveMessage(queueKey string, msg *models.SqsMessage) { }) } +// SaveMessageSync — сохраняет ОДНО сообщение в Redis СИНХРОННО и возвращает ошибку. +// Используется на пути SendMessage: ответ 200 клиенту должен выдаваться ТОЛЬКО +// после фактической записи сообщения в Redis, иначе краш пода в окне между +// ответом и async-записью приведёт к потере подтверждённого сообщения. +// Ключ: ssq:msg:{queueKey}, поле: msg.Uuid, значение: JSON(SqsMessage). +// Если Client == nil (memory-only режим) — возвращает nil (персистить некуда). +func SaveMessageSync(queueKey string, msg *models.SqsMessage) error { + if Client == nil { + return nil + } + data, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("marshal message %s/%s: %w", queueKey, msg.Uuid, err) + } + hashKey := redisMsgHashPrefix + queueKey + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := Client.HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil { + return fmt.Errorf("HSet message %s/%s: %w", queueKey, msg.Uuid, err) + } + return nil +} + // SaveMessages — сохраняет несколько сообщений в Redis одним pipeline. // Вызывать под Lock. func SaveMessages(queueKey string, msgs []models.SqsMessage) {