v3: various schedule fixes (#1040)

* Improve the SDK function types and expose a new APIError instead of the APIResult type

* Skip triggering scheduled tasks if the task isn’t found in the current deployment

Also fixes an issue when editing the environments of a schedule

Author: ericallam

.changeset/ninety-pets-travel.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Improve the SDK function types and expose a new APIError instead of the APIResult type

apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts

@@ -6,6 +6,7 @@ import { prisma } from "~/db.server";
 import { ViewSchedulePresenter } from "~/presenters/v3/ViewSchedulePresenter.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { UpsertSchedule } from "~/v3/schedules";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { UpsertTaskScheduleService } from "~/v3/services/upsertTaskSchedule.server";
 
 const ParamsSchema = z.object({
@@ -87,6 +88,10 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
         return json(responseObject, { status: 200 });
       } catch (error) {
+        if (error instanceof ServiceValidationError) {
+          return json({ error: error.message }, { status: 422 });
+        }
+
         return json(
           { error: error instanceof Error ? error.message : "Internal Server Error" },
           { status: 500 }

apps/webapp/app/routes/api.v1.schedules.ts

@@ -5,6 +5,7 @@ import { z } from "zod";
 import { ScheduleListPresenter } from "~/presenters/v3/ScheduleListPresenter.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { UpsertSchedule } from "~/v3/schedules";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { UpsertTaskScheduleService } from "~/v3/services/upsertTaskSchedule.server";
 
 const SearchParamsSchema = z.object({
@@ -63,6 +64,10 @@ export async function action({ request }: ActionFunctionArgs) {
 
     return json(responseObject, { status: 200 });
   } catch (error) {
+    if (error instanceof ServiceValidationError) {
+      return json({ error: error.message }, { status: 422 });
+    }
+
     return json(
       { error: error instanceof Error ? error.message : "Internal Server Error" },
       { status: 500 }

apps/webapp/app/v3/services/baseService.server.ts

@@ -32,3 +32,10 @@ export abstract class BaseService {
     );
   }
 }
+
+export class ServiceValidationError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "ServiceValidationError";
+  }
+}

apps/webapp/app/v3/services/triggerScheduledTask.server.ts

@@ -3,8 +3,10 @@ import { BaseService } from "./baseService.server";
 import { workerQueue } from "~/services/worker.server";
 import { RegisterNextTaskScheduleInstanceService } from "./registerNextTaskScheduleInstance.server";
 import { TriggerTaskService } from "./triggerTask.server";
-import { logger, stringifyIO } from "@trigger.dev/core/v3";
+import { stringifyIO } from "@trigger.dev/core/v3";
 import { nextScheduledTimestamps } from "../utils/calculateNextSchedule.server";
+import { findCurrentWorkerDeployment } from "../models/workerDeployment.server";
+import { logger } from "~/services/logger.server";
 
 export class TriggerScheduledTaskService extends BaseService {
   public async call(instanceId: string) {
@@ -49,6 +51,40 @@ export class TriggerScheduledTaskService extends BaseService {
       shouldTrigger = false;
     }
 
+    if (instance.environment.type !== "DEVELOPMENT") {
+      // Get the current backgroundWorker for this environment
+      const currentWorkerDeployment = await findCurrentWorkerDeployment(instance.environment.id);
+
+      if (!currentWorkerDeployment) {
+        logger.debug("No current worker deployment found, skipping task trigger", {
+          instanceId,
+          scheduleId: instance.taskSchedule.friendlyId,
+          environmentId: instance.environment.id,
+        });
+
+        shouldTrigger = false;
+      } else if (
+        !currentWorkerDeployment.worker ||
+        !currentWorkerDeployment.worker.tasks.some(
+          (t) => t.id === instance.taskSchedule.taskIdentifier
+        )
+      ) {
+        logger.debug(
+          "Current worker deployment does not contain the scheduled task identifier, skipping task trigger",
+          {
+            instanceId,
+            scheduleId: instance.taskSchedule.friendlyId,
+            environmentId: instance.environment.id,
+            workerDeploymentId: currentWorkerDeployment.id,
+            workerId: currentWorkerDeployment.worker?.id,
+            taskIdentifier: instance.taskSchedule.taskIdentifier,
+          }
+        );
+
+        shouldTrigger = false;
+      }
+    }
+
     const registerNextService = new RegisterNextTaskScheduleInstanceService();
 
     if (shouldTrigger) {

apps/webapp/app/v3/services/upsertTaskSchedule.server.ts

@@ -4,7 +4,7 @@ import { ZodError } from "zod";
 import { $transaction, PrismaClientOrTransaction } from "~/db.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
 import { CronPattern, UpsertSchedule } from "../schedules";
-import { BaseService } from "./baseService.server";
+import { BaseService, ServiceValidationError } from "./baseService.server";
 import { RegisterNextTaskScheduleInstanceService } from "./registerNextTaskScheduleInstance.server";
 import cronstrue from "cronstrue";
 import { calculateNextScheduledTimestamp } from "../utils/calculateNextSchedule.server";
@@ -32,10 +32,10 @@ export class UpsertTaskScheduleService extends BaseService {
       CronPattern.parse(schedule.cron);
     } catch (e) {
       if (e instanceof ZodError) {
-        throw new Error(`Invalid cron expression: ${e.issues[0].message}`);
+        throw new ServiceValidationError(`Invalid cron expression: ${e.issues[0].message}`);
       }
 
-      throw new Error(
+      throw new ServiceValidationError(
         `Invalid cron expression: ${e instanceof Error ? e.message : JSON.stringify(e)}`
       );
     }
@@ -186,26 +186,15 @@ export class UpsertTaskScheduleService extends BaseService {
     });
 
     // create the new instances
-    let instances: InstanceWithEnvironment[] = [];
+    const newInstances: InstanceWithEnvironment[] = [];
+    const updatingInstances: InstanceWithEnvironment[] = [];
 
     for (const environmentId of options.environments) {
       const existingInstance = existingInstances.find((i) => i.environmentId === environmentId);
 
       if (existingInstance) {
-        if (!existingInstance.active) {
-          // If the instance is not active, we need to activate it
-          await tx.taskScheduleInstance.update({
-            where: {
-              id: existingInstance.id,
-            },
-            data: {
-              active: true,
-            },
-          });
-        }
-
         // Update the existing instance
-        instances.push({ ...existingInstance, active: true });
+        updatingInstances.push(existingInstance);
       } else {
         // Create a new instance
         const instance = await tx.taskScheduleInstance.create({
@@ -226,35 +215,53 @@ export class UpsertTaskScheduleService extends BaseService {
           },
         });
 
-        instances.push(instance);
+        newInstances.push(instance);
       }
     }
 
     // find the instances that need to be removed
-    const instancesToDeactivate = existingInstances.filter(
+    const instancesToDeleted = existingInstances.filter(
       (i) => !options.environments.includes(i.environmentId)
     );
 
-    // deactivate the instances
-    for (const instance of instancesToDeactivate) {
-      await tx.taskScheduleInstance.update({
+    // delete the instances no longer selected
+    for (const instance of instancesToDeleted) {
+      await tx.taskScheduleInstance.delete({
         where: {
           id: instance.id,
         },
-        data: {
-          active: false,
-        },
       });
     }
 
-    if (scheduleHasChanged) {
-      const registerService = new RegisterNextTaskScheduleInstanceService(tx);
+    const registerService = new RegisterNextTaskScheduleInstanceService(tx);
+
+    for (const instance of newInstances) {
+      await registerService.call(instance.id);
+    }
 
-      for (const instance of existingInstances) {
+    if (scheduleHasChanged) {
+      for (const instance of updatingInstances) {
         await registerService.call(instance.id);
       }
     }
 
+    const instances = await tx.taskScheduleInstance.findMany({
+      where: {
+        taskScheduleId: scheduleRecord.id,
+      },
+      include: {
+        environment: {
+          include: {
+            orgMember: {
+              include: {
+                user: true,
+              },
+            },
+          },
+        },
+      },
+    });
+
     return { scheduleRecord, instances };
   }
 

docs/v3-openapi.json

@@ -40,6 +40,9 @@
           "400": {
             "description": "Invalid request parameters"
           },
+          "422": {
+            "description": "Unprocessable Entity"
+          },
           "401": {
             "description": "Unauthorized"
           }
@@ -216,6 +219,9 @@
           },
           "404": {
             "description": "Resource not found"
+          },
+          "422": {
+            "description": "Unprocessable Entity"
           }
         },
         "tags": [

packages/core/package.json

@@ -76,7 +76,8 @@
     "superjson": "^2.2.1",
     "ulidx": "^2.2.1",
     "zod": "3.22.3",
-    "zod-error": "1.5.0"
+    "zod-error": "1.5.0",
+    "zod-validation-error": "^1.5.0"
   },
   "devDependencies": {
     "@trigger.dev/tsconfig": "workspace:*",

packages/core/src/v3/apiClient/index.ts

@@ -1,5 +1,5 @@
 import { context, propagation } from "@opentelemetry/api";
-import { ZodFetchOptions, zodfetch } from "../../zodfetch";
+import { ZodFetchOptions, zodfetch } from "../zodfetch";
 import {
   BatchTriggerTaskRequestBody,
   BatchTriggerTaskResponse,
@@ -25,7 +25,7 @@ export type TriggerOptions = {
 
 const zodFetchOptions: ZodFetchOptions = {
   retry: {
-    maxAttempts: 5,
+    maxAttempts: 3,
     minTimeoutInMs: 1000,
     maxTimeoutInMs: 30_000,
     factor: 2,