diff --git a/HISTORY/2026-10-07-session-log.md b/HISTORY/2026-10-07-session-log.md index 8ce3340..8b5fb81 100644 --- a/HISTORY/2026-10-07-session-log.md +++ b/HISTORY/2026-10-07-session-log.md @@ -61,6 +61,19 @@ не вставляется. - Легаси `cmd/sqs-consumer`: вызов обновлён (`""`), файл помечен к удалению (P2-14). +### Ревью-фикс P0-7/P0-8 (найдено при перепроверке кода) +- ❗ Была реальная потеря данных: `flushAll` и `InsertTelemetry` (при заполнении + батча) вынимали строки из буфера и **подменяли буфер ДО записи**; при ошибке + `flushTenant` строки исчезали. Хуже — фоновый flusher мог «съесть» строки, + которые consumer ждал в `Flush`, тогда `Flush` вернул бы nil (буфер пуст) и + consumer **подтвердил бы** SQS-сообщение с незаписанными данными. +- Фикс: при ошибке `flushTenant` строки ВОЗВРАЩАЮТСЯ в буфер (в обоих путях) — + данные не теряются, повтор выполняет flusher/Flush. `ctx` проброшен в + `flushAll`/`flushTenant` (раньше `Flush(ctx)` игнорировал ctx). +- Consumer: receipt handle хранится значением (`string`), а не указателем SDK. +- Компромисс: при долгом сбое PG буфер растёт (backpressure) — осознанно, вместо + тихой потери телеметрии. + ## Осталось (не сделано) - P0-3 часть 1 (хэш паролей + EMQX), P1-11/P1-12 (тесты/CI), P2-14 (вычистить легаси `internal/api/`, `cmd/sqs-consumer`). diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go index 5d46b4d..572552f 100644 --- a/internal/service/consumer/consumer.go +++ b/internal/service/consumer/consumer.go @@ -59,14 +59,14 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * // успешного Flush — иначе краш/flush-ошибка = потеря подтверждённых // сообщений (P0-7). type pendingAck struct { - handle *string + handle string id string } var pending []pendingAck for _, msg := range resp.Messages { if msg.Body == nil || *msg.Body == "" { log.Warn("consumer: empty message body", "message_id", aws.ToString(msg.MessageId)) - pending = append(pending, pendingAck{msg.ReceiptHandle, aws.ToString(msg.MessageId)}) + pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)}) continue } if err := processTelemetry(ctx, *msg.Body, store, log); err != nil { @@ -84,7 +84,7 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * continue } processBackoff = time.Second - pending = append(pending, pendingAck{msg.ReceiptHandle, aws.ToString(msg.MessageId)}) + pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)}) } if len(pending) == 0 { @@ -107,10 +107,10 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * } // deleteMessage удаляет обработанное сообщение из очереди. -func deleteMessage(ctx context.Context, client *sqs.Client, queueURL string, receiptHandle *string, log *slog.Logger) error { +func deleteMessage(ctx context.Context, client *sqs.Client, queueURL string, receiptHandle string, log *slog.Logger) error { _, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{ QueueUrl: aws.String(queueURL), - ReceiptHandle: receiptHandle, + ReceiptHandle: aws.String(receiptHandle), }) return err } diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go index 3a58c49..5961f57 100644 --- a/internal/storage/iotpg/iot_telemetry_store.go +++ b/internal/storage/iotpg/iot_telemetry_store.go @@ -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()