diff --git a/internal/service/bridge/bridge.go b/internal/service/bridge/bridge.go index 535415b..c26bd78 100644 --- a/internal/service/bridge/bridge.go +++ b/internal/service/bridge/bridge.go @@ -104,28 +104,29 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand }) opts.SetOnConnectHandler(func(c mqtt.Client) { log.Info("bridge: MQTT connected") - // Подписка обязательна. Ретраим ограниченно. НЕ вызываем c.Disconnect(): - // в paho это ЯВНОЕ отключение, после которого автореконнект НЕ срабатывает - // — bridge оставался бы отключённым навсегда (ревью 2026-10-10, 🔴6). - // При сбое ждём следующего реконнекта (OnConnectHandler вызовется заново). - for attempt := 1; attempt <= 3; attempt++ { + // Подписка обязательна. Ретраим пока не удастся (или не оборвётся ctx/ + // соединение). НЕ вызываем c.Disconnect(): в paho это ЯВНОЕ отключение, + // после которого автореконнект НЕ срабатывает (ревью 2026-10-10, 🔴6). + // Раньше было 3 попытки и ожидание следующего реконнекта — но при живом + // соединении OnConnectHandler больше не вызовется, и bridge оставался бы + // без подписки навсегда (ревью 2026-10-11). + for { + if ctx.Err() != nil || !c.IsConnected() { + return + } token := c.Subscribe(telemetryTopicFilter, 1, handler) if token.WaitTimeout(10*time.Second) && token.Error() == nil { - log.Info("bridge: subscribed", "filter", telemetryTopicFilter, "attempt", attempt) + log.Info("bridge: subscribed", "filter", telemetryTopicFilter) setSubscribed(onSubscribed, true) return } - log.Error("bridge: subscribe failed", "filter", telemetryTopicFilter, "attempt", attempt, "err", token.Error()) + log.Error("bridge: subscribe failed, retrying", "filter", telemetryTopicFilter, "err", token.Error()) select { case <-ctx.Done(): return case <-time.After(subscribeRetryInterval): } } - // Подписка не удалась — bridge не готов (/ready=503) до следующего - // реконнекта. - setSubscribed(onSubscribed, false) - log.Error("bridge: subscribe failed after retries, will retry on next reconnect", "filter", telemetryTopicFilter) }) client := mqtt.NewClient(opts) diff --git a/internal/service/bridge/handler.go b/internal/service/bridge/handler.go index 738ec5c..7875ad9 100644 --- a/internal/service/bridge/handler.go +++ b/internal/service/bridge/handler.go @@ -5,13 +5,15 @@ package bridge import ( + "crypto/sha256" + "encoding/hex" "encoding/json" "log/slog" + "regexp" "strings" "time" mqtt "github.com/eclipse/paho.mqtt.golang" - "github.com/google/uuid" ) // TelemetryEnvelope — сообщение, перекладываемое в SQS (разбирается consumer). @@ -42,6 +44,13 @@ func newMessageHandler(dispatcher *sqsDispatcher, log *slog.Logger) mqtt.Message ns := parts[0] deviceID := parts[2] + // Defense-in-depth: EMQX ACL — основной барьер, но при его ошибке/отключении + // бридж принял бы произвольный namespace. Дополнительно проверяем формат. + if !identRe.MatchString(ns) || !identRe.MatchString(deviceID) { + log.Warn("bridge: invalid namespace/device in topic, skipping", "topic", topic) + return + } + // Не-JSON payload оборачиваем в строку. rawPayload := json.RawMessage(payload) if !json.Valid(payload) { @@ -55,7 +64,20 @@ func newMessageHandler(dispatcher *sqsDispatcher, log *slog.Logger) mqtt.Message Topic: topic, Payload: rawPayload, ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano), - MessageID: uuid.NewString(), + MessageID: messageIDFor(ns, deviceID, topic, payload), }) } } + +// identRe — безопасный идентификатор (namespace/device_id): буквы/цифры/дефис, +// не с дефиса, ≤64 символов. Defense-in-depth поверх EMQX ACL. +var identRe = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9-]{0,63}$`) + +// messageIDFor — детерминированный id сообщения для дедупа. +// Повторная QoS1-доставка того же (device, topic, payload) даёт тот же id, +// поэтому ON CONFLICT (device_id, message_id) в БД дедуплицирует дубликаты +// (раньше uuid на каждый колбэк — дедуп на повторе не срабатывал). +func messageIDFor(ns, deviceID, topic string, payload []byte) string { + h := sha256.Sum256([]byte(ns + "|" + deviceID + "|" + topic + "|" + string(payload))) + return hex.EncodeToString(h[:]) +} diff --git a/internal/service/bridge/sender.go b/internal/service/bridge/sender.go index 9195a84..7fe5859 100644 --- a/internal/service/bridge/sender.go +++ b/internal/service/bridge/sender.go @@ -12,6 +12,7 @@ package bridge import ( "context" "encoding/json" + "errors" "log/slog" "sync" "sync/atomic" @@ -19,6 +20,7 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/aws/smithy-go" ) // Константы диспатчера. @@ -176,6 +178,12 @@ func (d *sqsDispatcher) sendWithRetry(ctx context.Context, body string) error { if err == nil { return nil } + // Permanent ошибки (неверный параметр/размер/нет очереди/доступ) — ретрай + // бессмыслен: вернуть сразу, иначе воркер залип бы на poison-сообщении и + // заполнил буфер (дроп остальных). + if isPermanentSQSErr(err) { + return err + } if ctx.Err() != nil { // shutdown: ctx отменён, ретраить бессмысленно. return err @@ -190,3 +198,19 @@ func (d *sqsDispatcher) sendWithRetry(ctx context.Context, body string) error { } } } + +// isPermanentSQSErr — true для ошибок SQS, которые не пройдут ни при каком +// ретрае (валидация, размер, отсутствие очереди, доступ). Остальное (throttling, +// 5xx, сеть) — retryable. +func isPermanentSQSErr(err error) bool { + var apiErr smithy.APIError + if errors.As(err, &apiErr) { + switch apiErr.ErrorCode() { + case "InvalidParameterValue", "InvalidMessageContents", "MessageTooBig", + "QueueDoesNotExist", "InvalidAddress", "InvalidSecurity", + "AccessDenied", "InvalidBatchEntryId": + return true + } + } + return false +}