Почему зависает asyncio.Queue.join(): аудит вызовов taskdone

Аудит пропущенных подтверждений

Служба асинхронной обработки заказов не может штатно завершить смену: ожидание очереди зависает и никогда не разблокируется. Чтобы найти причину утечки, инспектор оборачивает очередь трассирующей прослойкой и непрерывно наблюдает за жизненным циклом каждого идентификатора. Прослойка точно фиксирует три события: момент постановки заказа в очередь, момент его фактического извлечения воркером и момент вызова подтверждения.

Если контрольное время ожидания завершения смены истекает, инспектор прерывает зависший процесс по таймауту. Все задачи, так и не получившие подтверждения, делятся на две непересекающиеся группы: заказы, которые воркеры уже успели забрать в работу, но так и не подтвердили из-за сбоев, и заказы, которые до сих пор остаются в теле очереди невостребованными.

Реализуйте функцию аудита конвейера, которая останавливает ожидание по таймауту и возвращает два раздельных списка идентификаторов — для извлечённых, но неподтверждённых заказов и для тех, что остались ждать в очереди.

Что уже дано

При выполнении кода доступен класс LifecycleQueue (наследник asyncio.Queue), который автоматически регистрирует операции в списке queue.events в виде кортежей ("put" | "get" | "done", identifier). Вам требуется реализовать корутину audit_ack_gaps.

Что нужно сделать

Реализуйте асинхронную функцию audit_ack_gaps(queue, timeout).

Параметры функции:

• queue — экземпляр LifecycleQueue, содержащий от 0 до 1000 уникальных хешируемых идентификаторов заказов, обработанных конечным числом воркеров (от 0 до 100). Вся работа продюсеров и воркеров к моменту аудита уже завершена, новых изменений в очереди не запланировано.
• timeout — положительное число секунд (от 0.001 до 1.0), задающее предельное время ожидания завершения очереди.

Требования к работе функции:

• Ожидайте вызов queue.join() в пределах заданного интервала timeout.
• Если время ожидания истекло, остановите ожидание и проанализируйте зафиксированный журнал queue.events.
• Функция должна вернуть пару (кортеж или список из двух элементов) списков идентификаторов (extracted_unacknowledged, still_waiting):

1. Первый список — идентификаторы заказов, которые были извлечены из очереди (событие "get"), но для которых так и не был вызван task_done (событие "done").
2. Второй список — идентификаторы заказов, которые были помещены в очередь (событие "put"), но ни разу не извлекались воркерами.

• Идентификаторы, для которых подтверждение состоялось, исключаются из обоих списков. Списки не должны пересекаться и не должны содержать дубликатов. Порядок элементов внутри списков произвольный.
• Если все задачи очереди подтверждены, функция возвращает два пустых списка ([], []). Сама по себе пустота очереди не гарантирует подтверждения всех заказов.
• Функция аудита не должна изменять содержимое очереди, список queue.events или количество вызовов task_done(), а также не должна отменять задачи или имитировать восстановление воркеров.

Ввод и вывод

Стандартные потоки ввода stdin и вывода stdout в решении не используются: считывать ввод и вызывать print() не нужно. После выполнения кода учащегося проверяющая система считывает из пространства имён модуля функцию верхнего уровня audit_ack_gaps и вызывает её с проверочными аргументами.

Пример использования

import asyncio
async def example():
    queue = LifecycleQueue()
    queue.put_nowait("order-1")
    return await audit_ack_gaps(queue, 0.01)
print(asyncio.run(example()))

Вывод:

([], ['order-1'])