fix(batch): SendMessageBatch пишет в Redis синхронно (как SendMessage) + не персистит дубли; SaveMessagesSync
This commit is contained in:
@@ -113,6 +113,7 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
|
||||
}
|
||||
|
||||
newMsgs := make([]models.SqsMessage, 0, len(sendEntries))
|
||||
appendedIdx := make([]int, 0, len(sendEntries))
|
||||
for _, sendEntry := range sendEntries {
|
||||
msg := models.SqsMessage{MessageBody: sendEntry.MessageBody}
|
||||
if len(sendEntry.MessageAttributes) > 0 {
|
||||
@@ -130,7 +131,8 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
|
||||
fifoSeqNumber = queue.NextSequenceNumber(sendEntry.MessageGroupId)
|
||||
}
|
||||
if !queue.IsDuplicate(sendEntry.MessageDeduplicationId) {
|
||||
queue.AppendMessage(msg)
|
||||
appendedIdx = append(appendedIdx, queue.AppendMessage(msg))
|
||||
newMsgs = append(newMsgs, msg)
|
||||
} else {
|
||||
log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", sendEntry.MessageDeduplicationId, queueName)
|
||||
}
|
||||
@@ -143,12 +145,18 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
|
||||
MD5OfMessageAttributes: msg.MD5OfMessageAttributes,
|
||||
SequenceNumber: fifoSeqNumber,
|
||||
})
|
||||
newMsgs = append(newMsgs, msg)
|
||||
// Логируем только метаданные — тело сообщения не логируется (perf + security)
|
||||
log.Debugf("Queue: %s, MessageId: %s, Size: %d bytes", queueName, msg.Uuid, len(sendEntry.MessageBody))
|
||||
}
|
||||
// Персистим новые сообщения отдельно — O(batch_size*msg_size) вместо O(N*msg_size)
|
||||
persistence.SaveMessages(key, newMsgs)
|
||||
// Персистим СИНХРОННО до ответа (как одиночный SendMessage): иначе краш в окне
|
||||
// между 200 OK и async-записью потеряет подтверждённые сообщения.
|
||||
if err := persistence.SaveMessagesSync(key, newMsgs); err != nil {
|
||||
log.Errorf("SendMessageBatch: не удалось персистить батч в %s: %v", key, err)
|
||||
for _, idx := range appendedIdx {
|
||||
queue.RemoveIndex(idx)
|
||||
}
|
||||
return utils.CreateErrorResponseV1("InternalError", true)
|
||||
}
|
||||
// Метаданные очереди (FIFO state, duplicates) — без Messages
|
||||
persistence.SaveQueue(key, queue)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user