merge: main → test(反向合)

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

@@ -4,6 +4,7 @@ import { CronJob } from 'cron';
 import * as path from 'path';
 import { SyncStatus } from '@pac/types';
 import { PrismaService } from '../prisma/prisma.service';
+import { AlertService } from '../common/alerting/alert.service';
 import {
   ColdImportService,
   SyncAlreadyRunningError,
@@ -91,8 +92,27 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
     private readonly persona: PersonaService,
     private readonly planEngine: PlanEngineService,
     private readonly scheduler: SchedulerRegistry,
+    private readonly alerter: AlertService,
   ) {}
 
+  /**
+   * 告警的统一出口。⛔ 必须吞异常:推送失败**绝不能**反过来打断同步 ——
+   * 告警是观测,不是业务。同理 alerter 未注入(老单测按位置构造)时也只记日志。
+   *
+   * 🔴 为什么这个文件到 2026-09-05 才有告警:此前整条增量链路**一次都不推送**,
+   *   `runHostSafe` 只 `logger.error` 就吞掉。轮次失败的唯一出口是每天 09:07 的
+   *   每日健康报告 —— 最坏晚 24 小时。09-04 生产宕 11.8h 全程零告警有这一份。
+   *   加了 compose 的 mem_limit(8g)之后这个洞更要命:容器被 SIGKILL → 轮次失败 →
+   *   HTTP 30 秒内恢复 → 外部探针一路绿 → **系统在稳定地失败,而所有指示灯是绿的**。
+   */
+  private async alertSafe(level: 'warning' | 'critical', title: string, body: string): Promise<void> {
+    try {
+      await this.alerter.send({ level, title, body });
+    } catch (err) {
+      this.logger.error(`sync-incremental: 告警推送失败(不影响同步): ${(err as Error).message}`);
+    }
+  }
+
   async onModuleInit(): Promise<void> {
     // ⭐ 写侧总闸:PAC_SCHEDULER_DISABLED=1 → 本实例不跑任何**会改数据**的定时任务。
     //   用于「新旧实例并存」的窗口(服务器迁移 / 蓝绿 / 备用实例):新实例要能起来接流量、
@@ -242,6 +262,27 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
       this.logger.log(
         `sync-incremental: 回收 ${count} 个僵尸同步锁(${processStartedAt ? '启动态' : '周期态'})`,
       );
+      // ⭐ 回收本身就是「上一轮没正常终结」的**唯一确定性信号**,必须推出去。
+      //   它比"轮次失败"更早、更准:失败可能只是一次网络抖动,而留下僵尸锁意味着
+      //   进程是被**外力打断**的(SIGKILL / 部署 / 卡死后重启),没走到 finally。
+      await this.alertSafe(
+        'warning',
+        `增量回收了 ${count} 个僵尸同步锁`,
+        [
+          `上一轮**没有正常终结**(${processStartedAt ? '启动态' : '周期态'}回收)。`,
+          ...stale.map(
+            (x) => `· host=${x.hostId} 起跑=${x.startedAt.toISOString()} trigger=${x.triggeredBy}`,
+          ),
+          '',
+          '按概率排查:',
+          '① 容器撞 mem_limit(8g)被 SIGKILL 后重启 —— 查 RestartCount 与 `dmesg -T | grep CONSTRAINT_MEMCG`',
+          '  ⚠️ 别看 docker inspect 的 ExitCode/OOMKilled,restart:always 下那是重启后那条命',
+          '② 部署/人工重启打断了在飞的轮次(预期内,不用处理)',
+          '③ 进程卡死后被外力重启',
+          '',
+          '数据没丢:游标未推进,下一轮靠 48h 回看窗补齐。',
+        ].join('\n'),
+      );
       return count;
     } catch (err) {
       // 回收失败不阻断启动(cron 仍注册);最坏退回人工清理,不比现状糟
@@ -258,6 +299,20 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
         `sync-incremental: host=${host} **跳过本轮** —— 上一轮仍在运行(防套圈)。` +
           `连续出现说明单轮已撑不下 cron 间隔,需要查 plan 段耗时。`,
       );
