feat(api): hold GET /machine/jobs open with ?wait= and wake it when work appears

A host found new work on its next poll, 8 to 21 seconds on average. With wait (at most 25 s) a call that finds nothing is held until a run is requested, a stop is asked for, a launch is added or stopped, or a run ends and frees a slot, then answers. A held call also looks again every 5 s so a wake in another process costs seconds, not the job. Workers answer at once. Bun's idle timeout is raised so the held call is not cut.
2026-11-03 22:22:39 +02:00 · 2026-10-18 07:58:03 +03:00
parent dbc53b469c
commit d5c5a43d0f
5 changed files with 481 additions and 7 deletions
@@ -16,6 +16,7 @@ import { Trial } from '../trial/index.js';
 import { Launch } from './launch.js';
 import { LaunchTable } from './launch.sql.js';
 import { SessionState, SessionTable } from './session.sql.js';
+import { JobWake } from './wake.js';

 /**
 * One run of one box, and the unit that gets billed.
@@ -196,6 +197,9 @@ export namespace Session {
 					gameId: session.gameId,
 					first: true
 				});
+				// A run waiting is work for its host. After the commit, so the
+				// host's next look finds the row.
+				await Database.effect(() => JobWake.wakeBox(input.boxId));
 				return { ...session, launches: [first] };
 			});
 		}
@@ -595,6 +599,34 @@ export namespace Session {
 		return [...stops, ...launches.stops, ...starts, ...launches.starts];
 	});

+	/**
+	 * {@link listJobsForMachine}, held open when there is nothing.
+	 *
+	 * Answers at once when there is work. Otherwise waits for a wake (see
+	 * {@link JobWake}) for up to `waitMs`, looking again on every wake and every
+	 * {@link JobWake.RECHECK_MS}, and answers empty at the deadline. The wake
+	 * is listened for before each look, so one that lands between the look and
+	 * the wait is not lost. `signal` ends the wait early, when the host has
+	 * hung up. No database connection is held while waiting.
+	 */
+	export async function holdJobsForMachine(
+		machineId: string,
+		options: { waitMs: number; signal?: AbortSignal }
+	): Promise<Job[]> {
+		const deadline = Date.now() + options.waitMs;
+		for (;;) {
+			const listener = JobWake.listen(machineId);
+			try {
+				const jobs = await listJobsForMachine(machineId);
+				const remaining = deadline - Date.now();
+				if (jobs.length > 0 || remaining <= 0 || options.signal?.aborted) return jobs;
+				await listener.until(Math.min(remaining, JobWake.RECHECK_MS), options.signal);
+			} finally {
+				listener.cancel();
+			}
+		}
+	}
+
 	/** The runs a host is holding: claimed and not yet over. */
 	const holding = (machineId: string) =>
 		Database.use((tx) =>
@@ -854,7 +886,10 @@ export namespace Session {
 				.set({ timeStopRequested: sql`coalesce(${SessionTable.timeStopRequested}, now())` })
 				.where(and(scope, inArray(SessionTable.state, ['starting', 'live'])))
 				.returning();
-			if (marked) return serialize(marked);
+			if (marked) {
+				await Database.effect(() => JobWake.wakeBox(marked.boxId));
+				return serialize(marked);
+			}
 			const [over] = await tx.select().from(SessionTable).where(scope);
 			return over ? serialize(over) : null;
 		});
@@ -949,8 +984,13 @@ export namespace Session {
 						)
 					)
 					.returning()
-					.then((rows) => {
+					.then(async (rows) => {
 						const row = rows.at(0);
+						// A run that is over frees a slot, and a run waiting for
+						// one is work the host can now be handed.
+						if (row && (input.to === 'ended' || input.to === 'failed')) {
+							await Database.effect(() => JobWake.wakeBox(row.boxId));
+						}
 						return row ? serialize(row) : null;
 					});
 			});
@@ -1283,6 +1323,7 @@ export namespace Session {
 					gameId: input.gameId,
 					first: false
 				});
+				await Database.effect(() => JobWake.wakeBox(run.boxId));
 				return { outcome: 'created', launch };
 			});
 		}
