Skip to content

Commit cc19156

Browse files
D-K-PD-K-P
authored
Merge branch 'main' into posthog-proxy-app-analytics
2 parents 67f71ac + 8bf5879 commit cc19156

8 files changed

Lines changed: 305 additions & 186 deletions

‎apps/webapp/test/runsReplicationService.part1.test.ts‎

Lines changed: 62 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -98,18 +98,23 @@ describe("RunsReplicationService (part 1/7)", () => {
9898
},
9999
});
100100

101-
await setTimeout(1000);
102-
103101
const queryRuns = clickhouse.reader.query({
104102
name: "runs-replication",
105103
query: "SELECT * FROM trigger_dev.task_runs_v2",
106104
schema: z.any(),
107105
});
108106

109-
const [queryError, result] = await queryRuns({});
107+
const result = await vi.waitFor(
108+
async () => {
109+
const [queryError, rows] = await queryRuns({});
110110

111-
expect(queryError).toBeNull();
112-
expect(result?.length).toBe(1);
111+
expect(queryError).toBeNull();
112+
expect(rows?.length).toBe(1);
113+
114+
return rows;
115+
},
116+
{ timeout: 30_000, interval: 250 }
117+
);
113118
expect(result?.[0]).toEqual(
114119
expect.objectContaining({
115120
run_id: taskRun.id,
@@ -228,18 +233,23 @@ describe("RunsReplicationService (part 1/7)", () => {
228233
},
229234
});
230235

231-
await setTimeout(1000);
232-
233236
const queryRuns = clickhouse.reader.query({
234237
name: "runs-replication",
235238
query: "SELECT * FROM trigger_dev.task_runs_v2",
236239
schema: z.any(),
237240
});
238241

239-
const [queryError, result] = await queryRuns({});
242+
const result = await vi.waitFor(
243+
async () => {
244+
const [queryError, rows] = await queryRuns({});
245+
246+
expect(queryError).toBeNull();
247+
expect(rows?.length).toBe(1);
240248

241-
expect(queryError).toBeNull();
242-
expect(result?.length).toBe(1);
249+
return rows;
250+
},
251+
{ timeout: 30_000, interval: 250 }
252+
);
243253
expect(result?.[0]).toEqual(
244254
expect.objectContaining({
245255
run_id: taskRun.id,
@@ -260,10 +270,17 @@ describe("RunsReplicationService (part 1/7)", () => {
260270
params: z.object({ run_id: z.string() }),
261271
});
262272

263-
const [payloadQueryError, payloadResult] = await queryPayloads({ run_id: taskRun.id });
273+
const payloadResult = await vi.waitFor(
274+
async () => {
275+
const [payloadQueryError, payloadRows] = await queryPayloads({ run_id: taskRun.id });
264276

265-
expect(payloadQueryError).toBeNull();
266-
expect(payloadResult?.length).toBe(1);
277+
expect(payloadQueryError).toBeNull();
278+
expect(payloadRows?.length).toBe(1);
279+
280+
return payloadRows;
281+
},
282+
{ timeout: 30_000, interval: 250 }
283+
);
267284
expect(payloadResult?.[0]).toEqual(
268285
expect.objectContaining({
269286
run_id: taskRun.id,
@@ -428,19 +445,24 @@ describe("RunsReplicationService (part 1/7)", () => {
428445
},
429446
});
430447

431-
await setTimeout(1000);
432-
433448
const queryRuns = clickhouse.reader.query({
434449
name: "runs-replication-batching",
435450
query: "SELECT * FROM trigger_dev.task_runs_v2 WHERE run_id = {run_id:String}",
436451
schema: z.any(),
437452
params: z.object({ run_id: z.string() }),
438453
});
439454

440-
const [queryError, result] = await queryRuns({ run_id: taskRun.id });
455+
const result = await vi.waitFor(
456+
async () => {
457+
const [queryError, rows] = await queryRuns({ run_id: taskRun.id });
441458

442-
expect(queryError).toBeNull();
443-
expect(result?.length).toBe(1);
459+
expect(queryError).toBeNull();
460+
expect(rows?.length).toBe(1);
461+
462+
return rows;
463+
},
464+
{ timeout: 30_000, interval: 250 }
465+
);
444466
expect(result?.[0]).toEqual(
445467
expect.objectContaining({
446468
run_id: taskRun.id,
@@ -533,19 +555,25 @@ describe("RunsReplicationService (part 1/7)", () => {
533555
},
534556
});
535557

536-
await setTimeout(1000);
537-
538558
const queryPayloads = clickhouse.reader.query({
539559
name: "runs-replication-payload",
540560
query: "SELECT * FROM trigger_dev.raw_task_runs_payload_v1 WHERE run_id = {run_id:String}",
541561
schema: z.any(),
542562
params: z.object({ run_id: z.string() }),
543563
});
544564

545-
const [queryError, result] = await queryPayloads({ run_id: taskRun.id });
565+
const result = await vi.waitFor(
566+
async () => {
567+
const [queryError, rows] = await queryPayloads({ run_id: taskRun.id });
568+
569+
expect(queryError).toBeNull();
570+
expect(rows?.length).toBe(1);
571+
572+
return rows;
573+
},
574+
{ timeout: 30_000, interval: 250 }
575+
);
546576

547-
expect(queryError).toBeNull();
548-
expect(result?.length).toBe(1);
549577
expect(result?.[0]).toEqual(
550578
expect.objectContaining({
551579
run_id: taskRun.id,
@@ -639,19 +667,25 @@ describe("RunsReplicationService (part 1/7)", () => {
639667
},
640668
});
641669

642-
await setTimeout(1000);
643-
644670
const queryPayloads = clickhouse.reader.query({
645671
name: "runs-replication-payload",
646672
query: "SELECT * FROM trigger_dev.raw_task_runs_payload_v1 WHERE run_id = {run_id:String}",
647673
schema: z.any(),
648674
params: z.object({ run_id: z.string() }),
649675
});
650676

651-
const [queryError, result] = await queryPayloads({ run_id: taskRun.id });
677+
const result = await vi.waitFor(
678+
async () => {
679+
const [queryError, rows] = await queryPayloads({ run_id: taskRun.id });
680+
681+
expect(queryError).toBeNull();
682+
expect(rows?.length).toBe(1);
683+
684+
return rows;
685+
},
686+
{ timeout: 30_000, interval: 250 }
687+
);
652688

653-
expect(queryError).toBeNull();
654-
expect(result?.length).toBe(1);
655689
expect(result?.[0]).toEqual(
656690
expect.objectContaining({
657691
run_id: taskRun.id,

‎apps/webapp/test/runsReplicationService.part2.test.ts‎

Lines changed: 44 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -109,19 +109,24 @@ describe("RunsReplicationService (part 2/7)", () => {
109109
},
110110
});
111111

112-
await setTimeout(10_000);
113-
114112
// Check that the row was replicated to clickhouse
115113
const queryRuns = clickhouse.reader.query({
116114
name: "runs-replication",
117115
query: "SELECT * FROM trigger_dev.task_runs_v2",
118116
schema: z.any(),
119117
});
120118

121-
const [queryError, result] = await queryRuns({});
119+
const result = await vi.waitFor(
120+
async () => {
121+
const [queryError, rows] = await queryRuns({});
122+
123+
expect(queryError).toBeNull();
124+
expect(rows?.length).toBe(1);
122125

123-
expect(queryError).toBeNull();
124-
expect(result?.length).toBe(1);
126+
return rows;
127+
},
128+
{ timeout: 30_000, interval: 250 }
129+
);
125130
expect(result?.[0]).toEqual(
126131
expect.objectContaining({
127132
run_id: taskRun.id,
@@ -221,19 +226,23 @@ describe("RunsReplicationService (part 2/7)", () => {
221226
const created = await prisma.taskRun.createMany({ data: runsData });
222227
expect(created.count).toBe(1000);
223228

224-
// Wait for replication
225-
await setTimeout(5000);
226-
227229
// Query ClickHouse for all runs using FINAL
228230
const queryRuns = clickhouse.reader.query({
229231
name: "runs-replication-stress-bulk-insert",
230232
query: `SELECT run_id, friendly_id, trace_id, task_identifier FROM trigger_dev.task_runs_v2 FINAL`,
231233
schema: z.any(),
232234
});
233235

234-
const [queryError, result] = await queryRuns({});
235-
expect(queryError).toBeNull();
236-
expect(result?.length).toBe(1000);
236+
const result = await vi.waitFor(
237+
async () => {
238+
const [queryError, rows] = await queryRuns({});
239+
expect(queryError).toBeNull();
240+
expect(rows?.length).toBe(1000);
241+
242+
return rows;
243+
},
244+
{ timeout: 30_000, interval: 250 }
245+
);
237246

238247
// Check a few random runs for correctness
239248
for (let i = 0; i < 10; i++) {
@@ -341,35 +350,37 @@ describe("RunsReplicationService (part 2/7)", () => {
341350
data: { status: "COMPLETED_SUCCESSFULLY" },
342351
});
343352

344-
// Wait for replication
345-
await setTimeout(5000);
346-
347353
// Query ClickHouse for all runs using FINAL
348354
const queryRuns = clickhouse.reader.query({
349355
name: "runs-replication-stress-bulk-insert",
350356
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
351357
schema: z.any(),
352358
});
353359

354-
const [queryError, result] = await queryRuns({});
355-
expect(queryError).toBeNull();
356-
expect(result?.length).toBe(1000);
357-
358-
// Check a few random runs for correctness
359-
for (let i = 0; i < 10; i++) {
360-
const idx = Math.floor(Math.random() * 1000);
361-
const expected = runsData[idx];
362-
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
363-
expect(found).toBeDefined();
364-
expect(found).toEqual(
365-
expect.objectContaining({
366-
friendly_id: expected.friendlyId,
367-
trace_id: expected.traceId,
368-
task_identifier: expected.taskIdentifier,
369-
status: "COMPLETED_SUCCESSFULLY",
370-
})
371-
);
372-
}
360+
await vi.waitFor(
361+
async () => {
362+
const [queryError, result] = await queryRuns({});
363+
expect(queryError).toBeNull();
364+
expect(result?.length).toBe(1000);
365+
366+
// Check a few random runs for correctness
367+
for (let i = 0; i < 10; i++) {
368+
const idx = Math.floor(Math.random() * 1000);
369+
const expected = runsData[idx];
370+
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
371+
expect(found).toBeDefined();
372+
expect(found).toEqual(
373+
expect.objectContaining({
374+
friendly_id: expected.friendlyId,
375+
trace_id: expected.traceId,
376+
task_identifier: expected.taskIdentifier,
377+
status: "COMPLETED_SUCCESSFULLY",
378+
})
379+
);
380+
}
381+
},
382+
{ timeout: 30_000, interval: 250 }
383+
);
373384

374385
await runsReplicationService.stop();
375386
}

‎apps/webapp/test/runsReplicationService.part3.test.ts‎

Lines changed: 35 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import { ClickHouse, getTaskRunField } from "@internal/clickhouse";
22
import { replicationContainerTest } from "@internal/testcontainers";
33
import { readFile } from "node:fs/promises";
4-
import { setTimeout } from "node:timers/promises";
54
import { z } from "zod";
65
import { RunsReplicationService } from "~/services/runsReplicationService.server";
76
import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
@@ -140,40 +139,42 @@ describe("RunsReplicationService (part 3/7)", () => {
140139
},
141140
});
142141

143-
// Wait for replication
144-
await setTimeout(5000);
145-
146142
// Query ClickHouse for all runs using FINAL
147143
const queryRuns = clickhouse.reader.query({
148144
name: "runs-replication-stress-bulk-insert",
149145
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
150146
schema: z.any(),
151147
});
152148

153-
const [queryError, result] = await queryRuns({});
154-
expect(queryError).toBeNull();
155-
expect(result?.length).toBe(10);
156-
157-
// Check a few random runs for correctness
158-
for (let i = 0; i < 9; i++) {
159-
const expected = runsData[i];
160-
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
161-
expect(found).toBeDefined();
162-
expect(found).toEqual(
163-
expect.objectContaining({
164-
friendly_id: expected.friendlyId,
165-
trace_id: expected.traceId,
166-
task_identifier: expected.taskIdentifier,
167-
status: "COMPLETED_SUCCESSFULLY",
168-
})
169-
);
170-
expect(found?.output).toBeDefined();
171-
}
172-
173-
// Check the run with the bad JSON
174-
const foundBad = result?.find((r: any) => r.span_id === "bulk-10");
175-
expect(foundBad).toBeDefined();
176-
expect(foundBad?.output).toStrictEqual({});
149+
await vi.waitFor(
150+
async () => {
151+
const [queryError, result] = await queryRuns({});
152+
expect(queryError).toBeNull();
153+
expect(result?.length).toBe(10);
154+
155+
// Check a few random runs for correctness
156+
for (let i = 0; i < 9; i++) {
157+
const expected = runsData[i];
158+
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
159+
expect(found).toBeDefined();
160+
expect(found).toEqual(
161+
expect.objectContaining({
162+
friendly_id: expected.friendlyId,
163+
trace_id: expected.traceId,
164+
task_identifier: expected.taskIdentifier,
165+
status: "COMPLETED_SUCCESSFULLY",
166+
})
167+
);
168+
expect(found?.output).toBeDefined();
169+
}
170+
171+
// Check the run with the bad JSON
172+
const foundBad = result?.find((r: any) => r.span_id === "bulk-10");
173+
expect(foundBad).toBeDefined();
174+
expect(foundBad?.output).toStrictEqual({});
175+
},
176+
{ timeout: 30_000, interval: 250 }
177+
);
177178

178179
await runsReplicationService.stop();
179180
}
@@ -287,9 +288,12 @@ describe("RunsReplicationService (part 3/7)", () => {
287288
data: { status: "COMPLETED_SUCCESSFULLY" },
288289
});
289290

290-
await setTimeout(1000);
291-
292-
expect(batchFlushedEvents?.[0].taskRunInserts).toHaveLength(2);
291+
await vi.waitFor(
292+
() => {
293+
expect(batchFlushedEvents?.[0].taskRunInserts).toHaveLength(2);
294+
},
295+
{ timeout: 30_000, interval: 250 }
296+
);
293297
// Use getTaskRunField for type-safe array access
294298
expect(getTaskRunField(batchFlushedEvents![0].taskRunInserts[0], "run_id")).toEqual(run.id);
295299
expect(getTaskRunField(batchFlushedEvents![0].taskRunInserts[0], "status")).toEqual(

0 commit comments

Comments
 (0)