mtfyun/inkwatch
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(落库): 重试槽改为多轮合并，短暂抖动不再"每 2 秒丢一批"（E2 复核改判）

Browse Source 
## 改判内容（team-lead 复核结论）
上一版（58f82a6）在"重试槽已被占用"时把新失败批**直接丢弃**，只让最早那一批
享有重试特权。幂运算一下就看出问题：PG 抖动 10 秒（2 秒一轮 → 5 轮）时，
轮 1 批 A 进槽（有救），轮 2/3/4 的批 B/C/D 全丢，轮 5 槽 A 超限丢、批 E 也丢 ——
**10 秒抖动只救回一批**，等于把"每批都丢"降级成"每 2 秒丢一批"，
而短暂抖动恰恰是重试最有效的场景。故改为**合并进槽**。

## 实现
- 槽的数据形态不变（`{ batch, attempts, lastError }`），但 `batch` 现在是
  **多轮合并累积**的帧数组：失败时一律 `concat`，不再"占用即丢弃"。
- **有界性靠槽容量，不靠丢新批**：合并后若超过 `_storageQueueMax`（默认 5000，
  与主队列同一上界）→ **截断**，保留**较早**的帧（先保已经等了更久的数据），
  超出部分走 `_discardStorageBatch(..., '超出重试槽上限 N 帧')`：计入
  `_storageFailed` + 告警，**明确丢弃不静默**。于是重试槽内存上界 = 主队列上界，
  重试不会让内存膨胀。
- `attempts` 语义改为**槽的重试轮次**（每轮 flush +1，首个失败轮记为 1），
  不再是"某一批的尝试次数"。超过 `_storageMaxRetries` → 整槽丢弃。
- 顺序不变（上一版写对的地方保留）：**先试槽 → 无论成败都照常处理主队列**，
  旧批失败绝不阻塞新数据；不在同一次 flush 内忙等重试。

## 附带发现：默认重试窗也要放宽（这一条请复核）
改完合并后我用临时量具实测"PG 抖动 10 秒"（5 轮 × 4 帧 = 20 帧，第 6 轮恢复）：
- 默认 `storageMaxRetries = 3` → 只救回 **8/20** 帧（第 4 轮 ≈8 秒整槽被丢，丢 12 帧）
- 因为合并后 attempts 记的是**轮次**，上限 3 意味着"只覆盖 4 轮 ≈ 8 秒"，
  与"要覆盖一次真实抖动"的意图不匹配。
故默认值改为 **9 轮 ≈ 20 秒窗口**（2 秒/轮 × (1+9)）。放宽不增加内存：槽容量由
`_storageQueueMax` 硬约束，窗口长短只影响"抖动期间攒多久"。改后同一场景
**20/20 全部救回、`storage_failed` = 0**。上限机制本身仍由测试 1（显式传
`storageMaxRetries: 3`）钉住。

三个版本在同一量具下的实测对照：
| 版本 | 落库 | storage_failed |
|---|---|---|
| 最初（只 log） | 0/20 | 字段不存在（静默） |
| 58f82a6（槽占用即丢） | 4/20 | 16 |
| 本次（合并 + 20 秒窗） | **20/20** | **0** |

## 测试（tests/storage-failure-test.js：29 → 41 项，全绿）
新增：
- **3b 跨 flush 合并**：3 个周期各失败 3/4/2 帧 → 槽内 **9 帧**、`attempts` = **3**
  （不是 9）；第 4 轮 PG 恢复 → **9 帧全部落库**、seq 1..9 顺序完整、`_storageFailed` = 0
- **3c 合并仍受容量约束**：槽上限 10，连续失败凑 15 帧 → 槽里只留 10 帧、
  另 5 帧计入 `_storageFailed`、**只告警一次**、保留的是较早帧（1..10）
- **3d 抖动 10 秒（默认窗口）**：5 轮 20 帧全攒槽、恢复后 20 帧一次落库
原有 5 条全绿（含"失败不阻塞新数据"毒丸隔离、重试成功不丢数据、计数/告警节奏、
队列满丢最旧语义不变、`drainStorage` 刷净）。
另：把 1/3/4 段的重试上限**显式**传 3，让"上限机制"与"抖动窗口"两件事在测试里
各自独立可读。

