fix(api): a box gets one run, and a stopped run keeps no address

Two invariants the session endpoint stated but did not hold.

A box runs one thing at a time. `POST /session` read `activeForBox` and
refused when something was already running, but the read and the insert
are two statements with nothing between them: two requests that both saw
"nothing is running" each got a row, and the job poll then handed the
host the same box to start twice. Demonstrated at 2 rows and 2 jobs from
one box. That is the failure the state claim exists to prevent, one step
earlier, and it takes the same answer — a partial unique index on the
predicate the read asks about, so the database refuses the second insert.
`Session.request` turns that refusal into the same 409 in the same words,
so a caller cannot tell which of the two caught it.

The migration resolves any existing duplicates before creating the index,
keeping each box's newest unstopped run because that is the one a person
is waiting on, and stopping the rest rather than deleting them.

Separately, a run that reached `ended` or `failed` kept the last ticket
it published. Publishing a new one is already refused, so the stale
address was both the only ticket a client could read for a dead run and
the one nothing was allowed to replace — and a client that polls would
dial it. Terminal transitions now clear it, in `setState` as well as in
the compare-and-set, so the invariant does not depend on which writer
stopped the run.

Seven tests, each checked against the unfixed code first. The published
descriptions for the ticket field and the read endpoint now say that a
stopped run has no address.

apps/api/app/routes/session.ts

