Skip to content

test(webapp): poll for replicated rows instead of fixed sleeps in runs replication tests #4181

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
matt-aitken merged 2 commits into from
Jul 7, 2026
Merged

test(webapp): poll for replicated rows instead of fixed sleeps in runs replication tests #4181

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8e45b08
test(webapp): poll for replicated rows instead of fixed sleeps in run…
claude Jul 7, 2026
e78f1dd
test(webapp): poll for deletion replication in runs replication part4…
claude Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 62 additions & 28 deletions apps/webapp/test/runsReplicationService.part1.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -428,19 +445,24 @@ 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}",
schema: z.any(),
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,
Expand Down Expand Up @@ -533,19 +555,25 @@ 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}",
schema: z.any(),
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,
Expand Down Expand Up @@ -639,19 +667,25 @@ 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}",
schema: z.any(),
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,
Expand Down
77 changes: 44 additions & 33 deletions apps/webapp/test/runsReplicationService.part2.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,19 +109,24 @@ 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",
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,
Expand Down Expand Up @@ -221,19 +226,23 @@ 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",
query: `SELECT run_id, friendly_id, trace_id, task_identifier FROM trigger_dev.task_runs_v2 FINAL`,
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++) {
Expand Down Expand Up @@ -341,35 +350,37 @@ 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",
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
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();
}
Expand Down
66 changes: 35 additions & 31 deletions apps/webapp/test/runsReplicationService.part3.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -140,40 +139,42 @@ 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",
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
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();
}
Expand Down Expand Up @@ -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(
Expand Down
Loading
Loading