From 54d5c81edb2bdf0b872bdf8d153275c3920e9a1f Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sat, 5 Sep 2026 13:02:57 +0300 Subject: [PATCH 1/2] feat(api): hold a run to the attempt that claimed it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The agent side sends a claim token on every write; this side rejected the field outright, so every state report and every ticket publish answered 400. Both bodies now take it. Underneath that, nothing compared a holder. A run was reachable by any caller on the right machine, and a box names exactly one machine — so two attempts polling the same job presented identical credentials and were told apart only by which one's select landed first. That is timing, not a rule, and no caller could be told which case it was in. The row now remembers which attempt holds it. Taking a claim requires there to be no holder; every write after it requires the caller to be the holder. The same state reported by a different attempt is a lost race and not a retry, and is refused whatever the state is - which is the only thing that separates the two 200s from the 409s. The ticket is held to the claim too, for a worse reason than a double start: the client re-reads the address rather than keeping the first, so a ticket written by a losing attempt produces a client that connects, successfully, to a machine running nothing. The holder is never cleared, including on a terminal state, so a settled claim cannot be replayed and a finished run still records which attempt ran it. It is not in what goes out - holding one permits writing to a run, and the owner reading their own session is not the holder. --- apps/api/app/routes/session.ts | 25 ++- apps/api/test/session.test.ts | 159 +++++++++++-- packages/core/src/examples.ts | 3 + packages/core/src/session/index.ts | 157 ++++++++++--- packages/core/src/session/session.test.ts | 257 +++++++++++++++++++--- 5 files changed, 511 insertions(+), 90 deletions(-) diff --git a/apps/api/app/routes/session.ts b/apps/api/app/routes/session.ts index eed46f48..d42ee17a 100644 --- a/apps/api/app/routes/session.ts +++ b/apps/api/app/routes/session.ts @@ -56,12 +56,24 @@ export namespace SessionApi { return actor.properties.userID; } + /** + * The value that says which attempt is speaking. + * + * Described rather than explained. This description is served publicly, so + * it says what to send and not what it defends against. + */ + const ClaimTokenField = Session.ClaimToken.meta({ + description: 'The token this attempt claimed the run with', + example: Examples.Session.claimToken + }); + const StateReport = z .object({ state: Session.ReportableState.meta({ description: 'Where the run has got to', example: 'starting' }), + claimToken: ClaimTokenField, errorMessage: z.string().max(1024).nullable().optional().meta({ description: 'Why it failed. Kept only for a run that did', example: Examples.Session.errorMessage @@ -281,6 +293,7 @@ export namespace SessionApi { id: c.req.valid('param').id, machineId: Actor.machineID, state: body.state, + claimToken: body.claimToken, errorMessage: body.errorMessage ?? null }); @@ -289,6 +302,12 @@ export namespace SessionApi { notYours(); case 'illegal': conflict(`A run in state ${result.session?.state} cannot become ${body.state}`); + case 'notHolder': + // The same answer whatever the state, including the state the + // run is already in. A report from an attempt that does not + // hold the run is that attempt losing a race, and telling + // that apart from a retry is the entire job of the token. + conflict('Another attempt holds this run'); case 'lost': conflict('Another caller moved this run first'); default: @@ -324,7 +343,8 @@ export namespace SessionApi { ticket: z.string().min(1).meta({ description: 'The current connect ticket', example: Examples.Session.ticket - }) + }), + claimToken: ClaimTokenField }) .strict() ), @@ -332,6 +352,7 @@ export namespace SessionApi { const result = await Session.publishTicket({ id: c.req.valid('param').id, machineId: Actor.machineID, + claimToken: c.req.valid('json').claimToken, ticket: c.req.valid('json').ticket }); @@ -340,6 +361,8 @@ export namespace SessionApi { notYours(); case 'unclaimed': conflict('Claim this run by reporting `starting` before publishing an address'); + case 'notHolder': + conflict('Another attempt holds this run'); case 'closed': conflict('That run has stopped, so it has no address to publish'); default: diff --git a/apps/api/test/session.test.ts b/apps/api/test/session.test.ts index c63e12a2..42c6fd92 100644 --- a/apps/api/test/session.test.ts +++ b/apps/api/test/session.test.ts @@ -338,6 +338,13 @@ describe('POST /session', () => { }); }); +/** + * Two attempts. The rival exists so that "the holder" is a claim a test can + * actually fail, rather than a value every request in the file shares. + */ +const HOLDER = 'h'.repeat(32); +const RIVAL = 'r'.repeat(32); + describe('GET /session/:id', () => { test('the owner reads their own run, ticket and all', async () => { const s = await scene('route-read', 5510); @@ -346,12 +353,12 @@ describe('GET /session/:id', () => { await app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state: 'starting' }) + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) }); await app.request(`/session/${body.data.id}/ticket`, { method: 'POST', headers: s.host, - body: JSON.stringify({ ticket: 'nodeaaa-one' }) + body: JSON.stringify({ ticket: 'nodeaaa-one', claimToken: HOLDER }) }); const res = await app.request(`/session/${body.data.id}`, { headers: s.user }); @@ -444,7 +451,7 @@ describe('POST /session/:id/state', () => { const res = await app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state: 'starting' }) + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) }); expect(res.status).toBe(200); expect(((await res.json()) as any).data.state).toBe('starting'); @@ -453,6 +460,58 @@ describe('POST /session/:id/state', () => { expect(((await jobs.json()) as any).data).toEqual([]); }); + test('a report without a holder is refused before it reaches the run', async () => { + const s = await scene('route-claim-tokenless', 5532); + const { body } = await requestSession(s); + + const res = await app.request(`/session/${body.data.id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state: 'starting' }) + }); + expect(res.status).toBe(400); + // And the run did not move on the way to being refused. + expect((await Session.fromID(body.data.id))?.state).toBe('requested'); + }); + + test('the same state from a second attempt is 409, not the retry’s 200', async () => { + const s = await scene('route-claim-rival', 5533); + const { body } = await requestSession(s); + + const report = (claimToken: string) => + app.request(`/session/${body.data.id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state: 'starting', claimToken }) + }); + + expect((await report(HOLDER)).status).toBe(200); + // Identical credentials, identical state, identical everything except + // the holder — which is the only thing that can separate an agent + // retrying a lost response from one that lost the race, and the reason + // this endpoint takes a token at all. + expect((await report(RIVAL)).status).toBe(409); + expect((await report(HOLDER)).status).toBe(200); + }); + + test('a run does not move for an attempt that does not hold it', async () => { + const s = await scene('route-claim-rival-move', 5534); + const { body } = await requestSession(s); + await app.request(`/session/${body.data.id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) + }); + + const res = await app.request(`/session/${body.data.id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state: 'live', claimToken: RIVAL }) + }); + expect(res.status).toBe(409); + expect((await Session.fromID(body.data.id))?.state).toBe('starting'); + }); + test('the same host re-reporting a state it already reported is fine', async () => { const s = await scene('route-claim-retry', 5531); const { body } = await requestSession(s); @@ -461,7 +520,7 @@ describe('POST /session/:id/state', () => { app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state: 'starting' }) + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) }); expect((await report()).status).toBe(200); @@ -480,12 +539,12 @@ describe('POST /session/:id/state', () => { const other = await app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: mine.host, - body: JSON.stringify({ state: 'starting' }) + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) }); const unknown = await app.request(`/session/${Identifier.ascending('session')}/state`, { method: 'POST', headers: mine.host, - body: JSON.stringify({ state: 'starting' }) + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) }); expect(other.status).toBe(403); @@ -501,7 +560,7 @@ describe('POST /session/:id/state', () => { const skipped = await app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state: 'live' }) + body: JSON.stringify({ state: 'live', claimToken: HOLDER }) }); expect(skipped.status).toBe(409); expect((await Session.fromID(body.data.id))?.state).toBe('requested'); @@ -514,7 +573,7 @@ describe('POST /session/:id/state', () => { app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state, errorMessage }) + body: JSON.stringify({ state, claimToken: HOLDER, errorMessage }) }); expect((await report('starting')).status).toBe(200); @@ -533,7 +592,7 @@ describe('POST /session/:id/state', () => { app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state }) + body: JSON.stringify({ state, claimToken: HOLDER }) }); await report('starting'); @@ -550,7 +609,7 @@ describe('POST /session/:id/state', () => { const res = await app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state: 'exploded' }) + body: JSON.stringify({ state: 'exploded', claimToken: HOLDER }) }); expect(res.status).toBe(400); }); @@ -561,7 +620,7 @@ describe('POST /session/:id/state', () => { const res = await app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.user, - body: JSON.stringify({ state: 'starting' }) + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) }); // Terminal states are written by the agent alone; a person closing the // app is not the same fact as a run that stopped. @@ -576,14 +635,14 @@ describe('POST /session/:id/ticket', () => { await app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state: 'starting' }) + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) }); const publish = (ticket: string) => app.request(`/session/${body.data.id}/ticket`, { method: 'POST', headers: s.host, - body: JSON.stringify({ ticket }) + body: JSON.stringify({ ticket, claimToken: HOLDER }) }); const first = await publish('nodeaaa-one'); @@ -595,6 +654,60 @@ describe('POST /session/:id/ticket', () => { expect(await Session.listByBox(s.box.id)).toHaveLength(1); }); + test('a second attempt cannot publish an address over the first’s', async () => { + const s = await scene('route-ticket-rival', 5547); + const { body } = await requestSession(s); + await app.request(`/session/${body.data.id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) + }); + + const publish = (ticket: string, claimToken: string) => + app.request(`/session/${body.data.id}/ticket`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ ticket, claimToken }) + }); + + expect((await publish('nodeaaa-winner', HOLDER)).status).toBe(200); + // The failure this prevents is quieter than a box started twice. The + // client re-reads the address rather than keeping the first one, so a + // ticket written here by the wrong attempt produces a client that + // connects, successfully, to a machine running nothing. + expect((await publish('nodeaaa-rival', RIVAL)).status).toBe(409); + + const read = await app.request(`/session/${body.data.id}`, { headers: s.user }); + expect(((await read.json()) as any).data.ticket).toBe('nodeaaa-winner'); + }); + + test('the holder is not in what the person reading their run gets back', async () => { + const s = await scene('route-ticket-noleak', 5548); + const { body } = await requestSession(s); + await app.request(`/session/${body.data.id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) + }); + + // Holding it permits writing to this run, and the owner is not the + // holder. This asserts the whole shape rather than the one field, so a + // column added later has to be added here too before it goes out. + const read = await app.request(`/session/${body.data.id}`, { headers: s.user }); + const data = ((await read.json()) as any).data; + expect(Object.keys(data).sort()).toEqual([ + 'boxId', + 'errorMessage', + 'gameId', + 'id', + 'linkedAccountId', + 'state', + 'ticket', + 'timeStarted', + 'timeStopped' + ]); + }); + test('a run nobody has claimed has no address to publish', async () => { const s = await scene('route-ticket-early', 5546); const { body } = await requestSession(s); @@ -602,7 +715,7 @@ describe('POST /session/:id/ticket', () => { const early = await app.request(`/session/${body.data.id}/ticket`, { method: 'POST', headers: s.host, - body: JSON.stringify({ ticket: 'nodeaaa-too-soon' }) + body: JSON.stringify({ ticket: 'nodeaaa-too-soon', claimToken: HOLDER }) }); // Publishing before reporting `starting` means the agent skipped the // claim, which is the only mutual exclusion in the design. @@ -612,12 +725,12 @@ describe('POST /session/:id/ticket', () => { await app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state: 'starting' }) + body: JSON.stringify({ state: 'starting', claimToken: HOLDER }) }); const now = await app.request(`/session/${body.data.id}/ticket`, { method: 'POST', headers: s.host, - body: JSON.stringify({ ticket: 'nodeaaa-in-time' }) + body: JSON.stringify({ ticket: 'nodeaaa-in-time', claimToken: HOLDER }) }); expect(now.status).toBe(200); }); @@ -630,7 +743,7 @@ describe('POST /session/:id/ticket', () => { const res = await app.request(`/session/${body.data.id}/ticket`, { method: 'POST', headers: mine.host, - body: JSON.stringify({ ticket: 'nodeaaa-stolen' }) + body: JSON.stringify({ ticket: 'nodeaaa-stolen', claimToken: HOLDER }) }); expect(res.status).toBe(403); expect((await Session.fromID(body.data.id))?.ticket).toBeNull(); @@ -643,7 +756,7 @@ describe('POST /session/:id/ticket', () => { app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state }) + body: JSON.stringify({ state, claimToken: HOLDER }) }); await report('starting'); await report('live'); @@ -652,7 +765,7 @@ describe('POST /session/:id/ticket', () => { const res = await app.request(`/session/${body.data.id}/ticket`, { method: 'POST', headers: s.host, - body: JSON.stringify({ ticket: 'nodeaaa-late' }) + body: JSON.stringify({ ticket: 'nodeaaa-late', claimToken: HOLDER }) }); expect(res.status).toBe(409); expect((await Session.fromID(body.data.id))?.ticket).toBeNull(); @@ -665,14 +778,14 @@ describe('POST /session/:id/ticket', () => { app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state }) + body: JSON.stringify({ state, claimToken: HOLDER }) }); 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' }) + body: JSON.stringify({ ticket: 'nodeaaa-live', claimToken: HOLDER }) }); expect(((await published.json()) as any).data.ticket).toBe('nodeaaa-live'); @@ -693,7 +806,7 @@ describe('POST /session/:id/ticket', () => { const res = await app.request(`/session/${body.data.id}/ticket`, { method: 'POST', headers: s.host, - body: JSON.stringify({ ticket: '' }) + body: JSON.stringify({ ticket: '', claimToken: HOLDER }) }); expect(res.status).toBe(400); }); @@ -707,7 +820,7 @@ describe('The box a run happens on', () => { app.request(`/session/${body.data.id}/state`, { method: 'POST', headers: s.host, - body: JSON.stringify({ state, errorMessage }) + body: JSON.stringify({ state, claimToken: HOLDER, errorMessage }) }); expect((await Box.fromID(s.box.id))?.state).toBe('created'); diff --git a/packages/core/src/examples.ts b/packages/core/src/examples.ts index 5dafeed6..3eb6eb8f 100644 --- a/packages/core/src/examples.ts +++ b/packages/core/src/examples.ts @@ -146,6 +146,9 @@ export namespace Examples { linkedAccountId: Id('linkedAccount'), state: 'live' as const, ticket: 'nodeaaqf…', + // Shaped like the real thing so the docs do not teach a shorter one, + // and obviously not one. + claimToken: '00000000000000000000000000000000', timeStarted: '2026-07-28T12:00:00.000Z', timeStopped: null, errorMessage: null diff --git a/packages/core/src/session/index.ts b/packages/core/src/session/index.ts index 82f94d4c..d6a2cb19 100644 --- a/packages/core/src/session/index.ts +++ b/packages/core/src/session/index.ts @@ -250,6 +250,18 @@ export namespace Session { * `requested` is missing on purpose: it is written once, when the row is * created, and nothing may put a run back there. */ + /** + * The holder of a claim, as the agent minted it. + * + * Opaque here on purpose: this end never generates one, never parses one + * and never shows one to anybody. The only property it needs is that two + * attempts cannot present the same value, and no length check can prove + * that — the floor exists to refuse something obviously degenerate, not to + * measure entropy. 22 characters is 128 bits at the tightest encoding + * anyone would reasonably use. + */ + export const ClaimToken = z.string().min(22).max(255); + export const ReportableState = z.enum(['starting', 'live', 'ended', 'failed']); export type ReportableState = z.infer; @@ -360,28 +372,40 @@ export namespace Session { }); }); + /** + * The same run as {@link forMachine}, as the row rather than as the shape + * that goes out. + * + * The claim holder is not in {@link Info} and must not be: `serialize` is + * what every caller of this resource is answered with, a token in it would + * reach the person's own `GET /session/:id`, and holding one permits + * writing to a run. So the two callers that compare a holder read the row, + * and nothing that leaves this file ever carries the column. + */ + const rowForMachine = async (id: string, machineId: string) => { + return Database.use(async (tx) => { + return tx + .select({ session: SessionTable }) + .from(SessionTable) + .innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id)) + .where( + and( + eq(SessionTable.id, id), + eq(BoxTable.machineId, machineId), + isNull(SessionTable.timeDeleted), + isNull(BoxTable.timeDeleted) + ) + ) + .then((rows) => rows.at(0)?.session ?? null); + }); + }; + /** One run, visible only to the host its box is placed on. */ export const forMachine = fn( z.object({ id: Info.shape.id, machineId: z.string() }), async (input) => { - return Database.use(async (tx) => { - return tx - .select({ session: SessionTable }) - .from(SessionTable) - .innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id)) - .where( - and( - eq(SessionTable.id, input.id), - eq(BoxTable.machineId, input.machineId), - isNull(SessionTable.timeDeleted), - isNull(BoxTable.timeDeleted) - ) - ) - .then((rows) => { - const row = rows.at(0); - return row ? serialize(row.session) : null; - }); - }); + const row = await rowForMachine(input.id, input.machineId); + return row ? serialize(row) : null; } ); @@ -435,15 +459,25 @@ export namespace Session { machineId: z.string(), from: z.enum(SessionState.enumValues), to: z.enum(SessionState.enumValues), + claimToken: ClaimToken, errorMessage: Info.shape.errorMessage }), async (input) => { const now = sql`now()`; + // The claim is the one transition that takes the holder rather than + // presenting it, and it is exactly this pair — derived rather than + // passed, because a flag that can disagree with the states either + // side of it is a flag that eventually will. + const isClaim = input.from === 'requested' && input.to === 'starting'; return Database.use(async (tx) => { return tx .update(SessionTable) .set({ state: input.to, + // Taken on the claim and never written again: a terminal row + // still records which attempt ran it, and a holder that goes + // back to null lets a settled claim be replayed. + ...(isClaim ? { claimToken: input.claimToken } : {}), errorMessage: input.to === 'failed' ? (input.errorMessage ?? null) : null, ...(input.to === 'live' ? { timeStarted: sql`coalesce(${SessionTable.timeStarted}, ${now})` } @@ -464,6 +498,15 @@ export namespace Session { and( eq(SessionTable.id, input.id), eq(SessionTable.state, input.from), + // The holder is in the write and not only in the read + // above it. Taking a claim requires there to be none; + // everything after requires this caller to hold it. Two + // agents that both read an unheld row still leave here + // with one winner, because the second one's `where` no + // longer matches. + isClaim + ? isNull(SessionTable.claimToken) + : eq(SessionTable.claimToken, input.claimToken), isNull(SessionTable.timeDeleted), inArray(SessionTable.boxId, boxesOn(tx, input.machineId)) ) @@ -488,10 +531,23 @@ export namespace Session { * - `unchanged` — already in that state. A retry after a lost response is * not a broken agent and must not be told it is. * - `illegal` — not a transition that exists. The row does not move. + * - `notHolder` — a report from an attempt that does not hold this run, + * which is also what a second attempt to claim one looks like: taking + * the claim and leaving `requested` are a single write, so by the time + * a rival arrives the run is `starting` and held, and there is no + * separate answer to give it. Distinct from `lost` because only one of + * them is a race — this one is decided by the request, and is the answer + * whatever state the run is in, including the state being reported. * - `lost` — a legal transition that something else got to first. * - `moved` — it happened. */ - export type TransitionOutcome = 'forbidden' | 'unchanged' | 'illegal' | 'lost' | 'moved'; + export type TransitionOutcome = + | 'forbidden' + | 'unchanged' + | 'illegal' + | 'notHolder' + | 'lost' + | 'moved'; export interface TransitionResult { outcome: TransitionOutcome; @@ -536,6 +592,7 @@ export namespace Session { id: Info.shape.id, machineId: z.string(), state: z.enum(SessionState.enumValues), + claimToken: ClaimToken, errorMessage: Info.shape.errorMessage }), async (input): Promise => { @@ -544,11 +601,34 @@ export namespace Session { // first without the second is how a box gets stuck `running` with // nothing running on it, and nothing here would ever correct it. return Database.transaction(async (): Promise => { - const current = await forMachine({ id: input.id, machineId: input.machineId }); - if (!current) return { outcome: 'forbidden', session: null }; - if (current.state === input.state) return { outcome: 'unchanged', session: current }; - if (!NEXT_STATES[current.state].includes(input.state)) { - return { outcome: 'illegal', session: current }; + const row = await rowForMachine(input.id, input.machineId); + if (!row) return { outcome: 'forbidden', session: null }; + const current = serialize(row); + + const isClaim = current.state === 'requested' && input.state === 'starting'; + if (isClaim) { + // A `requested` run has no holder, because the two are written + // together — so this is an invariant guard and not a case the + // endpoint can produce. A rival never sees `requested`; it sees + // `starting` and is answered below. + if (row.claimToken !== null) return { outcome: 'notHolder', session: current }; + } else { + const same = current.state === input.state; + // Legality first, because it is a property of the run and not + // of the caller: `requested → live` is the same mistake whoever + // makes it, and answering it with the holder would hide from an + // agent that it skipped the claim entirely. + if (!same && !NEXT_STATES[current.state].includes(input.state)) { + return { outcome: 'illegal', session: current }; + } + // Then the holder, and it comes before `unchanged` — that order + // is the whole point of the column. The same state reported by a + // different attempt is a lost race, not a retry, and nothing + // else in the request can tell those two apart. + if (row.claimToken !== input.claimToken) { + return { outcome: 'notHolder', session: current }; + } + if (same) return { outcome: 'unchanged', session: current }; } const moved = await compareAndSetState({ @@ -556,6 +636,7 @@ export namespace Session { machineId: input.machineId, from: current.state, to: input.state, + claimToken: input.claimToken, errorMessage: input.errorMessage }); // The state read above is not the state written below, and the @@ -575,7 +656,7 @@ export namespace Session { ); export interface TicketResult { - outcome: 'forbidden' | 'unclaimed' | 'closed' | 'published'; + outcome: 'forbidden' | 'unclaimed' | 'notHolder' | 'closed' | 'published'; session: Info | null; } @@ -599,17 +680,31 @@ export namespace Session { * refused, and the two refusals are separate answers because they are * different mistakes: a run not yet claimed is an agent that skipped a * step, and a run that stopped is one that has nothing left to reach. + * + * Held to the claim for a worse reason than the double start. A losing + * attempt that could publish here would overwrite the winner's address + * with its own; the client re-reads rather than caching, so it would take + * the new one and connect, successfully, to a box nobody is running. Two + * boxes started is waste, and it is visible. A working client pointed at + * the wrong machine looks exactly like everything having worked. */ export const publishTicket = fn( z.object({ id: Info.shape.id, machineId: z.string(), + claimToken: ClaimToken, ticket: z.string().min(1) }), async (input): Promise => { - const current = await forMachine({ id: input.id, machineId: input.machineId }); - if (!current) return { outcome: 'forbidden', session: null }; + const row = await rowForMachine(input.id, input.machineId); + if (!row) return { outcome: 'forbidden', session: null }; + const current = serialize(row); + // Before the holder, because a run nobody has claimed has no holder + // to fail against and "claim it first" is the answer that helps. if (current.state === 'requested') return { outcome: 'unclaimed', session: current }; + if (row.claimToken !== input.claimToken) { + return { outcome: 'notHolder', session: current }; + } return Database.use(async (tx) => { return tx @@ -618,10 +713,12 @@ export namespace Session { .where( and( eq(SessionTable.id, input.id), - // The state is in the write and not only in the check - // above it, so a run that stops underneath this call - // does not acquire an address on the way out. + // The state and the holder are in the write and not only + // in the checks above it, so a run that stops underneath + // this call does not acquire an address on the way out, + // and neither does one whose claim moved. inArray(SessionTable.state, [...ADDRESSABLE]), + eq(SessionTable.claimToken, input.claimToken), isNull(SessionTable.timeDeleted), inArray(SessionTable.boxId, boxesOn(tx, input.machineId)) ) diff --git a/packages/core/src/session/session.test.ts b/packages/core/src/session/session.test.ts index 84b16347..75575d38 100644 --- a/packages/core/src/session/session.test.ts +++ b/packages/core/src/session/session.test.ts @@ -70,6 +70,13 @@ async function requestedRun(label: string, steamAppId: number) { return { ...s, session }; } +/** + * Two attempts, so a test that means "the holder" cannot pass by accident on a + * value that every caller in the file happens to share. + */ +const HOLDER = 'h'.repeat(32); +const RIVAL = 'r'.repeat(32); + describe('Session', () => { test('a session starts requested, with no ticket and no times', async () => { const { owner, box, gameId } = await scene('ses-defaults', 5400); @@ -240,6 +247,7 @@ describe('Session jobs', () => { expect(await Session.listJobsForMachine(machineId)).toHaveLength(1); await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'starting', @@ -265,6 +273,7 @@ describe('Session claim', () => { const { machineId, session } = await requested('ses-cas', 5420); const won = await Session.compareAndSetState({ + claimToken: HOLDER, id: session.id, machineId, from: 'requested', @@ -273,10 +282,12 @@ describe('Session claim', () => { }); expect(won?.state).toBe('starting'); - // The same attempt again. The row is no longer `requested`, so the - // update matches nothing — which is what stops two agents from both - // starting the same box. Updating on the id alone would succeed twice. + // A second attempt, with a token of its own. The row is no longer + // `requested` and it already has a holder, so the update matches + // nothing — which is what stops two agents from both starting the same + // box. Updating on the id alone would succeed twice. const lost = await Session.compareAndSetState({ + claimToken: RIVAL, id: session.id, machineId, from: 'requested', @@ -292,6 +303,7 @@ describe('Session claim', () => { const other = await scene('ses-cas-other', 5422); const result = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId: other.machineId, state: 'starting', @@ -304,6 +316,7 @@ describe('Session claim', () => { // classification above it. expect( await Session.compareAndSetState({ + claimToken: HOLDER, id: session.id, machineId: other.machineId, from: 'requested', @@ -316,6 +329,7 @@ describe('Session claim', () => { test('a session that does not exist is refused the same way as one that is not yours', async () => { const other = await scene('ses-cas-ghost', 5423); const result = await Session.transition({ + claimToken: HOLDER, // A well-formed id for a row that was never written. id: Identifier.ascending('session'), machineId: other.machineId, @@ -330,8 +344,9 @@ describe('Session claim', () => { test('re-reporting the state you already reported changes nothing', async () => { const { machineId, session } = await requested('ses-repeat', 5424); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); const again = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'starting', @@ -349,6 +364,7 @@ describe('Session claim', () => { // is the only mutual exclusion here — so it is refused however tempting // the shortcut looks. const skipped = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'live', @@ -357,11 +373,12 @@ describe('Session claim', () => { expect(skipped.outcome).toBe('illegal'); expect((await Session.fromID(session.id))?.state).toBe('requested'); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); - await Session.transition({ id: session.id, machineId, state: 'failed', errorMessage: 'no' }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'failed', errorMessage: 'no' }); // Terminal is terminal: a dead session cannot be resurrected. const raised = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'live', @@ -373,8 +390,9 @@ describe('Session claim', () => { test('the timestamps survive a duplicate report, which is what billing rests on', async () => { const { machineId, session } = await requested('ses-idempotent', 5426); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); const live = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'live', @@ -383,6 +401,7 @@ describe('Session claim', () => { expect(live.session?.timeStarted).not.toBeNull(); const repeat = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'live', @@ -395,9 +414,10 @@ describe('Session claim', () => { const { machineId, session } = await requested('ses-ticket-scope', 5427); const other = await scene('ses-ticket-other', 5428); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); const refused = await Session.publishTicket({ + claimToken: HOLDER, id: session.id, machineId: other.machineId, ticket: 'stolen' @@ -406,22 +426,182 @@ describe('Session claim', () => { expect((await Session.fromID(session.id))?.ticket).toBeNull(); // A ticket may appear while the state is still `starting`. - const first = await Session.publishTicket({ id: session.id, machineId, ticket: 'one' }); + const first = await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'one' }); expect(first.outcome).toBe('published'); expect(first.session?.ticket).toBe('one'); expect(first.session?.state).toBe('starting'); - const second = await Session.publishTicket({ id: session.id, machineId, ticket: 'two' }); + const second = await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'two' }); expect(second.session?.ticket).toBe('two'); }); + test('the claim writes a holder, and the holder never goes out', async () => { + const { machineId, session } = await requested('ses-holder', 5440); + + const claimed = await Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }); + expect(claimed.outcome).toBe('moved'); + // Holding one permits writing to a run, including publishing the + // address a client will dial. It is carried by nothing that is + // serialized, and this is the assertion that keeps it that way. + expect(claimed.session).not.toHaveProperty('claimToken'); + expect(await Session.fromID(session.id)).not.toHaveProperty('claimToken'); + }); + + test('a rival claiming the same run is refused, and sees only that it is held', async () => { + const { machineId, session } = await requested('ses-held', 5441); + await Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }); + + // Taking the claim and leaving `requested` are one write, so a rival + // polling the same job never finds a `requested` run with a holder — + // it finds a `starting` one it does not hold. There is no second + // answer to give it, and this is the only one. + const second = await Session.transition({ + claimToken: RIVAL, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }); + expect(second.outcome).toBe('notHolder'); + }); + + test('the same state from a different attempt is a lost race and not a retry', async () => { + const { machineId, session } = await requested('ses-rival-same', 5442); + await Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }); + + // Same machine, same credentials, same state the run is already in. + // The holder is the only thing separating this call from the one below + // it, and without the column both would be answered the same way. + const rival = await Session.transition({ + claimToken: RIVAL, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }); + expect(rival.outcome).toBe('notHolder'); + + const retry = await Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }); + expect(retry.outcome).toBe('unchanged'); + }); + + test('a run does not move onwards for an attempt that does not hold it', async () => { + const { machineId, session } = await requested('ses-rival-move', 5443); + await Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }); + + const stolen = await Session.transition({ + claimToken: RIVAL, + id: session.id, + machineId, + state: 'live', + errorMessage: null + }); + expect(stolen.outcome).toBe('notHolder'); + expect((await Session.fromID(session.id))?.state).toBe('starting'); + }); + + test('the holder outlives the run, so a settled claim cannot be replayed', async () => { + const { machineId, session } = await requested('ses-holder-kept', 5444); + for (const state of ['starting', 'live', 'ended'] as const) { + await Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state, + errorMessage: null + }); + } + + // Reported at the state the run is already in, so the answer turns on + // the holder and on nothing else. Were the column cleared on a terminal + // state, the rival would match a null holder and be told `unchanged` — + // a dead claim answered as though it were the live one. + const replay = await Session.transition({ + claimToken: RIVAL, + id: session.id, + machineId, + state: 'ended', + errorMessage: null + }); + expect(replay.outcome).toBe('notHolder'); + + const holder = await Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state: 'ended', + errorMessage: null + }); + expect(holder.outcome).toBe('unchanged'); + }); + + test('a losing attempt cannot publish an address over the winner’s', async () => { + const { machineId, session } = await requested('ses-ticket-holder', 5445); + await Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }); + await Session.publishTicket({ + claimToken: HOLDER, + id: session.id, + machineId, + ticket: 'the-winner' + }); + + // The same machine, so the scope check passes and lets this through to + // the holder. Without that check the write would land, the client would + // re-read it rather than keeping the first, and it would connect — + // successfully — to a box nobody is running. + const stolen = await Session.publishTicket({ + claimToken: RIVAL, + id: session.id, + machineId, + ticket: 'the-wrong-machine' + }); + expect(stolen.outcome).toBe('notHolder'); + expect((await Session.fromID(session.id))?.ticket).toBe('the-winner'); + }); + test('a stopped session has no address to publish', async () => { const { machineId, session } = await requested('ses-ticket-dead', 5429); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); - await Session.transition({ id: session.id, machineId, state: 'live', errorMessage: null }); - await Session.transition({ id: session.id, machineId, state: 'ended', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'live', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'ended', errorMessage: null }); - const result = await Session.publishTicket({ id: session.id, machineId, ticket: 'late' }); + const result = await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'late' }); expect(result.outcome).toBe('closed'); expect((await Session.fromID(session.id))?.ticket).toBeNull(); }); @@ -465,9 +645,9 @@ describe('Session one active run per box', () => { }); 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 }); + await Session.transition({ claimToken: HOLDER, id: first.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: first.id, machineId, state: 'live', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, 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 @@ -481,14 +661,15 @@ describe('Session one active run per box', () => { 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 }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'live', errorMessage: null }); expect( - (await Session.publishTicket({ id: session.id, machineId, ticket: 'live-address' })).session + (await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'live-address' })).session ?.ticket ).toBe('live-address'); const ended = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'ended', @@ -504,10 +685,11 @@ describe('Session tickets and the end of a run', () => { 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' }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'starting-address' }); const failed = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'failed', @@ -538,12 +720,12 @@ describe('Session and the box underneath it', () => { // `created` while a run on it was `live`. expect((await Box.fromID(box.id))?.state).toBe('created'); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); // `starting` is deliberately not a box state: that transition is // synchronous from the agent's side, so nothing would ever write it. expect((await Box.fromID(box.id))?.state).toBe('created'); - await Session.transition({ id: session.id, machineId, state: 'live', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'live', errorMessage: null }); const running = await Box.fromID(box.id); expect(running?.state).toBe('running'); expect(running?.stopReason).toBeNull(); @@ -552,9 +734,9 @@ describe('Session and the box underneath it', () => { test('a run that ends stops its box, cleanly', async () => { const { machineId, box, session } = await requestedRun('ses-box-ended', 5461); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); - await Session.transition({ id: session.id, machineId, state: 'live', errorMessage: null }); - await Session.transition({ id: session.id, machineId, state: 'ended', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'live', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'ended', errorMessage: null }); const stopped = await Box.fromID(box.id); expect(stopped?.state).toBe('stopped'); @@ -564,8 +746,9 @@ describe('Session and the box underneath it', () => { test('a run that fails stops its box in the words the agent used', async () => { const { machineId, box, session } = await requestedRun('ses-box-failed', 5462); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'failed', @@ -585,6 +768,7 @@ describe('Session and the box underneath it', () => { const other = await scene('ses-box-otherhost', 5464); const refused = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId: other.machineId, state: 'starting', @@ -595,6 +779,7 @@ describe('Session and the box underneath it', () => { // An illegal transition does not move the run, so it must not move the // box either — otherwise the box records a run that never happened. const illegal = await Session.transition({ + claimToken: HOLDER, id: session.id, machineId, state: 'live', @@ -612,24 +797,24 @@ describe('Session tickets need a claim first', () => { // A ticket is the address of something being brought up, so publishing // one for a `requested` run means the agent skipped the claim — the // step that is the only mutual exclusion in the design. - const early = await Session.publishTicket({ id: session.id, machineId, ticket: 'too-soon' }); + const early = await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'too-soon' }); expect(early.outcome).toBe('unclaimed'); expect((await Session.fromID(session.id))?.ticket).toBeNull(); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); - const now = await Session.publishTicket({ id: session.id, machineId, ticket: 'in-time' }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); + const now = await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'in-time' }); expect(now.outcome).toBe('published'); expect(now.session?.ticket).toBe('in-time'); }); test('the two refusals are different answers, because they are different mistakes', async () => { const { machineId, session } = await requestedRun('ses-ticket-refusals', 5466); - const unclaimed = await Session.publishTicket({ id: session.id, machineId, ticket: 'a' }); + const unclaimed = await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'a' }); - await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null }); - await Session.transition({ id: session.id, machineId, state: 'live', errorMessage: null }); - await Session.transition({ id: session.id, machineId, state: 'ended', errorMessage: null }); - const closed = await Session.publishTicket({ id: session.id, machineId, ticket: 'b' }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'starting', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'live', errorMessage: null }); + await Session.transition({ claimToken: HOLDER, id: session.id, machineId, state: 'ended', errorMessage: null }); + const closed = await Session.publishTicket({ claimToken: HOLDER, id: session.id, machineId, ticket: 'b' }); // One is an agent that has not claimed the work; the other is a run // with nothing left to reach. Collapsing them would tell an agent From 647e5c5264b390af3a10c5f21d8aa79ae9b807ea Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sat, 5 Sep 2026 13:26:55 +0300 Subject: [PATCH 2/2] test(core): claim two attempts at once, not one after the other The mutual exclusion was asserted only through sequential calls, where the winner had already committed before the rival began. That never reaches the case the design is for: both attempts reading the run as unclaimed before either writes. Two tests, because the first can pass for the wrong reason. The concurrent transitions depend on how the transactions interleave; the paired updates skip the read entirely, so nothing but the predicate in the where clause can refuse the second. Both fail with two winners if the check is moved out of the write and left in the read above it. --- packages/core/src/session/session.test.ts | 94 +++++++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/packages/core/src/session/session.test.ts b/packages/core/src/session/session.test.ts index 75575d38..cf1f9087 100644 --- a/packages/core/src/session/session.test.ts +++ b/packages/core/src/session/session.test.ts @@ -435,6 +435,100 @@ describe('Session claim', () => { expect(second.session?.ticket).toBe('two'); }); + test('two attempts claiming at once produce exactly one winner', async () => { + const { machineId, session } = await requested('ses-cas-race', 5446); + + // Both launched before either has finished, so this is the case the + // sequential tests cannot reach: two attempts that may each read the + // run as unclaimed before either writes. Postgres holds the second + // update on the row lock until the first commits and then re-checks + // the predicate, so the loser matches nothing. + const [a, b] = await Promise.all([ + Session.transition({ + claimToken: HOLDER, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }), + Session.transition({ + claimToken: RIVAL, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }) + ]); + + const outcomes = [a.outcome, b.outcome]; + // The assertion that matters, and the only one that would catch the + // predicate being weakened: not which one won, but that one did. + expect(outcomes.filter((o) => o === 'moved')).toHaveLength(1); + // Which refusal the loser gets depends on whether its read landed + // before or after the winner's commit, and that is timing. Both mean + // the same thing to an agent — stop, this run is not yours. + expect(['lost', 'notHolder']).toContain(outcomes.find((o) => o !== 'moved')); + expect((await Session.fromID(session.id))?.state).toBe('starting'); + + // And the row holds the winner, not merely somebody: the attempt that + // was told it moved can report again and the other still cannot. + const winner = a.outcome === 'moved' ? HOLDER : RIVAL; + const loser = winner === HOLDER ? RIVAL : HOLDER; + expect( + ( + await Session.transition({ + claimToken: winner, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }) + ).outcome + ).toBe('unchanged'); + expect( + ( + await Session.transition({ + claimToken: loser, + id: session.id, + machineId, + state: 'starting', + errorMessage: null + }) + ).outcome + ).toBe('notHolder'); + }); + + test('the guarded update alone refuses the second claim, without the read', async () => { + const { machineId, session } = await requested('ses-cas-race-raw', 5447); + + // The test above depends on how two transactions interleave, so it can + // pass for the wrong reason on a run where one finishes first. This one + // cannot: it skips the read and fires both guarded updates, so the only + // thing that can refuse the second is the predicate in the `where` + // clause. Separating the check from the write would fail here. + const both = await Promise.all([ + Session.compareAndSetState({ + claimToken: HOLDER, + id: session.id, + machineId, + from: 'requested', + to: 'starting', + errorMessage: null + }), + Session.compareAndSetState({ + claimToken: RIVAL, + id: session.id, + machineId, + from: 'requested', + to: 'starting', + errorMessage: null + }) + ]); + + expect(both.filter((row) => row !== null)).toHaveLength(1); + expect((await Session.fromID(session.id))?.state).toBe('starting'); + }); + test('the claim writes a holder, and the holder never goes out', async () => { const { machineId, session } = await requested('ses-holder', 5440);