Журнал сессии — 2026-10-07 (IoT: P0-доработки до production-ready)

Рабочая папка: /home/naeel/nubes/SQS_und_IoT/IoT Ветка: feature/replace-kafka-with-sqs

Основание: doc/thinking/remediation-plan-2026-10-05.md (на ветке master) — план по двум анализам готовности. Работа только по явной команде; после каждой правки — коммит.

Сделано

P0-1/P0-2 — tenant-граница + обязательная подпись JWT (`5e1cd28`)

internal/service/api/middleware/auth.go:
- validateJWT теперь возвращает sub;
- middleware сверяет sub с {namespace} из URL (mux.Vars) → 403 при несовпадении (закрыт горизонтальный доступ к чужим namespace);
- вне AUTH_TEST_MODE подпись HS256 ОБЯЗАТЕЛЬНА: без JWT_HMAC_SECRET middleware отказывает (fail-closed, 500), а не проверяет «структурно».
internal/service/config/config.go: JWT_HMAC_SECRET добавлен в список обязательных (когда AUTH_TEST_MODE=false) — отказ на старте.

P1-4 — убраны неаутентифицированные `/internal/mqtt/*` (`cd47258`)

internal/service/api/router.go: удалены маршруты /internal/mqtt/auth|acl. EMQX ходит в PostgreSQL напрямую (backend=postgresql), HTTP-fallback был не нужен и висел на публичном порту без auth (перебор паролей/ACL).
Примечание: те же эндпоинты остаются в ЛЕГАСИ internal/api/ (не исполняется).

P0-3 (часть 2) — пароль устройства не отдаётся повторно (`f480be3`)

internal/service/api/handler/devices.go: GetIoTDevice больше не возвращает mqtt_password (передаём "" в deviceToResponse). Пароль — только при создании.
Часть 1 (хэш паролей в БД + перевод EMQX на password_hash_algorithm) НЕ сделана: требует правки emqx/emqx.conf и миграции существующих строк.

P0-7 — подтверждение из SQS только после записи в БД (`c84cbae`)

Проблема: InsertTelemetry лишь буферизует и возвращает nil до фактической записи; consumer удалял сообщение сразу после processTelemetry == nil → краш/flush-ошибка в окне = безвозвратная потеря телеметрии.
internal/storage/iotpg/iot_telemetry_store.go: добавлен Flush(ctx).
internal/service/consumer/consumer.go: копим receipt handles, вызываем store.Flush(ctx), и только при успехе — DeleteMessage. При ошибке flush сообщения НЕ удаляются (вернутся после visibility timeout).

P1-9 — `/ready` отражает состояние подсистем

internal/service/api/readiness.go — тип Readiness (atomic-флаги bridge/consumer).
internal/service/api/router.go — NewRouter принимает *Readiness; добавлен /ready (200 если обе подсистемы в строю, иначе 503). /health остался liveness-пробой.
cmd/iot-service/main.go — bridge/consumer поднимают/опускают свои флаги; если горутина завершилась (ошибка/падение), /ready вернёт 503.

P0-8 — идемпотентность телеметрии (dedup по message_id)

internal/service/bridge/handler.go: в TelemetryEnvelope добавлен MessageID (UUID на каждое MQTT-сообщение).
internal/service/consumer/processor.go: MessageID передаётся в InsertTelemetry.
internal/storage/iotpg/iot_telemetry_store.go: в таблицу добавлен столбец message_id + UNIQUE INDEX (device_id, message_id); батч-INSERT теперь ON CONFLICT DO NOTHING. Пустой message_id → NULL (NULL в unique-индексе не конфликтует → без ложного дедупа). При at-least-once повторе из SQS дубликат не вставляется.
Легаси cmd/sqs-consumer: вызов обновлён (""), файл помечен к удалению (P2-14).

Ревью-фикс P0-7/P0-8 (найдено при перепроверке кода)

❗ Была реальная потеря данных: flushAll и InsertTelemetry (при заполнении батча) вынимали строки из буфера и подменяли буфер ДО записи; при ошибке flushTenant строки исчезали. Хуже — фоновый flusher мог «съесть» строки, которые consumer ждал в Flush, тогда Flush вернул бы nil (буфер пуст) и consumer подтвердил бы SQS-сообщение с незаписанными данными.
Фикс: при ошибке flushTenant строки ВОЗВРАЩАЮТСЯ в буфер (в обоих путях) — данные не теряются, повтор выполняет flusher/Flush. ctx проброшен в flushAll/flushTenant (раньше Flush(ctx) игнорировал ctx).
Consumer: receipt handle хранится значением (string), а не указателем SDK.
Компромисс: при долгом сбое PG буфер растёт (backpressure) — осознанно, вместо тихой потери телеметрии.

Осталось (не сделано)

P0-3 часть 1 (хэш паролей + EMQX), P1-11/P1-12 (тесты/CI), P2-14 (вычистить легаси internal/api/, cmd/sqs-consumer).
Платформенные проблемы (wss ~150с, MSS/MTU) — вне кода.

Архитекторский разбор (Дипсик Про, по итогам перепроверки)

Перечитана цепочка bridge → SQS → consumer → iotpg целиком. Найдены 5 проблем (часть — мой же недочёт при кодинге P0-7/P0-8):

🔴 Залипание на «ядовитом» батче. После моего фикса «возвращаем строки в буфер при ошибке» один payload, падающий не по ON CONFLICT, зацикливает flush этого namespace навсегда (flusher каждые 200мс). Так «тихая потеря» превратилась в «вечное залипание». ← моя ошибка: фикс без различения типов ошибок.
🔴 Битый envelope теряется безвозвратно. processTelemetry на битый JSON возвращает nil → consumer подтверждает (DeleteMessage) → мусор исчезает без следа. Если мусор шлёт bridge по багу — данные теряются навсегда, без DLQ.
🟠 Нарушение порядка телеметрии. ts ставится now() в БД при INSERT, а не в момент приёма (ReceivedAt из envelope не идёт в батч). При батчинге и нескольких потоках порядок для device_id не гарантирован.
🟠 Дублирование / нет дедупа для message_id="". ON CONFLICT по (device_id, message_id) не ловит NULL: legacy cmd/sqs-consumer передаёт "" → дубли вставляются. Плюс гонка InsertTelemetry(full) ↔ flusher на одном namespace = два параллельных INSERT.
🟡 Close() может потерять последний батч при ошибке shutdown-flush (_ = flushAll(...) — ошибка игнорируется).

Решение (как принято)

п.1 + п.3 (битые/ядовитые → DLQ) — делаю я (Дипсик Про), это дизайн-развилки.
п.2 + п.4 (ReceivedAt; message_id NOT NULL) — передаются Флашу, с критериями.
Битое сообщение — в DLQ (стандартный паттерн: valid→process, invalid→DLQ).
Ядовитый SQL-батч — различать retryable (сеть/PG лежит → ретрай в буфер) от permanent (нарушение constraint → не вешать namespace, выброс в лог/DLQ).

Статус

Реализация п.1 (разделение ошибок flush) и п.3 (ббитые → DLQ) — см. ниже.

Проверки

go build ./... и go vet ./internal/... — зелёные на каждом шаге.

9.1 KiB Raw Blame History Unescape Escape