test(webapp): de-flake the task_run_v2 replication streaming test

Poll for the ClickHouse row with a bounded deadline instead of a fixed sleep, which is flaky under replication lag variance, and stop the replication service in a finally block so a failing assertion cannot leak it into later tests.

Author: d-cs

apps/webapp/test/runsReplicationService.taskRunV2.test.ts

@@ -47,79 +47,88 @@ describe("RunsReplicationService (task_run_v2)", () => {
 
       await runsReplicationService.start();
 
-      const organization = await prisma.organization.create({
-        data: { title: "test", slug: "test" },
-      });
-      const project = await prisma.project.create({
-        data: {
-          name: "test",
-          slug: "test",
-          organizationId: organization.id,
-          externalRef: "test",
-        },
-      });
-      const runtimeEnvironment = await prisma.runtimeEnvironment.create({
-        data: {
-          slug: "test",
-          type: "DEVELOPMENT",
-          projectId: project.id,
-          organizationId: organization.id,
-          apiKey: "test",
-          pkApiKey: "test",
-          shortcode: "test",
-        },
-      });
-
-      // A v2 run lives in task_run_v2, keyed by a KSUID id.
-      const ksuid = RunId.generateKsuid();
-      const run = await prisma.taskRunV2.create({
-        data: {
-          id: ksuid.id,
-          friendlyId: ksuid.friendlyId,
-          taskIdentifier: "my-task",
-          payload: JSON.stringify({ foo: "bar" }),
-          payloadType: "application/json",
-          traceId: "v2trace",
-          spanId: "v2span",
-          queue: "test",
-          workerQueue: "us-east-1-next",
-          region: "us-east-1",
-          planType: "free",
-          runtimeEnvironmentId: runtimeEnvironment.id,
-          projectId: project.id,
-          organizationId: organization.id,
-          environmentType: "DEVELOPMENT",
-          engine: "V2",
-        },
-      });
-
-      await setTimeout(1000);
+      try {
+        const organization = await prisma.organization.create({
+          data: { title: "test", slug: "test" },
+        });
+        const project = await prisma.project.create({
+          data: {
+            name: "test",
+            slug: "test",
+            organizationId: organization.id,
+            externalRef: "test",
+          },
+        });
+        const runtimeEnvironment = await prisma.runtimeEnvironment.create({
+          data: {
+            slug: "test",
+            type: "DEVELOPMENT",
+            projectId: project.id,
+            organizationId: organization.id,
+            apiKey: "test",
+            pkApiKey: "test",
+            shortcode: "test",
+          },
+        });
 
-      const queryRuns = clickhouse.reader.query({
-        name: "runs-replication",
-        query: "SELECT * FROM trigger_dev.task_runs_v2 WHERE run_id = {runId: String}",
-        schema: z.any(),
-        params: z.object({ runId: z.string() }),
-      });
+        // A v2 run lives in task_run_v2, keyed by a KSUID id.
+        const ksuid = RunId.generateKsuid();
+        const run = await prisma.taskRunV2.create({
+          data: {
+            id: ksuid.id,
+            friendlyId: ksuid.friendlyId,
+            taskIdentifier: "my-task",
+            payload: JSON.stringify({ foo: "bar" }),
+            payloadType: "application/json",
+            traceId: "v2trace",
+            spanId: "v2span",
+            queue: "test",
+            workerQueue: "us-east-1-next",
+            region: "us-east-1",
+            planType: "free",
+            runtimeEnvironmentId: runtimeEnvironment.id,
+            projectId: project.id,
+            organizationId: organization.id,
+            environmentType: "DEVELOPMENT",
+            engine: "V2",
+          },
+        });
 
-      const [queryError, result] = await queryRuns({ runId: run.id });
+        const queryRuns = clickhouse.reader.query({
+          name: "runs-replication",
+          query: "SELECT * FROM trigger_dev.task_runs_v2 WHERE run_id = {runId: String}",
+          schema: z.any(),
+          params: z.object({ runId: z.string() }),
+        });
 
-      expect(queryError).toBeNull();
-      expect(result?.length).toBe(1);
-      expect(result?.[0]).toEqual(
-        expect.objectContaining({
-          run_id: run.id,
-          friendly_id: run.friendlyId,
-          task_identifier: "my-task",
-          environment_id: runtimeEnvironment.id,
-          project_id: project.id,
-          organization_id: organization.id,
-          environment_type: "DEVELOPMENT",
-          engine: "V2",
-        })
-      );
+        // ClickHouse replication is asynchronous: poll until the row lands
+        // (bounded) instead of a fixed sleep, which is flaky under lag variance.
+        let queryError: unknown = null;
+        let result: Array<Record<string, unknown>> | undefined;
+        const deadline = Date.now() + 10_000;
+        do {
+          [queryError, result] = await queryRuns({ runId: run.id });
+          if (!queryError && result?.length === 1) break;
+          await setTimeout(200);
+        } while (Date.now() < deadline);
 
-      await runsReplicationService.stop();
+        expect(queryError).toBeNull();
+        expect(result?.length).toBe(1);
+        expect(result?.[0]).toEqual(
+          expect.objectContaining({
+            run_id: run.id,
+            friendly_id: run.friendlyId,
+            task_identifier: "my-task",
+            environment_id: runtimeEnvironment.id,
+            project_id: project.id,
+            organization_id: organization.id,
+            environment_type: "DEVELOPMENT",
+            engine: "V2",
+          })
+        );
+      } finally {
+        await runsReplicationService.stop();
+      }
     }
   );
 });