构建分布式任务调度器

项目概述

本项目实现一个面向 scheduled job execution 的完整系统，核心要解决的问题是 running jobs at their scheduled time across a cluster so each job runs exactly once even during failover。整体由 Job queue、Time trigger、Lock acquirer、Worker pool、Miss detector 五个模块组成，每个模块单一职责、可独立测试，最后通过集成测试验证端到端正确性。设计上把 a worker that claims a job, crashes, and leaves the lock held so the job never runs until the lock expires 作为第一类需求先行处理，而非事后再补。

架构与关键决策

数据模型为 a queue of jobs with a next-run time, claimed by a worker through a distributed lock so a failover does not double-run the job。Job queue 定义其余阶段共同依赖的数据形状并在边界处校验输入；Time trigger 承载 scheduled job execution 的正确性逻辑；Lock acquirer 专门处理 a worker that claims a job, crashes, and leaves the lock held so the job never runs until the lock expires，在失败路径上给出明确的拒绝、重试或降级策略；Worker pool 提供可观测性，每次操作输出带关联 id 的结构化事件；Miss detector 在过载时按既定方式降级，避免过载演变为数据损坏。配置与代码分离，使风险变更可经 diff 审阅。

实施与测试

按五个阶段依次实现 Job queue、Time trigger、Lock acquirer、Worker pool、Miss detector，每阶段配套聚焦单测。集成测试组合各模块并驱动真实 scheduled job execution 场景：先验证正常路径，再刻意触发 a worker that claims a job, crashes, and leaves the lock held so the job never runs until the lock expires 并断言系统降级而非损坏。所有依赖均在进程内或被桩替换，用例在 CI 中确定可复现。

限制与后续

• 单区域设计，多区域复制超出范围。
• Job queue 默认进程内，未水平共享。
• 无鉴权层，假定部署在强制身份的边缘之后。

代码示例

# setup for job-scheduler-distributed
pnpm init
pnpm add redis fastify
pnpm add -D typescript tsx vitest

job_scheduler_distributed/
├── src/index.ts
├── src/job-queue.ts
├── src/time-trigger.ts
├── tests/job_scheduler_distributed.test.ts
└── package.json

// Job queue: scheduled job execution core
export function jobqueue(input: JobqueueInput): JobqueueResult {
  // Build the job queue that anchors scheduled job execution and defines the core data shape.
  return processJobqueue(input);
}

// Time trigger — handles scheduled job execution Add the time trigger so
export class Timetrigger {
  constructor(private opts: TimetriggerOptions) {}
  async run(ctx: Context): Promise<TimetriggerOutput> {
    return this.execute(ctx);
  }
}

// Lock acquirer: scheduled job execution stage
export const lockacquirer = (next: Next) => async (req: Request) => {
  const out = await transform(req);
  return next(out);
};

// Worker pool for scheduled job execution
export async function workerpool(stream: AsyncIterable<Chunk>): Promise<Aggregate> {
  let acc = init();
  for await (const c of stream) acc = merge(acc, c);
  return acc;
}

// Miss detector: scheduled job execution finalization
export function missdetector(state: State): Result {
  validate(state);
  return finalize(state);
}

# run the scheduled job execution test suite
pnpm vitest run