diff --git a/AGENTS.md b/AGENTS.md index da555bb..bd35c25 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -353,7 +353,7 @@ const rows = await db.sql`select * from pgconductor._private_executions`; ### Controlling Time in Tests -`pgconductor._private_current_time()` returns `current_setting('pgconductor.fake_now')` when set. Test databases use a single connection (`max: 1`), so a session-level setting is seen by the test, conductor and workers alike: +`pgconductor._private_current_time()` returns `clock_timestamp()` shifted by the `pgconductor.fake_offset` setting when set, so fake time keeps running from the time it was set to and rows written one after another get later times, as in production (event waits match only events created after their subscription). Test databases use a single connection (`max: 1`), so a session-level setting is seen by the test, conductor and workers alike: ```typescript await db.clock.set(new Date("2024-01-01T12:00:00Z")); @@ -361,9 +361,9 @@ await db.clock.advance("1h"); await db.clock.reset(); ``` -Useful for sleeps, timeouts, backoff schedules and time windows. Each test database is dropped afterwards, so fake time needs no cleanup. `db.client.setFakeTime()` sets the same setting but does not move the workers' clock. +Useful for sleeps, timeouts, backoff schedules and time windows. Fake time runs on, so assert timestamps with bounds, not exact values. Each test database is dropped afterwards, so fake time needs no cleanup. `db.client.setFakeTime()` sets the same setting but does not move the workers' clock. -Cron scheduling and time windows (worker fetch filtering, `ctx.step()` and `ctx.checkpoint()`) use the worker's `Clock` (`src/lib/clock.ts`): local time corrected by an offset sampled from the database when the worker starts and every 10 minutes. `Conductor.create({ clock })` replaces the local time. `db.clock` (`createTestClock(sql)` from `src/testing.ts`) sets `fake_now` and returns the same time from `now()`, so the offset stays zero; pass `clock: db.clock` to the conductor and running workers follow `db.clock.set()` and `advance()`. Use `vi.setSystemTime()` only for code that reads `Date` directly. Without `vi.useFakeTimers()` it only fakes `Date` and timers keep running; `vi.useRealTimers()` restores it. `vi.useFakeTimers()` fakes timers and `Date` but not `setImmediate`, which postgres.js needs to send queries (`fakeTimers` in `vitest.config.ts`). +Cron scheduling and time windows (worker fetch filtering, `ctx.step()` and `ctx.checkpoint()`) use the worker's `Clock` (`src/lib/clock.ts`): local time corrected by an offset sampled from the database when the worker starts and every 10 minutes. `Conductor.create({ clock })` replaces the local time. `db.clock` (`createTestClock(sql)` from `src/testing.ts`) sets `fake_offset` and applies the same offset in `now()`, so the sampled offset stays zero; pass `clock: db.clock` to the conductor and running workers follow `db.clock.set()` and `advance()`. Use `vi.setSystemTime()` only for code that reads `Date` directly. Without `vi.useFakeTimers()` it only fakes `Date` and timers keep running; `vi.useRealTimers()` restores it. `vi.useFakeTimers()` fakes timers and `Date` but not `setImmediate`, which postgres.js needs to send queries (`fakeTimers` in `vitest.config.ts`). ## Development Environment diff --git a/docs/content/api/testing.md b/docs/content/api/testing.md index 45768be..a2756d3 100644 --- a/docs/content/api/testing.md +++ b/docs/content/api/testing.md @@ -26,7 +26,7 @@ The faked time, or the system time when time is not faked. ## clock.set(date) -Fake the time. It stands still until it is moved again. +Fake the time. It keeps running from `date`, like real time. ```typescript await clock.set(new Date("2024-01-01T12:00:00Z")); diff --git a/docs/content/crafting-tasks/testing.md b/docs/content/crafting-tasks/testing.md index 0b9f08e..a3cf274 100644 --- a/docs/content/crafting-tasks/testing.md +++ b/docs/content/crafting-tasks/testing.md @@ -66,12 +66,12 @@ test("sends the reminder an hour later", async () => { The clock moves two things together: database time, which decides when sleeps end and retries are due, and the time workers use for cron schedules and time windows. Pass it to `Conductor.create({ clock })` so workers use it: ```typescript -await clock.set(new Date("2024-01-01T12:00:00Z")); // Freeze time +await clock.set(new Date("2024-01-01T12:00:00Z")); // 12:00:00 await clock.advance("15m"); // 12:15:00 await clock.reset(); // Back to real time ``` -Fake time stands still until you move it. Without `set()`, `advance()` starts from the current real time. +Fake time keeps running from where you set it, like real time, so things that happen one after another get later times. Assert timestamps with bounds, such as `toBeGreaterThanOrEqual()`, rather than exact values. Without `set()`, `advance()` starts from the current real time. **Cron:** a worker schedules the next occurrence when it starts and after each run. @@ -99,7 +99,7 @@ await orchestrator.drain(); // Attempt 2 ## How Time Works -Postgres Conductor reads time from `pgconductor._private_current_time()`, which returns the `pgconductor.fake_now` session setting when it is set. Because the setting belongs to one connection, the `sql` you pass to `createTestClock()` must have exactly one connection (`max: 1`), and everything that should see fake time must use it: the conductor, the orchestrator and your assertions. +Postgres Conductor reads time from `pgconductor._private_current_time()`, which adds the `pgconductor.fake_offset` session setting to the real time when it is set. Because the setting belongs to one connection, the `sql` you pass to `createTestClock()` must have exactly one connection (`max: 1`), and everything that should see fake time must use it: the conductor, the orchestrator and your assertions. A single connection runs one query at a time. A transaction on `sql` holds the connection until it ends, so do not wait inside it for a query that runs on `sql` outside the transaction. A second connection to the same database sees real time. diff --git a/migrations/0000000001_setup.sql b/migrations/0000000001_setup.sql index 021f804..0f8afd4 100644 --- a/migrations/0000000001_setup.sql +++ b/migrations/0000000001_setup.sql @@ -1,19 +1,19 @@ create extension if not exists btree_gist; --- Returns either the actual current timestamp or a fake one for tests. --- Uses session variable (current_setting) for test time control. +-- Returns the current timestamp, shifted by the pgconductor.fake_offset session setting +-- in tests. Fake time keeps running, so rows written one after another get later times. create function pgconductor._private_current_time () returns timestamptz language plpgsql volatile as $$ declare - v_fake text; + v_offset text; begin - v_fake := current_setting('pgconductor.fake_now', true); - if v_fake is not null and length(trim(v_fake)) > 0 then - return v_fake::timestamptz; + v_offset := current_setting('pgconductor.fake_offset', true); + if v_offset is not null and length(trim(v_offset)) > 0 then + return clock_timestamp() + v_offset::interval; end if; return clock_timestamp(); diff --git a/packages/pgconductor-js/src/database-client.ts b/packages/pgconductor-js/src/database-client.ts index 4974d93..5bde58d 100644 --- a/packages/pgconductor-js/src/database-client.ts +++ b/packages/pgconductor-js/src/database-client.ts @@ -826,22 +826,28 @@ export class DatabaseClient { /** * Set fake time for testing purposes. - * All calls to pgconductor._private_current_time() will return this value. + * pgconductor._private_current_time() returns this value and keeps running from it. * * IMPORTANT: Test database connection pool must have max: 1 for this to work. */ async setFakeTime({ date }: SetFakeTimeArgs, opts?: QueryMethodOptions): Promise { - await this.query((sql) => sql.unsafe(`set pgconductor.fake_now = '${date.toISOString()}'`), { - label: "setFakeTime", - ...opts, - }); + await this.query( + (sql) => sql` + select set_config( + 'pgconductor.fake_offset', + (${date.toISOString()}::timestamptz - clock_timestamp())::text, + false + ) + `, + { label: "setFakeTime", ...opts }, + ); } /** * Clear fake time, returning to real clock_timestamp(). */ async clearFakeTime(opts?: QueryMethodOptions): Promise { - await this.query((sql) => sql.unsafe(`set pgconductor.fake_now = ''`), { + await this.query((sql) => sql.unsafe(`set pgconductor.fake_offset = ''`), { label: "clearFakeTime", ...opts, }); diff --git a/packages/pgconductor-js/src/generated/sql.ts b/packages/pgconductor-js/src/generated/sql.ts index 9625bcb..c2ff685 100644 --- a/packages/pgconductor-js/src/generated/sql.ts +++ b/packages/pgconductor-js/src/generated/sql.ts @@ -15,19 +15,19 @@ create table if not exists pgconductor.schema_migrations ( create extension if not exists btree_gist; --- Returns either the actual current timestamp or a fake one for tests. --- Uses session variable (current_setting) for test time control. +-- Returns the current timestamp, shifted by the pgconductor.fake_offset session setting +-- in tests. Fake time keeps running, so rows written one after another get later times. create function pgconductor._private_current_time () returns timestamptz language plpgsql volatile as $$ declare - v_fake text; + v_offset text; begin - v_fake := current_setting('pgconductor.fake_now', true); - if v_fake is not null and length(trim(v_fake)) > 0 then - return v_fake::timestamptz; + v_offset := current_setting('pgconductor.fake_offset', true); + if v_offset is not null and length(trim(v_offset)) > 0 then + return clock_timestamp() + v_offset::interval; end if; return clock_timestamp(); diff --git a/packages/pgconductor-js/src/testing.ts b/packages/pgconductor-js/src/testing.ts index f13ffa8..2e82592 100644 --- a/packages/pgconductor-js/src/testing.ts +++ b/packages/pgconductor-js/src/testing.ts @@ -1,10 +1,10 @@ -import type { Sql } from "postgres"; +import type { PendingQuery, Row, Sql } from "postgres"; import { parseDuration, type DurationInput } from "./lib/duration.js"; export type TestClock = { /** Current time: the faked time, or the system time */ now(): Date; - /** Fake database time */ + /** Fake database time, which keeps running from `date` */ set(date: Date): Promise; /** Move time forward, starting from the system time if it is not faked */ advance(duration: DurationInput): Promise; @@ -17,19 +17,29 @@ export type TestClock = { * Fake time is a session setting, so `sql` must have a single connection (`max: 1`). */ export function createTestClock(sql: Sql): TestClock { - let fakeNow: Date | null = null; + let offsetMs = 0; - const setFakeNow = async (date: Date | null) => { - const copy = date && new Date(date); - await sql`select set_config('pgconductor.fake_now', ${copy?.toISOString() || ""}, false)`; - fakeNow = copy; + // the database clock keeps the offset, so database and worker time run on together + const setOffset = async (offset: PendingQuery) => { + const [row] = await sql<{ offset_ms: number }[]>` + with o as materialized (select ${offset} as value) + select + set_config('pgconductor.fake_offset', coalesce(o.value::text, ''), false), + coalesce(extract(epoch from o.value) * 1000, 0)::float8 as offset_ms + from o + `; + // Date has millisecond precision; rounding up keeps now() from reading before the time set + offsetMs = Math.ceil(row?.offset_ms || 0); }; return { - now: () => new Date(fakeNow || Date.now()), - set: (date) => setFakeNow(date), + now: () => new Date(Date.now() + offsetMs), + set: (date) => setOffset(sql`${date.toISOString()}::timestamptz - clock_timestamp()`), advance: (duration) => - setFakeNow(new Date((fakeNow || new Date()).getTime() + parseDuration(duration))), - reset: () => setFakeNow(null), + setOffset(sql` + coalesce(nullif(current_setting('pgconductor.fake_offset', true), '')::interval, interval '0') + + ${parseDuration(duration)}::float8 * interval '1 millisecond' + `), + reset: () => setOffset(sql`null::interval`), }; } diff --git a/packages/pgconductor-js/tests/integration/batch-processing.test.ts b/packages/pgconductor-js/tests/integration/batch-processing.test.ts index f8aa295..7eba27f 100644 --- a/packages/pgconductor-js/tests/integration/batch-processing.test.ts +++ b/packages/pgconductor-js/tests/integration/batch-processing.test.ts @@ -243,10 +243,10 @@ describe("Batch Processing", () => { await drain(); await conductor.invoke({ name: "batch-sleep-mixed" }, { value: 2 }); - await db.client.setFakeTime({ date: new Date("2024-01-01T12:00:01Z") }); + await db.client.setFakeTime({ date: new Date("2024-01-01T12:00:02Z") }); await drain(); - await db.client.setFakeTime({ date: new Date("2024-01-01T12:00:02Z") }); + await db.client.setFakeTime({ date: new Date("2024-01-01T12:00:04Z") }); await drain(); await db.client.clearFakeTime(); diff --git a/packages/pgconductor-js/tests/integration/dead-letter.test.ts b/packages/pgconductor-js/tests/integration/dead-letter.test.ts index b9b4ae8..a32682e 100644 --- a/packages/pgconductor-js/tests/integration/dead-letter.test.ts +++ b/packages/pgconductor-js/tests/integration/dead-letter.test.ts @@ -134,9 +134,10 @@ describe("dead-letter queues (Postgres integration)", () => { sourceTask: "charge", attempts: 2, error: "card declined", - failedAt, + failedAt: expect.any(Date), }, ]); + expect(seen[0]?.failedAt.getTime()).toBeGreaterThanOrEqual(failedAt.getTime()); }, 60_000); test("ignores duplicate settlements and wrong worker identity", async () => { diff --git a/packages/pgconductor-js/tests/integration/orchestrator-cleanup.test.ts b/packages/pgconductor-js/tests/integration/orchestrator-cleanup.test.ts index 07c63b0..71e44e3 100644 --- a/packages/pgconductor-js/tests/integration/orchestrator-cleanup.test.ts +++ b/packages/pgconductor-js/tests/integration/orchestrator-cleanup.test.ts @@ -435,7 +435,9 @@ test.each(["stop()", "a shutdown signal"])( const [row] = await db.sql<{ last_heartbeat_at: Date }[]>` select last_heartbeat_at from pgconductor._private_orchestrators where id = ${orch.info.id}::uuid `; - expect(row?.last_heartbeat_at).toEqual(new Date("2024-01-01T12:10:00Z")); + expect(row?.last_heartbeat_at.getTime()).toBeGreaterThanOrEqual( + new Date("2024-01-01T12:10:00Z").getTime(), + ); await db.client.recoverStaleOrchestrators({ maxAge: "5 minutes" }); const locked = await db.sql` diff --git a/packages/pgconductor-js/tests/integration/testing.test.ts b/packages/pgconductor-js/tests/integration/testing.test.ts index 9a9190a..542c398 100644 --- a/packages/pgconductor-js/tests/integration/testing.test.ts +++ b/packages/pgconductor-js/tests/integration/testing.test.ts @@ -94,10 +94,11 @@ test("resumes ctx.sleep when time advances", async () => { await clock.advance("1h"); await orchestrator.drain(); expect(sent).toEqual(["ada@example.com"]); - expect(await conductor.getExecution(id)).toMatchObject({ - status: "completed", - completedAt: new Date("2024-01-01T13:00:00Z"), - }); + const execution = await conductor.getExecution(id); + expect(execution?.status).toBe("completed"); + expect(execution?.completedAt?.getTime()).toBeGreaterThanOrEqual( + new Date("2024-01-01T13:00:00Z").getTime(), + ); }); test("retries after the backoff", async () => { @@ -165,9 +166,21 @@ test("set keeps its own copy of the date", async () => { await setting; date.setUTCFullYear(2031); - const [row] = await sql<{ now: Date }[]>` - select current_setting('pgconductor.fake_now')::timestamptz as now - `; - expect(row?.now).toEqual(new Date("2024-06-01T00:00:00Z")); - expect(clock.now()).toEqual(new Date("2024-06-01T00:00:00Z")); + const [row] = await sql<{ now: Date }[]>`select pgconductor._private_current_time() as now`; + const start = new Date("2024-06-01T00:00:00Z").getTime(); + expect(row?.now.getTime()).toBeGreaterThanOrEqual(start); + expect(row?.now.getTime()).toBeLessThan(start + 60_000); + expect(clock.now().getTime()).toBeGreaterThanOrEqual(start); + expect(clock.now().getTime()).toBeLessThan(start + 60_000); +}); + +test("fake time keeps running from the time it was set to", async () => { + const now = () => sql<{ now: Date }[]>`select pgconductor._private_current_time() as now`; + const [first] = await now(); + await setTimeout(10); + const [second] = await now(); + const start = new Date("2024-01-01T12:00:00Z").getTime(); + expect(first?.now.getTime()).toBeGreaterThanOrEqual(start); + expect(second?.now.getTime()).toBeGreaterThan(first?.now.getTime() || 0); + expect(second?.now.getTime()).toBeLessThan(start + 60_000); }); diff --git a/packages/pgconductor-js/tests/integration/wait-deadlines.test.ts b/packages/pgconductor-js/tests/integration/wait-deadlines.test.ts index a39d002..206fc6f 100644 --- a/packages/pgconductor-js/tests/integration/wait-deadlines.test.ts +++ b/packages/pgconductor-js/tests/integration/wait-deadlines.test.ts @@ -504,6 +504,8 @@ describe("durations", () => { const [row] = await db.sql<{ run_at: Date }[]>` select run_at from pgconductor._private_executions where id = ${id}::uuid `; - expect(row?.run_at).toEqual(new Date("2024-01-01T00:01:00Z")); + const due = new Date("2024-01-01T00:01:00Z").getTime(); + expect(row?.run_at.getTime()).toBeGreaterThanOrEqual(due); + expect(row?.run_at.getTime()).toBeLessThan(due + 10_000); }); }); diff --git a/packages/pgconductor-js/tests/integration/wait-for-event.test.ts b/packages/pgconductor-js/tests/integration/wait-for-event.test.ts index ce3c88a..653a144 100644 --- a/packages/pgconductor-js/tests/integration/wait-for-event.test.ts +++ b/packages/pgconductor-js/tests/integration/wait-for-event.test.ts @@ -1341,4 +1341,127 @@ describe("waitForEvent", () => { expect(result).toEqual(["order", "reply"]); await waiting(database, 0); }); + + describe.each(["fake", "real"])("with %s time", (time) => { + async function clockedDb() { + const database = await db(); + if (time === "fake") await database.clock.set(new Date("2024-01-01T12:00:00Z")); + return database; + } + + test("waitForEvent receives an event emitted after it", async () => { + const database = await clockedDb(); + const result: string[] = []; + const { conductor } = await setup( + database, + async (id, ctx) => { + const value = await ctx.waitForEvent(`clock:${id}`, { event }); + result.push(value.payload.id); + }, + orchestrators, + ); + + await conductor.invoke({ name: "wait.task" }, { id: "run" }); + await suspendedAt(database, "clock:run"); + await conductor.emit("wait.order", { id: "after", kind: "match" }); + await until(async () => result.length === 1); + expect(result).toEqual(["after"]); + }); + + test("waitForAny receives an event emitted after subscribe", async () => { + const database = await clockedDb(); + const winners: unknown[] = []; + const { conductor } = await setup( + database, + async (_id, ctx) => { + const decision = await ctx.subscribe("decision", { event }); + const replied = await ctx.subscribe("reply", { event: reply }); + await ctx.step("post-card", () => ctx.emit("wait.reply", { id: "hello", thread: "t" })); + winners.push(await ctx.waitForAny("any", { decision, reply: replied })); + }, + orchestrators, + ); + + await conductor.invoke({ name: "wait.task" }, { id: "run" }); + await until(async () => winners.length === 1); + expect(winners).toEqual([ + { key: "reply", event: { name: "wait.reply", payload: { id: "hello", thread: "t" } } }, + ]); + }); + + test("waitForAny prefers the branch stored first", async () => { + const database = await clockedDb(); + const winners: unknown[] = []; + const stored = async (key: string) => { + const [row] = + await database.sql`select 1 from pgconductor._private_steps where key = ${key}`; + return row !== undefined; + }; + const { conductor } = await setup( + database, + async (_id, ctx) => { + const first = await ctx.subscribe("first", { event, filter: { code: ["first"] } }); + const second = await ctx.subscribe("second", { event, filter: { code: ["second"] } }); + await ctx.step("post-cards", async () => { + await ctx.emit("wait.order", { id: "earlier", kind: "match", code: "second" }); + await until(() => stored("second")); + await ctx.emit("wait.order", { id: "later", kind: "match", code: "first" }); + await until(() => stored("first")); + }); + winners.push(await ctx.waitForAny("any", { first, second })); + }, + orchestrators, + ); + + await conductor.invoke({ name: "wait.task" }, { id: "run" }); + await until(async () => winners.length === 1); + expect(winners).toEqual([ + { + key: "second", + event: { name: "wait.order", payload: { id: "earlier", kind: "match", code: "second" } }, + }, + ]); + }); + + test("ignores an event emitted before the wait and dispatched after it", async () => { + const database = await clockedDb(); + const result: string[] = []; + const earlier = await database.client.emitEvent({ + eventKey: "wait.order", + payload: { id: "earlier", kind: "match" }, + }); + await database.sql` + update pgconductor._private_executions set run_at = 'infinity' where id = ${earlier}::uuid + `; + const { conductor } = await setup( + database, + async (id, ctx) => { + const value = await ctx.waitForEvent(`clock:${id}`, { event }); + result.push(value.payload.id); + }, + orchestrators, + ); + + const executionId = await conductor.invoke({ name: "wait.task" }, { id: "run" }); + if (!executionId) throw new Error("invoke did not return an execution id"); + await suspendedAt(database, "clock:run"); + await database.sql` + update pgconductor._private_executions + set run_at = pgconductor._private_current_time() + where id = ${earlier}::uuid + `; + await until(async () => { + const [row] = await database.sql<{ completed: boolean }[]>` + select completed_at is not null as completed + from pgconductor._private_executions where id = ${earlier}::uuid + `; + return row?.completed === true; + }); + expect(await step(database, executionId, "clock:run")).toBeUndefined(); + + await conductor.emit("wait.order", { id: "later", kind: "match" }); + await until(async () => result.length === 1); + expect(result).toEqual(["later"]); + }); + }); }); diff --git a/packages/pgconductor-js/tests/integration/window-execution.test.ts b/packages/pgconductor-js/tests/integration/window-execution.test.ts index f12d213..fea86d0 100644 --- a/packages/pgconductor-js/tests/integration/window-execution.test.ts +++ b/packages/pgconductor-js/tests/integration/window-execution.test.ts @@ -94,7 +94,9 @@ describe("Window Execution", () => { expect(execution.completed_at).toBeNull(); // run_at should be scheduled for next day at 09:00 - expect(execution.run_at).toEqual(new Date("2024-01-02T09:00:00Z")); + const opens = new Date("2024-01-02T09:00:00Z").getTime(); + expect(execution.run_at.getTime()).toBeGreaterThanOrEqual(opens); + expect(execution.run_at.getTime()).toBeLessThan(opens + 1000); await orchestrator.stop(); }, 30000);