diff --git a/.changeset/smooth-schedule-windows.md b/.changeset/smooth-schedule-windows.md new file mode 100644 index 0000000000..6e7bba2eb7 --- /dev/null +++ b/.changeset/smooth-schedule-windows.md @@ -0,0 +1,6 @@ +--- +"@trigger.dev/core": patch +"@trigger.dev/sdk": patch +--- + +Define stable execution windows on declarative scheduled tasks. Schedule API responses now expose both the nominal CRON time and its assigned time, while the dashboard shows configured windows and upcoming assignments. diff --git a/apps/webapp/app/components/runs/v3/TaskRunsTable.tsx b/apps/webapp/app/components/runs/v3/TaskRunsTable.tsx index 0019711785..ef0c6e25f5 100644 --- a/apps/webapp/app/components/runs/v3/TaskRunsTable.tsx +++ b/apps/webapp/app/components/runs/v3/TaskRunsTable.tsx @@ -418,13 +418,13 @@ export function TaskRunsTable({ {run.isPending ? ( "–" ) : run.startedAt ? ( - formatDuration(new Date(run.createdAt), new Date(run.startedAt), { + formatDuration(new Date(run.triggeredAt), new Date(run.startedAt), { style: "short", }) ) : run.isCancellable ? ( - + ) : ( - formatDuration(new Date(run.createdAt), new Date(run.updatedAt), { + formatDuration(new Date(run.triggeredAt), new Date(run.updatedAt), { style: "short", }) )} diff --git a/apps/webapp/app/components/schedules/ScheduleInspector.tsx b/apps/webapp/app/components/schedules/ScheduleInspector.tsx index 1d0de51dc9..b2a64450f1 100644 --- a/apps/webapp/app/components/schedules/ScheduleInspector.tsx +++ b/apps/webapp/app/components/schedules/ScheduleInspector.tsx @@ -55,13 +55,14 @@ export type ScheduleInspectorData = { cron: string; cronDescription: string; timezone: string; + window?: string; externalId: string | null; deduplicationKey: string | null; userProvidedDeduplicationKey: boolean; active: boolean; environments: EnvironmentRow[]; runs: RunRow[]; - nextRuns: Date[]; + nextRuns: Array<{ nominalAt: Date; effectiveAt: Date }>; }; type Props = { @@ -142,6 +143,10 @@ export function ScheduleInspector({ Timezone {schedule.timezone} + + Window + {schedule.window ?? "-"} + Environment @@ -210,11 +215,11 @@ export function ScheduleInspector({ {!isUtc && ( - + )} - + )) @@ -249,8 +254,8 @@ export function ScheduleInspector({ } panelClassName="max-w-full" > - You can only edit a declarative schedule by updating your schedules.task and then - running the CLI dev and deploy commands. + You can only edit a declarative schedule, including its window, by updating your + schedules.task and then running the CLI dev and deploy commands. )} diff --git a/apps/webapp/app/presenters/v3/EditSchedulePresenter.server.ts b/apps/webapp/app/presenters/v3/EditSchedulePresenter.server.ts index 610ebb24d2..3fd7fec8a6 100644 --- a/apps/webapp/app/presenters/v3/EditSchedulePresenter.server.ts +++ b/apps/webapp/app/presenters/v3/EditSchedulePresenter.server.ts @@ -6,6 +6,7 @@ import { filterOrphanedEnvironments } from "~/utils/environmentSort"; import { getTimezones } from "~/utils/timezones.server"; import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server"; import { ServiceValidationError } from "~/v3/services/baseService.server"; +import { formatScheduleWindow } from "~/v3/scheduleWindow.server"; type EditScheduleOptions = { userId: string; @@ -124,6 +125,8 @@ export class EditSchedulePresenter { deduplicationKey: true, userProvidedDeduplicationKey: true, timezone: true, + windowDurationSeconds: true, + windowPercentage: true, taskIdentifier: true, instances: { select: { @@ -144,6 +147,7 @@ export class EditSchedulePresenter { return { ...schedule, cron: schedule.generatorExpression, + window: formatScheduleWindow(schedule), environments: schedule.instances.flatMap((instance) => { const environment = possibleEnvironments.find((env) => env.id === instance.environmentId); if (!environment) { diff --git a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts index fe6499b8e1..c9a120334f 100644 --- a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts @@ -24,6 +24,7 @@ import { regionForDisplay } from "~/runEngine/concerns/workerQueueSplit.server"; import { machinePresetFromRun } from "~/v3/machinePresets.server"; import { ServiceValidationError } from "~/v3/services/baseService.server"; import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus"; +import { runTriggeredAt } from "~/v3/runTimestamps"; // Positive-only cache: only envs known to have runs are stored (empty envs are re-checked), // so "has runs" is monotonic and the TTL can be very long. Tiered memory + Redis. @@ -299,12 +300,14 @@ export class NextRunListPresenter { const hasFinished = isFinalRunStatus(run.status); const startedAt = run.startedAt ?? run.lockedAt; + const triggeredAt = runTriggeredAt(run); return { id: run.id, number: 1, friendlyId: run.friendlyId, createdAt: run.createdAt.toISOString(), + triggeredAt: triggeredAt.toISOString(), updatedAt: run.updatedAt.toISOString(), startedAt: startedAt ? startedAt.toISOString() : undefined, delayUntil: run.delayUntil ? run.delayUntil.toISOString() : undefined, diff --git a/apps/webapp/app/presenters/v3/RunPresenter.server.ts b/apps/webapp/app/presenters/v3/RunPresenter.server.ts index 18557e5173..8a99f3e60f 100644 --- a/apps/webapp/app/presenters/v3/RunPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/RunPresenter.server.ts @@ -11,6 +11,7 @@ import { env } from "~/env.server"; import { getEventRepositoryForStore } from "~/v3/eventRepository/index.server"; import { runStore } from "~/v3/runStore.server"; import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server"; +import { runTriggeredAt } from "~/v3/runTimestamps"; type Result = Awaited>; export type Run = Result["run"]; @@ -92,6 +93,8 @@ export class RunPresenter { friendlyId: true, status: true, startedAt: true, + queueTimestamp: true, + scheduleId: true, completedAt: true, logsDeletedAt: true, annotations: true, @@ -151,6 +154,7 @@ export class RunPresenter { } const showLogs = showDeletedLogs || !run.logsDeletedAt; + const triggeredAt = runTriggeredAt(run); const runData = { id: run.id, @@ -240,7 +244,7 @@ export class RunPresenter { message: run.taskIdentifier, style: { icon: "task", variant: "primary" }, events: [], - startTime: run.createdAt, + startTime: triggeredAt, duration: 0, isError: run.status === "COMPLETED_WITH_ERRORS" || @@ -364,7 +368,7 @@ export class RunPresenter { rootStartedAt: tree?.data.startTime, startedAt: run.startedAt, queuedDuration: run.startedAt - ? millisecondsToNanoseconds(run.startedAt.getTime() - run.createdAt.getTime()) + ? millisecondsToNanoseconds(run.startedAt.getTime() - triggeredAt.getTime()) : undefined, overridesBySpanId: traceSummary.overridesBySpanId, linkedRunIdBySpanId, diff --git a/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts b/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts index 22b9821bab..c0eb6fc7a3 100644 --- a/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts @@ -5,12 +5,10 @@ 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 { calculateNextScheduleRunTimes, formatScheduleWindow } from "~/v3/scheduleWindow.server"; import { CheckScheduleService } from "~/v3/services/checkSchedule.server"; -import { - calculateNextScheduledTimestampFromNow, - previousScheduledTimestamp, -} from "~/v3/utils/calculateNextSchedule.server"; +import { previousScheduledTimestamp } from "~/v3/utils/calculateNextSchedule.server"; +import { env } from "~/env.server"; import { BasePresenter } from "./basePresenter.server"; type ScheduleListOptions = { @@ -35,6 +33,7 @@ export type ScheduleListItem = { window?: string; externalId: string | null; nextRun: Date; + nextRunEffectiveAt: Date; lastRun: Date | undefined; active: boolean; environments: { @@ -223,6 +222,7 @@ export class ScheduleListPresenter extends BasePresenter { instances: { select: { environmentId: true, + schedulePhase: true, }, }, active: true, @@ -300,6 +300,23 @@ export class ScheduleListPresenter extends BasePresenter { } } + const instance = schedule.instances.find( + (instance) => instance.environmentId === environmentId + ); + if (!instance) { + throw new Error(`Schedule instance not found for environment: ${environmentId}`); + } + const [nextRun] = calculateNextScheduleRunTimes({ + cron: schedule.generatorExpression, + timezone: schedule.timezone, + deduplicationKey: schedule.deduplicationKey, + environmentId, + schedulePhase: instance.schedulePhase, + phaseSecret: env.ENCRYPTION_KEY, + windowDurationSeconds: schedule.windowDurationSeconds, + windowPercentage: schedule.windowPercentage, + }); + return { id: schedule.id, type: schedule.type, @@ -314,10 +331,8 @@ export class ScheduleListPresenter extends BasePresenter { active: schedule.active, externalId: schedule.externalId, lastRun, - nextRun: calculateNextScheduledTimestampFromNow( - schedule.generatorExpression, - schedule.timezone - ), + nextRun: nextRun.nominalAt, + nextRunEffectiveAt: nextRun.effectiveAt, environments: schedule.instances.map((instance) => { const environment = project.environments.find((env) => env.id === instance.environmentId); if (!environment) { diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts index b680a1c6a8..11b923566c 100644 --- a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts @@ -33,6 +33,7 @@ import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.se import { buildSyntheticSpanRun } from "~/v3/mollifier/syntheticSpanRun.server"; import { engine } from "~/v3/runEngine.server"; import { runStore } from "~/v3/runStore.server"; +import { runTriggeredAt } from "~/v3/runTimestamps"; import { getTaskEventStoreTableForRun, type TaskEventStoreTable } from "~/v3/taskEventStore.server"; import { isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus"; import { BasePresenter } from "./basePresenter.server"; @@ -387,7 +388,7 @@ export class SpanPresenter extends BasePresenter { friendlyId: run.friendlyId, status: run.status, statusReason: run.statusReason ?? undefined, - createdAt: run.createdAt, + createdAt: runTriggeredAt(run), startedAt: run.startedAt, executedAt: run.executedAt, updatedAt: run.updatedAt, @@ -559,6 +560,7 @@ export class SpanPresenter extends BasePresenter { startedAt: true, executedAt: true, createdAt: true, + queueTimestamp: true, updatedAt: true, queuedAt: true, completedAt: true, diff --git a/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts b/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts index bc7d0388b0..61716b7364 100644 --- a/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts @@ -3,10 +3,10 @@ import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; import { displayableEnvironment } from "~/models/runtimeEnvironment.server"; import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server"; -import { nextScheduledTimestamps } from "~/v3/utils/calculateNextSchedule.server"; import { NextRunListPresenter } from "./NextRunListPresenter.server"; import { scheduleWhereClause } from "~/models/schedules.server"; -import { formatScheduleWindow } from "~/v3/scheduleWindow.server"; +import { calculateNextScheduleRunTimes, formatScheduleWindow } from "~/v3/scheduleWindow.server"; +import { env } from "~/env.server"; type ViewScheduleOptions = { userId?: string; @@ -52,6 +52,8 @@ export class ViewSchedulePresenter { }, instances: { select: { + environmentId: true, + schedulePhase: true, environment: { select: { id: true, @@ -82,8 +84,25 @@ export class ViewSchedulePresenter { return; } + const instance = schedule.instances.find( + (instance) => instance.environmentId === environmentId + ); + if (!instance && schedule.instances.length > 0) { + return; + } + const nextRuns = schedule.active - ? nextScheduledTimestamps(schedule.generatorExpression, schedule.timezone, new Date(), 5) + ? calculateNextScheduleRunTimes({ + cron: schedule.generatorExpression, + timezone: schedule.timezone, + deduplicationKey: schedule.deduplicationKey, + environmentId, + schedulePhase: instance?.schedulePhase ?? null, + phaseSecret: env.ENCRYPTION_KEY, + windowDurationSeconds: schedule.windowDurationSeconds, + windowPercentage: schedule.windowPercentage, + count: 5, + }) : []; const runs = includeRunHistory @@ -101,6 +120,7 @@ export class ViewSchedulePresenter { timezone: schedule.timezone, cron: schedule.generatorExpression, cronDescription: schedule.generatorDescription, + window: formatScheduleWindow(schedule), nextRuns, runs, environments: schedule.instances.map((instance) => { @@ -146,14 +166,15 @@ export class ViewSchedulePresenter { type: result.schedule.type, task: result.schedule.taskIdentifier, active: result.schedule.active, - nextRun: result.schedule.nextRuns[0], + nextRun: result.schedule.nextRuns[0]?.nominalAt ?? null, + nextRunEffectiveAt: result.schedule.nextRuns[0]?.effectiveAt ?? null, generator: { type: "CRON", expression: result.schedule.cron, description: result.schedule.cronDescription, }, timezone: result.schedule.timezone, - window: formatScheduleWindow(result.schedule), + window: result.schedule.window, externalId: result.schedule.externalId ?? undefined, deduplicationKey: result.schedule.userProvidedDeduplicationKey ? (result.schedule.deduplicationKey ?? undefined) diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.tasks.scheduled.$taskParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.tasks.scheduled.$taskParam/route.tsx index a8d40b56f1..60a3aaf104 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.tasks.scheduled.$taskParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.tasks.scheduled.$taskParam/route.tsx @@ -867,7 +867,7 @@ function ScheduledTaskDetailSidebar({ - CRON + Cron {firstSchedule ? (
@@ -889,7 +889,7 @@ function ScheduledTaskDetailSidebar({ Next run {firstSchedule ? ( - + ) : ( )} @@ -972,8 +972,9 @@ type ScheduleRow = { type: "DECLARATIVE" | "IMPERATIVE"; cron: string; cronDescription: string; + window?: string; externalId: string | null; - nextRun: Date; + nextRunEffectiveAt: Date; lastRun: Date | undefined; active: boolean; }; @@ -993,7 +994,7 @@ function SchedulesMiniTable({ return ( - + No schedules attached to this task yet. @@ -1010,6 +1011,7 @@ function SchedulesMiniTable({ Schedule ID Type Cron + Window External ID Next run Last run @@ -1036,6 +1038,9 @@ function SchedulesMiniTable({ {schedule.cron} + + {schedule.window} + {schedule.externalId ? ( {schedule.externalId} @@ -1044,7 +1049,7 @@ function SchedulesMiniTable({ )} - + {schedule.lastRun ? ( diff --git a/apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts b/apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts index f98707eecb..c601d0e621 100644 --- a/apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts +++ b/apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts @@ -130,6 +130,7 @@ export async function action({ request, params }: ActionFunctionArgs) { deduplicationKey: schedule.deduplicationKey, environments: schedule.environments, nextRun: schedule.nextRun, + nextRunEffectiveAt: schedule.nextRunEffectiveAt, }; return json(responseObject, { status: 200 }); diff --git a/apps/webapp/app/routes/api.v1.schedules.ts b/apps/webapp/app/routes/api.v1.schedules.ts index 277033dd94..aaa49d1c02 100644 --- a/apps/webapp/app/routes/api.v1.schedules.ts +++ b/apps/webapp/app/routes/api.v1.schedules.ts @@ -72,6 +72,7 @@ export async function action({ request }: ActionFunctionArgs) { deduplicationKey: schedule.deduplicationKey, environments: schedule.environments, nextRun: schedule.nextRun, + nextRunEffectiveAt: schedule.nextRunEffectiveAt, }; return json(responseObject, { status: 200 }); @@ -130,6 +131,7 @@ export async function loader({ request }: LoaderFunctionArgs) { externalId: schedule.externalId, active: schedule.active, nextRun: schedule.nextRun, + nextRunEffectiveAt: schedule.nextRunEffectiveAt, environments: schedule.environments, })), pagination: { diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.schedules.new/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.schedules.new/route.tsx index 61ce814dbd..74b1a1f295 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.schedules.new/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.schedules.new/route.tsx @@ -1,5 +1,6 @@ import { getFormProps, getInputProps, getSelectProps, useForm } from "@conform-to/react"; import { parseWithZod } from "@conform-to/zod"; +import { ScheduleWindow } from "@trigger.dev/core/v3"; import { CheckIcon, XMarkIcon } from "@heroicons/react/20/solid"; import { type FetcherWithComponents, @@ -52,6 +53,7 @@ import { requireUserId } from "~/services/session.server"; import { cn } from "~/utils/cn"; import { EnvironmentParamSchema, docsPath, v3EnvironmentPath } from "~/utils/pathBuilder"; import { CronPattern, UpsertSchedule } from "~/v3/schedules"; +import { ServiceValidationError } from "~/v3/services/baseService.server"; import { UpsertTaskScheduleService } from "~/v3/services/upsertTaskSchedule.server"; import { AIGeneratedCronField } from "../resources.orgs.$organizationSlug.projects.$projectParam.schedules.new.natural-language"; @@ -114,10 +116,21 @@ export const action = async ({ request, params }: ActionFunctionArgs) => { message ); } catch (error: any) { - logger.error("Failed to create schedule", error); + if (!(error instanceof ServiceValidationError)) { + logger.error("Failed to create schedule", error); + } - const errorMessage = `Something went wrong. Please try again.`; + const errorMessage = + error instanceof ServiceValidationError + ? error.message + : `Something went wrong. Please try again.`; if (wantsJson) { + if (error instanceof ServiceValidationError) { + return json(submission.reply({ formErrors: [error.message] }), { + status: error.status ?? 422, + }); + } + return json({ ok: false as const, message: errorMessage }, { status: 500 }); } return redirectWithErrorMessage( @@ -138,6 +151,15 @@ type CronPatternResult = error: string; }; +type ScheduleWindowResult = + | { + isValid: true; + } + | { + isValid: false; + error: string; + }; + export function UpsertScheduleForm({ schedule, possibleTasks, @@ -167,6 +189,7 @@ export function UpsertScheduleForm({ const [selectedTimezone, setSelectedTimezone] = useState(schedule?.timezone ?? "UTC"); const isUtc = selectedTimezone === "UTC"; const [cronPattern, setCronPattern] = useState(schedule?.cron ?? ""); + const [scheduleWindowValue, setScheduleWindowValue] = useState(schedule?.window ?? ""); const navigation = useNavigation(); const isLoading = submitFetcher ? submitFetcher.state !== "idle" : navigation.state !== "idle"; const organization = useOrganization(); @@ -174,22 +197,40 @@ export function UpsertScheduleForm({ const environment = useEnvironment(); const location = useLocation(); - const [form, { taskIdentifier, cron, timezone, externalId, environments, deduplicationKey }] = - useForm({ - // Disambiguate per-schedule so both sheets (create + edit) can - // coexist without duplicate DOM ids breaking `htmlFor` / conform. - id: schedule?.friendlyId ? `edit-schedule-${schedule.friendlyId}` : "create-schedule", - // TODO: type this - lastResult: lastSubmission as any, - shouldRevalidate: "onSubmit", - onValidate({ formData }) { - return parseWithZod(formData, { schema: UpsertSchedule }); - }, - }); + const [ + form, + { + taskIdentifier, + cron, + timezone, + window: scheduleWindow, + externalId, + environments, + deduplicationKey, + }, + ] = useForm({ + // Disambiguate per-schedule so both sheets (create + edit) can + // coexist without duplicate DOM ids breaking `htmlFor` / conform. + id: schedule?.friendlyId ? `edit-schedule-${schedule.friendlyId}` : "create-schedule", + // TODO: type this + lastResult: lastSubmission as any, + shouldRevalidate: "onSubmit", + onValidate({ formData }) { + return parseWithZod(formData, { schema: UpsertSchedule }); + }, + }); let cronPatternResult: CronPatternResult | undefined = undefined; + let scheduleWindowResult: ScheduleWindowResult | undefined = undefined; let nextRuns: Date[] | undefined = undefined; + if (scheduleWindowValue !== "") { + const result = ScheduleWindow.safeParse(scheduleWindowValue); + scheduleWindowResult = result.success + ? { isValid: true } + : { isValid: false, error: result.error.errors[0].message }; + } + if (cronPattern !== "") { const result = CronPattern.safeParse(cronPattern); @@ -305,9 +346,19 @@ export function UpsertScheduleForm({ {cronPatternResult === undefined ? ( Enter a CRON pattern or use natural language above. ) : cronPatternResult.isValid ? ( - + ) : ( - + )} @@ -335,9 +386,52 @@ export function UpsertScheduleForm({ {timezone.errors} + + + setScheduleWindowValue(event.target.value)} + /> + {scheduleWindowResult === undefined ? ( + + Assigns each run a stable time after its CRON time. Use minutes, hours, or a + percentage of the interval. + + ) : scheduleWindowResult.isValid ? ( + + ) : ( + + )} + {nextRuns !== undefined && (
Next 5 runs + {scheduleWindowValue !== "" && ( + + Actual run times will get a fixed offset based on the window, displayed after + creation. + + )}
@@ -486,9 +580,21 @@ function buttonText(mode: "edit" | "new", isLoading: boolean) { } } -function ValidCronMessage({ isValid, message }: { isValid: boolean; message: string }) { +function ValidationMessage({ + id, + isValid, + validLabel, + invalidLabel, + message, +}: { + id?: string; + isValid: boolean; + validLabel: string; + invalidLabel: string; + message: string; +}) { return ( - + {isValid ? ( @@ -496,7 +602,7 @@ function ValidCronMessage({ isValid, message }: { isValid: boolean; message: str )} - {isValid ? "Valid pattern:" : "Invalid pattern:"} + {isValid ? validLabel : invalidLabel} {message} diff --git a/apps/webapp/app/runEngine/concerns/traceEvents.server.ts b/apps/webapp/app/runEngine/concerns/traceEvents.server.ts index a0efcd1ac0..7e8734164f 100644 --- a/apps/webapp/app/runEngine/concerns/traceEvents.server.ts +++ b/apps/webapp/app/runEngine/concerns/traceEvents.server.ts @@ -2,6 +2,7 @@ import { SemanticInternalAttributes } from "@trigger.dev/core/v3/semanticInterna import type { TaskRun } from "@trigger.dev/database"; import type { IEventRepository } from "~/v3/eventRepository/eventRepository.types"; import { getEventRepository } from "~/v3/eventRepository/index.server"; +import { runTriggeredAt } from "~/v3/runTimestamps"; import type { TracedEventSpan, TraceEventConcern, TriggerTaskRequest } from "../types"; export class DefaultTraceEventsConcern implements TraceEventConcern { @@ -22,6 +23,13 @@ export class DefaultTraceEventsConcern implements TraceEventConcern { callback: (span: TracedEventSpan, store: string) => Promise ): Promise { const { repository, store } = await this.#getEventRepository(request, parentStore); + const startTime = request.options?.overrideCreatedAt + ? runTriggeredAt({ + createdAt: request.options.overrideCreatedAt, + queueTimestamp: request.options.queueTimestamp, + scheduleId: request.options.scheduleId, + }) + : undefined; return await repository.traceEvent( request.taskId, @@ -39,9 +47,7 @@ export class DefaultTraceEventsConcern implements TraceEventConcern { }, incomplete: true, immediate: true, - startTime: request.options?.overrideCreatedAt - ? BigInt(request.options.overrideCreatedAt.getTime()) * BigInt(1000000) - : undefined, + startTime: startTime ? BigInt(startTime.getTime()) * BigInt(1000000) : undefined, }, async (event, traceContext, traceparent) => { return await callback( diff --git a/apps/webapp/app/runEngine/types.ts b/apps/webapp/app/runEngine/types.ts index 339f1a36e1..14c992a285 100644 --- a/apps/webapp/app/runEngine/types.ts +++ b/apps/webapp/app/runEngine/types.ts @@ -16,6 +16,8 @@ export type TriggerTaskServiceOptions = { runFriendlyId?: string; skipChecks?: boolean; oneTimeUseToken?: string; + scheduleId?: string; + queueTimestamp?: Date; overrideCreatedAt?: Date; planType?: string; }; diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts index f9db41e4f0..808e30431e 100644 --- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts @@ -280,6 +280,8 @@ export class ClickHouseRunsRepository implements IRunsRepository { runtimeEnvironmentId: true, status: true, createdAt: true, + queueTimestamp: true, + scheduleId: true, startedAt: true, lockedAt: true, delayUntil: true, diff --git a/apps/webapp/app/v3/runTimestamps.ts b/apps/webapp/app/v3/runTimestamps.ts new file mode 100644 index 0000000000..23bcac6681 --- /dev/null +++ b/apps/webapp/app/v3/runTimestamps.ts @@ -0,0 +1,11 @@ +export function runTriggeredAt({ + createdAt, + queueTimestamp, + scheduleId, +}: { + createdAt: Date; + queueTimestamp?: Date | null; + scheduleId?: string | null; +}) { + return scheduleId && queueTimestamp ? queueTimestamp : createdAt; +} diff --git a/apps/webapp/app/v3/scheduleWindow.server.ts b/apps/webapp/app/v3/scheduleWindow.server.ts index e89123488e..c0bf8cea69 100644 --- a/apps/webapp/app/v3/scheduleWindow.server.ts +++ b/apps/webapp/app/v3/scheduleWindow.server.ts @@ -1,5 +1,10 @@ -import { parseScheduleWindow } from "@internal/schedule-engine"; -import type { ScheduleWindow } from "@trigger.dev/core/v3"; +import { calculateEffectiveScheduleTime, calculateSchedulePhase } from "@internal/schedule-engine"; +import { + ScheduleWindow, + parseScheduleWindow, + type NormalizedScheduleWindow, +} from "@trigger.dev/core/v3"; +import { nextScheduledTimestamps } from "./utils/calculateNextSchedule.server"; const SECONDS_PER_UNIT = { m: 60, @@ -11,9 +16,12 @@ export type ScheduleWindowDatabaseFields = { windowPercentage: number | null; }; -export function normalizeScheduleWindow( - window: ScheduleWindow | undefined -): ScheduleWindowDatabaseFields { +export type ScheduleRunTiming = { + nominalAt: Date; + effectiveAt: Date; +}; + +export function normalizeScheduleWindow(window: string | undefined): ScheduleWindowDatabaseFields { if (window === undefined) { return { windowDurationSeconds: null, @@ -39,7 +47,7 @@ export function normalizeScheduleWindow( export function formatScheduleWindow({ windowDurationSeconds, windowPercentage, -}: ScheduleWindowDatabaseFields): ScheduleWindow | undefined { +}: ScheduleWindowDatabaseFields): string | undefined { if (windowPercentage !== null) { return `${windowPercentage}%`; } @@ -59,20 +67,73 @@ export function formatScheduleWindow({ return `${windowDurationSeconds / SECONDS_PER_UNIT.m}m`; } +export function calculateNextScheduleRunTimes({ + cron, + timezone, + deduplicationKey, + environmentId, + schedulePhase, + phaseSecret, + windowDurationSeconds, + windowPercentage, + from = new Date(), + count = 1, +}: { + cron: string; + timezone: string | null; + deduplicationKey: string; + environmentId: string; + schedulePhase: number | null; + phaseSecret: string; + windowDurationSeconds: number | null; + windowPercentage: number | null; + from?: Date; + count?: number; +}): ScheduleRunTiming[] { + if (count <= 0) { + return []; + } + + const phase = + schedulePhase ?? + calculateSchedulePhase({ + secret: phaseSecret, + environmentId, + deduplicationKey, + }); + const window: NormalizedScheduleWindow | undefined = + windowPercentage !== null + ? { type: "percentage", percentage: windowPercentage } + : windowDurationSeconds !== null + ? { type: "duration", durationSeconds: windowDurationSeconds } + : undefined; + const nominalTimes = nextScheduledTimestamps(cron, timezone, from, count + 1); + + return nominalTimes.slice(0, count).map((nominalAt, index) => ({ + nominalAt, + effectiveAt: calculateEffectiveScheduleTime({ + nominalAt, + nextNominalAt: nominalTimes[index + 1], + schedulePhase: phase, + window, + }).effectiveAt, + })); +} + export function validateScheduleWindowSyntax( - window: ScheduleWindow | undefined + window: string | undefined ): { valid: true } | { valid: false; message: string } { if (window === undefined) { return { valid: true }; } - try { - parseScheduleWindow(window); + const result = ScheduleWindow.safeParse(window); + if (result.success) { return { valid: true }; - } catch (error) { - return { - valid: false, - message: error instanceof Error ? error.message : String(error), - }; } + + return { + valid: false, + message: result.error.issues[0]?.message ?? "Invalid schedule window", + }; } diff --git a/apps/webapp/app/v3/schedules.ts b/apps/webapp/app/v3/schedules.ts index bb1d3af55d..d2355a5f54 100644 --- a/apps/webapp/app/v3/schedules.ts +++ b/apps/webapp/app/v3/schedules.ts @@ -57,7 +57,7 @@ export const UpsertSchedule = z.object({ externalId: z.string().optional(), deduplicationKey: z.string().optional(), timezone: z.string().optional(), - window: ScheduleWindow.optional(), + window: z.preprocess((value) => (value === "" ? undefined : value), 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 ea7be4e39e..946cfad080 100644 --- a/apps/webapp/app/v3/services/checkSchedule.server.ts +++ b/apps/webapp/app/v3/services/checkSchedule.server.ts @@ -5,7 +5,6 @@ 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 { boundedIn, type PrismaClientOrTransaction } from "@trigger.dev/database"; import { validateScheduleWindowSyntax } from "../scheduleWindow.server"; @@ -14,7 +13,7 @@ type Schedule = { timezone?: string; taskIdentifier: string; friendlyId?: string; - window?: ScheduleWindow; + window?: string; }; export class CheckScheduleService extends BaseService { diff --git a/apps/webapp/app/v3/services/upsertTaskSchedule.server.ts b/apps/webapp/app/v3/services/upsertTaskSchedule.server.ts index 567e8269d3..cf72873b37 100644 --- a/apps/webapp/app/v3/services/upsertTaskSchedule.server.ts +++ b/apps/webapp/app/v3/services/upsertTaskSchedule.server.ts @@ -3,12 +3,16 @@ import cronstrue from "cronstrue"; import { nanoid } from "nanoid"; import { generateFriendlyId } from "../friendlyIdentifiers"; import { type UpsertSchedule } from "../schedules"; -import { calculateNextScheduledTimestampFromNow } from "../utils/calculateNextSchedule.server"; import { BaseService, ServiceValidationError } from "./baseService.server"; import { CheckScheduleService } from "./checkSchedule.server"; import { scheduleEngine } from "../scheduleEngine.server"; -import { formatScheduleWindow, normalizeScheduleWindow } from "../scheduleWindow.server"; +import { + calculateNextScheduleRunTimes, + formatScheduleWindow, + normalizeScheduleWindow, +} from "../scheduleWindow.server"; import { scheduleWhereClause } from "~/models/schedules.server"; +import { env } from "~/env.server"; export type UpsertTaskScheduleServiceOptions = UpsertSchedule; @@ -81,7 +85,7 @@ export class UpsertTaskScheduleService extends BaseService { }, }); - return this.#createReturnObject(scheduleRecord, instances); + return this.#createReturnObject(scheduleRecord, instances, schedule.environments[0]); } async #createNewSchedule( @@ -237,7 +241,27 @@ export class UpsertTaskScheduleService extends BaseService { return { scheduleRecord }; } - #createReturnObject(taskSchedule: TaskSchedule, instances: InstanceWithEnvironment[]) { + #createReturnObject( + taskSchedule: TaskSchedule, + instances: InstanceWithEnvironment[], + environmentId: string + ) { + const instance = instances.find((instance) => instance.environmentId === environmentId); + if (!instance) { + throw new ServiceValidationError("Failed to find the schedule instance"); + } + + const [nextRun] = calculateNextScheduleRunTimes({ + cron: taskSchedule.generatorExpression, + timezone: taskSchedule.timezone, + deduplicationKey: taskSchedule.deduplicationKey, + environmentId: instance.environmentId, + schedulePhase: instance.schedulePhase, + phaseSecret: env.ENCRYPTION_KEY, + windowDurationSeconds: taskSchedule.windowDurationSeconds, + windowPercentage: taskSchedule.windowPercentage, + }); + return { id: taskSchedule.friendlyId, type: taskSchedule.type, @@ -251,10 +275,8 @@ export class UpsertTaskScheduleService extends BaseService { cronDescription: taskSchedule.generatorDescription, timezone: taskSchedule.timezone, window: formatScheduleWindow(taskSchedule), - nextRun: calculateNextScheduledTimestampFromNow( - taskSchedule.generatorExpression, - taskSchedule.timezone - ), + nextRun: nextRun.nominalAt, + nextRunEffectiveAt: nextRun.effectiveAt, environments: instances.map((instance) => ({ id: instance.environment.id, shortcode: instance.environment.shortcode, diff --git a/apps/webapp/test/runTimestamps.test.ts b/apps/webapp/test/runTimestamps.test.ts new file mode 100644 index 0000000000..30298a23ee --- /dev/null +++ b/apps/webapp/test/runTimestamps.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from "vitest"; +import { runTriggeredAt } from "~/v3/runTimestamps"; + +describe("runTriggeredAt", () => { + const createdAt = new Date("2026-08-14T08:30:00.000Z"); + const queueTimestamp = new Date("2026-08-14T08:38:25.000Z"); + + it("uses the effective queue timestamp for scheduled runs", () => { + expect( + runTriggeredAt({ + createdAt, + queueTimestamp, + scheduleId: "schedule_123", + }) + ).toEqual(queueTimestamp); + }); + + it("preserves the creation time for non-scheduled delayed runs", () => { + expect( + runTriggeredAt({ + createdAt, + queueTimestamp, + scheduleId: null, + }) + ).toEqual(createdAt); + }); + + it("falls back to the creation time when a scheduled run has no queue timestamp", () => { + expect( + runTriggeredAt({ + createdAt, + queueTimestamp: null, + scheduleId: "schedule_123", + }) + ).toEqual(createdAt); + }); +}); diff --git a/apps/webapp/test/scheduleWindow.test.ts b/apps/webapp/test/scheduleWindow.test.ts index afc4a0c088..df19bb8919 100644 --- a/apps/webapp/test/scheduleWindow.test.ts +++ b/apps/webapp/test/scheduleWindow.test.ts @@ -1,5 +1,7 @@ +import { SCHEDULE_PHASE_DENOMINATOR } from "@internal/schedule-engine"; import { describe, expect, it } from "vitest"; import { + calculateNextScheduleRunTimes, formatScheduleWindow, normalizeScheduleWindow, validateScheduleWindowSyntax, @@ -62,4 +64,47 @@ describe("schedule window persistence", () => { it("accepts an absolute window independently of the cron interval", () => { expect(validateScheduleWindowSyntax("30m")).toEqual({ valid: true }); }); + + it("calculates stable nominal and effective times", () => { + const [first, second] = calculateNextScheduleRunTimes({ + cron: "*/5 * * * *", + timezone: "UTC", + deduplicationKey: "five-minute-task", + environmentId: "env_123", + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + phaseSecret: "test-secret", + windowDurationSeconds: null, + windowPercentage: 30, + from: new Date("2026-08-11T09:59:00.000Z"), + count: 2, + }); + + expect(first).toEqual({ + nominalAt: new Date("2026-08-11T10:00:00.000Z"), + effectiveAt: new Date("2026-08-11T10:00:45.000Z"), + }); + expect(second).toEqual({ + nominalAt: new Date("2026-08-11T10:05:00.000Z"), + effectiveAt: new Date("2026-08-11T10:05:45.000Z"), + }); + }); + + it("derives a stable phase when one has not been persisted", () => { + const input = { + cron: "0 * * * *", + timezone: "UTC", + deduplicationKey: "hourly-task", + environmentId: "env_123", + schedulePhase: null, + phaseSecret: "test-secret", + windowDurationSeconds: null, + windowPercentage: null, + from: new Date("2026-08-11T09:59:00.000Z"), + }; + + expect(calculateNextScheduleRunTimes(input)).toEqual(calculateNextScheduleRunTimes(input)); + expect(calculateNextScheduleRunTimes(input)[0].effectiveAt.getTime()).toBeGreaterThanOrEqual( + calculateNextScheduleRunTimes(input)[0].nominalAt.getTime() + ); + }); }); diff --git a/apps/webapp/test/schedules-api.e2e.full.test.ts b/apps/webapp/test/schedules-api.e2e.full.test.ts index 2924c30e10..a2f6ab72e3 100644 --- a/apps/webapp/test/schedules-api.e2e.full.test.ts +++ b/apps/webapp/test/schedules-api.e2e.full.test.ts @@ -29,15 +29,26 @@ describe("Schedules API windows", () => { timezone: "UTC", window: "30%", }); + expectAssignedTime(created, 18 * 60_000); 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({ + const retrieved = await retrieveResponse.json(); + expect(retrieved).toMatchObject({ id: created.id, window: "30%", }); + expectAssignedTime(retrieved, 18 * 60_000); + + const listResponse = await server.webapp.fetch("/api/v1/schedules", { + headers: authHeaders(apiKey), + }); + expect(listResponse.status).toBe(200); + const listed = await listResponse.json(); + expect(listed.data[0]).toMatchObject({ id: created.id }); + expectAssignedTime(listed.data[0], 18 * 60_000); const updateResponse = await server.webapp.fetch(`/api/v1/schedules/${created.id}`, { method: "PUT", @@ -49,10 +60,12 @@ describe("Schedules API windows", () => { }), }); expect(updateResponse.status).toBe(200); - await expect(updateResponse.json()).resolves.toMatchObject({ + const updated = await updateResponse.json(); + expect(updated).toMatchObject({ id: created.id, window: "2h", }); + expectAssignedTime(updated, 2 * 60 * 60_000); const clearResponse = await server.webapp.fetch(`/api/v1/schedules/${created.id}`, { method: "PUT", @@ -67,6 +80,26 @@ describe("Schedules API windows", () => { expect(cleared.id).toBe(created.id); expect(cleared).not.toHaveProperty("window"); + const deactivateResponse = await server.webapp.fetch( + `/api/v1/schedules/${created.id}/deactivate`, + { method: "POST", headers: authHeaders(apiKey) } + ); + expect(deactivateResponse.status).toBe(200); + await expect(deactivateResponse.json()).resolves.toMatchObject({ + active: false, + nextRun: null, + nextRunEffectiveAt: null, + }); + + const activateResponse = await server.webapp.fetch(`/api/v1/schedules/${created.id}/activate`, { + method: "POST", + headers: authHeaders(apiKey), + }); + expect(activateResponse.status).toBe(200); + const activated = await activateResponse.json(); + expect(activated.active).toBe(true); + expectAssignedTime(activated, 60_000); + const stored = await server.prisma.taskSchedule.findUniqueOrThrow({ where: { friendlyId: created.id }, select: { windowDurationSeconds: true, windowPercentage: true }, @@ -111,14 +144,9 @@ describe("Schedules API windows", () => { 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 }, - ]; + const invalidWindows = [30, "30.5%", "1d", "25h"]; - for (const [index, { window, expectedStatus }] of invalidRequests.entries()) { + for (const [index, window] of invalidWindows.entries()) { const response = await server.webapp.fetch("/api/v1/schedules", { method: "POST", headers: authHeaders(apiKey), @@ -130,12 +158,22 @@ describe("Schedules API windows", () => { }), }); - expect(response.status).toBe(expectedStatus); - await expect(response.json()).resolves.toHaveProperty("error"); + expect(response.status).toBe(400); } }); }); +function expectAssignedTime( + schedule: { nextRun: string; nextRunEffectiveAt: string }, + maximumDelayMs: number +) { + const nominalAt = new Date(schedule.nextRun).getTime(); + const effectiveAt = new Date(schedule.nextRunEffectiveAt).getTime(); + + expect(effectiveAt).toBeGreaterThanOrEqual(nominalAt); + expect(effectiveAt).toBeLessThan(nominalAt + maximumDelayMs); +} + function authHeaders(apiKey: string) { return { Authorization: `Bearer ${apiKey}`, diff --git a/internal-packages/schedule-engine/src/engine/scheduleTiming.ts b/internal-packages/schedule-engine/src/engine/scheduleTiming.ts index 003240e952..3b8931cef3 100644 --- a/internal-packages/schedule-engine/src/engine/scheduleTiming.ts +++ b/internal-packages/schedule-engine/src/engine/scheduleTiming.ts @@ -1,16 +1,19 @@ +import { + MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS, + parseScheduleWindow, + type NormalizedScheduleWindow, +} from "@trigger.dev/core/v3"; import { createHmac } from "node:crypto"; +export { MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS, parseScheduleWindow }; +export type { NormalizedScheduleWindow }; + 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; @@ -28,42 +31,6 @@ export type EffectiveScheduleTime = { 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 ( diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index f8de04bb4f..6cd100f7c3 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -1079,7 +1079,10 @@ export const ScheduleObject = z.object({ generator: ScheduleGenerator, timezone: z.string(), window: ScheduleWindow.optional(), + /** The next nominal CRON time. */ nextRun: z.coerce.date().nullish(), + /** The stable assigned time for the next nominal CRON time. */ + nextRunEffectiveAt: z.coerce.date().nullish(), environments: z.array( z.object({ id: z.string(), diff --git a/packages/core/src/v3/schemas/scheduleWindow.test.ts b/packages/core/src/v3/schemas/scheduleWindow.test.ts new file mode 100644 index 0000000000..f5812b2ed0 --- /dev/null +++ b/packages/core/src/v3/schemas/scheduleWindow.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, expectTypeOf, it } from "vitest"; +import { + ScheduleWindow, + parseScheduleWindow, + type ValidatedScheduleWindow, +} from "./scheduleWindow.js"; + +describe("ScheduleWindow", () => { + it.each(["0m", "30m", "1440m", "0h", "2h", "24h", "0%", "30%", "100%"])( + "accepts %s", + (window) => { + expect(ScheduleWindow.safeParse(window).success).toBe(true); + } + ); + + it.each([ + "", + "00m", + "01m", + "1.5h", + "0d", + "1d", + "25h", + "1441m", + "30s", + "-1m", + "0.5%", + "101%", + "1e2%", + " 30m", + "30m ", + ])("rejects %j", (window) => { + expect(ScheduleWindow.safeParse(window).success).toBe(false); + }); + + it("reports percentages above 100% precisely", () => { + expect(() => parseScheduleWindow("110%")).toThrow( + "Schedule window percentage cannot exceed 100%" + ); + }); + + it("normalizes valid windows", () => { + expect(parseScheduleWindow("30m")).toEqual({ + type: "duration", + durationSeconds: 1_800, + }); + expect(parseScheduleWindow("25%")).toEqual({ + type: "percentage", + percentage: 25, + }); + }); + + it("validates literal types without rejecting runtime strings", () => { + expectTypeOf>().toEqualTypeOf<"30m">(); + expectTypeOf>().toEqualTypeOf<"24h">(); + expectTypeOf>().toEqualTypeOf<"100%">(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf< + ValidatedScheduleWindow<"25h"> + >().toEqualTypeOf<"⛔ window duration cannot exceed 24 hours">(); + expectTypeOf< + ValidatedScheduleWindow<"101%"> + >().toEqualTypeOf<"⛔ percentage cannot exceed 100%">(); + expectTypeOf< + ValidatedScheduleWindow<"1d"> + >().toEqualTypeOf<'⛔ window must look like "30m", "2h", or "50%"'>(); + }); +}); diff --git a/packages/core/src/v3/schemas/scheduleWindow.ts b/packages/core/src/v3/schemas/scheduleWindow.ts new file mode 100644 index 0000000000..ce89d76404 --- /dev/null +++ b/packages/core/src/v3/schemas/scheduleWindow.ts @@ -0,0 +1,135 @@ +import { z } from "zod"; + +export const MAX_ABSOLUTE_SCHEDULE_WINDOW_SECONDS = 24 * 60 * 60; + +export type NormalizedScheduleWindow = + | { type: "duration"; durationSeconds: number } + | { type: "percentage"; percentage: number }; + +type Digit = "0" | "1" | "2" | "3" | "4" | "5" | "6" | "7" | "8" | "9"; + +type DigitsBelow = { + "0": never; + "1": "0"; + "2": "0" | "1"; + "3": "0" | "1" | "2"; + "4": "0" | "1" | "2" | "3"; + "5": "0" | "1" | "2" | "3" | "4"; + "6": "0" | "1" | "2" | "3" | "4" | "5"; + "7": "0" | "1" | "2" | "3" | "4" | "5" | "6"; + "8": "0" | "1" | "2" | "3" | "4" | "5" | "6" | "7"; + "9": "0" | "1" | "2" | "3" | "4" | "5" | "6" | "7" | "8"; +}; + +type DigitLength< + Value extends string, + Result extends 0[] = [], +> = Value extends `${Digit}${infer Rest}` ? DigitLength : Result; + +type CompareEqualLength< + A extends string, + B extends string, +> = A extends `${infer ADigit extends Digit}${infer ARest}` + ? B extends `${infer BDigit extends Digit}${infer BRest}` + ? ADigit extends BDigit + ? CompareEqualLength + : ADigit extends DigitsBelow[BDigit] + ? "lt" + : "gt" + : "eq" + : "eq"; + +type DecimalStringLTE = + DigitLength extends DigitLength + ? CompareEqualLength extends "gt" + ? false + : true + : DigitLength extends [...DigitLength, ...0[]] + ? true + : false; + +type IsCanonicalUnsignedInteger = Value extends `${bigint}` + ? Value extends `-${string}` + ? false + : true + : false; + +/** The literal validation error for a configured schedule window, or `never` when valid. */ +export type ScheduleWindowError = string extends Window + ? never + : Window extends `${infer Amount}m` + ? IsCanonicalUnsignedInteger extends false + ? "⛔ window must be a whole non-negative number" + : DecimalStringLTE extends true + ? never + : "⛔ window duration cannot exceed 24 hours" + : Window extends `${infer Amount}h` + ? IsCanonicalUnsignedInteger extends false + ? "⛔ window must be a whole non-negative number" + : DecimalStringLTE extends true + ? never + : "⛔ window duration cannot exceed 24 hours" + : Window extends `${infer Amount}%` + ? IsCanonicalUnsignedInteger extends false + ? "⛔ percentage must be a whole non-negative number" + : DecimalStringLTE extends true + ? never + : "⛔ percentage cannot exceed 100%" + : '⛔ window must look like "30m", "2h", or "50%"'; + +/** + * Preserves valid schedule-window literals and replaces invalid literals with a descriptive type + * error. Wide `string` values pass through for authoritative runtime validation. + */ +export type ValidatedScheduleWindow = Window extends string + ? [ScheduleWindowError] extends [never] + ? Window + : ScheduleWindowError + : Window; + +/** Parses and normalizes the public schedule-window syntax. */ +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 durationSeconds = amount * (unit === "m" ? 60 : 3_600); + + 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*)%$/.exec(value); + if (percentageMatch) { + const percentage = Number(percentageMatch[1]); + + if (!Number.isSafeInteger(percentage) || percentage > 100) { + throw new RangeError("Schedule window percentage cannot exceed 100%"); + } + + return { type: "percentage", percentage }; + } + + throw new TypeError( + 'Schedule window must be a whole duration such as "30m" or "2h", or a percentage such as "30%"' + ); +} + +/** Runtime authority for public schedule-window values. */ +export const ScheduleWindow = z.string().superRefine((value, ctx) => { + try { + parseScheduleWindow(value); + } catch (error) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: error instanceof Error ? error.message : String(error), + }); + } +}); diff --git a/packages/core/src/v3/schemas/schemas.ts b/packages/core/src/v3/schemas/schemas.ts index 7e95224f42..72b6d9ed58 100644 --- a/packages/core/src/v3/schemas/schemas.ts +++ b/packages/core/src/v3/schemas/schemas.ts @@ -7,6 +7,9 @@ import { TaskRunExecution, V3TaskRunExecution, } from "./common.js"; +import { ScheduleWindow } from "./scheduleWindow.js"; + +export * from "./scheduleWindow.js"; /* WARNING: Never import anything from ./messages here. If it's needed in both, put it here instead. @@ -174,15 +177,6 @@ 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(), diff --git a/packages/trigger-sdk/src/v3/schedules/api.ts b/packages/trigger-sdk/src/v3/schedules/api.ts index 65fe4d4294..4264b0d3ef 100644 --- a/packages/trigger-sdk/src/v3/schedules/api.ts +++ b/packages/trigger-sdk/src/v3/schedules/api.ts @@ -1,6 +1,23 @@ -export type { - CreateScheduleOptions, - ScheduledTaskPayload, +import type { + CreateScheduleOptions as CoreCreateScheduleOptions, ListScheduleOptions, - UpdateScheduleOptions, + ScheduledTaskPayload, + UpdateScheduleOptions as CoreUpdateScheduleOptions, + ValidatedScheduleWindow, } from "@trigger.dev/core/v3"; + +export type { ListScheduleOptions, ScheduledTaskPayload }; + +export type CreateScheduleOptions = Omit< + CoreCreateScheduleOptions, + "window" +> & { + window?: ValidatedScheduleWindow; +}; + +export type UpdateScheduleOptions = Omit< + CoreUpdateScheduleOptions, + "window" +> & { + window?: ValidatedScheduleWindow; +}; diff --git a/packages/trigger-sdk/src/v3/schedules/index.test.ts b/packages/trigger-sdk/src/v3/schedules/index.test.ts new file mode 100644 index 0000000000..f224b90c18 --- /dev/null +++ b/packages/trigger-sdk/src/v3/schedules/index.test.ts @@ -0,0 +1,55 @@ +import { resourceCatalog } from "@trigger.dev/core/v3"; +import { StandardResourceCatalog } from "@trigger.dev/core/v3/workers"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { task } from "./index.js"; + +describe("declarative schedule windows", () => { + beforeEach(() => { + resourceCatalog.disable(); + resourceCatalog.setGlobalResourceCatalog(new StandardResourceCatalog()); + resourceCatalog.setCurrentFileContext("scheduled.ts", "scheduled.ts"); + }); + + afterEach(() => { + resourceCatalog.clearCurrentFileContext(); + resourceCatalog.disable(); + }); + + it.each(["0m", "30m", "2h", "24h", "30%"] as const)( + "serializes the %s window into task metadata", + (window) => { + task({ + id: "daily-report", + cron: { + pattern: "0 9 * * *", + timezone: "Europe/London", + window, + environments: ["PRODUCTION"], + }, + run: async () => undefined, + }); + + expect(resourceCatalog.getTaskManifest("daily-report")?.schedule).toEqual({ + cron: "0 9 * * *", + timezone: "Europe/London", + window, + environments: ["PRODUCTION"], + }); + } + ); + + it("leaves the window undefined when it is omitted", () => { + task({ + id: "hourly-report", + cron: { pattern: "0 * * * *" }, + run: async () => undefined, + }); + + expect(resourceCatalog.getTaskManifest("hourly-report")?.schedule).toEqual({ + cron: "0 * * * *", + timezone: "UTC", + window: undefined, + environments: undefined, + }); + }); +}); diff --git a/packages/trigger-sdk/src/v3/schedules/index.ts b/packages/trigger-sdk/src/v3/schedules/index.ts index 7ab27e00e4..ad9f1538e8 100644 --- a/packages/trigger-sdk/src/v3/schedules/index.ts +++ b/packages/trigger-sdk/src/v3/schedules/index.ts @@ -5,6 +5,7 @@ import type { InitOutput, OffsetLimitPagePromise, ScheduleObject, + ValidatedScheduleWindow, } from "@trigger.dev/core/v3"; import { TimezonesResult, @@ -23,6 +24,7 @@ export type ScheduleOptions< TIdentifier extends string, TOutput, TInitOutput extends InitOutput, + TWindow extends string | undefined = string | undefined, > = TaskOptions & { /** You can optionally specify a CRON schedule on your task. You can also dynamically add a schedule in the dashboard or using the SDK functions. * @@ -31,11 +33,12 @@ export type ScheduleOptions< * "0 0 * * *" * ``` * - * 2. Or an object with a pattern, optional timezone, and optional environments + * 2. Or an object with a pattern, optional timezone, window, and environments * ```ts * { * pattern: "0 0 * * *", * timezone: "America/Los_Angeles", + * window: "30m", * environments: ["PRODUCTION", "STAGING"] * } * ``` @@ -47,6 +50,10 @@ export type ScheduleOptions< | { pattern: string; timezone?: string; + /** Optionally assign each run a stable time after its CRON time. + * Use a whole duration such as `"30m"` or `"2h"`, or a percentage such as `"30%"`. + */ + window?: ValidatedScheduleWindow; /** You can optionally specify which environments this schedule should run in. * When not specified, the schedule will run in all environments. * @@ -64,8 +71,13 @@ export type ScheduleOptions< }; }; -export function task( - params: ScheduleOptions +export function task< + TIdentifier extends string, + TOutput, + TInitOutput extends InitOutput, + const TWindow extends string | undefined = undefined, +>( + params: ScheduleOptions ): Task { const task = createTask(params); @@ -78,6 +90,7 @@ export function task( + options: SchedulesAPI.CreateScheduleOptions, requestOptions?: ApiRequestOptions ): ApiPromise { const apiClient = apiClientManager.clientOrThrow(); @@ -177,9 +191,9 @@ export function retrieve( * @param options.externalId - An optional external identifier for the schedule * @returns The updated schedule */ -export function update( +export function update( scheduleId: string, - options: SchedulesAPI.UpdateScheduleOptions, + options: SchedulesAPI.UpdateScheduleOptions, requestOptions?: ApiRequestOptions ): ApiPromise { const apiClient = apiClientManager.clientOrThrow(); diff --git a/packages/trigger-sdk/src/v3/schedules/index.types.test.ts b/packages/trigger-sdk/src/v3/schedules/index.types.test.ts new file mode 100644 index 0000000000..9db62ef0bc --- /dev/null +++ b/packages/trigger-sdk/src/v3/schedules/index.types.test.ts @@ -0,0 +1,98 @@ +import { describe, it } from "vitest"; +import { create, task, update } from "./index.js"; + +declare const runtimeWindow: string; + +function assertDeclarativeScheduleWindowTypes() { + task({ + id: "valid-window", + cron: { pattern: "*/5 * * * *", window: "30m" }, + run: async () => undefined, + }); + task({ + id: "runtime-window", + cron: { pattern: "*/5 * * * *", window: runtimeWindow }, + run: async () => undefined, + }); + + task({ + id: "duration-too-large", + // @ts-expect-error Schedule windows cannot exceed 24 hours. + cron: { pattern: "*/5 * * * *", window: "25h" }, + run: async () => undefined, + }); + task({ + id: "percentage-too-large", + // @ts-expect-error Schedule window percentages cannot exceed 100%. + cron: { pattern: "*/5 * * * *", window: "101%" }, + run: async () => undefined, + }); + task({ + id: "unsupported-unit", + // @ts-expect-error Days are not a supported schedule-window unit. + cron: { pattern: "*/5 * * * *", window: "1d" }, + run: async () => undefined, + }); + task({ + id: "fractional-window", + // @ts-expect-error Schedule windows must use whole numbers. + cron: { pattern: "*/5 * * * *", window: "1.5m" }, + run: async () => undefined, + }); + task({ + id: "negative-window", + // @ts-expect-error Schedule windows must be non-negative. + cron: { pattern: "*/5 * * * *", window: "-1m" }, + run: async () => undefined, + }); + task({ + id: "leading-zero-window", + // @ts-expect-error Schedule windows must use canonical whole numbers. + cron: { pattern: "*/5 * * * *", window: "01m" }, + run: async () => undefined, + }); +} + +function assertImperativeScheduleWindowTypes() { + create({ + task: "scheduled-task", + cron: "*/5 * * * *", + deduplicationKey: "valid", + window: "50%", + }); + create({ + task: "scheduled-task", + cron: "*/5 * * * *", + deduplicationKey: "runtime", + window: runtimeWindow, + }); + update("schedule_123", { + task: "scheduled-task", + cron: "*/5 * * * *", + window: "1440m", + }); + + create({ + task: "scheduled-task", + cron: "*/5 * * * *", + deduplicationKey: "invalid", + // @ts-expect-error Schedule windows cannot exceed 24 hours. + window: "1441m", + }); + update("schedule_123", { + task: "scheduled-task", + cron: "*/5 * * * *", + // @ts-expect-error Schedule windows must use a supported unit. + window: "30s", + }); +} + +describe("schedule window types", () => { + it("validates declarative schedule literals", () => { + void assertDeclarativeScheduleWindowTypes; + }); + + it("validates imperative schedule literals", () => { + void assertImperativeScheduleWindowTypes; + }); +});