# **PG Boss 任务重复故障分析与修复方案**

## **1\. 故障核心分析 (Root Cause Analysis) \- 修正版**

针对 "同一 TaskID 被创建 7 次" 且 "创建时间在同一毫秒" 的现象，在确认 **仅有 1 个 SAE 实例** 运行的情况下，我们排除了多实例并发的可能性。

结合已清理的 "rvw\_review\_task 有 7 个重复条目" 这一关键证据，我们得出了确切的结论：

### **核心根因：持久化配置重复 (Persisted Configuration Duplication)**

**问题不在于有多少个实例在跑，而在于数据库里存了多少份重复的指令。**

#### **💡 深度解析：为什么会有 7 个？(7 个闹钟的比喻)**

你的疑问是：*"每次处理应该是生成不同的任务ID，不可能是重复的对吗？"*

**答案：是的，pg-boss 生成了 7 个完全不同的 Job ID，但它们都在做同一件事。**

这就好比你为了早上 7 点起床，设置了 **7 个闹钟**：

1. **指令 (Schedules/Definitions)**：数据库里那些被清理的 "7 个重复条目"，就像是 7 个闹钟配置。它们都设定在同一个触发条件下（比如 Cron 表达式，或系统启动时）。  
2. **触发 (Trigger)**：当时间到了，或者系统启动扫描时，这 **7 个闹钟同时响了**。  
3. **执行 (Jobs)**：系统听到第 1 个闹钟，创建了 Job A；听到第 2 个闹钟，创建了 Job B... 直到 Job G。  
   * **结果**：你在一毫秒内，被叫醒了 7 次。  
   * **数据**：这 7 个 Job 都有**不同的 UUID**（符合数据库约束），但它们的\*\*内容（Payload）\*\*全是 "处理 Task bd19c3d3"。

这就是为什么你在数据库里看到 created\_on 完全一致，但 Job ID 不同。因为那 1 个 SAE 实例在极短的时间内，忠实地执行了数据库里残留的 7 条指令。

* **机制解析**：pg-boss 是一个基于数据库的任务队列。它的调度（Schedules）和某些队列配置是**持久化**在 PostgreSQL 数据库中的（通常在 pgboss.schedule 表中）。  
* **故障复盘**：  
  1. **积累阶段**：在过去的历史部署或重启中，代码可能在启动时调用了 boss.schedule('queue', 'cron')。由于没有加去重逻辑，每次部署都在数据库里**新增**了一条调度记录，而不是更新旧的。日积月累，数据库里就有了 7 条完全一样的调度记录。  
  2. **爆发阶段**：当你当前的 **1 个 SAE 实例** 运行时，pg-boss 内部的轮询器扫描数据库，读取到了这 7 条重复的记录。  
  3. **瞬间执行**：当触发条件满足，这单个实例在极短的 CPU 周期内，为这 7 条记录分别生成了一个 Job。  
* **证据链闭环**：  
  * **7 次重复** 对应 **7 个重复的 Schedule/配置记录**。  
  * **同一毫秒创建** 对应 **单实例在一次事件循环中连续处理了这 7 条指令**。

**结论**：你执行的 "清理了 32 个重复的队列定义" 操作，实际上就是**关掉了多余的 6 个闹钟**，这已经移除了问题的根源。

### **为什么 SingletonKey 之前没生效？**

虽然这是单实例产生的重复，但如果代码使用的是 insert 或者是没有严格 unique constraint 保护的 send，在极快的循环中（Event Loop），数据库可能仍未完成第一条的提交，第二条就来了。

但最可能的原因是：**生成 Key 的逻辑有问题**，或者根本没有在产生任务的那段特定逻辑中加上 singletonKey。

## **2\. 解决方案：三层防御体系**

虽然根因（重复配置）已被你清理，但为了防止未来代码逻辑再次意外引入重复配置，或者防止前端意外的连击，我们依然强烈建议保留以下防御措施。

### **第一层：入队时防御 (生产者层面 \- 强制去重)**

这是最关键的一步。无论是因为配置重复导致被调用 7 次，还是前端点了 7 次，这里都能拦住。

