146 lines
4.6 KiB
Go
146 lines
4.6 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"
|
||
"log/slog"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"github.com/aws/aws-sdk-go-v2/aws"
|
||
"github.com/aws/aws-sdk-go-v2/service/sqs"
|
||
)
|
||
|
||
// Константы диспатчера.
|
||
const (
|
||
// dispatcherQueueSize — буфер envelope перед SQS (~100 msg/s * 100с буфера).
|
||
dispatcherQueueSize = 10000
|
||
// dispatcherWorkers — горутин, отправляющих в SQS.
|
||
dispatcherWorkers = 4
|
||
// sendMaxAttempts — попыток SendMessage на одно сообщение.
|
||
sendMaxAttempts = 3
|
||
|
||
// 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
|
||
}
|
||
|
||
// newSQSDispatcher создаёт диспатчер. start() запускает worker-пул.
|
||
func newSQSDispatcher(client *sqs.Client, queueURL string, log *slog.Logger) *sqsDispatcher {
|
||
return &sqsDispatcher{
|
||
ch: make(chan TelemetryEnvelope, dispatcherQueueSize),
|
||
client: client,
|
||
queueURL: queueURL,
|
||
log: log,
|
||
}
|
||
}
|
||
|
||
// start запускает worker-пул.
|
||
func (d *sqsDispatcher) start(ctx context.Context) {
|
||
for i := 0; i < dispatcherWorkers; i++ {
|
||
d.wg.Add(1)
|
||
go func(wid int) {
|
||
defer d.wg.Done()
|
||
d.worker(ctx, wid)
|
||
}(i)
|
||
}
|
||
}
|
||
|
||
// enqueue кладёт envelope в канал НЕблокирующе. false = буфер полон (дроп).
|
||
func (d *sqsDispatcher) enqueue(e TelemetryEnvelope) bool {
|
||
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 закрывает канал и ждёт завершения worker'ов (flush).
|
||
func (d *sqsDispatcher) closeAndWait() {
|
||
close(d.ch)
|
||
d.wg.Wait()
|
||
d.log.Info("bridge: SQS dispatcher stopped", "dropped_total", d.dropped.Load())
|
||
}
|
||
|
||
// worker — цикл отправки из канала с ограниченными ретраями.
|
||
func (d *sqsDispatcher) worker(ctx context.Context, 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(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 с backoff 1с/2с/4с.
|
||
func (d *sqsDispatcher) sendWithRetry(ctx context.Context, body string) error {
|
||
var lastErr error
|
||
backoff := time.Second
|
||
for attempt := 1; attempt <= sendMaxAttempts; attempt++ {
|
||
_, err := d.client.SendMessage(ctx, &sqs.SendMessageInput{
|
||
QueueUrl: aws.String(d.queueURL),
|
||
MessageBody: aws.String(body),
|
||
})
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
lastErr = err
|
||
if ctx.Err() != nil {
|
||
return lastErr
|
||
}
|
||
if attempt < sendMaxAttempts {
|
||
select {
|
||
case <-ctx.Done():
|
||
return lastErr
|
||
case <-time.After(backoff):
|
||
}
|
||
backoff *= 2
|
||
}
|
||
}
|
||
return lastErr
|
||
}
|