Added ability to timeout triggers so they don’t run too late after the event time

.changeset/good-pillows-drum.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Added triggerTTL option that prevents old events from running a workflow

apps/webapp/app/components/runs/runStatus.tsx

@@ -21,6 +21,8 @@ export function runStatusTitle(status: WorkflowRunStatus): string {
       return "Disconnected";
     case "ERROR":
       return "Error";
+    case "TIMED_OUT":
+      return "Timed out";
   }
 }
 
@@ -36,6 +38,8 @@ export function runStatusLabel(status: WorkflowRunStatus): ReactNode {
       return <span className="text-amber-300">{runStatusTitle(status)}</span>;
     case "ERROR":
       return <span className="text-rose-500">{runStatusTitle(status)}</span>;
+    case "TIMED_OUT":
+      return <span className="text-amber-300">{runStatusTitle(status)}</span>;
   }
 }
 
@@ -91,5 +95,14 @@ export function runStatusIcon(
           )}
         />
       );
+    case "TIMED_OUT":
+      return (
+        <ExclamationTriangleIcon
+          className={classNames(
+            iconSize === "small" ? smallClasses : largeClasses,
+            "relative text-amber-300"
+          )}
+        />
+      );
   }
 }

apps/webapp/app/models/workflowRun.server.ts

@@ -19,6 +19,7 @@ export async function findWorklowRunById(id: string) {
     include: {
       event: true,
       environment: true,
+      workflow: true,
     },
   });
 }

apps/webapp/app/services/messageBroker.server.ts

@@ -43,6 +43,7 @@ import { WaitForConnection } from "./requests/waitForConnection.server";
 import { WorkflowRunDisconnected } from "./runs/runDisconnected.server";
 import { DeliverScheduledEvent } from "./scheduler/deliverScheduledEvent.server";
 import { RegisterSchedulerSource } from "./scheduler/registerSchedulerSource.server";
+import { WorkflowRunTriggerTimeout } from "./runs/runTriggerTimeout.server";
 
 let pulsarClient: PulsarClient;
 let triggerPublisher: ZodPublisher<TriggerCatalog>;
@@ -206,6 +207,13 @@ function createCommandSubscriber() {
 
         return !!success;
       },
+      WORKFLOW_RUN_TRIGGER_TIMEOUT: async (id, data, properties) => {
+        const service = new WorkflowRunTriggerTimeout();
+
+        await service.call(data);
+
+        return true;
+      },
       SEND_INTEGRATION_REQUEST: async (id, data, properties) => {
         const service = new CreateIntegrationRequest();
 
@@ -526,6 +534,10 @@ function createTaskQueue() {
             "x-workflow-id": run.workflowId,
             "x-env": run.environment.slug,
             "x-workflow-run-id": run.id,
+            "x-ttl": run.workflow.triggerTtlInSeconds,
+          },
+          {
+            eventTimestamp: run.event.timestamp.getTime(),
           }
         );
 

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

@@ -0,0 +1,37 @@
+import type { PrismaClient } from "~/db.server";
+import { prisma } from "~/db.server";
+
+export class WorkflowRunTriggerTimeout {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  async call(data: { id: string; ttl: number; elapsedSeconds: number }) {
+    const workflowRun = await this.#prismaClient.workflowRun.findUnique({
+      where: { id: data.id },
+      include: {
+        event: true,
+        environment: true,
+      },
+    });
+
+    if (!workflowRun) {
+      throw new Error("Workflow run not found");
+    }
+
+    if (workflowRun.status !== "PENDING") {
+      return;
+    }
+
+    await this.#prismaClient.workflowRun.update({
+      where: { id: data.id },
+      data: {
+        status: "TIMED_OUT",
+        timedOutAt: new Date(),
+        timedOutReason: `Trigger timed out after ${data.elapsedSeconds}s because it exceeded the TTL of ${data.ttl}s`,
+      },
+    });
+  }
+}

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

@@ -67,6 +67,7 @@ export class DeliverScheduledEvent {
         context: {},
         organizationId: schedulerSource.organizationId,
         environmentId: schedulerSource.environmentId,
+        timestamp: payload.scheduledTime,
       },
     });
 

apps/webapp/app/services/workflows/registerWorkflow.server.ts