## 变异自证
把失败分支改回旧实现（只 `log.error`、批次消失）→ **41 项里 27 项红**：
第 1 段 8 项全红、第 2 段 2 项、第 3 段 2 项、**3b 4 项、3c 5 项、3d 2 项**、
第 4 段 3 项、`drainStorage` 1 项；恢复后 41/41 绿。
（顺带加固了 3b/3c 的空槽判定为 `?.`，让回归时"报红"而不是抛异常中断整个套件。）

## 三侧数字
- GATEWAY_PORT=3107 bash tests/run-all.sh：**29 通过 / 1 失败**（30 套件；唯一红仍是隐私扫描，未触碰）
- cargo test --release：**61 / 0**
- cd client && flutter test：**485 / 0**
This commit is contained in:
PMon Dev committed 2026-10-27 13:29:45 +08:00
1 parent 748785bb8e
commit f543739b8d
3 files changed
+170 -37
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/gateway/gateway.js
+74 -32
View File
@@ -52,17 +52,28 @@ class Gateway {
// ① 主队列满时 `_enqueueForStorage` 会 shift 掉**队首**——把失败批次放回去，
// 等于让"最该被救的旧数据"变成"最该被丢的旧数据"，两套语义直接打架；
// ② 毒丸（某帧的值始终违反约束）会永久占据队首，把后面所有新数据一起堵死。
// 故：失败批次单独存放、每轮 flush 只试一次（不忙等，否则会拖住 2 秒定时器），
// 且**每轮先试它、再照常处理主队列** —— 旧批失败绝不阻塞新数据。
// 槽内是**多轮合并**后的一批（`_mergeIntoRetry`）：短暂抖动期间各轮失败的批次
// 全并进来，PG 一恢复就一次写回 —— 否则只让"最早那一批"享有重试特权，
// 10 秒抖动只救回 1/5 批，等于把"每批都丢"降级成"每 2 秒丢一批"。
// 每轮 flush 只试一次（不忙等，否则会拖住 2 秒定时器），且**每轮先试它、
// 再照常处理主队列** —— 旧批失败绝不阻塞新数据。
/** @type {{batch: object[], attempts: number, lastError: string}|null} */
this._storageRetry = null;
// 超过重试上限、被**明确丢弃**的帧数（与 _storageDropped 的区别见 _discardStorageBatch）
// 超过重试上限 / 超出槽容量、被**明确丢弃**的帧数（与 _storageDropped 的区别见 _discardStorageBatch）
this._storageFailed = 0;
this._storageFailedBatches = 0;
// 上次告警时的累计丢弃帧数（告警节奏：首丢必报，之后每 100 帧一条）
this._storageFailedAlertedAt = 0;
// 单批最多重试几次（不含首次尝试）。2 秒一轮 → 默认覆盖约 8 秒的 PG 抖动
this._storageMaxRetries = options.storageMaxRetries ?? 3;
// 槽最多重试几轮（含首个失败轮）。2 秒一轮 → 默认 1+9 ≈ **20 秒抖动窗口**。
//
// 为什么默认从 3 放宽到 9（实测驱动）：合并语义下 attempts 记的是**轮次**，
// 上限 3 意味着第 4 轮（约 8 秒）就整槽丢弃 —— 用 /tmp 量具跑"PG 抖动 10 秒"
// （5 轮各 4 帧）实测：上限 3 时只救回 8/20 帧（第 4 轮丢掉前 12 帧），
// 上限 9 时 20/20 全部救回。
// 放宽不会增加内存：槽的容量由 `_storageQueueMax`（与主队列同量级）硬约束，
// 窗口长短只影响"抖动期间攒多久"，不影响上界。长时间不可用仍会攒满后
// 明确丢弃并计数告警 —— 有界与可见性都不破。
this._storageMaxRetries = options.storageMaxRetries ?? 9;
// 查看端账号 → 编码偏好（Redis 持久，跨实例共享；进程内另有旁路缓存）
// 以及「已向该账号下发过 msgpack schema 首帧」的标记，避免重复补发。
@@ -1019,28 +1030,30 @@ class Gateway {
* （`_storageDropped` + 周期 warn），这条是裸的。
*
* 现在的语义：
* ① 失败批次进**独立重试槽**（不放回主队列，理由见构造函数注释）；
* ① 失败批次**合并进独立重试槽**（不放回主队列，理由见构造函数注释）；
* 合并而非"槽被占用就丢新批"—— 否则 10 秒抖动里只有最早一批有救
* （详见 `_mergeIntoRetry` 的幂运算）；
* ② 每轮 flush **先试重试槽，再无论如何都照常处理主队列** ——
* 旧批失败绝阻塞新数据；
* ③ 有界重试：attempts 超过 _storageMaxRetries → **明确丢弃**、计数、告警
* （宁可"明确丢弃并记录"，也不要"静默消失"）；
* ③ 有界：槽的轮次超过 `_storageMaxRetries`，或合并后超出 `_storageQueueMax`
* → **明确丢弃**、计数、告警（宁可"明确丢弃并记录"，也不要"静默消失"）；
* ④ 不在同一次 flush 里忙等重试 —— 那会把 2 秒定时器拖住，
* 让"暂时抖动"自己滚成"持续积压"。
*/
async _flushStorage() {
if (!this.pgEnabled) return;
// ① 先试重试槽（若在有限重试窗口内）
if (this._storageRetry) {
// ① 先试重试槽（槽里是**多轮合并**后的一批，见 _mergeIntoRetry）
if (this._storageRetry && this._storageRetry.batch.length > 0) {
const slot = this._storageRetry;
slot.attempts++; // 槽的重试轮次（不区分具体是哪一批）
try {
await this.repo.insertMetricsBatch(slot.batch);
this._storageRetry = null; // 补上了，缺口填平
this._storageRetry = null; // 全救回，缺口填平
} catch (e) {
slot.attempts++;
slot.lastError = e.message;
if (slot.attempts > this._storageMaxRetries) {
this._discardStorageBatch(slot);
this._discardStorageBatch(slot, '超过重试上限');
this._storageRetry = null;
}
}
@@ -1052,25 +1065,54 @@ class Gateway {
try {
await this.repo.insertMetricsBatch(batch);
} catch (e) {
if (this._storageRetry) {
// 重试槽里已经压着一批（它还在窗口内）—— 新失败的这批不能挤掉它
// （挤掉等于"更早的数据先死"），也不能攒着（内存必须立刻有界），
// 故按明确的丢弃处理：计数 + 告警，不静默。
this._discardStorageBatch({
batch,
attempts: this._storageMaxRetries + 1,
lastError: e.message,
});
} else {
this._storageRetry = { batch, attempts: 1, lastError: e.message };
this.log.warn?.(`[gateway] 指标批量落库失败（${batch.length} 帧），`
+ `进入有限重试（最多 ${this._storageMaxRetries} 次）: ${e.message}`);
}
this._mergeIntoRetry(batch, e.message);
}
}
/**
* **明确丢弃**一批落库失败的数据（超过重试上限，或重试槽已被占用）。
* 把一批写库失败的数据**合并进重试槽**（E2 复核后改判：合并，而不是丢弃）。
*
* ── 为什么必须合并（幂运算一下就看出来）──────────────────────────
* 若"槽已被占用就把新失败批直接丢掉"，PG 抖动 10 秒（2 秒一轮 → 5 轮）时：
* 轮1 批A进槽（有救）；轮2/3/4 批B/C/D 全丢；轮5 槽A超限丢、批E 也丢。
* → 10 秒抖动只救回一批，等于把"每批都丢"降级成"每 2 秒丢一批"，
* 而**短暂抖动恰恰是重试最有效的场景** —— 那样改等于没解决 E2。
* 合并后：5 轮全并进槽里，PG 一恢复就一次写回（受下面的容量上限约束）。
*
* ── 有界性不受影响 ────────────────────────────────────────────────
* 合并后超过 `_storageQueueMax` 就**截断**（保留较早的，丢尾部）：
* 超出部分计入 `_storageFailed` 并按同一套口径告警 —— 明确丢弃，不静默。
* 于是重试槽的内存上界 = 主队列上界（5000 帧），不会因为重试而膨胀。
*
* `attempts` 语义：槽的**重试轮次**（每轮 flush +1），不是帧数、也不按批分。
*/
_mergeIntoRetry(batch, lastError) {
if (!this._storageRetry) {
// 首次失败：这一轮已经试过了，故轮次从 1 起算
this._storageRetry = { batch: batch.slice(), attempts: 1, lastError };
this.log.warn?.(`[gateway] 指标批量落库失败（${batch.length} 帧），进入有限重试`
+ `（最多 ${this._storageMaxRetries} 轮，槽上限 ${this._storageQueueMax} 帧）: ${lastError}`);
} else {
// 合并：旧批仍在有限重试窗口内，新失败批**并入**而不是挤掉/丢弃
this._storageRetry.batch = this._storageRetry.batch.concat(batch);
this._storageRetry.lastError = lastError;
}
const slot = this._storageRetry;
const cap = this._storageQueueMax;
if (slot.batch.length <= cap) return;
// 超出容量：截断（保留较早的），超出部分明确丢弃并计数
const overflow = slot.batch.slice(cap);
slot.batch = slot.batch.slice(0, cap);
this._discardStorageBatch(
{ batch: overflow, attempts: slot.attempts, lastError },
`超出重试槽上限 ${cap} 帧`
);
}
/**
* **明确丢弃**一批落库失败的数据（超过重试上限，或合并后超出槽容量）。
*
* 为什么丢弃本身不可避免：内存必须有界，PG 长时间不可用时只能舍 ——
* 但**必须留痕**。这条方法就是"不许静默消失"的落地：
@@ -1078,10 +1120,10 @@ class Gateway {
* `/health.storage_failed` 直接暴露它，>0 就意味着"真的丢了数据"；
* · 首丢必报一条 warn，之后每累计 100 帧一条（对齐 `_storageDropped`
* 的节奏，避免持续故障时每 2 秒刷一条把日志淹了）；
* · 告警里带帧数、重试次数、错误原文、以及前几帧的 `device_id@seq` ——
* 事后能判断"丢了哪台设备的哪一段"，才知道要不要补。
* · 告警里带帧数、丢弃原因、已尝试轮次、错误原文、以及前几帧的
* `device_id@seq` —— 事后能判断"丢了哪台设备的哪一段"，才知道要不要补。
*/
_discardStorageBatch(slot) {
_discardStorageBatch(slot, why = '超过重试上限') {
const n = slot.batch.length;
this._storageFailed += n;
this._storageFailedBatches++;
@@ -1096,7 +1138,7 @@ class Gateway {
.map((f) => `${f.device_id}@${f.seq}`)
.join(',');
this.log.warn?.(`[gateway] 指标落库**丢弃** ${n} 帧`
+ `（累计 ${this._storageFailed} 帧 / ${this._storageFailedBatches} 批）：`
+ `（${why}；累计 ${this._storageFailed} 帧 / ${this._storageFailedBatches} 批）：`
+ `已尝试 ${slot.attempts} 次 ~ ${slot.lastError}；样本 ${sample}`);
}
tests/run-all.sh
+2 -2
View File
@@ -28,7 +28,7 @@
# 中继 30（四道闸 + 真传不落盘 + 慢传不误删 + 限速定时器不叠加）、
# 会话索引 50（O(1) 寻址与 sessions 表同生共死 + 角色翻转摘索引）、
# 登录限流兜底有界 22（防"限流器自己变成泄漏点"+ 喷洒不冲掉热点 key）。
# 落库失败不丢批 29（E2）：用注入 repo 跑，无需服务；末尾的 /health 校验
# 落库失败不丢批 41（E2）：用注入 repo 跑，无需服务；末尾的 /health 校验
# 需要 PG + Redis，起不来则明确跳过（不影响其余断言）。
#
# 用法：bash tests/run-all.sh
@@ -106,7 +106,7 @@ run "设备身份判据（23 项）" node device-identity-test.js
# 落库失败不许静默丢整批（E2）：旧实现一次失败就丢掉整批（最多 5000 帧），
# 不留计数、不重试、/health 全绿。本套件用注入 repo 验证有限重试槽、毒丸隔离、
# 丢弃计数与告警节奏；末尾再起一个真网关确认 /health 的 storage_failed 可读。
run "落库失败不丢批（29 项）" node storage-failure-test.js
run "落库失败不丢批（41 项）" node storage-failure-test.js
# Rust 侧：单测 + 真实加载编译产物的 FFI 冒烟
run "Rust 核心单测（59 项）" cargo test --release --manifest-path ../rust-core/Cargo.toml
tests/storage-failure-test.js
+94 -3
View File
@@ -173,8 +173,9 @@ async function main() {
/* ─────────── ① 有界重试 → 明确丢弃 ─────────── */
console.log('\n=== 1. 有界重试：失败批次不消失，超限后明确丢弃并计数 ===');
{
// 显式把上限压到 3，专测"有界"这件事本身（默认 9 是抖动窗口，见 gateway.js）
const repo = makeRepo(() => new Error('PG 挂了'));
const { gw, warns } = makeGateway(repo);
const { gw, warns } = makeGateway(repo, { storageMaxRetries: 3 });
for (let i = 1; i <= 6; i++) gw._enqueueForStorage(frame('dev-a', i));
await gw._flushStorage(); // 首次尝试失败 → 进重试槽
@@ -237,7 +238,7 @@ async function main() {
// 只要批次里含 dev-poison 就抛错 → 造一颗"毒丸"，且它永远救不回来
const repo = makeRepo((batch) =>
(batch.some((f) => f.device_id === 'dev-poison') ? new Error('值违反约束') : null));
const { gw } = makeGateway(repo);
const { gw } = makeGateway(repo, { storageMaxRetries: 3 });
gw._enqueueForStorage(frame('dev-poison', 1));
gw._enqueueForStorage(frame('dev-poison', 2));
@@ -265,11 +266,101 @@ async function main() {
`实得 ${repo.rows.length}`);
}
/* ─────────── ③b 跨 flush 合并（复核改判：合并而不是丢弃） ─────────── */
console.log('\n=== 3b. 跨 flush 合并：短暂抖动的每一批都要有救 ===');
{
// 前 5 次调用失败（= 3 个周期里槽/主队列各自的尝试），第 6 次成功
const repo = makeRepo((_b, n) => (n <= 5 ? new Error('PG 抖动') : null));
const { gw } = makeGateway(repo);
const sizes = [3, 4, 2]; // 三个周期各失败一批
let seq = 0;
for (const n of sizes) {
for (let i = 0; i < n; i++) gw._enqueueForStorage(frame('dev-m', ++seq));
await gw._flushStorage();
}
ok('三个周期的失败批合并进同一个槽（3 + 4 + 2 = 9 帧）',
gw._storageRetry !== null && gw._storageRetry.batch.length === 9,
gw._storageRetry ? String(gw._storageRetry.batch.length) : '槽为空');
ok('attempts 记的是**重试轮次**（3），不是帧数（9）',
gw._storageRetry?.attempts === 3,
gw._storageRetry ? String(gw._storageRetry.attempts) : '槽为空');
ok('合并期间一帧都没丢（storage_failed 仍为 0）',
gw._storageFailed === 0, String(gw._storageFailed));
await gw._flushStorage(); // 第 4 轮：PG 恢复
ok('★ PG 恢复后一次写回，9 帧全部落库',
repo.rows.length === 9, `实得 ${repo.rows.length}`);
ok('救回后槽清空、seq 顺序完整（1..9）、计数仍为 0',
gw._storageRetry === null && gw._storageFailed === 0
&& repo.rows.map((r) => r.seq).join(',') === '1,2,3,4,5,6,7,8,9',
repo.rows.map((r) => r.seq).join(','));
}
/* ─────────── ③c 合并仍受槽容量约束 ─────────── */
console.log('\n=== 3c. 合并不得破坏有界性：超容量明确丢弃并计数 ===');
{
// 槽容量复用 storageQueueMax（10）；maxRetries 调大，避免中途整槽被丢
const repo = makeRepo(() => new Error('持续不可用'));
const { gw, warns } = makeGateway(repo,
{ storageQueueMax: 10, storageMaxRetries: 20 });
const sizes = [4, 4, 4, 3]; // 合计 15 帧
let seq = 0;
for (const n of sizes) {
for (let i = 0; i < n; i++) gw._enqueueForStorage(frame('dev-cap', ++seq));
await gw._flushStorage();
}
ok('槽被截断到容量上限 10 帧（内存上界 = 主队列上界）',
gw._storageRetry !== null && gw._storageRetry.batch.length === 10,
gw._storageRetry ? String(gw._storageRetry.batch.length) : '槽为空');
ok('超出的 5 帧计入 storage_failed（明确丢弃，不静默）',
gw._storageFailed === 5, String(gw._storageFailed));
ok('保留的是**较早**的帧（尾部先丢，先保已经等了更久的数据）',
gw._storageRetry?.batch[0]?.seq === 1 && gw._storageRetry?.batch[9]?.seq === 10,
gw._storageRetry
? `${gw._storageRetry.batch[0]?.seq}..${gw._storageRetry.batch[9]?.seq}`
: '槽为空');
ok('attempts 仍是轮次（4），不受合并/截断影响',
gw._storageRetry?.attempts === 4,
gw._storageRetry ? String(gw._storageRetry.attempts) : '槽为空');
ok('告警只 1 条（首丢必报，之后按 100 帧节奏）',
discardWarns(warns).length === 1, String(discardWarns(warns).length));
}
/* ─────────── ③d 复核点名的场景：PG 抖动 10 秒（5 轮） ─────────── */
console.log('\n=== 3d. PG 抖动 10 秒（2 秒一轮 × 5 轮）：必须全救回 ===');
{
// 用**默认**上限跑，专测"默认窗口够不够覆盖一次真实抖动"
let failing = true;
const repo = makeRepo(() => (failing ? new Error('PG 抖动') : null));
const { gw } = makeGateway(repo);
let seq = 0;
for (let round = 1; round <= 5; round++) {
for (let i = 0; i < 4; i++) gw._enqueueForStorage(frame('dev-10s', ++seq));
await gw._flushStorage();
}
ok('抖动期间不丢：5 轮共 20 帧全部攒在槽里',
gw._storageRetry !== null && gw._storageRetry.batch.length === 20
&& gw._storageFailed === 0,
gw._storageRetry ? `槽内 ${gw._storageRetry.batch.length} 帧` : '槽为空');
// 第 6 轮：PG 恢复
failing = false;
await gw._flushStorage();
ok('★ PG 恢复后 20 帧一次全部落库（默认重试窗覆盖 10 秒抖动）',
repo.rows.length === 20 && gw._storageFailed === 0,
`落库 ${repo.rows.length} 帧 / 丢弃 ${gw._storageFailed}`);
}
/* ─────────── ④ 多次丢弃：计数累计、告警不刷屏 ─────────── */
console.log('\n=== 4. 多次丢弃：计数准确、告警不刷屏 ===');
{
const repo = makeRepo(() => new Error('磁盘满'));
const { gw, warns } = makeGateway(repo);
const { gw, warns } = makeGateway(repo, { storageMaxRetries: 3 });
for (let i = 1; i <= 2; i++) gw._enqueueForStorage(frame('dev-c', i));
await flushTimes(gw, 4); // 首次 + 3 次重试 → 丢第 1 批（2 帧）