Skip to content

test: synchronize integration outcomes #26

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
psteinroe merged 1 commit into from
Sep 17, 2026
Merged

test: synchronize integration outcomes #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
106 changes: 59 additions & 47 deletions packages/pgconductor-js/tests/integration/cron-scheduling.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<Array<{ dedupe_key: string; run_at: Date }>>`
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<Array<{ count: number }>>`
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();
Expand Down Expand Up @@ -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<Array<{ dedupe_key: string }>>`
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<Array<{ dedupe_key: string }>>`
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();
Expand Down Expand Up @@ -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;
},
);

Expand All @@ -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<Array<{ count: number }>>`
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<Array<{ id: string }>>`
SELECT id
Expand Down Expand Up @@ -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<Array<{ run_at: Date; dedupe_key: string }>>`
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<Array<{ run_at: Date; dedupe_key: string }>>`
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();
Expand Down
17 changes: 10 additions & 7 deletions packages/pgconductor-js/tests/integration/field-selection-types.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -48,6 +49,7 @@ describe("Field Selection - Type Safety & Runtime", () => {
payload: z.object({}),
});

const handled = new Deferred<void>();
const taskFn = mock(async (event) => {
// Type-level: TypeScript should know these fields exist with correct types
const str: string = event.payload.stringField;
Expand All @@ -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({
Expand Down Expand Up @@ -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();
Expand All @@ -135,6 +136,7 @@ describe("Field Selection - Type Safety & Runtime", () => {
payload: z.object({}),
});

const handled = new Deferred<void>();
const taskFn = mock(async (event) => {
// Type-level: Selected fields with correct camelCase
const userId: string = event.payload.userId;
Expand All @@ -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({
Expand Down Expand Up @@ -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();
Expand All @@ -202,6 +204,7 @@ describe("Field Selection - Type Safety & Runtime", () => {
payload: z.object({}),
});

const handled = new Deferred<void>();
const taskFn = mock(async (event) => {
// Type-level: All fields should be available
const f1: string = event.payload.field1;
Expand All @@ -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({
Expand All @@ -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();
Expand Down
Loading
Loading