diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go index 36eac09..6064b02 100644 --- a/internal/storage/iotpg/iot_telemetry_store.go +++ b/internal/storage/iotpg/iot_telemetry_store.go @@ -209,9 +209,40 @@ func (s *IoTPostgresStore) flushTenant(ctx context.Context, ns string, rows []te } sb.WriteString(" ON CONFLICT DO NOTHING") _, err = tenantDB.ExecContext(ctx, sb.String(), args...) + if err != nil && !isRetryableFlushErr(err) { + // Построчный fallback: одна «ядовитая» строка (NULL/пустой payload, + // \u0000 в JSONB, битый JSON) роняет весь multi-VALUES батч (до + // telemetryBatchSize валидных строк). Вставляем по одной — валидные + // сохраняются, битая логируется, namespace не блокируется (ревью 🔴2). + s.flushRowsIndividually(ctx, tenantDB, ns, rows) + } return err } +// flushRowsIndividually — построчная вставка при ПОСТОЯННОЙ ошибке батча. +// Вызывается только из flushTenant, когда ошибка НЕ retryable (т.е. это данные, +// а не сеть/PG). Каждая строка вставляется отдельным INSERT ... ON CONFLICT DO +// NOTHING: валидные сохраняются, битая — логируется и пропускается (не роняет +// соседние и не блокирует namespace вечными ретраями). +func (s *IoTPostgresStore) flushRowsIndividually(ctx context.Context, tenantDB *sql.DB, ns string, rows []telemetryInsert) { + for _, r := range rows { + var mid any + if r.messageID != "" { + mid = r.messageID + } + ts := r.ts + if ts.IsZero() { + ts = time.Now().UTC() + } + _, err := tenantDB.ExecContext(ctx, + `INSERT INTO iot_telemetry (device_id, payload, message_id, ts) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING`, + r.deviceID, r.payload, mid, ts) + if err != nil { + s.log.Error("iotpg: poison row dropped", "namespace", ns, "device", r.deviceID, "err", err) + } + } +} + // NewFromEnv создаёт store из env var IOT_PG_DSN. // Возвращает (nil, nil) если переменная не задана — IoT Postgres опционален. func NewFromEnv(log *slog.Logger) (*IoTPostgresStore, error) {