@@ -110,6 +110,7 @@ export class RegisterWorkflow {
         type: payload.trigger.type,
         service: payload.trigger.service,
         eventNames: payload.trigger.name,
+        triggerTtlInSeconds: payload.triggerTTL,
       },
       create: {
         organizationId: organization.id,
@@ -120,6 +121,7 @@ export class RegisterWorkflow {
         status: payload.trigger.service === "trigger" ? "READY" : "CREATED",
         service: payload.trigger.service,
         eventNames: payload.trigger.name,
+        triggerTtlInSeconds: payload.triggerTTL,
       },
       include: {
         externalSource: true,

apps/webapp/prisma/migrations/20230120010921_add_trigger_ttl_in_seconds/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Workflow" ADD COLUMN     "triggerTtlInSeconds" INTEGER NOT NULL DEFAULT 3600;

apps/webapp/prisma/migrations/20230120014606_add_timed_out_status_for_workflow_runs/migration.sql

@@ -0,0 +1,2 @@
+-- AlterEnum
+ALTER TYPE "WorkflowRunStatus" ADD VALUE 'TIMED_OUT';

apps/webapp/prisma/migrations/20230120014641_add_timed_out_columns/migration.sql

@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE "WorkflowRun" ADD COLUMN     "timedOutAt" TIMESTAMP(3),
+ADD COLUMN     "timedOutReason" TEXT;

apps/webapp/prisma/schema.prisma

@@ -142,6 +142,8 @@ model Workflow {
   archivedAt DateTime?
   isArchived Boolean   @default(false)
 
+  triggerTtlInSeconds Int @default(3600)
+
   @@unique([organizationId, slug])
 }
 
@@ -406,6 +408,9 @@ model WorkflowRun {
   startedAt  DateTime?
   finishedAt DateTime?
 
+  timedOutAt     DateTime?
+  timedOutReason String?
+
   isTest   Boolean              @default(false)
   requests IntegrationRequest[]
   delays   DurableDelay[]
@@ -417,6 +422,7 @@ enum WorkflowRunStatus {
   DISCONNECTED
   SUCCESS
   ERROR
+  TIMED_OUT
 }
 
 model WorkflowRunStep {

apps/wss/src/server.ts

@@ -324,6 +324,7 @@ export class TriggerServer {
           name: data.packageName,
           version: data.packageVersion,
         },
+        triggerTTL: data.triggerTTL,
       });
 
       this.#workflowId = response.id;
@@ -340,7 +341,7 @@ export class TriggerServer {
           subscriptionInitialPosition: "Earliest",
         },
         handlers: {
-          TRIGGER_WORKFLOW: async (id, data, properties) => {
+          TRIGGER_WORKFLOW: async (id, data, properties, messageAttributes) => {
             this.#logger.debug("Received trigger", id, data, properties);
             // If the API keys don't match, then we should ignore it
             // This ensures the workflow is triggered for the correct environment
@@ -366,6 +367,33 @@ export class TriggerServer {
               );
             }
 
+            if (properties["x-ttl"] && messageAttributes.eventTimestamp) {
+              const ttl = properties["x-ttl"];
+              const eventTimestamp = messageAttributes.eventTimestamp;
+              const now = Date.now();
+              const elapsedMilliseconds = now - eventTimestamp.getTime();
+              const elapsedSeconds = elapsedMilliseconds / 1000;
+
+              if (elapsedSeconds > ttl) {
+                this.#logger.debug("Message is expired, ignoring", {
+                  messageAttributes,
+                  properties,
+                });
+
+                await this.#commandPublisher.publish(
+                  "WORKFLOW_RUN_TRIGGER_TIMEOUT",
+                  {
+                    id: data.id,
+                    ttl,
+                    elapsedSeconds,
+                  },
+                  { ...properties, "x-timestamp": String(Date.now()) }
+                );
+
+                return true;
+              }
+            }
+
             this.#logger.debug("Triggering workflow", data, properties);
 
             const runController = new WorkflowRunController({

examples/playground/src/index.ts

@@ -427,6 +427,7 @@ new Trigger({
   apiKey: "trigger_dev_zC25mKNn6c0q",
   endpoint: "ws://localhost:8889/ws",
   logLevel: "debug",
+  triggerTTL: 5,
   on: scheduleEvent({ rateOf: { minutes: 4 } }),
   run: async (event, ctx) => {
     await ctx.logger.info("Received the scheduled event", {

examples/smoke-test/src/index.ts

@@ -12,6 +12,7 @@ const trigger = new Trigger({
   apiKey: "trigger_dev_zC25mKNn6c0q",
   endpoint: "ws://localhost:8889/ws",
   logLevel: "log",
+  triggerTTL: 60 * 60 * 24,
   on: customEvent({ name: "user.created", schema: userCreatedEvent }),
   run: async (event, ctx) => {
     await ctx.logger.info("Inside the smoke test workflow, received event", {

packages/internal-bridge/src/schemas/server.ts

@@ -58,6 +58,7 @@ export const ServerRPCSchema = {
       trigger: TriggerMetadataSchema,
       packageVersion: z.string(),
       packageName: z.string(),
+      triggerTTL: z.number().optional(),
     }),
     response: z
       .discriminatedUnion("type", [

packages/internal-platform/src/messages/catalogs/triggers.ts

@@ -1,10 +1,13 @@
+import { z } from "zod";
 import { TriggerWorkflowMessageSchema } from "../schemas/workflows";
 import { WorkflowRunEventPropertiesSchema } from "../sharedSchemas";
 
 const Catalog = {
   TRIGGER_WORKFLOW: {
     data: TriggerWorkflowMessageSchema,
-    properties: WorkflowRunEventPropertiesSchema,
+    properties: WorkflowRunEventPropertiesSchema.extend({
+      "x-ttl": z.coerce.number().optional(),
+    }),
   },
 };
 

packages/internal-platform/src/messages/schemas/workflowRuns.ts

@@ -27,4 +27,12 @@ export const commands = {
     }),
     properties: WorkflowSendRunEventPropertiesSchema,
   },
+  WORKFLOW_RUN_TRIGGER_TIMEOUT: {
+    data: z.object({
+      id: z.string(),
+      ttl: z.number(),
+      elapsedSeconds: z.number(),
+    }),
+    properties: WorkflowSendRunEventPropertiesSchema,
+  },
 };

packages/internal-platform/src/messages/zodPublisher.ts

@@ -16,6 +16,7 @@ export type PublishOptions = {
   partitionKey?: string;
   orderingKey?: string;
   id?: string;
+  eventTimestamp?: number;
 };
 
 type PendingMessages<PublisherSchema extends MessageCatalogSchema> = Array<{
@@ -216,6 +217,7 @@ export class ZodPublisher<PublisherSchema extends MessageCatalogSchema> {
       deliverAt: options?.deliverAt,
       partitionKey: options?.partitionKey,
       orderingKey: options?.orderingKey,
+      eventTimestamp: options?.eventTimestamp,
     });
 
     return response.toString();

packages/internal-platform/src/messages/zodSubscriber.ts

@@ -14,13 +14,21 @@ import {
 import { z, ZodError } from "zod";
 import { ZodPubSubStatus } from "./types";
 
+export type SubscriberMessageAttributes = {
+  messageId: string;
+  eventTimestamp?: Date;
+  publishedTimestamp: Date;
+  redeliveryCount: number;
+};
+
 export type ZodSubscriberHandlers<
   TConsumerSchema extends MessageCatalogSchema
 > = {
   [K in keyof TConsumerSchema]: (
     id: string,
     data: z.infer<TConsumerSchema[K]["data"]>,
-    properties: z.infer<TConsumerSchema[K]["properties"]>
+    properties: z.infer<TConsumerSchema[K]["properties"]>,
+    attributes: SubscriberMessageAttributes
   ) => Promise<boolean>;
 };
 
@@ -109,8 +117,32 @@ export class ZodSubscriber<SubscriberSchema extends MessageCatalogSchema> {
 
     const properties = this.#getRawProperties(msg);
 
+    const messageId = msg.getMessageId();
+    const publishedTimestamp = msg.getPublishTimestamp();
+    const eventTimestamp = msg.getEventTimestamp();
+    const redeliveryCount = msg.getRedeliveryCount();
+
+    this.#logger.debug("#onMessage", {
+      messageId,
+      publishedTimestamp,
+      eventTimestamp,
+      redeliveryCount,
+    });
+
+    const messageAttributes = {
+      eventTimestamp:
+        eventTimestamp === 0 ? undefined : new Date(eventTimestamp),
+      messageId: messageId.toString(),
+      publishedTimestamp: new Date(publishedTimestamp),
+      redeliveryCount,
+    };
+
     try {
-      const wasHandled = await this.#handleMessage(messageData, properties);
+      const wasHandled = await this.#handleMessage(
+        messageData,
+        properties,
+        messageAttributes
+      );
 
       if (wasHandled) {
         await consumer.acknowledge(msg);
@@ -134,7 +166,8 @@ export class ZodSubscriber<SubscriberSchema extends MessageCatalogSchema> {
 
   async #handleMessage<K extends keyof SubscriberSchema>(
     rawMessage: MessageData,
-    rawProperties: Record<string, string> = {}
+    rawProperties: Record<string, string> = {},
+    messageAttributes: SubscriberMessageAttributes
   ): Promise<boolean> {
     const subscriberSchema = this.#schema;
     type TypeKeys = keyof typeof subscriberSchema;
@@ -160,7 +193,12 @@ export class ZodSubscriber<SubscriberSchema extends MessageCatalogSchema> {
 
     const handler = this.#handlers[typeName];
 
-    const returnValue = await handler(rawMessage.id, message, properties);
+    const returnValue = await handler(
+      rawMessage.id,
+      message,
+      properties,
+      messageAttributes
+    );
 
     return returnValue;
   }

packages/internal-platform/src/schemas/workflows.ts

@@ -11,6 +11,7 @@ export const WorkflowMetadataSchema = z.object({
   name: z.string(),
   trigger: TriggerMetadataSchema,
   package: PackageMetadataSchema,
+  triggerTTL: z.number().optional(),
 });
 
 export type WorkflowMetadata = z.infer<typeof WorkflowMetadataSchema>;

packages/trigger-sdk/src/client.ts

@@ -443,6 +443,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
       trigger: this.#trigger.on.metadata,
       packageVersion: pkg.version,
       packageName: pkg.name,
+      triggerTTL: this.#options.triggerTTL,
     });
 
     if (response?.type === "error") {

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

@@ -12,6 +12,13 @@ export type TriggerOptions<TSchema extends z.ZodTypeAny> = {
   apiKey?: string;
   endpoint?: string;
   logLevel?: LogLevel;
+
+  /**
+   * The TTL for the trigger in seconds. If the trigger is not run within this time, it will be aborted. Defaults to 3600 seconds (1 hour).
+   * @type {number}
+   */
+  triggerTTL?: number;
+
   run: (event: z.infer<TSchema>, ctx: TriggerContext) => Promise<any>;
 };