P1.2: индексация сообщений — tombstone-удаление O(1), receipt/uuid-индексы, tick по in-flight
- models.Queue: deadIdx/receiptIdx/uuidIdx/liveCount/deadCount + примитивы (AppendMessage/RemoveIndex/Compact/CompactIfNeeded/Reset/EnsureIndexes/ IndexOfReceipt/IndexOfUUID/MarkInFlight/ClearInFlight/InFlightIndexes/LiveCount). Удаление — tombstone за O(1), физическая чистка амортизированная. - DeleteMessage/ChangeMessageVisibility(±Batch): поиск по индексу вместо скана среза; удаление без сдвига остатка. - DeleteMessageBatch: tombstone вместо полной пересборки среза. - processQueueTick: visibility/DLQ по индексу in-flight (O(in-flight)), retention — tombstone, в конце — CompactIfNeeded. - send-пути: AppendMessage + LiveCount (OOM-лимит по живым). - admin/metrics/persistence: учёт живых сообщений, индексы после загрузки из Redis. - Redis-схема не менялась (сообщения и так в per-message HASH).
This commit is contained in:
+16
-9
@@ -461,8 +461,11 @@ func (h *Handler) listTenantQueues(w http.ResponseWriter, r *http.Request) {
|
||||
p.q.RLockMu()
|
||||
visible := 0
|
||||
notVisible := 0
|
||||
for _, msg := range p.q.Messages {
|
||||
if msg.ReceiptHandle != "" {
|
||||
for i := range p.q.Messages {
|
||||
if p.q.IsDead(i) {
|
||||
continue
|
||||
}
|
||||
if p.q.Messages[i].ReceiptHandle != "" {
|
||||
notVisible++
|
||||
} else {
|
||||
visible++
|
||||
@@ -598,11 +601,15 @@ func (h *Handler) peekQueueMessages(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
q.RLockMu()
|
||||
result := make([]peekMessageItem, 0, len(q.Messages))
|
||||
for i, msg := range q.Messages {
|
||||
if i >= limit {
|
||||
result := make([]peekMessageItem, 0, limit)
|
||||
for i := range q.Messages {
|
||||
if len(result) >= limit {
|
||||
break
|
||||
}
|
||||
if q.IsDead(i) {
|
||||
continue
|
||||
}
|
||||
msg := &q.Messages[i]
|
||||
result = append(result, peekMessageItem{
|
||||
ID: msg.Uuid,
|
||||
Body: msg.MessageBody,
|
||||
@@ -652,7 +659,7 @@ func (h *Handler) sendMessageToQueue(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
msg := gosqs.BuildSqsMessage(req.Body)
|
||||
q.Messages = append(q.Messages, msg)
|
||||
q.AppendMessage(msg)
|
||||
// Персистим одно сообщение отдельно
|
||||
persistence.SaveMessage(key, &msg)
|
||||
persistence.SaveQueue(key, q)
|
||||
@@ -678,7 +685,7 @@ func (h *Handler) purgeQueue(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
q.LockMu()
|
||||
q.Messages = q.Messages[:0]
|
||||
q.Reset()
|
||||
// Удаляем все сообщения из Redis одной командой
|
||||
persistence.PurgeMessagesPersist(key)
|
||||
persistence.SaveQueue(key, q)
|
||||
@@ -720,7 +727,7 @@ func (h *Handler) detailedHealth(w http.ResponseWriter, r *http.Request) {
|
||||
models.SyncQueues.RUnlock()
|
||||
for _, q := range uiQueues {
|
||||
q.RLockMu()
|
||||
msgCount += len(q.Messages)
|
||||
msgCount += q.LiveCount()
|
||||
q.RUnlockMu()
|
||||
}
|
||||
|
||||
@@ -745,7 +752,7 @@ func (h *Handler) detailedHealth(w http.ResponseWriter, r *http.Request) {
|
||||
msgCount := 0
|
||||
for _, q := range allQueues {
|
||||
q.RLockMu()
|
||||
msgCount += len(q.Messages)
|
||||
msgCount += q.LiveCount()
|
||||
q.RUnlockMu()
|
||||
}
|
||||
|
||||
|
||||
@@ -56,48 +56,43 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo
|
||||
messageFound := false
|
||||
var changedMsgUuid string
|
||||
var msgRemoved bool
|
||||
for i := 0; i < len(queue.Messages); i++ {
|
||||
msgs := queue.Messages
|
||||
if msgs[i].ReceiptHandle == receiptHandle {
|
||||
// O(1)-поиск по индексу ReceiptHandle (P1.2) вместо скана всего среза.
|
||||
if i, ok := queue.IndexOfReceipt(receiptHandle); ok {
|
||||
msg := &queue.Messages[i]
|
||||
timeout := queue.VisibilityTimeout
|
||||
if visibilityTimeout == 0 {
|
||||
msgs[i].ReceiptTime = time.Now().UTC()
|
||||
msgs[i].ReceiptHandle = ""
|
||||
msgs[i].VisibilityTimeout = time.Now().Add(time.Duration(timeout) * time.Second)
|
||||
msgs[i].Retry++
|
||||
// Возврат видимости: сброс handle + снятие с индекса in-flight.
|
||||
queue.ClearInFlight(i, receiptHandle)
|
||||
msg.ReceiptTime = time.Now().UTC()
|
||||
msg.ReceiptHandle = ""
|
||||
msg.VisibilityTimeout = time.Now().Add(time.Duration(timeout) * time.Second)
|
||||
msg.Retry++
|
||||
changedMsgUuid = msg.Uuid
|
||||
if queue.MaxReceiveCount > 0 &&
|
||||
queue.DeadLetterQueue != nil &&
|
||||
msgs[i].Retry >= queue.MaxReceiveCount {
|
||||
changedMsgUuid = msgs[i].Uuid
|
||||
msg.Retry >= queue.MaxReceiveCount {
|
||||
// Перенос в DLQ + удаление из исходной очереди (tombstone, O(1)).
|
||||
dlqMsg := *msg
|
||||
queue.DeadLetterQueue.LockMu()
|
||||
queue.DeadLetterQueue.Messages = append(queue.DeadLetterQueue.Messages, msgs[i])
|
||||
queue.DeadLetterQueue.AppendMessage(dlqMsg)
|
||||
queue.DeadLetterQueue.UnlockMu()
|
||||
queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...)
|
||||
queue.RemoveIndex(i)
|
||||
msgRemoved = true
|
||||
} else {
|
||||
changedMsgUuid = msgs[i].Uuid
|
||||
}
|
||||
} else {
|
||||
msgs[i].VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second)
|
||||
changedMsgUuid = msgs[i].Uuid
|
||||
msg.VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second)
|
||||
changedMsgUuid = msg.Uuid
|
||||
}
|
||||
messageFound = true
|
||||
break
|
||||
}
|
||||
}
|
||||
// Персистим изменения в Redis под Lock (сетевые записи асинхронны).
|
||||
if messageFound {
|
||||
if msgRemoved {
|
||||
// Сообщение удалено (перемещено в DLQ) — удаляем из Redis
|
||||
persistence.DeleteMessagePersist(key, changedMsgUuid)
|
||||
} else {
|
||||
} else if i, ok := queue.IndexOfUUID(changedMsgUuid); ok {
|
||||
// Сообщение обновлено — перезаписываем в Redis
|
||||
for i := range queue.Messages {
|
||||
if queue.Messages[i].Uuid == changedMsgUuid {
|
||||
persistence.SaveMessage(key, &queue.Messages[i])
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
persistence.SaveQueue(key, queue)
|
||||
}
|
||||
|
||||
@@ -80,22 +80,22 @@ func ChangeMessageVisibilityBatchV1(req *http.Request) (int, interfaces.Abstract
|
||||
}
|
||||
|
||||
messageFound := false
|
||||
for i := 0; i < len(queue.Messages); i++ {
|
||||
if queue.Messages[i].ReceiptHandle == entry.ReceiptHandle {
|
||||
// O(1)-поиск по индексу ReceiptHandle (P1.2) вместо скана всего среза.
|
||||
if i, ok := queue.IndexOfReceipt(entry.ReceiptHandle); ok {
|
||||
msg := &queue.Messages[i]
|
||||
if entry.VisibilityTimeout == 0 {
|
||||
// Сброс: сообщение снова видимо, счётчик retry++
|
||||
queue.Messages[i].ReceiptTime = time.Now().UTC()
|
||||
queue.Messages[i].ReceiptHandle = ""
|
||||
queue.Messages[i].VisibilityTimeout = time.Now().Add(time.Duration(queue.VisibilityTimeout) * time.Second)
|
||||
queue.Messages[i].Retry++
|
||||
queue.ClearInFlight(i, entry.ReceiptHandle)
|
||||
msg.ReceiptTime = time.Now().UTC()
|
||||
msg.ReceiptHandle = ""
|
||||
msg.VisibilityTimeout = time.Now().Add(time.Duration(queue.VisibilityTimeout) * time.Second)
|
||||
msg.Retry++
|
||||
} else {
|
||||
queue.Messages[i].VisibilityTimeout = time.Now().Add(time.Duration(entry.VisibilityTimeout) * time.Second)
|
||||
msg.VisibilityTimeout = time.Now().Add(time.Duration(entry.VisibilityTimeout) * time.Second)
|
||||
}
|
||||
// Персистим изменённое сообщение отдельно
|
||||
persistence.SaveMessage(key, &queue.Messages[i])
|
||||
persistence.SaveMessage(key, msg)
|
||||
messageFound = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if messageFound {
|
||||
|
||||
@@ -49,14 +49,16 @@ func DeleteMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||||
}
|
||||
defer queue.UnlockMu()
|
||||
|
||||
for i, msg := range queue.Messages {
|
||||
if msg.ReceiptHandle == receiptHandle {
|
||||
msgUuid := msg.Uuid
|
||||
// O(1)-поиск сообщения по индексу ReceiptHandle → индекс в Messages (P1.2).
|
||||
// Раньше — линейный скан всего среза + сдвиг остатка (O(N) на большое тело).
|
||||
if i, ok := queue.IndexOfReceipt(receiptHandle); ok {
|
||||
msg := queue.Messages[i] // копия ДО мутации: RemoveIndex может сдвинуть срез
|
||||
queue.ClearInFlight(i, receiptHandle)
|
||||
queue.UnlockGroup(msg.GroupID)
|
||||
queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...)
|
||||
queue.RemoveIndex(i) // tombstone за O(1) + подрезка хвоста/амортизированная компакция
|
||||
delete(queue.Duplicates, msg.DeduplicationID)
|
||||
// Удаляем одно сообщение из Redis — O(1)
|
||||
persistence.DeleteMessagePersist(key, msgUuid)
|
||||
persistence.DeleteMessagePersist(key, msg.Uuid)
|
||||
// Метаданные очереди (duplicates, FIFO state)
|
||||
persistence.SaveQueue(key, queue)
|
||||
respStruct := models.DeleteMessageResponse{
|
||||
@@ -65,7 +67,6 @@ func DeleteMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||||
}
|
||||
return 200, &respStruct
|
||||
}
|
||||
}
|
||||
log.Warning("Receipt Handle not found")
|
||||
|
||||
return utils.CreateErrorResponseV1("MessageDoesNotExist", true)
|
||||
|
||||
@@ -73,22 +73,29 @@ func DeleteMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBo
|
||||
|
||||
deletedEntries := make([]models.DeleteMessageBatchResultEntry, 0)
|
||||
deletedUuids := make([]string, 0)
|
||||
remainingMessages := make([]models.SqsMessage, 0, len(queue.Messages))
|
||||
|
||||
for _, message := range queue.Messages {
|
||||
if de, found := deleteMessageMap[message.ReceiptHandle]; found {
|
||||
// Идём по индексам, не пересобирая срез (P1.2): удаление = tombstone за O(1),
|
||||
// индексы не сдвигаются, поэтому итерация остаётся корректной.
|
||||
for i := range queue.Messages {
|
||||
if queue.IsDead(i) {
|
||||
continue
|
||||
}
|
||||
message := &queue.Messages[i]
|
||||
de, found := deleteMessageMap[message.ReceiptHandle]
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
log.Debugf("FIFO Queue %s unlocking group %s:", queueName, message.GroupID)
|
||||
queue.UnlockGroup(message.GroupID)
|
||||
delete(queue.Duplicates, message.DeduplicationID)
|
||||
// Читаем поля ДО RemoveIndex: компакция может перезаписать слот.
|
||||
groupID, dedupID, uuid := message.GroupID, message.DeduplicationID, message.Uuid
|
||||
queue.ClearInFlight(i, message.ReceiptHandle)
|
||||
queue.UnlockGroup(groupID)
|
||||
delete(queue.Duplicates, dedupID)
|
||||
de.Deleted = true
|
||||
deletedEntries = append(deletedEntries, models.DeleteMessageBatchResultEntry{Id: de.Id})
|
||||
deletedUuids = append(deletedUuids, message.Uuid)
|
||||
} else {
|
||||
remainingMessages = append(remainingMessages, message)
|
||||
deletedUuids = append(deletedUuids, uuid)
|
||||
queue.RemoveIndex(i)
|
||||
}
|
||||
}
|
||||
|
||||
queue.Messages = remainingMessages
|
||||
// Удаляем сообщения из Redis отдельно — O(batch_size)
|
||||
persistence.DeleteMessagesPersist(key, deletedUuids)
|
||||
// Метаданные очереди (duplicates, FIFO state)
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
// app/gosqs/delete_middle_test.go
|
||||
// Тест P1.2: удаление сообщения из СЕРЕДИНЫ очереди не ломает остальные —
|
||||
// индексы receipt/uuid корректны (tombstone), удалённое не возвращается,
|
||||
// оставшиеся находятся по своим ReceiptHandle.
|
||||
package gosqs
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"testing"
|
||||
|
||||
"shared-sqs/app/models"
|
||||
)
|
||||
|
||||
func TestDeleteMessageV1_MiddleKeepsOthersReceivable(t *testing.T) {
|
||||
resetTestEnv(t)
|
||||
tnt := testTenant("SSAK-aaa")
|
||||
injectRequest(t, func(v interface{}) { v.(*models.CreateQueueRequest).QueueName = "q" })
|
||||
if code, _ := CreateQueueV1(reqWithTenant("/", tnt)); code != http.StatusOK {
|
||||
t.Fatal("create failed")
|
||||
}
|
||||
q := models.SyncQueues.Queues["SSAK-aaa:q"]
|
||||
q.MaximumMessageSize = 262144
|
||||
|
||||
// Три сообщения.
|
||||
for _, body := range []string{"m1", "m2", "m3"} {
|
||||
b := body
|
||||
injectRequest(t, func(v interface{}) {
|
||||
r := v.(*models.SendMessageRequest)
|
||||
r.QueueUrl = "http://x/tenant/q"
|
||||
r.MessageBody = b
|
||||
})
|
||||
if code, _ := SendMessageV1(reqWithTenant("/", tnt)); code != http.StatusOK {
|
||||
t.Fatal("send failed")
|
||||
}
|
||||
}
|
||||
|
||||
// Получаем все три разом.
|
||||
injectRequest(t, func(v interface{}) {
|
||||
r := v.(*models.ReceiveMessageRequest)
|
||||
r.QueueUrl = "http://x/tenant/q"
|
||||
r.MaxNumberOfMessages = 10
|
||||
})
|
||||
_, resp := ReceiveMessageV1(reqWithTenant("/", tnt))
|
||||
rr := resp.(models.ReceiveMessageResponse)
|
||||
if len(rr.Result.Messages) != 3 {
|
||||
t.Fatalf("receive вернул %d сообщений, want 3", len(rr.Result.Messages))
|
||||
}
|
||||
|
||||
// Удаляем СРЕДНЕЕ.
|
||||
middle := rr.Result.Messages[1]
|
||||
injectRequest(t, func(v interface{}) {
|
||||
r := v.(*models.DeleteMessageRequest)
|
||||
r.QueueUrl = "http://x/tenant/q"
|
||||
r.ReceiptHandle = middle.ReceiptHandle
|
||||
})
|
||||
if code, _ := DeleteMessageV1(reqWithTenant("/", tnt)); code != http.StatusOK {
|
||||
t.Fatal("delete failed")
|
||||
}
|
||||
if q.LiveCount() != 2 {
|
||||
t.Fatalf("LiveCount = %d, want 2", q.LiveCount())
|
||||
}
|
||||
if _, ok := q.IndexOfUUID(middle.MessageId); ok {
|
||||
t.Fatal("удалённое сообщение осталось в uuid-индексе")
|
||||
}
|
||||
// Оставшиеся по-прежнему in-flight и находятся по своим handle (индексы не «съехали»).
|
||||
for _, m := range []*models.ResultMessage{rr.Result.Messages[0], rr.Result.Messages[2]} {
|
||||
if _, ok := q.IndexOfReceipt(m.ReceiptHandle); !ok {
|
||||
t.Fatalf("handle оставшегося сообщения потерян: %s", m.ReceiptHandle)
|
||||
}
|
||||
}
|
||||
|
||||
// Возвращаем видимость оставшимся и получаем их снова — удалённого быть не должно.
|
||||
for _, m := range []*models.ResultMessage{rr.Result.Messages[0], rr.Result.Messages[2]} {
|
||||
handle := m.ReceiptHandle
|
||||
injectRequest(t, func(v interface{}) {
|
||||
r := v.(*models.ChangeMessageVisibilityRequest)
|
||||
r.QueueUrl = "http://x/tenant/q"
|
||||
r.ReceiptHandle = handle
|
||||
r.VisibilityTimeout = 0
|
||||
})
|
||||
if code, _ := ChangeMessageVisibilityV1(reqWithTenant("/", tnt)); code != http.StatusOK {
|
||||
t.Fatal("change visibility failed")
|
||||
}
|
||||
}
|
||||
|
||||
injectRequest(t, func(v interface{}) {
|
||||
r := v.(*models.ReceiveMessageRequest)
|
||||
r.QueueUrl = "http://x/tenant/q"
|
||||
r.MaxNumberOfMessages = 10
|
||||
})
|
||||
_, resp2 := ReceiveMessageV1(reqWithTenant("/", tnt))
|
||||
rr2 := resp2.(models.ReceiveMessageResponse)
|
||||
if len(rr2.Result.Messages) != 2 {
|
||||
t.Fatalf("повторный receive вернул %d, want 2", len(rr2.Result.Messages))
|
||||
}
|
||||
for _, m := range rr2.Result.Messages {
|
||||
if m.MessageId == middle.MessageId {
|
||||
t.Fatal("удалённое сообщение вернулось в receive")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -87,7 +87,7 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo
|
||||
queueAttributes = append(queueAttributes, models.Attribute{Name: "VisibilityTimeout", Value: strconv.Itoa(queue.VisibilityTimeout)})
|
||||
}
|
||||
if shouldInclude("ApproximateNumberOfMessages") {
|
||||
queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessages", Value: strconv.Itoa(len(queue.Messages))})
|
||||
queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessages", Value: strconv.Itoa(queue.LiveCount())})
|
||||
}
|
||||
if shouldInclude("ApproximateNumberOfMessagesNotVisible") {
|
||||
queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessagesNotVisible", Value: strconv.Itoa(numberOfHiddenMessagesInQueue(queue))})
|
||||
|
||||
+40
-20
@@ -74,28 +74,34 @@ func PeriodicTasks(d time.Duration, quit chan bool) {
|
||||
// - если Retry >= MaxReceiveCount и настроена DLQ — переносим сообщение
|
||||
// в DLQ и удаляем из исходной очереди.
|
||||
func processQueueTick(key string, queue *models.Queue) {
|
||||
// Ленивая починка индексов (P1.2): очередь могла быть собрана литералом или
|
||||
// загружена из Redis — тогда индексы строятся здесь один раз.
|
||||
queue.EnsureIndexes()
|
||||
|
||||
// 0. Эвикция просроченных сообщений по MessageRetentionPeriod.
|
||||
// Retention == 0 означает «без истечения» (не эвиктируем).
|
||||
// Удаление — tombstone за O(1); физическая чистка — CompactIfNeeded ниже.
|
||||
if queue.MessageRetentionPeriod > 0 {
|
||||
cutoff := time.Now().Add(-time.Duration(queue.MessageRetentionPeriod) * time.Second)
|
||||
expired := make([]string, 0) // uuid просроченных сообщений для удаления из Redis
|
||||
// Компакция НА МЕСТЕ: читаем значение в i, пишем в n (n <= i) — без
|
||||
// алиасинга на переиспользуемом срезе и без лишней аллокации.
|
||||
n := 0
|
||||
for i := range queue.Messages {
|
||||
if queue.IsDead(i) {
|
||||
continue
|
||||
}
|
||||
msg := queue.Messages[i]
|
||||
if msg.SentTime.Before(cutoff) {
|
||||
if !msg.SentTime.Before(cutoff) {
|
||||
continue
|
||||
}
|
||||
expired = append(expired, msg.Uuid)
|
||||
if msg.GroupID != "" {
|
||||
queue.UnlockGroup(msg.GroupID)
|
||||
}
|
||||
continue
|
||||
if msg.ReceiptHandle != "" {
|
||||
queue.ClearInFlight(i, msg.ReceiptHandle)
|
||||
}
|
||||
queue.Messages[n] = msg
|
||||
n++
|
||||
queue.RemoveIndex(i)
|
||||
}
|
||||
if len(expired) > 0 {
|
||||
queue.Messages = queue.Messages[:n]
|
||||
log.Infof("Queue [%s]: evicted %d messages by MessageRetentionPeriod", queue.Name, len(expired))
|
||||
// Удаляем просроченные из Redis, чтобы не воскресли после рестарта.
|
||||
persistence.DeleteMessagesPersist(key, expired)
|
||||
@@ -111,18 +117,25 @@ func processQueueTick(key string, queue *models.Queue) {
|
||||
}
|
||||
|
||||
// 2. Возврат видимости истёкших сообщений + перенос в DLQ.
|
||||
log.Debugf("Queue [%s] length [%d]", queue.Name, len(queue.Messages))
|
||||
for i := 0; i < len(queue.Messages); i++ {
|
||||
msg := &queue.Messages[i]
|
||||
|
||||
if msg.ReceiptHandle == "" {
|
||||
continue // сообщение не в полёте
|
||||
// Итерируем ТОЛЬКО по in-flight (receiptIdx), а не по всему срезу — стоимость
|
||||
// пропорциональна числу сообщений «в полёте», а не размеру очереди (P1.2).
|
||||
log.Debugf("Queue [%s] live [%d]", queue.Name, queue.LiveCount())
|
||||
now := time.Now()
|
||||
for _, i := range queue.InFlightIndexes() {
|
||||
if i < 0 || i >= len(queue.Messages) || queue.IsDead(i) {
|
||||
continue
|
||||
}
|
||||
if msg.VisibilityTimeout.After(time.Now()) {
|
||||
msg := &queue.Messages[i]
|
||||
rh := msg.ReceiptHandle
|
||||
if rh == "" {
|
||||
continue // уже возвращено в видимость
|
||||
}
|
||||
if msg.VisibilityTimeout.After(now) {
|
||||
continue // таймаут видимости ещё действует
|
||||
}
|
||||
|
||||
log.Debugf("Making message visible again %s", msg.ReceiptHandle)
|
||||
log.Debugf("Making message visible again %s", rh)
|
||||
queue.ClearInFlight(i, rh)
|
||||
queue.UnlockGroup(msg.GroupID)
|
||||
msg.ReceiptHandle = ""
|
||||
msg.ReceiptTime = time.Now().UTC()
|
||||
@@ -132,18 +145,25 @@ func processQueueTick(key string, queue *models.Queue) {
|
||||
if queue.MaxReceiveCount > 0 &&
|
||||
queue.DeadLetterQueue != nil &&
|
||||
msg.Retry >= queue.MaxReceiveCount {
|
||||
dlqMsg := *msg
|
||||
queue.DeadLetterQueue.LockMu()
|
||||
queue.DeadLetterQueue.Messages = append(queue.DeadLetterQueue.Messages, *msg)
|
||||
queue.DeadLetterQueue.AppendMessage(dlqMsg)
|
||||
queue.DeadLetterQueue.UnlockMu()
|
||||
queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...)
|
||||
i-- // элемент на позиции i удалён — повторно проверяем новый элемент на ней
|
||||
queue.RemoveIndex(i)
|
||||
}
|
||||
}
|
||||
|
||||
// 4. Компакция tombstone-слотов (амортизированно): срез не растёт бесконечно.
|
||||
queue.CompactIfNeeded()
|
||||
}
|
||||
|
||||
func numberOfHiddenMessagesInQueue(queue *models.Queue) int {
|
||||
num := 0
|
||||
for _, m := range queue.Messages {
|
||||
for i := range queue.Messages {
|
||||
if queue.IsDead(i) {
|
||||
continue
|
||||
}
|
||||
m := &queue.Messages[i]
|
||||
if m.ReceiptHandle != "" || m.DelaySecs > 0 && time.Now().Before(m.SentTime.Add(time.Duration(m.DelaySecs)*time.Second)) {
|
||||
num++
|
||||
}
|
||||
|
||||
@@ -60,7 +60,7 @@ func PurgeQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||||
}
|
||||
|
||||
log.Infof("Purging Queue: %s (tenant: %s)", queueName, t.ID)
|
||||
queue.Messages = nil
|
||||
queue.Reset()
|
||||
queue.Duplicates = make(map[string]time.Time)
|
||||
// Фиксируем время purge — защита от повторного вызова в течение 60 секунд.
|
||||
queue.LastPurgeTime = time.Now()
|
||||
|
||||
@@ -154,7 +154,7 @@ func ReceiveMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody)
|
||||
// - для FIFO-очереди блокируется группа (LockGroup) до DeleteMessage/таймаута.
|
||||
func receiveMessagesUnderLock(key string, visibilityTimeout, maxNumberOfMessages int) ([]*models.ResultMessage, bool) {
|
||||
queue := models.SyncQueues.GetLocked(key)
|
||||
if queue == nil || len(queue.Messages) == 0 {
|
||||
if queue == nil || queue.LiveCount() == 0 {
|
||||
return nil, false
|
||||
}
|
||||
defer queue.UnlockMu()
|
||||
@@ -169,6 +169,9 @@ func receiveMessagesUnderLock(key string, visibilityTimeout, maxNumberOfMessages
|
||||
if numMsg >= maxNumberOfMessages {
|
||||
break
|
||||
}
|
||||
if queue.IsDead(i) {
|
||||
continue // tombstone: сообщение логически удалено (P1.2)
|
||||
}
|
||||
msg := &queue.Messages[i]
|
||||
if msg.ReceiptHandle != "" {
|
||||
continue // сообщение уже в полёте у другого получателя
|
||||
@@ -192,6 +195,9 @@ func receiveMessagesUnderLock(key string, visibilityTimeout, maxNumberOfMessages
|
||||
msg.VisibilityTimeout = time.Now().Add(time.Duration(queue.VisibilityTimeout) * time.Second)
|
||||
}
|
||||
msg.NumberOfReceives++
|
||||
// Регистрируем handle→index: по нему DeleteMessage/ChangeMessageVisibility
|
||||
// работают за O(1), а visibility-тик итерирует только in-flight (P1.2).
|
||||
queue.MarkInFlight(i, msg.ReceiptHandle)
|
||||
|
||||
messages = append(messages, buildResultMessage(msg))
|
||||
numMsg++
|
||||
@@ -246,6 +252,9 @@ func buildResultMessage(m *models.SqsMessage) *models.ResultMessage {
|
||||
// читает общее состояние очереди.
|
||||
func queueHasReceivableMessages(queue *models.Queue) bool {
|
||||
for i := range queue.Messages {
|
||||
if queue.IsDead(i) {
|
||||
continue // tombstone (P1.2)
|
||||
}
|
||||
msg := &queue.Messages[i]
|
||||
if msg.ReceiptHandle != "" {
|
||||
continue
|
||||
|
||||
@@ -90,7 +90,7 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||||
defer queue.UnlockMu()
|
||||
|
||||
maxMessageSize := queue.MaximumMessageSize
|
||||
currentMsgCount := len(queue.Messages)
|
||||
currentMsgCount := queue.LiveCount()
|
||||
queueIsFIFO := queue.IsFIFO
|
||||
delaySecs := queue.DelaySeconds
|
||||
|
||||
@@ -129,7 +129,7 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||||
|
||||
// Повторная проверка лимита под per-queue Lock (между снимком выше и текущим
|
||||
// моментом другие горутины могли добавить сообщения — мягкий OOM-лимит).
|
||||
if len(queue.Messages) >= MaxMessagesForQueue(queue.IsFIFO) {
|
||||
if queue.LiveCount() >= MaxMessagesForQueue(queue.IsFIFO) {
|
||||
return utils.CreateErrorResponseV1("OverLimit", true)
|
||||
}
|
||||
|
||||
@@ -140,14 +140,14 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||||
|
||||
persistErr := false
|
||||
if !queue.IsDuplicate(messageDeduplicationID) {
|
||||
queue.Messages = append(queue.Messages, msg)
|
||||
idx := queue.AppendMessage(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]
|
||||
queue.RemoveIndex(idx)
|
||||
persistErr = true
|
||||
}
|
||||
} else {
|
||||
|
||||
@@ -48,7 +48,7 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
|
||||
return utils.CreateErrorResponseV1("QueueNotFound", true)
|
||||
}
|
||||
maxMsgSize := queue.MaximumMessageSize
|
||||
currentMsgCount := len(queue.Messages)
|
||||
currentMsgCount := queue.LiveCount()
|
||||
queueIsFIFO := queue.IsFIFO
|
||||
queue.RUnlockMu()
|
||||
|
||||
@@ -122,7 +122,7 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
|
||||
fifoSeqNumber = queue.NextSequenceNumber(sendEntry.MessageGroupId)
|
||||
}
|
||||
if !queue.IsDuplicate(sendEntry.MessageDeduplicationId) {
|
||||
queue.Messages = append(queue.Messages, msg)
|
||||
queue.AppendMessage(msg)
|
||||
} else {
|
||||
log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", sendEntry.MessageDeduplicationId, queueName)
|
||||
}
|
||||
|
||||
@@ -53,7 +53,7 @@ func updateGauges() {
|
||||
tenantKey := parts[0]
|
||||
tenantQueues[tenantKey]++
|
||||
queue.RLockMu()
|
||||
tenantMessages[tenantKey] += len(queue.Messages)
|
||||
tenantMessages[tenantKey] += queue.LiveCount()
|
||||
queue.RUnlockMu()
|
||||
}
|
||||
|
||||
|
||||
@@ -61,6 +61,21 @@ type Queue struct {
|
||||
MaximumMessageSize int
|
||||
MessageRetentionPeriod int // seconds (реализовано в processQueueTick)
|
||||
Messages []SqsMessage
|
||||
// ─── Индексация сообщений (P1.2) ────────────────────────────────────────
|
||||
// deadIdx — параллельный Messages слайс tombstone-флагов: true = слот
|
||||
// логически удалён. Удаление НЕ сдвигает элементы (O(1)), поэтому индексы
|
||||
// (receiptIdx/uuidIdx) остаются валидными; физическая чистка — Compact.
|
||||
// Менять Messages напрямую НЕЛЬЗЯ: только через AppendMessage/RemoveIndex/
|
||||
// Compact/Reset (см. ниже). Прямая замена допустима лишь для полной очистки
|
||||
// и загрузки из Redis (с последующим RebuildIndexes).
|
||||
deadIdx []bool `json:"-"`
|
||||
// receiptIdx — ReceiptHandle → индекс в Messages (только in-flight сообщения).
|
||||
receiptIdx map[string]int `json:"-"`
|
||||
// uuidIdx — Uuid → индекс в Messages (только живые сообщения).
|
||||
uuidIdx map[string]int `json:"-"`
|
||||
// liveCount — число живых сообщений; deadCount — число tombstone-слотов.
|
||||
liveCount int `json:"-"`
|
||||
deadCount int `json:"-"`
|
||||
DeadLetterQueue *Queue
|
||||
// DeadLetterQueueName — имя DLQ. В Redis-метаданных хранится ТОЛЬКО имя
|
||||
// (а не вложенный объект): после загрузки указатель DeadLetterQueue
|
||||
@@ -163,3 +178,234 @@ func (q *Queue) InitDuplicatation(deduplicationId string) {
|
||||
q.Duplicates[deduplicationId] = time.Now()
|
||||
}
|
||||
}
|
||||
|
||||
// ─── P1.2: индексация сообщений (tombstone + индексы) ──────────────────────
|
||||
//
|
||||
// Все методы ниже мутируют состояние очереди → вызывать ТОЛЬКО под
|
||||
// per-queue write-блокировкой (queue.LockMu). Читающие методы (LiveCount,
|
||||
// IsDead) безопасны под RLock.
|
||||
|
||||
// compactDeadThreshold — минимальное число tombstone-слотов, после которого
|
||||
// компакция при удалении разрешена. Защищает от O(N)-чистки на мелких очередях:
|
||||
// при удалении слот подрезается с хвоста, остальная чистка — в processQueueTick.
|
||||
const compactDeadThreshold = 1024
|
||||
|
||||
// ensureIndexes — ленивая инициализация/починка индексов.
|
||||
// При корректном состоянии — две дешёвые проверки; O(N) — если Messages
|
||||
// изменили в обход методов (загрузка из Redis, тесты, старый JSON-формат).
|
||||
// Вызывать под write-блокировкой.
|
||||
func (q *Queue) ensureIndexes() {
|
||||
if len(q.deadIdx) == len(q.Messages) &&
|
||||
q.receiptIdx != nil && q.uuidIdx != nil &&
|
||||
q.liveCount == len(q.Messages)-q.deadCount {
|
||||
return
|
||||
}
|
||||
q.RebuildIndexes()
|
||||
}
|
||||
|
||||
// RebuildIndexes — перестраивает deadIdx/receiptIdx/uuidIdx/liveCount/deadCount
|
||||
// по текущему Messages. Существующие tombstone СОХРАНЯЮТСЯ по позициям (если
|
||||
// deadIdx длиннее Messages — обрезается, короче — добивается false).
|
||||
// Вызывать под write-блокировкой.
|
||||
func (q *Queue) RebuildIndexes() {
|
||||
if len(q.deadIdx) > len(q.Messages) {
|
||||
q.deadIdx = q.deadIdx[:len(q.Messages)]
|
||||
}
|
||||
for len(q.deadIdx) < len(q.Messages) {
|
||||
q.deadIdx = append(q.deadIdx, false)
|
||||
}
|
||||
q.receiptIdx = make(map[string]int, len(q.Messages))
|
||||
q.uuidIdx = make(map[string]int, len(q.Messages))
|
||||
q.liveCount = 0
|
||||
q.deadCount = 0
|
||||
for i := range q.Messages {
|
||||
if q.deadIdx[i] {
|
||||
q.deadCount++
|
||||
continue
|
||||
}
|
||||
q.liveCount++
|
||||
if u := q.Messages[i].Uuid; u != "" {
|
||||
q.uuidIdx[u] = i
|
||||
}
|
||||
if rh := q.Messages[i].ReceiptHandle; rh != "" {
|
||||
q.receiptIdx[rh] = i
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// AppendMessage — добавляет сообщение в конец очереди, поддерживая индексы.
|
||||
// Возвращает индекс добавленного сообщения. Вызывать под write-блокировкой.
|
||||
func (q *Queue) AppendMessage(m SqsMessage) int {
|
||||
q.ensureIndexes()
|
||||
q.Messages = append(q.Messages, m)
|
||||
q.deadIdx = append(q.deadIdx, false)
|
||||
idx := len(q.Messages) - 1
|
||||
q.liveCount++
|
||||
if m.Uuid != "" {
|
||||
q.uuidIdx[m.Uuid] = idx
|
||||
}
|
||||
if m.ReceiptHandle != "" {
|
||||
q.receiptIdx[m.ReceiptHandle] = idx
|
||||
}
|
||||
return idx
|
||||
}
|
||||
|
||||
// RemoveIndex — логически удаляет сообщение по индексу (tombstone) за O(1).
|
||||
// ВАЖНО: длина q.Messages НЕ меняется (слот остаётся tombstone) — поэтому
|
||||
// индексы, взятые до вызова, остаются валидными, а итерации по срезу
|
||||
// безопасны. Физическая чистка — Compact/CompactIfNeeded (в processQueueTick).
|
||||
// Вызывать под write-блокировкой.
|
||||
func (q *Queue) RemoveIndex(i int) {
|
||||
q.ensureIndexes()
|
||||
if i < 0 || i >= len(q.Messages) || q.deadIdx[i] {
|
||||
return
|
||||
}
|
||||
msg := &q.Messages[i]
|
||||
q.deadIdx[i] = true
|
||||
q.deadCount++
|
||||
if q.liveCount > 0 {
|
||||
q.liveCount--
|
||||
}
|
||||
if msg.Uuid != "" {
|
||||
if idx, ok := q.uuidIdx[msg.Uuid]; ok && idx == i {
|
||||
delete(q.uuidIdx, msg.Uuid)
|
||||
}
|
||||
}
|
||||
if rh := msg.ReceiptHandle; rh != "" {
|
||||
if idx, ok := q.receiptIdx[rh]; ok && idx == i {
|
||||
delete(q.receiptIdx, rh)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// CompactIfNeeded — компакция при заметной доле tombstone-слотов.
|
||||
// Амортизированно: копия среза делается не на каждое удаление, а когда число
|
||||
// «мёртвых» слотов становится заметным (≥ 1024 или ≥ половины среза).
|
||||
// Вызывать под write-блокировкой ВНЕ итераций по Messages (меняет индексы).
|
||||
func (q *Queue) CompactIfNeeded() {
|
||||
if q.deadCount == 0 {
|
||||
return
|
||||
}
|
||||
if q.deadCount >= compactDeadThreshold || q.deadCount*2 >= len(q.Messages) {
|
||||
q.Compact()
|
||||
}
|
||||
}
|
||||
|
||||
// Compact — физически удаляет tombstone-слоты, сохраняя порядок живых, и
|
||||
// перестраивает индексы. O(N). Вызывать под write-блокировкой.
|
||||
func (q *Queue) Compact() {
|
||||
if q.deadCount == 0 {
|
||||
return
|
||||
}
|
||||
if len(q.deadIdx) != len(q.Messages) {
|
||||
q.RebuildIndexes()
|
||||
if q.deadCount == 0 {
|
||||
return
|
||||
}
|
||||
}
|
||||
n := 0
|
||||
for i := range q.Messages {
|
||||
if q.deadIdx[i] {
|
||||
continue
|
||||
}
|
||||
if n != i {
|
||||
q.Messages[n] = q.Messages[i]
|
||||
}
|
||||
n++
|
||||
}
|
||||
// Обнуляем «хвост», чтобы не держать ссылки на удалённые сообщения (GC).
|
||||
tail := q.Messages[n:]
|
||||
for i := range tail {
|
||||
tail[i] = SqsMessage{}
|
||||
}
|
||||
q.Messages = q.Messages[:n]
|
||||
// После чистки tombstone нет — сбрасываем флаги, чтобы RebuildIndexes
|
||||
// построил индексы только по живым (иначе старые deadIdx-флаги «съедут»).
|
||||
q.deadIdx = q.deadIdx[:0]
|
||||
q.deadCount = 0
|
||||
q.RebuildIndexes()
|
||||
}
|
||||
|
||||
// Reset — полностью очищает сообщения и индексы (purge).
|
||||
// Вызывать под write-блокировкой.
|
||||
func (q *Queue) Reset() {
|
||||
q.Messages = nil
|
||||
q.deadIdx = nil
|
||||
q.receiptIdx = make(map[string]int)
|
||||
q.uuidIdx = make(map[string]int)
|
||||
q.liveCount = 0
|
||||
q.deadCount = 0
|
||||
}
|
||||
|
||||
// EnsureIndexes — экспортируемая обёртка над ensureIndexes: строит индексы,
|
||||
// если очередь была собрана в обход методов (литералом, из Redis, тестами).
|
||||
// Вызывать под write-блокировкой.
|
||||
func (q *Queue) EnsureIndexes() {
|
||||
q.ensureIndexes()
|
||||
}
|
||||
|
||||
// IndexOfReceipt — индекс сообщения по ReceiptHandle (только in-flight) за O(1).
|
||||
// Вызывать под write-блокировкой (ensureIndexes может перестроить индексы).
|
||||
func (q *Queue) IndexOfReceipt(receiptHandle string) (int, bool) {
|
||||
q.ensureIndexes()
|
||||
i, ok := q.receiptIdx[receiptHandle]
|
||||
if !ok || i < 0 || i >= len(q.Messages) || q.deadIdx[i] {
|
||||
return 0, false
|
||||
}
|
||||
return i, true
|
||||
}
|
||||
|
||||
// IndexOfUUID — индекс живого сообщения по Uuid за O(1).
|
||||
// Вызывать под write-блокировкой.
|
||||
func (q *Queue) IndexOfUUID(uuid string) (int, bool) {
|
||||
q.ensureIndexes()
|
||||
i, ok := q.uuidIdx[uuid]
|
||||
if !ok || i < 0 || i >= len(q.Messages) || q.deadIdx[i] {
|
||||
return 0, false
|
||||
}
|
||||
return i, true
|
||||
}
|
||||
|
||||
// MarkInFlight — регистрирует ReceiptHandle → индекс (для tick по in-flight).
|
||||
// Вызывать под write-блокировкой.
|
||||
func (q *Queue) MarkInFlight(i int, receiptHandle string) {
|
||||
q.ensureIndexes()
|
||||
if receiptHandle != "" {
|
||||
q.receiptIdx[receiptHandle] = i
|
||||
}
|
||||
}
|
||||
|
||||
// ClearInFlight — снимает регистрацию ReceiptHandle (возврат видимости).
|
||||
// Вызывать под write-блокировкой.
|
||||
func (q *Queue) ClearInFlight(i int, receiptHandle string) {
|
||||
if receiptHandle == "" {
|
||||
return
|
||||
}
|
||||
if idx, ok := q.receiptIdx[receiptHandle]; ok && idx == i {
|
||||
delete(q.receiptIdx, receiptHandle)
|
||||
}
|
||||
}
|
||||
|
||||
// InFlightIndexes — снимок индексов in-flight сообщений (для visibility-тика).
|
||||
// Возвращает срез индексов в Messages. Вызывать под write-блокировкой.
|
||||
func (q *Queue) InFlightIndexes() []int {
|
||||
idx := make([]int, 0, len(q.receiptIdx))
|
||||
for _, i := range q.receiptIdx {
|
||||
idx = append(idx, i)
|
||||
}
|
||||
return idx
|
||||
}
|
||||
|
||||
// IsDead — tombstone-ли слот. Безопасно под RLock (не мутирует состояние).
|
||||
func (q *Queue) IsDead(i int) bool {
|
||||
if i < 0 || i >= len(q.deadIdx) {
|
||||
return false
|
||||
}
|
||||
return q.deadIdx[i]
|
||||
}
|
||||
|
||||
// LiveCount — число живых (не tombstone) сообщений.
|
||||
// Безопасно под RLock. Эквивалент прежнего len(queue.Messages).
|
||||
func (q *Queue) LiveCount() int {
|
||||
return len(q.Messages) - q.deadCount
|
||||
}
|
||||
|
||||
@@ -482,6 +482,9 @@ func LoadAllQueues() (map[string]*models.Queue, error) {
|
||||
// Пустая очередь — Messages уже nil, оставляем
|
||||
q.Messages = nil
|
||||
}
|
||||
// Строим индексы (P1.2): receiptIdx/uuidIdx/deadIdx по загруженным
|
||||
// сообщениям. Без этого первый же receive/delete перестроил бы их лениво.
|
||||
q.RebuildIndexes()
|
||||
}
|
||||
|
||||
// 4. Релинк DLQ: в метаданных хранится только имя DLQ — связываем указатель
|
||||
|
||||
Reference in New Issue
Block a user