Commit 8d6664c8 by luoqi

fix(sync): CLI 启动会清掉正在跑的同步锁 —— 生产实测夭折一轮增量

🔴 事故(2026-08-30 生产):在 pac-service 容器里 docker exec 跑 recompute-plans,
   08:17:11 起进程 → 08:17:13 正在跑的 08:15 那轮同步被标 failed。前 19 轮全 success,
   只死了撞上的这一轮。数据未丢(cursor_after=null,下轮同水位 catchup),但白丢一轮。

根因:每个 CLI 都 createApplicationContext(AppModule) → 跑一遍
SyncIncrementalScheduler.onModuleInit → reapStaleRunningLocks。
原判据是「startedAt < 本进程启动 = 僵尸锁」,注释里写着"两者的进程都不可能比本进程
启动得更早还活着"—— 但**长驻的 pac-service 恰恰就是那个更早启动还活着的进程**。
理由写反了方向,而且只在"回收逻辑跑在长驻服务里"时才成立。

两道防线:
① scheduler 加年龄阈值 REAP_MIN_AGE_MS=3h —— 真僵尸锁必然躺很久,正在跑的不会。
   用年龄区分,不靠猜进程身份。(生产单轮摄入实测 28~52 分钟)
② 新增 src/cli/bootstrap-flags.ts,16 个 CLI 在建上下文**之前**设
   PAC_SCHEDULER_DISABLED=1(该总闸本就会跳过回收,只是没人用)。
   豁免 sync-incremental.cli(它就是要触发同步),由 ① 兜底。

