diff --git a/cmd/iot-service/main.go b/cmd/iot-service/main.go index 12cd2b5..35a5233 100644 --- a/cmd/iot-service/main.go +++ b/cmd/iot-service/main.go @@ -30,6 +30,22 @@ import ( "gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" ) +// shutdownGrace — максимум ожидания остановки подсистем при завершении. Если +// что-то зависло — не блокируем выход (ревью 2026-10-10, «ограничить wg.Wait»). +const shutdownGrace = 15 * time.Second + +// waitWithTimeout ждёт wg не дольше d; false = не дождались. +func waitWithTimeout(wg *sync.WaitGroup, d time.Duration) bool { + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + return true + case <-time.After(d): + return false + } +} + // version — подставляется через ldflags при сборке. var version = "dev" @@ -136,9 +152,10 @@ func run() int { go func() { defer wg.Done() - ready.SetConsumer(true) + // ready.SetConsumer управляется consumer'ом: true после успешного + // ReceiveMessage, false при ошибке SQS (находка «readiness»). defer ready.SetConsumer(false) - if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) { + if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log, ready.SetConsumer); err != nil && !errors.Is(err, context.Canceled) { log.Error("consumer stopped", "err", err) fatal.Store(true) cancel() @@ -151,7 +168,10 @@ func run() int { // (фикс MEDIUM из ревью 2026-08-16). go func() { <-ctx.Done() - wg.Wait() + // Ждём остановки подсистем, но не бесконечно (зависшую не ждём). + if !waitWithTimeout(&wg, shutdownGrace) { + log.Warn("iot-service: подсистемы не завершились за отведённое время", "grace", shutdownGrace) + } shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second) defer shutdownCancel() _ = server.Shutdown(shutdownCtx) @@ -161,11 +181,13 @@ func run() int { if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { log.Error("http server", "err", err) cancel() // останавливаем bridge/consumer - wg.Wait() + waitWithTimeout(&wg, shutdownGrace) return 1 } - wg.Wait() + if !waitWithTimeout(&wg, shutdownGrace) { + log.Warn("iot-service: подсистемы не завершились за отведённое время", "grace", shutdownGrace) + } if fatal.Load() { // Не os.Exit: возвращаем код, чтобы отработали defer-ы (iotStore.Close → // финальный flush последнего батча). os.Exit их бы обошёл. diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go index de8f0f2..fd0f293 100644 --- a/internal/service/consumer/consumer.go +++ b/internal/service/consumer/consumer.go @@ -32,8 +32,9 @@ const maxProcessBackoff = 60 * time.Second // уходит туда на первом же приёме. const maxReceiveCount = 5 -// Run — бесконечный long-poll цикл потребителя. -func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *iotpg.IoTPostgresStore, log *slog.Logger) error { +// Run — бесконечный long-poll цикл потребителя. onReady — колбэк готовности: +// true после успешного ReceiveMessage, false при ошибке SQS (отражается в /ready). +func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *iotpg.IoTPostgresStore, log *slog.Logger, onReady func(bool)) error { queueURL, err := sqsclient.ResolveQueueURLWithRetry(ctx, sqsClient, cfg.SQSQueueName) if err != nil { return err @@ -80,11 +81,13 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * break } log.Error("consumer: ReceiveMessage failed", "err", err) + setReady(onReady, false) if !sleepCtx(ctx, errorBackoff) { break } continue } + setReady(onReady, true) processBackoff := time.Second // Копим подтверждения, но НЕ удаляем из SQS сразу: InsertTelemetry лишь @@ -205,6 +208,13 @@ func receiveCount(msg sqstypes.Message) int { return n } +// setReady — nil-safe вызов колбэка готовности. +func setReady(fn func(bool), v bool) { + if fn != nil { + fn(v) + } +} + // sleepCtx — ожидание с прерыванием по ctx. false = ctx завершён. func sleepCtx(ctx context.Context, d time.Duration) bool { select { diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go index 4a32fa1..e2d85c3 100644 --- a/internal/storage/iotpg/iot_telemetry_store.go +++ b/internal/storage/iotpg/iot_telemetry_store.go @@ -52,6 +52,10 @@ type IoTPostgresStore struct { batches map[string][]telemetryInsert stopFlush chan struct{} flushWG sync.WaitGroup + + // closeOnce — Close идемпотентен: повторный вызов не паникует на close(stopFlush) + // (ревью 2026-10-10). + closeOnce sync.Once } // telemetryInsert — строка для батч-вставки. @@ -602,17 +606,22 @@ func isDBNotExistErr(err error) bool { } // Close останавливает flusher, дописывает остаток и закрывает все подключения. +// Идемпотентен (closeOnce): повторный вызов не паникует на close(stopFlush). func (s *IoTPostgresStore) Close() error { - close(s.stopFlush) - s.flushWG.Wait() - _ = s.flushAll(context.Background()) - s.tenants.Range(func(_, value any) bool { - if db, ok := value.(*sql.DB); ok { - db.Close() - } - return true + var err error + s.closeOnce.Do(func() { + close(s.stopFlush) + s.flushWG.Wait() + _ = s.flushAll(context.Background()) + s.tenants.Range(func(_, value any) bool { + if db, ok := value.(*sql.DB); ok { + db.Close() + } + return true + }) + err = s.adminDB.Close() }) - return s.adminDB.Close() + return err } // getTenantDB возвращает *sql.DB для tenant DB из кэша или открывает новый.