From 7407de5fd3768b218959f77c7732afc4bcd4aa33 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 10 Oct 2026 18:52:43 +0300 Subject: [PATCH] =?UTF-8?q?fix(iot):=20=D1=81=D0=B5=D1=80=D0=B8=D0=B0?= =?UTF-8?q?=D0=BB=D0=B8=D0=B7=D0=BE=D0=B2=D0=B0=D1=82=D1=8C=20flush=20(?= =?UTF-8?q?=D0=B3=D0=BE=D0=BD=D0=BA=D0=B0=20=D1=81=20=D1=84=D0=BE=D0=BD?= =?UTF-8?q?=D0=BE=D0=B2=D1=8B=D0=BC=20flusher'=D0=BE=D0=BC=20=E2=80=94=20?= =?UTF-8?q?=D0=BF=D0=BE=D1=82=D0=B5=D1=80=D1=8F=20=D0=B4=D0=B0=D0=BD=D0=BD?= =?UTF-8?q?=D1=8B=D1=85)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🔴1 ревью: flushAll и синхронная запись полного батча в InsertTelemetry писали параллельно с фоновым flusher'ом (каждые 200мс). Flush() из consumer видел пустой буфер (фоновый flusher уже забрал строки) и возвращал nil → consumer подтверждал SQS-сообщение, пока INSERT ещё в полёте/упал. Добавлен flushMu: сериализует grab буфера + INSERT между фоновым flusher'ом, Flush() и полным батчом InsertTelemetry. Теперь Flush дожидается in-flight записи и возвращает реальную ошибку. --- internal/storage/iotpg/iot_telemetry_store.go | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go index 43e7a9d..36eac09 100644 --- a/internal/storage/iotpg/iot_telemetry_store.go +++ b/internal/storage/iotpg/iot_telemetry_store.go @@ -43,6 +43,11 @@ type IoTPostgresStore struct { // батчинг вставок (фикс MEDIUM ревью 2026-08-16: 1000 msg/s = 1000 INSERT) mu sync.Mutex + // flushMu сериализует фактическую запись (grab буфера + INSERT) между + // фоновым flusher'ом, Flush() из consumer и синхронной записью полного + // батча в InsertTelemetry. Без него Flush видел бы пустой буфер и возвращал + // nil, пока INSERT ещё в полёте → подтверждение SQS до записи (потеря). + flushMu sync.Mutex batches map[string][]telemetryInsert stopFlush chan struct{} flushWG sync.WaitGroup @@ -125,6 +130,12 @@ func (s *IoTPostgresStore) startFlusher() { // в буфер и будут повторены. Иначе фоновый flusher мог бы молча потерять // телеметрию, а consumer, увидев пустой буфер в Flush, подтвердил бы SQS-сообщение. func (s *IoTPostgresStore) flushAll(ctx context.Context) error { + // flushMu: см. поле. Сериализует запись целиком, чтобы Flush() из consumer + // дождался in-flight INSERT фонового flusher'а и вернул РЕАЛЬНУЮ ошибку + // (или nil только если всё записано), а не «буфер пуст, пока идёт запись». + s.flushMu.Lock() + defer s.flushMu.Unlock() + s.mu.Lock() if len(s.batches) == 0 { s.mu.Unlock() @@ -342,7 +353,11 @@ func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, devic s.mu.Unlock() if full { - if err := s.flushTenant(ctx, namespace, rows); err != nil { + // Синхронная запись полного батча — под тем же flushMu, что и фоновый + // flusher и Flush: чтобы не писать параллельно (см. поле flushMu). + s.flushMu.Lock() + err := s.flushTenant(ctx, namespace, rows) + if err != nil { if isRetryableFlushErr(err) { // Не теряем строки: возвращаем в буфер (сеть/PG временно лежит). s.mu.Lock() @@ -351,6 +366,9 @@ func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, devic } else { s.log.Error("iotpg: permanent error, batch dropped", "namespace", namespace, "rows", len(rows), "err", err) } + } + s.flushMu.Unlock() + if err != nil { return err } }