- apiBase → window.location.origin (same-origin с API: консоль и /v1 раздаёт один
iot-service).
- mqttBase → wss://exqx.containerk8s.dev.nubes.ru/mqtt (как в landing.html).
- Убраны упоминания «оператор»/k8s в комментариях; MQTTX-инструкция — новый хост.
Убран ранний дроп пустого тела; теперь оно идёт через processTelemetry →
errInvalidEnvelope → DLQ (или дроп после maxReceiveCount, если DLQ недоступна).
bridge пишет, consumer парсит RFC3339Nano (субсекундная точность). Раньше RFC3339
(секунды) + non-FIFO очередь → показания в одну секунду могли встать не в том
порядке. Parse RFC3339Nano совместим со старыми RFC3339-сообщениями (доля секунды
опциональна).
- bridge: SetMaxReconnectInterval(30с) — дефолт ~10м слишком долог для телеметрии.
- logging middleware: /health,/healthz,/ready → Debug (не засорять лог).
- processor: 'telemetry buffered' вместо Info 'saved' (данные ещё буферизованы).
store.Create классифицирует 23505 по имени constraint: (namespace,device_id) →
ErrDeviceIDExists, иначе ErrAlreadyExists. Handler отдаёт разные 409-сообщения.
Помогает Terraform понять, какое поле конфликтует.
- handler CreateIoTDevice: пароль устройства хранится bcrypt-хэшем (hashPassword),
открытый пароль отдаётся один раз в 201.
- store.EnsureBridgeDevice: пароль бриджа тоже bcrypt-хэшем.
- emqx/emqx.conf: password_hash_algorithm name=bcrypt, salt_rounds=10 (подтверждено
по emqx/emqx: HOCON-ключ salt_rounds; Go bcrypt.DefaultCost=10).
- Добавлена зависимость golang.org/x/crypto v0.42.0.
iotDeviceUpdateRequest получил поле metadata; UpdateIoTDevice обновляет переданные
поля (enabled и/или metadata). Добавлен store.UpdateMetadata. Хотя бы одно поле
обязательно. Раньше metadata нельзя было изменить (только enabled).
- QueryTelemetry: ORDER BY ts DESC, id DESC (был без tiebreaker → нестабильная
пагинация при одинаковых ts).
- ListIoTDevices: limit дефолт 100, потолок 1000 (ошибка парсинга больше не даёт
безлимитный запрос).
- Логи 'batch dropped' → 'валидные строки — построчно' (после fallback батч НЕ
теряется целиком).
- consumer.Run получил onReady: /ready теперь false при ошибке ReceiveMessage
(раньше флаг consumer'а выставлялся до старта и не отражал SQS).
- main.go: wg.Wait заменён на waitWithTimeout (shutdownGrace=15с) — зависшую
подсистему не ждём бесконечно.
- iotpg.Close идемпотентен (sync.Once) — повторный вызов не паникует.
- Удалён internal/service/api/handler/mqtt_auth.go: методы MQTTAuth/MQTTAcl
нигде не подключены (мёртвый код; при подключении небезопасен — доверял
клиентскому clientid и разрешал subscribe на '#').
- admin.go: сравнение ADMIN_STATS_TOKEN через subtle.ConstantTimeCompare.
- replaceDSNDatabase: возвращает ОШИБКУ для key=value DSN (раньше молча отдавал
исходный DSN → телеметрия всех тенантов ушла бы в management-БД). Query
отделяется первым '?', поэтому sslrootcert=/path больше не ломает поиск БД.
- EnsureTenantDB: guard safeNamespaceRe (defense-in-depth, совпадает с API-валидацией);
небезопасный namespace → ErrInvalidNamespace → в consumer классифицируется как
постоянная ошибка (DLQ, не вечный ретрай).
Было: main выставлял ready.SetBridge(true) ПЕРЕД bridge.Run, поэтому «соединён,
но подписка не прошла» давал /ready=200 и молча терял телеметрию до реконнекта.
Теперь bridge сам управляет флагом (колбэк onSubscribed): true только после
успешного SUBSCRIBE, false при потере соединения/сбое подписки.
Было: при недоступной DLQ битое сообщение НЕ подтверждалось и крутилось на
основной очереди бесконечно. Теперь читается ApproximateReceiveCount; после
maxReceiveCount=5 битое сообщение подтверждается (дроп с громким логом), а не
зацикливается. При доступной DLQ — уходит туда на первом приёме.
os.Exit обходит defer-ы: на fatal-пути и при ошибке HTTP-сервера iotStore.Close()
(финальный flushAll последнего батча телеметрии) не выполнялся. Логика вынесена в
run() int, main = os.Exit(run()); все выходы через return — defer-ы отрабатывают.
- Добавлен ResolveQueueURLWithRetry (10 попыток × 2с backoff): временная ошибка
GetQueueUrl больше не роняет bridge/consumer без ретрая.
- main.go: при ПОСТОЯННОЙ ошибке bridge/consumer процесс завершается с кодом ≠ 0
(fatal-флаг + cancel), чтобы платформа перезапустила контейнер — раньше процесс
оставался жив, а /health отдавал 200.
- Диспатчер получил собственный ctx (не сигнальный): при shutdown воркеры
успевают отправить накопленное в SQS (раньше сразу возвращали «context
canceled» и дропали остаток). Флаш ограничен bridgeFlushTimeout=10с.
- enqueue/closeAndWait синхронизированы мьютексом (closed-флаг): поздний колбэк
paho больше не паникует на «send on closed channel».
- sendWithRetry больше не дропает после 3 попыток (~7с) при почти пустом буфере:
ретраит до ctx.Done() (shutdown) или успеха, backoff капится на 8с.
- Добавлен sendAttemptTimeout=10с на одну попытку — зависшее соединение SQS
не блокирует воркер навсегда.
- ResolveQueueURL для DLQ теперь ретраится 3 раза с backoff: временная ошибка
больше не отключает DLQ навсегда.
- Битый envelope подтверждается только если DLQ реально приняла его; при
недоступной DLQ/сбое sendToDLQ сообщение НЕ подтверждается (вернётся после
visibility timeout), а не дропается.
OnConnectHandler вызывал c.Disconnect(100) при таймауте/ошибке подписки. В paho
это ЯВНОЕ отключение — автореконнект после него не срабатывает, и bridge
«глохнет» (соединения нет, подписки нет, /ready=true).
Теперь: подписка ретраится 3 раза с паузой 2с; при сбое — ждём следующего
реконнекта (OnConnectHandler снова вызовется). Disconnect не вызывается.
processTelemetry теперь отбрасывает (в DLQ) envelope с пустым namespace,
device_id или payload ДО записи в БД — иначе пустой namespace создавал бы
tenant-БД «tenant_», а пустой payload давал NULL (23502)/битый JSONB и вечный
ретрай. Дополняет построчный fallback в flushTenant (🔴2).
Одна «ядовитая» строка (NULL/пустой payload, \u0000 в JSONB, битый JSON)
ронила весь multi-VALUES батч (до telemetryBatchSize валидных строк), который
классифицировался как permanent и выбрасывался целиком.
Добавлен flushRowsIndividually: при не-retryable ошибке батча вставляем строки
по одной (ON CONFLICT DO NOTHING) — валидные сохраняются, битая логируется.
🔴1 ревью: flushAll и синхронная запись полного батча в InsertTelemetry писали
параллельно с фоновым flusher'ом (каждые 200мс). Flush() из consumer видел
пустой буфер (фоновый flusher уже забрал строки) и возвращал nil → consumer
подтверждал SQS-сообщение, пока INSERT ещё в полёте/упал.
Добавлен flushMu: сериализует grab буфера + INSERT между фоновым flusher'ом,
Flush() и полным батчом InsertTelemetry. Теперь Flush дожидается in-flight
записи и возвращает реальную ошибку.
- iotDeviceResponse: добавлено telemetry_topic = «ns/telemetry/device» (явный
топик для Terraform; topic_prefix = «ns/» был неполным).
- metadata теперь без omitempty и всегда «{}» при пустом — убран drift
«{}» против null (ревью 2026-10-10, «Что стабилизировать для Terraform»).
- deviceToResponse: created_at теперь RFC3339 (был «2006-01-02 15:04:05 UTC»)
— единый формат с last_connected, без drift для Terraform.
- CreateIoTDevice: после INSERT перечитываем устройство из БД (GetByName),
чтобы в 201 отдавался реальный created_at (in-memory структура его не имеет).
- handler/devices.go: CreateIoTDevice отдаёт mqtt_password один раз в 201
(было deviceToResponse(device, "") — пароль терялся, устройство нельзя
подключить; блокер для Terraform). Заголовок поправлен (GET — без пароля).
- store/devices.go: Create пишет phase явно через phaseOf(enabled) — иначе
enabled=false лежал в БД как 'Active' (DEFAULT), расходясь с POST/GET.
- iot-admin.html: секции Kafka/Pods заменены на Очередь shared-SQS и Рантайм
(бэкенд admin.go давно отдаёт sqs/runtime, а фронт рисовал мёртвые секции);
удалены CSS .pod-*/.badge-*/.dot-*/.lag-bar*, исправлен заголовок.
- .gitignore: удалены сталые пути бинарников старого IoT (/iot-operator,
/mqtt-bridge, /kafka-consumer, /sqs-consumer).
Не собирался Dockerfile/Makefile (только ./cmd/iot-service/) → в образ/деплой
не попадал; логика поглощена internal/service/consumer/. Был источником пустого
message_id (обход дедупа). message_id оставлен nullable — единственный пишущий
путь (bridge) всегда ставит UUID. LEGACY.md обновлён. Плюс исправлена неточность
в HISTORY про поведение без DLQ.
Раньше ts проставлялся now() в БД при INSERT → при батчинге и параллельных
записях порядок телеметрии (ORDER BY ts DESC) ломался. Теперь в ts идёт
ReceivedAt из envelope (серверное время приёма, ставит bridge); битый/пустой
received_at → zero → стор подставляет now(). Схема не меняется (используется
существующая колонка ts). Обновлены callers: consumer/processor.go и legacy
cmd/sqs-consumer.
isRetryableFlushErr: '08' (connection) и 57P01/02/03 (shutdown/crash) — retryable
(возврат в буфер); '22' (data exception) и '23' (integrity violation) — permanent
(батч выбрасывается с логом, НЕ возвращается в буфер). Иначе один невалидный
payload зацикливал flush namespace навсегда (вечные 200мс-ретраи). Относится к
обоим путям: flushAll и InsertTelemetry(full).
processTelemetry теперь возвращает sentinel errInvalidEnvelope при битом JSON
(раньше — молчаливый nil → mусор подтверждался и терялся). Consumer при этой
ошибке кладёт тело в DLQ (SQS_DLQ_NAME, дефолт iot-telemetry-dlq) и подтверждает
основное сообщение; без DLQ — только логирует, но НЕ подтверждает (оседает).
Добавлены SQSDLQName в config + DefaultSQSDLQName. Реальные ошибки PG (не invalid)
по-прежнему не подтверждаются и уходят в ретрай после visibility timeout.
flushAll и InsertTelemetry (путь заполнения батча) вынимали строки из буфера и
подменяли буфер ДО записи: при ошибке flushTenant строки исчезали. Хуже —
фоновый flusher мог забрать строки, которые consumer ждал в Flush, тогда Flush
видел пустой буфер, возвращал nil, и consumer подтверждал SQS-сообщение с
незаписанными данными (потеря). Теперь при ошибке строки возвращаются в буфер
в обоих путях. ctx проброшен в flushAll/flushTenant. Consumer хранит receipt
handle значением (string), не указателем SDK.
Envelope получил MessageID (UUID); таблица iot_telemetry — столбец message_id +
UNIQUE(device_id,message_id); батч-INSERT ON CONFLICT DO NOTHING. Дубли при
at-least-once повторе из SQS больше не вставляются. Пустой message_id → NULL
(без ложного дедупа).
/health остаётся liveness (процесс жив); /ready возвращает 503, если bridge или
consumer не в строю (горутина упала/не поднялась). Добавлен api.Readiness с
atomic-флагами, NewRouter принимает *Readiness, main поднимает/опускает флаги
вокруг фоновых горутин.
InsertTelemetry лишь буферизует и возвращает nil до фактической записи. Consumer
удалял сообщение сразу после processTelemetry == nil — краш/flush-ошибка в этом
окне = безвозвратная потеря. Теперь: копим receipt handles, вызываем store.Flush,
и только при успехе DeleteMessage; при ошибке flush — не удаляем (сообщения
вернутся после visibility timeout). Добавлен iotpg.Flush().
GetIoTDevice больше не возвращает mqtt_password (deviceToResponse получает ).
Пароль остаётся доступен только один раз — при создании устройства.
Хэширование паролей в БД + перевод EMQX на алгоритм хэша (часть 1) требует
изменения emqx.conf и миграции — отдельным шагом.
EMQX ходит в PostgreSQL напрямую (backend=postgresql), эти HTTP-эндпоинты были
неиспользуемым fallback и висели на публичном порту без аутентификации (перебор
паролей устройств / выдача ACL). Удалены из роутера.
validateJWT теперь возвращает sub; middleware сверяет sub с {namespace} из URL
(несовпадение → 403) — закрывает горизонтальный доступ к чужим namespace.
Вне AUTH_TEST_MODE подпись HS256 обязательна: без JWT_HMAC_SECRET middleware
отказывает (fail-closed, 500), а config.Load требует секрет на старте.
Repinoid