diff --git a/cmd/iot-service/main.go b/cmd/iot-service/main.go index 35a5233..2995e2c 100644 --- a/cmd/iot-service/main.go +++ b/cmd/iot-service/main.go @@ -131,6 +131,11 @@ func run() int { var wg sync.WaitGroup wg.Add(2) + // shutdownDone — закрывается после завершения server.Shutdown (дренаж + // in-flight HTTP-запросов). run() ждёт его перед выходом, чтобы deferred + // Close (iotStore/devices) не сработали, пока запросы ещё дренируются. + shutdownDone := make(chan struct{}) + // fatal — подсистема (bridge/consumer) завершилась с ПОСТОЯННОЙ ошибкой. // Тогда после штатной остановки выходим с кодом ≠ 0, чтобы платформа // перезапустила контейнер (ревью 🔴5: раньше процесс оставался жив, а @@ -175,6 +180,7 @@ func run() int { shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second) defer shutdownCancel() _ = server.Shutdown(shutdownCtx) + close(shutdownDone) }() log.Info("iot-service listening", "addr", server.Addr) @@ -182,9 +188,12 @@ func run() int { log.Error("http server", "err", err) cancel() // останавливаем bridge/consumer waitWithTimeout(&wg, shutdownGrace) + <-shutdownDone return 1 } + // Ждём завершения Shutdown (дренаж in-flight HTTP) до deferred Close. + <-shutdownDone if !waitWithTimeout(&wg, shutdownGrace) { log.Warn("iot-service: подсистемы не завершились за отведённое время", "grace", shutdownGrace) } diff --git a/internal/service/api/handler/devices.go b/internal/service/api/handler/devices.go index 0d9f5cf..dcf7b03 100644 --- a/internal/service/api/handler/devices.go +++ b/internal/service/api/handler/devices.go @@ -234,8 +234,11 @@ func (h *Handler) CreateIoTDevice(w http.ResponseWriter, r *http.Request) { // стабилизировать в REST для Terraform»). created, err := h.Devices.GetByName(r.Context(), ns, req.Name) if err != nil { - h.Log.Error("reload device after create", "namespace", ns, "name", req.Name, "err", err) - writeJSON(w, http.StatusInternalServerError, errResp("failed to reload device")) + // Устройство УЖЕ создано (INSERT прошёл): не отвечаем 500 — иначе пароль + // (отдаётся один раз) потерялся бы, а повтор дал бы 409. Отдаём созданный + // объект из памяти (created_at будет нулевым, уточнится следующим GET). + h.Log.Error("reload device after create (возвращаю in-memory без created_at)", "namespace", ns, "name", req.Name, "err", err) + writeJSON(w, http.StatusCreated, deviceToResponse(device, password)) return } diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go index 6c54538..68e21f0 100644 --- a/internal/service/consumer/consumer.go +++ b/internal/service/consumer/consumer.go @@ -136,6 +136,22 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * continue } log.Error("consumer: process telemetry", "err", err, "message_id", aws.ToString(msg.MessageId)) + // Ошибка обработки (не envelope, напр. PG). Применяем тот же лимит + // повторов, что и для битых envelope — иначе постоянная ошибка крутила + // бы сообщение бесконечно (visibility timeout → повтор → снова ошибка). + recv := receiveCount(msg) + if recv >= maxReceiveCount { + log.Error("consumer: dropped after maxReceiveCount", + "message_id", aws.ToString(msg.MessageId), "receive_count", recv, "err", err) + if dlqURL != "" { + if derr := sendToDLQ(ctx, sqsClient, dlqURL, body, log); derr == nil { + pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)}) + } + } else { + pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)}) + } + continue + } // Сообщение НЕ подтверждаем — вернётся после visibility timeout. // Backoff с jitter — не долбить упавший PG в цикле // (фикс HIGH из ревью 2026-08-16: лавина ретраев).