diff --git a/.server-changes/schedule-windows.md b/.server-changes/schedule-windows.md new file mode 100644 index 00000000000..ecdd209c1c2 --- /dev/null +++ b/.server-changes/schedule-windows.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: feature +--- + +Add backend support for delaying cron schedules within a specified window with a minimum of 60 seconds. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 5bd5b5f4978..3ce98dd521f 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -1684,6 +1684,11 @@ const EnvironmentSchema = z SCHEDULE_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50), SCHEDULE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(30_000), SCHEDULE_WORKER_DISTRIBUTION_WINDOW_SECONDS: z.coerce.number().int().default(30), + SCHEDULE_WORKER_CRON_SPREAD_FRACTION: z.coerce + .number() + .catch(0) + .default(0) + .transform((value) => (Number.isFinite(value) ? Math.min(1, Math.max(0, value)) : 0)), SCHEDULE_WORKER_REDIS_HOST: z .string() diff --git a/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts b/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts index ab394b76ec1..22b9821bab9 100644 --- a/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts @@ -5,6 +5,7 @@ import { getTaskIdentifiers } from "~/models/task.server"; import { getCurrentPlan, getPlans } from "~/services/platform.v3.server"; import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server"; import { ServiceValidationError } from "~/v3/services/baseService.server"; +import { formatScheduleWindow } from "~/v3/scheduleWindow.server"; import { CheckScheduleService } from "~/v3/services/checkSchedule.server"; import { calculateNextScheduledTimestampFromNow, @@ -31,6 +32,7 @@ export type ScheduleListItem = { cron: string; cronDescription: string; timezone: string; + window?: string; externalId: string | null; nextRun: Date; lastRun: Date | undefined; @@ -215,6 +217,8 @@ export class ScheduleListPresenter extends BasePresenter { generatorExpression: true, generatorDescription: true, timezone: true, + windowDurationSeconds: true, + windowPercentage: true, externalId: true, instances: { select: { @@ -306,6 +310,7 @@ export class ScheduleListPresenter extends BasePresenter { cron: schedule.generatorExpression, cronDescription: schedule.generatorDescription, timezone: schedule.timezone, + window: formatScheduleWindow(schedule), active: schedule.active, externalId: schedule.externalId, lastRun, diff --git a/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts b/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts index 318b6da4921..bc7d0388b0e 100644 --- a/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts @@ -6,12 +6,14 @@ import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstan import { nextScheduledTimestamps } from "~/v3/utils/calculateNextSchedule.server"; import { NextRunListPresenter } from "./NextRunListPresenter.server"; import { scheduleWhereClause } from "~/models/schedules.server"; +import { formatScheduleWindow } from "~/v3/scheduleWindow.server"; type ViewScheduleOptions = { userId?: string; projectId: string; friendlyId: string; environmentId: string; + includeRunHistory?: boolean; }; export class ViewSchedulePresenter { @@ -21,7 +23,13 @@ export class ViewSchedulePresenter { this.#prismaClient = prismaClient; } - public async call({ userId, projectId, friendlyId, environmentId }: ViewScheduleOptions) { + public async call({ + userId, + projectId, + friendlyId, + environmentId, + includeRunHistory = true, + }: ViewScheduleOptions) { const schedule = await this.#prismaClient.taskSchedule.findFirst({ select: { id: true, @@ -30,6 +38,8 @@ export class ViewSchedulePresenter { generatorExpression: true, generatorDescription: true, timezone: true, + windowDurationSeconds: true, + windowPercentage: true, externalId: true, deduplicationKey: true, userProvidedDeduplicationKey: true, @@ -76,17 +86,14 @@ export class ViewSchedulePresenter { ? nextScheduledTimestamps(schedule.generatorExpression, schedule.timezone, new Date(), 5) : []; - const clickhouse = await clickhouseFactory.getClickhouseForOrganization( - schedule.project.organizationId, - "standard" - ); - const runPresenter = new NextRunListPresenter(this.#prismaClient, clickhouse); - const { runs } = await runPresenter.call(schedule.project.organizationId, environmentId, { - projectId: schedule.project.id, - scheduleId: schedule.id, - pageSize: 5, - period: "31d", - }); + const runs = includeRunHistory + ? await this.#getRunHistory({ + organizationId: schedule.project.organizationId, + environmentId, + projectId: schedule.project.id, + scheduleId: schedule.id, + }) + : []; return { schedule: { @@ -107,6 +114,32 @@ export class ViewSchedulePresenter { }; } + async #getRunHistory({ + organizationId, + environmentId, + projectId, + scheduleId, + }: { + organizationId: string; + environmentId: string; + projectId: string; + scheduleId: string; + }) { + const clickhouse = await clickhouseFactory.getClickhouseForOrganization( + organizationId, + "standard" + ); + const runPresenter = new NextRunListPresenter(this.#prismaClient, clickhouse); + const { runs } = await runPresenter.call(organizationId, environmentId, { + projectId, + scheduleId, + pageSize: 5, + period: "31d", + }); + + return runs; + } + public toJSONResponse(result: NonNullable>>) { const response: ScheduleObject = { id: result.schedule.friendlyId, @@ -120,6 +153,7 @@ export class ViewSchedulePresenter { description: result.schedule.cronDescription, }, timezone: result.schedule.timezone, + window: formatScheduleWindow(result.schedule), externalId: result.schedule.externalId ?? undefined, deduplicationKey: result.schedule.userProvidedDeduplicationKey ? (result.schedule.deduplicationKey ?? undefined) diff --git a/apps/webapp/app/routes/api.v1.schedules.$scheduleId.activate.ts b/apps/webapp/app/routes/api.v1.schedules.$scheduleId.activate.ts index 99ca3159954..2a7bc70b25d 100644 --- a/apps/webapp/app/routes/api.v1.schedules.$scheduleId.activate.ts +++ b/apps/webapp/app/routes/api.v1.schedules.$scheduleId.activate.ts @@ -64,6 +64,7 @@ export async function action({ request, params }: ActionFunctionArgs) { projectId: authenticationResult.environment.projectId, friendlyId: parsedParams.data.scheduleId, environmentId: authenticationResult.environment.id, + includeRunHistory: false, }); if (!result) { diff --git a/apps/webapp/app/routes/api.v1.schedules.$scheduleId.deactivate.ts b/apps/webapp/app/routes/api.v1.schedules.$scheduleId.deactivate.ts index 3c9514ef8e3..af223022057 100644 --- a/apps/webapp/app/routes/api.v1.schedules.$scheduleId.deactivate.ts +++ b/apps/webapp/app/routes/api.v1.schedules.$scheduleId.deactivate.ts @@ -64,6 +64,7 @@ export async function action({ request, params }: ActionFunctionArgs) { projectId: authenticationResult.environment.projectId, friendlyId: parsedParams.data.scheduleId, environmentId: authenticationResult.environment.id, + includeRunHistory: false, }); if (!result) { diff --git a/apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts b/apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts index 4f7e8d8c164..f98707eecbb 100644 --- a/apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts +++ b/apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts @@ -107,6 +107,7 @@ export async function action({ request, params }: ActionFunctionArgs) { taskIdentifier: body.data.task, cron: body.data.cron, timezone: body.data.timezone, + window: body.data.window, environments: [authenticationResult.environment.id], externalId: body.data.externalId, }; @@ -124,6 +125,7 @@ export async function action({ request, params }: ActionFunctionArgs) { description: schedule.cronDescription, }, timezone: schedule.timezone, + window: schedule.window, externalId: schedule.externalId ?? undefined, deduplicationKey: schedule.deduplicationKey, environments: schedule.environments, @@ -176,6 +178,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) { projectId: authenticationResult.environment.projectId, friendlyId: parsedParams.data.scheduleId, environmentId: authenticationResult.environment.id, + includeRunHistory: false, }); if (!result) { diff --git a/apps/webapp/app/routes/api.v1.schedules.ts b/apps/webapp/app/routes/api.v1.schedules.ts index b5fd2fd968c..277033dd942 100644 --- a/apps/webapp/app/routes/api.v1.schedules.ts +++ b/apps/webapp/app/routes/api.v1.schedules.ts @@ -51,6 +51,7 @@ export async function action({ request }: ActionFunctionArgs) { externalId: body.data.externalId, deduplicationKey: body.data.deduplicationKey, timezone: body.data.timezone, + window: body.data.window, }; const schedule = await service.call(authenticationResult.environment.projectId, options); @@ -66,6 +67,7 @@ export async function action({ request }: ActionFunctionArgs) { description: schedule.cronDescription, }, timezone: schedule.timezone, + window: schedule.window, externalId: schedule.externalId ?? undefined, deduplicationKey: schedule.deduplicationKey, environments: schedule.environments, @@ -121,6 +123,7 @@ export async function loader({ request }: LoaderFunctionArgs) { description: schedule.cronDescription, }, timezone: schedule.timezone, + window: schedule.window, deduplicationKey: schedule.userProvidedDeduplicationKey ? schedule.deduplicationKey : undefined, diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts index 4ddeb2af171..5604c107a68 100644 --- a/apps/webapp/app/services/runsReplicationService.server.ts +++ b/apps/webapp/app/services/runsReplicationService.server.ts @@ -1307,6 +1307,7 @@ export class RunsReplicationService { run.id, // run_id run.updatedAt.getTime(), // updated_at run.createdAt.getTime(), // created_at + run.queueTimestamp?.getTime() ?? null, // queue_timestamp run.status, // status environmentType, // environment_type run.friendlyId, // friendly_id diff --git a/apps/webapp/app/v3/scheduleEngine.server.ts b/apps/webapp/app/v3/scheduleEngine.server.ts index 68f78af376e..9939c0c26ca 100644 --- a/apps/webapp/app/v3/scheduleEngine.server.ts +++ b/apps/webapp/app/v3/scheduleEngine.server.ts @@ -72,6 +72,8 @@ function createScheduleEngine() { distributionWindow: { seconds: env.SCHEDULE_WORKER_DISTRIBUTION_WINDOW_SECONDS, }, + schedulePhaseSecret: env.ENCRYPTION_KEY, + cronSpreadFraction: env.SCHEDULE_WORKER_CRON_SPREAD_FRACTION, tracer, meter, onTriggerScheduledTask: async ({ @@ -81,6 +83,7 @@ function createScheduleEngine() { scheduleInstanceId, scheduleId, exactScheduleTime, + effectiveScheduleTime, }) => { try { // v3 (engine V1) is retired: skip firing V1 schedules instead of triggering into a guaranteed rejection every tick. @@ -104,6 +107,7 @@ function createScheduleEngine() { scheduleInstanceId, scheduleId, exactScheduleTime, + effectiveScheduleTime, }); const result = await triggerService.call( @@ -114,7 +118,7 @@ function createScheduleEngine() { customIcon: "scheduled", scheduleId, scheduleInstanceId, - queueTimestamp: exactScheduleTime, + queueTimestamp: effectiveScheduleTime, overrideCreatedAt: exactScheduleTime, triggerSource: "schedule", triggerAction: "trigger", diff --git a/apps/webapp/app/v3/scheduleWindow.server.ts b/apps/webapp/app/v3/scheduleWindow.server.ts new file mode 100644 index 00000000000..e89123488ef --- /dev/null +++ b/apps/webapp/app/v3/scheduleWindow.server.ts @@ -0,0 +1,78 @@ +import { parseScheduleWindow } from "@internal/schedule-engine"; +import type { ScheduleWindow } from "@trigger.dev/core/v3"; + +const SECONDS_PER_UNIT = { + m: 60, + h: 3_600, +} as const; + +export type ScheduleWindowDatabaseFields = { + windowDurationSeconds: number | null; + windowPercentage: number | null; +}; + +export function normalizeScheduleWindow( + window: ScheduleWindow | undefined +): ScheduleWindowDatabaseFields { + if (window === undefined) { + return { + windowDurationSeconds: null, + windowPercentage: null, + }; + } + + const parsedWindow = parseScheduleWindow(window); + + if (parsedWindow.type === "percentage") { + return { + windowDurationSeconds: null, + windowPercentage: parsedWindow.percentage, + }; + } + + return { + windowDurationSeconds: parsedWindow.durationSeconds, + windowPercentage: null, + }; +} + +export function formatScheduleWindow({ + windowDurationSeconds, + windowPercentage, +}: ScheduleWindowDatabaseFields): ScheduleWindow | undefined { + if (windowPercentage !== null) { + return `${windowPercentage}%`; + } + + if (windowDurationSeconds === null) { + return undefined; + } + + if (windowDurationSeconds === 0) { + return "0m"; + } + + if (windowDurationSeconds % SECONDS_PER_UNIT.h === 0) { + return `${windowDurationSeconds / SECONDS_PER_UNIT.h}h`; + } + + return `${windowDurationSeconds / SECONDS_PER_UNIT.m}m`; +} + +export function validateScheduleWindowSyntax( + window: ScheduleWindow | undefined +): { valid: true } | { valid: false; message: string } { + if (window === undefined) { + return { valid: true }; + } + + try { + parseScheduleWindow(window); + return { valid: true }; + } catch (error) { + return { + valid: false, + message: error instanceof Error ? error.message : String(error), + }; + } +} diff --git a/apps/webapp/app/v3/schedules.ts b/apps/webapp/app/v3/schedules.ts index 1653e053544..bb1d3af55d7 100644 --- a/apps/webapp/app/v3/schedules.ts +++ b/apps/webapp/app/v3/schedules.ts @@ -1,3 +1,4 @@ +import { ScheduleWindow } from "@trigger.dev/core/v3"; import { parseExpression } from "cron-parser"; import { z } from "zod"; @@ -56,6 +57,7 @@ export const UpsertSchedule = z.object({ externalId: z.string().optional(), deduplicationKey: z.string().optional(), timezone: z.string().optional(), + window: ScheduleWindow.optional(), }); export type UpsertSchedule = z.infer; diff --git a/apps/webapp/app/v3/services/checkSchedule.server.ts b/apps/webapp/app/v3/services/checkSchedule.server.ts index bc28ee9c357..fb365a78244 100644 --- a/apps/webapp/app/v3/services/checkSchedule.server.ts +++ b/apps/webapp/app/v3/services/checkSchedule.server.ts @@ -5,13 +5,16 @@ import { resolveProjectScopedEnvironments } from "./resolveProjectScopedEnvironm import { getLimit } from "~/services/platform.v3.server"; import { getTimezones } from "~/utils/timezones.server"; import { env } from "~/env.server"; +import type { ScheduleWindow } from "@trigger.dev/core/v3"; import { type PrismaClientOrTransaction } from "@trigger.dev/database"; +import { validateScheduleWindowSyntax } from "../scheduleWindow.server"; type Schedule = { cron: string; timezone?: string; taskIdentifier: string; friendlyId?: string; + window?: ScheduleWindow; }; export class CheckScheduleService extends BaseService { @@ -39,6 +42,11 @@ export class CheckScheduleService extends BaseService { } } + const windowValidation = validateScheduleWindowSyntax(schedule.window); + if (!windowValidation.valid) { + throw new ServiceValidationError(windowValidation.message); + } + //check the task exists const task = await this._prisma.backgroundWorkerTask.findFirst({ where: { diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index dc5c79129ac..f73f177270c 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -29,6 +29,7 @@ import { updateQueueConcurrencyLimits, } from "../runQueue.server"; import { scheduleEngine } from "../scheduleEngine.server"; +import { normalizeScheduleWindow } from "../scheduleWindow.server"; import { calculateNextBuildVersion } from "../utils/calculateNextBuildVersion"; import { clampMaxDuration } from "../utils/maxDuration"; import { BaseService, ServiceValidationError } from "./baseService.server"; @@ -665,6 +666,10 @@ export async function syncDeclarativeSchedules( id: true, friendlyId: true, taskIdentifier: true, + generatorExpression: true, + timezone: true, + windowDurationSeconds: true, + windowPercentage: true, instances: { select: { environmentId: true, @@ -710,11 +715,18 @@ export async function syncDeclarativeSchedules( timezone: task.schedule.timezone, taskIdentifier: task.id, friendlyId: existingSchedule?.friendlyId, + window: task.schedule.window, }, [environment.id] ); if (existingSchedule) { + const normalizedWindow = normalizeScheduleWindow(task.schedule.window); + const timingChanged = + existingSchedule.generatorExpression !== task.schedule.cron || + existingSchedule.timezone !== task.schedule.timezone || + existingSchedule.windowDurationSeconds !== normalizedWindow.windowDurationSeconds || + existingSchedule.windowPercentage !== normalizedWindow.windowPercentage; const schedule = await prisma.taskSchedule.update({ where: { id: existingSchedule.id, @@ -723,6 +735,7 @@ export async function syncDeclarativeSchedules( generatorExpression: task.schedule.cron, generatorDescription: cronstrue.toString(task.schedule.cron), timezone: task.schedule.timezone, + ...normalizedWindow, }, include: { instances: true, @@ -732,7 +745,10 @@ export async function syncDeclarativeSchedules( missingSchedules.delete(existingSchedule.id); const instance = schedule.instances.at(0); if (instance) { - await scheduleEngine.registerNextTaskScheduleInstance({ instanceId: instance.id }); + await scheduleEngine.registerNextTaskScheduleInstance({ + instanceId: instance.id, + preserveExistingJob: !timingChanged, + }); } else { throw new CreateDeclarativeScheduleError( `Missing instance for declarative schedule ${schedule.id}` @@ -748,6 +764,7 @@ export async function syncDeclarativeSchedules( generatorDescription: cronstrue.toString(task.schedule.cron), timezone: task.schedule.timezone, type: "DECLARATIVE", + ...normalizeScheduleWindow(task.schedule.window), instances: { create: [ { diff --git a/apps/webapp/app/v3/services/upsertTaskSchedule.server.ts b/apps/webapp/app/v3/services/upsertTaskSchedule.server.ts index d9d8b6c0a46..567e8269d3a 100644 --- a/apps/webapp/app/v3/services/upsertTaskSchedule.server.ts +++ b/apps/webapp/app/v3/services/upsertTaskSchedule.server.ts @@ -7,6 +7,7 @@ import { calculateNextScheduledTimestampFromNow } from "../utils/calculateNextSc import { BaseService, ServiceValidationError } from "./baseService.server"; import { CheckScheduleService } from "./checkSchedule.server"; import { scheduleEngine } from "../scheduleEngine.server"; +import { formatScheduleWindow, normalizeScheduleWindow } from "../scheduleWindow.server"; import { scheduleWhereClause } from "~/models/schedules.server"; export type UpsertTaskScheduleServiceOptions = UpsertSchedule; @@ -100,6 +101,7 @@ export class UpsertTaskScheduleService extends BaseService { generatorDescription: cronstrue.toString(options.cron), timezone: options.timezone ?? "UTC", externalId: options.externalId ? options.externalId : undefined, + ...normalizeScheduleWindow(options.window), }, }); @@ -161,12 +163,15 @@ export class UpsertTaskScheduleService extends BaseService { generatorDescription: cronstrue.toString(options.cron), timezone: options.timezone ?? "UTC", externalId: options.externalId ? options.externalId : null, + ...normalizeScheduleWindow(options.window), }, }); const scheduleHasChanged = scheduleRecord.generatorExpression !== existingSchedule.generatorExpression || - scheduleRecord.timezone !== existingSchedule.timezone; + scheduleRecord.timezone !== existingSchedule.timezone || + scheduleRecord.windowDurationSeconds !== existingSchedule.windowDurationSeconds || + scheduleRecord.windowPercentage !== existingSchedule.windowPercentage; // create the new instances const newInstances: InstanceWithEnvironment[] = []; @@ -245,6 +250,7 @@ export class UpsertTaskScheduleService extends BaseService { cron: taskSchedule.generatorExpression, cronDescription: taskSchedule.generatorDescription, timezone: taskSchedule.timezone, + window: formatScheduleWindow(taskSchedule), nextRun: calculateNextScheduledTimestampFromNow( taskSchedule.generatorExpression, taskSchedule.timezone diff --git a/apps/webapp/test/engine/triggerTask.test.ts b/apps/webapp/test/engine/triggerTask.test.ts index 190fa151631..07e43b9c202 100644 --- a/apps/webapp/test/engine/triggerTask.test.ts +++ b/apps/webapp/test/engine/triggerTask.test.ts @@ -124,6 +124,78 @@ describe("RunEngineTriggerTaskService", () => { expect(queueLength).toBe(1); }); + containerTest( + "persists distinct nominal and effective schedule times", + async ({ prisma, redisOptions }) => { + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0005, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + onTestFinished(() => engine.quit()); + + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + const taskIdentifier = "scheduled-task"; + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const traceEventConcern = new MockTraceEventConcern(); + const triggerTaskService = new RunEngineTriggerTaskService({ + engine, + prisma, + payloadProcessor: new MockPayloadProcessor(), + queueConcern: new DefaultQueueManager(prisma, engine), + idempotencyKeyConcern: new IdempotencyKeyConcern(prisma, engine, traceEventConcern), + validator: new MockTriggerTaskValidator(), + traceEventConcern, + tracer: trace.getTracer("test", "0.0.0"), + metadataMaximumSize: 1024 * 1024, + }); + + const nominalAt = new Date(Date.now() - 30_000); + const effectiveAt = new Date(Date.now() + 60_000); + const result = await triggerTaskService.call({ + taskId: taskIdentifier, + environment: authenticatedEnvironment, + body: { payload: { timestamp: nominalAt } }, + options: { + overrideCreatedAt: nominalAt, + queueTimestamp: effectiveAt, + triggerSource: "schedule", + triggerAction: "trigger", + }, + }); + + const run = await prisma.taskRun.findUniqueOrThrow({ + where: { id: result!.run.id }, + }); + expect(run.createdAt).toEqual(nominalAt); + expect(run.queueTimestamp).toEqual(effectiveAt); + } + ); + containerTest( "routes scheduled-lineage runs to a separate worker queue that dequeues independently", async ({ prisma, redisOptions }) => { diff --git a/apps/webapp/test/runsReplicationService.part1.test.ts b/apps/webapp/test/runsReplicationService.part1.test.ts index be194b4dd40..7be2976df0e 100644 --- a/apps/webapp/test/runsReplicationService.part1.test.ts +++ b/apps/webapp/test/runsReplicationService.part1.test.ts @@ -73,6 +73,7 @@ describe("RunsReplicationService (part 1/7)", () => { }, }); + const queueTimestamp = new Date("2026-08-11T12:34:56.789Z"); const taskRun = await prisma.taskRun.create({ data: { friendlyId: "run_1234", @@ -81,6 +82,7 @@ describe("RunsReplicationService (part 1/7)", () => { traceId: "1234", spanId: "1234", queue: "test", + queueTimestamp, workerQueue: "us-east-1-next", region: "us-east-1", planType: "free", @@ -100,7 +102,8 @@ describe("RunsReplicationService (part 1/7)", () => { const queryRuns = clickhouse.reader.query({ name: "runs-replication", - query: "SELECT * FROM trigger_dev.task_runs_v2", + query: + "SELECT *, toString(toUnixTimestamp64Milli(queue_timestamp)) AS queue_timestamp_ms FROM trigger_dev.task_runs_v2", schema: z.any(), }); @@ -125,6 +128,7 @@ describe("RunsReplicationService (part 1/7)", () => { organization_id: organization.id, environment_type: "DEVELOPMENT", engine: "V2", + queue_timestamp_ms: queueTimestamp.getTime().toString(), trigger_source: "api", root_trigger_source: "dashboard", is_warm_start: 1, diff --git a/apps/webapp/test/scheduleWindow.test.ts b/apps/webapp/test/scheduleWindow.test.ts new file mode 100644 index 00000000000..afc4a0c0888 --- /dev/null +++ b/apps/webapp/test/scheduleWindow.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, it } from "vitest"; +import { + formatScheduleWindow, + normalizeScheduleWindow, + validateScheduleWindowSyntax, +} from "~/v3/scheduleWindow.server"; + +describe("schedule window persistence", () => { + it("normalizes duration and percentage windows", () => { + expect(normalizeScheduleWindow("30m")).toEqual({ + windowDurationSeconds: 1_800, + windowPercentage: null, + }); + expect(normalizeScheduleWindow("0m")).toEqual({ + windowDurationSeconds: 0, + windowPercentage: null, + }); + expect(normalizeScheduleWindow("30%")).toEqual({ + windowDurationSeconds: null, + windowPercentage: 30, + }); + expect(normalizeScheduleWindow(undefined)).toEqual({ + windowDurationSeconds: null, + windowPercentage: null, + }); + }); + + it("formats stored windows canonically", () => { + expect( + formatScheduleWindow({ + windowDurationSeconds: 0, + windowPercentage: null, + }) + ).toBe("0m"); + expect( + formatScheduleWindow({ + windowDurationSeconds: 86_400, + windowPercentage: null, + }) + ).toBe("24h"); + expect( + formatScheduleWindow({ + windowDurationSeconds: 7_200, + windowPercentage: null, + }) + ).toBe("2h"); + expect( + formatScheduleWindow({ + windowDurationSeconds: null, + windowPercentage: 30, + }) + ).toBe("30%"); + }); + + it.each(["30.5%", "1d", "25h"])( + "rejects invalid syntax through the authoritative timing parser: %s", + (window) => { + expect(validateScheduleWindowSyntax(window)).toMatchObject({ valid: false }); + } + ); + + it("accepts an absolute window independently of the cron interval", () => { + expect(validateScheduleWindowSyntax("30m")).toEqual({ valid: true }); + }); +}); diff --git a/apps/webapp/test/schedules-api.e2e.full.test.ts b/apps/webapp/test/schedules-api.e2e.full.test.ts new file mode 100644 index 00000000000..2924c30e10e --- /dev/null +++ b/apps/webapp/test/schedules-api.e2e.full.test.ts @@ -0,0 +1,173 @@ +import type { PrismaClient } from "@trigger.dev/database"; +import { describe, expect, it } from "vitest"; +import { seedTestEnvironment } from "./helpers/seedTestEnvironment"; +import { getTestServer } from "./helpers/sharedTestServer"; + +const TASK_IDENTIFIER = "scheduled-task"; + +describe("Schedules API windows", () => { + it("creates, retrieves, updates, and clears a window", async () => { + const server = getTestServer(); + const { apiKey, project, environment } = await seedTestEnvironment(server.prisma); + await seedScheduledTask(server.prisma, project.id, environment.id); + + const createResponse = await server.webapp.fetch("/api/v1/schedules", { + method: "POST", + headers: authHeaders(apiKey), + body: JSON.stringify({ + task: TASK_IDENTIFIER, + cron: "0 * * * *", + deduplicationKey: "window-lifecycle", + window: "30%", + }), + }); + + expect(createResponse.status).toBe(200); + const created = await createResponse.json(); + expect(created).toMatchObject({ + task: TASK_IDENTIFIER, + timezone: "UTC", + window: "30%", + }); + + const retrieveResponse = await server.webapp.fetch(`/api/v1/schedules/${created.id}`, { + headers: authHeaders(apiKey), + }); + expect(retrieveResponse.status).toBe(200); + await expect(retrieveResponse.json()).resolves.toMatchObject({ + id: created.id, + window: "30%", + }); + + const updateResponse = await server.webapp.fetch(`/api/v1/schedules/${created.id}`, { + method: "PUT", + headers: authHeaders(apiKey), + body: JSON.stringify({ + task: TASK_IDENTIFIER, + cron: "0 0 * * *", + window: "2h", + }), + }); + expect(updateResponse.status).toBe(200); + await expect(updateResponse.json()).resolves.toMatchObject({ + id: created.id, + window: "2h", + }); + + const clearResponse = await server.webapp.fetch(`/api/v1/schedules/${created.id}`, { + method: "PUT", + headers: authHeaders(apiKey), + body: JSON.stringify({ + task: TASK_IDENTIFIER, + cron: "0 0 * * *", + }), + }); + expect(clearResponse.status).toBe(200); + const cleared = await clearResponse.json(); + expect(cleared.id).toBe(created.id); + expect(cleared).not.toHaveProperty("window"); + + const stored = await server.prisma.taskSchedule.findUniqueOrThrow({ + where: { friendlyId: created.id }, + select: { windowDurationSeconds: true, windowPercentage: true }, + }); + expect(stored).toEqual({ + windowDurationSeconds: null, + windowPercentage: null, + }); + }); + + it("accepts zero windows and absolute windows longer than the cron interval", async () => { + const server = getTestServer(); + const { apiKey, project, environment } = await seedTestEnvironment(server.prisma); + await seedScheduledTask(server.prisma, project.id, environment.id); + + const windows = [ + ["0m", "0m"], + ["0h", "0m"], + ["0%", "0%"], + ["2h", "2h"], + ] as const; + + for (const [index, [window, expectedWindow]] of windows.entries()) { + const response = await server.webapp.fetch("/api/v1/schedules", { + method: "POST", + headers: authHeaders(apiKey), + body: JSON.stringify({ + task: TASK_IDENTIFIER, + cron: "0 * * * *", + deduplicationKey: `zero-window-${index}`, + window, + }), + }); + + expect(response.status).toBe(200); + await expect(response.json()).resolves.toMatchObject({ window: expectedWindow }); + } + }); + + it("returns safe errors for invalid windows", async () => { + const server = getTestServer(); + const { apiKey, project, environment } = await seedTestEnvironment(server.prisma); + await seedScheduledTask(server.prisma, project.id, environment.id); + + const invalidRequests = [ + { window: 30, expectedStatus: 400 }, + { window: "30.5%", expectedStatus: 422 }, + { window: "1d", expectedStatus: 422 }, + { window: "25h", expectedStatus: 422 }, + ]; + + for (const [index, { window, expectedStatus }] of invalidRequests.entries()) { + const response = await server.webapp.fetch("/api/v1/schedules", { + method: "POST", + headers: authHeaders(apiKey), + body: JSON.stringify({ + task: TASK_IDENTIFIER, + cron: "0 * * * *", + deduplicationKey: `invalid-window-${index}`, + window, + }), + }); + + expect(response.status).toBe(expectedStatus); + await expect(response.json()).resolves.toHaveProperty("error"); + } + }); +}); + +function authHeaders(apiKey: string) { + return { + Authorization: `Bearer ${apiKey}`, + "Content-Type": "application/json", + }; +} + +async function seedScheduledTask( + prisma: PrismaClient, + projectId: string, + runtimeEnvironmentId: string +) { + const worker = await prisma.backgroundWorker.create({ + data: { + friendlyId: `worker_${runtimeEnvironmentId}`, + contentHash: `hash_${runtimeEnvironmentId}`, + version: "20260811.1", + metadata: {}, + projectId, + runtimeEnvironmentId, + }, + }); + + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_${runtimeEnvironmentId}`, + slug: TASK_IDENTIFIER, + filePath: "src/trigger/scheduled-task.ts", + workerId: worker.id, + projectId, + runtimeEnvironmentId, + triggerSource: "SCHEDULED", + }, + }); +} diff --git a/apps/webapp/test/syncDeclarativeSchedules.test.ts b/apps/webapp/test/syncDeclarativeSchedules.test.ts index 6bd6aaa363f..d6835692532 100644 --- a/apps/webapp/test/syncDeclarativeSchedules.test.ts +++ b/apps/webapp/test/syncDeclarativeSchedules.test.ts @@ -4,8 +4,17 @@ import { describe, expect, vi } from "vitest"; import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { syncDeclarativeSchedules } from "~/v3/services/createBackgroundWorker.server"; +const { registerNextTaskScheduleInstance } = vi.hoisted(() => ({ + registerNextTaskScheduleInstance: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("~/v3/scheduleEngine.server", () => ({ + scheduleEngine: { registerNextTaskScheduleInstance }, +})); + vi.setConfig({ testTimeout: 60_000 }); +type TasksArg = Parameters[0]; type WorkerArg = Parameters[1]; const noWorker = {} as unknown as WorkerArg; @@ -82,6 +91,82 @@ function countingPrisma(prisma: PrismaClient) { const asEnv = (env: { id: string; projectId: string; type: string }) => env as unknown as AuthenticatedEnvironment; +function declarativeTasks(schedule: { cron: string; timezone: string; window?: string }): TasksArg { + return [{ id: "my-task", schedule }] as TasksArg; +} + +async function seedScheduledTask( + prisma: PrismaClient, + projectId: string, + runtimeEnvironmentId: string +) { + const worker = await prisma.backgroundWorker.create({ + data: { + friendlyId: `worker_${runtimeEnvironmentId}`, + contentHash: `hash_${runtimeEnvironmentId}`, + version: "20260811.1", + metadata: {}, + projectId, + runtimeEnvironmentId, + }, + }); + + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_${runtimeEnvironmentId}`, + slug: "my-task", + filePath: "src/trigger/my-task.ts", + workerId: worker.id, + projectId, + runtimeEnvironmentId, + triggerSource: "SCHEDULED", + }, + }); +} + +describe("syncDeclarativeSchedules registration", () => { + containerTest( + "preserves an existing Redis job when declarative timing is unchanged", + async ({ prisma }) => { + registerNextTaskScheduleInstance.mockClear(); + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + const schedule = await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id]); + await seedScheduledTask(prisma, project.id, prodEnv.id); + + await syncDeclarativeSchedules( + declarativeTasks({ cron: "0 * * * *", timezone: "UTC" }), + noWorker, + asEnv(prodEnv), + prisma + ); + + expect(registerNextTaskScheduleInstance).toHaveBeenCalledWith({ + instanceId: schedule.instances[0].id, + preserveExistingJob: true, + }); + } + ); + + containerTest("replaces the Redis job when declarative timing changes", async ({ prisma }) => { + registerNextTaskScheduleInstance.mockClear(); + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + const schedule = await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id]); + await seedScheduledTask(prisma, project.id, prodEnv.id); + + await syncDeclarativeSchedules( + declarativeTasks({ cron: "30 * * * *", timezone: "UTC", window: "30m" }), + noWorker, + asEnv(prodEnv), + prisma + ); + + expect(registerNextTaskScheduleInstance).toHaveBeenCalledWith({ + instanceId: schedule.instances[0].id, + preserveExistingJob: false, + }); + }); +}); + describe("syncDeclarativeSchedules deletion path", () => { containerTest( "does not issue any instance delete when the env owns no instance of the missing schedules", diff --git a/internal-packages/clickhouse/schema/037_add_queue_timestamp_to_task_runs_v2.sql b/internal-packages/clickhouse/schema/037_add_queue_timestamp_to_task_runs_v2.sql new file mode 100644 index 00000000000..e020c4e7de7 --- /dev/null +++ b/internal-packages/clickhouse/schema/037_add_queue_timestamp_to_task_runs_v2.sql @@ -0,0 +1,7 @@ +-- +goose Up +ALTER TABLE trigger_dev.task_runs_v2 + ADD COLUMN IF NOT EXISTS queue_timestamp Nullable(DateTime64(3)) AFTER created_at; + +-- +goose Down +ALTER TABLE trigger_dev.task_runs_v2 + DROP COLUMN IF EXISTS queue_timestamp; diff --git a/internal-packages/clickhouse/src/taskRuns.test.ts b/internal-packages/clickhouse/src/taskRuns.test.ts index 461b2d38281..3595146a986 100644 --- a/internal-packages/clickhouse/src/taskRuns.test.ts +++ b/internal-packages/clickhouse/src/taskRuns.test.ts @@ -29,6 +29,7 @@ describe("Task Runs V2", () => { }); const now = Date.now(); + const queueTimestamp = now + 30_000; const taskRunData: TaskRunInsertArray = [ "env_1234", // environment_id "org_1234", // organization_id @@ -36,6 +37,7 @@ describe("Task Runs V2", () => { "run_1234", // run_id now, // updated_at now, // created_at + queueTimestamp, // queue_timestamp "PENDING", // status "DEVELOPMENT", // environment_type "friendly_1234", // friendly_id @@ -105,6 +107,7 @@ describe("Task Runs V2", () => { schema: z.object({ environment_id: z.string(), run_id: z.string(), + queue_timestamp: z.coerce.date().nullable(), concurrency_key: z.string(), bulk_action_group_ids: z.array(z.string()), }), @@ -121,6 +124,7 @@ describe("Task Runs V2", () => { expect.objectContaining({ environment_id: "env_1234", run_id: "run_1234", + queue_timestamp: new Date(queueTimestamp), concurrency_key: "concurrency_key_1234", bulk_action_group_ids: ["bulk_action_group_id_1234", "bulk_action_group_id_1235"], }), @@ -183,6 +187,7 @@ describe("Task Runs V2", () => { "run_mixed", // run_id now, // updated_at now, // created_at + null, // queue_timestamp "COMPLETED_SUCCESSFULLY", // status "DEVELOPMENT", // environment_type "friendly_mixed", // friendly_id @@ -282,6 +287,7 @@ describe("Task Runs V2", () => { "cma45oli70002qrdy47w0j4n7", // run_id createdAt, // updated_at createdAt, // created_at + null, // queue_timestamp "PENDING", // status "PRODUCTION", // environment_type "run_cma45oli70002qrdy47w0j4n7", // friendly_id @@ -339,6 +345,7 @@ describe("Task Runs V2", () => { "cma45oli70002qrdy47w0j4n7", // run_id createdAt, // updated_at createdAt, // created_at + null, // queue_timestamp "COMPLETED_SUCCESSFULLY", // status "PRODUCTION", // environment_type "run_cma45oli70002qrdy47w0j4n7", // friendly_id @@ -443,6 +450,7 @@ describe("Task Runs V2", () => { "cma45oli70002qrdy47w0j4n7", // run_id createdAt, // updated_at createdAt, // created_at + null, // queue_timestamp "PENDING", // status "PRODUCTION", // environment_type "run_cma45oli70002qrdy47w0j4n7", // friendly_id @@ -555,6 +563,7 @@ describe("Task Runs V2", () => { "root_run_1", // run_id baseCreatedAt, // updated_at baseCreatedAt, // created_at + null, // queue_timestamp "EXECUTING", // status "DEVELOPMENT", // environment_type "run_root_1", // friendly_id @@ -612,6 +621,7 @@ describe("Task Runs V2", () => { "child_a", baseCreatedAt + 1_000, baseCreatedAt + 1_000, + null, // queue_timestamp "PENDING", "DEVELOPMENT", "run_child_a", @@ -673,6 +683,7 @@ describe("Task Runs V2", () => { "child_b", baseCreatedAt + 2_000, baseCreatedAt + 2_000, + null, // queue_timestamp "EXECUTING", "DEVELOPMENT", "run_child_b", @@ -730,6 +741,7 @@ describe("Task Runs V2", () => { "child_deleted", baseCreatedAt + 3_000, baseCreatedAt + 3_000, + null, // queue_timestamp "PENDING", "DEVELOPMENT", "run_child_deleted", @@ -907,6 +919,7 @@ describe("Task Runs V2", () => { "cma45oli70002qrdy47w0j4n7", createdAt, createdAt, + null, // queue_timestamp "PENDING", "PRODUCTION", "run_cma45oli70002qrdy47w0j4n7", @@ -1010,6 +1023,7 @@ describe("Task Runs V2", () => { "cma45oli70002qrdy47w0j4n7", createdAt, createdAt, + null, // queue_timestamp "PENDING", "PRODUCTION", "run_cma45oli70002qrdy47w0j4n7", @@ -1113,6 +1127,7 @@ describe("Task Runs V2", () => { "cma45oli70002qrdy47w0j4n7", createdAt, createdAt, + null, // queue_timestamp "PENDING", "PRODUCTION", "run_cma45oli70002qrdy47w0j4n7", diff --git a/internal-packages/clickhouse/src/taskRuns.ts b/internal-packages/clickhouse/src/taskRuns.ts index a0c5f8c4f9b..f561be04d9e 100644 --- a/internal-packages/clickhouse/src/taskRuns.ts +++ b/internal-packages/clickhouse/src/taskRuns.ts @@ -9,6 +9,7 @@ export const TaskRunV2 = z.object({ run_id: z.string(), updated_at: z.number().int(), created_at: z.number().int(), + queue_timestamp: z.number().int().nullish(), status: z.string(), environment_type: z.string(), friendly_id: z.string(), @@ -69,6 +70,7 @@ export const TASK_RUN_COLUMNS = [ "run_id", "updated_at", "created_at", + "queue_timestamp", "status", "environment_type", "friendly_id", @@ -138,6 +140,7 @@ export type TaskRunFieldTypes = { run_id: string; updated_at: number; created_at: number; + queue_timestamp: number | null; status: string; environment_type: string; friendly_id: string; @@ -306,6 +309,7 @@ export type TaskRunInsertArray = [ run_id: string, updated_at: number, created_at: number, + queue_timestamp: number | null, status: string, environment_type: string, friendly_id: string, diff --git a/internal-packages/database/prisma/migrations/20260810130446_add_cron_spread_fields/migration.sql b/internal-packages/database/prisma/migrations/20260810130446_add_cron_spread_fields/migration.sql new file mode 100644 index 00000000000..da12959b9bf --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260810130446_add_cron_spread_fields/migration.sql @@ -0,0 +1,15 @@ +-- AlterTable +ALTER TABLE "public"."TaskSchedule" + ADD COLUMN "windowDurationSeconds" INTEGER, + ADD COLUMN "windowPercentage" INTEGER; + +ALTER TABLE "public"."TaskSchedule" + ADD CONSTRAINT "TaskSchedule_window_exclusive" + CHECK ( + "windowDurationSeconds" IS NULL + OR "windowPercentage" IS NULL + ) NOT VALID; + +-- AlterTable +ALTER TABLE "public"."TaskScheduleInstance" + ADD COLUMN "schedulePhase" INTEGER; diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index 0c77ac3aa24..868e02eb27e 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/internal-packages/database/prisma/schema.prisma @@ -2270,6 +2270,10 @@ model TaskSchedule { /// These are IANA format string, or the default "UTC". E.g. "America/New_York" timezone String @default("UTC") + // Cron spread + windowDurationSeconds Int? + windowPercentage Int? + ///Can be provided by the user then accessed inside a run externalId String? @@ -2317,6 +2321,9 @@ model TaskScheduleInstance { project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade) projectId String + // Durable cron spread phase + schedulePhase Int? + createdAt DateTime @default(now()) updatedAt DateTime @updatedAt diff --git a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts index 28918ce6f47..8887c850a56 100644 --- a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts @@ -156,6 +156,10 @@ export class DequeueSystem { const orgId = message.message.orgId; const runId = message.messageId; + const queueWaitMs = + typeof message.message.eligibleAtMs === "number" + ? Math.max(0, Date.now() - message.message.eligibleAtMs) + : undefined; this.$.logger.info("DequeueSystem.dequeueFromWorkerQueue dequeued message", { runId, @@ -174,6 +178,9 @@ export class DequeueSystem { span.setAttribute("consumer_id", consumerId); span.setAttribute("worker_queue", workerQueue); span.setAttribute("blocking_pop", blockingPop ?? true); + if (queueWaitMs !== undefined) { + span.setAttribute("queue_wait_ms", queueWaitMs); + } //lock the run so nothing else can modify it try { diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 58225cc5051..cd6a8ce3bd8 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -2133,8 +2133,10 @@ export class RunQueue { const messageId = message.runId; const messageData = JSON.stringify(message); const messageScore = String(message.timestamp); - const currentTime = String(Date.now()); - const enableFastPathArg = enableFastPath ? "1" : "0"; + const currentTimeMs = Date.now(); + const shouldEnableFastPath = enableFastPath && message.timestamp <= currentTimeMs; + const currentTime = String(currentTimeMs); + const enableFastPathArg = shouldEnableFastPath ? "1" : "0"; const metricsGaugeArg = this.#queueMetricsGaugeArg(); const defaultEnvConcurrencyLimit = String(this.options.defaultEnvConcurrency); const defaultEnvConcurrencyBurstFactor = String( @@ -2155,6 +2157,7 @@ export class RunQueue { messageScore, masterQueueKey, enableFastPath, + shouldEnableFastPath, ttlInfo, service: this.name, }); diff --git a/internal-packages/run-engine/src/run-queue/tests/enqueueMessage.test.ts b/internal-packages/run-engine/src/run-queue/tests/enqueueMessage.test.ts index a12755b5fec..15023aa7e88 100644 --- a/internal-packages/run-engine/src/run-queue/tests/enqueueMessage.test.ts +++ b/internal-packages/run-engine/src/run-queue/tests/enqueueMessage.test.ts @@ -179,6 +179,53 @@ describe("RunQueue.enqueueMessage fast path", () => { } ); + redisTest("should not fast-path a future-scored message", async ({ redisContainer }) => { + const queue = createQueue(redisContainer, "runqueue:fp-future-score:"); + + try { + await queue.updateEnvConcurrencyLimits(authenticatedEnvDev); + + const futureMessage: InputPayload = { + ...messageDev, + runId: "r_future_score", + timestamp: Date.now() + 60_000, + }; + + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: futureMessage, + workerQueue: authenticatedEnvDev.id, + enableFastPath: true, + }); + + const queueLength = await queue.lengthOfQueue(authenticatedEnvDev, futureMessage.queue); + const queueConcurrency = await queue.currentConcurrencyOfQueue( + authenticatedEnvDev, + futureMessage.queue + ); + const dequeued = await queue.dequeueMessageFromWorkerQueue( + "test_12345", + authenticatedEnvDev.id, + { blockingPop: false } + ); + + expect({ + // A future-scored message must remain in the sorted set until it is eligible. + queueLength, + // It must not claim concurrency before it becomes eligible. + queueConcurrency, + // It must not be visible to a worker before its timestamp. + dequeuedMessageId: dequeued?.messageId, + }).toEqual({ + queueLength: 1, + queueConcurrency: 0, + dequeuedMessageId: undefined, + }); + } finally { + await queue.quit(); + } + }); + redisTest("should take slow path when enableFastPath is false", async ({ redisContainer }) => { const queue = createQueue(redisContainer, "runqueue:fp2:"); diff --git a/internal-packages/schedule-engine/src/engine/distributedScheduling.ts b/internal-packages/schedule-engine/src/engine/distributedScheduling.ts index 4c9b6b440d5..df85e21a0fc 100644 --- a/internal-packages/schedule-engine/src/engine/distributedScheduling.ts +++ b/internal-packages/schedule-engine/src/engine/distributedScheduling.ts @@ -1,16 +1,16 @@ /** * Calculates a distributed execution time for a scheduled task. - * Tasks are distributed across a time window before the exact schedule time + * Tasks are distributed across a time window before their target time * to prevent thundering herd issues while maintaining schedule accuracy. */ export function calculateDistributedExecutionTime( - exactScheduleTime: Date, + targetTime: Date, distributionWindowSeconds: number = 30, instanceId?: string ): Date { // Create seed by combining ISO timestamp with optional instanceId // This ensures different instances get different distributions even with same schedule time - const timeSeed = exactScheduleTime.toISOString(); + const timeSeed = targetTime.toISOString(); const seed = instanceId ? `${timeSeed}:${instanceId}` : timeSeed; // Use a better hash function (FNV-1a variant) for more uniform distribution @@ -30,6 +30,6 @@ export function calculateDistributedExecutionTime( // Calculate offset in milliseconds (0 to distributionWindowSeconds * 1000) const offsetMs = Math.floor(normalized * distributionWindowSeconds * 1000); - // Return time that's offsetMs before the exact schedule time - return new Date(exactScheduleTime.getTime() - offsetMs); + // Return time that's offsetMs before the target time + return new Date(targetTime.getTime() - offsetMs); } diff --git a/internal-packages/schedule-engine/src/engine/index.ts b/internal-packages/schedule-engine/src/engine/index.ts index 277f4c3d87e..e4930736cdb 100644 --- a/internal-packages/schedule-engine/src/engine/index.ts +++ b/internal-packages/schedule-engine/src/engine/index.ts @@ -5,7 +5,7 @@ import type { PrismaClient } from "@trigger.dev/database"; import { Worker, type JobHandlerParams } from "@trigger.dev/redis-worker"; import { calculateDistributedExecutionTime } from "./distributedScheduling.js"; import { - calculateNextScheduledTimestamp, + calculateNextSchedulableOccurrence, nextScheduledTimestamps, previousScheduledTimestamp, } from "./scheduleCalculation.js"; @@ -15,6 +15,11 @@ import type { TriggerScheduledTaskCallback, TriggerScheduleParams, } from "./types.js"; +import { + calculateSchedulePhase, + SCHEDULE_PHASE_DENOMINATOR, + type NormalizedScheduleWindow, +} from "./scheduleTiming.js"; import { scheduleWorkerCatalog } from "./workerCatalog.js"; import { tryCatch } from "@trigger.dev/core/utils"; @@ -31,6 +36,8 @@ export class ScheduleEngine { private scheduleExecutionDuration: Histogram; private scheduleExecutionFailureCounter: Counter; private distributionOffsetHistogram: Histogram; + private scheduleWindowCappedCounter: Counter; + private schedulePhasePersistedCounter: Counter; private devEnvironmentCheckCounter: Counter; prisma: PrismaClient; @@ -71,11 +78,22 @@ export class ScheduleEngine { this.distributionOffsetHistogram = this.meter.createHistogram( "schedule_distribution_offset_ms", { - description: "Distribution offset from exact schedule time in milliseconds", + description: "Distribution offset from effective schedule time in milliseconds", unit: "ms", } ); + this.scheduleWindowCappedCounter = this.meter.createCounter("schedule_windows_capped_total", { + description: "Total number of absolute schedule windows capped at the next nominal interval", + }); + + this.schedulePhasePersistedCounter = this.meter.createCounter( + "schedule_phase_persisted_total", + { + description: "Total number of schedule phases persisted during registration", + } + ); + this.devEnvironmentCheckCounter = this.meter.createCounter("dev_environment_checks_total", { description: "Total number of development environment connectivity checks", }); @@ -168,24 +186,113 @@ export class ScheduleEngine { instance.taskSchedule.generatorExpression ); - const fromTimestamp = params.fromTimestamp ?? new Date(); - span.setAttribute("from_timestamp", fromTimestamp.toISOString()); + const scheduleWindow = normalizedScheduleWindow(instance.taskSchedule); + const schedulePhase = + instance.schedulePhase ?? + calculateSchedulePhase({ + secret: this.options.schedulePhaseSecret, + environmentId: instance.environmentId, + deduplicationKey: instance.taskSchedule.deduplicationKey, + }); - const nextScheduledTimestamp = calculateNextScheduledTimestamp( - instance.taskSchedule.generatorExpression, - instance.taskSchedule.timezone, - fromTimestamp + const cronSpreadActive = this.#isCronSpreadActive(schedulePhase); + + let persisted = false; + if (cronSpreadActive && instance.schedulePhase === null) { + await this.prisma.taskScheduleInstance.updateMany({ + where: { + id: instance.id, + schedulePhase: null, + }, + data: { + schedulePhase, + }, + }); + persisted = true; + this.schedulePhasePersistedCounter.add(1, { + environment_type: instance.environment.type, + schedule_type: instance.taskSchedule.type, + }); + } + + span.setAttribute( + "schedule_phase_source", + instance.schedulePhase !== null ? "db" : persisted ? "persisted" : "ephemeral" ); + span.setAttribute("schedule_phase", schedulePhase); - span.setAttribute("next_scheduled_timestamp", nextScheduledTimestamp.toISOString()); + const registrationTime = new Date(); + const fromTimestamp = params.fromTimestamp ?? registrationTime; + span.setAttribute("from_timestamp", fromTimestamp.toISOString()); - const schedulingDelayMs = nextScheduledTimestamp.getTime() - Date.now(); + const { + nominalAt, + candidateEffectiveAt, + effectiveAt, + effectiveRangeMs, + windowMs, + offsetMs: candidateDelayMs, + intervalMs, + windowWasCappedToInterval, + skippedExpiredOccurrences, + } = calculateNextSchedulableOccurrence({ + schedule: instance.taskSchedule.generatorExpression, + timezone: instance.taskSchedule.timezone, + afterNominal: fromTimestamp, + now: registrationTime, + schedulePhase, + window: scheduleWindow, + cronSpreadEnabled: cronSpreadActive, + }); + const appliedDelayMs = effectiveAt.getTime() - nominalAt.getTime(); + + span.setAttribute("cron_spread_fraction", this.options.cronSpreadFraction); + span.setAttribute("cron_spread_active", cronSpreadActive); + span.setAttribute("schedule_window_type", scheduleWindow?.type ?? "none"); + span.setAttribute("next_scheduled_timestamp", nominalAt.toISOString()); + span.setAttribute("candidate_effective_schedule_time", candidateEffectiveAt.toISOString()); + span.setAttribute("effective_schedule_time", effectiveAt.toISOString()); + span.setAttribute("candidate_delay_ms", candidateDelayMs); + span.setAttribute("applied_delay_ms", appliedDelayMs); + span.setAttribute("schedule_window_ms", windowMs); + span.setAttribute("effective_range_ms", effectiveRangeMs); + span.setAttribute("schedule_window_was_capped_to_interval", windowWasCappedToInterval); + span.setAttribute("schedule_expired_occurrences_skipped", skippedExpiredOccurrences); + + if (skippedExpiredOccurrences) { + span.addEvent("schedule_expired_occurrences_skipped", { + from_nominal_time: fromTimestamp.toISOString(), + selected_nominal_time: nominalAt.toISOString(), + }); + } + + if (windowWasCappedToInterval) { + span.addEvent("schedule_window_capped_to_interval", { + requested_window_ms: windowMs, + nominal_interval_ms: intervalMs, + }); + this.scheduleWindowCappedCounter.add(1, { + environment_type: instance.environment.type, + schedule_type: instance.taskSchedule.type, + }); + } + + const schedulingDelayMs = effectiveAt.getTime() - registrationTime.getTime(); span.setAttribute("scheduling_delay_ms", schedulingDelayMs); - this.logger.debug("Calculated next schedule timestamp", { + this.logger.debug("Calculated next schedule timestamps", { instanceId: params.instanceId, taskIdentifier: instance.taskSchedule.taskIdentifier, - nextScheduledTimestamp: nextScheduledTimestamp.toISOString(), + nominalAt: nominalAt.toISOString(), + candidateEffectiveAt: candidateEffectiveAt.toISOString(), + effectiveAt: effectiveAt.toISOString(), + cronSpreadActive, + scheduleWindowType: scheduleWindow?.type ?? "none", + candidateDelayMs, + appliedDelayMs, + effectiveRangeMs, + windowWasCappedToInterval, + skippedExpiredOccurrences, schedulingDelayMs, generatorExpression: instance.taskSchedule.generatorExpression, timezone: instance.taskSchedule.timezone, @@ -224,11 +331,13 @@ export class ScheduleEngine { } } - await this.enqueueScheduledTask( - params.instanceId, - nextScheduledTimestamp, - lastScheduleTime - ); + await this.enqueueScheduledTask({ + instanceId: params.instanceId, + exactScheduleTime: nominalAt, + effectiveScheduleTime: effectiveAt, + lastScheduleTime, + preserveExistingJob: params.preserveExistingJob, + }); // Record metrics this.scheduleRegistrationCounter.add(1, { @@ -268,6 +377,7 @@ export class ScheduleEngine { instanceId: payload.instanceId, finalAttempt: false, // TODO: implement retry logic exactScheduleTime: payload.exactScheduleTime, + effectiveScheduleTime: payload.effectiveScheduleTime, lastScheduleTime: payload.lastScheduleTime, }); } @@ -281,14 +391,17 @@ export class ScheduleEngine { span.setAttribute("instanceId", params.instanceId); span.setAttribute("finalAttempt", params.finalAttempt); - if (params.exactScheduleTime) { - span.setAttribute("exactScheduleTime", params.exactScheduleTime.toISOString()); - } + const exactScheduleTime = params.exactScheduleTime ?? new Date(); + const effectiveScheduleTime = params.effectiveScheduleTime ?? exactScheduleTime; + + span.setAttribute("exactScheduleTime", exactScheduleTime.toISOString()); + span.setAttribute("effectiveScheduleTime", effectiveScheduleTime.toISOString()); this.logger.debug("Starting scheduled task trigger", { instanceId: params.instanceId, finalAttempt: params.finalAttempt, - exactScheduleTime: params.exactScheduleTime?.toISOString(), + exactScheduleTime: exactScheduleTime.toISOString(), + effectiveScheduleTime: effectiveScheduleTime.toISOString(), }); let taskIdentifier: string | undefined; @@ -412,9 +525,6 @@ export class ScheduleEngine { span.setAttribute("skip_reason", skipReason); } - // Calculate the schedule timestamp that will be used (regardless of whether we trigger or not) - const scheduleTimestamp = params.exactScheduleTime ?? new Date(); - if (shouldTrigger) { // payload.lastTimestamp is the actual previous fire time. Sources, in // order: @@ -427,25 +537,48 @@ export class ScheduleEngine { // 3. undefined — first-ever fire (no previous fire to point at). const lastTimestamp = params.lastScheduleTime ?? instance.lastScheduledTimestamp ?? undefined; + const actualExecutionTime = new Date(); + const scheduleWindow = normalizedScheduleWindow(instance.taskSchedule); + const schedulePhase = + instance.schedulePhase ?? + calculateSchedulePhase({ + secret: this.options.schedulePhaseSecret, + environmentId: instance.environmentId, + deduplicationKey: instance.taskSchedule.deduplicationKey, + }); + const cronSpreadActive = this.#isCronSpreadActive(schedulePhase); + span.setAttribute("cron_spread_active", cronSpreadActive); + const nextOccurrence = calculateNextSchedulableOccurrence({ + schedule: instance.taskSchedule.generatorExpression, + timezone: instance.taskSchedule.timezone, + afterNominal: exactScheduleTime, + now: actualExecutionTime, + schedulePhase, + window: scheduleWindow, + cronSpreadEnabled: cronSpreadActive, + }); + const upcoming = [ + nextOccurrence.nominalAt, + ...nextScheduledTimestamps( + instance.taskSchedule.generatorExpression, + instance.taskSchedule.timezone, + nextOccurrence.nominalAt, + 9 + ), + ]; const payload = { scheduleId: instance.taskSchedule.friendlyId, type: instance.taskSchedule.type as "DECLARATIVE" | "IMPERATIVE", - timestamp: scheduleTimestamp, + timestamp: exactScheduleTime, lastTimestamp, externalId: instance.taskSchedule.externalId ?? undefined, timezone: instance.taskSchedule.timezone, - upcoming: nextScheduledTimestamps( - instance.taskSchedule.generatorExpression, - instance.taskSchedule.timezone, - scheduleTimestamp, - 10 - ), + upcoming, }; // Calculate execution timing metrics - const actualExecutionTime = new Date(); - const schedulingAccuracyMs = actualExecutionTime.getTime() - scheduleTimestamp.getTime(); + const schedulingAccuracyMs = actualExecutionTime.getTime() - exactScheduleTime.getTime(); span.setAttribute("scheduling_accuracy_ms", schedulingAccuracyMs); span.setAttribute("actual_execution_time", actualExecutionTime.toISOString()); @@ -453,7 +586,8 @@ export class ScheduleEngine { this.logger.debug("Triggering scheduled task", { instanceId: params.instanceId, taskIdentifier: instance.taskSchedule.taskIdentifier, - scheduleTimestamp: scheduleTimestamp.toISOString(), + exactScheduleTime: exactScheduleTime.toISOString(), + effectiveScheduleTime: effectiveScheduleTime.toISOString(), actualExecutionTime: actualExecutionTime.toISOString(), schedulingAccuracyMs, lastTimestamp: lastTimestamp?.toISOString(), @@ -469,7 +603,8 @@ export class ScheduleEngine { payload, scheduleInstanceId: instance.id, scheduleId: instance.taskSchedule.id, - exactScheduleTime: scheduleTimestamp, + exactScheduleTime, + effectiveScheduleTime, }) ); @@ -573,21 +708,22 @@ export class ScheduleEngine { }); } - // Register the next run. `fromTimestamp` advances on every tick so - // the next cron slot keeps marching forward even through skips. + // Register the next run. `fromTimestamp` anchors nominal chaining; + // registration preserves an upcoming effective occurrence and skips + // expired intermediate ticks after downtime. // `lastScheduleTime` is the actual previous fire time the next job // will report as `payload.lastTimestamp` — only advance it when we // actually triggered, otherwise carry forward the existing value so // a long pause/disconnect doesn't quietly overwrite the real // last-fire timestamp with a series of skipped slots. const carriedLastScheduleTime = shouldTrigger - ? scheduleTimestamp + ? exactScheduleTime : (params.lastScheduleTime ?? instance.lastScheduledTimestamp ?? undefined); const [nextRunError] = await tryCatch( this.registerNextTaskScheduleInstance({ instanceId: params.instanceId, - fromTimestamp: scheduleTimestamp, + fromTimestamp: exactScheduleTime, lastScheduleTime: carriedLastScheduleTime, }) ); @@ -641,28 +777,49 @@ export class ScheduleEngine { }); } + /** + * Per-schedule rollout gate for cron spread. The schedule's deterministic + * phase doubles as a stable sampling key: raising the fraction is strictly + * additive (a schedule never leaves the rollout once included), and 0/1 map + * to fully off/on. + */ + #isCronSpreadActive(schedulePhase: number): boolean { + return schedulePhase < this.options.cronSpreadFraction * SCHEDULE_PHASE_DENOMINATOR; + } + /** * Enqueues a scheduled task with distributed execution timing */ - private async enqueueScheduledTask( - instanceId: string, - exactScheduleTime: Date, - lastScheduleTime?: Date - ) { + private async enqueueScheduledTask({ + instanceId, + exactScheduleTime, + effectiveScheduleTime, + lastScheduleTime, + preserveExistingJob = false, + }: { + instanceId: string; + exactScheduleTime: Date; + effectiveScheduleTime: Date; + lastScheduleTime?: Date; + preserveExistingJob?: boolean; + }) { return startSpan(this.tracer, "enqueueScheduledTask", async (span) => { span.setAttribute("instanceId", instanceId); span.setAttribute("exactScheduleTime", exactScheduleTime.toISOString()); + span.setAttribute("effectiveScheduleTime", effectiveScheduleTime.toISOString()); + span.setAttribute("preserveExistingJob", preserveExistingJob); if (lastScheduleTime) { span.setAttribute("lastScheduleTime", lastScheduleTime.toISOString()); } const distributedExecutionTime = calculateDistributedExecutionTime( - exactScheduleTime, + effectiveScheduleTime, this.distributionWindowSeconds, instanceId ); - const distributionOffsetMs = exactScheduleTime.getTime() - distributedExecutionTime.getTime(); + const distributionOffsetMs = + effectiveScheduleTime.getTime() - distributedExecutionTime.getTime(); span.setAttribute("distributedExecutionTime", distributedExecutionTime.toISOString()); span.setAttribute("distributionOffsetMs", distributionOffsetMs); @@ -675,29 +832,42 @@ export class ScheduleEngine { this.logger.debug("Enqueuing scheduled task with distributed execution", { instanceId, exactScheduleTime: exactScheduleTime.toISOString(), + effectiveScheduleTime: effectiveScheduleTime.toISOString(), distributedExecutionTime: distributedExecutionTime.toISOString(), distributionOffsetMs, distributionWindowSeconds: this.distributionWindowSeconds, + preserveExistingJob, }); try { - await this.worker.enqueue({ + const job = { id: `scheduled-task-instance:${instanceId}`, - job: "schedule.triggerScheduledTask", + job: "schedule.triggerScheduledTask" as const, payload: { instanceId, exactScheduleTime, + effectiveScheduleTime, lastScheduleTime, }, availableAt: distributedExecutionTime, - }); + }; + let enqueued = true; + if (preserveExistingJob) { + enqueued = await this.worker.enqueueOnce(job); + } else { + await this.worker.enqueue(job); + } span.setAttribute("enqueue_success", true); + span.setAttribute("existing_job_preserved", !enqueued); - this.logger.debug("Successfully enqueued scheduled task", { - instanceId, - jobId: `scheduled-task-instance:${instanceId}`, - }); + this.logger.debug( + enqueued ? "Successfully enqueued scheduled task" : "Preserved existing scheduled task", + { + instanceId, + jobId: job.id, + } + ); } catch (error) { this.logger.error("Failed to enqueue scheduled task", { instanceId, @@ -865,3 +1035,21 @@ export class ScheduleEngine { } } } + +function normalizedScheduleWindow({ + windowDurationSeconds, + windowPercentage, +}: { + windowDurationSeconds: number | null; + windowPercentage: number | null; +}): NormalizedScheduleWindow | undefined { + if (windowPercentage !== null) { + return { type: "percentage", percentage: windowPercentage }; + } + + if (windowDurationSeconds !== null) { + return { type: "duration", durationSeconds: windowDurationSeconds }; + } + + return undefined; +} diff --git a/internal-packages/schedule-engine/src/engine/scheduleCalculation.test.ts b/internal-packages/schedule-engine/src/engine/scheduleCalculation.test.ts new file mode 100644 index 00000000000..cfa9922df2e --- /dev/null +++ b/internal-packages/schedule-engine/src/engine/scheduleCalculation.test.ts @@ -0,0 +1,145 @@ +import { + calculateNextNominalTimestamp, + calculateNextSchedulableOccurrence, + nextScheduledTimestamps, +} from "./scheduleCalculation.js"; +import { SCHEDULE_PHASE_DENOMINATOR } from "./scheduleTiming.js"; + +describe("calculateNextNominalTimestamp", () => { + it("advances from the previous nominal tick instead of wall-clock time", () => { + const next = calculateNextNominalTimestamp( + "* * * * *", + "UTC", + new Date("2024-01-01T09:00:00.000Z") + ); + + expect(next).toEqual(new Date("2024-01-01T09:01:00.000Z")); + }); + + it("uses the 23-hour elapsed interval across spring DST", () => { + const nominalAt = new Date("2026-03-08T05:00:00.000Z"); + const next = calculateNextNominalTimestamp("0 0 * * *", "America/New_York", nominalAt); + + expect(next).toEqual(new Date("2026-03-09T04:00:00.000Z")); + expect(next.getTime() - nominalAt.getTime()).toBe(23 * 60 * 60 * 1_000); + }); + + it("uses the 25-hour elapsed interval across autumn DST", () => { + const nominalAt = new Date("2026-11-01T04:00:00.000Z"); + const next = calculateNextNominalTimestamp("0 0 * * *", "America/New_York", nominalAt); + + expect(next).toEqual(new Date("2026-11-02T05:00:00.000Z")); + expect(next.getTime() - nominalAt.getTime()).toBe(25 * 60 * 60 * 1_000); + }); + + it("preserves cron-parser calendar semantics across month boundaries", () => { + const next = calculateNextNominalTimestamp( + "0 23 L * *", + "UTC", + new Date("2027-01-31T23:00:00.000Z") + ); + + expect(next).toEqual(new Date("2027-02-28T23:00:00.000Z")); + }); +}); + +describe("calculateNextSchedulableOccurrence", () => { + const hourlySchedule = "0 * * * *"; + const window = { type: "percentage", percentage: 100 } as const; + + it("restores wall-clock catch-up behavior when spreading is disabled", () => { + const occurrence = calculateNextSchedulableOccurrence({ + schedule: hourlySchedule, + timezone: "UTC", + afterNominal: new Date("2026-08-11T09:00:00.000Z"), + now: new Date("2026-08-11T12:30:00.000Z"), + schedulePhase: (SCHEDULE_PHASE_DENOMINATOR * 3) / 4, + window, + cronSpreadEnabled: false, + }); + + expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T13:00:00.000Z")); + expect(occurrence.effectiveAt).toEqual(occurrence.nominalAt); + expect(occurrence.skippedExpiredOccurrences).toBe(true); + }); + + it("keeps strict nominal chaining when the next effective time is upcoming", () => { + const occurrence = calculateNextSchedulableOccurrence({ + schedule: hourlySchedule, + timezone: "UTC", + afterNominal: new Date("2026-08-11T09:00:00.000Z"), + now: new Date("2026-08-11T10:00:01.000Z"), + schedulePhase: (SCHEDULE_PHASE_DENOMINATOR * 3) / 4, + window, + cronSpreadEnabled: true, + }); + + expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T10:00:00.000Z")); + expect(occurrence.effectiveAt).toEqual(new Date("2026-08-11T10:45:00.000Z")); + expect(occurrence.skippedExpiredOccurrences).toBe(false); + }); + + it("keeps the latest nominal occurrence when its effective time is upcoming", () => { + const occurrence = calculateNextSchedulableOccurrence({ + schedule: hourlySchedule, + timezone: "UTC", + afterNominal: new Date("2026-08-11T09:00:00.000Z"), + now: new Date("2026-08-11T12:30:00.000Z"), + schedulePhase: (SCHEDULE_PHASE_DENOMINATOR * 3) / 4, + window, + cronSpreadEnabled: true, + }); + + expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T12:00:00.000Z")); + expect(occurrence.effectiveAt).toEqual(new Date("2026-08-11T12:45:00.000Z")); + expect(occurrence.skippedExpiredOccurrences).toBe(true); + }); + + it("skips to the next future nominal occurrence when the latest effective time expired", () => { + const occurrence = calculateNextSchedulableOccurrence({ + schedule: hourlySchedule, + timezone: "UTC", + afterNominal: new Date("2026-08-11T09:00:00.000Z"), + now: new Date("2026-08-11T12:30:00.000Z"), + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 4, + window, + cronSpreadEnabled: true, + }); + + expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T13:00:00.000Z")); + expect(occurrence.effectiveAt).toEqual(new Date("2026-08-11T13:15:00.000Z")); + expect(occurrence.skippedExpiredOccurrences).toBe(true); + }); + + it("includes a nominal occurrence exactly at now when it is still eligible", () => { + const occurrence = calculateNextSchedulableOccurrence({ + schedule: hourlySchedule, + timezone: "UTC", + afterNominal: new Date("2026-08-11T09:00:00.000Z"), + now: new Date("2026-08-11T12:00:00.000Z"), + schedulePhase: 0, + window, + cronSpreadEnabled: true, + }); + + expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T12:00:00.000Z")); + expect(occurrence.effectiveAt).toEqual(new Date("2026-08-11T12:00:00.000Z")); + }); +}); + +describe("nextScheduledTimestamps", () => { + it("advances every timestamp from the preceding nominal tick", () => { + const upcoming = nextScheduledTimestamps( + "* * * * *", + "UTC", + new Date("2024-01-01T09:00:00.000Z"), + 3 + ); + + expect(upcoming).toEqual([ + new Date("2024-01-01T09:01:00.000Z"), + new Date("2024-01-01T09:02:00.000Z"), + new Date("2024-01-01T09:03:00.000Z"), + ]); + }); +}); diff --git a/internal-packages/schedule-engine/src/engine/scheduleCalculation.ts b/internal-packages/schedule-engine/src/engine/scheduleCalculation.ts index 140ea4e285f..074aae16042 100644 --- a/internal-packages/schedule-engine/src/engine/scheduleCalculation.ts +++ b/internal-packages/schedule-engine/src/engine/scheduleCalculation.ts @@ -1,9 +1,22 @@ import { parseExpression } from "cron-parser"; +import { + calculateEffectiveScheduleTime, + type EffectiveScheduleTime, + type NormalizedScheduleWindow, +} from "./scheduleTiming.js"; export function calculateNextScheduledTimestampFromNow(schedule: string, timezone: string | null) { return calculateNextScheduledTimestamp(schedule, timezone, new Date()); } +export function calculateNextNominalTimestamp( + schedule: string, + timezone: string | null, + nominalTimestamp: Date +) { + return calculateNextStep(schedule, timezone, nominalTimestamp); +} + export function calculateNextScheduledTimestamp( schedule: string, timezone: string | null, @@ -29,6 +42,82 @@ function calculateNextStep(schedule: string, timezone: string | null, currentDat .toDate(); } +type SchedulableOccurrence = Omit & { + candidateEffectiveAt: Date; + effectiveAt: Date; + skippedExpiredOccurrences: boolean; +}; + +/** + * Selects the next occurrence that has not passed its actual eligibility time. + * + * The usual path advances strictly from the preceding nominal tick. If that occurrence expired + * during downtime, selection jumps directly to the latest nominal tick that could still be + * eligible, or to the first future nominal tick. This preserves one late catch-up without + * replaying every missed occurrence. + */ +export function calculateNextSchedulableOccurrence({ + schedule, + timezone, + afterNominal, + now, + schedulePhase, + window, + cronSpreadEnabled, +}: { + schedule: string; + timezone: string | null; + afterNominal: Date; + now: Date; + schedulePhase: number; + window?: NormalizedScheduleWindow; + cronSpreadEnabled: boolean; +}): SchedulableOccurrence { + const occurrenceAt = ( + nominalAt: Date + ): Omit => { + const nextNominalAt = calculateNextNominalTimestamp(schedule, timezone, nominalAt); + const { effectiveAt: candidateEffectiveAt, ...timing } = calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt, + schedulePhase, + window, + }); + + return { + ...timing, + candidateEffectiveAt, + effectiveAt: cronSpreadEnabled ? candidateEffectiveAt : nominalAt, + }; + }; + + const firstNominalAt = calculateNextNominalTimestamp(schedule, timezone, afterNominal); + const firstOccurrence = occurrenceAt(firstNominalAt); + + if (firstOccurrence.effectiveAt.getTime() >= now.getTime()) { + return { ...firstOccurrence, skippedExpiredOccurrences: false }; + } + + // `prev()` is strictly before its current date. Advancing by one millisecond includes a cron + // tick exactly at `now`, whose effective time may still be upcoming. + const latestNominalAt = previousScheduledTimestamp( + schedule, + timezone, + new Date(now.getTime() + 1) + ); + + if (latestNominalAt.getTime() > afterNominal.getTime()) { + const latestOccurrence = occurrenceAt(latestNominalAt); + + if (latestOccurrence.effectiveAt.getTime() >= now.getTime()) { + return { ...latestOccurrence, skippedExpiredOccurrences: true }; + } + } + + const nextOccurrence = occurrenceAt(calculateNextNominalTimestamp(schedule, timezone, now)); + return { ...nextOccurrence, skippedExpiredOccurrences: true }; +} + /** * Cron's previous slot relative to `fromTimestamp`. For a continuously- * running schedule this equals the actual last fire time; for paused or @@ -59,11 +148,7 @@ export function nextScheduledTimestamps( let nextScheduledTimestamp = lastScheduledTimestamp; for (let i = 0; i < count; i++) { - nextScheduledTimestamp = calculateNextScheduledTimestamp( - cron, - timezone, - nextScheduledTimestamp - ); + nextScheduledTimestamp = calculateNextNominalTimestamp(cron, timezone, nextScheduledTimestamp); result.push(nextScheduledTimestamp); } diff --git a/internal-packages/schedule-engine/src/engine/scheduleTiming.test.ts b/internal-packages/schedule-engine/src/engine/scheduleTiming.test.ts new file mode 100644 index 00000000000..e2fd7cfd00c --- /dev/null +++ b/internal-packages/schedule-engine/src/engine/scheduleTiming.test.ts @@ -0,0 +1,266 @@ +import { + MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS, + MAX_SCHEDULE_PHASE, + MINIMUM_SCHEDULE_RANGE_MS, + SCHEDULE_PHASE_DENOMINATOR, + calculateEffectiveScheduleTime, + calculateSchedulePhase, + parseScheduleWindow, + resolveScheduleWindowMs, + validateScheduleWindow, +} from "./scheduleTiming.js"; + +describe("parseScheduleWindow", () => { + it.each([ + ["30m", { type: "duration", durationSeconds: 1_800 }], + ["2h", { type: "duration", durationSeconds: 7_200 }], + ["24h", { type: "duration", durationSeconds: 86_400 }], + ["0m", { type: "duration", durationSeconds: 0 }], + ["0h", { type: "duration", durationSeconds: 0 }], + ["0%", { type: "percentage", percentage: 0 }], + ["12%", { type: "percentage", percentage: 12 }], + ["100%", { type: "percentage", percentage: 100 }], + ] as const)("normalizes %s", (input, expected) => { + expect(parseScheduleWindow(input)).toEqual(expected); + }); + + it.each([ + "", + "00m", + "01m", + "1.5h", + "0d", + "1d", + "25h", + "1441m", + "30s", + "0.01%", + "1.0%", + "12.3%", + "100.01%", + "101%", + "1.234%", + "1e2%", + " 30m", + "30m ", + ])("rejects %j", (input) => { + expect(() => parseScheduleWindow(input)).toThrow(); + }); + + it("rejects normalized durations over 24 hours", () => { + expect(() => + validateScheduleWindow({ + type: "duration", + durationSeconds: MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS + 1, + }) + ).toThrow("up to 24 hours"); + }); +}); + +describe("schedule window validation", () => { + it.each([0, 100])("allows %s percent", (percentage) => { + expect(() => validateScheduleWindow({ type: "percentage", percentage })).not.toThrow(); + }); + + it("allows a zero-duration window", () => { + expect(() => validateScheduleWindow({ type: "duration", durationSeconds: 0 })).not.toThrow(); + }); + + it.each([ + { type: "duration", durationSeconds: -1 }, + { type: "duration", durationSeconds: 1.5 }, + { type: "percentage", percentage: -100 }, + { type: "percentage", percentage: 101 }, + { type: "percentage", percentage: 1.5 }, + ] as const)("rejects an invalid normalized window: %o", (window) => { + expect(() => validateScheduleWindow(window)).toThrow(); + }); +}); + +describe("resolveScheduleWindowMs", () => { + it("returns zero when no window was configured", () => { + expect(resolveScheduleWindowMs(undefined, 5 * 60_000)).toBe(0); + }); + + it("resolves percentage windows using integer arithmetic", () => { + expect(resolveScheduleWindowMs({ type: "percentage", percentage: 33 }, 5 * 60_000)).toBe( + 99_000 + ); + }); +}); + +describe("calculateEffectiveScheduleTime", () => { + const nominalAt = new Date("2026-08-10T10:00:00.000Z"); + + it("uses the 60-second baseline when no window was configured", () => { + const timing = calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt: new Date("2026-08-10T10:05:00.000Z"), + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + }); + + expect(timing).toEqual({ + nominalAt, + nextNominalAt: new Date("2026-08-10T10:05:00.000Z"), + effectiveAt: new Date("2026-08-10T10:00:30.000Z"), + intervalMs: 300_000, + windowMs: 0, + effectiveRangeMs: MINIMUM_SCHEDULE_RANGE_MS, + offsetMs: 30_000, + windowWasCappedToInterval: false, + }); + }); + + it.each([ + [0, 0], + [10, 30_000], + ])("uses the 60-second baseline when %s percent resolves to %sms", (percentage, windowMs) => { + const timing = calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt: new Date("2026-08-10T10:05:00.000Z"), + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + window: { type: "percentage", percentage }, + }); + + expect(timing.windowMs).toBe(windowMs); + expect(timing.effectiveRangeMs).toBe(60_000); + expect(timing.offsetMs).toBe(30_000); + }); + + it("uses 30% of a five-minute interval", () => { + const timing = calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt: new Date("2026-08-10T10:05:00.000Z"), + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + window: { type: "percentage", percentage: 30 }, + }); + + expect(timing.windowMs).toBe(90_000); + expect(timing.effectiveRangeMs).toBe(90_000); + expect(timing.offsetMs).toBe(45_000); + expect(timing.effectiveAt).toEqual(new Date("2026-08-10T10:00:45.000Z")); + }); + + it("keeps a 100% window half-open at the maximum phase", () => { + const nextNominalAt = new Date("2026-08-10T10:05:00.000Z"); + const timing = calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt, + schedulePhase: MAX_SCHEDULE_PHASE, + window: { type: "percentage", percentage: 100 }, + }); + + expect(timing.effectiveRangeMs).toBe(300_000); + expect(timing.offsetMs).toBe(299_999); + expect(timing.effectiveAt).toEqual(new Date(nextNominalAt.getTime() - 1)); + expect(timing.effectiveAt.getTime()).toBeLessThan(nextNominalAt.getTime()); + }); + + it("preserves cadence for consecutive occurrences with a stable 100% phase", () => { + const phase = 1_610_612_735; + const first = calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt: new Date("2026-08-10T10:05:00.000Z"), + schedulePhase: phase, + window: { type: "percentage", percentage: 100 }, + }); + const second = calculateEffectiveScheduleTime({ + nominalAt: new Date("2026-08-10T10:05:00.000Z"), + nextNominalAt: new Date("2026-08-10T10:10:00.000Z"), + schedulePhase: phase, + window: { type: "percentage", percentage: 100 }, + }); + + expect(second.effectiveAt.getTime() - first.effectiveAt.getTime()).toBe(5 * 60_000); + }); + + it("allows an effective time to cross a calendar boundary", () => { + const timing = calculateEffectiveScheduleTime({ + nominalAt: new Date("2026-12-31T23:00:00.000Z"), + nextNominalAt: new Date("2027-01-01T23:00:00.000Z"), + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + window: { type: "duration", durationSeconds: 3 * 60 * 60 }, + }); + + expect(timing.effectiveAt).toEqual(new Date("2027-01-01T00:30:00.000Z")); + }); + + it("caps an absolute window at the interval to the next nominal tick", () => { + const timing = calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt: new Date("2026-08-10T10:05:00.000Z"), + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + window: { type: "duration", durationSeconds: 30 * 60 }, + }); + + expect(timing.windowMs).toBe(1_800_000); + expect(timing.effectiveRangeMs).toBe(300_000); + expect(timing.windowWasCappedToInterval).toBe(true); + expect(timing.effectiveAt).toEqual(new Date("2026-08-10T10:02:30.000Z")); + }); + + it.each([-1, 1.5, SCHEDULE_PHASE_DENOMINATOR])( + "rejects invalid schedule phase %s", + (schedulePhase) => { + expect(() => + calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt: new Date("2026-08-10T10:05:00.000Z"), + schedulePhase, + }) + ).toThrow("Schedule phase must be an integer"); + } + ); + + it("rejects a non-positive nominal interval", () => { + expect(() => + calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt: nominalAt, + schedulePhase: 0, + }) + ).toThrow("Nominal schedule interval must be a positive integer"); + }); +}); + +describe("calculateSchedulePhase", () => { + const input = { + secret: "test-secret", + environmentId: "env_789", + deduplicationKey: "daily-report", + }; + + it("uses the agreed domain-separated HMAC input", () => { + expect(calculateSchedulePhase(input)).toBe(43_063_717); + }); + + it("is stable for the same logical schedule instance", () => { + expect(calculateSchedulePhase(input)).toBe(calculateSchedulePhase(input)); + }); + + it.each(["environmentId", "deduplicationKey"] as const)("changes when %s changes", (field) => { + expect(calculateSchedulePhase({ ...input, [field]: `${input[field]}_other` })).not.toBe( + calculateSchedulePhase(input) + ); + }); + + it("changes when the secret changes", () => { + expect(calculateSchedulePhase({ ...input, secret: "other-secret" })).not.toBe( + calculateSchedulePhase(input) + ); + }); + + it("always returns a non-negative signed 31-bit integer", () => { + for (let index = 0; index < 1_000; index++) { + const phase = calculateSchedulePhase({ ...input, deduplicationKey: `schedule-${index}` }); + expect(phase).toBeGreaterThanOrEqual(0); + expect(phase).toBeLessThan(SCHEDULE_PHASE_DENOMINATOR); + } + }); + + it("rejects an empty secret", () => { + expect(() => calculateSchedulePhase({ ...input, secret: "" })).toThrow( + "secret must not be empty" + ); + }); +}); diff --git a/internal-packages/schedule-engine/src/engine/scheduleTiming.ts b/internal-packages/schedule-engine/src/engine/scheduleTiming.ts new file mode 100644 index 00000000000..003240e9524 --- /dev/null +++ b/internal-packages/schedule-engine/src/engine/scheduleTiming.ts @@ -0,0 +1,198 @@ +import { createHmac } from "node:crypto"; + +export const SCHEDULE_PHASE_DENOMINATOR = 2_147_483_648; +export const MAX_SCHEDULE_PHASE = SCHEDULE_PHASE_DENOMINATOR - 1; +export const MINIMUM_SCHEDULE_RANGE_MS = 60_000; +export const MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS = 24 * 60 * 60; + +const PERCENTAGE_DENOMINATOR = 100; + +export type NormalizedScheduleWindow = + | { type: "duration"; durationSeconds: number } + | { type: "percentage"; percentage: number }; + +export type SchedulePhaseInput = { + secret: string | Buffer; + environmentId: string; + deduplicationKey: string; +}; + +export type EffectiveScheduleTime = { + nominalAt: Date; + nextNominalAt: Date; + effectiveAt: Date; + intervalMs: number; + windowMs: number; + effectiveRangeMs: number; + offsetMs: number; + windowWasCappedToInterval: boolean; +}; + +/** + * Parses the public schedule-window syntax. + * + * Durations are non-negative whole minutes or hours up to 24 hours. + * Percentages are whole numbers from 0% through 100%. + */ +export function parseScheduleWindow(value: string): NormalizedScheduleWindow { + const durationMatch = /^(0|[1-9]\d*)([mh])$/.exec(value); + + if (durationMatch) { + const amount = Number(durationMatch[1]); + const unit = durationMatch[2] as "m" | "h"; + const unitSeconds = unit === "m" ? 60 : 3_600; + const durationSeconds = amount * unitSeconds; + + if ( + !Number.isSafeInteger(durationSeconds) || + durationSeconds > MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS + ) { + throw new RangeError("Schedule window duration cannot exceed 24 hours"); + } + + return { type: "duration", durationSeconds }; + } + + const percentageMatch = /^(0|[1-9]\d?|100)%$/.exec(value); + + if (percentageMatch) { + return { type: "percentage", percentage: Number(percentageMatch[1]) }; + } + + throw new TypeError( + 'Schedule window must be a whole duration such as "0m", "30m", or "24h", or a percentage such as "30%"' + ); +} + +export function validateScheduleWindow(window: NormalizedScheduleWindow): void { + if (window.type === "duration") { + if ( + !Number.isSafeInteger(window.durationSeconds) || + window.durationSeconds < 0 || + window.durationSeconds > MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS + ) { + throw new RangeError( + "Schedule window duration must be a non-negative integer up to 24 hours" + ); + } + + return; + } + + if ( + !Number.isInteger(window.percentage) || + window.percentage < 0 || + window.percentage > PERCENTAGE_DENOMINATOR + ) { + throw new RangeError( + "Schedule window percentage must be a whole percentage from 0% through 100%" + ); + } +} + +export function resolveScheduleWindowMs( + window: NormalizedScheduleWindow | undefined, + intervalMs: number +): number { + assertPositiveInterval(intervalMs); + + if (!window) { + return 0; + } + + validateScheduleWindow(window); + + if (window.type === "duration") { + return window.durationSeconds * 1_000; + } + + return Number((BigInt(intervalMs) * BigInt(window.percentage)) / BigInt(PERCENTAGE_DENOMINATOR)); +} + +/** + * Calculates the stable effective time for one nominal occurrence using integer arithmetic. + * + * An absolute window is a maximum. Each occurrence caps it at the interval to its next nominal + * tick, guaranteeing that the effective time never reaches or passes the next occurrence. + */ +export function calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt, + schedulePhase, + window, +}: { + nominalAt: Date; + nextNominalAt: Date; + schedulePhase: number; + window?: NormalizedScheduleWindow; +}): EffectiveScheduleTime { + assertValidDate(nominalAt, "nominalAt"); + assertValidDate(nextNominalAt, "nextNominalAt"); + assertValidSchedulePhase(schedulePhase); + + const intervalMs = nextNominalAt.getTime() - nominalAt.getTime(); + assertPositiveInterval(intervalMs); + + const windowMs = resolveScheduleWindowMs(window, intervalMs); + const requestedRangeMs = Math.max(MINIMUM_SCHEDULE_RANGE_MS, windowMs); + const effectiveRangeMs = Math.min(intervalMs, requestedRangeMs); + const windowWasCappedToInterval = effectiveRangeMs !== requestedRangeMs; + const offsetMs = Number( + (BigInt(schedulePhase) * BigInt(effectiveRangeMs)) / BigInt(SCHEDULE_PHASE_DENOMINATOR) + ); + const effectiveAtMs = nominalAt.getTime() + offsetMs; + + if (!Number.isSafeInteger(effectiveAtMs)) { + throw new RangeError("Calculated effective schedule time is outside the safe date range"); + } + + return { + nominalAt, + nextNominalAt, + effectiveAt: new Date(effectiveAtMs), + intervalMs, + windowMs, + effectiveRangeMs, + offsetMs, + windowWasCappedToInterval, + }; +} + +/** Calculates the durable, domain-separated phase stored on a schedule instance. */ +export function calculateSchedulePhase({ + secret, + environmentId, + deduplicationKey, +}: SchedulePhaseInput): number { + if ( + (typeof secret === "string" && secret.length === 0) || + (Buffer.isBuffer(secret) && !secret.length) + ) { + throw new RangeError("Schedule phase secret must not be empty"); + } + + const input = JSON.stringify(["cron-phase-v1", environmentId, deduplicationKey]); + const digest = createHmac("sha256", secret).update(input).digest(); + + return digest.readUInt32BE(0) & MAX_SCHEDULE_PHASE; +} + +function assertValidSchedulePhase(schedulePhase: number): void { + if (!Number.isInteger(schedulePhase) || schedulePhase < 0 || schedulePhase > MAX_SCHEDULE_PHASE) { + throw new RangeError(`Schedule phase must be an integer from 0 to ${MAX_SCHEDULE_PHASE}`); + } +} + +function assertPositiveInterval(intervalMs: number): void { + if (!Number.isSafeInteger(intervalMs) || intervalMs <= 0) { + throw new RangeError( + "Nominal schedule interval must be a positive integer number of milliseconds" + ); + } +} + +function assertValidDate(value: Date, name: string): void { + if (!Number.isFinite(value.getTime())) { + throw new RangeError(`${name} must be a valid date`); + } +} diff --git a/internal-packages/schedule-engine/src/engine/types.ts b/internal-packages/schedule-engine/src/engine/types.ts index bf0aeab4d76..58e089dab03 100644 --- a/internal-packages/schedule-engine/src/engine/types.ts +++ b/internal-packages/schedule-engine/src/engine/types.ts @@ -21,7 +21,8 @@ export type TriggerScheduledTaskParams = { }; scheduleInstanceId: string; scheduleId: string; - exactScheduleTime?: Date; + exactScheduleTime: Date; + effectiveScheduleTime: Date; }; export type TriggerScheduledTaskErrorType = "QUEUE_LIMIT" | "OUT_OF_ENTITLEMENTS" | "SYSTEM_ERROR"; @@ -50,6 +51,14 @@ export interface ScheduleEngineOptions { distributionWindow?: { seconds: number; }; + schedulePhaseSecret: string | Buffer; + /** + * Fraction of schedules (0 to 1) with cron spread active, gated on each + * schedule's deterministic phase. 0 disables spreading entirely; 1 enables + * it for every schedule. Raising the fraction is strictly additive — phases + * are stable, so a schedule never leaves the rollout once included. + */ + cronSpreadFraction: number; tracer?: Tracer; meter?: Meter; onTriggerScheduledTask: TriggerScheduledTaskCallback; @@ -74,15 +83,16 @@ export interface TriggerScheduleParams { instanceId: string; finalAttempt: boolean; exactScheduleTime?: Date; + effectiveScheduleTime?: Date; lastScheduleTime?: Date; } export interface RegisterScheduleInstanceParams { instanceId: string; /** - * Anchor for computing the next cron slot. Defaults to now() when omitted. - * This advances on every tick (fired or skipped) so the next slot keeps - * marching forward regardless of skip reasons. + * Nominal anchor for selecting the next non-expired cron occurrence. Defaults + * to now() when omitted. The engine advances from this timestamp when the + * next occurrence is still eligible and skips expired intermediate ticks. */ fromTimestamp?: Date; /** @@ -92,4 +102,9 @@ export interface RegisterScheduleInstanceParams { * disconnected, etc.) do NOT advance this — only real fires do. */ lastScheduleTime?: Date; + /** + * Keep an existing stable-ID Redis job unchanged, while still creating it + * when missing. Intended for no-op reconciliation of unchanged schedules. + */ + preserveExistingJob?: boolean; } diff --git a/internal-packages/schedule-engine/src/engine/workerCatalog.test.ts b/internal-packages/schedule-engine/src/engine/workerCatalog.test.ts new file mode 100644 index 00000000000..2d44dc1aad9 --- /dev/null +++ b/internal-packages/schedule-engine/src/engine/workerCatalog.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from "vitest"; +import { scheduleWorkerCatalog } from "./workerCatalog.js"; + +const schema = scheduleWorkerCatalog["schedule.triggerScheduledTask"].schema; + +describe("scheduleWorkerCatalog", () => { + it("accepts legacy payloads without an effective schedule time", () => { + const exactScheduleTime = "2026-08-11T10:00:00.000Z"; + + const payload = schema.parse({ + instanceId: "instance_123", + exactScheduleTime, + }); + + expect(payload.exactScheduleTime).toEqual(new Date(exactScheduleTime)); + expect(payload.effectiveScheduleTime).toBeUndefined(); + }); + + it("coerces nominal and effective schedule times for new payloads", () => { + const exactScheduleTime = "2026-08-11T10:00:00.000Z"; + const effectiveScheduleTime = "2026-08-11T10:00:42.123Z"; + + const payload = schema.parse({ + instanceId: "instance_123", + exactScheduleTime, + effectiveScheduleTime, + }); + + expect(payload.exactScheduleTime).toEqual(new Date(exactScheduleTime)); + expect(payload.effectiveScheduleTime).toEqual(new Date(effectiveScheduleTime)); + }); +}); diff --git a/internal-packages/schedule-engine/src/engine/workerCatalog.ts b/internal-packages/schedule-engine/src/engine/workerCatalog.ts index c960f458f88..e351ed00dd0 100644 --- a/internal-packages/schedule-engine/src/engine/workerCatalog.ts +++ b/internal-packages/schedule-engine/src/engine/workerCatalog.ts @@ -4,7 +4,12 @@ export const scheduleWorkerCatalog = { "schedule.triggerScheduledTask": { schema: z.object({ instanceId: z.string(), + // The nominal cron occurrence. Keep this field name for compatibility + // with jobs enqueued before effective schedule times were introduced. exactScheduleTime: z.coerce.date(), + // Optional for compatibility with in-flight jobs. Missing means the + // effective time is the nominal exactScheduleTime. + effectiveScheduleTime: z.coerce.date().optional(), // Optional for backward compat with in-flight jobs enqueued by older // engines. After deploy, every newly-enqueued job populates this with // the just-fired schedule time so the next dequeue can report diff --git a/internal-packages/schedule-engine/src/index.ts b/internal-packages/schedule-engine/src/index.ts index 6c96f2cd54b..dcffbf17428 100644 --- a/internal-packages/schedule-engine/src/index.ts +++ b/internal-packages/schedule-engine/src/index.ts @@ -1,4 +1,21 @@ export { ScheduleEngine } from "./engine/index.js"; +export { calculateNextNominalTimestamp } from "./engine/scheduleCalculation.js"; +export { + MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS, + MAX_SCHEDULE_PHASE, + MINIMUM_SCHEDULE_RANGE_MS, + SCHEDULE_PHASE_DENOMINATOR, + calculateEffectiveScheduleTime, + calculateSchedulePhase, + parseScheduleWindow, + resolveScheduleWindowMs, + validateScheduleWindow, +} from "./engine/scheduleTiming.js"; +export type { + EffectiveScheduleTime, + NormalizedScheduleWindow, + SchedulePhaseInput, +} from "./engine/scheduleTiming.js"; export type { ScheduleEngineOptions, TriggerScheduleParams, diff --git a/internal-packages/schedule-engine/test/scheduleEngine.test.ts b/internal-packages/schedule-engine/test/scheduleEngine.test.ts index 59598988338..4e957d10498 100644 --- a/internal-packages/schedule-engine/test/scheduleEngine.test.ts +++ b/internal-packages/schedule-engine/test/scheduleEngine.test.ts @@ -22,6 +22,8 @@ describe("ScheduleEngine Integration", () => { prisma, redis: redisOptions, distributionWindow: { seconds: 10 }, + schedulePhaseSecret: "test-schedule-phase-secret", + cronSpreadFraction: 1, worker: { concurrency: 1, disabled: false, // Enable worker for full integration test @@ -95,6 +97,9 @@ describe("ScheduleEngine Integration", () => { environmentId: environment.id, projectId: project.id, active: true, + // Keep the lifecycle test fast and deterministic. Non-zero phase + // behavior is covered by the focused registration tests. + schedulePhase: 0, }, }); @@ -209,6 +214,7 @@ describe("ScheduleEngine Integration", () => { scheduleInstanceId: scheduleInstance.id, scheduleId: taskSchedule.id, exactScheduleTime: firstScheduledTime, + effectiveScheduleTime: firstScheduledTime, }); // Verify the second execution parameters @@ -232,6 +238,7 @@ describe("ScheduleEngine Integration", () => { scheduleInstanceId: scheduleInstance.id, scheduleId: taskSchedule.id, exactScheduleTime: secondScheduledTime, + effectiveScheduleTime: secondScheduledTime, }); } finally { // Clean up: stop the worker diff --git a/internal-packages/schedule-engine/test/scheduleEngine2.test.ts b/internal-packages/schedule-engine/test/scheduleEngine2.test.ts index 94274673b04..32767cb098c 100644 --- a/internal-packages/schedule-engine/test/scheduleEngine2.test.ts +++ b/internal-packages/schedule-engine/test/scheduleEngine2.test.ts @@ -2,7 +2,14 @@ import { containerTest } from "@internal/testcontainers"; import { trace } from "@internal/tracing"; import { describe, expect, vi } from "vitest"; import type { TriggerScheduledTaskParams } from "../src/engine/types.js"; -import { ScheduleEngine } from "../src/index.js"; +import { + calculateEffectiveScheduleTime, + calculateNextNominalTimestamp, + calculateSchedulePhase, + SCHEDULE_PHASE_DENOMINATOR, + ScheduleEngine, +} from "../src/index.js"; +import { calculateDistributedExecutionTime } from "../src/engine/distributedScheduling.js"; describe("ScheduleEngine Integration (part 2)", () => { // Deploy-moment backward compatibility. At deploy time, in-flight Redis jobs @@ -20,6 +27,8 @@ describe("ScheduleEngine Integration (part 2)", () => { prisma, redis: redisOptions, distributionWindow: { seconds: 10 }, + schedulePhaseSecret: "test-schedule-phase-secret", + cronSpreadFraction: 0, worker: { concurrency: 1, disabled: true, // Don't actually run the worker — calling triggerScheduledTask directly @@ -72,6 +81,7 @@ describe("ScheduleEngine Integration (part 2)", () => { type: "DECLARATIVE", active: true, externalId: "legacy-ext", + windowDurationSeconds: 60, }, }); @@ -88,24 +98,395 @@ describe("ScheduleEngine Integration (part 2)", () => { }, }); - // Call triggerScheduledTask directly without lastScheduleTime, - // simulating an in-flight Redis job enqueued by the old engine. + // Call triggerScheduledTask directly without lastScheduleTime or an + // effective time, simulating an in-flight Redis job from the old engine. const exactScheduleTime = new Date("2026-04-30T10:05:00.000Z"); + const beforeTrigger = new Date(); await engine.triggerScheduledTask({ instanceId: scheduleInstance.id, finalAttempt: false, exactScheduleTime, - // lastScheduleTime intentionally omitted — legacy payload shape + // effectiveScheduleTime and lastScheduleTime intentionally omitted }); expect(triggerCalls.length).toBe(1); expect(triggerCalls[0].payload.timestamp).toEqual(exactScheduleTime); + expect(triggerCalls[0].exactScheduleTime).toEqual(exactScheduleTime); + expect(triggerCalls[0].effectiveScheduleTime).toEqual(exactScheduleTime); // Falls back to instance.lastScheduledTimestamp from the DB rather // than reporting undefined for this one transitional fire. expect(triggerCalls[0].payload.lastTimestamp).toEqual(preDeployLastFire); + expect(triggerCalls[0].payload.upcoming).toHaveLength(10); + expect( + triggerCalls[0].payload.upcoming.every( + (timestamp) => timestamp.getTime() > beforeTrigger.getTime() + ) + ).toBe(true); + + const nextJob = await engine.getJob(`scheduled-task-instance:${scheduleInstance.id}`); + const nextJobPayload = nextJob!.item as unknown as { + exactScheduleTime: string; + effectiveScheduleTime: string; + }; + const nextNominalAt = new Date(nextJobPayload.exactScheduleTime); + + // The legacy occurrence fires once, then expired intermediate ticks are + // skipped instead of being replayed. With spread disabled, eligibility + // remains nominal and the next job is in the future. + expect(nextNominalAt.getTime()).toBeGreaterThan(beforeTrigger.getTime()); + expect(new Date(nextJobPayload.effectiveScheduleTime)).toEqual(nextNominalAt); + expect(nextJob!.timestamp).toEqual( + calculateDistributedExecutionTime(nextNominalAt, 10, scheduleInstance.id) + ); + expect(new Date((nextJob!.item as { lastScheduleTime: string }).lastScheduleTime)).toEqual( + exactScheduleTime + ); + + const updatedInstance = await prisma.taskScheduleInstance.findUniqueOrThrow({ + where: { id: scheduleInstance.id }, + select: { schedulePhase: true }, + }); + expect(updatedInstance.schedulePhase).toBeNull(); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "should assign a stable schedule phase once when spreading is active", + { timeout: 30_000 }, + async ({ prisma, redisOptions }) => { + const schedulePhaseSecret = "test-schedule-phase-secret"; + const triggerCalls: TriggerScheduledTaskParams[] = []; + const engine = new ScheduleEngine({ + prisma, + redis: redisOptions, + distributionWindow: { seconds: 10 }, + schedulePhaseSecret, + cronSpreadFraction: 1, + worker: { + concurrency: 1, + disabled: true, + pollIntervalMs: 1000, + }, + tracer: trace.getTracer("test", "0.0.0"), + onTriggerScheduledTask: async (params) => { + triggerCalls.push(params); + return { success: true }; + }, + isDevEnvironmentConnectedHandler: vi.fn().mockResolvedValue(true), + }); + + try { + const organization = await prisma.organization.create({ + data: { title: "Schedule Phase Org", slug: "schedule-phase-org" }, + }); + const project = await prisma.project.create({ + data: { + name: "Schedule Phase Project", + slug: "schedule-phase-project", + externalRef: "schedule-phase-ref", + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + slug: "schedule-phase-env", + type: "PRODUCTION", + projectId: project.id, + organizationId: organization.id, + apiKey: "tr_schedule_phase", + pkApiKey: "pk_schedule_phase", + shortcode: "phase", + }, + }); + const taskSchedule = await prisma.taskSchedule.create({ + data: { + friendlyId: "sched_phase", + taskIdentifier: "schedule-phase-task", + projectId: project.id, + deduplicationKey: "schedule-phase-dedup", + generatorExpression: "*/5 * * * *", + generatorDescription: "Every 5 minutes", + timezone: "UTC", + type: "DECLARATIVE", + }, + }); + const scheduleInstance = await prisma.taskScheduleInstance.create({ + data: { + taskScheduleId: taskSchedule.id, + environmentId: environment.id, + projectId: project.id, + }, + }); + + // Atomic preserve mode still creates the stable-ID job when it is missing. + await engine.registerNextTaskScheduleInstance({ + instanceId: scheduleInstance.id, + preserveExistingJob: true, + }); + + const unwindowedInstance = await prisma.taskScheduleInstance.findUniqueOrThrow({ + where: { id: scheduleInstance.id }, + select: { schedulePhase: true }, + }); + expect(unwindowedInstance.schedulePhase).toBe( + calculateSchedulePhase({ + secret: schedulePhaseSecret, + environmentId: environment.id, + deduplicationKey: taskSchedule.deduplicationKey, + }) + ); + + const unwindowedJob = await engine.getJob(`scheduled-task-instance:${scheduleInstance.id}`); + const unwindowedPayload = unwindowedJob!.item as unknown as { + exactScheduleTime: string; + effectiveScheduleTime: string; + }; + const unwindowedNominalAt = new Date(unwindowedPayload.exactScheduleTime); + const unwindowedNextNominalAt = calculateNextNominalTimestamp( + taskSchedule.generatorExpression, + taskSchedule.timezone, + unwindowedNominalAt + ); + const unwindowedPhase = calculateSchedulePhase({ + secret: schedulePhaseSecret, + environmentId: environment.id, + deduplicationKey: taskSchedule.deduplicationKey, + }); + const { effectiveAt: unwindowedEffectiveAt } = calculateEffectiveScheduleTime({ + nominalAt: unwindowedNominalAt, + nextNominalAt: unwindowedNextNominalAt, + schedulePhase: unwindowedPhase, + }); + expect(new Date(unwindowedPayload.effectiveScheduleTime)).toEqual(unwindowedEffectiveAt); + expect(unwindowedJob!.timestamp).toEqual( + calculateDistributedExecutionTime(unwindowedEffectiveAt, 10, scheduleInstance.id) + ); + + await prisma.taskSchedule.update({ + where: { id: taskSchedule.id }, + data: { windowDurationSeconds: 60 }, + }); + + const expectedPhase = calculateSchedulePhase({ + secret: schedulePhaseSecret, + environmentId: environment.id, + deduplicationKey: taskSchedule.deduplicationKey, + }); + + await Promise.all([ + engine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id }), + engine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id }), + engine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id }), + ]); + + const assignedInstance = await prisma.taskScheduleInstance.findUniqueOrThrow({ + where: { id: scheduleInstance.id }, + select: { schedulePhase: true }, + }); + expect(assignedInstance.schedulePhase).toBe(expectedPhase); + + const pinnedPhase = 1_234_567_890; + await prisma.taskScheduleInstance.update({ + where: { id: scheduleInstance.id }, + data: { schedulePhase: pinnedPhase }, + }); + + await engine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id }); + + const preservedInstance = await prisma.taskScheduleInstance.findUniqueOrThrow({ + where: { id: scheduleInstance.id }, + select: { schedulePhase: true }, + }); + expect(preservedInstance.schedulePhase).toBe(pinnedPhase); + + const pendingBeforeNoop = await engine.getJob( + `scheduled-task-instance:${scheduleInstance.id}` + ); + + // No-op reconciliation preserves the existing payload and score atomically. + await engine.registerNextTaskScheduleInstance({ + instanceId: scheduleInstance.id, + preserveExistingJob: true, + }); + const pendingAfterNoop = await engine.getJob( + `scheduled-task-instance:${scheduleInstance.id}` + ); + expect(pendingAfterNoop).toEqual(pendingBeforeNoop); + + await prisma.taskSchedule.update({ + where: { id: taskSchedule.id }, + data: { windowDurationSeconds: 120 }, + }); + + // Normal registration still replaces the job when timing changed. + await engine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id }); + const pendingAfterTimingChange = await engine.getJob( + `scheduled-task-instance:${scheduleInstance.id}` + ); + expect(pendingAfterTimingChange).not.toEqual(pendingBeforeNoop); + + const intervalMs = 5 * 60_000; + const exactScheduleTime = new Date(Math.floor(Date.now() / intervalMs) * intervalMs); + const effectiveScheduleTime = new Date(exactScheduleTime.getTime() + 45_000); + await engine.triggerScheduledTask({ + instanceId: scheduleInstance.id, + finalAttempt: false, + exactScheduleTime, + effectiveScheduleTime, + }); + + expect(triggerCalls).toHaveLength(1); + expect(triggerCalls[0].payload.timestamp).toEqual(exactScheduleTime); + expect(triggerCalls[0].exactScheduleTime).toEqual(exactScheduleTime); + expect(triggerCalls[0].effectiveScheduleTime).toEqual(effectiveScheduleTime); + + const nextJob = await engine.getJob(`scheduled-task-instance:${scheduleInstance.id}`); + const nextJobPayload = nextJob!.item as unknown as { + exactScheduleTime: string; + effectiveScheduleTime: string; + }; + const nextNominalAt = new Date(exactScheduleTime.getTime() + intervalMs); + const followingNominalAt = new Date(nextNominalAt.getTime() + intervalMs); + const { effectiveAt: nextEffectiveAt } = calculateEffectiveScheduleTime({ + nominalAt: nextNominalAt, + nextNominalAt: followingNominalAt, + schedulePhase: pinnedPhase, + window: { type: "duration", durationSeconds: 120 }, + }); + + expect(new Date(nextJobPayload.exactScheduleTime)).toEqual(nextNominalAt); + expect(new Date(nextJobPayload.effectiveScheduleTime)).toEqual(nextEffectiveAt); + expect(nextJob!.timestamp).toEqual( + calculateDistributedExecutionTime(nextEffectiveAt, 10, scheduleInstance.id) + ); } finally { await engine.quit(); } } ); + + containerTest( + "gates cron spread per schedule via the rollout fraction", + { timeout: 30_000 }, + async ({ prisma, redisOptions }) => { + const schedulePhaseSecret = "test-schedule-phase-secret"; + + const organization = await prisma.organization.create({ + data: { title: "Spread Fraction Org", slug: "spread-fraction-org" }, + }); + const project = await prisma.project.create({ + data: { + name: "Spread Fraction Project", + slug: "spread-fraction-project", + externalRef: "spread-fraction-ref", + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + slug: "spread-fraction-env", + type: "PRODUCTION", + projectId: project.id, + organizationId: organization.id, + apiKey: "tr_spread_fraction", + pkApiKey: "pk_spread_fraction", + shortcode: "spread", + }, + }); + const taskSchedule = await prisma.taskSchedule.create({ + data: { + friendlyId: "sched_spread_fraction", + taskIdentifier: "spread-fraction-task", + projectId: project.id, + deduplicationKey: "spread-fraction-dedup", + generatorExpression: "*/5 * * * *", + generatorDescription: "Every 5 minutes", + timezone: "UTC", + type: "DECLARATIVE", + }, + }); + const scheduleInstance = await prisma.taskScheduleInstance.create({ + data: { + taskScheduleId: taskSchedule.id, + environmentId: environment.id, + projectId: project.id, + }, + }); + + const phase = calculateSchedulePhase({ + secret: schedulePhaseSecret, + environmentId: environment.id, + deduplicationKey: taskSchedule.deduplicationKey, + }); + + // The gate is `phase < fraction * DENOMINATOR`. Dividing and multiplying + // by 2^31 is exact in floating point, so `phase / DENOMINATOR` excludes + // this schedule and `(phase + 1) / DENOMINATOR` includes it. + const excludingFraction = phase / SCHEDULE_PHASE_DENOMINATOR; + const includingFraction = (phase + 1) / SCHEDULE_PHASE_DENOMINATOR; + + const createEngine = (cronSpreadFraction: number) => + new ScheduleEngine({ + prisma, + redis: redisOptions, + distributionWindow: { seconds: 10 }, + schedulePhaseSecret, + cronSpreadFraction, + worker: { + concurrency: 1, + disabled: true, + pollIntervalMs: 1000, + }, + tracer: trace.getTracer("test", "0.0.0"), + onTriggerScheduledTask: async () => ({ success: true }), + isDevEnvironmentConnectedHandler: vi.fn().mockResolvedValue(true), + }); + + const jobId = `scheduled-task-instance:${scheduleInstance.id}`; + + const excludedEngine = createEngine(excludingFraction); + try { + await excludedEngine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id }); + const job = await excludedEngine.getJob(jobId); + const payload = job!.item as unknown as { + exactScheduleTime: string; + effectiveScheduleTime: string; + }; + // Spread inactive: the effective time is the nominal tick. + expect(new Date(payload.effectiveScheduleTime)).toEqual( + new Date(payload.exactScheduleTime) + ); + } finally { + await excludedEngine.quit(); + } + + const includedEngine = createEngine(includingFraction); + try { + await includedEngine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id }); + const job = await includedEngine.getJob(jobId); + const payload = job!.item as unknown as { + exactScheduleTime: string; + effectiveScheduleTime: string; + }; + const nominalAt = new Date(payload.exactScheduleTime); + const nextNominalAt = calculateNextNominalTimestamp( + taskSchedule.generatorExpression, + taskSchedule.timezone, + nominalAt + ); + // Spread active with no window configured: the 60s baseline applies. + const { effectiveAt } = calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt, + schedulePhase: phase, + }); + expect(new Date(payload.effectiveScheduleTime)).toEqual(effectiveAt); + } finally { + await includedEngine.quit(); + } + } + ); }); diff --git a/internal-packages/schedule-engine/test/scheduleRecovery.test.ts b/internal-packages/schedule-engine/test/scheduleRecovery.test.ts index 518e0ff3eba..4d5f3bbc23a 100644 --- a/internal-packages/schedule-engine/test/scheduleRecovery.test.ts +++ b/internal-packages/schedule-engine/test/scheduleRecovery.test.ts @@ -16,6 +16,8 @@ describe("Schedule Recovery", () => { prisma, redis: redisOptions, distributionWindow: { seconds: 10 }, + schedulePhaseSecret: "test-schedule-phase-secret", + cronSpreadFraction: 1, worker: { concurrency: 1, disabled: true, // Disable worker to prevent automatic execution @@ -118,6 +120,8 @@ describe("Schedule Recovery", () => { prisma, redis: redisOptions, distributionWindow: { seconds: 10 }, + schedulePhaseSecret: "test-schedule-phase-secret", + cronSpreadFraction: 1, worker: { concurrency: 1, disabled: true, // Disable worker to prevent automatic execution @@ -223,6 +227,8 @@ describe("Schedule Recovery", () => { prisma, redis: redisOptions, distributionWindow: { seconds: 10 }, + schedulePhaseSecret: "test-schedule-phase-secret", + cronSpreadFraction: 1, worker: { concurrency: 1, disabled: true, // Disable worker to prevent automatic execution @@ -334,6 +340,8 @@ describe("Schedule Recovery", () => { prisma, redis: redisOptions, distributionWindow: { seconds: 10 }, + schedulePhaseSecret: "test-schedule-phase-secret", + cronSpreadFraction: 1, worker: { concurrency: 1, disabled: true, // Disable worker to prevent automatic execution @@ -404,6 +412,8 @@ describe("Schedule Recovery", () => { prisma, redis: redisOptions, distributionWindow: { seconds: 10 }, + schedulePhaseSecret: "test-schedule-phase-secret", + cronSpreadFraction: 1, worker: { concurrency: 1, disabled: true, pollIntervalMs: 1000 }, tracer: trace.getTracer("test", "0.0.0"), onTriggerScheduledTask: async () => ({ success: true }), @@ -505,6 +515,8 @@ describe("Schedule Recovery", () => { prisma, redis: redisOptions, distributionWindow: { seconds: 10 }, + schedulePhaseSecret: "test-schedule-phase-secret", + cronSpreadFraction: 1, worker: { concurrency: 1, disabled: true, pollIntervalMs: 1000 }, tracer: trace.getTracer("test", "0.0.0"), onTriggerScheduledTask: async () => ({ success: true }), diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index bf32ad74ebc..f8de04bb4fa 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -10,6 +10,7 @@ import { import { BackgroundWorkerMetadata } from "./resources.js"; import { DequeuedMessage, MachineResources } from "./runEngine.js"; import { QueueTypeName } from "./queues.js"; +import { ScheduleWindow } from "./schemas.js"; export const RunEngineVersion = z.union([z.literal("V1"), z.literal("V2")]); @@ -1045,6 +1046,13 @@ export const CreateScheduleOptions = z.object({ * */ timezone: z.string().optional(), + /** Optionally delay each occurrence by a stable amount within this window. + * Absolute windows use whole minutes or hours up to 24 hours and are capped at the next + * nominal interval. Percentages are relative to each nominal interval. + * + * @example "30m", "2h", "24h", "30%", "100%" + */ + window: ScheduleWindow.optional(), }); export type CreateScheduleOptions = z.infer; @@ -1070,6 +1078,7 @@ export const ScheduleObject = z.object({ externalId: z.string().nullish(), generator: ScheduleGenerator, timezone: z.string(), + window: ScheduleWindow.optional(), nextRun: z.coerce.date().nullish(), environments: z.array( z.object({ diff --git a/packages/core/src/v3/schemas/schemas.ts b/packages/core/src/v3/schemas/schemas.ts index 0b12e7ae3db..7e95224f429 100644 --- a/packages/core/src/v3/schemas/schemas.ts +++ b/packages/core/src/v3/schemas/schemas.ts @@ -174,10 +174,20 @@ export const QueueManifest = z.object({ export type QueueManifest = z.infer; +/** + * A delay window after a nominal cron tick. + * + * The server's schedule timing domain validates and normalizes the public syntax. + */ +export const ScheduleWindow = z.string().min(1); + +export type ScheduleWindow = z.infer; + export const ScheduleMetadata = z.object({ cron: z.string(), timezone: z.string(), environments: z.array(EnvironmentType).optional(), + window: ScheduleWindow.optional(), }); const AgentConfig = z.object({