fix(telemetry): не терять строки при ошибке flush (flushAll + InsertTelemetry)
flushAll и InsertTelemetry (путь заполнения батча) вынимали строки из буфера и подменяли буфер ДО записи: при ошибке flushTenant строки исчезали. Хуже — фоновый flusher мог забрать строки, которые consumer ждал в Flush, тогда Flush видел пустой буфер, возвращал nil, и consumer подтверждал SQS-сообщение с незаписанными данными (потеря). Теперь при ошибке строки возвращаются в буфер в обоих путях. ctx проброшен в flushAll/flushTenant. Consumer хранит receipt handle значением (string), не указателем SDK.
This commit is contained in:
@@ -113,14 +113,17 @@ func (s *IoTPostgresStore) startFlusher() {
|
||||
case <-s.stopFlush:
|
||||
return
|
||||
case <-t.C:
|
||||
s.flushAll(false)
|
||||
s.flushAll(context.Background())
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// flushAll — flush всех накопленных батчей. reportErr=false: только лог.
|
||||
func (s *IoTPostgresStore) flushAll(reportErr bool) error {
|
||||
// flushAll — flush всех накопленных батчей под переданным ctx.
|
||||
// ВАЖНО: при ошибке записи тенанта строки НЕ теряются — возвращаются обратно
|
||||
// в буфер и будут повторены. Иначе фоновый flusher мог бы молча потерять
|
||||
// телеметрию, а consumer, увидев пустой буфер в Flush, подтвердил бы SQS-сообщение.
|
||||
func (s *IoTPostgresStore) flushAll(ctx context.Context) error {
|
||||
s.mu.Lock()
|
||||
if len(s.batches) == 0 {
|
||||
s.mu.Unlock()
|
||||
@@ -132,11 +135,15 @@ func (s *IoTPostgresStore) flushAll(reportErr bool) error {
|
||||
|
||||
var firstErr error
|
||||
for ns, rows := range batches {
|
||||
if err := s.flushTenant(ns, rows); err != nil {
|
||||
if err := s.flushTenant(ctx, ns, rows); err != nil {
|
||||
s.log.Error("iotpg: flush batch", "namespace", ns, "rows", len(rows), "err", err)
|
||||
if firstErr == nil {
|
||||
firstErr = err
|
||||
}
|
||||
// Возвращаем строки в буфер — не теряем данные, повторим позже.
|
||||
s.mu.Lock()
|
||||
s.batches[ns] = append(s.batches[ns], rows...)
|
||||
s.mu.Unlock()
|
||||
}
|
||||
}
|
||||
return firstErr
|
||||
@@ -147,12 +154,12 @@ func (s *IoTPostgresStore) flushAll(reportErr bool) error {
|
||||
// ТОЛЬКО фактически записанные данные: без этого краш/flush-ошибка после
|
||||
// InsertTelemetry (которая лишь буферизует и возвращает nil) = потеря сообщений.
|
||||
func (s *IoTPostgresStore) Flush(ctx context.Context) error {
|
||||
return s.flushAll(true)
|
||||
return s.flushAll(ctx)
|
||||
}
|
||||
|
||||
// flushTenant — INSERT батча строк одного тенанта (multi-VALUES).
|
||||
func (s *IoTPostgresStore) flushTenant(ns string, rows []telemetryInsert) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
func (s *IoTPostgresStore) flushTenant(ctx context.Context, ns string, rows []telemetryInsert) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
tenantDB, err := s.getTenantDB(ctx, ns)
|
||||
if err != nil {
|
||||
@@ -320,7 +327,15 @@ func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, devic
|
||||
s.mu.Unlock()
|
||||
|
||||
if full {
|
||||
return s.flushTenant(namespace, rows)
|
||||
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()
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -488,7 +503,7 @@ func isDBNotExistErr(err error) bool {
|
||||
func (s *IoTPostgresStore) Close() error {
|
||||
close(s.stopFlush)
|
||||
s.flushWG.Wait()
|
||||
_ = s.flushAll(true)
|
||||
_ = s.flushAll(context.Background())
|
||||
s.tenants.Range(func(_, value any) bool {
|
||||
if db, ok := value.(*sql.DB); ok {
|
||||
db.Close()
|
||||
|
||||
Reference in New Issue
Block a user