Как организовать тестирование конкурентного доступа к данным (гонки, потерянные обновления) в backend-сервисе, учитывая, что такие баги плохо воспроизводятся обычными последовательными тестами?

Короткий ответ

Нужно сочетать явные тесты конкурентного сценария на реальной тестовой БД (запуск нескольких параллельных операций над одной записью и проверка итогового состояния) с осознанным выбором механизма защиты (оптимистичная блокировка через version-поле, пессимистичная блокировка SELECT ... FOR UPDATE, сериализуемый уровень изоляции транзакций) — тест должен проверять именно тот механизм, который реализован, а не абстрактную «защищённость от гонок».

Как это работает подробнее

Тест запускает несколько потоков или процессов (или два отдельных соединения/сессии к одной тестовой БД), которые одновременно читают одну запись, изменяют её и пытаются сохранить, затем проверяет, что итоговое значение соответствует ожидаемому при выбранной стратегии защиты — например, что при оптимистичной блокировке один из конкурентных апдейтов явно завершается ошибкой конфликта версии, а не молча перезаписывает чужие изменения.

Сравнение

• Компромисс. Тесты с принудительной синхронизацией гонки дают высокую уверенность, но сложны в реализации и поддержке, специфичны к конкретной СУБД и уровню изоляции. Более простой вариант — полагаться на статический анализ и ревью кода критичных секций плюс мониторинг конфликтов в production — дешевле, но не даёт автоматической регрессионной защиты. Критерий выбора: для операций с прямой денежной или юридической ценой (списания, бронирования ограниченного ресурса) оправдан явный тест на гонку с синхронизацией; для менее критичных операций может быть достаточно явного ревью кода и мониторинга.

Пример кода

testing_example_326.pyКопировать

def test_optimistic_locking_detects_concurrent_update_conflict(db_engine):
    conn_a = db_engine.connect()
    conn_b = db_engine.connect()
    account_a = load_account(conn_a, account_id=1)  # version=1
    account_b = load_account(conn_b, account_id=1)  # version=1
    save_account(conn_a, account_a, new_balance=90)   # успешно, version становится 2
    with pytest.raises(StaleDataError):
        save_account(conn_b, account_b, new_balance=80)  # version устарела -> конфликт, а не тихая перезапись

Что использовать на практике

• Практика. Такие тесты по своей природе недетерминированы по времени выполнения гонки, поэтому часто используют искусственную синхронизацию (barriers, явные паузы в определённых точках через монкипатчинг или специальные hooks) — иначе гонка может просто не успеть произойти в большинстве прогонов, и тест будет ложно-зелёным, создавая иллюзию защищённости.

Подводные камни

• Риск. Писать «тест на гонку» без принудительной синхронизации потоков, который в 99% прогонов проходит просто потому, что операции физически не пересеклись по времени — такой тест не доказывает отсутствие проблемы. Путать уровень защиты в тесте с уровнем, реально используемым в production-коде (например, тестировать с SERIALIZABLE, когда в проде используется READ COMMITTED по умолчанию).