From c19a056de373afff6f1985b7c79b1c78b3f9e430 Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Tue, 6 Oct 2026 18:04:22 +0300
Subject: [PATCH] =?UTF-8?q?P0.3:=20=D1=81=D0=B8=D0=BD=D1=85=D1=80=D0=BE?=
 =?UTF-8?q?=D0=BD=D0=BD=D1=8B=D0=B9=20=D0=BF=D0=B5=D1=80=D1=81=D0=B8=D1=81?=
 =?UTF-8?q?=D1=82=20SendMessage=20=D0=B4=D0=BE=20=D0=BE=D1=82=D0=B2=D0=B5?=
 =?UTF-8?q?=D1=82=D0=B0=20200?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Добавлен persistence.SaveMessageSync (синхронный HSet, возвращает ошибку).
SendMessage теперь пишет сообщение в Redis ДО возврата клиенту; при ошибке Redis
сообщение откатывается из памяти и возвращается InternalError (новая ошибка в
models/errors.go). Устраняет окно потери подтверждённого сообщения при краше пода
между 200 OK и асинхронной записью.
---
 app/gosqs/send_message.go | 16 ++++++++++++++--
 app/models/errors.go      |  2 ++
 app/persistence/redis.go  | 23 +++++++++++++++++++++++
 3 files changed, 39 insertions(+), 2 deletions(-)

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) {