fix(iot): пагинация — tiebreaker ts DESC,id DESC + потолок limit; точный лог (Этап 5)

- QueryTelemetry: ORDER BY ts DESC, id DESC (был без tiebreaker → нестабильная
  пагинация при одинаковых ts).
- ListIoTDevices: limit дефолт 100, потолок 1000 (ошибка парсинга больше не даёт
  безлимитный запрос).
- Логи 'batch dropped' → 'валидные строки — построчно' (после fallback батч НЕ
  теряется целиком).
This commit is contained in:
Repinoid
2026-10-10 19:17:25 +03:00
parent f3927fdf11
commit 38e3f65c21
2 changed files with 22 additions and 10 deletions
@@ -171,10 +171,10 @@ func (s *IoTPostgresStore) flushAll(ctx context.Context) error {
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)
// Постоянная ошибка: валидные строки уже вставлены построчным
// fallback'ом (см. flushTenant), битые отброшены с логом. Повтор
// бессмыслен — в буфер НЕ возвращаем (иначе залипание namespace).
s.log.Error("iotpg: permanent flush error (валидные строки — построчно, битые отброшены)", "namespace", ns, "rows", len(rows), "err", err)
}
}
}
@@ -412,7 +412,7 @@ func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, devic
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)
s.log.Error("iotpg: permanent error (валидные строки — построчно, битые отброшены)", "namespace", namespace, "rows", len(rows), "err", err)
}
}
s.flushMu.Unlock()
@@ -473,13 +473,13 @@ func (s *IoTPostgresStore) QueryTelemetry(ctx context.Context, namespace, device
if deviceID != "" {
rows, err = tenantDB.QueryContext(ctx,
`SELECT id, device_id, ts, payload FROM iot_telemetry
WHERE device_id = $1 ORDER BY ts DESC LIMIT $2 OFFSET $3`,
WHERE device_id = $1 ORDER BY ts DESC, id DESC LIMIT $2 OFFSET $3`,
deviceID, limit, offset,
)
} else {
rows, err = tenantDB.QueryContext(ctx,
`SELECT id, device_id, ts, payload FROM iot_telemetry
ORDER BY ts DESC LIMIT $1 OFFSET $2`,
ORDER BY ts DESC, id DESC LIMIT $1 OFFSET $2`,
limit, offset,
)
}