Move terminal sessions to worker process
This commit is contained in:
Mohamed Boudra
@@ -0,0 +1,527 @@
|
||||
import { fork, type ChildProcess } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { TerminalState } from "../shared/messages.js";
|
||||
import type {
|
||||
ClientMessage,
|
||||
ServerMessage,
|
||||
TerminalCommandFinishedInfo,
|
||||
TerminalExitInfo,
|
||||
TerminalSession,
|
||||
} from "./terminal.js";
|
||||
import type {
|
||||
TerminalListItem,
|
||||
TerminalManager,
|
||||
TerminalsChangedEvent,
|
||||
TerminalsChangedListener,
|
||||
} from "./terminal-manager.js";
|
||||
import type {
|
||||
TerminalWorkerRequest,
|
||||
TerminalWorkerResponse,
|
||||
TerminalWorkerToParentMessage,
|
||||
WorkerCreateTerminalOptions,
|
||||
WorkerTerminalInfo,
|
||||
} from "./terminal-worker-protocol.js";
|
||||
|
||||
const REQUEST_TIMEOUT_MS = 10000;
|
||||
|
||||
type TerminalWorkerRequestInput = TerminalWorkerRequest extends infer Request
|
||||
? Request extends TerminalWorkerRequest
|
||||
? Omit<Request, "requestId">
|
||||
: never
|
||||
: never;
|
||||
|
||||
interface PendingRequest {
|
||||
resolve: (value: unknown) => void;
|
||||
reject: (error: Error) => void;
|
||||
timeout: ReturnType<typeof setTimeout>;
|
||||
}
|
||||
|
||||
interface WorkerTerminalRecord {
|
||||
info: WorkerTerminalInfo;
|
||||
state: TerminalState;
|
||||
exitInfo: TerminalExitInfo | null;
|
||||
messageListeners: Set<(msg: ServerMessage) => void>;
|
||||
exitListeners: Set<(info: TerminalExitInfo) => void>;
|
||||
commandFinishedListeners: Set<(info: TerminalCommandFinishedInfo) => void>;
|
||||
titleChangeListeners: Set<(title?: string) => void>;
|
||||
session: TerminalSession;
|
||||
}
|
||||
|
||||
interface TerminalWorkerProcess {
|
||||
connected: boolean;
|
||||
killed: boolean;
|
||||
send(message: TerminalWorkerRequest, callback: (error: Error | null) => void): boolean;
|
||||
disconnect(): void;
|
||||
kill(): boolean;
|
||||
on(event: "message", listener: (message: TerminalWorkerToParentMessage) => void): this;
|
||||
on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this;
|
||||
}
|
||||
|
||||
interface WorkerTerminalManagerOptions {
|
||||
requestTimeoutMs?: number;
|
||||
forkWorker?: () => TerminalWorkerProcess;
|
||||
}
|
||||
|
||||
function resolveWorkerUrl(): URL {
|
||||
const currentUrl = import.meta.url;
|
||||
if (currentUrl.endsWith(".ts")) {
|
||||
return new URL("./terminal-worker-process.ts", currentUrl);
|
||||
}
|
||||
return new URL("./terminal-worker-process.js", currentUrl);
|
||||
}
|
||||
|
||||
function resolveWorkerExecArgv(): string[] {
|
||||
if (!import.meta.url.endsWith(".ts")) {
|
||||
return [];
|
||||
}
|
||||
const loaderUrl = new URL("./terminal-ts-loader.mjs", import.meta.url).href;
|
||||
const importSource = [
|
||||
'import { register } from "node:module";',
|
||||
'import { pathToFileURL } from "node:url";',
|
||||
`register(${JSON.stringify(loaderUrl)}, pathToFileURL("./"));`,
|
||||
].join(" ");
|
||||
return [
|
||||
"--experimental-strip-types",
|
||||
"--import",
|
||||
`data:text/javascript,${encodeURIComponent(importSource)}`,
|
||||
];
|
||||
}
|
||||
|
||||
function isResponse(message: TerminalWorkerToParentMessage): message is TerminalWorkerResponse {
|
||||
return message.type === "response";
|
||||
}
|
||||
|
||||
function cloneTerminalInfo(info: WorkerTerminalInfo): WorkerTerminalInfo {
|
||||
return {
|
||||
id: info.id,
|
||||
name: info.name,
|
||||
cwd: info.cwd,
|
||||
...(info.title ? { title: info.title } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function forkTerminalWorker(): TerminalWorkerProcess {
|
||||
return fork(fileURLToPath(resolveWorkerUrl()), [], {
|
||||
execArgv: resolveWorkerExecArgv(),
|
||||
serialization: "advanced",
|
||||
stdio: ["ignore", "ignore", "inherit", "ipc"],
|
||||
}) as ChildProcess as TerminalWorkerProcess;
|
||||
}
|
||||
|
||||
export function createWorkerTerminalManager(
|
||||
managerOptions: WorkerTerminalManagerOptions = {},
|
||||
): TerminalManager {
|
||||
const worker = managerOptions.forkWorker ? managerOptions.forkWorker() : forkTerminalWorker();
|
||||
const requestTimeoutMs = managerOptions.requestTimeoutMs ?? REQUEST_TIMEOUT_MS;
|
||||
const pendingRequests = new Map<string, PendingRequest>();
|
||||
const recordsById = new Map<string, WorkerTerminalRecord>();
|
||||
const terminalIdsByCwd = new Map<string, Set<string>>();
|
||||
const terminalsChangedListeners = new Set<TerminalsChangedListener>();
|
||||
let workerExited = false;
|
||||
let workerShutdownTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
function emitTerminalsChanged(event: TerminalsChangedEvent): void {
|
||||
for (const listener of Array.from(terminalsChangedListeners)) {
|
||||
try {
|
||||
listener(event);
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function listTerminalItemsForCwd(cwd: string): TerminalListItem[] {
|
||||
const terminalIds = terminalIdsByCwd.get(cwd);
|
||||
if (!terminalIds) {
|
||||
return [];
|
||||
}
|
||||
const terminals: TerminalListItem[] = [];
|
||||
for (const terminalId of terminalIds) {
|
||||
const record = recordsById.get(terminalId);
|
||||
if (!record) {
|
||||
continue;
|
||||
}
|
||||
terminals.push({
|
||||
id: record.info.id,
|
||||
name: record.info.name,
|
||||
cwd: record.info.cwd,
|
||||
...(record.info.title ? { title: record.info.title } : {}),
|
||||
});
|
||||
}
|
||||
return terminals;
|
||||
}
|
||||
|
||||
function registerRecord(input: {
|
||||
info: WorkerTerminalInfo;
|
||||
state: TerminalState;
|
||||
}): TerminalSession {
|
||||
const existing = recordsById.get(input.info.id);
|
||||
if (existing) {
|
||||
existing.info = cloneTerminalInfo(input.info);
|
||||
existing.state = input.state;
|
||||
return existing.session;
|
||||
}
|
||||
|
||||
const record: WorkerTerminalRecord = {
|
||||
info: cloneTerminalInfo(input.info),
|
||||
state: input.state,
|
||||
exitInfo: null,
|
||||
messageListeners: new Set(),
|
||||
exitListeners: new Set(),
|
||||
commandFinishedListeners: new Set(),
|
||||
titleChangeListeners: new Set(),
|
||||
session: undefined as unknown as TerminalSession,
|
||||
};
|
||||
|
||||
const session: TerminalSession = {
|
||||
get id() {
|
||||
return record.info.id;
|
||||
},
|
||||
get name() {
|
||||
return record.info.name;
|
||||
},
|
||||
get cwd() {
|
||||
return record.info.cwd;
|
||||
},
|
||||
send(message: ClientMessage): void {
|
||||
sendBestEffortRequest({ type: "send", terminalId: record.info.id, message });
|
||||
},
|
||||
subscribe(listener: (msg: ServerMessage) => void): () => void {
|
||||
record.messageListeners.add(listener);
|
||||
queueMicrotask(() => {
|
||||
if (record.messageListeners.has(listener)) {
|
||||
listener({ type: "snapshot", state: record.state });
|
||||
}
|
||||
});
|
||||
return () => {
|
||||
record.messageListeners.delete(listener);
|
||||
};
|
||||
},
|
||||
onExit(listener: (info: TerminalExitInfo) => void): () => void {
|
||||
if (record.exitInfo) {
|
||||
queueMicrotask(() => listener(record.exitInfo!));
|
||||
return () => {};
|
||||
}
|
||||
record.exitListeners.add(listener);
|
||||
return () => {
|
||||
record.exitListeners.delete(listener);
|
||||
};
|
||||
},
|
||||
onCommandFinished(listener: (info: TerminalCommandFinishedInfo) => void): () => void {
|
||||
record.commandFinishedListeners.add(listener);
|
||||
return () => {
|
||||
record.commandFinishedListeners.delete(listener);
|
||||
};
|
||||
},
|
||||
onTitleChange(listener: (title?: string) => void): () => void {
|
||||
record.titleChangeListeners.add(listener);
|
||||
if (record.info.title !== undefined) {
|
||||
queueMicrotask(() => {
|
||||
if (record.titleChangeListeners.has(listener)) {
|
||||
listener(record.info.title);
|
||||
}
|
||||
});
|
||||
}
|
||||
return () => {
|
||||
record.titleChangeListeners.delete(listener);
|
||||
};
|
||||
},
|
||||
getSize(): { rows: number; cols: number } {
|
||||
return {
|
||||
rows: record.state.rows,
|
||||
cols: record.state.cols,
|
||||
};
|
||||
},
|
||||
getState(): TerminalState {
|
||||
return record.state;
|
||||
},
|
||||
getTitle(): string | undefined {
|
||||
return record.info.title;
|
||||
},
|
||||
getExitInfo(): TerminalExitInfo | null {
|
||||
return record.exitInfo;
|
||||
},
|
||||
kill(): void {
|
||||
sendBestEffortRequest({ type: "killTerminal", terminalId: record.info.id });
|
||||
},
|
||||
killAndWait(options?: {
|
||||
gracefulTimeoutMs?: number;
|
||||
forceTimeoutMs?: number;
|
||||
}): Promise<void> {
|
||||
return sendRequest({
|
||||
type: "killTerminalAndWait",
|
||||
terminalId: record.info.id,
|
||||
...(options ? { options } : {}),
|
||||
}).then(() => undefined);
|
||||
},
|
||||
};
|
||||
|
||||
record.session = session;
|
||||
recordsById.set(record.info.id, record);
|
||||
const terminalIds = terminalIdsByCwd.get(record.info.cwd) ?? new Set<string>();
|
||||
terminalIds.add(record.info.id);
|
||||
terminalIdsByCwd.set(record.info.cwd, terminalIds);
|
||||
return session;
|
||||
}
|
||||
|
||||
function removeRecord(terminalId: string): WorkerTerminalRecord | undefined {
|
||||
const record = recordsById.get(terminalId);
|
||||
if (!record) {
|
||||
return undefined;
|
||||
}
|
||||
recordsById.delete(terminalId);
|
||||
const terminalIds = terminalIdsByCwd.get(record.info.cwd);
|
||||
if (terminalIds) {
|
||||
terminalIds.delete(terminalId);
|
||||
if (terminalIds.size === 0) {
|
||||
terminalIdsByCwd.delete(record.info.cwd);
|
||||
}
|
||||
}
|
||||
return record;
|
||||
}
|
||||
|
||||
function handleWorkerEvent(message: TerminalWorkerToParentMessage): void {
|
||||
switch (message.type) {
|
||||
case "terminalCreated": {
|
||||
registerRecord({ info: message.terminal, state: message.state });
|
||||
return;
|
||||
}
|
||||
|
||||
case "terminalRemoved": {
|
||||
removeRecord(message.terminalId);
|
||||
emitTerminalsChanged({
|
||||
cwd: message.cwd,
|
||||
terminals: listTerminalItemsForCwd(message.cwd),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
case "terminalMessage": {
|
||||
const record = recordsById.get(message.terminalId);
|
||||
if (!record) {
|
||||
return;
|
||||
}
|
||||
if (message.message.type === "snapshot") {
|
||||
record.state = message.message.state;
|
||||
}
|
||||
for (const listener of Array.from(record.messageListeners)) {
|
||||
listener(message.message);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
case "terminalExit": {
|
||||
const record = recordsById.get(message.terminalId);
|
||||
if (!record) {
|
||||
return;
|
||||
}
|
||||
record.exitInfo = message.info;
|
||||
for (const listener of Array.from(record.exitListeners)) {
|
||||
listener(message.info);
|
||||
}
|
||||
record.exitListeners.clear();
|
||||
removeRecord(message.terminalId);
|
||||
emitTerminalsChanged({
|
||||
cwd: record.info.cwd,
|
||||
terminals: listTerminalItemsForCwd(record.info.cwd),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
case "terminalTitleChange": {
|
||||
const record = recordsById.get(message.terminalId);
|
||||
if (!record) {
|
||||
return;
|
||||
}
|
||||
record.info = {
|
||||
...record.info,
|
||||
...(message.title ? { title: message.title } : { title: undefined }),
|
||||
};
|
||||
for (const listener of Array.from(record.titleChangeListeners)) {
|
||||
listener(message.title);
|
||||
}
|
||||
emitTerminalsChanged({
|
||||
cwd: record.info.cwd,
|
||||
terminals: listTerminalItemsForCwd(record.info.cwd),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
case "terminalCommandFinished": {
|
||||
const record = recordsById.get(message.terminalId);
|
||||
if (!record) {
|
||||
return;
|
||||
}
|
||||
for (const listener of Array.from(record.commandFinishedListeners)) {
|
||||
listener(message.info);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
case "terminalsChanged": {
|
||||
emitTerminalsChanged({
|
||||
cwd: message.cwd,
|
||||
terminals: message.terminals.map((terminal) => ({
|
||||
id: terminal.id,
|
||||
name: terminal.name,
|
||||
cwd: terminal.cwd,
|
||||
...(terminal.title ? { title: terminal.title } : {}),
|
||||
})),
|
||||
});
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function rejectPendingRequests(error: Error): void {
|
||||
for (const [requestId, pending] of pendingRequests) {
|
||||
clearTimeout(pending.timeout);
|
||||
pending.reject(error);
|
||||
pendingRequests.delete(requestId);
|
||||
}
|
||||
}
|
||||
|
||||
worker.on("message", (message: TerminalWorkerToParentMessage) => {
|
||||
if (isResponse(message)) {
|
||||
const pending = pendingRequests.get(message.requestId);
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
clearTimeout(pending.timeout);
|
||||
pendingRequests.delete(message.requestId);
|
||||
if (message.ok) {
|
||||
pending.resolve(message.result);
|
||||
} else {
|
||||
pending.reject(new Error(message.error));
|
||||
}
|
||||
return;
|
||||
}
|
||||
handleWorkerEvent(message);
|
||||
});
|
||||
|
||||
worker.on("exit", (code, signal) => {
|
||||
workerExited = true;
|
||||
if (workerShutdownTimer) {
|
||||
clearTimeout(workerShutdownTimer);
|
||||
workerShutdownTimer = null;
|
||||
}
|
||||
rejectPendingRequests(new Error(`Terminal worker exited (${signal ?? code ?? "unknown"})`));
|
||||
});
|
||||
|
||||
function sendRequest(input: TerminalWorkerRequestInput): Promise<unknown> {
|
||||
if (workerExited || !worker.connected) {
|
||||
return Promise.reject(new Error("Terminal worker is not running"));
|
||||
}
|
||||
const requestId = randomUUID();
|
||||
const message = { ...input, requestId } as TerminalWorkerRequest;
|
||||
return new Promise((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
pendingRequests.delete(requestId);
|
||||
reject(new Error(`Terminal worker request timed out: ${input.type}`));
|
||||
}, requestTimeoutMs);
|
||||
pendingRequests.set(requestId, { resolve, reject, timeout });
|
||||
worker.send(message, (error) => {
|
||||
if (!error) {
|
||||
return;
|
||||
}
|
||||
clearTimeout(timeout);
|
||||
pendingRequests.delete(requestId);
|
||||
reject(error);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function sendBestEffortRequest(input: TerminalWorkerRequestInput): void {
|
||||
void sendRequest(input).catch(() => {
|
||||
// The public terminal methods that call this are intentionally synchronous.
|
||||
// Worker failures are surfaced through awaitable manager methods and worker
|
||||
// lifecycle state; do not let fire-and-forget sends crash the daemon.
|
||||
});
|
||||
}
|
||||
|
||||
function toSessions(terminals: WorkerTerminalInfo[]): TerminalSession[] {
|
||||
return terminals
|
||||
.map((terminal) => recordsById.get(terminal.id)?.session)
|
||||
.filter((session): session is TerminalSession => Boolean(session));
|
||||
}
|
||||
|
||||
return {
|
||||
async getTerminals(cwd: string): Promise<TerminalSession[]> {
|
||||
const result = (await sendRequest({ type: "getTerminals", cwd })) as WorkerTerminalInfo[];
|
||||
return toSessions(result);
|
||||
},
|
||||
|
||||
async createTerminal(options: WorkerCreateTerminalOptions): Promise<TerminalSession> {
|
||||
const result = (await sendRequest({ type: "createTerminal", options })) as {
|
||||
terminal: WorkerTerminalInfo;
|
||||
state: TerminalState;
|
||||
};
|
||||
return registerRecord({ info: result.terminal, state: result.state });
|
||||
},
|
||||
|
||||
registerCwdEnv(options: { cwd: string; env: Record<string, string> }): void {
|
||||
sendBestEffortRequest({
|
||||
type: "registerCwdEnv",
|
||||
cwd: options.cwd,
|
||||
env: options.env,
|
||||
});
|
||||
},
|
||||
|
||||
getTerminal(id: string): TerminalSession | undefined {
|
||||
return recordsById.get(id)?.session;
|
||||
},
|
||||
|
||||
killTerminal(id: string): void {
|
||||
void sendRequest({ type: "killTerminal", terminalId: id }).catch(() => {
|
||||
// no-op; kill is intentionally best-effort and synchronous in the public interface.
|
||||
});
|
||||
},
|
||||
|
||||
async killTerminalAndWait(
|
||||
id: string,
|
||||
options?: { gracefulTimeoutMs?: number; forceTimeoutMs?: number },
|
||||
): Promise<void> {
|
||||
await sendRequest({
|
||||
type: "killTerminalAndWait",
|
||||
terminalId: id,
|
||||
...(options ? { options } : {}),
|
||||
});
|
||||
},
|
||||
|
||||
listDirectories(): string[] {
|
||||
return Array.from(terminalIdsByCwd.keys());
|
||||
},
|
||||
|
||||
killAll(): void {
|
||||
void sendRequest({ type: "killAll" })
|
||||
.catch(() => {
|
||||
// no-op
|
||||
})
|
||||
.finally(() => {
|
||||
if (worker.connected) {
|
||||
worker.disconnect();
|
||||
}
|
||||
if (!worker.killed && !workerShutdownTimer) {
|
||||
workerShutdownTimer = setTimeout(() => {
|
||||
worker.kill();
|
||||
}, 1000);
|
||||
}
|
||||
});
|
||||
for (const terminalId of Array.from(recordsById.keys())) {
|
||||
removeRecord(terminalId);
|
||||
}
|
||||
},
|
||||
|
||||
subscribeTerminalsChanged(listener: TerminalsChangedListener): () => void {
|
||||
terminalsChangedListeners.add(listener);
|
||||
return () => {
|
||||
terminalsChangedListeners.delete(listener);
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function terminateWorkerTerminalManager(manager: TerminalManager): void {
|
||||
manager.killAll();
|
||||
}
|
||||
Reference in New Issue
Block a user