From 6048777a9ed6e671b97004537a7733caada41797 Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Sat, 10 Oct 2026 18:55:34 +0300
Subject: [PATCH] =?UTF-8?q?fix(iot):=20=D0=BD=D0=B5=20=D1=80=D0=B2=D0=B0?=
 =?UTF-8?q?=D1=82=D1=8C=20=D0=B0=D0=B2=D1=82=D0=BE=D1=80=D0=B5=D0=BA=D0=BE?=
 =?UTF-8?q?=D0=BD=D0=BD=D0=B5=D0=BA=D1=82=20=D0=BF=D1=80=D0=B8=20=D1=81?=
 =?UTF-8?q?=D0=B1=D0=BE=D0=B5=20=D0=BF=D0=BE=D0=B4=D0=BF=D0=B8=D1=81=D0=BA?=
 =?UTF-8?q?=D0=B8=20(=F0=9F=94=B46)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

OnConnectHandler вызывал c.Disconnect(100) при таймауте/ошибке подписки. В paho
это ЯВНОЕ отключение — автореконнект после него не срабатывает, и bridge
«глохнет» (соединения нет, подписки нет, /ready=true).

Теперь: подписка ретраится 3 раза с паузой 2с; при сбое — ждём следующего
реконнекта (OnConnectHandler снова вызовется). Disconnect не вызывается.
---
 internal/service/bridge/bridge.go | 32 ++++++++++++++++++++-----------
 1 file changed, 21 insertions(+), 11 deletions(-)

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)