perf(push): reduce synchronous ingestion overhead and trace latency

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

@@ -215,6 +215,15 @@ const sig  = require('crypto').createHmac('sha256', SECRET).update(`${ts}.${body
 
 ### 9.1 回执
 
+正式 `/push/rows` 在业务写入完成后返回原回执，不需要额外模式或参数。
+验签、请求留档、必要字段校验、主档/流水/事实写入和画像任务入队均在返回前完成。
+画像、召回计算仍在后台；历史字段对比和外部告警通知也不阻塞回执。
+
+`failed`、`factsFailed`、`hasFailures`、`personaEnqueueFailed` 反映已知失败，不把部分失败当全成功。
+正式回执的 `suspectFields` 包含当前批次校验发现的问题；历史列增减结果补充到推送记录。
+`dryRun=1` 仍同步等待完整预检。
+
+
 统一返回 `{ code, msg, data }`,HTTP 恒为 200(仅进程级故障返回 5xx)。按 `code` 判断结果:
 
 | 场景                 | `code`  | 处理                                                                     |
@@ -250,7 +259,7 @@ const sig  = require('crypto').createHmac('sha256', SECRET).update(`${ts}.${body
 | `failed` | 装配 / 写入失败计数,按派生资源累计 | 修正后重推,或由宿主持久化隔离后处理;不能直接忽略 |
 | `factsFailed` | 流水已保存,但部分派生事实被校验拒收的数量 | 看 `factRejects` 定位与修正 |
 | `mappingMisses` | 存在未映射的原值 | 暂停推送,发样本给 PAC |
-| `suspectFields` | 必填映射列整批缺席，或与历史原文相比列发生增减 | 确认表结构是否变更,同步 PAC；可选字段无历史基线且未提供时不计缺列 |
+| `suspectFields` | 当前批必填映射列缺席；历史列增减在预检回执或后台推送记录查看 | 确认表结构是否变更,同步 PAC；可选字段无历史基线且未提供时不计缺列 |
 
 一条原生行可能拆成多条流水或被业务规则过滤,上述计数不保证相加等于 `accepted`。
 `status` 在推送记录中查询,不是 `/push/rows` 回执中的字段。

apps/pac-docs/content/docs/integration/friday-push-payload.mdx

@@ -4,6 +4,9 @@ description: FRIDAY SaaS 按形态 A 推送的 10 个 source 及字段定义;宿
 icon: FileBraces
 ---
 
+> `/push/rows` 地址、正文、签名不变，业务落库完成后返回原回执。历史字段对比和告警通知后台执行；失败明细按 `syncLogId` 查询。
+
+
 > 本文是 [Push 通道](/docs/integration/channel-push)**形态 A** 在 FRIDAY SaaS 宿主的具体化:每个 `source` 的 JSON 字段清单。
 > 已接入字段与 PAC 存量摄入口径一致；标注「待接入」的扩展需双方完成联调后启用。宿主按契约推送,与存量自动衔接:
 > 同行按幂等键去重,行更新(`updated_gmt_at` 变)自动版本演进。**重叠无害,宁多勿漏**。

apps/pac-service/prisma/migrations/20261014110100_push_drift_index/migration.sql

+-- 独立语句，无事务包装。生产构建索引不阻塞业务写入。
+CREATE INDEX CONCURRENTLY IF NOT EXISTS patient_txn_drift_sample_idx
+ ON patient_transactions(host_id,subject_type,created_at DESC);

apps/pac-service/prisma/schema.prisma

@@ -625,6 +625,7 @@ model PatientTransaction {
   ///   真正的杠杆是 CLI 的 `--patients-file` 按受影响患者收窄(实测最高 250 倍)。
   ///   这条胜在每次 reparse / 每个 host 都受益,代价只有一次建索引。
   @@index([hostId, subjectType, patientId])
+  @@index([hostId, subjectType, createdAt(sort: Desc)], map: "patient_txn_drift_sample_idx")
   /// 幂等键 partial UNIQUE (host_id, tenant_id, source_event_id) WHERE source_event_id IS NOT NULL
   /// Prisma 不支持声明式 partial unique;migration 20260528000002 是单一真理源。
   /// (init migration 漏建,20260528000002 补 dedup + 建索引)

apps/pac-service/src/modules/sync/push/push-audit-queue.ts

+import { Logger } from '@nestjs/common';
+
+/** 仅用于辅助审计，禁止放入业务写入或重算投递。限并发、限积压，防慢告警拖垮服务。 */
+export class PushAuditQueue {
+  private readonly logger = new Logger(PushAuditQueue.name);
+  private readonly pending: Array<{ id: string; run: () => Promise<void> }> = [];
+  private running = false;
+
+  enqueue(id: string, run: () => Promise<void>): boolean {
+    if (this.pending.length >= 100) {
+      this.logger.error(`push audit skipped syncLogId=${id} reason=backlog_limit`);
+      return false;
+    }
+    this.pending.push({ id, run });
+    if (!this.running) {
+      this.running = true;
+      setImmediate(() => { void this.drain(); });
+    }
+    return true;
+  }
+
+  private async drain(): Promise<void> {
+    try {
+      let next;
+      while ((next = this.pending.shift())) {
+        try { await next.run(); }
+        catch (error) {
+          this.logger.error(`push audit failed syncLogId=${next.id} type=${error instanceof Error ? error.name : 'unknown'}`);
+        }
+      }
+    } finally { this.running = false; }
+  }
+}

apps/pac-service/src/modules/sync/push/push-receiver.service.ts

@@ -51,15 +51,24 @@ export class PushReceiverService {
     hostName: string;
     source: string;
     rows: Record<string, unknown>[];
+    traceId?: string;
     rawBody?: string; // 控制器传入验签原文，不重新序列化
     /// 预检(?dryRun=1):不写业务数据、不派生事实、不触发重算;仍保留预检推送记录。
     dryRun?: boolean;
   }): Promise<PushRowsResponse> {
     const { hostId, hostName, source, rows } = input;
     const dryRun = !!input.dryRun;
+    let phaseAt = Date.now();
+    const timing = (phase: string) => {
+      const now = Date.now();
+      this.logger.log(`push timing traceId=${input.traceId ?? 'internal'} phase=${phase} durationMs=${now - phaseAt}`);
+      phaseAt = now;
+    };
 
     const evidenceId = await this.evidence.capture({ hostId, source, rowCount: rows.length,
       rawBody: input.rawBody ?? JSON.stringify({ source, rows }) });
+    timing('evidence-saved');
+    this.logger.log(`push traceId=${input.traceId ?? 'internal'} evidenceId=${evidenceId}`);
     let hasFailures = false;
     try {
       const r = await this.coldImport.ingestRawTables({
@@ -71,6 +80,8 @@ export class PushReceiverService {
         triggeredBy: `${dryRun ? 'push-dryrun' : 'push'}:${hostName}:${source}`,
       });
 
+      timing('business-pipeline-done');
+      this.logger.log(`push traceId=${input.traceId ?? 'internal'} syncLogId=${r.syncLogId}`);
       const agg = r.perResource.reduce(
         (a, s) => ({
           upserted: a.upserted + s.patientsUpserted,
@@ -111,6 +122,7 @@ export class PushReceiverService {
       // ⚠️ plan 不在此触发(persona 完成后链式 enqueue),与单患者刷新路径有意区分。
       // 预检只保留推送记录,不写业务数据、不入队(touched 本就为空,此处显式短路)。
       let personaEnqueued = 0;
+      let personaEnqueueFailed = 0;
       for (const { patientId, tenantId } of dryRun ? [] : r.touched) {
         try {
           await this.queue.enqueuePersonaRecompute(
@@ -119,12 +131,15 @@ export class PushReceiverService {
           );
           personaEnqueued++;
         } catch (err) {
+          personaEnqueueFailed++;
+          hasFailures = true;
           this.logger.warn(
             `enqueue persona-recompute failed patient=${patientId}: ${err instanceof Error ? err.message : err}`,
           );
         }
       }
 
+      timing('persona-enqueue-done');
       this.logger.log(
         `push/rows host=${hostName} source=${source} rows=${rows.length} ` +
           `txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} stubs=${agg.stubs} ` +
@@ -147,6 +162,7 @@ export class PushReceiverService {
         : [];
 
       return {
+        hasFailures,
         syncLogId: r.syncLogId,
         source,
         accepted: rows.length,
@@ -158,6 +174,7 @@ export class PushReceiverService {
         /// 上面那些拒收的定位明细(封顶若干条),宿主据此定位到自己源库的具体行
         factRejects,
         personaEnqueued,
+        personaEnqueueFailed,
         /// 本批新建的空壳患者主档数(引用的患者号在 PAC 找不到)。持续 >0 = 患者号对不上
         patientStubsCreated: agg.stubs,
         mappingMisses: r.mappingMisses.length,

apps/pac-service/src/modules/sync/push/push.controller.ts

+import { randomUUID } from 'node:crypto';
+import { performance } from 'node:perf_hooks';
 import {
   Body,
+  Logger,
   Controller,
   Get,
   Headers,
@@ -44,6 +47,7 @@ import {
 @ApiTags('push')
 @Controller('push')
 export class PushController {
+  private readonly logger = new Logger(PushController.name);
   constructor(
     private readonly verifier: HmacVerifier,
     private readonly receiver: PushReceiverService,
@@ -160,6 +164,15 @@ export class PushController {
     @Query('dryRun') dryRunQuery: string | undefined,
     @Body() body: PushRowsRequest,
   ) {
+    const traceId = randomUUID();
+    const started = performance.now();
+    const loopStart = performance.eventLoopUtilization();
+    this.logger.log(`push request traceId=${traceId} phase=start`);
+    const finish = (event: string) => this.logger.log(
+      `push request traceId=${traceId} event=${event} httpStatus=${req.res?.statusCode ?? 0} durationMs=${Math.round(performance.now() - started)} ` +
+      `eventLoopUtilization=${performance.eventLoopUtilization(loopStart).utilization.toFixed(3)} rssMiB=${Math.round(process.memoryUsage().rss / 1048576)}`);
+    req.res?.once('finish', () => finish('response-finished'));
+    req.res?.once('close', () => { if (!req.res?.writableFinished) finish('client-disconnected'); });
     const rawBody = req.rawBody ? req.rawBody.toString('utf8') : JSON.stringify(body);
 
     const { hostId, hostName } = await this.verifier.verify({
@@ -169,6 +182,7 @@ export class PushController {
       rawBody,
     });
 
+    this.logger.log(`push request traceId=${traceId} phase=verified durationMs=${Math.round(performance.now() - started)}`);
     const parsed = this.parseOrReject(() => PushRowsRequestSchema.parse(body));
 
     // ?dryRun=1 / true / yes 均视为预检(宿主客户端拼 query 习惯不一)
@@ -176,6 +190,7 @@ export class PushController {
 
     return this.withHostLock(hostId, hostName, () =>
       this.receiver.receiveRows({
+        traceId,
         hostId,
         hostName,
         source: parsed.source,

apps/pac-service/src/modules/sync/push/push.schema.ts

@@ -80,6 +80,8 @@ export const PushRowsRequestSchema = z.object({
 export type PushRowsRequest = z.infer<typeof PushRowsRequestSchema>;
 
 export const PushRowsResponseSchema = z.object({
+  personaEnqueueFailed: z.number().int().optional().describe('画像入队失败数'),
+  hasFailures: z.boolean().optional().describe('管线或后续画像入队包含失败，不能仅凭 HTTP 成功忽略'),
   syncLogId: z.string().uuid(),
   source: z.string(),
   accepted: z.number().int().describe('收到的原生行数,不是成功写入数'),

apps/pac-service/tests/assembler-column-drift.spec.ts

@@ -35,7 +35,7 @@ test('已提供的可选字段类型错误仍拒收，不能把可选误当不
 test('历史原文已存在的列消失仍能检测，不关闭历史漂移告警', async () => {
   const baseline = { ...good, new_optional_path: 'old/path.jpg' };
   const db = { patientTransaction: { findMany: jest.fn().mockResolvedValue(
-    Array.from({ length: 5 }, () => ({ rawPayload: baseline }))) } };
+    Array.from({ length: 5 }, () => ({ rawPayload: baseline, createdAt: new Date() }))) } };
   const cold = new ColdImportService(db as never, {} as never, {} as never, {} as never,
     {} as never, {} as never, {} as never);
   const detector = cold as unknown as { detectRawColumnDrift(args: {

apps/pac-service/tests/push-audit-queue.spec.ts

+import { PushAuditQueue } from '../src/modules/sync/push/push-audit-queue';
+import { Logger } from '@nestjs/common';
+Logger.overrideLogger(false);
+test('审计单任务在途、积压有上限，失败后继续处理', async () => {
+  const queue = new PushAuditQueue();
+  let release!: () => void;
+  const order: string[] = [];
+  queue.enqueue('slow', () => new Promise<void>(resolve => { release = resolve; }));
+  await new Promise(resolve => setImmediate(resolve));
+  queue.enqueue('bad', async () => { order.push('bad'); throw new Error('network'); });
+  for (let i = 0; i < 99; i++) expect(queue.enqueue(String(i), async () => { order.push(String(i)); })).toBe(true);
+  expect(queue.enqueue('overflow', async () => {})).toBe(false);
+  expect(order).toEqual([]);
+  release();
+  await new Promise(resolve => setImmediate(resolve));
+  expect(order).toHaveLength(100);
+  expect(order[0]).toBe('bad');
+  expect(order[99]).toBe('98');
+});

apps/pac-service/tests/push-patient-index.spec.ts

+import { ColdImportService } from '../src/modules/sync/cold-import/cold-import.service';
+import { Logger } from '@nestjs/common';
+Logger.overrideLogger(false);
+
+test('小批事件只读取引用患者，保留跨品牌同号隔离', async () => {
+  const patient = { findMany: jest.fn(async ({ where }: { where: { sourceUnit: string } }) =>
+    [{ id: `pac-${where.sourceUnit}`, externalId: 'same-id', sourceUnit: where.sourceUnit }]),
+    upsert: jest.fn() };
+  const resolutions: string[] = [];
+  const synth = { synthesize: jest.fn((input: { patientIdResolver: (id: string) => string }) => {
+    resolutions.push(input.patientIdResolver('same-id')); return null;
+  }) };
+  const assembler = { assemble: () => ({ rows: ['brand-a', 'brand-b'].map((unit) => ({
+    canonical: { patientExternalId: 'same-id' }, rawSource: { unit },
+  })), emits: {}, stats: { fetched: 2, failed: 0, mappingMisses: [], suspectFields: [] }, rejections: [] }) };
+  const cold = new ColdImportService({ patient } as never, synth as never, {} as never,
+    assembler as never, {} as never, {} as never, {} as never);
+  const subject = cold as unknown as { processSubject(...args: unknown[]): Promise<unknown> };
+  await subject.processSubject({}, { canonical: 'appointment' }, 'host',
+    { resolve: () => 'tenant' }, { resolve: (raw: { unit: string }) => raw.unit }, new Set(), {}, undefined, false);
+  expect(patient.findMany).toHaveBeenCalledTimes(2);
+  for (const unit of ['brand-a', 'brand-b']) {
+    expect(patient.findMany).toHaveBeenCalledWith({
+      where: { hostId: 'host', tenantId: 'tenant', sourceUnit: unit, externalId: { in: ['same-id'] } },
+      select: { id: true, externalId: true, sourceUnit: true },
+    });
+  }
+  expect(resolutions).toEqual(['pac-brand-a', 'pac-brand-b']);
+  expect(patient.upsert).not.toHaveBeenCalled();
+});

apps/pac-service/tests/push-upsert-batch-status.spec.ts

@@ -15,6 +15,7 @@ type UpsertResource = 'patient' | 'patient_relation' | 'patient_return_visit';
 
 function setup(resource: UpsertResource) {
   const prisma = {
+    $executeRaw: jest.fn().mockResolvedValue(1),
     host: { findFirst: jest.fn().mockResolvedValue({ id: 'host-demo' }) },
     syncLog: {
       create: jest.fn().mockResolvedValue({ id: 'sync-demo' }),
@@ -228,3 +229,50 @@ test.each([
   } as never);
   expect(await s.send([s.good, s.bad])).toMatchObject({ ...progress, failed: 1 });
 });
+
+
+test('正式推送等待业务写入完成，不能只留原文就返回成功', async () => {
+  const s = setup('patient');
+  let finish!: (v: { id: string }) => void;
+  s.prisma.patient.upsert.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
+  let returned = false;
+  const response = s.send([s.good]).then(value => { returned = true; return value; });
+  await new Promise(resolve => setImmediate(resolve));
+  expect(returned).toBe(false);
+  expect(s.prisma.syncLog.update).not.toHaveBeenCalled();
+  finish({ id: 'patient-1' });
+  expect(await response).toMatchObject({ failed: 0, hasFailures: false });
+});
+
+test('慢历史检查和慢告警不阻塞正常业务回执', async () => {
+  const s = setup('patient');
+  let finish!: (v: []) => void;
+  jest.spyOn(s.dependencies, 'detectRawColumnDrift').mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
+  const result = await s.send([s.good]);
+  expect(result).toMatchObject({ failed: 0, hasFailures: false });
+  await new Promise(resolve => setImmediate(resolve));
+  expect(s.dependencies.detectRawColumnDrift).toHaveBeenCalled();
+  finish([]);
+  await new Promise(resolve => setImmediate(resolve));
+  expect(s.prisma.$executeRaw).toHaveBeenCalled();
+});
+
+test('辅助审计失败不把已落库成功改成推送失败', async () => {
+  const s = setup('patient');
+  jest.spyOn(s.dependencies, 'detectRawColumnDrift').mockRejectedValueOnce(new Error('audit unavailable'));
+  expect(await s.send([s.good])).toMatchObject({ failed: 0, hasFailures: false });
+  await new Promise(resolve => setImmediate(resolve));
+  expect(s.evidence.markFailed).not.toHaveBeenCalled();
+});
+
+test('预检仍等待完整历史字段检查', async () => {
+  const s = setup('patient');
+  let finish!: (v: []) => void;
+  jest.spyOn(s.dependencies, 'detectRawColumnDrift').mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
+  let returned = false;
+  const response = s.send([s.good], true).then(value => { returned = true; return value; });
+  await new Promise(resolve => setImmediate(resolve));
+  expect(returned).toBe(false);
+  finish([]);
+  expect(await response).toMatchObject({ dryRun: true });
+});

docs/qa/2026-10-14-push-latency-local.md

+# Push 关键路径优化与排障（本地）
+
+分支 `codex/push-latency-fix`，基于 main `99fb2648cacdafcd0f26ec8540d04c8465ecbb87`。独立 worktree，主目录及正在运行的本地服务未变。未提交、未部署。
+
+## 最终边界
+
+- 删除此前新增的异步受理参数、receipt 接口、收件箱及其迁移/测试。旧方案仅在 /tmp 留临时备份。
+- 正式推送继续等待请求留档、校验、主档/流水/事实写入、同步结果记录及画像任务入队。
+- 事实写入仍使用原管线及事务，未把业务写入挪到内存任务。部分失败、事实拒收、画像入队失败可见。
+- 历史原文列对比、外部告警通知不等待；辅助审计仅一个在途任务、最多积压 100 个。队满/失败记错误日志。
+- 辅助审计是非持久任务，进程退出可能丢失未执行检查；业务原文、写入与失败记录已持久保留。历史检查完成后原子补充 metadata，不覆盖逐行失败证据。
+- dryRun 仍同步完成完整校验；正式响应 suspectFields 是当前批校验结果，历史对比见后台记录。
+
+## 已确认的优化
+
+- 小批事件仅查询本批患者，按 host/tenant/sourceUnit/externalId 保留隔离，避免拉整租户 13 万患者。
+- 历史检查按事件类型读取各自最近 50 条，合并最近 50 条。唯一新增迁移为 patient_transactions(host_id,subject_type,created_at DESC) 并发索引。
+- 该索引此前在独立临时库真实执行过迁移及 10 万合成流水验证，命中索引；此数据不代表线上耗时。
+- 请求 traceId 关联 evidenceId/syncLogId，记录验签、客户端断连、响应完成、事件循环利用率和 RSS。
+- pipeline 分阶段耗时进入同步 metadata；流水写入、查回、事实写入开始/结束另有日志，耗时包含等待实际写入返回。
+
+## 本次验证
+
+- 14 个测试集，137 项通过；源码及 tests 类型检查通过；git diff --check 通过。
+- 新增覆盖：业务写入被阻塞时不得返回；历史检查被阻塞时可返回已完成业务结果；辅助审计异常不把业务成功改成失败；预检仍等待完整检查；审计串行、容量上限及异常后继续消费。
+- 保留 Friday 病历、专属客服、重解析、推送隔离及字段漂移回归。
+
+## 测试服务器只读现场（2026-10-14，北京时间）
+
+- 18:34 左右：可用内存约 9.2GiB，PAC 容器约 852MiB，CPU 快照约 100%；不能由这些数值断言进程卡死。
+- vmstat 两个实时采样：I/O wait 42% / 58%，读取约 150MB/s，swap in/out 当时为 0（已有 swap 占用不等于正在换页）。
+- 18:36:26 /health JSON status=ok，约 9ms。
+- pidstat 显示 PostgreSQL 主查询及两个并行进程为主要读盘来源；宿主 PID 与容器 NSpid 已对应。
+- pg_stat_activity 抓到同组 3 个进程：patient_transactions SELECT，IO/DataFileRead，已运行约 131 秒。当时 sync_logs 没有 running 项。
+- 再取完整 SQL 时该连接已消失，未能获得完整语句和对应业务编号。因此已确认当前存在数据库读盘压力，但不能将其直接认定为下午预约超时的具体 SQL，也不能断言是定时同步造成。
+- 未取消查询、重启、写服务器数据或触发新推送。新阶段日志部署后才能用于下一次慢批次关联。
+
+## 后续验证
+
+上线前执行唯一索引迁移；CONCURRENTLY 不包事务，中断后检查索引有效性。上线后按断连 traceId → evidenceId/syncLogId → 阶段耗时，与同期 pg_stat_activity/磁盘读写关联。不能用本地通过或健康接口快替代真实慢批次验证。