chore: update ignore rules

- ContentBasedDeduplication=true: dedupId computed from body SHA-256 when not provided
- do NOT drop dedupId on message delete: AWS keeps the 5-minute window regardless
  (previously re-send after delete delivered the message again)
- utils: GetSHA256Hash

.gitignore

@@ -35,3 +35,10 @@ cmd
 # Python cache
 __pycache__/
 *.pyc
+/HISTORY/2026-08-13-session-log.md
+/HISTORY/2026-08-13-sonnet-answer-v2.md
+/HISTORY/2026-08-14-session-log.md
+/app/gosqs/tenant_flow_test.go
+/deployments/helm/shared-sqs/values.yaml
+/doc/2026-08-13-architecture-and-migration-plan.md
+/tests/run_full_regression.sh

HISTORY/2026-10-10-session-log.md

@@ -152,6 +152,72 @@
 - Док (коммит `865db49`): почему C, механика, замеры, ограничения, пошаговое включение, откат,
   путь к варианту A (5 шагов) и триггеры перехода.
 
+### 2.17. Код-ревью Opus + критический разбор
+- Дал промпт Opus: полное код-ревью + советы по архитектуре. Дословный ответ:
+  `doc/thinking/opus-report-2026-10-10.md`; критический разбор + что подтверждено:
+  `doc/thinking/opus-report-analysis-2026-10-10.md`.
+- Opus указал ~50 пунктов (🔴/🟡/🟢). Моя оценка: силён по конкуренции/durability, но смешивает
+  баги / осознанный дизайн / отклонения от семантики AWS; часть 🔴 надо понизить.
+- **Подтверждено по коду (чинится без владельца):**
+  - 🔴1 фатал `concurrent map iteration and map write`: `create_queue.go:84` зовёт `SaveQueue`
+    под `SyncQueues.Lock` без `queue.mu`; `SaveQueue`→`json.Marshal` итерирует `Duplicates`/`FIFO*`/`Tags`,
+    а мутаторы держат только `queue.mu`. Условие: идемпотентный CreateQueue существующей FIFO-очереди + SendMessage.
+  - 🔴5 неполный откат dedup при сбое `SaveMessageSync` (`send_message.go:151-160`) → ретрай 200 без сохранения.
+  - 🔴8 пустой `ReceiptHandle` в DeleteMessageBatch удаляет все невыданные сообщения — **только
+    через JSON-протокол** (`x-amz-json-1.0`); Query-форма защищена (`requests.go:568` не на этом пути).
+  - Порядок локов + I/O под глобальным локом; fail-fast на старте без Redis отсутствует.
+- **Нужны решения владельца:**
+  - JWT: проверяет ли шлюз подпись (trusted-perimeter из `jwt.go:7-11`)? Независимо чинятся:
+    три шлюза prod/test/dev → prod; потолок grace в `cache.go:68-75`; идентичность по `sub`, не email.
+  - Лимит по байтам (OOM, 🔴14): нужны RAM ВМ + число тенантов → cap на тенанта/очередь + глобальный предохранитель по RSS.
+- Порядок исправления — §5 дока анализа.
+
+### 2.18. Правки по ревью (применено, 3 коммита)
+- `654e315` — 🔴1 (SaveQueue под queue.mu), 🔴8 (пустой ReceiptHandle), 🔴5 (dedup только после
+  успешной записи, single+batch), глобальный предохранитель HeapAlloc (`memory_guard.go`,
+  `SHARED_SQS_MAX_HEAP_MB`, 0=выключен).
+- `1230149` — JWT: только prod-шлюз (убран test/dev); grace в `cache.go` ограничен `graceDeadline`
+  от последней успешной проверки (не продлевается бесконечно).
+- `2187027` — fail-fast: при заданном `REDIS_ADDR` и недоступном Redis → `log.Fatalf`.
+- build + vet + test — зелёные. Не задеплоено на ВМ (жду команды).
+- НЕ сделано (нужны решения владельца): цифры лимита по байтам (RAM/тенанты); идентичность
+  JWT по `sub` вместо email (зависит от того, проверяет ли шлюз подпись). Порядок локов и I/O
+  под глобальным локом — осталось (более крупный рефакторинг).
+
+### 2.19. Деплой правок на ВМ 213 + проверка
+- Обновлён `/home/naeel/sqs-test/shared-sqs` (prev-бинарник сохранён, лог бэкапнут
+  `sqs.log.before-deploy-*`), сервис `active`, `/health` 200, `/ready` 200.
+- В логе подтверждено: `leader-election выключен`, два листенера (4100 public / 4101 localhost),
+  fail-fast не срабатывал (Redis доступен).
+- **Функциональная проверка 🔴8 (изолированная очередь `probe-empty-handle`):**
+  - отправлено 2 сообщения (не выдавались → `ReceiptHandle == ""`), `ApproximateNumberOfMessages` = 2;
+  - `delete-message-batch` с `ReceiptHandle: ""` → ошибка
+    `AWS.SimpleQueueService.InvalidParameterValue`, **сообщения остались (2)** → фикс работает;
+  - CLI v2 подтверждённо шлёт JSON-протокол (ошибку отдал наш новый валидатор, а не парсер формы);
+  - легитимный `delete-message-batch` по реальному handle → `Successful`, счётчик 2→1;
+  - probe-очередь удалена.
+- Регрессий нет: `list-queues`, `send-message`, `receive-message` через `https://sqs.sinefo.ru` — OK.
+
+### 2.20. AWS-совместимость для Terraform-провайдера (цель дня уточнена владельцем)
+Контекст: это **не глобальный прод с высокой нагрузкой**; цель — рабочие SQS+IoT и **Terraform-провайдер**,
+чтобы юзеры Nubes подключали сервисы в своих конфигурациях. Приоритет — **надёжность и соответствие
+AWS**, а не масштабирование. Поэтому HA/перф-оптимизации отложены, в фокус — совместимость.
+- `95bce88` — атрибуты очереди для провайдера: `GetQueueAttributes` теперь отдаёт `FifoQueue`,
+  `ContentBasedDeduplication`, `Policy` и **реальные** `CreatedTimestamp`/`LastModifiedTimestamp`
+  (были отсутствующие/заглушка `0000000000`); в `Queue` добавлены `CreatedAt`/`LastModifiedAt`/`Policy`/
+  `ContentBasedDeduplication`; разбор FIFO-атрибутов в обоих протоколах (`StringToBool`);
+  `Policy`/`RedriveAllowPolicy` нормализуются из строки в объект (JSON-протокол присылает строку →
+  раньше `SetQueueAttributes` падал с `InvalidParameterValue`); бэкфилл пустых timestamps при загрузке.
+- `988a6a6` — FIFO-дедупликация по AWS: дедуп — свойство FIFO (гейт по `IsFIFO`, а не по env-флагу
+  `EnableDuplicates=false`, из-за которого дедуп не работал вообще); `ContentBasedDeduplication`
+  считает dedupId из тела (SHA-256); dedup-окно **не сбрасывается при удалении** сообщения (5 мин).
+- **Проверено локально и на проде** (`aws` CLI v2, JSON-протокол):
+  - `GetQueueAttributes All` отдаёт полный набор (см. лог);
+  - FIFO: два send с одним dedupId → 1 сообщение; `ContentBasedDeduplication` (два одинаковых тела) → 1;
+  - receive+delete, затем повторный send с тем же dedupId → 0 (окно живо) — раньше доставлялся повторно;
+  - `set-queue-attributes` с `Policy` → OK, `get-queue-attributes Policy` возвращает канонический JSON.
+- На ВМ задеплоено (prev-бинарник + бэкап лога сохранены), `active`, `/health` 200.
+
 ## 3. Грабли (для повторного использования)
 
 1. **`proxy_set_header Host $host` теряет порт** → SigV4 `SignatureDoesNotMatch`.
@@ -216,3 +282,9 @@ pkill -x shared-sqs
    вторая ВМ, LB с проверкой `/ready`. Пока всё это не готово — флаг выключен.
 5. Дальше по мануалу: обновить устаревший `MANUAL-NEXT-2026-10-08.md` (ещё говорит «Шаг 1 — проверить
    гипотезу» и «вопрос по домену открыт»); решить вопрос push в origin (18 коммитов впереди).
+6. **JWT (решение владельца):** проверяет ли шлюз подпись и отдаёт 200 только для валидных токенов?
+   От этого — 🔴 или «доверяем шлюзу». См. `opus-report-analysis-2026-10-10.md` §3.
+7. **Лимит по байтам (решение владельца):** нужны RAM ВМ 213 и целевое число тенантов для цифр cap.
+   См. `opus-report-analysis-2026-10-10.md` §4.
+8. 🔴1/🔴5/🔴8 + fail-fast + JWT + heap-guard — **применены** (коммиты 654e315/1230149/2187027),
+   не задеплоено. Осталось: порядок локов + I/O под глобальным локом (рефакторинг).

HISTORY/2026-10-11-session-log.md

@@ -0,0 +1,124 @@
+# История правок shared-sqs — 2026-10-11 (внешнее код-ревью и фиксы)
+
+> Запись по требованию «документировать всё, включая собственные ошибки» (приказ 2026-09-24).
+> Предыдущие: `HISTORY/2026-10-10-session-log.md`, `doc/thinking/*-2026-10-08.md`.
+
+## 1. Задача
+
+Провести **внешнее код-ревью** двух сервисов (SQS-service + IoT) моделью Claude Sonnet
+(в режиме Plan — правки не вносит, только находит), затем своими силами исправить
+найденное. Промпт для ревью — `CLAUDE-REVIEW-PROMPT.md` в корне монорепо
+(со списком файлов, фокус-точками и строгим форматом ответа; явно запрещены
+правки, оценка тестов/UI/legacy/стиля).
+
+Ревью дало **40 пунктов** (10 🔴 / 29 🟡 / 1 🟢). Разбор вопросов ревьюера и
+уточнение некорректных выводов — в п.3.
+
+## 2. Что исправлено (SQS-service)
+
+### 🔴 Критичные (10/10)
+
+| Пункт | Файл | Суть | Коммит |
+|-------|------|------|--------|
+| 🔴1 | `app/admin/admin.go` | `jwtMiddleware` пропускал любой путь с суффиксом `/auth` (обход JWT + IDOR на `DELETE .../queues/auth`). Теперь — точное сравнение `/ui/api/auth` | `1241093` |
+| 🔴2 | `app/gosqs/queue_attributes.go` | Запрет self-/циклического `RedrivePolicy`: перенос в DLQ берёт второй `mu` под первым → deadlock | `9b9bbe1` |
+| 🔴3 | `app/gosqs/gosqs.go`, `models.go`, `change_message_visibility*.go` | Перенос в DLQ не персистился (сообщение «воскрешало» в исходной и пропадало из DLQ) | `1e334c2` |
+| 🔴4 | `app/persistence/redis.go`, `app/tenant/tenant_store.go` | `DeleteTenant` не чистил индекс `ssq:tenant_ak` — отозванный тенант возвращался через read-through | `f68b0a4` |
+| 🔴5 | `app/persistence/redis.go`, `app/cmd/goaws.go` | Глобальный `Client` перезаписывался в `reconnect()` без синхронизации (data race). → `atomic.Pointer` | `d03ce36` |
+| 🔴6 | `app/conf/config.go` | `return ports` без `SyncQueues.Unlock()` при ошибке `setQueueRedrivePolicy` → вечный deadlock | `1520753` |
+| 🔴7 | `app/billing/billing.go` | `Close()` писал `db=nil` параллельно с `usageWorker` (nil-паника). → мьютекс + остановка воркера до `db.Close` | `4e36ba0` |
+| 🔴10 | `app/auth/auth_middleware.go` | `X-Amz-Content-Sha256` не сверялся с телом — подмена тела под подписанным заголовком | `2918a41` |
+
+(🔴8/🔴9 — на стороне IoT, см. `IoT/HISTORY/2026-10-11-session-log.md`.)
+
+### 🟡 Важные
+
+| Пункт | Файл | Суть | Коммит |
+|-------|------|------|--------|
+| 🟡20 | `app/admin/admin.go` | `createTenantQueue` не валидировал имя очереди (`:`,`/` ломали ключ) | `efd8c8e` |
+| 🟡23 | `app/auth/jwt.go` | Токен без `exp` принимался бессрочно → требуем `exp` | `efd8c8e` |
+| 🟡24 | `app/leader/leader.go` | При сбое `RefreshLock` инстанс оставался лидером (split-brain) → сброс лидерства после TTL | `efd8c8e` |
+| 🟡21 | `app/tenant/tenant_store.go` | Запрет `:` в BYOC `access_key` (коллизии ключа `{ak}:{name}`) | `dc4815d` |
+| 🟡22 | `app/admin/admin.go` | `sendMessageToQueue` без лимитов + async persist → лимит `MaxMessagesForQueue` + `SaveMessageSync` | `dc4815d` |
+| 🟡33 | `app/models/models.go` | `InitDuplicatation` падал на nil-map `Duplicates` → ленивая инициализация | `dc4815d` |
+| 🟡35 | `app/models/models.go` | `Reset()` не чистил `FIFOMessages` (после purge группы не выдавались) | `dc4815d` |
+| 🟡36 | `app/gosqs/receive_message.go` | `ApproximateReceiveCount` был завышен на 1 | `dc4815d` |
+| 🟡38 | `app/cmd/goaws.go` | `billing.Close()` вызывался ДО `Shutdown` серверов → перенесён после | `dc4815d` |
+| 🟡26 | `app/gosqs/delete_message_batch.go` | Ключ map по `ReceiptHandle` (два entry с одним handle теряли Id) → по `Id` | `e0b5f24` |
+| 🟡27 | `app/gosqs/send_message.go` | `DelaySeconds` не валидировался → 0–900 | `e0b5f24` |
+| 🟡25 | `app/gosqs/send_message_batch.go` | Dedup регистрировался после цикла (два entry с одним id оба добавлялись); `NextSequenceNumber` до проверки дубликата; per-entry `DelaySeconds` не применялся | `6f69755` |
+| 🟡29 | `app/gosqs/delete_message*.go`, `persistence/redis.go` | Ошибка `HDel` только логировалась (клиент получал 200, после рестарта сообщение воскрешало) → `DeleteMessage(s)Persist` возвращает ошибку, 500 клиенту | `5d38118` |
+| 🟡31 | `app/persistence/redis.go`, `app/cmd/goaws.go` | Ошибка загрузки сообщений очереди давала `continue` (очередь стартовала пустой) → fail-fast (`log.Fatalf`) | `5d38118` |
+| 🟡30 | `app/persistence/redis.go` | Async `SaveMessage` без порядковой нумерации (старое состояние могло перезаписать новое) → seq по `{queueKey}|{uuid}` + очистка при удалении | `7a2340d` |
+| 🟡32 | `app/gosqs/gosqs.go` | `PeriodicTasks` без `recover` (паника убивала процесс) | `dc4815d` |
+| 🟢 | `app/utils/utils.go` | `SqsErrors[errKey]` без проверки → nil-deref на неизвестном ключе | `5d38118` |
+| (🟡 new) | `app/gosqs/create_queue.go` | Очередь из API получала `VisibilityTimeout=0` и пр. (дефолты не подставлялись) | `5d38118` |
+
+### Осознанно НЕ менялось — РЕШЕНИЯ (2026-10-11)
+
+- 🟡34 (атрибуты «0 = не передано»): **решение — не менять** совместимость.
+  Различать «не передано» и явный 0 потребовало бы перевести `QueueAttributes`
+  на указатели — ломает JSON-протокол и Terraform-drift. Явный 0 — редкий сценарий.
+- 🟡37 (дефолт `SHARED_SQS_MAX_HEAP_MB`): **решение — оставить выключенным (0)**.
+  Безопасное число зависит от RAM конкретной ВМ и числа тенантов — выбор владельца,
+  не хардкод в коде.
+- 🟡28 (сеть под `queue.mu`): **решение — оставить как есть (вариант A)**. См. п.4.
+
+## 3. Уточнение выводов ревьюера
+
+Ревьюер задал вопросы по коду, ответы найдены в коде (файл:строка):
+
+- **ACL EMQX — единственный барьер для топиков** `{ns}/telemetry/{deviceId}`? Да:
+  `bridge/handler.go` берёт `ns`/`deviceID` из топика без сверки принадлежности,
+  барьер — EMQX authz (`emqx/emqx.conf`). Добавлена defense-in-depth проверка
+  формата (IoT).
+- **Self-/циклический `RedrivePolicy` реально задаётся через API?** Да
+  (`setQueueAttributesV1` не проверял) — отсюда фикс 🔴2.
+- **Пустой `adminToken`?** Недопустим — `goaws.go:~74` `log.Fatal` при отсутствии.
+- **Дефолты очереди расходятся** (`create_queue.go` vs `conf/config.go:100-109`)?
+  Да — отсюда фикс (🟡 new).
+- **`utils.go` паника на неизвестном ключе?** Текущей паники нет (все ключи есть
+  в `models.SqsErrors`), но добавлена защита.
+
+Также ревьюер отметил некорректный пункт (удалён из списка по лимиту 40): nil-map
+`Duplicates` (🟡33) фактически не воспроизводился в текущем коде — но защита
+добавлена как дешёвая страховка.
+
+## 4. 🟡28 — РЕШЕНИЕ (2026-10-11)
+
+- **Проблема**: синхронный `SaveMessageSync` (до 3 с) выполняется под `queue.mu`
+  (`send_message.go:150`, `delete_message.go`, `admin.go`). Медленный Redis блокирует
+  всю очередь.
+- **Варианты**: A) оставить; B) укоротить таймаут Redis-записи; C) полный рефакторинг
+  с per-message синхронизацией.
+- **Выбрано A — оставить.** Обоснование:
+  1. in-memory ядро очередей — одиночка, блокировка `queue.mu` локальна (одна очередь).
+  2. Синхронный персист под `mu` = гарантия «нет потери подтверждённого сообщения
+     при крахе пода» — приоритет проекта (надёжность, а не масштабирование).
+  3. Вынос сети за `mu` вносит окно между `Unlock`/`Lock`: два конкурентных
+     `SendMessage` с одним `deduplicationId` оба пройдут `IsDuplicate` (дубль),
+     либо при откате `dedup` теряется сообщение — хуже кратковременной блокировки.
+  4. Redis-авария — глобальная: деградирует всё, 2–3 с на очередь не критично.
+- 🟡34/🟡37 — см. п.2.
+
+## 5. Верификация
+
+- `go build ./...`, `go vet ./...`, `go test -race ./...` — зелёные на всех коммитах.
+- Push: `2c6d2d7..f80e1b4` в `origin/master`.
+
+## 6. Деплой на прод (ВМ 213) и smoke
+
+- `make build` → `scp shared-sqs → /home/naeel/sqs-test/shared-sqs.new` →
+  `systemctl stop shared-sqs; mv .new → shared-sqs; start`. Бэкап — `shared-sqs.prev`.
+- Проверка: `systemctl is-active` = active; `/health` → 200 `v0.1.35`;
+  в логе «загружено 5 очередей из Redis» (fail-fast-загрузка сработала, не упала).
+- Smoke по `https://sqs.sinefo.ru` (`tests/api_test.sh`, aws cli, тенант `demo-service`):
+  **PASS=22 FAIL=0** (17 базовых операций + FIFO CreateQueue/Send/Receive/Delete/DeleteQueue).
+  Вывод: 18 фиксов ревью не сломали AWS-совместимость на проде.
+
+## 7. Образы → Gitea registry
+
+- Реестр — **Gitea** `gitea.services.ngcloud.ru` (НЕ Docker Hub). Makefile/Dockerfile/доки исправлены.
+- Залито: `nail/shared-sqs:v0.1.35` (+latest; было только v0.1.25), `nail/iot-service:v0.1.13`,
+  `nail/iot-emqx:v0.2.4`.
+- Push: `e30bf74..f80e1b4`.

