259 lines
8.7 KiB
TypeScript
259 lines
8.7 KiB
TypeScript
/**
|
||
* 引擎进程管理（任务 A3.10，ADR-0006）。
|
||
*
|
||
* 两个引擎共用一套：会话级资源登记、空闲回收、并发上限、崩溃退避重启、日志（stderr）环形缓冲。
|
||
* 平台不关心引擎是子进程还是别的形态——只管「谁在用、用了多久、坏了怎么办」。
|
||
*
|
||
* 默认值（实施设计 9.5）：一进程一会话、空闲 15 分钟回收、并发上限 4、连续重启上限 5。
|
||
* 重启上限是「连续」的：一次正常回合会把计数清零（markHealthy）。
|
||
*/
|
||
import { AppError } from '../app/errors.js';
|
||
import type { RuntimeSession } from './runtime.js';
|
||
|
||
export interface SupervisorOptions {
|
||
maxConcurrent: number;
|
||
idleTimeoutMs: number;
|
||
sweepIntervalMs: number;
|
||
/** 单个会话允许的连续重启次数 */
|
||
maxRestarts: number;
|
||
/** 退避基数：第 n 次重启等待 base * 2^(n-1) */
|
||
baseBackoffMs: number;
|
||
/** 引擎日志（stderr 尾部）输出口 */
|
||
onLog?: (message: string) => void;
|
||
/** 便于测试注入时钟 */
|
||
now?: () => number;
|
||
}
|
||
|
||
const DEFAULTS: Omit<SupervisorOptions, 'onLog' | 'now'> = {
|
||
maxConcurrent: 4,
|
||
idleTimeoutMs: 15 * 60_000,
|
||
sweepIntervalMs: 60_000,
|
||
maxRestarts: 5,
|
||
baseBackoffMs: 1_000,
|
||
};
|
||
|
||
const STDERR_TAIL_CHARS = 2000;
|
||
|
||
interface Entry {
|
||
/** 死了的条目保留为「墓碑」：记着重试次数与退避截止时间 */
|
||
dead: boolean;
|
||
runtimeSession: RuntimeSession | undefined;
|
||
fingerprint: string;
|
||
lastUsedAt: number;
|
||
restarts: number;
|
||
/** 到点之前不接受新请求；Infinity 表示已达重启上限 */
|
||
retryAfter: number;
|
||
stderrTail: string;
|
||
stderrOffset: number;
|
||
}
|
||
|
||
export interface SupervisorStats {
|
||
active: number;
|
||
evictedIdle: number;
|
||
evictedFailed: number;
|
||
evictedClosed: number;
|
||
evictedStale: number;
|
||
rejectedBusy: number;
|
||
rejectedBackoff: number;
|
||
}
|
||
|
||
export class RuntimeSupervisor {
|
||
private readonly entries = new Map<string, Entry>();
|
||
private readonly options: SupervisorOptions;
|
||
private sweeper: NodeJS.Timeout | undefined;
|
||
private readonly counters = { evictedIdle: 0, evictedFailed: 0, evictedClosed: 0, evictedStale: 0, rejectedBusy: 0, rejectedBackoff: 0 };
|
||
|
||
constructor(options: Partial<SupervisorOptions> = {}) {
|
||
this.options = { ...DEFAULTS, ...options };
|
||
}
|
||
|
||
/** 活跃（还活着的）引擎会话数，墓碑不算 */
|
||
get size(): number {
|
||
return [...this.entries.values()].filter((entry) => !entry.dead).length;
|
||
}
|
||
|
||
stats(): SupervisorStats {
|
||
return { active: this.size, ...this.counters };
|
||
}
|
||
|
||
get(key: string): RuntimeSession | undefined {
|
||
const entry = this.entries.get(key);
|
||
return entry?.dead === true ? undefined : entry?.runtimeSession;
|
||
}
|
||
|
||
/**
|
||
* 取会话对应的引擎资源：命中且指纹一致就复用；墓碑到点后重建；并发满或退避中则明确拒绝。
|
||
*/
|
||
async acquire(key: string, fingerprint: string, create: () => Promise<RuntimeSession>): Promise<RuntimeSession> {
|
||
const now = this.now();
|
||
|
||
let entry = this.entries.get(key);
|
||
if (entry && entry.fingerprint !== fingerprint) {
|
||
await this.evict(key, 'stale', '工具集或身份变化');
|
||
entry = this.entries.get(key);
|
||
}
|
||
|
||
if (entry?.dead === true) {
|
||
if (entry.retryAfter === Number.POSITIVE_INFINITY) {
|
||
this.counters.rejectedBackoff += 1;
|
||
throw new AppError('unavailable', `引擎连续启动失败已达上限（${this.options.maxRestarts} 次），停止重试`, {
|
||
status: 503,
|
||
retryable: false,
|
||
});
|
||
}
|
||
if (entry.retryAfter > now) {
|
||
this.counters.rejectedBackoff += 1;
|
||
const waitSeconds = Math.max(0.1, Math.round(((entry.retryAfter - now) / 1000) * 10) / 10);
|
||
throw new AppError('rate_limited', `引擎正在退避重启（第 ${entry.restarts} 次），请 ${waitSeconds} 秒后重试`, {
|
||
status: 429,
|
||
retryable: true,
|
||
});
|
||
}
|
||
const runtimeSession = await create();
|
||
this.entries.set(key, {
|
||
dead: false,
|
||
runtimeSession,
|
||
fingerprint,
|
||
lastUsedAt: now,
|
||
restarts: entry.restarts,
|
||
retryAfter: 0,
|
||
stderrTail: entry.stderrTail,
|
||
stderrOffset: 0,
|
||
});
|
||
return runtimeSession;
|
||
}
|
||
|
||
if (entry) {
|
||
entry.lastUsedAt = now;
|
||
this.drainStderr(key, entry);
|
||
return entry.runtimeSession as RuntimeSession;
|
||
}
|
||
|
||
if (this.size >= this.options.maxConcurrent) {
|
||
this.counters.rejectedBusy += 1;
|
||
throw new AppError('rate_limited', `同时活跃的引擎会话已达上限（${this.options.maxConcurrent}），请稍后重试`, {
|
||
status: 429,
|
||
retryable: true,
|
||
});
|
||
}
|
||
|
||
const runtimeSession = await create();
|
||
this.entries.set(key, {
|
||
dead: false,
|
||
runtimeSession,
|
||
fingerprint,
|
||
lastUsedAt: now,
|
||
restarts: 0,
|
||
retryAfter: 0,
|
||
stderrTail: '',
|
||
stderrOffset: 0,
|
||
});
|
||
return runtimeSession;
|
||
}
|
||
|
||
touch(key: string): void {
|
||
const entry = this.entries.get(key);
|
||
if (entry) entry.lastUsedAt = this.now();
|
||
}
|
||
|
||
/** 一次正常回合之后调用：连续重启计数清零（上限管的是「连续」失败） */
|
||
markHealthy(key: string): void {
|
||
const entry = this.entries.get(key);
|
||
if (entry && !entry.dead) entry.restarts = 0;
|
||
}
|
||
|
||
/**
|
||
* 回收：idle（空闲）、failed（崩溃或回合失败）、closed（平台关停）、
|
||
* stale（工具集或身份变化，旧会话作废——这不是故障，立即重建且不进退避）
|
||
*/
|
||
async evict(key: string, reason: 'idle' | 'failed' | 'closed' | 'stale', detail?: string): Promise<void> {
|
||
const entry = this.entries.get(key);
|
||
if (!entry) return;
|
||
|
||
if (reason !== 'failed') {
|
||
this.entries.delete(key);
|
||
if (reason === 'idle') this.counters.evictedIdle += 1;
|
||
else if (reason === 'stale') {
|
||
this.counters.evictedStale += 1;
|
||
this.logTail(key, entry, detail ?? '配置变化');
|
||
} else {
|
||
this.counters.evictedClosed += 1;
|
||
this.logTail(key, entry, detail ?? '平台关停');
|
||
}
|
||
await this.closeEntry(entry);
|
||
return;
|
||
}
|
||
|
||
this.counters.evictedFailed += 1;
|
||
const restarts = entry.restarts + 1;
|
||
await this.closeEntry(entry);
|
||
|
||
if (restarts > this.options.maxRestarts) {
|
||
this.options.onLog?.(`[process] 会话 ${key} 连续启动失败已达上限（${this.options.maxRestarts}），停止重试：${detail ?? ''}`);
|
||
this.entries.set(key, { ...entry, dead: true, runtimeSession: undefined, restarts, retryAfter: Number.POSITIVE_INFINITY });
|
||
return;
|
||
}
|
||
const backoff = this.options.baseBackoffMs * 2 ** (restarts - 1);
|
||
this.options.onLog?.(`[process] 会话 ${key} 异常回收（${detail ?? '失败'}），第 ${restarts} 次重启将在 ${backoff}ms 后允许`);
|
||
this.entries.set(key, { ...entry, dead: true, runtimeSession: undefined, restarts, retryAfter: this.now() + backoff });
|
||
}
|
||
|
||
/** 平台关停：全部回收，避免子进程残留 */
|
||
async disposeAll(): Promise<void> {
|
||
for (const key of [...this.entries.keys()]) await this.evict(key, 'closed');
|
||
this.stopSweeper();
|
||
}
|
||
|
||
startSweeper(): void {
|
||
if (this.sweeper) return;
|
||
this.sweeper = setInterval(() => {
|
||
void this.sweepIdle();
|
||
}, this.options.sweepIntervalMs);
|
||
this.sweeper.unref();
|
||
}
|
||
|
||
stopSweeper(): void {
|
||
if (this.sweeper) clearInterval(this.sweeper);
|
||
this.sweeper = undefined;
|
||
}
|
||
|
||
async sweepIdle(): Promise<void> {
|
||
const now = this.now();
|
||
for (const [key, entry] of [...this.entries]) {
|
||
if (entry.dead) continue; // 墓碑由 acquire 处理
|
||
if (now - entry.lastUsedAt >= this.options.idleTimeoutMs) {
|
||
await this.evict(key, 'idle');
|
||
}
|
||
}
|
||
}
|
||
|
||
/** 引擎 stderr 环形缓冲：只留最近一段，回收时打日志 */
|
||
private drainStderr(key: string, entry: Entry): void {
|
||
if (!entry.runtimeSession) return;
|
||
const snapshot = entry.runtimeSession.stderrTail?.() ?? '';
|
||
if (snapshot.length > entry.stderrOffset) {
|
||
const delta = snapshot.slice(entry.stderrOffset);
|
||
entry.stderrOffset = snapshot.length;
|
||
entry.stderrTail = `${entry.stderrTail}${delta}`.slice(-STDERR_TAIL_CHARS);
|
||
}
|
||
}
|
||
|
||
private logTail(key: string, entry: Entry, detail: string): void {
|
||
if (entry.stderrTail.trim() === '') return;
|
||
this.options.onLog?.(`[process] 会话 ${key} ${detail}，引擎 stderr 尾部：${entry.stderrTail.slice(-400)}`);
|
||
}
|
||
|
||
private async closeEntry(entry: Entry): Promise<void> {
|
||
if (!entry.runtimeSession) return;
|
||
try {
|
||
this.drainStderr('', entry);
|
||
await entry.runtimeSession.dispose();
|
||
} catch {
|
||
/* 回收失败不阻塞：进程可能已经死了 */
|
||
}
|
||
}
|
||
|
||
private now(): number {
|
||
return this.options.now?.() ?? Date.now();
|
||
}
|
||
} |