fix(iot): слить остаток буфера bridge при shutdown + безопасное закрытие канала (🔴3)

- Диспатчер получил собственный ctx (не сигнальный): при shutdown воркеры
  успевают отправить накопленное в SQS (раньше сразу возвращали «context
  canceled» и дропали остаток). Флаш ограничен bridgeFlushTimeout=10с.
- enqueue/closeAndWait синхронизированы мьютексом (closed-флаг): поздний колбэк
  paho больше не паникует на «send on closed channel».

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)
 

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)