ClickHouse replication improvements (retrying, strip bad unicode chars) (#2205)

* Add retry logic for insert operations

Add a generic retry mechanism for task run and payload inserts to handle
transient connection errors. The new #insertWithRetry method retries up to
three times with exponential backoff and jitter on retryable connection
errors such as connection resets or timeouts. Errors are logged and
recorded in tracing spans to improve observability and robustness of the
replication service.

* Replication settings are configurable

* Log out the runIds for failed batches

* Detecting bad JSON in run replication and ignoring it

* Reproduced split unicode error

* Move output file

* Massively improved the performance

* Minor performance improvements

* Unskip tests

* Remove unused test in CH package

* Fix for the ClickHouse UI explorer

* RunReplication keepAlive defaults to false

* Add concurrency_key and bulk_action_group_ids to ClickHouse task runs

* ClickHouse package doesn't need to be built anymore for the webapp

* Set the concurrency_key from the run replication service

Author: matt-aitken

apps/webapp/app/env.server.ts

@@ -842,9 +842,13 @@ const EnvironmentSchema = z.object({
   RUN_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS: z.coerce.number().int().default(10_000),
   RUN_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS: z.coerce.number().int().default(500),
   RUN_REPLICATION_WAIT_FOR_ASYNC_INSERT: z.string().default("0"),
-  RUN_REPLICATION_KEEP_ALIVE_ENABLED: z.string().default("1"),
+  RUN_REPLICATION_KEEP_ALIVE_ENABLED: z.string().default("0"),
   RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
   RUN_REPLICATION_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
+  // Retry configuration for insert operations
+  RUN_REPLICATION_INSERT_MAX_RETRIES: z.coerce.number().int().default(3),
+  RUN_REPLICATION_INSERT_BASE_DELAY_MS: z.coerce.number().int().default(100),
+  RUN_REPLICATION_INSERT_MAX_DELAY_MS: z.coerce.number().int().default(2000),
 
   // Clickhouse
   CLICKHOUSE_URL: z.string().optional(),

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

@@ -62,6 +62,9 @@ function initializeRunsReplicationInstance() {
     logLevel: env.RUN_REPLICATION_LOG_LEVEL,
     waitForAsyncInsert: env.RUN_REPLICATION_WAIT_FOR_ASYNC_INSERT === "1",
     tracer: provider.getTracer("runs-replication-service"),
+    insertMaxRetries: env.RUN_REPLICATION_INSERT_MAX_RETRIES,
+    insertBaseDelayMs: env.RUN_REPLICATION_INSERT_BASE_DELAY_MS,
+    insertMaxDelayMs: env.RUN_REPLICATION_INSERT_MAX_DELAY_MS,
   });
 
   if (env.RUN_REPLICATION_ENABLED === "1") {

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

@@ -15,6 +15,8 @@ import { TaskRun } from "@trigger.dev/database";
 import { nanoid } from "nanoid";
 import EventEmitter from "node:events";
 import pLimit from "p-limit";
+import { logger } from "./logger.server";
+import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
 
 interface TransactionEvent<T = any> {
   tag: "insert" | "update" | "delete";
@@ -51,6 +53,10 @@ export type RunsReplicationServiceOptions = {
   logLevel?: LogLevel;
   tracer?: Tracer;
   waitForAsyncInsert?: boolean;
+  // Retry configuration for insert operations
+  insertMaxRetries?: number;
+  insertBaseDelayMs?: number;
+  insertMaxDelayMs?: number;
 };
 
 type TaskRunInsert = { _version: bigint; run: TaskRun; event: "insert" | "update" | "delete" };
@@ -80,6 +86,10 @@ export class RunsReplicationService {
   private _latestCommitEndLsn: string | null = null;
   private _lastAcknowledgedLsn: string | null = null;
   private _acknowledgeInterval: NodeJS.Timeout | null = null;
+  // Retry configuration
+  private _insertMaxRetries: number;
+  private _insertBaseDelayMs: number;
+  private _insertMaxDelayMs: number;
 
   public readonly events: EventEmitter<RunsReplicationServiceEvents>;
 
@@ -151,6 +161,11 @@ export class RunsReplicationService {
     this._replicationClient.events.on("leaderElection", (isLeader) => {
       this.logger.info("Leader election", { isLeader });
     });
+
+    // Initialize retry configuration
+    this._insertMaxRetries = options.insertMaxRetries ?? 3;
+    this._insertBaseDelayMs = options.insertBaseDelayMs ?? 100;
+    this._insertMaxDelayMs = options.insertMaxDelayMs ?? 2000;
   }
 
   public async shutdown() {
@@ -445,8 +460,37 @@ export class RunsReplicationService {
         payloadInserts: payloadInserts.length,
       });
 
-      await this.#insertTaskRunInserts(taskRunInserts);
-      await this.#insertPayloadInserts(payloadInserts);
+      // Insert task runs and payloads with retry logic for connection errors
+      const [taskRunError, taskRunResult] = await this.#insertWithRetry(
+        () => this.#insertTaskRunInserts(taskRunInserts),
+        "task run inserts",
+        flushId
+      );
+
+      const [payloadError, payloadResult] = await this.#insertWithRetry(
+        () => this.#insertPayloadInserts(payloadInserts),
+        "payload inserts",
+        flushId
+      );
+
+      // Log any errors that occurred
+      if (taskRunError) {
+        this.logger.error("Error inserting task run inserts", {
+          error: taskRunError,
+          flushId,
+          runIds: taskRunInserts.map((r) => r.run_id),
+        });
+        recordSpanError(span, taskRunError);
+      }
+
+      if (payloadError) {
+        this.logger.error("Error inserting payload inserts", {
+          error: payloadError,
+          flushId,
+          runIds: payloadInserts.map((r) => r.run_id),
+        });
+        recordSpanError(span, payloadError);
+      }
 
       this.logger.debug("Flushed inserts", {
         flushId,
@@ -456,6 +500,73 @@ export class RunsReplicationService {
     });
   }
 
+  // New method to handle inserts with retry logic for connection errors
+  async #insertWithRetry<T>(
+    insertFn: () => Promise<T>,
+    operationName: string,
+    flushId: string
+  ): Promise<[Error | null, T | null]> {
+    let lastError: Error | null = null;
+
+    for (let attempt = 1; attempt <= this._insertMaxRetries; attempt++) {
+      try {
+        const result = await insertFn();
+        return [null, result];
+      } catch (error) {
+        lastError = error instanceof Error ? error : new Error(String(error));
+
+        // Check if this is a retryable connection error
+        if (this.#isRetryableConnectionError(lastError) && attempt < this._insertMaxRetries) {
+          const delay = this.#calculateConnectionRetryDelay(attempt);
+
+          this.logger.warn(`Retrying ${operationName} due to connection error`, {
+            flushId,
+            attempt,
+            maxRetries: this._insertMaxRetries,
+            error: lastError.message,
+            delay,
+          });
+
+          await new Promise((resolve) => setTimeout(resolve, delay));
+          continue;
+        }
+        break;
+      }
+    }
+
+    return [lastError, null];
+  }
+
+  // New method to check if an error is a retryable connection error
+  #isRetryableConnectionError(error: Error): boolean {
+    const errorMessage = error.message.toLowerCase();
+    const retryableConnectionPatterns = [
+      "socket hang up",
+      "econnreset",
+      "connection reset",
+      "connection refused",
+      "connection timeout",
+      "network error",
+      "read econnreset",
+      "write econnreset",
+    ];
+
+    return retryableConnectionPatterns.some((pattern) => errorMessage.includes(pattern));
+  }
+
+  // New method to calculate retry delay for connection errors
+  #calculateConnectionRetryDelay(attempt: number): number {
+    // Exponential backoff: baseDelay, baseDelay*2, baseDelay*4, etc.
+    const delay = Math.min(
+      this._insertBaseDelayMs * Math.pow(2, attempt - 1),
+      this._insertMaxDelayMs
+    );
+
+    // Add some jitter to prevent thundering herd
+    const jitter = Math.random() * 100;
+    return delay + jitter;
+  }
+
   async #insertTaskRunInserts(taskRunInserts: TaskRunV2[]) {
     return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => {
       const [insertError, insertResult] = await this.options.clickhouse.taskRuns.insert(
@@ -604,6 +715,7 @@ export class RunsReplicationService {
       idempotency_key: run.idempotencyKey ?? "",
       expiration_ttl: run.ttl ?? "",
       output,
+      concurrency_key: run.concurrencyKey ?? "",
       _version: _version.toString(),
       _is_deleted: event === "delete" ? 1 : 0,
     };
@@ -631,6 +743,14 @@ export class RunsReplicationService {
       return { data: undefined };
     }
 
+    if (detectBadJsonStrings(data)) {
+      this.logger.warn("Detected bad JSON strings", {
+        data,
+        dataType,
+      });
+      return { data: undefined };
+    }
+
     const packet = {
       data,
       dataType,

apps/webapp/app/utils/detectBadJsonStrings.ts

@@ -0,0 +1,68 @@
+export function detectBadJsonStrings(jsonString: string): boolean {
+  // Fast path: skip everything if no \u
+  let idx = jsonString.indexOf("\\u");
+  if (idx === -1) return false;
+
+  // Use a more efficient scanning strategy
+  const length = jsonString.length;
+
+  while (idx !== -1 && idx < length - 5) {
+    // Only check if we have enough characters left
+    if (idx + 6 > length) break;
+
+    if (jsonString[idx + 1] === "u" && jsonString[idx + 2] === "d") {
+      const third = jsonString[idx + 3];
+
+      // High surrogate check
+      if (
+        /[89ab]/.test(third) &&
+        /[0-9a-f]/.test(jsonString[idx + 4]) &&
+        /[0-9a-f]/.test(jsonString[idx + 5])
+      ) {
+        // Check for low surrogate after (need at least 6 more chars)
+        if (idx + 12 > length) {
+          return true; // Incomplete high surrogate (not enough chars left)
+        }
+
+        if (
+          jsonString[idx + 6] !== "\\" ||
+          jsonString[idx + 7] !== "u" ||
+          jsonString[idx + 8] !== "d" ||
+          !/[cd]/.test(jsonString[idx + 9]) ||
+          !/[0-9a-f]/.test(jsonString[idx + 10]) ||
+          !/[0-9a-f]/.test(jsonString[idx + 11])
+        ) {
+          return true; // Incomplete high surrogate
+        }
+      }
+
+      // Low surrogate check
+      if (
+        (third === "c" || third === "d") &&
+        /[0-9a-f]/.test(jsonString[idx + 4]) &&
+        /[0-9a-f]/.test(jsonString[idx + 5])
+      ) {
+        // Check for high surrogate before (need at least 6 chars before)
+        if (idx < 6) {
+          return true; // Incomplete low surrogate (not enough chars before)
+        }
+
+        if (
+          jsonString[idx - 6] !== "\\" ||
+          jsonString[idx - 5] !== "u" ||
+          jsonString[idx - 4] !== "d" ||
+          !/[89ab]/.test(jsonString[idx - 3]) ||
+          !/[0-9a-f]/.test(jsonString[idx - 2]) ||
+          !/[0-9a-f]/.test(jsonString[idx - 1])
+        ) {
+          return true; // Incomplete low surrogate
+        }
+      }
+    }
+
+    // More efficient next search - skip ahead by 2 to avoid overlapping matches
+    idx = jsonString.indexOf("\\u", idx + 2);
+  }
+
+  return false;
+}

apps/webapp/test/bad-clickhouse-output.json

@@ -0,0 +1,3 @@
+{
+  "title": "❜ 𝐒 𝐏𝗈𝗌𝗍 . . . 𝐍𝖾𝗐 𝐂𝗈𝗇𝗍𝖾𝗇𝗍 ꒰ ⚔️ ꒱ 𝐒𝐋 ❜ 𝐔𝐋\n\n꒰ ❤️ ꒱ 𓃊 𝐋𝗲𝗮𝘃𝗲 𝖺 𝗹𝗶𝗸𝗲 𝖺𝗇\ud835"
+}