From 2cca7cb8561fc464a7431def4c0555cd0243eab5 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 10 Oct 2026 18:53:59 +0300 Subject: [PATCH] =?UTF-8?q?fix(iot):=20=D1=81=D0=B5=D0=BC=D0=B0=D0=BD?= =?UTF-8?q?=D1=82=D0=B8=D1=87=D0=B5=D1=81=D0=BA=D0=B0=D1=8F=20=D0=B2=D0=B0?= =?UTF-8?q?=D0=BB=D0=B8=D0=B4=D0=B0=D1=86=D0=B8=D1=8F=20envelope=20(=D0=BF?= =?UTF-8?q?=D1=83=D1=81=D1=82=D1=8B=D0=B5=20=D0=BF=D0=BE=D0=BB=D1=8F=20?= =?UTF-8?q?=E2=86=92=20DLQ)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit processTelemetry теперь отбрасывает (в DLQ) envelope с пустым namespace, device_id или payload ДО записи в БД — иначе пустой namespace создавал бы tenant-БД «tenant_», а пустой payload давал NULL (23502)/битый JSONB и вечный ретрай. Дополняет построчный fallback в flushTenant (🔴2). --- internal/service/consumer/processor.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/internal/service/consumer/processor.go b/internal/service/consumer/processor.go index 254947b..abf0c24 100644 --- a/internal/service/consumer/processor.go +++ b/internal/service/consumer/processor.go @@ -26,6 +26,17 @@ func processTelemetry(ctx context.Context, body string, store *iotpg.IoTPostgres return errInvalidEnvelope } + // Семантическая валидация envelope ДО записи в БД: пустые namespace/device_id + // или пустой payload — битое сообщение, ретраем не исправится. Пустой + // namespace → EnsureTenantDB создал бы БД «tenant_», пустой payload → NULL + // (23502) или битый JSONB. Такое шлём в DLQ (errInvalidEnvelope), а не + // бесконечно ретраим (ревью 2026-10-10, 🔴2). + if envelope.Namespace == "" || envelope.DeviceID == "" || len(envelope.Payload) == 0 { + log.Warn("consumer: invalid envelope (empty namespace/device_id/payload), routing to DLQ", + "namespace", envelope.Namespace, "device", envelope.DeviceID) + return errInvalidEnvelope + } + // EnsureTenantDB идемпотентен, кэшируется после первого вызова. if err := store.EnsureTenantDB(ctx, envelope.Namespace); err != nil { return err