From 54d5c81edb2bdf0b872bdf8d153275c3920e9a1f Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sat, 5 Sep 2026 13:02:57 +0300 Subject: [PATCH] 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