From 20d4be331e05f5cfcb2aafd9ffcaa954f565e352 Mon Sep 17 00:00:00 2001 From: hy Date: Fri, 21 Aug 2026 15:16:05 +0800 Subject: [PATCH] feat: enhance scheduled task handling to skip missed runs during sleep and adjust related tests --- doc/prd/workbench.md | 2 + doc/test-plan/workbench-test.md | 8 +- .../scheduler/ScheduledTaskService.ts | 85 +++++++++++------ .../case/unit/scheduled-task-service.test.ts | 93 ++++++++++++++++--- ...sk-trigger-and-session-refresh.e2e.test.ts | 26 +++++- 5 files changed, 169 insertions(+), 45 deletions(-) diff --git a/doc/prd/workbench.md b/doc/prd/workbench.md index 5b42e59e..218cb5d1 100644 --- a/doc/prd/workbench.md +++ b/doc/prd/workbench.md @@ -729,8 +729,10 @@ #### ST-RUN-003 程序关闭错过不补跑 - 系统行为：程序关闭期间错过的调度时间不补跑，重新打开后只等待下一次到点。 +- 系统行为：程序运行期间因系统休眠跨过触发点（超过 60 秒宽限）同样视为错过，不补跑并顺延到下一周期；到点后 60 秒内的正常延迟仍触发。 - 验收： - `ST-RUN-003-AC1` 程序关闭期间错过的调度时间，重新打开后不会立即触发补跑。 + - `ST-RUN-003-AC2` 系统休眠跨过触发点后唤醒，不补跑并顺延到下一周期。 ## 9. 非目标 diff --git a/doc/test-plan/workbench-test.md b/doc/test-plan/workbench-test.md index 29af66b3..cd969cc4 100644 --- a/doc/test-plan/workbench-test.md +++ b/doc/test-plan/workbench-test.md @@ -610,13 +610,13 @@ tests/case/workbench/ - Why：这是定时任务的核心闭环，验证真实调度、真实 AI run 与前端会话列表刷新联动。 - How：在真实 UI 建立当前会话后，配置并启用一个每天 + 当前系统时间下一分钟的任务（提示词触发一次轻量真实 AI 请求）；等待到点，通过 `chatSessionList` 轮询新会话出现并运行完成，同时确认当前会话仍为激活项。 - Oracle：新会话创建与标题取提示词前 20 字符对齐 `ST-RUN-001-AC1`；运行完成后会话列表刷新且不切换当前会话对齐 `ST-RUN-001-AC2` / `ST-RUN-002-AC1/AC2`。 -- Not：不验证跨天 / 跨周边界调度，不验证程序关闭期间错过补跑（该行为由 backend 单测覆盖）。 +- Not：不验证跨天 / 跨周边界调度，不验证程序关闭 / 休眠期间错过补跑（该行为由 backend 单测覆盖）。 #### `scheduled-task` backend 单测 - 测试类型：`backend` -- What：`computeNextRunAt` 的每天 / 每周 / 时间格式 / 星期边界；`ScheduledTaskService.save/setEnabled` 必填校验与 `nextRunAt` 计算；`fireTask` 成功建会话与提交载荷、失败清理空会话、工作区缺失跳过；`start` 对错过档位只前移不补跑。 +- What：`computeNextRunAt` 的每天 / 每周 / 时间格式 / 星期边界；`ScheduledTaskService.save/setEnabled` 必填校验与 `nextRunAt` 计算；`fireTask` 成功建会话与提交载荷、失败清理空会话、工作区缺失跳过；`start` 对错过档位只前移不补跑；`runDue` 对休眠错过（超过 60 秒宽限）只前移不补跑；自排 `setTimeout` 到最近 `nextRunAt` 与变更后重排。 - Why：调度边界与副作用清理不适合用 UI E2E 稳定复现，下沉到单测保证确定性。 -- How：注入 fake task/workspace/chat store、submitTask、clock，驱动 `tick` / `start` / `save` / `setEnabled`。 -- Oracle：跳过不可用工作区不产生空会话对齐 `ST-RUN-001-AC3`；程序关闭错过不补跑对齐 `ST-RUN-003-AC1`。 +- How：注入 fake task/workspace/chat store、submitTask、clock 与 timer，驱动 `runDue` / `start` / `save` / `setEnabled`。 +- Oracle：跳过不可用工作区不产生空会话对齐 `ST-RUN-001-AC3`；程序关闭错过不补跑对齐 `ST-RUN-003-AC1`；休眠错过不补跑对齐 `ST-RUN-003-AC2`。 diff --git a/src/main/services/scheduler/ScheduledTaskService.ts b/src/main/services/scheduler/ScheduledTaskService.ts index 1ee95f99..f9d2a4e0 100644 --- a/src/main/services/scheduler/ScheduledTaskService.ts +++ b/src/main/services/scheduler/ScheduledTaskService.ts @@ -14,9 +14,9 @@ import type { ChatStore } from '../storage/domains/ChatStore' import type { ScheduledTaskStore } from '../storage/domains/ScheduledTaskStore' import type { WorkspaceStore } from '../storage/domains/WorkspaceStore' -const TICK_INTERVAL_MS = 30_000 +const MISFIRE_GRACE_MS = 60_000 -type RepeatHandle = ReturnType +type TimerHandle = ReturnType type ScheduledTaskServiceDeps = { taskStore: Pick @@ -24,8 +24,8 @@ type ScheduledTaskServiceDeps = { chatStore: Pick submitTask: (payload: ChatSubmitRequest) => Promise emitSessionListRefresh: () => void - repeat?: (callback: () => void, intervalMs: number) => RepeatHandle - clearRepeat?: (handle: RepeatHandle) => void + setTimer?: (callback: () => void, delayMs: number) => TimerHandle + clearTimer?: (handle: TimerHandle) => void now?: () => number } @@ -92,55 +92,81 @@ export const computeNextRunAt = ( } export class ScheduledTaskService { - private readonly deps: Required> & - Omit - private timer: RepeatHandle | null = null - private ticking = false + private readonly deps: Required> & + Omit + private timer: TimerHandle | null = null + private running = false public constructor(deps: ScheduledTaskServiceDeps) { this.deps = { - repeat: (callback, intervalMs) => setInterval(callback, intervalMs), - clearRepeat: (handle) => clearInterval(handle), + setTimer: (callback, delayMs) => setTimeout(callback, delayMs), + clearTimer: (handle) => clearTimeout(handle), now: () => Date.now(), ...deps } } public async start(): Promise { - if (this.timer) return try { await this.reconcileMissed() } catch (error) { console.error('[SCHEDULED-TASK] reconcile missed failed:', error) } - if (this.timer) return - this.timer = this.deps.repeat(() => { - void this.tick() - }, TICK_INTERVAL_MS) - await this.tick() + await this.scheduleNext() } - public async tick(): Promise { - if (this.ticking) return - this.ticking = true + public stop(): void { + if (this.timer === null) return + this.deps.clearTimer(this.timer) + this.timer = null + } + + public async runDue(): Promise { + if (this.running) return + this.running = true try { const now = this.deps.now() const tasks = await this.deps.taskStore.listEnabled() for (const task of tasks) { if (task.nextRunAt === null || now < task.nextRunAt) continue - await this.fireTask(task, now) + if (now - task.nextRunAt <= MISFIRE_GRACE_MS) { + await this.fireTask(task, now) + } else { + console.warn( + '[SCHEDULED-TASK] missed slot, skip:', + JSON.stringify({ id: task.id, nextRunAt: task.nextRunAt, now, gapMs: now - task.nextRunAt }) + ) + await this.advanceNextRun(task, now) + } } } catch (error) { - console.error('[SCHEDULED-TASK] tick failed:', error) + console.error('[SCHEDULED-TASK] run due failed:', error) } finally { - this.ticking = false + this.running = false } + await this.scheduleNext() } - public stop(): void { - if (!this.timer) return - this.deps.clearRepeat(this.timer) - this.timer = null + private async scheduleNext(): Promise { + if (this.timer !== null) { + this.deps.clearTimer(this.timer) + this.timer = null + } + const now = this.deps.now() + const tasks = await this.deps.taskStore.listEnabled() + const next = tasks + .map((task) => task.nextRunAt) + .filter((value): value is number => value !== null) + .sort((a, b) => a - b)[0] + if (next === undefined) return + const delayMs = Math.max(0, next - now) + console.warn( + '[SCHEDULED-TASK] arm timer:', + JSON.stringify({ next, now, delayMs }) + ) + this.timer = this.deps.setTimer(() => { + void this.runDue() + }, delayMs) } public async list(): Promise { @@ -151,6 +177,7 @@ export class ScheduledTaskService { try { const task = this.buildTask(input) await this.deps.taskStore.save(task) + await this.scheduleNext() return { success: true, task } } catch (error) { return { success: false, error: error instanceof Error ? error.message : String(error) } @@ -160,6 +187,7 @@ export class ScheduledTaskService { public async delete(id: string): Promise { try { await this.deps.taskStore.delete(String(id || '').trim()) + await this.scheduleNext() return { success: true } } catch (error) { return { success: false, error: error instanceof Error ? error.message : String(error) } @@ -181,6 +209,7 @@ export class ScheduledTaskService { updatedAt: now } await this.deps.taskStore.save(task) + await this.scheduleNext() return { success: true, task } } catch (error) { return { success: false, error: error instanceof Error ? error.message : String(error) } @@ -276,6 +305,10 @@ export class ScheduledTaskService { return } submitted = true + console.warn( + '[SCHEDULED-TASK] fired:', + JSON.stringify({ id: task.id, sessionId, nextRunAt: task.nextRunAt, now }) + ) this.deps.emitSessionListRefresh() } catch (error) { console.error('[SCHEDULED-TASK] fire failed:', error) diff --git a/tests/case/unit/scheduled-task-service.test.ts b/tests/case/unit/scheduled-task-service.test.ts index 0c53b6f3..0d8c833c 100644 --- a/tests/case/unit/scheduled-task-service.test.ts +++ b/tests/case/unit/scheduled-task-service.test.ts @@ -15,6 +15,7 @@ const localAt = ( ): number => new Date(year, month, day, hours, minutes).getTime() const NOW = localAt(2026, 0, 1, 8, 0) +const HOUR = 60 * 60 * 1000 const makeTask = (overrides: Partial = {}): ScheduledTask => ({ id: 'task-1', @@ -67,12 +68,14 @@ type ServiceHarness = { chatStore: { saveChatSession: ReturnType; deleteChatSession: ReturnType } submitTask: ReturnType emitSessionListRefresh: ReturnType - repeat: ReturnType - clearRepeat: ReturnType - now: () => number + setTimer: ReturnType + clearTimer: ReturnType + now: ReturnType service: ScheduledTaskService } +let nextTimerId = 0 + const createService = (): ServiceHarness => { const taskStore = new FakeTaskStore() const workspaceStore = { @@ -90,8 +93,8 @@ const createService = (): ServiceHarness => { async (): Promise => ({ success: true, mode: 'ai' }) ) const emitSessionListRefresh = vi.fn() - const repeat = vi.fn(() => 1 as ReturnType) - const clearRepeat = vi.fn() + const setTimer = vi.fn(() => ++nextTimerId) + const clearTimer = vi.fn() const now = vi.fn(() => NOW) const service = new ScheduledTaskService({ taskStore, @@ -99,8 +102,8 @@ const createService = (): ServiceHarness => { chatStore, submitTask, emitSessionListRefresh, - repeat, - clearRepeat, + setTimer, + clearTimer, now }) return { @@ -109,8 +112,8 @@ const createService = (): ServiceHarness => { chatStore, submitTask, emitSessionListRefresh, - repeat, - clearRepeat, + setTimer, + clearTimer, now, service } @@ -183,7 +186,7 @@ describe('ScheduledTaskService', () => { it('fires a due task, creates a session, submits with the configured context and refreshes the list', async () => { const harness = createService() harness.taskStore.set(makeTask({ nextRunAt: NOW - 1 })) - await harness.service.tick() + await harness.service.runDue() expect(harness.chatStore.saveChatSession).toHaveBeenCalledTimes(1) expect(harness.chatStore.saveChatSession).toHaveBeenCalledWith({ @@ -208,12 +211,21 @@ describe('ScheduledTaskService', () => { expect(harness.chatStore.deleteChatSession).not.toHaveBeenCalled() }) + it('re-arms the timer to the next run after firing a due task', async () => { + const harness = createService() + harness.taskStore.set(makeTask({ nextRunAt: NOW - 1 })) + await harness.service.runDue() + + expect(harness.setTimer).toHaveBeenCalledTimes(1) + expect(harness.setTimer).toHaveBeenCalledWith(expect.any(Function), HOUR) + }) + it('cleans up the empty session and skips refresh when submit fails', async () => { const harness = createService() harness.taskStore.set(makeTask({ nextRunAt: NOW - 1 })) harness.submitTask.mockResolvedValueOnce({ success: false, error: '模型失效' }) - await harness.service.tick() + await harness.service.runDue() expect(harness.chatStore.saveChatSession).toHaveBeenCalledTimes(1) expect(harness.chatStore.deleteChatSession).toHaveBeenCalledTimes(1) @@ -223,7 +235,7 @@ describe('ScheduledTaskService', () => { it('skips without creating a session when the workspace is missing', async () => { const harness = createService() harness.taskStore.set(makeTask({ workspaceId: 'gone', nextRunAt: NOW - 1 })) - await harness.service.tick() + await harness.service.runDue() expect(harness.chatStore.saveChatSession).not.toHaveBeenCalled() expect(harness.submitTask).not.toHaveBeenCalled() @@ -241,4 +253,61 @@ describe('ScheduledTaskService', () => { expect(saved.nextRunAt).not.toBeNull() expect(saved.nextRunAt).toBeGreaterThan(NOW) }) + + it('skips and advances a run that was missed while asleep', async () => { + const harness = createService() + harness.taskStore.set(makeTask({ nextRunAt: NOW - 2 * HOUR })) + await harness.service.runDue() + + expect(harness.submitTask).not.toHaveBeenCalled() + expect(harness.chatStore.saveChatSession).not.toHaveBeenCalled() + const stored = harness.taskStore.list() + const saved = (await stored)[0] + expect(saved.nextRunAt).toBe(localAt(2026, 0, 1, 9, 0)) + }) + + it('fires every due task when multiple tasks share the same slot', async () => { + const harness = createService() + harness.taskStore.set(makeTask({ id: 'task-1', nextRunAt: NOW - 1 })) + harness.taskStore.set( + makeTask({ id: 'task-2', name: '晚间摘要', timeOfDay: '20:00', nextRunAt: NOW - 1 }) + ) + await harness.service.runDue() + + expect(harness.submitTask).toHaveBeenCalledTimes(2) + expect(harness.chatStore.saveChatSession).toHaveBeenCalledTimes(2) + expect(harness.emitSessionListRefresh).toHaveBeenCalledTimes(2) + }) + + it('arms a timer when a task is enabled', async () => { + const harness = createService() + harness.taskStore.set(makeTask({ enabled: false, nextRunAt: null })) + await harness.service.start() + expect(harness.setTimer).not.toHaveBeenCalled() + + await harness.service.setEnabled({ id: 'task-1', enabled: true }) + expect(harness.setTimer).toHaveBeenCalledTimes(1) + expect(harness.setTimer).toHaveBeenCalledWith(expect.any(Function), HOUR) + }) + + it('does not arm a timer when there are no enabled tasks', async () => { + const harness = createService() + harness.taskStore.set(makeTask({ enabled: false, nextRunAt: null })) + await harness.service.start() + + expect(harness.setTimer).not.toHaveBeenCalled() + }) + + it('re-arms to the earliest enabled task after a delete', async () => { + const harness = createService() + harness.taskStore.set(makeTask({ id: 'task-1', nextRunAt: NOW + HOUR })) + harness.taskStore.set( + makeTask({ id: 'task-2', name: '晚间摘要', timeOfDay: '20:00', nextRunAt: NOW + 12 * HOUR }) + ) + await harness.service.start() + expect(harness.setTimer).toHaveBeenLastCalledWith(expect.any(Function), HOUR) + + await harness.service.delete('task-1') + expect(harness.setTimer).toHaveBeenLastCalledWith(expect.any(Function), 12 * HOUR) + }) }) diff --git a/tests/case/workbench/scheduled-task/scheduled-task-trigger-and-session-refresh.e2e.test.ts b/tests/case/workbench/scheduled-task/scheduled-task-trigger-and-session-refresh.e2e.test.ts index 278355b8..a9d3e0fb 100644 --- a/tests/case/workbench/scheduled-task/scheduled-task-trigger-and-session-refresh.e2e.test.ts +++ b/tests/case/workbench/scheduled-task/scheduled-task-trigger-and-session-refresh.e2e.test.ts @@ -22,7 +22,7 @@ const TASK_PROMPT = '定时任务触发验证，请只回复 ok。' const padTimePart = (value: number): string => String(value).padStart(2, '0') const nextMinuteHhmm = (): string => { - const next = new Date(Date.now() + 120_000) + const next = new Date(Date.now() + 50_000) return `${padTimePart(next.getHours())}:${padTimePart(next.getMinutes())}` } @@ -59,33 +59,50 @@ describe('workbench scheduled-task e2e', () => { runMode: 'single', execute: async (runtime) => { const { page } = runtime + const mark = (label: string): void => { + runtime.signal({ name: 'scheduled-task-trigger-step', message: label }) + } + + mark('setup-start') const setup = await setupWorkbenchWithTempWorkspace(page, { caseId: CASE_ID, workspaceName: `Workbench ${CASE_ID}`, createSession: false }) + mark('setup-done') await page.getByRole('button', { name: `在 ${setup.workspaceName} 中新建对话` }).click() + mark('new-chat-clicked') await sendWorkbenchMessage(page, FIRST_PROMPT) + mark('first-message-sent') await waitForWorkbenchAiRunCompletion(runtime, Date.now(), setup.env.modelName) + mark('first-run-completed') const sessionsAfterFirst = await listSessions(page) + mark('first-list-sessions-done') const firstSession = sessionsAfterFirst.find( (session) => session.title === FIRST_PROMPT.slice(0, 20) ) if (!firstSession) throw new Error('failed to locate the first session') await openScheduledTasksView(page) + mark('view-opened') await getWorkbenchLocator(page, scheduledTaskSelectors.addButton).click() await getWorkbenchLocator(page, scheduledTaskSelectors.formName).fill('触发验证任务') await getWorkbenchLocator(page, scheduledTaskSelectors.formPrompt).fill(TASK_PROMPT) + mark('form-name-prompt-filled') await selectScheduledTaskWorkspace(page, setup.workspaceName) + mark('workspace-selected') await selectScheduledTaskModel(page, setup.env.modelName) + mark('model-selected') const scheduledHhmm = nextMinuteHhmm() await fillScheduledTaskTime(page, scheduledHhmm) + mark('time-filled') await getWorkbenchLocator(page, scheduledTaskSelectors.formSave).click() + mark('form-saved') const createdTasks = await listTasks(page) + mark('list-tasks-done') const createdTask = createdTasks.find((task) => task.name === '触发验证任务') if (!createdTask) throw new Error('failed to create the scheduled task') const taskCreatedAt = Date.now() @@ -93,9 +110,10 @@ describe('workbench scheduled-task e2e', () => { page, `${scheduledTaskSelectors.itemTogglePrefix}-${createdTask.id}` ).click() + mark('toggle-clicked') const triggeredStartedAt = Date.now() - const deadline = triggeredStartedAt + 240_000 + const deadline = triggeredStartedAt + 150_000 let triggeredSession: ChatSessionListItem | null = null while (Date.now() < deadline) { const sessions = await listSessions(page) @@ -109,12 +127,13 @@ describe('workbench scheduled-task e2e', () => { if (triggeredSession) break await page.waitForTimeout(5_000) } + mark('trigger-session-poll-done') const activeSessionId = await readActiveSessionId(page) let triggeredRunCompleted = false if (triggeredSession) { - const runDeadline = Date.now() + 180_000 + const runDeadline = Date.now() + 120_000 while (Date.now() < runDeadline) { const sessions = await listSessions(page) const current = sessions.find((session) => session.id === triggeredSession?.id) @@ -125,6 +144,7 @@ describe('workbench scheduled-task e2e', () => { await page.waitForTimeout(3_000) } } + mark('trigger-run-completion-poll-done') return { scheduledHhmm,