feat(session): runs wait in line for a free slot on their host

A host holds only so many runs at once, and until now every waiting run
was handed to it regardless; the ones it could not hold failed there.

The host now says on its heartbeat how many runs it holds, and is handed
only as many waiting runs as it has free. The rest wait in line: paid
runs before free-weekend runs, then first come first served. Reading a
waiting run returns its place in line, and says when a paid run that
asked later has gone ahead, so a free player is never reordered quietly.

A waiting run nobody has asked after for a minute was abandoned, and
leaves the line rather than holding up everyone behind it. A host that
does not report its slots is handed every run, as before.

apps/api/app/routes/machine.ts

@@ -455,6 +455,11 @@ export namespace MachineApi {
 							description:
 								'Where this host can be reached, as its own endpoint id. Omit it and the stored value is left alone — a host that does not mention where it is has not moved, and an absent field must never read as "nowhere"',
 							example: Examples.Machine.endpointId
+						}),
+						slots: z.number().int().min(0).max(64).optional().meta({
+							description:
+								'How many runs this host will hold at once. Omit it and the stored value is left alone. Runs beyond it wait their turn rather than being handed to a host that would refuse them',
+							example: 2
 						})
 					})
 					// A host that has nothing to add sends no body at all, which
@@ -465,7 +470,8 @@ export namespace MachineApi {
 				const body = c.req.valid('json');
 				const lastSeen = await Machine.touchLastSeen({
 					id: Actor.machineID,
-					endpointId: body?.endpointId
+					endpointId: body?.endpointId,
+					slots: body?.slots
 				});
 				if (!lastSeen) {
 					// The credentials authenticated but the row is gone — a host

apps/api/app/routes/session.ts

@@ -333,7 +333,7 @@ export namespace SessionApi {
 						'No such session, or it is not yours'
 					);
 				}
-				return c.json({ data: session });
+				return c.json({ data: { ...session, queue: await Session.queueFor(session.id) } });
 			}
 		)
 		.post(

apps/api/test/queue.test.ts

@@ -0,0 +1,158 @@
+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 { app } from '../app/index';
+
+const sql = testDb();
+const createdUserIds: string[] = [];
+const createdGameIds: string[] = [];
+
+afterAll(async () => {
+	await sql`delete from session where box_id in (select id from box where user_id = any(${createdUserIds}))`;
+	await sql`delete from box where user_id = any(${createdUserIds})`;
+	await sql`delete from machine where owner_user_id = any(${createdUserIds})`;
+	await sql`delete from game where id = any(${createdGameIds})`;
+	await sql`delete from "user" where id = any(${createdUserIds})`;
+});
+
+/** One host, and a way to put a player in line on it. */
+async function host(label: string, steamAppId: number) {
+	const operator = await Fixtures.owner(`${label}-host`);
+	createdUserIds.push(operator.userId);
+	const machine = await Machine.register({
+		id: Identifier.ascending('machine'),
+		ownerUserId: operator.userId,
+		teamId: operator.teamId,
+		label
+	});
+	const [game] = await Game.upsert({
+		id: Identifier.ascending('game'),
+		steamAppId,
+		slug: `queue-${steamAppId}`,
+		name: `Queue ${steamAppId}`
+	});
+	createdGameIds.push(game!.id);
+	const headers = {
+		'x-nestri-machine-id': machine.id,
+		'x-nestri-machine-secret': machine.secret,
+		'content-type': 'application/json'
+	};
+
+	async function player(name: string, paid: boolean) {
+		const owner = await Fixtures.owner(`${label}-${name}`);
+		createdUserIds.push(owner.userId);
+		const box = await Box.create({
+			id: Identifier.ascending('box'),
+			userId: owner.userId,
+			machineId: machine.id,
+			label: name,
+			tier: 'sm'
+		});
+		const run = await Session.request({
+			id: Identifier.ascending('session'),
+			boxId: box.id,
+			gameId: game!.id,
+			linkedAccountId: owner.linkedAccountId,
+			trial: !paid
+		});
+		const pat = await AccessToken.create({
+			id: Identifier.ascending('accessToken'),
+			ownerUserId: owner.userId,
+			teamId: null,
+			name
+		});
+		const read = async () => {
+			const res = await app.request(`/session/${run.id}`, {
+				headers: { authorization: `Bearer ${pat.token}` }
+			});
+			return ((await res.json()) as any).data;
+		};
+		return { run, read };
+	}
+
+	const beat = (slots: number) =>
+		app.request('/machine/heartbeat', {
+			method: 'POST',
+			headers,
+			body: JSON.stringify({ slots })
+		});
+	const jobs = async () => {
+		const res = await app.request('/machine/jobs', { headers });
+		return (((await res.json()) as any).data as any[]).filter((j) => j.kind === 'session.start');
+	};
+	return { machine, player, beat, jobs };
+}
+
+describe('the line for a full host', () => {
+	test('a host is handed only as many runs as it has free slots', async () => {
+		const h = await host('queue-slots', 7701);
+		const a = await h.player('a', true);
+		const b = await h.player('b', true);
+		const c = await h.player('c', true);
+
+		// An agent that never said how many it holds gets everything, as before.
+		expect(await h.jobs()).toHaveLength(3);
+
+		expect((await h.beat(2)).status).toBe(200);
+		expect((await h.jobs()).map((j) => j.sessionId)).toEqual([a.run.id, b.run.id]);
+
+		await Session.setState({ id: a.run.id, state: 'live' });
+		expect((await h.jobs()).map((j) => j.sessionId)).toEqual([b.run.id]);
+		await Session.setState({ id: b.run.id, state: 'live' });
+		expect(await h.jobs()).toHaveLength(0);
+		expect((await c.read()).queue).toEqual({ position: 1, paidFirst: false });
+
+		// A slot frees and the next in line is handed it.
+		await Session.setState({ id: a.run.id, state: 'ended' });
+		expect((await h.jobs()).map((j) => j.sessionId)).toEqual([c.run.id]);
+	});
+
+	test('paid runs go first, and a free player is told so', async () => {
+		const h = await host('queue-paid', 7702);
+		await h.beat(1);
+		const playing = await h.player('playing', true);
+		await Session.setState({ id: playing.run.id, state: 'live' });
+
+		const free = await h.player('free', false);
+		const paid = await h.player('paid', true);
+
+		expect(await free.read()).toMatchObject({ queue: { position: 2, paidFirst: true } });
+		expect(await paid.read()).toMatchObject({ queue: { position: 1, paidFirst: false } });
+
+		await Session.setState({ id: playing.run.id, state: 'ended' });
+		expect((await h.jobs()).map((j) => j.sessionId)).toEqual([paid.run.id]);
+	});
+
+	test('a run nobody asks after leaves the line', async () => {
+		const h = await host('queue-gone', 7703);
+		await h.beat(1);
+		const playing = await h.player('playing', true);
+		await Session.setState({ id: playing.run.id, state: 'live' });
+		const gone = await h.player('gone', true);
+		const here = await h.player('here', true);
+
+		await sql`update session set time_created = now() - interval '6 minutes' where id = ${gone.run.id}`;
+		await sql`update session set time_created = now() - interval '5 minutes' where id = ${here.run.id}`;
+		expect((await here.read()).queue.position).toBe(2);
+
+		await Session.setState({ id: playing.run.id, state: 'ended' });
+		expect((await h.jobs()).map((j) => j.sessionId)).toEqual([here.run.id]);
+		const left = await Session.fromID(gone.run.id);
+		expect(left).toMatchObject({ state: 'ended', errorMessage: 'Left the queue' });
+	});
+
+	test('a run with a slot is not in line', async () => {
+		const h = await host('queue-none', 7704);
+		const a = await h.player('a', true);
+		await Session.setState({ id: a.run.id, state: 'live' });
+		expect((await a.read()).queue).toBeNull();
+	});
+});

apps/api/test/session.test.ts

@@ -725,6 +725,7 @@ describe('POST /session/:id/ticket', () => {
 			'gameId',
 			'id',
 			'linkedAccountId',
+			'queue',
 			'state',
 			'ticket',
 			'tier',

packages/core/migrations/0025_run_queue.sql

@@ -0,0 +1,2 @@
+ALTER TABLE "machine" ADD COLUMN "slots" integer;--> statement-breakpoint
+ALTER TABLE "session" ADD COLUMN "time_seen" timestamp with time zone;

packages/core/migrations/meta/_journal.json

@@ -176,6 +176,13 @@
       "when": 1791634593049,
       "tag": "0024_box_sizes_move_up",
       "breakpoints": true
+    },
+    {
+      "idx": 25,
+      "version": "7",
+      "when": 1791715304549,
+      "tag": "0025_run_queue",
+      "breakpoints": true
     }
   ]
 }

