diff --git a/apps/webapp/test/runsReplicationService.part1.test.ts b/apps/webapp/test/runsReplicationService.part1.test.ts index a09ca74f43..be194b4dd4 100644 --- a/apps/webapp/test/runsReplicationService.part1.test.ts +++ b/apps/webapp/test/runsReplicationService.part1.test.ts @@ -98,18 +98,23 @@ describe("RunsReplicationService (part 1/7)", () => { }, }); - await setTimeout(1000); - const queryRuns = clickhouse.reader.query({ name: "runs-replication", query: "SELECT * FROM trigger_dev.task_runs_v2", schema: z.any(), }); - const [queryError, result] = await queryRuns({}); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({}); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1); + expect(queryError).toBeNull(); + expect(rows?.length).toBe(1); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); expect(result?.[0]).toEqual( expect.objectContaining({ run_id: taskRun.id, @@ -228,18 +233,23 @@ describe("RunsReplicationService (part 1/7)", () => { }, }); - await setTimeout(1000); - const queryRuns = clickhouse.reader.query({ name: "runs-replication", query: "SELECT * FROM trigger_dev.task_runs_v2", schema: z.any(), }); - const [queryError, result] = await queryRuns({}); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({}); + + expect(queryError).toBeNull(); + expect(rows?.length).toBe(1); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1); + return rows; + }, + { timeout: 30_000, interval: 250 } + ); expect(result?.[0]).toEqual( expect.objectContaining({ run_id: taskRun.id, @@ -260,10 +270,17 @@ describe("RunsReplicationService (part 1/7)", () => { params: z.object({ run_id: z.string() }), }); - const [payloadQueryError, payloadResult] = await queryPayloads({ run_id: taskRun.id }); + const payloadResult = await vi.waitFor( + async () => { + const [payloadQueryError, payloadRows] = await queryPayloads({ run_id: taskRun.id }); - expect(payloadQueryError).toBeNull(); - expect(payloadResult?.length).toBe(1); + expect(payloadQueryError).toBeNull(); + expect(payloadRows?.length).toBe(1); + + return payloadRows; + }, + { timeout: 30_000, interval: 250 } + ); expect(payloadResult?.[0]).toEqual( expect.objectContaining({ run_id: taskRun.id, @@ -428,8 +445,6 @@ describe("RunsReplicationService (part 1/7)", () => { }, }); - await setTimeout(1000); - const queryRuns = clickhouse.reader.query({ name: "runs-replication-batching", query: "SELECT * FROM trigger_dev.task_runs_v2 WHERE run_id = {run_id:String}", @@ -437,10 +452,17 @@ describe("RunsReplicationService (part 1/7)", () => { params: z.object({ run_id: z.string() }), }); - const [queryError, result] = await queryRuns({ run_id: taskRun.id }); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({ run_id: taskRun.id }); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1); + expect(queryError).toBeNull(); + expect(rows?.length).toBe(1); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); expect(result?.[0]).toEqual( expect.objectContaining({ run_id: taskRun.id, @@ -533,8 +555,6 @@ describe("RunsReplicationService (part 1/7)", () => { }, }); - await setTimeout(1000); - const queryPayloads = clickhouse.reader.query({ name: "runs-replication-payload", query: "SELECT * FROM trigger_dev.raw_task_runs_payload_v1 WHERE run_id = {run_id:String}", @@ -542,10 +562,18 @@ describe("RunsReplicationService (part 1/7)", () => { params: z.object({ run_id: z.string() }), }); - const [queryError, result] = await queryPayloads({ run_id: taskRun.id }); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryPayloads({ run_id: taskRun.id }); + + expect(queryError).toBeNull(); + expect(rows?.length).toBe(1); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1); expect(result?.[0]).toEqual( expect.objectContaining({ run_id: taskRun.id, @@ -639,8 +667,6 @@ describe("RunsReplicationService (part 1/7)", () => { }, }); - await setTimeout(1000); - const queryPayloads = clickhouse.reader.query({ name: "runs-replication-payload", query: "SELECT * FROM trigger_dev.raw_task_runs_payload_v1 WHERE run_id = {run_id:String}", @@ -648,10 +674,18 @@ describe("RunsReplicationService (part 1/7)", () => { params: z.object({ run_id: z.string() }), }); - const [queryError, result] = await queryPayloads({ run_id: taskRun.id }); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryPayloads({ run_id: taskRun.id }); + + expect(queryError).toBeNull(); + expect(rows?.length).toBe(1); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1); expect(result?.[0]).toEqual( expect.objectContaining({ run_id: taskRun.id, diff --git a/apps/webapp/test/runsReplicationService.part2.test.ts b/apps/webapp/test/runsReplicationService.part2.test.ts index dda3fe8e8d..a29b963847 100644 --- a/apps/webapp/test/runsReplicationService.part2.test.ts +++ b/apps/webapp/test/runsReplicationService.part2.test.ts @@ -109,8 +109,6 @@ describe("RunsReplicationService (part 2/7)", () => { }, }); - await setTimeout(10_000); - // Check that the row was replicated to clickhouse const queryRuns = clickhouse.reader.query({ name: "runs-replication", @@ -118,10 +116,17 @@ describe("RunsReplicationService (part 2/7)", () => { schema: z.any(), }); - const [queryError, result] = await queryRuns({}); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({}); + + expect(queryError).toBeNull(); + expect(rows?.length).toBe(1); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1); + return rows; + }, + { timeout: 30_000, interval: 250 } + ); expect(result?.[0]).toEqual( expect.objectContaining({ run_id: taskRun.id, @@ -221,9 +226,6 @@ describe("RunsReplicationService (part 2/7)", () => { const created = await prisma.taskRun.createMany({ data: runsData }); expect(created.count).toBe(1000); - // Wait for replication - await setTimeout(5000); - // Query ClickHouse for all runs using FINAL const queryRuns = clickhouse.reader.query({ name: "runs-replication-stress-bulk-insert", @@ -231,9 +233,16 @@ describe("RunsReplicationService (part 2/7)", () => { schema: z.any(), }); - const [queryError, result] = await queryRuns({}); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1000); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({}); + expect(queryError).toBeNull(); + expect(rows?.length).toBe(1000); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); // Check a few random runs for correctness for (let i = 0; i < 10; i++) { @@ -341,9 +350,6 @@ describe("RunsReplicationService (part 2/7)", () => { data: { status: "COMPLETED_SUCCESSFULLY" }, }); - // Wait for replication - await setTimeout(5000); - // Query ClickHouse for all runs using FINAL const queryRuns = clickhouse.reader.query({ name: "runs-replication-stress-bulk-insert", @@ -351,25 +357,30 @@ describe("RunsReplicationService (part 2/7)", () => { schema: z.any(), }); - const [queryError, result] = await queryRuns({}); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1000); - - // Check a few random runs for correctness - for (let i = 0; i < 10; i++) { - const idx = Math.floor(Math.random() * 1000); - const expected = runsData[idx]; - const found = result?.find((r: any) => r.friendly_id === expected.friendlyId); - expect(found).toBeDefined(); - expect(found).toEqual( - expect.objectContaining({ - friendly_id: expected.friendlyId, - trace_id: expected.traceId, - task_identifier: expected.taskIdentifier, - status: "COMPLETED_SUCCESSFULLY", - }) - ); - } + await vi.waitFor( + async () => { + const [queryError, result] = await queryRuns({}); + expect(queryError).toBeNull(); + expect(result?.length).toBe(1000); + + // Check a few random runs for correctness + for (let i = 0; i < 10; i++) { + const idx = Math.floor(Math.random() * 1000); + const expected = runsData[idx]; + const found = result?.find((r: any) => r.friendly_id === expected.friendlyId); + expect(found).toBeDefined(); + expect(found).toEqual( + expect.objectContaining({ + friendly_id: expected.friendlyId, + trace_id: expected.traceId, + task_identifier: expected.taskIdentifier, + status: "COMPLETED_SUCCESSFULLY", + }) + ); + } + }, + { timeout: 30_000, interval: 250 } + ); await runsReplicationService.stop(); } diff --git a/apps/webapp/test/runsReplicationService.part3.test.ts b/apps/webapp/test/runsReplicationService.part3.test.ts index 938a7cd151..9f38164683 100644 --- a/apps/webapp/test/runsReplicationService.part3.test.ts +++ b/apps/webapp/test/runsReplicationService.part3.test.ts @@ -1,7 +1,6 @@ import { ClickHouse, getTaskRunField } from "@internal/clickhouse"; import { replicationContainerTest } from "@internal/testcontainers"; import { readFile } from "node:fs/promises"; -import { setTimeout } from "node:timers/promises"; import { z } from "zod"; import { RunsReplicationService } from "~/services/runsReplicationService.server"; import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings"; @@ -140,9 +139,6 @@ describe("RunsReplicationService (part 3/7)", () => { }, }); - // Wait for replication - await setTimeout(5000); - // Query ClickHouse for all runs using FINAL const queryRuns = clickhouse.reader.query({ name: "runs-replication-stress-bulk-insert", @@ -150,30 +146,35 @@ describe("RunsReplicationService (part 3/7)", () => { schema: z.any(), }); - const [queryError, result] = await queryRuns({}); - expect(queryError).toBeNull(); - expect(result?.length).toBe(10); - - // Check a few random runs for correctness - for (let i = 0; i < 9; i++) { - const expected = runsData[i]; - const found = result?.find((r: any) => r.friendly_id === expected.friendlyId); - expect(found).toBeDefined(); - expect(found).toEqual( - expect.objectContaining({ - friendly_id: expected.friendlyId, - trace_id: expected.traceId, - task_identifier: expected.taskIdentifier, - status: "COMPLETED_SUCCESSFULLY", - }) - ); - expect(found?.output).toBeDefined(); - } - - // Check the run with the bad JSON - const foundBad = result?.find((r: any) => r.span_id === "bulk-10"); - expect(foundBad).toBeDefined(); - expect(foundBad?.output).toStrictEqual({}); + await vi.waitFor( + async () => { + const [queryError, result] = await queryRuns({}); + expect(queryError).toBeNull(); + expect(result?.length).toBe(10); + + // Check a few random runs for correctness + for (let i = 0; i < 9; i++) { + const expected = runsData[i]; + const found = result?.find((r: any) => r.friendly_id === expected.friendlyId); + expect(found).toBeDefined(); + expect(found).toEqual( + expect.objectContaining({ + friendly_id: expected.friendlyId, + trace_id: expected.traceId, + task_identifier: expected.taskIdentifier, + status: "COMPLETED_SUCCESSFULLY", + }) + ); + expect(found?.output).toBeDefined(); + } + + // Check the run with the bad JSON + const foundBad = result?.find((r: any) => r.span_id === "bulk-10"); + expect(foundBad).toBeDefined(); + expect(foundBad?.output).toStrictEqual({}); + }, + { timeout: 30_000, interval: 250 } + ); await runsReplicationService.stop(); } @@ -287,9 +288,12 @@ describe("RunsReplicationService (part 3/7)", () => { data: { status: "COMPLETED_SUCCESSFULLY" }, }); - await setTimeout(1000); - - expect(batchFlushedEvents?.[0].taskRunInserts).toHaveLength(2); + await vi.waitFor( + () => { + expect(batchFlushedEvents?.[0].taskRunInserts).toHaveLength(2); + }, + { timeout: 30_000, interval: 250 } + ); // Use getTaskRunField for type-safe array access expect(getTaskRunField(batchFlushedEvents![0].taskRunInserts[0], "run_id")).toEqual(run.id); expect(getTaskRunField(batchFlushedEvents![0].taskRunInserts[0], "status")).toEqual( diff --git a/apps/webapp/test/runsReplicationService.part4.test.ts b/apps/webapp/test/runsReplicationService.part4.test.ts index ce68cc4ac1..a653561819 100644 --- a/apps/webapp/test/runsReplicationService.part4.test.ts +++ b/apps/webapp/test/runsReplicationService.part4.test.ts @@ -93,8 +93,6 @@ describe("RunsReplicationService (part 4/7)", () => { data: { status: TaskRunStatus.COMPLETED_SUCCESSFULLY }, }); - await setTimeout(1000); - const queryRuns = clickhouse.reader.query({ name: "runs-replication-update", query: "SELECT * FROM trigger_dev.task_runs_v2 FINAL WHERE run_id = {run_id:String}", @@ -102,15 +100,20 @@ describe("RunsReplicationService (part 4/7)", () => { params: z.object({ run_id: z.string() }), }); - const [queryError, result] = await queryRuns({ run_id: taskRun.id }); + await vi.waitFor( + async () => { + const [queryError, result] = await queryRuns({ run_id: taskRun.id }); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1); - expect(result?.[0]).toEqual( - expect.objectContaining({ - run_id: taskRun.id, - status: TaskRunStatus.COMPLETED_SUCCESSFULLY, - }) + expect(queryError).toBeNull(); + expect(result?.length).toBe(1); + expect(result?.[0]).toEqual( + expect.objectContaining({ + run_id: taskRun.id, + status: TaskRunStatus.COMPLETED_SUCCESSFULLY, + }) + ); + }, + { timeout: 30_000, interval: 250 } ); await runsReplicationService.stop(); @@ -193,14 +196,6 @@ describe("RunsReplicationService (part 4/7)", () => { }, }); - await setTimeout(1000); - - await prisma.taskRun.delete({ - where: { id: taskRun.id }, - }); - - await setTimeout(1000); - const queryRuns = clickhouse.reader.query({ name: "runs-replication-delete", query: "SELECT * FROM trigger_dev.task_runs_v2 FINAL WHERE run_id = {run_id:String}", @@ -208,10 +203,33 @@ describe("RunsReplicationService (part 4/7)", () => { params: z.object({ run_id: z.string() }), }); - const [queryError, result] = await queryRuns({ run_id: taskRun.id }); + // Wait for the insert to replicate so there is a row to delete (length transitions to 1). + // Asserting presence first keeps the deletion poll below sound: length 0 can only mean the + // DELETE propagated, not that the insert never arrived. + await vi.waitFor( + async () => { + const [queryError, result] = await queryRuns({ run_id: taskRun.id }); + + expect(queryError).toBeNull(); + expect(result?.length).toBe(1); + }, + { timeout: 30_000, interval: 250 } + ); + + await prisma.taskRun.delete({ + where: { id: taskRun.id }, + }); - expect(queryError).toBeNull(); - expect(result?.length).toBe(0); + // Wait for the deletion to replicate (length transitions from 1 to 0). + await vi.waitFor( + async () => { + const [queryError, result] = await queryRuns({ run_id: taskRun.id }); + + expect(queryError).toBeNull(); + expect(result?.length).toBe(0); + }, + { timeout: 30_000, interval: 250 } + ); await runsReplicationService.stop(); } @@ -320,16 +338,21 @@ describe("RunsReplicationService (part 4/7)", () => { }, }); - await setTimeout(1000); - const queryRuns = clickhouse.reader.query({ name: "runs-replication-shutdown-handover", query: "SELECT * FROM trigger_dev.task_runs_v2 FINAL ORDER BY created_at ASC", schema: z.any(), }); - const [queryError, result] = await queryRuns({}); - expect(queryError).toBeNull(); - expect(result?.length).toBe(1); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({}); + expect(queryError).toBeNull(); + expect(rows?.length).toBe(1); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); expect(result?.[0]).toEqual(expect.objectContaining({ run_id: taskRun1.id })); // Service B @@ -351,12 +374,18 @@ describe("RunsReplicationService (part 4/7)", () => { await runsReplicationServiceB.start(); - await setTimeout(1000); + const resultB = await vi.waitFor( + async () => { + const [queryErrorB, rows] = await queryRuns({}); - const [queryErrorB, resultB] = await queryRuns({}); + expect(queryErrorB).toBeNull(); + expect(rows?.length).toBe(2); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); - expect(queryErrorB).toBeNull(); - expect(resultB?.length).toBe(2); expect(resultB).toEqual( expect.arrayContaining([ expect.objectContaining({ run_id: taskRun1.id }), @@ -445,8 +474,6 @@ describe("RunsReplicationService (part 4/7)", () => { }, }); - await setTimeout(1000); - const queryRuns = clickhouse.reader.query({ name: "runs-replication-shutdown-after-processed", query: "SELECT * FROM trigger_dev.task_runs_v2 FINAL WHERE run_id = {run_id:String}", @@ -454,9 +481,16 @@ describe("RunsReplicationService (part 4/7)", () => { params: z.object({ run_id: z.string() }), }); - const [queryErrorA, resultA] = await queryRuns({ run_id: taskRun1.id }); - expect(queryErrorA).toBeNull(); - expect(resultA?.length).toBe(1); + const resultA = await vi.waitFor( + async () => { + const [queryErrorA, rows] = await queryRuns({ run_id: taskRun1.id }); + expect(queryErrorA).toBeNull(); + expect(rows?.length).toBe(1); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); expect(resultA?.[0]).toEqual(expect.objectContaining({ run_id: taskRun1.id })); await runsReplicationServiceA.shutdown(); @@ -500,11 +534,16 @@ describe("RunsReplicationService (part 4/7)", () => { await runsReplicationServiceB.start(); - await setTimeout(1000); + const resultB = await vi.waitFor( + async () => { + const [queryErrorB, rows] = await queryRuns({ run_id: taskRun2.id }); + expect(queryErrorB).toBeNull(); + expect(rows?.length).toBe(1); - const [queryErrorB, resultB] = await queryRuns({ run_id: taskRun2.id }); - expect(queryErrorB).toBeNull(); - expect(resultB?.length).toBe(1); + return rows; + }, + { timeout: 30_000, interval: 250 } + ); expect(resultB?.[0]).toEqual(expect.objectContaining({ run_id: taskRun2.id })); await runsReplicationServiceB.stop(); diff --git a/apps/webapp/test/runsReplicationService.part5.test.ts b/apps/webapp/test/runsReplicationService.part5.test.ts index 839bc69d49..b19cefbeaf 100644 --- a/apps/webapp/test/runsReplicationService.part5.test.ts +++ b/apps/webapp/test/runsReplicationService.part5.test.ts @@ -1,6 +1,5 @@ import { ClickHouse } from "@internal/clickhouse"; import { replicationContainerTest } from "@internal/testcontainers"; -import { setTimeout } from "node:timers/promises"; import { z } from "zod"; import { RunsReplicationService } from "~/services/runsReplicationService.server"; import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory"; @@ -115,9 +114,6 @@ describe("RunsReplicationService (part 5/7)", () => { return [run1, run2]; }); - // Wait for replication - await setTimeout(1000); - // Query ClickHouse for both runs using FINAL const queryRuns = clickhouse.reader.query({ name: "runs-replication-multi-event-tx", @@ -126,9 +122,16 @@ describe("RunsReplicationService (part 5/7)", () => { 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); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({ run_id_1: run1.id, run_id_2: run2.id }); + expect(queryError).toBeNull(); + expect(rows?.length).toBe(2); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); const run1Result = result?.find((r: any) => r.run_id === run1.id); const run2Result = result?.find((r: any) => r.run_id === run2.id); expect(run1Result).toBeDefined(); diff --git a/apps/webapp/test/runsReplicationService.part6.test.ts b/apps/webapp/test/runsReplicationService.part6.test.ts index 90004c1126..757244ec0e 100644 --- a/apps/webapp/test/runsReplicationService.part6.test.ts +++ b/apps/webapp/test/runsReplicationService.part6.test.ts @@ -1,6 +1,5 @@ import { ClickHouse, getPayloadField, getTaskRunField } from "@internal/clickhouse"; import { replicationContainerTest } from "@internal/testcontainers"; -import { setTimeout } from "node:timers/promises"; import { z } from "zod"; import { RunsReplicationService } from "~/services/runsReplicationService.server"; import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory"; @@ -162,10 +161,13 @@ describe("RunsReplicationService (part 6/7)", () => { }, }); - await setTimeout(1000); - - expect(batchFlushedEvents[0]?.taskRunInserts.length).toBeGreaterThan(1); - expect(batchFlushedEvents[0]?.payloadInserts.length).toBeGreaterThan(1); + await vi.waitFor( + () => { + expect(batchFlushedEvents[0]?.taskRunInserts.length).toBeGreaterThan(1); + expect(batchFlushedEvents[0]?.payloadInserts.length).toBeGreaterThan(1); + }, + { timeout: 30_000, interval: 250 } + ); // Verify sorting order: organization_id, project_id, environment_id, created_at, run_id for (let i = 1; i < batchFlushedEvents[0]?.taskRunInserts.length; i++) { @@ -410,9 +412,6 @@ describe("RunsReplicationService (part 6/7)", () => { }, }); - // Wait for replication - await setTimeout(1500); - // Query ClickHouse directly to get all columns const queryRuns = clickhouse.reader.query({ name: "exhaustive-replication-test", @@ -421,10 +420,17 @@ describe("RunsReplicationService (part 6/7)", () => { params: z.object({ run_id: z.string() }), }); - const [queryError, result] = await queryRuns({ run_id: taskRun.id }); + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({ run_id: taskRun.id }); - expect(queryError).toBeNull(); - expect(result).toHaveLength(1); + expect(queryError).toBeNull(); + expect(rows).toHaveLength(1); + + return rows; + }, + { timeout: 30_000, interval: 250 } + ); const clickhouseRun = result![0]; @@ -519,10 +525,17 @@ describe("RunsReplicationService (part 6/7)", () => { params: z.object({ run_id: z.string() }), }); - const [payloadError, payloadResult] = await queryPayloads({ run_id: taskRun.id }); + const payloadResult = await vi.waitFor( + async () => { + const [payloadError, payloadRows] = await queryPayloads({ run_id: taskRun.id }); + + expect(payloadError).toBeNull(); + expect(payloadRows).toHaveLength(1); - expect(payloadError).toBeNull(); - expect(payloadResult).toHaveLength(1); + return payloadRows; + }, + { timeout: 30_000, interval: 250 } + ); expect(payloadResult![0].run_id).toBe(taskRun.id); expect(parseClickhouseTimestamp(payloadResult![0].created_at)).toBe(createdAt.getTime()); expect(payloadResult![0].payload).toEqual({ data: { input: "test-payload" } }); diff --git a/apps/webapp/test/runsReplicationService.part7.test.ts b/apps/webapp/test/runsReplicationService.part7.test.ts index 77ca3bcc0f..608683c53c 100644 --- a/apps/webapp/test/runsReplicationService.part7.test.ts +++ b/apps/webapp/test/runsReplicationService.part7.test.ts @@ -96,9 +96,6 @@ describe("RunsReplicationService (part 7/7)", () => { // Stop the interval clearInterval(interval); - // Wait for replication - await setTimeout(1000); - // Query ClickHouse for all runs using FINAL const queryRuns = clickhouse.reader.query({ name: "runs-replication-long-tx", @@ -106,10 +103,15 @@ describe("RunsReplicationService (part 7/7)", () => { schema: z.any(), }); - const [queryError, result] = await queryRuns({}); - expect(queryError).toBeNull(); + await vi.waitFor( + async () => { + const [queryError, result] = await queryRuns({}); + expect(queryError).toBeNull(); - expect(result?.length).toBeGreaterThanOrEqual(50); + expect(result?.length).toBeGreaterThanOrEqual(50); + }, + { timeout: 30_000, interval: 250 } + ); await runsReplicationService.stop(); } diff --git a/apps/webapp/test/runsReplicationService.part8.test.ts b/apps/webapp/test/runsReplicationService.part8.test.ts index 0b4afab2d3..cd6dca1ffb 100644 --- a/apps/webapp/test/runsReplicationService.part8.test.ts +++ b/apps/webapp/test/runsReplicationService.part8.test.ts @@ -159,9 +159,6 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => { } await seedFkRows(prisma, "legacy", TaskRunStatus.PENDING, "run_dual_legacy"); - - // Wait for BOTH streams to flush into ClickHouse. - await setTimeout(3000); } finally { await newPrisma.$disconnect(); } @@ -177,15 +174,21 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => { }), }); - const [queryError, result] = await queryRuns({}); - - expect(queryError).toBeNull(); - expect(result).toHaveLength(1); - expect(result?.[0]).toEqual( - expect.objectContaining({ - run_id: sharedRunId, - status: "COMPLETED_SUCCESSFULLY", - }) + // Poll until BOTH streams have flushed and the gen-1 winner has settled in ClickHouse. + await vi.waitFor( + async () => { + const [queryError, result] = await queryRuns({}); + + expect(queryError).toBeNull(); + expect(result).toHaveLength(1); + expect(result?.[0]).toEqual( + expect.objectContaining({ + run_id: sharedRunId, + status: "COMPLETED_SUCCESSFULLY", + }) + ); + }, + { timeout: 30_000, interval: 250 } ); } finally { await runsReplicationService?.stop(); @@ -329,7 +332,6 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => { // THEN flush the LEGACY (gen-0, PENDING) snapshot. await seedFkRows(prisma, "legacy", TaskRunStatus.PENDING, "run_dual_legacy"); - await setTimeout(3000); } finally { await newPrisma.$disconnect(); } @@ -345,15 +347,21 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => { }), }); - const [queryError, result] = await queryRuns({}); - - expect(queryError).toBeNull(); - expect(result).toHaveLength(1); - expect(result?.[0]).toEqual( - expect.objectContaining({ - run_id: sharedRunId, - status: "COMPLETED_SUCCESSFULLY", - }) + // Poll until BOTH streams have flushed and the gen-1 winner has settled in ClickHouse. + await vi.waitFor( + async () => { + const [queryError, result] = await queryRuns({}); + + expect(queryError).toBeNull(); + expect(result).toHaveLength(1); + expect(result?.[0]).toEqual( + expect.objectContaining({ + run_id: sharedRunId, + status: "COMPLETED_SUCCESSFULLY", + }) + ); + }, + { timeout: 30_000, interval: 250 } ); } finally { await runsReplicationService?.stop(); @@ -498,9 +506,6 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => { // Seed run X ONLY in legacy and run Y ONLY in new. await seedRun(prisma, "legacy", legacyRunId, TaskRunStatus.PENDING); await seedRun(newPrisma, "new", newRunId, TaskRunStatus.COMPLETED_SUCCESSFULLY); - - // Wait for BOTH streams to flush into ClickHouse. - await setTimeout(3000); } finally { await newPrisma.$disconnect(); } @@ -514,10 +519,18 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => { }), }); - const [queryError, result] = await queryRuns({}); + // Poll until BOTH sources have independently flushed their run into ClickHouse. + const result = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({}); + + expect(queryError).toBeNull(); + expect(rows).toHaveLength(2); - expect(queryError).toBeNull(); - expect(result).toHaveLength(2); + return rows; + }, + { timeout: 30_000, interval: 250 } + ); const byRunId = new Map(result?.map((row) => [row.run_id, row.status])); expect(byRunId.get(legacyRunId)).toBe("PENDING");