perf(runs-replication): Improve the CPU efficiency and throughput of the runs replication to clickhouse

apps/webapp/test/concurrentFlushScheduler.test.ts

@@ -0,0 +1,161 @@
+import { ConcurrentFlushScheduler } from "~/services/runsReplicationService.server";
+
+vi.setConfig({ testTimeout: 10_000 });
+
+type TestItem = {
+  id: string;
+  event: "insert" | "update";
+  version: number;
+};
+
+describe("ConcurrentFlushScheduler", () => {
+  it("should deduplicate items by key, keeping the latest version", async () => {
+    const flushedBatches: TestItem[][] = [];
+
+    const scheduler = new ConcurrentFlushScheduler<TestItem>({
+      batchSize: 100,
+      flushInterval: 50,
+      maxConcurrency: 1,
+      callback: async (_flushId, batch) => {
+        flushedBatches.push([...batch]);
+      },
+      getKey: (item) => `${item.event}_${item.id}`,
+      shouldReplace: (existing, incoming) => incoming.version >= existing.version,
+    });
+
+    scheduler.start();
+
+    // Add items with duplicate keys but different versions
+    scheduler.addToBatch([
+      { id: "run_1", event: "insert", version: 1 },
+      { id: "run_1", event: "update", version: 2 },
+      { id: "run_2", event: "insert", version: 1 },
+    ]);
+
+    // Add more items - should merge with existing
+    scheduler.addToBatch([
+      { id: "run_1", event: "insert", version: 3 }, // Higher version, should replace
+      { id: "run_1", event: "update", version: 1 }, // Lower version, should NOT replace
+      { id: "run_2", event: "update", version: 4 },
+    ]);
+
+    // Wait for flush
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    scheduler.shutdown();
+
+    // Should have flushed once with deduplicated items
+    expect(flushedBatches.length).toBeGreaterThanOrEqual(1);
+
+    const allFlushed = flushedBatches.flat();
+
+    // Find items by their key
+    const insertRun1 = allFlushed.find((i) => i.id === "run_1" && i.event === "insert");
+    const updateRun1 = allFlushed.find((i) => i.id === "run_1" && i.event === "update");
+    const insertRun2 = allFlushed.find((i) => i.id === "run_2" && i.event === "insert");
+    const updateRun2 = allFlushed.find((i) => i.id === "run_2" && i.event === "update");
+
+    // Verify correct versions were kept
+    expect(insertRun1?.version).toBe(3); // Latest version for insert_run_1
+    expect(updateRun1?.version).toBe(2); // Original update_run_1 (v1 didn't replace v2)
+    expect(insertRun2?.version).toBe(1); // Only version for insert_run_2
+    expect(updateRun2?.version).toBe(4); // Only version for update_run_2
+  });
+
+  it("should skip items where getKey returns null", async () => {
+    const flushedBatches: TestItem[][] = [];
+
+    const scheduler = new ConcurrentFlushScheduler<TestItem>({
+      batchSize: 100,
+      flushInterval: 50,
+      maxConcurrency: 1,
+      callback: async (_flushId, batch) => {
+        flushedBatches.push([...batch]);
+      },
+      getKey: (item) => {
+        if (!item.id) {
+          return null;
+        }
+        return `${item.event}_${item.id}`;
+      },
+      shouldReplace: (existing, incoming) => incoming.version >= existing.version,
+    });
+
+    scheduler.start();
+
+    scheduler.addToBatch([
+      { id: "run_1", event: "insert", version: 1 },
+      { id: "", event: "insert", version: 2 }, // Should be skipped (null key)
+      { id: "run_2", event: "insert", version: 1 },
+    ]);
+
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    scheduler.shutdown();
+
+    const allFlushed = flushedBatches.flat();
+    expect(allFlushed).toHaveLength(2);
+    expect(allFlushed.map((i) => i.id).sort()).toEqual(["run_1", "run_2"]);
+  });
+
+  it("should flush when batch size threshold is reached", async () => {
+    const flushedBatches: TestItem[][] = [];
+
+    const scheduler = new ConcurrentFlushScheduler<TestItem>({
+      batchSize: 3,
+      flushInterval: 10000, // Long interval so timer doesn't trigger
+      maxConcurrency: 1,
+      callback: async (_flushId, batch) => {
+        flushedBatches.push([...batch]);
+      },
+      getKey: (item) => `${item.event}_${item.id}`,
+      shouldReplace: (existing, incoming) => incoming.version >= existing.version,
+    });
+
+    scheduler.start();
+
+    // Add 3 unique items - should trigger flush
+    scheduler.addToBatch([
+      { id: "run_1", event: "insert", version: 1 },
+      { id: "run_2", event: "insert", version: 1 },
+      { id: "run_3", event: "insert", version: 1 },
+    ]);
+
+    await new Promise((resolve) => setTimeout(resolve, 50));
+
+    expect(flushedBatches.length).toBe(1);
+    expect(flushedBatches[0]).toHaveLength(3);
+
+    scheduler.shutdown();
+  });
+
+  it("should respect shouldReplace returning false", async () => {
+    const flushedBatches: TestItem[][] = [];
+
+    const scheduler = new ConcurrentFlushScheduler<TestItem>({
+      batchSize: 100,
+      flushInterval: 50,
+      maxConcurrency: 1,
+      callback: async (_flushId, batch) => {
+        flushedBatches.push([...batch]);
+      },
+      getKey: (item) => `${item.event}_${item.id}`,
+      // Never replace - first item wins
+      shouldReplace: () => false,
+    });
+
+    scheduler.start();
+
+    scheduler.addToBatch([{ id: "run_1", event: "insert", version: 10 }]);
+
+    scheduler.addToBatch([{ id: "run_1", event: "insert", version: 999 }]);
+
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    scheduler.shutdown();
+
+    const allFlushed = flushedBatches.flat();
+    const insertRun1 = allFlushed.find((i) => i.id === "run_1" && i.event === "insert");
+    expect(insertRun1?.version).toBe(10); // First one wins
+  });
+});

