diff --git a/apps/api/app/routes/machine.ts b/apps/api/app/routes/machine.ts index 60ee408a..09adae39 100644 --- a/apps/api/app/routes/machine.ts +++ b/apps/api/app/routes/machine.ts @@ -1,10 +1,11 @@ import { Actor } from '@nestri/core/actor'; +import { Box } from '@nestri/core/box/index'; import { ErrorCodes, VisibleError } from '@nestri/core/error'; import { Examples } from '@nestri/core/examples'; import { Identifier } from '@nestri/core/id'; import { Machine } from '@nestri/core/machine/index'; -import { Member } from '@nestri/core/team/member'; import { Team } from '@nestri/core/team/index'; +import { Member } from '@nestri/core/team/member'; import { Hono } from 'hono'; import { describeRoute } from 'hono-openapi'; import { z } from 'zod'; @@ -260,6 +261,82 @@ export namespace MachineApi { }); } ) + .post( + '/report', + machineOnly, + describeRoute({ + tags: ['Machine'], + summary: 'Say what the host is running', + description: + 'Records one full snapshot of the boxes on the calling host. Separate from the beat because the two have different loss tolerance: a dropped report is corrected by the next one, where a dropped beat moves a host towards offline. Send one when a box changes lifecycle, and send one anyway every so often so a single lost snapshot cannot leave this record permanently wrong. Never send a delta — a retrying agent cannot promise ordering, and out-of-order deltas describe a host that never existed.', + responses: { + 200: { + content: { 'application/json': { schema: Result(Box.ReportOutcome) } }, + description: 'The snapshot was recorded' + }, + 400: ErrorResponses[400], + 403: ErrorResponses[403], + 404: ErrorResponses[404] + } + }), + validator( + 'json', + z + .object({ + agentPid: z.number().int().meta({ + description: 'The reporting agent’s own process id, in its own namespace' + }), + boxesKnown: z.number().int().meta({ description: 'How many boxes the host holds' }), + boxesRunning: z.number().int().meta({ + description: 'How many of them are running' + }), + boxes: z.array(Box.Reported).meta({ + description: 'Every box the host holds. A full snapshot, never a delta' + }) + }) + // Strict, so a field this cannot act on is a validation error a + // host operator sees rather than one quietly dropped. Capacity + // belongs here eventually and it has no honest fields yet; + // refusing the ones nobody measures is how it stays that way. + .strict() + ), + async (c) => { + const machine = await Machine.fromID(Actor.machineID); + if (!machine) { + // Same answer as the beat gives, for the same reason: the + // credentials authenticated but the row is gone, and a host + // must re-register rather than keep reporting into nothing. + throw new VisibleError( + 'not_found', + ErrorCodes.NotFound.RESOURCE_NOT_FOUND, + 'This machine no longer exists' + ); + } + + const { boxes } = c.req.valid('json'); + const outcome = await Box.applyHostReport({ machineId: Actor.machineID, boxes }); + + // `agentPid`, `boxesKnown` and `boxesRunning` are read and not + // stored. They are a summary of the list that follows them, and a + // stored copy is a second answer to a question the list already + // answers — one that goes stale the first time the two disagree. + // They stay on the wire because a host that cannot enumerate its + // boxes can still say how many it has. + if (outcome.unknown.length > 0) { + // Loudly, per the contract this endpoint is built to: a host + // holding boxes nobody placed there is a bug to surface, not a + // state to reconcile quietly. Nothing is created for them. + // eslint-disable-next-line no-console + console.warn( + 'host report named boxes that are not placed here:', + Actor.machineID, + outcome.unknown.join(', ') + ); + } + + return c.json({ data: outcome }); + } + ) .get( '/me', machineOnly, diff --git a/apps/api/test/report.test.ts b/apps/api/test/report.test.ts new file mode 100644 index 00000000..39550997 --- /dev/null +++ b/apps/api/test/report.test.ts @@ -0,0 +1,183 @@ +import { afterAll, describe, expect, test } from 'bun:test'; + +import { Box } from '@nestri/core/box/index'; +import { Fixtures } from '@nestri/core/db/fixtures'; +import { testDb } from '@nestri/core/db/test'; +import { Identifier } from '@nestri/core/id'; +import { Machine } from '@nestri/core/machine/index'; + +import { app } from '../app/index'; +import './setup'; + +const sql = testDb(); + +const createdUserIds: string[] = []; + +async function registeredHost(label: string) { + const owner = await Fixtures.owner(label); + createdUserIds.push(owner.userId); + const registered = await Machine.register({ + id: Identifier.ascending('machine'), + ownerUserId: owner.userId, + teamId: owner.teamId, + label + }); + return { + id: registered.id, + userId: owner.userId, + headers: { + 'x-nestri-machine-id': registered.id, + 'x-nestri-machine-secret': registered.secret, + 'content-type': 'application/json' + } + }; +} + +async function boxOn(host: { id: string; userId: string }, label: string) { + return Box.create({ + id: Identifier.ascending('box'), + userId: host.userId, + machineId: host.id, + label, + tier: 'sm' + }); +} + +function report(host: { headers: Record }, body: unknown) { + return app.request('/machine/report', { + method: 'POST', + headers: host.headers, + body: JSON.stringify(body) + }); +} + +afterAll(async () => { + if (createdUserIds.length > 0) { + await sql`delete from "box" where user_id in ${sql(createdUserIds)}`; + await sql`delete from "user" where id in ${sql(createdUserIds)}`; + createdUserIds.length = 0; + } +}); + +describe('POST /machine/report', () => { + test('the body is the shape the agent sends, flat and camelCased', async () => { + const host = await registeredHost('report-shape'); + const running = await boxOn(host, 'one'); + const idle = await boxOn(host, 'two'); + + // Written from the contract and not from the handler: host fields flat at + // the top, boxes under `boxes`, the lifecycle tag flattened into each box, + // and `uptimeS` rather than `uptime_s`. A rename on either side of this + // seam produces a host that reports, is understood by nothing, and is + // told it succeeded. + const res = await report(host, { + agentPid: 4711, + boxesKnown: 2, + boxesRunning: 1, + boxes: [ + { boxId: running.id, tier: 'sm', state: 'running', pid: 1234, uptimeS: 45 }, + { boxId: idle.id, tier: 'sm', state: 'created' } + ] + }); + + expect(res.status).toBe(200); + const body = (await res.json()) as any; + expect(body.data.recorded).toBe(2); + expect(body.data.unknown).toEqual([]); + expect((await Box.fromID(running.id))!.state).toBe('running'); + }); + + test('a request is not wrapped in an envelope, but the response is', async () => { + const host = await registeredHost('report-envelope'); + + const wrapped = await report(host, { + data: { agentPid: 1, boxesKnown: 0, boxesRunning: 0, boxes: [] } + }); + expect(wrapped.status).toBe(400); + + const flat = await report(host, { + agentPid: 1, + boxesKnown: 0, + boxesRunning: 0, + boxes: [] + }); + expect(flat.status).toBe(200); + expect(await flat.json()).toHaveProperty('data'); + }); + + test('a stop carries its reason all the way to the row', async () => { + const host = await registeredHost('report-stop'); + const box = await boxOn(host, 'faulty'); + + await report(host, { + agentPid: 1, + boxesKnown: 1, + boxesRunning: 0, + boxes: [ + { boxId: box.id, tier: 'sm', state: 'stopped', reason: 'guest exited 1', clean: false } + ] + }); + + const row = (await Box.fromID(box.id))!; + expect(row.state).toBe('stopped'); + expect(row.stopReason).toBe('guest exited 1'); + expect(row.stopClean).toBe(false); + }); + + test('a host cannot report on another host’s boxes', async () => { + // The access rule that matters here. A machine credential is a long-lived + // secret on hardware in somebody's living room, so the scope is in the + // query and not in the agent asking politely about its own boxes. + const mine = await registeredHost('report-mine'); + const theirs = await registeredHost('report-theirs'); + const victim = await boxOn(theirs, 'not yours'); + await Box.setState({ id: victim.id, state: 'running' }); + + const res = await report(mine, { + agentPid: 1, + boxesKnown: 1, + boxesRunning: 1, + boxes: [{ boxId: victim.id, tier: 'sm', state: 'stopped', reason: 'mine now', clean: false }] + }); + + expect(res.status).toBe(200); + expect(((await res.json()) as any).data.unknown).toEqual([victim.id]); + expect((await Box.fromID(victim.id))!.state).toBe('running'); + }); + + test('a field nobody can act on is refused rather than dropped', async () => { + const host = await registeredHost('report-strict'); + + // Capacity belongs on this call eventually and has no honest fields yet. + // Refusing an invented one is how an unmeasured number stays out of a + // record a placement decision will one day read. + const res = await report(host, { + agentPid: 1, + boxesKnown: 0, + boxesRunning: 0, + boxes: [], + gpusFree: 4 + }); + expect(res.status).toBe(400); + }); + + test('wrong credentials are indistinguishable from none', async () => { + const host = await registeredHost('report-wrongsecret'); + const body = JSON.stringify({ agentPid: 1, boxesKnown: 0, boxesRunning: 0, boxes: [] }); + + const none = await app.request('/machine/report', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body + }); + const wrong = await app.request('/machine/report', { + method: 'POST', + headers: { ...host.headers, 'x-nestri-machine-secret': 'msk_wrong' }, + body + }); + + expect(wrong.status).toBe(403); + expect(none.status).toBe(403); + expect(await wrong.json()).toEqual(await none.json()); + }); +}); diff --git a/packages/core/src/box/box.test.ts b/packages/core/src/box/box.test.ts index e28039b5..533598f9 100644 --- a/packages/core/src/box/box.test.ts +++ b/packages/core/src/box/box.test.ts @@ -139,3 +139,137 @@ describe('Box', () => { expect((await Box.listByMachine(machineB)).map((b) => b.label)).toEqual(['b1']); }); }); + +describe('Box.applyHostReport', () => { + async function placed(label: string, count: number) { + const owner = await newOwner(label); + const machineId = await Fixtures.machine(owner); + const boxes = []; + for (let i = 0; i < count; i++) { + boxes.push( + await Box.create({ + id: Identifier.ascending('box'), + userId: owner.userId, + machineId, + label: `${label}-${i}`, + tier: 'sm' + }) + ); + } + return { owner, machineId, boxes }; + } + + test('a snapshot moves the boxes it names', async () => { + const { machineId, boxes } = await placed('report-moves', 2); + + const outcome = await Box.applyHostReport({ + machineId, + boxes: [ + { boxId: boxes[0]!.id, tier: 'sm', state: 'running', pid: 1234, uptimeS: 45 }, + { boxId: boxes[1]!.id, tier: 'sm', state: 'stopped', reason: 'guest exited 0', clean: true } + ] + }); + + expect(outcome.recorded).toBe(2); + expect(outcome.unknown).toEqual([]); + expect(outcome.markedStopped).toEqual([]); + + expect((await Box.fromID(boxes[0]!.id))!.state).toBe('running'); + const stopped = (await Box.fromID(boxes[1]!.id))!; + expect(stopped.state).toBe('stopped'); + expect(stopped.stopReason).toBe('guest exited 0'); + expect(stopped.stopClean).toBe(true); + }); + + test('pid and uptime are accepted and not stored', async () => { + const { machineId, boxes } = await placed('report-drops', 1); + + await Box.applyHostReport({ + machineId, + boxes: [{ boxId: boxes[0]!.id, tier: 'sm', state: 'running', pid: 4711, uptimeS: 900 }] + }); + + // The columns do not exist, so what this pins is that the row is still + // readable and carries nothing invented in the two nullable columns it + // does have. + const row = (await Box.fromID(boxes[0]!.id))!; + expect(row.state).toBe('running'); + expect(row.stopReason).toBeNull(); + expect(row.stopClean).toBeNull(); + }); + + test('a running box the snapshot omits is stopped, and says why', async () => { + const { machineId, boxes } = await placed('report-omits', 2); + await Box.setState({ id: boxes[0]!.id, state: 'running' }); + await Box.setState({ id: boxes[1]!.id, state: 'running' }); + + const outcome = await Box.applyHostReport({ + machineId, + boxes: [{ boxId: boxes[0]!.id, tier: 'sm', state: 'running', uptimeS: 5 }] + }); + + expect(outcome.markedStopped).toEqual([boxes[1]!.id]); + const gone = (await Box.fromID(boxes[1]!.id))!; + expect(gone.state).toBe('stopped'); + expect(gone.stopReason).toBe(Box.OMITTED_FROM_REPORT); + expect(gone.stopClean).toBe(false); + }); + + test('a created box the snapshot omits is left alone', async () => { + // The ordinary path, not a divergence: a person creates a box here before + // its host has been told anything about it. + const { machineId, boxes } = await placed('report-created', 1); + + const outcome = await Box.applyHostReport({ machineId, boxes: [] }); + + expect(outcome.markedStopped).toEqual([]); + expect((await Box.fromID(boxes[0]!.id))!.state).toBe('created'); + }); + + test('an empty snapshot from a host holding nothing stops only what was running', async () => { + const { machineId, boxes } = await placed('report-empty', 2); + await Box.setState({ id: boxes[0]!.id, state: 'running' }); + + const outcome = await Box.applyHostReport({ machineId, boxes: [] }); + + expect(outcome.recorded).toBe(0); + expect(outcome.markedStopped).toEqual([boxes[0]!.id]); + expect((await Box.fromID(boxes[1]!.id))!.state).toBe('created'); + }); + + test('a box the snapshot names that is not placed here is never created', async () => { + const { machineId } = await placed('report-unknown', 0); + const invented = Identifier.ascending('box'); + + const outcome = await Box.applyHostReport({ + machineId, + boxes: [{ boxId: invented, tier: 'lg', state: 'running', uptimeS: 1 }] + }); + + expect(outcome.unknown).toEqual([invented]); + expect(outcome.recorded).toBe(0); + expect(await Box.fromID(invented)).toBeNull(); + }); + + test('a snapshot cannot move a box placed on another host', async () => { + const mine = await placed('report-mine', 1); + const theirs = await placed('report-theirs', 1); + await Box.setState({ id: theirs.boxes[0]!.id, state: 'running' }); + + const outcome = await Box.applyHostReport({ + machineId: mine.machineId, + boxes: [ + { + boxId: theirs.boxes[0]!.id, + tier: 'sm', + state: 'stopped', + reason: 'mine now', + clean: false + } + ] + }); + + expect(outcome.unknown).toEqual([theirs.boxes[0]!.id]); + expect((await Box.fromID(theirs.boxes[0]!.id))!.state).toBe('running'); + }); +}); diff --git a/packages/core/src/box/index.ts b/packages/core/src/box/index.ts index bfcd36c7..906421e7 100644 --- a/packages/core/src/box/index.ts +++ b/packages/core/src/box/index.ts @@ -1,4 +1,4 @@ -import { and, eq, isNull, sql } from 'drizzle-orm'; +import { and, eq, isNull, notInArray, sql } from 'drizzle-orm'; import z from 'zod'; import { Database } from '../db/index.js'; @@ -189,10 +189,157 @@ export namespace Box { export const remove = fn(Info.shape.id, async (id) => { await Database.use(async (tx) => { - await tx.update(BoxTable).set({ timeDeleted: sql`now()` }).where(eq(BoxTable.id, id)); + await tx + .update(BoxTable) + .set({ timeDeleted: sql`now()` }) + .where(eq(BoxTable.id, id)); }); }); + /** + * What a host says about one box, in the shape the wire carries. + * + * Flat rather than nested: the agent's own type is an internally-tagged + * enum, and flattening it here is what lets one box be one object on both + * sides. `pid` and `uptimeS` are accepted and deliberately not stored — + * a pid is a number in another machine's namespace and means nothing here, + * and uptime is derivable from the run's `timeStarted`, which is already + * kept and already trustworthy. Storing a plausible number instead of a + * measured one is how a scheduler learns to trust a field nobody produced. + */ + export const Reported = z.discriminatedUnion('state', [ + z.object({ + boxId: z.string(), + tier: z.enum(BoxTier.enumValues), + state: z.literal('created') + }), + z.object({ + boxId: z.string(), + tier: z.enum(BoxTier.enumValues), + state: z.literal('running'), + pid: z.number().int().optional(), + uptimeS: z.number().int() + }), + z.object({ + boxId: z.string(), + tier: z.enum(BoxTier.enumValues), + state: z.literal('stopped'), + reason: z.string(), + clean: z.boolean() + }) + ]); + + export type Reported = z.infer; + + /** What a snapshot changed, and what it disagreed with us about. */ + export const ReportOutcome = z.object({ + recorded: z.number().meta({ description: 'Boxes in the snapshot that we know and updated' }), + unknown: z.array(z.string()).meta({ + description: 'Boxes the host is holding that are not placed here. Recorded, never created' + }), + markedStopped: z.array(z.string()).meta({ + description: 'Boxes placed here that the snapshot did not mention, and are now stopped' + }) + }); + + export type ReportOutcome = z.infer; + + /** + * Written on a box its host did not mention. + * + * A sentence rather than a code because it is read by whoever is looking at + * a box that stopped for no reason they can see, and "the host stopped + * mentioning it" is the fact they need. + */ + export const OMITTED_FROM_REPORT = 'not in its host’s last report'; + + /** + * Record one full snapshot of what a host is running. + * + * Scoped to the calling machine in the `where` clause and not by trusting + * the ids in the body: a machine credential is a long-lived secret on + * hardware in somebody's living room, and a snapshot naming another host's + * boxes must move nothing. + * + * Three rules, and the second two are why this is one function rather than + * a loop of `setState` in a route: + * + * - A box we know, that the snapshot names, takes the reported state. + * - A box we know, that the snapshot omits, is stopped — absence inside a + * received snapshot is information. (Silence from the host is not, and is + * not this function's input at all: no report means this is never called.) + * - A box the snapshot names that is not placed here is **not created**. An + * agent that can conjure a row is an agent that can mint owned resources, + * and a box belongs to somebody this snapshot cannot name. It is returned + * as a divergence to be surfaced loudly instead. + * + * One transaction, because a snapshot is one observation: applying half of + * it leaves a state the host was never in. + */ + export const applyHostReport = fn( + z.object({ machineId: Info.shape.machineId, boxes: z.array(Reported) }), + async (input): Promise => { + return Database.transaction(async (tx) => { + const placed = await tx + .select({ id: BoxTable.id }) + .from(BoxTable) + .where(and(eq(BoxTable.machineId, input.machineId), isNull(BoxTable.timeDeleted))) + .then((rows) => new Set(rows.map((row) => row.id))); + + const seen: string[] = []; + const unknown: string[] = []; + for (const box of input.boxes) { + if (!placed.has(box.boxId)) { + unknown.push(box.boxId); + continue; + } + seen.push(box.boxId); + const stopped = box.state === 'stopped'; + await tx + .update(BoxTable) + .set({ + state: box.state, + stopReason: stopped ? box.reason : null, + stopClean: stopped ? box.clean : null + }) + .where(and(eq(BoxTable.id, box.boxId), eq(BoxTable.machineId, input.machineId))); + } + + // Everything placed here that the snapshot did not mention, and + // that we believed was running. + // + // **`running` and not "anything not stopped"**, which is narrower + // than it first looks it should be. A box is created here, by a + // person, before its host has ever been told about it — so between + // creation and the job that starts it there is a `created` box the + // host correctly does not mention, and stopping it on that basis + // would break the ordinary path rather than catch a divergence. A + // box we were told was running and that has since vanished from its + // host's own inventory is the real disagreement, and it is the one + // that leaves a person looking at a box nothing is running. + // + // `notInArray` on an empty list matches nothing in SQL rather than + // everything, so the empty snapshot — a host that has just come up + // holding no boxes — is spelt out instead of relying on that. + const missing = await tx + .update(BoxTable) + .set({ state: 'stopped', stopReason: OMITTED_FROM_REPORT, stopClean: false }) + .where( + and( + eq(BoxTable.machineId, input.machineId), + isNull(BoxTable.timeDeleted), + eq(BoxTable.state, 'running'), + seen.length > 0 ? notInArray(BoxTable.id, seen) : undefined + ) + ) + .returning({ id: BoxTable.id }) + .then((rows) => rows.map((row) => row.id)); + + return { recorded: seen.length, unknown, markedStopped: missing }; + }); + } + ); + export function serialize(input: typeof BoxTable.$inferSelect): z.infer { return { id: input.id,