111 lines
6.9 KiB
JavaScript
111 lines
6.9 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import fs from 'node:fs';
|
|
import http from 'node:http';
|
|
import { join } from 'node:path';
|
|
import { tmpdir } from 'node:os';
|
|
import { spawn } from 'node:child_process';
|
|
import { once } from 'node:events';
|
|
import { Credentials } from '../src/credentials.mjs';
|
|
import { ModelKeys } from '../src/model-keys.mjs';
|
|
import { createProxyLock } from '../src/proxy-lock.mjs';
|
|
import { ManagementClient } from '../../model-proxy/src/client.mjs';
|
|
|
|
async function fixture(t) {
|
|
const root = fs.mkdtempSync(join(tmpdir(), 'shared-proxy-'));
|
|
const configDir = join(root, 'config'); fs.mkdirSync(configDir, { mode: 0o700 });
|
|
const stateDir = join(root, 'proxy'); fs.mkdirSync(stateDir, { mode: 0o700 });
|
|
const state = { keys: ['fixture-foreign'], putDelay: 0, putStarted: null };
|
|
const server = http.createServer(async (req, res) => {
|
|
if (req.headers.authorization !== 'Bearer fixture-management' && req.url !== '/v1/models') { res.writeHead(401); res.end('{}'); return; }
|
|
if (req.url === '/v1/models') { res.writeHead(state.keys.includes(req.headers.authorization?.slice(7)) ? 200 : 401); res.end('{}'); return; }
|
|
if (req.method === 'GET') { res.end(JSON.stringify(state.keys)); return; }
|
|
const chunks = []; for await (const chunk of req) chunks.push(chunk);
|
|
const next = JSON.parse(Buffer.concat(chunks)); state.putStarted?.();
|
|
if (state.putDelay) await new Promise(resolve => setTimeout(resolve, state.putDelay));
|
|
state.keys = next; res.end('{"status":"ok"}');
|
|
});
|
|
await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
|
|
const baseUrl = `http://127.0.0.1:${server.address().port}`;
|
|
const identity = `${configDir}:${server.address().port}`;
|
|
const lock = createProxyLock({ stateDir, identity });
|
|
const make = (org, timeout = 5000) => {
|
|
fs.mkdirSync(join(root, org, 'secrets'), { recursive: true, mode: 0o700 });
|
|
const credentials = new Credentials({ storeRoot: join(root, org, 'secrets'), dataRoot: join(root, org), modelAgents: [`${org}/worker`] });
|
|
t.after(() => credentials.close());
|
|
const client = new ManagementClient({ baseUrl, key: 'fixture-management', timeout });
|
|
return { client, manager: new ModelKeys({ credentials, client, configDir, baseUrl, org, lock }) };
|
|
};
|
|
t.after(async () => { server.closeAllConnections(); await new Promise(resolve => server.close(resolve)); fs.rmSync(root, { recursive: true, force: true }); });
|
|
return { root, configDir, stateDir, state, server, baseUrl, lock, make };
|
|
}
|
|
function worker(f, org) {
|
|
fs.mkdirSync(join(f.root, org, 'secrets'), { recursive: true, mode: 0o700 });
|
|
const child = spawn(process.execPath, [new URL('./shared-proxy-worker.mjs', import.meta.url).pathname, f.root, f.configDir, f.baseUrl, org], { stdio: ['ignore', 'pipe', 'pipe'] });
|
|
let out = '', err = ''; child.stdout.on('data', c => { out += c; }); child.stderr.on('data', c => { err += c; });
|
|
return { child, result: once(child, 'close').then(([code, signal]) => { if (code !== 0) throw Error(`worker ${org} exited ${code}/${signal}: ${err}`); return JSON.parse(out); }) };
|
|
}
|
|
test('two processes register and issue concurrently without losing sibling keys', async t => {
|
|
const f = await fixture(t), a = worker(f, 'alpha'), b = worker(f, 'beta');
|
|
const [alpha, beta] = await Promise.all([a.result, b.result]);
|
|
assert.ok(f.state.keys.includes(alpha.key)); assert.ok(f.state.keys.includes(beta.key));
|
|
assert.equal(await f.make('gamma').client.probeKey(alpha.key), 200);
|
|
assert.equal(f.state.keys.filter(key => key.includes('-alpha-sentinel-')).length, 1);
|
|
assert.equal(f.state.keys.filter(key => key.includes('-beta-sentinel-')).length, 1);
|
|
});
|
|
|
|
test('a delayed PUT leaves the fence set until restart-stated recovery and registration retries', async t => {
|
|
const f = await fixture(t), { manager, client } = f.make('alpha', 30);
|
|
f.state.putDelay = 150;
|
|
await assert.rejects(manager.reconcile(), { code: 'proxy-uncertain' });
|
|
assert.throws(() => manager.assertLaunchable('alpha/worker'), { code: 'proxy-uncertain' });
|
|
await assert.rejects(manager.ensure('alpha/worker'), { code: 'proxy-uncertain' });
|
|
await assert.rejects(f.lock.clear({ restarted: false, read: () => client.getKeys() }), { code: 'proxy-restart-required' });
|
|
await new Promise(resolve => setTimeout(resolve, 200));
|
|
assert.equal(f.lock.status().op, 'sentinel.register');
|
|
await f.lock.clear({ restarted: true, read: () => client.getKeys() });
|
|
f.state.putDelay = 0;
|
|
await manager.reconcile(); await manager.reconcile();
|
|
assert.equal(f.state.keys.filter(key => key.includes('-alpha-sentinel-')).length, 1);
|
|
});
|
|
|
|
test('a process killed during registration recovers the existing sentinel idempotently', async t => {
|
|
const f = await fixture(t); f.state.putDelay = 150;
|
|
let arrived; const started = new Promise(resolve => { arrived = resolve; }); f.state.putStarted = arrived;
|
|
const w = worker(f, 'alpha'); w.result.catch(() => {});
|
|
await started; w.child.kill('SIGKILL'); await once(w.child, 'close');
|
|
await new Promise(resolve => setTimeout(resolve, 200));
|
|
assert.equal(f.lock.status().op, 'sentinel.register');
|
|
const { manager, client } = f.make('alpha');
|
|
await assert.rejects(manager.reconcile(), { code: 'proxy-uncertain' });
|
|
await f.lock.clear({ restarted: true, read: () => client.getKeys() });
|
|
f.state.putDelay = 0; f.state.putStarted = null;
|
|
await manager.reconcile();
|
|
assert.equal(f.state.keys.filter(key => key.includes('-alpha-sentinel-')).length, 1);
|
|
});
|
|
|
|
test('a registered organization refuses a missing sentinel instead of trusting a replacement proxy', async t => {
|
|
const f = await fixture(t), { manager } = f.make('alpha');
|
|
await manager.reconcile();
|
|
f.state.keys = f.state.keys.filter(key => !key.includes('-alpha-sentinel-'));
|
|
await assert.rejects(manager.reconcile(), { code: 'proxy-open' });
|
|
assert.equal(f.state.keys.some(key => key.includes('-alpha-sentinel-')), false);
|
|
});
|
|
|
|
test('organization IDs ending in sentinel or key cannot collide with a sibling namespace', async t => {
|
|
const f = await fixture(t), alpha = f.make('alpha'), sibling = f.make('alpha-sentinel');
|
|
await alpha.manager.reconcile(); await sibling.manager.reconcile();
|
|
const { key } = await sibling.manager.ensure('alpha-sentinel/worker');
|
|
await alpha.manager.reconcile();
|
|
assert.equal(await sibling.client.probeKey(key), 200);
|
|
assert.equal(f.state.keys.filter(value => value.includes('-alpha-sentinel-sentinel-')).length, 1);
|
|
});
|
|
|
|
test('an uncertain sibling write blocks completing a local revocation', async t => {
|
|
const f = await fixture(t), { manager } = f.make('alpha');
|
|
await manager.reconcile(); await manager.ensure('alpha/worker'); await manager.prepareRevoke('alpha/worker');
|
|
await f.lock.run(async () => f.lock.fence({ org: 'beta', op: 'keys.put' }));
|
|
assert.throws(() => manager.completeRevoke('alpha/worker'), { code: 'proxy-uncertain' });
|
|
assert.equal(manager.hasRecord('alpha/worker'), true);
|
|
});
|