apps/webapp/test/runsReplicationService.part1.test.ts

@@ -1,6 +1,5 @@
 import { ClickHouse } from "@internal/clickhouse";
 import { containerTest } from "@internal/testcontainers";
-import { Logger } from "@trigger.dev/core/logger";
 import { setTimeout } from "node:timers/promises";
 import { z } from "zod";
 import { TaskRunStatus } from "~/database-types";
@@ -22,6 +21,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         compression: {
           request: true,
         },
+        logLevel: "warn",
       });
 
       const { tracer, exporter } = createInMemoryTracing();
@@ -40,6 +40,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
         tracer,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();
@@ -135,6 +136,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         compression: {
           request: true,
         },
+        logLevel: "warn",
       });
 
       const { tracer, exporter } = createInMemoryTracing();
@@ -153,6 +155,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
         tracer,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();
@@ -271,6 +274,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication",
+        logLevel: "warn",
       });
 
       const { tracer, exporter } = createInMemoryTracing();
@@ -289,6 +293,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
         tracer,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();
@@ -341,6 +346,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication-batching",
+        logLevel: "warn",
       });
 
       const runsReplicationService = new RunsReplicationService({
@@ -356,6 +362,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();
@@ -443,6 +450,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication-payload",
+        logLevel: "warn",
       });
 
       const runsReplicationService = new RunsReplicationService({
@@ -458,6 +466,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();
@@ -542,6 +551,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication-payload",
+        logLevel: "warn",
       });
 
       const runsReplicationService = new RunsReplicationService({
@@ -557,6 +567,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();
@@ -646,6 +657,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication-update",
+        logLevel: "warn",
       });
 
       const runsReplicationService = new RunsReplicationService({
@@ -661,6 +673,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();
@@ -751,6 +764,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication-delete",
+        logLevel: "warn",
       });
 
       const runsReplicationService = new RunsReplicationService({
@@ -766,6 +780,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();
@@ -849,6 +864,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication-shutdown-handover",
+        logLevel: "warn",
       });
 
       // Service A
@@ -865,6 +881,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationServiceA.start();
@@ -968,6 +985,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationServiceB.start();
@@ -997,6 +1015,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication-shutdown-after-processed",
+        logLevel: "warn",
       });
 
       // Service A
@@ -1013,6 +1032,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationServiceA.start();
@@ -1114,6 +1134,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         leaderLockTimeoutMs: 5000,
         leaderLockExtendIntervalMs: 1000,
         ackIntervalSeconds: 5,
+        logLevel: "warn",
       });
 
       await runsReplicationServiceB.start();
@@ -1137,6 +1158,7 @@ describe("RunsReplicationService (part 1/2)", () => {
       const clickhouse = new ClickHouse({
         url: clickhouseContainer.getConnectionUrl(),
         name: "runs-replication-metrics",
+        logLevel: "warn",
       });
 
       const { tracer } = createInMemoryTracing();
@@ -1157,6 +1179,7 @@ describe("RunsReplicationService (part 1/2)", () => {
         ackIntervalSeconds: 5,
         tracer,
         meter: metricsHelper.meter,
+        logLevel: "warn",
       });
 
       await runsReplicationService.start();