回归测试 tests/cli-scheduler-guard.spec.ts:遍历所有会建上下文的 CLI,
断言调用存在**且位置早于** createApplicationContext;并锁住年龄阈值 ≥2h。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
parent 0c76093a
Pipeline #3630 failed in 0 seconds
......@@ -15,6 +15,7 @@ import { Logger } from '@nestjs/common';
import { AppModule } from '../app.module';
import { PrismaService } from '../prisma/prisma.service';
import { PlanScriptOrchestrator } from '../modules/ai/orchestrators/plan-script.orchestrator';
import { disableSchedulersForCli } from './bootstrap-flags';
interface CliArgs {
planId?: string;
......@@ -81,6 +82,8 @@ async function bootstrap() {
}
const logger = new Logger('ai:gen-script');
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['warn', 'error', 'log'],
});
......
......@@ -25,6 +25,7 @@ import { NestFactory } from '@nestjs/core';
import { Logger } from '@nestjs/common';
import { AppModule } from '../app.module';
import { PlanLabelService } from '../modules/plan/plan-label.service';
import { disableSchedulersForCli } from './bootstrap-flags';
async function main(): Promise<void> {
const log = new Logger('backfill-plan-labels');
......@@ -32,6 +33,10 @@ async function main(): Promise<void> {
const all = argv.includes('--all');
const checkOnly = argv.includes('--check');
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['error', 'warn', 'log'],
});
......
/**
* CLI 启动前的写侧总闸 —— **必须在 `NestFactory.createApplicationContext` 之前调用**。
*
* 🔴 为什么存在(2026-08-30 生产事故):
* 每个 CLI 都会 `createApplicationContext(AppModule)`,于是把**整个应用的 onModuleInit
* 全跑一遍** —— 包括 SyncIncrementalSchedulerService 的僵尸锁回收和 cron 注册。
* 在长驻的 pac-service 容器里 `docker exec` 跑一个 CLI,那个 CLI 就会把 service 里
* **正在跑**的那轮同步的锁当成僵尸清掉,那轮增量当场夭折。
* 实测:08:17:11 跑 recompute-plans → 08:17:13 正常跑着的 08:15 那轮被标 failed。
*
* 所以:凡是**不以调度器为目的**的 CLI,一律先调本函数。
* ⚠️ 例外只有 `sync-incremental.cli`(它本身就是要触发同步)—— 但它也不该回收别人的锁,
* 那一层由 scheduler 的年龄阈值(REAP_MIN_AGE_MS)兜底。
*/
export function disableSchedulersForCli(): void {
process.env.PAC_SCHEDULER_DISABLED = '1';
}
......@@ -15,6 +15,7 @@ import {
ColdImportService,
SyncAlreadyRunningError,
} from '../modules/sync/cold-import/cold-import.service';
import { disableSchedulersForCli } from './bootstrap-flags';
interface CliArgs {
dir?: string;
......@@ -112,6 +113,10 @@ async function bootstrap() {
`${args.since ? `, since=${args.since}${args.months ? `(--months=${args.months})` : ''}` : ''})`,
);
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['log', 'warn', 'error'],
});
......
......@@ -21,6 +21,7 @@ import { NestFactory } from '@nestjs/core';
import { Logger } from '@nestjs/common';
import { AppModule } from '../app.module';
import { HostsService } from '../modules/admin/hosts.service';
import { disableSchedulersForCli } from './bootstrap-flags';
interface ParsedArgs {
command: string;
......@@ -90,6 +91,8 @@ async function main() {
}
const logger = new Logger('pac:host');
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['warn', 'error'],
});
......
......@@ -20,6 +20,7 @@ import { ColdImportService } from '../modules/sync/cold-import/cold-import.servi
import { PersonaService } from '../modules/persona/persona.service';
import { PlanEngineService } from '../modules/plan/engine/plan-engine.service';
import * as path from 'node:path';
import { disableSchedulersForCli } from './bootstrap-flags';
interface Args {
host: string;
......@@ -73,6 +74,8 @@ async function bootstrap() {
}
const logger = new Logger('import-patient');
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['log', 'warn', 'error'],
});
......
......@@ -16,6 +16,7 @@ import { NestFactory } from '@nestjs/core';
import { Logger } from '@nestjs/common';
import { AppModule } from '../app.module';
import { PrismaService } from '../prisma/prisma.service';
import { disableSchedulersForCli } from './bootstrap-flags';
interface Row {
fileNum: string;
......@@ -48,6 +49,10 @@ async function main(): Promise<void> {
const rows = parseCsv(csvArg);
logger.log(`对照表 ${rows.length} 行(去表头/垃圾后)`);
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, { logger: ['error', 'warn'] });
const prisma = app.get(PrismaService);
try {
......
......@@ -17,6 +17,7 @@ import { PrismaService } from '../prisma/prisma.service';
import { RedisService } from '../redis/redis.service';
import { doctorOptionsCacheKey } from '../modules/plan/plan.service';
import { runPool } from '../common/run-pool';
import { disableSchedulersForCli } from './bootstrap-flags';
interface Args {
host: string;
......@@ -60,6 +61,8 @@ async function bootstrap() {
if (args.concurrency > 1 && !process.env.PAC_DB_CONCURRENCY) {
process.env.PAC_DB_CONCURRENCY = String(args.concurrency);
}
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['log', 'warn', 'error'],
});
......
......@@ -16,6 +16,7 @@ import { Logger } from '@nestjs/common';
import { AppModule } from '../app.module';
import { PlanEngineService } from '../modules/plan/engine/plan-engine.service';
import { PrismaService } from '../prisma/prisma.service';
import { disableSchedulersForCli } from './bootstrap-flags';
interface Args {
host: string;
......@@ -47,6 +48,8 @@ async function bootstrap() {
const n = Math.max(1, Number(process.env.PAC_PLAN_BATCH_CONCURRENCY) || 8);
if (n > 1) process.env.PAC_DB_CONCURRENCY = String(n);
}
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['log', 'warn', 'error'],
});
......
......@@ -25,6 +25,7 @@ import { createClient } from '@clickhouse/client';
import { AppModule } from '../app.module';
import { PrismaService } from '../prisma/prisma.service';
import { ColdImportManifestSchema } from '../modules/sync/cold-import/manifest.schema';
import { disableSchedulersForCli } from './bootstrap-flags';
// CLI 是短命进程,不需要 org-tree 启动预热(且会在 app.close() 时跟后台 warmAll 抢连接报噪音)。
process.env.PAC_ORGTREE_WARM_ON_BOOT = 'false';
......@@ -118,6 +119,10 @@ async function main(): Promise<void> {
process.exit(0);
}
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, { logger: ['warn', 'error'] });
try {
const prisma = app.get(PrismaService);
......
......@@ -38,6 +38,7 @@ import { PrismaService } from '../prisma/prisma.service';
import { ColdImportService } from '../modules/sync/cold-import/cold-import.service';
import { PersonaService } from '../modules/persona/persona.service';
import { PlanEngineService } from '../modules/plan/engine/plan-engine.service';
import { disableSchedulersForCli } from './bootstrap-flags';
interface CliArgs {
dir: string;
......@@ -81,6 +82,10 @@ async function main(): Promise<void> {
return;
}
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, { logger: ['log', 'warn', 'error'] });
try {
const prisma = app.get(PrismaService);
......
......@@ -34,6 +34,7 @@ import {
} from '@pac/types';
import { AppModule } from '../app.module';
import { PrismaService } from '../prisma/prisma.service';
import { disableSchedulersForCli } from './bootstrap-flags';
interface Args {
host: string;
......@@ -77,6 +78,8 @@ function mulberry32(seed: number): () => number {
async function main(): Promise<void> {
const logger = new Logger('SeedAssignment');
const args = parseArgs(process.argv.slice(2));
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, { logger: ['warn', 'error'] });
const prisma = app.get(PrismaService);
......
......@@ -13,9 +13,12 @@ import { NestFactory } from '@nestjs/core';
import { Logger } from '@nestjs/common';
import { AppModule } from '../app.module';
import { StaleScanService } from '../queues/stale-scan.service';
import { disableSchedulersForCli } from './bootstrap-flags';
async function main(): Promise<void> {
const logger = new Logger('stale-scan-cli');
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, { logger });
try {
......
......@@ -22,6 +22,7 @@ import { PullOrchestrator } from '../modules/sync/pull/pull.orchestrator';
import { ReconcileOrchestrator } from '../modules/sync/reconcile/reconcile.orchestrator';
import { HmacVerifier } from '../modules/sync/push/hmac-verifier.service';
import { randomUUID, createHash } from 'node:crypto';
import { disableSchedulersForCli } from './bootstrap-flags';
interface CliArgs {
cmd: 'pull-setup' | 'pull' | 'reconcile' | 'push' | 'help';
......@@ -70,6 +71,8 @@ async function bootstrap() {
process.exit(0);
}
const logger = new Logger('sync:test');
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, { logger: ['warn', 'error'] });
try {
......
......@@ -15,6 +15,7 @@ import { AppModule } from '../app.module';
import { PrismaService } from '../prisma/prisma.service';
import { PatientService } from '../modules/patient/patient.service';
import type { TenantScopeContext } from '../common/decorators/tenant-scope.decorator';
import { disableSchedulersForCli } from './bootstrap-flags';
interface CliArgs {
pid?: string; // host externalId
......@@ -91,6 +92,8 @@ async function bootstrap() {
}
const logger = new Logger('timeline:cli');
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['warn', 'error'],
});
......
......@@ -6,6 +6,7 @@ import { NestFactory } from '@nestjs/core';
import { AppModule } from '../app.module';
import { PrismaService } from '../prisma/prisma.service';
import { ChainComposerService } from '../modules/plan/engine/chain-composer.service';
import { disableSchedulersForCli } from './bootstrap-flags';
async function main() {
const id = process.argv.find((a) => a.startsWith('--id='))?.slice('--id='.length);
......@@ -13,6 +14,8 @@ async function main() {
console.error('Usage: --id=<patientId>');
process.exit(1);
}
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, { logger: ['error'] });
const prisma = app.get(PrismaService);
const composer = app.get(ChainComposerService);
......
......@@ -31,6 +31,7 @@ import { TreatmentInitiationRecallScenario } from '../modules/plan/engine/scenar
import { PotentialTreatmentSelector } from '../modules/clinical-gap/potential-treatment.selector';
import type { ScenarioScope } from '../modules/plan/engine/scenario.interface';
import type { GapVariant } from '../modules/clinical-gap/potential-treatment-gap.sql';
import { disableSchedulersForCli } from './bootstrap-flags';
interface Args {
host: string;
......@@ -79,6 +80,8 @@ async function bootstrap(): Promise<number> {
// 而这个 CLI 的输出就是它的全部产物 —— 不能被日志级别吃掉。
const out = (m: string): void => console.log(m);
const bad_ = (m: string): void => console.error(m);
// ⚠️ 必须在建应用上下文之前:否则会清掉 service 里正在跑的同步锁(见 bootstrap-flags)
disableSchedulersForCli();
const app = await NestFactory.createApplicationContext(AppModule, {
logger: ['warn', 'error'],
});
......
......@@ -37,6 +37,12 @@ import { schedulerDisabled } from './scheduler-switch';
*
* 跑失败:cursor 不前进 → 下次自动 catchup;log ERROR 不抛。
*/
/**
* 僵尸锁的最小年龄。比「一轮同步的正常耗时」留足余量 —— 生产实测单轮摄入 28~52 分钟,
* 取 3 小时:真崩溃留下的锁必然远超此值,而正常在跑的绝不会。
*/
const REAP_MIN_AGE_MS = 3 * 60 * 60 * 1000;
@Injectable()
export class SyncIncrementalSchedulerService implements OnModuleInit {
private readonly logger = new Logger(SyncIncrementalSchedulerService.name);
......@@ -144,11 +150,21 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
/**
* 回收僵尸同步锁 —— 把 startedAt 早于本进程启动的 running sync_log 标 failed。
*
* 为什么这样判据安全:sync 只在两处跑 —— 本 service 进程的 cron 回调,或一次性 CLI
* (`sync-incremental.cli` / `cold-import.cli`,跑完即退)。两者的进程都不可能比本进程
* 启动得更早还活着。所以 `startedAt < PROCESS_STARTED_AT` 的 running 行 = 上一个已死进程的残留。
* 反过来,本进程启动后新建的 running 行(startedAt >= PROCESS_STARTED_AT)绝不碰 —— 那可能是
* 正在跑的真锁(例如运维手动触发的 CLI 与本进程并存),误清会把在跑的同步" orphan"掉。
* 🔴 2026-08-30 生产事故:本判据**曾经是错的**,理由写反了方向。
* 原注释说「sync 只在 service 进程的 cron 或一次性 CLI 里跑,两者的进程都不可能比本进程
* 启动得更早还活着」——【但长驻的 pac-service 恰恰就是「启动得更早还活着」的那个】。
* 任何 CLI(recompute-plans / recompute-persona / reparse …)都会
* `createApplicationContext(AppModule)`,于是也跑一遍本 onModuleInit;
* 此时 CLI 进程的 PROCESS_STARTED_AT = 现在,而 service 里**正在跑**的那轮 sync
* startedAt 更早 → 被当成僵尸锁清掉 → 那一轮增量当场夭折。
* 实测:08:17:11 在生产容器里跑 recompute-plans,08:17:13(Nest 启动 2 秒后)
* 08:15 那轮正常同步就被标 failed。前 19 轮全 success,只死了撞上的这一轮。
* 数据没丢(cursor_after=null,下轮按同一水位 catchup),但白丢一轮、晚 2 小时落库。
*
* 现在的判据:`startedAt < 本进程启动` **且** `startedAt < now - REAP_MIN_AGE_MS`。
* 后半条是真正的防线 —— 真僵尸锁是上个进程崩溃留下的,必然已经躺了很久;
* 而被误伤的那种,是"刚起没多久还在正常跑"的。用年龄区分,不靠进程身份猜。
* 另外 CLI 侧统一设 PAC_SCHEDULER_DISABLED=1(见 src/cli/bootstrap-flags.ts),双保险。
*
* 幂等:被标 failed 只是让并发锁释放;数据侧不受影响(游标没推进,下次增量靠 48h 回看窗补齐)。
*
......@@ -157,14 +173,17 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
*/
private async reapStaleRunningLocks(processStartedAt: Date = PROCESS_STARTED_AT): Promise<void> {
try {
// 双条件取更早的那个界:既要早于本进程启动,又要已经躺够 REAP_MIN_AGE_MS。
const ageCutoff = new Date(Date.now() - REAP_MIN_AGE_MS);
const cutoff = ageCutoff < processStartedAt ? ageCutoff : processStartedAt;
const stale = await this.prisma.syncLog.findMany({
where: { status: SyncStatus.RUNNING, startedAt: { lt: processStartedAt } },
where: { status: SyncStatus.RUNNING, startedAt: { lt: cutoff } },
select: { id: true, hostId: true, startedAt: true, triggeredBy: true },
});
if (stale.length === 0) return;
const { count } = await this.prisma.syncLog.updateMany({
where: { status: SyncStatus.RUNNING, startedAt: { lt: processStartedAt } },
where: { status: SyncStatus.RUNNING, startedAt: { lt: cutoff } },
data: {
status: SyncStatus.FAILED,
endedAt: new Date(),
......
/**
* CLI 启动不得清掉正在跑的同步锁(2026-08-30 生产事故回归测试)
*
* 事故:在生产容器里 `docker exec` 跑 recompute-plans,CLI 会
* `createApplicationContext(AppModule)` → 跑一遍 SyncIncrementalScheduler.onModuleInit
* → reapStaleRunningLocks 把 service 里**正在跑**的那轮同步当僵尸锁清掉,那轮增量夭折。
* 实测 08:17:11 起 CLI,08:17:13 正常跑着的 08:15 那轮被标 failed(前 19 轮全 success)。
*
* 两道防线,这里各锁一条。
*/
import * as fs from 'node:fs';
import * as path from 'node:path';
const CLI_DIR = path.resolve(__dirname, '../src/cli');
/// 唯一豁免:它本身就是要触发同步(靠 scheduler 的年龄阈值兜底)
const EXEMPT = new Set(['sync-incremental.cli.ts']);
describe('CLI 调度器总闸', () => {
const cliFiles = fs
.readdirSync(CLI_DIR)
.filter((f) => f.endsWith('.cli.ts'))
.filter((f) => fs.readFileSync(path.join(CLI_DIR, f), 'utf8').includes('createApplicationContext'));
it('存在会启动完整应用上下文的 CLI(否则本 spec 形同虚设)', () => {
expect(cliFiles.length).toBeGreaterThan(5);
});
it.each(cliFiles.filter((f) => !EXEMPT.has(f)))(
'%s 在建上下文之前调用 disableSchedulersForCli()',
(file) => {
const src = fs.readFileSync(path.join(CLI_DIR, file), 'utf8');
const guard = src.indexOf('disableSchedulersForCli()');
const ctx = src.indexOf('NestFactory.createApplicationContext');
expect(guard).toBeGreaterThanOrEqual(0);
// 顺序也要对:晚于建上下文就没意义了(onModuleInit 已经跑完)
expect(guard).toBeLessThan(ctx);
},
);
it('僵尸锁回收带年龄阈值 —— 不能只靠"比本进程早"这一个判据', () => {
const sched = fs.readFileSync(
path.resolve(__dirname, '../src/queues/sync-incremental.scheduler.ts'),
'utf8',
);
expect(sched).toContain('REAP_MIN_AGE_MS');
// 阈值必须显著大于一轮同步的正常耗时(生产实测 28~52 分钟)
const m = sched.match(/const REAP_MIN_AGE_MS = ([^;]+);/);
expect(m).not.toBeNull();
// eslint-disable-next-line no-eval
const ms = eval(m![1]!) as number;
expect(ms).toBeGreaterThanOrEqual(2 * 60 * 60 * 1000);
});
});
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment