feat: implement cron window spread backend (#4566)

- New DB fields on Schedule and ScheduleInstance
- Use `queueTimestamp` for the "effectiveAt" delayed start time,
propagate it to Clickhouse TaskRun table
- Disable fastpath for delayed jobs
- Add schedule timing logic, API endpoints with windows, persistence
- Calculate phase for every schedule, only persist when window is
non-null
- Additional o11y for phased rollout

Author: carderne

.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

@@ -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()

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,
+    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",