feat: replicate trigger_source, root_trigger_source, and is_warm_start to ClickHouse (#3274)

Adds three new top-level columns to the ClickHouse task_runs_v2 table primarily for analytics: - `trigger_source` / `root_trigger_source` - extracted from the existing TaskRun.annotations JSON during WAL replication - `is_warm_start` - new nullable boolean on TaskRun in Postgres, set in the existing taskRun.update() at attempt start (no additional write). null until the first attempt starts. Run region is already available via the existing `worker_queue` column in ClickHouse.
2026-03-26 10:27:31 +01:00
parent efcafdf388
commit 38559480c9
8 changed files with 72 additions and 5 deletions
@@ -22,6 +22,7 @@ import { Logger, type LogLevel } from "@trigger.dev/core/logger";
 import { tryCatch } from "@trigger.dev/core/utils";
 import { parsePacketAsJson } from "@trigger.dev/core/v3/utils/ioSerialization";
 import { unsafeExtractIdempotencyKeyScope, unsafeExtractIdempotencyKeyUser } from "@trigger.dev/core/v3/serverOnly";
+import { RunAnnotations } from "@trigger.dev/core/v3";
 import { type TaskRun } from "@trigger.dev/database";
 import { nanoid } from "nanoid";
 import EventEmitter from "node:events";
@@ -866,6 +867,8 @@ export class RunsReplicationService {
      ? calculateErrorFingerprint(run.error)
      : '';

+    const annotations = this.#parseAnnotations(run.annotations);
+
    // Return array matching TASK_RUN_COLUMNS order
    return [
      run.runtimeEnvironmentId, // environment_id
@@ -916,9 +919,16 @@ export class RunsReplicationService {
      run.bulkActionGroupIds ?? [], // bulk_action_group_ids
      run.masterQueue ?? "", // worker_queue
      run.maxDurationInSeconds ?? null, // max_duration_in_seconds
+      annotations?.triggerSource ?? "", // trigger_source
+      annotations?.rootTriggerSource ?? "", // root_trigger_source
+      run.isWarmStart ?? null, // is_warm_start
    ];
  }

+  #parseAnnotations(annotations: unknown) {
+    return RunAnnotations.safeParse(annotations).data;
+  }
+
  async #preparePayloadInsert(run: TaskRun, _version: bigint): Promise<PayloadInsertArray> {
    const payload = await this.#prepareJson(run.payload, run.payloadType);

@@ -86,6 +86,12 @@ describe("RunsReplicationService (part 1/2)", () => {
          organizationId: organization.id,
          environmentType: "DEVELOPMENT",
          engine: "V2",
+          annotations: {
+            triggerSource: "api",
+            triggerAction: "trigger",
+            rootTriggerSource: "dashboard",
+          },
+          isWarmStart: true,
        },
      });

@@ -111,6 +117,9 @@ describe("RunsReplicationService (part 1/2)", () => {
          organization_id: organization.id,
          environment_type: "DEVELOPMENT",
          engine: "V2",
+          trigger_source: "api",
+          root_trigger_source: "dashboard",
+          is_warm_start: 1,
        })
      );

