Вы точно знаете, где упал продакшен. Стектрейс указывает на строку 147 в invoice_service.py. Исключение — KeyError на "customer_id". Вы вытягиваете код, запускаете тесты, всё проходит. Вы бьёте вручную по эндпоинту с примером полезной нагрузки — работает.

Баг реальный. Клиенты на него натыкаются. Вы не можете воспроизвести его на своей машине.

Это стандартный опыт отладки по отчётам об ошибках. Стектрейсы говорят где. Они не говорят, что туда пришло. Без точных входных данных, вызвавших сбой, вы реконструируете место преступления по фотографии мела на асфальте.

Что на самом деле означает воспроизведение краша

Воспроизведение краша — это практика захвата полного входного состояния, вызвавшего продакшен-сбой, и повторного выполнения пути кода с этим точным состоянием в локальном окружении. Цель — превратить «упало на строке 147» в «вот тест-кейс, который заставляет строку 147 падать каждый раз».

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

Разница между угадыванием и воспроизведением — в сериализации. Вы должны захватить точные входные данные функции, точные ответы базы данных и точные возвращаемые значения внешних API в момент сбоя. Затем подать их обратно.

Как захватывать продакшен-входные данные для воспроизведения

Простейший эффективный паттерн — это декоратор, который перехватывает аргументы функции, сериализует их на диск и отправляет куда-то, куда вы позже сможете получить доступ. Когда функция падает, у вас есть застывший снимок мира, который его произвёл.

Вот рабочая реализация на Python:

import json
import functools
import traceback
from pathlib import Path
from datetime import datetime, timezone

CAPTURE_DIR = Path("/var/crash-captures")


def capture_for_replay(func):
    """Decorator that captures inputs and outputs for replay debugging."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        capture = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "function": func.__qualname__,
            "module": func.__module__,
            "args": args,
            "kwargs": kwargs,
            "exception": None,
            "traceback": None,
        }

        try:
            result = func(*args, **kwargs)
            capture["result"] = result
            return result
        except Exception as exc:
            capture["exception"] = {
                "type": type(exc).__name__,
                "message": str(exc),
            }
            capture["traceback"] = traceback.format_exc()

            # Write crash capture to disk
            CAPTURE_DIR.mkdir(parents=True, exist_ok=True)
            filename = (
                f"{func.__name__}_"
                f"{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.json"
            )
            capture_path = CAPTURE_DIR / filename

            with open(capture_path, "w") as f:
                json.dump(capture, f, default=str, indent=2)

            raise  # Re-raise so normal error handling continues

    return wrapper

Примените его к функции, которая падает:

@capture_for_replay
def generate_invoice(customer_data: dict, line_items: list) -> dict:
    customer_id = customer_data["customer_id"]  # This is line 147
    # ... rest of invoice logic
    return {"invoice_id": "INV-123", "total": 0}

Когда KeyError: "customer_id" срабатывает в продакшене, вы получаете JSON-файл, который выглядит так:

{
  "timestamp": "2026-08-15T14:32:11+00:00",
  "function": "generate_invoice",
  "module": "billing.invoice_service",
  "args": [
    {},
    [{"sku": "PRO-1", "price": 99.0}]
  ],
  "kwargs": {},
  "exception": {
    "type": "KeyError",
    "message": "'customer_id'"
  },
  "traceback": "..."
}

Первый аргумент был пустым словарём. Вот и весь баг. Вызывающий наверху передал {} вместо записи о клиенте.

Теперь у вас есть тест-кейс:

def test_generate_invoice_with_empty_customer():
    with pytest.raises(KeyError, match="customer_id"):
        generate_invoice({}, [{"sku": "PRO-1", "price": 99.0}])

Этот тест падает до фикса и проходит после добавления валидации входных данных. Что важнее, вам не пришлось угадывать. Краш точно сказал, что тестировать.

Почему захват пропускает зависимости, которые он не видит

Этот паттерн захватывает аргументы функции, а не глобальное состояние. Если generate_invoice читает из базы данных, вызывает внешний API или проверяет переменную окружения, этих значений нет в args и kwargs. Воспроизведение сработает только если ваше локальное окружение случайно совпадает.

Вы можете расширить декоратор, чтобы явно захватывать внешние зависимости:

@capture_for_replay
def generate_invoice(
    customer_data: dict,
    line_items: list,
    db_conn
) -> dict:
    tax_rate = db_conn.execute(
        "SELECT rate FROM tax_rates WHERE region = ?",
        (customer_data["region"],)
    ).fetchone()[0]
    # ...

Но db_conn — это объект соединения. Вы не можете сериализовать его в JSON. То, что вы можете сериализовать, — это запрос и результат. Более принципиальный подход — отделить чистую логику от побочных эффектов. Передавайте результат запроса как аргумент, а не соединение:

@capture_for_replay
def generate_invoice(
    customer_data: dict,
    line_items: list,
    tax_rate: float
) -> dict:
    # Pure function. All inputs are serializable.
    total = sum(item["price"] for item in line_items)
    total_with_tax = total * (1 + tax_rate)
    return {
        "invoice_id": "INV-123",
        "total": round(total_with_tax, 2),
    }

Это функциональное ядро, императивная оболочка. Это делает захват тривиальным, потому что нет скрытого состояния. Каждый вход, который имеет значение, находится в списке аргументов.

Проблема недетерминизма, которую вы не захватите

Даже с идеальным захватом входных данных некоторые краши невоспроизводимы. Состояния гонки зависят от тайминга потоков. Повреждение памяти зависит от состояния аллокатора. Внешние API возвращают разные данные при каждом вызове. Генераторы случайных чисел выдают разные последовательности, если вы не задаёте seed.

Если ваш краш — это состояние гонки, воспроизведение тех же входных данных в однопоточном локальном тесте не вызовет его. Вам нужен реальный паттерн конкурентности, а это означает запуск оригинальных потоков, а это означает, что вы перешли от «воспроизведения» к «распределённому трейсингу плюс нагрузочное тестирование». Это другой инструмент.

Для большинства багов уровня приложения входное воспроизведение достаточно. Для гейзенбагов — нет. Знайте, с чем вы имеете дело, прежде чем потратить три часа на попытку воспроизвести проблему тайминга.

Делаем захват рабочим в продакшене

Декоратор выше пишет на локальный диск. В продакшене вы хотите, чтобы эти захваты отправлялись в объектное хранилище или ваш трекер ошибок. Интеграция проста: замените вызов open(capture_path, "w") на загрузку в S3 или вложение к вашей задаче в Sentry.

Большинство команд должны начать с одной функции. Выберите сервис, который падает чаще всего. Добавьте декоратор. Дождитесь следующего краша. Когда он придёт, у вас будет JSON-полезная нагрузка, которая превращает тридцатиминутную сессию угадывания в пятиминутное написание теста.

Если вы уже используете Sentry, функция Breadcrumbs захватывает часть этого контекста автоматически. Для кастомного подхода паттерн умещается в тридцать строк Python и работает в любом рантайме, который поддерживает декораторы или middleware.

Захватите один эндпоинт на этой неделе

Добавьте захват к вашему эндпоинту с наибольшим числом ошибок на этой неделе. Не каждый эндпоинт. Не каждую функцию. Один. Когда он упадёт, напишите тест-кейс из захвата до того, как исправлять баг. Запустите тест, посмотрите, как он падает, примените фикс, посмотрите, как он проходит.

Этот цикл — от продакшен-краша до воспроизводимого тест-кейса — вот что отличает отладку от археологии.