import { setImmediate } from 'node:timers/promises'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { newFixture, type Fixture } from '../helpers/fixture.js'; vi.mock('../../src/lib/server/hosts/preparation.js', () => ({ reconcileHostPreparation: vi.fn() })); let fixture: Fixture; const inventory = { protocolVersion: 1, dockerRunning: true, directoriesWritable: true, storageAvailable: 100 * 1024 ** 3, runtimeImages: ['plutainer:test'], libraries: [{ id: 'test-library', state: 'ready' }], capabilities: ['instance.cancel-create', 'operations.health-lane'] }; beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); }); afterEach(async () => { vi.restoreAllMocks(); vi.useRealTimers(); await fixture.cleanup(); }); async function enrolled() { const hosts = await import('../../src/lib/server/hosts/index.js'); const host = hosts.createHost({ name: 'Delivery host' }); const credentials = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory); const { GET } = await import('../../src/routes/api/agent/operations/+server.js'); const request = (query = '') => new Request(`https://ec.test/api/agent/operations${query}`, { headers: { authorization: `Bearer ${credentials.credential}` } }); return { ...hosts, host, GET, request }; } describe('agent operation delivery', () => { it('streams large persisted launch files without parsing payloads and yields to controller work', async () => { const api = await enrolled(); const payload = { // Place a Unicode character across the first payload byte chunk. value: 'x'.repeat(64 * 1024 - '{"value":"'.length - 1) + '🧟 café "quoted"\n', files: [{ path: 'plugins/runtime.dll', contentBase64: 'A'.repeat(8 * 1024 ** 2) }] }; const operation = api.queueHostOperation(api.host.id, 'instance.create', payload, { instanceId: 'instance-one', generation: 2 }); const result = { phase: 'prepared', description: '世界', details: [null, true, 42] }; api.reportHostOperation(api.host.id, operation.id, { state: 'running', result }); const expected = api.getHostOperation(operation.id)!; const serialized = JSON.stringify(payload); const originalParse = JSON.parse; const parse = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => { if (value === serialized) throw new Error('Delivery must not deserialize launch files.'); return originalParse(value, reviver); }); const response = await api.GET({ request: api.request('?lane=write') }); expect(response.headers.get('cache-control')).toBe('no-store'); expect(response.headers.get('content-type')).toContain('application/json'); let complete = false; let interleaved = false; const controllerWork = setImmediate().then(() => { interleaved = !complete; }); const chunks: Uint8Array[] = []; const reader = response.body!.getReader(); while (true) { const next = await reader.read(); if (next.done) break; expect(next.value.byteLength).toBeLessThanOrEqual(64 * 1024); chunks.push(next.value); } complete = true; await controllerWork; expect(interleaved).toBe(true); expect(chunks.length).toBeGreaterThan(100); parse.mockRestore(); const body = JSON.parse(Buffer.concat(chunks).toString('utf8')); expect(body).toEqual({ operations: [expected], serverTime: expect.any(String) }); expect(Number.isFinite(Date.parse(body.serverTime))).toBe(true); }); it('selects without payloads, loads one payload at a time, and retains the selected receipt snapshot', async () => { const api = await enrolled(); const payload = Buffer.from(JSON.stringify({ file: 'A'.repeat(512 * 1024) })); api.queueHostOperationSerialized(api.host.id, 'instance.create', payload); const second = api.queueHostOperationSerialized(api.host.id, 'instance.create', payload); api.reportHostOperationStatus(api.host.id, second.id, { state: 'accepted', result: { phase: 'selected' } }); const { db } = await import('../../src/lib/server/db/index.js'); const prepare = vi.spyOn(db(), 'prepare'); const payloadReads = () => prepare.mock.calls.filter(([sql]) => sql.startsWith('SELECT CAST(') && sql.includes(' AS BLOB) AS payload')).length; const response = await api.GET({ request: api.request('?lane=write') }); const selection = prepare.mock.calls.find(([sql]) => sql.includes('FROM ec_host_operations o'))![0]; const projection = selection.split(' FROM ')[0]; expect(projection).not.toContain('payload'); expect(projection).not.toContain('*'); expect(payloadReads()).toBe(0); // A receipt arriving during a streamed response must not mix a newer // result with the state and timestamps chosen for this delivery. api.reportHostOperationStatus(api.host.id, second.id, { state: 'running', result: { phase: 'newer' } }); const reader = response.body!.getReader(); const first = await reader.read(); expect(first.done).toBe(false); expect(payloadReads()).toBe(0); const chunks: Uint8Array[] = [first.value!]; let checkedNextPayload = false; while (true) { const next = await reader.read(); if (next.done) break; if (payloadReads() === 2 && !checkedNextPayload) { expect(Buffer.concat(chunks).includes(payload)).toBe(true); checkedNextPayload = true; } chunks.push(next.value); } expect(payloadReads()).toBe(2); expect(checkedNextPayload).toBe(true); const body = JSON.parse(Buffer.concat(chunks).toString('utf8')); expect(body.operations).toHaveLength(2); expect(body.operations[1]).toMatchObject({ id: second.id, state: 'accepted', result: { phase: 'selected' } }); expect(body.operations[0].payload).toEqual(JSON.parse(payload.toString('utf8'))); expect(body.operations[1].payload).toEqual(body.operations[0].payload); const other = api.createHost({ name: 'Other host' }); expect(() => api.getHostOperationPayload(other.id, second.id)).toThrow('not found for this host'); }); it('retains health lanes, delivery rotation, and cancellation suppression through the streamed route', async () => { const api = await enrolled(); const operations = Array.from({ length: 27 }, (_, index) => api.queueHostOperation(api.host.id, 'instance.restart', { index })); const inspect = api.queueHostOperation(api.host.id, 'instance.inspect', {}); const first = await (await api.GET({ request: api.request('?lane=write') })).json(); expect(first.operations.map((operation: { id: string }) => operation.id)).toEqual(operations.slice(0, 25).map(operation => operation.id)); const second = await (await api.GET({ request: api.request('?lane=write') })).json(); expect(second.operations.slice(0, 2).map((operation: { id: string }) => operation.id)).toEqual(operations.slice(25).map(operation => operation.id)); const health = await (await api.GET({ request: api.request('?lane=health') })).json(); expect(health.operations).toEqual([inspect]); const create = api.queueHostOperation(api.host.id, 'instance.create', {}); const cancellation = api.queueHostOperation(api.host.id, 'instance.cancel-create', { createOperationId: create.id }); const cancelled = await (await api.GET({ request: api.request('?lane=write') })).json(); expect(cancelled.operations[0].id).toBe(cancellation.id); expect(cancelled.operations.some((operation: { id: string }) => operation.id === create.id)).toBe(false); }); it('rechecks credentials after waking a long poll before sending work', async () => { const api = await enrolled(); const waiting = api.GET({ request: api.request('?lane=health&wait=20') }); api.revokeCredentials(api.host.id); api.queueHostOperation(api.host.id, 'instance.inspect', {}); await expect(waiting).rejects.toMatchObject({ status: 401 }); }); });