From 4ee0fc3b6144eb35cf31ab65312e5fbec61fd97d Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Wed, 7 Oct 2026 07:41:05 +0300
Subject: [PATCH] =?UTF-8?q?fix(telemetry):=20=D1=80=D0=B0=D0=B7=D0=BB?=
 =?UTF-8?q?=D0=B8=D1=87=D0=B0=D1=82=D1=8C=20retryable/permanent=20=D0=BE?=
 =?UTF-8?q?=D1=88=D0=B8=D0=B1=D0=BA=D0=B8=20flush=20(=D0=BD=D0=B5=20=D0=B7?=
 =?UTF-8?q?=D0=B0=D0=BB=D0=B8=D0=BF=D0=B0=D1=82=D1=8C=20=D0=BD=D0=B0=20?=
 =?UTF-8?q?=D1=8F=D0=B4=D0=BE=D0=B2=D0=B8=D1=82=D0=BE=D0=BC=20=D0=B1=D0=B0?=
 =?UTF-8?q?=D1=82=D1=87=D0=B5)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

isRetryableFlushErr: '08' (connection) и 57P01/02/03 (shutdown/crash) — retryable
(возврат в буфер); '22' (data exception) и '23' (integrity violation) — permanent
(батч выбрасывается с логом, НЕ возвращается в буфер). Иначе один невалидный
payload зацикливал flush namespace навсегда (вечные 200мс-ретраи). Относится к
обоим путям: flushAll и InsertTelemetry(full).
---
 internal/storage/iotpg/iot_telemetry_store.go | 52 +++++++++++++++----
 1 file changed, 42 insertions(+), 10 deletions(-)

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 — смещение для пагинации.