diff --git a/internal/service/bridge/bridge.go b/internal/service/bridge/bridge.go index 50ef565..0b8cef4 100644 --- a/internal/service/bridge/bridge.go +++ b/internal/service/bridge/bridge.go @@ -19,6 +19,10 @@ import ( // telemetryTopicFilter — все namespace, все устройства. const telemetryTopicFilter = "+/telemetry/+" +// subscribeRetryInterval — пауза между попытками подписки при сбое. +// Подписка обязательна (иначе bridge «глохнет»: соединение есть, а топик не слушается). +const subscribeRetryInterval = 2 * time.Second + // Run — бесконечный цикл bridge: подключение → подписка → ожидание сигнала завершения. func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *slog.Logger) error { log.Info("bridge: starting", "broker", cfg.MQTTBrokerURL, "client_id", cfg.MQTTClientID) @@ -86,18 +90,24 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand }) opts.SetOnConnectHandler(func(c mqtt.Client) { log.Info("bridge: MQTT connected") - token := c.Subscribe(telemetryTopicFilter, 1, handler) - if !token.WaitTimeout(10 * time.Second) { - log.Error("bridge: resubscribe timeout", "filter", telemetryTopicFilter) - c.Disconnect(100) // принудительный реконнект (подписка обязательна) - return + // Подписка обязательна. Ретраим ограниченно. НЕ вызываем c.Disconnect(): + // в paho это ЯВНОЕ отключение, после которого автореконнект НЕ срабатывает + // — bridge оставался бы отключённым навсегда (ревью 2026-10-10, 🔴6). + // При сбое ждём следующего реконнекта (OnConnectHandler вызовется заново). + for attempt := 1; attempt <= 3; attempt++ { + token := c.Subscribe(telemetryTopicFilter, 1, handler) + if token.WaitTimeout(10*time.Second) && token.Error() == nil { + log.Info("bridge: subscribed", "filter", telemetryTopicFilter, "attempt", attempt) + return + } + log.Error("bridge: subscribe failed", "filter", telemetryTopicFilter, "attempt", attempt, "err", token.Error()) + select { + case <-ctx.Done(): + return + case <-time.After(subscribeRetryInterval): + } } - if token.Error() != nil { - log.Error("bridge: resubscribe failed", "filter", telemetryTopicFilter, "err", token.Error()) - c.Disconnect(100) // принудительный реконнект (подписка обязательна) - return - } - log.Info("bridge: subscribed", "filter", telemetryTopicFilter) + log.Error("bridge: subscribe failed after retries, will retry on next reconnect", "filter", telemetryTopicFilter) }) client := mqtt.NewClient(opts)