Makefile

@@ -1,8 +1,10 @@
 # Makefile — shared-sqs
 # Created: 2026-04-09
-# Registry: Docker Hub (naeel/shared-sqs)
+# Registry: Gitea (gitea.services.ngcloud.ru) — managed-платформа тянет образы отсюда.
+# Проверено 2026-10-11: gitea.services.ngcloud.ru/nail/shared-sqs существует (login naeel).
 
-IMAGE_REPO=naeel/shared-sqs
+REGISTRY ?= gitea.services.ngcloud.ru/nail
+IMAGE_REPO=$(REGISTRY)/shared-sqs
 VERSION=v0.1.35
 LDFLAGS=-X shared-sqs/app/models.Version=$(VERSION)
 BINARY=shared-sqs

app/admin/admin.go

@@ -219,8 +219,10 @@ func (h *Handler) uiCredentials(w http.ResponseWriter, r *http.Request) {
 // НЕ вызывает PingNubesAPI повторно — токен уже провалидирован при /ui/api/auth.
 func (h *Handler) jwtMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		// /ui/api/auth — публичный endpoint, пропускаем без проверки JWT
+		// /ui/api/auth — публичный endpoint, пропускаем без проверки JWT.
-		if strings.HasSuffix(r.URL.Path, "/auth") {
+		// Сравниваем ТОЧНЫЙ путь, а не суффикс "/auth": иначе, например,
+		// DELETE /ui/api/tenants/{id}/queues/auth пропускался бы без JWT и IDOR-проверки.
+		if r.URL.Path == "/ui/api/auth" {
 			next.ServeHTTP(w, r)
 			return
 		}
@@ -508,6 +510,12 @@ func (h *Handler) createTenantQueue(w http.ResponseWriter, r *http.Request) {
 		jsonErr(w, http.StatusBadRequest, "name is required")
 		return
 	}
+	// Та же валидация имени, что и в SQS CreateQueue (иначе имена с ':'/'/'
+	// ломали бы ключ "{accessKey}:{name}").
+	if err := gosqs.ValidateQueueName(req.Name); err != nil {
+		jsonErr(w, http.StatusBadRequest, "invalid queue name")
+		return
+	}
 	key := t.AccessKey + ":" + req.Name
 	models.SyncQueues.Lock()
 	if _, exists := models.SyncQueues.Queues[key]; exists {
@@ -650,18 +658,28 @@ func (h *Handler) sendMessageToQueue(w http.ResponseWriter, r *http.Request) {
 		jsonErr(w, http.StatusBadRequest, "body is required")
 		return
 	}
-	// Сообщение строится через общий билдер (UUID + MD5), + проверка размера —
+	// Сообщение строится через общий билдер (UUID + MD5), + проверка размера и
-	// как в SQS SendMessage.
+	// лимита — как в SQS SendMessage (иначе UI обходил бы OOM-защиту).
 	q.LockMu()
 	if q.MaximumMessageSize > 0 && len(req.Body) > q.MaximumMessageSize {
 		q.UnlockMu()
 		jsonErr(w, http.StatusBadRequest, "message too big")
 		return
 	}
+	if q.LiveCount() >= gosqs.MaxMessagesForQueue(q.IsFIFO) {
+		q.UnlockMu()
+		jsonErr(w, http.StatusTooManyRequests, "queue is full")
+		return
+	}
 	msg := gosqs.BuildSqsMessage(req.Body)
-	q.AppendMessage(msg)
+	idx := q.AppendMessage(msg)
-	// Персистим одно сообщение отдельно
+	// Персистим СИНХРОННО (как SendMessage) — иначе 201 без гарантии записи.
-	persistence.SaveMessage(key, &msg)
+	if err := persistence.SaveMessageSync(key, &msg); err != nil {
+		q.RemoveIndex(idx)
+		q.UnlockMu()
+		jsonErr(w, http.StatusInternalServerError, "failed to persist message")
+		return
+	}
 	persistence.SaveQueue(key, q)
 	q.UnlockMu()
 	w.Header().Set("Content-Type", "application/json")

app/auth/auth_middleware.go

@@ -163,9 +163,14 @@ func verifyHeaderSigV4(r *http.Request, t *tenant.Tenant, authHeader string) err
 	if err != nil {
 		return err
 	}
+	actualHash := sha256Hex(body)
 	payloadHash := r.Header.Get("X-Amz-Content-Sha256")
 	if payloadHash == "" {
-		payloadHash = sha256Hex(body)
+		payloadHash = actualHash
+	} else if !strings.EqualFold(payloadHash, actualHash) && !strings.EqualFold(payloadHash, "UNSIGNED-PAYLOAD") {
+		// Заголовок заявлен, но не совпадает с фактическим хэшем тела: клиент
+		// не должен иметь возможность подменить тело под подписанным заголовком.
+		return errors.New("X-Amz-Content-Sha256 does not match body hash")
 	}
 
 	canonicalRequest, err := buildCanonicalRequest(r, signedHeaders, payloadHash, true)

app/auth/cache.go

@@ -28,6 +28,7 @@ const cacheCleanupInterval = 5 * time.Minute
 
 type tokenCacheEntry struct {
 	validUntil    time.Time
+	graceDeadline time.Time // потолок grace от последней УСПЕШНОЙ проверки (не продлевается)
 }
 
 var (
@@ -66,18 +67,20 @@ func PingNubesAPICached(ctx context.Context, endpoint, token string) error {
 		return err
 	}
 	if err != nil {
-		// Сетевая ошибка шлюза: если токен ранее был валиден — даём grace.
+		// Сетевая ошибка шлюза: если токен ранее был валиден — даём grace, но
-		if ok {
+		// НЕ дальше graceDeadline. Раньше validUntil сдвигали вперёд на каждый
-			tokenCacheMu.Lock()
+		// сетевой сбой — отозванный токен жил бы вечно при повторных ошибках.
-			tokenCache[key] = tokenCacheEntry{validUntil: time.Now().Add(cacheGraceTTL)}
+		if ok && now.Before(entry.graceDeadline) {
-			tokenCacheMu.Unlock()
 			return nil
 		}
 		return err
 	}
 
 	tokenCacheMu.Lock()
-	tokenCache[key] = tokenCacheEntry{validUntil: time.Now().Add(cacheTTL)}
+	tokenCache[key] = tokenCacheEntry{
+		validUntil:    now.Add(cacheTTL),
+		graceDeadline: now.Add(cacheGraceTTL),
+	}
 	tokenCacheMu.Unlock()
 	return nil
 }
@@ -85,8 +88,12 @@ func PingNubesAPICached(ctx context.Context, endpoint, token string) error {
 // WarmTokenCache — записывает токен в кэш как валидный без вызова шлюза.
 // Используется после успешной свежей проверки при логине (jwtAuth).
 func WarmTokenCache(token string) {
+	now := time.Now()
 	tokenCacheMu.Lock()
-	tokenCache[tokenCacheKey(token)] = tokenCacheEntry{validUntil: time.Now().Add(cacheTTL)}
+	tokenCache[tokenCacheKey(token)] = tokenCacheEntry{
+		validUntil:    now.Add(cacheTTL),
+		graceDeadline: now.Add(cacheGraceTTL),
+	}
 	tokenCacheMu.Unlock()
 }
 

app/auth/jwt.go

@@ -59,7 +59,10 @@ func ParseJWTClaims(token string) (*JWTClaims, error) {
 	if claims.Sub == "" {
 		return nil, fmt.Errorf("JWT missing 'sub' claim")
 	}
-	if claims.Exp > 0 && claims.Exp < time.Now().Unix() {
+	if claims.Exp == 0 {
+		return nil, fmt.Errorf("JWT missing 'exp' claim")
+	}
+	if claims.Exp < time.Now().Unix() {
 		return nil, fmt.Errorf("JWT token expired")
 	}
 	return &claims, nil
@@ -71,11 +74,11 @@ var ErrTokenRejected = errors.New("token rejected by nubes API")
 
 // nubesAPIEndpoints — стенды Nubes API Gateway, по которым валидируется токен.
 // Токен принимается, если ХОТЯ БЫ ОДИН стенд его принял.
-// Юзер может вводить токен от любого стенда — главное, что токен действительный.
+// Безопасность: только prod-шлюз. Тестовый/dev-шлюзы принимают токены своих
+// стендов, и при их наличии токен тестового стенда считался бы валидным на проде.
+// Специфичный endpoint можно переопределить через NUBES_ENDPOINT.
 var nubesAPIEndpoints = []string{
 	"https://lk-api-gateway.ngcloud.ru/api/v1/svc",
-	"https://lk-api-gateway-test.ngcloud.ru/api/v1/svc",
-	"https://lk-api-gateway-dev.ngcloud.ru/api/v1/svc",
 }
 
 // nubesHTTPClient — общий клиент для валидации токенов: переиспользует

app/billing/billing.go

@@ -9,6 +9,7 @@ import (
 	"database/sql"
 	"fmt"
 	"os"
+	"sync"
 
 	// PostgreSQL драйвер — регистрируется в database/sql через init()
 	_ "github.com/lib/pq"
@@ -16,8 +17,26 @@ import (
 )
 
 // db — подключение к PostgreSQL для записи usage-данных.
-// nil если billing отключён.
+// nil если billing отключён. Доступ — только через getDB/setDB (мьютекс):
-var db *sql.DB
+// usageWorker читает db параллельно с Close/Init — иначе data race и nil-паника.
+var (
+	db   *sql.DB
+	dbMu sync.Mutex
+)
+
+// getDB — потокобезопасное чтение db.
+func getDB() *sql.DB {
+	dbMu.Lock()
+	defer dbMu.Unlock()
+	return db
+}
+
+// setDB — потокобезопасная установка db.
+func setDB(d *sql.DB) {
+	dbMu.Lock()
+	db = d
+	dbMu.Unlock()
+}
 
 // Init — подключается к PostgreSQL и создаёт таблицу sqs_usage_records если не существует.
 // Env переменные: BILLING_PG_HOST, BILLING_PG_PORT, BILLING_PG_DATABASE, BILLING_PG_USER,
@@ -46,28 +65,28 @@ func Init() {
 		host, port, dbname, user, password, sslmode)
 
 	var err error
-	db, err = sql.Open("postgres", dsn)
+	d, err := sql.Open("postgres", dsn)
 	if err != nil {
 		log.Errorf("billing: failed to open PostgreSQL: %v", err)
 		return
 	}
 
 	// Проверяем реальное подключение (Open не подключается)
-	if err = db.Ping(); err != nil {
+	if err = d.Ping(); err != nil {
 		log.Errorf("billing: failed to connect to PostgreSQL: %v", err)
-		db.Close()
+		d.Close()
-		db = nil
 		return
 	}
 
 	// Ограничиваем пул — billing не должен отжирать коннекты у основной БД
-	db.SetMaxOpenConns(5)
+	d.SetMaxOpenConns(5)
-	db.SetMaxIdleConns(2)
+	d.SetMaxIdleConns(2)
 
+	setDB(d)
 	if err = autoMigrate(); err != nil {
 		log.Errorf("billing: failed to create table: %v", err)
-		db.Close()
+		setDB(nil)
-		db = nil
+		d.Close()
 		return
 	}
 
@@ -77,7 +96,7 @@ func Init() {
 // autoMigrate — создаёт таблицу и индекс если не существуют.
 // Идемпотентно — безопасно вызывать при каждом старте.
 func autoMigrate() error {
-	_, err := db.Exec(`
+	_, err := getDB().Exec(`
 		CREATE TABLE IF NOT EXISTS sqs_usage_records (
 			id          BIGSERIAL PRIMARY KEY,
 			tenant_id   TEXT NOT NULL,
@@ -112,37 +131,70 @@ type usageRecord struct {
 // один воркер, читающий из буферизованного канала.
 var usageChan = make(chan usageRecord, 1024)
 
+// done — сигнал остановки воркера при graceful shutdown.
+var done = make(chan struct{})
+
+// closeOnce гарантирует однократный close(done).
+var closeOnce sync.Once
+
+// workerWg — ожидание завершения воркера перед закрытием db (иначе Close обнулил
+// бы db, пока воркер пишет — nil-паника + потеря остатка буфера usageChan).
+var workerWg sync.WaitGroup
+
 // init запускает воркер-горутину биллинга. Воркер стартует всегда; если db == nil
 // (billing отключён) — он просто дропает записи.
 func init() {
+	workerWg.Add(1)
 	go usageWorker()
 }
 
 // usageWorker — единственная горутина, пишущая usage-записи в PostgreSQL.
 func usageWorker() {
-	for rec := range usageChan {
+	defer workerWg.Done()
-		if db == nil {
+	for {
-			continue
+		select {
+		case rec := <-usageChan:
+			processUsage(rec)
+		case <-done:
+			// Дренируем остаток буфера перед выходом (не теряем usage).
+			for {
+				select {
+				case rec := <-usageChan:
+					processUsage(rec)
+				default:
+					return
 				}
-		_, err := db.Exec(
+			}
+		}
+	}
+}
+
+// processUsage — одна запись usage в PostgreSQL (db == nil → дроп).
+func processUsage(rec usageRecord) {
+	d := getDB()
+	if d == nil {
+		return
+	}
+	_, err := d.Exec(
 		`INSERT INTO sqs_usage_records (tenant_id, operation, queue_name, msg_count, msg_bytes) VALUES ($1, $2, $3, $4, $5)`,
 		rec.tenantID, rec.operation, rec.queueName, rec.msgCount, rec.msgBytes,
 	)
 	if err != nil {
 		log.Errorf("billing: failed to record usage [%s/%s]: %v", rec.tenantID, rec.operation, err)
 	}
-	}
 }
 
 // RecordUsage — ставит запись в очередь биллинга (асинхронно, без блокировки SQS).
 // Если billing отключён — no-op. Если буфер переполнен — запись дропается (биллинг
 // жертвуем ради стабильности основного пути), ошибка логируется.
 func RecordUsage(tenantID, operation, queueName string, msgCount int, msgBytes int64) {
-	if db == nil {
+	if getDB() == nil {
 		return
 	}
 	select {
 	case usageChan <- usageRecord{tenantID, operation, queueName, msgCount, msgBytes}:
+	case <-done:
+		// Воркер остановлен (shutdown) — запись не примется.
 	default:
 		log.Warnf("billing: буфер полон, запись [%s/%s] дропнута", tenantID, operation)
 	}
@@ -150,14 +202,20 @@ func RecordUsage(tenantID, operation, queueName string, msgCount int, msgBytes i
 
 // Enabled — возвращает true если billing подключён к PostgreSQL
 func Enabled() bool {
-	return db != nil
+	return getDB() != nil
 }
 
-// Close — закрывает подключение к PostgreSQL. Вызывается при graceful shutdown.
+// Close — останавливает воркер, дожидается его завершения и закрывает db.
+// Вызывается при graceful shutdown. Порядок обязателен: сначала воркер,
+// потом db.Close — иначе гонка db=nil с Exec воркера (nil-паника).
 func Close() {
+	closeOnce.Do(func() { close(done) })
+	workerWg.Wait()
+	dbMu.Lock()
 	if db != nil {
 		db.Close()
 		db = nil
-		log.Info("billing: PostgreSQL connection closed")
 	}
+	dbMu.Unlock()
+	log.Info("billing: PostgreSQL connection closed")
 }

app/cmd/goaws.go

@@ -101,12 +101,15 @@ func main() {
 	redisPass := os.Getenv("REDIS_PASSWORD")
 	if redisAddr != "" {
 		if err := persistence.Connect(redisAddr, redisUser, redisPass); err != nil {
-			log.Warnf("Не удалось подключиться к Redis: %v — работаем в memory-only режиме", err)
+			// Fail-fast: при явно заданном REDIS_ADDR, но недоступном Redis,
+			// продолжать опасно — SaveQueue перезапишет Redis пустотой, read-through
+			// и failover сломаются. Падаем, чтобы systemd перезапустил.
+			log.Fatalf("Не удалось подключиться к Redis (%s): %v", redisAddr, err)
 		}
 	}
 
 	// Восстановление состояния из Redis (тенанты + очереди)
-	if persistence.Client != nil {
+	if persistence.Client.Load() != nil {
 		// Загружаем тенантов
 		tenantsRaw, err := persistence.LoadAllTenantsRaw()
 		if err != nil {
@@ -124,7 +127,7 @@ func main() {
 		// Загружаем очереди
 		queues, err := persistence.LoadAllQueues()
 		if err != nil {
-			log.Warnf("Ошибка загрузки очередей из Redis: %v", err)
+			log.Fatalf("Ошибка загрузки очередей из Redis: %v", err)
 		} else {
 			models.SyncQueues.Lock()
 			for k, q := range queues {
@@ -136,7 +139,7 @@ func main() {
 
 	// Бэкфилл индекса accessKey→tenant для read-through (идемпотентно; нужен для
 	// тенантов, созданных до появления индекса, и для multi-instance сценария).
-	if persistence.Client != nil {
+	if persistence.Client.Load() != nil {
 		for _, t := range tenantStore.List() {
 			if data, err := json.Marshal(t); err == nil {
 				persistence.SaveTenantRaw(t.ID, data)
@@ -254,9 +257,6 @@ func main() {
 		elector.Stop()
 	}
 
-	// Закрыть billing (если был подключён)
-	billing.Close()
-
 	// Дать 10 секунд на завершение текущих HTTP запросов
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
@@ -266,5 +266,9 @@ func main() {
 	if err := controlSrv.Shutdown(ctx); err != nil {
 		log.Errorf("Control-plane shutdown error: %v", err)
 	}
+
+	// Закрыть billing ПОСЛЕ Shutdown — иначе in-flight запросы писали бы usage
+	// в уже закрытую БД (ревью 2026-10-11).
+	billing.Close()
 	log.Info("shared-sqs stopped")
 }

app/conf/config.go

@@ -129,6 +129,9 @@ func LoadYamlConfig(filename string, env string) []string {
 			err := setQueueRedrivePolicy(models.SyncQueues.Queues, q, queue.RedrivePolicy)
 			if err != nil {
 				log.Errorf("err: %s", err)
+				// Разблокировать глобальный мьютекс перед выходом — иначе все
+				// операции с очередями зависнут навсегда (deadlock).
+				models.SyncQueues.Unlock()
 				return ports
 			}
 		}

app/gosqs/change_message_visibility.go

@@ -56,12 +56,13 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo
 	messageFound := false
 	var changedMsgUuid string
 	var msgRemoved bool
+	var dlqMsg *models.SqsMessage
 	// O(1)-поиск по индексу ReceiptHandle (P1.2) вместо скана всего среза.
 	if i, ok := queue.IndexOfReceipt(receiptHandle); ok {
 		if visibilityTimeout == 0 {
 			// Возврат видимости (+ DLQ-перенос при исчерпании MaxReceiveCount) —
 			// единый путь с батчевой версией.
-			changedMsgUuid, msgRemoved = queue.ResetVisibilityAndTryDLQ(i, receiptHandle)
+			changedMsgUuid, msgRemoved, dlqMsg = queue.ResetVisibilityAndTryDLQ(i, receiptHandle)
 		} else {
 			queue.Messages[i].VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second)
 			changedMsgUuid = queue.Messages[i].Uuid
@@ -71,8 +72,12 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo
 	// Персистим изменения в Redis под Lock (сетевые записи асинхронны).
 	if messageFound {
 		if msgRemoved {
-			// Сообщение удалено (перемещено в DLQ) — удаляем из Redis
+			// Сообщение удалено (перемещено в DLQ) — удаляем из Redis исходной
-			persistence.DeleteMessagePersist(key, changedMsgUuid)
+			// очереди и сохраняем в DLQ, иначе после рестарта оно пропадёт из DLQ.
+			_ = persistence.DeleteMessagePersist(key, changedMsgUuid)
+			if dlqMsg != nil {
+				persistence.SaveMessage(dlqKeyFor(key, queue.DeadLetterQueue.Name), dlqMsg)
+			}
 		} else if i, ok := queue.IndexOfUUID(changedMsgUuid); ok {
 			// Сообщение обновлено — перезаписываем в Redis
 			persistence.SaveMessage(key, &queue.Messages[i])

app/gosqs/change_message_visibility_batch.go

@@ -86,10 +86,14 @@ func ChangeMessageVisibilityBatchV1(req *http.Request) (int, interfaces.Abstract
 				// Сброс + DLQ-перенос при исчерпании MaxReceiveCount — единый путь
 				// с одиночной версией (раньше батч лишь инкрементировал Retry и не
 				// переносил в DLQ — счётчик и перенос могли расходиться).
-				uuid, moved := queue.ResetVisibilityAndTryDLQ(i, entry.ReceiptHandle)
+				uuid, moved, dlqMsg := queue.ResetVisibilityAndTryDLQ(i, entry.ReceiptHandle)
 				if moved {
-					// Сообщение перенесено в DLQ и удалено из очереди — чистим Redis.
+					// Сообщение перенесено в DLQ и удалено из очереди — чистим Redis
-					persistence.DeleteMessagePersist(key, uuid)
+					// исходной и сохраняем в DLQ (иначе пропадёт после рестарта).
+					_ = persistence.DeleteMessagePersist(key, uuid)
+					if dlqMsg != nil {
+						persistence.SaveMessage(dlqKeyFor(key, queue.DeadLetterQueue.Name), dlqMsg)
+					}
 				} else if j, ok := queue.IndexOfUUID(uuid); ok {
 					persistence.SaveMessage(key, &queue.Messages[j])
 				}

app/gosqs/create_queue.go

@@ -59,6 +59,16 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 		IsFIFO:           utils.HasFIFOQueueName(queueName),
 		EnableDuplicates: models.CurrentEnvironment.EnableDuplicates,
 		Duplicates:       make(map[string]time.Time),
+		// Дефолты (как в InitQueues): без них очередь из API получала бы
+		// VisibilityTimeout=0 и т.д. — сообщение сразу видимо, retention 0 = нет
+		// эвикции. Переданные атрибуты перезапишут дефолты в setQueueAttributesV1.
+		VisibilityTimeout:             models.CurrentEnvironment.QueueAttributeDefaults.VisibilityTimeout,
+		MaximumMessageSize:            models.CurrentEnvironment.QueueAttributeDefaults.MaximumMessageSize,
+		MessageRetentionPeriod:        models.CurrentEnvironment.QueueAttributeDefaults.MessageRetentionPeriod,
+		ReceiveMessageWaitTimeSeconds: models.CurrentEnvironment.QueueAttributeDefaults.ReceiveMessageWaitTimeSeconds,
+		// CreatedTimestamp/LastModifiedTimestamp в GetQueueAttributes (AWS их отдаёт).
+		CreatedAt:      time.Now().UTC(),
+		LastModifiedAt: time.Now().UTC(),
 	}
 	// Список имён переданных атрибутов — для whitelist-валидации
 	// (Query-протокол: Attribute.N.Name; JSON: nil — проверка пропускается).
@@ -71,7 +81,8 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 	}
 
 	models.SyncQueues.Lock()
-	if _, exists := models.SyncQueues.Queues[key]; !exists {
+	existing, exists := models.SyncQueues.Queues[key]
+	if !exists {
 		// Проверка лимита очередей тенанта
 		if t.MaxQueues > 0 && countTenantQueues(t.AccessKey) >= t.MaxQueues {
 			models.SyncQueues.Unlock()
@@ -79,10 +90,18 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 		}
 		log.Infof("Creating Queue: %s (tenant: %s)", queueName, t.ID)
 		models.SyncQueues.AddQueue(key, queue)
+		existing = queue
 	}
-	// Сохраняем очередь в Redis пока держим Lock — консистентный снапшот
-	persistence.SaveQueue(key, models.SyncQueues.Queues[key])
 	models.SyncQueues.Unlock()
+
+	// SaveQueue маршалит map'ы очереди (Duplicates/FIFO*/Tags) в JSON. Это нужно
+	// делать под per-queue блокировкой, а НЕ под глобальным SyncQueues.Lock:
+	// мутаторы (send/tag/CMV) пишут в эти map под queue.mu, и идемпотентный
+	// CreateQueue существующей очереди без queue.mu даёт fatal
+	// "concurrent map iteration and map write" (краш всего процесса).
+	existing.LockMu()
+	persistence.SaveQueue(key, existing)
+	existing.UnlockMu()
 	respStruct := models.CreateQueueResponse{
 		Xmlns:    models.BaseXmlns,
 		Result:   models.CreateQueueResult{QueueUrl: queueUrl},

app/gosqs/delete_message.go

@@ -56,9 +56,15 @@ func DeleteMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 		queue.ClearInFlight(i, receiptHandle)
 		queue.UnlockGroup(msg.GroupID)
 		queue.RemoveIndex(i) // tombstone за O(1) + подрезка хвоста/амортизированная компакция
-		delete(queue.Duplicates, msg.DeduplicationID)
+		// ВАЖНО: запись в Duplicates НЕ удаляем — AWS держит dedup-окно 5 минут
-		// Удаляем одно сообщение из Redis — O(1)
+		// независимо от удаления сообщения (иначе повторный send с тем же
-		persistence.DeleteMessagePersist(key, msg.Uuid)
+		// dedupId доставлялся бы повторно). Чистит только истечение окна (PeriodicTasks).
+		// Удаляем одно сообщение из Redis — O(1). Ошибка → 500: иначе клиент
+		// получил бы 200, а после рестарта сообщение «воскресло» бы.
+		if err := persistence.DeleteMessagePersist(key, msg.Uuid); err != nil {
+			log.Errorf("DeleteMessage: не удалось удалить сообщение из Redis: %v", err)
+			return utils.CreateErrorResponseV1("InternalError", true)
+		}
 		// Метаданные очереди (duplicates, FIFO state)
 		persistence.SaveQueue(key, queue)
 		respStruct := models.DeleteMessageResponse{

app/gosqs/delete_message_batch.go

@@ -54,6 +54,12 @@ func DeleteMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBo
 			return utils.CreateErrorResponseV1("BatchEntryIdsNotDistinct", true)
 		}
 		ids[v.Id] = true
+		// Пустой ReceiptHandle опасен: в JSON-протоколе (x-amz-json-1.0) он не
+		// отсекается парсером и совпадает с ReceiptHandle НЕвыданных сообщений (""),
+		// что удаляло бы все такие сообщения очереди одним запросом.
+		if v.ReceiptHandle == "" {
+			return utils.CreateErrorResponseV1("InvalidParameterValue", true)
+		}
 	}
 
 	queue := models.SyncQueues.GetLocked(key)
@@ -62,13 +68,19 @@ func DeleteMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBo
 	}
 	defer queue.UnlockMu()
 
-	deleteMessageMap := make(map[string]*deleteEntry)
+	// Ключ — Id, а НЕ ReceiptHandle: два entry с одинаковым ReceiptHandle и разными
+	// Id не должны перезаписывать друг друга (первый исчезал бы и не попадал ни в
+	// Successful, ни в Failed — ревью 2026-10-11). entriesByHandle — обратный индекс.
+	entriesByID := make(map[string]*deleteEntry)
+	entriesByHandle := make(map[string][]*deleteEntry)
 	for _, entry := range requestBody.Entries {
-		deleteMessageMap[entry.ReceiptHandle] = &deleteEntry{
+		de := &deleteEntry{
 			Id:            entry.Id,
 			ReceiptHandle: entry.ReceiptHandle,
 			Deleted:       false,
 		}
+		entriesByID[entry.Id] = de
+		entriesByHandle[entry.ReceiptHandle] = append(entriesByHandle[entry.ReceiptHandle], de)
 	}
 
 	deletedEntries := make([]models.DeleteMessageBatchResultEntry, 0)
@@ -81,28 +93,33 @@ func DeleteMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBo
 			continue
 		}
 		message := &queue.Messages[i]
-		de, found := deleteMessageMap[message.ReceiptHandle]
+		des, found := entriesByHandle[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
+		groupID, uuid := message.GroupID, message.Uuid
 		queue.ClearInFlight(i, message.ReceiptHandle)
 		queue.UnlockGroup(groupID)
-		delete(queue.Duplicates, dedupID)
+		// Duplicates НЕ чистим — dedup-окно 5 минут живёт независимо от удаления (AWS).
+		for _, de := range des {
 			de.Deleted = true
 			deletedEntries = append(deletedEntries, models.DeleteMessageBatchResultEntry{Id: de.Id})
+		}
 		deletedUuids = append(deletedUuids, uuid)
 		queue.RemoveIndex(i)
 	}
-	// Удаляем сообщения из Redis отдельно — O(batch_size)
+	// Удаляем сообщения из Redis отдельно — O(batch_size). Ошибка → InternalError.
-	persistence.DeleteMessagesPersist(key, deletedUuids)
+	if err := persistence.DeleteMessagesPersist(key, deletedUuids); err != nil {
+		log.Errorf("DeleteMessageBatch: не удалось удалить сообщения из Redis: %v", err)
+		return utils.CreateErrorResponseV1("InternalError", true)
+	}
 	// Метаданные очереди (duplicates, FIFO state)
 	persistence.SaveQueue(key, queue)
 
 	notFoundEntries := make([]models.BatchResultErrorEntry, 0)
-	for _, de := range deleteMessageMap {
+	for _, de := range entriesByID {
 		if !de.Deleted {
 			notFoundEntries = append(notFoundEntries, models.BatchResultErrorEntry{
 				Code:        "1",

app/gosqs/get_queue_attributes.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
 	"shared-sqs/app/interfaces"
 	"shared-sqs/app/models"
@@ -86,6 +87,13 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo
 	if shouldInclude("VisibilityTimeout") {
 		queueAttributes = append(queueAttributes, models.Attribute{Name: "VisibilityTimeout", Value: strconv.Itoa(queue.VisibilityTimeout)})
 	}
+	// FIFO-атрибуты: читаются Terraform (aws_sqs_queue) и aws cli; без них — drift.
+	if shouldInclude("FifoQueue") {
+		queueAttributes = append(queueAttributes, models.Attribute{Name: "FifoQueue", Value: strconv.FormatBool(queue.IsFIFO)})
+	}
+	if shouldInclude("ContentBasedDeduplication") {
+		queueAttributes = append(queueAttributes, models.Attribute{Name: "ContentBasedDeduplication", Value: strconv.FormatBool(queue.ContentBasedDeduplication)})
+	}
 	if shouldInclude("ApproximateNumberOfMessages") {
 		queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessages", Value: strconv.Itoa(queue.LiveCount())})
 	}
@@ -93,10 +101,10 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo
 		queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessagesNotVisible", Value: strconv.Itoa(numberOfHiddenMessagesInQueue(queue))})
 	}
 	if shouldInclude("CreatedTimestamp") {
-		queueAttributes = append(queueAttributes, models.Attribute{Name: "CreatedTimestamp", Value: "0000000000"})
+		queueAttributes = append(queueAttributes, models.Attribute{Name: "CreatedTimestamp", Value: epochSeconds(queue.CreatedAt)})
 	}
 	if shouldInclude("LastModifiedTimestamp") {
-		queueAttributes = append(queueAttributes, models.Attribute{Name: "LastModifiedTimestamp", Value: "0000000000"})
+		queueAttributes = append(queueAttributes, models.Attribute{Name: "LastModifiedTimestamp", Value: epochSeconds(queue.LastModifiedAt)})
 	}
 	if shouldInclude("QueueArn") {
 		queueAttributes = append(queueAttributes, models.Attribute{Name: "QueueArn", Value: queue.Arn})
@@ -107,6 +115,10 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo
 			Value: fmt.Sprintf(`{"maxReceiveCount":"%d", "deadLetterTargetArn":"%s"}`, queue.MaxReceiveCount, queue.DeadLetterQueue.Arn),
 		})
 	}
+	// Policy — отдаём сохранённую JSON-строку как есть (совместимость с Terraform).
+	if shouldInclude("Policy") && queue.Policy != "" {
+		queueAttributes = append(queueAttributes, models.Attribute{Name: "Policy", Value: queue.Policy})
+	}
 
 	respStruct := models.GetQueueAttributesResponse{
 		Xmlns:    models.BaseXmlns,
@@ -115,3 +127,12 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo
 	}
 	return http.StatusOK, respStruct
 }
+
+// epochSeconds — timestamp в формате AWS (unix-секунды строкой).
+// Нулевое время отдаём как "0" (стабильно, без drift при чтении).
+func epochSeconds(t time.Time) string {
+	if t.IsZero() {
+		return "0"
+	}
+	return strconv.FormatInt(t.Unix(), 10)
+}

app/gosqs/gosqs.go

@@ -48,9 +48,18 @@ func PeriodicTasks(d time.Duration, quit chan bool) {
 			// Каждую очередь обрабатываем под ОТДЕЛЬНЫМ Lock — блокировка одной
 			// очереди не мешает обработке остальных и HTTP-запросам.
 			for _, qi := range queues {
+				func() {
+					// recover: паника в processQueueTick одной очереди не должна
+					// убивать весь процесс; mutex освобождается в defer.
+					defer qi.queue.UnlockMu()
+					defer func() {
+						if r := recover(); r != nil {
+							log.Errorf("gosqs: panic in processQueueTick for queue %q: %v", qi.key, r)
+						}
+					}()
 					qi.queue.LockMu()
 					processQueueTick(qi.key, qi.queue)
-				qi.queue.UnlockMu()
+				}()
 			}
 		case <-quit:
 			ticker.Stop()
@@ -104,7 +113,7 @@ func processQueueTick(key string, queue *models.Queue) {
 		if len(expired) > 0 {
 			log.Infof("Queue [%s]: evicted %d messages by MessageRetentionPeriod", queue.Name, len(expired))
 			// Удаляем просроченные из Redis, чтобы не воскресли после рестарта.
-			persistence.DeleteMessagesPersist(key, expired)
+			_ = persistence.DeleteMessagesPersist(key, expired)
 		}
 	}
 
@@ -146,10 +155,16 @@ func processQueueTick(key string, queue *models.Queue) {
 			queue.DeadLetterQueue != nil &&
 			msg.Retry >= queue.MaxReceiveCount {
 			dlqMsg := *msg
+			msgUuid := msg.Uuid
+			dlqName := queue.DeadLetterQueue.Name
 			queue.DeadLetterQueue.LockMu()
 			queue.DeadLetterQueue.AppendMessage(dlqMsg)
 			queue.DeadLetterQueue.UnlockMu()
 			queue.RemoveIndex(i)
+			// Персист переноса: иначе после рестарта сообщение «воскреснет»
+			// в исходной очереди (не удалено из Redis) и пропадёт из DLQ.
+			_ = persistence.DeleteMessagePersist(key, msgUuid)
+			persistence.SaveMessage(dlqKeyFor(key, dlqName), &dlqMsg)
 		}
 	}
 

app/gosqs/memory_guard.go

@@ -0,0 +1,37 @@
+// app/gosqs/memory_guard.go
+// Глобальный предохранитель памяти процесса (защита от OOM одним тенантом).
+//
+// Лимиты по числу сообщений (MaxMessagesForQueue) не ограничивают СУММАРНЫЙ
+// объём: 120k × 256 КБ ≈ 30 ГБ на одну очередь. Этот предохранитель отказывает
+// в SendMessage/SendMessageBatch по HeapAlloc выше порога — единственная защита
+// процесса от суммы всех тенантов.
+//
+// Задаётся env `SHARED_SQS_MAX_HEAP_MB` (0 = выключен). По умолчанию выключен:
+// конкретное число зависит от RAM ВМ и целевого числа тенантов (решение владельца).
+// Рекомендация на старт: ~70% от RAM ВМ, напр. 1024 для 1.5 ГБ ВМ.
+package gosqs
+
+import (
+	"os"
+	"runtime"
+	"strconv"
+)
+
+// maxHeapBytes — порог HeapAlloc в байтах; 0 = предохранитель выключен.
+var maxHeapBytes = func() int64 {
+	mb, err := strconv.ParseInt(os.Getenv("SHARED_SQS_MAX_HEAP_MB"), 10, 64)
+	if err != nil || mb <= 0 {
+		return 0
+	}
+	return mb * 1024 * 1024
+}()
+
+// heapOverLimit — true, если текущий HeapAlloc превышает порог.
+func heapOverLimit() bool {
+	if maxHeapBytes <= 0 {
+		return false
+	}
+	var ms runtime.MemStats
+	runtime.ReadMemStats(&ms)
+	return int64(ms.HeapAlloc) > maxHeapBytes
+}

app/gosqs/queue_attributes.go

@@ -1,8 +1,10 @@
 package gosqs
 
 import (
+	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 
@@ -82,8 +84,42 @@ func setQueueAttributesV1(q *models.Queue, attr models.QueueAttributes, provided
 			log.Error("Invalid RedrivePolicy Attribute")
 			return fmt.Errorf("InvalidAttributeValue")
 		}
+		// Запрет self-DLQ (q → q) и циклов (A → B → A): при переносе в DLQ
+		// переносящий код берёт DeadLetterQueue.LockMu() под уже взятым q.mu —
+		// self/цикл даёт немедленный self-deadlock (или AB/BA-deadlock).
+		if deadLetterQueue == q {
+			log.Error("Invalid RedrivePolicy Attribute: self DLQ")
+			return fmt.Errorf("InvalidAttributeValue")
+		}
+		// Обход цепочки DLQ для детекта циклов. Лимит 100 — защита от уже
+		// существующего в данных цикла (если бы он там был).
+		for cur, steps := deadLetterQueue, 0; cur != nil && steps < 100; cur, steps = cur.DeadLetterQueue, steps+1 {
+			if cur == q {
+				log.Error("Invalid RedrivePolicy Attribute: cyclic DLQ")
+				return fmt.Errorf("InvalidAttributeValue")
+			}
+		}
 		q.DeadLetterQueue = deadLetterQueue
 		q.MaxReceiveCount = attr.RedrivePolicy.MaxReceiveCount.Int()
 	}
+
+	// Шаг 4: FIFO-атрибуты и Policy — храним как есть (совместимость с Terraform).
+	// FifoQueue только включаем (выключить FIFO у существующей очереди нельзя —
+	// как и в AWS). ContentBasedDeduplication и Policy не исполняются,
+	// но должны возвращаться в GetQueueAttributes, иначе drift в Terraform.
+	if attr.FifoQueue.Bool() {
+		q.IsFIFO = true
+	}
+	if attr.ContentBasedDeduplication.Bool() {
+		q.ContentBasedDeduplication = true
+	}
+	if attr.Policy != nil {
+		// Сохраняем канонический JSON (map → marshal сортирует ключи) —
+		// детерминированно, без drift при чтении.
+		if data, err := json.Marshal(attr.Policy); err == nil {
+			q.Policy = string(data)
+		}
+	}
+	q.LastModifiedAt = time.Now().UTC()
 	return nil
 }

app/gosqs/receive_message.go

@@ -244,7 +244,7 @@ func buildResultMessage(m *models.SqsMessage) *models.ResultMessage {
 		Attributes: map[string]string{
 			"ApproximateFirstReceiveTimestamp": fmt.Sprintf("%d", m.ReceiptTime.UnixNano()/int64(time.Millisecond)),
 			"SenderId":                         models.CurrentEnvironment.AccountID,
-			"ApproximateReceiveCount":          fmt.Sprintf("%d", m.NumberOfReceives+1),
+			"ApproximateReceiveCount":          fmt.Sprintf("%d", m.NumberOfReceives),
 			"SentTimestamp":                    fmt.Sprintf("%d", m.SentTime.UnixNano()/int64(time.Millisecond)), // Фикс: реальное время отправки
 		},
 	}

app/gosqs/send_message.go

@@ -92,6 +92,7 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 	maxMessageSize := queue.MaximumMessageSize
 	currentMsgCount := queue.LiveCount()
 	queueIsFIFO := queue.IsFIFO
+	contentBasedDedup := queue.ContentBasedDeduplication
 	delaySecs := queue.DelaySeconds
 
 	// FIFO-очередь БЕЗ MessageGroupId — ошибка: AWS требует GroupId для каждой
@@ -100,6 +101,12 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 	if queueIsFIFO && messageGroupID == "" {
 		return utils.CreateErrorResponseV1("MissingParameter", true)
 	}
+	// ContentBasedDeduplication: если MessageDeduplicationId не передан, AWS сам
+	// вычисляет его из тела сообщения (SHA-256). Без этого включённый флаг
+	// ничего не делал бы и FIFO-дедупликация не работала.
+	if queueIsFIFO && contentBasedDedup && messageDeduplicationID == "" {
+		messageDeduplicationID = utils.GetSHA256Hash(messageBody)
+	}
 
 	if maxMessageSize > 0 && len(messageBody) > maxMessageSize {
 		return utils.CreateErrorResponseV1("MessageTooBig", true)
@@ -110,6 +117,9 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 		return utils.CreateErrorResponseV1("OverLimit", true)
 	}
 
+	if requestBody.DelaySeconds < 0 || requestBody.DelaySeconds > 900 {
+		return utils.CreateErrorResponseV1("InvalidParameterValue", true)
+	}
 	if requestBody.DelaySeconds != 0 {
 		delaySecs = requestBody.DelaySeconds
 	}
@@ -132,12 +142,12 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 	if queue.LiveCount() >= MaxMessagesForQueue(queue.IsFIFO) {
 		return utils.CreateErrorResponseV1("OverLimit", true)
 	}
-
+	// Глобальный предохранитель памяти (HeapAlloc) — защита от OOM суммой тенантов.
-	fifoSeqNumber := ""
+	if heapOverLimit() {
-	if queue.IsFIFO {
+		return utils.CreateErrorResponseV1("OverLimit", true)
-		fifoSeqNumber = queue.NextSequenceNumber(messageGroupID)
 	}
 
+	fifoSeqNumber := ""
 	persistErr := false
 	if !queue.IsDuplicate(messageDeduplicationID) {
 		idx := queue.AppendMessage(msg)
@@ -149,15 +159,21 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 			log.Errorf("SendMessage: не удалось персистить сообщение %s в очередь %s: %v", msg.Uuid, key, err)
 			queue.RemoveIndex(idx)
 			persistErr = true
+		} else {
+			// Регистрируем dedupId и порядковый номер ТОЛЬКО при успешной записи.
+			// Иначе ретрай с тем же dedupId получил бы 200 без сохранённого сообщения
+			// (InitDuplicatation после отката append помечал бы id как обработанный).
+			queue.InitDuplicatation(messageDeduplicationID)
+			if queue.IsFIFO {
+				fifoSeqNumber = queue.NextSequenceNumber(messageGroupID)
+			}
+			// Метаданные очереди (FIFO state, duplicates) — без Messages
+			persistence.SaveQueue(key, queue)
 		}
 	} else {
 		log.Debugf("Duplicate message deduplicationId [%s] in queue [%s]", messageDeduplicationID, queueName)
 	}
 
-	queue.InitDuplicatation(messageDeduplicationID)
-	// Сохраняем метаданные очереди (FIFO state, duplicates) — без Messages
-	persistence.SaveQueue(key, queue)
-
 	if persistErr {
 		return utils.CreateErrorResponseV1("InternalError", true)
 	}

app/gosqs/send_message_batch.go

@@ -111,9 +111,14 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
 	if queue.LiveCount()+len(sendEntries) > MaxMessagesForQueue(queue.IsFIFO) {
 		return utils.CreateErrorResponseV1("OverLimit", true)
 	}
+	// Глобальный предохранитель памяти (HeapAlloc) — защита от OOM суммой тенантов.
+	if heapOverLimit() {
+		return utils.CreateErrorResponseV1("OverLimit", true)
+	}
 
 	newMsgs := make([]models.SqsMessage, 0, len(sendEntries))
 	appendedIdx := make([]int, 0, len(sendEntries))
+	appendedDedupIDs := make([]string, 0, len(sendEntries))
 	for _, sendEntry := range sendEntries {
 		msg := models.SqsMessage{MessageBody: sendEntry.MessageBody}
 		if len(sendEntry.MessageAttributes) > 0 {
@@ -122,21 +127,41 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
 		}
 		msg.MD5OfMessageBody = utils.GetMD5Hash(sendEntry.MessageBody)
 		msg.GroupID = sendEntry.MessageGroupId
-		msg.DeduplicationID = sendEntry.MessageDeduplicationId
+		msg.DelaySecs = sendEntry.DelaySeconds
+		// ContentBasedDeduplication: dedupId вычисляется из тела (SHA-256),
+		// если не передан явно (AWS-семантика FIFO).
+		dedupID := sendEntry.MessageDeduplicationId
+		if queue.IsFIFO && queue.ContentBasedDeduplication && dedupID == "" {
+			dedupID = utils.GetSHA256Hash(sendEntry.MessageBody)
+		}
+		msg.DeduplicationID = dedupID
 		msg.Uuid = uuid.NewString()
 		msg.SentTime = time.Now()
 
+		// Дубликат проверяем ДО NextSequenceNumber/Append: иначе seq инкрементировался
+		// бы и для отброшенного дубликата, а два entry с одним dedupID в одном батче
+		// оба добавились бы (InitDuplicatation был после цикла — ревью 2026-10-11).
+		if queue.IsDuplicate(dedupID) {
+			log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", dedupID, queueName)
+			sentEntries = append(sentEntries, models.SendMessageBatchResultEntry{
+				Id:                     sendEntry.Id,
+				MessageId:              msg.Uuid,
+				MD5OfMessageBody:       msg.MD5OfMessageBody,
+				MD5OfMessageAttributes: msg.MD5OfMessageAttributes,
+				SequenceNumber:         "",
+			})
+			continue
+		}
+
 		fifoSeqNumber := ""
 		if queue.IsFIFO {
 			fifoSeqNumber = queue.NextSequenceNumber(sendEntry.MessageGroupId)
 		}
-		if !queue.IsDuplicate(sendEntry.MessageDeduplicationId) {
+		// Регистрируем dedupId сразу — второй entry с тем же id увидит его.
+		queue.InitDuplicatation(dedupID)
 		appendedIdx = append(appendedIdx, queue.AppendMessage(msg))
 		newMsgs = append(newMsgs, msg)
-		} else {
+		appendedDedupIDs = append(appendedDedupIDs, dedupID)
-			log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", sendEntry.MessageDeduplicationId, queueName)
-		}
-		queue.InitDuplicatation(sendEntry.MessageDeduplicationId)
 
 		sentEntries = append(sentEntries, models.SendMessageBatchResultEntry{
 			Id:                     sendEntry.Id,
@@ -155,6 +180,11 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
 		for _, idx := range appendedIdx {
 			queue.RemoveIndex(idx)
 		}
+		// Откат dedup-регистраций: иначе retry с тем же dedupId получил бы
+		// «duplicate» без сохранённых сообщений.
+		for _, id := range appendedDedupIDs {
+			delete(queue.Duplicates, id)
+		}
 		return utils.CreateErrorResponseV1("InternalError", true)
 	}
 	// Метаданные очереди (FIFO state, duplicates) — без Messages

app/gosqs/tenant_helpers.go

@@ -18,8 +18,15 @@ func tenantQueueKey(tenantAccessKey, queueName string) string {
 return tenantAccessKey + ":" + queueName
 }
 
-// getTenantFromContext — извлекает тенанта из request context.
+// dlqKeyFor — ключ DLQ по ключу исходной очереди (формат "{accessKey}:{name}")
-// Возвращает nil если тенант не найден (не должно быть — auth middleware должен это поймать раньше).
+// и имени DLQ. Используется для персиста перенесённых в DLQ сообщений.
+func dlqKeyFor(srcKey, dlqName string) string {
+	idx := strings.LastIndex(srcKey, ":")
+	if idx < 0 {
+		return srcKey + ":" + dlqName
+	}
+	return srcKey[:idx] + ":" + dlqName
+}
 func getTenantFromContext(r *http.Request) *tenant.Tenant {
 t, _ := r.Context().Value(auth.TenantContextKey).(*tenant.Tenant)
 return t

app/leader/leader.go

@@ -33,6 +33,10 @@ type Elector struct {
 	isLeader bool
 	stopCh   chan struct{}
 	stopped  bool
+
+	// lastRefresh — время последнего успешного refresh/acquire лока.
+	// Нужен для сброса лидерства при длительном сбое refresh (split-brain).
+	lastRefresh time.Time
 }
 
 // New — создаёт Elector. key — ключ лока, id — уникальный id инстанса.
@@ -43,6 +47,7 @@ func New(key, id string, ttl, interval time.Duration) *Elector {
 		ttl:         ttl,
 		interval:    interval,
 		stopCh:      make(chan struct{}),
+		lastRefresh: time.Now(),
 	}
 }
 
@@ -108,8 +113,15 @@ func (e *Elector) step() {
 		ok, err := persistence.RefreshLock(e.key, e.id, e.ttl)
 		if err != nil {
 			log.Errorf("leader: refresh: %v", err)
+			// Refresh не удаётся: если с последнего успеха прошло > TTL — лок
+			// гарантированно истёк, и другой инстанс мог стать лидером (split-brain).
+			// Сбрасываем лидерство (fail-safe), а не остаёмся «лидером» без лока.
+			if time.Since(e.lastRefreshAt()) > e.ttl {
+				e.setLeader(false)
+			}
 			return
 		}
+		e.markRefreshed()
 		if !ok {
 			e.setLeader(false)
 		}
@@ -122,5 +134,18 @@ func (e *Elector) step() {
 	}
 	if ok {
 		e.setLeader(true)
+		e.markRefreshed()
 	}
 }
+
+func (e *Elector) lastRefreshAt() time.Time {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+	return e.lastRefresh
+}
+
+func (e *Elector) markRefreshed() {
+	e.mu.Lock()
+	e.lastRefresh = time.Now()
+	e.mu.Unlock()
+}

app/models/conversions.go

@@ -34,3 +34,31 @@ func (s *StringToInt) UnmarshalJSON(data []byte) error {
 func (s *StringToInt) Int() int {
 	return int(*s)
 }
+
+// StringToBool — аналог StringToInt для булевых атрибутов: принимает и JSON-bool
+// (true/false), и строку ("true"/"false"). Query-протокол шлёт строки,
+// JSON-протокол — булевы значения; без этого FifoQueue/ContentBasedDeduplication
+// из JSON-запроса молча терялись бы.
+type StringToBool bool
+
+func (s *StringToBool) UnmarshalJSON(data []byte) error {
+	var b bool
+	if err := json.Unmarshal(data, &b); err == nil {
+		*s = StringToBool(b)
+		return nil
+	}
+	var str string
+	if err := json.Unmarshal(data, &str); err != nil {
+		return err
+	}
+	tmp, err := strconv.ParseBool(str)
+	if err != nil {
+		return err
+	}
+	*s = StringToBool(tmp)
+	return nil
+}
+
+func (s *StringToBool) Bool() bool {
+	return bool(*s)
+}

app/models/models.go

@@ -92,6 +92,16 @@ type Queue struct {
 	// Используется для запрета повторного 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
@@ -150,13 +160,12 @@ func (q *Queue) UnlockGroup(groupId string) {
 }
 
 // IsDuplicate — проверяет, было ли сообщение с таким deduplicationId уже
-// добавлено в очередь в пределах deduplication-окна.
+// добавлено в очередь в пределах deduplication-окна (5 минут).
-//
+// Дедупликация — НЕОТЪЕМЛЕМОЕ свойство FIFO-очереди (AWS): раньше она зависела
-// Логическая схема: дубликаты отслеживаются ТОЛЬКО для FIFO-очередей с
+// от env-флага EnableDuplicates=false по умолчанию, и FIFO-очереди не
-// включённой дупликацией (EnableDuplicates) и непустым deduplicationId.
+// дедуплицировали вообще. Теперь — только по IsFIFO и непустому id.
-// Для остальных случаев дубликаты не отслеживаются (AWS-семантика).
 func (q *Queue) IsDuplicate(deduplicationId string) bool {
-	if !q.EnableDuplicates || !q.IsFIFO || deduplicationId == "" {
+	if !q.IsFIFO || deduplicationId == "" {
 		return false
 	}
 
@@ -170,10 +179,15 @@ func (q *Queue) IsDuplicate(deduplicationId string) bool {
 // фоновой задачей PeriodicTasks (deduplication period expired).
 // Для не-FIFO очередей или пустого deduplicationId — no-op.
 func (q *Queue) InitDuplicatation(deduplicationId string) {
-	if !q.EnableDuplicates || !q.IsFIFO || deduplicationId == "" {
+	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()
 	}
@@ -335,6 +349,10 @@ func (q *Queue) Reset() {
 	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: строит индексы,
@@ -396,7 +414,17 @@ func (q *Queue) ClearInFlight(i int, receiptHandle string) {
 // этого retry-счётчик и DLQ-перенос могли расходиться.
 // Вызывать под write-блокировкой q. Перенос в DLQ выполняется под её блокировкой.
 // Возвращает: msgUuid, movedToDLQ.
-func (q *Queue) ResetVisibilityAndTryDLQ(i int, receiptHandle string) (msgUuid string, movedToDLQ bool) {
+// 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]
@@ -411,9 +439,9 @@ func (q *Queue) ResetVisibilityAndTryDLQ(i int, receiptHandle string) (msgUuid s
 		q.DeadLetterQueue.AppendMessage(dlqMsg)
 		q.DeadLetterQueue.UnlockMu()
 		q.RemoveIndex(i)
-		return msgUuid, true
+		return msgUuid, true, &dlqMsg
 	}
-	return msgUuid, false
+	return msgUuid, false, nil
 }
 
 // InFlightIndexes — снимок индексов in-flight сообщений (для visibility-тика).

app/models/requests.go

@@ -113,6 +113,20 @@ func (r *CreateQueueRequest) SetAttributesFromForm(values url.Values) {
 				continue
 			}
 			r.Attributes.RedriveAllowPolicy = tmp
+		case "FifoQueue":
+			tmp, err := strconv.ParseBool(attrValue)
+			if err != nil {
+				log.Debugf("Failed to parse form attribute - %s: %s", attrName, attrValue)
+				continue
+			}
+			r.Attributes.FifoQueue = StringToBool(tmp)
+		case "ContentBasedDeduplication":
+			tmp, err := strconv.ParseBool(attrValue)
+			if err != nil {
+				log.Debugf("Failed to parse form attribute - %s: %s", attrName, attrValue)
+				continue
+			}
+			r.Attributes.ContentBasedDeduplication = StringToBool(tmp)
 		}
 	}
 	return
@@ -381,6 +395,20 @@ func (r *SetQueueAttributesRequest) SetAttributesFromForm(values url.Values) {
 				continue
 			}
 			r.Attributes.RedriveAllowPolicy = tmp
+		case "FifoQueue":
+			tmp, err := strconv.ParseBool(attrValue)
+			if err != nil {
+				log.Debugf("Failed to parse form attribute - %s: %s", attrName, attrValue)
+				continue
+			}
+			r.Attributes.FifoQueue = StringToBool(tmp)
+		case "ContentBasedDeduplication":
+			tmp, err := strconv.ParseBool(attrValue)
+			if err != nil {
+				log.Debugf("Failed to parse form attribute - %s: %s", attrName, attrValue)
+				continue
+			}
+			r.Attributes.ContentBasedDeduplication = StringToBool(tmp)
 		}
 	}
 	return
@@ -393,12 +421,17 @@ type QueueAttributes struct {
 	DelaySeconds                  StringToInt            `json:"DelaySeconds"`
 	MaximumMessageSize            StringToInt            `json:"MaximumMessageSize"`
 	MessageRetentionPeriod        StringToInt            `json:"MessageRetentionPeriod"` // NOTE: not implemented
-	Policy                        map[string]interface{} `json:"Policy"`                 // NOTE: not implemented
+	Policy                        map[string]interface{} `json:"Policy"`                 // хранится, не исполняется
 		ReceiveMessageWaitTimeSeconds StringToInt            `json:"ReceiveMessageWaitTimeSeconds"`
 		VisibilityTimeout             StringToInt            `json:"VisibilityTimeout"`
 		// Dead Letter Queues Only
 		RedrivePolicy      RedrivePolicy          `json:"RedrivePolicy"`
 		RedriveAllowPolicy map[string]interface{} `json:"RedriveAllowPolicy"` // NOTE: not implemented
+	// FIFO-атрибуты. FifoQueue у нас также выводится из суффикса .fifo имени,
+	// но значение от клиента нужно СОХРАНИТЬ и вернуть в GetQueueAttributes —
+	// иначе Terraform (aws_sqs_queue) видел бы постоянный drift.
+	FifoQueue                 StringToBool `json:"FifoQueue"`
+	ContentBasedDeduplication StringToBool `json:"ContentBasedDeduplication"`
 }
 
 // AttrNameWhitelist — имена атрибутов, допустимые в SetQueueAttributes/CreateQueue.
@@ -444,8 +477,29 @@ func (q *QueueAttributes) UnmarshalJSON(data []byte) error {
 			return fmt.Errorf("invalid attribute name: %s", name)
 		}
 	}
+	// Policy/RedriveAllowPolicy в SQS API — СТРОКА с JSON внутри, а поля у нас —
+	// map. JSON-протокол присылает их именно строкой, и unmarshal в map падал бы
+	// (SetQueueAttributes → InvalidParameterValue). Разворачиваем строку в объект.
+	// Также принимаем вариант, где уже пришёл объект — он и так валиден.
+	for _, name := range []string{"Policy", "RedriveAllowPolicy"} {
+		rv, ok := raw[name]
+		if !ok {
+			continue
+		}
+		var s string
+		if err := json.Unmarshal(rv, &s); err == nil {
+			if !json.Valid([]byte(s)) {
+				return fmt.Errorf("invalid JSON in attribute %s", name)
+			}
+			raw[name] = json.RawMessage(s)
+		}
+	}
+	normalized, err := json.Marshal(raw)
+	if err != nil {
+		return err
+	}
 	type plain QueueAttributes
-	return json.Unmarshal(data, (*plain)(q))
+	return json.Unmarshal(normalized, (*plain)(q))
 }
 
 type RedrivePolicy struct {

app/persistence/redis.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/redis/go-redis/v9"
@@ -30,7 +31,9 @@ import (
 
 // Client — глобальный Redis клиент.
 // nil означает режим "только память" — все функции тихо no-op.
-var Client *redis.Client
+// atomic.Pointer: reconnect() пишет новый клиент из health-check горутины,
+// а остальные горутины читают — без атомарности это data race.
+var Client atomic.Pointer[redis.Client]
 
 // Параметры подключения — сохраняются для переподключения после обрыва.
 var (
@@ -43,6 +46,11 @@ var (
 	seqMu          sync.Mutex
 	queueWriteSeq  = map[string]uint64{}
 	tenantWriteSeq = map[string]uint64{}
+	// msgWriteSeq — порядковая нумерация async-записей ОТДЕЛЬНЫХ сообщений
+	// (ключ "{queueKey}|{uuid}"). Без неё два SaveMessage одного сообщения из
+	// разных горутин могли завершиться вразнобой: старое состояние перезаписало
+	// бы новое. Записи удаляются при удалении сообщения (DeleteMessagePersist и др.).
+	msgWriteSeq = map[string]uint64{}
 )
 
 const (
@@ -78,7 +86,7 @@ func Connect(addr, username, password string) error {
 	if err := rdb.Ping(ctx).Err(); err != nil {
 		return fmt.Errorf("redis ping %s: %w", addr, err)
 	}
-	Client = rdb
+	Client.Store(rdb)
 	log.Infof("persistence: подключились к Redis %s", addr)
 	startRedisHealthCheck()
 	return nil
@@ -93,11 +101,11 @@ func startRedisHealthCheck() {
 		ticker := time.NewTicker(30 * time.Second)
 		defer ticker.Stop()
 		for range ticker.C {
-			if Client == nil {
+			if Client.Load() == nil {
 				continue
 			}
 			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-			err := Client.Ping(ctx).Err()
+			err := Client.Load().Ping(ctx).Err()
 			cancel()
 			if err == nil {
 				continue
@@ -128,7 +136,7 @@ func reconnect() error {
 	if err := rdb.Ping(ctx).Err(); err != nil {
 		return err
 	}
-	Client = rdb
+	Client.Store(rdb)
 	return nil
 }
 
@@ -167,6 +175,11 @@ type queueMetaDTO struct {
 	Tags                          map[string]string
 	LastPurgeTime                 time.Time
 	DeadLetterQueueName           string `json:"dead_letter_queue_name,omitempty"`
+	// Создание/изменение + совместимые атрибуты (GetQueueAttributes/Terraform).
+	CreatedAt                 time.Time
+	LastModifiedAt            time.Time
+	Policy                    string
+	ContentBasedDeduplication bool
 }
 
 // queueToMetaDTO — строит снимок очереди без сообщений, НЕ мутируя исходную очередь.
@@ -194,6 +207,10 @@ func queueToMetaDTO(q *models.Queue) *queueMetaDTO {
 		Tags:                          q.Tags,
 		LastPurgeTime:                 q.LastPurgeTime,
 		DeadLetterQueueName:           q.DeadLetterQueueName,
+		CreatedAt:                     q.CreatedAt,
+		LastModifiedAt:                q.LastModifiedAt,
+		Policy:                        q.Policy,
+		ContentBasedDeduplication:     q.ContentBasedDeduplication,
 	}
 	if q.DeadLetterQueue != nil {
 		dto.DeadLetterQueueName = q.DeadLetterQueue.Name
@@ -213,7 +230,7 @@ func marshalQueueMeta(queue *models.Queue) ([]byte, error) {
 // создаёт консистентный снапшот. Горутина только делает сетевой вызов.
 // Для сохранения сообщений используй SaveMessage/SaveMessages.
 func SaveQueue(key string, queue *models.Queue) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return
 	}
 	// Сериализуем метаданные синхронно под мьютексом вызывающего → консистентный снапшот
@@ -229,7 +246,7 @@ func SaveQueue(key string, queue *models.Queue) {
 		}
 		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 		defer cancel()
-		if err := Client.HSet(ctx, redisHashQueues, key, string(data)).Err(); err != nil {
+		if err := Client.Load().HSet(ctx, redisHashQueues, key, string(data)).Err(); err != nil {
 			log.Errorf("persistence: HSet queue meta %q: %v", key, err)
 		}
 	})
@@ -239,7 +256,7 @@ func SaveQueue(key string, queue *models.Queue) {
 // Ключ: ssq:msg:{queueKey}, поле: msg.Uuid, значение: JSON(SqsMessage).
 // Вызывать под Lock — json.Marshal делает консистентный снапшот сообщения.
 func SaveMessage(queueKey string, msg *models.SqsMessage) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return
 	}
 	data, err := json.Marshal(msg)
@@ -248,10 +265,15 @@ func SaveMessage(queueKey string, msg *models.SqsMessage) {
 		return
 	}
 	hashKey := redisMsgHashPrefix + queueKey
+	msgKey := queueKey + "|" + msg.Uuid
+	writeSeq := nextMsgWriteSeq(msgKey)
 	asyncWrite(func() {
+		if !isLatestMsgWriteSeq(msgKey, writeSeq) {
+			return
+		}
 		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 		defer cancel()
-		if err := Client.HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil {
+		if err := Client.Load().HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil {
 			log.Errorf("persistence: HSet message %s/%s: %v", queueKey, msg.Uuid, err)
 		}
 	})
@@ -264,7 +286,7 @@ func SaveMessage(queueKey string, msg *models.SqsMessage) {
 // Ключ: ssq:msg:{queueKey}, поле: msg.Uuid, значение: JSON(SqsMessage).
 // Если Client == nil (memory-only режим) — возвращает nil (персистить некуда).
 func SaveMessageSync(queueKey string, msg *models.SqsMessage) error {
-	if Client == nil {
+	if Client.Load() == nil {
 		return nil
 	}
 	data, err := json.Marshal(msg)
@@ -274,7 +296,7 @@ func SaveMessageSync(queueKey string, msg *models.SqsMessage) error {
 	hashKey := redisMsgHashPrefix + queueKey
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
-	if err := Client.HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil {
+	if err := Client.Load().HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil {
 		return fmt.Errorf("HSet message %s/%s: %w", queueKey, msg.Uuid, err)
 	}
 	return nil
@@ -283,7 +305,7 @@ func SaveMessageSync(queueKey string, msg *models.SqsMessage) error {
 // SaveMessages — сохраняет несколько сообщений в Redis одним pipeline.
 // Вызывать под Lock.
 func SaveMessages(queueKey string, msgs []models.SqsMessage) {
-	if Client == nil || len(msgs) == 0 {
+	if Client.Load() == nil || len(msgs) == 0 {
 		return
 	}
 	// Сериализуем все сообщения синхронно под Lock
@@ -308,7 +330,7 @@ func SaveMessages(queueKey string, msgs []models.SqsMessage) {
 		for k, v := range fields {
 			args = append(args, k, v)
 		}
-		if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil {
+		if err := Client.Load().HSet(ctx, hashKey, args...).Err(); err != nil {
 			log.Errorf("persistence: HSet messages %s (%d msgs): %v", queueKey, len(fields), err)
 		}
 	})
@@ -318,7 +340,7 @@ func SaveMessages(queueKey string, msgs []models.SqsMessage) {
 // Используется на пути SendMessageBatch: ответ 200 выдаётся только после записи
 // всех сообщений (как в одиночном SendMessage). Возвращает ошибку для отката.
 func SaveMessagesSync(queueKey string, msgs []models.SqsMessage) error {
-	if Client == nil || len(msgs) == 0 {
+	if Client.Load() == nil || len(msgs) == 0 {
 		return nil
 	}
 	fields := make(map[string]string, len(msgs))
@@ -339,7 +361,7 @@ func SaveMessagesSync(queueKey string, msgs []models.SqsMessage) error {
 	for k, v := range fields {
 		args = append(args, k, v)
 	}
-	if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil {
+	if err := Client.Load().HSet(ctx, hashKey, args...).Err(); err != nil {
 		return fmt.Errorf("HSet messages %s (%d msgs): %w", queueKey, len(fields), err)
 	}
 	return nil
@@ -349,50 +371,55 @@ func SaveMessagesSync(queueKey string, msgs []models.SqsMessage) error {
 // Синхронность обязательна: если удаление потеряется при рестарте пода,
 // сообщение «воскреснет» из Redis (факт: удалённые очереди возвращались
 // после рестарта, когда удаления были асинхронными).
-func DeleteMessagePersist(queueKey string, msgUuid string) {
+func DeleteMessagePersist(queueKey string, msgUuid string) error {
-	if Client == nil || msgUuid == "" {
+	if Client.Load() == nil || msgUuid == "" {
-		return
+		return nil
 	}
 	hashKey := redisMsgHashPrefix + queueKey
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
-	if err := Client.HDel(ctx, hashKey, msgUuid).Err(); err != nil {
+	if err := Client.Load().HDel(ctx, hashKey, msgUuid).Err(); err != nil {
-		log.Errorf("persistence: HDel message %s/%s: %v", queueKey, msgUuid, err)
+		return fmt.Errorf("HDel message %s/%s: %w", queueKey, msgUuid, err)
 	}
+	clearMsgWriteSeq(queueKey, msgUuid)
+	return nil
 }
 
 // DeleteMessagesPersist — удаляет несколько сообщений из Redis СИНХРОННО
 // (иначе удалённые сообщения воскреснут после рестарта пода).
-func DeleteMessagesPersist(queueKey string, uuids []string) {
+func DeleteMessagesPersist(queueKey string, uuids []string) error {
-	if Client == nil || len(uuids) == 0 {
+	if Client.Load() == nil || len(uuids) == 0 {
-		return
+		return nil
 	}
 	hashKey := redisMsgHashPrefix + queueKey
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	if err := Client.HDel(ctx, hashKey, uuids...).Err(); err != nil {
+	if err := Client.Load().HDel(ctx, hashKey, uuids...).Err(); err != nil {
-		log.Errorf("persistence: HDel messages %s (%d): %v", queueKey, len(uuids), err)
+		return fmt.Errorf("HDel messages %s (%d): %w", queueKey, len(uuids), err)
 	}
+	clearMsgWriteSeq(queueKey, uuids...)
+	return nil
 }
 
 // PurgeMessagesPersist — удаляет ВСЕ сообщения очереди из Redis СИНХРОННО.
 // Синхронность обязательна: при рестарте пода удалённые сообщения не должны
 // воскреснуть (факт: асинхронное удаление очередей приводило к их возврату).
 func PurgeMessagesPersist(queueKey string) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return
 	}
 	hashKey := redisMsgHashPrefix + queueKey
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	if err := Client.Del(ctx, hashKey).Err(); err != nil {
+	if err := Client.Load().Del(ctx, hashKey).Err(); err != nil {
 		log.Errorf("persistence: DEL messages hash %s: %v", queueKey, err)
 	}
+	clearAllMsgWriteSeq(queueKey)
 }
 
 // DeleteQueue — удаляет метаданные очереди И все её сообщения из Redis асинхронно.
 func DeleteQueue(key string) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return
 	}
 	writeSeq := nextQueueWriteSeq(key)
@@ -406,11 +433,11 @@ func DeleteQueue(key string) {
 		return
 	}
 	// Удаляем метаданные из общего HASH
-	if err := Client.HDel(ctx, redisHashQueues, key).Err(); err != nil {
+	if err := Client.Load().HDel(ctx, redisHashQueues, key).Err(); err != nil {
 		log.Errorf("persistence: HDel queue meta %q: %v", key, err)
 	}
 	// Удаляем весь HASH с сообщениями
-	if err := Client.Del(ctx, msgHashKey).Err(); err != nil {
+	if err := Client.Load().Del(ctx, msgHashKey).Err(); err != nil {
 		log.Errorf("persistence: DEL messages hash %q: %v", key, err)
 	}
 }
@@ -420,14 +447,14 @@ func DeleteQueue(key string) {
 // Миграция v1→v2: если в ssq:queues лежит JSON со встроенными Messages,
 // они извлекаются в отдельный HASH и метаданные пересохраняются без Messages.
 func LoadAllQueues() (map[string]*models.Queue, error) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return nil, nil
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
 	// 1. Загружаем метаданные очередей
-	raw, err := Client.HGetAll(ctx, redisHashQueues).Result()
+	raw, err := Client.Load().HGetAll(ctx, redisHashQueues).Result()
 	if err != nil {
 		return nil, fmt.Errorf("redis HGetAll queues: %w", err)
 	}
@@ -452,6 +479,16 @@ func LoadAllQueues() (map[string]*models.Queue, error) {
 			q.FIFOSequenceNumbers = make(map[string]int)
 		}
 
+		// Бэкфилл timestamp-ов: у очередей, созданных до появления полей, они нулевые.
+		// Без этого Terraform видел бы CreatedTimestamp=0 и постоянный drift.
+		// Заполняем сейчас и сохраним при ближайшем SaveQueue.
+		if q.CreatedAt.IsZero() {
+			q.CreatedAt = time.Now().UTC()
+		}
+		if q.LastModifiedAt.IsZero() {
+			q.LastModifiedAt = q.CreatedAt
+		}
+
 		// Миграция v1→v2: если в JSON есть Messages — это старый формат
 		if len(q.Messages) > 0 {
 			migrateKeys = append(migrateKeys, k)
@@ -477,7 +514,7 @@ func LoadAllQueues() (map[string]*models.Queue, error) {
 				args = append(args, q.Messages[i].Uuid, string(data))
 			}
 			if len(args) > 0 {
-				if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil {
+				if err := Client.Load().HSet(ctx, hashKey, args...).Err(); err != nil {
 					log.Errorf("persistence: migrate HSet messages %q: %v", k, err)
 				}
 			}
@@ -485,7 +522,7 @@ func LoadAllQueues() (map[string]*models.Queue, error) {
 		// Пересохраняем метаданные без Messages
 		metaData, err := marshalQueueMeta(q)
 		if err == nil {
-			if err := Client.HSet(ctx, redisHashQueues, k, string(metaData)).Err(); err != nil {
+			if err := Client.Load().HSet(ctx, redisHashQueues, k, string(metaData)).Err(); err != nil {
 				log.Errorf("persistence: migrate HSet meta %q: %v", k, err)
 			}
 		}
@@ -495,10 +532,11 @@ func LoadAllQueues() (map[string]*models.Queue, error) {
 	// 3. Загружаем сообщения из отдельных HASH'ов для каждой очереди
 	for k, q := range queues {
 		hashKey := redisMsgHashPrefix + k
-		msgRaw, err := Client.HGetAll(ctx, hashKey).Result()
+		msgRaw, err := Client.Load().HGetAll(ctx, hashKey).Result()
 		if err != nil {
-			log.Errorf("persistence: HGetAll messages %q: %v", k, err)
+			// Fail-fast: скрытая пустая очередь при недоступных сообщениях
-			continue
+			// хуже явной ошибки — сообщения остались бы невидимыми в Redis.
+			return nil, fmt.Errorf("redis HGetAll messages %q: %w", k, err)
 		}
 		if len(msgRaw) > 0 {
 			msgs := make([]models.SqsMessage, 0, len(msgRaw))
@@ -546,7 +584,7 @@ func LoadAllQueues() (map[string]*models.Queue, error) {
 // Принимает []byte чтобы избежать циклического импорта с пакетом tenant.
 // Сериализацию делает вызывающий (tenant_store.go).
 func SaveTenantRaw(id string, jsonData []byte) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return
 	}
 	// Копируем bytes — вызывающий может переиспользовать буфер
@@ -564,11 +602,11 @@ func SaveTenantRaw(id string, jsonData []byte) {
 		}
 		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 		defer cancel()
-		if err := Client.HSet(ctx, redisHashTenants, id, string(dataCopy)).Err(); err != nil {
+		if err := Client.Load().HSet(ctx, redisHashTenants, id, string(dataCopy)).Err(); err != nil {
 			log.Errorf("persistence: HSet tenant %q: %v", id, err)
 		}
 		if meta.AccessKey != "" {
-			if err := Client.HSet(ctx, redisHashTenantAK, meta.AccessKey, string(dataCopy)).Err(); err != nil {
+			if err := Client.Load().HSet(ctx, redisHashTenantAK, meta.AccessKey, string(dataCopy)).Err(); err != nil {
 				log.Errorf("persistence: HSet tenant_ak %q: %v", meta.AccessKey, err)
 			}
 		}
@@ -578,12 +616,12 @@ func SaveTenantRaw(id string, jsonData []byte) {
 // LoadTenantByAccessKey — read-through тенанта по accessKey (для auth-middleware).
 // Возвращает сырой JSON; десериализацию делает вызывающий.
 func LoadTenantByAccessKey(accessKey string) ([]byte, bool) {
-	if Client == nil || accessKey == "" {
+	if Client.Load() == nil || accessKey == "" {
 		return nil, false
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
-	v, err := Client.HGet(ctx, redisHashTenantAK, accessKey).Result()
+	v, err := Client.Load().HGet(ctx, redisHashTenantAK, accessKey).Result()
 	if err != nil {
 		return nil, false
 	}
@@ -591,8 +629,10 @@ func LoadTenantByAccessKey(accessKey string) ([]byte, bool) {
 }
 
 // DeleteTenant — удаляет тенанта из Redis асинхронно.
-func DeleteTenant(id string) {
+// Удаляет из ОБОИХ индексов: ssq:tenants (по id) и ssq:tenant_ak (по accessKey),
-	if Client == nil {
+// иначе LoadTenantByAccessKey вернёт удалённого тенанта и отзыв ключей не сработает.
+func DeleteTenant(id, accessKey string) {
+	if Client.Load() == nil {
 		return
 	}
 	writeSeq := nextTenantWriteSeq(id)
@@ -602,9 +642,14 @@ func DeleteTenant(id string) {
 		}
 		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 		defer cancel()
-		if err := Client.HDel(ctx, redisHashTenants, id).Err(); err != nil {
+		if err := Client.Load().HDel(ctx, redisHashTenants, id).Err(); err != nil {
 			log.Errorf("persistence: HDel tenant %q: %v", id, err)
 		}
+		if accessKey != "" {
+			if err := Client.Load().HDel(ctx, redisHashTenantAK, accessKey).Err(); err != nil {
+				log.Errorf("persistence: HDel tenant_ak %q: %v", accessKey, err)
+			}
+		}
 	})
 }
 
@@ -621,6 +666,41 @@ func isLatestQueueWriteSeq(key string, seq uint64) bool {
 	return queueWriteSeq[key] == seq
 }
 
+func nextMsgWriteSeq(key string) uint64 {
+	seqMu.Lock()
+	defer seqMu.Unlock()
+	msgWriteSeq[key]++
+	return msgWriteSeq[key]
+}
+
+func isLatestMsgWriteSeq(key string, seq uint64) bool {
+	seqMu.Lock()
+	defer seqMu.Unlock()
+	return msgWriteSeq[key] == seq
+}
+
+// clearMsgWriteSeq — удаляет записи нумерации для удалённых сообщений
+// (иначе msgWriteSeq рос бы бесконечно).
+func clearMsgWriteSeq(queueKey string, uuids ...string) {
+	seqMu.Lock()
+	defer seqMu.Unlock()
+	for _, u := range uuids {
+		delete(msgWriteSeq, queueKey+"|"+u)
+	}
+}
+
+// clearAllMsgWriteSeq — удаляет ВСЕ записи нумерации для очереди (purge).
+func clearAllMsgWriteSeq(queueKey string) {
+	prefix := queueKey + "|"
+	seqMu.Lock()
+	defer seqMu.Unlock()
+	for k := range msgWriteSeq {
+		if strings.HasPrefix(k, prefix) {
+			delete(msgWriteSeq, k)
+		}
+	}
+}
+
 func nextTenantWriteSeq(id string) uint64 {
 	seqMu.Lock()
 	defer seqMu.Unlock()
@@ -637,12 +717,12 @@ func isLatestTenantWriteSeq(id string, seq uint64) bool {
 // LoadAllTenantsRaw — загружает всех тенантов из Redis при старте.
 // Возвращает map[tenantID]rawJSON — десериализацию делает tenant_store.go.
 func LoadAllTenantsRaw() (map[string][]byte, error) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return nil, nil
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	raw, err := Client.HGetAll(ctx, redisHashTenants).Result()
+	raw, err := Client.Load().HGetAll(ctx, redisHashTenants).Result()
 	if err != nil {
 		return nil, fmt.Errorf("redis HGetAll tenants: %w", err)
 	}
@@ -664,12 +744,12 @@ func LoadAllTenantsRaw() (map[string][]byte, error) {
 
 // TryAcquireLock — пытается взять лок (SETNX с TTL). true — лок наш.
 func TryAcquireLock(key, value string, ttl time.Duration) (bool, error) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return true, nil
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
-	return Client.SetNX(ctx, key, value, ttl).Result()
+	return Client.Load().SetNX(ctx, key, value, ttl).Result()
 }
 
 // refreshLockScript — продлить TTL, ТОЛЬКО если лок всё ещё наш (атомарно).
@@ -690,12 +770,12 @@ end`)
 
 // RefreshLock — продлевает TTL лока, если он всё ещё наш. false — лок потерян.
 func RefreshLock(key, value string, ttl time.Duration) (bool, error) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return true, nil
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
-	n, err := refreshLockScript.Run(ctx, Client, []string{key}, value, ttl.Milliseconds()).Int()
+	n, err := refreshLockScript.Run(ctx, Client.Load(), []string{key}, value, ttl.Milliseconds()).Int()
 	if err != nil {
 		return false, err
 	}
@@ -704,12 +784,12 @@ func RefreshLock(key, value string, ttl time.Duration) (bool, error) {
 
 // ReleaseLock — освобождает лок (если он наш). Вызывается при graceful shutdown.
 func ReleaseLock(key, value string) {
-	if Client == nil {
+	if Client.Load() == nil {
 		return
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
-	if _, err := releaseLockScript.Run(ctx, Client, []string{key}, value).Int(); err != nil {
+	if _, err := releaseLockScript.Run(ctx, Client.Load(), []string{key}, value).Int(); err != nil {
 		log.Errorf("persistence: release lock %q: %v", key, err)
 	}
 }

app/tenant/tenant_store.go

@@ -7,6 +7,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -121,6 +122,11 @@ func (s *TenantStore) CreateFixedAutoID(name string, maxQueues int, accessKey, s
 }
 
 func (s *TenantStore) CreateFixed(name string, maxQueues int, tenantID, accessKey, secretKey string) (*Tenant, error) {
+	// access_key используется как префикс ключа "{accessKey}:{queueName}" —
+	// ':' в нём ломает изоляцию тенантов (ListQueues/коллизии ключей).
+	if strings.Contains(accessKey, ":") {
+		return nil, fmt.Errorf("access key must not contain ':'")
+	}
 	s.mu.Lock()
 	if _, exists := s.byID[tenantID]; exists {
 		s.mu.Unlock()
@@ -183,8 +189,8 @@ func (s *TenantStore) Delete(id string) bool {
 	}
 	s.mu.Unlock()
 
-	// Удаляем из Redis асинхронно
+	// Удаляем из Redis асинхронно (оба индекса: по id и по accessKey)
-	persistence.DeleteTenant(id)
+	persistence.DeleteTenant(id, t.AccessKey)
 	return true
 }
 

app/utils/utils.go

@@ -2,6 +2,7 @@ package utils
 
 import (
 	"crypto/md5"
+	"crypto/sha256"
 	"encoding/base64"
 	"encoding/binary"
 	"encoding/hex"
@@ -82,7 +83,11 @@ func ExtractQueueAttributes(u url.Values) map[string]string {
 // CreateErrorResponseV1 — формирует SQS error response по ключу ошибки.
 // Параметр isSqs оставлен для обратной совместимости (всегда true).
 func CreateErrorResponseV1(errKey string, isSqs bool) (int, interfaces.AbstractResponseBody) {
-	err := models.SqsErrors[errKey]
+	err, ok := models.SqsErrors[errKey]
+	if !ok {
+		// Неизвестный ключ → InternalError вместо nil-deref (ревью 2026-10-11).
+		err = models.SqsErrors["InternalError"]
+	}
 
 	respStruct := models.ErrorResponse{
 		Result:    err.Response(),
@@ -97,6 +102,13 @@ func GetMD5Hash(text string) string {
 	return hex.EncodeToString(hasher.Sum(nil))
 }
 
+// GetSHA256Hash — SHA-256 (hex) от значения. Используется для
+// ContentBasedDeduplication: AWS сам вычисляет MessageDeduplicationId из тела.
+func GetSHA256Hash(text string) string {
+	sum := sha256.Sum256([]byte(text))
+	return hex.EncodeToString(sum[:])
+}
+
 func HashAttributes(attributes map[string]models.MessageAttribute) string {
 	hasher := md5.New()
 

doc/thinking/opus-report-2026-10-10.md

@@ -0,0 +1,231 @@
+# Код-ревью shared-sqs (Opus, 2026-10-10)
+
+> Ответ Opus на промпт «сделай код-ревью и посоветуй по архитектуре».
+> Дословно, без правок.
+
+Все файлы из списка прочитаны. Строки взяты из листингов, часть пунктов опирается на прочитанное ранее. Пункты, которые стоит перепроверить перед правкой, помечены «проверить». Теги [Ф1–Ф6] указывают на ваши фокусные области.
+
+## 🔴 Критично: потеря данных, падение процесса, обход безопасности
+
+1. **[Ф1] Fatal `concurrent map iteration and map write` при повторном CreateQueue.** `create_queue.go:84` вызывает `SaveQueue` для существующей очереди под глобальным локом, но без `queue.mu`. `SaveQueue` маршалит `Duplicates`, `FIFO*` и `Tags`, а `send_message` пишет в них под `queue.mu`. Это не panic, а fatal runtime, который `recover` не ловит и который убивает весь процесс. Идемпотентный CreateQueue — частый вызов SDK. Тот же класс ошибки: `admin.go:498-535` и `seed.go`.
+2. **[Ф1] `list_queue_tags.go:49-60` отдаёт в ответ ссылку на `queue.Tags`.** `RUnlock` срабатывает по `defer` при return, а XML-маршалинг идёт позже, в router. Параллельный `TagQueue` даёт тот же fatal. Нужна копия мапы.
+3. **[Ф1] Инверсия порядка локов и дедлок через ожидающего писателя.**
+   - `queue.mu` → `SyncQueues.RLock`: `set_queue_attributes.go:48,63` → `queue_attributes.go:78`.
+   - `SyncQueues.RLock` → `queue.RLock`: скан в `admin.go:435-486` (подтверждено), `gauge_updater` (проверить, держит ли глобальный RLock).
+   - Любой писатель в `SyncQueues` (`create_queue.go:73`, `delete_queue.go:45`, admin) блокирует новых читателей Go RWMutex. Получается цикл, и сервис встаёт целиком.
+   - Отдельно: `models.go:410` — ABBA и self-дедлок на DLQ (цикл A→B→A или DLQ=self не отсекается в `queue_attributes.go:85`).
+   - `conf/config.go:126-134` — `return` без `Unlock` и дедлок на старте.
+4. **[Ф1, Ф3] Redis-I/O под глобальным локом.** `delete_queue.go:45-56` выполняет синхронный `DeleteQueue` под `SyncQueues.Lock`. Комментарии «асинхронно» (`:24,55`) устарели. Redis-пауза 3–5 с останавливает все тенанты. То же в `admin.go:498` и `gosqs.go:107`.
+5. **[Ф3] Подтверждённое сообщение теряется при сбое Redis.**
+   - `send_message.go:157` и `send_message_batch.go:133-139,153-158`: при ошибке `SaveMessage(s)Sync` откатывается только `RemoveIndex`.
+   - `Duplicates` и `FIFOSequenceNumbers` не откатываются, `InitDuplicatation` уже сработал.
+   - Ретрай клиента с тем же dedupId получает 200 без сохранения.
+   - Таймаут Redis неоднозначен: запись могла пройти, и после рестарта появится фантом.
+6. **[Ф3] Receive, visibility-return и DLQ-перенос не персистятся.**
+   - Затронуты `gosqs.go:124-154`, `change_message_visibility*.go`, `receive_message.go`.
+   - DLQ-копия не пишется (`models.go:409-413`), а `change_message_visibility.go:72-81` делает только Delete в источнике.
+   - После рестарта сообщение воскресает в источнике и пропадает из DLQ.
+   - Порядок при загрузке теряется: `redis.go:504-513` итерирует HGETALL без сортировки, FIFO ломается.
+7. **[Ф1, Ф2] Зависание FIFO-групп навсегда.**
+   - `ResetVisibilityAndTryDLQ` (`models.go:389-417`) не делает `UnlockGroup`, путь из `change_message_visibility.go:64` и batch-версии.
+   - `purge_queue.go:63-64` сбрасывает `Duplicates`, но не `FIFOMessages`.
+   - После рестарта `FIFOMessages` из Redis (`redis.go:163,448`) не согласован с in-flight, а тик снимает лок только по `receiptIdx`.
+8. **DeleteMessageBatch с пустым `ReceiptHandle` удаляет все неполученные сообщения.**
+   - `delete_message_batch.go:79-97` сопоставляет handle сканом. У невыданных сообщений handle `""` и совпадает с записью `""`.
+   - Query-парсер (`requests.go:568`) отсекает пустой handle, JSON-протокол — нет.
+   - Одиночный delete безопасен: индекс пропускает `""`.
+9. **[Ф4] HA — это не HA.**
+   - `leader.go:106-126`: ошибка Refresh не демотирует лидера, lease-проверки нет.
+   - `goaws.go:171-191`: пассив выполняет `PeriodicTasks` и пишет в Redis. Handlers лидерство не проверяют, защита — только `/ready`.
+   - При повышении пассива состояние из Redis не перечитывается: он работает со снимком своего старта.
+   - Без Redis `TryAcquireLock` и `RefreshLock` возвращают true (`redis.go:666-715`), каждый инстанс становится «лидером».
+   - Нет fencing-токена.
+   - `redis.go:61-85`: при неудачном Connect клиент остаётся nil навсегда, reconnect не запускается.
+10. **[Ф3, Ф4] Старт с пустым состоянием.** `goaws.go:103-106,125-134` при недоступном Redis или ошибке `LoadAllQueues` продолжает работу. Следом `SaveQueue` перезапишет Redis пустотой. Нужен fail-fast.
+11. **[Ф6] Тело подменяется при валидной подписи.** `auth_middleware.go:166-169,234-243` берёт `X-Amz-Content-Sha256` из заголовка и не сверяет с реальным хешем тела.
+12. **[Ф6] JWT без проверки подписи.**
+    - `jwt.go`: идентичность берётся из непроверенного payload, `Exp==0` пропускается.
+    - Валидность подтверждает GET на любой из трёх шлюзов, включая dev и test.
+    - `tenant_store.go:213-291`: тенант находится по email из такого токена, возможен захват чужого тенанта.
+    - `cache.go:68-75`: при ошибке сети grace продлевает отозванный токен.
+13. **[Ф3] Удалённый тенант воскресает.** `redis.go:594-609` `DeleteTenant` не чистит `ssq:tenant_ak`, а read-through в auth (`:63-74`) его возвращает. Найденный тенант кэшируется навсегда.
+14. **[Ф5] OOM-потолок на уровне процесса.**
+    - Лимит только по числу сообщений: 120k × 256 КБ = ~30 ГБ на одну очередь (`validation.go:43-44`).
+    - Лимитов по байтам или памяти нет ни на очередь, ни на тенанта, ни глобально.
+    - Размер атрибутов не входит в проверку (`MaxMessageAttributeSize` объявлен и не используется).
+    - Один тенант валит процесс всем.
+15. **[Ф3] `redis.go:33,116-133`: `Client` глобален, `reconnect()` подменяет его без синхронизации.** Это data race, старый клиент не закрывается. go-redis и так переподключается сам, `reconnect()` не нужен.
+
+## 🟡 Существенно
+
+**Гонки и конкурентность**
+- **[Ф1]** `globals.go:77-100`: `GetLocked` отпускает глобальный лок до `LockMu`, и после взятия лока нет проверки, что очередь ещё в map.
+  - Запись в удалённую очередь возвращает 200, а сообщение пропало.
+  - `SaveMessagesSync` пишет `ssq:msg:{key}` уже после `DeleteQueue`, это осиротевшие ключи в Redis.
+  - Аналог: `send_message_batch.go:46-53` → `:105`.
+- **[Ф3]** `redis.go:215-236`: seq-guard проверяется до HSET (TOCTOU). `:394-416`: проверка seq сразу после инкремента — тавтология, запущенная async-`SaveQueue` воскрешает метаданные после `HDel`. Карты seq растут вечно (`:44-46`).
+- **[Ф3]** `purge_queue.go:68`: синхронный Redis под `queue.mu`, ошибка не возвращается, память очищена, а Redis нет. Ранее запущенные async `SaveMessage` могут записать сообщение после DEL.
+
+**Tombstone и индексы**
+- **[Ф2]** `models.go:201-208`: `ensureIndexes` тавтологичен. Проверка `liveCount == len-deadCount` не ловит рассинхрон map-индексов.
+- **[Ф2]** `models.go:262-283`: `RemoveIndex` не обнуляет слот, тело сообщения живёт до компакции. Порог 1024 слота или половина — до 256 МБ мусора на очередь.
+- **[Ф2]** `RebuildIndexes` на старте разреженный, после загрузки `deadCount=0`: компакции нет, но и мусора нет. Это нормально, если порядок восстановлен сортировкой (см. 🔴6).
+- `models.go:113-141`: `FIFOSequenceNumbers` и `FIFOMessages` растут без чистки.
+- `models.go:172-180`: запись в nil-map при `Duplicates==nil`.
+
+**Семантика SQS**
+- Dedup:
+  - `delete_message.go:59` и `delete_message_batch.go:93` делают `delete(Duplicates)` при удалении, окно 5 мин нарушено.
+  - `requests.go:422-423`: `ContentBasedDeduplication` принимается и игнорируется, FIFO без dedupId не дедуплицируется.
+  - Дубль возвращает новый MessageId.
+- Receive и счётчики:
+  - `Retry++` на таймауте видимости (`gosqs.go:142`): `MaxReceiveCount` считается по таймаутам.
+  - `receive_message.go:203,247`: `ApproximateReceiveCount` завышен на 1.
+  - `ReceiptTime` перезаписывается, `FirstReceiveTimestamp` неверен.
+- `models.go:405`: `ChangeMessageVisibility(0)` скрывает сообщение ещё на `q.VisibilityTimeout`, а должно быть видно сразу (проверить по receive).
+- `change_message_visibility_batch.go:72`: нет проверки отрицательных значений.
+- `queue_attributes.go:41-70`: 0 означает «не передан», и поэтому нельзя вернуть `DelaySeconds`, `ReceiveMessageWaitTimeSeconds` или `VisibilityTimeout` в 0.
+- `requests.go:35-37,30`: невалидные значения и пустая строка молча игнорируются вместо ошибки.
+- `queue_attributes.go:73-87`:
+  - нет проверки self-ссылки, цикла, совпадения FIFO-типа и диапазона `maxReceiveCount`;
+  - `DeadLetterQueueName` здесь не выставляется (проверить, заполняет ли его `SaveQueue`, иначе DLQ теряется после рестарта);
+  - `delete_queue.go:53` не отвязывает DLQ-ссылки других очередей.
+- `create_queue.go:73-85`: повторный CreateQueue с другими атрибутами возвращает 200 вместо `QueueNameExists`.
+- `create_queue.go:66`, `set_queue_attributes.go:59`: точное сравнение Content-Type. Для `json-1.1` или с `charset` whitelist обходится.
+- Batch:
+  - `send_message_batch.go:87-103` валидирует пакет целиком, вместо per-entry `Failed`.
+  - Нет суммарного лимита 256 КБ.
+  - Per-message `DelaySeconds` не применяется (`:118-127`, проверить паритет с single).
+  - `delete_message_batch.go:65-72`: дубль handle теряет Id из ответа.
+  - `delete_message_batch.go:79` — O(N) скан вместо `IndexOfReceipt`, при том что `change_message_visibility_batch.go:84` индекс использует.
+  - Код ошибки `"1"`, порядок ответа недетерминирован.
+  - Query-парсеры (`requests.go:235,568,602`) молча обрезают пакет на первой пустой записи.
+- `get_queue_attributes.go:89-100`:
+  - `Approximate…` включает in-flight и delayed;
+  - timestamp-заглушки `"0000000000"`;
+  - нет `FifoQueue`, `ContentBasedDeduplication`, `Policy`, `…Delayed`, что ломает Terraform и мониторинг.
+- `list_queues.go:34-43`: скан всех очередей сервиса под глобальным RLock при готовом tenant-индексе. Пагинации нет (`MaxResults` и `NextToken` игнорируются), порядок случайный.
+- `tag_queue.go:56-60`: нет лимитов на теги (AWS: 50 шт., ключ 128, значение 256). Теги сериализуются на каждый `SaveQueue`, это DoS-вектор.
+
+**Производительность [Ф5]**
+- `receive_message.go:164-216`: два полных скана под write-локом. Long-poll опрашивает каждые 100 мс с эксклюзивным локом на каждого поллера.
+- `SaveQueue` на каждый send, delete и CMV (`send_message.go:159`, `delete_message*.go`, `change_message_visibility_batch.go:117`): O(Duplicates + FIFO + Tags) под локом.
+- `gosqs.go:50-54`: один горутин-тик на все очереди последовательно, с синхронным Redis (`:107`).
+- `redis.go:422-543`:
+  - `LoadAllQueues` грузит всё последовательно, общий таймаут 30 с;
+  - ошибка очереди → `continue`, очередь грузится пустой;
+  - время старта и RAM пропорциональны полному состоянию.
+
+**HA и жизненный цикл [Ф4]**
+- `router.go:214-228`: `/ready` не учитывает Redis, загрузку состояния и shutdown. `ReadyFunc` — глобальная переменная.
+- `goaws.go:248-268`:
+  - `billing.Close()` идёт до дренажа HTTP;
+  - async Redis-записи не дренируются;
+  - не-ready на время дренажа нет;
+  - `log.Fatalf` в `:245` не отпускает лок;
+  - `leader.go:63-76`: `Stop` освобождает лок до дренажа.
+- `goaws.go:204-211`: нет `ReadTimeout` (медленное тело), лимитов соединений и rate-limit. Тело читается целиком до проверки подписи.
+
+**Безопасность [Ф6]**
+- SigV4:
+  - `auth_middleware.go:296-326` не требует подписи `host`, `x-amz-date`, `x-amz-target`;
+  - `:158` clock skew 15 мин (в AWS — 5);
+  - `:215` presigned не проверяет дату из будущего;
+  - `:290` `canonicalURI` зависит от нормализации nginx и заголовка Host;
+  - превышение лимита тела даёт 403 `SignatureDoesNotMatch`, а не 413 (`router.go:124-133`);
+  - rate-limit по неудачным попыткам нет, ошибки раскрывают причину (`:81`).
+- `auth_middleware.go:63-74`: каждый запрос с неизвестным accessKey ходит в Redis (HGET до 3 с) без негативного кэша. `jwt.go:90-133` — до трёх последовательных запросов по 10 с на токен, без singleflight.
+- `admin.go`:
+  - статичный Bearer без rate-limit;
+  - `:223` `jwtMiddleware` пропускает путь по `HasSuffix("/auth")`;
+  - `:263-267` IDOR-защита только если в `mux.Vars` есть id;
+  - `:409-420` `deleteTenant`: in-flight handlers пишут в удалённые очереди;
+  - `:194,207-212` токен и `SecretKey` отдаются в теле ответа;
+  - `:640-669` `sendMessageToQueue` без лимита `MaxMessagesForQueue` и с async-сохранением.
+- `tenant_store.go`: секреты в Redis открытым текстом. Нет проверки уникальности email (`:58-108`), `MaxQueues` перезаписывается при каждом логине (`:240`), `LoadTenant` затирает свежие правки (`:194-202`).
+- `router.go:72-74`: маршруты `/{account}` и `/{account}/{queueName}` перекрывают `/admin/*` и `/health` для POST. Bypass в auth по этим путям fail-open, но nil-проверка tenant присутствует во всех 17 просмотренных handlers.
+- `goaws.go:148-150`: при `SEED_DEMO` креды демо-тенанта пишутся в лог.
+
+**Прочее**
+- `router.go:250-277`: `countBatchEntries` работает только с form, для JSON-протокола возвращает 0 и биллинг батчей занижен. `:297` `msgBytes` — размер запроса, а не ответа.
+- `billing.go`:
+  - канал 1024, переполнение дропает записи;
+  - `db.Exec` без таймаута;
+  - `Close()` обнуляет `db` без синхронизации с воркером (data race);
+  - DSN собирается через Sprintf.
+- `metrics`:
+  - label tenant: accessKey в gauge и `tenant.ID` в counters, accessKey утекает в метрики;
+  - `Reset()` gauge даёт окно пустых значений;
+  - label не чистится при удалении тенанта.
+- `conf/config.go`:
+  - `:111` yaml-очереди кладутся без префикса `accessKey:` и обходят tenant-изоляцию;
+  - `:189-196` при пустой RedrivePolicy `queues[deadLetterQueueName]` ищется без tenant-префикса.
+- `seed.go`: ARN собирается с `t.ID` вместо `AccountID`, очереди без FIFO/DLQ-полей, `SaveMessages` асинхронный.
+
+## 🟢 Мелочи
+
+- Расхождение дефолтов:
+  - `build.go:19-32` (admin) задаёт `VisibilityTimeout`, `MaxSize`, `Retention` напрямую;
+  - `CreateQueueV1` берёт их из `NewCreateQueueRequest` по yaml;
+  - при нулях в yaml пути разойдутся.
+- Мёртвый код и дубли:
+  - `validation.go`: `ClampInt`, `ValidateMessageAttributes`, `MaxRequestBodySize=1MB` (router — 3 МБ), `MaxTenantsDefault` дублирует `MaxTenantsGlobal`;
+  - `AvailableQueueAttributes` и `AttrNameWhitelist` — два списка;
+  - `DeduplicationPeriod` — `var`;
+  - `send_message.go:73` — мёртвая ветка `mux.Vars`.
+- `BaseResponseMetadata.RequestId` — нулевой UUID во всех ответах: корреляция невозможна.
+- `models/helpers.go`: `rand.Intn(max-min)` паникует при `max<min`; `Sprintf` и `ParseDuration` на каждый вызов; `IsReadyForReceipt` пересчитывает latency на каждый вызов.
+- Устаревшие комментарии: `redis.go:1-13`, `send_message.go:38`, `goaws.yaml` (retention «not implemented»).
+- INFO-логи на каждый вызов: `delete_message.go:43` (с ReceiptHandle), `get_queue_attributes.go:64`, `list_queues.go:30`.
+- `models/errors.go`:
+  - `delete_message.go:48` возвращает `MessageDoesNotExist` при отсутствии очереди;
+  - `InvalidAttributeValue` и `MissingParameter` имеют жёсткий текст сообщения;
+  - `QueueExists` не используется.
+- `tenant_helpers.go` и `configuration.go` не прогнаны через gofmt. Имя очереди берётся из последнего сегмента URL, `tenantID` в пути игнорируется.
+- Версия `xml.Marshal` с `http.Error` после `WriteHeader` (`router.go:175-184`).
+
+## Архитектура: потолок и вынос в Redis (вариант A)
+
+**Потолок сегодня.** Он определяется тремя вещами:
+- RAM процесса при отсутствии лимита по байтам;
+- время старта: полная загрузка последовательным HGETALL;
+- последовательный тик и O(N)-сканы в receive и gauge.
+
+Ориентировочно потолок — десятки тысяч сообщений на горячую очередь. Измерить p99 receive при 100k in-flight и время `LoadAllQueues` на проде.
+
+**Что Redis не гарантирует.** «Синхронно» в коде означает «принято Redis в память», а не fsync и не репликация. Redis на той же ВМ — общий SPOF. Для durability нужны `appendfsync always` или `everysec` с осознанной потерей до 1 с, либо `WAIT`, и отдельный реплицируемый Redis.
+
+**Порядок перехода (по риску):**
+1. **Сначала дёшево, без смены архитектуры:**
+   - сделать `SaveQueue` синхронным или write-behind с batch;
+   - вынести FIFO-lock из сообщений в метаданные и персистить;
+   - сортировать при загрузке (по seq или `SentTime`);
+   - персистить receive, visibility и DLQ;
+   - fail-fast при старте без Redis;
+   - лимиты по байтам и по тегам.
+2. **Починить конкурентность:**
+   - единый порядок локов (глобальный до per-queue, не наоборот);
+   - убрать I/O под глобальным локом;
+   - проверка живости очереди после `LockMu`;
+   - копии при отдаче map в ответ.
+3. **Интерфейс `QueueStore`** поверх `SyncQueues`. Сейчас handlers лезут в `Messages`, индексы и `Duplicates` напрямую, поэтому без интерфейса вынос невозможен.
+4. **Redis как источник истины:**
+   - Lua-скрипты (или Streams) на send, receive, delete, change-visibility, DLQ-перенос;
+   - visibility через ZSET по времени;
+   - dedup через ключ с TTL;
+   - FIFO-групп-лок как ключ с lease.
+5. **Sweeper вместо `PeriodicTasks`**, под отдельным локом и с fencing-токеном.
+6. **Снять leader-lock.** До этого этапа HA остаётся «один активный, пассив безопасен только за `/ready`». Для безопасного active/passive нужны fencing-токен, перезагрузка состояния при повышении, запрет записи без лидерства и lease-проверка на лидере.
+
+**Главный риск варианта A** — сетевой round-trip на каждую операцию и сложность FIFO и DLQ в Lua. Дешёвая промежуточная ступень — шардирование по тенантам (вариант B) с оставленным in-memory ядром. Оно не даёт HA, но снимает потолок по RAM и тику.
+
+## Порядок исправления
+
+1. Fatal-гонки: п. 1–2 из 🔴.
+2. Lock order и I/O под глобальным локом: п. 3–4.
+3. Откат dedup при сбое Redis: п. 5.
+4. Пустой handle в DeleteMessageBatch: п. 8.
+5. Подпись, тело и JWT: п. 11–12.
+6. Fail-fast на старте и lease у лидера: п. 9–10.
+7. Лимит по байтам: п. 14.
+8. Персист receive, visibility и DLQ вместе с порядком при загрузке: п. 6–7.

doc/thinking/opus-report-analysis-2026-10-10.md

@@ -0,0 +1,90 @@
+# Критический разбор код-ревью Opus (2026-10-10)
+
+> Дополняет `opus-report-2026-10-10.md` (дословный отчёт Opus).
+> Здесь — проверка ключевых пунктов по коду + что подтверждено/отклонено + решения, которые нужны.
+
+## 1. Вердикт
+
+Отчёт сильный по конкуренции и durability. Главный дефект отчёта — смешивает три категории
+без разделения: **(а)** реальные баги, **(б)** осознанный дизайн (JWT trusted-perimeter),
+**(в)** отклонения от семантики AWS (в основном косметика). Часть 🔴 надо понизить.
+
+## 2. Подтверждено по коду (критично, чинится без владельца)
+
+### 2.1. 🔴1 — фатал `concurrent map iteration and map write` в CreateQueue
+- `create_queue.go:73-84`: ветка «очередь уже существует» вызывает `SaveQueue(key, queue)`
+  под `SyncQueues.Lock`, **без** `queue.mu`.
+- `SaveQueue` → `json.Marshal` (`redis.go:208,220`) итерирует `Duplicates`, `FIFOSequenceNumbers`,
+  `FIFOMessages`, `Tags`.
+- Мутаторы этих map (send/tag/CMV) держат только `queue.mu` (`GetLocked` берёт глобальный RLock
+  на миг и отпускает — `globals.go:77-88`), поэтому глобальный Lock их не блокирует.
+- Условие: идемпотентный `CreateQueue` существующей FIFO-очереди (обычный вызов SDK при старте)
+  + параллельный `SendMessage`. Вероятность растёт с размером map. Необратимый краш процесса
+  (`recover` не ловит fatal).
+- Шапка `redis.go:211-213` («вызывать под `SyncQueues.Lock`») описывает неверный контракт.
+
+### 2.2. 🔴5 — неполный откат dedup при сбое Redis
+- `send_message.go:151-160`: при ошибке `SaveMessageSync` откатывается только `RemoveIndex`,
+  а `InitDuplicatation` выполняется всё равно → ретрай с тем же dedupId даёт 200, но сообщение
+  не сохранено. Аналог в `send_message_batch.go`.
+- `FIFOSequenceNumbers` (инкремент в `models.go:117`) тоже не откатывается.
+
+### 2.3. 🔴8 — пустой `ReceiptHandle` в DeleteMessageBatch (protocol-gated)
+- Срабатывает **только** при `Content-Type: application/x-amz-json-1.0`: тело декодируется
+  напрямую (`utils.go:35-46`), валидация `receiptHandle == ""` (`requests.go:568`, Query-форма)
+  на этом пути не вызывается.
+- `delete_message_batch.go:51-57` проверяет только уникальность `Id`, handle не валидируется.
+- `deleteMessageMap[""]` совпадает с невыданными сообщениями (у них `ReceiptHandle == ""`),
+  цикл `:79-98` удаляет все такие сообщения очереди.
+- Одиночный `DeleteMessage` безопасен: пустой handle не кладётся в `receiptIdx`
+  (`models.go:233,251,373`).
+
+### 2.4. Дополнительно подтверждено (не тестировалось, статический разбор)
+- Инверсия порядка локов (`queue.mu` ↔ `SyncQueues.RLock`) и Redis-I/O под глобальным локом
+  (`delete_queue.go:45-56`).
+- Fail-fast отсутствует: старт с недоступным Redis продолжает работу и следом `SaveQueue`
+  перезаписывает Redis пустотой (`goaws.go:103-106,125-134`).
+
+## 3. Решение владельца: JWT
+
+`jwt.go:7-11` прямо документирует trusted-perimeter: «Подпись JWT НЕ проверяется — аналогично
+sless», валидация через `PingNubesAPI` (GET на 3 шлюза, приём по 2xx).
+
+**Ключевой факт, который не проверен:** проверяет ли шлюз подпись и отдаёт 200 только для
+валидных токенов? От этого зависит классификация:
+- шлюз проверяет подпись → «доверяем шлюзу», не уязвимость;
+- нет → 🔴, чинить первым делом.
+
+Независимо от ответа чинятся три вещи:
+1. `jwt.go:90-92` — три шлюза (prod/test/dev): токен тестового стенда принимается как prod.
+   Оставить prod-шлюз.
+2. `cache.go:68-75` — grace при сетевой ошибке продлевается повторно → отозванный токен живёт
+   бесконечно. Нужен потолок суммарного grace (например, 2 мин от последней успешной проверки).
+3. `tenant_store.go:213-291` — идентичность по email; привязать к стабильному `sub`/`iss`,
+   иначе смена email на IdP отдаёт чужой тенант.
+
+## 4. Решение владельца: лимит по байтам (OOM, 🔴14)
+
+Сейчас лимит только по числу сообщений (`validation.go:43-44`), по байтам/памяти — нет.
+Нужны входные данные владельца: RAM ВМ и целевое число тенантов.
+
+Структура (рекомендация):
+- **Cap на тенанта** (суммарные байты тел+атрибутов по всем очередям) — изоляция.
+- **Cap на очередь** — производный (доля тенантского лимита).
+- **Глобальный предохранитель** — отказ send при RSS > порога (напр. 70% RAM), единственная
+  защита от суммы тенантов.
+
+Реализация — счётчик байт в `Queue` (append/delete/purge/compact/restore) + тенантский
+аккумулятор → требует `QueueStore`-интерфейса (план A).
+
+Минимальный шаг без решения по квотам: снизить `MaxMessagesPerQueue` + глобальный
+предохранитель по RSS.
+
+## 5. Порядок исправления (после решений владельца)
+
+1. 🔴1 — убрать `SaveQueue` без `queue.mu` в `create_queue.go:84` (и `admin.go`/`seed.go`).
+2. 🔴8 — валидация `ReceiptHandle != ""` в `delete_message_batch.go`.
+3. 🔴5 — полный откат dedup/FIFO-seq при сбое `SaveMessage(s)Sync`.
+4. Lock order + I/O под глобальным локом.
+5. Fail-fast на старте без Redis.
+6. JWT (после решения п.3) + лимит по байтам (после п.4).