fix(P1.2): утечка per-queue блокировки при ReceiveMessage на пустой очереди
Возврат до defer queue.UnlockMu() оставлял блокировку захваченной навсегда — очередь «зависала»: все последующие Lock/RLock ждали вечно. Проявлялось при LiveCount()==0 (все сообщения удалены/tombstone). Найдено стресс-тестом (2000 сообщений × циклы): стек горутин показал массовое ожидание RLockMu. - receive: defer до проверки пустоты. - регрессионный тест TestReceiveEmptyQueue_DoesNotLeakLock (падает на дефекте, проходит на фиксе).
This commit is contained in:
@@ -154,10 +154,16 @@ func ReceiveMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody)
|
|||||||
// - для FIFO-очереди блокируется группа (LockGroup) до DeleteMessage/таймаута.
|
// - для FIFO-очереди блокируется группа (LockGroup) до DeleteMessage/таймаута.
|
||||||
func receiveMessagesUnderLock(key string, visibilityTimeout, maxNumberOfMessages int) ([]*models.ResultMessage, bool) {
|
func receiveMessagesUnderLock(key string, visibilityTimeout, maxNumberOfMessages int) ([]*models.ResultMessage, bool) {
|
||||||
queue := models.SyncQueues.GetLocked(key)
|
queue := models.SyncQueues.GetLocked(key)
|
||||||
if queue == nil || queue.LiveCount() == 0 {
|
if queue == nil {
|
||||||
return nil, false
|
return nil, false
|
||||||
}
|
}
|
||||||
|
// ВАЖНО: defer — ДО любых return ниже. Иначе возврат на пустой очереди
|
||||||
|
// оставлял бы per-queue блокировку захваченной навсегда (очередь «зависала»:
|
||||||
|
// все последующие Lock/RLock ждали бы вечно).
|
||||||
defer queue.UnlockMu()
|
defer queue.UnlockMu()
|
||||||
|
if queue.LiveCount() == 0 {
|
||||||
|
return nil, false
|
||||||
|
}
|
||||||
// Быстрая предпроверка: есть ли вообще хотя бы одно готовое сообщение.
|
// Быстрая предпроверка: есть ли вообще хотя бы одно готовое сообщение.
|
||||||
if !queueHasReceivableMessages(queue) {
|
if !queueHasReceivableMessages(queue) {
|
||||||
return nil, false
|
return nil, false
|
||||||
|
|||||||
Reference in New Issue
Block a user