@@ -1336,9 +1377,9 @@ export namespace Session {
 			}
 			// A host that does not handle launches would never end this one.
 			if (!(await Launch.hostTakesLaunches(run.id))) return 'hostTooOld';
-			return (
-				(await Launch.markStop(row.id)) ?? Launch.serialize((await Launch.row(run.id, row.id))!)
-			);
+			const marked = await Launch.markStop(row.id);
+			if (marked) await JobWake.wakeBox(run.boxId);
+			return marked ?? Launch.serialize((await Launch.row(run.id, row.id))!);
 		}
 	);

@@ -0,0 +1,95 @@
+import { eq } from 'drizzle-orm';
+
+import { BoxTable } from '../box/box.sql.js';
+import { Database } from '../db/index.js';
+
+/**
+ * Waking a host that is waiting for work.
+ *
+ * A host holds `GET /machine/jobs?wait=` open when there is nothing for it.
+ * Whatever creates or changes work for a machine calls {@link wake} after the
+ * write has committed, and the held call looks again at once.
+ *
+ * In this process only. A second API process would not hear it, which is why a
+ * held call also looks again every {@link RECHECK_MS} on its own: a wake that
+ * lands in another process costs a host a few seconds, never the job. The
+ * `redis` package is a dependency of the API but nothing uses it and no
+ * deployment configures one, so a shared channel is left until there is a
+ * second process to share it between.
+ */
+export namespace JobWake {
+	/** How often a held call looks again with nobody having woken it. */
+	export const RECHECK_MS = 5_000;
+
+	const waiting = new Map<string, Set<() => void>>();
+
+	/** Tell every call held for this machine to look again. */
+	export function wake(machineId: string) {
+		const set = waiting.get(machineId);
+		if (!set) return;
+		waiting.delete(machineId);
+		for (const resolve of set) resolve();
+	}
+
+	/**
+	 * Wake the machine a box is placed on. Looked up, because the places that
+	 * change a run know its box and not its host. Never throws: a failed
+	 * lookup must not fail the write it follows, and the recheck covers it.
+	 */
+	export async function wakeBox(boxId: string) {
+		try {
+			const machineId = await Database.use((tx) =>
+				tx
+					.select({ machineId: BoxTable.machineId })
+					.from(BoxTable)
+					.where(eq(BoxTable.id, boxId))
+					.then((rows) => rows[0]?.machineId ?? null)
+			);
+			if (machineId) wake(machineId);
+		} catch (err) {
+			console.error('[jobs] could not wake the host for a box:', err);
+		}
+	}
+
+	/**
+	 * Start listening for a wake. Taken *before* the caller reads the work, so
+	 * a wake between the read and the wait is not lost.
+	 */
+	export function listen(machineId: string) {
+		let done!: () => void;
+		const woken = new Promise<void>((resolve) => (done = resolve));
+		let set = waiting.get(machineId);
+		if (!set) waiting.set(machineId, (set = new Set()));
+		set.add(done);
+
+		return {
+			/** Resolves on a wake, after `ms`, or when `signal` aborts. */
+			async until(ms: number, signal?: AbortSignal) {
+				let timer: ReturnType<typeof setTimeout> | undefined;
+				let onAbort: (() => void) | undefined;
+				await Promise.race([
+					woken,
+					new Promise<void>((resolve) => (timer = setTimeout(resolve, ms))),
+					new Promise<void>((resolve) => {
+						if (!signal) return;
+						onAbort = () => resolve();
+						if (signal.aborted) resolve();
+						else signal.addEventListener('abort', onAbort, { once: true });
+					})
+				]);
+				clearTimeout(timer);
+				if (signal && onAbort) signal.removeEventListener('abort', onAbort);
+			},
+			cancel() {
+				const current = waiting.get(machineId);
+				current?.delete(done);
+				if (current?.size === 0) waiting.delete(machineId);
+			}
+		};
+	}
+
+	/** How many calls are held, for tests. */
+	export function held(machineId: string) {
+		return waiting.get(machineId)?.size ?? 0;
+	}
+}