企业官网建设流程全解析

Wasp 后台任务实践：用 PgBoss 执行器实现延迟与周期性 Job

【免费下载链接】waspThe batteries-included full-stack framework for the AI era. Develop JS/TS web apps (React, Node.js, and Prisma) using declarative code that abstracts away complex full-stack features like auth, background jobs, RPC, email sending, end-to-end type safety, single-command deployment, and more.项目地址: https://gitcode.com/GitHub_Trending/wa/wasp

在多数 Web 应用中，用户向服务器发送请求、服务器快速返回数据，应用体验才是顺滑的。但一旦服务端需要额外时间处理（发邮件、调用慢速外部 API），最佳做法是尽快响应用户，把剩余工作放到后台执行。Wasp 的后台任务（Jobs）正是为此设计，核心能力包括：

• 任务在服务器重启后依然持久化（persist between restarts）；
• 任务失败后可以自动重试（retry）；
• 任务可以延迟到未来某个时间点执行（delay）；
• 任务可以按 cron 周期性地重复执行（recurring schedule）。

本文以 Wasp v0.13 的 Jobs 文档为主体，完整讲清 Job 的声明、worker 函数实现、提交调用、周期调度与全部 API 字段，并结合当前仓库中的 SDK 源码模板（waspc/data/Generator/templates/sdk/wasp/server/jobs）和examples/kitchen-sink示例项目，说明这些声明在运行时到底发生了什么。

Job 声明与 worker 函数

第一步：在应用声明中定义 Job

先创建一个示例 Job：它会向控制台打印一条消息，并从数据库读取任务列表返回。在main.wasp中声明：

job mySpecialJob { executor: PgBoss, perform: { fn: import { foo } from "@src/workers/bar" }, entities: [Task], }

其中executor: PgBoss指定由 pg-boss 负责调度的持久化、监控与执行；perform.fn指向执行实际工作的 NodeJS 函数；entities: [Task]表示 worker 函数里允许使用的实体（用法与 Queries/Actions 中声明 entities 相同，参见 Operations 概述 与 Queries 文档）。

第二步：实现 worker 函数

JavaScript 版本（src/workers/bar.js）：

export const foo = async ({ name }, context) => { console.log(`Hello ${name}!`) const tasks = await context.entities.Task.findMany({}) return { tasks } }

TypeScript 版本（src/workers/bar.ts）：

import { type MySpecialJob } from 'wasp/server/jobs' import { type Task } from 'wasp/entities' type Input = { name: string; } type Output = { tasks: Task[]; } export const foo: MySpecialJob<Input, Output> = async ({ name }, context) => { console.log(`Hello ${name}!`) const tasks = await context.entities.Task.findMany({}) return { tasks } }

worker 函数有两条硬性约定：

1. 必须是async函数，其返回值就是 Job 的执行结果（result）；
2. 它接受两个参数：
   • args：提交任务时传入的数据；
   • context: { entities }：包含声明中列出的实体的上下文对象。

TypeScript 中，MySpecialJob<Input, Output>是 Wasp 为每个 Job 声明生成的泛型类型，用于精确标注 worker 函数的入参与返回值，保证端到端类型安全（详见下文 JavaScript API 一节）。

这一约定在 SDK 源码中有直接印证：pg-boss 注册 worker 时，Wasp 用一层包装器（pgBossCallbackWrapper）把用户的jobFn和声明的entities组合起来——它从 pg-boss 的回调参数中取出data作为args，构造context = { entities }后调用jobFn(args.data, context)，见 pgBossJob.ts。这也解释了为什么 worker 收到的args与提交时传入的数据完全一致：pg-boss 内部的data包裹层被剥离了。

第三步：提交任务

Job 定义成功后，可以在 Operations、setupFn（见 server-config 文档）或任何其他 NodeJS 代码中提交：

import { mySpecialJob } from 'wasp/server/jobs' const submittedJob = await mySpecialJob.submit({ job: "Johnny" }) // 若希望稍后执行，只需链式追加 .delay()。 // 参数可以是秒数、Date 或 ISO 日期字符串。 await mySpecialJob .delay(10) .submit({ name: "Johnny" })

到这一步就完成了：foo会像被直接调用foo({ name: "Johnny" })一样由 PgBoss 执行。示例中foo接收参数，但给 Job 传参并非必须——取决于你的 worker 函数如何编写。

delay()的具体语义（整数秒 / ISO 字符串 / Date）在实现里对应PgBossJob类的startAfter字段：delay(startAfter)会返回一个携带该值的新 Job 实例，submit()时把它并入 pg-boss 的发送选项，见 pgBossJob.ts。

周期性任务（Recurring Jobs）

对于需要按固定节奏运行的工作，只要在 Job 声明中增加schedule块即可：

