From 120a44cb6acd6eb8d1aaab487a3849862844ef6c Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sun, 11 Oct 2026 08:54:21 +0300 Subject: [PATCH] =?UTF-8?q?SQS:=20sendMessageToQueue=20=D0=BB=D0=B8=D0=BC?= =?UTF-8?q?=D0=B8=D1=82=D1=8B+sync=20persist,=20Reset=20=D1=87=D0=B8=D1=81?= =?UTF-8?q?=D1=82=D0=B8=D1=82=20FIFO,=20nil-map=20Duplicates,=20Approximat?= =?UTF-8?q?eReceiveCount=20=D0=B1=D0=B5=D0=B7=20+1,=20billing.Close=20?= =?UTF-8?q?=D0=BF=D0=BE=D1=81=D0=BB=D0=B5=20Shutdown,=20=D0=B7=D0=B0=D0=BF?= =?UTF-8?q?=D1=80=D0=B5=D1=82=20':'=20=D0=B2=20access=5Fkey,=20recover=20?= =?UTF-8?q?=D0=B2=20PeriodicTasks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/admin/admin.go | 20 +++++++++++++++----- app/cmd/goaws.go | 7 ++++--- app/gosqs/gosqs.go | 15 ++++++++++++--- app/gosqs/receive_message.go | 2 +- app/models/models.go | 9 +++++++++ app/tenant/tenant_store.go | 6 ++++++ 6 files changed, 47 insertions(+), 12 deletions(-) diff --git a/app/admin/admin.go b/app/admin/admin.go index 8977ee4..bc16db1 100644 --- a/app/admin/admin.go +++ b/app/admin/admin.go @@ -658,18 +658,28 @@ func (h *Handler) sendMessageToQueue(w http.ResponseWriter, r *http.Request) { jsonErr(w, http.StatusBadRequest, "body is required") return } - // Сообщение строится через общий билдер (UUID + MD5), + проверка размера — - // как в SQS SendMessage. + // Сообщение строится через общий билдер (UUID + MD5), + проверка размера и + // лимита — как в SQS SendMessage (иначе UI обходил бы OOM-защиту). q.LockMu() if q.MaximumMessageSize > 0 && len(req.Body) > q.MaximumMessageSize { q.UnlockMu() jsonErr(w, http.StatusBadRequest, "message too big") return } + if q.LiveCount() >= gosqs.MaxMessagesForQueue(q.IsFIFO) { + q.UnlockMu() + jsonErr(w, http.StatusTooManyRequests, "queue is full") + return + } msg := gosqs.BuildSqsMessage(req.Body) - q.AppendMessage(msg) - // Персистим одно сообщение отдельно - persistence.SaveMessage(key, &msg) + idx := q.AppendMessage(msg) + // Персистим СИНХРОННО (как SendMessage) — иначе 201 без гарантии записи. + if err := persistence.SaveMessageSync(key, &msg); err != nil { + q.RemoveIndex(idx) + q.UnlockMu() + jsonErr(w, http.StatusInternalServerError, "failed to persist message") + return + } persistence.SaveQueue(key, q) q.UnlockMu() w.Header().Set("Content-Type", "application/json") diff --git a/app/cmd/goaws.go b/app/cmd/goaws.go index 86abb4e..15daa85 100644 --- a/app/cmd/goaws.go +++ b/app/cmd/goaws.go @@ -257,9 +257,6 @@ func main() { elector.Stop() } - // Закрыть billing (если был подключён) - billing.Close() - // Дать 10 секунд на завершение текущих HTTP запросов ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -269,5 +266,9 @@ func main() { if err := controlSrv.Shutdown(ctx); err != nil { log.Errorf("Control-plane shutdown error: %v", err) } + + // Закрыть billing ПОСЛЕ Shutdown — иначе in-flight запросы писали бы usage + // в уже закрытую БД (ревью 2026-10-11). + billing.Close() log.Info("shared-sqs stopped") } diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go index c032c03..81df8a5 100644 --- a/app/gosqs/gosqs.go +++ b/app/gosqs/gosqs.go @@ -48,9 +48,18 @@ func PeriodicTasks(d time.Duration, quit chan bool) { // Каждую очередь обрабатываем под ОТДЕЛЬНЫМ Lock — блокировка одной // очереди не мешает обработке остальных и HTTP-запросам. for _, qi := range queues { - qi.queue.LockMu() - processQueueTick(qi.key, qi.queue) - qi.queue.UnlockMu() + func() { + // recover: паника в processQueueTick одной очереди не должна + // убивать весь процесс; mutex освобождается в defer. + defer qi.queue.UnlockMu() + defer func() { + if r := recover(); r != nil { + log.Errorf("gosqs: panic in processQueueTick for queue %q: %v", qi.key, r) + } + }() + qi.queue.LockMu() + processQueueTick(qi.key, qi.queue) + }() } case <-quit: ticker.Stop() diff --git a/app/gosqs/receive_message.go b/app/gosqs/receive_message.go index fe7974f..735cf2c 100644 --- a/app/gosqs/receive_message.go +++ b/app/gosqs/receive_message.go @@ -244,7 +244,7 @@ func buildResultMessage(m *models.SqsMessage) *models.ResultMessage { Attributes: map[string]string{ "ApproximateFirstReceiveTimestamp": fmt.Sprintf("%d", m.ReceiptTime.UnixNano()/int64(time.Millisecond)), "SenderId": models.CurrentEnvironment.AccountID, - "ApproximateReceiveCount": fmt.Sprintf("%d", m.NumberOfReceives+1), + "ApproximateReceiveCount": fmt.Sprintf("%d", m.NumberOfReceives), "SentTimestamp": fmt.Sprintf("%d", m.SentTime.UnixNano()/int64(time.Millisecond)), // Фикс: реальное время отправки }, } diff --git a/app/models/models.go b/app/models/models.go index 3e4649e..0977e86 100644 --- a/app/models/models.go +++ b/app/models/models.go @@ -183,6 +183,11 @@ func (q *Queue) InitDuplicatation(deduplicationId string) { return } + // Ленивая инициализация: очередь, собранная литералом или загруженная из + // Redis до появления поля, может иметь nil-map — иначе запись паникует. + if q.Duplicates == nil { + q.Duplicates = make(map[string]time.Time) + } if _, ok := q.Duplicates[deduplicationId]; !ok { q.Duplicates[deduplicationId] = time.Now() } @@ -344,6 +349,10 @@ func (q *Queue) Reset() { q.uuidIdx = make(map[string]int) q.liveCount = 0 q.deadCount = 0 + // Чистим FIFO-состояние: иначе блокировки групп и порядковые номера остались + // бы, и сообщения этих групп не выдавались после purge (ревью 2026-10-11). + q.FIFOMessages = make(map[string]int) + q.FIFOSequenceNumbers = make(map[string]int) } // EnsureIndexes — экспортируемая обёртка над ensureIndexes: строит индексы, diff --git a/app/tenant/tenant_store.go b/app/tenant/tenant_store.go index 9f72ea0..de34410 100644 --- a/app/tenant/tenant_store.go +++ b/app/tenant/tenant_store.go @@ -7,6 +7,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "strings" "sync" "time" @@ -121,6 +122,11 @@ func (s *TenantStore) CreateFixedAutoID(name string, maxQueues int, accessKey, s } func (s *TenantStore) CreateFixed(name string, maxQueues int, tenantID, accessKey, secretKey string) (*Tenant, error) { + // access_key используется как префикс ключа "{accessKey}:{queueName}" — + // ':' в нём ломает изоляцию тенантов (ListQueues/коллизии ключей). + if strings.Contains(accessKey, ":") { + return nil, fmt.Errorf("access key must not contain ':'") + } s.mu.Lock() if _, exists := s.byID[tenantID]; exists { s.mu.Unlock()