github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

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

Browse Source 
<!-- ccr-slack-attribution -->
_Requested by **Matt Aitken** · [Slack
thread](https://triggerdotdev.slack.com/archives/C032WA2S43F/p1783430373189849?thread_ts=1783430373.189849&cid=C032WA2S43F)_

## ✅ Checklist

- [x] The PR title follows the convention.
- [x] I ran and tested the code works (typecheck of the edited files is
clean; see Testing)

---

## Testing

**Before:** the webapp run-replication test shard failed on nearly every
PR because assertions waited a fixed 1s for rows to replicate from
Postgres → ClickHouse and intermittently checked before the row arrived
under CI load.

**After:** those assertions poll (up to 30s, 250ms interval) until the
rows land, so they pass as soon as replication completes and stop
flaking, without slowing the happy path.

These tests are testcontainers-backed (need Docker + Postgres +
ClickHouse), so the full suite is exercised in CI. Locally I confirmed
the edited `runsReplicationService.part1..part8.test.ts` files
type-check with no new errors.

---

## Changelog

**How:** wrapped the ~21 present-row assertions across
`runsReplicationService.part1..part8.test.ts` in `vi.waitFor`, matching
the existing poll pattern in `part9.test.ts`. Left absence assertions
(expecting 0 rows / no spans) on a fixed settle delay since there is
nothing to poll for. Tests only — no production code changed.

Note: this does NOT touch the `subscribe()` startup race in
`internal-packages/replication/src/client.ts` (a riskier, separate
follow-up).

💯

---
_Generated by [Claude
Code](https://claude.ai/code/session_01KtUdSLKrK17eFVuRYXT6uj)_

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
claude[bot]
2026-07-07 16:15:15 +01:00
committed by GitHub
parent aa74e68c71
commit 8bf5879b60
8 changed files with 300 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/test/runsReplicationService.part1.test.ts
+62 -28
View File
@@ -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(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,
@@ -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,
apps/webapp/test/runsReplicationService.part2.test.ts
+43 -32
View File
@@ -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(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,
@@ -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);
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",
})
);
}
// 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();
}
apps/webapp/test/runsReplicationService.part3.test.ts
+33 -29
View File
@@ -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);
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 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({});
// 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(
apps/webapp/test/runsReplicationService.part4.test.ts
+78 -39
View File
@@ -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(0);
expect(queryError).toBeNull();
expect(result?.length).toBe(1);
},
{ timeout: 30_000, interval: 250 }
);
await prisma.taskRun.delete({
where: { id: taskRun.id },
});
// 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();
apps/webapp/test/runsReplicationService.part5.test.ts
+10 -7
View File
@@ -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();
apps/webapp/test/runsReplicationService.part6.test.ts
+27 -14
View File
@@ -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(payloadResult).toHaveLength(1);
expect(payloadError).toBeNull();
expect(payloadRows).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" } });
apps/webapp/test/runsReplicationService.part7.test.ts
+8 -6
View File
@@ -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();
}
apps/webapp/test/runsReplicationService.part8.test.ts
+39 -26
View File
@@ -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({});
// 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",
})
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({});
// 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",
})
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(result).toHaveLength(2);
expect(queryError).toBeNull();
expect(rows).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");