From c5b2c11f5d8111b81055f2f9fbf2b77b00c19554 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Wed, 7 Oct 2026 08:01:49 +0300 Subject: [PATCH] =?UTF-8?q?P1.2:=20=D0=B8=D0=BD=D0=B4=D0=B5=D0=BA=D1=81?= =?UTF-8?q?=D0=B0=D1=86=D0=B8=D1=8F=20=D1=81=D0=BE=D0=BE=D0=B1=D1=89=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D0=B9=20=E2=80=94=20tombstone-=D1=83=D0=B4=D0=B0?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=D0=B8=D0=B5=20O(1),=20receipt/uuid-=D0=B8?= =?UTF-8?q?=D0=BD=D0=B4=D0=B5=D0=BA=D1=81=D1=8B,=20tick=20=D0=BF=D0=BE=20i?= =?UTF-8?q?n-flight?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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). --- app/admin/admin.go | 25 +- app/gosqs/change_message_visibility.go | 61 ++--- app/gosqs/change_message_visibility_batch.go | 30 +-- app/gosqs/delete_message.go | 31 +-- app/gosqs/delete_message_batch.go | 33 ++- app/gosqs/delete_middle_test.go | 101 +++++++ app/gosqs/get_queue_attributes.go | 2 +- app/gosqs/gosqs.go | 68 +++-- app/gosqs/purge_queue.go | 2 +- app/gosqs/receive_message.go | 11 +- app/gosqs/send_message.go | 8 +- app/gosqs/send_message_batch.go | 4 +- app/metrics/gauge_updater.go | 2 +- app/models/models.go | 268 ++++++++++++++++++- app/persistence/redis.go | 3 + 15 files changed, 519 insertions(+), 130 deletions(-) create mode 100644 app/gosqs/delete_middle_test.go diff --git a/app/admin/admin.go b/app/admin/admin.go index 9d3893a..a19e394 100644 --- a/app/admin/admin.go +++ b/app/admin/admin.go @@ -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() } diff --git a/app/gosqs/change_message_visibility.go b/app/gosqs/change_message_visibility.go index 4aca164..e46ccb8 100644 --- a/app/gosqs/change_message_visibility.go +++ b/app/gosqs/change_message_visibility.go @@ -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 { - 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++ - if queue.MaxReceiveCount > 0 && - queue.DeadLetterQueue != nil && - msgs[i].Retry >= queue.MaxReceiveCount { - changedMsgUuid = msgs[i].Uuid - queue.DeadLetterQueue.LockMu() - queue.DeadLetterQueue.Messages = append(queue.DeadLetterQueue.Messages, msgs[i]) - queue.DeadLetterQueue.UnlockMu() - queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...) - msgRemoved = true - } else { - changedMsgUuid = msgs[i].Uuid - } - } else { - msgs[i].VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second) - changedMsgUuid = msgs[i].Uuid + // O(1)-поиск по индексу ReceiptHandle (P1.2) вместо скана всего среза. + if i, ok := queue.IndexOfReceipt(receiptHandle); ok { + msg := &queue.Messages[i] + timeout := queue.VisibilityTimeout + if visibilityTimeout == 0 { + // Возврат видимости: сброс 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 && + msg.Retry >= queue.MaxReceiveCount { + // Перенос в DLQ + удаление из исходной очереди (tombstone, O(1)). + dlqMsg := *msg + queue.DeadLetterQueue.LockMu() + queue.DeadLetterQueue.AppendMessage(dlqMsg) + queue.DeadLetterQueue.UnlockMu() + queue.RemoveIndex(i) + msgRemoved = true } - messageFound = true - break + } else { + msg.VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second) + changedMsgUuid = msg.Uuid } + messageFound = true } // Персистим изменения в 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.SaveMessage(key, &queue.Messages[i]) } persistence.SaveQueue(key, queue) } diff --git a/app/gosqs/change_message_visibility_batch.go b/app/gosqs/change_message_visibility_batch.go index 620bd73..c4ef84f 100644 --- a/app/gosqs/change_message_visibility_batch.go +++ b/app/gosqs/change_message_visibility_batch.go @@ -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 { - 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++ - } else { - queue.Messages[i].VisibilityTimeout = time.Now().Add(time.Duration(entry.VisibilityTimeout) * time.Second) - } - // Персистим изменённое сообщение отдельно - persistence.SaveMessage(key, &queue.Messages[i]) - messageFound = true - break + // O(1)-поиск по индексу ReceiptHandle (P1.2) вместо скана всего среза. + if i, ok := queue.IndexOfReceipt(entry.ReceiptHandle); ok { + msg := &queue.Messages[i] + if entry.VisibilityTimeout == 0 { + // Сброс: сообщение снова видимо, счётчик 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 { + msg.VisibilityTimeout = time.Now().Add(time.Duration(entry.VisibilityTimeout) * time.Second) } + // Персистим изменённое сообщение отдельно + persistence.SaveMessage(key, msg) + messageFound = true } if messageFound { diff --git a/app/gosqs/delete_message.go b/app/gosqs/delete_message.go index b56c413..2163369 100644 --- a/app/gosqs/delete_message.go +++ b/app/gosqs/delete_message.go @@ -49,22 +49,23 @@ func DeleteMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { } defer queue.UnlockMu() - for i, msg := range queue.Messages { - if msg.ReceiptHandle == receiptHandle { - msgUuid := msg.Uuid - queue.UnlockGroup(msg.GroupID) - queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...) - delete(queue.Duplicates, msg.DeduplicationID) - // Удаляем одно сообщение из Redis — O(1) - persistence.DeleteMessagePersist(key, msgUuid) - // Метаданные очереди (duplicates, FIFO state) - persistence.SaveQueue(key, queue) - respStruct := models.DeleteMessageResponse{ - Xmlns: models.BaseXmlns, - Metadata: models.BaseResponseMetadata, - } - return 200, &respStruct + // 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.RemoveIndex(i) // tombstone за O(1) + подрезка хвоста/амортизированная компакция + delete(queue.Duplicates, msg.DeduplicationID) + // Удаляем одно сообщение из Redis — O(1) + persistence.DeleteMessagePersist(key, msg.Uuid) + // Метаданные очереди (duplicates, FIFO state) + persistence.SaveQueue(key, queue) + respStruct := models.DeleteMessageResponse{ + Xmlns: models.BaseXmlns, + Metadata: models.BaseResponseMetadata, } + return 200, &respStruct } log.Warning("Receipt Handle not found") diff --git a/app/gosqs/delete_message_batch.go b/app/gosqs/delete_message_batch.go index b0c1182..32ef866 100644 --- a/app/gosqs/delete_message_batch.go +++ b/app/gosqs/delete_message_batch.go @@ -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 { - log.Debugf("FIFO Queue %s unlocking group %s:", queueName, message.GroupID) - queue.UnlockGroup(message.GroupID) - delete(queue.Duplicates, message.DeduplicationID) - de.Deleted = true - deletedEntries = append(deletedEntries, models.DeleteMessageBatchResultEntry{Id: de.Id}) - deletedUuids = append(deletedUuids, message.Uuid) - } else { - remainingMessages = append(remainingMessages, message) + // Идём по индексам, не пересобирая срез (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) + // Читаем поля ДО 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, uuid) + queue.RemoveIndex(i) } - - queue.Messages = remainingMessages // Удаляем сообщения из Redis отдельно — O(batch_size) persistence.DeleteMessagesPersist(key, deletedUuids) // Метаданные очереди (duplicates, FIFO state) diff --git a/app/gosqs/delete_middle_test.go b/app/gosqs/delete_middle_test.go new file mode 100644 index 0000000..39bf3a1 --- /dev/null +++ b/app/gosqs/delete_middle_test.go @@ -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") + } + } +} diff --git a/app/gosqs/get_queue_attributes.go b/app/gosqs/get_queue_attributes.go index 61bc53c..5e79445 100644 --- a/app/gosqs/get_queue_attributes.go +++ b/app/gosqs/get_queue_attributes.go @@ -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))}) diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go index 2bda081..abe8af7 100644 --- a/app/gosqs/gosqs.go +++ b/app/gosqs/gosqs.go @@ -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 { - msg := queue.Messages[i] - if msg.SentTime.Before(cutoff) { - expired = append(expired, msg.Uuid) - if msg.GroupID != "" { - queue.UnlockGroup(msg.GroupID) - } + if queue.IsDead(i) { continue } - queue.Messages[n] = msg - n++ + msg := queue.Messages[i] + if !msg.SentTime.Before(cutoff) { + continue + } + expired = append(expired, msg.Uuid) + if msg.GroupID != "" { + queue.UnlockGroup(msg.GroupID) + } + if msg.ReceiptHandle != "" { + queue.ClearInFlight(i, msg.ReceiptHandle) + } + 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++ } diff --git a/app/gosqs/purge_queue.go b/app/gosqs/purge_queue.go index 1248b92..fc2f023 100644 --- a/app/gosqs/purge_queue.go +++ b/app/gosqs/purge_queue.go @@ -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() diff --git a/app/gosqs/receive_message.go b/app/gosqs/receive_message.go index 622aebc..24afa58 100644 --- a/app/gosqs/receive_message.go +++ b/app/gosqs/receive_message.go @@ -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 diff --git a/app/gosqs/send_message.go b/app/gosqs/send_message.go index 9ada36b..1b4a094 100644 --- a/app/gosqs/send_message.go +++ b/app/gosqs/send_message.go @@ -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 { diff --git a/app/gosqs/send_message_batch.go b/app/gosqs/send_message_batch.go index ec10979..1bd8ef3 100644 --- a/app/gosqs/send_message_batch.go +++ b/app/gosqs/send_message_batch.go @@ -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) } diff --git a/app/metrics/gauge_updater.go b/app/metrics/gauge_updater.go index 4360014..cf79c2f 100644 --- a/app/metrics/gauge_updater.go +++ b/app/metrics/gauge_updater.go @@ -53,7 +53,7 @@ func updateGauges() { tenantKey := parts[0] tenantQueues[tenantKey]++ queue.RLockMu() - tenantMessages[tenantKey] += len(queue.Messages) + tenantMessages[tenantKey] += queue.LiveCount() queue.RUnlockMu() } diff --git a/app/models/models.go b/app/models/models.go index 4ef67f3..33fd4fa 100644 --- a/app/models/models.go +++ b/app/models/models.go @@ -61,18 +61,33 @@ type Queue struct { MaximumMessageSize int MessageRetentionPeriod int // seconds (реализовано в processQueueTick) Messages []SqsMessage - DeadLetterQueue *Queue + // ─── Индексация сообщений (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 // релинкуется на объект из общей map (см. persistence.LoadAllQueues). DeadLetterQueueName string `json:"dead_letter_queue_name,omitempty"` MaxReceiveCount int - IsFIFO bool - FIFOMessages map[string]int - FIFOSequenceNumbers map[string]int - EnableDuplicates bool - Duplicates map[string]time.Time - Tags map[string]string + IsFIFO bool + FIFOMessages map[string]int + FIFOSequenceNumbers map[string]int + EnableDuplicates bool + Duplicates map[string]time.Time + Tags map[string]string // LastPurgeTime — время последнего успешного PurgeQueue. // Используется для запрета повторного purge ранее чем через 60 секунд // (требование AWS: PurgeQueueInProgress). Нулевое значение = purge ещё не выполнялся. @@ -81,10 +96,10 @@ type Queue struct { // LockMu/UnlockMu/RLockMu/RUnlockMu — доступ к per-queue блокировке извне пакета models // (поле mu неэкспортируемо). -func (q *Queue) LockMu() { q.mu.Lock() } -func (q *Queue) UnlockMu() { q.mu.Unlock() } -func (q *Queue) RLockMu() { q.mu.RLock() } -func (q *Queue) RUnlockMu() { q.mu.RUnlock() } +func (q *Queue) LockMu() { q.mu.Lock() } +func (q *Queue) UnlockMu() { q.mu.Unlock() } +func (q *Queue) RLockMu() { q.mu.RLock() } +func (q *Queue) RUnlockMu() { q.mu.RUnlock() } // NextSequenceNumber — возвращает следующий порядковый номер сообщения FIFO-группы. // @@ -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 +} diff --git a/app/persistence/redis.go b/app/persistence/redis.go index 7ac89ae..9710dd2 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -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 — связываем указатель