diff --git a/internal/service/bridge/sender.go b/internal/service/bridge/sender.go index 856b465..63ca8b9 100644 --- a/internal/service/bridge/sender.go +++ b/internal/service/bridge/sender.go @@ -27,8 +27,13 @@ const ( dispatcherQueueSize = 10000 // dispatcherWorkers — горутин, отправляющих в SQS. dispatcherWorkers = 4 - // sendMaxAttempts — попыток SendMessage на одно сообщение. - sendMaxAttempts = 3 + + // sendAttemptTimeout — таймаут ОДНОЙ попытки SendMessage. Без него зависшее + // соединение (SQS висит, но не отвечает) блокировало бы воркера навсегда и + // забило бы весь пул (ревью 🔴4). + sendAttemptTimeout = 10 * time.Second + // sendMaxBackoff — потолок паузы между попытками (экспоненциальный backoff). + sendMaxBackoff = 8 * time.Second // maxSQSBodyBytes — потолок размера envelope для SQS. // SQS (и shared-sqs) режут сообщения >256KB — запас на оверхед. @@ -116,30 +121,34 @@ func (d *sqsDispatcher) worker(ctx context.Context, wid int) { } } -// sendWithRetry — SendMessage с backoff 1с/2с/4с. +// sendWithRetry — SendMessage с ретраями до остановки (ctx) и таймаутом на +// попытку. НЕ дропаем после фиксированного числа попыток (было 3 ≈ 7с): буфер +// на dispatcherQueueSize рассчитан на поглощение временной недоступности SQS, +// поэтому ретраим до ctx.Done() (shutdown) или успеха — иначе потеря была бы +// безвозвратной (ревью 🔴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{ + 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 } - lastErr = err if ctx.Err() != nil { - return lastErr + // shutdown: ctx отменён, ретраить бессмысленно. + return err } - if attempt < sendMaxAttempts { - select { - case <-ctx.Done(): - return lastErr - case <-time.After(backoff): - } + select { + case <-ctx.Done(): + return err + case <-time.After(backoff): + } + if backoff < sendMaxBackoff { backoff *= 2 } } - return lastErr }