From a0b0a642468c2ae3a7b0e3d113b8fee2a0cc110c Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 13:44:48 +0800 Subject: [PATCH] refactor(schedules): register pi.token_refresh and auth.cleanup as scheduled jobs Pi token refresh (10 min) and auth session cleanup (1 h) no longer run their own setInterval; both register in the maintain group of the schedules registry with the same default intervals. Co-Authored-By: Claude Opus 5.5 --- .../server/src/modules/auth-settings/index.ts | 27 ++++++---- apps/server/src/modules/pi/index.ts | 33 ++++++------ apps/server/src/modules/schedules/index.ts | 3 -- .../server/test/schedules/module-jobs.test.ts | 51 +++++++++++++++++++ docs/runbook/schedules.md | 5 +- 5 files changed, 90 insertions(+), 29 deletions(-) create mode 100644 apps/server/test/schedules/module-jobs.test.ts diff --git a/apps/server/src/modules/auth-settings/index.ts b/apps/server/src/modules/auth-settings/index.ts index 235bff0b..6240c0a7 100644 --- a/apps/server/src/modules/auth-settings/index.ts +++ b/apps/server/src/modules/auth-settings/index.ts @@ -1,4 +1,5 @@ import { sql } from "@umt/db"; +import { registerScheduledJob } from "../schedules/registry.ts"; import type { ServerContext, ServerModule } from "../types.ts"; import { registerAuthRoutes } from "./auth-routes.ts"; import { ensureSetupToken } from "./auth-service.ts"; @@ -18,7 +19,7 @@ export interface AuthSettingsModuleOptions { engineSchema?: string; /** api 角色启动时若未初始化管理员则生成并打印 setup token(默认 true;测试关掉自己调 ensureSetupToken) */ setupTokenOnStart?: boolean; - /** 过期会话/登录尝试清理间隔,默认 1 小时 */ + /** 定时任务 auth.cleanup(过期会话/登录尝试清理)的默认周期,默认 1 小时;<=0 不登记 */ cleanupIntervalMs?: number; } @@ -30,7 +31,6 @@ export async function cleanupAuthTables(ctx: Pick): Promise } export function createAuthSettingsModule(opts: AuthSettingsModuleOptions = {}): ServerModule { - let timer: ReturnType | null = null; return { name: "auth-settings", registerRoutes(app, ctx) { @@ -44,14 +44,23 @@ export function createAuthSettingsModule(opts: AuthSettingsModuleOptions = {}): } }, registerTasks() {}, + /** 清理周期登记为定时任务 auth.cleanup(维护组,由 schedules 调度器驱动) */ start(ctx) { - const run = () => cleanupAuthTables(ctx).catch((err) => ctx.logger.warn({ err }, "认证表清理失败")); - timer = setInterval(run, opts.cleanupIntervalMs ?? 3600_000); - void run(); - }, - stop() { - if (timer) clearInterval(timer); - timer = null; + const everyMs = opts.cleanupIntervalMs ?? 3600_000; + if (everyMs <= 0) return; + registerScheduledJob(ctx, { + key: "auth.cleanup", + defaults: { + name: "登录会话清理", + description: "删除 7 天前的登录尝试和 30 天前过期或吊销的会话", + group: "maintain", + schedule: { type: "interval", everyMs }, + }, + run: async () => { + await cleanupAuthTables(ctx); + return { summary: "已清理" }; + }, + }); }, }; } diff --git a/apps/server/src/modules/pi/index.ts b/apps/server/src/modules/pi/index.ts index db1f55db..b5672bee 100644 --- a/apps/server/src/modules/pi/index.ts +++ b/apps/server/src/modules/pi/index.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import type { FakeOAuthControl } from "@umt/agent"; import { defineTask } from "@umt/tasks"; import { z } from "zod"; +import { registerScheduledJob } from "../schedules/registry.ts"; import type { ServerContext, ServerModule } from "../types.ts"; import { type OAuthOptions, OAuthSessions } from "./oauth.ts"; import { refreshPiTokens } from "./refresh.ts"; @@ -11,14 +12,14 @@ export { type RunUpstream, resolveRunUpstream } from "./service.ts"; /** * W7-A Pi 模型上游:pi_upstream CRUD、能力表、测试、模型表、服务端订阅登录(挂起式 interaction)、 - * 定时任务 refreshPiTokens(OAuth 过期前刷新,行锁内写回)。runAgent 经 ./service.ts 的 resolveRunUpstream 取上游。 + * 持久任务 refreshPiTokens(周期登记为定时任务 pi.token_refresh;OAuth 过期前刷新,行锁内写回)。runAgent 经 ./service.ts 的 resolveRunUpstream 取上游。 * 用法见 docs/runbook/agent.md「上游与登录」。 */ export interface PiModuleOptions { /** provider = fake 的模拟 OAuth(测试注入) */ fakeOAuth?: FakeOAuthControl; oauth?: Omit; - /** 刷新任务入队间隔(默认 10 分钟;<=0 关闭) */ + /** 定时任务 pi.token_refresh 的默认周期(默认 10 分钟;<=0 不登记) */ refreshIntervalMs?: number; /** 剩余有效期少于它就刷新(默认 30 分钟) */ refreshWindowMs?: number; @@ -37,7 +38,6 @@ export function createPiModule( } return s; }; - let timer: ReturnType | null = null; const live: OAuthSessions[] = []; const interval = opts.refreshIntervalMs ?? 10 * 60_000; const windowMs = opts.refreshWindowMs ?? 30 * 60_000; @@ -67,25 +67,28 @@ export function createPiModule( }), ); }, + /** 刷新周期登记为定时任务 pi.token_refresh(维护组,由 schedules 调度器驱动),执行体只入队持久任务 */ start(ctx) { if (interval <= 0) return; - const enqueue = async () => { - try { - await ctx.engine.enqueue( + registerScheduledJob(ctx, { + key: "pi.token_refresh", + defaults: { + name: "上游登录刷新", + description: "订阅登录即将过期的 Agent 上游,提前刷新 token", + group: "maintain", + schedule: { type: "interval", everyMs: interval }, + }, + run: async (c) => { + const r = await ctx.engine.enqueue( "refreshPiTokens", {}, - { idempotencyKey: `refreshPiTokens:${Math.floor(Date.now() / interval)}`, traceId: randomUUID() }, + { idempotencyKey: `refreshPiTokens:${c.slot}`, traceId: randomUUID() }, ); - } catch (err) { - ctx.logger.warn({ err: { message: (err as Error).message } }, "refreshPiTokens 入队失败"); - } - }; - void enqueue(); - timer = setInterval(() => void enqueue(), interval); + return { summary: r.deduplicated ? "本周期已入队" : "已入队刷新" }; + }, + }); }, async stop() { - if (timer) clearInterval(timer); - timer = null; await Promise.all(live.splice(0).map((s) => s.shutdown())); }, }; diff --git a/apps/server/src/modules/schedules/index.ts b/apps/server/src/modules/schedules/index.ts index 200cdb5a..e3abf78f 100644 --- a/apps/server/src/modules/schedules/index.ts +++ b/apps/server/src/modules/schedules/index.ts @@ -12,9 +12,6 @@ export { registerScheduledJob, scheduleRegistry } from "./registry.ts"; * 各模块 start() 用 registerScheduledJob 登记周期任务;本模块在 io worker 上起唯一调度器(PG 咨询锁)按表驱动, * 并登记内置 Agent 巡检模板(默认关闭)与备份占位。看板 API 见 routes.ts(全部只接受控制台会话)。 * 须放在 SERVER_MODULES 末尾:start 时其它模块已登记完(之后再登记的也会在下一轮同步)。 - * - * TODO(W7-A):pi 模块的 refreshPiTokens 目前在 pi/index.ts 自建 setInterval, - * 待其改为 registerScheduledJob({ key: "pi.token_refresh", defaults: { group: "maintain", ... } })。 */ export interface SchedulesModuleOptions { scheduler?: SchedulerOptions; diff --git a/apps/server/test/schedules/module-jobs.test.ts b/apps/server/test/schedules/module-jobs.test.ts new file mode 100644 index 00000000..5166167d --- /dev/null +++ b/apps/server/test/schedules/module-jobs.test.ts @@ -0,0 +1,51 @@ +import { describe, expect, it } from "vitest"; +import { createAuthSettingsModule } from "../../src/modules/auth-settings/index.ts"; +import { createPiModule } from "../../src/modules/pi/index.ts"; +import { scheduleRegistry } from "../../src/modules/schedules/registry.ts"; +import type { ServerContext } from "../../src/modules/types.ts"; + +// W8-A:pi token 刷新与认证表清理不再自建计时器,改为登记到定时任务注册表(维护组、默认开启、周期不变)。 + +describe("模块周期任务登记", () => { + it("pi.token_refresh:10 分钟,执行体按时间槽幂等入队 refreshPiTokens", async () => { + const enqueued: { name: string; key: string }[] = []; + const ctx = { + engine: { + enqueue: (name: string, _i: unknown, o: { idempotencyKey: string }) => { + enqueued.push({ name, key: o.idempotencyKey }); + return Promise.resolve({ deduplicated: false }); + }, + }, + } as unknown as ServerContext; + const mod = createPiModule(); + await mod.start?.(ctx); + const def = scheduleRegistry(ctx).get("pi.token_refresh"); + expect(def?.defaults).toMatchObject({ group: "maintain", schedule: { type: "interval", everyMs: 600_000 } }); + expect(def?.defaults.enabled).not.toBe(false); + await def?.run({ + jobId: "pi.token_refresh", + runId: "r", + trigger: "schedule", + now: new Date(), + config: {}, + slot: "42", + }); + expect(enqueued).toEqual([{ name: "refreshPiTokens", key: "refreshPiTokens:42" }]); + await mod.stop?.(); + }); + + it("refreshIntervalMs<=0 不登记", async () => { + const ctx = {} as ServerContext; + await createPiModule({ refreshIntervalMs: 0 }).start?.(ctx); + expect(scheduleRegistry(ctx).get("pi.token_refresh")).toBeUndefined(); + }); + + it("auth.cleanup:1 小时,维护组", async () => { + const ctx = {} as ServerContext; + await createAuthSettingsModule({ setupTokenOnStart: false }).start?.(ctx); + expect(scheduleRegistry(ctx).get("auth.cleanup")?.defaults).toMatchObject({ + group: "maintain", + schedule: { type: "interval", everyMs: 3_600_000 }, + }); + }); +}); diff --git a/docs/runbook/schedules.md b/docs/runbook/schedules.md index 878c5b18..1fd8af0e 100644 --- a/docs/runbook/schedules.md +++ b/docs/runbook/schedules.md @@ -28,6 +28,8 @@ | `library.reconcile` | 资产对账 | 整理 | 6 小时 | 入队 `reconcileLibrary`，幂等键是 `auto:<时间槽>` | | `cache.threshold` | 缓存阈值检查 | 维护 | 同 qB 同步 | 超过硬阈值时入队 `cleanupDownloadCache` | | `trash.purge` | 回收站清理 | 维护 | 24 小时 | 入队 `purgeTrash` | +| `pi.token_refresh` | 上游登录刷新 | 维护 | 10 分钟 | 入队 `refreshPiTokens`,幂等键 `refreshPiTokens:<时间槽>`;对 30 分钟内过期的订阅登录上游刷新 token | +| `auth.cleanup` | 登录会话清理 | 维护 | 1 小时 | 直接执行 `cleanupAuthTables`:删 7 天前的登录尝试、30 天前过期或吊销的会话 | | `maintain.backup` | 备份 | 维护 | 每天 03:30(占位) | 未配置。备份目前由宿主机上的 `scripts/backup.sh` 执行 | | `check.pending` | 待处理巡检 | Agent | 6 小时，默认关闭 | 只读。有未关闭事项时才会运行 | | `check.downloads` | 下载诊断 | Agent | 2 小时，默认关闭 | 默认只读。打开 `allowRecover` 后，在授权范围内重试或重新校验 | @@ -37,8 +39,7 @@ - 自动化总开关关闭，或处于只读对账模式时，编排类任务会记为 `skipped`，不执行。 - 各项默认周期和改造前的计时器一致。测试时可以通过模块参数调短，比如 `syncIntervalMs`、`schedulerIntervalMs`。 -- auth-settings 的会话清理(每小时)目前还在模块里自建计时器，尚未登记。 -- Pi 的 token 刷新(`refreshPiTokens`)目前也还是自建计时器，待 W7-A 登记为 `pi.token_refresh`。 +- 所有周期任务都已登记，模块里不再有自建计时器(W8-A 把 `pi.token_refresh`、`auth.cleanup` 也迁了进来)。 ## 调度语义