217 lines
8.7 KiB
Go
217 lines
8.7 KiB
Go
// sender.go — асинхронная отправка envelope в SQS из бриджа.
|
||
//
|
||
// Мотивация (Sonnet-ревью 2026-08-16, находка CRITICAL): SendMessage был
|
||
// синхронным внутри MQTT-колбэка paho. При недоступности SQS (таймаут ~30с)
|
||
// блокировался приём ВСЕХ MQTT-сообщений → потери телеметрии.
|
||
//
|
||
// Теперь: колбэк кладёт envelope в буферизованный канал НЕблокирующе
|
||
// (переполнение — дроп со счётчиком), worker-пул отправляет в SQS с
|
||
// ограниченными ретраями и backoff.
|
||
package bridge
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"log/slog"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"github.com/aws/aws-sdk-go-v2/aws"
|
||
"github.com/aws/aws-sdk-go-v2/service/sqs"
|
||
"github.com/aws/smithy-go"
|
||
)
|
||
|
||
// Константы диспатчера.
|
||
const (
|
||
// dispatcherQueueSize — буфер envelope перед SQS (~100 msg/s * 100с буфера).
|
||
dispatcherQueueSize = 10000
|
||
// dispatcherWorkers — горутин, отправляющих в SQS.
|
||
dispatcherWorkers = 4
|
||
|
||
// sendAttemptTimeout — таймаут ОДНОЙ попытки SendMessage. Без него зависшее
|
||
// соединение (SQS висит, но не отвечает) блокировало бы воркера навсегда и
|
||
// забило бы весь пул (ревью 🔴4).
|
||
sendAttemptTimeout = 10 * time.Second
|
||
// 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
|
||
)
|
||
|
||
// sqsDispatcher — канал + worker-пул для отправки в SQS.
|
||
type sqsDispatcher struct {
|
||
ch chan TelemetryEnvelope
|
||
client *sqs.Client
|
||
queueURL string
|
||
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-пул. Воркеры используют 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(wid)
|
||
}(i)
|
||
}
|
||
}
|
||
|
||
// 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
|
||
default:
|
||
n := d.dropped.Add(1)
|
||
if n%100 == 1 {
|
||
d.log.Error("bridge: SQS dispatch queue full, dropping",
|
||
"namespace", e.Namespace, "device", e.DeviceID,
|
||
"dropped_total", n)
|
||
}
|
||
return false
|
||
}
|
||
}
|
||
|
||
// 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 — цикл отправки из канала с ретраями. Использует d.ctx (не сигнальный),
|
||
// чтобы при shutdown остаток буфера успел сойти в SQS.
|
||
func (d *sqsDispatcher) worker(wid int) {
|
||
for e := range d.ch {
|
||
body, err := json.Marshal(e)
|
||
if err != nil {
|
||
d.log.Error("bridge: marshal envelope", "worker", wid,
|
||
"namespace", e.Namespace, "err", err)
|
||
continue
|
||
}
|
||
if len(body) > maxSQSBodyBytes {
|
||
// Лимит SQS 256KB — такие сообщения не пройдут в принципе
|
||
// (проверено 2026-08-16: InvalidParameterValue message size
|
||
// exceeds the limit). Дроп с явным логом.
|
||
d.log.Error("bridge: payload too large for SQS, dropping",
|
||
"worker", wid, "namespace", e.Namespace,
|
||
"device", e.DeviceID, "bytes", len(body),
|
||
"limit_bytes", maxSQSBodyBytes)
|
||
continue
|
||
}
|
||
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)
|
||
}
|
||
}
|
||
}
|
||
|
||
// sendWithRetry — SendMessage с ретраями до остановки (ctx) и таймаутом на
|
||
// попытку. НЕ дропаем после фиксированного числа попыток (было 3 ≈ 7с): буфер
|
||
// на dispatcherQueueSize рассчитан на поглощение временной недоступности SQS,
|
||
// поэтому ретраим до ctx.Done() (shutdown) или успеха — иначе потеря была бы
|
||
// безвозвратной (ревью 🔴4).
|
||
func (d *sqsDispatcher) sendWithRetry(ctx context.Context, body string) error {
|
||
backoff := time.Second
|
||
for {
|
||
attemptCtx, cancel := context.WithTimeout(ctx, sendAttemptTimeout)
|
||
_, err := d.client.SendMessage(attemptCtx, &sqs.SendMessageInput{
|
||
QueueUrl: aws.String(d.queueURL),
|
||
MessageBody: aws.String(body),
|
||
})
|
||
cancel()
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
// Permanent ошибки (неверный параметр/размер/нет очереди/доступ) — ретрай
|
||
// бессмыслен: вернуть сразу, иначе воркер залип бы на poison-сообщении и
|
||
// заполнил буфер (дроп остальных).
|
||
if isPermanentSQSErr(err) {
|
||
return err
|
||
}
|
||
if ctx.Err() != nil {
|
||
// shutdown: ctx отменён, ретраить бессмысленно.
|
||
return err
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return err
|
||
case <-time.After(backoff):
|
||
}
|
||
if backoff < sendMaxBackoff {
|
||
backoff *= 2
|
||
}
|
||
}
|
||
}
|
||
|
||
// 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
|
||
}
|