fix(iot): семантическая валидация envelope (пустые поля → DLQ)
processTelemetry теперь отбрасывает (в DLQ) envelope с пустым namespace,
device_id или payload ДО записи в БД — иначе пустой namespace создавал бы
tenant-БД «tenant_», а пустой payload давал NULL (23502)/битый JSONB и вечный
ретрай. Дополняет построчный fallback в flushTenant (🔴2).
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user