Better handle interactive transaction errors and adding increased timeouts

apps/webapp/app/db.server.ts

@@ -1,6 +1,7 @@
 import { PrismaClient, Prisma } from "@trigger.dev/database";
 import invariant from "tiny-invariant";
 import { z } from "zod";
+import { logger } from "./services/logger.server";
 
 export type PrismaTransactionClient = Omit<
   PrismaClient,
@@ -15,15 +16,54 @@ function isTransactionClient(
   return !("$transaction" in prisma);
 }
 
-export function $transaction<R>(
+function isPrismaKnownError(
+  error: unknown
+): error is Prisma.PrismaClientKnownRequestError {
+  return (
+    typeof error === "object" &&
+    error !== null &&
+    "code" in error &&
+    typeof error.code === "string"
+  );
+}
+
+export type PrismaTransactionOptions = {
+  /** The maximum amount of time (in ms) Prisma Client will wait to acquire a transaction from the database. The default value is 2000ms. */
+  maxWait?: number;
+
+  /** The maximum amount of time (in ms) the interactive transaction can run before being canceled and rolled back. The default value is 5000ms. */
+  timeout?: number;
+
+  /**  Sets the transaction isolation level. By default this is set to the value currently configured in your database. */
+  isolationLevel?: Prisma.TransactionIsolationLevel;
+};
+
+export async function $transaction<R>(
   prisma: PrismaClientOrTransaction,
-  fn: (prisma: PrismaTransactionClient) => Promise<R>
-): Promise<R> {
+  fn: (prisma: PrismaTransactionClient) => Promise<R>,
+  options?: PrismaTransactionOptions
+): Promise<R | undefined> {
   if (isTransactionClient(prisma)) {
     return fn(prisma);
   }
 
-  return (prisma as PrismaClient).$transaction(fn);
+  try {
+    return await (prisma as PrismaClient).$transaction(fn, options);
+  } catch (error) {
+    if (isPrismaKnownError(error)) {
+      logger.debug("prisma.$transaction error", {
+        code: error.code,
+        meta: error.meta,
+        stack: error.stack,
+        message: error.message,
+        name: error.name,
+      });
+
+      return;
+    }
+
+    throw error;
+  }
 }
 
 export { Prisma };

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx

@@ -43,7 +43,10 @@ import { useOrganization } from "~/hooks/useOrganizations";
 import { usePathName } from "~/hooks/usePathName";
 import { useProject } from "~/hooks/useProject";
 import { JobRunStatus } from "~/models/job.server";
-import { redirectWithSuccessMessage } from "~/models/message.server";
+import {
+  redirectBackWithErrorMessage,
+  redirectWithSuccessMessage,
+} from "~/models/message.server";
 import { RunPresenter } from "~/presenters/RunPresenter.server";
 import { ContinueRunService } from "~/services/runs/continueRun.server";
 import { ReRunService } from "~/services/runs/reRun.server";
@@ -114,6 +117,10 @@ export const action: ActionFunction = async ({ request, params }) => {
       const rerunService = new ReRunService();
       const run = await rerunService.call({ runId: runParam });
 
+      if (!run) {
+        return redirectBackWithErrorMessage(request, "Unable to retry run");
+      }
+
       return redirectWithSuccessMessage(
         runDashboardPath(
           { slug: organizationSlug },

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx

@@ -12,10 +12,7 @@ import { HowToRunATest } from "~/components/helpContent/HelpContentText";
 import { Button, ButtonContent } from "~/components/primitives/Buttons";
 import { Callout } from "~/components/primitives/Callout";
 import { FormError } from "~/components/primitives/FormError";
-import { HelpTrigger } from "~/components/primitives/Help";
-import { HelpContent } from "~/components/primitives/Help";
-import { Help } from "~/components/primitives/Help";
-import { Paragraph } from "~/components/primitives/Paragraph";
+import { Help, HelpContent, HelpTrigger } from "~/components/primitives/Help";
 import { Popover, PopoverContent } from "~/components/primitives/Popover";
 import {
   Select,
@@ -25,18 +22,16 @@ import {
   SelectTrigger,
   SelectValue,
 } from "~/components/primitives/Select";
-import { redirectWithSuccessMessage } from "~/models/message.server";
+import {
+  redirectBackWithErrorMessage,
+  redirectWithSuccessMessage,
+} from "~/models/message.server";
 import { TestJobPresenter } from "~/presenters/TestJobPresenter.server";
 import { TestJobService } from "~/services/jobs/testJob.server";
 import { requireUserId } from "~/services/session.server";
 import { cn } from "~/utils/cn";
-import { formDataAsObject } from "~/utils/formData";
 import { Handle } from "~/utils/handle";
-import {
-  JobParamsSchema,
-  jobTestPath,
-  runDashboardPath,
-} from "~/utils/pathBuilder";
+import { JobParamsSchema, runDashboardPath } from "~/utils/pathBuilder";
 
 export const loader = async ({ request, params }: LoaderArgs) => {
   const userId = await requireUserId(request);
@@ -100,6 +95,13 @@ export const action: ActionFunction = async ({ request, params }) => {
     versionId: submission.value.versionId,
   });
 
+  if (!run) {
+    return redirectBackWithErrorMessage(
+      request,
+      "Unable to start a test run: Something went wrong"
+    );
+  }
+
   return redirectWithSuccessMessage(
     runDashboardPath(
       { slug: organizationSlug },

apps/webapp/app/routes/api.v1.$endpointSlug.schedules.$id.registrations.ts

@@ -58,6 +58,10 @@ export async function action({ request, params }: ActionArgs) {
       id: parsedParams.data.id,
     });
 
+    if (!registration) {
+      return json({ error: "Something went wrong" }, { status: 500 });
+    }
+
     return json(
       RegisterScheduleResponseBodySchema.parse({
         id: registration.key,

apps/webapp/app/routes/api.v1.$endpointSlug.triggers.$id.registrations.$key.ts

@@ -57,6 +57,10 @@ export async function action({ request, params }: ActionArgs) {
       key: parsedParams.data.key,
     });
 
+    if (!registration) {
+      return json({ error: "Could not register trigger" }, { status: 500 });
+    }
+
     return json(registration);
   } catch (error) {
     if (error instanceof Error) {

apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts

@@ -63,13 +63,19 @@ export async function action({ request, params }: ActionArgs) {
   const service = new IndexEndpointService();
 
   try {
-    const { data, ...index } = await service.call(
+    const indexing = await service.call(
       endpoint.id,
       "API",
       parsedBody.data.reason,
       parsedBody.data.data
     );
 
+    if (!indexing) {
+      return json({ error: "Something went wrong" }, { status: 500 });
+    }
+
+    const { data, ...index } = indexing;
+
     return json(index);
   } catch (error) {
     if (error instanceof Error) {

apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts

@@ -149,6 +149,10 @@ export async function action({ request, params }: ActionArgs) {
       task,
     });
 
+    if (!task) {
+      return json({ error: "Something went wrong" }, { status: 500 });
+    }
+
     return json(task);
   } catch (error) {
     if (error instanceof Error) {
@@ -170,7 +174,7 @@ export class RunTaskService {
     runId: string,
     idempotencyKey: string,
     taskBody: RunTaskBodyOutput
-  ): Promise<ServerTask> {
+  ): Promise<ServerTask | undefined> {
     const task = await $transaction(this.#prismaClient, async (tx) => {
       const existingTask = await tx.task.findUnique({
         where: {
@@ -260,6 +264,6 @@ export class RunTaskService {
       return task;
     });
 
-    return taskWithAttemptsToServerTask(task);
+    return task ? taskWithAttemptsToServerTask(task) : undefined;
   }
 }

apps/webapp/app/services/endpoints/indexEndpoint.server.ts

@@ -51,90 +51,94 @@ export class IndexEndpointService {
       dynamicSchedules: 0,
     };
 
-    return await $transaction(this.#prismaClient, async (tx) => {
-      for (const job of jobs) {
-        if (!job.enabled) {
-          continue;
+    return await $transaction(
+      this.#prismaClient,
+      async (tx) => {
+        for (const job of jobs) {
+          if (!job.enabled) {
+            continue;
+          }
+
+          indexStats.jobs++;
+
+          await workerQueue.enqueue(
+            "registerJob",
+            {
+              job,
+              endpointId: endpoint.id,
+            },
+            {
+              queueName,
+              tx,
+            }
+          );
         }
 
-        indexStats.jobs++;
+        for (const source of sources) {
+          indexStats.sources++;
 
-        await workerQueue.enqueue(
-          "registerJob",
-          {
-            job,
-            endpointId: endpoint.id,
-          },
-          {
-            queueName,
-            tx,
-          }
-        );
-      }
+          await workerQueue.enqueue(
+            "registerSource",
+            {
+              source,
+              endpointId: endpoint.id,
+            },
+            {
+              queueName,
+              tx,
+            }
+          );
+        }
 
-      for (const source of sources) {
-        indexStats.sources++;
+        for (const dynamicTrigger of dynamicTriggers) {
+          indexStats.dynamicTriggers++;
 
-        await workerQueue.enqueue(
-          "registerSource",
-          {
-            source,
-            endpointId: endpoint.id,
-          },
-          {
-            queueName,
-            tx,
-          }
-        );
-      }
+          await workerQueue.enqueue(
+            "registerDynamicTrigger",
+            {
+              dynamicTrigger,
+              endpointId: endpoint.id,
+            },
+            {
+              queueName,
+              tx,
+            }
+          );
+        }
 
-      for (const dynamicTrigger of dynamicTriggers) {
-        indexStats.dynamicTriggers++;
+        for (const dynamicSchedule of dynamicSchedules) {
+          indexStats.dynamicSchedules++;
 
-        await workerQueue.enqueue(
-          "registerDynamicTrigger",
-          {
-            dynamicTrigger,
-            endpointId: endpoint.id,
-          },
-          {
-            queueName,
-            tx,
-          }
-        );
-      }
+          await workerQueue.enqueue(
+            "registerDynamicSchedule",
+            {
+              dynamicSchedule,
+              endpointId: endpoint.id,
+            },
+            {
+              queueName,
+              tx,
+            }
+          );
+        }
 
-      for (const dynamicSchedule of dynamicSchedules) {
-        indexStats.dynamicSchedules++;
-
-        await workerQueue.enqueue(
-          "registerDynamicSchedule",
-          {
-            dynamicSchedule,
-            endpointId: endpoint.id,
-          },
-          {
-            queueName,
-            tx,
-          }
-        );
-      }
-
-      return await tx.endpointIndex.create({
-        data: {
-          endpointId: endpoint.id,
-          stats: indexStats,
+        return await tx.endpointIndex.create({
           data: {
-            jobs,
-            sources,
-            dynamicTriggers,
-            dynamicSchedules,
+            endpointId: endpoint.id,
+            stats: indexStats,
+            data: {
+              jobs,
+              sources,
+              dynamicTriggers,
+              dynamicSchedules,
+            },
+            source,
+            sourceData,
+            reason,
           },
-          source,
-          sourceData,
-          reason,
-        },
-      });
-    });
+        });
+      },
+      { timeout: 15000 }
+    );
   }
 }

apps/webapp/app/services/events/deliverEvent.server.ts

@@ -13,76 +13,80 @@ export class DeliverEventService {
   }
 
   public async call(id: string) {
-    await $transaction(this.#prismaClient, async (tx) => {
-      const eventRecord = await tx.eventRecord.findUniqueOrThrow({
-        where: {
-          id,
-        },
-        include: {
-          environment: {
-            include: {
-              organization: true,
-              project: true,
+    await $transaction(
+      this.#prismaClient,
+      async (tx) => {
+        const eventRecord = await tx.eventRecord.findUniqueOrThrow({
+          where: {
+            id,
+          },
+          include: {
+            environment: {
+              include: {
+                organization: true,
+                project: true,
+              },
             },
           },
-        },
-      });
+        });
 
-      const possibleEventDispatchers = await tx.eventDispatcher.findMany({
-        where: {
-          environmentId: eventRecord.environmentId,
-          event: eventRecord.name,
-          source: eventRecord.source,
-          enabled: true,
-          manual: false,
-        },
-      });
+        const possibleEventDispatchers = await tx.eventDispatcher.findMany({
+          where: {
+            environmentId: eventRecord.environmentId,
+            event: eventRecord.name,
+            source: eventRecord.source,
+            enabled: true,
+            manual: false,
+          },
+        });
 
-      logger.debug("Found possible event dispatchers", {
-        possibleEventDispatchers,
-        eventRecord: eventRecord.id,
-      });
-
-      const matchingEventDispatchers = possibleEventDispatchers.filter(
-        (eventDispatcher) =>
-          this.#evaluateEventRule(eventDispatcher, eventRecord)
-      );
-
-      if (matchingEventDispatchers.length === 0) {
-        logger.debug("No matching event dispatchers", {
+        logger.debug("Found possible event dispatchers", {
+          possibleEventDispatchers,
           eventRecord: eventRecord.id,
         });
 
-        return;
-      }
+        const matchingEventDispatchers = possibleEventDispatchers.filter(
+          (eventDispatcher) =>
+            this.#evaluateEventRule(eventDispatcher, eventRecord)
+        );
 
-      logger.debug("Found matching event dispatchers", {
-        matchingEventDispatchers,
-        eventRecord: eventRecord.id,
-      });
+        if (matchingEventDispatchers.length === 0) {
+          logger.debug("No matching event dispatchers", {
+            eventRecord: eventRecord.id,
+          });
 
-      await Promise.all(
-        matchingEventDispatchers.map((eventDispatcher) =>
-          workerQueue.enqueue(
-            "events.invokeDispatcher",
-            {
-              id: eventDispatcher.id,
-              eventRecordId: eventRecord.id,
-            },
-            { tx }
+          return;
+        }
+
+        logger.debug("Found matching event dispatchers", {
+          matchingEventDispatchers,
+          eventRecord: eventRecord.id,
+        });
+
+        await Promise.all(
+          matchingEventDispatchers.map((eventDispatcher) =>
+            workerQueue.enqueue(
+              "events.invokeDispatcher",
+              {
+                id: eventDispatcher.id,
+                eventRecordId: eventRecord.id,
+              },
+              { tx }
+            )
           )
-        )
-      );
+        );
 
-      await tx.eventRecord.update({
-        where: {
-          id: eventRecord.id,
-        },
-        data: {
-          deliveredAt: new Date(),
-        },
-      });
-    });
+        await tx.eventRecord.update({
+          where: {
+            id: eventRecord.id,
+          },
+          data: {
+            deliveredAt: new Date(),
+          },
+        });
+      },
+      { timeout: 10000 }
+    );
   }
 
   #evaluateEventRule(

apps/webapp/app/services/externalApis/integrationConnectionCreated.server.ts

@@ -1,8 +1,8 @@
-import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { MISSING_CONNECTION_RESOLVED_NOTIFICATION } from "@trigger.dev/internal";
+import { PrismaClientOrTransaction, prisma } from "~/db.server";
 import { IngestSendEvent } from "../events/ingestSendEvent.server";
 import { logger } from "../logger.server";
 import { workerQueue } from "../worker.server";
-import { MISSING_CONNECTION_RESOLVED_NOTIFICATION } from "@trigger.dev/internal";
 
 export class IntegrationConnectionCreatedService {
   #prismaClient: PrismaClientOrTransaction;
@@ -14,9 +14,9 @@ export class IntegrationConnectionCreatedService {
   public async call(id: string) {
     logger.debug("IntegrationConnectionCreatedService.call", { id });
 
-    return await $transaction(this.#prismaClient, async (tx) => {
-      // first, deliver the event through the dispatcher
-      const connection = await tx.integrationConnection.findUniqueOrThrow({
+    // first, deliver the event through the dispatcher
+    const connection =
+      await this.#prismaClient.integrationConnection.findUniqueOrThrow({
         where: {
           id,
         },
@@ -26,7 +26,8 @@ export class IntegrationConnectionCreatedService {
         },
       });
 
-      const missingConnection = await tx.missingConnection.findUnique({
+    const missingConnection =
+      await this.#prismaClient.missingConnection.findUnique({
         where: {
           integrationId_connectionType_accountIdentifier: {
             integrationId: connection.integrationId,
@@ -55,68 +56,63 @@ export class IntegrationConnectionCreatedService {
         },
       });
 
-      if (!missingConnection) {
-        return;
-      }
+    if (!missingConnection) {
+      return;
+    }
 
-      if (missingConnection.resolved) {
-        return;
-      }
+    if (missingConnection.resolved) {
+      return;
+    }
 
-      const firstRun = missingConnection.runs[0];
+    const firstRun = missingConnection.runs[0];
 
-      if (!firstRun) {
-        return;
-      }
+    if (!firstRun) {
+      return;
+    }
 
-      const eventId = `${missingConnection.id}-resolved`;
+    const eventId = `${missingConnection.id}-resolved`;
 
-      const eventService = new IngestSendEvent(tx);
+    const eventService = new IngestSendEvent();
 
-      await eventService.call(firstRun.environment, {
-        id: eventId,
-        name: MISSING_CONNECTION_RESOLVED_NOTIFICATION,
-        payload: {
-          id: missingConnection.id,
-          type: missingConnection.connectionType,
-          client: {
-            id: missingConnection.integration.slug,
-            title: missingConnection.integration.title,
-            scopes: missingConnection.integration.scopes,
-            createdAt: missingConnection.integration.createdAt,
-            updatedAt: missingConnection.integration.updatedAt,
-          },
-          expiresAt: connection.expiresAt ?? undefined,
-          account: missingConnection.externalAccount
-            ? {
-                id: missingConnection.externalAccount.identifier,
-                metadata: missingConnection.externalAccount.metadata,
-              }
-            : undefined,
+    await eventService.call(firstRun.environment, {
+      id: eventId,
+      name: MISSING_CONNECTION_RESOLVED_NOTIFICATION,
+      payload: {
+        id: missingConnection.id,
+        type: missingConnection.connectionType,
+        client: {
+          id: missingConnection.integration.slug,
+          title: missingConnection.integration.title,
+          scopes: missingConnection.integration.scopes,
+          createdAt: missingConnection.integration.createdAt,
+          updatedAt: missingConnection.integration.updatedAt,
         },
-        context: {},
-      });
-
-      await tx.missingConnection.delete({
-        where: {
-          id: missingConnection.id,
-        },
-      });
-
-      for (const run of missingConnection.runs) {
-        logger.debug("[IntegrationConnectionCreatedService] restarting run", {
-          run,
-        });
-
-        // We need to start the run again
-        await workerQueue.enqueue(
-          "startRun",
-          {
-            id: run.id,
-          },
-          { tx }
-        );
-      }
+        expiresAt: connection.expiresAt ?? undefined,
+        account: missingConnection.externalAccount
+          ? {
+              id: missingConnection.externalAccount.identifier,
+              metadata: missingConnection.externalAccount.metadata,
+            }
+          : undefined,
+      },
+      context: {},
     });
+
+    await this.#prismaClient.missingConnection.delete({
+      where: {
+        id: missingConnection.id,
+      },
+    });
+
+    for (const run of missingConnection.runs) {
+      logger.debug("[IntegrationConnectionCreatedService] restarting run", {
+        run,
+      });
+
+      // We need to start the run again
+      await workerQueue.enqueue("startRun", {
+        id: run.id,
+      });
+    }
   }
 }

apps/webapp/app/services/runs/continueRun.server.ts

@@ -12,88 +12,92 @@ export class ContinueRunService {
   }
 
   public async call({ runId }: { runId: string }) {
-    return await $transaction(this.#prismaClient, async (tx) => {
-      const run = await tx.jobRun.findUniqueOrThrow({
-        where: { id: runId },
-        include: {
-          queue: true,
-        },
-      });
-
-      if (!RESUMABLE_STATUSES.includes(run.status)) {
-        throw new Error("Run is not resumable");
-      }
-
-      if (run.queue.jobCount >= run.queue.maxJobs) {
-        await tx.jobRun.update({
+    return await $transaction(
+      this.#prismaClient,
+      async (tx) => {
+        const run = await tx.jobRun.findUniqueOrThrow({
           where: { id: runId },
-          data: {
-            status: "QUEUED",
-            queuedAt: new Date(),
-            startedAt: null,
-            completedAt: null,
-            output: Prisma.DbNull,
-            timedOutAt: null,
-            timedOutReason: null,
+          include: {
+            queue: true,
           },
         });
-      } else {
-        await tx.jobRun.update({
-          where: { id: runId },
-          data: {
-            status: "STARTED",
-            queuedAt: null,
-            startedAt: new Date(),
-            completedAt: null,
-            output: Prisma.DbNull,
-            timedOutAt: null,
-            timedOutReason: null,
-            queue: {
-              update: {
-                jobCount: {
-                  increment: 1,
+
+        if (!RESUMABLE_STATUSES.includes(run.status)) {
+          throw new Error("Run is not resumable");
+        }
+
+        if (run.queue.jobCount >= run.queue.maxJobs) {
+          await tx.jobRun.update({
+            where: { id: runId },
+            data: {
+              status: "QUEUED",
+              queuedAt: new Date(),
+              startedAt: null,
+              completedAt: null,
+              output: Prisma.DbNull,
+              timedOutAt: null,
+              timedOutReason: null,
+            },
+          });
+        } else {
+          await tx.jobRun.update({
+            where: { id: runId },
+            data: {
+              status: "STARTED",
+              queuedAt: null,
+              startedAt: new Date(),
+              completedAt: null,
+              output: Prisma.DbNull,
+              timedOutAt: null,
+              timedOutReason: null,
+              queue: {
+                update: {
+                  jobCount: {
+                    increment: 1,
+                  },
                 },
               },
             },
-          },
-        });
+          });
 
-        const execution = await tx.jobRunExecution.create({
-          data: {
-            run: {
-              connect: {
-                id: runId,
+          const execution = await tx.jobRunExecution.create({
+            data: {
+              run: {
+                connect: {
+                  id: runId,
+                },
               },
+              status: "PENDING",
+              reason: "EXECUTE_JOB",
+              retryLimit: EXECUTE_JOB_RETRY_LIMIT,
             },
-            status: "PENDING",
-            reason: "EXECUTE_JOB",
-            retryLimit: EXECUTE_JOB_RETRY_LIMIT,
-          },
-        });
+          });
 
-        const job = await workerQueue.enqueue(
-          "performRunExecution",
-          {
-            id: execution.id,
-          },
-          { tx }
-        );
+          const job = await workerQueue.enqueue(
+            "performRunExecution",
+            {
+              id: execution.id,
+            },
+            { tx }
+          );
 
-        await tx.jobRunExecution.update({
-          where: { id: execution.id },
-          data: {
-            graphileJobId: job.id,
-          },
-        });
+          await tx.jobRunExecution.update({
+            where: { id: execution.id },
+            data: {
+              graphileJobId: job.id,
+            },
+          });
 
-        await workerQueue.enqueue(
-          "startQueuedRuns",
-          {
-            id: run.queueId,
-          },
-          { tx }
-        );
-      }
-    });
+          await workerQueue.enqueue(
+            "startQueuedRuns",
+            {
+              id: run.queueId,
+            },
+            { tx }
+          );
+        }
+      },
+      { timeout: 10000 }
+    );
   }
 }

apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts

@@ -13,72 +13,82 @@ export class DeliverScheduledEventService {
   }
 
   public async call(id: string, payload: ScheduledPayload) {
-    return await $transaction(this.#prismaClient, async (tx) => {
-      // first, deliver the event through the dispatcher
-      const scheduleSource = await tx.scheduleSource.findUniqueOrThrow({
-        where: {
-          id,
-        },
-        include: {
-          dispatcher: true,
-          environment: {
-            include: {
-              organization: true,
-              project: true,
-            },
+    return await $transaction(
+      this.#prismaClient,
+      async (tx) => {
+        // first, deliver the event through the dispatcher
+        const scheduleSource = await tx.scheduleSource.findUniqueOrThrow({
+          where: {
+            id,
           },
-          externalAccount: true,
-        },
-      });
+          include: {
+            dispatcher: true,
+            environment: {
+              include: {
+                organization: true,
+                project: true,
+              },
+            },
+            externalAccount: true,
+          },
+        });
 
-      if (!scheduleSource.active) {
-        return;
-      }
-
-      const eventId = `${scheduleSource.id}:${payload.ts.getTime()}`;
-
-      // false prevents send event from delivering the event to dispatchers
-      // since we are going to control that ourselves
-      const eventService = new IngestSendEvent(tx, false);
-
-      const eventRecord = await eventService.call(
-        scheduleSource.environment,
-        {
-          id: eventId,
-          name: SCHEDULED_EVENT,
-          payload,
-        },
-        { accountId: scheduleSource.externalAccount?.identifier },
-        {
-          id: scheduleSource.key,
-          metadata: scheduleSource.metadata,
+        if (!scheduleSource.active) {
+          return;
         }
-      );
 
-      const invokeDispatcherService = new InvokeDispatcherService(tx);
+        const eventId = `${scheduleSource.id}:${payload.ts.getTime()}`;
 
-      await invokeDispatcherService.call(
-        scheduleSource.dispatcher.id,
-        eventRecord.id
-      );
+        // false prevents send event from delivering the event to dispatchers
+        // since we are going to control that ourselves
+        const eventService = new IngestSendEvent(tx, false);
 
-      logger.debug("updating lastEventTimestamp", {
-        id,
-        lastEventTimestamp: payload.ts,
-      });
+        const eventRecord = await eventService.call(
+          scheduleSource.environment,
+          {
+            id: eventId,
+            name: SCHEDULED_EVENT,
+            payload,
+          },
+          { accountId: scheduleSource.externalAccount?.identifier },
+          {
+            id: scheduleSource.key,
+            metadata: scheduleSource.metadata,
+          }
+        );
 
-      await tx.scheduleSource.update({
-        where: {
+        if (!eventRecord) {
+          throw new Error(
+            `Unable to create an event record when delivering scheduled event for scheduleSource.id = ${scheduleSource.id}`
+          );
+        }
+
+        const invokeDispatcherService = new InvokeDispatcherService(tx);
+
+        await invokeDispatcherService.call(
+          scheduleSource.dispatcher.id,
+          eventRecord.id
+        );
+
+        logger.debug("updating lastEventTimestamp", {
           id,
-        },
-        data: {
           lastEventTimestamp: payload.ts,
-        },
-      });
+        });
 
-      const nextScheduledEventService = new NextScheduledEventService(tx);
+        await tx.scheduleSource.update({
+          where: {
+            id,
+          },
+          data: {
+            lastEventTimestamp: payload.ts,
+          },
+        });
 
-      await nextScheduledEventService.call(scheduleSource.id);
-    });
+        const nextScheduledEventService = new NextScheduledEventService(tx);
+
+        await nextScheduledEventService.call(scheduleSource.id);
+      },
+      { timeout: 10000 }
+    );
   }
 }

apps/webapp/app/services/sources/registerSource.server.ts

@@ -64,7 +64,7 @@ export class RegisterSourceService {
       .filter(Boolean)
       .join(":");
 
-    const { id, orphanedEvents } = await $transaction(
+    const source = await $transaction(
       this.#prismaClient,
       async (tx) => {
         const integration = await this.#findOrCreateIntegration(
@@ -215,9 +215,16 @@ export class RegisterSourceService {
           id: triggerSource.id,
           orphanedEvents: Array.from(orphanedEvents),
         };
-      }
+      },
+      { timeout: 15000 }
     );
 
+    if (!source) {
+      return;
+    }
+
+    const { id, orphanedEvents } = source;
+
     // We need to activate the source if:
     // 1. It's not active
     // 2. There are orphaned events

apps/webapp/app/services/triggers/initializeTrigger.server.ts

@@ -74,6 +74,10 @@ export class InitializeTriggerService {
       registrationMetadata: payload.metadata,
     });
 
+    if (!registration) {
+      return;
+    }
+
     await this.#sendEvent.call(
       environment,
       {

apps/webapp/app/services/triggers/registerDynamicSchedule.server.ts

@@ -2,12 +2,7 @@ import {
   RegisterDynamicSchedulePayload,
   SCHEDULED_EVENT,
 } from "@trigger.dev/internal";
-import {
-  $transaction,
-  PrismaClient,
-  PrismaClientOrTransaction,
-} from "~/db.server";
-import { prisma } from "~/db.server";
+import { PrismaClientOrTransaction, prisma } from "~/db.server";
 
 export class RegisterDynamicScheduleService {
   #prismaClient: PrismaClientOrTransaction;
@@ -20,90 +15,88 @@ export class RegisterDynamicScheduleService {
     endpointId: string,
     metadata: RegisterDynamicSchedulePayload
   ) {
-    await $transaction(this.#prismaClient, async (tx) => {
-      const dynamicTrigger = await tx.dynamicTrigger.upsert({
-        where: {
-          endpointId_slug_type: {
-            endpointId: endpointId,
-            slug: metadata.id,
-            type: "SCHEDULE",
-          },
-        },
-        create: {
+    const dynamicTrigger = await this.#prismaClient.dynamicTrigger.upsert({
+      where: {
+        endpointId_slug_type: {
+          endpointId: endpointId,
           slug: metadata.id,
           type: "SCHEDULE",
-          endpoint: {
-            connect: {
-              id: endpointId,
-            },
+        },
+      },
+      create: {
+        slug: metadata.id,
+        type: "SCHEDULE",
+        endpoint: {
+          connect: {
+            id: endpointId,
           },
         },
-        update: {},
-        include: {
-          jobs: true,
-          endpoint: true,
-        },
-      });
+      },
+      update: {},
+      include: {
+        jobs: true,
+        endpoint: true,
+      },
+    });
 
-      // Now we need to connect the jobs
-      const jobs = await tx.job.findMany({
-        where: {
-          slug: {
-            in: metadata.jobs.map((job) => job.id),
-          },
-          versions: {
-            some: {
-              endpointId,
-            },
+    // Now we need to connect the jobs
+    const jobs = await this.#prismaClient.job.findMany({
+      where: {
+        slug: {
+          in: metadata.jobs.map((job) => job.id),
+        },
+        versions: {
+          some: {
+            endpointId,
           },
         },
-      });
+      },
+    });
 
-      // Update all the jobs that are associated with this dynamic trigger
-      await tx.dynamicTrigger.update({
-        where: {
+    // Update all the jobs that are associated with this dynamic trigger
+    await this.#prismaClient.dynamicTrigger.update({
+      where: {
+        id: dynamicTrigger.id,
+      },
+      data: {
+        jobs: {
+          connect: jobs.map((job) => ({
+            id: job.id,
+          })),
+          disconnect: dynamicTrigger.jobs.filter(
+            (job) => !jobs.find((j) => j.id === job.id)
+          ),
+        },
+      },
+    });
+
+    await this.#prismaClient.eventDispatcher.upsert({
+      where: {
+        dispatchableId_environmentId: {
+          dispatchableId: dynamicTrigger.id,
+          environmentId: dynamicTrigger.endpoint.environmentId,
+        },
+      },
+      create: {
+        event: SCHEDULED_EVENT,
+        source: "trigger.dev",
+        payloadFilter: {},
+        contextFilter: {},
+        environmentId: dynamicTrigger.endpoint.environmentId,
+        enabled: true,
+        dispatchable: {
+          type: "DYNAMIC_TRIGGER",
           id: dynamicTrigger.id,
         },
-        data: {
-          jobs: {
-            connect: jobs.map((job) => ({
-              id: job.id,
-            })),
-            disconnect: dynamicTrigger.jobs.filter(
-              (job) => !jobs.find((j) => j.id === job.id)
-            ),
-          },
+        dispatchableId: dynamicTrigger.id,
+        manual: true,
+      },
+      update: {
+        dispatchable: {
+          type: "DYNAMIC_TRIGGER",
+          id: dynamicTrigger.id,
         },
-      });
-
-      const eventDispatcher = await tx.eventDispatcher.upsert({
-        where: {
-          dispatchableId_environmentId: {
-            dispatchableId: dynamicTrigger.id,
-            environmentId: dynamicTrigger.endpoint.environmentId,
-          },
-        },
-        create: {
-          event: SCHEDULED_EVENT,
-          source: "trigger.dev",
-          payloadFilter: {},
-          contextFilter: {},
-          environmentId: dynamicTrigger.endpoint.environmentId,
-          enabled: true,
-          dispatchable: {
-            type: "DYNAMIC_TRIGGER",
-            id: dynamicTrigger.id,
-          },
-          dispatchableId: dynamicTrigger.id,
-          manual: true,
-        },
-        update: {
-          dispatchable: {
-            type: "DYNAMIC_TRIGGER",
-            id: dynamicTrigger.id,
-          },
-        },
-      });
+      },
     });
   }
 }

apps/webapp/app/services/triggers/registerTriggerSource.server.ts

@@ -32,7 +32,7 @@ export class RegisterTriggerSourceService {
     key: string;
     accountId?: string;
     registrationMetadata?: any;
-  }): Promise<RegisterSourceEvent> {
+  }): Promise<RegisterSourceEvent | undefined> {
     const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({
       where: {
         environmentId_slug: {
@@ -53,97 +53,105 @@ export class RegisterTriggerSourceService {
         },
       });
 
-    return await $transaction(this.#prismaClient, async (tx) => {
-      const service = new RegisterSourceService(tx);
+    return await $transaction(
+      this.#prismaClient,
+      async (tx) => {
+        const service = new RegisterSourceService(tx);
 
-      const triggerSource = await service.call(
-        endpoint.id,
-        payload.source,
-        dynamicTrigger.id,
-        accountId,
-        { id: key, metadata: registrationMetadata }
-      );
+        const triggerSource = await service.call(
+          endpoint.id,
+          payload.source,
+          dynamicTrigger.id,
+          accountId,
+          { id: key, metadata: registrationMetadata }
+        );
 
-      const eventDispatcher = await tx.eventDispatcher.upsert({
-        where: {
-          dispatchableId_environmentId: {
+        if (!triggerSource) {
+          return;
+        }
+
+        const eventDispatcher = await tx.eventDispatcher.upsert({
+          where: {
+            dispatchableId_environmentId: {
+              dispatchableId: triggerSource.id,
+              environmentId: environment.id,
+            },
+          },
+          create: {
             dispatchableId: triggerSource.id,
             environmentId: environment.id,
+            event: payload.rule.event,
+            source: payload.rule.source,
+            payloadFilter: payload.rule.payload,
+            contextFilter: payload.rule.context,
+            dispatchable: {
+              type: "DYNAMIC_TRIGGER",
+              id: dynamicTrigger.id,
+            },
           },
-        },
-        create: {
-          dispatchableId: triggerSource.id,
-          environmentId: environment.id,
-          event: payload.rule.event,
-          source: payload.rule.source,
-          payloadFilter: payload.rule.payload,
-          contextFilter: payload.rule.context,
-          dispatchable: {
-            type: "DYNAMIC_TRIGGER",
-            id: dynamicTrigger.id,
+          update: {
+            event: payload.rule.event,
+            source: payload.rule.source,
+            payloadFilter: payload.rule.payload,
+            contextFilter: payload.rule.context,
+            dispatchable: {
+              type: "DYNAMIC_TRIGGER",
+              id: dynamicTrigger.id,
+            },
           },
-        },
-        update: {
-          event: payload.rule.event,
-          source: payload.rule.source,
-          payloadFilter: payload.rule.payload,
-          contextFilter: payload.rule.context,
-          dispatchable: {
-            type: "DYNAMIC_TRIGGER",
-            id: dynamicTrigger.id,
-          },
-        },
-      });
+        });
 
-      const registration = await tx.dynamicTriggerRegistration.upsert({
-        where: {
-          key_dynamicTriggerId: {
+        const registration = await tx.dynamicTriggerRegistration.upsert({
+          where: {
+            key_dynamicTriggerId: {
+              key,
+              dynamicTriggerId: dynamicTrigger.id,
+            },
+          },
+          create: {
             key,
             dynamicTriggerId: dynamicTrigger.id,
+            sourceId: triggerSource.id,
+            eventDispatcherId: eventDispatcher.id,
+            metadata: registrationMetadata,
           },
-        },
-        create: {
-          key,
-          dynamicTriggerId: dynamicTrigger.id,
-          sourceId: triggerSource.id,
-          eventDispatcherId: eventDispatcher.id,
-          metadata: registrationMetadata,
-        },
-        update: {
-          metadata: registrationMetadata,
-        },
-      });
-
-      const secretStore = getSecretStore(
-        triggerSource.secretReference.provider,
-        { prismaClient: tx }
-      );
-
-      const { secret } = await secretStore.getSecretOrThrow(
-        z.object({
-          secret: z.string(),
-        }),
-        triggerSource.secretReference.key
-      );
-
-      return {
-        id: registration.id,
-        source: {
-          key: triggerSource.key,
-          active: triggerSource.active,
-          params: triggerSource.params,
-          secret,
-          data: triggerSource.channelData as any,
-          channel: {
-            type: "HTTP",
-            url: `${env.APP_ORIGIN}/api/v1/sources/http/${triggerSource.id}`,
+          update: {
+            metadata: registrationMetadata,
           },
-          clientId: triggerSource.integration.slug,
-        },
-        events: triggerSource.events.map((e) => e.name),
-        missingEvents: [],
-        orphanedEvents: [],
-      };
-    });
+        });
+
+        const secretStore = getSecretStore(
+          triggerSource.secretReference.provider,
+          { prismaClient: tx }
+        );
+
+        const { secret } = await secretStore.getSecretOrThrow(
+          z.object({
+            secret: z.string(),
+          }),
+          triggerSource.secretReference.key
+        );
+
+        return {
+          id: registration.id,
+          source: {
+            key: triggerSource.key,
+            active: triggerSource.active,
+            params: triggerSource.params,
+            secret,
+            data: triggerSource.channelData as any,
+            channel: {
+              type: "HTTP",
+              url: `${env.APP_ORIGIN}/api/v1/sources/http/${triggerSource.id}`,
+            },
+            clientId: triggerSource.integration.slug,
+          },
+          events: triggerSource.events.map((e) => e.name),
+          missingEvents: [],
+          orphanedEvents: [],
+        };
+      },
+      { timeout: 15000 }
+    );
   }
 }