diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go index 5961f57..2d0dcbb 100644 --- a/internal/storage/iotpg/iot_telemetry_store.go +++ b/internal/storage/iotpg/iot_telemetry_store.go @@ -140,10 +140,17 @@ func (s *IoTPostgresStore) flushAll(ctx context.Context) error { if firstErr == nil { firstErr = err } - // Возвращаем строки в буфер — не теряем данные, повторим позже. - s.mu.Lock() - s.batches[ns] = append(s.batches[ns], rows...) - s.mu.Unlock() + if isRetryableFlushErr(err) { + // Сеть/PG временно недоступен — возвращаем строки в буфер, повторим. + s.mu.Lock() + s.batches[ns] = append(s.batches[ns], rows...) + s.mu.Unlock() + } else { + // Постоянная ошибка (неверные данные/constraint) — повтор бессмыслен: + // ВЫБРАСЫВАЕМ батч, иначе он зависнет в буфере навсегда и заблокирует + // весь namespace (вечные 200мс-ретраи). Логируем подробно. + s.log.Error("iotpg: permanent flush error, batch dropped", "namespace", ns, "rows", len(rows), "err", err) + } } } return firstErr @@ -328,18 +335,43 @@ func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, devic if full { if err := s.flushTenant(ctx, namespace, rows); err != nil { - // Не теряем строки: возвращаем в буфер. Иначе ранее забуференные - // сообщения, которые consumer собирался подтвердить через Flush, - // оказались бы записаны «мимо» и потеряны. - s.mu.Lock() - s.batches[namespace] = append(s.batches[namespace], rows...) - s.mu.Unlock() + if isRetryableFlushErr(err) { + // Не теряем строки: возвращаем в буфер (сеть/PG временно лежит). + s.mu.Lock() + s.batches[namespace] = append(s.batches[namespace], rows...) + s.mu.Unlock() + } else { + s.log.Error("iotpg: permanent error, batch dropped", "namespace", namespace, "rows", len(rows), "err", err) + } return err } } return nil } +// isRetryableFlushErr — классифицирует ошибку INSERT: retryable (сеть, падение +// PG, таймаут) или permanent (неверные данные/нарушение constraint). Permanent +// НЕ должно возвращаться в буфер — иначе один «ядовитый» payload зациклит flush +// namespace навсегда. +func isRetryableFlushErr(err error) bool { + var pqErr *pq.Error + if errors.As(err, &pqErr) { + switch { + case strings.HasPrefix(string(pqErr.Code), "08"): // connection exception + return true + case pqErr.Code == "57P01" || pqErr.Code == "57P02" || pqErr.Code == "57P03": // admin shutdown / crash / cannot connect now + return true + case strings.HasPrefix(string(pqErr.Code), "22"): // data exception (invalid data) + return false + case strings.HasPrefix(string(pqErr.Code), "23"): // integrity constraint violation + return false + } + } + // Не-pq ошибка (например, context.DeadlineExceeded, net timeout) — считаем + // retryable: это среда, а не данные. + return true +} + // QueryTelemetry читает телеметрию из tenant DB (ts DESC). // deviceID — фильтр (пустая строка = все устройства). limit — max записей // (50..1000), offset — смещение для пагинации.