184 lines
6.9 KiB
JavaScript
184 lines
6.9 KiB
JavaScript
/**
|
||
* 引擎进程管理用例（任务 A3.10）。
|
||
*
|
||
* 五条验收：可启停、超时中断、崩溃按退避重启且上限可配、日志轮转、平台退出时子进程回收。
|
||
* 超时中断与真进程回收在 pi-runtime / 关停用例里覆盖，这里覆盖调度、重建与退避逻辑。
|
||
*/
|
||
import assert from 'node:assert/strict';
|
||
import { test } from 'node:test';
|
||
|
||
import { RuntimeSupervisor } from '@aps/server/agent/process-manager';
|
||
|
||
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
||
|
||
/** 造一个可观测的假运行时会话 */
|
||
function fakeSession(label, log) {
|
||
return {
|
||
id: label,
|
||
toolExecution: 'platform',
|
||
async *send() {
|
||
yield { type: 'turn_end', reason: 'stop' };
|
||
},
|
||
resolveToolResult() {},
|
||
abort() {
|
||
log.push(`abort:${label}`);
|
||
},
|
||
async dispose() {
|
||
log.push(`dispose:${label}`);
|
||
},
|
||
stderrTail: () => log.filter((item) => item.startsWith('stderr')).join(''),
|
||
};
|
||
}
|
||
|
||
function countingFactory(log) {
|
||
let created = 0;
|
||
return async () => {
|
||
created += 1;
|
||
const label = `s${created}`;
|
||
log.push(`create:${label}`);
|
||
return fakeSession(label, log);
|
||
};
|
||
}
|
||
|
||
test('空闲回收：到点回收，下一次请求重建', async () => {
|
||
const log = [];
|
||
const supervisor = new RuntimeSupervisor({ idleTimeoutMs: 30, sweepIntervalMs: 10, maxConcurrent: 4 });
|
||
const create = countingFactory(log);
|
||
|
||
await supervisor.acquire('k1', 'fp', create);
|
||
assert.equal(supervisor.size, 1);
|
||
await sleep(80);
|
||
await supervisor.sweepIdle();
|
||
assert.equal(supervisor.size, 0, '空闲会话应被回收');
|
||
assert.equal(supervisor.stats().evictedIdle, 1);
|
||
assert.ok(log.includes('dispose:s1'), '回收要真的释放引擎资源');
|
||
|
||
await supervisor.acquire('k1', 'fp', create);
|
||
assert.equal(supervisor.stats().active, 1);
|
||
assert.ok(log.includes('create:s2'), '下一次请求应重建');
|
||
});
|
||
|
||
test('并发上限：满了明确拒绝（rate_limited 429），回收后可再用', async () => {
|
||
const log = [];
|
||
const supervisor = new RuntimeSupervisor({ maxConcurrent: 1, idleTimeoutMs: 60_000 });
|
||
const create = countingFactory(log);
|
||
|
||
await supervisor.acquire('a', 'fp', create);
|
||
await assert.rejects(
|
||
() => supervisor.acquire('b', 'fp', create),
|
||
(error) => {
|
||
assert.equal(error.code, 'rate_limited');
|
||
assert.equal(error.status, 429);
|
||
assert.equal(error.retryable, true);
|
||
return true;
|
||
},
|
||
);
|
||
assert.equal(supervisor.stats().rejectedBusy, 1);
|
||
|
||
await supervisor.evict('a', 'idle');
|
||
await supervisor.acquire('b', 'fp', create);
|
||
assert.equal(supervisor.stats().active, 1);
|
||
});
|
||
|
||
test('崩溃退避重启：按指数退避等待，超过上限停止重试，正常回合清零', async () => {
|
||
const log = [];
|
||
const supervisor = new RuntimeSupervisor({ maxConcurrent: 4, idleTimeoutMs: 60_000, maxRestarts: 2, baseBackoffMs: 20 });
|
||
const create = countingFactory(log);
|
||
|
||
await supervisor.acquire('k', 'fp', create);
|
||
await supervisor.evict('k', 'failed', '进程退出');
|
||
// 第一次退避：立即请求被拒
|
||
await assert.rejects(() => supervisor.acquire('k', 'fp', create), /退避重启/);
|
||
await sleep(40);
|
||
await supervisor.acquire('k', 'fp', create);
|
||
assert.equal(supervisor.stats().active, 1, '退避到点后应重建');
|
||
|
||
// 正常回合清零连续失败计数
|
||
supervisor.markHealthy('k');
|
||
await supervisor.evict('k', 'failed');
|
||
await sleep(40);
|
||
await supervisor.acquire('k', 'fp', create);
|
||
assert.equal(supervisor.stats().active, 1);
|
||
|
||
// 连续失败到上限：不再重试
|
||
supervisor.touch('k');
|
||
await supervisor.evict('k', 'failed');
|
||
await sleep(90);
|
||
await supervisor.acquire('k', 'fp', create);
|
||
await supervisor.evict('k', 'failed');
|
||
await assert.rejects(
|
||
() => supervisor.acquire('k', 'fp', create),
|
||
(error) => {
|
||
assert.equal(error.code, 'unavailable');
|
||
assert.equal(error.retryable, false);
|
||
return true;
|
||
},
|
||
'超过重启上限后应停止重试',
|
||
);
|
||
assert.ok(supervisor.stats().rejectedBackoff >= 1);
|
||
});
|
||
|
||
test('工具集/身份变化：旧会话作废并立即重建，不进退避也不计崩溃', async () => {
|
||
const log = [];
|
||
const supervisor = new RuntimeSupervisor({ maxConcurrent: 4, idleTimeoutMs: 60_000, maxRestarts: 1, baseBackoffMs: 5_000 });
|
||
const create = countingFactory(log);
|
||
|
||
await supervisor.acquire('k', 'fp-a', create);
|
||
// 指纹变化（例如新注册了 Provider）：立刻换新，不能像崩溃那样把用户的请求挡在退避里
|
||
await supervisor.acquire('k', 'fp-b', create);
|
||
assert.ok(log.includes('create:s2'), '指纹变化应立刻重建运行时会话');
|
||
assert.equal(supervisor.stats().active, 1);
|
||
assert.equal(supervisor.stats().evictedStale, 1);
|
||
assert.equal(supervisor.stats().evictedFailed, 0, '更换工具集不是故障，不该计入崩溃回收');
|
||
assert.ok(log.includes('dispose:s1'), '旧会话要真的释放');
|
||
});
|
||
test('平台关停：全部回收并记录引擎日志尾部', async () => {
|
||
const log = [];
|
||
const supervisor = new RuntimeSupervisor({ maxConcurrent: 4, idleTimeoutMs: 60_000, onLog: (message) => log.push(message) });
|
||
const create = countingFactory(log);
|
||
await supervisor.acquire('a', 'fp', create);
|
||
await supervisor.acquire('b', 'fp', create);
|
||
|
||
await supervisor.disposeAll();
|
||
assert.equal(supervisor.stats().active, 0);
|
||
assert.equal(supervisor.stats().evictedClosed, 2);
|
||
assert.ok(log.includes('dispose:s1') && log.includes('dispose:s2'), '关停要把所有引擎资源释放');
|
||
});
|
||
|
||
test('服务层：关停平台时引擎会话被释放（无残留进程）', async () => {
|
||
const log = [];
|
||
let created = 0;
|
||
const runtime = {
|
||
name: 'counting',
|
||
toolExecution: 'platform',
|
||
async createSession() {
|
||
created += 1;
|
||
return fakeSession(`svc${created}`, log);
|
||
},
|
||
};
|
||
|
||
process.env.APS_DEV_PROVIDER = '1';
|
||
const { startServer } = await import('../server/dist/index.js');
|
||
const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test', databasePath: ':memory:', engine: 'fake', runtime });
|
||
try {
|
||
const created1 = await (
|
||
await fetch(`${server.url}/v1/sessions`, {
|
||
method: 'POST',
|
||
headers: { 'content-type': 'application/json' },
|
||
body: JSON.stringify({ scope: 'project', projectId: 'p-default', title: '关停用例' }),
|
||
})
|
||
).json();
|
||
const sessionId = created1.session.sessionId;
|
||
await fetch(`${server.url}/v1/sessions/${sessionId}/messages`, {
|
||
method: 'POST',
|
||
headers: { 'content-type': 'application/json' },
|
||
body: JSON.stringify({ text: '你好', idempotencyKey: 'idem-proc-0001' }),
|
||
});
|
||
await sleep(200);
|
||
assert.equal(server.service.processStats().active, 1, '发过消息后应有一个活跃引擎会话');
|
||
} finally {
|
||
await server.close();
|
||
delete process.env.APS_DEV_PROVIDER;
|
||
}
|
||
assert.ok(log.includes('dispose:svc1'), `关停平台应释放引擎会话，实际日志：${log.join(',')}`);
|
||
}); |