Не собирался Dockerfile/Makefile (только ./cmd/iot-service/) → в образ/деплой не попадал; логика поглощена internal/service/consumer/. Был источником пустого message_id (обход дедупа). message_id оставлен nullable — единственный пишущий путь (bridge) всегда ставит UUID. LEGACY.md обновлён. Плюс исправлена неточность в HISTORY про поведение без DLQ.
13 KiB
13 KiB
Журнал сессии — 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_SECRETmiddleware отказывает (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: legacycmd/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) — см. ниже.
Реализовано (п.3) — битые envelope → DLQ (42f0045)
processor.go:processTelemetryвозвращает sentinelerrInvalidEnvelopeпри битом JSON (вместо молчаливого nil, из-за которого мусор подтверждался и исчезал навсегда).consumer.go: приerrInvalidEnvelopeтело кладётся в DLQ (SQS_DLQ_NAME, дефолтiot-telemetry-dlq) и основное сообщение подтверждается; без DLQ — только лог (НЕ подтверждаем). Реальные ошибки PG — по-прежнему ретрай.config.go/defaults.go: добавленSQSDLQName+DefaultSQSDLQName.
Реализовано (п.1) — разделение retryable/permanent ошибок flush (f1a7266)
iot_telemetry_store.go:isRetryableFlushErr—'08'(connection) и 57P01/02/03 (shutdown/crash) retryable (возврат в буфер);'22'(data exception) и'23'(integrity violation) permanent (батч выбрасывается с логом, НЕ в буфер).- Закрыто замеченное залипание: один «ядовитый» payload больше не зацикливает flush namespace навсегда.
Реализовано (п.2) — ts = время приёма (a386d22)
iot_telemetry_store.go:telemetryInsertполучилts;flushTenantпишетtsявно изReceivedAt(серверное время приёма, ставит bridge), а неnow()БД. Zero →now(). Схема не менялась (существующая колонкаts).- Callers обновлены:
consumer/processor.go(парситReceivedAt), legacycmd/sqs-consumer.
ОТКРЫТО (п.4) — message_id NOT NULL / дедуп legacy
- Проблема:
UNIQUE(device_id, message_id)не ловит NULL → legacycmd/sqs-consumer(передаёт"") не дедуплицируется; возможны дубли. - Развилка, где я не уверен (ОСТАНОВКА, без догадок):
- Сделать
message_id NOT NULL: нужна миграция (backfilllegacy-'||id+SET NOT NULL). Безопасна, но legacy-консьюмер тогда должен передавать SQSMessageId(иначе его строки не пройдут NOT NULL и будут «дропнуты» как permanent). Legacy и так идёт под удаление (P2-14). - Оставить nullable, задокументировать отсутствие дедупа для legacy и удалить legacy (P2-14). Минимальный риск, но формально «не NOT NULL».
- Сделать
- Нужно решение владельца.
Решено (п.4) — удалён legacy cmd/sqs-consumer
- Проверено:
Dockerfile/Makefileсобирают ТОЛЬКО./cmd/iot-service/→cmd/sqs-consumerне попадает в образ/деплой (пустойmessage_idдо прода не доходил). Логика поглощенаinternal/service/consumer/. - Удалён (
git rm -r cmd/sqs-consumer); сборка/vet зелёные.message_idоставлен nullable — единственный пишущий путь (bridge) всегда ставит UUID. - LEGACY.md обновлён:
cmd/sqs-consumerперенесён в «удалено». - Примечание-противоречие: LEGACY.md до этого называл
cmd/sqs-consumer«НЕ легаси, переиспользуется»; при сверке с Dockerfile/Makefile подтверждено, что standalone-бинарник не используется → удаление корректно.
Исправление неточности (моей)
- В коммите п.3 и в этой HISTORY я написал «без DLQ — НЕ подтверждаем». По коду НЕ так: даже без DLQ битый envelope ПОДТВЕРЖДАЕТСЯ (удаляется с логом) — и это правильно (иначе он зациклился бы). Формулировка была неверной, исправлено.
Проверки
go build ./...иgo vet ./...— зелёные на каждом шаге.