fix(telemetry): различать retryable/permanent ошибки flush (не залипать на ядовитом батче)
isRetryableFlushErr: '08' (connection) и 57P01/02/03 (shutdown/crash) — retryable (возврат в буфер); '22' (data exception) и '23' (integrity violation) — permanent (батч выбрасывается с логом, НЕ возвращается в буфер). Иначе один невалидный payload зацикливал flush namespace навсегда (вечные 200мс-ретраи). Относится к обоим путям: flushAll и InsertTelemetry(full).
This commit is contained in:
@@ -140,10 +140,17 @@ func (s *IoTPostgresStore) flushAll(ctx context.Context) error {
|
|||||||
if firstErr == nil {
|
if firstErr == nil {
|
||||||
firstErr = err
|
firstErr = err
|
||||||
}
|
}
|
||||||
// Возвращаем строки в буфер — не теряем данные, повторим позже.
|
if isRetryableFlushErr(err) {
|
||||||
|
// Сеть/PG временно недоступен — возвращаем строки в буфер, повторим.
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
s.batches[ns] = append(s.batches[ns], rows...)
|
s.batches[ns] = append(s.batches[ns], rows...)
|
||||||
s.mu.Unlock()
|
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
|
return firstErr
|
||||||
@@ -328,18 +335,43 @@ func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, devic
|
|||||||
|
|
||||||
if full {
|
if full {
|
||||||
if err := s.flushTenant(ctx, namespace, rows); err != nil {
|
if err := s.flushTenant(ctx, namespace, rows); err != nil {
|
||||||
// Не теряем строки: возвращаем в буфер. Иначе ранее забуференные
|
if isRetryableFlushErr(err) {
|
||||||
// сообщения, которые consumer собирался подтвердить через Flush,
|
// Не теряем строки: возвращаем в буфер (сеть/PG временно лежит).
|
||||||
// оказались бы записаны «мимо» и потеряны.
|
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
s.batches[namespace] = append(s.batches[namespace], rows...)
|
s.batches[namespace] = append(s.batches[namespace], rows...)
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
|
} else {
|
||||||
|
s.log.Error("iotpg: permanent error, batch dropped", "namespace", namespace, "rows", len(rows), "err", err)
|
||||||
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return nil
|
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).
|
// QueryTelemetry читает телеметрию из tenant DB (ts DESC).
|
||||||
// deviceID — фильтр (пустая строка = все устройства). limit — max записей
|
// deviceID — фильтр (пустая строка = все устройства). limit — max записей
|
||||||
// (50..1000), offset — смещение для пагинации.
|
// (50..1000), offset — смещение для пагинации.
|
||||||
|
|||||||
Reference in New Issue
Block a user