Skip to content

Commit e66a3d7

Browse files
jumskijumski
authored
fix(edge-worker): await worker startup in integration tests (#680)
## Summary Make the integration-test harness await worker startup. `startWorker()` in `pkgs/edge-worker/tests/integration/_helpers.ts` fired `worker.startOnlyOnce()` fire-and-forget, so the startup transaction (`ensure_flow_compiled`: `create_flow`/`add_step`, or destructive local recompile via `delete_flow_and_data`) could race the test body and commit across test boundaries, interleaving with the next test's `reset_db`. A stray commit then left a listed PGMQ queue with no owning flow row, and a later `create_flow` rejected it (`queue "X" is already in use by another owner`). Startup errors now fail at the call site instead of surfacing via a later `worker.stop()`. Test-only change: `_helpers.ts`, 31 call sites across 9 integration test files, and `performanceMapFlow`'s `startWorkers`. No product code, SQL, or migrations. Base of the #679 stack (queue identity feature); that PR's CI originally failed on this race. ## Checks - `mapFlow.test.ts` focused runs pass twice consecutively (no orphan-queue leak across runs); `queueIdentity.test.ts` likewise - Full `pnpm nx test:integration edge-worker`: 63 passed, 0 failed - `nx affected --target=prepush`: 32/32 successful Part of #650
2 parents 4c09f51 + cae4772 commit e66a3d7

9 files changed

Lines changed: 39 additions & 33 deletions

‎pkgs/edge-worker/tests/integration/_helpers.ts‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ export function createTestPlatformAdapter(sql: postgres.Sql): PlatformAdapter<Su
4949
};
5050
}
5151

