import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { newFixture, type Fixture } from '../helpers/fixture.js'; import { randomBytes } from 'node:crypto'; let fixture: Fixture; const inventory = { hostname: 'games-1', agentVersion: '1.0.0', protocolVersion: 1, os: 'Debian 13', architecture: 'amd64', dockerRunning: true, directoriesWritable: true, storageAvailable: 100 * 1024 ** 3, preparedRuntimeImages: ['plutainer:test'], libraries: [{ id: 'test-library', state: 'ready' }] }; beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); }); afterEach(async () => { vi.useRealTimers(); vi.unstubAllEnvs(); await fixture.cleanup(); }); async function enrolled() { const hosts = await import('../../src/lib/server/hosts/index.js'); const host = hosts.createHost({ name: 'Game host' }); const enrollment = hosts.createEnrollment(host.id); return { ...hosts, host, enrollment, credentials: hosts.enrollHost(enrollment.token, inventory) }; } describe('native host enrollment and scoped authentication', () => { it('downloads the installer from the fixed repository and quotes the enrollment token', async () => { const { installerCommand } = await import('../../src/lib/server/hosts/http.js'); expect(installerCommand('https://controller.example/hosts/new', "token'quote")).toBe( "curl --proto '=https' --tlsv1.2 --fail --silent --show-error --connect-timeout 25 --max-time 120 'https://gitea.thewrightserver.net/josh/electric-cherry/raw/branch/main/agent/install.sh' -o electric-cherry-install.sh && sudo bash ./electric-cherry-install.sh --controller 'https://controller.example' --token 'token'\\''quote'" ); }); it.each(['instance.inspect', 'rcon.status-batch'])( 'expires undelivered %s on timeout and delivers the next fresh check', async (kind) => { const api = await enrolled(); const { db } = await import('../../src/lib/server/db/index.js'); api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] }); const stale = api.queueHostOperation(api.host.id, kind, {}); await expect(api.waitHostOperationStatus(stale.id, 0)).rejects.toThrow( 'expired before delivery' ); expect(api.getHostOperationStatus(stale.id)).toMatchObject({ state: 'failed', error: 'Expired before delivery.' }); expect(await api.waitHostOperationStatus(stale.id, 0)).toMatchObject({ state: 'failed' }); expect( db() .prepare( "SELECT state,error FROM ec_host_operation_events WHERE operation_id=? AND state='failed'" ) .all(stale.id) ).toEqual([{ state: 'failed', error: 'Expired before delivery.' }]); const fresh = api.queueHostOperation(api.host.id, kind, {}); expect(api.pendingHostOperations(api.host.id, 'health').map((op) => op.id)).toEqual([ fresh.id ]); } ); it.each(['pending', 'accepted', 'running'])( 'preserves delivered health checks in %s state after the caller times out', async (state) => { const api = await enrolled(); const operation = api.queueHostOperation(api.host.id, 'instance.inspect', {}); api.pendingHostOperations(api.host.id, 'health'); if (state !== 'pending') api.reportHostOperationStatus(api.host.id, operation.id, { state }); await expect(api.waitHostOperationStatus(operation.id, 0)).rejects.toThrow( 'awaiting acknowledgement' ); expect(api.getHostOperationStatus(operation.id)?.state).toBe(state); api.reportHostOperationStatus(api.host.id, operation.id, { state: 'completed', result: { healthy: true } }); expect(await api.waitHostOperationStatus(operation.id)).toMatchObject({ state: 'completed', result: { healthy: true } }); } ); it.each(['instance.restart', 'files.stage', 'rcon.execute'])( 'preserves undelivered %s when its caller times out', async (kind) => { const api = await enrolled(); const operation = api.queueHostOperation(api.host.id, kind, { command: 'status' }); await expect(api.waitHostOperationStatus(operation.id, 0)).rejects.toThrow( 'awaiting acknowledgement' ); expect(api.getHostOperationStatus(operation.id)?.state).toBe('pending'); } ); it('keeps the host online and delivers work when individual state observations are invalid', async () => { const api = await enrolled(); const good = { instanceId: 'healthy-server', generation: 1, state: 'running', healthy: true, observedAt: new Date().toISOString() }; const operation = api.queueHostOperation(api.host.id, 'host.inspect', {}); const host = api.heartbeat(api.host.id, { ...inventory, instanceStates: [ good, { ...good, instanceId: 'crashing', state: 'restarting' }, { ...good, instanceId: 'paused', state: 'paused' }, null, { ...good, generation: 0 }, { ...good, observedAt: 'invalid' } ] }); expect(host.inventory.instanceStates).toEqual([good]); expect(host.readiness).toBe('ready'); expect(api.pendingHostOperations(api.host.id).map((row) => row.id)).toContain(operation.id); }); it('queues serialized launch files and processes receipts without hydrating plugin payloads', async () => { const api = await enrolled(); const { db } = await import('../../src/lib/server/db/index.js'); const payload = { files: [ { path: 'plugins/fixture.dll', contentBase64: Buffer.alloc(8 * 1024 ** 2, 42).toString('base64') } ] }; const { putFileBlob } = await import('../../src/lib/server/db/file-blobs.js'); const hash = putFileBlob(Buffer.from(payload.files[0].contentBase64, 'base64')); // Launch preparation transfers bytes separately and serializes only references. const bytes = Buffer.from( JSON.stringify({ files: [{ path: payload.files[0].path, contentSha256: hash }] }) ); const options = { instanceId: 'immutable-instance', generation: 1, idempotencyKey: 'large-launch' }; const parse = JSON.parse; const parsing = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => { if (value.length > 1024 * 1024) throw new Error('Launch payload was unnecessarily parsed.'); return parse(value, reviver); }); let id: string; try { const operation = api.queueHostOperationSerialized( api.host.id, 'instance.create', bytes, options ); id = operation.id; expect(operation).not.toHaveProperty('payload'); expect( api.queueHostOperationSerialized(api.host.id, 'instance.create', bytes, options).id ).toBe(id); expect(() => api.queueHostOperationSerialized( api.host.id, 'instance.create', Buffer.from('{}'), options ) ).toThrow('different request'); const stored = db() .prepare( 'SELECT typeof(payload) AS kind,count(*) AS count FROM ec_host_operations WHERE idempotency_key=?' ) .get(options.idempotencyKey); expect(stored).toEqual({ kind: 'text', count: 1 }); const waiting = api.waitHostOperationStatus(id); expect( api.reportHostOperationStatus(api.host.id, id, { state: 'accepted' }).state ).toBe('accepted'); expect(api.reportHostOperationStatus(api.host.id, id, { state: 'running' }).state).toBe( 'running' ); api.reportHostOperationStatus(api.host.id, id, { state: 'completed', result: { healthy: true } }); expect(await waiting).toMatchObject({ id, state: 'completed', result: { healthy: true } }); expect(api.reportHostOperationStatus(api.host.id, id, { state: 'running' }).state).toBe( 'completed' ); expect(api.getHostOperationStatus(id)).not.toHaveProperty('payload'); const other = api.createHost({ name: 'Other host' }); expect(() => api.reportHostOperationStatus(other.id, id, { state: 'completed' }) ).toThrow('this host'); } finally { parsing.mockRestore(); } // Full-data APIs retain their original contract for explicit readers. expect(api.getHostOperation(id!)?.payload).toEqual(payload); expect(api.reportHostOperation(api.host.id, id!, { state: 'running' }).payload).toEqual( payload ); expect((await api.waitHostOperation(id!)).payload).toEqual(payload); }); it('keeps serialized launch routing rows small while full-data APIs hydrate separate immutable bytes', async () => { const api = await enrolled(); const { db } = await import('../../src/lib/server/db/index.js'); const routing = { libraryId: 'test-library', name: 't5_zombie_theater_1', game: 't5' }; const payload = { ...routing, files: [{ contentBase64: 'A'.repeat(512 * 1024) }] }; const bytes = Buffer.from(JSON.stringify(payload)); const launch = api.queueHostOperationSerialized( api.host.id, 'instance.create', bytes, {}, routing ); const inline = api.queueHostOperation(api.host.id, 'instance.inspect', { name: 'inline-legacy' }); expect(api.getHostOperation(launch.id)?.payload).toEqual(payload); expect( api.listHostOperations(api.host.id).find((operation) => operation.id === launch.id) ?.payload ).toEqual(payload); expect( api.pendingHostOperations(api.host.id, 'write').map((operation) => operation.payload) ).toEqual([payload]); expect(api.pendingHostOperations(api.host.id, 'read')).toEqual([inline]); expect(api.pendingHostOperationRows(api.host.id, 'write', false)[0]).not.toHaveProperty( 'payload' ); api.reportHostOperationStatus(api.host.id, launch.id, { state: 'accepted' }); api.reportHostOperationStatus(api.host.id, launch.id, { state: 'running', result: { phase: 'starting' } }); const stored = db() .prepare(`SELECT payload,last_delivered_at FROM ec_host_operations WHERE id=?`) .get(launch.id) as { payload: string; last_delivered_at: string | null }; expect(stored.payload).toBe(JSON.stringify(routing)); expect(stored.last_delivered_at).not.toBeNull(); const saved = db() .prepare( 'SELECT length(CAST(payload AS BLOB)) AS bytes FROM ec_host_operation_payloads WHERE operation_id=?' ) .get(launch.id) as { bytes: number }; expect(saved.bytes).toBeLessThan(1024); expect(api.getHostOperationPayload(api.host.id, launch.id)).toEqual(bytes); expect(api.getHostOperationPayload(api.host.id, inline.id).toString('utf8')).toBe( JSON.stringify(inline.payload) ); expect( db() .prepare('SELECT 1 FROM ec_host_operation_payloads WHERE operation_id=?') .get(inline.id) ).toBeUndefined(); }); it('narrows cancellation lookups by kind instead of scanning completed health history', async () => { const api = await enrolled(); api.heartbeat(api.host.id, { ...inventory, capabilities: ['instance.cancel-create', 'operations.health-lane'] }); const { db } = await import('../../src/lib/server/db/index.js'); const handle = db(); const insert = handle.prepare(`INSERT INTO ec_host_operations (id,host_id,kind,payload,state,created_at,updated_at) VALUES(?,?,'instance.inspect','{}','completed',?,?)`); const at = new Date().toISOString(); handle.transaction(() => { for (let i = 0; i < 10_000; i++) insert.run(`history-${i}`, api.host.id, at, at); })(); const inspect = api.queueHostOperation(api.host.id, 'instance.inspect', {}); const prepare = vi.spyOn(handle, 'prepare'); expect(api.pendingHostOperations(api.host.id, 'health').map((op) => op.id)).toEqual([ inspect.id ]); const query = prepare.mock.calls .map(([sql]) => sql) .find((sql) => sql.includes('FROM ec_host_operations o WHERE host_id=?'))!; prepare.mockRestore(); // Assert the actual poll's access path, avoiding hardware-dependent timing thresholds. const plan = handle .prepare(`EXPLAIN QUERY PLAN ${query}`) .all(api.host.id, 1, 1, new Date().toISOString()) as { detail: string }[]; expect(plan.find((row) => row.detail.includes('SEARCH cancellation'))?.detail).toContain( '(host_id=? AND kind=?)' ); }); it.each(['accepted', 'running'])( 'withholds %s health work and replays its original ID after the recovery lease', async (state) => { const api = await enrolled(); vi.useFakeTimers({ toFake: ['Date'] }); const at = Date.now(); api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] }); const busy = api.queueHostOperation(api.host.id, 'instance.inspect', {}); api.pendingHostOperations(api.host.id, 'health'); api.reportHostOperation(api.host.id, busy.id, { state }); const next = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'status' }); expect(api.pendingHostOperations(api.host.id, 'health').map((op) => op.id)).toEqual([ next.id ]); api.reportHostOperation(api.host.id, next.id, { state: 'completed' }); expect(api.pendingHostOperations(api.host.id, 'health')).toEqual([]); vi.setSystemTime(at + 30_001); expect(api.pendingHostOperations(api.host.id, 'health').map((op) => op.id)).toEqual([ busy.id ]); expect(api.pendingHostOperations(api.host.id, 'health')).toEqual([]); api.reportHostOperation(api.host.id, busy.id, { state: 'unknown' }); expect(api.pendingHostOperations(api.host.id, 'health').map((op) => op.id)).toEqual([ busy.id ]); } ); it('reserves health delivery while a file backlog is pending and bounds background batches', async () => { const api = await enrolled(); api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] }); const files = Array.from({ length: 30 }, (_, i) => api.queueHostOperation(api.host.id, 'files.read', { path: `file-${i}` }) ); const status = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'status' }); const inspect = api.queueHostOperation(api.host.id, 'instance.inspect', {}); const mutation = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'map_restart' }); expect(api.pendingHostOperations(api.host.id, 'health').map((op) => op.id)).toEqual([ status.id ]); api.reportHostOperation(api.host.id, status.id, { state: 'completed', result: {} }); expect(api.pendingHostOperations(api.host.id, 'health').map((op) => op.id)).toEqual([ inspect.id ]); expect(api.pendingHostOperations(api.host.id, 'read').map((op) => op.id)).toEqual( files.slice(0, 4).map((op) => op.id) ); expect(api.pendingHostOperations(api.host.id, 'write').map((op) => op.id)).toEqual([ mutation.id ]); // Rolling upgrades: older agents still receive health through read. api.heartbeat(api.host.id, inventory); files.forEach((file) => api.reportHostOperation(api.host.id, file.id, { state: 'completed', result: {} }) ); expect(api.pendingHostOperations(api.host.id, 'read').map((op) => op.id)).toEqual([ inspect.id ]); }); it('separates observation and mutation delivery and wakes long polls on newly queued work', async () => { const api = await enrolled(); const inspect = api.queueHostOperation(api.host.id, 'files.manifest', { roots: ['configs'] }); const restart = api.queueHostOperation(api.host.id, 'instance.restart', {}); const status = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'status' }); const command = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'map_restart' }); expect( api .pendingHostOperations(api.host.id, 'read') .map((op) => op.id) .sort() ).toEqual([inspect.id, status.id].sort()); expect( api .pendingHostOperations(api.host.id, 'write') .map((op) => op.id) .sort() ).toEqual([restart.id, command.id].sort()); // Let earlier enqueue notifications finish before subscribing. await Promise.resolve(); const controller = new AbortController(); const waiting = api.waitForHostWork(api.host.id, controller.signal, 20_000); api.queueHostOperation(api.host.id, 'host.inspect', {}); await waiting; const aborted = api.waitForHostWork(api.host.id, controller.signal, 20_000); controller.abort(); await aborted; }); it('consumes an enrollment exactly once and stores only credential hashes', async () => { const api = await enrolled(); expect(api.authenticateCredential(api.credentials.credential)?.id).toBe(api.host.id); expect(api.authenticateCredential(api.enrollment.token)).toBeNull(); expect(() => api.enrollHost(api.enrollment.token, inventory)).toThrow('invalid or expired'); const { db } = await import('../../src/lib/server/db/index.js'); const stored = JSON.stringify(db().prepare('SELECT * FROM ec_host_credentials').all()) + JSON.stringify(db().prepare('SELECT * FROM ec_host_enrollments').all()); expect(stored).not.toContain(api.credentials.credential); expect(stored).not.toContain(api.enrollment.token); }); it('expires enrollments and invalidates earlier unused enrollment tokens', async () => { const api = await enrolled(); const first = api.createEnrollment(api.host.id); const next = api.createEnrollment(api.host.id, 10); expect(() => api.enrollHost(first.token, inventory)).toThrow('invalid or expired'); vi.useFakeTimers(); vi.setSystemTime(Date.now() + 20); expect(() => api.enrollHost(next.token, inventory)).toThrow('invalid or expired'); }); it('rotates credentials independently and immediately revokes access', async () => { const api = await enrolled(); const other = api.createHost({ name: 'Other' }); const second = api.enrollHost(api.createEnrollment(other.id).token, inventory); const rotated = api.rotateCredential(api.host.id); expect(api.authenticateCredential(api.credentials.credential)).toBeNull(); expect(api.authenticateCredential(rotated)?.id).toBe(api.host.id); api.revokeCredentials(api.host.id); expect(api.authenticateCredential(rotated)).toBeNull(); expect(api.authenticateCredential(second.credential)?.id).toBe(other.id); }); it('requires HTTPS and blocks cross-origin management and agent credentials on management APIs', async () => { const api = await enrolled(); const { authenticateAgent, requireManagementOrigin } = await import('../../src/lib/server/hosts/http.js'); const headers = { authorization: `Bearer ${api.credentials.credential}` }; expect( authenticateAgent(new Request('https://ec.test/api/agent/operations', { headers })).id ).toBe(api.host.id); expect(() => authenticateAgent(new Request('http://ec.test/api/agent/operations', { headers })) ).toThrow(); expect(() => authenticateAgent( new Request('https://ec.test/api/agent/operations', { headers: { authorization: `Bearer ${api.enrollment.token}` } }) ) ).toThrow(); expect(() => requireManagementOrigin(new Request('https://ec.test/api/hosts', { headers })) ).toThrow(); expect(() => requireManagementOrigin( new Request('https://ec.test/api/hosts', { headers: { Origin: 'https://attacker.test' } }) ) ).toThrow(); }); it('recovers credential rotation when acknowledgement is lost using the agent-persisted replacement', async () => { const api = await enrolled(); const next = 'ec_host_' + randomBytes(32).toString('base64url'); expect(api.rotateCredential(api.host.id, next)).toBe(next); expect(api.authenticateCredential(api.credentials.credential)).toBeNull(); expect(api.authenticateCredential(next)?.id).toBe(api.host.id); expect(api.rotateCredential(api.host.id, next)).toBe(next); api.revokeCredentials(api.host.id); expect(() => api.rotateCredential(api.host.id, next)).toThrow('cannot be reused'); }); it('bounds chunked request bodies before consuming the complete upload', async () => { const { objectBody } = await import('../../src/lib/server/hosts/http.js'); let cancelled = false; const body = new ReadableStream({ pull(controller) { controller.enqueue(new TextEncoder().encode('x'.repeat(100))); }, cancel() { cancelled = true; } }); const request = new Request('https://ec.test/api/agent/operations/x', { method: 'POST', body, duplex: 'half' } as RequestInit); await expect(objectBody(request, 50)).rejects.toMatchObject({ status: 413 }); expect(cancelled).toBe(true); }); }); describe('native host readiness and reservations', () => { it.each([ { preparedRuntimeImages: 'plutainer:test' }, { preparedRuntimeImages: [123] }, { preparedRuntimeImages: null } ])( 'rejects malformed prepared runtime inventory: $preparedRuntimeImages', async ({ preparedRuntimeImages }) => { const api = await enrolled(); expect(() => api.heartbeat(api.host.id, { ...inventory, preparedRuntimeImages }) ).toThrow(/runtime image inventory/i); expect(api.requireReadyHost(api.host.id).readiness).toBe('ready'); } ); it('separates a connected agent from ready Docker, storage, runtime and library state', async () => { const api = await enrolled(); expect(api.requireReadyHost(api.host.id).readiness).toBe('ready'); expect(() => api.requireReadyHost(api.host.id, { image: 'missing:image' })).toThrow( 'runtime' ); expect(() => api.requireReadyHost(api.host.id, { libraryId: 'missing' })).toThrow( 'library' ); expect(api.heartbeat(api.host.id, { ...inventory, libraries: [] }).readiness).toBe( 'needs-game-files' ); const host = api.heartbeat(api.host.id, { ...inventory, dockerRunning: false }); expect(host.connected).toBe(true); expect(host.readiness).toBe('degraded'); expect(() => api.requireReadyHost(host.id)).toThrow('Docker'); api.heartbeat(host.id, inventory); vi.useFakeTimers(); vi.setSystemTime(Date.now() + 91_000); expect(api.getHost(host.id)?.connectionState).toBe('offline'); expect(() => api.requireReadyHost(host.id)).toThrow('stale'); }); it.each([ { missing: { libraries: [] }, expected: 'needs-game-files' }, { missing: { preparedRuntimeImages: [] }, expected: 'needs-runtime' }, { missing: { preparedRuntimeImages: [], libraries: [] }, expected: 'needs-runtime' } ])( 'names missing content as $expected until a prepared heartbeat arrives', async ({ missing, expected }) => { const api = await enrolled(); const host = api.heartbeat(api.host.id, { ...inventory, ...missing }); expect(host.connected).toBe(true); expect(host.readiness).toBe(expected); expect(() => api.requireReadyHost(host.id)).toThrow(); expect(api.heartbeat(host.id, inventory).readiness).toBe('ready'); expect(api.requireReadyHost(host.id).id).toBe(host.id); } ); it('keeps setup work deliverable while new instance creation waits for missing game files', async () => { const api = await enrolled(); const create = api.queueHostOperation(api.host.id, 'instance.create', {}); api.heartbeat(api.host.id, { ...inventory, libraries: [] }); expect(() => api.queueHostOperation(api.host.id, 'instance.create', {})).toThrow( 'game-file library' ); const download = api.queueHostOperation(api.host.id, 'library.download', {}); const runtime = api.queueHostOperation(api.host.id, 'runtime.pull', {}); expect(api.pendingHostOperations(api.host.id).map((operation) => operation.id)).toEqual([ download.id, runtime.id ]); api.heartbeat(api.host.id, { ...inventory, preparedRuntimeImages: [], libraries: [] }); expect(api.pendingHostOperations(api.host.id).map((operation) => operation.id)).toEqual([ download.id, runtime.id ]); api.heartbeat(api.host.id, inventory); expect(api.pendingHostOperations(api.host.id).map((operation) => operation.id)).toContain( create.id ); }); it('gives enrollment, infrastructure failure and host modes priority over missing content', async () => { const api = await enrolled(); expect(api.createHost({ name: 'New host' }).readiness).toBe('preparing'); expect( api.heartbeat(api.host.id, { ...inventory, libraries: [], dockerRunning: false }) .readiness ).toBe('degraded'); api.heartbeat(api.host.id, { ...inventory, libraries: [] }); expect(api.updateHost(api.host.id, { mode: 'draining' }).readiness).toBe('draining'); expect(api.updateHost(api.host.id, { mode: 'disabled' }).readiness).toBe('disabled'); }); it('blocks new deployments on a draining or disabled host without altering existing operations', async () => { const api = await enrolled(); const existing = api.queueHostOperation( api.host.id, 'instance.inspect', {}, { instanceId: 'immutable-id', generation: 1 } ); api.updateHost(api.host.id, { mode: 'draining' }); expect(() => api.queueHostOperation(api.host.id, 'instance.create', {})).toThrow( 'draining' ); expect(api.getHostOperation(existing.id)?.state).toBe('pending'); api.updateHost(api.host.id, { mode: 'disabled' }); expect(api.pendingHostOperations(api.host.id)).toEqual([]); }); it('allocates per host, bind address and protocol while protecting wildcard bindings', async () => { const api = await enrolled(); const second = api.createHost({ name: 'Other' }); expect(api.reserveHostPort(api.host.id, 'a', { port: 28960 })).toBe(28960); expect(api.reserveHostPort(second.id, 'b', { port: 28960 })).toBe(28960); expect(api.reserveHostPort(api.host.id, 'c', { port: 28960, protocol: 'tcp' })).toBe(28960); expect(() => api.reserveHostPort(api.host.id, 'd', { port: 28960, bindAddress: '127.0.0.1' }) ).toThrow('available'); expect(api.reserveHostPort(api.host.id, 'a', { port: 28960 })).toBe(28960); api.releaseHostPorts('a'); expect( api.reserveHostPort(api.host.id, 'e', { port: 28960, bindAddress: '127.0.0.1' }) ).toBe(28960); expect( api.reserveHostPort(api.host.id, 'f', { port: 28960, bindAddress: '127.0.0.2' }) ).toBe(28960); expect(() => api.reserveHostPort(api.host.id, 'g', { port: 28960, bindAddress: '0.0.0.0' }) ).toThrow('available'); expect(() => api.reserveHostPort(api.host.id, 'h', { port: 28000 })).toThrow('available'); }); }); describe('durable host operations', () => { it('drains an old agent for a cancellation update while allowing receipts, heartbeats, and other hosts', async () => { const api = await enrolled(); const original = api.queueHostOperation( api.host.id, 'instance.create', {}, { instanceId: 'cancelled-instance', generation: 1 } ); const inspect = api.queueHostOperation(api.host.id, 'instance.inspect', {}); expect(api.pendingHostOperations(api.host.id).map((op) => op.id)).toContain(original.id); api.reportHostOperation(api.host.id, original.id, { state: 'running' }); api.reportHostOperation(api.host.id, inspect.id, { state: 'unknown' }); const cleanup = api.queueHostOperation( api.host.id, 'instance.cancel-create', { createOperationId: original.id }, { instanceId: 'cancelled-instance', generation: 1 } ); const logs = api.queueHostOperation(api.host.id, 'logs.read', {}); const preparation = api.queueHostOperation(api.host.id, 'library.download', {}); const other = api.createHost({ name: 'Unaffected host' }); api.enrollHost(api.createEnrollment(other.id).token, inventory); const otherWork = api.queueHostOperation(other.id, 'host.inspect', {}); const { db } = await import('../../src/lib/server/db/index.js'); const before = db() .prepare( 'SELECT id,state,last_delivered_at FROM ec_host_operations WHERE host_id=? ORDER BY id' ) .all(api.host.id); for (let poll = 0; poll < 3; poll++) expect(api.pendingHostOperations(api.host.id)).toEqual([]); expect( db() .prepare( 'SELECT id,state,last_delivered_at FROM ec_host_operations WHERE host_id=? ORDER BY id' ) .all(api.host.id) ).toEqual(before); expect(api.pendingHostOperations(other.id).map((op) => op.id)).toEqual([otherWork.id]); api.reportHostOperation(api.host.id, inspect.id, { state: 'completed', result: { healthy: true } }); api.heartbeat(api.host.id, inventory); expect(api.pendingHostOperations(api.host.id)).toEqual([]); // The controller hold must clear on capability, without waiting for cleanup // or the original creation to finish (which would recreate the deadlock). api.heartbeat(api.host.id, { ...inventory, capabilities: ['instance.cancel-create'] }); expect(api.pendingHostOperations(api.host.id).map((op) => op.id)).toEqual([ cleanup.id, logs.id, preparation.id ]); expect(api.getHostOperation(original.id)?.state).toBe('running'); expect(api.getHostOperation(cleanup.id)?.state).toBe('pending'); }); it.each(['completed', 'failed'] as const)( 'does not pause an old host for a terminal %s cancellation', async (state) => { const api = await enrolled(); const cleanup = api.queueHostOperation(api.host.id, 'instance.cancel-create', { createOperationId: 'creation' }); api.reportHostOperation(api.host.id, cleanup.id, { state }); const work = api.queueHostOperation(api.host.id, 'host.inspect', {}); expect(api.pendingHostOperations(api.host.id).map((op) => op.id)).toEqual([work.id]); } ); it('rotates unresolved operation delivery so an old batch cannot starve new work', async () => { const api = await enrolled(); const operations = Array.from({ length: 30 }, () => api.queueHostOperation(api.host.id, 'instance.inspect', {}) ); operations.forEach((operation) => api.reportHostOperation(api.host.id, operation.id, { state: 'unknown' }) ); const first = api.pendingHostOperations(api.host.id).map((operation) => operation.id); const second = api.pendingHostOperations(api.host.id).map((operation) => operation.id); expect(new Set([...first, ...second]).size).toBe(30); expect( operations.every((operation) => api.getHostOperation(operation.id)?.state === 'unknown') ).toBe(true); }); it('reconciles the same operation ID after loss of acknowledgement and prevents duplicate side effects', async () => { const api = await enrolled(); const options = { instanceId: 'immutable-id', generation: 4, idempotencyKey: 'provisioning:123' }; const operation = api.queueHostOperation(api.host.id, 'instance.restart', {}, options); api.reportHostOperation(api.host.id, operation.id, { state: 'accepted' }); api.reportHostOperation(api.host.id, operation.id, { state: 'running' }); expect(api.queueHostOperation(api.host.id, 'instance.restart', {}, options).id).toBe( operation.id ); expect(api.pendingHostOperations(api.host.id).map((value) => value.id)).toEqual([ operation.id ]); expect(() => api.queueHostOperation(api.host.id, 'instance.stop', {}, options)).toThrow( 'different request' ); api.reportHostOperation(api.host.id, operation.id, { state: 'completed', result: { restarted: true } }); expect(api.reportHostOperation(api.host.id, operation.id, { state: 'running' }).state).toBe( 'completed' ); expect(api.pendingHostOperations(api.host.id)).toEqual([]); expect((await api.waitHostOperation(operation.id)).result).toEqual({ restarted: true }); }); it('does not allow another host to report a result, and timeouts retain the original operation', async () => { const api = await enrolled(); const other = api.createHost({ name: 'Other' }); const operation = api.queueHostOperation(api.host.id, 'instance.stop', {}); expect(() => api.reportHostOperation(other.id, operation.id, { state: 'completed' }) ).toThrow('this host'); await expect(api.waitHostOperation(operation.id, 0)).rejects.toThrow(operation.id); expect(api.getHostOperation(operation.id)?.state).toBe('pending'); }); }); it('limits preparing hosts at delivery, preserves replays, and releases capacity after completion', async () => { const api = await enrolled(); const second = api.createHost({ name: 'Second' }); api.enrollHost(api.createEnrollment(second.id).token, inventory); const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js'); saveFleetSettings({ ...fleetSettings(), preparationConcurrency: 1 }); const first = api.queueHostOperation(api.host.id, 'library.download', { libraryVersionId: 'one' }); const next = api.queueHostOperation(second.id, 'library.download', { libraryVersionId: 'two' }); const inspect = api.queueHostOperation(second.id, 'host.inspect', {}); expect(api.pendingHostOperations(api.host.id).map((row) => row.id)).toContain(first.id); expect(api.pendingHostOperations(second.id).map((row) => row.id)).not.toContain(next.id); expect(api.pendingHostOperations(second.id).map((row) => row.id)).toContain(inspect.id); expect(api.pendingHostOperations(api.host.id).map((row) => row.id)).toContain(first.id); api.reportHostOperation(api.host.id, first.id, { state: 'completed', result: {} }); expect(api.pendingHostOperations(second.id).map((row) => row.id)).toContain(next.id); });