Use the new schedule engine, create runs with future created ats, filter that out in the lists

ericallam · ericallam · commit 2d4fe8578b33 · 2025-06-15T20:02:31.000+01:00
diff --git a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts
@@ -190,6 +190,11 @@ export class NextRunListPresenter {
       prisma: this.replica as PrismaClient,
     });
 
+    function clampToNow(date: Date): Date {
+      const now = new Date();
+      return date > now ? now : date;
+    }
+
     const { runs, pagination } = await runsRepository.listRuns({
       organizationId,
       environmentId,
@@ -200,8 +205,8 @@ export class NextRunListPresenter {
       tags,
       scheduleId,
       period: periodMs ?? undefined,
-      from,
-      to,
+      from: time.from ? time.from.getTime() : undefined,
+      to: time.to ? clampToNow(time.to).getTime() : undefined,
       isTest,
       rootOnly,
       batchId,
diff --git a/apps/webapp/app/presenters/v3/RunListPresenter.server.ts b/apps/webapp/app/presenters/v3/RunListPresenter.server.ts
@@ -179,6 +179,12 @@ export class RunListPresenter extends BasePresenter {
 
     const periodMs = time.period ? parse(time.period) : undefined;
 
+    function clampToNow(date: Date): Date {
+      const now = new Date();
+
+      return date > now ? now : date;
+    }
+
     //get the runs
     const runs = await this._replica.$queryRaw<
       {
@@ -282,7 +288,9 @@ WHERE
         : Prisma.empty
     }
     ${
-      time.to ? Prisma.sql`AND tr."createdAt" <= ${time.to.toISOString()}::timestamp` : Prisma.empty
+      time.to
+        ? Prisma.sql`AND tr."createdAt" <= ${clampToNow(time.to).toISOString()}::timestamp`
+        : Prisma.sql`AND tr."createdAt" <= CURRENT_TIMESTAMP`
     }
     ${
       tags && tags.length > 0
diff --git a/apps/webapp/app/runEngine/services/triggerTask.server.ts b/apps/webapp/app/runEngine/services/triggerTask.server.ts
@@ -308,6 +308,7 @@ export class RunEngineTriggerTaskService {
                   runChainState,
                   scheduleId: options.scheduleId,
                   scheduleInstanceId: options.scheduleInstanceId,
+                  createdAt: options.overrideCreatedAt,
                 },
                 this.prisma
               );
diff --git a/apps/webapp/app/services/runsRepository.server.ts b/apps/webapp/app/services/runsRepository.server.ts
@@ -91,6 +91,10 @@ export class RunsRepository {
 
     if (options.to) {
       queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", { to: options.to });
+    } else {
+      queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", {
+        to: Date.now(),
+      });
     }
 
     if (typeof options.isTest === "boolean") {
diff --git a/apps/webapp/app/v3/scheduleEngine.server.ts b/apps/webapp/app/v3/scheduleEngine.server.ts
@@ -12,50 +12,6 @@ export const scheduleEngine = singleton("ScheduleEngine", createScheduleEngine);
 
 export type { ScheduleEngine };
 
-const triggerScheduledTaskCallback: TriggerScheduledTaskCallback = async ({
-  taskIdentifier,
-  environment,
-  payload,
-  scheduleInstanceId,
-  scheduleId,
-  exactScheduleTime,
-}) => {
-  try {
-    // Use TriggerTaskServiceV1 for now (can be updated to use TriggerTaskService when ready)
-    const triggerService = new TriggerTaskService();
-
-    const payloadPacket = await stringifyIO(payload);
-
-    logger.debug("Triggering scheduled task", {
-      taskIdentifier,
-      environment,
-      payload,
-      scheduleInstanceId,
-      scheduleId,
-      exactScheduleTime,
-    });
-
-    const result = await triggerService.call(
-      taskIdentifier,
-      environment,
-      { payload: payloadPacket.data, options: { payloadType: payloadPacket.dataType } },
-      {
-        customIcon: "scheduled",
-        scheduleId,
-        scheduleInstanceId,
-        queueTimestamp: exactScheduleTime,
-      }
-    );
-
-    return { success: !!result };
-  } catch (error) {
-    return {
-      success: false,
-      error: error instanceof Error ? error.message : String(error),
-    };
-  }
-};
-
 async function isDevEnvironmentConnectedHandler(environmentId: string) {
   const environment = await prisma.runtimeEnvironment.findFirst({
     where: {
@@ -114,7 +70,50 @@ function createScheduleEngine() {
     },
     tracer,
     meter,
-    onTriggerScheduledTask: triggerScheduledTaskCallback,
+    onTriggerScheduledTask: async ({
+      taskIdentifier,
+      environment,
+      payload,
+      scheduleInstanceId,
+      scheduleId,
+      exactScheduleTime,
+    }) => {
+      try {
+        // Use TriggerTaskServiceV1 for now (can be updated to use TriggerTaskService when ready)
+        const triggerService = new TriggerTaskService();
+
+        const payloadPacket = await stringifyIO(payload);
+
+        logger.debug("Triggering scheduled task", {
+          taskIdentifier,
+          environment,
+          payload,
+          scheduleInstanceId,
+          scheduleId,
+          exactScheduleTime,
+        });
+
+        const result = await triggerService.call(
+          taskIdentifier,
+          environment,
+          { payload: payloadPacket.data, options: { payloadType: payloadPacket.dataType } },
+          {
+            customIcon: "scheduled",
+            scheduleId,
+            scheduleInstanceId,
+            queueTimestamp: exactScheduleTime,
+            overrideCreatedAt: exactScheduleTime,
+          }
+        );
+
+        return { success: !!result };
+      } catch (error) {
+        return {
+          success: false,
+          error: error instanceof Error ? error.message : String(error),
+        };
+      }
+    },
     isDevEnvironmentConnectedHandler: isDevEnvironmentConnectedHandler,
   });
 
diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts
@@ -23,9 +23,9 @@ import { clampMaxDuration } from "../utils/maxDuration";
 import { BaseService, ServiceValidationError } from "./baseService.server";
 import { CheckScheduleService } from "./checkSchedule.server";
 import { projectPubSub } from "./projectPubSub.server";
-import { RegisterNextTaskScheduleInstanceService } from "./registerNextTaskScheduleInstance.server";
 import { tryCatch } from "@trigger.dev/core/v3";
 import { engine } from "../runEngine.server";
+import { scheduleEngine } from "../scheduleEngine.server";
 
 export class CreateBackgroundWorkerService extends BaseService {
   public async call(
@@ -510,7 +510,6 @@ export async function syncDeclarativeSchedules(
   });
 
   const checkSchedule = new CheckScheduleService(prisma);
-  const registerNextService = new RegisterNextTaskScheduleInstanceService(prisma);
 
   //start out by assuming they're all missing
   const missingSchedules = new Set<string>(
@@ -569,7 +568,7 @@ export async function syncDeclarativeSchedules(
       missingSchedules.delete(existingSchedule.id);
       const instance = schedule.instances.at(0);
       if (instance) {
-        await registerNextService.call(instance.id);
+        await scheduleEngine.registerNextTaskScheduleInstance({ instanceId: instance.id });
       } else {
         throw new CreateDeclarativeScheduleError(
           `Missing instance for declarative schedule ${schedule.id}`
@@ -601,7 +600,7 @@ export async function syncDeclarativeSchedules(
       const instance = newSchedule.instances.at(0);
 
       if (instance) {
-        await registerNextService.call(instance.id);
+        await scheduleEngine.registerNextTaskScheduleInstance({ instanceId: instance.id });
       } else {
         throw new CreateDeclarativeScheduleError(
           `Missing instance for declarative schedule ${newSchedule.id}`
diff --git a/apps/webapp/app/v3/services/triggerTask.server.ts b/apps/webapp/app/v3/services/triggerTask.server.ts
@@ -32,6 +32,7 @@ export type TriggerTaskServiceOptions = {
   scheduleId?: string;
   scheduleInstanceId?: string;
   queueTimestamp?: Date;
+  overrideCreatedAt?: Date;
 };
 
 export class OutOfEntitlementError extends Error {
diff --git a/apps/webapp/app/v3/services/triggerTaskV1.server.ts b/apps/webapp/app/v3/services/triggerTaskV1.server.ts
@@ -439,6 +439,7 @@ export class TriggerTaskServiceV1 extends BaseService {
                     machinePreset: body.options?.machine,
                     scheduleId: options.scheduleId,
                     scheduleInstanceId: options.scheduleInstanceId,
+                    createdAt: options.overrideCreatedAt,
                   },
                 });
 
diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts
@@ -366,6 +366,7 @@ export class RunEngine {
       runChainState,
       scheduleId,
       scheduleInstanceId,
+      createdAt,
     }: TriggerParams,
     tx?: PrismaClientOrTransaction
   ): Promise<TaskRun> {
@@ -439,6 +440,7 @@ export class RunEngine {
               runChainState,
               scheduleId,
               scheduleInstanceId,
+              createdAt,
               executionSnapshots: {
                 create: {
                   engine: "V2",
diff --git a/internal-packages/run-engine/src/engine/types.ts b/internal-packages/run-engine/src/engine/types.ts
@@ -133,6 +133,7 @@ export type TriggerParams = {
   runChainState?: RunChainState;
   scheduleId?: string;
   scheduleInstanceId?: string;
+  createdAt?: Date;
 };
 
 export type EngineWorker = Worker<typeof workerCatalog>;
diff --git a/internal-packages/schedule-engine/src/engine/index.ts b/internal-packages/schedule-engine/src/engine/index.ts
@@ -1,20 +1,13 @@
-import {
-  Counter,
-  getMeter,
-  Histogram,
-  Meter,
-  startSpan,
-  Tracer,
-  UpDownCounter,
-} from "@internal/tracing";
+import { Counter, getMeter, Histogram, Meter, startSpan, Tracer } from "@internal/tracing";
 import { Logger } from "@trigger.dev/core/logger";
 import { PrismaClient } from "@trigger.dev/database";
-import { Worker } from "@trigger.dev/redis-worker";
+import { Worker, type JobHandlerParams } from "@trigger.dev/redis-worker";
 import { calculateDistributedExecutionTime } from "./distributedScheduling.js";
 import { calculateNextScheduledTimestamp, nextScheduledTimestamps } from "./scheduleCalculation.js";
 import {
   RegisterScheduleInstanceParams,
   ScheduleEngineOptions,
+  TriggerScheduledTaskCallback,
   TriggerScheduleParams,
 } from "./types.js";
 import { scheduleWorkerCatalog } from "./workerCatalog.js";
@@ -37,11 +30,14 @@ export class ScheduleEngine {
 
   prisma: PrismaClient;
 
+  private onTriggerScheduledTask: TriggerScheduledTaskCallback;
+
   constructor(private readonly options: ScheduleEngineOptions) {
     this.logger =
       options.logger ?? new Logger("ScheduleEngine", (this.options.logLevel ?? "info") as any);
     this.prisma = options.prisma;
     this.distributionWindowSeconds = options.distributionWindow?.seconds ?? 30;
+    this.onTriggerScheduledTask = options.onTriggerScheduledTask;
 
     this.tracer = options.tracer ?? (startSpan as any).tracer;
     this.meter = options.meter ?? getMeter("schedule-engine");
@@ -93,13 +89,7 @@ export class ScheduleEngine {
       shutdownTimeoutMs: options.worker.shutdownTimeoutMs,
       logger: new Logger("ScheduleEngineWorker", "debug"),
       jobs: {
-        "schedule.triggerScheduledTask": async ({ payload }) => {
-          await this.triggerScheduledTask({
-            instanceId: payload.instanceId,
-            finalAttempt: false, // TODO: implement retry logic
-            exactScheduleTime: payload.exactScheduleTime,
-          });
-        },
+        "schedule.triggerScheduledTask": this.#handleTriggerScheduledTaskJob.bind(this),
       },
     });
 
@@ -224,6 +214,16 @@ export class ScheduleEngine {
     });
   }
 
+  async #handleTriggerScheduledTaskJob({
+    payload,
+  }: JobHandlerParams<typeof scheduleWorkerCatalog, "schedule.triggerScheduledTask">) {
+    await this.triggerScheduledTask({
+      instanceId: payload.instanceId,
+      finalAttempt: false, // TODO: implement retry logic
+      exactScheduleTime: payload.exactScheduleTime,
+    });
+  }
+
   /**
    * Triggers a scheduled task (called by the Redis worker)
    */
@@ -410,7 +410,7 @@ export class ScheduleEngine {
 
           // Rewritten try/catch to use tryCatch utility
           const [triggerError, result] = await tryCatch(
-            this.options.onTriggerScheduledTask({
+            this.onTriggerScheduledTask({
               taskIdentifier: instance.taskSchedule.taskIdentifier,
               environment: instance.environment,
               payload,
diff --git a/internal-packages/schedule-engine/src/engine/scheduleCalculation.ts b/internal-packages/schedule-engine/src/engine/scheduleCalculation.ts
@@ -7,9 +7,16 @@ export function calculateNextScheduledTimestampFromNow(schedule: string, timezon
 export function calculateNextScheduledTimestamp(
   schedule: string,
   timezone: string | null,
-  currentDate: Date = new Date()
+  lastScheduledTimestamp: Date = new Date()
 ) {
-  return calculateNextStep(schedule, timezone, currentDate);
+  const nextStep = calculateNextStep(schedule, timezone, lastScheduledTimestamp);
+
+  if (nextStep.getTime() < Date.now()) {
+    // If the next step is in the past, we just need to calculate the next step from now
+    return calculateNextStep(schedule, timezone, new Date());
+  }
+
+  return nextStep;
 }
 
 function calculateNextStep(schedule: string, timezone: string | null, currentDate: Date) {
diff --git a/packages/redis-worker/src/worker.ts b/packages/redis-worker/src/worker.ts
@@ -34,13 +34,17 @@ type QueueCatalogFromWorkerCatalog<Catalog extends WorkerCatalog> = {
   [K in keyof Catalog]: Catalog[K]["schema"];
 };
 
-type JobHandler<Catalog extends WorkerCatalog, K extends keyof Catalog> = (params: {
+export type JobHandlerParams<Catalog extends WorkerCatalog, K extends keyof Catalog> = {
   id: string;
   payload: z.infer<Catalog[K]["schema"]>;
   visibilityTimeoutMs: number;
   attempt: number;
   deduplicationKey?: string;
-}) => Promise<void>;
+};
+
+export type JobHandler<Catalog extends WorkerCatalog, K extends keyof Catalog> = (
+  params: JobHandlerParams<Catalog, K>
+) => Promise<void>;
 
 export type WorkerConcurrencyOptions = {
   workers?: number;
diff --git a/references/hello-world/src/trigger/schedule.ts b/references/hello-world/src/trigger/schedule.ts
@@ -2,7 +2,8 @@ import { schedules } from "@trigger.dev/sdk/v3";
 
 export const simpleSchedule = schedules.task({
   id: "simple-schedule",
-  cron: "0 0 * * *",
+  // Every other minute
+  cron: "*/2 * * * *",
   run: async (payload, { ctx }) => {
     return {
       message: "Hello, world!",

Original file line number	Diff line number	Diff line change
`@@ -91,6 +91,10 @@ export class RunsRepository {`
`91`	`91`
`92`	`92`	`if (options.to) {`
`93`	`93`	`queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", { to: options.to });`
	`94`	`+ } else {`
	`95`	`+ queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", {`
	`96`	`+ to: Date.now(),`
	`97`	`+ });`
`94`	`98`	`}`
`95`	`99`
`96`	`100`	`if (typeof options.isTest === "boolean") {`