fix(testing): keep fake time running so event waits match under a set test clock

AGENTS.md

@@ -353,17 +353,17 @@ 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"));
 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
 

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"));

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.
 

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();

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<void> {
-		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<void> {
-		await this.query((sql) => sql.unsafe(`set pgconductor.fake_now = ''`), {
+		await this.query((sql) => sql.unsafe(`set pgconductor.fake_offset = ''`), {
 			label: "clearFakeTime",
 			...opts,
 		});

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();

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<void>;
 	/** Move time forward, starting from the system time if it is not faked */
 	advance(duration: DurationInput): Promise<void>;
@@ -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<Row[]>) => {
+		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`),
 	};
 }

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();

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 () => {

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`

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);
 });

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);
 	});
 });