Убран ранний дроп пустого тела; теперь оно идёт через 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' (данные ещё буферизованы).
- consumer.Run получил onReady: /ready теперь false при ошибке ReceiveMessage
(раньше флаг consumer'а выставлялся до старта и не отражал SQS).
- main.go: wg.Wait заменён на waitWithTimeout (shutdownGrace=15с) — зависшую
подсистему не ждём бесконечно.
- iotpg.Close идемпотентен (sync.Once) — повторный вызов не паникует.
- replaceDSNDatabase: возвращает ОШИБКУ для key=value DSN (раньше молча отдавал
исходный DSN → телеметрия всех тенантов ушла бы в management-БД). Query
отделяется первым '?', поэтому sslrootcert=/path больше не ломает поиск БД.
- EnsureTenantDB: guard safeNamespaceRe (defense-in-depth, совпадает с API-валидацией);
небезопасный namespace → ErrInvalidNamespace → в consumer классифицируется как
постоянная ошибка (DLQ, не вечный ретрай).
Было: при недоступной DLQ битое сообщение НЕ подтверждалось и крутилось на
основной очереди бесконечно. Теперь читается ApproximateReceiveCount; после
maxReceiveCount=5 битое сообщение подтверждается (дроп с громким логом), а не
зацикливается. При доступной DLQ — уходит туда на первом приёме.
- Добавлен ResolveQueueURLWithRetry (10 попыток × 2с backoff): временная ошибка
GetQueueUrl больше не роняет bridge/consumer без ретрая.
- main.go: при ПОСТОЯННОЙ ошибке bridge/consumer процесс завершается с кодом ≠ 0
(fatal-флаг + cancel), чтобы платформа перезапустила контейнер — раньше процесс
оставался жив, а /health отдавал 200.
- ResolveQueueURL для DLQ теперь ретраится 3 раза с backoff: временная ошибка
больше не отключает DLQ навсегда.
- Битый envelope подтверждается только если DLQ реально приняла его; при
недоступной DLQ/сбое sendToDLQ сообщение НЕ подтверждается (вернётся после
visibility timeout), а не дропается.
processTelemetry теперь отбрасывает (в DLQ) envelope с пустым namespace,
device_id или payload ДО записи в БД — иначе пустой namespace создавал бы
tenant-БД «tenant_», а пустой payload давал NULL (23502)/битый JSONB и вечный
ретрай. Дополняет построчный fallback в flushTenant (🔴2).
Раньше ts проставлялся now() в БД при INSERT → при батчинге и параллельных
записях порядок телеметрии (ORDER BY ts DESC) ломался. Теперь в ts идёт
ReceivedAt из envelope (серверное время приёма, ставит bridge); битый/пустой
received_at → zero → стор подставляет now(). Схема не меняется (используется
существующая колонка ts). Обновлены callers: consumer/processor.go и legacy
cmd/sqs-consumer.
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
(без ложного дедупа).
InsertTelemetry лишь буферизует и возвращает nil до фактической записи. Consumer
удалял сообщение сразу после processTelemetry == nil — краш/flush-ошибка в этом
окне = безвозвратная потеря. Теперь: копим receipt handles, вызываем store.Flush,
и только при успехе DeleteMessage; при ошибке flush — не удаляем (сообщения
вернутся после visibility timeout). Добавлен iotpg.Flush().
Repinoid