P0.3: синхронный персист SendMessage до ответа 200

Добавлен persistence.SaveMessageSync (синхронный HSet, возвращает ошибку).
SendMessage теперь пишет сообщение в Redis ДО возврата клиенту; при ошибке Redis
сообщение откатывается из памяти и возвращается InternalError (новая ошибка в
models/errors.go). Устраняет окно потери подтверждённого сообщения при краше пода
между 200 OK и асинхронной записью.
This commit is contained in:
Repinoid
2026-10-06 18:04:22 +03:00
parent 928a3fb85d
commit c19a056de3
3 changed files with 39 additions and 2 deletions
+14 -2
View File
@@ -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))