diff --git a/internal/service/bridge/bridge.go b/internal/service/bridge/bridge.go index 0b8cef4..1c8d313 100644 --- a/internal/service/bridge/bridge.go +++ b/internal/service/bridge/bridge.go @@ -42,7 +42,7 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *sl // Отправка в SQS — через диспатчер (канал + worker-пул): колбэк MQTT // не блокируется сетью (фикс CRITICAL из ревью 2026-08-16). dispatcher := newSQSDispatcher(sqsClient, queueURL, log) - dispatcher.start(ctx) + dispatcher.start() handler := newMessageHandler(dispatcher, log) diff --git a/internal/service/bridge/sender.go b/internal/service/bridge/sender.go index 63ca8b9..9195a84 100644 --- a/internal/service/bridge/sender.go +++ b/internal/service/bridge/sender.go @@ -35,6 +35,11 @@ const ( // sendMaxBackoff — потолок паузы между попытками (экспоненциальный backoff). sendMaxBackoff = 8 * time.Second + // bridgeFlushTimeout — сколько ждать слива остатка буфера в SQS при shutdown. + // Если SQS лежит и ретраи не завершаются, по этому таймауту ctx отменяется, + // воркеры обрывают отправку (остаток логируется как дропнутый). + bridgeFlushTimeout = 10 * time.Second + // maxSQSBodyBytes — потолок размера envelope для SQS. // SQS (и shared-sqs) режут сообщения >256KB — запас на оверхед. maxSQSBodyBytes = 250 * 1024 @@ -48,31 +53,54 @@ type sqsDispatcher struct { log *slog.Logger wg sync.WaitGroup dropped atomic.Int64 + + // mu защищает закрытие канала: не даёт enqueue писать в уже закрытый ch + // (паника «send on closed channel» от позднего колбэка paho — ревью 🔴3). + mu sync.Mutex + closed bool + + // ctx для отправки НЕ связан с сигнальным ctx bridge: иначе при shutdown + // воркеры сразу вернули бы «context canceled» и дропнули остаток буфера + // (ревью 🔴3). Отменяется в closeAndWait по bridgeFlushTimeout, если воркеры + // не успели слить остаток. + ctx context.Context + cancel context.CancelFunc } // newSQSDispatcher создаёт диспатчер. start() запускает worker-пул. func newSQSDispatcher(client *sqs.Client, queueURL string, log *slog.Logger) *sqsDispatcher { + // ctx независимый от сигнала bridge: см. поле ctx. Отменяется только в + // closeAndWait (по таймауту), чтобы при shutdown воркеры успели слить буфер. + ctx, cancel := context.WithCancel(context.Background()) return &sqsDispatcher{ ch: make(chan TelemetryEnvelope, dispatcherQueueSize), client: client, queueURL: queueURL, log: log, + ctx: ctx, + cancel: cancel, } } -// start запускает worker-пул. -func (d *sqsDispatcher) start(ctx context.Context) { +// start запускает worker-пул. Воркеры используют d.ctx (независимый от сигнала). +func (d *sqsDispatcher) start() { for i := 0; i < dispatcherWorkers; i++ { d.wg.Add(1) go func(wid int) { defer d.wg.Done() - d.worker(ctx, wid) + d.worker(wid) }(i) } } -// enqueue кладёт envelope в канал НЕблокирующе. false = буфер полон (дроп). +// enqueue кладёт envelope в канал НЕблокирующе. false = буфер полон (дроп) или +// канал уже закрыт (shutdown). Под mu, чтобы не писать в закрытый ch. func (d *sqsDispatcher) enqueue(e TelemetryEnvelope) bool { + d.mu.Lock() + defer d.mu.Unlock() + if d.closed { + return false + } select { case d.ch <- e: return true @@ -87,15 +115,25 @@ func (d *sqsDispatcher) enqueue(e TelemetryEnvelope) bool { } } -// closeAndWait закрывает канал и ждёт завершения worker'ов (flush). +// closeAndWait закрывает канал и ждёт, пока воркеры сольют остаток буфера в SQS. +// Закрытие — под mu (см. поле mu): после него поздний колбэк paho не упадёт на +// «send on closed channel». Флаш ограничен bridgeFlushTimeout: если SQS лежит, +// отменяем d.ctx, чтобы sendWithRetry оборвался и wg.Wait() вернулся. func (d *sqsDispatcher) closeAndWait() { + d.mu.Lock() + d.closed = true close(d.ch) + d.mu.Unlock() + + t := time.AfterFunc(bridgeFlushTimeout, d.cancel) d.wg.Wait() + t.Stop() d.log.Info("bridge: SQS dispatcher stopped", "dropped_total", d.dropped.Load()) } -// worker — цикл отправки из канала с ограниченными ретраями. -func (d *sqsDispatcher) worker(ctx context.Context, wid int) { +// worker — цикл отправки из канала с ретраями. Использует d.ctx (не сигнальный), +// чтобы при shutdown остаток буфера успел сойти в SQS. +func (d *sqsDispatcher) worker(wid int) { for e := range d.ch { body, err := json.Marshal(e) if err != nil { @@ -113,7 +151,7 @@ func (d *sqsDispatcher) worker(ctx context.Context, wid int) { "limit_bytes", maxSQSBodyBytes) continue } - if err := d.sendWithRetry(ctx, string(body)); err != nil { + if err := d.sendWithRetry(d.ctx, string(body)); err != nil { d.log.Error("bridge: SQS send failed after retries, dropping", "worker", wid, "namespace", e.Namespace, "device", e.DeviceID, "err", err)