prepare for using clickhouse cloud and now running ch migrations during boot in the entrypoint.sh

Author: ericallam

apps/webapp/app/env.server.ts

@@ -754,6 +754,18 @@ const EnvironmentSchema = z.object({
     .optional()
     .transform((v) => v ?? process.env.REDIS_PASSWORD),
   RUN_REPLICATION_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
+
+  RUN_REPLICATION_CLICKHOUSE_URL: z.string().optional(),
+  RUN_REPLICATION_ENABLED: z.string().default("0"),
+  RUN_REPLICATION_SLOT_NAME: z.string().default("task_runs_to_clickhouse_v1"),
+  RUN_REPLICATION_PUBLICATION_NAME: z.string().default("task_runs_to_clickhouse_v1_publication"),
+  RUN_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(100),
+  RUN_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
+  RUN_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
+  RUN_REPLICATION_INSERT_STRATEGY: z.enum(["streaming", "batching"]).default("batching"),
+  RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
+  RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
+  RUN_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/routes/admin.api.v1.runs-replication.start.ts

@@ -34,7 +34,7 @@ export async function action({ request }: ActionFunctionArgs) {
     const body = await request.json();
     const { insertStrategy } = schema.parse(body);
 
-    await runsReplicationInstance.start(insertStrategy);
+    await runsReplicationInstance?.start(insertStrategy);
 
     return json({
       success: true,

apps/webapp/app/routes/admin.api.v1.runs-replication.stop.ts

@@ -26,7 +26,7 @@ export async function action({ request }: ActionFunctionArgs) {
   }
 
   try {
-    await runsReplicationInstance.stop();
+    await runsReplicationInstance?.stop();
 
     return json({
       success: true,

apps/webapp/app/routes/admin.api.v1.runs-replication.teardown.ts

@@ -26,7 +26,7 @@ export async function action({ request }: ActionFunctionArgs) {
   }
 
   try {
-    await runsReplicationInstance.teardown();
+    await runsReplicationInstance?.teardown();
 
     return json({
       success: true,

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

@@ -4,6 +4,7 @@ import { singleton } from "~/utils/singleton";
 import invariant from "tiny-invariant";
 import { env } from "~/env.server";
 import { metricsRegister } from "~/metrics.server";
+import { logger } from "./logger.server";
 
 export const runsReplicationInstance = singleton(
   "runsReplicationInstance",
@@ -14,14 +15,22 @@ function initializeRunsReplicationInstance() {
   const { DATABASE_URL } = process.env;
   invariant(typeof DATABASE_URL === "string", "DATABASE_URL env var not set");
 
-  const clickhouse = ClickHouse.fromEnv();
+  if (!env.RUN_REPLICATION_CLICKHOUSE_URL) {
+    logger.info("🗃️ Runs replication service not enabled");
+    return;
+  }
+
+  const clickhouse = new ClickHouse({
+    url: env.RUN_REPLICATION_CLICKHOUSE_URL,
+    name: "runs-replication",
+  });
 
   const service = new RunsReplicationService({
     clickhouse: clickhouse,
     pgConnectionUrl: DATABASE_URL,
     serviceName: "runs-replication",
-    slotName: "task_runs_to_clickhouse_v1",
-    publicationName: "task_runs_to_clickhouse_v1_publication",
+    slotName: env.RUN_REPLICATION_SLOT_NAME,
+    publicationName: env.RUN_REPLICATION_PUBLICATION_NAME,
     redisOptions: {
       keyPrefix: "runs-replication:",
       port: env.RUN_REPLICATION_REDIS_PORT ?? undefined,
@@ -32,7 +41,27 @@ function initializeRunsReplicationInstance() {
       ...(env.RUN_REPLICATION_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
     },
     metricsRegister: metricsRegister,
+    maxFlushConcurrency: env.RUN_REPLICATION_MAX_FLUSH_CONCURRENCY,
+    flushIntervalMs: env.RUN_REPLICATION_FLUSH_INTERVAL_MS,
+    flushBatchSize: env.RUN_REPLICATION_FLUSH_BATCH_SIZE,
+    insertStrategy: env.RUN_REPLICATION_INSERT_STRATEGY,
+    leaderLockTimeoutMs: env.RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS,
+    leaderLockExtendIntervalMs: env.RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS,
+    ackIntervalSeconds: env.RUN_REPLICATION_ACK_INTERVAL_SECONDS,
   });
 
+  if (env.RUN_REPLICATION_ENABLED === "1") {
+    service
+      .start()
+      .then(() => {
+        logger.info("🗃️ Runs replication service started");
+      })
+      .catch((error) => {
+        logger.error("🗃️ Runs replication service failed to start", {
+          error,
+        });
+      });
+  }
+
   return service;
 }

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

@@ -23,9 +23,12 @@ export type RunsReplicationServiceOptions = {
   maxFlushConcurrency?: number;
   flushIntervalMs?: number;
   flushBatchSize?: number;
+  leaderLockTimeoutMs?: number;
+  leaderLockExtendIntervalMs?: number;
+  ackIntervalSeconds?: number;
 };
 
-type TaskRunInsert = { _version: bigint; run: TaskRun; event: "insert" | "update" };
+type TaskRunInsert = { _version: bigint; run: TaskRun; event: "insert" | "update" | "delete" };
 
 export class RunsReplicationService {
   private _lastLsn: string | null = null;
@@ -61,9 +64,9 @@ export class RunsReplicationService {
       autoAcknowledge: false,
       publicationActions: ["insert", "update"],
       logger: new Logger("RunsReplicationService", "debug"),
-      leaderLockTimeoutMs: 30_000,
-      leaderLockExtendIntervalMs: 10_000,
-      ackIntervalSeconds: 10,
+      leaderLockTimeoutMs: options.leaderLockTimeoutMs ?? 30_000,
+      leaderLockExtendIntervalMs: options.leaderLockExtendIntervalMs ?? 10_000,
+      ackIntervalSeconds: options.ackIntervalSeconds ?? 10,
     });
 
     this._concurrentFlushScheduler = new ConcurrentFlushScheduler<TaskRunInsert>({
@@ -217,20 +220,6 @@ export class RunsReplicationService {
       return;
     }
 
-    const relevantEvents = transaction.events.filter(
-      (event) => event.tag === "insert" || event.tag === "update"
-    );
-
-    if (relevantEvents.length === 0) {
-      this.logger.debug("No relevant events", {
-        transaction,
-      });
-
-      await this._replicationClient.acknowledge(transaction.commitEndLsn);
-
-      return;
-    }
-
     this.logger.debug("Handling transaction", {
       transaction,
     });
@@ -242,20 +231,20 @@ export class RunsReplicationService {
 
     if (this._insertStrategy === "streaming") {
       await this._concurrentFlushScheduler.addToBatch(
-        relevantEvents.map((event) => ({
+        transaction.events.map((event) => ({
           _version,
           run: event.data,
-          event: event.tag as "insert" | "update",
+          event: event.tag,
         }))
       );
     } else {
       const [flushError] = await tryCatch(
         this.#flushBatch(
           nanoid(),
-          relevantEvents.map((event) => ({
+          transaction.events.map((event) => ({
             _version,
             run: event.data,
-            event: event.tag as "insert" | "update",
+            event: event.tag,
           }))
         )
       );
@@ -376,11 +365,12 @@ export class RunsReplicationService {
       };
     }
 
-    if (event === "update") {
+    if (event === "update" || event === "delete") {
       const taskRunInsert = await this.#prepareTaskRunInsert(
         run,
         run.organizationId,
         run.environmentType,
+        event,
         _version
       );
 
@@ -391,7 +381,7 @@ export class RunsReplicationService {
     }
 
     const [taskRunInsert, payloadInsert] = await Promise.all([
-      this.#prepareTaskRunInsert(run, run.organizationId, run.environmentType, _version),
+      this.#prepareTaskRunInsert(run, run.organizationId, run.environmentType, event, _version),
       this.#preparePayloadInsert(run, _version),
     ]);
 
@@ -405,6 +395,7 @@ export class RunsReplicationService {
     run: TaskRun,
     organizationId: string,
     environmentType: string,
+    event: "insert" | "update" | "delete",
     _version: bigint
   ): Promise<TaskRunV1> {
     const output = await this.#prepareJson(run.output, run.outputType);
@@ -424,10 +415,10 @@ export class RunsReplicationService {
       queue: run.queue,
       span_id: run.spanId,
       trace_id: run.traceId,
-      error: run.error ? (run.error as TaskRunError) : undefined,
+      error: { data: run.error },
       attempt: run.attemptNumber ?? 1,
-      schedule_id: run.scheduleId,
-      batch_id: run.batchId,
+      schedule_id: run.scheduleId ?? "",
+      batch_id: run.batchId ?? "",
       completed_at: run.completedAt?.getTime(),
       started_at: run.startedAt?.getTime(),
       executed_at: run.executedAt?.getTime(),
@@ -438,18 +429,19 @@ export class RunsReplicationService {
       cost_in_cents: run.costInCents,
       base_cost_in_cents: run.baseCostInCents,
       tags: run.runTags ?? [],
-      task_version: run.taskVersion,
-      sdk_version: run.sdkVersion,
-      cli_version: run.cliVersion,
-      machine_preset: run.machinePreset,
-      root_run_id: run.rootTaskRunId,
-      parent_run_id: run.parentTaskRunId,
+      task_version: run.taskVersion ?? "",
+      sdk_version: run.sdkVersion ?? "",
+      cli_version: run.cliVersion ?? "",
+      machine_preset: run.machinePreset ?? "",
+      root_run_id: run.rootTaskRunId ?? "",
+      parent_run_id: run.parentTaskRunId ?? "",
       depth: run.depth,
       is_test: run.isTest,
-      idempotency_key: run.idempotencyKey,
-      expiration_ttl: run.ttl,
+      idempotency_key: run.idempotencyKey ?? "",
+      expiration_ttl: run.ttl ?? "",
       output,
       _version: _version.toString(),
+      _is_deleted: event === "delete" ? 1 : 0,
     };
   }
 

docker/Dockerfile

@@ -1,5 +1,8 @@
 ARG NODE_IMAGE=node:20.11.1-bullseye-slim@sha256:5a5a92b3a8d392691c983719dbdc65d9f30085d6dcd65376e7a32e6fe9bf4cbe
 
+FROM golang:1.23-alpine AS goose_builder
+RUN go install github.com/pressly/goose/v3/cmd/goose@latest
+
 FROM ${NODE_IMAGE} AS pruner
 
 WORKDIR /triggerdotdev
@@ -43,6 +46,11 @@ WORKDIR /triggerdotdev
 # Corepack is used to install pnpm
 RUN corepack enable
 
+# Goose and schemas
+COPY --from=goose_builder /go/bin/goose /usr/local/bin/goose
+RUN chmod +x /usr/local/bin/goose
+COPY --chown=node:node internal-packages/clickhouse/schema /triggerdotdev/internal-packages/clickhouse/schema
+
 COPY --from=pruner --chown=node:node /triggerdotdev/out/full/ .
 COPY --from=dev-deps --chown=node:node /triggerdotdev/ .
 COPY --chown=node:node turbo.json turbo.json
@@ -70,6 +78,10 @@ COPY --from=builder --chown=node:node /triggerdotdev/apps/webapp/public ./apps/w
 COPY --from=builder --chown=node:node /triggerdotdev/apps/webapp/prisma/seed.js ./apps/webapp/prisma/seed.js
 COPY --from=builder --chown=node:node /triggerdotdev/scripts ./scripts
 
+# Goose and schemas
+COPY --from=builder /usr/local/bin/goose /usr/local/bin/goose
+COPY --from=builder --chown=node:node /triggerdotdev/internal-packages/clickhouse/schema /triggerdotdev/internal-packages/clickhouse/schema
+
 EXPOSE 3000
 
 USER node