diff --git a/app/cmd/goaws.go b/app/cmd/goaws.go index 15daa85..a92c939 100644 --- a/app/cmd/goaws.go +++ b/app/cmd/goaws.go @@ -127,7 +127,7 @@ func main() { // Загружаем очереди queues, err := persistence.LoadAllQueues() if err != nil { - log.Warnf("Ошибка загрузки очередей из Redis: %v", err) + log.Fatalf("Ошибка загрузки очередей из Redis: %v", err) } else { models.SyncQueues.Lock() for k, q := range queues { diff --git a/app/gosqs/change_message_visibility.go b/app/gosqs/change_message_visibility.go index cee0deb..ee1f71c 100644 --- a/app/gosqs/change_message_visibility.go +++ b/app/gosqs/change_message_visibility.go @@ -74,7 +74,7 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo if msgRemoved { // Сообщение удалено (перемещено в DLQ) — удаляем из Redis исходной // очереди и сохраняем в DLQ, иначе после рестарта оно пропадёт из DLQ. - persistence.DeleteMessagePersist(key, changedMsgUuid) + _ = persistence.DeleteMessagePersist(key, changedMsgUuid) if dlqMsg != nil { persistence.SaveMessage(dlqKeyFor(key, queue.DeadLetterQueue.Name), dlqMsg) } diff --git a/app/gosqs/change_message_visibility_batch.go b/app/gosqs/change_message_visibility_batch.go index 64a1a4a..c6fba30 100644 --- a/app/gosqs/change_message_visibility_batch.go +++ b/app/gosqs/change_message_visibility_batch.go @@ -90,7 +90,7 @@ func ChangeMessageVisibilityBatchV1(req *http.Request) (int, interfaces.Abstract if moved { // Сообщение перенесено в DLQ и удалено из очереди — чистим Redis // исходной и сохраняем в DLQ (иначе пропадёт после рестарта). - persistence.DeleteMessagePersist(key, uuid) + _ = persistence.DeleteMessagePersist(key, uuid) if dlqMsg != nil { persistence.SaveMessage(dlqKeyFor(key, queue.DeadLetterQueue.Name), dlqMsg) } diff --git a/app/gosqs/create_queue.go b/app/gosqs/create_queue.go index 1378007..3ede598 100644 --- a/app/gosqs/create_queue.go +++ b/app/gosqs/create_queue.go @@ -59,6 +59,13 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { IsFIFO: utils.HasFIFOQueueName(queueName), EnableDuplicates: models.CurrentEnvironment.EnableDuplicates, Duplicates: make(map[string]time.Time), + // Дефолты (как в InitQueues): без них очередь из API получала бы + // VisibilityTimeout=0 и т.д. — сообщение сразу видимо, retention 0 = нет + // эвикции. Переданные атрибуты перезапишут дефолты в setQueueAttributesV1. + VisibilityTimeout: models.CurrentEnvironment.QueueAttributeDefaults.VisibilityTimeout, + MaximumMessageSize: models.CurrentEnvironment.QueueAttributeDefaults.MaximumMessageSize, + MessageRetentionPeriod: models.CurrentEnvironment.QueueAttributeDefaults.MessageRetentionPeriod, + ReceiveMessageWaitTimeSeconds: models.CurrentEnvironment.QueueAttributeDefaults.ReceiveMessageWaitTimeSeconds, // CreatedTimestamp/LastModifiedTimestamp в GetQueueAttributes (AWS их отдаёт). CreatedAt: time.Now().UTC(), LastModifiedAt: time.Now().UTC(), diff --git a/app/gosqs/delete_message.go b/app/gosqs/delete_message.go index a5c0ae0..4357b22 100644 --- a/app/gosqs/delete_message.go +++ b/app/gosqs/delete_message.go @@ -59,8 +59,12 @@ func DeleteMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { // ВАЖНО: запись в Duplicates НЕ удаляем — AWS держит dedup-окно 5 минут // независимо от удаления сообщения (иначе повторный send с тем же // dedupId доставлялся бы повторно). Чистит только истечение окна (PeriodicTasks). - // Удаляем одно сообщение из Redis — O(1) - persistence.DeleteMessagePersist(key, msg.Uuid) + // Удаляем одно сообщение из Redis — O(1). Ошибка → 500: иначе клиент + // получил бы 200, а после рестарта сообщение «воскресло» бы. + if err := persistence.DeleteMessagePersist(key, msg.Uuid); err != nil { + log.Errorf("DeleteMessage: не удалось удалить сообщение из Redis: %v", err) + return utils.CreateErrorResponseV1("InternalError", true) + } // Метаданные очереди (duplicates, FIFO state) persistence.SaveQueue(key, queue) respStruct := models.DeleteMessageResponse{ diff --git a/app/gosqs/delete_message_batch.go b/app/gosqs/delete_message_batch.go index 64af928..0abd990 100644 --- a/app/gosqs/delete_message_batch.go +++ b/app/gosqs/delete_message_batch.go @@ -110,8 +110,11 @@ func DeleteMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBo deletedUuids = append(deletedUuids, uuid) queue.RemoveIndex(i) } - // Удаляем сообщения из Redis отдельно — O(batch_size) - persistence.DeleteMessagesPersist(key, deletedUuids) + // Удаляем сообщения из Redis отдельно — O(batch_size). Ошибка → InternalError. + if err := persistence.DeleteMessagesPersist(key, deletedUuids); err != nil { + log.Errorf("DeleteMessageBatch: не удалось удалить сообщения из Redis: %v", err) + return utils.CreateErrorResponseV1("InternalError", true) + } // Метаданные очереди (duplicates, FIFO state) persistence.SaveQueue(key, queue) diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go index 81df8a5..9a69314 100644 --- a/app/gosqs/gosqs.go +++ b/app/gosqs/gosqs.go @@ -113,7 +113,7 @@ func processQueueTick(key string, queue *models.Queue) { if len(expired) > 0 { log.Infof("Queue [%s]: evicted %d messages by MessageRetentionPeriod", queue.Name, len(expired)) // Удаляем просроченные из Redis, чтобы не воскресли после рестарта. - persistence.DeleteMessagesPersist(key, expired) + _ = persistence.DeleteMessagesPersist(key, expired) } } @@ -163,7 +163,7 @@ func processQueueTick(key string, queue *models.Queue) { queue.RemoveIndex(i) // Персист переноса: иначе после рестарта сообщение «воскреснет» // в исходной очереди (не удалено из Redis) и пропадёт из DLQ. - persistence.DeleteMessagePersist(key, msgUuid) + _ = persistence.DeleteMessagePersist(key, msgUuid) persistence.SaveMessage(dlqKeyFor(key, dlqName), &dlqMsg) } } diff --git a/app/persistence/redis.go b/app/persistence/redis.go index 8de0839..4350ece 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -361,30 +361,32 @@ func SaveMessagesSync(queueKey string, msgs []models.SqsMessage) error { // Синхронность обязательна: если удаление потеряется при рестарте пода, // сообщение «воскреснет» из Redis (факт: удалённые очереди возвращались // после рестарта, когда удаления были асинхронными). -func DeleteMessagePersist(queueKey string, msgUuid string) { +func DeleteMessagePersist(queueKey string, msgUuid string) error { if Client.Load() == nil || msgUuid == "" { - return + return nil } hashKey := redisMsgHashPrefix + queueKey ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() if err := Client.Load().HDel(ctx, hashKey, msgUuid).Err(); err != nil { - log.Errorf("persistence: HDel message %s/%s: %v", queueKey, msgUuid, err) + return fmt.Errorf("HDel message %s/%s: %w", queueKey, msgUuid, err) } + return nil } // DeleteMessagesPersist — удаляет несколько сообщений из Redis СИНХРОННО // (иначе удалённые сообщения воскреснут после рестарта пода). -func DeleteMessagesPersist(queueKey string, uuids []string) { +func DeleteMessagesPersist(queueKey string, uuids []string) error { if Client.Load() == nil || len(uuids) == 0 { - return + return nil } hashKey := redisMsgHashPrefix + queueKey ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := Client.Load().HDel(ctx, hashKey, uuids...).Err(); err != nil { - log.Errorf("persistence: HDel messages %s (%d): %v", queueKey, len(uuids), err) + return fmt.Errorf("HDel messages %s (%d): %w", queueKey, len(uuids), err) } + return nil } // PurgeMessagesPersist — удаляет ВСЕ сообщения очереди из Redis СИНХРОННО. @@ -519,8 +521,9 @@ func LoadAllQueues() (map[string]*models.Queue, error) { hashKey := redisMsgHashPrefix + k msgRaw, err := Client.Load().HGetAll(ctx, hashKey).Result() if err != nil { - log.Errorf("persistence: HGetAll messages %q: %v", k, err) - continue + // Fail-fast: скрытая пустая очередь при недоступных сообщениях + // хуже явной ошибки — сообщения остались бы невидимыми в Redis. + return nil, fmt.Errorf("redis HGetAll messages %q: %w", k, err) } if len(msgRaw) > 0 { msgs := make([]models.SqsMessage, 0, len(msgRaw)) diff --git a/app/utils/utils.go b/app/utils/utils.go index a4bb288..8456e59 100644 --- a/app/utils/utils.go +++ b/app/utils/utils.go @@ -83,7 +83,11 @@ func ExtractQueueAttributes(u url.Values) map[string]string { // CreateErrorResponseV1 — формирует SQS error response по ключу ошибки. // Параметр isSqs оставлен для обратной совместимости (всегда true). func CreateErrorResponseV1(errKey string, isSqs bool) (int, interfaces.AbstractResponseBody) { - err := models.SqsErrors[errKey] + err, ok := models.SqsErrors[errKey] + if !ok { + // Неизвестный ключ → InternalError вместо nil-deref (ревью 2026-10-11). + err = models.SqsErrors["InternalError"] + } respStruct := models.ErrorResponse{ Result: err.Response(),