mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-07 05:56:21 -10:00
fix: contain terminal WebSocket attach failures (#907)
A rejected web-terminal attach closes only that viewer with 1011 instead of exiting the daemon, and the failed broker is disposed so a retry starts fresh.
This commit is contained in:
1 parent
98e643b6d0
commit
b00a536b82
4 files changed
+128
-2
No files matched your search
@@ -111,7 +111,19 @@ export function registerTerminalWs(
|
||||
close: (code: number, reason: string) => { try { ws.close(code, reason); } catch { /* already closed */ } },
|
||||
};
|
||||
subscriber = sub;
|
||||
const b = await getRegistry(tmux as unknown as BrokerTmux).attach(sessionName, sub);
|
||||
let b: TerminalSessionBroker;
|
||||
try {
|
||||
b = await getRegistry(tmux as unknown as BrokerTmux).attach(sessionName, sub);
|
||||
} catch {
|
||||
// The WebSocket adapter does not await onOpen. Contain attach failures
|
||||
// here; the registry disposes the failed broker before rejecting.
|
||||
closed = true;
|
||||
earlyFrames.length = 0;
|
||||
earlyFrameBytes = 0;
|
||||
subscriber = null;
|
||||
sub.close(1011, "terminal attach failed");
|
||||
return;
|
||||
}
|
||||
broker = b;
|
||||
// If the socket closed while attach was in flight, detach now so the
|
||||
// broker does not retain a dead subscriber (detach is idempotent).
|
||||
|
||||
@@ -616,7 +616,14 @@ export class TerminalBrokerRegistry {
|
||||
});
|
||||
this.brokers.set(sessionName, broker);
|
||||
}
|
||||
await broker.attach(sub);
|
||||
try {
|
||||
await broker.attach(sub);
|
||||
} catch (error) {
|
||||
// A rejected open never registers the subscriber, so detach cannot evict
|
||||
// this broker or remove a partially created output file.
|
||||
broker.dispose();
|
||||
throw error;
|
||||
}
|
||||
return broker;
|
||||
}
|
||||
}
|
||||
@@ -298,6 +298,49 @@ describe("TerminalSessionBroker", () => {
|
||||
// ---- registry: create-if-absent + eviction ---------------------------------
|
||||
|
||||
describe("TerminalBrokerRegistry", () => {
|
||||
it.each(["probe", "pipe"])("evicts a rejected %s attach, cleans resources, and permits retry", async (stage) => {
|
||||
let fail = true;
|
||||
let outputPath: string | null = null;
|
||||
const fault = new Error("attach unavailable");
|
||||
const reg = new TerminalBrokerRegistry(makeTmux({
|
||||
hasSession: async (name) => {
|
||||
if (name === "failed@rig" && fail && stage === "probe") throw fault;
|
||||
return true;
|
||||
},
|
||||
startPipePane: async (name, file) => {
|
||||
if (name === "failed@rig") {
|
||||
outputPath = file;
|
||||
if (fail && stage === "pipe") throw fault;
|
||||
}
|
||||
return { ok: true };
|
||||
},
|
||||
}));
|
||||
const healthy = track(await reg.attach("healthy@rig", makeSub()));
|
||||
const first = reg.attach("failed@rig", makeSub());
|
||||
const failed = track(reg.get("failed@rig")!);
|
||||
const attempts = await Promise.allSettled([first, reg.attach("failed@rig", makeSub())]);
|
||||
expect(attempts).toEqual([
|
||||
{ status: "rejected", reason: fault },
|
||||
{ status: "rejected", reason: fault },
|
||||
]);
|
||||
expect(failed.subscriberCount).toBe(0);
|
||||
expect(failed.pipeOutputPath).toBeNull();
|
||||
if (stage === "pipe") {
|
||||
expect(outputPath).toBeTruthy();
|
||||
expect(fs.existsSync(outputPath!)).toBe(false);
|
||||
}
|
||||
expect(reg.get("failed@rig")).toBeUndefined();
|
||||
expect(reg.size).toBe(1);
|
||||
expect(reg.get("healthy@rig")).toBe(healthy);
|
||||
expect(healthy.subscriberCount).toBe(1);
|
||||
|
||||
fail = false;
|
||||
const recovered = track(await reg.attach("failed@rig", makeSub()));
|
||||
expect(recovered).not.toBe(failed);
|
||||
expect(recovered.subscriberCount).toBe(1);
|
||||
expect(reg.size).toBe(2);
|
||||
});
|
||||
|
||||
it("create-if-absent: two subscribers on one session share ONE broker / ONE pipe", async () => {
|
||||
const startPipePane = vi.fn(async () => ({ ok: true as const }));
|
||||
const reg = new TerminalBrokerRegistry(makeTmux({ startPipePane }), { pollMs: 10 });
|
||||
|
||||
@@ -568,3 +568,67 @@ describe("terminal WebSocket send-at-open buffering (initialText race)", () => {
|
||||
expect(sentTexts).toEqual([]);
|
||||
}, 10000);
|
||||
});
|
||||
|
||||
// Exercise the real WebSocket callback: node-ws does not await async onOpen.
|
||||
describe("terminal WebSocket attach rejection", () => {
|
||||
it.each(["probe", "pipe"])("closes a rejected %s viewer and accepts a later viewer", async (stage) => {
|
||||
let fail = true;
|
||||
const sendText = vi.fn(async () => ({ ok: true as const }));
|
||||
const app = new Hono();
|
||||
app.use("*", async (c, next) => {
|
||||
c.set("tmuxAdapter" as never, {
|
||||
hasSession: async () => {
|
||||
if (fail && stage === "probe") throw new Error("probe unavailable");
|
||||
return true;
|
||||
},
|
||||
setWindowOption: async () => ({ ok: true }),
|
||||
resizeWindow: async () => ({ ok: true }),
|
||||
startPipePane: async () => {
|
||||
if (fail && stage === "pipe") throw new Error("pipe unavailable");
|
||||
return { ok: true };
|
||||
},
|
||||
stopPipePane: async () => ({ ok: true }),
|
||||
sendKeys: async () => ({ ok: true }),
|
||||
sendText,
|
||||
capturePaneScreen: async () => "ready",
|
||||
getPaneCursorPosition: async () => null,
|
||||
});
|
||||
await next();
|
||||
});
|
||||
app.use("/api/*", browserBoundary({ webUiEnabled: true, bearerTokens: [TOKEN], warn: () => {} }));
|
||||
app.get("/healthz", (c) => c.text("ok"));
|
||||
const { injectWebSocket, upgradeWebSocket } = createNodeWebSocket({ app });
|
||||
registerTerminalWs(app, upgradeWebSocket as never, { bearerToken: TOKEN });
|
||||
const localServer = serve({ fetch: app.fetch, port: 0, hostname: "127.0.0.1" });
|
||||
injectWebSocket(localServer);
|
||||
await new Promise<void>((resolve) => {
|
||||
if (localServer.listening) resolve();
|
||||
else localServer.once("listening", resolve);
|
||||
});
|
||||
const port = (localServer.address() as { port: number }).port;
|
||||
const url = `ws://127.0.0.1:${port}/api/terminal/retry@rig?token=${TOKEN}`;
|
||||
const viewers: WebSocket[] = [];
|
||||
try {
|
||||
const failed = new WebSocket(url);
|
||||
viewers.push(failed);
|
||||
failed.onopen = () => failed.send(JSON.stringify({ type: "text", text: "early input" }));
|
||||
const closed = await new Promise<CloseEvent>((resolve) => { failed.onclose = resolve; });
|
||||
expect(closed.code).toBe(1011);
|
||||
expect(closed.reason).toBe("terminal attach failed");
|
||||
expect(sendText).not.toHaveBeenCalled();
|
||||
expect((await fetch(`http://127.0.0.1:${port}/healthz`)).status).toBe(200);
|
||||
|
||||
fail = false;
|
||||
const retry = new WebSocket(url);
|
||||
viewers.push(retry);
|
||||
const frame = await new Promise<MessageEvent>((resolve) => { retry.onmessage = resolve; });
|
||||
expect(String(frame.data)).toContain("ready");
|
||||
const retryClosed = new Promise<void>((resolve) => { retry.onclose = () => resolve(); });
|
||||
retry.close();
|
||||
await retryClosed;
|
||||
} finally {
|
||||
for (const ws of viewers) ws.close();
|
||||
localServer.close();
|
||||
}
|
||||
}, 5000);
|
||||
});
|
||||
Reference in new issue
Block a user