job mySpecialJob { executor: PgBoss, perform: { fn: import { foo } from "@src/workers/bar" }, schedule: { cron: "0 * * * *", args: {=json { "job": "args" } json=} // optional } }

此时你不需要在任何 JS/TS 代码中调用任何东西——foo({ job: "args" })会按照 cron 表达式自动被调度和执行。"0 * * * *"表示每小时的第 0 分钟触发。

运行时行为可在源码中确认：registerJob在 pg-boss 实例启动后，除了用boss.work(jobName, callback)注册 worker，还会检查job.jobSchedule，若存在则调用boss.schedule(jobName, cron, args, options)注册周期计划；同名 schedule 已存在时会被更新为新的 cron 表达式、参数与选项，见 pgBossJob.ts。

仓库中examples/kitchen-sink就有一个真实用例：mySpecialScheduledJob声明了cron: "0 * * * *"、args: { foo: "bar" }，并在 schedule 级覆盖了retryLimit: 2（新版本采用@wasp.sh/spec的job()声明写法，字段语义与上述main.wasp一致），参见 jobs.wasp.ts：

job(mySpecialScheduledJob, { executor: "PgBoss", schedule: { cron: "0 * * * *", args: { foo: "bar" }, executorOptions: { pgBoss: { retryLimit: 2 } }, }, }),

API 参考

Job 声明的全部字段

一份完整的声明长这样：

job mySpecialJob { executor: PgBoss, perform: { fn: import { foo } from "@src/workers/bar", executorOptions: { pgBoss: {=json { "retryLimit": 1 } json=} } }, schedule: { cron: "*/5 * * * *", args: {=json { "foo": "bar" } json=}, executorOptions: { pgBoss: {=json { "retryLimit": 0 } json=} } }, entities: [Task], }

各字段说明如下：

• executor: JobExecutor（必填）
  • Job 需要一个执行器来处理调度、监控与执行。
  • PgBoss目前是唯一可选的执行器，适合低负载（low-volume）的生产场景，要求app.db.system为PostgreSQL。
• perform: dict（必填）
  • fn: ExtImport（必填）：执行工作的async函数。由于 Wasp 在服务端运行 Job，导入路径必须指向 NodeJS 文件。它接收args: Input（提交时传入的数据）和context: { entities: Entities }（声明中列出的实体）。
  • executorOptions: dict（可选）：提交任务时使用的执行器默认选项，直接透传给 pg-boss 的send（可参考 pg-boss 官方send(name, data, options)文档）。这些默认值可以在调用submit()时或在schedule中被覆盖。
• schedule: dict（可选）
  • cron: string（必填）：5 段占位符格式的 cron 表达式。pg-boss 的调度精度为分钟级（可参考 pg-boss 官方关于 scheduling 的说明）；不确定写法时可用 Crontab Guru 之类的工具校验。
  • args: JSON（可选）：调度触发时传给perform.fn的参数。
  • executorOptions: dict（可选）：调度提交时的执行器选项。perform.executorOptions是默认值，schedule.executorOptions会覆盖/扩展它。
• entities: [Entity]（可选）：Job 内部允许使用的实体列表，用法与 Queries/Actions 相同。

从源码可以确认选项的合并规则：submit()时按defaultJobOptions（即perform.executorOptions）→startAfter（来自delay()）→ 调用方传入的jobOptions的顺序展开合并，后者覆盖前者，见 pgBossJob.ts；schedule 的选项合并逻辑同样是defaultJobOptions打底、jobSchedule.options覆盖，见 pgBossJob.ts。因此文档中“schedule 可覆盖 perform 默认选项”的说法在实现层面是确定的。

JavaScript API

导入 Job：

import { mySpecialJob } from 'wasp/server/jobs'

import { mySpecialJob, type MySpecialJob } from 'wasp/server/jobs'

类型安全说明：Wasp 为每个 Job 声明生成一个以声明名命名的泛型类型（本例为MySpecialJob），位于wasp/server/jobs模块，接收两个类型参数——Input（perform.fn的args类型）与Output（perform.fn的返回值类型）。

submit(jobArgs, executorOptions)

• jobArgs: Input
• executorOptions: object

把任务提交给执行器，可选地传入 JSON 格式的 job 参数（由 worker 函数接收）以及执行器特定的提交选项：

const submittedJob = await mySpecialJob.submit({ job: "args" })

delay(startAfter)

• startAfter: int | string | Date（必填）

延迟 worker 的执行时机，支持三种形式：

• 整数：延迟的秒数（默认 0）；
• 字符串：ISO 日期字符串，表示在该时刻执行；
• Date：在该日期时刻执行。

const submittedJob = await mySpecialJob .delay(10) .submit({ job: "args" }, { "retryLimit": 2 })

任务跟踪（Tracking）

submit()的返回值是一个SubmittedJob实例，包含：

• jobId：该任务在执行器中的 ID；
• jobName：.wasp声明中使用的 Job 名；
• executorName：执行器名称的 Symbol。

此外还有按执行器命名空间划分的方法。对 pg-boss 而言，可以访问submittedJob.pgBoss：

• details()：获取 pg-boss 的任务详细信息（对应boss.getJobById(id)）；
• cancel()：尝试取消任务（对应boss.cancel(id)）；
• resume()：尝试恢复已取消的任务（对应boss.resume(id)）。

SDK 源码中，SubmittedJob基类只持有job（Job 定义）与jobId（见 job.ts）；pg-boss 特有的PgBossSubmittedJob在其上追加了pgBoss.cancel/resume/details三个方法。值得注意的是，details()的返回类型PgBossDetails是 Wasp 对 pg-boss 原始JobWithMetadata的类型收窄：当state为completed时output精确为 Job 的Output类型（若Output是基本类型，pg-boss 会用{ value }包裹）；failed时output为 object；其余状态（created、retry、active、expired、cancelled）下output为null，见 pgBossJob.ts。

examples/kitchen-sink的 serverSetup.ts 展示了完整用法：在setupFn中await mySpecialJob.submit(...)，打印submittedJob.jobId、jobName、executorName，并调用submittedJob.pgBoss.details()查看任务详情；旁边还留了mySpecialJob.delay(10).submit({ something: "here" })的延迟用法注释。

PgBoss 执行器的工作原理与注意事项

Wasp 选择 [pg-boss] 作为首个 Job 执行器，用来处理大多数 Web 应用常见的低负载基础任务队列需求。pg-boss 直接以 PostgreSQL（利用SELECT ... SKIP LOCKED）作为存储与同步机制，使 Wasp 在不引入任何额外基础设施（如 Redis、独立 broker 进程）的情况下提供完整的队列能力：任务持久化、失败重试、延迟执行、周期调度全部落在数据库里。

examples/kitchen-sink的端到端测试 async-jobs.spec.ts 验证了这条链路：提交一个“文本转大写”任务后，断言任务状态先变为pending，worker 执行完成后变为success，且输出为输入的大写形式——说明任务确实经过 pg-boss 队列异步流转，而不是同步执行。

需要注意以下实现层面的约束：

• 与服务端共享 CPU。Wasp 在 Web 服务器应用启动的同时启动 pg-boss（生成的服务器入口会调用startPgBoss并导入wasp/server/jobs/core/allJobs.js完成注册，见 server.ts 模板），两者同时运行、共享 CPU 资源。因此 Job 不适合 CPU 密集型任务，也不要把重计算塞进 worker。从源码结构看，Wasp 目前也不支持把 pg-boss 单独水平扩展为独立 worker 进程。
• pgboss schema。pg-boss 接入后会自动在数据库创建名为pgboss的新 schema，包含job、schedule等内部跟踪表。这些表大多有name列，与你在.wasp文件中定义的 Job 标识符一一对应，并维护参数、状态、返回值、重试信息、开始与过期时间等元数据。
• 谨慎修改带 schedule 的 Job 名。.wasp文件中的 Job 名就是 pg-boss 表中name列的值。如果你改了一个原本带schedule的 Job 名，pg-boss 会继续按旧名调度，但找不到对应 handler，任务将变成过期失效的“僵尸”任务；若移除了schedule，同样需要处理。解决办法是到数据库中pgbossschema 的schedule表里删除对应的行。
• 自定义 pg-boss 实例。如果需要定制 pg-boss 的初始化，可设置环境变量PG_BOSS_NEW_OPTIONS为 JSON 字符串（对应 pg-boss 的new()初始化参数）。注意：该设置会覆盖 Wasp 的全部默认值，因此必须自行包含数据库连接信息。
• Heroku 部署：需额外把PG_BOSS_NEW_OPTIONS设为{"connectionString":"<REGULAR_HEROKU_DATABASE_URL>","ssl":{"rejectUnauthorized":false}}。原因是 pg-boss 依赖的pg扩展默认不通过 SSL 连接 Heroku Postgres，而 Heroku 强制 SSL 且使用自签证书。

最后，worker 注册时还有一个值得了解的设计：registerJob不会阻塞等待pgBossStarted完成（否则 Node 模块引导会死锁在startServer()的运行时 promise 上）；即便submit()先于 worker 注册发生也无妨，因为 pg-boss 允许在没有 worker 注册时先入队，worker 注册后从队列第一个任务开始执行。源码中的注释明确说明了这一容错设计，见 pgBossJob.ts。

小结

Wasp 的 Jobs 用一套声明式配置（.wasp中的job块）加上一个asyncworker 函数，就覆盖了后台任务的核心需求：submit()即时提交、delay()延迟执行、schedule周期触发，失败重试与持久化由 PgBoss 基于 PostgreSQL 免费获得，TypeScript 项目中还有MySpecialJob<Input, Output>泛型类型保证入参与返回值的端到端类型安全。结合SubmittedJob提供的jobId与pgBoss.details()/cancel()/resume()，你还可以对单个任务做跟踪、取消与恢复。对于不想自建队列基础设施的 Wasp 应用，这就是完整的后台任务方案。

【免费下载链接】waspThe batteries-included full-stack framework for the AI era. Develop JS/TS web apps (React, Node.js, and Prisma) using declarative code that abstracts away complex full-stack features like auth, background jobs, RPC, email sending, end-to-end type safety, single-command deployment, and more.项目地址: https://gitcode.com/GitHub_Trending/wa/wasp