浸泡测试（Soak Tests）

浸泡测试发现什么

浸泡测试（也称耐久测试）以稳定、真实且中等的负载运行数小时甚至数天。许多缺陷只在时间维度上出现：缓存不断增长却不淘汰，连接池每次失败请求泄漏一个连接，日志缓冲区填满磁盘，堆碎片化导致垃圾回收停顿变长。短时负载测试在这些效应成熟前就结束，因此报告健康数字，而生产环境在周末逐渐劣化。

负载曲线

使用接近生产的中等负载组合，而不是最大吞吐量；目标是持续时间而非峰值。要包含后台任务、定时任务和周期性重连，因为它们常触发泄漏。常见模式是爬升到预期负载、长时间保持平台期、短暂下降，并持续到跨越至少一个每日周期。与尖峰测试压测突变不同，浸泡测试压测的是持续性。

观测指标

跟踪内存和垃圾回收、打开的文件描述符、线程或工作进程数、连接池饱和度、磁盘使用以及随时间变化的延迟分位数。吞吐量持平而 p99 缓慢上升是典型的劣化信号。也要观察下游系统：浸泡测试可能耗尽共享数据库，或填满其他服务依赖的队列。应基于趋势告警，而不只是阈值。

持续时间与局限

没有通用时长。存在已知慢速泄漏的服务可能需要二十四小时，批处理管线可能需要多次运行。在重大发布前、依赖升级后以及按计划在生产类环境中运行浸泡测试，并让它远离快速反馈回路。浸泡测试昂贵且不能证明不存在泄漏。例如，只在罕见请求组合下出现的泄漏可能躲过一次干净运行。应把耐久测试与堆分析和生产内存告警结合。

代码示例

export class FakeClock {
  constructor(start = 0) {
    if (!Number.isFinite(start)) throw new TypeError("start must be a finite number");
    this.now = start;
    this.timers = [];
    this.seq = 0;
  }

  setTimeout(fn, delay) {
    if (!Number.isFinite(delay) || delay <= 0) {
      throw new RangeError("delay must be a positive number");
    }
    const entry = { id: ++this.seq, due: this.now + delay, fn };
    this.timers.push(entry);
    this.timers.sort((a, b) => a.due - b.due || a.id - b.id);
    return entry.id;
  }

  clearTimeout(id) {
    this.timers = this.timers.filter((t) => t.id !== id);
  }

  advance(ms) {
    if (!Number.isFinite(ms) || ms < 0) throw new RangeError("ms must be >= 0");
    const target = this.now + ms;
    for (;;) {
      const next = this.timers[0];
      if (!next || next.due > target) break;
      this.timers.shift();
      this.now = next.due;
      next.fn();
    }
    this.now = target;
  }

  get pending() {
    return this.timers.length;
  }
}

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Protocol


class Clock(Protocol):
    def now(self) -> datetime: ...


class SystemClock:
    def now(self) -> datetime:
        return datetime.now(timezone.utc)


class FakeClock:
    def __init__(self, start: datetime) -> None:
        if start.tzinfo is None:
            raise ValueError("start must be timezone aware")
        self._current = start

    def now(self) -> datetime:
        return self._current

    def advance(self, seconds: float) -> None:
        self._current += timedelta(seconds=seconds)


@dataclass
class CachedValue:
    clock: Clock
    ttl: timedelta
    value: str | None = None
    expires_at: datetime | None = None

    def get(self) -> str | None:
        now = self.clock.now()
        if self.value is None or self.expires_at is None or now >= self.expires_at:
            return None
        return self.value

    def put(self, value: str) -> None:
        self.value = value
        self.expires_at = self.clock.now() + self.ttl


def main() -> None:
    clock = FakeClock(datetime(2026, 10, 21, 12, 0, tzinfo=timezone.utc))
    cache = CachedValue(clock=clock, ttl=timedelta(seconds=30))
    cache.put("fresh")
    print("immediately:", cache.get())
    clock.advance(29)
    print("after 29s:", cache.get())
    clock.advance(2)
    print("after 31s:", cache.get())
    print("system clock tz:", SystemClock().now().tzinfo)


if __name__ == "__main__":
    main()