fix(dedup): session_id 贯通写入链路，并实测推翻"DB 唯一索引"方案

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

@@ -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 可选加速（缺失即优雅降级）====================

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;

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,
+        });
       }
     }
   }

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) {

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,
 };

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