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}]`);