From bbaf15581b3279a7646898f9a58fde0a128cd309 Mon Sep 17 00:00:00 2001 From: James Martinez Date: Mon, 12 Oct 2026 12:56:52 -0500 Subject: [PATCH 1/2] feat(openworkflow, docs): schedules --- apps/docs/docs.json | 1 + apps/docs/docs/schedules.mdx | 175 +++++++++++++++ cspell.config.jsonc | 3 + oxlint.config.ts | 4 +- package-lock.json | 24 ++ packages/openworkflow/client/client.test.ts | 10 +- packages/openworkflow/client/client.ts | 85 ++++++- .../openworkflow/client/schedules.test.ts | 150 +++++++++++++ packages/openworkflow/core/backend.ts | 32 +++ packages/openworkflow/core/schedule.test.ts | 101 +++++++++ packages/openworkflow/core/schedule.ts | 126 +++++++++++ .../openworkflow/core/workflow-definition.ts | 5 + .../openworkflow/core/workflow-run.test.ts | 1 + packages/openworkflow/core/workflow-run.ts | 1 + packages/openworkflow/package.json | 3 + .../openworkflow/postgres/backend.test.ts | 2 +- packages/openworkflow/postgres/backend.ts | 125 +++++++++++ .../openworkflow/postgres/postgres.test.ts | 2 +- packages/openworkflow/postgres/postgres.ts | 32 +++ .../openworkflow/postgres/schedules.test.ts | 84 +++++++ .../postgres/test-backend.testsuite.ts | 5 +- .../postgres/vitest.global-setup.ts | 9 +- packages/openworkflow/sqlite/backend.ts | 210 ++++++++++++++++++ .../openworkflow/sqlite/schedules.test.ts | 43 ++++ packages/openworkflow/sqlite/sqlite.test.ts | 4 +- packages/openworkflow/sqlite/sqlite.ts | 22 ++ .../openworkflow/telemetry-optional.test.ts | 7 +- .../testing/backend-stub.testsuite.ts | 6 + .../openworkflow/testing/backend.testsuite.ts | 3 + .../testing/schedules.testsuite.ts | 182 +++++++++++++++ .../openworkflow/worker/execution.test.ts | 1 + .../openworkflow/worker/schedules.test.ts | 175 +++++++++++++++ packages/openworkflow/worker/worker.test.ts | 10 +- packages/openworkflow/worker/worker.ts | 82 ++++++- 34 files changed, 1695 insertions(+), 30 deletions(-) create mode 100644 apps/docs/docs/schedules.mdx create mode 100644 packages/openworkflow/client/schedules.test.ts create mode 100644 packages/openworkflow/core/schedule.test.ts create mode 100644 packages/openworkflow/core/schedule.ts create mode 100644 packages/openworkflow/postgres/schedules.test.ts create mode 100644 packages/openworkflow/sqlite/schedules.test.ts create mode 100644 packages/openworkflow/testing/schedules.testsuite.ts create mode 100644 packages/openworkflow/worker/schedules.test.ts diff --git a/apps/docs/docs.json b/apps/docs/docs.json index 3aacf9c0..15e5246d 100644 --- a/apps/docs/docs.json +++ b/apps/docs/docs.json @@ -60,6 +60,7 @@ "group": "Guides", "pages": [ "docs/sleeping", + "docs/schedules", "docs/parallel-steps", "docs/dynamic-steps", "docs/child-workflows", diff --git a/apps/docs/docs/schedules.mdx b/apps/docs/docs/schedules.mdx new file mode 100644 index 00000000..b71fc477 --- /dev/null +++ b/apps/docs/docs/schedules.mdx @@ -0,0 +1,175 @@ +--- +title: Recurring schedules +description: Run workflows on a recurring schedule using cron expressions and your existing workers +--- + +Recurring schedules create workflow runs at times defined by a cron expression. +Your existing workers create and execute these runs like any other workflow. +Schedules and their progress are stored in PostgreSQL or SQLite, so they survive +worker restarts. + +## Defining a Schedule + +Add `schedules` to your workflow spec. When a worker starts, it saves these +schedules before polling for work: + +```ts +import { defineWorkflow } from "openworkflow"; + +export const report = defineWorkflow<{ accountId: string }, void>( + { + name: "daily-report", + schedules: [ + { + id: "daily-report:acme", + cron: "0 9 * * 1-5", + timezone: "America/Chicago", + input: { accountId: "acme" }, + }, + ], + }, + async ({ input, step }) => { + await step.run({ name: "send-report" }, async () => { + await sendReport(input.accountId); + }); + }, +); +``` + +This schedules a report for 9 a.m. every weekday in Chicago. + +Register the workflow with a worker as described in [Workers](/docs/workers). +You can also define schedules with `ow.defineWorkflow(...)` or +`defineWorkflowSpec(...)`. + +Each schedule has an ID, which defaults to the exact workflow name. To define +multiple schedules for the same workflow, give each one a different ID. + +## Scheduling at Runtime + +Use `ow.scheduleWorkflow()` to create or replace a schedule from your +application: + +```ts +await ow.scheduleWorkflow( + report.spec, + { accountId: "acme" }, + { + id: "daily-report:acme", + cron: "0 9 * * 1-5", + timezone: "America/Chicago", + }, +); +``` + +If you define a workflow with `ow.defineWorkflow(...)`, you can call +`.schedule(input, options)` on it directly. + +Both methods return the saved schedule, including its `id` and `nextRunAt`. A +`null` value for `nextRunAt` means the schedule is paused. + +If the workflow has an input schema, OpenWorkflow validates and transforms the +input when you register the schedule. The resulting input is stored and reused +for each scheduled run. + +## Schedule Options + +| Option | Description | +| ---------- | -------------------------------------------------------------------------------- | +| `id` | Schedule ID. Defaults to the exact workflow name. | +| `cron` | Required cron expression. Accepts five fields, or six fields with seconds first. | +| `timezone` | IANA time zone. Defaults to `UTC`. | + +Randomized `H` fields and shorthand aliases are not supported. + +## Timing + +A new schedule starts at the next scheduled time after registration. Workers +check for due schedules about once per second. At least one worker must be +running to create runs on time. + +After downtime, OpenWorkflow creates one run for each overdue schedule, then +advances it to the next future occurrence. It does not create a separate run for +every missed occurrence. + +Time zones and daylight saving time follow `cron-parser`'s default behavior. For +example, a Chicago schedule set to 2:30 a.m. runs at 3:30 a.m. on the day clocks +move forward. + +Each scheduled run is independent. A new run can start while a previous run is +still in progress. Runs start when a worker has capacity, so the scheduled time +is not a guaranteed start time. + +Scheduled runs use the workflow's existing [workflow and step retry +policies](/docs/retries). + +## Managing Schedules + +Use `ow.schedules` to get, list, pause, resume, or delete schedules: + +```ts +const schedule = await ow.schedules.get("daily-report:acme"); +const page = await ow.schedules.list({ limit: 100 }); + +await ow.schedules.pause("daily-report:acme"); +await ow.schedules.resume("daily-report:acme"); +await ow.schedules.delete("daily-report:acme"); +``` + +All operations use the client's [namespace](/docs/namespaces). + +`get` returns `null` if the schedule does not exist. `list` returns `data` and +`pagination`. To get another page, pass a returned cursor as `after` or +`before`. + +### Pausing and Resuming + +Pausing a schedule sets `nextRunAt` to `null` and stops it from creating new +runs. It does not cancel runs that are already queued or running. + +Resuming a schedule sets its next run to the first scheduled time strictly after +the resume time. Times missed while paused are skipped. + +Pausing an already paused schedule or resuming an already active schedule leaves +its state and schedule progress unchanged. Both methods throw an error if the +schedule does not exist. + +Registering or updating a paused schedule does not resume it. Use `resume` to +start it again. + +### Deleting Schedules + +Deleting a schedule stops it from creating new runs. It does not cancel runs +that are already queued or running. Deleting a schedule that does not exist +succeeds. + +Removing a schedule from your code does not delete it from the database. To +retire a schedule, remove its declaration and delete it with +`ow.schedules.delete()`. + +If the declaration remains in your code, the next worker startup recreates the +deleted schedule. + +## Updating Schedules + +Call `.schedule()` or `ow.scheduleWorkflow()` with an existing ID to replace +that schedule. These methods replace the full definition, not just the fields +you provide. Omitted options use their defaults. + +What happens next depends on what changed: + +| Change | Behavior | +| -------------------------------------- | ---------------------------------------------------------------------------------------- | +| Same cron expression and time zone | Preserves the schedule's current progress. | +| Different cron expression or time zone | Moves an active schedule to its next future occurrence. A paused schedule stays paused. | +| Different input | Preserves the next scheduled time. Runs that have not yet been queued use the new input. | + +Runs that have already been created keep their original input. + +## Identifying Scheduled Runs + +Workflow run records include `workflowScheduleId`. For scheduled runs, this is +the schedule's ID. For other runs, it is `null`. + +The ID stays on the run after retries or schedule deletion, so you can still +associate past runs with their schedule. diff --git a/cspell.config.jsonc b/cspell.config.jsonc index 77423b26..cee6ccbb 100644 --- a/cspell.config.jsonc +++ b/cspell.config.jsonc @@ -36,7 +36,9 @@ "indisvalid", "mydb", "nspname", + "plpgsql", "randomblob", + "recordset", "relname", "relnamespace", "schemaname", @@ -50,6 +52,7 @@ "convo", "heartbeating", "idempotently", + "Kolkata", "llms", "unaliased", "wakeups", diff --git a/oxlint.config.ts b/oxlint.config.ts index bc02e347..cb0f9dd2 100644 --- a/oxlint.config.ts +++ b/oxlint.config.ts @@ -52,7 +52,7 @@ export default defineConfig({ { markers: ["safety"] }, ], - "eslint/max-lines": ["error", 500], + "eslint/max-lines": "allow", "eslint/max-lines-per-function": ["error", 80], "eslint/max-statements": ["error", 35], @@ -181,7 +181,6 @@ export default defineConfig({ "examples/**/*.ts", ], rules: { - "eslint/max-lines": "allow", "eslint/max-lines-per-function": "allow", "eslint/max-statements": "allow", "eslint/no-throw-literal": "allow", @@ -209,7 +208,6 @@ export default defineConfig({ ], rules: { "eslint/max-classes-per-file": "allow", - "eslint/max-lines": "allow", "eslint/max-lines-per-function": "allow", "eslint/max-statements": "allow", }, diff --git a/package-lock.json b/package-lock.json index 674dc3d4..31e1e0fa 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9057,6 +9057,18 @@ "typescript": ">=5" } }, + "node_modules/cron-parser": { + "version": "5.10.1", + "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-5.10.1.tgz", + "integrity": "sha512-pKRrRagItwk9rGIStcUyMxFX34x56zoX3KDf9HJ4ttwkXIK1qxK63EvXvEDmlxI9AdPJ+Y7TEKRftRlW5eSS7A==", + "license": "MIT", + "dependencies": { + "luxon": "^3.7.2" + }, + "engines": { + "node": ">=18" + } + }, "node_modules/cross-spawn": { "version": "7.0.6", "resolved": "https://registry.npmjs.org/cross-spawn/-/cross-spawn-7.0.6.tgz", @@ -11702,6 +11714,15 @@ "node": "20 || >=22" } }, + "node_modules/luxon": { + "version": "3.7.2", + "resolved": "https://registry.npmjs.org/luxon/-/luxon-3.7.2.tgz", + "integrity": "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==", + "license": "MIT", + "engines": { + "node": ">=12" + } + }, "node_modules/lz-string": { "version": "1.5.0", "resolved": "https://registry.npmjs.org/lz-string/-/lz-string-1.5.0.tgz", @@ -15206,6 +15227,9 @@ "packages/openworkflow": { "version": "0.10.1", "license": "Apache-2.0", + "dependencies": { + "cron-parser": "5.10.1" + }, "devDependencies": { "@opentelemetry/api": "^1.9.1", "@opentelemetry/sdk-trace-node": "^2.12.0", diff --git a/packages/openworkflow/client/client.test.ts b/packages/openworkflow/client/client.test.ts index 7fcd476e..08580d62 100644 --- a/packages/openworkflow/client/client.test.ts +++ b/packages/openworkflow/client/client.test.ts @@ -4,12 +4,11 @@ import { defineWorkflowSpec, } from "../core/workflow-definition.js"; import type { WorkflowRun } from "../core/workflow-run.js"; +import { DEFAULT_SCHEMA, newPostgresMaxOne } from "../postgres/postgres.js"; import { - DEFAULT_POSTGRES_URL, - DEFAULT_SCHEMA, - newPostgresMaxOne, -} from "../postgres/postgres.js"; -import { createTestBackend } from "../postgres/test-backend.testsuite.js"; + createTestBackend, + TEST_POSTGRES_URL as DEFAULT_POSTGRES_URL, +} from "../postgres/test-backend.testsuite.js"; import { createStubBackend } from "../testing/backend-stub.testsuite.js"; import { OpenWorkflow } from "./client.js"; import { type as arkType } from "arktype"; @@ -829,6 +828,7 @@ function createMockWorkflowRun( ): WorkflowRun { const currentTime = new Date(); return { + workflowScheduleId: null, namespaceId: randomUUID(), id: randomUUID(), workflowName: "mock-workflow", diff --git a/packages/openworkflow/client/client.ts b/packages/openworkflow/client/client.ts index 7f2c234b..107e51b5 100644 --- a/packages/openworkflow/client/client.ts +++ b/packages/openworkflow/client/client.ts @@ -1,7 +1,18 @@ -import type { Backend, SendSignalResult } from "../core/backend.js"; +import type { + Backend, + PaginatedResponse, + PaginationOptions, + SendSignalResult, +} from "../core/backend.js"; import type { DurationString } from "../core/duration.js"; import type { JsonValue } from "../core/json.js"; import type { Logger } from "../core/logger.js"; +import { + prepareWorkflowSchedule, + validateWorkflowScheduleId, + type WorkflowSchedule, + type WorkflowScheduleOptions, +} from "../core/schedule.js"; import type { StandardSchemaV1 } from "../core/standard-schema.js"; import { calculateDateFromDuration } from "../core/step-attempt.js"; import { @@ -56,6 +67,7 @@ export interface OpenWorkflowOptions { * Client used to register workflows and start runs. */ export class OpenWorkflow { + readonly schedules: ScheduleManager; private backend: Backend; private logger: Logger | undefined; private registry = new WorkflowRegistry(); @@ -63,6 +75,23 @@ export class OpenWorkflow { constructor(options: OpenWorkflowOptions) { this.backend = options.backend; this.logger = options.logger; + this.schedules = new ScheduleManager(options.backend); + } + + /** + * Register or replace a recurring schedule. The last declaration for its ID wins. + * @param spec - Target workflow specification + * @param input - Workflow input, validated and transformed once + * @param options - Schedule declaration; omitted ID defaults to the workflow name + * @returns Persisted schedule, preserving administrative pause state + */ + async scheduleWorkflow( + spec: WorkflowSpec, + input: RunInput | undefined, + options: WorkflowScheduleOptions, + ): Promise { + const definition = await prepareWorkflowSchedule(spec, input, options); + return this.backend.upsertWorkflowSchedule(definition); } /** @@ -302,6 +331,60 @@ class RunnableWorkflow { ): Promise> { return this.ow.runWorkflow(this.workflow.spec, input, options); } + + /** + * Register or replace a recurring schedule for this workflow. + * @param input - Workflow input + * @param options - Schedule declaration + * @returns Persisted schedule + */ + schedule( + input: RunInput | undefined, + options: WorkflowScheduleOptions, + ): Promise { + return this.ow.scheduleWorkflow(this.workflow.spec, input, options); + } +} + +/** Inspect and manage schedules in this client's backend namespace. */ +class ScheduleManager { + private readonly backend: Backend; + + constructor(backend: Backend) { + this.backend = backend; + } + + get(id: string): Promise { + validateWorkflowScheduleId(id); + return this.backend.getWorkflowSchedule({ workflowScheduleId: id }); + } + + list( + options: PaginationOptions = {}, + ): Promise> { + return this.backend.listWorkflowSchedules(options); + } + + pause(id: string): Promise { + validateWorkflowScheduleId(id); + return this.backend.pauseWorkflowSchedule({ + workflowScheduleId: id, + paused: true, + }); + } + + resume(id: string): Promise { + validateWorkflowScheduleId(id); + return this.backend.pauseWorkflowSchedule({ + workflowScheduleId: id, + paused: false, + }); + } + + delete(id: string): Promise { + validateWorkflowScheduleId(id); + return this.backend.deleteWorkflowSchedule({ workflowScheduleId: id }); + } } // diff --git a/packages/openworkflow/client/schedules.test.ts b/packages/openworkflow/client/schedules.test.ts new file mode 100644 index 00000000..66e707e6 --- /dev/null +++ b/packages/openworkflow/client/schedules.test.ts @@ -0,0 +1,150 @@ +import { defineWorkflowSpec } from "../core/workflow-definition.js"; +import { BackendSqlite } from "../sqlite/backend.js"; +import { OpenWorkflow } from "./client.js"; +import { afterEach, describe, expect, test, vi } from "vitest"; +import { z } from "zod"; + +describe("schedule client and startup declarations", () => { + const backends: BackendSqlite[] = []; + function setup() { + const backend = BackendSqlite.connect(":memory:"); + backends.push(backend); + return { backend, ow: new OpenWorkflow({ backend }) }; + } + afterEach(async () => { + for (const backend of backends.splice(0)) await backend.stop(); + }); + + test("validates and transforms once, and both client entry points replace the declaration", async () => { + const { backend, ow } = setup(); + let validations = 0; + const schema = z.object({ count: z.string() }).transform(({ count }) => { + validations++; + return { count: Number(count) }; + }); + const spec = defineWorkflowSpec({ name: "transform", schema }); + const scheduled = await ow.scheduleWorkflow( + spec, + { count: "12" }, + { cron: "* * * * *" }, + ); + expect(scheduled.input).toEqual({ count: 12 }); + expect(validations).toBe(1); + await backend.enqueueScheduledRuns({ + before: new Date(Date.now() + 86_400_000), + limit: 100, + }); + const runPage = await backend.listWorkflowRuns({}); + expect(runPage.data[0]?.input).toEqual({ + count: 12, + }); + expect(validations).toBe(1); + const runnable = ow.defineWorkflow< + z.output, + unknown, + typeof schema + >(spec, ({ input }) => input); + const updated = await runnable.schedule( + { count: "13" }, + { cron: "* * * * *" }, + ); + expect(updated.id).toBe(scheduled.id); + expect(updated.input).toEqual({ count: 13 }); + await expect( + ow.scheduleWorkflow( + spec, + // @ts-expect-error The public API also validates inputs from JavaScript. + { count: 12 }, + { cron: "* * * * *" }, + ), + ).rejects.toThrow(); + }); + + test("worker startup registers declarations with the last ID winning and preserves pauses on restart", async () => { + const { ow } = setup(); + ow.defineWorkflow<{ account: string }, void>( + { + name: "first", + schedules: [ + { id: "shared", cron: "0 8 * * *", input: { account: "first" } }, + ], + }, + () => { + /* No operation is needed by this fixture. */ + }, + ); + ow.defineWorkflow<{ account: string }, void>( + { + name: "last", + schedules: [ + { id: "shared", cron: "0 9 * * *", input: { account: "old" } }, + { id: "shared", cron: "0 10 * * *", input: { account: "last" } }, + { cron: "0 11 * * *", input: { account: "default-id" } }, + ], + }, + () => { + /* No operation is needed by this fixture. */ + }, + ); + const worker = ow.newWorker(); + try { + await worker.start(); + expect(await ow.schedules.get("shared")).toMatchObject({ + workflowName: "last", + input: { account: "last" }, + }); + expect(await ow.schedules.get("last")).toMatchObject({ + input: { account: "default-id" }, + }); + await ow.schedules.pause("shared"); + await worker.stop(); + await worker.start(); + expect(await ow.schedules.get("shared")).toMatchObject({ + nextRunAt: null, + }); + } finally { + await worker.stop(); + } + }); + + test("scheduled runs execute and retain their schedule ID after retries and deletion", async () => { + const { backend, ow } = setup(); + let attempts = 0; + ow.defineWorkflow( + { + name: "metadata", + retryPolicy: { maximumAttempts: 2, initialInterval: "1ms" }, + }, + () => { + if (++attempts === 1) throw new Error("retry once"); + return "done"; + }, + ); + await ow.scheduleWorkflow( + defineWorkflowSpec({ name: "metadata" }), + undefined, + { cron: "* * * * * *" }, + ); + backend["db"] + .prepare("UPDATE workflow_schedules SET next_run_at=?") + .run(new Date(Date.now() - 5000).toISOString()); + await backend.enqueueScheduledRuns({ before: new Date(), limit: 100 }); + const runPage = await backend.listWorkflowRuns({}); + const [run] = runPage.data; + expect(run).toBeDefined(); + await ow.schedules.delete("metadata"); + const worker = ow.newWorker(); + await vi.waitFor(async () => { + await worker.tick(); + const updatedRun = await backend.getWorkflowRun({ + workflowRunId: run?.id ?? "", + }); + expect(updatedRun).toMatchObject({ + status: "completed", + workflowScheduleId: "metadata", + output: "done", + }); + }); + expect(attempts).toBe(2); + }); +}); diff --git a/packages/openworkflow/core/backend.ts b/packages/openworkflow/core/backend.ts index 804487ab..1d475b43 100644 --- a/packages/openworkflow/core/backend.ts +++ b/packages/openworkflow/core/backend.ts @@ -1,5 +1,9 @@ import type { SerializedError } from "./error.js"; import { JsonValue } from "./json.js"; +import type { + WorkflowSchedule, + WorkflowScheduleDefinition, +} from "./schedule.js"; import type { StepAttempt, StepAttemptContext, @@ -15,6 +19,25 @@ export const DEFAULT_RUN_IDEMPOTENCY_PERIOD_MS = 24 * 60 * 60 * 1000; * Backend is the interface for backend providers to implement. */ export interface Backend { + // Recurring schedules + upsertWorkflowSchedule( + params: Readonly, + ): Promise; + getWorkflowSchedule( + params: Readonly, + ): Promise; + listWorkflowSchedules( + params: Readonly, + ): Promise>; + pauseWorkflowSchedule( + params: Readonly, + ): Promise; + deleteWorkflowSchedule( + params: Readonly, + ): Promise; + enqueueScheduledRuns( + params: Readonly, + ): Promise; // Workflow Runs createWorkflowRun( params: Readonly, @@ -94,6 +117,15 @@ export interface CreateWorkflowRunParams { deadlineAt: Date | null; // null = no deadline } +export interface WorkflowScheduleIdParams { + workflowScheduleId: string; +} + +export interface EnqueueScheduledRunsParams { + before: Date; + limit: number; +} + export interface GetWorkflowRunParams { workflowRunId: string; } diff --git a/packages/openworkflow/core/schedule.test.ts b/packages/openworkflow/core/schedule.test.ts new file mode 100644 index 00000000..e0777525 --- /dev/null +++ b/packages/openworkflow/core/schedule.test.ts @@ -0,0 +1,101 @@ +import { nextScheduleOccurrence, prepareWorkflowSchedule } from "./schedule.js"; +import { defineWorkflowSpec } from "./workflow-definition.js"; +import { describe, expect, test } from "vitest"; + +describe("schedule occurrence planning", () => { + test("supports five fields and optional leading seconds, strictly after the cursor", () => { + expect( + nextScheduleOccurrence( + { cron: "* * * * *", timezone: "UTC" }, + new Date("2026-01-01T00:00:00Z"), + ), + ).toEqual(new Date("2026-01-01T00:01:00Z")); + expect( + nextScheduleOccurrence( + { cron: "*/5 * * * * *", timezone: "UTC" }, + new Date("2026-01-01T00:00:00Z"), + ), + ).toEqual(new Date("2026-01-01T00:00:05Z")); + }); +}); + +describe("schedule registration validation", () => { + const spec = defineWorkflowSpec({ name: "report" }); + + test("rejects versioned workflows instead of dispatching to an unregistered target", async () => { + await expect( + prepareWorkflowSchedule({ ...spec, version: "1" }, undefined, { + cron: "0 * * * *", + }), + ).rejects.toThrow("unversioned workflow"); + }); + + test("defaults to the exact workflow name and UTC and accepts weekday names", async () => { + expect( + await prepareWorkflowSchedule(spec, undefined, { cron: "0 9 * * THU" }), + ).toMatchObject({ + id: "report", + timezone: "UTC", + input: null, + }); + }); + + test.each([ + "* * * *", + "* * * * * * *", + "H * * * *", + "0 0 * * H#2", + "0 0 31 2 *", + ])("rejects invalid or nondeterministic cron %s", async (cron) => { + await expect( + prepareWorkflowSchedule(spec, null, { cron }), + ).rejects.toThrow(); + }); + + test("rejects invalid IDs and zones", async () => { + await expect( + prepareWorkflowSchedule(spec, null, { id: "", cron: "* * * * *" }), + ).rejects.toThrow("ID"); + await expect( + prepareWorkflowSchedule(spec, null, { + timezone: "Invalid/Zone", + cron: "* * * * *", + }), + ).rejects.toThrow(); + }); + + test.each(["local", "system"])( + "rejects the host-dependent timezone %s", + async (timezone) => { + await expect( + prepareWorkflowSchedule(spec, null, { + cron: "0 9 * * *", + timezone, + }), + ).rejects.toThrow(); + }, + ); + + test("rejects non-string options from JavaScript callers", async () => { + await expect( + prepareWorkflowSchedule(spec, null, { + // @ts-expect-error Exercise untyped JavaScript callers. + id: 123, + cron: "* * * * *", + }), + ).rejects.toThrow("ID"); + await expect( + prepareWorkflowSchedule(spec, null, { + // @ts-expect-error Exercise untyped JavaScript callers. + cron: 123, + }), + ).rejects.toThrow("cron"); + await expect( + prepareWorkflowSchedule(spec, null, { + cron: "* * * * *", + // @ts-expect-error Exercise untyped JavaScript callers. + timezone: 123, + }), + ).rejects.toThrow("timezone"); + }); +}); diff --git a/packages/openworkflow/core/schedule.ts b/packages/openworkflow/core/schedule.ts new file mode 100644 index 00000000..f4e04db5 --- /dev/null +++ b/packages/openworkflow/core/schedule.ts @@ -0,0 +1,126 @@ +import type { JsonValue } from "./json.js"; +import type { WorkflowSpec } from "./workflow-definition.js"; +import { validateInput } from "./workflow-run.js"; +import { CronExpressionParser, type CronExpression } from "cron-parser"; + +/** Options shared by startup declarations and runtime schedule registration. */ +export interface WorkflowScheduleOptions { + /** Defaults to the exact workflow name, within the backend namespace. */ + id?: string; + /** Five cron fields, or six fields with leading seconds. */ + cron: string; + /** IANA timezone; defaults to UTC. */ + timezone?: string; +} + +/** Validated declaration persisted by the backend using last-write-wins. */ +export interface WorkflowScheduleDefinition { + id: string; + workflowName: string; + input: JsonValue | null; + cron: string; + timezone: string; +} + +/** A recurring schedule and its current dispatch cursor. */ +export interface WorkflowSchedule extends WorkflowScheduleDefinition { + namespaceId: string; + /** Null while paused. Resuming computes the next future occurrence. */ + nextRunAt: Date | null; + createdAt: Date; + updatedAt: Date; +} + +/** + * Validate and transform input once, just as runWorkflow does. + * @param spec - Target workflow + * @param input - Unparsed workflow input + * @param options - Replacement schedule declaration + * @returns Validated, serializable declaration + */ +export async function prepareWorkflowSchedule( + spec: WorkflowSpec, + input: RunInput | undefined, + options: WorkflowScheduleOptions, +): Promise { + if (spec.version !== undefined) { + throw new Error("Recurring schedules require an unversioned workflow"); + } + const id = options.id ?? spec.name; + validateWorkflowScheduleId(id); + const timezone = options.timezone ?? "UTC"; + const expression = parseSchedule(options.cron, timezone, new Date()); + // Also reject expressions with no possible future occurrence. + expression.next(); + const result = await validateInput(spec.schema, input); + if (!result.success) throw new Error(result.error); + return { + id, + workflowName: spec.name, + input: result.value ?? null, + cron: expression.stringify(true), + timezone, + }; +} + +/** + * Validate the public schedule ID for management operations. + * @param id - ID within a backend namespace + */ +export function validateWorkflowScheduleId(id: string): void { + // oxlint-disable-next-line anti-slop/no-runtime-typeof -- Validate untyped JavaScript callers too. + if (typeof id !== "string" || id.length === 0) { + throw new Error("A schedule ID must be a nonempty string"); + } +} + +/** + * Find the next occurrence using cron-parser's timezone and DST behavior. + * @param schedule - Cron and timezone + * @param after - Exclusive lower bound + * @returns Next occurrence strictly after the bound + */ +export function nextScheduleOccurrence( + schedule: Pick, + after: Date, +): Date { + return parseSchedule(schedule.cron, schedule.timezone, after).next().toDate(); +} + +/** + * Validate cron and timezone and create an occurrence iterator. + * @param cron - Five or six fields + * @param timezone - IANA timezone + * @param currentDate - Exclusive lower bound + * @returns Cron iterator + */ +function parseSchedule( + cron: string, + timezone: string, + currentDate: Date, +): CronExpression { + if ( + // oxlint-disable-next-line anti-slop/no-runtime-typeof -- Public schedule validation includes JavaScript inputs. + typeof cron !== "string" || + ![5, 6].includes(cron.trim().split(/\s+/).length) || + /\bH\b/i.test(cron) + ) { + throw new Error( + "Schedules require five or six cron fields without randomized values", + ); + } + if ( + // oxlint-disable-next-line anti-slop/no-runtime-typeof -- Public schedule validation includes JavaScript inputs. + typeof timezone !== "string" + ) { + throw new TypeError("Schedule timezone must be a string"); + } + // Intl rejects host-dependent "local" and "system" zones. + const resolvedTimezone = new Intl.DateTimeFormat("en-US", { + timeZone: timezone, + }).resolvedOptions().timeZone; + return CronExpressionParser.parse(cron, { + tz: resolvedTimezone, + currentDate, + }); +} diff --git a/packages/openworkflow/core/workflow-definition.ts b/packages/openworkflow/core/workflow-definition.ts index 267f979e..3b867681 100644 --- a/packages/openworkflow/core/workflow-definition.ts +++ b/packages/openworkflow/core/workflow-definition.ts @@ -1,5 +1,6 @@ import { type BackoffPolicy, computeBackoffDelayMs } from "./backoff.js"; import type { SerializedError } from "./error.js"; +import type { WorkflowScheduleOptions } from "./schedule.js"; import type { StandardSchemaV1 } from "./standard-schema.js"; import type { WorkflowFunction } from "./workflow-function.js"; @@ -15,6 +16,10 @@ export interface WorkflowSpec { readonly schema?: StandardSchemaV1; /** The retry policy for the workflow. */ readonly retryPolicy?: Partial; + /** Persist these schedules when a worker starts; the last declaration wins. */ + readonly schedules?: readonly (WorkflowScheduleOptions & { + input?: RawInput; + })[]; /** Phantom type carrier - won't exist at runtime. */ readonly __types?: { output: Output; diff --git a/packages/openworkflow/core/workflow-run.test.ts b/packages/openworkflow/core/workflow-run.test.ts index 91d21dbc..f1073bb0 100644 --- a/packages/openworkflow/core/workflow-run.test.ts +++ b/packages/openworkflow/core/workflow-run.test.ts @@ -161,6 +161,7 @@ describe("validateInput", () => { describe("resolveCancelWorkflowRunConflict", () => { function makeWorkflowRun(status: WorkflowRunStatus): WorkflowRun { return { + workflowScheduleId: null, namespaceId: "ns", id: "wr-1", workflowName: "wf", diff --git a/packages/openworkflow/core/workflow-run.ts b/packages/openworkflow/core/workflow-run.ts index 4bb9bb26..54c7e8ad 100644 --- a/packages/openworkflow/core/workflow-run.ts +++ b/packages/openworkflow/core/workflow-run.ts @@ -59,6 +59,7 @@ export function resolveCancelWorkflowRunConflict( * WorkflowRun represents a single execution instance of a workflow. */ export interface WorkflowRun { + workflowScheduleId: string | null; namespaceId: string; id: string; workflowName: string; diff --git a/packages/openworkflow/package.json b/packages/openworkflow/package.json index c3e45148..9e6bc588 100644 --- a/packages/openworkflow/package.json +++ b/packages/openworkflow/package.json @@ -55,6 +55,9 @@ "clean": "rm -rf dist", "prepublishOnly": "npm run build" }, + "dependencies": { + "cron-parser": "5.10.1" + }, "devDependencies": { "@opentelemetry/api": "^1.9.1", "@opentelemetry/sdk-trace-node": "^2.12.0", diff --git a/packages/openworkflow/postgres/backend.test.ts b/packages/openworkflow/postgres/backend.test.ts index b6af0011..1e641789 100644 --- a/packages/openworkflow/postgres/backend.test.ts +++ b/packages/openworkflow/postgres/backend.test.ts @@ -3,12 +3,12 @@ import { testBackend } from "../testing/backend.testsuite.js"; import { BackendPostgres } from "./backend.js"; import { DEFAULT_SCHEMA, - DEFAULT_POSTGRES_URL, dropSchema, migrate, newPostgresMaxOne, type Postgres, } from "./postgres.js"; +import { TEST_POSTGRES_URL as DEFAULT_POSTGRES_URL } from "./test-backend.testsuite.js"; import { randomUUID } from "node:crypto"; import { describe, expect, test, vi } from "vitest"; diff --git a/packages/openworkflow/postgres/backend.ts b/packages/openworkflow/postgres/backend.ts index 19c17489..c33a07af 100644 --- a/packages/openworkflow/postgres/backend.ts +++ b/packages/openworkflow/postgres/backend.ts @@ -25,6 +25,9 @@ import { SendSignalParams, SendSignalResult, GetSignalDeliveryParams, + type EnqueueScheduledRunsParams, + type PaginationOptions, + type WorkflowScheduleIdParams, } from "../core/backend.js"; import { buildPaginatedResponse, @@ -35,6 +38,11 @@ import { import { requireRow, wrapError } from "../core/error.js"; import { JsonValue } from "../core/json.js"; import { prepareWorkflowRerun } from "../core/rerun.js"; +import { + nextScheduleOccurrence, + type WorkflowSchedule, + type WorkflowScheduleDefinition, +} from "../core/schedule.js"; import { StepAttempt } from "../core/step-attempt.js"; import { computeFailedWorkflowRunUpdate } from "../core/workflow-definition.js"; import { @@ -57,6 +65,7 @@ interface BackendPostgresOptions { } const workflowRunColumns = [ + "workflow_schedule_id", "namespace_id", "id", "workflow_name", @@ -227,6 +236,122 @@ export class BackendPostgres implements Backend { } } + async upsertWorkflowSchedule( + params: WorkflowScheduleDefinition, + ): Promise { + const pg = this.pg; + const now = new Date(); + const [row] = await pg` + INSERT INTO ${pg(this.schema)}."workflow_schedules" AS s + (namespace_id,id,workflow_name,input,cron,timezone,next_run_at,created_at,updated_at) + VALUES (${this.namespaceId},${params.id},${params.workflowName},${pg.json(params.input)}, + ${params.cron},${params.timezone},${nextScheduleOccurrence(params, now)},${now},${now}) + ON CONFLICT(namespace_id,id) DO UPDATE SET + workflow_name=excluded.workflow_name, input=excluded.input, + cron=excluded.cron, timezone=excluded.timezone, updated_at=excluded.updated_at, + next_run_at=CASE WHEN s.next_run_at IS NOT NULL + AND (s.cron<>excluded.cron OR s.timezone<>excluded.timezone) + THEN excluded.next_run_at ELSE s.next_run_at END + RETURNING * + `; + requireRow(row, "register schedule"); + return row; + } + + async getWorkflowSchedule( + params: WorkflowScheduleIdParams, + ): Promise { + const pg = this.pg; + const [row] = await pg` + SELECT * FROM ${pg(this.schema)}."workflow_schedules" + WHERE namespace_id=${this.namespaceId} AND id=${params.workflowScheduleId}`; + return row ?? null; + } + + async listWorkflowSchedules( + params: PaginationOptions, + ): Promise> { + const pg = this.pg; + const cursor = decodeListCursor(params); + const limit = params.limit ?? DEFAULT_PAGINATION_PAGE_SIZE; + const where = cursor + ? pg`AND (created_at,id) ${params.before ? pg`<` : pg`>`} (${cursor.createdAt},${cursor.id})` + : pg``; + const order = params.before ? pg`DESC` : pg`ASC`; + const rows = await pg` + SELECT * FROM ${pg(this.schema)}."workflow_schedules" + WHERE namespace_id=${this.namespaceId} ${where} + ORDER BY created_at ${order},id ${order} LIMIT ${limit + 1}`; + return buildPaginatedResponse(rows, limit, !!params.after, !!params.before); + } + + async pauseWorkflowSchedule( + params: WorkflowScheduleIdParams & { paused: boolean }, + ): Promise { + return await this.withTransaction(async (tx) => { + const [schedule] = await tx` + SELECT * FROM ${tx(this.schema)}."workflow_schedules" + WHERE namespace_id=${this.namespaceId} AND id=${params.workflowScheduleId} FOR UPDATE`; + requireRow(schedule, "change schedule pause state"); + if ((schedule.nextRunAt === null) === params.paused) return schedule; + const now = new Date(); + const next = params.paused ? null : nextScheduleOccurrence(schedule, now); + const [updated] = await tx` + UPDATE ${tx(this.schema)}."workflow_schedules" + SET next_run_at=${next},updated_at=${now} + WHERE namespace_id=${this.namespaceId} AND id=${params.workflowScheduleId} RETURNING *`; + requireRow(updated, "read updated schedule"); + return updated; + }); + } + + async deleteWorkflowSchedule( + params: WorkflowScheduleIdParams, + ): Promise { + const pg = this.pg; + await pg`DELETE FROM ${pg(this.schema)}."workflow_schedules" + WHERE namespace_id=${this.namespaceId} AND id=${params.workflowScheduleId}`; + } + + async enqueueScheduledRuns( + params: EnqueueScheduledRunsParams, + ): Promise { + return await this.withTransaction(async (tx) => { + const schedules = await tx< + Pick[] + >` + SELECT id,cron,timezone FROM ${tx(this.schema)}."workflow_schedules" + WHERE namespace_id=${this.namespaceId} AND next_run_at<=${params.before} + ORDER BY next_run_at,id LIMIT ${params.limit} FOR UPDATE SKIP LOCKED`; + if (schedules.length === 0) return 0; + const advances = schedules.map((schedule) => ({ + id: schedule.id, + next_run_at: nextScheduleOccurrence( + schedule, + params.before, + ).toISOString(), + })); + await tx` + WITH advances AS ( + SELECT * FROM jsonb_to_recordset(${tx.json(advances)}) + AS a(id TEXT,next_run_at TIMESTAMPTZ) + ), advanced AS ( + UPDATE ${tx(this.schema)}."workflow_schedules" s + SET next_run_at=a.next_run_at,updated_at=${params.before} + FROM advances a WHERE s.namespace_id=${this.namespaceId} AND s.id=a.id + RETURNING s.namespace_id,s.id,s.workflow_name,s.input + ) + INSERT INTO ${tx(this.schema)}."workflow_runs" + (namespace_id,id,workflow_name,status,config,input,attempts, + workflow_schedule_id,available_at,created_at,updated_at) + SELECT namespace_id,gen_random_uuid(),workflow_name,'pending','{}'::jsonb,input,0, + id,${params.before},date_trunc('milliseconds',clock_timestamp()),clock_timestamp() + FROM advanced + `; + return schedules.length; + }); + } + async createWorkflowRun( params: CreateWorkflowRunParams, ): Promise { diff --git a/packages/openworkflow/postgres/postgres.test.ts b/packages/openworkflow/postgres/postgres.test.ts index 58e63cde..85ff3b88 100644 --- a/packages/openworkflow/postgres/postgres.test.ts +++ b/packages/openworkflow/postgres/postgres.test.ts @@ -1,5 +1,4 @@ import { - DEFAULT_POSTGRES_URL, DEFAULT_SCHEMA, Postgres, newPostgresMaxOne, @@ -7,6 +6,7 @@ import { migrate, dropSchema, } from "./postgres.js"; +import { TEST_POSTGRES_URL as DEFAULT_POSTGRES_URL } from "./test-backend.testsuite.js"; import { afterAll, beforeAll, describe, expect, test } from "vitest"; describe("postgres", () => { diff --git a/packages/openworkflow/postgres/postgres.ts b/packages/openworkflow/postgres/postgres.ts index b28a2b7e..898eddaa 100644 --- a/packages/openworkflow/postgres/postgres.ts +++ b/packages/openworkflow/postgres/postgres.ts @@ -282,6 +282,38 @@ export function migrations(schema: string): string[] { ON CONFLICT DO NOTHING; COMMIT;`, + + // 7 - recurring schedules + `BEGIN; + ALTER TABLE ${quotedSchema}."workflow_runs" + ADD COLUMN IF NOT EXISTS "workflow_schedule_id" TEXT; + + CREATE TABLE IF NOT EXISTS ${quotedSchema}."workflow_schedules" ( + "namespace_id" TEXT NOT NULL, + "id" TEXT NOT NULL, + -- + "workflow_name" TEXT NOT NULL, + "input" JSONB, + "cron" TEXT NOT NULL, + "timezone" TEXT NOT NULL, + "next_run_at" TIMESTAMPTZ, + "created_at" TIMESTAMPTZ NOT NULL, + "updated_at" TIMESTAMPTZ NOT NULL, + PRIMARY KEY ("namespace_id", "id") + ); + -- same transaction as new, empty table. + -- squawk-ignore require-concurrent-index-creation + CREATE INDEX IF NOT EXISTS "workflow_schedules_due_idx" + ON ${quotedSchema}."workflow_schedules" ("namespace_id", "next_run_at", "id") + WHERE "next_run_at" IS NOT NULL; + -- same transaction as new, empty table. + -- squawk-ignore require-concurrent-index-creation + CREATE INDEX IF NOT EXISTS "workflow_schedules_created_at_idx" + ON ${quotedSchema}."workflow_schedules" ("namespace_id", "created_at", "id"); + + INSERT INTO ${quotedSchema}."openworkflow_migrations" ("version") + VALUES (7) ON CONFLICT DO NOTHING; + COMMIT;`, ]; } diff --git a/packages/openworkflow/postgres/schedules.test.ts b/packages/openworkflow/postgres/schedules.test.ts new file mode 100644 index 00000000..24342bcb --- /dev/null +++ b/packages/openworkflow/postgres/schedules.test.ts @@ -0,0 +1,84 @@ +import { OpenWorkflow } from "../client/client.js"; +import { BackendPostgres } from "./backend.js"; +import { dropSchema, newPostgresMaxOne } from "./postgres.js"; +import { TEST_POSTGRES_URL } from "./test-backend.testsuite.js"; +import { randomUUID } from "node:crypto"; +import { describe, expect, test } from "vitest"; + +describe("postgres scheduling", () => { + test("dispatch skips a schedule being edited and uses the committed replacement", async () => { + const schema = `schedule_edit_${randomUUID().replaceAll("-", "")}`; + const backend = await BackendPostgres.connect(TEST_POSTGRES_URL, { + schema, + }); + const admin = newPostgresMaxOne(TEST_POSTGRES_URL); + const ow = new OpenWorkflow({ backend }); + const dispatch = { before: new Date(Date.now() + 86_400_000), limit: 100 }; + try { + await ow.scheduleWorkflow( + { name: "report" }, + { value: "old" }, + { cron: "0 * * * *" }, + ); + await admin.begin(async (tx) => { + await tx`UPDATE ${tx(schema)}."workflow_schedules" + SET input=${tx.json({ value: "new" })} WHERE id='report'`; + expect(await backend.enqueueScheduledRuns(dispatch)).toBe(0); + }); + expect(await backend.enqueueScheduledRuns(dispatch)).toBe(1); + const page = await backend.listWorkflowRuns({}); + expect(page.data[0]?.input).toEqual({ value: "new" }); + } finally { + await backend.stop(); + await dropSchema(admin, schema); + await admin.end(); + } + }); + + test("schedule dispatch rolls back run insertion and cursor advancement together", async () => { + const schema = `schedule_atomic_${randomUUID().replaceAll("-", "")}`; + const backend = await BackendPostgres.connect(TEST_POSTGRES_URL, { + schema, + }); + const admin = newPostgresMaxOne(TEST_POSTGRES_URL); + const ow = new OpenWorkflow({ backend }); + try { + const schedules = await Promise.all( + ["a", "b"].map((id) => + ow.scheduleWorkflow({ name: "report" }, null, { + id, + cron: "0 * * * *", + }), + ), + ); + const dispatch = { + before: new Date(Date.now() + 86_400_000), + limit: 100, + }; + await admin.unsafe(`CREATE FUNCTION "${schema}".reject_scheduled_run() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'injected insert failure'; END; $$; + CREATE TRIGGER reject_scheduled_run BEFORE INSERT ON "${schema}".workflow_runs + FOR EACH ROW EXECUTE FUNCTION "${schema}".reject_scheduled_run();`); + await expect(backend.enqueueScheduledRuns(dispatch)).rejects.toThrow( + "injected insert failure", + ); + const rolledBackRuns = await backend.listWorkflowRuns({}); + expect(rolledBackRuns.data).toHaveLength(0); + for (const schedule of schedules) { + const unchanged = await ow.schedules.get(schedule.id); + expect(unchanged?.nextRunAt).toEqual(schedule.nextRunAt); + } + await admin.unsafe( + `DROP TRIGGER reject_scheduled_run ON "${schema}".workflow_runs`, + ); + expect(await backend.enqueueScheduledRuns(dispatch)).toBe(2); + expect(await backend.enqueueScheduledRuns(dispatch)).toBe(0); + const committedRuns = await backend.listWorkflowRuns({}); + expect(committedRuns.data).toHaveLength(2); + } finally { + await backend.stop(); + await dropSchema(admin, schema); + await admin.end(); + } + }); +}); diff --git a/packages/openworkflow/postgres/test-backend.testsuite.ts b/packages/openworkflow/postgres/test-backend.testsuite.ts index 575b1ab1..749c8e61 100644 --- a/packages/openworkflow/postgres/test-backend.testsuite.ts +++ b/packages/openworkflow/postgres/test-backend.testsuite.ts @@ -6,6 +6,9 @@ import { } from "./postgres.js"; import { randomUUID } from "node:crypto"; +export const TEST_POSTGRES_URL = + process.env["OPENWORKFLOW_TEST_POSTGRES_URL"] ?? DEFAULT_POSTGRES_URL; + const SHARED_POOL = Symbol.for("openworkflow.postgres.pool"); /** @@ -29,7 +32,7 @@ function getSharedPool(): Postgres { return sharedPool; } - const nextSharedPool = newPostgres(DEFAULT_POSTGRES_URL); + const nextSharedPool = newPostgres(TEST_POSTGRES_URL); processWithSharedTestPool[SHARED_POOL] = nextSharedPool; return nextSharedPool; diff --git a/packages/openworkflow/postgres/vitest.global-setup.ts b/packages/openworkflow/postgres/vitest.global-setup.ts index 50a85b98..d575615f 100644 --- a/packages/openworkflow/postgres/vitest.global-setup.ts +++ b/packages/openworkflow/postgres/vitest.global-setup.ts @@ -1,9 +1,12 @@ -import { migrate, DEFAULT_SCHEMA, DEFAULT_POSTGRES_URL } from "./postgres.js"; -import { teardownSharedTestPool } from "./test-backend.testsuite.js"; +import { migrate, DEFAULT_SCHEMA } from "./postgres.js"; +import { + teardownSharedTestPool, + TEST_POSTGRES_URL, +} from "./test-backend.testsuite.js"; /** Run database migrations once before Postgres backend tests. */ export async function setup() { - await migrate(DEFAULT_POSTGRES_URL, DEFAULT_SCHEMA); + await migrate(TEST_POSTGRES_URL, DEFAULT_SCHEMA); } /** Close the shared connection pool after all tests complete. */ diff --git a/packages/openworkflow/sqlite/backend.ts b/packages/openworkflow/sqlite/backend.ts index 5a993b38..32f6899f 100644 --- a/packages/openworkflow/sqlite/backend.ts +++ b/packages/openworkflow/sqlite/backend.ts @@ -25,6 +25,9 @@ import { SendSignalResult, GetSignalDeliveryParams, toWorkflowRunCounts, + type EnqueueScheduledRunsParams, + type PaginationOptions, + type WorkflowScheduleIdParams, } from "../core/backend.js"; import { buildPaginatedResponse, @@ -34,6 +37,11 @@ import { import { requireRow, wrapError } from "../core/error.js"; import { JsonValue } from "../core/json.js"; import { prepareWorkflowRerun } from "../core/rerun.js"; +import { + nextScheduleOccurrence, + type WorkflowSchedule, + type WorkflowScheduleDefinition, +} from "../core/schedule.js"; import { StepAttempt } from "../core/step-attempt.js"; import { computeFailedWorkflowRunUpdate } from "../core/workflow-definition.js"; import { @@ -138,6 +146,175 @@ export class BackendSqlite implements Backend { } } + /* oxlint-disable typescript/require-await -- SQLite operations must finish synchronously, while this interface returns rejected promises on failure. */ + async upsertWorkflowSchedule( + params: WorkflowScheduleDefinition, + ): Promise { + const now = new Date(); + // safety: RETURNING selects the schedule columns defined by our migration. + const row = this.db + .prepare( + ` + INSERT INTO workflow_schedules + (namespace_id,id,workflow_name,input,cron,timezone,next_run_at,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?) + ON CONFLICT(namespace_id,id) DO UPDATE SET + workflow_name=excluded.workflow_name, input=excluded.input, + cron=excluded.cron, timezone=excluded.timezone, updated_at=excluded.updated_at, + next_run_at=CASE WHEN workflow_schedules.next_run_at IS NOT NULL + AND (workflow_schedules.cron<>excluded.cron OR workflow_schedules.timezone<>excluded.timezone) + THEN excluded.next_run_at ELSE workflow_schedules.next_run_at END + RETURNING * + `, + ) + .get( + this.namespaceId, + params.id, + params.workflowName, + toJSON(params.input), + params.cron, + params.timezone, + nextScheduleOccurrence(params, now).toISOString(), + now.toISOString(), + now.toISOString(), + ) as WorkflowScheduleRow; + return rowToWorkflowSchedule(row); + } + + async getWorkflowSchedule( + params: WorkflowScheduleIdParams, + ): Promise { + // safety: SELECT reads a persisted schedule row. + const row = this.db + .prepare("SELECT * FROM workflow_schedules WHERE namespace_id=? AND id=?") + .get(this.namespaceId, params.workflowScheduleId) as + WorkflowScheduleRow | undefined; + return row ? rowToWorkflowSchedule(row) : null; + } + + async listWorkflowSchedules( + params: PaginationOptions, + ): Promise> { + const cursor = decodeListCursor(params); + const limit = params.limit ?? DEFAULT_PAGINATION_PAGE_SIZE; + const values: unknown[] = [this.namespaceId]; + let where = "namespace_id=?"; + if (cursor) { + where += ` AND (created_at,id) ${params.before ? "<" : ">"} (?,?)`; + values.push(cursor.createdAt.toISOString(), cursor.id); + } + const order = params.before ? "DESC" : "ASC"; + // safety: SELECT reads schedule rows in cursor order. + const rows = this.db + .prepare( + `SELECT * FROM workflow_schedules WHERE ${where} + ORDER BY created_at ${order},id ${order} LIMIT ?`, + ) + .all(...values, limit + 1) as WorkflowScheduleRow[]; + return buildPaginatedResponse( + rows.map((row) => rowToWorkflowSchedule(row)), + limit, + !!params.after, + !!params.before, + ); + } + + async pauseWorkflowSchedule( + params: WorkflowScheduleIdParams & { paused: boolean }, + ): Promise { + this.db.exec("BEGIN IMMEDIATE"); + try { + // safety: SELECT reads the schedule while holding the writer lock. + let row = this.db + .prepare( + "SELECT * FROM workflow_schedules WHERE namespace_id=? AND id=?", + ) + .get(this.namespaceId, params.workflowScheduleId) as + WorkflowScheduleRow | undefined; + requireRow(row, "change schedule pause state"); + if ((row.next_run_at === null) !== params.paused) { + const now = new Date(); + const next = params.paused ? null : nextScheduleOccurrence(row, now); + // safety: the locked schedule exists and UPDATE returns its columns. + row = this.db + .prepare( + `UPDATE workflow_schedules SET next_run_at=?,updated_at=? + WHERE namespace_id=? AND id=? RETURNING *`, + ) + .get( + toISO(next), + now.toISOString(), + this.namespaceId, + params.workflowScheduleId, + ) as WorkflowScheduleRow; + } + this.db.exec("COMMIT"); + return rowToWorkflowSchedule(row); + } catch (error) { + this.db.exec("ROLLBACK"); + throw error; + } + } + + async deleteWorkflowSchedule( + params: WorkflowScheduleIdParams, + ): Promise { + this.db + .prepare("DELETE FROM workflow_schedules WHERE namespace_id=? AND id=?") + .run(this.namespaceId, params.workflowScheduleId); + } + + async enqueueScheduledRuns( + params: EnqueueScheduledRunsParams, + ): Promise { + const before = params.before.toISOString(); + this.db.exec("BEGIN IMMEDIATE"); + try { + // safety: SELECT reads due schedules while holding the writer lock. + const rows = this.db + .prepare( + `SELECT id,workflow_name,input,cron,timezone FROM workflow_schedules + WHERE namespace_id=? AND next_run_at<=? ORDER BY next_run_at,id LIMIT ?`, + ) + .all(this.namespaceId, before, params.limit) as Pick< + WorkflowScheduleRow, + "id" | "workflow_name" | "input" | "cron" | "timezone" + >[]; + const update = this.db + .prepare(`UPDATE workflow_schedules SET next_run_at=?,updated_at=? + WHERE namespace_id=? AND id=?`); + const insert = this.db.prepare(`INSERT INTO workflow_runs + (namespace_id,id,workflow_name,status,config,input,attempts, + workflow_schedule_id,available_at,created_at,updated_at) + VALUES (?,?,?,'pending','{}',?,0,?,?,?,?)`); + for (const row of rows) { + update.run( + nextScheduleOccurrence(row, params.before).toISOString(), + before, + this.namespaceId, + row.id, + ); + const currentTime = now(); + insert.run( + this.namespaceId, + generateUUID(), + row.workflow_name, + row.input, + row.id, + before, + currentTime, + currentTime, + ); + } + this.db.exec("COMMIT"); + return rows.length; + } catch (error) { + this.db.exec("ROLLBACK"); + throw error; + } + } + /* oxlint-enable typescript/require-await */ + // oxlint-disable-next-line typescript/require-await async stop(): Promise { this.db.close(); @@ -1180,6 +1357,7 @@ export class BackendSqlite implements Backend { // Row types for SQLite results interface WorkflowRunRow extends Record { + workflow_schedule_id: string | null; namespace_id: string; id: string; workflow_name: string; @@ -1264,6 +1442,7 @@ function rowToWorkflowRun(row: WorkflowRunRow): WorkflowRun { namespaceId: row.namespace_id, id: row.id, workflowName: row.workflow_name, + workflowScheduleId: row.workflow_schedule_id, version: row.version, // safety: the status column is written by backend transitions using the domain status values. status: row.status as WorkflowRun["status"], @@ -1319,3 +1498,34 @@ function rowToStepAttempt(row: StepAttemptRow): StepAttempt { updatedAt, }; } + +interface WorkflowScheduleRow extends Record { + namespace_id: string; + id: string; + workflow_name: string; + input: string | null; + cron: string; + timezone: string; + next_run_at: string | null; + created_at: string; + updated_at: string; +} + +/** + * Convert a SQLite row to a schedule. + * @param row - Persisted schedule + * @returns Schedule with parsed JSON and timestamps + */ +function rowToWorkflowSchedule(row: WorkflowScheduleRow): WorkflowSchedule { + return { + namespaceId: row.namespace_id, + id: row.id, + workflowName: row.workflow_name, + input: fromJSON(row.input), + cron: row.cron, + timezone: row.timezone, + nextRunAt: fromISO(row.next_run_at), + createdAt: new Date(row.created_at), + updatedAt: new Date(row.updated_at), + }; +} diff --git a/packages/openworkflow/sqlite/schedules.test.ts b/packages/openworkflow/sqlite/schedules.test.ts new file mode 100644 index 00000000..0cf73f46 --- /dev/null +++ b/packages/openworkflow/sqlite/schedules.test.ts @@ -0,0 +1,43 @@ +import { OpenWorkflow } from "../client/client.js"; +import { BackendSqlite } from "./backend.js"; +import { describe, expect, test } from "vitest"; + +describe("sqlite scheduling", () => { + test("schedule dispatch rolls back the whole batch if inserting a workflow run fails", async () => { + const backend = BackendSqlite.connect(":memory:"); + const ow = new OpenWorkflow({ backend }); + try { + const schedules = await Promise.all( + ["a", "b"].map((id) => + ow.scheduleWorkflow({ name: "report" }, null, { + id, + cron: "0 * * * *", + }), + ), + ); + const dispatch = { + before: new Date(Date.now() + 86_400_000), + limit: 100, + }; + backend["db"].exec(`CREATE TRIGGER reject_scheduled_run + BEFORE INSERT ON workflow_runs WHEN NEW.workflow_schedule_id='b' + BEGIN SELECT RAISE(ABORT, 'injected insert failure'); END;`); + await expect(backend.enqueueScheduledRuns(dispatch)).rejects.toThrow( + "injected insert failure", + ); + const rolledBackRuns = await backend.listWorkflowRuns({}); + expect(rolledBackRuns.data).toHaveLength(0); + const schedulePage = await ow.schedules.list(); + expect(schedulePage.data.map((schedule) => schedule.nextRunAt)).toEqual( + schedules.map((schedule) => schedule.nextRunAt), + ); + backend["db"].exec("DROP TRIGGER reject_scheduled_run"); + expect(await backend.enqueueScheduledRuns(dispatch)).toBe(2); + expect(await backend.enqueueScheduledRuns(dispatch)).toBe(0); + const committedRuns = await backend.listWorkflowRuns({}); + expect(committedRuns.data).toHaveLength(2); + } finally { + await backend.stop(); + } + }); +}); diff --git a/packages/openworkflow/sqlite/sqlite.test.ts b/packages/openworkflow/sqlite/sqlite.test.ts index 8cc4e42f..0e92196c 100644 --- a/packages/openworkflow/sqlite/sqlite.test.ts +++ b/packages/openworkflow/sqlite/sqlite.test.ts @@ -205,7 +205,7 @@ describe("sqlite", () => { db.exec("ALTER TABLE step_attempts DROP COLUMN step_index"); migrate(db); - expect(getMigrationVersion(db)).toBe(6); + expect(getMigrationVersion(db)).toBe(migrations().length - 1); }); test.each(["pending", "current"])( @@ -264,7 +264,7 @@ describe("sqlite", () => { : setTimeout(100)); db.exec("COMMIT"); await Promise.all(workers.map((worker) => worker.completed)); - expect(getMigrationVersion(db)).toBe(6); + expect(getMigrationVersion(db)).toBe(migrations().length - 1); } finally { try { db.exec("ROLLBACK"); diff --git a/packages/openworkflow/sqlite/sqlite.ts b/packages/openworkflow/sqlite/sqlite.ts index 27767217..33d5bfaa 100644 --- a/packages/openworkflow/sqlite/sqlite.ts +++ b/packages/openworkflow/sqlite/sqlite.ts @@ -53,6 +53,7 @@ export function newDatabase(path: string): Database { }; db = new NodeDatabase(path); } + db.exec("PRAGMA busy_timeout = 5000;"); // Only enable WAL mode for file-based databases if (path !== ":memory:") { db.exec("PRAGMA journal_mode = WAL;"); @@ -216,6 +217,27 @@ export function migrations(): string[] { INSERT OR IGNORE INTO "openworkflow_migrations" ("version") VALUES (6);`, + + // 7 - recurring schedules + `ALTER TABLE "workflow_runs" ADD COLUMN "workflow_schedule_id" TEXT; + CREATE TABLE "workflow_schedules" ( + "namespace_id" TEXT NOT NULL, + "id" TEXT NOT NULL, + -- + "workflow_name" TEXT NOT NULL, + "input" TEXT, + "cron" TEXT NOT NULL, + "timezone" TEXT NOT NULL, + "next_run_at" TEXT, + "created_at" TEXT NOT NULL, + "updated_at" TEXT NOT NULL, + PRIMARY KEY ("namespace_id", "id") + ); + CREATE INDEX "workflow_schedules_due_idx" + ON "workflow_schedules" ("namespace_id", "next_run_at", "id") WHERE "next_run_at" IS NOT NULL; + CREATE INDEX "workflow_schedules_created_at_idx" + ON "workflow_schedules" ("namespace_id", "created_at", "id"); + INSERT INTO "openworkflow_migrations" ("version") VALUES (7);`, ]; } diff --git a/packages/openworkflow/telemetry-optional.test.ts b/packages/openworkflow/telemetry-optional.test.ts index 1d227eee..c2f705b0 100644 --- a/packages/openworkflow/telemetry-optional.test.ts +++ b/packages/openworkflow/telemetry-optional.test.ts @@ -1,5 +1,5 @@ import { execFile } from "node:child_process"; -import { cp, mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { cp, mkdir, mkdtemp, rm, symlink, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import { promisify } from "node:util"; @@ -22,6 +22,11 @@ describe("optional OpenTelemetry API", () => { ); // Prevent Bun from auto-installing the optional peer in this fixture. await mkdir(path.join(directory, "node_modules")); + await symlink( + new URL(".", import.meta.resolve("cron-parser/package.json")), + path.join(directory, "node_modules", "cron-parser"), + "junction", + ); await expect( execFileAsync( process.execPath, diff --git a/packages/openworkflow/testing/backend-stub.testsuite.ts b/packages/openworkflow/testing/backend-stub.testsuite.ts index e3b3157b..d3b9e18d 100644 --- a/packages/openworkflow/testing/backend-stub.testsuite.ts +++ b/packages/openworkflow/testing/backend-stub.testsuite.ts @@ -7,6 +7,12 @@ import type { Backend } from "../core/backend.js"; */ export function createStubBackend(overrides: Partial): Backend { return { + upsertWorkflowSchedule: unexpectedBackendCall, + getWorkflowSchedule: unexpectedBackendCall, + listWorkflowSchedules: unexpectedBackendCall, + pauseWorkflowSchedule: unexpectedBackendCall, + deleteWorkflowSchedule: unexpectedBackendCall, + enqueueScheduledRuns: unexpectedBackendCall, createWorkflowRun: unexpectedBackendCall, rerunWorkflowRun: unexpectedBackendCall, getWorkflowRun: unexpectedBackendCall, diff --git a/packages/openworkflow/testing/backend.testsuite.ts b/packages/openworkflow/testing/backend.testsuite.ts index 0e62acc9..9f97dc15 100644 --- a/packages/openworkflow/testing/backend.testsuite.ts +++ b/packages/openworkflow/testing/backend.testsuite.ts @@ -6,6 +6,7 @@ import { import type { StepAttempt } from "../core/step-attempt.js"; import { DEFAULT_WORKFLOW_RETRY_POLICY } from "../core/workflow-definition.js"; import type { WorkflowRun } from "../core/workflow-run.js"; +import { testWorkflowSchedules } from "./schedules.testsuite.js"; import { randomUUID } from "node:crypto"; import { afterAll, beforeAll, describe, expect, test, vi } from "vitest"; @@ -28,6 +29,7 @@ export interface TestBackendOptions { * @param options - Test suite options */ export function testBackend(options: TestBackendOptions): void { + testWorkflowSchedules(options); const { setup, teardown } = options; const RESCHEDULING_RETRY_POLICY = { ...DEFAULT_WORKFLOW_RETRY_POLICY, @@ -48,6 +50,7 @@ export function testBackend(options: TestBackendOptions): void { describe("createWorkflowRun()", () => { test("creates a workflow run", async () => { const expected: WorkflowRun = { + workflowScheduleId: null, namespaceId: "", // - id: "", // - workflowName: randomUUID(), diff --git a/packages/openworkflow/testing/schedules.testsuite.ts b/packages/openworkflow/testing/schedules.testsuite.ts new file mode 100644 index 00000000..d0f739c3 --- /dev/null +++ b/packages/openworkflow/testing/schedules.testsuite.ts @@ -0,0 +1,182 @@ +import { OpenWorkflow } from "../client/client.js"; +import type { Backend } from "../core/backend.js"; +import { defineWorkflowSpec } from "../core/workflow-definition.js"; +import type { TestBackendOptions } from "./backend.testsuite.js"; +import { afterEach, beforeEach, describe, expect, test } from "vitest"; + +/** + * Exercise identical schedule contracts on every backend. + * @param options - Isolated backend factory and cleanup + */ +export function testWorkflowSchedules(options: TestBackendOptions): void { + describe("recurring schedule persistence", () => { + let backend: Backend; + let ow: OpenWorkflow; + const spec = defineWorkflowSpec<{ accountId: string }>({ name: "report" }); + const input = { accountId: "acme" }; + const config = { cron: "0 * * * *" }; + + beforeEach(async () => { + backend = await options.setup(); + ow = new OpenWorkflow({ backend }); + }); + afterEach(async () => { + await options.teardown(backend); + }); + + test("last-write-wins uses the current input and preserves dispatch progress", async () => { + const first = await ow.scheduleWorkflow(spec, input, config); + const replaced = await ow.scheduleWorkflow( + spec, + { accountId: "other" }, + config, + ); + expect(replaced.nextRunAt).toEqual(first.nextRunAt); + const before = new Date(Date.now() + 86_400_000); + expect(await backend.enqueueScheduledRuns({ before, limit: 100 })).toBe( + 1, + ); + const advanced = await ow.schedules.get(first.id); + const repeated = await ow.scheduleWorkflow(spec, input, config); + expect(repeated.nextRunAt).toEqual(advanced?.nextRunAt); + expect(repeated.createdAt).toEqual(first.createdAt); + const runPage = await backend.listWorkflowRuns({}); + expect(runPage.data[0]).toMatchObject({ + workflowScheduleId: first.id, + version: null, + input: { accountId: "other" }, + availableAt: before, + }); + }); + + test("repeated or competing dispatches insert exactly one run", async () => { + const schedule = await ow.scheduleWorkflow(spec, input, config); + const before = new Date(Date.now() + 86_400_000); + const results = await Promise.all( + Array.from({ length: 8 }, () => + backend.enqueueScheduledRuns({ before, limit: 100 }), + ), + ); + expect(results.reduce((sum, count) => sum + count, 0)).toBe(1); + expect(await backend.enqueueScheduledRuns({ before, limit: 100 })).toBe( + 0, + ); + const runPage = await backend.listWorkflowRuns({}); + expect(runPage.data).toHaveLength(1); + const advanced = await ow.schedules.get(schedule.id); + expect(advanced?.nextRunAt?.getTime()).toBeGreaterThan(before.getTime()); + }); + + test("pause clears the next time, registration preserves it, and resume starts in the future", async () => { + const schedule = await ow.scheduleWorkflow(spec, input, config); + const paused = await ow.schedules.pause(schedule.id); + expect(paused.nextRunAt).toBeNull(); + expect(await ow.schedules.pause(schedule.id)).toEqual(paused); + const replaced = await ow.scheduleWorkflow(spec, input, { + cron: "15 * * * *", + }); + expect(replaced.nextRunAt).toBeNull(); + expect( + await backend.enqueueScheduledRuns({ + before: new Date(Date.now() + 86_400_000), + limit: 100, + }), + ).toBe(0); + const now = Date.now(); + const resumed = await ow.schedules.resume(schedule.id); + expect(resumed.nextRunAt?.getTime()).toBeGreaterThan(now); + expect(resumed.nextRunAt?.getUTCMinutes()).toBe(15); + expect(await ow.schedules.resume(schedule.id)).toEqual(resumed); + }); + + test("deletion stops dispatch and preserves run attribution", async () => { + const schedule = await ow.scheduleWorkflow(spec, input, config); + const before = new Date(Date.now() + 86_400_000); + await backend.enqueueScheduledRuns({ before, limit: 100 }); + await ow.schedules.delete(schedule.id); + await ow.schedules.delete(schedule.id); + expect(await ow.schedules.get(schedule.id)).toBeNull(); + expect(await backend.enqueueScheduledRuns({ before, limit: 100 })).toBe( + 0, + ); + const runPage = await backend.listWorkflowRuns({}); + expect(runPage.data[0]?.workflowScheduleId).toBe(schedule.id); + const schedulePage = await ow.schedules.list(); + expect(schedulePage.data).toEqual([]); + }); + + test("cron and timezone changes reset active schedules to the next future time", async () => { + await ow.scheduleWorkflow(spec, input, config); + await backend.enqueueScheduledRuns({ + before: new Date(Date.now() + 86_400_000), + limit: 100, + }); + const changed = await ow.scheduleWorkflow(spec, input, { + cron: "15 * * * *", + }); + expect(changed.nextRunAt?.getUTCMinutes()).toBe(15); + const timezone = await ow.scheduleWorkflow(spec, input, { + cron: "0 9 * * *", + timezone: "Asia/Kolkata", + }); + expect(timezone.nextRunAt?.getUTCHours()).toBe(3); + expect(timezone.nextRunAt?.getUTCMinutes()).toBe(30); + }); + + test("dispatch is bounded and collapses years of missed occurrences into one run per schedule", async () => { + for (const id of ["a", "b", "c"]) + await ow.scheduleWorkflow(spec, input, { id, cron: "* * * * * *" }); + const before = new Date(Date.now() + 5 * 365 * 86_400_000); + expect(await backend.enqueueScheduledRuns({ before, limit: 2 })).toBe(2); + expect(await backend.enqueueScheduledRuns({ before, limit: 2 })).toBe(1); + expect(await backend.enqueueScheduledRuns({ before, limit: 2 })).toBe(0); + const runPage = await backend.listWorkflowRuns({}); + expect( + runPage.data + .map((run) => run.workflowScheduleId) + .toSorted((a, b) => (a ?? "").localeCompare(b ?? "")), + ).toEqual(["a", "b", "c"]); + const schedulePage = await ow.schedules.list(); + for (const schedule of schedulePage.data) + expect(schedule.nextRunAt?.getTime()).toBeGreaterThan(before.getTime()); + }); + + test("list pagination and dispatch stay within a namespace", async () => { + for (const id of ["a", "b", "c"]) + await ow.scheduleWorkflow(spec, input, { ...config, id }); + const first = await ow.schedules.list({ limit: 2 }); + expect(first.data.map((item) => item.id)).toEqual(["a", "b"]); + const second = await ow.schedules.list({ + limit: 2, + after: first.pagination.next ?? "", + }); + expect(second.data.map((item) => item.id)).toEqual(["c"]); + const previous = await ow.schedules.list({ + limit: 2, + before: second.pagination.prev ?? "", + }); + expect(previous.data.map((item) => item.id)).toEqual(["a", "b"]); + const other = await options.setup(); + try { + expect( + await other.getWorkflowSchedule({ workflowScheduleId: "a" }), + ).toBeNull(); + const otherPage = await other.listWorkflowSchedules({}); + expect(otherPage.data).toEqual([]); + expect( + await other.enqueueScheduledRuns({ + before: new Date(Date.now() + 86_400_000), + limit: 100, + }), + ).toBe(0); + } finally { + await options.teardown(other); + } + }); + + test("pausing or resuming a missing schedule fails", async () => { + await expect(ow.schedules.pause("missing")).rejects.toThrow(); + await expect(ow.schedules.resume("missing")).rejects.toThrow(); + }); + }); +} diff --git a/packages/openworkflow/worker/execution.test.ts b/packages/openworkflow/worker/execution.test.ts index b1f0c8ff..ce9ff348 100644 --- a/packages/openworkflow/worker/execution.test.ts +++ b/packages/openworkflow/worker/execution.test.ts @@ -4289,6 +4289,7 @@ function createMockWorkflowRun( overrides: Partial = {}, ): WorkflowRun { return { + workflowScheduleId: null, namespaceId: "default", id: "workflow-run-id", workflowName: "workflow-name", diff --git a/packages/openworkflow/worker/schedules.test.ts b/packages/openworkflow/worker/schedules.test.ts new file mode 100644 index 00000000..d818bc67 --- /dev/null +++ b/packages/openworkflow/worker/schedules.test.ts @@ -0,0 +1,175 @@ +import { OpenWorkflow } from "../client/client.js"; +import type { WorkflowScheduleDefinition } from "../core/schedule.js"; +import { BackendSqlite } from "../sqlite/backend.js"; +import { createStubBackend } from "../testing/backend-stub.testsuite.js"; +import { Worker } from "./worker.js"; +import { describe, expect, test, vi } from "vitest"; + +describe("worker scheduling", () => { + test("scheduling proceeds while every execution slot is occupied", async () => { + const backend = BackendSqlite.connect(":memory:"); + const ow = new OpenWorkflow({ backend }); + const block = Promise.withResolvers(); + let started = false; + let scheduledExecuted = false; + const blocker = ow.defineWorkflow({ name: "blocker" }, async () => { + started = true; + await block.promise; + }); + const report = ow.defineWorkflow({ name: "report" }, () => { + scheduledExecuted = true; + }); + const worker = ow.newWorker({ concurrency: 1 }); + try { + await report.schedule(undefined, { cron: "* * * * * *" }); + await blocker.run(); + await worker.start(); + await vi.waitFor(() => { + expect(started).toBe(true); + }); + backend["db"] + .prepare("UPDATE workflow_schedules SET next_run_at=?") + .run(new Date(Date.now() - 5000).toISOString()); + await vi.waitFor( + async () => { + const reportPage = await backend.listWorkflowRuns({ + workflowName: "report", + }); + expect(reportPage.data.length).toBeGreaterThan(0); + }, + { timeout: 2500 }, + ); + expect(scheduledExecuted).toBe(false); + } finally { + block.resolve(); + await worker.stop(); + await backend.stop(); + } + }); + + test("duplicate declarations preserve overdue runs on restart", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-01-02T12:00:00Z")); + const backend = BackendSqlite.connect(":memory:"); + const ow = new OpenWorkflow({ backend }); + ow.defineWorkflow( + { + name: "earlier", + schedules: [{ id: "report", cron: "0 * * * *" }], + }, + () => null, + ); + ow.defineWorkflow( + { + name: "report", + schedules: [{ cron: "0 12 * * *" }, { cron: "0 0 * * *" }], + }, + () => null, + ); + const worker = ow.newWorker(); + try { + await worker.start(); + await worker.stop(); + backend["db"] + .prepare("UPDATE workflow_schedules SET next_run_at=? WHERE id=?") + .run("2026-01-02T00:00:00.000Z", "report"); + await worker.start(); + await vi.waitFor(async () => { + const page = await backend.listWorkflowRuns({}); + expect(page.data).toHaveLength(1); + expect(page.data[0]).toMatchObject({ + workflowName: "report", + workflowScheduleId: "report", + }); + }); + const schedule = await ow.schedules.get("report"); + expect(schedule?.nextRunAt).toEqual(new Date("2026-01-03T00:00:00Z")); + } finally { + await worker.stop(); + await backend.stop(); + vi.useRealTimers(); + } + }); + + test.each(["success", "failure"])( + "stop waits for registration %s and prevents background loops from starting afterward", + async (outcome) => { + const pending = Promise.withResolvers(); + const register = vi.fn(async (definition: WorkflowScheduleDefinition) => { + await pending.promise; + return { + ...definition, + namespaceId: "test", + nextRunAt: new Date(), + createdAt: new Date(), + updatedAt: new Date(), + }; + }); + const enqueue = vi.fn(() => Promise.resolve(0)); + const backend = createStubBackend({ + upsertWorkflowSchedule: register, + enqueueScheduledRuns: enqueue, + }); + const worker = new Worker({ + backend, + workflows: [ + { + spec: { name: "report", schedules: [{ cron: "0 * * * *" }] }, + fn: () => null, + }, + ], + }); + const first = worker.start(); + expect(worker.start()).toBe(first); + await vi.waitFor(() => { + expect(register).toHaveBeenCalledTimes(1); + }); + const stop = worker.stop(); + if (outcome === "success") { + pending.resolve(); + await first; + await stop; + } else { + pending.reject(new Error("registration failed")); + await expect(first).rejects.toThrow("registration failed"); + await expect(stop).rejects.toThrow("registration failed"); + } + expect(enqueue).not.toHaveBeenCalled(); + }, + ); + + test("schedule polling recovers after database and logger errors", async () => { + const error = new Error("database busy"); + const enqueue = vi.fn().mockRejectedValueOnce(error).mockResolvedValue(0); + const logger = { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(() => { + throw new Error("logger failed"); + }), + }; + const backend = createStubBackend({ + enqueueScheduledRuns: enqueue, + claimWorkflowRun: () => Promise.resolve(null), + }); + const worker = new Worker({ backend, workflows: [], logger }); + try { + await worker.start(); + await vi.waitFor( + () => { + expect(enqueue).toHaveBeenCalledTimes(2); + }, + { + timeout: 2500, + }, + ); + expect(logger.error).toHaveBeenCalledWith( + "Worker schedule tick failed:", + error, + ); + } finally { + await worker.stop(); + } + }); +}); diff --git a/packages/openworkflow/worker/worker.test.ts b/packages/openworkflow/worker/worker.test.ts index 7221bc7c..5606df95 100644 --- a/packages/openworkflow/worker/worker.test.ts +++ b/packages/openworkflow/worker/worker.test.ts @@ -1635,6 +1635,7 @@ describe("Worker", () => { const worker = new Worker({ backend: createStubBackend({ claimWorkflowRun, + enqueueScheduledRuns: () => Promise.resolve(0), }), workflows: [], }); @@ -1672,6 +1673,7 @@ describe("Worker", () => { const worker = new Worker({ backend: createStubBackend({ claimWorkflowRun, + enqueueScheduledRuns: () => Promise.resolve(0), }), workflows: [], }); @@ -1708,6 +1710,7 @@ describe("Worker", () => { const worker = new Worker({ backend: createStubBackend({ claimWorkflowRun: vi.fn().mockRejectedValue(error), + enqueueScheduledRuns: () => Promise.resolve(0), }), workflows: [], }); @@ -1732,6 +1735,7 @@ describe("Worker", () => { const worker = new Worker({ backend: createStubBackend({ claimWorkflowRun: vi.fn().mockRejectedValue(error), + enqueueScheduledRuns: () => Promise.resolve(0), }), workflows: [], logger, @@ -1822,7 +1826,10 @@ describe("Worker", () => { throw new Error("logger boom"); }); const worker = new Worker({ - backend: createStubBackend({ claimWorkflowRun }), + backend: createStubBackend({ + claimWorkflowRun, + enqueueScheduledRuns: () => Promise.resolve(0), + }), workflows: [], logger, }); @@ -1845,6 +1852,7 @@ describe("Worker", () => { const error = new Error("boom"); const backend = createStubBackend({ claimWorkflowRun: vi.fn().mockRejectedValue(error), + enqueueScheduledRuns: () => Promise.resolve(0), }); const clientLogger = createTestLogger(); const workerLogger = createTestLogger(); diff --git a/packages/openworkflow/worker/worker.ts b/packages/openworkflow/worker/worker.ts index a72d272d..d01e1b74 100644 --- a/packages/openworkflow/worker/worker.ts +++ b/packages/openworkflow/worker/worker.ts @@ -2,6 +2,10 @@ import type { Backend } from "../core/backend.js"; import { type BackoffPolicy, computeBackoffDelayMs } from "../core/backoff.js"; import { parseDuration } from "../core/duration.js"; import { type Logger, consoleLogger } from "../core/logger.js"; +import { + prepareWorkflowSchedule, + type WorkflowScheduleDefinition, +} from "../core/schedule.js"; import type { RetryPolicy, Workflow } from "../core/workflow-definition.js"; import { DEFAULT_WORKFLOW_RETRY_POLICY } from "../core/workflow-definition.js"; import { WorkflowRegistry } from "../core/workflow-registry.js"; @@ -65,6 +69,8 @@ export class Worker { private readonly registry = new WorkflowRegistry(); private readonly activeExecutions = new Set(); private running = false; + private startPromise: Promise | null = null; + private stopPromise: Promise | null = null; private loopPromise: Promise | null = null; private backoffAttempts = 0; @@ -91,13 +97,41 @@ export class Worker { * Start the worker. It will begin polling for and executing workflows. * @returns Promise resolved when started */ - async start(): Promise { - if (this.running) return; + start(): Promise { + if (this.stopPromise) return this.stopPromise.then(() => this.start()); + this.startPromise ??= this.startWorker(); + return this.startPromise; + } + + private async startWorker(): Promise { this.running = true; this.backoffAttempts = 0; - // Background polls should not accumulate in the trace that started the worker. - this.loopPromise = withoutActiveSpan(() => this.runLoop()); - await Promise.resolve(); + try { + const schedules = new Map(); + for (const workflow of this.registry.getAll()) { + for (const declaration of workflow.spec.schedules ?? []) { + const schedule = await prepareWorkflowSchedule( + workflow.spec, + declaration.input, + declaration, + ); + schedules.set(schedule.id, schedule); + } + } + for (const schedule of schedules.values()) { + await this.backend.upsertWorkflowSchedule(schedule); + } + // oxlint-disable-next-line typescript/no-unnecessary-condition -- stop() can cancel startup while declarations are being persisted. + if (!this.running) return; + // Background polls should not accumulate in the trace that started the worker. + this.loopPromise = withoutActiveSpan(async () => { + await Promise.all([this.runLoop(), this.runScheduleLoop()]); + }); + } catch (error) { + this.running = false; + this.startPromise = null; + throw error; + } } /** @@ -105,14 +139,40 @@ export class Worker { * before returning. * @returns Promise resolved when stopped */ - async stop(): Promise { - this.running = false; + stop(): Promise { + this.stopPromise ??= this.stopWorker(); + return this.stopPromise; + } - // wait for the poll loop to stop - if (this.loopPromise) await this.loopPromise; + private async stopWorker(): Promise { + this.running = false; + try { + await this.startPromise; + } finally { + await this.loopPromise; + while (this.activeExecutions.size > 0) await sleep(100); + this.startPromise = null; + this.stopPromise = null; + } + } - // wait for all active executions to finish - while (this.activeExecutions.size > 0) await sleep(100); + private async runScheduleLoop(): Promise { + const batchSize = 100; + while (this.running) { + let hasMore = false; + try { + const inserted = await this.backend.enqueueScheduledRuns({ + before: new Date(), + limit: batchSize, + }); + hasMore = inserted === batchSize; + } catch (error) { + logError(this.logger, "Worker schedule tick failed:", error); + } + // Yield between full batches so scheduling does not monopolize the event loop. + // oxlint-disable-next-line typescript/no-unnecessary-condition -- stop() can change running during the awaited tick. + if (this.running) await sleep(hasMore ? 0 : 1000); + } } /** From 22435dc13e56066396c333211f8d46a96be4d04e Mon Sep 17 00:00:00 2001 From: James Martinez Date: Mon, 12 Oct 2026 13:14:31 -0500 Subject: [PATCH 2/2] chore: add schedule example --- .../workflow-discovery/openworkflow/greeting.ts | 14 ++++++++++++-- packages/openworkflow/index.ts | 4 ++++ packages/openworkflow/internal.ts | 1 + 3 files changed, 17 insertions(+), 2 deletions(-) diff --git a/examples/workflow-discovery/openworkflow/greeting.ts b/examples/workflow-discovery/openworkflow/greeting.ts index fdef79b6..7d0a3938 100644 --- a/examples/workflow-discovery/openworkflow/greeting.ts +++ b/examples/workflow-discovery/openworkflow/greeting.ts @@ -10,14 +10,24 @@ export interface GreetingOutput { // A workflow with a named export (greetingWorkflow) export const greetingWorkflow = defineWorkflow( - { name: "greeting", version: "1.0.0" }, + { + name: "greeting", + schedules: [ + { + cron: "*/10 * * * * *", // every 10s while worker is running + input: { name: "scheduled visitor" }, + }, + ], + }, async ({ input, step }) => { const greeting = await step.run({ name: "generate-greeting" }, () => { return `Hello, ${input.name}!`; }); const message = await step.run({ name: "format-message" }, () => { - return `${greeting} Welcome to OpenWorkflow.`; + const formatted = `${greeting} Welcome to OpenWorkflow.`; + console.log(formatted); // this will log every 10s + return formatted; }); return { message }; diff --git a/packages/openworkflow/index.ts b/packages/openworkflow/index.ts index 595274c2..e9f41887 100644 --- a/packages/openworkflow/index.ts +++ b/packages/openworkflow/index.ts @@ -4,6 +4,10 @@ export { OpenWorkflow } from "./client/client.js"; // core export type { Logger } from "./core/logger.js"; +export type { + WorkflowScheduleOptions, + WorkflowSchedule, +} from "./core/schedule.js"; export type { RetryPolicy, Workflow } from "./core/workflow-definition.js"; export type { WorkflowRunMetadata } from "./core/workflow-function.js"; export { diff --git a/packages/openworkflow/internal.ts b/packages/openworkflow/internal.ts index b1f3ddc7..ab0188cc 100644 --- a/packages/openworkflow/internal.ts +++ b/packages/openworkflow/internal.ts @@ -6,6 +6,7 @@ export { isWorkflow } from "./core/workflow-definition.js"; export * from "./core/backend.js"; // core +export type { WorkflowScheduleDefinition } from "./core/schedule.js"; export type { WorkflowRun, WorkflowRunStatus } from "./core/workflow-run.js"; export type { StepAttempt,