187 lines
8.6 KiB
TypeScript
187 lines
8.6 KiB
TypeScript
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,
|
|
preparedRuntimeImages: ['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 });
|
|
});
|
|
});
|