merge: feat/scheduler-kill-switch → main(迁移期定时任务写侧总闸)

apps/pac-service/src/queues/scheduler-switch.ts

+/**
+ * 定时任务写侧总闸 —— `PAC_SCHEDULER_DISABLED=1` 时本实例不跑任何**会改数据**的定时任务。
+ *
+ * 【为什么需要】服务器迁移 / 蓝绿发布 / 备用实例这类「新旧实例并存」的窗口里,新实例要能
+ * 起来接流量、验证配置,但绝不能同时对 DW 做增量摄入 + persona/plan 重算:
+ *   - 各连各库:两边各拉各的,待迁移的数据快照持续偏移,dump 出来就是旧的
+ *   - 同连一库:互抢 sync_logs 的 partial UNIQUE(host_id) WHERE status='running' 锁,
+ *     还会把对方**正在跑**的锁当"僵尸"回收掉(回收判据只看 startedAt 早于本进程启动)
+ *
+ * 【为什么必须是代码开关,不能用 env 现有机制】
+ *   - jvs-dw 的 manifest 写死 `auto_sync: true`,scheduler 启动即自动发现并注册
+ *   - cron 表达式也写在 manifest(`incremental_cron`),优先级高于全局 env
+ *   - `PAC_INCREMENTAL_HOSTS=` 留空只会 fallback 到自动发现,关不掉
+ *   - 改 manifest 能关,但 `deploy-prod.sh` 会 `git pull`,改动被覆盖
+ *
+ * 【边界:只关"写",不关"读"】本闸刻意**不覆盖** dw-lag-monitor / daily-health-report ——
+ * 那两个只发告警和报表、不碰业务数据,备用实例照常跑反而能多一双眼睛。真要静音它们,
+ * 各自的 `PAC_*_CRON` 不设即为「永不执行」(代码默认 `0 0 31 12 *`)。
+ *
+ * 【别忘了关掉】迁移完成、旧实例下线后,**务必移除该 env 并重启** —— 否则新生产静默不摄入,
+ * 表现只有"数据越来越旧",没有任何报错。DW 滞后监控能兜住(它不受本闸影响),但那是 24h 后的事。
+ */
+export function schedulerDisabled(): boolean {
+  return process.env.PAC_SCHEDULER_DISABLED === '1';
+}

apps/pac-service/src/queues/stale-scan.service.ts

@@ -2,6 +2,7 @@ import { Injectable, Logger } from '@nestjs/common';
 import { Cron } from '@nestjs/schedule';
 import { PrismaService } from '../prisma/prisma.service';
 import { QueueProducer } from './queue-producer.service';
