fix(iot): ретрай GetQueueUrl на старте + выход ≠0 при сбое подсистемы (🔴5)
- Добавлен ResolveQueueURLWithRetry (10 попыток × 2с backoff): временная ошибка GetQueueUrl больше не роняет bridge/consumer без ретрая. - main.go: при ПОСТОЯННОЙ ошибке bridge/consumer процесс завершается с кодом ≠ 0 (fatal-флаг + cancel), чтобы платформа перезапустила контейнер — раньше процесс оставался жив, а /health отдавал 200.
This commit is contained in:
@@ -16,6 +16,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -102,12 +103,20 @@ func main() {
|
|||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
wg.Add(2)
|
wg.Add(2)
|
||||||
|
|
||||||
|
// fatal — подсистема (bridge/consumer) завершилась с ПОСТОЯННОЙ ошибкой.
|
||||||
|
// Тогда после штатной остановки выходим с кодом ≠ 0, чтобы платформа
|
||||||
|
// перезапустила контейнер (ревью 🔴5: раньше процесс оставался жив, а
|
||||||
|
// /health отдавал 200).
|
||||||
|
var fatal atomic.Bool
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
ready.SetBridge(true)
|
ready.SetBridge(true)
|
||||||
defer ready.SetBridge(false)
|
defer ready.SetBridge(false)
|
||||||
if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) {
|
if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) {
|
||||||
log.Error("bridge stopped", "err", err)
|
log.Error("bridge stopped", "err", err)
|
||||||
|
fatal.Store(true)
|
||||||
|
cancel() // штатный shutdown, затем выход ≠ 0
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
@@ -117,6 +126,8 @@ func main() {
|
|||||||
defer ready.SetConsumer(false)
|
defer ready.SetConsumer(false)
|
||||||
if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) {
|
if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) {
|
||||||
log.Error("consumer stopped", "err", err)
|
log.Error("consumer stopped", "err", err)
|
||||||
|
fatal.Store(true)
|
||||||
|
cancel()
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
@@ -139,5 +150,9 @@ func main() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
wg.Wait()
|
wg.Wait()
|
||||||
|
if fatal.Load() {
|
||||||
|
log.Error("iot-service: подсистема завершилась с ошибкой — выход ≠ 0")
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
log.Info("iot-service stopped")
|
log.Info("iot-service stopped")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -28,7 +28,7 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *sl
|
|||||||
log.Info("bridge: starting", "broker", cfg.MQTTBrokerURL, "client_id", cfg.MQTTClientID)
|
log.Info("bridge: starting", "broker", cfg.MQTTBrokerURL, "client_id", cfg.MQTTClientID)
|
||||||
|
|
||||||
// URL очереди получаем один раз при старте (как старый consumer).
|
// URL очереди получаем один раз при старте (как старый consumer).
|
||||||
queueURL, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSQueueName)
|
queueURL, err := sqsclient.ResolveQueueURLWithRetry(ctx, sqsClient, cfg.SQSQueueName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -26,7 +26,7 @@ const maxProcessBackoff = 60 * time.Second
|
|||||||
|
|
||||||
// Run — бесконечный long-poll цикл потребителя.
|
// Run — бесконечный long-poll цикл потребителя.
|
||||||
func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *iotpg.IoTPostgresStore, log *slog.Logger) error {
|
func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *iotpg.IoTPostgresStore, log *slog.Logger) error {
|
||||||
queueURL, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSQueueName)
|
queueURL, err := sqsclient.ResolveQueueURLWithRetry(ctx, sqsClient, cfg.SQSQueueName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ package sqsclient
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/aws/aws-sdk-go-v2/aws"
|
"github.com/aws/aws-sdk-go-v2/aws"
|
||||||
"github.com/aws/aws-sdk-go-v2/credentials"
|
"github.com/aws/aws-sdk-go-v2/credentials"
|
||||||
@@ -36,3 +37,29 @@ func ResolveQueueURL(ctx context.Context, client *sqs.Client, queueName string)
|
|||||||
}
|
}
|
||||||
return aws.ToString(out.QueueUrl), nil
|
return aws.ToString(out.QueueUrl), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ResolveQueueURLWithRetry — ResolveQueueURL с ретраями на старте. Временная
|
||||||
|
// ошибка GetQueueUrl (SQS ещё поднимается, сетевой сбой) не должна фатально
|
||||||
|
// ронять bridge/consumer без ретрая (ревью 2026-10-10, 🔴5). После исчерпания
|
||||||
|
// попыток возвращает ошибку — вызывающий должен завершить процесс ≠ 0 (платформа
|
||||||
|
// перезапустит контейнер).
|
||||||
|
func ResolveQueueURLWithRetry(ctx context.Context, client *sqs.Client, queueName string) (string, error) {
|
||||||
|
const attempts = 10
|
||||||
|
const backoff = 2 * time.Second
|
||||||
|
var lastErr error
|
||||||
|
for attempt := 1; attempt <= attempts; attempt++ {
|
||||||
|
url, err := ResolveQueueURL(ctx, client, queueName)
|
||||||
|
if err == nil {
|
||||||
|
return url, nil
|
||||||
|
}
|
||||||
|
lastErr = err
|
||||||
|
if attempt < attempts {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return "", lastErr
|
||||||
|
case <-time.After(backoff):
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return "", lastErr
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user