diff --git a/cmd/iot-service/main.go b/cmd/iot-service/main.go index 609f527..5c1febe 100644 --- a/cmd/iot-service/main.go +++ b/cmd/iot-service/main.go @@ -16,6 +16,7 @@ import ( "os" "os/signal" "sync" + "sync/atomic" "syscall" "time" @@ -102,12 +103,20 @@ func main() { var wg sync.WaitGroup wg.Add(2) + // fatal — подсистема (bridge/consumer) завершилась с ПОСТОЯННОЙ ошибкой. + // Тогда после штатной остановки выходим с кодом ≠ 0, чтобы платформа + // перезапустила контейнер (ревью 🔴5: раньше процесс оставался жив, а + // /health отдавал 200). + var fatal atomic.Bool + go func() { defer wg.Done() ready.SetBridge(true) defer ready.SetBridge(false) if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) { log.Error("bridge stopped", "err", err) + fatal.Store(true) + cancel() // штатный shutdown, затем выход ≠ 0 } }() @@ -117,6 +126,8 @@ func main() { defer ready.SetConsumer(false) if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) { log.Error("consumer stopped", "err", err) + fatal.Store(true) + cancel() } }() @@ -139,5 +150,9 @@ func main() { } wg.Wait() + if fatal.Load() { + log.Error("iot-service: подсистема завершилась с ошибкой — выход ≠ 0") + os.Exit(1) + } log.Info("iot-service stopped") } diff --git a/internal/service/bridge/bridge.go b/internal/service/bridge/bridge.go index 1c8d313..5d1995e 100644 --- a/internal/service/bridge/bridge.go +++ b/internal/service/bridge/bridge.go @@ -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) // URL очереди получаем один раз при старте (как старый consumer). - queueURL, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSQueueName) + queueURL, err := sqsclient.ResolveQueueURLWithRetry(ctx, sqsClient, cfg.SQSQueueName) if err != nil { return err } diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go index bec668e..6171b24 100644 --- a/internal/service/consumer/consumer.go +++ b/internal/service/consumer/consumer.go @@ -26,7 +26,7 @@ const maxProcessBackoff = 60 * time.Second // Run — бесконечный long-poll цикл потребителя. 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 { return err } diff --git a/internal/service/sqsclient/sqsclient.go b/internal/service/sqsclient/sqsclient.go index 72ff8b8..9ea9735 100644 --- a/internal/service/sqsclient/sqsclient.go +++ b/internal/service/sqsclient/sqsclient.go @@ -7,6 +7,7 @@ package sqsclient import ( "context" "fmt" + "time" "github.com/aws/aws-sdk-go-v2/aws" "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 } + +// 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 +}