52-
export function startWorker<TFlow extends AnyFlow>(
52+
export async function startWorker<TFlow extends AnyFlow>(
5353
sql: postgres.Sql,
5454
flow: TFlow,
5555
options: FlowWorkerConfig
@@ -84,7 +84,13 @@ export function startWorker<TFlow extends AnyFlow>(
8484

8585
const worker = createFlowWorker(flow, mergedOptions, () => consoleLogger, createTestPlatformAdapter(sql));
8686

87-
worker.startOnlyOnce({
87+
// Await startup: ensure_flow_compiled may create the flow, its queue, or
88+
// destructively recompile (dropping queues). Fire-and-forget startup lets
89+
// that transaction race the test body's own create_flow/add_step calls and,
90+
// under load, commit after the test finished — leaving a listed queue with
91+
// no owning flow row, which the create_flow ownership guard then rejects in
92+
// a later test (#650).
93+
await worker.startOnlyOnce({
8894
edgeFunctionName: 'test_flow',
8995
workerId: crypto.randomUUID(),
9096
});

‎pkgs/edge-worker/tests/integration/flow/conditionalFlow.test.ts‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ Deno.test(
4949
}
5050
);
5151

52-
const worker = startWorker(sql, ConditionalIfMetFlow, workerConfig);
52+
const worker = await startWorker(sql, ConditionalIfMetFlow, workerConfig);
5353

5454
try {
5555
await createFlowInDb(sql, ConditionalIfMetFlow);
@@ -118,7 +118,7 @@ Deno.test(
118118
}
119119
);
120120

121-
const worker = startWorker(sql, ConditionalIfUnmetFlow, workerConfig);
121+
const worker = await startWorker(sql, ConditionalIfUnmetFlow, workerConfig);
122122

123123
try {
124124
await createFlowInDb(sql, ConditionalIfUnmetFlow);
@@ -192,7 +192,7 @@ Deno.test(
192192
}
193193
);
194194

195-
const worker = startWorker(sql, ConditionalIfNotMetFlow, workerConfig);
195+
const worker = await startWorker(sql, ConditionalIfNotMetFlow, workerConfig);
196196

197197
try {
198198
await createFlowInDb(sql, ConditionalIfNotMetFlow);
@@ -260,7 +260,7 @@ Deno.test(
260260
}
261261
);
262262

263-
const worker = startWorker(sql, ConditionalIfNotUnmetFlow, workerConfig);
263+
const worker = await startWorker(sql, ConditionalIfNotUnmetFlow, workerConfig);
264264

265265
try {
266266
await createFlowInDb(sql, ConditionalIfNotUnmetFlow);
@@ -333,7 +333,7 @@ Deno.test(
333333
}
334334
);
335335

336-
const worker = startWorker(sql, SkipCascadeFlow, workerConfig);
336+
const worker = await startWorker(sql, SkipCascadeFlow, workerConfig);
337337

338338
try {
339339
await createFlowInDb(sql, SkipCascadeFlow);
@@ -434,7 +434,7 @@ Deno.test(
434434
}
435435
);
436436

437-
const worker = startWorker(sql, NonCascadeSkipFlow, workerConfig);
437+
const worker = await startWorker(sql, NonCascadeSkipFlow, workerConfig);
438438

439439
try {
440440
await createFlowInDb(sql, NonCascadeSkipFlow);
@@ -529,7 +529,7 @@ Deno.test(
529529
}
530530
);
531531

532-
const worker = startWorker(sql, FailOnUnmetFlow, workerConfig);
532+
const worker = await startWorker(sql, FailOnUnmetFlow, workerConfig);
533533

534534
try {
535535
await createFlowInDb(sql, FailOnUnmetFlow);

‎pkgs/edge-worker/tests/integration/flow/contextResourcesFlow.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ Deno.test(
5555
withPgNoTransaction(async (sql) => {
5656
await sql`select pgflow_tests.reset_db();`;
5757

58-
const worker = startWorker(sql, ContextResourcesFlow, {
58+
const worker = await startWorker(sql, ContextResourcesFlow, {
5959
maxConcurrent: 1,
6060
batchSize: 10,
6161
maxPollSeconds: 1,

‎pkgs/edge-worker/tests/integration/flow/mapFlow.test.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ Deno.test(
6666
withPgNoTransaction(async (sql) => {
6767
await sql`select pgflow_tests.reset_db();`;
6868

69-
const worker = startWorker(sql, RootMapFlow, {
69+
const worker = await startWorker(sql, RootMapFlow, {
7070
maxConcurrent: 3,
7171
batchSize: 10,
7272
maxPollSeconds: 1,
@@ -114,7 +114,7 @@ Deno.test(
114114
withPgNoTransaction(async (sql) => {
115115
await sql`select pgflow_tests.reset_db();`;
116116

117-
const worker = startWorker(sql, DependentMapFlow, {
117+
const worker = await startWorker(sql, DependentMapFlow, {
118118
maxConcurrent: 3,
119119
batchSize: 10,
120120
maxPollSeconds: 1,
@@ -176,7 +176,7 @@ Deno.test(
176176
withPgNoTransaction(async (sql) => {
177177
await sql`select pgflow_tests.reset_db();`;
178178

179-
const worker = startWorker(sql, EmptyArrayMapFlow, {
179+
const worker = await startWorker(sql, EmptyArrayMapFlow, {
180180
maxConcurrent: 3,
181181
batchSize: 10,
182182
maxPollSeconds: 1,
@@ -226,7 +226,7 @@ Deno.test(
226226
withPgNoTransaction(async (sql) => {
227227
await sql`select pgflow_tests.reset_db();`;
228228

229-
const worker = startWorker(sql, SlotRefillFlow, {
229+
const worker = await startWorker(sql, SlotRefillFlow, {
230230
maxConcurrent: 2,
231231
batchSize: 2,
232232
maxPollSeconds: 1,

‎pkgs/edge-worker/tests/integration/flow/minimalFlow.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ Deno.test(
3434
withPgNoTransaction(async (sql) => {
3535
await sql`select pgflow_tests.reset_db();`;
3636

37-
const worker = startWorker(sql, MinimalFlow, {
37+
const worker = await startWorker(sql, MinimalFlow, {
3838
maxConcurrent: 1,
3939
batchSize: 10,
4040
maxPollSeconds: 1,

‎pkgs/edge-worker/tests/integration/flow/performanceMapFlow.test.ts‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -253,7 +253,7 @@ interface WorkerConfig {
253253
[key: string]: unknown;
254254
}
255255

256-
const startWorkers = (
256+
const startWorkers = async (
257257
sql: postgres.Sql,
258258
flow: typeof LargeArrayMapFlow,
259259
numWorkers: number,
@@ -262,7 +262,7 @@ const startWorkers = (
262262
const workers = [];
263263

264264
for (let i = 0; i < numWorkers; i++) {
265-
const worker = startWorker(sql, flow, {
265+
const worker = await startWorker(sql, flow, {
266266
...config,
267267
// Add slight variation to worker configs to simulate real-world
268268
maxConcurrent: config.maxConcurrent + (i % 2) * 5,
@@ -342,7 +342,7 @@ Deno.test(
342342
const NUM_WORKERS = 4;
343343
const ELEMENTS_PER_FLOW = 100;
344344

345-
const workers = startWorkers(sql, LargeArrayMapFlow, NUM_WORKERS, {
345+
const workers = await startWorkers(sql, LargeArrayMapFlow, NUM_WORKERS, {
346346
maxConcurrent: 25,
347347
batchSize: 10,
348348
maxPollSeconds: 1,
@@ -498,7 +498,7 @@ Deno.test(
498498
console.log = () => {};
499499
console.debug = () => {};
500500

501-
const worker = startWorker(sql, LargeArrayMapFlow, {
501+
const worker = await startWorker(sql, LargeArrayMapFlow, {
502502
maxConcurrent: 50, // Higher concurrency for performance test
503503
batchSize: 25, // Larger batch size for better throughput
504504
maxPollSeconds: 1,
@@ -725,7 +725,7 @@ Deno.test(
725725
console.log = () => {};
726726
console.debug = () => {};
727727

728-
const worker = startWorker(sql, LargeArrayMapFlow, {
728+
const worker = await startWorker(sql, LargeArrayMapFlow, {
729729
maxConcurrent: 100, // Even higher concurrency for stress test
730730
batchSize: 50,
731731
maxPollSeconds: 2,

‎pkgs/edge-worker/tests/integration/flow/startDelayFlow.test.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,7 @@ Deno.test(
7272
withPgNoTransaction(async (sql) => {
7373
await sql`select pgflow_tests.reset_db();`;
7474

75-
const worker = startWorker(sql, RootStepDelayFlow, {
75+
const worker = await startWorker(sql, RootStepDelayFlow, {
7676
maxConcurrent: 1,
7777
batchSize: 10,
7878
maxPollSeconds: 1,
@@ -127,7 +127,7 @@ Deno.test(
127127
withPgNoTransaction(async (sql) => {
128128
await sql`select pgflow_tests.reset_db();`;
129129

130-
const worker = startWorker(sql, NormalStepDelayFlow, {
130+
const worker = await startWorker(sql, NormalStepDelayFlow, {
131131
maxConcurrent: 2,
132132
batchSize: 10,
133133
maxPollSeconds: 1,
@@ -187,7 +187,7 @@ Deno.test(
187187
withPgNoTransaction(async (sql) => {
188188
await sql`select pgflow_tests.reset_db();`;
189189

190-
const worker = startWorker(sql, CascadedDelayFlow, {
190+
const worker = await startWorker(sql, CascadedDelayFlow, {
191191
maxConcurrent: 3,
192192
batchSize: 10,
193193
maxPollSeconds: 1,
@@ -293,7 +293,7 @@ Deno.test(
293293
return `Success on attempt ${attemptCount}`;
294294
});
295295

296-
const worker = startWorker(sql, FlowWithRetry, {
296+
const worker = await startWorker(sql, FlowWithRetry, {
297297
maxConcurrent: 1,
298298
batchSize: 10,
299299
maxPollSeconds: 1,

‎pkgs/edge-worker/tests/integration/flow/whenExhausted.test.ts‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ Deno.test(
5353
}
5454
);
5555

56-
const worker = startWorker(sql, FailOnErrorFlow, workerConfig);
56+
const worker = await startWorker(sql, FailOnErrorFlow, workerConfig);
5757

5858
try {
5959
await createFlowInDb(sql, FailOnErrorFlow);
@@ -125,7 +125,7 @@ Deno.test(
125125
}
126126
);
127127

128-
const worker = startWorker(sql, SkipOnErrorFlow, workerConfig);
128+
const worker = await startWorker(sql, SkipOnErrorFlow, workerConfig);
129129

130130
try {
131131
await createFlowInDb(sql, SkipOnErrorFlow);
@@ -194,7 +194,7 @@ Deno.test(
194194
}
195195
);
196196

197-
const worker = startWorker(sql, SkipCascadeOnErrorFlow, workerConfig);
197+
const worker = await startWorker(sql, SkipCascadeOnErrorFlow, workerConfig);
198198

199199
try {
200200
await createFlowInDb(sql, SkipCascadeOnErrorFlow);
@@ -267,7 +267,7 @@ Deno.test(
267267
}
268268
);
269269

270-
const worker = startWorker(sql, RetryThenSkipFlow, workerConfig);
270+
const worker = await startWorker(sql, RetryThenSkipFlow, workerConfig);
271271

272272
try {
273273
await createFlowInDb(sql, RetryThenSkipFlow);
@@ -325,7 +325,7 @@ Deno.test(
325325
return { result: deps.maybe_skip ?? null };
326326
});
327327

328-
const worker = startWorker(sql, SuccessfulHandlerFlow, workerConfig);
328+
const worker = await startWorker(sql, SuccessfulHandlerFlow, workerConfig);
329329

330330
try {
331331
await createFlowInDb(sql, SuccessfulHandlerFlow);
@@ -396,7 +396,7 @@ Deno.test(
396396
}
397397
);
398398

399-
const worker = startWorker(sql, CombinedConditionsFlow, workerConfig);
399+
const worker = await startWorker(sql, CombinedConditionsFlow, workerConfig);
400400

401401
try {
402402
await createFlowInDb(sql, CombinedConditionsFlow);

‎pkgs/edge-worker/tests/integration/flow_deprecation.test.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ Deno.test(
4040
await sql`select pgflow.add_step(${flowSlug}::text, 'step2'::text, ARRAY['step1']::text[], null, null, null, null)`;
4141

4242
// Start the flow worker
43-
const worker = startWorker(sql, flow, {
43+
const worker = await startWorker(sql, flow, {
4444
maxPollSeconds: 1,
4545
pollIntervalMs: 50,
4646
});
@@ -160,12 +160,12 @@ Deno.test(
160160
await sql`select pgflow.add_step(${flowSlug}::text, 'step2'::text, ARRAY['step1']::text[], null, null, null, null)`;
161161

162162
// Start two workers for the same flow
163-
const worker1 = startWorker(sql, flow1, {
163+
const worker1 = await startWorker(sql, flow1, {
164164
maxPollSeconds: 1,
165165
pollIntervalMs: 50,
166166
});
167167

168-
const worker2 = startWorker(sql, flow2, {
168+
const worker2 = await startWorker(sql, flow2, {
169169
maxPollSeconds: 1,
170170
pollIntervalMs: 50,
171171
});

0 commit comments

Comments
 (0)