ClickHouse worker_queue on task runs

Author: matt-aitken

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

@@ -59,7 +59,13 @@ export type RunsReplicationServiceOptions = {
   insertMaxDelayMs?: number;
 };
 
-type TaskRunInsert = { _version: bigint; run: TaskRun; event: "insert" | "update" | "delete" };
+type PostgresTaskRun = TaskRun & { masterQueue: string };
+
+type TaskRunInsert = {
+  _version: bigint;
+  run: PostgresTaskRun;
+  event: "insert" | "update" | "delete";
+};
 
 export type RunsReplicationServiceEvents = {
   message: [{ lsn: string; message: PgoutputMessage; service: RunsReplicationService }];
@@ -243,7 +249,7 @@ export class RunsReplicationService {
     }
   }
 
-  async backfill(runs: TaskRun[]) {
+  async backfill(runs: PostgresTaskRun[]) {
     // divide into batches of 50 to get data from Postgres
     const flushId = nanoid();
     // Use current timestamp as LSN (high enough to be above existing data)
@@ -352,7 +358,7 @@ export class RunsReplicationService {
         const replicationLagMs = Date.now() - Number(message.commitTime / 1000n);
         this._currentTransaction.commitEndLsn = message.commitEndLsn;
         this._currentTransaction.replicationLagMs = replicationLagMs;
-        const transaction = this._currentTransaction as Transaction<TaskRun>;
+        const transaction = this._currentTransaction as Transaction<PostgresTaskRun>;
         this._currentTransaction = null;
 
         if (transaction.commitEndLsn) {
@@ -370,7 +376,7 @@ export class RunsReplicationService {
     }
   }
 
-  #handleTransaction(transaction: Transaction<TaskRun>) {
+  #handleTransaction(transaction: Transaction<PostgresTaskRun>) {
     if (this._isShutDownComplete) return;
 
     if (this._isShuttingDown) {
@@ -764,7 +770,7 @@ export class RunsReplicationService {
   }
 
   async #prepareTaskRunInsert(
-    run: TaskRun,
+    run: PostgresTaskRun,
     organizationId: string,
     environmentType: string,
     event: "insert" | "update" | "delete",
@@ -814,6 +820,7 @@ export class RunsReplicationService {
       output,
       concurrency_key: run.concurrencyKey ?? "",
       bulk_action_group_ids: run.bulkActionGroupIds ?? [],
+      worker_queue: run.masterQueue,
       _version: _version.toString(),
       _is_deleted: event === "delete" ? 1 : 0,
     };

internal-packages/clickhouse/schema/006_add_task_runs_v2_workerqueue.sql

@@ -0,0 +1,10 @@
+-- +goose Up
+/*
+Add worker_queue column.
+ */
+ALTER TABLE trigger_dev.task_runs_v2
+ADD COLUMN worker_queue String DEFAULT '';
+
+-- +goose Down
+ALTER TABLE trigger_dev.task_runs_v2
+DROP COLUMN worker_queue;

internal-packages/clickhouse/src/taskRuns.ts

@@ -44,6 +44,7 @@ export const TaskRunV2 = z.object({
   is_test: z.boolean().default(false),
   concurrency_key: z.string().default(""),
   bulk_action_group_ids: z.array(z.string()).default([]),
+  worker_queue: z.string().default(""),
   _version: z.string(),
   _is_deleted: z.number().int().default(0),
 });