feat(api): hold GET /machine/jobs while every job is one the host has seen (?seen=)

apps/api/app/routes/session.ts

@@ -735,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. 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. 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.',
+				'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)) } },
@@ -756,11 +756,21 @@ export namespace SessionApi {
 					.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) => {
-			const { wait } = c.req.valid('query');
+			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.
@@ -771,7 +781,8 @@ export namespace SessionApi {
 					waitMs > 0
 						? await Session.holdJobsForMachine(Actor.machineID, {
 								waitMs,
-								signal: c.req.raw.signal
+								signal: c.req.raw.signal,
+								seen: new Set(seen?.split(',').filter(Boolean))
 							})
 						: await Session.listJobsForMachine(Actor.machineID)
 			});

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

@@ -287,6 +287,48 @@ describe('GET /machine/jobs?wait=', () => {
 		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();

packages/core/src/session/index.ts

@@ -600,18 +600,32 @@ export namespace Session {
 	});
 
 	/**
-	 * {@link listJobsForMachine}, held open when there is nothing.
+	 * 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. 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
+	 * 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 }
+		options: { waitMs: number; signal?: AbortSignal; seen?: ReadonlySet<string> }
 	): Promise<Job[]> {
 		const deadline = Date.now() + options.waitMs;
 		for (;;) {
@@ -619,7 +633,8 @@ export namespace Session {
 			try {
 				const jobs = await listJobsForMachine(machineId);
 				const remaining = deadline - Date.now();
-				if (jobs.length > 0 || remaining <= 0 || options.signal?.aborted) return jobs;
+				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();