existing

myftija · myftija · commit fdaae622d5a4 · 2025-07-24T10:48:49.000+02:00
diff --git a/apps/webapp/test/runsReplicationService.part2.test.ts b/apps/webapp/test/runsReplicationService.part2.test.ts
@@ -783,4 +783,308 @@ describe("RunsReplicationService (part 2/2)", () => {
       await runsReplicationService.stop();
     }
   );
+
+  containerTest(
+    "should merge duplicate event+run.id combinations keeping the latest version",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public.\"TaskRun\" REPLICA IDENTITY FULL;`);
+
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "runs-replication-merge-batch",
+      });
+
+      const runsReplicationService = new RunsReplicationService({
+        clickhouse,
+        pgConnectionUrl: postgresContainer.getConnectionUri(),
+        serviceName: "runs-replication-merge-batch",
+        slotName: "task_runs_to_clickhouse_v1",
+        publicationName: "task_runs_to_clickhouse_v1_publication",
+        redisOptions,
+        maxFlushConcurrency: 1,
+        flushIntervalMs: 100,
+        flushBatchSize: 10, // Higher batch size to test merging
+        leaderLockTimeoutMs: 5000,
+        leaderLockExtendIntervalMs: 1000,
+        ackIntervalSeconds: 5,
+        logger: new Logger("runs-replication-merge-batch", "info"),
+      });
+
+      await runsReplicationService.start();
+
+      const organization = await prisma.organization.create({
+        data: {
+          title: "test-merge-batch",
+          slug: "test-merge-batch",
+        },
+      });
+
+      const project = await prisma.project.create({
+        data: {
+          name: "test-merge-batch",
+          slug: "test-merge-batch",
+          organizationId: organization.id,
+          externalRef: "test-merge-batch",
+        },
+      });
+
+      const runtimeEnvironment = await prisma.runtimeEnvironment.create({
+        data: {
+          slug: "test-merge-batch",
+          type: "DEVELOPMENT",
+          projectId: project.id,
+          organizationId: organization.id,
+          apiKey: "test-merge-batch",
+          pkApiKey: "test-merge-batch",
+          shortcode: "test-merge-batch",
+        },
+      });
+
+      // Create a run and rapidly update it multiple times in a transaction
+      // This should create multiple events for the same run that get merged
+      const [taskRun] = await prisma.$transaction(async (tx) => {
+        const run = await tx.taskRun.create({
+          data: {
+            friendlyId: `run_merge_${Date.now()}`,
+            taskIdentifier: "my-task-merge",
+            payload: JSON.stringify({ version: 1 }),
+            payloadType: "application/json",
+            traceId: `merge-${Date.now()}`,
+            spanId: `merge-${Date.now()}`,
+            queue: "test-merge-batch",
+            runtimeEnvironmentId: runtimeEnvironment.id,
+            projectId: project.id,
+            organizationId: organization.id,
+            environmentType: "DEVELOPMENT",
+            engine: "V2",
+            status: "PENDING",
+          },
+        });
+
+        await tx.taskRun.update({
+          where: { id: run.id },
+          data: { status: "EXECUTING", payload: JSON.stringify({ version: 2 }) },
+        });
+
+        await tx.taskRun.update({
+          where: { id: run.id },
+          data: { status: "COMPLETED_SUCCESSFULLY", payload: JSON.stringify({ version: 3 }) },
+        });
+
+        return [run];
+      });
+
+      // Wait for replication
+      await setTimeout(2000);
+
+      // Query ClickHouse for the run using FINAL
+      const queryRuns = clickhouse.reader.query({
+        name: "runs-replication-merge-batch",
+        query: "SELECT * FROM trigger_dev.task_runs_v2 FINAL WHERE run_id = {run_id:String}",
+        schema: z.any(),
+        params: z.object({ run_id: z.string() }),
+      });
+
+      const [queryError, result] = await queryRuns({ run_id: taskRun.id });
+
+      expect(queryError).toBeNull();
+      expect(result?.length).toBe(1);
+
+      // Should have the final status from the last update
+      expect(result?.[0]).toEqual(
+        expect.objectContaining({
+          run_id: taskRun.id,
+          status: "COMPLETED_SUCCESSFULLY",
+        })
+      );
+
+      // Check payload was also updated to latest version
+      const queryPayloads = clickhouse.reader.query({
+        name: "runs-replication-merge-batch",
+        query: "SELECT * FROM trigger_dev.raw_task_runs_payload_v1 WHERE run_id = {run_id:String}",
+        schema: z.any(),
+        params: z.object({ run_id: z.string() }),
+      });
+
+      const [payloadError, payloadResult] = await queryPayloads({ run_id: taskRun.id });
+
+      expect(payloadError).toBeNull();
+      expect(payloadResult?.length).toBe(1);
+      expect(payloadResult?.[0]).toEqual(
+        expect.objectContaining({
+          run_id: taskRun.id,
+          payload: expect.objectContaining({
+            data: { version: 3 },
+          }),
+        })
+      );
+
+      await runsReplicationService.stop();
+    }
+  );
+
+  containerTest(
+    "should sort batch inserts according to table schema ordering for optimal performance",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public.\"TaskRun\" REPLICA IDENTITY FULL;`);
+
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "runs-replication-sorting",
+      });
+
+      const runsReplicationService = new RunsReplicationService({
+        clickhouse,
+        pgConnectionUrl: postgresContainer.getConnectionUri(),
+        serviceName: "runs-replication-sorting",
+        slotName: "task_runs_to_clickhouse_v1",
+        publicationName: "task_runs_to_clickhouse_v1_publication",
+        redisOptions,
+        maxFlushConcurrency: 1,
+        flushIntervalMs: 100,
+        flushBatchSize: 10,
+        leaderLockTimeoutMs: 5000,
+        leaderLockExtendIntervalMs: 1000,
+        ackIntervalSeconds: 5,
+        logger: new Logger("runs-replication-sorting", "info"),
+      });
+
+      await runsReplicationService.start();
+
+      // Create two organizations to test sorting by organization_id
+      const org1 = await prisma.organization.create({
+        data: { title: "org-z", slug: "org-z" },
+      });
+
+      const org2 = await prisma.organization.create({
+        data: { title: "org-a", slug: "org-a" },
+      });
+
+      const project1 = await prisma.project.create({
+        data: {
+          name: "test-sorting-z",
+          slug: "test-sorting-z",
+          organizationId: org1.id,
+          externalRef: "test-sorting-z",
+        },
+      });
+
+      const project2 = await prisma.project.create({
+        data: {
+          name: "test-sorting-a",
+          slug: "test-sorting-a",
+          organizationId: org2.id,
+          externalRef: "test-sorting-a",
+        },
+      });
+
+      const env1 = await prisma.runtimeEnvironment.create({
+        data: {
+          slug: "test-sorting-z",
+          type: "DEVELOPMENT",
+          projectId: project1.id,
+          organizationId: org1.id,
+          apiKey: "test-sorting-z",
+          pkApiKey: "test-sorting-z",
+          shortcode: "test-sorting-z",
+        },
+      });
+
+      const env2 = await prisma.runtimeEnvironment.create({
+        data: {
+          slug: "test-sorting-a",
+          type: "DEVELOPMENT",
+          projectId: project2.id,
+          organizationId: org2.id,
+          apiKey: "test-sorting-a",
+          pkApiKey: "test-sorting-a",
+          shortcode: "test-sorting-a",
+        },
+      });
+
+      const now = Date.now();
+
+      // Create runs in reverse alphabetical order by organization
+      // The sorting should put org2 (org-a) before org1 (org-z)
+      const [run1, run2] = await prisma.$transaction(async (tx) => {
+        const run1 = await tx.taskRun.create({
+          data: {
+            friendlyId: `run_sort_org_z_${now}`,
+            taskIdentifier: "my-task-sort",
+            payload: JSON.stringify({ org: "z" }),
+            payloadType: "application/json",
+            traceId: `sort-z-${now}`,
+            spanId: `sort-z-${now}`,
+            queue: "test-sorting",
+            runtimeEnvironmentId: env1.id,
+            projectId: project1.id,
+            organizationId: org1.id,
+            environmentType: "DEVELOPMENT",
+            engine: "V2",
+            status: "PENDING",
+            createdAt: new Date(now + 100), // Later timestamp
+          },
+        });
+
+        const run2 = await tx.taskRun.create({
+          data: {
+            friendlyId: `run_sort_org_a_${now}`,
+            taskIdentifier: "my-task-sort",
+            payload: JSON.stringify({ org: "a" }),
+            payloadType: "application/json",
+            traceId: `sort-a-${now}`,
+            spanId: `sort-a-${now}`,
+            queue: "test-sorting",
+            runtimeEnvironmentId: env2.id,
+            projectId: project2.id,
+            organizationId: org2.id,
+            environmentType: "DEVELOPMENT",
+            engine: "V2",
+            status: "PENDING",
+            createdAt: new Date(now), // Earlier timestamp
+          },
+        });
+
+        return [run1, run2];
+      });
+
+      // Wait for replication
+      await setTimeout(2000);
+
+      // Query ClickHouse for both runs
+      const queryRuns = clickhouse.reader.query({
+        name: "runs-replication-sorting",
+        query: `SELECT run_id, organization_id, project_id, environment_id, created_at, friendly_id 
+                FROM trigger_dev.task_runs_v2 FINAL 
+                WHERE run_id IN ({run_id_1:String}, {run_id_2:String})
+                ORDER BY organization_id, project_id, environment_id, created_at, run_id`,
+        schema: z.any(),
+        params: z.object({ run_id_1: z.string(), run_id_2: z.string() }),
+      });
+
+      const [queryError, result] = await queryRuns({ run_id_1: run1.id, run_id_2: run2.id });
+
+      expect(queryError).toBeNull();
+      expect(result?.length).toBe(2);
+
+      // Due to sorting, org2 (org-a) should come first even though it was created second
+      expect(result?.[0]).toEqual(
+        expect.objectContaining({
+          run_id: run2.id,
+          organization_id: org2.id,
+          friendly_id: `run_sort_org_a_${now}`,
+        })
+      );
+
+      expect(result?.[1]).toEqual(
+        expect.objectContaining({
+          run_id: run1.id,
+          organization_id: org1.id,
+          friendly_id: `run_sort_org_z_${now}`,
+        })
+      );
+
+      await runsReplicationService.stop();
+    }
+  );
 });