packages/core/src/machine/index.ts

@@ -338,7 +338,10 @@ export namespace Machine {
 	 * alone and leave the two facts disagreeing.
 	 */
 	export const touchLastSeen = fn(
-		Info.pick({ id: true }).extend({ endpointId: EndpointId.optional() }),
+		Info.pick({ id: true }).extend({
+			endpointId: EndpointId.optional(),
+			slots: z.number().int().min(0).optional()
+		}),
 		async (input) => {
 			try {
 				return await Database.use(async (tx) => {
@@ -350,7 +353,8 @@ export namespace Machine {
 							// that does not mention where it is has not moved, and
 							// clearing the column would deregister a working host
 							// from every route that reads it.
-							...(input.endpointId ? { endpointId: input.endpointId } : {})
+							...(input.endpointId ? { endpointId: input.endpointId } : {}),
+							...(input.slots !== undefined ? { slots: input.slots } : {})
 						})
 						.where(eq(MachineTable.id, input.id))
 						.returning({ lastSeen: MachineTable.lastSeen })

packages/core/src/machine/machine.sql.ts

@@ -1,5 +1,5 @@
 import { sql } from 'drizzle-orm';
-import { check, index, jsonb, pgTable, text, uniqueIndex } from 'drizzle-orm/pg-core';
+import { check, index, integer, jsonb, pgTable, text, uniqueIndex } from 'drizzle-orm/pg-core';
 
 import { id, timestamps, ulid, utc } from '../db/types.js';
 import { OrganisationTable } from '../organisation/organisation.sql.js';
@@ -101,7 +101,11 @@ export const MachineTable = pgTable(
 		// calls the machine ("AMD Radeon RX 9060 XT"), and the first measured part
 		// of the capacity a scheduler will place by. Null until a report carries
 		// them; never filled in by guessing.
-		gpus: jsonb('gpus').$type<{ model: string; vramMib?: number; pciId: string }[]>()
+		gpus: jsonb('gpus').$type<{ model: string; vramMib?: number; pciId: string }[]>(),
+		// How many runs this host will hold at once, as its last heartbeat said.
+		// Null for an agent that does not say, which is handed every waiting
+		// run as before; a number is what makes the rest wait their turn.
+		slots: integer('slots')
 	},
 	(t) => [
 		uniqueIndex('machine_secret_hash_unique').on(t.secretHash),

packages/core/src/session/index.ts

@@ -1,4 +1,4 @@
-import { and, desc, eq, inArray, isNotNull, isNull, sql } from 'drizzle-orm';
+import { and, count, desc, eq, inArray, isNotNull, isNull, sql } from 'drizzle-orm';
 import z from 'zod';
 
 import { Burn } from '../billing/burn.js';
@@ -10,6 +10,7 @@ import { ErrorCodes, VisibleError } from '../error.js';
 import { Examples } from '../examples.js';
 import { fn } from '../fn.js';
 import { GameTable } from '../game/game.sql.js';
+import { MachineTable } from '../machine/machine.sql.js';
 import { Trial } from '../trial/index.js';
 import { SessionState, SessionTable } from './session.sql.js';
 
@@ -32,6 +33,29 @@ export namespace Session {
 	 */
 	export const BOX_BUSY = 'That box already has a run that has not stopped';
 
+	/**
+	 * A run waiting for a slot on its host.
+	 *
+	 * Runs on Nestri GPUs that are paid for go ahead of free-weekend runs, and
+	 * within each, first come first served. That reorders a free player, so it
+	 * is said rather than done quietly: `paidFirst` is true when a paid run that
+	 * asked after this one is ahead of it.
+	 */
+	export const Queue = z.object({
+		position: z.number().int().min(1).meta({ description: 'Place in line, 1 is next', example: 3 }),
+		paidFirst: z.boolean().meta({
+			description: 'Whether a paid run that asked later has gone ahead of this one',
+			example: false
+		})
+	});
+
+	/**
+	 * A waiting run nobody has asked after for this long has been abandoned —
+	 * the app was closed — and leaves the line. A client polls every few
+	 * seconds while it waits, so a minute is many missed polls, not one.
+	 */
+	export const ABANDONED_AFTER_SECONDS = 60;
+
 	export const Info = z
 		.object({
 			id: z.string().meta({
@@ -84,6 +108,10 @@ export namespace Session {
 			errorMessage: z.string().nullable().optional().meta({
 				description: 'Why it failed, when it did',
 				example: Examples.Session.errorMessage
+			}),
+			queue: Queue.nullable().optional().meta({
+				description:
+					'Where this run is in line for its host, while every slot there is taken. Null once it is not waiting'
 			})
 		})
 		.meta({
@@ -426,6 +454,8 @@ export namespace Session {
 				.filter((run) => !seen.has(run.sessionId))
 				.map((run): Job => ({ kind: 'session.stop', ...run }))
 		];
+		await leaveAbandoned(machineId);
+		const free = await freeSlots(machineId);
 		const starts = await Database.use(async (tx) => {
 			return tx
 				.select({ session: SessionTable, box: BoxTable, game: GameTable })
@@ -440,7 +470,9 @@ export namespace Session {
 						isNull(BoxTable.timeDeleted)
 					)
 				)
-				.orderBy(SessionTable.timeCreated)
+				// Paid ahead of free, then first come first served. See {@link Queue}.
+				.orderBy(SessionTable.trial, SessionTable.timeCreated)
+				.limit(free)
 				.then((rows) =>
 					rows.map(
 						(row): Job => ({
@@ -460,6 +492,113 @@ export namespace Session {
 		return [...stops, ...starts];
 	});
 
+	/** The runs a host is holding: claimed and not yet over. */
+	const holding = (machineId: string) =>
+		Database.use((tx) =>
+			tx
+				.select({ n: count() })
+				.from(SessionTable)
+				.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
+				.where(
+					and(
+						eq(BoxTable.machineId, machineId),
+						inArray(SessionTable.state, ['starting', 'live']),
+						isNull(SessionTable.timeStopped),
+						isNull(SessionTable.timeDeleted)
+					)
+				)
+				.then((rows) => rows[0]?.n ?? 0)
+		);
+
+	/**
+	 * How many waiting runs a host may be handed now.
+	 *
+	 * Its own count of slots, less what it already holds. A host that has not
+	 * said how many it holds is handed every waiting run, as before there was a
+	 * line; a run it cannot hold fails there, which is what it did then.
+	 */
+	export const freeSlots = fn(z.string(), async (machineId) => {
+		const slots = await Database.use((tx) =>
+			tx
+				.select({ slots: MachineTable.slots })
+				.from(MachineTable)
+				.where(eq(MachineTable.id, machineId))
+				.then((rows) => rows[0]?.slots ?? null)
+		);
+		if (slots === null) return 1000;
+		return Math.max(0, slots - (await holding(machineId)));
+	});
+
+	/**
+	 * End the waiting runs on a host that nobody has asked after. See
+	 * {@link ABANDONED_AFTER_SECONDS}. Only waiting runs: a run that has a slot
+	 * is ended by its host, never from here.
+	 */
+	export const leaveAbandoned = fn(z.string(), async (machineId) => {
+		await Database.use((tx) =>
+			tx
+				.update(SessionTable)
+				.set({
+					state: 'ended',
+					timeStopped: sql`now()`,
+					errorMessage: 'Left the queue'
+				})
+				.where(
+					and(
+						eq(SessionTable.state, 'requested'),
+						isNull(SessionTable.timeDeleted),
+						inArray(
+							SessionTable.boxId,
+							tx.select({ id: BoxTable.id }).from(BoxTable).where(eq(BoxTable.machineId, machineId))
+						),
+						sql`coalesce(${SessionTable.timeSeen}, ${SessionTable.timeCreated}) < now() - make_interval(secs => ${ABANDONED_AFTER_SECONDS})`
+					)
+				)
+		);
+	});
+
+	/**
+	 * Where a waiting run is in line, or null when it is not waiting.
+	 *
+	 * Ahead of it: every waiting run on the same host that the host would be
+	 * handed first — paid before free, then older first. Reading it also marks
+	 * the run as asked after, which is what keeps it in line.
+	 */
+	export const queueFor = fn(z.string(), async (id): Promise<z.infer<typeof Queue> | null> => {
+		return Database.use(async (tx) => {
+			const mine = await tx
+				.update(SessionTable)
+				.set({ timeSeen: sql`now()` })
+				.where(and(eq(SessionTable.id, id), eq(SessionTable.state, 'requested')))
+				.returning()
+				.then((rows) => rows.at(0));
+			if (!mine) return null;
+			const machine = await tx
+				.select({ machineId: BoxTable.machineId })
+				.from(BoxTable)
+				.where(eq(BoxTable.id, mine.boxId))
+				.then((rows) => rows[0]!.machineId);
+			const waiting = await tx
+				.select({ trial: SessionTable.trial, timeCreated: SessionTable.timeCreated })
+				.from(SessionTable)
+				.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
+				.where(
+					and(
+						eq(BoxTable.machineId, machine),
+						eq(SessionTable.state, 'requested'),
+						isNull(SessionTable.timeDeleted),
+						isNull(BoxTable.timeDeleted)
+					)
+				);
+			const before = (r: { trial: boolean; timeCreated: Date }) =>
+				r.trial !== mine.trial ? !r.trial : r.timeCreated < mine.timeCreated;
+			return {
+				position: waiting.filter(before).length + 1,
+				paidFirst: mine.trial && waiting.some((r) => !r.trial && r.timeCreated > mine.timeCreated)
+			};
+		});
+	});
+
 	/**
 	 * The same run as {@link forMachine}, as the row rather than as the shape
 	 * that goes out.

packages/core/src/session/session.sql.ts

@@ -88,7 +88,13 @@ export const SessionTable = pgTable(
 		 * and the database agree; without it the next generated migration adds a
 		 * column that already exists and fails wherever it has run once.
 		 */
-		claimToken: text('claim_token')
+		claimToken: text('claim_token'),
+		/**
+		 * When the person waiting for this run last asked after it. A run that
+		 * is still waiting for a free slot and nobody has asked about in a while
+		 * has been abandoned — the app was closed — and leaves the line.
+		 */
+		timeSeen: utc('time_seen')
 	},
 	(t) => [
 		index('session_box_idx').on(t.boxId),