fix(webapp): worker actions now catch service validation errors and respond properly (#2481)

This also stops all the unnecessary error logging when throwing 
ServiceValidationErrors

Author: ericallam

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -1,3 +1,4 @@
+import { EngineServiceValidationError } from "@internal/run-engine";
 import { json } from "@remix-run/server-runtime";
 import {
   generateJWT as internal_generateJWT,
@@ -8,7 +9,6 @@ import { TaskRun } from "@trigger.dev/database";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { env } from "~/env.server";
-import { EngineServiceValidationError } from "~/runEngine/concerns/errors";
 import { ApiAuthenticationResultSuccess, getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";

apps/webapp/app/runEngine/concerns/errors.ts

@@ -3,7 +3,7 @@ import { PayloadProcessor, TriggerTaskRequest } from "../types";
 import { env } from "~/env.server";
 import { startActiveSpan } from "~/v3/tracer.server";
 import { uploadPacketToObjectStore } from "~/v3/r2.server";
-import { EngineServiceValidationError } from "./errors";
+import { ServiceValidationError } from "~/v3/services/common.server";
 
 export class DefaultPayloadProcessor implements PayloadProcessor {
   async process(request: TriggerTaskRequest): Promise<IOPacket> {
@@ -36,10 +36,7 @@ export class DefaultPayloadProcessor implements PayloadProcessor {
       );
 
       if (uploadError) {
-        throw new EngineServiceValidationError(
-          "Failed to upload large payload to object store",
-          500
-        ); // This is retryable
+        throw new ServiceValidationError("Failed to upload large payload to object store", 500); // This is retryable
       }
 
       return {

apps/webapp/app/runEngine/concerns/payloads.server.ts

@@ -13,8 +13,8 @@ import {
 import { WorkerGroupService } from "~/v3/services/worker/workerGroupService.server";
 import type { RunEngine } from "~/v3/runEngine.server";
 import { env } from "~/env.server";
-import { EngineServiceValidationError } from "./errors";
 import { tryCatch } from "@trigger.dev/core/v3";
+import { ServiceValidationError } from "~/v3/services/common.server";
 
 export class DefaultQueueManager implements QueueManager {
   constructor(
@@ -45,7 +45,7 @@ export class DefaultQueueManager implements QueueManager {
         });
 
         if (!specifiedQueue) {
-          throw new EngineServiceValidationError(
+          throw new ServiceValidationError(
             `Specified queue '${specifiedQueueName}' not found or not associated with locked version '${
               lockedBackgroundWorker.version ?? "<unknown>"
             }'.`
@@ -68,7 +68,7 @@ export class DefaultQueueManager implements QueueManager {
         });
 
         if (!lockedTask) {
-          throw new EngineServiceValidationError(
+          throw new ServiceValidationError(
             `Task '${request.taskId}' not found on locked version '${
               lockedBackgroundWorker.version ?? "<unknown>"
             }'.`
@@ -83,7 +83,7 @@ export class DefaultQueueManager implements QueueManager {
             workerId: lockedBackgroundWorker.id,
             version: lockedBackgroundWorker.version,
           });
-          throw new EngineServiceValidationError(
+          throw new ServiceValidationError(
             `Default queue configuration for task '${request.taskId}' missing on locked version '${
               lockedBackgroundWorker.version ?? "<unknown>"
             }'.`
@@ -97,7 +97,7 @@ export class DefaultQueueManager implements QueueManager {
       // Task is not locked to a specific version, use regular logic
       if (request.body.options?.lockToVersion) {
         // This should only happen if the findFirst failed, indicating the version doesn't exist
-        throw new EngineServiceValidationError(
+        throw new ServiceValidationError(
           `Task locked to version '${request.body.options.lockToVersion}', but no worker found with that version.`
         );
       }
@@ -221,11 +221,11 @@ export class DefaultQueueManager implements QueueManager {
     );
 
     if (error) {
-      throw new EngineServiceValidationError(error.message);
+      throw new ServiceValidationError(error.message);
     }
 
     if (!workerGroup) {
-      throw new EngineServiceValidationError("No worker group found");
+      throw new ServiceValidationError("No worker group found");
     }
 
     return workerGroup.masterQueue;

apps/webapp/app/runEngine/concerns/queues.server.ts

@@ -31,7 +31,6 @@ import type {
 } from "../../v3/services/triggerTask.server";
 import { getTaskEventStore } from "../../v3/taskEventStore.server";
 import { clampMaxDuration } from "../../v3/utils/maxDuration";
-import { EngineServiceValidationError } from "../concerns/errors";
 import { IdempotencyKeyConcern } from "../concerns/idempotencyKeys.server";
 import type {
   PayloadProcessor,
@@ -41,6 +40,7 @@ import type {
   TriggerTaskRequest,
   TriggerTaskValidator,
 } from "../types";
+import { ServiceValidationError } from "~/v3/services/common.server";
 
 export class RunEngineTriggerTaskService {
   private readonly queueConcern: QueueManager;
@@ -157,7 +157,7 @@ export class RunEngineTriggerTaskService {
       const [parseDelayError, delayUntil] = await tryCatch(parseDelay(body.options?.delay));
 
       if (parseDelayError) {
-        throw new EngineServiceValidationError(`Invalid delay ${body.options?.delay}`);
+        throw new ServiceValidationError(`Invalid delay ${body.options?.delay}`);
       }
 
       const ttl =
@@ -210,7 +210,7 @@ export class RunEngineTriggerTaskService {
         });
 
         if (!queueSizeGuard.ok) {
-          throw new EngineServiceValidationError(
+          throw new ServiceValidationError(
             `Cannot trigger ${taskId} as the queue size limit for this environment has been reached. The maximum size is ${queueSizeGuard.maximumSize}`
           );
         }
@@ -351,7 +351,7 @@ export class RunEngineTriggerTaskService {
           );
 
           if (result?.error) {
-            throw new EngineServiceValidationError(
+            throw new ServiceValidationError(
               taskRunErrorToString(taskRunErrorEnhancer(result.error))
             );
           }
@@ -365,7 +365,7 @@ export class RunEngineTriggerTaskService {
         }
 
         if (error instanceof RunOneTimeUseTokenError) {
-          throw new EngineServiceValidationError(
+          throw new ServiceValidationError(
             `Cannot trigger ${taskId} with a one-time use token as it has already been used.`
           );
         }

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -3,7 +3,6 @@ import { logger } from "~/services/logger.server";
 import { getEntitlement } from "~/services/platform.v3.server";
 import { MAX_ATTEMPTS, OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { isFinalRunStatus } from "~/v3/taskStatus";
-import { EngineServiceValidationError } from "../concerns/errors";
 import type {
   EntitlementValidationParams,
   EntitlementValidationResult,
@@ -13,6 +12,7 @@ import type {
   TriggerTaskValidator,
   ValidationResult,
 } from "../types";
+import { ServiceValidationError } from "~/v3/services/common.server";
 
 export class DefaultTriggerTaskValidator implements TriggerTaskValidator {
   validateTags(params: TagValidationParams): ValidationResult {
@@ -29,7 +29,7 @@ export class DefaultTriggerTaskValidator implements TriggerTaskValidator {
     if (tags.length > MAX_TAGS_PER_RUN) {
       return {
         ok: false,
-        error: new EngineServiceValidationError(
+        error: new ServiceValidationError(
           `Runs can only have ${MAX_TAGS_PER_RUN} tags, you're trying to set ${tags.length}.`
         ),
       };
@@ -65,7 +65,7 @@ export class DefaultTriggerTaskValidator implements TriggerTaskValidator {
     if (attempt > MAX_ATTEMPTS) {
       return {
         ok: false,
-        error: new EngineServiceValidationError(
+        error: new ServiceValidationError(
           `Failed to trigger ${taskId} after ${MAX_ATTEMPTS} attempts.`
         ),
       };
@@ -95,7 +95,7 @@ export class DefaultTriggerTaskValidator implements TriggerTaskValidator {
 
       return {
         ok: false,
-        error: new EngineServiceValidationError(
+        error: new ServiceValidationError(
           `Cannot trigger ${taskId} as the parent run has a status of ${parentRun.status}`
         ),
       };

apps/webapp/app/runEngine/validators/triggerTaskValidator.ts

@@ -23,6 +23,8 @@ import {
 } from "~/v3/services/worker/workerGroupTokenService.server";
 import { API_VERSIONS, getApiVersion } from "~/api/versions";
 import { WORKER_HEADERS } from "@trigger.dev/core/v3/runEngineWorker";
+import { ServiceValidationError } from "~/v3/services/common.server";
+import { EngineServiceValidationError } from "@internal/run-engine";
 
 type AnyZodSchema = z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>;
 
@@ -1040,11 +1042,18 @@ export function createActionWorkerApiRoute<
       });
       return result;
     } catch (error) {
-      console.error("Error in API route:", error);
       if (error instanceof Response) {
         return error;
       }
 
+      if (error instanceof EngineServiceValidationError) {
+        return json({ error: error.message }, { status: error.status ?? 422 });
+      }
+
+      if (error instanceof ServiceValidationError) {
+        return json({ error: error.message }, { status: error.status ?? 422 });
+      }
+
       logger.error("Error in action", {
         error:
           error instanceof Error

apps/webapp/app/services/routeBuilders/apiBuilder.server.ts

@@ -1,4 +1,3 @@
-import { BillingCache } from "./billingCache.js";
 import { createRedisClient, Redis } from "@internal/redis";
 import { getMeter, Meter, startSpan, trace, Tracer } from "@internal/tracing";
 import { Logger } from "@trigger.dev/core/logger";
@@ -32,6 +31,7 @@ import { FairQueueSelectionStrategy } from "../run-queue/fairQueueSelectionStrat
 import { RunQueue } from "../run-queue/index.js";
 import { RunQueueFullKeyProducer } from "../run-queue/keyProducer.js";
 import { MinimalAuthenticatedEnvironment } from "../shared/index.js";
+import { BillingCache } from "./billingCache.js";
 import { NotImplementedError, RunDuplicateIdempotencyKeyError } from "./errors.js";
 import { EventBus, EventBusEvents } from "./eventBus.js";
 import { RunLocker } from "./locking.js";

internal-packages/run-engine/src/engine/index.ts

@@ -15,6 +15,7 @@ import {
   Attributes,
   Histogram,
 } from "@internal/tracing";
+import { ServiceValidationError } from "./errors.js";
 
 const SemanticAttributes = {
   LOCK_TYPE: "run_engine.lock.type",
@@ -174,6 +175,10 @@ export class RunLocker {
         );
 
         if (error) {
+          if (error instanceof ServiceValidationError) {
+            throw error;
+          }
+
           // Record failed lock acquisition
           const lockDuration = performance.now() - lockStartTime;
           this.lockDurationHistogram.record(lockDuration, {
@@ -186,6 +191,7 @@ export class RunLocker {
             resources,
             duration: this.duration,
           });
+
           throw error;
         }
 

internal-packages/run-engine/src/engine/locking.ts

@@ -1,4 +1,8 @@
 export { RunEngine } from "./engine/index.js";
-export { RunDuplicateIdempotencyKeyError, RunOneTimeUseTokenError } from "./engine/errors.js";
+export {
+  RunDuplicateIdempotencyKeyError,
+  RunOneTimeUseTokenError,
+  ServiceValidationError as EngineServiceValidationError,
+} from "./engine/errors.js";
 export type { EventBusEventArgs, EventBusEvents } from "./engine/eventBus.js";
 export type { AuthenticatedEnvironment } from "./shared/index.js";