+import { schedulerDisabled } from './scheduler-switch';
 
 /**
  * 凌晨兜底扫描 — 每日 02:00 跑。
@@ -50,6 +51,12 @@ export class StaleScanService {
     timeZone: 'Asia/Shanghai',
   })
   async scanAndEnqueueStale(): Promise<void> {
+    // 写侧总闸(见 scheduler-switch):迁移/备用实例不重算 persona —— 它会 enqueue 重算任务、
+    // 改业务数据,与旧实例并跑会让待迁移的快照持续偏移。
+    if (schedulerDisabled()) {
+      this.logger.warn('stale-scan: ⏸ PAC_SCHEDULER_DISABLED=1,跳过本轮(迁移/备用实例)');
+      return;
+    }
     const startedAt = Date.now();
     this.logger.log('stale-scan: START');
 

apps/pac-service/src/queues/sync-incremental.scheduler.ts

@@ -18,6 +18,7 @@ const PROCESS_STARTED_AT = new Date();
 import { OrgTreeService } from '../modules/auth/org-tree';
 import { PersonaService } from '../modules/persona/persona.service';
 import { PlanEngineService } from '../modules/plan/engine/plan-engine.service';
+import { schedulerDisabled } from './scheduler-switch';
 
 /**
  * SyncIncrementalSchedulerService — DW 直连增量自动跑(**每宿主独立 cron**)
@@ -50,6 +51,25 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
   ) {}
 
   async onModuleInit(): Promise<void> {
+    // ⭐ 写侧总闸:PAC_SCHEDULER_DISABLED=1 → 本实例不跑任何**会改数据**的定时任务。
+    //   用于「新旧实例并存」的窗口(服务器迁移 / 蓝绿 / 备用实例):新实例要能起来接流量、
+    //   验证配置,但**绝不能**同时对 DW 做增量摄入 + persona/plan 重算 —— 两边各拉各的,
+    //   会让待迁移的数据快照持续偏移;若两边连同一个库,还会互抢 sync_logs 的
+    //   partial UNIQUE(host_id) WHERE status='running' 锁、互相回收对方的"僵尸锁"。
+    //
+    //   ⚠️ 为什么必须是代码开关:jvs-dw 的 manifest 写死 auto_sync=true,cron 也写在
+    //   manifest(优先级高于 env),`PAC_INCREMENTAL_HOSTS=` 留空只会 fallback 到自动发现 ——
+    //   **靠 env 关不掉**。改 manifest 又会被 deploy 的 git pull 覆盖。
+    //
+    //   ⚠️ 连僵尸锁回收也一并跳过:回收判据是"startedAt 早于本进程启动",共库场景下
+    //   新实例会把老实例**正在跑**的锁误当僵尸回收掉。禁用态就该完全不碰 sync_logs。
+    if (schedulerDisabled()) {
+      this.logger.warn(
+        'sync-incremental: ⏸ PAC_SCHEDULER_DISABLED=1 —— 不注册 cron、不回收僵尸锁(迁移/备用实例)',
+      );
+      return;
+    }
+
     // 启动即回收上一个进程留下的僵尸锁(部署/崩溃重启时,running 行来不及标终态,
     // 会永久占住 sync_logs 的 partial UNIQUE(host_id) WHERE status='running',
     // 导致之后每次 cron 增量都被并发锁 skip → 游标不动 → DW 滞后告警)。

apps/pac-service/tests/scheduler-kill-switch.spec.ts

+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { schedulerDisabled } from '../src/queues/scheduler-switch';
+
+/**
+ * 定时任务写侧总闸(PAC_SCHEDULER_DISABLED)。
+ *
+ * 【为什么必须是代码开关】迁移窗口里新旧实例并存,新实例绝不能同时跑增量摄入 + 重算。
+ * 但 jvs-dw 的 manifest 写死 `auto_sync: true`、cron 也写在 manifest(优先级高于全局 env),
+ * `PAC_INCREMENTAL_HOSTS=` 留空只 fallback 到自动发现 —— **靠 env 关不掉**;
+ * 改 manifest 又会被 deploy 的 `git pull` 覆盖。
+ *
+ * 【边界】只关"写",不关"读":dw-lag-monitor / daily-health-report 只发告警和报表,
+ * 备用实例照常跑反而多一双眼睛。这条边界如果被后来人"顺手统一"掉,迁移期就会静默失去监控 ——
+ * 故用测试钉住。
+ */
+const read = (f: string) => readFileSync(join(__dirname, '../src/queues', f), 'utf-8');
+
+describe('PAC_SCHEDULER_DISABLED', () => {
+  const ORIG = process.env.PAC_SCHEDULER_DISABLED;
+  afterEach(() => {
+    if (ORIG === undefined) delete process.env.PAC_SCHEDULER_DISABLED;
+    else process.env.PAC_SCHEDULER_DISABLED = ORIG;
+  });
+
+  test('未设置 → false(默认跑,既有部署行为不变)', () => {
+    delete process.env.PAC_SCHEDULER_DISABLED;
+    expect(schedulerDisabled()).toBe(false);
+  });
+
+  test('=1 → true', () => {
+    process.env.PAC_SCHEDULER_DISABLED = '1';
+    expect(schedulerDisabled()).toBe(true);
+  });
+
+  test('⭐ 只认严格的 "1" —— "true"/"yes"/"0" 都不算,避免似是而非地半开', () => {
+    for (const v of ['true', 'yes', 'on', '0', '', ' 1 ']) {
+      process.env.PAC_SCHEDULER_DISABLED = v;
+      expect(schedulerDisabled()).toBe(false);
+    }
+  });
+});
+
+describe('闸门覆盖范围 — 只关写侧', () => {
+  test('⭐ 增量摄入(会改数据)受闸控,且连僵尸锁回收一并跳过', () => {
+    const s = read('sync-incremental.scheduler.ts');
+    expect(s).toContain('schedulerDisabled()');
+    // 闸判定必须在 reapStaleRunningLocks 之前 —— 共库场景下回收会误清老实例正在跑的锁
+    expect(s.indexOf('schedulerDisabled()')).toBeLessThan(s.indexOf('await this.reapStaleRunningLocks()'));
+  });
+
+  test('⭐ persona stale-scan(会 enqueue 重算)受闸控', () => {
+    expect(read('stale-scan.service.ts')).toContain('schedulerDisabled()');
+  });
+
+  test('⭐ 只读任务不受闸控 —— 迁移期仍要有监控和日报', () => {
+    expect(read('dw-lag-monitor.service.ts')).not.toContain('schedulerDisabled');
+    expect(read('daily-health-report.service.ts')).not.toContain('schedulerDisabled');
+  });
+});