feat(api): long-poll GET /machine/jobs (#435)

Cuts host job pickup from the heartbeat interval to under a second.

`GET /machine/jobs?wait=<s>` (capped at 25) answers at once when there
is work, else holds until woken or the wait is up, then answers empty.
Without `wait` nothing changes.

Wake (in-process, keyed by machine id, fired after commit through
`Database.effect`):
- run requested: `Session.create`
- stop asked for on a starting or live run: `stopRun`
- launch added: `addLaunch`; launch stop asked: `stopLaunch`
- run ended or failed (frees a slot for a queued run):
`compareAndSetState`

Not woken, covered by the 5 s recheck: trial and weekend-window expiry,
a heartbeat that raises slots.

Multi-process: `redis` is in `apps/api/package.json` but nothing imports
it and nothing configures a URL, so no pub/sub path here. Each held call
re-reads every 5 s, so a wake that lands in another process costs a few
seconds. Add a channel when a second process exists.

Worker runtime: answers at once (no hold). Bun: `idleTimeout: 60` added,
since the default 10 s would cut the held call.

Tests: `apps/api/test/jobs-wait.test.ts` (8, on the local test Postgres)
plus the full API suite, 216 pass. Pairs with the neslet PR in nescore
(`feat/jobs-long-poll`); an old neslet is unaffected and a new neslet
against an old API falls back to its interval.

apps/api/app/routes/session.ts

@@ -36,6 +36,9 @@ import {
  * credential is a long-lived secret on hardware in somebody's home, and what
  * one leaking can reach is decided here.
  */
+/** The longest a job call is held. Under the server's idle timeout and a tunnel's. */
+const MAX_WAIT_SECONDS = 25;
+
 export namespace SessionApi {
 	/**
 	 * One answer for "no such run" and "not your run".
@@ -732,7 +735,7 @@ export namespace SessionApi {
 			tags: ['Session'],
 			summary: 'Ask for work',
 			description:
-				'Returns the work for the calling host, and only that host — it comes from its own credentials and the scope is the query, so a box cannot see work for another. `session.start` is a run waiting to be started; `session.stop` is a run to end now, because its owner asked, its last launch is over, or the terms it started under ran out, repeated on every poll until the host reports it ended. `launch.start` is another game to start in a box that is already live; `launch.stop` is one launch to end, both repeated until the host reports that launch. Poll at the cadence the heartbeat hands down. An agent skips a kind it does not know.',
+				'Returns the work for the calling host, and only that host — it comes from its own credentials and the scope is the query, so a box cannot see work for another. `session.start` is a run waiting to be started; `session.stop` is a run to end now, because its owner asked, its last launch is over, or the terms it started under ran out, repeated on every poll until the host reports it ended. `launch.start` is another game to start in a box that is already live; `launch.stop` is one launch to end, both repeated until the host reports that launch. An agent skips a kind it does not know. With `wait`, a call that finds nothing is held open until work appears or the wait is up, then answers empty, so a host is handed work within a moment of it being asked for; poll again straight away. Add `seen` and a call is held, too, while all the work it would answer is work the host already has, since a stop is repeated until reported. An API that cannot hold a call answers at once whatever `wait` says, and then the cadence the heartbeat hands down is the one to poll at.',
 			responses: {
 				200: {
 					content: { 'application/json': { schema: Result(z.array(Session.Job)) } },
@@ -741,8 +744,48 @@ export namespace SessionApi {
 				403: ErrorResponses[403]
 			}
 		}),
+		validator(
+			'query',
+			z.object({
+				wait: z.coerce
+					.number()
+					.int()
+					.min(0)
+					.optional()
+					.catch(undefined)
+					.meta({
+						description: `Seconds to hold the call open when there is no work, at most ${MAX_WAIT_SECONDS}. A larger value is held for the most; absent or 0 answers at once`,
+						example: 25
+					}),
+				seen: z
+					.string()
+					.max(4096)
+					.optional()
+					.catch(undefined)
+					.meta({
+						description:
+							'Comma-separated keys of the jobs the host already has, from its last answer: `kind/sessionId`, and `kind/sessionId/launchId` for a launch job. With `wait`, work whose key is listed is not new, so a call is held while all of it is; at the end of the wait the answer is the full list. Absent, every job is new',
+						example: 'session.stop/ses_1,launch.stop/ses_1/lch_1'
+					})
+			})
+		),
 		async (c) => {
-			return c.json({ data: await Session.listJobsForMachine(Actor.machineID) });
+			const { wait, seen } = c.req.valid('query');
+			// A Worker has nowhere to keep a held call that another request can
+			// reach, so there the answer is the plain one and the host's own
+			// interval is what it polls at. Only a long-lived process holds.
+			const holds = typeof (globalThis as { Bun?: unknown }).Bun !== 'undefined';
+			const waitMs = holds ? Math.min(wait ?? 0, MAX_WAIT_SECONDS) * 1000 : 0;
+			return c.json({
+				data:
+					waitMs > 0
+						? await Session.holdJobsForMachine(Actor.machineID, {
+								waitMs,
+								signal: c.req.raw.signal,
+								seen: new Set(seen?.split(',').filter(Boolean))
+							})
+						: await Session.listJobsForMachine(Actor.machineID)
+			});
 		}
 	);
 }

apps/api/app/server.ts

@@ -31,6 +31,9 @@ const hostname = process.env.HOST ?? '127.0.0.1';
 Bun.serve({
 	port,
 	hostname,
+	// A host's job call is held open for up to 25 seconds with nothing sent,
+	// and Bun's default of ten would close it. See `GET /machine/jobs?wait=`.
+	idleTimeout: 60,
 	// A Worker runtime hands the handler a context whose `waitUntil` keeps the
 	// invocation alive past the response. A process does not need convincing to
 	// stay alive, so the equivalent is to let the promise run — with a catch,

apps/api/test/jobs-wait.test.ts

@@ -0,0 +1,345 @@
+import { afterAll, describe, expect, test } from 'bun:test';
+
+import { AccessToken } from '@nestri/core/access-token/index';
+import { Box } from '@nestri/core/box/index';
+import { Fixtures } from '@nestri/core/db/fixtures';
+import { testDb } from '@nestri/core/db/test';
+import { Game } from '@nestri/core/game/index';
+import { Identifier } from '@nestri/core/id';
+import { Machine } from '@nestri/core/machine/index';
+import { Session } from '@nestri/core/session/index';
+import { JobWake } from '@nestri/core/session/wake';
+import { Library } from '@nestri/core/user/library';
+import { LinkedAccount } from '@nestri/core/user/linked-account';
+
+import { app } from '../app/index';
+import './setup';
+
+const sql = testDb();
+
+const HOLDER = 'h'.repeat(32);
+
+const createdUserIds: string[] = [];
+const createdGameIds: string[] = [];
+
+async function newGame(steamAppId: number): Promise<string> {
+	const [row] = await Game.upsert({
+		id: Identifier.ascending('game'),
+		steamAppId,
+		slug: `session-route-${steamAppId}`,
+		name: `Session Route ${steamAppId}`
+	});
+	if (!row) throw new Error('expected a game row');
+	createdGameIds.push(row.id);
+	return row.id;
+}
+
+/**
+ * Everything one session needs, plus both sets of credentials that reach it.
+ *
+ * The person authenticates with a personal token, which is the one user
+ * credential a test can mint without an auth service; the host authenticates
+ * as itself with the secret registration hands back exactly once.
+ */
+async function scene(label: string, steamAppId: number) {
+	const owner = await Fixtures.owner(label);
+	createdUserIds.push(owner.userId);
+
+	const registered = await Machine.register({
+		id: Identifier.ascending('machine'),
+		ownerUserId: owner.userId,
+		teamId: owner.teamId,
+		label
+	});
+
+	const box = await Box.create({
+		id: Identifier.ascending('box'),
+		userId: owner.userId,
+		machineId: registered.id,
+		label,
+		tier: 'sm'
+	});
+
+	const pat = await AccessToken.create({
+		id: Identifier.ascending('accessToken'),
+		ownerUserId: owner.userId,
+		// Null on purpose: a token scoped to the user alone makes the caller a
+		// plain user actor, which is the credential a person browsing has.
+		teamId: null,
+		name: label
+	});
+
+	const gameId = await newGame(steamAppId);
+	// A run launches as a Steam account that owns the game, so the endpoint
+	// refuses one outside the caller's library. Every scene here is about
+	// something else, so the game is stocked.
+	await Library.upsert({
+		id: Identifier.ascending('userLibrary'),
+		userId: owner.userId,
+		gameId,
+		playtime2w: null,
+		playtimeForever: null,
+		lastPlayed: null
+	});
+
+	return {
+		owner,
+		box,
+		machineId: registered.id,
+		gameId,
+		user: {
+			authorization: `Bearer ${pat.token}`,
+			'content-type': 'application/json'
+		} as Record<string, string>,
+		host: {
+			'x-nestri-machine-id': registered.id,
+			'x-nestri-machine-secret': registered.secret,
+			'content-type': 'application/json'
+		} as Record<string, string>
+	};
+}
+
+async function requestSession(s: Awaited<ReturnType<typeof scene>>) {
+	const res = await app.request('/session', {
+		method: 'POST',
+		headers: s.user,
+		body: JSON.stringify({
+			boxId: s.box.id,
+			gameId: s.gameId,
+			linkedAccountId: s.owner.linkedAccountId
+		})
+	});
+	const body = (await res.json()) as any;
+	return { res, body };
+}
+
+afterAll(async () => {
+	if (createdUserIds.length > 0) {
+		await sql`delete from "burn_segment" where session_id in (
+			select s.id from "session" s
+			join "box" b on b.id = s.box_id
+			where b.user_id in ${sql(createdUserIds)}
+		)`;
+		await sql`delete from "box" where user_id in ${sql(createdUserIds)}`;
+		await sql`delete from "user" where id in ${sql(createdUserIds)}`;
+	}
+	if (createdGameIds.length > 0) {
+		await sql`delete from "game" where id in ${sql(createdGameIds)}`;
+	}
+});
+
+type Scene = Awaited<ReturnType<typeof scene>>;
+
+const report = (s: Scene, id: string, state: string, extra: object = {}) =>
+	app.request(`/session/${id}/state`, {
+		method: 'POST',
+		headers: s.host,
+		body: JSON.stringify({ state, claimToken: HOLDER, ...extra })
+	});
+
+const heldJobs = async (s: Scene, query: string) => {
+	const at = Date.now();
+	const res = await app.request(`/machine/jobs${query}`, { headers: s.host });
+	return { status: res.status, data: ((await res.json()) as any).data as any[], took: Date.now() - at };
+};
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+const kinds = (data: any[]) => data.map((j) => j.kind);
+
+async function liveRun(label: string, appId: number) {
+	const s = await scene(label, appId);
+	const { body } = await requestSession(s);
+	const id = body.data.id as string;
+	await app.request('/machine/heartbeat', {
+		method: 'POST',
+		headers: s.host,
+		body: JSON.stringify({ capabilities: ['launches'] })
+	});
+	expect((await report(s, id, 'starting')).status).toBe(200);
+	expect((await report(s, id, 'live')).status).toBe(200);
+	const second = await newGame(appId + 1000);
+	await Library.upsert({
+		id: Identifier.ascending('userLibrary'),
+		userId: s.owner.userId,
+		gameId: second,
+		playtime2w: null,
+		playtimeForever: null,
+		lastPlayed: null
+	});
+	return { s, id, second };
+}
+
+describe('GET /machine/jobs?wait=', () => {
+	test('with no work, it is held for the wait and then answers empty', async () => {
+		const s = await scene('wait-empty', 8100);
+		const r = await heldJobs(s, '?wait=1');
+		expect(r.status).toBe(200);
+		expect(r.data).toEqual([]);
+		expect(r.took).toBeGreaterThanOrEqual(900);
+		expect(r.took).toBeLessThan(3000);
+		expect(JobWake.held(s.machineId)).toBe(0);
+	});
+
+	test('without wait, or with one that is not a number, it answers at once', async () => {
+		const s = await scene('wait-absent', 8101);
+		expect((await heldJobs(s, '')).took).toBeLessThan(500);
+		const r = await heldJobs(s, '?wait=soon');
+		expect(r.status).toBe(200);
+		expect(r.took).toBeLessThan(500);
+		expect((await heldJobs(s, '?wait=0')).took).toBeLessThan(500);
+	});
+
+	test('with work waiting, it answers at once', async () => {
+		const s = await scene('wait-ready', 8102);
+		await requestSession(s);
+		const r = await heldJobs(s, '?wait=20');
+		expect(kinds(r.data)).toEqual(['session.start']);
+		expect(r.took).toBeLessThan(1000);
+	});
+
+	test('a run requested while held is handed over at once', async () => {
+		const s = await scene('wait-start', 8103);
+		const held = heldJobs(s, '?wait=20');
+		await sleep(300);
+		expect(JobWake.held(s.machineId)).toBe(1);
+		const { body } = await requestSession(s);
+		const r = await held;
+		expect(r.data.map((j) => j.sessionId)).toEqual([body.data.id]);
+		expect(r.took).toBeLessThan(2000);
+		expect(JobWake.held(s.machineId)).toBe(0);
+	});
+
+	test('a stop asked for while held is handed over at once', async () => {
+		const { s, id } = await liveRun('wait-stop', 8104);
+		const held = heldJobs(s, '?wait=20');
+		await sleep(300);
+		const stop = await app.request(`/session/${id}/stop`, { method: 'POST', headers: s.user });
+		expect(stop.status).toBe(200);
+		const r = await held;
+		expect(r.data).toEqual([
+			{ kind: 'session.stop', sessionId: id, boxId: s.box.id, reason: 'user' }
+		]);
+		expect(r.took).toBeLessThan(2000);
+	});
+
+	test('a launch added, and a launch stopped, while held are each handed over at once', async () => {
+		const { s, id, second } = await liveRun('wait-launch', 8105);
+		const addHeld = heldJobs(s, '?wait=20');
+		await sleep(300);
+		const added = await app.request(`/session/${id}/launch`, {
+			method: 'POST',
+			headers: s.user,
+			body: JSON.stringify({ gameId: second })
+		});
+		expect(added.status).toBe(201);
+		const launch = ((await added.json()) as any).data;
+		const start = await addHeld;
+		expect(start.data.map((j) => [j.kind, j.launchId])).toEqual([['launch.start', launch.id]]);
+		expect(start.took).toBeLessThan(2000);
+
+		const running = await app.request(`/session/${id}/launch/${launch.id}/state`, {
+			method: 'POST',
+			headers: s.host,
+			body: JSON.stringify({ state: 'running', claimToken: HOLDER })
+		});
+		expect(running.status).toBe(200);
+
+		const stopHeld = heldJobs(s, '?wait=20');
+		await sleep(300);
+		const stopped = await app.request(`/session/${id}/launch/${launch.id}/stop`, {
+			method: 'POST',
+			headers: s.user
+		});
+		expect(stopped.status).toBe(200);
+		const stop = await stopHeld;
+		expect(stop.data.map((j) => [j.kind, j.launchId])).toEqual([['launch.stop', launch.id]]);
+		expect(stop.took).toBeLessThan(2000);
+	});
+
+	test('a run that ends frees a slot, and the next in line is handed it at once', async () => {
+		const { s, id } = await liveRun('wait-slot', 8106);
+		await app.request('/machine/heartbeat', {
+			method: 'POST',
+			headers: s.host,
+			body: JSON.stringify({ slots: 1 })
+		});
+		// The one slot is held by the live run, so a second box's run waits.
+		const other = await Box.create({
+			id: Identifier.ascending('box'),
+			userId: s.owner.userId,
+			machineId: s.machineId,
+			label: 'second',
+			tier: 'sm'
+		});
+		const queued = await Session.request({
+			id: Identifier.ascending('session'),
+			boxId: other.id,
+			gameId: s.gameId,
+			linkedAccountId: s.owner.linkedAccountId
+		});
+		const held = heldJobs(s, '?wait=20');
+		await sleep(300);
+		expect(JobWake.held(s.machineId)).toBe(1);
+		expect((await report(s, id, 'ended')).status).toBe(200);
+		const r = await held;
+		expect(r.data.map((j) => j.sessionId)).toEqual([queued.id]);
+		expect(r.took).toBeLessThan(2000);
+	});
+
+	test('a stop the host has seen is held, and answered in full at the deadline', async () => {
+		const { s, id } = await liveRun('wait-seen-hold', 8108);
+		await app.request(`/session/${id}/stop`, { method: 'POST', headers: s.user });
+		const key = `session.stop/${id}`;
+		const unseen = await heldJobs(s, '?wait=20');
+		expect(unseen.took).toBeLessThan(1000);
+		expect(kinds(unseen.data)).toEqual(['session.stop']);
+		const r = await heldJobs(s, `?wait=1&seen=${encodeURIComponent(key)}`);
+		expect(r.took).toBeGreaterThanOrEqual(900);
+		expect(r.took).toBeLessThan(3000);
+		expect(kinds(r.data)).toEqual(['session.stop']);
+		expect(JobWake.held(s.machineId)).toBe(0);
+	});
+
+	test('a new job among the seen ones is answered at once, with all of them', async () => {
+		const { s, id, second } = await liveRun('wait-seen-new', 8109);
+		const added = await app.request(`/session/${id}/launch`, {
+			method: 'POST',
+			headers: s.user,
+			body: JSON.stringify({ gameId: second })
+		});
+		const launch = ((await added.json()) as any).data;
+		const running = await app.request(`/session/${id}/launch/${launch.id}/state`, {
+			method: 'POST',
+			headers: s.host,
+			body: JSON.stringify({ state: 'running', claimToken: HOLDER })
+		});
+		expect(running.status).toBe(200);
+		await app.request(`/session/${id}/launch/${launch.id}/stop`, {
+			method: 'POST',
+			headers: s.user
+		});
+		const seen = encodeURIComponent(`launch.stop/${id}/${launch.id}`);
+		const held = heldJobs(s, `?wait=20&seen=${seen}`);
+		await sleep(300);
+		expect(JobWake.held(s.machineId)).toBe(1);
+		await app.request(`/session/${id}/stop`, { method: 'POST', headers: s.user });
+		const r = await held;
+		expect(r.took).toBeLessThan(2000);
+		expect(kinds(r.data).sort()).toEqual(['launch.stop', 'session.stop']);
+	});
+
+	test('a host that hangs up stops being held', async () => {
+		const s = await scene('wait-abort', 8107);
+		const controller = new AbortController();
+		const call = Promise.resolve(
+			app.request('/machine/jobs?wait=20', { headers: s.host, signal: controller.signal })
+		).catch(() => null);
+		await sleep(300);
+		expect(JobWake.held(s.machineId)).toBe(1);
+		controller.abort();
+		await call;
+		await sleep(200);
+		expect(JobWake.held(s.machineId)).toBe(0);
+	});
+});

packages/core/src/session/index.ts

@@ -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,49 @@ export namespace Session {
 		return [...stops, ...launches.stops, ...starts, ...launches.starts];
 	});
 
+	/**
+	 * The identity of a job, stable across the repeats of it: its kind, its
+	 * run, and for a launch job its launch. A host names the jobs it already
+	 * has in `?seen=` by these.
+	 */
+	export function jobKey(job: Job): string {
+		return job.kind === 'launch.start' || job.kind === 'launch.stop'
+			? `${job.kind}/${job.sessionId}/${job.launchId}`
+			: `${job.kind}/${job.sessionId}`;
+	}
+
+	/**
+	 * {@link listJobsForMachine}, held open when there is nothing new.
+	 *
+	 * Answers at once when there is work the host has not got. A stop, and a
+	 * launch, are repeated until the host reports them, so work whose
+	 * {@link jobKey} is in `seen` is not new: when every job is in `seen`, or
+	 * there are none, the call waits for a wake (see {@link JobWake}) for up to
+	 * `waitMs`, looking again on every wake and every {@link JobWake.RECHECK_MS},
+	 * and at the deadline answers with whatever there is, the full list. 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; seen?: ReadonlySet<string> }
+	): 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();
+				const fresh = jobs.some((job) => !options.seen?.has(jobKey(job)));
+				if (fresh || 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 +901,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 +999,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 +1338,7 @@ export namespace Session {
 					gameId: input.gameId,
 					first: false
 				});
+				await Database.effect(() => JobWake.wakeBox(run.boxId));
 				return { outcome: 'created', launch };
 			});
 		}
@@ -1336,9 +1392,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))!);
 		}
 	);
 

packages/core/src/session/wake.ts

@@ -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;
+	}
+}