merge: fix/push-body-limit → main(push 请求体上限 10MB + 超限归 10002 + 画像圈人 14 条偏索引 23.4s→84ms)

apps/pac-docs/content/docs/integration/channel-push.mdx

@@ -221,7 +221,7 @@ const sig  = require('crypto').createHmac('sha256', SECRET).update(`${ts}.${body
 | ------------------ | ------- | ----------------------------------------------------------------------- |
 | 成功                 | `0`     | — |
 | 验签失败 | `10106` | 检查 secret、时钟、签名串 |
-| 字段校验失败 / 批量超限          | `10002` | 修正后重发 |
+| 字段校验失败 / 批量超限(条数或字节) | `10002` | 修正后重发(字节超限见 §10,减小单批行数) |
 | source 表名不识别            | `10001` | 核对表名 |
 | 数据形态漂移(含整批全部失败)   | `30802` | 核对数据字典,修正后原批重推 |
 | 并发超限 | `10003` | 退避后重发本批 |
@@ -249,11 +249,22 @@ PAC 侧对推送中断有告警。
 
 | 项 | 约定 |
 |---|---|
-| 批量 | 两形态 ≤ 500 条/请求 |
+| 批量(条数) | 两形态 ≤ 500 条/请求 |
+| 批量(字节) | **请求体 ≤ 10 MB**,超出返 `10002`,减小单批行数后重发 |
 | 并发 | 同 host 有并发额度,超限返 `10003`,退避重发即可 |
 | 编码 | UTF-8 |
 | 重试 | 幂等,可安全重试,建议指数退避 |
 
+⚠️ **两条上限都要满足,先到先约束** —— 行的大小按表差很多,别只按条数切批:
+
+| 表类型 | 单行典型大小 | 10 MB 约能装 | 建议单批 |
+|---|---|---|---|
+| 病历 / 诊断(含自由文本:主诉、检查、处置、医嘱) | 1.5-2.2 KB | ~5000 行 | **50-100 行** |
+| 预约 / 接诊 / 结算(纯结构化字段) | ~400 B | ~25000 行 | 500 行(条数先到顶) |
+
+病历类按 500 条切批很容易做出 700 KB 以上的包;虽然仍在 10 MB 内,但单请求处理时间会显著变长
+(实测有整批超时被客户端断连的情况),所以建议按上表压到 50-100 行。
+
 ---
 
 ## 11. 接入清单

apps/pac-service/prisma/migrations/20260728020000_persona_features_tag_filter_indexes/migration.sql

+-- CreateIndex —— 画像圈人(personaTags)筛选索引,每个可筛维度一条偏索引
+--
+-- 【症状】生产 2026-07-28 实测:召回池按「禁忌:手术禁忌」筛人
+--     GET /pac/v1/plans?view=pool&personaTags=contraindication:surgery
+--   前端等 20+ 秒。服务端复现:count 23,379ms + findMany 22,208ms(两句并发,故墙钟 ≈23s)。
+--   **不是禁忌特有** —— 测试服同口径按「价值分群:重要价值」(rfm)更慢,57s。
+--   14 个圈人维度全都踩同一个坑,禁忌只是用户先点到的那个。
+--
+-- 【根因】Prisma 的 relation filter 编译成两层 EXISTS,最内层是:
+--     EXISTS (SELECT persona_id FROM persona_features t2
+--             WHERE t2.key = 'contraindication'
+--               AND (t2.data #> '{domains}') @> '"surgery"'
+--               AND t1.id = t2.persona_id)
+--   现有索引只有 persona_features_key_idx(单列 key),于是 planner 只能:
+--     按 key 取出该维度**全部**特征行 → 逐行回堆读 data(jsonb) → 在堆上过滤 JSON 值。
+--
+--   生产 EXPLAIN (ANALYZE, BUFFERS) 实测(persona_features 7,657,944 行 / 5.4GB,堆 3.4GB):
+--     Index Scan using persona_features_key_idx  actual time=3567..3954ms
+--       Index Cond: (key = 'contraindication')
+--       Filter:     ((data #> '{domains}') @> '"surgery"')
+--       rows=2,156,**Rows Removed by Filter: 188,054**      ← 命中率 1.1%,98.9% 是白读
+--       Buffers: shared read=162,830(约 1.3GB 随机堆读)   ← 冷缓存下就是这 20 秒
+--   即:为了捞 2,156 行,把 190,210 行的 jsonb 全从磁盘搬了一遍。
+--   维度越大越惨:rfm / gender / lifecycle_stage 各 869,043 行,一次筛人搬 869k 行 data。
+--
+-- 【修法】给每个可筛维度建一条**偏索引**(WHERE key = '<维度>'),表达式与 Prisma 生成的
+--   完全一致(data #> '{<dataPath>}'),让 JSON 条件下推到索引:
+--     · 数组维度(isArray,array_contains → @>)→ GIN jsonb_path_ops,只支持也只需要 @>
+--     · 标量维度(equals → =)             → btree (表达式, persona_id) 复合
+--       第二列 persona_id 不是为了排序,是为了 **index-only scan**:内层 EXISTS 只 SELECT
+--       persona_id,索引自带就不用回堆。rfm 这种低选择性维度(164,442 行命中)光靠索引定位
+--       没用 —— 测试服实测走上 btree 后 Bitmap Index Scan 只花 217ms,但回堆 143,187 个块
+--       花了 103s(还因 work_mem 不够退化成 lossy,Rows Removed by Index Recheck 1,564,553)。
+--       盖住 persona_id 才是这类维度真正的解。
+--
+-- 【为什么不写进 schema.prisma】Prisma schema 表达不了表达式索引 / 偏索引,只能落在迁移 SQL 里。
+--   副作用:本地 `prisma migrate dev` 会把这些索引报成 drift(它不认识),**别按提示 reset**;
+--   `migrate deploy` 不受影响。同类先例见 20260727070000 / 20260727110000。
+--
+-- 【维度增删纪律】可筛维度的单一真理源是 packages/types/src/persona-tag-filters.ts 的
+--   PERSONA_TAG_FILTER_DIMS。往那里加维度时,必须在这里补一条对应索引(数组走 GIN、标量走 btree),
+--   否则新维度一上线就是 20 秒起步 —— 这正是禁忌维度这次踩的坑。
+--
+-- 【体积与写入代价】偏索引只覆盖本 key 的行,14 条加起来 ≈ 全表一遍(每行至多落进一条)。
+--   生产实测 14 条合计 **220MB**(表 5.4GB,占 4%)。写入侧同理:插一行特征至多多维护 1 条索引,
+--   画像重算不会被拖垮。
+--
+-- 【CONCURRENTLY 与上线方式】同 20260727110000:普通 CREATE INDEX 会 ACCESS EXCLUSIVE 锁全表,
+--   画像重算与摄入全挂,故用 CONCURRENTLY。**推荐部署前手动先建**(deploy 时 pac-service 要等
+--   pac-migrate 退出,14 条索引串行构建 = 白等的停机);手动建好后本文件全部 IF NOT EXISTS 空跑。
+--   生产实测单条 4-17s,14 条共 **88 秒**(不锁表,期间业务照常);测试服磁盘慢,单条约 42s。
+--
+-- 【上线后实测】生产 2026-07-28 已手动建完,同口径复测(count + findMany 并发,墙钟):
+--     contraindication:surgery      23,379ms → **84ms**    (278×,命中 1,347 单)
+--     contraindication:implant                    842ms    (25,900 单)
+--     rfm:important_value                       1,558ms    (61,300 单)
+--     treatment_history:implant_history           854ms    (17,179 单)
+--     gender:female                             2-4s       (105,110 单)
+--   剩下的耗时不再是 JSON 过滤,而是**结果集本身大**:count(*) 要数完十万级 plan。
+--   若哪天 gender 这类宽维度也嫌慢,方向是给 count 做近似/缓存,不是继续加索引。
+--   测试服同日也已手动建完(14 条 / 238s / 195MB):rfm 57,299ms → **8,576ms**,
+--   禁忌 4,210ms → **344ms**。那台盘慢,绝对值别拿去跟生产比,看倍数即可。
+
+-- ── 标量维度(equals):btree(表达式, persona_id),第二列盖住 EXISTS 的输出列 ──────────────
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_rfm_segment_idx"
+  ON "persona_features" (((data #> '{segment}'::text[])), "persona_id") WHERE "key" = 'rfm';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_lifecycle_stage_stage_idx"
+  ON "persona_features" (((data #> '{stage}'::text[])), "persona_id") WHERE "key" = 'lifecycle_stage';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_urgency_level_level_idx"
+  ON "persona_features" (((data #> '{level}'::text[])), "persona_id") WHERE "key" = 'urgency_level';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_age_bracket_bracket_idx"
+  ON "persona_features" (((data #> '{bracket}'::text[])), "persona_id") WHERE "key" = 'age_bracket';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_gender_gender_idx"
+  ON "persona_features" (((data #> '{gender}'::text[])), "persona_id") WHERE "key" = 'gender';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_family_structure_structure_idx"
+  ON "persona_features" (((data #> '{structure}'::text[])), "persona_id") WHERE "key" = 'family_structure';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_referral_champion_type_idx"
+  ON "persona_features" (((data #> '{type}'::text[])), "persona_id") WHERE "key" = 'referral_champion';
+
+-- ── 数组维度(array_contains → @>):GIN jsonb_path_ops ────────────────────────────────────
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_contraindication_domains_idx"
+  ON "persona_features" USING gin (((data #> '{domains}'::text[])) jsonb_path_ops) WHERE "key" = 'contraindication';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_entitlement_status_types_idx"
+  ON "persona_features" USING gin (((data #> '{types}'::text[])) jsonb_path_ops) WHERE "key" = 'entitlement_status';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_treatment_history_types_idx"
+  ON "persona_features" USING gin (((data #> '{types}'::text[])) jsonb_path_ops) WHERE "key" = 'treatment_history';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_treatment_sensitivity_types_idx"
+  ON "persona_features" USING gin (((data #> '{types}'::text[])) jsonb_path_ops) WHERE "key" = 'treatment_sensitivity';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_special_attention_types_idx"
+  ON "persona_features" USING gin (((data #> '{types}'::text[])) jsonb_path_ops) WHERE "key" = 'special_attention';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_time_preference_types_idx"
+  ON "persona_features" USING gin (((data #> '{types}'::text[])) jsonb_path_ops) WHERE "key" = 'time_preference';
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "persona_features_potential_treatment_types_idx"
+  ON "persona_features" USING gin (((data #> '{types}'::text[])) jsonb_path_ops) WHERE "key" = 'potential_treatment';

apps/pac-service/src/common/filters/all-exceptions.filter.ts

@@ -30,9 +30,11 @@ import { BizError } from '../errors/biz-error';
  * Code mapping policy (priority highest → lowest):
  *   1. BizError              → use its explicit 5-digit code
  *   2. ZodValidationException → CLIENT_VALIDATION_FAILED (10002)
- *   3. ContractDriftError    → SYNC_CONTRACT_DRIFT (30802)
- *   4. NestJS HttpException  → mapped by class (UnauthorizedException → 10106, etc.)
- *   5. Plain Error           → INTERNAL_ERROR (90000), HTTP 500
+ *   3. PayloadTooLargeError  → CLIENT_VALIDATION_FAILED (10002), HTTP 200 —— 裸 Error,
+ *                              必须排在第 5 条之前拦下,否则对面只看到 500/90000
+ *   4. ContractDriftError    → SYNC_CONTRACT_DRIFT (30802)
+ *   5. NestJS HttpException  → mapped by class (UnauthorizedException → 10106, etc.)
+ *   6. Plain Error           → INTERNAL_ERROR (90000), HTTP 500
  */
 @Catch()
 export class AllExceptionsFilter implements ExceptionFilter {
@@ -63,6 +65,18 @@ export class AllExceptionsFilter implements ExceptionFilter {
           message: i.message,
         }));
       }
+    } else if (isPayloadTooLargeError(exception)) {
+      // body-parser 的 PayloadTooLargeError 是**裸 Error**(不是 HttpException)→ 不拦的话掉进
+      //   下面 500/90000 分支,对面只看到"PAC 内部错误"、按文档去退避重试,永远重试不好
+      //   (2026-07 FRIDAY 增量 push 整点重试连挂 25 次就是这么来的)。
+      // 归 10002:接入文档 §9.1 把「字段校验失败 / 批量超限」定为 10002「修正后重发」——
+      //   包超限正是"这批得改小了再发",语义与处置动作都对得上。
+      // HTTP 保持 200:同文档「HTTP 恒为 200(仅进程级故障返回 5xx)」,客户端只认 body.code。
+      // 也**不上报 Sentry**:这是对面发包过大,不是 PAC 故障。
+      code = ApiCode.CLIENT_VALIDATION_FAILED;
+      msg = `请求体超出上限(${PAC_BODY_LIMIT_HINT}),请减小单批行数后重发`;
+      details = { limit: PAC_BODY_LIMIT_HINT, hint: '病历类行含自由文本,建议单批 50-100 行' };
+      this.logger.warn(`${req.method} ${req.url} → 请求体超限,已按 10002 回执`);
     } else if (exception instanceof ContractDriftError) {
       code = ApiCode.SYNC_CONTRACT_DRIFT;
       msg = exception.message;
@@ -122,3 +136,18 @@ function mapNestExceptionToCode(exc: HttpException, status: number): number {
   if (status === 429) return ApiCode.CLIENT_RATE_LIMITED;
   return ApiCode.INTERNAL_ERROR;
 }
+
+/// 请求体上限的展示串 —— 仅用于回执文案 / details。真值在 main.ts 的 PAC_BODY_LIMIT,
+/// 两处都改才对得上(这里不 import main.ts:那会把 bootstrap 副作用拖进过滤器)。
+const PAC_BODY_LIMIT_HINT = '10MB';
+
+/**
+ * body-parser(raw-body)在请求体超过 limit 时抛的错 —— **裸 Error,不是 HttpException**,
+ * 也不导出类型可供 instanceof。稳定特征是 `type === 'entity.too.large'`(raw-body 固定写入),
+ * 辅以 status/statusCode 413 兜底(不同版本字段名有出入)。
+ */
+function isPayloadTooLargeError(e: unknown): boolean {
+  if (!(e instanceof Error)) return false;
+  const x = e as Error & { type?: string; status?: number; statusCode?: number };
+  return x.type === 'entity.too.large' || x.status === 413 || x.statusCode === 413;
+}

apps/pac-service/src/main.ts

@@ -4,6 +4,7 @@ import { NestFactory } from '@nestjs/core';
 import { Logger } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { apiReference } from '@scalar/nestjs-api-reference';
+import type { NestExpressApplication } from '@nestjs/platform-express';
 import helmet from 'helmet';
 import { AppModule } from './app.module';
 import { AllExceptionsFilter } from './common/filters/all-exceptions.filter';
@@ -13,9 +14,26 @@ import {
   buildPacOpenApiDocument,
 } from './openapi/build-document';
 
+/// 请求体上限(json / urlencoded 共用)。见 bootstrap 内注释:不显式设就是 100KB 默认值。
+const PAC_BODY_LIMIT = '10mb';
+
 async function bootstrap() {
   // rawBody: true → req.rawBody 保留原始 bytes,push HMAC 验签需要(parsed JSON 字段顺序不稳)
-  const app = await NestFactory.create(AppModule, { bufferLogs: true, rawBody: true });
+  const app = await NestFactory.create<NestExpressApplication>(AppModule, {
+    bufferLogs: true,
+    rawBody: true,
+  });
+
+  // ⭐ 请求体上限 —— **必须显式设**:不设就吃 body-parser 出厂默认 100KB。
+  //   2026-07 线上(测试机)踩雷:FRIDAY 增量 push `med_emr_info` 整点重试连挂 25 次,
+  //   报 `PayloadTooLargeError` → 兜成 500/90000,对面完全看不出是自己包太大。
+  //   接入文档 §10 只约束"≤500 条/请求",没约束字节 —— 而病历行带自由文本(实测均 1.4KB、
+  //   峰值 2.2KB),500 条 ≈ 710KB,超默认值 7 倍必挂;连 405B 的预约行 500 条(≈198KB)也超。
+  //   取 10MB:覆盖 500 条病历十几倍余量,又远低于网关 client_max_body_size 50M(不会把
+  //   拦截点推到网关,那层返 HTML 413,对面更难排查)。rawBody 由上面的 create 选项保留,
+  //   useBodyParser 重设 limit 不影响验签。
+  app.useBodyParser('json', { limit: PAC_BODY_LIMIT });
+  app.useBodyParser('urlencoded', { limit: PAC_BODY_LIMIT, extended: true });
 
   app.use(helmet({ contentSecurityPolicy: false }));
 

apps/pac-service/tests/exceptions-filter-payload-too-large.spec.ts

+import { ArgumentsHost, HttpStatus } from '@nestjs/common';
+import { ApiCode } from '@pac/types';
+import { AllExceptionsFilter } from '../src/common/filters/all-exceptions.filter';
+
+/**
+ * 请求体超限回执契约(2026-07 FRIDAY 增量 push 线上问题)。
+ *
+ * body-parser 的 PayloadTooLargeError 是**裸 Error**,不拦就掉进 500/90000 分支 ——
+ * 对面按接入文档 §9.1 把 9xxxx 当"PAC 内部错误 → 退避重试",于是整点重试永远重试不好。
+ * 本文件锁住:超限必须回 10002 + HTTP 200,且带可执行的提示。
+ */
+function makeHost() {
+  const json = jest.fn();
+  const status = jest.fn().mockReturnValue({ json });
+  const host = {
+    switchToHttp: () => ({
+      getResponse: () => ({ status }),
+      getRequest: () => ({ method: 'POST', url: '/pac/v1/push/rows' }),
+    }),
+  } as unknown as ArgumentsHost;
+  return { host, status, json };
+}
+
+/** 复刻 raw-body 抛的形状(它不导出类型,只能按特征字段构造) */
+function payloadTooLarge(shape: 'type' | 'status' | 'statusCode') {
+  const e = new Error('request entity too large') as Error & Record<string, unknown>;
+  if (shape === 'type') e.type = 'entity.too.large';
+  if (shape === 'status') e.status = 413;
+  if (shape === 'statusCode') e.statusCode = 413;
+  return e;
+}
+
+describe('AllExceptionsFilter — 请求体超限', () => {
+  const filter = new AllExceptionsFilter();
+
+  it.each(['type', 'status', 'statusCode'] as const)(
+    '⭐ 按 %s 特征识别 → 10002 + HTTP 200(不是 90000/500)',
+    (shape) => {
+      const { host, status, json } = makeHost();
+      filter.catch(payloadTooLarge(shape), host);
+
+      expect(status).toHaveBeenCalledWith(HttpStatus.OK); // 文档:HTTP 恒为 200
+      const body = json.mock.calls[0][0];
+      expect(body.code).toBe(ApiCode.CLIENT_VALIDATION_FAILED); // 10002 = 批量超限,修正后重发
+      expect(body.code).not.toBe(ApiCode.INTERNAL_ERROR); // 不能再是 90000
+      expect(body.data).toBeNull();
+    },
+  );
+
+  it('回执要能指导对面动作(带上限 + 建议批量)', () => {
+    const { host, json } = makeHost();
+    filter.catch(payloadTooLarge('type'), host);
+    const body = json.mock.calls[0][0];
+    expect(body.msg).toContain('10MB');
+    expect(body.msg).toContain('减小单批行数');
+    expect(body.details).toMatchObject({ limit: '10MB' });
+  });
+
+  it('普通 Error 不受影响 —— 仍是 90000 + HTTP 500(基础设施要能看见真故障)', () => {
+    const { host, status, json } = makeHost();
+    filter.catch(new Error('boom'), host);
+    expect(status).toHaveBeenCalledWith(HttpStatus.INTERNAL_SERVER_ERROR);
+    expect(json.mock.calls[0][0].code).toBe(ApiCode.INTERNAL_ERROR);
+  });
+
+  it('不能误伤:仅消息里含 "too large" 但无特征字段的普通 Error 仍走 500', () => {
+    const { host, status } = makeHost();
+    filter.catch(new Error('some payload is too large for the queue'), host);
+    expect(status).toHaveBeenCalledWith(HttpStatus.INTERNAL_SERVER_ERROR);
+  });
+});

packages/types/src/persona-tag-filters.ts

@@ -9,6 +9,12 @@
  *
  * 筛选语义(plan.service 实现):同一维度多选 = OR;跨维度 = AND;
  * 匹配的是患者**当前版**画像(personas.supersededAt IS NULL)。
+ *
+ * ⚠️ **加维度必须同步补库索引** —— 每个维度的 (key, dataPath) 在
+ * `prisma/migrations/20260728020000_persona_features_tag_filter_indexes` 里各有一条偏索引
+ * (数组维度走 GIN、标量维度走 btree 且第二列盖 persona_id)。漏建的维度会退化成
+ * 「按 key 捞全量特征行 + 回堆过滤 jsonb」:生产实测禁忌维度 190,210 行里只中 2,156 行,
+ * 读了 1.3GB 堆、单次筛人 20+ 秒(rfm 在测试服 57 秒)。这不是能忍的慢,是不可用。
  */
 export interface PersonaTagOption {
   /** data 里的结构化取值(英文 code) */