134 lines
10 KiB
Markdown
134 lines
10 KiB
Markdown
# Журнал сессии — 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):
|
||
|
||
1. 🔴 **Залипание на «ядовитом» батче.** После моего фикса «возвращаем строки в
|
||
буфер при ошибке» один payload, падающий не по `ON CONFLICT`, зацикливает
|
||
flush этого namespace навсегда (flusher каждые 200мс). Так «тихая потеря»
|
||
превратилась в «вечное залипание». ← моя ошибка: фикс без различения типов ошибок.
|
||
2. 🔴 **Битый envelope теряется безвозвратно.** `processTelemetry` на битый JSON
|
||
возвращает `nil` → consumer подтверждает (DeleteMessage) → мусор исчезает без
|
||
следа. Если мусор шлёт bridge по багу — данные теряются навсегда, без DLQ.
|
||
3. 🟠 **Нарушение порядка телеметрии.** `ts` ставится `now()` в БД при INSERT, а
|
||
не в момент приёма (`ReceivedAt` из envelope не идёт в батч). При батчинге и
|
||
нескольких потоках порядок для device_id не гарантирован.
|
||
4. 🟠 **Дублирование / нет дедупа для `message_id=""`.** `ON CONFLICT` по
|
||
`(device_id, message_id)` не ловит NULL: legacy `cmd/sqs-consumer` передаёт
|
||
`""` → дубли вставляются. Плюс гонка InsertTelemetry(full) ↔ flusher на одном
|
||
namespace = два параллельных INSERT.
|
||
5. 🟡 **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` возвращает sentinel `errInvalidEnvelope`
|
||
при битом 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 навсегда.
|
||
|
||
## Проверки
|
||
- `go build ./...` и `go vet ./internal/...` — зелёные на каждом шаге.
|