你确切知道生产环境在哪里崩溃了。堆栈追踪指向 invoice_service.py 的第 147 行。异常是一个针对 "customer_id"KeyError。你拉下代码,运行测试,全部通过。你用一个示例 payload 手动命中端点,它工作正常。

Bug 是真实的。客户在触发它。但你就是没法在自己的机器上复现。

这是基于错误报告进行调试的标准体验。堆栈追踪告诉你”在哪里”,却不告诉你”什么到达了那里”。没有触发失败的确切输入,你就像是在根据一张粉笔画轮廓的照片重建犯罪现场。

崩溃复现到底意味着什么

崩溃复现是一种实践:捕获触发生产环境失败的完整输入状态,并在本地环境中用那个确切的状态重新执行代码路径。目标是把”它在第 147 行崩溃了”变成”这是一个每次都能让第 147 行崩溃的测试用例”。

大多数开发者已经在手动做这件事的弱化版。你读取堆栈追踪,猜测是哪个请求导致的,尝试从日志重建 payload,并指望你的本地数据库有类似的数据。它失败的原因和占星术失败的原因一样:你在信息不足的情况下匹配模式。

猜测和复现之间的区别是序列化。你必须在失败发生的瞬间捕获确切的函数输入、确切的数据库响应和确切的外部 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": "..."
}

第一个参数是一个空字典。这就是整个 bug。上游调用者传了 {} 而不是客户记录。

你现在有了一条测试用例:

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 或检查环境变量,这些值不在 argskwargs 中。复现只有在你的本地环境碰巧匹配时才会生效。

你可以扩展装饰器来显式捕获外部依赖:

@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),
    }

这就是 functional core, imperative shell。它让捕获变得轻而易举,因为没有隐藏状态。每个重要的输入都在参数列表里。

你无法通过捕获消除的非确定性问题

即使有完美的输入捕获,有些崩溃也是不可复现的。竞态条件依赖于线程时序。内存损坏依赖于分配器状态。外部 API 每次调用返回不同数据。随机数生成器除非你设置种子,否则产生不同序列。

如果你的崩溃是竞态条件,在单线程本地测试中重放相同输入不会触发它。你需要实际的并发模式,这意味着运行原始线程,也就是说你已经从”复现”变成了”分布式追踪加负载测试”。那是另一种工具。

对于大多数应用级别的 bug,输入复现就足够了。对于 heisenbugs,则不够。在花费三小时试图复现一个时序问题之前,先弄清楚你面对的是哪一种。

让捕获在生产环境中可运营

上面的装饰器写入本地磁盘。在生产环境中,你希望这些捕获被发送到对象存储或你的错误追踪器。集成很直接:把 open(capture_path, "w") 调用替换为 S3 上传或作为 Sentry issue 的附件。

大多数团队应该从一个函数开始。选那个崩溃最频繁的服务。加上装饰器。等待下一次崩溃。当它到来时,你会得到一个 JSON payload,把三十分钟的猜测会议变成五分钟的测试编写。

如果你已经在使用 Sentry,Breadcrumbs 功能会自动捕获部分这类上下文。对于自定义方案,这个模式只需三十行 Python,并能在任何支持装饰器或中间件的运行时中工作。

本周就给一个端点加上捕获

本周就给错误率最高的端点加上捕获。不是每个端点。不是每个函数。一个就够了。当它崩溃时,在修复 bug 之前先从捕获写出测试用例。运行测试,看着它失败,应用修复,看着它通过。

这个从生产环境崩溃到可复现测试用例的循环,就是调试与考古学之间的区别。