+      // 套圈是 2026-08-29 雪崩的前兆信号,也是 09-04 那次的中间态 —— 当时整整 11.8 小时
+      // 每轮都在这里 return,而没有任何人知道。防套圈闸救了系统,却也**吞掉了症状**。
+      await this.alertSafe(
+        'warning',
+        `增量跳过本轮(上一轮还在跑):${host}`,
+        [
+          '上一轮已超过一个 cron 间隔仍未结束,本轮跳过(防套圈)。',
+          '',
+          '⚠️ **连续出现 = 系统在恶化**,不是稳态:',
+          '· 查 plan 段耗时 —— `docker logs pac-pac-service-1 | grep "\\[plan\\] 阶段耗时"`',
+          '· 查内存水位 —— 同一批日志里的 `heapUsed=`',
+          '· 若上一轮其实是死的,下一轮开跑前的僵尸锁回收会另行告警',
+        ].join('\n'),
+      );
       return;
     }
     this.runningHosts.add(host);
@@ -279,7 +334,19 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
       if (err instanceof SyncAlreadyRunningError) {
         this.logger.warn(`sync-incremental: host=${host} skip(并发锁拦截):${err.message}`);
       } else {
-        this.logger.error(`sync-incremental: host=${host} failed: ${(err as Error).message}`);
+        const msg = (err as Error).message;
+        this.logger.error(`sync-incremental: host=${host} failed: ${msg}`);
+        await this.alertSafe(
+          'critical',
+          `增量轮次失败:${host}`,
+          [
+            msg,
+            '',
+            '游标未推进,下一轮会按同一水位 catchup(48h 回看窗)——',
+            '**单次失败通常不用人工介入,连续失败才要查**。',
+            '连续多轮同一错误 → 看 `docker logs --tail=300 pac-pac-service-1`。',
+          ].join('\n'),
+        );
       }
     } finally {
       // ⚠️ 必须在 finally —— 抛异常时不释放会把该 host 永久锁死到进程重启

apps/pac-service/tests/sync-lock-reap.spec.ts

@@ -36,6 +36,12 @@ function makeService(rows: Array<{ id: string; hostId: string; startedAt: Date;
   const findMany = jest.fn().mockImplementation(async ({ where }) => match(where));
   const updateMany = jest.fn().mockImplementation(async ({ where }) => ({ count: match(where).length }));
   const prisma = { syncLog: { findMany, updateMany } };
+  // 假告警器:记下每一条推送。⚠️ 必须真的注入 —— 不注入时 alertSafe 会吞掉 TypeError,
+  //   于是"告警没发"和"告警发了"在断言上无法区分,用例就成了摆设。
+  const sent: Array<{ level: string; title: string; body: string }> = [];
+  const send = jest.fn().mockImplementation(async (a) => {
+    sent.push(a);
+  });
   const svc = new SyncIncrementalSchedulerService(
     prisma as never,
     {} as never,
@@ -43,8 +49,9 @@ function makeService(rows: Array<{ id: string; hostId: string; startedAt: Date;
     {} as never,
     {} as never,
     {} as never,
+    { send } as never,
   );
-  return { svc, findMany, updateMany };
+  return { svc, findMany, updateMany, sent, send };
 }
 
 // 触达 private 方法(纯编排,无需走 onModuleInit 以免顺带注册 cron);时间界显式注入
@@ -114,6 +121,7 @@ describe('reapStaleRunningLocks', () => {
     };
     const svc = new SyncIncrementalSchedulerService(
       prisma as never, {} as never, {} as never, {} as never, {} as never, {} as never,
+      { send: jest.fn() } as never,
     );
     // 契约是「不抛」;返回值 2026-09-04 从 void 改成「回收了几条」,失败路径返回 0。
     await expect(reap(svc)).resolves.toBe(0);
@@ -212,3 +220,91 @@ describe('reapStaleRunningLocks — deploy 孤儿锁(年龄闸的盲区)', () =>
     expect(updateMany).not.toHaveBeenCalled();
   });
 });
+
+
+/**
+ * 2026-09-05 新增:增量链路的告警出口。
+ *
+ * 🔴 为什么补:此前 sync-incremental.scheduler **一条告警都不推**(不注入 AlertService),
+ *   轮次失败只 logger.error 就吞了,唯一出口是次日 09:07 的每日健康报告 —— 最坏晚 24 小时。
+ *   而 compose 刚加的 mem_limit(8g)会让"容器被 SIGKILL → 轮次失败 → HTTP 30 秒内恢复"
+ *   成为常见路径:外部探针一路绿,系统却在**稳定地失败**。三条信号必须自己会喊。
+ */
+describe('增量链路告警', () => {
+  const FAR = new Date('2099-01-01T00:00:00.000Z');
+  const farBefore = (ms: number) => new Date(FAR.getTime() - ms);
+
+  beforeEach(() => {
+    jest.useFakeTimers().setSystemTime(FAR);
+  });
+  afterEach(() => {
+    jest.useRealTimers();
+  });
+
+  it('回收到僵尸锁 → 推 warning,正文带 host / 起跑时刻 / trigger', async () => {
+    const { svc, sent } = makeService([
+      { id: 'a', hostId: 'h1', startedAt: farBefore(4 * 3600_000), triggeredBy: 'sync:jvs-dw:r1' },
+    ]);
+    const n = await reapPeriodic(svc, 'sync:jvs-dw:');
+    expect(n).toBe(1);
+    expect(sent).toHaveLength(1);
+    expect(sent[0].level).toBe('warning');
+    expect(sent[0].title).toContain('1 个僵尸同步锁');
+    expect(sent[0].body).toContain('h1');
+    expect(sent[0].body).toContain('sync:jvs-dw:r1');
+    // 排查指引必须指向 mem_limit 那条路径,否则运维会照 ExitCode 得出反结论
+    expect(sent[0].body).toContain('CONSTRAINT_MEMCG');
+  });
+
+  it('没回收到任何锁 → 不推(常态每轮都会跑,推了就是噪音)', async () => {
+    const { svc, sent } = makeService([
+      // 太年轻,不该被回收
+      { id: 'a', hostId: 'h1', startedAt: farBefore(60_000), triggeredBy: 'sync:jvs-dw:r1' },
+    ]);
+    expect(await reapPeriodic(svc, 'sync:jvs-dw:')).toBe(0);
+    expect(sent).toHaveLength(0);
+  });
+
+  it('⛔ 告警推送炸了不能反过来打断回收 —— 回收值照常返回', async () => {
+    const { svc } = makeService([
+      { id: 'a', hostId: 'h1', startedAt: farBefore(4 * 3600_000), triggeredBy: 'sync:jvs-dw:r1' },
+    ]);
+    (svc as unknown as { alerter: { send: unknown } }).alerter = {
+      send: jest.fn().mockRejectedValue(new Error('webhook 挂了')),
+    };
+    await expect(reapPeriodic(svc, 'sync:jvs-dw:')).resolves.toBe(1);
+  });
+
+  it('轮次抛异常 → 推 critical,且不向外抛(cron 回调必须吞)', async () => {
+    const { svc, sent } = makeService([]);
+    const s = svc as unknown as {
+      runOne: unknown;
+      dataDir: unknown;
+      runHostSafe: (h: string) => Promise<void>;
+    };
+    s.dataDir = () => '/tmp/data';
+    s.runOne = jest.fn().mockRejectedValue(new Error('DW 连不上'));
+    await expect(s.runHostSafe('jvs-dw')).resolves.toBeUndefined();
+    const crit = sent.filter((x) => x.level === 'critical');
+    expect(crit).toHaveLength(1);
+    expect(crit[0].title).toContain('jvs-dw');
+    expect(crit[0].body).toContain('DW 连不上');
+  });
+
+  it('套圈跳过 → 推 warning,且**不**执行本轮(防套圈语义不能被告警改掉)', async () => {
+    const { svc, sent } = makeService([]);
+    const s = svc as unknown as {
+      runOne: jest.Mock;
+      dataDir: unknown;
+      runningHosts: Set<string>;
+      runHostSafe: (h: string) => Promise<void>;
+    };
+    s.dataDir = () => '/tmp/data';
+    s.runOne = jest.fn();
+    s.runningHosts.add('jvs-dw');
+    await s.runHostSafe('jvs-dw');
+    expect(s.runOne).not.toHaveBeenCalled();
+    expect(sent.map((x) => x.level)).toEqual(['warning']);
+    expect(sent[0].title).toContain('跳过本轮');
+  });
+});

deploy/health-watch.sh

@@ -30,6 +30,14 @@
 #                         防「探针机自己断网 → 误报生产挂了」。默认空=不启用
 #                         (连续 3 次 × 5 分钟 = 15 分钟已能滤掉绝大多数网络抖动)
 #   WATCH_STATE_DIR       状态文件目录(默认 /var/lib/pac-health-watch)
+#   WATCH_HEARTBEAT_HOUR  每日心跳的小时(0-23,默认 9;设 -1 关闭)。见下方「为什么要心跳」
+#
+# 【为什么要心跳(dead-man's switch)】探针本身也会死:测试机宕、cron 被删、脚本被改坏、
+#   /etc/pac-health-watch.env 丢了 —— 这些情况下探针**静静地停掉**,而所有人以为它还看着。
+#   这正是 09-04 的同款失败:告警器坏了没人告诉你。
+#   所以每天固定时刻推一条「我还活着 + 过去 24h 探了多少次/失败几次」。
+#   ⇒ **没收到心跳本身就是告警**。它刻意排在生产每日健康报告(09:07)旁边:
+#     两条都到 = 两侧都活;只到一条 = 立刻知道是哪侧断了。
 #
 # crontab 示例(测试机 root,每 5 分钟):
 #   */5 * * * * ALERT_WEBHOOK_URL='https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx' \
@@ -49,6 +57,7 @@ WATCH_REPEAT_EVERY="${WATCH_REPEAT_EVERY:-12}"
 WATCH_TIMEOUT="${WATCH_TIMEOUT:-10}"
 WATCH_CONTROL_URL="${WATCH_CONTROL_URL:-}"
 WATCH_STATE_DIR="${WATCH_STATE_DIR:-/var/lib/pac-health-watch}"
+WATCH_HEARTBEAT_HOUR="${WATCH_HEARTBEAT_HOUR:-9}"
 WATCH_DRY_RUN="${WATCH_DRY_RUN:-0}"
 
 case "${1:-}" in
@@ -86,15 +95,19 @@ probe() {
 }
 
 # ── 状态读写 ───────────────────────────────────────────────────────────
-fails=0; alerted=0; since=""
+fails=0; alerted=0; since=""; cnt_ok=0; cnt_fail=0; hb_date=""
 # shellcheck disable=SC1090
 [[ -f "$STATE" ]] && . "$STATE"
-[[ "$fails" =~ ^[0-9]+$ ]] || fails=0
-[[ "$alerted" =~ ^[01]$ ]] || alerted=0
+# ⛔ 每个字段都要校验:状态文件是被 source 的,写坏一次(磁盘满/并发)会让后面的算术
+#   直接语法错退出 —— 探针从此静默死亡,而这正是本脚本要防的那类失败。
+[[ "$fails"    =~ ^[0-9]+$ ]] || fails=0
+[[ "$alerted"  =~ ^[01]$   ]] || alerted=0
+[[ "$cnt_ok"   =~ ^[0-9]+$ ]] || cnt_ok=0
+[[ "$cnt_fail" =~ ^[0-9]+$ ]] || cnt_fail=0
 
 save_state() {
-  printf "fails=%s\nalerted=%s\nsince='%s'\nlast='%s'\n" \
-    "$fails" "$alerted" "$since" "$(ts)" > "$STATE"
+  printf "fails=%s\nalerted=%s\nsince='%s'\ncnt_ok=%s\ncnt_fail=%s\nhb_date='%s'\nlast='%s'\n" \
+    "$fails" "$alerted" "$since" "$cnt_ok" "$cnt_fail" "$hb_date" "$(ts)" > "$STATE"
 }
 
 # ── 推送 ───────────────────────────────────────────────────────────────
@@ -137,6 +150,21 @@ if diag=$(probe "$WATCH_URL"); then
   fi
   [[ "$fails" -gt 0 ]] && say "恢复正常(此前连续失败 $fails 次)"
   fails=0; alerted=0; since=""
+  cnt_ok=$((cnt_ok + 1))
+
+  # 每日心跳。⚠️ 判据是「今天还没推过 且 已到点」而不是「正好等于某分钟」——
+  #   cron 可能因为负载/机器重启错过某一格,按"分钟相等"判会整天不推,
+  #   而心跳漏推 = 误报"探针死了"。所以用日期去重 + 到点即可。
+  today="$(date +%F)"
+  if [[ "$WATCH_HEARTBEAT_HOUR" != "-1" && "$hb_date" != "$today" ]] \
+     && [[ "$(date +%-H)" -ge "$WATCH_HEARTBEAT_HOUR" ]]; then
+    hb_date="$today"
+    push "🟢" "info" "$WATCH_NAME 探针存活(每日心跳)" \
+      "$(printf '目标正常。\n上一个周期:探测 %s 次成功 / %s 次失败。\n\n⚠️ **收不到这条 = 探针本身死了**(测试机宕/cron 被删/配置丢失),\n这时生产是死是活没有任何人在看,请手工确认。' \
+        "$cnt_ok" "$cnt_fail")"
+    cnt_ok=0; cnt_fail=0
+  fi
+
   save_state
   say "OK $diag"
   exit 0
@@ -152,6 +180,7 @@ if [[ -n "$WATCH_CONTROL_URL" ]]; then
 fi
 
 fails=$((fails + 1))
+cnt_fail=$((cnt_fail + 1))
 [[ -z "$since" ]] && since="$(ts)"
 say "失败 #$fails  $diag"