Раньше 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