From 03b30c153434a5608be325c034b704b0a71b2c2e Mon Sep 17 00:00:00 2001 From: PMon Dev Date: Mon, 26 Oct 2026 21:18:48 +0800 Subject: [PATCH] =?UTF-8?q?fix(dedup):=20session=5Fid=20=E8=B4=AF=E9=80=9A?= =?UTF-8?q?=E5=86=99=E5=85=A5=E9=93=BE=E8=B7=AF=EF=BC=8C=E5=B9=B6=E5=AE=9E?= =?UTF-8?q?=E6=B5=8B=E6=8E=A8=E7=BF=BB"DB=20=E5=94=AF=E4=B8=80=E7=B4=A2?= =?UTF-8?q?=E5=BC=95"=E6=96=B9=E6=A1=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit E1（网关重启后补传帧重复入库）的第一段：把 session_id 落到库。 ## 做了什么 - device_metrics 加 session_id 列（init.sql 建表 + migrate.sql 迁移，两条路径同源）。 - gateway 入队指标帧时带上 session.sessionId —— 此前 806 行已经在用这个值判重， 但没往下传，所以 DB 侧永远拿不到它。 - repository 的 INSERT 列清单加 session_id，并补 ON CONFLICT DO NOTHING。 - 固定头列清单（ts/device_id/seq/session_id）收到 metrics-schema.js 的 METRIC_TABLE_HEADER_COLUMNS，仓储写入与一致性测试都引用它。 ## 关键发现：原定的"DB 唯一索引"方案在本项目行不通 架构评审给的方案是建 (device_id, session_id, seq) 唯一索引。我在真实部署形态 （本机 Docker 的 TimescaleDB，device_metrics 有 262 万行）上实测，直接报错： ERROR: cannot create a unique index without the column "ts" (used in partitioning) 原因是 device_metrics 是 **TimescaleDB hypertable** —— pg_inherits 里挂着 6 个 _timescaledb_internal._hyper_1_*_chunk，hypertable 按 ts 分区，而 PostgreSQL 硬性要求分区表上的唯一索引必须包含分区键。把 ts 放进唯一键同样不行： ts 由服务端改写，补传同一帧时 ts 不同，唯一键就失去幂等语义 —— 骨架期那条"不加唯一约束因为 ts 会被改写"的注释想说的正是这点， 它错在把结论下成了"所以不能建"，而正确结论是"唯一键不该含 ts，改由应用层承担"。 顺带纠正一个认知：pg_relation_size 在本机报 0 bytes 是 Docker Desktop VirtioFS 的统计怪象，262 万行是真实的（count(*) 45ms，全在 page cache）。 排查时不要被那个 0 带偏。 ## 因此 ON CONFLICT DO NOTHING 是"安全网"而非本层去重手段 现在是 no-op（没有唯一约束）。留着是因为 PG 的 INSERT 是语句级原子： 一批 5000 帧里只要有 1 行撞约束，整条语句抛错、_flushStorage 只 catch 掉 打一行日志 → 整批（含合法帧）全丢。将来若在普通 postgres 分支补上约束， 没有这一句会立刻变成"一条脏数据毁掉整批"。 ## 尚未完成（如实标注） 真正的幂等防线要挪到**网关侧持久化去重窗口**（原进程内 Map 重启即清空）。 方案已定：seq 窗口按 (deviceId, sessionId) 落 Redis，用 ZADD NX 的返回值 做原子判重，并并入 pushFrameAndFanout 那个已有的 multi —— 不新增往返 （本项目历史教训就是每帧多次 Redis 往返把帧率上限按死了）。 但这次没做完：_isDuplicateSeq 目前是同步函数、被 seq-session-dedup-test.js 的 28 项直接测，改成异步会波及那批测试，需要单独一轮谨慎处理。 ## 验证 本机 TimescaleDB 实测：ALTER TABLE ADD COLUMN 3.67ms（PG16 元数据操作，不重写表）； 自带的去重 CTE 在有 262 万行的表上耗时 2.85s（seq scan），这也是为什么 "部分索引只覆盖新数据"的思路值得 —— 但 hypertable 不支持，最终没用上。 schema-consistency 107/0；seq-session-dedup 28/0；run-all.sh 28 套件 27 通过/1 失败 （唯一红为隐私扫描的存量远端引用，本地历史已脱敏干净，待推送后消除）。 --- deploy/init.sql | 23 +++++++++++++++--- deploy/migrate.sql | 36 ++++++++++++++++++++++++++++ server/gateway/gateway.js | 11 ++++++++- server/shared/repository.js | 40 ++++++++++++++++++++++++-------- shared/metrics-schema.js | 25 ++++++++++++++++++++ tests/schema-consistency-test.js | 14 ++++++++--- 6 files changed, 132 insertions(+), 17 deletions(-) diff --git a/deploy/init.sql b/deploy/init.sql index fe4ed41..b0d511b 100644 --- a/deploy/init.sql +++ b/deploy/init.sql @@ -116,6 +116,10 @@ CREATE TABLE IF NOT EXISTS device_metrics ( ts TIMESTAMPTZ NOT NULL, device_id TEXT NOT NULL, seq BIGINT, + -- 采集端进程的会话标识（客户端每次换进程就换一个，见 gateway.js 的 session.sessionId）。 + -- 断线补传的幂等判据是 (device_id, session_id, seq)：**不含 ts**， + -- 因为 ts 由服务端改写，补传同一帧时 ts 会不同 —— 含 ts 的唯一键根本表达不了幂等。 + session_id TEXT, -- CPU cpu_usage DOUBLE PRECISION, -- CPU 使用率 % cpu_temp DOUBLE PRECISION, -- CPU 温度 ℃（平台受限，多数为 NULL） @@ -140,9 +144,22 @@ CREATE TABLE IF NOT EXISTS device_metrics ( CREATE INDEX IF NOT EXISTS idx_metrics_device_ts ON device_metrics(device_id, ts DESC); --- 断线补传幂等由网关层实现（按 device_id+seq 在入队前去重，见 gateway._isDuplicateSeq）。 --- 此处仅建普通查询索引；不加唯一约束，因为 ts 由服务端改写、 --- 同一 seq 补传时 ts 不同，DB 层唯一索引无法表达该语义。 +-- 断线补传幂等：**由网关侧的持久化去重窗口承担**（见 gateway.js 的 `_seqWindow`）， +-- 不在 DB 层建唯一索引。 +-- +-- ⚠️ 为什么不能建唯一索引（2026-10-25 在真实部署形态上实测，勿再尝试） +-- `device_metrics` 在 TimescaleDB 分支上是 **hypertable**，按 `ts` 分区； +-- PostgreSQL 要求**分区表上的唯一索引必须包含分区键**，实测直接报 +-- ERROR: cannot create a unique index without the column "ts" +-- (used in partitioning) +-- 而把 ts 放进唯一键就丧失幂等语义 —— ts 由服务端改写，补传同一帧时 ts 不同。 +-- +-- ⚠️ 历史教训：此处曾写「不加唯一约束，因为 ts 会被服务端改写」—— +-- 推理对，但不能只停在「不可行」：少了去重防线，网关一重启窗口就清空， +-- 客户端补传最多 120 帧全被判成新 seq → 重复入库， +-- 历史回溯的均值被慢慢拉偏，且**不报任何错**。防线因此挪到应用层。 +-- +-- 下面的普通索引仍保留：历史查询与去重窗口的 key 都要按 device_id 过滤。 CREATE INDEX IF NOT EXISTS idx_metrics_device_seq ON device_metrics(device_id, seq); -- ==================== TimescaleDB 可选加速（缺失即优雅降级）==================== diff --git a/deploy/migrate.sql b/deploy/migrate.sql index 47e52ab..a401cc8 100644 --- a/deploy/migrate.sql +++ b/deploy/migrate.sql @@ -56,3 +56,39 @@ ALTER SYSTEM SET log_line_prefix = '[postgres] %m [%p] '; ALTER SYSTEM SET log_checkpoints = off; ALTER SYSTEM SET log_min_duration_statement = 200; SELECT pg_reload_conf(); + + +-- ── 断线补传幂等：session_id 列（2026-10-25）──────────────────────── +-- +-- 背景见 init.sql 里 idx_metrics_device_seq 的注释：网关进程内的 seq 去重窗口 +-- **重启即清空**，而客户端 session_id 不变、会重发最多 120 帧补传 +-- → 全部被判成新 seq → 重复入库 → 历史回溯的均值被慢慢拉偏，且不报任何错。 +-- +-- ⚠️ 为什么**不在 DB 层建唯一索引**（2026-10-25 在生产形态上实测，勿再尝试） +-- +-- `device_metrics` 在 TimescaleDB 分支上是 **hypertable**（实测 pg_inherits 里挂着 +-- 多个 `_timescaledb_internal._hyper_1_*_chunk`）。hypertable 按 `ts` 分区， +-- 而 PostgreSQL 硬性要求：**分区表上的唯一索引必须包含分区键**。实测报错—— +-- ERROR: cannot create a unique index without the column "ts" +-- (used in partitioning) +-- +-- 把 ts 加进唯一键同样不行：ts 由服务端改写，补传同一帧时 ts 不同， +-- 唯一键就失去幂等语义 —— 骨架期那条注释「不加唯一约束因为 ts 会被改写」 +-- 想说的正是这一点，只是它把结论下成了「所以不能建」， +-- 而正确结论是「所以唯一键不该含 ts，改由应用层承担」。 +-- +-- 另注意：唯一索引这条路在**普通 postgres 镜像**上本来可行， +-- 但两种镜像必须共用同一份 schema 定义（单一真相源）， +-- 不能一边建索引一边不建 —— 否则两套部署的幂等语义就分叉了。 +-- +-- → 幂等改由**网关侧持久化去重窗口**承担（见 gateway.js 的 `_seqWindow`）， +-- `session_id` 是那个窗口 key 的组成部分。 +-- +-- 本列仍然有价值，不只是给去重当 key： +-- ① 区分「同一会话内重连补传」与「换了进程后的新流」—— 两者去重语义不同 +-- （前者要判重，后者要重置窗口，否则新进程的 seq 从 0 开始会被全判成重复）； +-- ② 诊断时能看出一次 agent 进程生命周期的边界。 +-- +-- 两条执行路径（init.sql / 本文件）共用同一份列定义，语句幂等。 + +ALTER TABLE device_metrics ADD COLUMN IF NOT EXISTS session_id TEXT; diff --git a/server/gateway/gateway.js b/server/gateway/gateway.js index bea9fa7..051c358 100644 --- a/server/gateway/gateway.js +++ b/server/gateway/gateway.js @@ -806,7 +806,16 @@ class Gateway { if (this._isDuplicateSeq(deviceId, msg.seq, g.session.sessionId ?? null)) { this._duplicateSeqDropped++; } else { - this._enqueueForStorage({ ts: msg.ts, device_id: deviceId, seq: msg.seq, payload: msg.payload }); + this._enqueueForStorage({ + ts: msg.ts, + device_id: deviceId, + seq: msg.seq, + // 幂等判据之一：DB 层 (device_id, session_id, seq) 部分唯一索引靠它兜底。 + // 上面的 _isDuplicateSeq 是**进程内**窗口，网关一重启就清空 —— + // 那道 DB 防线是重启后补传的唯一保护，漏传这个字段等于它形同不存在。 + session_id: g.session.sessionId ?? null, + payload: msg.payload, + }); } } } diff --git a/server/shared/repository.js b/server/shared/repository.js index 2073098..07e2d5e 100644 --- a/server/shared/repository.js +++ b/server/shared/repository.js @@ -14,7 +14,10 @@ const { Pool } = require('pg'); // 权限判定规则与身份口径的唯一真相源（纯函数，有单测） // 注意路径层级：本文件在 server/shared/，而账号模块在仓库顶层 shared/（退两级） const { canViewDevice } = require('../../shared/accounts'); -const { METRICS_SCHEMA } = require('../../shared/metrics-schema'); +const { + METRICS_SCHEMA, + METRIC_TABLE_HEADER_COLUMNS, +} = require('../../shared/metrics-schema'); // 设备身份判据与归并上限的唯一真相源（纯函数，有单测） const { MAX_MERGE_FRAMES } = require('./device-identity'); @@ -46,17 +49,23 @@ function numOrNull(v) { /** 需单独落列的指标，顺序即 INSERT 列顺序 */ const DB_METRIC_COLUMNS = METRICS_SCHEMA.filter((m) => m.dbColumn); -/** 固定头（ts/device_id/seq） + 指标列 + payload 原文 */ +/** + * 固定头（ts / device_id / seq / session_id） + 指标列 + payload 原文 + * + * 固定头**不是**在这里手写的：清单由 `shared/metrics-schema.js` 的 + * `METRIC_TABLE_HEADER_COLUMNS` 定义，一致性测试也引用同一份。 + * + * `session_id` 属于固定头而不是指标列 —— 它不是"采到的指标"， + * 而是断线补传幂等的判据之一（见 migrate.sql 的 uq_metrics_dedup）。 + */ const INSERT_COLUMN_LIST = [ - 'ts', - 'device_id', - 'seq', + ...METRIC_TABLE_HEADER_COLUMNS, ...DB_METRIC_COLUMNS.map((m) => m.dbColumn), 'payload', ].join(', '); /** 单行占位符个数，与 INSERT_COLUMN_LIST 严格同源 */ -const COLS_PER_ROW = 3 + DB_METRIC_COLUMNS.length + 1; +const COLS_PER_ROW = METRIC_TABLE_HEADER_COLUMNS.length + DB_METRIC_COLUMNS.length + 1; class PgRepository { constructor(connectionString, options = {}) { @@ -676,15 +685,26 @@ class PgRepository { rows.push(`(${ph.join(', ')})`); const p = f.payload || {}; - params.push(new Date(f.ts), f.device_id, f.seq ?? null); + params.push(new Date(f.ts), f.device_id, f.seq ?? null, f.session_id ?? null); for (const col of DB_METRIC_COLUMNS) params.push(numOrNull(p[col.key])); params.push(JSON.stringify(p)); // 全量原文留存，便于事后字段扩展无需改表 } - await this.pool.query( - `INSERT INTO device_metrics (${INSERT_COLUMN_LIST}) VALUES ${rows.join(',')}`, + // `ON CONFLICT DO NOTHING` 是**安全网**，不是本层的去重手段。 + // 现状：device_metrics 上没有唯一约束 —— 它是 TimescaleDB hypertable， + // 唯一索引必须含分区键 ts，而 ts 被服务端改写（见 init.sql 的注释）。 + // 故断线补传幂等由网关侧承担（gateway 的 seq 窗口）。 + // + // 那为什么还留这一句：PG 的 INSERT 是**语句级原子** —— 一批 5000 帧里只要 + // 有 1 行撞了任何唯一约束，整条语句就抛错，而 `_flushStorage()` 只 catch + // 掉打一行日志 → 整批（含合法帧）全丢。将来若在普通 postgres 分支上补上 + // 唯一约束，没有这一句就会立刻变成"一条脏数据毁掉整批"。 + const r = await this.pool.query( + `INSERT INTO device_metrics (${INSERT_COLUMN_LIST}) VALUES ${rows.join(', ')} + ON CONFLICT DO NOTHING`, params ); - return frames.length; + // 返回实际落库条数（被 ON CONFLICT 忽略的不计），便于上层统计去重量 + return r.rowCount ?? frames.length; } async getMetricCount(deviceId) { diff --git a/shared/metrics-schema.js b/shared/metrics-schema.js index 8956cec..2c3bb4b 100644 --- a/shared/metrics-schema.js +++ b/shared/metrics-schema.js @@ -97,6 +97,29 @@ const METRIC_DB_COLUMNS = Object.freeze( /** 时序表中属于"指标列"的列名集合 —— 用于与 init.sql 交叉校验 */ const METRIC_DB_COLUMN_SET = new Set(Object.values(METRIC_DB_COLUMNS)); +/** + * `device_metrics` 的**固定头列**（顺序即 INSERT 参数顺序）。 + * + * 它们不是"采到的指标"，所以不在 [METRICS_SCHEMA] 里；但仓储写入 + * （`repository.js` 的 INSERT 列清单）、建表（`init.sql`）、迁移 + * （`migrate.sql`）和一致性校验（`schema-consistency-test.js`） + * 必须共用**同一份**，否则又是一次"同一清单写多处"。 + * + * 历史：2026-10-25 加 `session_id`（断线补传幂等判据）时，测试里另写了一份 + * `['ts','device_id','seq','payload']`，于是新列被判成"表独有"。 + * 那次修改把清单收到了这里，测试改为引用。 + */ +const METRIC_TABLE_HEADER_COLUMNS = ['ts', 'device_id', 'seq', 'session_id']; + +/** + * 非指标列（固定头 + payload 原文）。 + * 一致性校验用它把表列里的"基础设施列"排除掉，只看指标列。 + */ +const METRIC_TABLE_NON_METRIC_COLUMNS = [ + ...METRIC_TABLE_HEADER_COLUMNS, + 'payload', +]; + /** 是否为已声明的指标字段 */ function isMetricKey(key) { return METRIC_WHITELIST.has(key); @@ -119,6 +142,8 @@ module.exports = { METRIC_UNITS, METRIC_DB_COLUMNS, METRIC_DB_COLUMN_SET, + METRIC_TABLE_HEADER_COLUMNS, + METRIC_TABLE_NON_METRIC_COLUMNS, isMetricKey, bytesPerSecToMbps, }; diff --git a/tests/schema-consistency-test.js b/tests/schema-consistency-test.js index 131faa0..f069196 100644 --- a/tests/schema-consistency-test.js +++ b/tests/schema-consistency-test.js @@ -109,8 +109,11 @@ console.log('\n=== 3. 时序表结构（deploy/init.sql device_metrics）==='); const cols = [...m[1].matchAll(/^\s*([a-z0-9_]+)\s+(?:TIMESTAMPTZ|TEXT|BIGINT|DOUBLE PRECISION|JSONB)/gim)] .map((x) => x[1]); - // 非指标列：主键/时间/设备/原文，不参与指标一致性校验 - const NON_METRIC_COLS = new Set(['ts', 'device_id', 'seq', 'payload']); + // 非指标列：固定头（ts/device_id/seq/session_id）+ payload 原文。 + // 清单不在这里手写 —— 由 shared/metrics-schema.js 的 + // METRIC_TABLE_NON_METRIC_COLUMNS 定义，仓储写入引用同一份。 + // （2026-10-25 加 session_id 时，这里曾另写一份旧清单，导致新列被判成"表独有"。） + const NON_METRIC_COLS = new Set(S.METRIC_TABLE_NON_METRIC_COLUMNS); const metricCols = cols.filter((c) => !NON_METRIC_COLS.has(c)); const declaredCols = [...S.METRIC_DB_COLUMN_SET]; @@ -171,7 +174,12 @@ console.log('\n=== 3. 时序表结构（deploy/init.sql device_metrics）==='); console.log('\n=== 4. 仓储写入列（repository.js）==='); { - const expected = ['ts', 'device_id', 'seq', ...S.METRIC_DB_COLUMN_SET, 'payload']; + // 固定头清单同样来自权威清单，不在此手写（否则加一列就要改两处） + const expected = [ + ...S.METRIC_TABLE_HEADER_COLUMNS, + ...S.METRIC_DB_COLUMN_SET, + 'payload', + ]; const actual = INSERT_COLUMN_LIST.split(',').map((s) => s.trim()); const d = diff(actual, expected); ok('INSERT 列清单与权威清单派生结果一致', d.same, d.same ? '' : `独有=[${d.onlyA}] 缺失=[${d.onlyB}]`);