Asyncio Queue: объединение парных событий по идентификатору

Парный клиринг проводок

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

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

Спроектируйте конвейер парной обработки проводок, который удерживает подтверждение в очереди до сбора полной транзакции и исключает появление частичных записей в реестре балансов.

Что уже дано

В заготовке представлена сигнатура асинхронной функции match_transaction_pairs(queue, ledger, lock), которую вам предстоит реализовать.

В среде исполнения уже доступны вспомогательные структуры данных:

• класс TransactionHalf(transaction, account, delta) — половина проводки с идентификатором транзакции, счётом и целочисленным изменением баланса;
• класс BalanceLedger(initial_balances, lock) — реестр балансов, поддерживающий метод apply_pair(first, second).

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

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

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

• queue — готовый экземпляр asyncio.Queue, предварительно заполненный объектами TransactionHalf (от 0 до 1000 полных пар транзакций). Новые элементы в очередь в процессе работы не добавляются. Каждая транзакция строго состоит ровно из двух половин: одной с отрицательным и одной с равным положительным значением delta (по модулю от 1 до 1 000 000). Половины могут поступать в произвольном порядке, в том числе затрагивать один и тот же счёт;
• ledger — балансовый реестр BalanceLedger, совместимый с Mapping. Метод ledger.apply_pair(first, second) выполняет атомарное обновление балансов для обеих переданных половин (для отсутствующих счетов берётся нулевой начальный баланс);
• lock — предоставленный экземпляр asyncio.Lock (или его совместимый подкласс).

Правила обработки:

• Используйте одного агрегатора и локальный буфер ожидания, где ключом выступает идентификатор транзакции transaction. Не вызывайте подтверждение очереди для первой встреченной половины, пока не получена вторая.
• При поступлении парной половины примените обе части транзакции через ledger.apply_pair(first, second) строго под защитой предоставленного замка lock (корутина агрегатора должна удерживать замок в момент вызова).
• Только после применения пары в реестре вызовите queue.task_done() ровно два раза — по одному разу для каждой из двух половин транзакции. Подтверждения разрешено вызывать как внутри блока удержания замка, так и сразу после выхода из него.
• Дождитесь полного завершения обработки всех задач очереди с помощью await queue.join().
• Функция должна вернуть совместимый с Mapping отчёт с итоговыми балансами счетов (значения балансов должны быть целыми числами и совпадать с текущим состоянием ledger).
• Запрещено обрабатывать транзакции поодиночке без пары, выдумывать комиссии, таймауты или повторные попытки.

Ввод и вывод

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

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

import asyncio
async def example():
    queue = asyncio.Queue()
    queue.put_nowait(TransactionHalf("t", "source", -3))
    queue.put_nowait(TransactionHalf("t", "target", 3))
    lock = asyncio.Lock()
    ledger = BalanceLedger({"source": 10, "target": 0}, lock)
    return await match_transaction_pairs(queue, ledger, lock)
print(asyncio.run(example()))

Вывод:

{'source': 7, 'target': 3}