@@ -164,12 +164,17 @@ export namespace SessionApi {
 				// A box runs one thing at a time. Refusing is the honest answer;
 				// starting a second run would leave two rows that both think they
 				// own the same hardware.
+				//
+				// This read is the message, not the guarantee — two callers can
+				// both pass it. `Session.request` is refused by a unique index on
+				// the same predicate, and answers with the same 409 in the same
+				// words, so which one caught it is not visible from here.
 				const active = await Session.activeForBox(box.id);
 				if (active) {
-					conflict('That box already has a run that has not stopped');
+					conflict(Session.BOX_BUSY);
 				}
 
-				const session = await Session.create({
+				const session = await Session.request({
 					id: Identifier.ascending('session'),
 					boxId: box.id,
 					gameId: game.id,
@@ -185,7 +190,7 @@ export namespace SessionApi {
 				tags: ['Session'],
 				summary: 'Read a run you asked for',
 				description:
-					'Poll this while a run starts. The ticket appears part-way through and is republished as addresses are discovered, so re-read it rather than keeping the first one — a client that treats the first ticket as final works on a local network and fails from anywhere else.',
+					'Poll this while a run starts. The ticket appears part-way through and is republished as addresses are discovered, so re-read it rather than keeping the first one — a client that treats the first ticket as final works on a local network and fails from anywhere else. Once the run reaches a terminal state the ticket is null: stop polling and stop dialling it.',
 				responses: {
 					200: {
 						content: { 'application/json': { schema: Result(Session.Info) } },

apps/api/test/session.test.ts

@@ -191,6 +191,29 @@ describe('POST /session', () => {
 		expect(second.body.type).toBe('already_exists');
 	});
 
+	test('two requests racing for one box still start it once', async () => {
+		const s = await scene('route-race', 5509);
+		const [a, b] = await Promise.all([requestSession(s), requestSession(s)]);
+
+		// Which request wins is a timing detail; that exactly one does is not.
+		// The pre-check and the unique index answer identically, so the loser
+		// cannot tell which caught it.
+		//
+		// This asserts the endpoint's answer, not the invariant: two requests
+		// in one process usually interleave such that the pre-check catches
+		// the second, so it passes with the unique index dropped. The index is
+		// pinned in the core tests, where both callers can be made to read
+		// before either writes.
+		const statuses = [a.res.status, b.res.status].sort();
+		expect(statuses).toEqual([201, 409]);
+		expect([a.body, b.body].find((x) => x.type)?.type).toBe('already_exists');
+
+		expect(await Session.listByBox(s.box.id)).toHaveLength(1);
+		// The failure this prevents: the host offered the same box twice.
+		const jobs = await app.request('/machine/jobs', { headers: s.host });
+		expect(((await jobs.json()) as any).data).toHaveLength(1);
+	});
+
 	test('you can only play as an account you have linked', async () => {
 		const mine = await scene('route-account-mine', 5505);
 		const theirs = await scene('route-account-theirs', 5506);
@@ -539,6 +562,35 @@ describe('POST /session/:id/ticket', () => {
 		expect((await Session.fromID(body.data.id))?.ticket).toBeNull();
 	});
 
+	test('a run that stops loses the address it published', async () => {
+		const s = await scene('route-ticket-cleared', 5545);
+		const { body } = await requestSession(s);
+		const report = (state: string) =>
+			app.request(`/session/${body.data.id}/state`, {
+				method: 'POST',
+				headers: s.host,
+				body: JSON.stringify({ state })
+			});
+		await report('starting');
+		await report('live');
+		const published = await app.request(`/session/${body.data.id}/ticket`, {
+			method: 'POST',
+			headers: s.host,
+			body: JSON.stringify({ ticket: 'nodeaaa-live' })
+		});
+		expect(((await published.json()) as any).data.ticket).toBe('nodeaaa-live');
+
+		await report('ended');
+
+		// The polling client is the reason. It reads this endpoint until it has
+		// an address, and an address left behind by a run that stopped is one
+		// it would dial — while publishing a replacement is already refused.
+		const read = await app.request(`/session/${body.data.id}`, { headers: s.user });
+		const after = (await read.json()) as any;
+		expect(after.data.state).toBe('ended');
+		expect(after.data.ticket).toBeNull();
+	});
+
 	test('a ticket has to say something', async () => {
 		const s = await scene('route-ticket-empty', 5544);
 		const { body } = await requestSession(s);

packages/core/migrations/0008_session_one_active_run_per_box.sql

@@ -0,0 +1,30 @@
+-- A box runs one thing at a time, and now the database is what says so.
+--
+-- `POST /session` reads `session.activeForBox` and refuses when something is
+-- already running, but the read and the insert are two statements. Two requests
+-- that both read "nothing is running" before either inserts each get a row, and
+-- `/machine/jobs` then hands the host the same box to start twice. A partial
+-- unique index on the same predicate the read asks about makes the second
+-- insert fail instead. ref(d-0048)
+--
+-- The index cannot be created while any box already has two unstopped runs, so
+-- the duplicates are resolved first. Keeping the newest is the only choice that
+-- matches what a person saw: their most recent request is the one they are
+-- waiting on. The older rows are stopped rather than deleted, because a session
+-- is the billing unit and rows that were once real do not vanish from it.
+--
+-- `ended` and not `failed`: nothing about these runs failed. They were work
+-- nobody picked up, and `failed` carries a reason there is none of.
+UPDATE "session" s
+SET "state" = 'ended', "time_stopped" = now()
+WHERE s."time_stopped" IS NULL
+	AND s."time_deleted" IS NULL
+	AND EXISTS (
+		SELECT 1 FROM "session" newer
+		WHERE newer."box_id" = s."box_id"
+			AND newer."time_stopped" IS NULL
+			AND newer."time_deleted" IS NULL
+			AND (newer."time_created", newer."id") > (s."time_created", s."id")
+	);--> statement-breakpoint
+
+CREATE UNIQUE INDEX "session_box_active_unique" ON "session" USING btree ("box_id") WHERE time_stopped is null and time_deleted is null;

packages/core/migrations/meta/_journal.json

@@ -57,6 +57,13 @@
       "when": 1788460224524,
       "tag": "0007_box_session_team_notnull",
       "breakpoints": true
+    },
+    {
+      "idx": 8,
+      "version": "7",
+      "when": 1788547836146,
+      "tag": "0008_session_one_active_run_per_box",
+      "breakpoints": true
     }
   ]
 }

packages/core/src/session/index.ts

@@ -3,6 +3,7 @@ import z from 'zod';
 
 import { BoxTable, BoxTier } from '../box/box.sql.js';
 import { Database } from '../db/index.js';
+import { ErrorCodes, VisibleError } from '../error.js';
 import { Examples } from '../examples.js';
 import { fn } from '../fn.js';
 import { GameTable } from '../game/game.sql.js';
@@ -18,6 +19,15 @@ import { SessionState, SessionTable } from './session.sql.js';
  * value once.
  */
 export namespace Session {
+	/**
+	 * One wording for "that box is busy", however it was discovered.
+	 *
+	 * Both the read in {@link activeForBox} and the unique index behind
+	 * {@link request} report it, and a caller must not be able to tell which,
+	 * because that would only tell it how close the race was.
+	 */
+	export const BOX_BUSY = 'That box already has a run that has not stopped';
+
 	export const Info = z
 		.object({
 			id: z.string().meta({
@@ -41,7 +51,8 @@ export namespace Session {
 				example: Examples.Session.state
 			}),
 			ticket: z.string().nullable().optional().meta({
-				description: 'Current iroh connect ticket, or null before neshub mints one',
+				description:
+					'Current connect ticket. Null before one has been minted, and null again once the run has stopped — a run that is not there has no address',
 				example: Examples.Session.ticket
 			}),
 			timeStarted: z.string().nullable().optional().meta({
@@ -83,6 +94,41 @@ export namespace Session {
 		}
 	);
 
+	/** Postgres refusing a second row for the same key. */
+	function isUniqueViolation(err: unknown): boolean {
+		const e = err as { code?: string; cause?: { code?: string } };
+		return e?.code === '23505' || e?.cause?.code === '23505';
+	}
+
+	/**
+	 * Ask for a run of a box, and let the database refuse a second one.
+	 *
+	 * The same argument as {@link compareAndSetState}, one step earlier. A
+	 * caller reading {@link activeForBox} first and the insert being refused
+	 * are different properties: the read is a nicer error message, the unique
+	 * index is the invariant. Two requests that both read "nothing is running"
+	 * before either inserts each get a row, and the host is then handed the
+	 * same box to start twice — which is exactly what the state claim exists
+	 * to prevent.
+	 *
+	 * The refusal is the same 409 a caller gets from the read, and worded
+	 * identically, because from outside they are the same fact and which one
+	 * answered is a timing detail.
+	 */
+	export const request = fn(
+		Info.pick({ id: true, boxId: true, gameId: true, linkedAccountId: true }),
+		async (input) => {
+			try {
+				return await create(input);
+			} catch (err) {
+				if (isUniqueViolation(err)) {
+					throw new VisibleError('already_exists', ErrorCodes.Validation.INVALID_STATE, BOX_BUSY);
+				}
+				throw err;
+			}
+		}
+	);
+
 	export const fromID = fn(Info.shape.id, async (id) => {
 		return Database.use(async (tx) => {
 			return tx
@@ -99,9 +145,11 @@ export namespace Session {
 	/**
 	 * The run currently occupying a box, if any.
 	 *
-	 * Newest first and limited to one: a box has at most one live session by
-	 * construction, and if that ever stops being true this is the query that
-	 * should start refusing rather than picking a winner silently.
+	 * At most one row can match: `session_box_active_unique` is a unique index
+	 * on this exact predicate, so "newest first, limited to one" describes the
+	 * query and not a choice being made. Reading this before inserting gives a
+	 * caller a better message than a constraint violation; it is not what makes
+	 * the answer single. See {@link request}.
 	 */
 	export const activeForBox = fn(Info.shape.boxId, async (boxId) => {
 		return Database.use(async (tx) => {
@@ -177,7 +225,12 @@ export namespace Session {
 							? { timeStarted: sql`coalesce(${SessionTable.timeStarted}, ${now})` }
 							: {}),
 						...(input.state === 'ended' || input.state === 'failed'
-							? { timeStopped: sql`coalesce(${SessionTable.timeStopped}, ${now})` }
+							? {
+									timeStopped: sql`coalesce(${SessionTable.timeStopped}, ${now})`,
+									// A stopped run has no address, whichever writer
+									// stopped it.
+									ticket: null
+								}
 							: {})
 					})
 					.where(and(eq(SessionTable.id, input.id), isNull(SessionTable.timeDeleted)))
@@ -395,7 +448,15 @@ export namespace Session {
 							? { timeStarted: sql`coalesce(${SessionTable.timeStarted}, ${now})` }
 							: {}),
 						...(input.to === 'ended' || input.to === 'failed'
-							? { timeStopped: sql`coalesce(${SessionTable.timeStopped}, ${now})` }
+							? {
+									timeStopped: sql`coalesce(${SessionTable.timeStopped}, ${now})`,
+									// A stopped run has no address. Publishing one is
+									// already refused, so keeping the last one would
+									// leave the only readable ticket for a dead run
+									// being the one nothing may replace — and a client
+									// that polls would dial it.
+									ticket: null
+								}
 							: {})
 					})
 					.where(

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

@@ -1,7 +1,8 @@
-import { index, pgEnum, pgTable, text } from 'drizzle-orm/pg-core';
+import { sql } from 'drizzle-orm';
+import { index, pgEnum, pgTable, text, uniqueIndex } from 'drizzle-orm/pg-core';
 
-import { id, timestamps, ulid, utc } from '../db/types.js';
 import { BoxTable } from '../box/box.sql.js';
+import { id, timestamps, ulid, utc } from '../db/types.js';
 import { GameTable } from '../game/game.sql.js';
 import { LinkedAccountTable } from '../user/linked-account.sql.js';
 
@@ -66,6 +67,18 @@ export const SessionTable = pgTable(
 	(t) => [
 		index('session_box_idx').on(t.boxId),
 		index('session_state_idx').on(t.state),
+		// A box runs one thing at a time, and this is where that is true.
+		//
+		// The caller checking first is a nicer error message; it is not the
+		// invariant. Two requests that both read "nothing is running" before
+		// either inserts each get a row, and the host is then handed the same
+		// box to start twice — the identical failure the state claim exists to
+		// prevent, one step earlier. So the predicate is exactly the one
+		// `Session.activeForBox` asks about, and the database refuses the
+		// second row rather than a caller remembering to.
+		uniqueIndex('session_box_active_unique')
+			.on(t.boxId)
+			.where(sql`time_stopped is null and time_deleted is null`),
 		// Metering reads "sessions in this window", and this table is what
 		// billing sums, so the time index is not speculative. ref(d-0048)
 		index('session_started_idx').on(t.timeStarted)

packages/core/src/session/session.test.ts

@@ -58,6 +58,18 @@ afterAll(async () => {
 	}
 });
 
+/** A scene with the run already requested. */
+async function requestedRun(label: string, steamAppId: number) {
+	const s = await scene(label, steamAppId);
+	const session = await Session.request({
+		id: Identifier.ascending('session'),
+		boxId: s.box.id,
+		gameId: s.gameId,
+		linkedAccountId: s.owner.linkedAccountId
+	});
+	return { ...s, session };
+}
+
 describe('Session', () => {
 	test('a session starts requested, with no ticket and no times', async () => {
 		const { owner, box, gameId } = await scene('ses-defaults', 5400);
@@ -414,3 +426,107 @@ describe('Session claim', () => {
 		expect((await Session.fromID(session.id))?.ticket).toBeNull();
 	});
 });
+
+describe('Session one active run per box', () => {
+	test('the database refuses a second unstopped run, not just the caller', async () => {
+		const { owner, machineId, box, gameId } = await scene('ses-one-active', 5450);
+		const mk = () =>
+			Session.request({
+				id: Identifier.ascending('session'),
+				boxId: box.id,
+				gameId,
+				linkedAccountId: owner.linkedAccountId
+			});
+
+		const first = await mk();
+		expect(first.state).toBe('requested');
+
+		// The interleaving `POST /session` permits: both callers read
+		// `activeForBox` and see nothing, then both insert. The read is a
+		// message; the unique index is the invariant, so the second insert is
+		// refused rather than producing a row.
+		const before = await Session.activeForBox(box.id);
+		expect(before?.id).toBe(first.id);
+		await expect(mk()).rejects.toThrow(Session.BOX_BUSY);
+
+		expect(await Session.listByBox(box.id)).toHaveLength(1);
+		// The point of all of it: the host is handed one launch, not two.
+		expect(await Session.listJobsForMachine(machineId)).toHaveLength(1);
+	});
+
+	test('a box that has stopped running is free to run again', async () => {
+		const { owner, machineId, box, gameId } = await scene('ses-one-active-reuse', 5451);
+		const mk = () =>
+			Session.request({
+				id: Identifier.ascending('session'),
+				boxId: box.id,
+				gameId,
+				linkedAccountId: owner.linkedAccountId
+			});
+
+		const first = await mk();
+		await Session.transition({ id: first.id, machineId, state: 'starting', errorMessage: null });
+		await Session.transition({ id: first.id, machineId, state: 'live', errorMessage: null });
+		await Session.transition({ id: first.id, machineId, state: 'ended', errorMessage: null });
+
+		// The index is partial for exactly this reason: a box is a durable
+		// thing and playing twice is the ordinary case, so a stopped run must
+		// not occupy the slot forever.
+		const second = await mk();
+		expect(second.id).not.toBe(first.id);
+		expect(await Session.listByBox(box.id)).toHaveLength(2);
+	});
+});
+
+describe('Session tickets and the end of a run', () => {
+	test('stopping a run takes its address away', async () => {
+		const { machineId, session } = await requestedRun('ses-ticket-cleared', 5452);
+		await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null });
+		await Session.transition({ id: session.id, machineId, state: 'live', errorMessage: null });
+		expect(
+			(await Session.publishTicket({ id: session.id, machineId, ticket: 'live-address' })).session
+				?.ticket
+		).toBe('live-address');
+
+		const ended = await Session.transition({
+			id: session.id,
+			machineId,
+			state: 'ended',
+			errorMessage: null
+		});
+		// Publishing an address for a stopped run is already refused, so a
+		// kept one would be the only ticket a client can read for a dead run
+		// and the one nothing is allowed to replace. It would be dialled.
+		expect(ended.outcome).toBe('moved');
+		expect(ended.session?.ticket).toBeNull();
+		expect((await Session.fromID(session.id))?.ticket).toBeNull();
+	});
+
+	test('a run that failed does not keep an address either', async () => {
+		const { machineId, session } = await requestedRun('ses-ticket-cleared-fail', 5453);
+		await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null });
+		await Session.publishTicket({ id: session.id, machineId, ticket: 'starting-address' });
+
+		const failed = await Session.transition({
+			id: session.id,
+			machineId,
+			state: 'failed',
+			errorMessage: 'the guest never came up'
+		});
+		expect(failed.session?.ticket).toBeNull();
+		// The reason survives; only the address goes.
+		expect(failed.session?.errorMessage).toBe('the guest never came up');
+	});
+
+	test('the unscoped primitive clears it too, whichever writer stops a run', async () => {
+		const { session } = await requestedRun('ses-ticket-cleared-setstate', 5454);
+		await Session.setTicket({ id: session.id, ticket: 'an-address' });
+
+		const ended = await Session.setState({
+			id: session.id,
+			state: 'ended',
+			errorMessage: null
+		});
+		expect(ended?.ticket).toBeNull();
+	});
+});