From 946600238b089cda3fa4793b6945c629aecc49f3 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 10 Oct 2026 19:05:08 +0300 Subject: [PATCH] =?UTF-8?q?fix(iot):=20=D1=80=D0=B5=D1=82=D1=80=D0=B0?= =?UTF-8?q?=D0=B9=20GetQueueUrl=20=D0=BD=D0=B0=20=D1=81=D1=82=D0=B0=D1=80?= =?UTF-8?q?=D1=82=D0=B5=20+=20=D0=B2=D1=8B=D1=85=D0=BE=D0=B4=20=E2=89=A00?= =?UTF-8?q?=20=D0=BF=D1=80=D0=B8=20=D1=81=D0=B1=D0=BE=D0=B5=20=D0=BF=D0=BE?= =?UTF-8?q?=D0=B4=D1=81=D0=B8=D1=81=D1=82=D0=B5=D0=BC=D1=8B=20(?= =?UTF-8?q?=F0=9F=94=B45)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Добавлен ResolveQueueURLWithRetry (10 попыток × 2с backoff): временная ошибка GetQueueUrl больше не роняет bridge/consumer без ретрая. - main.go: при ПОСТОЯННОЙ ошибке bridge/consumer процесс завершается с кодом ≠ 0 (fatal-флаг + cancel), чтобы платформа перезапустила контейнер — раньше процесс оставался жив, а /health отдавал 200. --- cmd/iot-service/main.go | 15 ++++++++++++++ internal/service/bridge/bridge.go | 2 +- internal/service/consumer/consumer.go | 2 +- internal/service/sqsclient/sqsclient.go | 27 +++++++++++++++++++++++++ 4 files changed, 44 insertions(+), 2 deletions(-) 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 +}