From 6ff5576e7d1b3805b140c54860dd048e48c76586 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 10 Oct 2026 19:10:44 +0300 Subject: [PATCH] =?UTF-8?q?fix(iot):=20/ready=20=D0=BE=D1=82=D1=80=D0=B0?= =?UTF-8?q?=D0=B6=D0=B0=D0=B5=D1=82=20=D1=84=D0=B0=D0=BA=D1=82=20=D0=BF?= =?UTF-8?q?=D0=BE=D0=B4=D0=BF=D0=B8=D1=81=D0=BA=D0=B8=20MQTT=20(=D0=BD?= =?UTF-8?q?=D0=B0=D1=85=D0=BE=D0=B4=D0=BA=D0=B0=20#3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Было: main выставлял ready.SetBridge(true) ПЕРЕД bridge.Run, поэтому «соединён, но подписка не прошла» давал /ready=200 и молча терял телеметрию до реконнекта. Теперь bridge сам управляет флагом (колбэк onSubscribed): true только после успешного SUBSCRIBE, false при потере соединения/сбое подписки. --- cmd/iot-service/main.go | 6 ++++-- internal/service/bridge/bridge.go | 23 +++++++++++++++++++---- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/cmd/iot-service/main.go b/cmd/iot-service/main.go index f24ac33..303ae38 100644 --- a/cmd/iot-service/main.go +++ b/cmd/iot-service/main.go @@ -118,9 +118,11 @@ func run() int { go func() { defer wg.Done() - ready.SetBridge(true) + // ready.SetBridge управляется самим bridge: true только после успешной + // подписки (находка #3). Иначе «соединён, но не подписан» давал /ready=200 + // и молча терял телеметрию. defer ready.SetBridge(false) - if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) { + if err := bridge.Run(ctx, cfg, sqsClient, log, ready.SetBridge); err != nil && !errors.Is(err, context.Canceled) { log.Error("bridge stopped", "err", err) fatal.Store(true) cancel() // штатный shutdown, затем выход ≠ 0 diff --git a/internal/service/bridge/bridge.go b/internal/service/bridge/bridge.go index 5d1995e..afafa86 100644 --- a/internal/service/bridge/bridge.go +++ b/internal/service/bridge/bridge.go @@ -23,8 +23,18 @@ const telemetryTopicFilter = "+/telemetry/+" // Подписка обязательна (иначе bridge «глохнет»: соединение есть, а топик не слушается). const subscribeRetryInterval = 2 * time.Second -// Run — бесконечный цикл bridge: подключение → подписка → ожидание сигнала завершения. -func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *slog.Logger) error { +// setSubscribed — nil-safe вызов колбэка состояния подписки. +func setSubscribed(fn func(bool), v bool) { + if fn != nil { + fn(v) + } +} + +// Run — бесконечный цикл bridge. onSubscribed — колбэк состояния подписки +// (true после успешного SUBSCRIBE, false при потере соединения/сбое подписки): +// /ready должен отражать реальную подписку, а не факт запуска горутины +// (находка #3: «соединён, но не подписан» молча терял телеметрию при /ready=200). +func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *slog.Logger, onSubscribed func(bool)) error { log.Info("bridge: starting", "broker", cfg.MQTTBrokerURL, "client_id", cfg.MQTTClientID) // URL очереди получаем один раз при старте (как старый consumer). @@ -46,7 +56,7 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *sl handler := newMessageHandler(dispatcher, log) - client, err := connectMQTT(ctx, cfg, log, handler) + client, err := connectMQTT(ctx, cfg, log, handler, onSubscribed) if err != nil { dispatcher.closeAndWait() return err @@ -70,7 +80,7 @@ func (e errBridge) Error() string { return string(e) } // connectMQTT устанавливает подключение к EMQX с автореконнектом. // handler передаётся в OnConnectHandler: подписка повторяется при каждом подключении. -func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, handler mqtt.MessageHandler) (mqtt.Client, error) { +func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, handler mqtt.MessageHandler, onSubscribed func(bool)) (mqtt.Client, error) { opts := mqtt.NewClientOptions() opts.AddBroker(cfg.MQTTBrokerURL) opts.SetClientID(cfg.MQTTClientID) @@ -84,6 +94,7 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand opts.SetConnectionLostHandler(func(_ mqtt.Client, err error) { log.Warn("bridge: MQTT connection lost, reconnecting...", "err", err) + setSubscribed(onSubscribed, false) }) opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) { log.Info("bridge: MQTT reconnecting...") @@ -98,6 +109,7 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand token := c.Subscribe(telemetryTopicFilter, 1, handler) if token.WaitTimeout(10*time.Second) && token.Error() == nil { log.Info("bridge: subscribed", "filter", telemetryTopicFilter, "attempt", attempt) + setSubscribed(onSubscribed, true) return } log.Error("bridge: subscribe failed", "filter", telemetryTopicFilter, "attempt", attempt, "err", token.Error()) @@ -107,6 +119,9 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand case <-time.After(subscribeRetryInterval): } } + // Подписка не удалась — bridge не готов (/ready=503) до следующего + // реконнекта. + setSubscribed(onSubscribed, false) log.Error("bridge: subscribe failed after retries, will retry on next reconnect", "filter", telemetryTopicFilter) })