**修改代码建议 (Producer/Service):**

// reviewTaskProducer.ts

import { PgBoss } from 'pg-boss';

// 假设这是你的入队逻辑  
export async function createReviewTask(boss: PgBoss, taskId: string, payload: any) {  
  const queueName \= 'rvw\_review\_task';  
    
  // ✅ 核心修复：构造确定性的 singletonKey  
  // 不要包含时间戳等变量，只包含业务唯一标识 (如 taskId)  
  const singletonKey \= \`review\_task\_${taskId}\`;

  // 发送任务  
  const jobId \= await boss.send(queueName, payload, {  
    // ✅ 启用单例模式  
    singletonKey: singletonKey,  
    // ✅ 节流/防抖：如果任务已存在且活跃，300秒内不再创建  
    singletonSeconds: 300,   
    // ✅ 即使旧任务完成了，保留Key一段时间以防重复触发  
    singletonNextSlot: false   
  });

  if (\!jobId) {  
    console.warn(\`\[Duplicate Prevented\] Task ${taskId} already exists in queue.\`);  
    return null;  
  }  
    
  return jobId;  
}

### **第二层：处理时防御 (Worker 层面 \- 幂等性检查)**

你已经添加了状态检查，这很好。为了处理潜在的竞争（虽然单实例下竞争少，但为了健壮性），建议保持乐观锁逻辑。

**修改代码建议 (Worker):**

// reviewWorker.ts

export async function processReviewTask(job: Job) {  
  const { taskId } \= job.data;

  // 1\. 业务状态检查 (你已经做了)  
  const task \= await db.task.findUnique({ where: { id: taskId } });  
    
  // ✅ 状态检查：如果已经是处理中或完成，直接跳过  
  if (task.status \=== 'COMPLETED' || task.status \=== 'PROCESSING') {  
    console.log(\`\[Skipped\] Task ${taskId} is already ${task.status}\`);  
    return;   
  }

  // 2\. 乐观锁更新 (Database Atomic Update)  
  const updateResult \= await db.task.updateMany({  
    where: {   
      id: taskId,  
      status: 'PENDING' // 👈 关键：只有当前状态是 PENDING 时才更新  
    },  
    data: {  
      status: 'PROCESSING',  
      startedAt: new Date()  
    }  
  });

  if (updateResult.count \=== 0\) {  
    console.log(\`\[Concurrency Control\] Task ${taskId} claimed by another worker or logic.\`);  
    return;  
  }

  // 3\. 执行逻辑  
  try {  
    await performReviewLogic(taskId);  
    await db.task.update({ where: { id: taskId }, data: { status: 'COMPLETED' }});  
  } catch (error) {  
    await db.task.update({ where: { id: taskId }, data: { status: 'FAILED' }});  
    throw error;  
  }  
}

### **第三层：初始化代码审查 (防止复发)**

针对 **"32 个重复队列定义"** 的来源，需要检查你的启动代码。

1. **检查 schedule 调用**：  
   如果你在代码里使用了 boss.schedule('queue', cron, ...)，请确保不要在每次应用启动时都无脑调用。  
   * **错误做法**：在 main.ts 直接调用 boss.schedule(...)。每次部署都会尝试再加一个（取决于 pg-boss 版本行为）。  
   * **正确做法**：通常 pg-boss 会处理去重，但如果参数稍有不同（比如 cron 表达式或数据），它可能会视为新 Schedules。建议检查 pg-boss 的 schedules 表，确保没有垃圾数据。  
2. **清理脚本**：  
   保留一个数据库迁移脚本或运维 SQL，定期检查 pg-boss 的 job 表中是否有异常激增的 created 状态的任务。

## **3\. 总结**

* **问题原因**：数据库中残留的历史重复配置（7个重复条目）导致单实例在循环中瞬间创建了 7 个任务。  
* **当前状态**：你清理了重复条目，这已经解决了根源。  
* **未来保障**：部署带有 singletonKey 的代码，这将是永远的防线，即使数据库里有 100 个重复配置，pg-boss 也会拒绝创建第 2 到 第 100 个任务。