From f219f0d7fdd2e03790521ddbfa3911bce46e4a08 Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 15 Sep 2026 11:38:47 +0000 Subject: [PATCH] test: synchronize integration outcomes --- .../tests/integration/cron-scheduling.test.ts | 106 ++++++----- .../integration/field-selection-types.test.ts | 17 +- .../tests/integration/invoke-support.test.ts | 168 ++++++++++++------ .../tests/integration/step-support.test.ts | 40 ++--- packages/pgconductor-js/tests/test-utils.ts | 12 ++ 5 files changed, 210 insertions(+), 133 deletions(-) create mode 100644 packages/pgconductor-js/tests/test-utils.ts diff --git a/packages/pgconductor-js/tests/integration/cron-scheduling.test.ts b/packages/pgconductor-js/tests/integration/cron-scheduling.test.ts index 6699181..18e5a20 100644 --- a/packages/pgconductor-js/tests/integration/cron-scheduling.test.ts +++ b/packages/pgconductor-js/tests/integration/cron-scheduling.test.ts @@ -6,6 +6,7 @@ import { defineTask } from "../../src/task-definition"; import { TestDatabasePool } from "../fixtures/test-database"; import { waitFor } from "../../src/lib/wait-for"; import { TaskSchemas } from "../../src/schemas"; +import { waitForCondition } from "../test-utils"; describe("Cron Scheduling", () => { let pool: TestDatabasePool; @@ -101,26 +102,20 @@ describe("Cron Scheduling", () => { await orchestrator.start(); - // Wait for first execution - await waitFor(4000); - expect(executions.mock.calls.length).toBeGreaterThanOrEqual(1); + await waitForCondition(() => executions.mock.calls.length >= 1, 7000); - // Check that next execution is scheduled - const schedules = await db.sql>` - SELECT dedupe_key, run_at - FROM pgconductor._private_executions - WHERE task_key = 'frequent-sync' - AND dedupe_key LIKE 'scheduled::%' - AND run_at > pgconductor._private_current_time() - ORDER BY run_at - LIMIT 1 - `; + // A completed cron execution durably creates a distinct next occurrence. + await waitForCondition(async () => { + const schedules = await db.sql>` + SELECT count(*)::integer AS count + FROM pgconductor._private_executions + WHERE task_key = 'frequent-sync' + AND dedupe_key LIKE 'scheduled::%' + `; + return (schedules[0]?.count ?? 0) >= 2; + }, 7000); - expect(schedules.length).toBe(1); - - // Wait for second execution - await waitFor(4000); - expect(executions.mock.calls.length).toBeGreaterThanOrEqual(2); + await waitForCondition(() => executions.mock.calls.length >= 2, 7000); await orchestrator.stop(); await db.destroy(); @@ -361,21 +356,23 @@ describe("Cron Scheduling", () => { await orchestrator.start(); await conductor.invoke({ name: "dynamic-scheduler" }, {}); - await waitFor(4000); - expect(targetExecutions.mock.calls.length).toBeGreaterThanOrEqual(1); - - const nextSchedules = await db.sql>` - SELECT dedupe_key - FROM pgconductor._private_executions - WHERE task_key = 'dynamic-target' - AND cron_expression IS NOT NULL - AND run_at > pgconductor._private_current_time() - ORDER BY run_at - LIMIT 1 - `; - - expect(nextSchedules.length).toBe(1); - expect(nextSchedules[0]!.dedupe_key).toMatch(/^scheduled::reporting::\d+$/); + await waitForCondition(() => targetExecutions.mock.calls.length >= 1, 20_000); + + let nextSchedule: { dedupe_key: string } | undefined; + await waitForCondition(async () => { + [nextSchedule] = await db.sql>` + SELECT dedupe_key + FROM pgconductor._private_executions + WHERE task_key = 'dynamic-target' + AND cron_expression IS NOT NULL + AND run_at > pgconductor._private_current_time() + ORDER BY run_at + LIMIT 1 + `; + return nextSchedule !== undefined; + }, 20_000); + + expect(nextSchedule?.dedupe_key).toMatch(/^scheduled::reporting::\d+$/); await orchestrator.stop(); await db.destroy(); @@ -424,11 +421,13 @@ describe("Cron Scheduling", () => { }, ); + let unscheduled = false; const unschedulerTask = conductor.createTask( { name: "dynamic-unscheduler" }, { invocable: true }, async (_event, ctx) => { await ctx.unschedule({ name: "dynamic-target" }, "reporting"); + unscheduled = true; }, ); @@ -443,12 +442,22 @@ describe("Cron Scheduling", () => { await orchestrator.start(); await conductor.invoke({ name: "dynamic-scheduler" }, {}); - await waitFor(4000); - expect(targetExecutions.mock.calls.length).toBeGreaterThanOrEqual(1); + await waitForCondition(() => targetExecutions.mock.calls.length >= 1, 7000); + await waitForCondition(async () => { + const active = await db.sql>` + SELECT count(*)::integer AS count + FROM pgconductor._private_executions + WHERE task_key = 'dynamic-target' + AND locked_at IS NOT NULL + AND completed_at IS NULL + AND failed_at IS NULL + `; + return (active[0]?.count ?? 0) === 0; + }); const runsBeforeUnschedule = targetExecutions.mock.calls.length; await conductor.invoke({ name: "dynamic-unscheduler" }, {}); - await waitFor(2000); + await waitForCondition(() => unscheduled); const futureSchedules = await db.sql>` SELECT id @@ -558,18 +567,21 @@ describe("Cron Scheduling", () => { // Should have at least attempted twice (fail + success) expect(attemptCount).toBeGreaterThanOrEqual(2); - // Verify next cron execution is scheduled - const nextExecution = await db.sql>` - SELECT run_at, dedupe_key - FROM pgconductor._private_executions - WHERE task_key = 'flaky-cron' - AND dedupe_key LIKE 'scheduled::%' - AND run_at > pgconductor._private_current_time() - ORDER BY run_at - LIMIT 1 - `; + // Verify next cron execution is scheduled without sampling at a claim boundary. + let nextExecution: Array<{ run_at: Date; dedupe_key: string }> = []; + await waitForCondition(async () => { + nextExecution = await db.sql>` + SELECT run_at, dedupe_key + FROM pgconductor._private_executions + WHERE task_key = 'flaky-cron' + AND dedupe_key LIKE 'scheduled::%' + AND run_at > pgconductor._private_current_time() + ORDER BY run_at + LIMIT 1 + `; + return nextExecution.length === 1; + }); - expect(nextExecution.length).toBe(1); expect(nextExecution[0]?.dedupe_key).toMatch(/^scheduled::.*::\d+$/); await orchestrator.stop(); diff --git a/packages/pgconductor-js/tests/integration/field-selection-types.test.ts b/packages/pgconductor-js/tests/integration/field-selection-types.test.ts index 717e994..709bd42 100644 --- a/packages/pgconductor-js/tests/integration/field-selection-types.test.ts +++ b/packages/pgconductor-js/tests/integration/field-selection-types.test.ts @@ -7,6 +7,7 @@ import { defineEvent } from "../../src/event-definition"; import { TaskSchemas, EventSchemas } from "../../src/schemas"; import { TestDatabasePool } from "../fixtures/test-database"; import type { TestDatabase } from "../fixtures/test-database"; +import { Deferred } from "../../src/lib/deferred"; describe("Field Selection - Type Safety & Runtime", () => { let pool: TestDatabasePool; @@ -48,6 +49,7 @@ describe("Field Selection - Type Safety & Runtime", () => { payload: z.object({}), }); + const handled = new Deferred(); const taskFn = mock(async (event) => { // Type-level: TypeScript should know these fields exist with correct types const str: string = event.payload.stringField; @@ -70,6 +72,7 @@ describe("Field Selection - Type Safety & Runtime", () => { // Non-selected fields should not exist at runtime expect(event.payload.extraString).toBeUndefined(); expect(event.payload.extraNumber).toBeUndefined(); + handled.resolve(); }); const conductor = Conductor.create({ @@ -107,9 +110,7 @@ describe("Field Selection - Type Safety & Runtime", () => { extraNumber: 999, }); - // Wait for task to execute - await new Promise((r) => setTimeout(r, 300)); - + await handled.promise; expect(taskFn).toHaveBeenCalledTimes(1); await orchestrator.stop(); @@ -135,6 +136,7 @@ describe("Field Selection - Type Safety & Runtime", () => { payload: z.object({}), }); + const handled = new Deferred(); const taskFn = mock(async (event) => { // Type-level: Selected fields with correct camelCase const userId: string = event.payload.userId; @@ -146,6 +148,7 @@ describe("Field Selection - Type Safety & Runtime", () => { expect(event.payload.firstName).toBeUndefined(); expect(event.payload.lastName).toBeUndefined(); expect(event.payload.accountType).toBeUndefined(); + handled.resolve(); }); const conductor = Conductor.create({ @@ -177,8 +180,7 @@ describe("Field Selection - Type Safety & Runtime", () => { accountType: "premium", }); - await new Promise((r) => setTimeout(r, 300)); - + await handled.promise; expect(taskFn).toHaveBeenCalledTimes(1); await orchestrator.stop(); @@ -202,6 +204,7 @@ describe("Field Selection - Type Safety & Runtime", () => { payload: z.object({}), }); + const handled = new Deferred(); const taskFn = mock(async (event) => { // Type-level: All fields should be available const f1: string = event.payload.field1; @@ -212,6 +215,7 @@ describe("Field Selection - Type Safety & Runtime", () => { expect(event.payload.field1).toBe("value1"); expect(event.payload.field2).toBe(123); expect(event.payload.field3).toBe(true); + handled.resolve(); }); const conductor = Conductor.create({ @@ -238,8 +242,7 @@ describe("Field Selection - Type Safety & Runtime", () => { field3: true, }); - await new Promise((r) => setTimeout(r, 300)); - + await handled.promise; expect(taskFn).toHaveBeenCalledTimes(1); await orchestrator.stop(); diff --git a/packages/pgconductor-js/tests/integration/invoke-support.test.ts b/packages/pgconductor-js/tests/integration/invoke-support.test.ts index 47d9b3f..c25f0ea 100644 --- a/packages/pgconductor-js/tests/integration/invoke-support.test.ts +++ b/packages/pgconductor-js/tests/integration/invoke-support.test.ts @@ -6,6 +6,8 @@ import { defineTask } from "../../src/task-definition"; import { TestDatabasePool } from "../fixtures/test-database"; import type { TestDatabase } from "../fixtures/test-database"; import { TaskSchemas } from "../../src/schemas"; +import { Deferred } from "../../src/lib/deferred"; +import { waitForCondition } from "../test-utils"; describe("Invoke Support", () => { let pool: TestDatabasePool; @@ -41,6 +43,7 @@ describe("Invoke Support", () => { }); const childFn = mock((n: number) => n * 2); + const childCompleted = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -54,6 +57,7 @@ describe("Invoke Support", () => { async (event, _ctx) => { if (event.name === "pgconductor.invoke") { const result = childFn(event.payload.input); + childCompleted.resolve(); return { output: result }; } throw new Error("Unexpected event type"); @@ -85,9 +89,7 @@ describe("Invoke Support", () => { await orchestrator.start(); await conductor.invoke({ name: "parent-task" }, { value: 5 }); - - await new Promise((r) => setTimeout(r, 300)); - + await childCompleted.promise; await orchestrator.stop(); expect(childFn).toHaveBeenCalledTimes(1); @@ -156,15 +158,24 @@ describe("Invoke Support", () => { await orchestrator.start(); await conductor.invoke({ name: "timeout-parent" }, {}); - - await new Promise((r) => setTimeout(r, 300)); + await waitForCondition(async () => { + const [child] = await db.sql<{ released: boolean; sleeping: boolean }[]>` + select + locked_by is null as released, + exists ( + select 1 from pgconductor._private_steps s + where s.execution_id = e.id and s.key = 'long-sleep' + ) as sleeping + from pgconductor._private_executions e + where task_key = 'slow-child' + `; + return child?.released === true && child.sleeping; + }); // Advance time past timeout (1 second) but before sleep completes (5 seconds) const afterTimeout = new Date(startTime.getTime() + 1500); await db.client.setFakeTime({ date: afterTimeout }); - - await new Promise((r) => setTimeout(r, 300)); - + await waitForCondition(async () => parentError.mock.calls.length === 1); await orchestrator.stop(); await db.client.clearFakeTime(); @@ -172,7 +183,7 @@ describe("Invoke Support", () => { expect(parentError).toHaveBeenCalledTimes(1); const errorMsg = parentError.mock.results[0]?.value; expect(errorMsg).toContain("timed out after 1000ms"); - }, 5000); + }, 30000); test("invoke() caches child result on retry", async () => { const db = await pool.child(); @@ -191,6 +202,7 @@ describe("Invoke Support", () => { }); const childFn = mock((n: number) => n * 3); + const parentCompleted = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -228,6 +240,7 @@ describe("Invoke Support", () => { throw new Error("First attempt fails"); } + parentCompleted.resolve(); return { result: childResult.output }; } throw new Error("Unexpected event type"); @@ -243,9 +256,7 @@ describe("Invoke Support", () => { await orchestrator.start(); await conductor.invoke({ name: "retry-parent" }, { value: 7 }); - - await new Promise((r) => setTimeout(r, 500)); - + await parentCompleted.promise; await orchestrator.stop(); expect(childFn).toHaveBeenCalledTimes(1); @@ -269,6 +280,7 @@ describe("Invoke Support", () => { }); const childFn = mock(() => "completed"); + const childCompleted = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -282,6 +294,7 @@ describe("Invoke Support", () => { async (_event, ctx) => { await ctx.sleep("moderate-sleep", 2000); const result = childFn(); + childCompleted.resolve(); return { result }; }, ); @@ -304,9 +317,7 @@ describe("Invoke Support", () => { await orchestrator.start(); await conductor.invoke({ name: "patient-parent" }, {}); - - await new Promise((r) => setTimeout(r, 2500)); - + await childCompleted.promise; await orchestrator.stop(); expect(childFn).toHaveBeenCalledTimes(1); @@ -372,17 +383,31 @@ describe("Invoke Support", () => { await conductor.invoke({ name: "cascade-parent" }, {}); - // Wait for first execution cycle: parent runs → invokes child → child fails - await new Promise((r) => setTimeout(r, 500)); + await waitForCondition(async () => { + const [child] = await db.sql<{ attempts: number; released: boolean }[]>` + select attempts, locked_by is null as released + from pgconductor._private_executions + where task_key = 'failing-child' + `; + return child?.attempts === 1 && child.released; + }); expect(childFn).toHaveBeenCalledTimes(1); - // Advance fake time past first backoff (15 seconds) - const afterFirstBackoff = new Date(startTime.getTime() + 15000); - await db.client.setFakeTime({ date: afterFirstBackoff }); - - // Wait for second execution cycle: child retries and fails permanently - await new Promise((r) => setTimeout(r, 500)); + const [retry] = await db.sql<{ run_at: Date }[]>` + select run_at from pgconductor._private_executions + where task_key = 'failing-child' + `; + if (!retry) throw new Error("expected persisted child retry"); + await db.client.setFakeTime({ date: new Date(retry.run_at.getTime() + 1) }); + await waitForCondition(async () => { + const [parent] = await db.sql<{ failed: boolean }[]>` + select failed_at is not null as failed + from pgconductor._private_executions + where task_key = 'cascade-parent' + `; + return parent?.failed === true; + }); await orchestrator.stop(); await db.client.clearFakeTime(); @@ -418,6 +443,7 @@ describe("Invoke Support", () => { }); const childFn = mock((n: number) => n * 2); + const childCalled = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -452,6 +478,7 @@ describe("Invoke Support", () => { async (event, _ctx) => { if (event.name === "pgconductor.invoke") { const result = childFn(event.payload.input); + childCalled.resolve(); return { output: result }; } throw new Error("Unexpected event type"); @@ -472,9 +499,7 @@ describe("Invoke Support", () => { await orchestrator.start(); await conductor.invoke({ queue: "parent-queue", name: "parent-task" }, { value: 5 }); - - await new Promise((r) => setTimeout(r, 4000)); - + await childCalled.promise; await orchestrator.stop(); expect(childFn).toHaveBeenCalledTimes(1); @@ -493,6 +518,7 @@ describe("Invoke Support", () => { const childDefinition = defineTask({ name: "slow-child-2", + queue: "pending-child-queue", payload: z.object({}), returns: z.object({ completed: z.boolean() }), }); @@ -503,32 +529,34 @@ describe("Invoke Support", () => { context: {}, }); - // Child that takes 5 seconds - const slowChildTask = conductor.createTask( - { name: "slow-child-2" }, - { invocable: true }, - async (_event, ctx) => { - await ctx.sleep("long-sleep", 5000); - return { completed: true }; - }, - ); - // Parent with 1 second timeout - let error throw const timeoutParentTask = conductor.createTask( { name: "timeout-parent-2" }, { invocable: true }, async (_event, ctx) => { - await ctx.invoke("invoke-slow", { name: "slow-child-2" }, {}, 1000); + await ctx.invoke( + "invoke-slow", + { name: "slow-child-2", queue: "pending-child-queue" }, + {}, + 1000, + ); return { success: true }; }, ); + await conductor.ensureInstalled(); + await db.sql` + insert into pgconductor._private_queues (name) + values ('pending-child-queue') + on conflict (name) do nothing + `; + const startTime = new Date("2024-01-01T00:00:00Z"); await db.client.setFakeTime({ date: startTime }); const orchestrator = Orchestrator.create({ conductor, - tasks: [timeoutParentTask, slowChildTask], + tasks: [timeoutParentTask], defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50, @@ -539,16 +567,29 @@ describe("Invoke Support", () => { await conductor.invoke({ name: "timeout-parent-2" }, {}); - // Wait for parent to invoke child - await new Promise((r) => setTimeout(r, 200)); + // Wait for the parent to persist the unclaimed child execution. + await waitForCondition(async () => { + const [child] = await db.sql<{ pending: boolean }[]>` + select locked_by is null as pending + from pgconductor._private_executions + where task_key = 'slow-child-2' + and queue = 'pending-child-queue' + `; + return child?.pending === true; + }); - // Advance time past timeout (child still pending due to 50ms intervals) + // Advance time past timeout while the child is still pending. const afterTimeout = new Date(startTime.getTime() + 1500); await db.client.setFakeTime({ date: afterTimeout }); - - // Wait for timeout to be processed - await new Promise((r) => setTimeout(r, 500)); - + await waitForCondition(async () => { + const [child] = await db.sql<{ failed: boolean }[]>` + select failed_at is not null as failed + from pgconductor._private_executions + where task_key = 'slow-child-2' + and queue = 'pending-child-queue' + `; + return child?.failed === true; + }); await orchestrator.stop(); await db.client.clearFakeTime(); @@ -563,6 +604,7 @@ describe("Invoke Support", () => { select cancelled, failed_at, last_error from pgconductor._private_executions where task_key = 'slow-child-2' + and queue = 'pending-child-queue' `; expect(children.length).toBe(1); @@ -668,8 +710,14 @@ describe("Invoke Support", () => { { dedupe_key: "locked-key" }, ); - // Wait for it to be locked - await new Promise((r) => setTimeout(r, 100)); + await waitForCondition(async () => { + const [execution] = await db.sql<{ locked: boolean }[]>` + select locked_by is not null as locked + from pgconductor._private_executions + where id = ${id1} + `; + return execution?.locked === true; + }); // Second invocation while first is locked - should create NEW execution const id2 = await conductor.invoke( @@ -681,9 +729,14 @@ describe("Invoke Support", () => { // Should be different IDs expect(id2).not.toBe(id1); - // Wait for both executions to complete and flush (100ms + 500ms + 500ms + 50ms + margin) - await new Promise((r) => setTimeout(r, 1200)); - + await waitForCondition(async () => { + const [execution] = await db.sql<{ completed: boolean }[]>` + select completed_at is not null as completed + from pgconductor._private_executions + where id = ${id2} + `; + return execution?.completed === true; + }); await orchestrator.stop(); // First execution should be marked as failed (superseded) @@ -755,6 +808,9 @@ describe("Invoke Support", () => { }, ); + const startTime = new Date("2024-01-01T00:00:00Z"); + await db.client.setFakeTime({ date: startTime }); + const orchestrator = Orchestrator.create({ defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50 }, conductor, @@ -764,7 +820,7 @@ describe("Invoke Support", () => { await orchestrator.start(); // Rapid invocations with same dedupe_key and delayed run_at - const futureTime = new Date(Date.now() + 1000); + const futureTime = new Date(startTime.getTime() + 1000); const id1 = await conductor.invoke( { name: "debounce-task" }, @@ -772,16 +828,12 @@ describe("Invoke Support", () => { { dedupe_key: "debounce-1", run_at: futureTime }, ); - await new Promise((r) => setTimeout(r, 50)); - const id2 = await conductor.invoke( { name: "debounce-task" }, { value: 2 }, { dedupe_key: "debounce-1", run_at: futureTime }, ); - await new Promise((r) => setTimeout(r, 50)); - const id3 = await conductor.invoke( { name: "debounce-task" }, { value: 3 }, @@ -800,10 +852,10 @@ describe("Invoke Support", () => { `; expect(executions[0]?.count).toBe(1); - // Wait for execution - await new Promise((r) => setTimeout(r, 1500)); - + await db.client.setFakeTime({ date: futureTime }); + await waitForCondition(async () => executionCount === 1); await orchestrator.stop(); + await db.client.clearFakeTime(); // Should have executed only once with the last value expect(executionCount).toBe(1); diff --git a/packages/pgconductor-js/tests/integration/step-support.test.ts b/packages/pgconductor-js/tests/integration/step-support.test.ts index 32e1638..b43fcb6 100644 --- a/packages/pgconductor-js/tests/integration/step-support.test.ts +++ b/packages/pgconductor-js/tests/integration/step-support.test.ts @@ -6,6 +6,7 @@ import { defineTask } from "../../src/task-definition"; import { TestDatabasePool } from "../fixtures/test-database"; import type { TestDatabase } from "../fixtures/test-database"; import { TaskSchemas } from "../../src/schemas"; +import { Deferred } from "../../src/lib/deferred"; describe("Step Support", () => { let pool: TestDatabasePool; @@ -35,6 +36,7 @@ describe("Step Support", () => { }); const expensiveFn = mock((n: number) => n * 2); + const taskCompleted = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -51,6 +53,7 @@ describe("Step Support", () => { return expensiveFn(event.payload.value); }); + taskCompleted.resolve(); return { result }; } throw new Error("Unexpected event type"); @@ -66,9 +69,7 @@ describe("Step Support", () => { await orchestrator.start(); await conductor.invoke({ name: "step-task" }, { value: 5 }); - - await new Promise((r) => setTimeout(r, 200)); - + await taskCompleted.promise; await orchestrator.stop(); // Expensive function should only be called once @@ -87,6 +88,8 @@ describe("Step Support", () => { }); const executionSteps = mock((step: string) => step); + const sleepStarted = new Deferred(); + const sleepFinished = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -100,8 +103,10 @@ describe("Step Support", () => { async (event, ctx) => { if (event.name === "pgconductor.invoke") { executionSteps("before-sleep"); + sleepStarted.resolve(); await ctx.sleep("wait", event.payload.delay); executionSteps("after-sleep"); + sleepFinished.resolve(); return { completed: true }; } throw new Error("Unexpected event type"); @@ -117,18 +122,12 @@ describe("Step Support", () => { await orchestrator.start(); await conductor.invoke({ name: "sleep-task" }, { delay: 2000 }); + await sleepStarted.promise; - // Wait for initial execution (should hit sleep and release) - await new Promise((r) => setTimeout(r, 200)); - - // At this point, should have seen "before-sleep" but not "after-sleep" expect(executionSteps).toHaveBeenCalledWith("before-sleep"); expect(executionSteps).not.toHaveBeenCalledWith("after-sleep"); - // Wait for sleep to complete and task to resume (2s sleep + buffer) - await new Promise((r) => setTimeout(r, 2200)); - - // Now should see "after-sleep" + await sleepFinished.promise; expect(executionSteps).toHaveBeenCalledWith("after-sleep"); await orchestrator.stop(); @@ -145,6 +144,7 @@ describe("Step Support", () => { }); const processedItems = mock((item: number) => item); + const checkpointCompleted = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -163,6 +163,7 @@ describe("Step Support", () => { // Simulate some work await new Promise((r) => setTimeout(r, 50)); } + checkpointCompleted.resolve(); return { processed: event.payload.items }; } throw new Error("Unexpected event type"); @@ -178,10 +179,7 @@ describe("Step Support", () => { await orchestrator.start(); await conductor.invoke({ name: "checkpoint-task" }, { items: 5 }); - - // Wait for task to complete - await new Promise((r) => setTimeout(r, 300)); - + await checkpointCompleted.promise; await orchestrator.stop(); // Should have processed all items @@ -201,6 +199,7 @@ describe("Step Support", () => { const step1Fn = mock((n: number) => n + 1); const step2Fn = mock((n: number) => n * 2); const step3Fn = mock((n: number) => n - 3); + const stepsCompleted = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -216,6 +215,7 @@ describe("Step Support", () => { const a = await ctx.step("step1", () => step1Fn(event.payload.x)); const b = await ctx.step("step2", () => step2Fn(a)); const c = await ctx.step("step3", () => step3Fn(b)); + stepsCompleted.resolve(); return { result: c }; } throw new Error("Unexpected event type"); @@ -231,9 +231,7 @@ describe("Step Support", () => { await orchestrator.start(); await conductor.invoke({ name: "multi-step-task" }, { x: 5 }); - - await new Promise((r) => setTimeout(r, 200)); - + await stepsCompleted.promise; await orchestrator.stop(); // All steps should execute once @@ -259,6 +257,7 @@ describe("Step Support", () => { let transformedValue: string[] | undefined; let countValue: number | undefined; + const completed = new Deferred(); const conductor = Conductor.create({ sql: db.sql, @@ -284,6 +283,7 @@ describe("Step Support", () => { return transformed.length; }); + completed.resolve(); return { count }; } throw new Error("Unexpected event type"); @@ -299,9 +299,7 @@ describe("Step Support", () => { await orchestrator.start(); await conductor.invoke({ name: "step-unwrap-task" }, { items: ["apple", "banana", "cherry"] }); - - await new Promise((r) => setTimeout(r, 300)); - + await completed.promise; await orchestrator.stop(); expect(transformedValue).toEqual(["APPLE", "BANANA", "CHERRY"]); diff --git a/packages/pgconductor-js/tests/test-utils.ts b/packages/pgconductor-js/tests/test-utils.ts new file mode 100644 index 0000000..a66b926 --- /dev/null +++ b/packages/pgconductor-js/tests/test-utils.ts @@ -0,0 +1,12 @@ +export async function waitForCondition( + condition: () => boolean | Promise, + timeoutMs = 20_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!(await condition())) { + if (Date.now() >= deadline) { + throw new Error(`condition was not met within ${timeoutMs}ms`); + } + await Bun.sleep(25); + } +}