IoT/doc/thinking/opus-review-2026-10-10.md

Ревью 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.