feat: implement cron window spread backend

.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.

apps/webapp/app/env.server.ts

@@ -1672,6 +1672,7 @@ 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_ENABLED: BoolEnv.default(false),
 
     SCHEDULE_WORKER_REDIS_HOST: z
       .string()

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,

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<Awaited<ReturnType<ViewSchedulePresenter["call"]>>>) {
     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)

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) {

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) {

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) {

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,

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

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,
+    cronSpreadEnabled: env.SCHEDULE_WORKER_CRON_SPREAD_ENABLED,
     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",

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),
+    };
+  }
+}

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<typeof UpsertSchedule>;

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: {