475 lines
22 KiB
Go
475 lines
22 KiB
Go
package models
|
||
|
||
import (
|
||
"strconv"
|
||
"sync"
|
||
"time"
|
||
|
||
log "github.com/sirupsen/logrus"
|
||
)
|
||
|
||
type MessageStructure string
|
||
type Protocol string
|
||
|
||
type MessageAttribute struct {
|
||
BinaryListValues []string `json:"BinaryListValues,omitempty" xml:"BinaryListValues,omitempty"` // currently unsupported by AWS
|
||
BinaryValue string `json:"BinaryValue,omitempty" xml:"BinaryValue,omitempty"`
|
||
DataType string `json:"DataType,omitempty" xml:"DataType,omitempty"`
|
||
StringListValues []string `json:"StringListValues,omitempty" xml:"StringListValues,omitempty"` // currently unsupported by AWS
|
||
StringValue string `json:"StringValue,omitempty" xml:"StringValue,omitempty"`
|
||
}
|
||
|
||
type SqsMessage struct {
|
||
MessageBody string
|
||
Uuid string
|
||
MD5OfMessageAttributes string
|
||
MD5OfMessageBody string
|
||
ReceiptHandle string
|
||
ReceiptTime time.Time
|
||
VisibilityTimeout time.Time
|
||
NumberOfReceives int
|
||
Retry int
|
||
MessageAttributes map[string]MessageAttribute
|
||
GroupID string
|
||
DeduplicationID string
|
||
SentTime time.Time
|
||
DelaySecs int
|
||
}
|
||
|
||
func (m *SqsMessage) IsReadyForReceipt() bool {
|
||
randomLatency, err := generateRandomLatency()
|
||
if err != nil {
|
||
log.Error(err)
|
||
return true
|
||
}
|
||
showAt := m.SentTime.Add(randomLatency).Add(time.Duration(m.DelaySecs) * time.Second)
|
||
return showAt.Before(time.Now())
|
||
}
|
||
|
||
type Queue struct {
|
||
// mu — per-queue блокировка содержимого очереди (Messages, Duplicates, FIFO-
|
||
// состояние, LastPurgeTime). Заменяет удержание глобального SyncQueues.Lock
|
||
// на время операций с одной очередью. sync.RWMutex сериализации не подлежит.
|
||
mu sync.RWMutex `json:"-"`
|
||
|
||
Name string
|
||
URL string
|
||
Arn string
|
||
VisibilityTimeout int // seconds
|
||
ReceiveMessageWaitTimeSeconds int
|
||
DelaySeconds int
|
||
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
|
||
// релинкуется на объект из общей 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
|
||
// LastPurgeTime — время последнего успешного PurgeQueue.
|
||
// Используется для запрета повторного purge ранее чем через 60 секунд
|
||
// (требование AWS: PurgeQueueInProgress). Нулевое значение = purge ещё не выполнялся.
|
||
LastPurgeTime time.Time
|
||
// CreatedAt/LastModifiedAt — для атрибутов CreatedTimestamp/LastModifiedTimestamp
|
||
// (AWS отдаёт их в GetQueueAttributes; Terraform читает и без них видел бы drift).
|
||
CreatedAt time.Time
|
||
LastModifiedAt time.Time
|
||
// Policy — resource policy (JSON-строка), храним и отдаём как есть для
|
||
// совместимости с Terraform. НЕ исполняется (нет enforcement).
|
||
Policy string
|
||
// ContentBasedDeduplication — FIFO-атрибут. Храним и отдаём значение,
|
||
// переданное клиентом (для совместимости), см. раздел про dedup.
|
||
ContentBasedDeduplication bool
|
||
}
|
||
|
||
// 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() }
|
||
|
||
// NextSequenceNumber — возвращает следующий порядковый номер сообщения FIFO-группы.
|
||
//
|
||
// Логическая схема: каждый вызов инкрементирует счётчик группы и возвращает
|
||
// его строковое представление (порядковые номера в FIFO должны расти монотонно).
|
||
//
|
||
// ВАЖНО (фикс): раньше при ПЕРВОМ обращении к группе map пересоздавалась целиком
|
||
// (q.FIFOSequenceNumbers = map[string]int{groupId: 0}) — счётчики ВСЕХ остальных
|
||
// групп терялись, и порядковые номера начинали дублироваться. Теперь map только
|
||
// инициализируется один раз (если nil), а счётчик конкретной группы инкрементируется.
|
||
func (q *Queue) NextSequenceNumber(groupId string) string {
|
||
if q.FIFOSequenceNumbers == nil {
|
||
q.FIFOSequenceNumbers = make(map[string]int)
|
||
}
|
||
q.FIFOSequenceNumbers[groupId]++
|
||
return strconv.Itoa(q.FIFOSequenceNumbers[groupId])
|
||
}
|
||
|
||
// IsLocked — проверяет, заблокирована ли FIFO-группа.
|
||
// Группа блокируется при выдаче её сообщения в ReceiveMessage и остаётся
|
||
// заблокированной, пока сообщение in-flight (до DeleteMessage или таймаута
|
||
// видимости) — это гарантирует порядок обработки внутри группы.
|
||
func (q *Queue) IsLocked(groupId string) bool {
|
||
_, ok := q.FIFOMessages[groupId]
|
||
return ok
|
||
}
|
||
|
||
// LockGroup — блокирует FIFO-группу (помечает как занятую).
|
||
//
|
||
// ВАЖНО (фикс): раньше при ПЕРВОМ обращении map пересоздавалась целиком
|
||
// (q.FIFOMessages = map[string]int{groupId: 0}) — блокировки ВСЕХ остальных
|
||
// групп терялись, и их сообщения могли выдаваться одновременно. Теперь map
|
||
// только инициализируется один раз, а для группы проставляется ключ.
|
||
func (q *Queue) LockGroup(groupId string) {
|
||
if q.FIFOMessages == nil {
|
||
q.FIFOMessages = make(map[string]int)
|
||
}
|
||
q.FIFOMessages[groupId] = 0
|
||
}
|
||
|
||
// UnlockGroup — снимает блокировку FIFO-группы.
|
||
// Вызывается после удаления сообщения или при истечении таймаута видимости
|
||
// (см. PeriodicTasks → processQueueTick).
|
||
func (q *Queue) UnlockGroup(groupId string) {
|
||
if _, ok := q.FIFOMessages[groupId]; ok {
|
||
delete(q.FIFOMessages, groupId)
|
||
}
|
||
}
|
||
|
||
// IsDuplicate — проверяет, было ли сообщение с таким deduplicationId уже
|
||
// добавлено в очередь в пределах deduplication-окна (5 минут).
|
||
// Дедупликация — НЕОТЪЕМЛЕМОЕ свойство FIFO-очереди (AWS): раньше она зависела
|
||
// от env-флага EnableDuplicates=false по умолчанию, и FIFO-очереди не
|
||
// дедуплицировали вообще. Теперь — только по IsFIFO и непустому id.
|
||
func (q *Queue) IsDuplicate(deduplicationId string) bool {
|
||
if !q.IsFIFO || deduplicationId == "" {
|
||
return false
|
||
}
|
||
|
||
_, ok := q.Duplicates[deduplicationId]
|
||
|
||
return ok
|
||
}
|
||
|
||
// InitDuplicatation — регистрирует deduplicationId с текущим временем.
|
||
// Эта отметка — начало deduplication-окна; просроченные записи удаляются
|
||
// фоновой задачей PeriodicTasks (deduplication period expired).
|
||
// Для не-FIFO очередей или пустого deduplicationId — no-op.
|
||
func (q *Queue) InitDuplicatation(deduplicationId string) {
|
||
if !q.IsFIFO || deduplicationId == "" {
|
||
return
|
||
}
|
||
|
||
// Ленивая инициализация: очередь, собранная литералом или загруженная из
|
||
// Redis до появления поля, может иметь nil-map — иначе запись паникует.
|
||
if q.Duplicates == nil {
|
||
q.Duplicates = make(map[string]time.Time)
|
||
}
|
||
if _, ok := q.Duplicates[deduplicationId]; !ok {
|
||
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-формат). ВАЖНО (инвариант): оценка «в порядке» проверяет лишь
|
||
// согласованность чисел и не перестраивает уже корректные map-индексы — поэтому
|
||
// мутировать Messages ВНЕ методов (AppendMessage/RemoveIndex/Compact/Reset),
|
||
// например менять ReceiptHandle на месте, НЕЛЬЗЯ.
|
||
// Вызывать под 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() {
|
||
// ensureIndexes гарантирует len(deadIdx)==len(Messages) (иначе слоты разъедутся)
|
||
// и корректность deadCount. При согласованном состоянии — две дешёвые проверки.
|
||
q.ensureIndexes()
|
||
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 заново посчитает liveCount/deadCount и заполнит receipt/uuid).
|
||
q.deadIdx = q.deadIdx[: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
|
||
// Чистим FIFO-состояние: иначе блокировки групп и порядковые номера остались
|
||
// бы, и сообщения этих групп не выдавались после purge (ревью 2026-10-11).
|
||
q.FIFOMessages = make(map[string]int)
|
||
q.FIFOSequenceNumbers = make(map[string]int)
|
||
}
|
||
|
||
// 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)
|
||
}
|
||
}
|
||
|
||
// ResetVisibilityAndTryDLQ — сбрасывает видимость сообщения по индексу и, если
|
||
// исчерпан MaxReceiveCount при настроенной DLQ, переносит его в DLQ и удаляет
|
||
// из исходной очереди (tombstone). Возвращает false, если сообщение на самом
|
||
// деле перенесено в DLQ (удалено из очереди).
|
||
//
|
||
// Единый путь для одиночной и батчевой ChangeMessageVisibility: раньше логика
|
||
// DLQ была только в одиночной версии, а батч лишь инкрементировал Retry — из-за
|
||
// этого retry-счётчик и DLQ-перенос могли расходиться.
|
||
// Вызывать под write-блокировкой q. Перенос в DLQ выполняется под её блокировкой.
|
||
// Возвращает: msgUuid, movedToDLQ.
|
||
// ResetVisibilityAndTryDLQ — сбрасывает видимость сообщения по индексу и, если
|
||
// исчерпан MaxReceiveCount, переносит его в DLQ. При переносе удаляет сообщение
|
||
// из исходной очереди (tombstone).
|
||
//
|
||
// Единый путь для одиночной и батчевой ChangeMessageVisibility: раньше логика
|
||
// DLQ была только в одиночной версии, а батч лишь инкрементировал Retry — из-за
|
||
// этого retry-счётчик и DLQ-перенос могли расходиться.
|
||
// Вызывать под write-блокировкой q. Перенос в DLQ выполняется под её блокировкой.
|
||
// Возвращает: msgUuid, movedToDLQ, dlqMsg (копия перенесённого сообщения, чтобы
|
||
// вызывающий мог персистить его в DLQ; nil если переноса не было).
|
||
func (q *Queue) ResetVisibilityAndTryDLQ(i int, receiptHandle string) (msgUuid string, movedToDLQ bool, dlqMsg *SqsMessage) {
|
||
q.ensureIndexes()
|
||
q.ClearInFlight(i, receiptHandle)
|
||
msg := &q.Messages[i]
|
||
msg.ReceiptTime = time.Now().UTC()
|
||
msg.ReceiptHandle = ""
|
||
msg.VisibilityTimeout = time.Now().Add(time.Duration(q.VisibilityTimeout) * time.Second)
|
||
msg.Retry++
|
||
msgUuid = msg.Uuid
|
||
if q.MaxReceiveCount > 0 && q.DeadLetterQueue != nil && msg.Retry >= q.MaxReceiveCount {
|
||
dlqMsg := *msg
|
||
q.DeadLetterQueue.LockMu()
|
||
q.DeadLetterQueue.AppendMessage(dlqMsg)
|
||
q.DeadLetterQueue.UnlockMu()
|
||
q.RemoveIndex(i)
|
||
return msgUuid, true, &dlqMsg
|
||
}
|
||
return msgUuid, false, nil
|
||
}
|
||
|
||
// InFlightIndexes — снимок индексов in-flight сообщений (для visibility-тика).
|
||
// Возвращает срез индексов в Messages. Вызывать под write-блокировкой.
|
||
//
|
||
// ВАЖНО: индексы — снимок на момент вызова. Некоторые могут стать неактуальными
|
||
// (handle переехал/сообщение удалено), поэтому потребитель ОБЯЗАН перепроверять
|
||
// каждое сообщение по индексу (живость: в границах среза, не tombstone, handle
|
||
// не пуст) — как это делает processQueueTick.
|
||
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
|
||
}
|