dilim/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-10 17:16:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ui): tolerate transient gateway stalls in terminal liveness probe (#109317)

Browse Source
This commit is contained in:
Peter Steinberger
2026-07-16 14:45:10 -07:00
committed by GitHub
parent 7bc2b96c82
commit d7fe43acc3
2 changed files with 155 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
ui/src/components/terminal/terminal-connection.test.ts
View File

@@ -7,7 +7,10 @@ import {
const TERMINAL_LIVENESS_IDLE_MS = 20_000;
const TERMINAL_LIVENESS_PROBE_TIMEOUT_MS = 5_000;
const TERMINAL_LIVENESS_FAILURE_RETRY_MS = 5_000;
const TERMINAL_OPEN_WATCHDOG_MS = 35_000;
// Idle window elapses, then one probe times out: the interval after which a probe resolves failed.
const IDLE_PLUS_PROBE_MS = TERMINAL_LIVENESS_IDLE_MS + TERMINAL_LIVENESS_PROBE_TIMEOUT_MS;
function deferred<T>() {
let resolve!: (value: T) => void;
@@ -69,6 +72,31 @@ function makeFakeClient() {
return client;
}
function setLivenessProbeOutcomes(
client: ReturnType<typeof makeFakeClient>,
outcomes: readonly ("success" | "timeout")[],
): void {
const baseRequest = client.request.bind(client);
let probeIndex = 0;
client.request = (<T>(
method: string,
params?: unknown,
options?: { timeoutMs?: number | null },
): Promise<T> => {
if (method !== "terminal.list") {
return baseRequest<T>(method, params, options);
}
client.requests.push({ method, params, ...(options ? { options } : {}) });
const outcome = outcomes[probeIndex++] ?? "timeout";
if (outcome === "success") {
return Promise.resolve({ sessions: [] } as T);
}
return new Promise<T>((_, reject) => {
setTimeout(() => reject(new Error("request timed out")), options?.timeoutMs ?? 0);
});
}) as typeof client.request;
}
describe("TerminalConnection", () => {
it("opens a session and routes its data to the registered sink", async () => {
const client = makeFakeClient();
@@ -963,34 +991,41 @@ describe("TerminalConnection", () => {
expect(await conn.list()).toEqual([]);
});
it("forces reconnect when a terminal liveness probe gets no inbound response", async () => {
it("keeps the socket after one failed liveness probe and retries on a short backoff", async () => {
vi.useFakeTimers();
try {
const client = makeFakeClient();
client.request = ((
method: string,
params: unknown,
options?: { timeoutMs?: number | null },
) => {
client.requests.push({ method, params, ...(options ? { options } : {}) });
if (method === "terminal.list") {
return new Promise((_, reject) => {
setTimeout(() => reject(new Error("request timed out")), options?.timeoutMs ?? 0);
});
}
return Promise.resolve(client.nextResponse);
}) as typeof client.request;
setLivenessProbeOutcomes(client, ["timeout"]);
const conn = new TerminalConnection(client);
await conn.open({ cols: 80, rows: 24 }, { onData: () => {}, onExit: () => {} });
const probes = () =>
client.requests.filter((request) => request.method === "terminal.list").length;
await vi.advanceTimersByTimeAsync(IDLE_PLUS_PROBE_MS);
expect(probes()).toBe(1);
// A single failure only schedules the short retry; it never tears down the socket.
expect(client.forceReconnects).toEqual([]);
await vi.advanceTimersByTimeAsync(TERMINAL_LIVENESS_FAILURE_RETRY_MS);
expect(probes()).toBe(2);
conn.dispose();
} finally {
vi.useRealTimers();
}
});
it("forces exactly one reconnect after two consecutive failed liveness probes", async () => {
vi.useFakeTimers();
try {
const client = makeFakeClient();
setLivenessProbeOutcomes(client, ["timeout", "timeout"]);
const conn = new TerminalConnection(client);
await conn.open({ cols: 80, rows: 24 }, { onData: () => {}, onExit: () => {} });
await vi.advanceTimersByTimeAsync(TERMINAL_LIVENESS_IDLE_MS);
expect(client.requests.at(-1)).toEqual({
method: "terminal.list",
params: undefined,
options: { timeoutMs: TERMINAL_LIVENESS_PROBE_TIMEOUT_MS },
});
await vi.advanceTimersByTimeAsync(TERMINAL_LIVENESS_PROBE_TIMEOUT_MS);
await vi.advanceTimersByTimeAsync(IDLE_PLUS_PROBE_MS);
expect(client.forceReconnects).toEqual([]);
await vi.advanceTimersByTimeAsync(
TERMINAL_LIVENESS_FAILURE_RETRY_MS + TERMINAL_LIVENESS_PROBE_TIMEOUT_MS,
);
expect(client.forceReconnects).toEqual(["terminal liveness timeout"]);
conn.dispose();
} finally {
@@ -998,29 +1033,37 @@ describe("TerminalConnection", () => {
}
});
it("resets liveness failures after a successful probe", async () => {
vi.useFakeTimers();
try {
const client = makeFakeClient();
setLivenessProbeOutcomes(client, ["timeout", "success", "timeout"]);
const conn = new TerminalConnection(client);
await conn.open({ cols: 80, rows: 24 }, { onData: () => {}, onExit: () => {} });
// Probes: timeout (fail), success (clears the streak), timeout (fail again).
await vi.advanceTimersByTimeAsync(IDLE_PLUS_PROBE_MS);
await vi.advanceTimersByTimeAsync(TERMINAL_LIVENESS_FAILURE_RETRY_MS + IDLE_PLUS_PROBE_MS);
// The middle success reset the streak, so the later lone failure cannot reconnect.
expect(client.forceReconnects).toEqual([]);
conn.dispose();
} finally {
vi.useRealTimers();
}
});
it("keeps the socket when other inbound traffic arrives during a failed probe", async () => {
vi.useFakeTimers();
try {
const client = makeFakeClient();
client.request = ((
method: string,
params: unknown,
options?: { timeoutMs?: number | null },
) => {
client.requests.push({ method, params, ...(options ? { options } : {}) });
if (method === "terminal.list") {
return new Promise((_, reject) => {
setTimeout(() => reject(new Error("request timed out")), options?.timeoutMs ?? 0);
});
}
return Promise.resolve(client.nextResponse);
}) as typeof client.request;
setLivenessProbeOutcomes(client, ["timeout", "timeout"]);
const conn = new TerminalConnection(client);
await conn.open({ cols: 80, rows: 24 }, { onData: () => {}, onExit: () => {} });
await vi.advanceTimersByTimeAsync(TERMINAL_LIVENESS_IDLE_MS);
// A frame delivered mid-probe proves the socket alive, so the probe timeout is not counted.
client.emitActivity();
await vi.advanceTimersByTimeAsync(TERMINAL_LIVENESS_PROBE_TIMEOUT_MS);
await vi.advanceTimersByTimeAsync(TERMINAL_LIVENESS_PROBE_TIMEOUT_MS + IDLE_PLUS_PROBE_MS);
expect(client.forceReconnects).toEqual([]);
conn.dispose();
@@ -1029,6 +1072,35 @@ describe("TerminalConnection", () => {
}
});
it("restarts the full idle window when inbound traffic arrives during the retry backoff", async () => {
vi.useFakeTimers();
try {
const client = makeFakeClient();
setLivenessProbeOutcomes(client, ["timeout", "timeout"]);
const conn = new TerminalConnection(client);
await conn.open({ cols: 80, rows: 24 }, { onData: () => {}, onExit: () => {} });
const probeCount = () =>
client.requests.filter((request) => request.method === "terminal.list").length;
// First probe fails and schedules the short 5s retry.
await vi.advanceTimersByTimeAsync(IDLE_PLUS_PROBE_MS);
expect(probeCount()).toBe(1);
// A non-terminal frame proves the socket alive during the backoff: the next check treats it
// as fresh activity and waits a full idle window instead of re-probing on the short retry, so
// no second probe fires and no reconnect happens.
client.emitActivity();
await vi.advanceTimersByTimeAsync(
TERMINAL_LIVENESS_FAILURE_RETRY_MS + TERMINAL_LIVENESS_PROBE_TIMEOUT_MS,
);
expect(probeCount()).toBe(1);
expect(client.forceReconnects).toEqual([]);
conn.dispose();
} finally {
vi.useRealTimers();
}
});
it("dispose() drops the gateway subscription and clears buffered state", async () => {
const client = makeFakeClient();
const conn = new TerminalConnection(client);

55
ui/src/components/terminal/terminal-connection.ts
View File

@@ -77,6 +77,8 @@ type PendingEvent =
const TERMINAL_LIVENESS_IDLE_MS = 20_000;
const TERMINAL_LIVENESS_PROBE_TIMEOUT_MS = 5_000;
const TERMINAL_LIVENESS_MAX_CONSECUTIVE_FAILURES = 2;
const TERMINAL_LIVENESS_FAILURE_RETRY_MS = 5_000;
// The Gateway owns the 30s open deadline. This longer browser watchdog only
// recovers a half-open socket when the Gateway's response cannot arrive.
const TERMINAL_OPEN_WATCHDOG_MS = 35_000;
@@ -112,6 +114,8 @@ export class TerminalConnection {
private pendingOpenCount = 0;
private livenessTimer: ReturnType<typeof setTimeout> | null = null;
private livenessProbeInFlight = false;
private livenessProbeFailures = 0;
private lastLivenessFailureActivityVersion: number | null = null;
private lastTerminalActivityAtMs = Date.now();
private inboundActivityVersion = 0;
@@ -200,7 +204,7 @@ export class TerminalConnection {
if (isTerminalOpenRequestTimeout(error)) {
// The server should answer first. A later browser timeout means this
// socket cannot carry the response, so disconnect to cancel ownership.
this.client.forceReconnect("terminal open watchdog timeout");
this.forceReconnect("terminal open watchdog timeout");
}
throw new TerminalOpenTimeoutError(error);
}
@@ -346,7 +350,7 @@ export class TerminalConnection {
// would duplicate bytes already rendered before the detected gap.
stream.recovering = false;
this.pending.delete(sessionId);
this.client.forceReconnect("terminal replay reset unavailable");
this.forceReconnect("terminal replay reset unavailable");
return;
}
// The ring may include both bytes already delivered and the gap's
@@ -384,7 +388,7 @@ export class TerminalConnection {
}
stream.recovering = false;
this.pending.delete(sessionId);
this.client.forceReconnect("terminal replay failed");
this.forceReconnect("terminal replay failed");
});
}
@@ -449,10 +453,21 @@ export class TerminalConnection {
/** Terminal traffic delays the next probe without resetting a timer per chunk. */
private noteTerminalActivity(): void {
this.resetLivenessProbeFailures();
this.lastTerminalActivityAtMs = Date.now();
this.inboundActivityVersion += 1;
}
private forceReconnect(reason: string): void {
this.resetLivenessProbeFailures();
this.client.forceReconnect(reason);
}
private resetLivenessProbeFailures(): void {
this.livenessProbeFailures = 0;
this.lastLivenessFailureActivityVersion = null;
}
private scheduleLivenessCheck(delayMs = TERMINAL_LIVENESS_IDLE_MS): void {
if (this.livenessTimer || this.livenessProbeInFlight || this.streams.size === 0) {
return;
@@ -476,29 +491,52 @@ export class TerminalConnection {
return;
}
const activityBefore = this.client.inboundActivitySeq ?? this.inboundActivityVersion;
if (
this.lastLivenessFailureActivityVersion !== null &&
activityBefore !== this.lastLivenessFailureActivityVersion
) {
// A frame arrived since the last failed probe, so the socket is proven alive. Treat it like
// any other activity: restart the full idle window instead of immediately re-probing on the
// short failure-retry backoff (matches the during-probe activity path).
this.resetLivenessProbeFailures();
this.lastTerminalActivityAtMs = Date.now();
this.scheduleLivenessCheck();
return;
}
let nextDelayMs = TERMINAL_LIVENESS_IDLE_MS;
this.livenessProbeInFlight = true;
void this.client
.request("terminal.list", undefined, { timeoutMs: TERMINAL_LIVENESS_PROBE_TIMEOUT_MS })
.then(() => {
// The response itself proves the inbound half of the socket is alive.
this.resetLivenessProbeFailures();
this.lastTerminalActivityAtMs = Date.now();
})
.catch(() => {
if (this.streams.size === 0) {
this.resetLivenessProbeFailures();
return;
}
const activityNow = this.client.inboundActivitySeq ?? this.inboundActivityVersion;
if (activityNow === activityBefore) {
this.lastTerminalActivityAtMs = Date.now();
this.client.forceReconnect("terminal liveness timeout");
} else {
if (activityNow !== activityBefore) {
// Any valid inbound frame proves the socket is not half-open.
this.resetLivenessProbeFailures();
this.lastTerminalActivityAtMs = Date.now();
return;
}
this.livenessProbeFailures += 1;
this.lastLivenessFailureActivityVersion = activityNow;
if (this.livenessProbeFailures >= TERMINAL_LIVENESS_MAX_CONSECUTIVE_FAILURES) {
// One probe cannot distinguish a dead socket from a stalled Gateway event loop.
this.forceReconnect("terminal liveness timeout");
return;
}
// Keep the old activity time so the short retry performs another probe.
nextDelayMs = TERMINAL_LIVENESS_FAILURE_RETRY_MS;
})
.finally(() => {
this.livenessProbeInFlight = false;
this.scheduleLivenessCheck();
this.scheduleLivenessCheck(nextDelayMs);
});
}
@@ -540,6 +578,7 @@ export class TerminalConnection {
}
private stopLiveness(): void {
this.resetLivenessProbeFailures();
if (this.livenessTimer) {
clearTimeout(this.livenessTimer);
this.livenessTimer = null;