fix(iot): готовность consumer по SQS + ограничение wg.Wait + идемпотентный Close (Этап 3)

- consumer.Run получил onReady: /ready теперь false при ошибке ReceiveMessage
  (раньше флаг consumer'а выставлялся до старта и не отражал SQS).
- main.go: wg.Wait заменён на waitWithTimeout (shutdownGrace=15с) — зависшую
  подсистему не ждём бесконечно.
- iotpg.Close идемпотентен (sync.Once) — повторный вызов не паникует.
This commit is contained in:
Repinoid
2026-10-10 19:16:30 +03:00
parent c725676f85
commit f3927fdf11
3 changed files with 57 additions and 16 deletions
+27 -5
View File
@@ -30,6 +30,22 @@ import (
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" "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 при сборке. // version — подставляется через ldflags при сборке.
var version = "dev" var version = "dev"
@@ -136,9 +152,10 @@ func run() int {
go func() { go func() {
defer wg.Done() defer wg.Done()
ready.SetConsumer(true) // ready.SetConsumer управляется consumer'ом: true после успешного
// ReceiveMessage, false при ошибке SQS (находка «readiness»).
defer ready.SetConsumer(false) 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) log.Error("consumer stopped", "err", err)
fatal.Store(true) fatal.Store(true)
cancel() cancel()
@@ -151,7 +168,10 @@ func run() int {
// (фикс MEDIUM из ревью 2026-08-16). // (фикс MEDIUM из ревью 2026-08-16).
go func() { go func() {
<-ctx.Done() <-ctx.Done()
wg.Wait() // Ждём остановки подсистем, но не бесконечно (зависшую не ждём).
if !waitWithTimeout(&wg, shutdownGrace) {
log.Warn("iot-service: подсистемы не завершились за отведённое время", "grace", shutdownGrace)
}
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second) shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer shutdownCancel() defer shutdownCancel()
_ = server.Shutdown(shutdownCtx) _ = server.Shutdown(shutdownCtx)
@@ -161,11 +181,13 @@ func run() int {
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Error("http server", "err", err) log.Error("http server", "err", err)
cancel() // останавливаем bridge/consumer cancel() // останавливаем bridge/consumer
wg.Wait() waitWithTimeout(&wg, shutdownGrace)
return 1 return 1
} }
wg.Wait() if !waitWithTimeout(&wg, shutdownGrace) {
log.Warn("iot-service: подсистемы не завершились за отведённое время", "grace", shutdownGrace)
}
if fatal.Load() { if fatal.Load() {
// Не os.Exit: возвращаем код, чтобы отработали defer-ы (iotStore.Close → // Не os.Exit: возвращаем код, чтобы отработали defer-ы (iotStore.Close →
// финальный flush последнего батча). os.Exit их бы обошёл. // финальный flush последнего батча). os.Exit их бы обошёл.
+12 -2
View File
@@ -32,8 +32,9 @@ const maxProcessBackoff = 60 * time.Second
// уходит туда на первом же приёме. // уходит туда на первом же приёме.
const maxReceiveCount = 5 const maxReceiveCount = 5
// Run — бесконечный long-poll цикл потребителя. // Run — бесконечный long-poll цикл потребителя. onReady — колбэк готовности:
func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *iotpg.IoTPostgresStore, log *slog.Logger) error { // 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) queueURL, err := sqsclient.ResolveQueueURLWithRetry(ctx, sqsClient, cfg.SQSQueueName)
if err != nil { if err != nil {
return err return err
@@ -80,11 +81,13 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
break break
} }
log.Error("consumer: ReceiveMessage failed", "err", err) log.Error("consumer: ReceiveMessage failed", "err", err)
setReady(onReady, false)
if !sleepCtx(ctx, errorBackoff) { if !sleepCtx(ctx, errorBackoff) {
break break
} }
continue continue
} }
setReady(onReady, true)
processBackoff := time.Second processBackoff := time.Second
// Копим подтверждения, но НЕ удаляем из SQS сразу: InsertTelemetry лишь // Копим подтверждения, но НЕ удаляем из SQS сразу: InsertTelemetry лишь
@@ -205,6 +208,13 @@ func receiveCount(msg sqstypes.Message) int {
return n return n
} }
// setReady — nil-safe вызов колбэка готовности.
func setReady(fn func(bool), v bool) {
if fn != nil {
fn(v)
}
}
// sleepCtx — ожидание с прерыванием по ctx. false = ctx завершён. // sleepCtx — ожидание с прерыванием по ctx. false = ctx завершён.
func sleepCtx(ctx context.Context, d time.Duration) bool { func sleepCtx(ctx context.Context, d time.Duration) bool {
select { select {
+10 -1
View File
@@ -52,6 +52,10 @@ type IoTPostgresStore struct {
batches map[string][]telemetryInsert batches map[string][]telemetryInsert
stopFlush chan struct{} stopFlush chan struct{}
flushWG sync.WaitGroup flushWG sync.WaitGroup
// closeOnce — Close идемпотентен: повторный вызов не паникует на close(stopFlush)
// (ревью 2026-10-10).
closeOnce sync.Once
} }
// telemetryInsert — строка для батч-вставки. // telemetryInsert — строка для батч-вставки.
@@ -602,7 +606,10 @@ func isDBNotExistErr(err error) bool {
} }
// Close останавливает flusher, дописывает остаток и закрывает все подключения. // Close останавливает flusher, дописывает остаток и закрывает все подключения.
// Идемпотентен (closeOnce): повторный вызов не паникует на close(stopFlush).
func (s *IoTPostgresStore) Close() error { func (s *IoTPostgresStore) Close() error {
var err error
s.closeOnce.Do(func() {
close(s.stopFlush) close(s.stopFlush)
s.flushWG.Wait() s.flushWG.Wait()
_ = s.flushAll(context.Background()) _ = s.flushAll(context.Background())
@@ -612,7 +619,9 @@ func (s *IoTPostgresStore) Close() error {
} }
return true return true
}) })
return s.adminDB.Close() err = s.adminDB.Close()
})
return err
} }
// getTenantDB возвращает *sql.DB для tenant DB из кэша или открывает новый. // getTenantDB возвращает *sql.DB для tenant DB из кэша или открывает новый.