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

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

@@ -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 };
   }