From 6aedd6eb53c0d3e1ba99d9de0cc1c893b5329ad9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E5=91=98=E9=98=BF=E6=B1=9F=28Relakkes?=
 =?UTF-8?q?=29?= <relakkes@gmail.com>
Date: Wed, 8 Apr 2026 21:14:18 +0800
Subject: [PATCH] fix: prevent cross-process duplicate execution of scheduled
 tasks
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

三层防护：内存 minuteKey 去重、文件 lastFiredAt 跨进程去重、启动时清理僵尸 running 条目。
将 updateLastFired 从任务完成后移至启动时，让其他调度器进程尽早感知。
---
 src/server/services/cronScheduler.ts | 68 ++++++++++++++++++++++++++--
 1 file changed, 63 insertions(+), 5 deletions(-)

diff --git a/src/server/services/cronScheduler.ts b/src/server/services/cronScheduler.ts
index 03d8bf29..f7f39423 100644
--- a/src/server/services/cronScheduler.ts
+++ b/src/server/services/cronScheduler.ts
@@ -246,6 +246,8 @@ export class CronScheduler {
     string,
     { proc: ReturnType<typeof Bun.spawn>; startedAt: number; runId: string }
   >()
+  /** Track which minute each task last fired (prevents same-process duplicate within a minute). */
+  private lastFiredMinuteKey = new Map<string, string>()
   private cronService: CronService
   private sessionService: SessionService
 
@@ -254,10 +256,19 @@ export class CronScheduler {
     this.sessionService = new SessionService()
   }
 
+  /** Return a string key representing the calendar minute of `date`. */
+  private static minuteKey(date: Date): string {
+    return `${date.getFullYear()}-${date.getMonth() + 1}-${date.getDate()}-${date.getHours()}-${date.getMinutes()}`
+  }
+
   /** Start the scheduler (called on server boot). */
   start(): void {
     if (this.intervalId) return // already running
     console.log('[CronScheduler] Starting — checking every 60 s')
+    // Clean up stale "running" entries left by previously crashed processes
+    this.cleanupStaleRuns().catch((err) =>
+      console.error('[CronScheduler] Error cleaning up stale runs:', err),
+    )
     this.intervalId = setInterval(() => this.tick(), 60_000)
     // Immediate first check
     this.tick()
@@ -285,15 +296,28 @@ export class CronScheduler {
     try {
       const tasks = await this.cronService.listTasks()
       const now = new Date()
+      const currentKey = CronScheduler.minuteKey(now)
 
       for (const task of tasks) {
         // Skip disabled tasks
         if (task.enabled === false) continue
 
-        // Skip if already running
+        // Skip if already running (in-memory guard — same process)
         if (this.runningTasks.has(task.id)) continue
 
+        // Skip if this process already fired the task in the current minute
+        if (this.lastFiredMinuteKey.get(task.id) === currentKey) continue
+
+        // Skip if ANY process already fired the task in the current minute
+        // (cross-process guard via file-persisted lastFiredAt)
+        if (task.lastFiredAt) {
+          const lastFiredKey = CronScheduler.minuteKey(new Date(task.lastFiredAt))
+          if (lastFiredKey === currentKey) continue
+        }
+
         if (cronMatches(task.cron, now)) {
+          // Record the minute key BEFORE firing to prevent double-fire
+          this.lastFiredMinuteKey.set(task.id, currentKey)
           // Fire and forget — don't await; we want all matching tasks to start
           this.executeTask(task).catch((err) => {
             console.error(
@@ -359,6 +383,10 @@ export class CronScheduler {
       sessionId,
     }
 
+    // Update lastFiredAt IMMEDIATELY so other scheduler processes see it
+    // and skip this task in the current minute (cross-process dedup).
+    await this.cronService.updateLastFired(task.id, startedAt)
+
     // Persist the "running" state
     await appendRun(run)
 
@@ -478,9 +506,6 @@ export class CronScheduler {
 
       await updateRun(completedRun)
 
-      // Update lastFiredAt on the task
-      await this.cronService.updateLastFired(task.id, startedAt)
-
       // If non-recurring, disable after first run
       if (!task.recurring) {
         await this.cronService.updateTask(task.id, { enabled: false }).catch(() => {
@@ -504,12 +529,45 @@ export class CronScheduler {
       }
 
       await updateRun(failedRun)
-      await this.cronService.updateLastFired(task.id, startedAt)
 
       return failedRun
     }
   }
 
+  // ─── Cleanup ───────────────────────────────────────────────────────────────
+
+  /**
+   * Mark stale "running" entries as "failed" on startup.
+   * These are leftover from previous process instances that crashed or were
+   * killed before they could update the run log.
+   */
+  private async cleanupStaleRuns(): Promise<void> {
+    const data = await readRunsFile()
+    let changed = false
+    const now = Date.now()
+
+    for (const run of data.runs) {
+      if (run.status !== 'running') continue
+      const startedAt = new Date(run.startedAt).getTime()
+      // If "running" for longer than the task timeout + 1-minute buffer,
+      // the owning process is certainly dead.
+      if (now - startedAt > TASK_TIMEOUT_MS + 60_000) {
+        run.status = 'failed'
+        run.error = 'Process terminated before task could complete'
+        run.completedAt = new Date().toISOString()
+        run.durationMs = now - startedAt
+        changed = true
+        console.log(
+          `[CronScheduler] Cleaned up stale run ${run.id} for task ${run.taskId}`,
+        )
+      }
+    }
+
+    if (changed) {
+      await writeRunsFile(data)
+    }
+  }
+
   // ─── Query helpers ─────────────────────────────────────────────────────────
 
   /** Get execution history for a specific task. */