Merge push latency optimization into test

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/cold-import/cold-import.service.ts

+import { PushAuditQueue } from '../push/push-audit-queue';
 import { Injectable, Logger } from '@nestjs/common';
 import * as fs from 'node:fs';
 import * as path from 'node:path';
@@ -9,7 +10,7 @@ import { recordAssemblerRejections, recordRowRejection, rowRejections, rowReject
 import { ApiCode, SyncDirection, SyncStatus } from '@pac/types';
 import type { Action, CanonicalResourceKey } from '@pac/types';
 import { BizError } from '../../../common/errors/biz-error';
-import type { Prisma } from '@prisma/client';
+import { Prisma } from '@prisma/client';
 import { PrismaService } from '../../../prisma/prisma.service';
 import { TransactionSynthesizer } from '../pipeline/transaction-synthesizer';
 import {
@@ -118,6 +119,7 @@ function patientIndexKey(sourceUnit: string, externalId: string): string {
  */
 @Injectable()
 export class ColdImportService {
+  private readonly pushAudit = new PushAuditQueue();
   private readonly logger = new Logger(ColdImportService.name);
   // ⚠️ source_unit 解析器**只能是局部值,逐层传参**,绝不能挂实例字段。
   //    本服务是 Nest 单例,而 push(webhook,按请求并发)与 cold-import / reparse / pull
@@ -373,6 +375,7 @@ export class ColdImportService {
     return { perResource, affectedPatientIds, dryRunDiffs };
   }
 
+
   /**
    * 形态 A push 入口:宿主推来某张【原生源表】的原生行 → 复用 transforms + assembler + processSubject 落库。
    *
@@ -462,10 +465,19 @@ export class ColdImportService {
       const inp = (tf as { input?: string }).input;
       if (inp && !tables[inp]) tables[inp] = [];
     }
+    let stageAt = Date.now();
+    const timingsMs: Record<string, number> = {};
+    const recordStage = (stage: string) => {
+      const now = Date.now();
+      timingsMs[stage] = now - stageAt;
+      this.logger.log(`push timing syncLogId=${syncLog.id} stage=${stage} durationMs=${now - stageAt}`);
+      stageAt = now;
+    };
     await this.hydratePushLookupTables(tables, transforms, host.id, sourceUnitResolver);
     const transformed =
       transforms.length > 0 ? this.transformEngine.run({ tables, transforms }) : tables;
 
+    recordStage('lookup-transform');
     const seenTenants = new Set<string>();
     const perResource: PerResourceStats[] = [];
     // 请求内收集,不放单例状态:归属移出者未出现在本批新 transaction 中也必须触发重算。
@@ -496,6 +508,7 @@ export class ColdImportService {
           );
         }
       }
+      recordStage('upsert');
       for (const cfg of targetCfgs) {
         const stats = await this.processSubject(
           transformed,
@@ -510,6 +523,7 @@ export class ColdImportService {
           affectedPatients,
         );
         perResource.push(stats);
+        recordStage(`subject:${cfg.canonical}`);
       }
     } catch (err) {
       firstError = err instanceof Error ? err.message : String(err);
@@ -540,15 +554,19 @@ export class ColdImportService {
     // ── raw 层列漂移检测(form A 专属,最可靠):本次推送行的列集 vs 历史同源 rawPayload 列集 ──
     //   transform 会掩盖 assembler 层的删列(补 key),故在原生行层面直接对比。基线 = 已存 transaction,
     //   不需新存储。基线样本太少(新 host 冷启)则跳过,避免误报。
-    const rawDrift = await this.detectRawColumnDrift({
+    recordStage('touched');
+    const auditInput = {
       hostId: host.id,
       source: opts.source,
       subjectTypes: targetCfgs.map((c) => c.emits!.subjectType),
       pushedCols,
       rowCount: opts.rows.length,
-      excludeSyncLogId: syncLog.id, // ⚠️ 基线排除本 run 刚写的行(否则加列会被自己污染成基线)
-    });
+      excludeSyncLogId: syncLog.id,
+    };
+    // 预检仍同步返回完整检查；正式推送不等待历史样本和外部通知。
+    const rawDrift = opts.dryRun ? await this.detectRawColumnDrift(auditInput) : [];
 
+    recordStage(opts.dryRun ? 'column-drift' : 'column-drift-deferred');
     // 摄入漂移(保障机制)→ SyncLog.metadata,供审计:
     //   mappingMisses = 落 _default 的原值(扩字典);suspectFields = 疑似宿主改表(列缺席/删/加)。
     const mappingMisses = mergeMappingMisses(perResource.map((s) => s.mappingMisses));
@@ -614,31 +632,17 @@ export class ColdImportService {
         failed: agg.failed,
         errorMessage: firstError ?? (allRowsFailed ? '整批全部行失败(疑似关键字段缺失/改名)' : null),
         endedAt: new Date(),
-        ...(driftMeta ? { metadata: driftMeta as unknown as Prisma.InputJsonValue } : {}),
+        metadata: { ...(driftMeta ?? {}), timingsMs } as unknown as Prisma.InputJsonValue,
       },
     });
 
-    // ── push 路径主动告警(纯 push 宿主没有每日 pull cron 的探针,这里是唯一的"有人知道"渠道)──
-    // 只挑两类硬信号:疑似改表 / 失败率超阈。mappingMisses 不告警(长尾常态,靠 admin 页巡检 + 响应透出)。
-    const failedRatio = opts.rows.length > 0 ? agg.failed / opts.rows.length : 0;
-    if (suspectFields.length > 0) {
-      await this.alerter.send({
-        level: 'warning',
-        title: `push 疑似宿主改表:${opts.hostName}/${opts.source}`,
-        body: suspectFields.map((s) => `[${s.reason}] ${s.resource}.${s.field}`).join('\n'),
-        context: { host: opts.hostName, syncLogId: syncLog.id },
-      });
-    }
-    if (agg.failed > 0 && failedRatio >= 0.1) {
-      await this.alerter.send({
-        level: allRowsFailed ? 'critical' : 'warning',
-        title: `push 摄入失败率 ${(failedRatio * 100).toFixed(0)}%:${opts.hostName}/${opts.source}`,
-        body:
-          `rows=${opts.rows.length} txn=${agg.txn} dup=${agg.dup} failed=${agg.failed}` +
-          (firstError ? `\nfirstError: ${firstError}` : ''),
-        context: { host: opts.hostName, syncLogId: syncLog.id },
-      });
-    }
+    recordStage('sync-log-finalize');
+    const audit = {
+      ...auditInput, hostName: opts.hostName, suspectFields,
+      failed: agg.failed, transactionsWritten: agg.txn, duplicates: agg.dup,
+      allRowsFailed, firstError, dryRun: !!opts.dryRun,
+    };
+    this.pushAudit.enqueue(syncLog.id, () => this.runPushAudit(audit));
 
     this.logger.log(
       `ingestRawTables host=${opts.hostName} source=${opts.source} rows=${opts.rows.length} ` +
@@ -797,6 +801,40 @@ export class ColdImportService {
    * 基线 = 已存 transaction 的 rawPayload(同 subjectType 抽样 50 条),不需新存储。
    * 基线样本 < 5 条(新 host 冷启)→ 跳过,避免误报。`_` 开头的内部 meta key 忽略。
    */
+  private async runPushAudit(input: {
+    hostId: string; hostName: string; source: string; subjectTypes: string[]; pushedCols: Set<string>;
+    rowCount: number; excludeSyncLogId: string; suspectFields: SuspectField[];
+    failed: number; transactionsWritten: number; duplicates: number;
+    allRowsFailed: boolean; firstError: string | null; dryRun: boolean;
+  }): Promise<void> {
+    const started = Date.now();
+    const context = { host: input.hostName, syncLogId: input.excludeSyncLogId };
+    let fields = input.suspectFields;
+    try {
+      if (!input.dryRun) {
+        fields = mergeSuspectFields([fields, await this.detectRawColumnDrift(input)]);
+        // 原子追加审计字段，不覆盖同步阶段的逐行失败等证据。
+        const patch = JSON.stringify({ suspectFields: fields, rawColumnDriftCheckedAt: new Date().toISOString() });
+        await this.prisma.$executeRaw`
+          UPDATE sync_logs SET metadata=COALESCE(metadata,'{}'::jsonb) || ${patch}::jsonb
+          WHERE id=${input.excludeSyncLogId}::uuid AND host_id=${input.hostId}::uuid`;
+      }
+      if (fields.length) await this.alerter.send({
+        level: 'warning', title: `push 疑似宿主改表:${input.hostName}/${input.source}`,
+        body: fields.map((s) => `[${s.reason}] ${s.resource}.${s.field}`).join('\n'), context,
+      });
+    } finally {
+      // 漂移检查失败也不能吞掉已确认的业务失败告警。
+      if (input.failed > 0 && input.failed / input.rowCount >= 0.1) await this.alerter.send({
+        level: input.allRowsFailed ? 'critical' : 'warning',
+        title: `push 摄入失败率 ${(input.failed / input.rowCount * 100).toFixed(0)}%:${input.hostName}/${input.source}`,
+        body: `rows=${input.rowCount} txn=${input.transactionsWritten} dup=${input.duplicates} failed=${input.failed}` +
+          (input.firstError ? `\nfirstError: ${input.firstError}` : ''), context,
+      });
+      this.logger.log(`push timing syncLogId=${input.excludeSyncLogId} stage=background-audit durationMs=${Date.now() - started}`);
+    }
+  }
+
   private async detectRawColumnDrift(args: {
     hostId: string;
     source: string;
@@ -806,16 +844,15 @@ export class ColdImportService {
     excludeSyncLogId: string;
   }): Promise<SuspectField[]> {
     const MIN_BASELINE = 5;
-    const sample = await this.prisma.patientTransaction.findMany({
-      where: {
-        hostId: args.hostId,
-        subjectType: { in: args.subjectTypes },
-        syncLogId: { not: args.excludeSyncLogId }, // 排除本 run 刚写的行(防自污染)
-      },
-      select: { rawPayload: true },
-      take: 50,
+    const candidates: Array<{ rawPayload: unknown; createdAt: Date }> = [];
+    for (const subjectType of new Set(args.subjectTypes)) {
+      candidates.push(...await this.prisma.patientTransaction.findMany({
+        where: { hostId: args.hostId, subjectType, syncLogId: { not: args.excludeSyncLogId } },
+        select: { rawPayload: true, createdAt: true }, take: 50,
         orderBy: { createdAt: 'desc' },
-    });
+      }));
+    }
+    const sample = candidates.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime()).slice(0, 50);
     if (sample.length < MIN_BASELINE) return [];
     // 同 subjectType 可能混多源(如 refund = 结算头 patient_settlement ∪ 结算明细
     // patient_settlement_spec,列集不同)。按「与本批列签名的 Jaccard 相似度」自选**同源**样本,
@@ -1982,6 +2019,17 @@ export class ColdImportService {
 
     // v2.3:per-tenant lazy patient index(多 tenant 时按需加载,避免一次拉全 host)
     const patientIndexByTenant = new Map<string, Map<string, string>>();
+    const patientRefsByTenant = new Map<string, Map<string, Set<string>>>();
+    for (const { canonical, rawSource } of assembled.rows) {
+      const tenant = tenantResolver.resolve(rawSource);
+      const ref = (canonical as Record<string, unknown>).patientExternalId;
+      if (!tenant || typeof ref !== 'string' || !ref) continue;
+      const unit = sourceUnitResolver.resolve(rawSource);
+      if (!patientRefsByTenant.has(tenant)) patientRefsByTenant.set(tenant, new Map());
+      const units = patientRefsByTenant.get(tenant)!;
+      if (!units.has(unit)) units.set(unit, new Set());
+      units.get(unit)!.add(ref);
+    }
     // ⚠️ sourceContext 必须跨 run 稳定(否则 source_event_id 永远不重复)
     const sourceContext = `cold-import`;
 
@@ -2033,7 +2081,11 @@ export class ColdImportService {
         // 懒加载该 tenant 的 patient index
         let patientIndex = patientIndexByTenant.get(tenantId);
         if (!patientIndex) {
-          patientIndex = await this.buildPatientIndex(hostId, tenantId);
+          patientIndex = new Map();
+          for (const [unit, refs] of patientRefsByTenant.get(tenantId) ?? []) {
+            const subset = await this.buildPatientIndex(hostId, tenantId, [...refs], unit);
+            for (const [key, id] of subset) patientIndex.set(key, id);
+          }
           patientIndexByTenant.set(tenantId, patientIndex);
         }
         const sourceUnit = sourceUnitResolver.resolve(rawSource);
@@ -2151,6 +2203,14 @@ export class ColdImportService {
     const { tenantId, buffer, stats, config, affectedPatients } = args;
     if (buffer.length === 0) return;
     const hostId = buffer[0]!.tx.hostId as string;
+    const syncLogId = buffer[0]!.tx.syncLogId ?? 'none';
+    let stageAt = Date.now();
+    const stage = (name: string) => {
+      const now = Date.now();
+      this.logger.log(`ingest write syncLogId=${syncLogId} resource=${config.canonical} stage=${name} durationMs=${now - stageAt} rows=${buffer.length}`);
+      stageAt = now;
+    };
+    stage('transaction-write-start');
 
     // 1. createMany tx(skipDuplicates 跳重)
     const data = buffer.map((b) => b.tx);
@@ -2173,6 +2233,7 @@ export class ColdImportService {
       await this.fallbackPerRowWrite(buffer, stats, config, affectedPatients);
       return;
     }
+    stage('transaction-write-done');
     stats.transactionsWritten += createdCount;
     stats.duplicates += buffer.length - createdCount;
 
@@ -2206,6 +2267,7 @@ export class ColdImportService {
         }),
       'findMany tx by sourceEventId',
     );
+    stage('transaction-read-done');
     const txBySourceEventId = new Map(txRows.map((t) => [t.sourceEventId!, t]));
 
     // 3. 构建 batch items,跑 parserPipeline.runForBatch
@@ -2236,7 +2298,9 @@ export class ColdImportService {
     }
     if (items.length === 0) return;
 
+    stage('fact-write-start');
     const metrics = await this.parserPipeline.runForBatch(items);
+    stage('fact-write-done');
     if (affectedPatients) {
       for (const item of items) {
         if (item.transaction.patientId) affectedPatients.set(item.transaction.patientId, tenantId);
@@ -2353,15 +2417,18 @@ export class ColdImportService {
   private async buildPatientIndex(
     hostId: string,
     tenantId: string,
+    externalIds?: string[],
+    sourceUnit?: string,
   ): Promise<Map<string, string>> {
-    const rows = await this.withDbRetry(
-      () =>
-        this.prisma.patient.findMany({
-          where: { hostId, tenantId },
+    const rows: Array<{ id: string; externalId: string; sourceUnit: string }> = [];
+    const chunks = externalIds === undefined ? [undefined] :
+      Array.from({ length: Math.ceil(externalIds.length / 1000) }, (_, i) => externalIds.slice(i * 1000, (i + 1) * 1000));
+    for (const refs of chunks) {
+      rows.push(...await this.withDbRetry(() => this.prisma.patient.findMany({
+        where: { hostId, tenantId, ...(sourceUnit !== undefined ? { sourceUnit } : {}), ...(refs ? { externalId: { in: refs } } : {}) },
         select: { id: true, externalId: true, sourceUnit: true },
-        }),
-      'buildPatientIndex',
-    );
+      }), 'buildPatientIndex'));
+    }
     // 集团内 external_id 可跨 source_unit 撞号 → 索引按 (source_unit, external_id) 建键
     return new Map(rows.map((p) => [patientIndexKey(p.sourceUnit, p.externalId), p.id]));
   }

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/磁盘读写关联。不能用本地通过或健康接口快替代真实慢批次验证。