同一用户，两种分组
用户早上在 Android 上把商品加入购物车，实验组是「新版结算页」；晚上换 iPhone 打开同一个账号，看到旧版结算页。两笔订单被记进两个分组，报告里却都算作「该用户」的贡献。
只要两端分流键不同、老版本还在用上一版实验盐、用户退出再登录后没有重算，或者重置后旧设备又上报缓存，就会发生。每个端单独看都「有分组」，放到一起才看得出矛盾。污染检测不能只看一条记录，必须把同一用户在多端、多版本上的记录放在一起看。
污染不是「数据脏了」，而是「归因断了」
污染的本质不是字段为空或格式错误，而是同一个分析单元在不同时间、不同端上接受了不同处理：效应被稀释、被平台差异放大，或者因为没有记录平台与版本而无法复现。因此记录必须带 platform、version、at 三个字段，否则检测函数无从下手。
六类污染信号
| 信号 | 触发条件 | 说明 |
|---|---|---|
variant-drift | 同一用户同一实验出现多个分组 | 最直接的污染 |
cross-platform | 漂移同时跨多个平台 | 双端分流键不一致 |
cross-version | 记录版本低于当前版本 | 老客户端仍用旧配置 |
mutex-overlap | 同一互斥组内命中多个实验 | 配置或分流顺序出错 |
residual | 最近记录超过阈值未刷新 | 本地缓存长期未更新 |
superseded | 存在早于重置时间的记录 | 重置后旧分组回流 |
信号不是错误日志，而是排查入口：一条 variant-drift 能告诉你是哪个用户、哪个实验、出现了哪几个分组。
互斥组与当前分组
互斥组解决「两个实验不能同时影响同一个用户」的问题。一个实验只能属于一个互斥组，重复登记直接抛错；同一用户在同一组内命中两个实验就是 mutex-overlap，与分组名是否相同无关。
同一用户有多条记录时，当前分组按三步确定：取时间最新的记录，时间相同取版本更高的；若存在不早于最新记录的重置标记，返回已重置；若最新时刻同时存在多个分组，标记冲突并返回 null。冲突时重新分流，比随便选一个更安全。
清理与重置
清理按年龄删除旧记录，是数据卫生；重置是业务动作，重置时间点之前的记录全部失效。两者不能混：把「例行清理」和「业务重置」写进同一个任务，容易误删刚重置后的新记录。
重置还要防回流。旧设备可能带着重置前的缓存重新上报，所以检测函数要识别「早于重置时间」的记录并给出 superseded 信号。只删数据库旧行不够，回流会把它写回来。
一个可运行的污染检测模块
下面的模块属于 L1：输入是记录、互斥组与重置标记，输出是信号列表，没有网络、没有定时器。
// file: src/experiment/interference.ts
// tier: L1
// verify: node --test test/experiment/interference.test.ts
export type Platform = "ios" | "android" | "web";
export interface AssignmentRecord {
unitId: string;
experiment: string;
variant: string;
platform: Platform;
version: number;
at: number;
}
export interface MutexGroup {
group: string;
experiments: string[];
}
export interface ResetMark {
unitId: string;
experiment: string;
at: number;
}
export type SignalCode =
| "variant-drift"
| "mutex-overlap"
| "cross-platform"
| "cross-version"
| "residual"
| "superseded";
export interface InterferenceSignal {
code: SignalCode;
unitId: string;
experiments: string[];
detail: string;
}
export interface VariantResolution {
variant: string | null;
conflict: boolean;
at: number | null;
source: "record" | "reset" | "none";
}
// 一个实验只能属于一个互斥组；重复登记属于配置错误，直接抛错。
export function groupIndex(groups: MutexGroup[]): Map<string, string> {
const index = new Map<string, string>();
for (const group of groups) {
if (group.group.length === 0) throw new Error("互斥组名不能为空");
for (const experiment of group.experiments) {
const previous = index.get(experiment);
if (previous !== undefined && previous !== group.group) {
throw new Error(
`实验 ${experiment} 同时属于互斥组 ${previous} 与 ${group.group}`,
);
}
index.set(experiment, group.group);
}
}
return index;
}
export function latestRecord(records: AssignmentRecord[]): AssignmentRecord | null {
let latest: AssignmentRecord | null = null;
for (const record of records) {
if (
latest === null ||
record.at > latest.at ||
(record.at === latest.at && record.version > latest.version)
) {
latest = record;
}
}
return latest;
}
function recordKey(record: AssignmentRecord): string {
return `${record.unitId}\u0000${record.experiment}`;
}
function sortedUnique(values: string[]): string[] {
return [...new Set(values)].sort();
}
function sortSignals(signals: InterferenceSignal[]): InterferenceSignal[] {
return signals.sort((left, right) => {
const leftKey = `${left.unitId}\u0000${left.code}\u0000${left.experiments.join(",")}`;
const rightKey = `${right.unitId}\u0000${right.code}\u0000${right.experiments.join(",")}`;
return leftKey < rightKey ? -1 : leftKey > rightKey ? 1 : 0;
});
}
// 六个信号覆盖：同实验分组漂移、跨端漂移、旧版本残留、互斥组重叠、
// 长时间未刷新、重置后仍有旧记录回流。
export function detectInterference(
records: AssignmentRecord[],
groups: MutexGroup[],
options: {
now?: number;
currentVersion?: number;
maxAgeMs?: number;
resets?: ResetMark[];
} = {},
): InterferenceSignal[] {
const index = groupIndex(groups);
const signals: InterferenceSignal[] = [];
const grouped = new Map<string, AssignmentRecord[]>();
for (const record of records) {
const key = recordKey(record);
grouped.set(key, [...(grouped.get(key) ?? []), record]);
}
for (const list of grouped.values()) {
const { unitId, experiment } = list[0];
const variants = sortedUnique(list.map((record) => record.variant));
const platforms = sortedUnique(list.map((record) => record.platform));
if (variants.length > 1) {
signals.push({
code: "variant-drift",
unitId,
experiments: [experiment],
detail: `${experiment} 出现 ${variants.join("/")} 多个分组`,
});
if (platforms.length > 1) {
signals.push({
code: "cross-platform",
unitId,
experiments: [experiment],
detail: `跨 ${platforms.join("/")} 出现分组漂移`,
});
}
}
if (options.currentVersion !== undefined) {
const outdated = list.filter((item) => item.version < options.currentVersion!);
if (outdated.length > 0) {
const oldest = Math.min(...outdated.map((item) => item.version));
signals.push({
code: "cross-version",
unitId,
experiments: [experiment],
detail: `存在 v${oldest} 记录，当前 v${options.currentVersion}`,
});
}
}
if (options.now !== undefined && options.maxAgeMs !== undefined) {
const latest = latestRecord(list)!;
if (options.now - latest.at > options.maxAgeMs) {
signals.push({
code: "residual",
unitId,
experiments: [experiment],
detail: `最近一次记录超过 ${options.maxAgeMs}ms 未刷新`,
});
}
}
if (options.resets !== undefined) {
const mark = options.resets
.filter((reset) => reset.unitId === unitId && reset.experiment === experiment)
.sort((left, right) => right.at - left.at)[0];
if (mark !== undefined && list.some((item) => item.at <= mark.at)) {
signals.push({
code: "superseded",
unitId,
experiments: [experiment],
detail: `存在早于重置时间 ${mark.at} 的旧记录`,
});
}
}
}
const byUnit = new Map<string, Set<string>>();
for (const record of records) {
if (!index.has(record.experiment)) continue;
const experiments = byUnit.get(record.unitId) ?? new Set<string>();
experiments.add(record.experiment);
byUnit.set(record.unitId, experiments);
}
for (const [unitId, experiments] of byUnit) {
const byGroup = new Map<string, string[]>();
for (const experiment of experiments) {
const group = index.get(experiment)!;
byGroup.set(group, [...(byGroup.get(group) ?? []), experiment]);
}
for (const [group, members] of byGroup) {
if (members.length > 1) {
signals.push({
code: "mutex-overlap",
unitId,
experiments: sortedUnique(members),
detail: `互斥组 ${group} 内同时命中 ${members.length} 个实验`,
});
}
}
}
return sortSignals(signals);
}
// 解析当前应生效的分组：重置标记优先，时间相同时并列分组视为冲突。
export function resolveVariant(
records: AssignmentRecord[],
unitId: string,
experiment: string,
resets: ResetMark[] = [],
): VariantResolution {
const list = records.filter(
(record) => record.unitId === unitId && record.experiment === experiment,
);
if (list.length === 0) {
return { variant: null, conflict: false, at: null, source: "none" };
}
const latest = latestRecord(list)!;
const mark = resets
.filter((reset) => reset.unitId === unitId && reset.experiment === experiment)
.sort((left, right) => right.at - left.at)[0];
if (mark !== undefined && mark.at >= latest.at) {
return { variant: null, conflict: false, at: mark.at, source: "reset" };
}
const sameInstant = list.filter((record) => record.at === latest.at);
const variants = sortedUnique(sameInstant.map((record) => record.variant));
return {
variant: variants.length === 1 ? latest.variant : null,
conflict: variants.length > 1,
at: latest.at,
source: "record",
};
}
export function planCleanup(
records: AssignmentRecord[],
now: number,
maxAgeMs: number,
): { keep: AssignmentRecord[]; drop: AssignmentRecord[] } {
if (!Number.isFinite(now) || !Number.isFinite(maxAgeMs) || maxAgeMs < 0) {
throw new Error("清理参数非法");
}
const keep: AssignmentRecord[] = [];
const drop: AssignmentRecord[] = [];
for (const record of records) {
(now - record.at > maxAgeMs ? drop : keep).push(record);
}
return { keep, drop };
}
// 重置会删除该用户在该实验上不晚于重置时间的记录；更晚的记录视为新事实。
export function applyReset(
records: AssignmentRecord[],
resets: ResetMark[],
): { records: AssignmentRecord[]; removed: number } {
const survivors = records.filter(
(record) =>
!resets.some(
(mark) =>
mark.unitId === record.unitId &&
mark.experiment === record.experiment &&
record.at <= mark.at,
),
);
return { records: survivors, removed: records.length - survivors.length };
}
三个设计点：记录分组以「用户 + 实验」为键，平台与版本只是属性；信号统一排序，保证同一批输入得到同一份报告；重置优先于记录，重置时间等于记录时间时也视为重置生效。
测试与真实输出
本机运行结果如下：
$ node --test test/experiment/interference.test.ts
✔ 一致的分组记录不产生信号 (1.056833ms)
✔ 同一实验出现两个分组时报告漂移 (0.150291ms)
✔ 跨端漂移同时报告跨端信号 (0.1465ms)
✔ 旧版本记录触发跨版本信号 (0.095375ms)
✔ 全部记录都是当前版本时不报跨版本 (0.08225ms)
✔ 同一互斥组内命中两个实验时报告重叠 (0.0965ms)
✔ 不同互斥组的实验可以同时命中 (0.068084ms)
✔ 实验重复登记到两个互斥组时拒绝建索引 (0.206542ms)
✔ resolveVariant 取时间最新的记录 (0.130167ms)
✔ 同一时刻出现两个分组时标记冲突 (0.100125ms)
✔ 重置标记晚于记录时不再返回分组 (0.071625ms)
✔ 没有记录时返回 none (0.070625ms)
✔ applyReset 删除不晚于重置时间的记录 (0.086125ms)
✔ planCleanup 按年龄拆分保留与丢弃 (0.082042ms)
✔ planCleanup 拒绝非法参数 (0.082167ms)
✔ 长时间未刷新触发 residual 信号 (0.073667ms)
✔ 重置后旧记录回流触发 superseded 信号 (0.071416ms)
✔ latestRecord 在时间相同时用版本决胜 (0.049625ms)
✔ 空输入返回空信号 (0.037667ms)
ℹ tests 19
ℹ suites 0
ℹ pass 19
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 138.312292
十九条用例覆盖：无信号、同端漂移、跨端漂移、跨版本、当前版本、互斥重叠、不同组并行、重复登记、最新记录、同刻冲突、重置优先、无记录、重置删除、年龄清理、非法参数、长期未刷新、重置回流、版本决胜、空输入。
一次双端哈希输入不一致的复盘
某交易客户端的 iOS 端用账号 id 分流，Android 端在未登录时用设备 id、登录后仍沿用设备 id。两端各自稳定，但同一账号在两台手机上算出了不同桶位。
大促期间问题集中暴露：用户在 Android 看到新版结算页，换 iPhone 看到旧版，两笔订单进入不同分组。报告显示「新版转化略高」，分平台拆开后，iOS 与 Android 方向相反。
复盘发现三个问题：两端分流键定义不同且没有共享测试向量；记录没有平台字段，报告无法拆开看；实验重置后旧设备仍带着缓存分组上报。
修复做了三件事：统一分流键为「登录账号 id，未登录持久化设备 id」；记录补上平台、版本与时间；检测侧增加跨端、跨版本与重置回流信号，看板按平台拆分。回归后，同一账号两端桶位一致，报告也能解释「为什么某个平台方向相反」。
这个复盘说明：污染往往不是某段代码写错，而是两端对「用户是谁」的回答不一致。 统一身份口径，比增加统计校正更有效。
小结
实验污染的核心是同一分析单元接受了不同处理。记录必须带平台、版本与时间；检测要覆盖分组漂移、跨端、跨版本、互斥重叠、长期未刷新与重置回流；当前分组按最新记录解析，重置标记优先；清理按年龄，重置按业务时间点。
下一章转向交易链路的另一端：客服与售后上下文如何组装。
阅读导航：上一节：10.2 实验配置的本地缓存与回退 · 下一节：11.1 客服上下文的组装 。
常见错误与边界
- 两端使用不同的分流键，账号与设备 id 混用，桶位天然不一致。
- 记录不保存平台与版本，出现漂移后无法定位是哪一端。
- 同一实验的多条记录只保留最后一条，历史漂移被覆盖掉。
- 互斥组只做 UI 提示，不做实际分流约束。
- 一个实验同时登记进多个互斥组，索引时静默覆盖。
- 同一时刻出现两个分组时随便取一个，把污染带进后续请求。
- 重置只删数据库旧行，不防旧设备缓存回流。
- 清理任务按固定天数删除，误删刚重置后的有效记录。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。