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

.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 existingInstances) {
+    for (const instance of newInstances) {
+      await registerService.call(instance.id);
+    }
+
+    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,

packages/core/src/v3/apiErrors.ts

@@ -0,0 +1,142 @@
+export type APIHeaders = Record<string, string | null | undefined>;
+
+export class APIError extends Error {
+  readonly status: number | undefined;
+  readonly headers: APIHeaders | undefined;
+  readonly error: Object | undefined;
+
+  readonly code: string | null | undefined;
+  readonly param: string | null | undefined;
+  readonly type: string | undefined;
+
+  constructor(
+    status: number | undefined,
+    error: Object | undefined,
+    message: string | undefined,
+    headers: APIHeaders | undefined
+  ) {
+    super(`${APIError.makeMessage(status, error, message)}`);
+    this.status = status;
+    this.headers = headers;
+
+    const data = error as Record<string, any>;
+    this.error = data;
+    this.code = data?.["code"];
+    this.param = data?.["param"];
+    this.type = data?.["type"];
+  }
+
+  private static makeMessage(status: number | undefined, error: any, message: string | undefined) {
+    const msg = error?.message
+      ? typeof error.message === "string"
+        ? error.message
+        : JSON.stringify(error.message)
+      : error
+      ? JSON.stringify(error)
+      : message;
+
+    if (status && msg) {
+      return `${status} ${msg}`;
+    }
+    if (status) {
+      return `${status} status code (no body)`;
+    }
+    if (msg) {
+      return msg;
+    }
+    return "(no status code or body)";
+  }
+
+  static generate(
+    status: number | undefined,
+    errorResponse: Object | undefined,
+    message: string | undefined,
+    headers: APIHeaders | undefined
+  ) {
+    if (!status) {
+      return new APIConnectionError({ cause: castToError(errorResponse) });
+    }
+
+    const error = (errorResponse as Record<string, any>)?.["error"];
+
+    if (status === 400) {
+      return new BadRequestError(status, error, message, headers);
+    }
+
+    if (status === 401) {
+      return new AuthenticationError(status, error, message, headers);
+    }
+
+    if (status === 403) {
+      return new PermissionDeniedError(status, error, message, headers);
+    }
+
+    if (status === 404) {
+      return new NotFoundError(status, error, message, headers);
+    }
+
+    if (status === 409) {
+      return new ConflictError(status, error, message, headers);
+    }
+
+    if (status === 422) {
+      return new UnprocessableEntityError(status, error, message, headers);
+    }
+
+    if (status === 429) {
+      return new RateLimitError(status, error, message, headers);
+    }
+
+    if (status >= 500) {
+      return new InternalServerError(status, error, message, headers);
+    }
+
+    return new APIError(status, error, message, headers);
+  }
+}
+
+export class APIConnectionError extends APIError {
+  override readonly status: undefined = undefined;
+
+  constructor({ message, cause }: { message?: string; cause?: Error | undefined }) {
+    super(undefined, undefined, message || "Connection error.", undefined);
+    // in some environments the 'cause' property is already declared
+    // @ts-ignore
+    if (cause) this.cause = cause;
+  }
+}
+
+export class BadRequestError extends APIError {
+  override readonly status: 400 = 400;
+}
+
+export class AuthenticationError extends APIError {
+  override readonly status: 401 = 401;
+}
+
+export class PermissionDeniedError extends APIError {
+  override readonly status: 403 = 403;
+}
+
+export class NotFoundError extends APIError {
+  override readonly status: 404 = 404;
+}
+
+export class ConflictError extends APIError {
+  override readonly status: 409 = 409;
+}
+
+export class UnprocessableEntityError extends APIError {
+  override readonly status: 422 = 422;
+}
+
+export class RateLimitError extends APIError {
+  override readonly status: 429 = 429;
+}
+
+export class InternalServerError extends APIError {}
+
+function castToError(err: any): Error {
+  if (err instanceof Error) return err;
+  return new Error(err);
+}

packages/core/src/v3/index.ts

@@ -7,6 +7,7 @@ export * from "./zodNamespace";
 export * from "./zodSocket";
 export * from "./zodIpc";
 export * from "./errors";
+export * from "./apiErrors";
 export * from "./runtime-api";
 export * from "./logger-api";
 export * from "./clock-api";

packages/core/src/v3/utils/ioSerialization.ts

@@ -110,27 +110,25 @@ async function exportPacket(packet: IOPacket, pathPrefix: string): Promise<IOPac
 
   const presignedResponse = await apiClientManager.client!.createUploadPayloadUrl(filename);
 
-  if (presignedResponse.ok) {
-    const uploadResponse = await fetch(presignedResponse.data.presignedUrl, {
-      method: "PUT",
-      headers: {
-        "Content-Type": packet.dataType,
-      },
-      body: packet.data,
-    });
+  const uploadResponse = await fetch(presignedResponse.presignedUrl, {
+    method: "PUT",
+    headers: {
+      "Content-Type": packet.dataType,
+    },
+    body: packet.data,
+  });
 
-    if (!uploadResponse.ok) {
-      throw new Error(
-        `Failed to upload output to ${presignedResponse.data.presignedUrl}: ${uploadResponse.statusText}`
-      );
-    }
-
-    return {
-      data: filename,
-      dataType: "application/store",
-    };
+  if (!uploadResponse.ok) {
+    throw new Error(
+      `Failed to upload output to ${presignedResponse.presignedUrl}: ${uploadResponse.statusText}`
+    );
   }
 
+  return {
+    data: filename,
+    dataType: "application/store",
+  };
+
   return packet;
 }
 
@@ -172,25 +170,23 @@ async function importPacket(packet: IOPacket, span?: Span): Promise<IOPacket> {
 
   const presignedResponse = await apiClientManager.client.getPayloadUrl(packet.data);
 
-  if (presignedResponse.ok) {
-    const response = await fetch(presignedResponse.data.presignedUrl);
+  const response = await fetch(presignedResponse.presignedUrl);
 
-    if (!response.ok) {
-      throw new Error(
-        `Failed to import packet ${presignedResponse.data.presignedUrl}: ${response.statusText}`
-      );
-    }
-
-    const data = await response.text();
-
-    span?.setAttribute("size", Buffer.byteLength(data, "utf8"));
-
-    return {
-      data,
-      dataType: response.headers.get("content-type") ?? "application/json",
-    };
+  if (!response.ok) {
+    throw new Error(
+      `Failed to import packet ${presignedResponse.presignedUrl}: ${response.statusText}`
+    );
   }
 
+  const data = await response.text();
+
+  span?.setAttribute("size", Buffer.byteLength(data, "utf8"));
+
+  return {
+    data,
+    dataType: response.headers.get("content-type") ?? "application/json",
+  };
+
   return packet;
 }
 

packages/core/src/v3/zodfetch.ts

@@ -0,0 +1,174 @@
+import { z } from "zod";
+import { fromZodError } from "zod-validation-error";
+import { APIConnectionError, APIError } from "./apiErrors";
+import { RetryOptions } from "./schemas";
+import { calculateNextRetryDelay } from "./utils/retries";
+
+export const defaultRetryOptions = {
+  maxAttempts: 3,
+  factor: 2,
+  minTimeoutInMs: 1000,
+  maxTimeoutInMs: 60000,
+  randomize: false,
+} satisfies RetryOptions;
+
+export type ZodFetchOptions = {
+  retry?: RetryOptions;
+};
+
+export async function zodfetch<TResponseBody extends any>(
+  schema: z.Schema<TResponseBody>,
+  url: string,
+  requestInit?: RequestInit,
+  options?: ZodFetchOptions
+): Promise<TResponseBody> {
+  return await _doZodFetch(schema, url, requestInit, options);
+}
+
+async function _doZodFetch<TResponseBody extends any>(
+  schema: z.Schema<TResponseBody>,
+  url: string,
+  requestInit?: RequestInit,
+  options?: ZodFetchOptions,
+  attempt = 1
+): Promise<TResponseBody> {
+  try {
+    const response = await fetch(url, requestInitWithCache(requestInit));
+
+    const responseHeaders = createResponseHeaders(response.headers);
+
+    if (!response.ok) {
+      const retryResult = shouldRetry(response, attempt, options?.retry);
+
+      if (retryResult.retry) {
+        await new Promise((resolve) => setTimeout(resolve, retryResult.delay));
+
+        return await _doZodFetch(schema, url, requestInit, options, attempt + 1);
+      } else {
+        const errText = await response.text().catch((e) => castToError(e).message);
+        const errJSON = safeJsonParse(errText);
+        const errMessage = errJSON ? undefined : errText;
+
+        throw APIError.generate(response.status, errJSON, errMessage, responseHeaders);
+      }
+    }
+
+    const jsonBody = await response.json();
+    const parsedResult = schema.safeParse(jsonBody);
+
+    if (parsedResult.success) {
+      return parsedResult.data;
+    }
+
+    throw fromZodError(parsedResult.error);
+  } catch (error) {
+    if (error instanceof APIError) {
+      throw error;
+    }
+
+    if (options?.retry) {
+      const retry = { ...defaultRetryOptions, ...options.retry };
+
+      const delay = calculateNextRetryDelay(retry, attempt);
+
+      if (delay) {
+        await new Promise((resolve) => setTimeout(resolve, delay));
+
+        return await _doZodFetch(schema, url, requestInit, options, attempt + 1);
+      }
+    }
+
+    throw new APIConnectionError({ cause: castToError(error) });
+  }
+}
+
+function castToError(err: any): Error {
+  if (err instanceof Error) return err;
+  return new Error(err);
+}
+
+type ShouldRetryResult =
+  | {
+      retry: false;
+    }
+  | {
+      retry: true;
+      delay: number;
+    };
+
+function shouldRetry(
+  response: Response,
+  attempt: number,
+  retryOptions?: RetryOptions
+): ShouldRetryResult {
+  function shouldRetryForOptions(): ShouldRetryResult {
+    const retry = { ...defaultRetryOptions, ...retryOptions };
+
+    const delay = calculateNextRetryDelay(retry, attempt);
+
+    if (delay) {
+      return { retry: true, delay };
+    } else {
+      return { retry: false };
+    }
+  }
+
+  // Note this is not a standard header.
+  const shouldRetryHeader = response.headers.get("x-should-retry");
+
+  // If the server explicitly says whether or not to retry, obey.
+  if (shouldRetryHeader === "true") return shouldRetryForOptions();
+  if (shouldRetryHeader === "false") return { retry: false };
+
+  // Retry on request timeouts.
+  if (response.status === 408) return shouldRetryForOptions();
+
+  // Retry on lock timeouts.
+  if (response.status === 409) return shouldRetryForOptions();
+
+  // Retry on rate limits.
+  if (response.status === 429) return shouldRetryForOptions();
+
+  // Retry internal errors.
+  if (response.status >= 500) return shouldRetryForOptions();
+
+  return { retry: false };
+}
+
+function safeJsonParse(text: string): any {
+  try {
+    return JSON.parse(text);
+  } catch (e) {
+    return undefined;
+  }
+}
+
+function createResponseHeaders(headers: Response["headers"]): Record<string, string> {
+  return new Proxy(
+    Object.fromEntries(
+      // @ts-ignore
+      headers.entries()
+    ),
+    {
+      get(target, name) {
+        const key = name.toString();
+        return target[key.toLowerCase()] || target[key];
+      },
+    }
+  );
+}
+
+function requestInitWithCache(requestInit?: RequestInit): RequestInit {
+  try {
+    const withCache: RequestInit = {
+      ...requestInit,
+      cache: "no-cache",
+    };
+
+    const _ = new Request("http://localhost", withCache);
+
+    return withCache;
+  } catch (error) {
+    return requestInit ?? {};
+  }
+}

packages/core/src/zodfetch.ts

@@ -1,120 +0,0 @@
-import { z } from "zod";
-import { RetryOptions, calculateNextRetryDelay, defaultRetryOptions } from "./v3";
-
-export type ApiResult<TSuccessResult> =
-  | { ok: true; data: TSuccessResult }
-  | {
-      ok: false;
-      error: string;
-    };
-
-export type ZodFetchOptions = {
-  retry?: RetryOptions;
-};
-
-export async function zodfetch<TResponseBody extends any>(
-  schema: z.Schema<TResponseBody>,
-  url: string,
-  requestInit?: RequestInit,
-  options?: ZodFetchOptions
-): Promise<ApiResult<TResponseBody>> {
-  return await _doZodFetch(schema, url, requestInit, options);
-}
-
-async function _doZodFetch<TResponseBody extends any>(
-  schema: z.Schema<TResponseBody>,
-  url: string,
-  requestInit?: RequestInit,
-  options?: ZodFetchOptions,
-  attempt = 1
-): Promise<ApiResult<TResponseBody>> {
-  try {
-    const response = await fetch(url, requestInit);
-
-    if ((!requestInit || requestInit.method === "GET") && response.status === 404) {
-      return {
-        ok: false,
-        error: `404: ${response.statusText}`,
-      };
-    }
-
-    if (response.status >= 400 && response.status < 500 && response.status !== 429) {
-      const body = await response.json();
-      if (!body.error) {
-        return { ok: false, error: "Something went wrong" };
-      }
-
-      return { ok: false, error: body.error };
-    }
-
-    // Retryable errors
-    if (response.status === 429 || response.status >= 500) {
-      if (!options?.retry) {
-        return {
-          ok: false,
-          error: `Failed to fetch ${url}, got status code ${response.status}`,
-        };
-      }
-
-      const retry = { ...defaultRetryOptions, ...options.retry };
-
-      if (attempt > retry.maxAttempts) {
-        return {
-          ok: false,
-          error: `Failed to fetch ${url}, got status code ${response.status}`,
-        };
-      }
-
-      const delay = calculateNextRetryDelay(retry, attempt);
-
-      await new Promise((resolve) => setTimeout(resolve, delay));
-
-      return await _doZodFetch(schema, url, requestInit, options, attempt + 1);
-    }
-
-    if (response.status !== 200) {
-      return {
-        ok: false,
-        error: `Failed to fetch ${url}, got status code ${response.status}`,
-      };
-    }
-
-    const jsonBody = await response.json();
-    const parsedResult = schema.safeParse(jsonBody);
-
-    if (parsedResult.success) {
-      return { ok: true, data: parsedResult.data };
-    }
-
-    if ("error" in jsonBody) {
-      return {
-        ok: false,
-        error: typeof jsonBody.error === "string" ? jsonBody.error : JSON.stringify(jsonBody.error),
-      };
-    }
-
-    return { ok: false, error: parsedResult.error.message };
-  } catch (error) {
-    if (options?.retry) {
-      const retry = { ...defaultRetryOptions, ...options.retry };
-
-      if (attempt > retry.maxAttempts) {
-        return {
-          ok: false,
-          error: error instanceof Error ? error.message : JSON.stringify(error),
-        };
-      }
-
-      const delay = calculateNextRetryDelay(retry, attempt);
-
-      await new Promise((resolve) => setTimeout(resolve, delay));
-
-      return await _doZodFetch(schema, url, requestInit, options, attempt + 1);
-    }
-
-    return {
-      ok: false,
-      error: error instanceof Error ? error.message : JSON.stringify(error),
-    };
-  }
-}

packages/trigger-sdk/src/v3/index.ts

@@ -8,7 +8,19 @@ export { queue } from "./shared";
 import type { Context } from "./shared";
 export type { Context };
 
-export { logger, type LogLevel } from "@trigger.dev/core/v3";
+export {
+  logger,
+  type LogLevel,
+  APIError,
+  BadRequestError,
+  AuthenticationError,
+  PermissionDeniedError,
+  NotFoundError,
+  ConflictError,
+  UnprocessableEntityError,
+  RateLimitError,
+  InternalServerError,
+} from "@trigger.dev/core/v3";
 
 export { runs } from "./management";
 export * as schedules from "./schedules";

packages/trigger-sdk/src/v3/management.ts

@@ -13,13 +13,7 @@ async function replayRun(runId: string): Promise<ReplayRunResponse> {
     throw apiClientMissingError();
   }
 
-  const response = await apiClient.replayRun(runId);
-
-  if (!response.ok) {
-    throw new Error(response.error);
-  }
-
-  return response.data;
+  return await apiClient.replayRun(runId);
 }
 
 async function cancelRun(runId: string): Promise<CanceledRunResponse> {
@@ -29,11 +23,5 @@ async function cancelRun(runId: string): Promise<CanceledRunResponse> {
     throw apiClientMissingError();
   }
 
-  const response = await apiClient.cancelRun(runId);
-
-  if (!response.ok) {
-    throw new Error(response.error);
-  }
-
-  return response.data;
+  return await apiClient.cancelRun(runId);
 }

packages/trigger-sdk/src/v3/schedules/index.ts

@@ -1,4 +1,11 @@
-import { InitOutput, apiClientManager, taskCatalog } from "@trigger.dev/core/v3";
+import {
+  DeletedScheduleObject,
+  InitOutput,
+  ListSchedulesResult,
+  ScheduleObject,
+  apiClientManager,
+  taskCatalog,
+} from "@trigger.dev/core/v3";
 import { Task, TaskOptions, apiClientMissingError, createTask } from "../shared";
 import * as SchedulesAPI from "./api";
 
@@ -23,7 +30,7 @@ export function task<TOutput, TInitOutput extends InitOutput>(
  * @param options.deduplicationKey - An optional deduplication key for the schedule
  * @returns The created schedule
  */
-export async function create(options: SchedulesAPI.CreateScheduleOptions) {
+export async function create(options: SchedulesAPI.CreateScheduleOptions): Promise<ScheduleObject> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {
@@ -38,7 +45,7 @@ export async function create(options: SchedulesAPI.CreateScheduleOptions) {
  * @param scheduleId - The ID of the schedule to retrieve
  * @returns The retrieved schedule
  */
-export async function retrieve(scheduleId: string) {
+export async function retrieve(scheduleId: string): Promise<ScheduleObject> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {
@@ -57,7 +64,10 @@ export async function retrieve(scheduleId: string) {
  * @param options.externalId - An optional external identifier for the schedule
  * @returns The updated schedule
  */
-export async function update(scheduleId: string, options: SchedulesAPI.UpdateScheduleOptions) {
+export async function update(
+  scheduleId: string,
+  options: SchedulesAPI.UpdateScheduleOptions
+): Promise<ScheduleObject> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {
@@ -71,7 +81,7 @@ export async function update(scheduleId: string, options: SchedulesAPI.UpdateSch
  * Deletes a schedule
  * @param scheduleId - The ID of the schedule to delete
  */
-export async function del(scheduleId: string) {
+export async function del(scheduleId: string): Promise<DeletedScheduleObject> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {
@@ -85,7 +95,7 @@ export async function del(scheduleId: string) {
  * Deactivates a schedule
  * @param scheduleId - The ID of the schedule to deactivate
  */
-export async function deactivate(scheduleId: string) {
+export async function deactivate(scheduleId: string): Promise<ScheduleObject> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {
@@ -99,7 +109,7 @@ export async function deactivate(scheduleId: string) {
  * Activates a schedule
  * @param scheduleId - The ID of the schedule to activate
  */
-export async function activate(scheduleId: string) {
+export async function activate(scheduleId: string): Promise<ScheduleObject> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {
@@ -116,7 +126,9 @@ export async function activate(scheduleId: string) {
  * @param options.perPage - The number of schedules per page
  * @returns The list of schedules
  */
-export async function list(options?: SchedulesAPI.ListScheduleOptions) {
+export async function list(
+  options?: SchedulesAPI.ListScheduleOptions
+): Promise<ListSchedulesResult> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {

packages/trigger-sdk/src/v3/shared.ts

@@ -245,13 +245,9 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
             { spanParentAsLink: true }
           );
 
-          if (!response.ok) {
-            throw new Error(response.error);
-          }
+          span.setAttribute("messaging.message.id", response.id);
 
-          span.setAttribute("messaging.message.id", response.data.id);
-
-          return response.data;
+          return response;
         },
         {
           kind: SpanKind.PRODUCER,
@@ -314,13 +310,9 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
             { spanParentAsLink: true }
           );
 
-          if (!response.ok) {
-            throw new Error(response.error);
-          }
+          span.setAttribute("messaging.message.id", response.batchId);
 
-          span.setAttribute("messaging.message.id", response.data.batchId);
-
-          return response.data;
+          return response;
         },
         {
           kind: SpanKind.PRODUCER,
@@ -384,14 +376,10 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
             },
           });
 
-          if (!response.ok) {
-            throw new Error(response.error);
-          }
-
-          span.setAttribute("messaging.message.id", response.data.id);
+          span.setAttribute("messaging.message.id", response.id);
 
           const result = await runtime.waitForTask({
-            id: response.data.id,
+            id: response.id,
             ctx,
           });
 
@@ -465,15 +453,11 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
             dependentAttempt: ctx.attempt.id,
           });
 
-          if (!response.ok) {
-            throw new Error(response.error);
-          }
-
-          span.setAttribute("messaging.message.id", response.data.batchId);
+          span.setAttribute("messaging.message.id", response.batchId);
 
           const result = await runtime.waitForBatch({
-            id: response.data.batchId,
-            runs: response.data.runs,
+            id: response.batchId,
+            runs: response.runs,
             ctx,
           });
 

pnpm-lock.yaml

@@ -1739,6 +1739,9 @@ importers:
       zod-error:
         specifier: 1.5.0
         version: 1.5.0
+      zod-validation-error:
+        specifier: ^1.5.0
+        version: 1.5.0(zod@3.22.3)
     devDependencies:
       '@trigger.dev/tsconfig':
         specifier: workspace:*

references/v3-catalog/src/management.ts

@@ -1,4 +1,4 @@
-import { runs, schedules } from "@trigger.dev/sdk/v3";
+import { APIError, runs, schedules } from "@trigger.dev/sdk/v3";
 import { simpleChildTask } from "./trigger/subtasks";
 import dotenv from "dotenv";
 import { firstScheduledTask } from "./trigger/scheduled";
@@ -6,77 +6,73 @@ import { firstScheduledTask } from "./trigger/scheduled";
 dotenv.config();
 
 export async function run() {
-  const run = await simpleChildTask.trigger({ payload: { message: "Hello, World!" } });
-  const canceled = await runs.cancel(run.id);
-  console.log("canceled run", canceled);
+  try {
+    const run = await simpleChildTask.trigger({ payload: { message: "Hello, World!" } });
+    const canceled = await runs.cancel(run.id);
+    console.log("canceled run", canceled);
 
-  const replayed = await runs.replay(run.id);
-  console.log("replayed run", replayed);
+    const replayed = await runs.replay(run.id);
+    console.log("replayed run", replayed);
 
-  const run2 = await simpleChildTask.trigger({
-    payload: { message: "Hello, World!" },
-    options: {
-      idempotencyKey: "mmvlgwcidiklyeygen4",
-    },
-  });
+    const run2 = await simpleChildTask.trigger({
+      payload: { message: "Hello, World!" },
+      options: {
+        idempotencyKey: "mmvlgwcidiklyeygen4",
+      },
+    });
 
-  const run3 = await simpleChildTask.trigger({
-    payload: { message: "Hello, World again!" },
-    options: {
-      idempotencyKey: "mmvlgwcidiklyeygen4",
-    },
-  });
+    const run3 = await simpleChildTask.trigger({
+      payload: { message: "Hello, World again!" },
+      options: {
+        idempotencyKey: "mmvlgwcidiklyeygen4",
+      },
+    });
 
-  console.log("run2", run2);
-  console.log("run3", run3);
+    console.log("run2", run2);
+    console.log("run3", run3);
 
-  const allSchedules = await schedules.list();
+    const allSchedules = await schedules.list();
 
-  // Create a schedule
-  const createdSchedule = await schedules.create({
-    task: firstScheduledTask.id,
-    cron: "0 0 * * *",
-    externalId: "ext_1234444",
-    deduplicationKey: "dedup_1234444",
-  });
+    console.log("all schedules", allSchedules);
 
-  if (createdSchedule.ok) {
-    console.log("created schedule", createdSchedule.data);
+    // Create a schedule
+    const createdSchedule = await schedules.create({
+      task: firstScheduledTask.id,
+      cron: "0 0 * * *",
+      externalId: "ext_1234444",
+      deduplicationKey: "dedup_1234444",
+    });
 
-    const retrievedSchedule = await schedules.retrieve(createdSchedule.data.id);
+    console.log("created schedule", createdSchedule);
 
-    if (retrievedSchedule.ok) {
-      console.log("retrieved schedule", retrievedSchedule.data);
+    const retrievedSchedule = await schedules.retrieve(createdSchedule.id);
 
-      const updatedSchedule = await schedules.update(createdSchedule.data.id, {
-        task: firstScheduledTask.id,
-        cron: "0 0 1 * *",
-        externalId: "ext_1234444",
-      });
+    console.log("retrieved schedule", retrievedSchedule);
 
-      if (updatedSchedule.ok) {
-        console.log("updated schedule", updatedSchedule.data);
+    const updatedSchedule = await schedules.update(createdSchedule.id, {
+      task: firstScheduledTask.id,
+      cron: "0 0 1 * *",
+      externalId: "ext_1234444",
+    });
 
-        const deactivatedSchedule = await schedules.deactivate(createdSchedule.data.id);
+    console.log("updated schedule", updatedSchedule);
 
-        if (deactivatedSchedule.ok) {
-          console.log("deactivated schedule", deactivatedSchedule.data);
-        } else {
-          console.error("failed to deactivate schedule", deactivatedSchedule.error);
-        }
+    const deactivatedSchedule = await schedules.deactivate(createdSchedule.id);
 
-        const activatedSchedule = await schedules.activate(createdSchedule.data.id);
+    console.log("deactivated schedule", deactivatedSchedule);
 
-        if (activatedSchedule.ok) {
-          console.log("activated schedule", activatedSchedule.data);
+    const activatedSchedule = await schedules.activate(createdSchedule.id);
 
-          const deletedSchedule = await schedules.del(createdSchedule.data.id);
+    console.log("activated schedule", activatedSchedule);
 
-          if (deletedSchedule.ok) {
-            console.log("deleted schedule", deletedSchedule.data);
-          }
-        }
-      }
+    const deletedSchedule = await schedules.del(createdSchedule.id);
+
+    console.log("deleted schedule", deletedSchedule);
+  } catch (error) {
+    if (error instanceof APIError) {
+      console.error("APIError", error);
+    } else {
+      console.error("Unknown error", error);
     }
   }
 }