@@ -0,0 +1,19 @@
+-- +goose Up
+ALTER TABLE trigger_dev.task_runs_v2
+  ADD COLUMN trigger_source LowCardinality(String) DEFAULT '';
+
+ALTER TABLE trigger_dev.task_runs_v2
+  ADD COLUMN root_trigger_source LowCardinality(String) DEFAULT '';
+
+ALTER TABLE trigger_dev.task_runs_v2
+  ADD COLUMN is_warm_start Nullable(UInt8) DEFAULT NULL;
+
+-- +goose Down
+ALTER TABLE trigger_dev.task_runs_v2
+  DROP COLUMN trigger_source;
+
+ALTER TABLE trigger_dev.task_runs_v2
+  DROP COLUMN root_trigger_source;
+
+ALTER TABLE trigger_dev.task_runs_v2
+  DROP COLUMN is_warm_start;
@@ -82,6 +82,9 @@ describe("Task Runs V2", () => {
      ["bulk_action_group_id_1234", "bulk_action_group_id_1235"], // bulk_action_group_ids
      "", // worker_queue
      null, // max_duration_in_seconds
+      "", // trigger_source
+      "", // root_trigger_source
+      null, // is_warm_start
    ];

    const [insertError, insertResult] = await insert([taskRunData]);
@@ -210,6 +213,9 @@ describe("Task Runs V2", () => {
      [], // bulk_action_group_ids
      "", // worker_queue
      null, // max_duration_in_seconds
+      "", // trigger_source
+      "", // root_trigger_source
+      null, // is_warm_start
    ];

    const run2: TaskRunInsertArray = [
@@ -261,6 +267,9 @@ describe("Task Runs V2", () => {
      [], // bulk_action_group_ids
      "", // worker_queue
      null, // max_duration_in_seconds
+      "", // trigger_source
+      "", // root_trigger_source
+      null, // is_warm_start
    ];

    const [insertError, insertResult] = await insert([run1, run2]);
@@ -359,6 +368,9 @@ describe("Task Runs V2", () => {
        [], // bulk_action_group_ids
        "", // worker_queue
        null, // max_duration_in_seconds
+        "", // trigger_source
+        "", // root_trigger_source
+        null, // is_warm_start
      ];

      const [insertError, insertResult] = await insert([taskRun]);
@@ -49,6 +49,9 @@ export const TaskRunV2 = z.object({
  bulk_action_group_ids: z.array(z.string()).default([]),
  worker_queue: z.string().default(""),
  max_duration_in_seconds: z.number().int().nullish(),
+  trigger_source: z.string().default(""),
+  root_trigger_source: z.string().default(""),
+  is_warm_start: z.boolean().nullish(),
  _version: z.string(),
  _is_deleted: z.number().int().default(0),
 });
@@ -105,6 +108,9 @@ export const TASK_RUN_COLUMNS = [
  "bulk_action_group_ids",
  "worker_queue",
  "max_duration_in_seconds",
+  "trigger_source",
+  "root_trigger_source",
+  "is_warm_start",
 ] as const;

 export type TaskRunColumnName = (typeof TASK_RUN_COLUMNS)[number];
@@ -168,6 +174,9 @@ export type TaskRunFieldTypes = {
  bulk_action_group_ids: string[];
  worker_queue: string;
  max_duration_in_seconds: number | null;
+  trigger_source: string;
+  root_trigger_source: string;
+  is_warm_start: boolean | null;
 };

 /**
@@ -302,6 +311,9 @@ export type TaskRunInsertArray = [
  bulk_action_group_ids: string[],
  worker_queue: string,
  max_duration_in_seconds: number | null,
+  trigger_source: string,
+  root_trigger_source: string,
+  is_warm_start: boolean | null,
 ];

 /**
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "public"."TaskRun" ADD COLUMN "isWarmStart" BOOLEAN;
@@ -537,13 +537,13 @@ model BackgroundWorkerFile {
 }

 model Prompt {
-  id         String @id @default(cuid())
-  friendlyId String @unique @map("friendly_id")
+  id          String  @id @default(cuid())
+  friendlyId  String  @unique @map("friendly_id")
  slug        String
  description String?
  type        String  @default("text") // "text" | "chat"

-  organization   Organization       @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  organization   Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade)
  organizationId String

  project   Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
@@ -558,7 +558,7 @@ model Prompt {
  defaultModel   String?
  defaultConfig  Json?

-  tags       String[] @default([])
+  tags       String[]  @default([])
  archivedAt DateTime?

  createdAt DateTime @default(now())
@@ -840,6 +840,9 @@ model TaskRun {
  /// Structured annotations: triggerSource, triggerAction, rootTriggerSource, rootScheduleId
  annotations Json?

+  /// Whether the latest attempt was a warm start. Null until first attempt starts.
+  isWarmStart Boolean?
+
  /// Run output
  output     String?
  outputType String  @default("application/json")
@@ -857,7 +860,6 @@ model TaskRun {
  /// Store the stream keys that are being used by the run
  realtimeStreams        String[] @default([])

-
  @@unique([oneTimeUseToken])
  @@unique([runtimeEnvironmentId, taskIdentifier, idempotencyKey])
  // Finding child runs
@@ -402,6 +402,7 @@ export class RunAttemptSystem {
                  status: "EXECUTING",
                  attemptNumber: nextAttemptNumber,
                  executedAt: taskRun.attemptNumber === null ? new Date() : undefined,
+                  isWarmStart: isWarmStart ?? false,
                },
                select: {
                  id: true,