From ecf6b41131ea5ff4277d10db378664a993ad353a Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 10 Oct 2026 18:02:59 +0300 Subject: [PATCH] docs(review): Opus code-review of iot-service (bridge->SQS->consumer) verbatim --- doc/thinking/opus-review-2026-10-10.md | 154 +++++++++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 doc/thinking/opus-review-2026-10-10.md diff --git a/doc/thinking/opus-review-2026-10-10.md b/doc/thinking/opus-review-2026-10-10.md new file mode 100644 index 0000000..c5ee20d --- /dev/null +++ b/doc/thinking/opus-review-2026-10-10.md @@ -0,0 +1,154 @@ +# Ревью iot-service (bridge → SQS → consumer → iotpg) + +> Ответ Opus на промпт «код-ревью IoT». Дословно, без правок. + +Код не запускал и тесты не писал. Поведение paho проверил по исходникам v1.5.1 в module cache. `emqx/emqx.conf` вне вашего списка, но я цитирую его строки, где от них зависит вывод. + +## 🔴 Критично + +1. **Подтверждение SQS до фактической записи (гонка с фоновым flusher).** + - `iot_telemetry_store.go:127-134`: `flushAll` забирает буфер целиком, а фоновый flusher (`:117`) идёт параллельно и игнорирует ошибку. + - `Flush` (`:164`), вызванный из `consumer.go:116`, видит пустой буфер и сразу возвращает `nil`. Тогда `consumer.go:120` удаляет сообщения, пока INSERT ещё в полёте или уже провалился. + - Если провал постоянный (`:149-154`), строки выброшены, а сообщения уже подтверждены. Это потеря данных. + - Тот же эффект в `InsertTelemetry` (`:345-354`). При полном батче и постоянной ошибке теряются до 99 строк от сообщений, которые уже лежат в `pending` и будут подтверждены. + - Нужна сериализация flush и ожидание in-flight записи, а `Flush` должен возвращать ошибку и фонового прохода. + +2. **Ядовитое сообщение (poison pill) валит весь батч, и его никуда не отправляют.** + - `:144-154` и `:345-354` классифицируют ошибку на уровне батча, а не строки. Одна плохая строка выбрасывает до 100 валидных строк того же tenant или вечно блокирует их. + - Триггеры, которые проходят проверки: + - envelope без `payload` (`processor.go:39`) даёт NULL и ошибку 23502; + - `\u0000` в JSON-строке: `json.Valid` в `handler.go:47` пропускает, JSONB отвергает (22P05); + - пустые `namespace` и `device_id`: `processor.go:23-27` проверяет только синтаксис JSON, а `EnsureTenantDB` (`:234`) создаёт БД `tenant_`. + - В `consumer.go:116-119` сообщение не подтверждается и не уходит в DLQ. Счётчик `ApproximateReceiveCount` не читается, поэтому оно крутится вечно, если на очереди нет RedrivePolicy. + - Нужны построчный fallback при постоянной ошибке, семантическая валидация envelope как `errInvalidEnvelope`, и отправка в DLQ после N приёмов. + +3. **Остаток буфера bridge при shutdown не доставляется.** + - `bridge.go:40-41` стартует воркеры с `ctx` от сигнала, а `bridge.go:54-55` вызывает `closeAndWait` уже после его отмены. + - `sender.go:124` и `:132` получают `context canceled` на первой попытке, и все накопленные сообщения логируются как «dropping». + - Комментарий в `main.go:123-126` про «flush SQS» не соответствует поведению. + - Дополнительно после `close(d.ch)` (`sender.go:87`) поздний колбэк paho паникует на `send on closed channel` (`:72`). Disconnect с quiesce 2 с не гарантирует остановку обработчиков. + +4. **Короткая недоступность SQS ведёт к потере данных.** + - `sender.go:31` и `:119-145`: 3 попытки с паузами 1 и 2 с, затем сообщение дропается, хотя буфер на 10 000 почти пуст. + - У `SendMessage` (`:124`) нет таймаута на попытку. Зависшее соединение блокирует все 4 воркера, канал переполняется и `enqueue` дропает (`:70-83`). + - MQTT QoS1 подтверждается брокеру сразу после `enqueue` (`handler.go:52`), то есть до записи в SQS. Сквозной гарантии at-least-once нет. + - Нужны ретраи до дедлайна или shutdown и таймаут на попытку. Опционально отложенный `msg.Ack()` после успешной отправки. + +5. **Пайплайн необратимо умирает после временной ошибки при старте, а `/health` остаётся 200.** + - `bridge.go:27-29` и `consumer.go:29-32`: ошибка `GetQueueUrl` фатальна без ретрая. + - `bridge.go:105-111`: таймаут подключения 30 с при `ConnectRetry=true` возвращает ошибку, но paho продолжает подключаться в фоне. Это утечка клиента и повод для паники на закрытом канале из п.3. + - `main.go:109-120` только логирует «stopped». Процесс не завершается и не перезапускается, а `/health` в `router.go:69` всегда отдаёт 200. + - Нужны ретрай старта с backoff или завершение процесса с кодом ≠ 0. + +6. **Bridge «глохнет» после неудачной подписки.** + - `bridge.go:92` и `:97` вызывают `c.Disconnect(100)` «для реконнекта». В paho это явное отключение (`client.go:460`), автореконнект после него не срабатывает. + - Подписки нет, а `/ready` остаётся true. + +7. **DLQ и подтверждение битых envelope (`consumer.go:33-39`, `:81-94`).** + - Любая ошибка `ResolveQueueURL` для DLQ, даже временная, навсегда превращает `dlqURL` в `""`, и битые сообщения подтверждаются и теряются (`:88-92`). + - Сбой `sendToDLQ` (`:85-87`) только логируется, затем следует подтверждение. Это потеря. + +8. **MQTT-пароль устройства нигде не возвращается.** + - `devices.go:152` отдаёт `deviceToResponse(device, "")`, хотя комментарии (`:37-39`, `:78`) обещают отдать пароль один раз при создании. GET тоже отдаёт `""`. + - Сгенерированный пароль не получить, устройства не смогут подключиться. Это блокер и для Terraform. + +## 🟡 Важно + +- **JWT (`middleware/auth.go`).** + - `:140` при отсутствии `exp` токен вечный. Нужно требовать `exp`, желательно проверять `nbf`, `iss` и `aud`. + - `:57` отдаёт 403 на невалидный и просроченный токен. Должно быть 401, иначе клиент не отличит протухший токен от чужого namespace. + - `:38-42` и `config.go:121`: `AUTH_TEST_MODE` обходит проверку namespace и работает даже при заданном секрете. Лучше запретить при заданном `JWT_HMAC_SECRET` или громко логировать. + - `:92` в тестовом режиме JWT-подобные токены с пустым секретом принимаются без подписи. + - `:61` жёсткое `sub == namespace`. Проверьте, что платформа выдаёт именно такие токены, иначе Terraform получит 403. + +- **Изоляция tenant и валидация входа.** + - `iot_telemetry_store.go:618-620`: `a-b` и `a_b` дают одну БД `tenant_a_b`. Длинные namespace (>56 символов) усекаются PG до 63 байт и тоже сливаются. + - `devices.go:106-113`: нет проверки `name`, `device_id` и namespace. + - `device_id` вида `+`, `#`, `/` превращает ACL-топик (`emqx.conf:75`) в wildcard. + - `_` в namespace делает username `ns_device` неоднозначным (`emqx.conf:46`, `LIMIT 1` без `ORDER BY`). Станет 🔴, если namespace допускают `_`. + - Префикс `__bridge` не зарезервирован. + +- **Consumer: давление и дубли.** + - `consumer.go:99-105`: паузы 1…60 с внутри цикла по батчу при visibility 120 с (`config.go:116`) дают redelivery ещё не обработанных сообщений. Спасает только дедуп. + - `:116-119`: после ошибки flush нет backoff, цикл крутится «горячо» и вычерпывает очередь в память. Во время простоя PG буфер растёт и накапливает дубликаты строк (`store:147`, `:348`). Он не ограничен. + - `processor.go:30-32`: ошибки `EnsureTenantDB` не классифицируются, постоянные ошибки крутятся бесконечно. + +- **`ts` и порядок.** + - `handler.go:57` пишет `received_at` с точностью до секунды. + - `iot_telemetry_store.go:410` и `:415`: `ORDER BY ts DESC` без tiebreaker плюс OFFSET даёт нестабильную пагинацию. Нужны RFC3339Nano и `id DESC`. + +- **Дедуп.** + - `handler.go:58`: `message_id` создаётся на каждый приём из MQTT, поэтому дубль QoS1 (DUP после реконнекта) получает новый uuid и не дедуплицируется. Дедуп защищает только от повторной доставки SQS и ретраев отправки. + - `store:190-192`: пустой `message_id` превращается в NULL, дедупа нет. Лучше отвергать такие envelope или брать `MessageId` из SQS как запасной ключ. + +- **Readiness.** + - `main.go:107` и `:116` выставляют `ready=true` до старта, а не после подключения MQTT и подписки. + - Потеря соединения MQTT (`bridge.go:81`), PG и SQS на `/ready` не влияют. + +- **Shutdown.** + - `main.go:129` не ограничивает `wg.Wait` по времени, а платформа убьёт процесс раньше. + - `main.go:138`: `os.Exit(1)` обходит defer-ы, то есть Close и flush. + - `iot_telemetry_store.go:546`: `Close` делает одну попытку flush, строки после retryable-ошибки теряются при выходе. `Close` не идемпотентен. + +- **MQTT.** + - `bridge.go:75-77`: `MaxReconnectInterval` по умолчанию 10 минут. Нужно ~30 с. + - Фиксированный `ClientID` при rolling-деплое с перекрытием даёт takeover и флаппинг двух инстансов. + +- **`replaceDSNDatabase` (`:596-613`).** + - Работает только с URL-DSN. Для key=value DSN молча подставляет management-БД, и телеметрия всех tenant смешивается. + - `LastIndex("/")` ломается на `sslrootcert=/path` в query (типично для RDS). + - Пулы tenant (`:559-591`) не вытесняются, по 2 idle-соединения на tenant, и это упрётся в `max_connections`. + +- **AWS-совместимость (`sqsclient.go:19-27`).** + - Только статические ключи: нет session token, IRSA и default chain. + - `SQS_ENDPOINT` нельзя оставить пустым (`config.go:109`), без override реальный AWS не используется. + - Нет таймаутов HTTP и настроек ретрайера. + - `SQS_LONG_POLL_SECONDS` не ограничен 20 (`consumer.go:50`). + - Envelope >250 КБ дропается без DLQ (`sender.go:101-110`). + +- **`store/devices.go:19-21`: INSERT не пишет `phase`.** + - Устройство, созданное с `enabled=false`, лежит в БД как `Active`. + - При этом ответ POST считает `Disabled` (`devices.go:127-129`), и GET потом даёт другое значение. + +- **P0-3 (коммент).** + - Хеширование пароля затронет `EnsureBridgeDevice` (`bridge.go:20-26`, перезапись при каждом старте) и `password_hash_algorithm` в `emqx.conf:42-46`. Менять нужно согласованно. + - `tenant_credentials.pg_password` (`:285`) тоже хранится открыто и нигде не используется для подключения. + +## 🟢 Мелочи + +- `handler/mqtt_auth.go` не подключён в `router.go`, то есть это мёртвый код. Если его подключить, он небезопасен: доверяет клиентскому `clientid` (`:123`) и разрешает subscribe на `#` (`:144`). Лучше удалить. +- `admin.go:23`: токен сравнивается не constant-time. +- Нет `MaxBytesReader` и `DisallowUnknownFields` (`devices.go:102`, `:233`). +- `main.go:95-99`: нет `WriteTimeout` и `IdleTimeout`. +- `processor.go:43` пишет «telemetry saved», хотя данные только буферизованы. Это Info на каждое сообщение. +- `logging.go:27` логирует пробы здоровья. +- `consumer.go:75-79`: пустое тело подтверждается без DLQ. +- `consumer.go:120`: сообщения удаляются по одному. Есть `DeleteMessageBatch`. +- При ошибке flush DLQ-сообщения отправляются повторно, и в DLQ появляются дубли. +- `defaults.go:6` обещает `/metrics`, маршрута нет. Нет счётчиков дропов (`sender.go:45`), глубины буфера и DLQ. +- Мёртвый код: `ns=="default"` в `devices.go`, nil-проверки `IoTPG`, `NewFromEnv`. +- `iot_telemetry_store.go:279-291`: сбой между `CREATE DATABASE` и записью credentials делает tenant невидимым в `GetAdminStats` (`:460`). + +## Что стабилизировать в REST для Terraform + +- **Пароль.** Вернуть `mqtt_password` один раз в 201 на POST, иначе ресурс не создать (п.8). Добавить ротацию (отдельный endpoint) или пароль, задаваемый клиентом. Решите заранее, что sensitive и что считается drift. +- **Идентичность ответов POST, GET и PATCH.** + - Сейчас при создании `created_at` пуст, потому что устройство не перечитывается из БД (`devices.go:152`). + - `created_at` в формате `2006-01-02 15:04:05 UTC` (`:73`) нужно привести к RFC3339, как `last_connected`. + - `phase` должен совпадать (см. 🟡 про INSERT). + - Пустой `metadata` (`{}`) не должен пропадать: из-за `omitempty` (`:50`) возникает drift `{}` против null. +- **Поля.** + - Детерминированные, `ForceNew`: `namespace`, `name`, `device_id`. Вычисляемые, стабильные: `mqtt_username` (`ns_device`) и `phase`. + - Нестабильное, только computed: `last_connected`. + - `topic_prefix` (`:66`) даёт `ns/`, а реальный топик `ns/telemetry/{device_id}`. Нужен явный `telemetry_topic`. +- **Изменяемость.** PATCH меняет только `enabled`, `metadata` обновить нельзя. Либо добавить update метаданных, либо зафиксировать `metadata` как `ForceNew`. +- **Контракт ошибок.** + - Токен: 401 против 403 (п. JWT выше). + - Ошибки middleware (`auth.go:27`, `:57`) идут как `text/plain`, а не JSON. + - 409 не различает конфликт `name` и `device_id`. + - Валидация входа возвращает 400 с понятным кодом и полем. + - Нужны единые `error`-структуры и идемпотентный DELETE (404 = «уже удалено»). +- **Валидация.** Регулярки для `name` и `device_id`, максимальная длина, запрет `+`, `#`, `/` и `_` в namespace. +- **Import.** GET по `{namespace}/{name}` уже есть, нужен стабильный ID. +- **List и telemetry.** Ограничения `limit` и `offset` молча игнорируются при ошибке парсинга (`devices.go:163-164`). Нужны явный потолок, `total` или cursor и tiebreaker сортировки. +- **Delete.** Телеметрия при удалении устройства остаётся, это нужно зафиксировать в документации API.