WARNING: THIS SITE IS A MIRROR OF GITHUB.COM / IT CANNOT LOGIN OR REGISTER ACCOUNTS / THE CONTENTS ARE PROVIDED AS-IS / THIS SITE ASSUMES NO RESPONSIBILITY FOR ANY DISPLAYED CONTENT OR LINKS / IF YOU FOUND SOMETHING MAY NOT GOOD FOR EVERYONE, CONTACT ADMIN AT ilovescratch@foxmail.com
Skip to content

fix: replace heartbeat with keep alive #236

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
christianalfoni wants to merge 1 commit into from
Closed

fix: replace heartbeat with keep alive #236

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 0 additions & 13 deletions src/AgentClient/AgentConnection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,19 +93,6 @@ export class AgentConnection {
connection.onDisconnected(() => {
this.state = "DISCONNECTED";
});

connection.onMissingHeartbeat(() => {
// Be more conservative about disconnection - only disconnect if we have no activity
// and no pending messages, indicating a truly dead connection
if (this.pendingMessages.size === 0) {
// Add a small delay to allow for network recovery before declaring disconnection
setTimeout(() => {
if (this.pendingMessages.size === 0 && this.state === "CONNECTED") {
this.state = "DISCONNECTED";
}
}, 1000);
}
});
}

onNotification<T extends PitcherNotification["method"]>(
Expand Down
80 changes: 1 addition & 79 deletions src/AgentClient/WebSocketClient.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,22 +11,8 @@ export type DisconnectedEvent = {
wasClean: boolean;
};

// With our PONG detection we rather use an offset of that to determine how often we ping. If not we result
// in never detecting PONG timeouts as we run a new ping, clearing the PONG timeout, before we get a chance
// to detect any offline condition
let WEBSOCKET_PING_OFFSET = 2000;
// During init of PitcherClient we have a much longer timeout for detection of the "pong" response. The
// reason is that during initialization Pitcher might do a lot of heavy lifting and we want to avoid unnecessary
// reconnects during this, but still have some fallback for detecting an actual disconnect
let INIT_DETECT_PONG_TIMEOUT = 20_000;

const readyStateToString = ["CONNECTING", "OPEN", "CLOSING", "CLOSED"];

if (typeof process !== "undefined" && process.env.NODE_ENV === "test") {
WEBSOCKET_PING_OFFSET = 20;
INIT_DETECT_PONG_TIMEOUT = 200;
}

/*
This WebsocketClient is responsible for a single connection. That means when a disconnect happens a new WebsocketClient will be created. The
responsibility of this client is to keep the connection alive, detect disconnects and of course send messages. The following scenarios will cause
Expand All @@ -50,7 +36,6 @@ if (typeof process !== "undefined" && process.env.NODE_ENV === "test") {
*/
export class WebSocketClient extends Disposable {
private ws: WebSocket;
private pongDetectionTimeout = INIT_DETECT_PONG_TIMEOUT;
private bufferQueue = new SerialQueue("websocket-buffer-queue");
private onMessageEmitter: Emitter<Uint8Array> = new Emitter();
/**
Expand All @@ -61,19 +46,9 @@ export class WebSocketClient extends Disposable {
wasClean: boolean;
reason: string;
}> = new Emitter();
/**
* Whenever the heartbeat is missing. This normally detects a disconnect, but
* there can be other reasons like a slow network or a big pending message causing it
*/
private onMissingHeartbeatEmitter: Emitter<void> = new Emitter();

// This timeout is triggered when we send a ping, if we do not get a pong back within
// the timeout, we dispose of the websocket and go into disconnected state
private detectDisconnectByPongTimeout: number | undefined;

onMessage = this.onMessageEmitter.event;
onDisconnected = this.onDisconnectedEmitter.event;
onMissingHeartbeat = this.onMissingHeartbeatEmitter.event;

lastActivity = Date.now();

Expand All @@ -88,35 +63,15 @@ export class WebSocketClient extends Disposable {
this.ws = ws;
this.lastActivity = Date.now();

/**
* Our heartbeat has two purposes:
* 1. Keep the connection alive by pinging Pitcher when there is no other activity
* 2. Detect the lack of a PONG response to identify the lack of a connection
*/
const onHeartbeatInterval = () => {
const timeSinceActivity = Date.now() - this.lastActivity;

if (timeSinceActivity > this.pingTimeout) {
this.ping();
}
};

const heartbeatInterval = setInterval(
onHeartbeatInterval,
this.pingTimeout
);

const onMessageListener = (event: { data: WebsocketData }) => {
this.lastActivity = Date.now();

const data = event.data;

// We clear the PONG detection regardless of what message we got
clearTimeout(this.detectDisconnectByPongTimeout);

// Browser environment
if (typeof window !== "undefined" && data instanceof window.Blob) {
// To ensure that messages are emitted in order we use a serial queue

this.bufferQueue.add(async () => {
this.emitMessage(new Uint8Array(await data.arrayBuffer()));
});
Expand Down Expand Up @@ -165,10 +120,6 @@ export class WebSocketClient extends Disposable {
ws.addEventListener("error", onErrorListener);

this.onWillDispose(() => {
clearInterval(heartbeatInterval);
clearTimeout(this.pongDetectionTimeout);
clearTimeout(this.detectDisconnectByPongTimeout);

ws.removeEventListener("close", onCloseListener);
ws.removeEventListener("message", onMessageListener);
ws.removeEventListener("error", onErrorListener);
Expand All @@ -190,34 +141,6 @@ export class WebSocketClient extends Disposable {
this.onMessageEmitter.fire(message);
}

private get pingTimeout() {
return this.pongDetectionTimeout + WEBSOCKET_PING_OFFSET;
}

setPongDetectionTimeout(ms: number) {
this.pongDetectionTimeout = ms;
}

/**
We use "PING" to both keep the session alive, but also detect disconnects. Certain interactions, like
focusing the application should trigger an immediate "ping", which is why this is a public method. An optional
pong timeout can be set. This is useful to detect disconnects faster for example when focusing the application
*/
ping(pongTimeout?: number) {
clearTimeout(this.detectDisconnectByPongTimeout);
this.detectDisconnectByPongTimeout = setTimeout(() => {
this.onMissingHeartbeatEmitter.fire();
}, pongTimeout || this.pongDetectionTimeout) as unknown as number;
// Empty string is the payload of a heartbeat. We do not use
// ws.ping() here because it does not produce a pong response that is detectable and its
// callback does not give an error when internet is down
try {
this.send("");
} catch {
// We do not care about errors here
}
}

send(data: WebsocketData) {
// To avoid showing a bunch of errors when we already know the connection is down, we return early. A closing
// handshake can take up to 30 seconds, so this will help to more quickly close the connection related to
Expand All @@ -241,7 +164,6 @@ export class WebSocketClient extends Disposable {
// This is an async operation in Node, but to avoid wrapping every send in a promise, we
// rely on the error listener to deal with any errors. Any unsent messages will be timed out
// by our PendingMessage logic

this.ws.send(data);
}

Expand Down
9 changes: 1 addition & 8 deletions src/AgentClient/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,8 @@ import {
PickRawFsResult,
} from "../agent-client-interface";
import { AgentConnection } from "./AgentConnection";
import { Emitter, Event } from "../utils/event";
import { Emitter } from "../utils/event";
import { DEFAULT_SUBSCRIPTIONS, SandboxSession } from "../types";
import { SandboxClient } from "../SandboxClient";
import { InitStatus } from "../pitcher-protocol/messages/system";

// Timeout for detecting a pong response, leading to a forced disconnect
Expand Down Expand Up @@ -414,9 +413,6 @@ export class AgentClient implements IAgentClient {
// Connection is fully established after successful client/join
agentConnection.state = "CONNECTED";

// Now that we have initialized we set an appropriate timeout to more efficiently detect disconnects
agentConnection.connection.setPongDetectionTimeout(PONG_DETECTION_TIMEOUT);

return {
client: new AgentClient(getSession, agentConnection, {
sandboxId: session.sandboxId,
Expand Down Expand Up @@ -458,9 +454,6 @@ export class AgentClient implements IAgentClient {
this.isUpToDate = params.isUpToDate;
this.reconnectToken = params.reconnectToken;
}
ping() {
this.agentConnection.connection.ping(FOCUS_PONG_DETECTION_TIMEOUT);
}
async disconnect(): Promise<void> {
await this.agentConnection.disconnect();
}
Expand Down
73 changes: 13 additions & 60 deletions src/SandboxClient/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,10 +45,14 @@ export class SandboxClient {
if (session.isPint) {
const pintClient = await PintClient.create(session);
const progress = await pintClient.setup.getProgress();
return new SandboxClient(pintClient, {
hostToken: session.hostToken,
tracer,
}, progress);
return new SandboxClient(
pintClient,
{
hostToken: session.hostToken,
tracer,
},
progress
);
}

const { client: agentClient, joinResult } = await AgentClient.create({
Expand Down Expand Up @@ -151,13 +155,6 @@ export class SandboxClient {
initialSetupProgress: setup.SetupProgress
) {
this.tracer = tracer;
// TODO: Bring this back once metrics polling does not reset inactivity
// const metricsDisposable = {
// dispose:
// this.pitcherClient.clients.system.startMetricsPollingAtInterval(5000),
// };

// this.addDisposable(metricsDisposable);
this.setup = new Setup(
this.disposable,
this.agentClient,
Expand Down Expand Up @@ -216,6 +213,10 @@ export class SandboxClient {
}
}
});

if (this.shouldKeepAlive) {
this.keepActiveWhileConnected(true);
}
}

private async withSpan<T>(
Expand Down Expand Up @@ -288,54 +289,6 @@ export class SandboxClient {
return `https://codesandbox.io/p/devbox/${this.id}`;
}

// TODO: Bring this back once metrics polling does not reset inactivity
// /**
// * Get the current system metrics. This return type may change in the future.
// */
// public async getMetrics(): Promise<SystemMetricsStatus> {
// await this.pitcherClient.clients.system.update();

// const barrier = new Barrier<_protocol.system.SystemMetricsStatus>();
// const initialMetrics = this.pitcherClient.clients.system.getMetrics();
// if (!initialMetrics) {
// const disposable = this.pitcherClient.clients.system.onMetricsUpdated(
// (metrics) => {
// if (metrics) {
// barrier.open(metrics);
// }
// }
// );
// disposable.dispose();
// } else {
// barrier.open(initialMetrics);
// }

// const barrierResult = await barrier.wait();
// if (barrierResult.status === "disposed") {
// throw new Error("Metrics not available");
// }

// const metrics = barrierResult.value;

// return {
// cpu: {
// cores: metrics.cpu.cores,
// used: metrics.cpu.used / 100,
// configured: metrics.cpu.configured,
// },
// memory: {
// usedKiB: metrics.memory.used * 1024 * 1024,
// totalKiB: metrics.memory.total * 1024 * 1024,
// configuredKiB: metrics.memory.total * 1024 * 1024,
// },
// storage: {
// usedKB: metrics.storage.used * 1000 * 1000,
// totalKB: metrics.storage.total * 1000 * 1000,
// configuredKB: metrics.storage.configured * 1000 * 1000,
// },
// };
// }

/**
* Disconnect from the sandbox, this does not hibernate the sandbox (it will
* automatically hibernate after hibernation timeout). Call "reconnect" to
Expand Down Expand Up @@ -414,7 +367,7 @@ export class SandboxClient {
}

private keepAliveInterval: NodeJS.Timeout | null = null;
private shouldKeepAlive = false;
private shouldKeepAlive = true;
private isExplicitlyDisconnected = false;
private keepAliveFailures = 0;
private maxKeepAliveFailures = 3;
Expand Down
1 change: 0 additions & 1 deletion src/agent-client-interface.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,6 @@ export interface IAgentClient {
setup: IAgentClientSetup;
tasks: IAgentClientTasks;
system: IAgentClientSystem;
ping(): void;
disconnect(): Promise<void>;
reconnect(): Promise<void>;
dispose(): void;
Expand Down
6 changes: 1 addition & 5 deletions src/browser/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,11 +43,7 @@ export async function connectToSandbox({
onFocusChange((isFocused) => {
// We immediately ping the connection when focusing, so that
// we detect a disconnect as early as possible
if (isFocused && client.state === "CONNECTED") {
client["agentClient"].ping();
// If we happen to be disconnected when focusing we try to reconnect, but only if we are currently
// hibernated and we did not do a manual disconnect
} else if (
if (
isFocused &&
(client.state === "DISCONNECTED" || client.state === "HIBERNATED")
) {
Expand Down
14 changes: 7 additions & 7 deletions src/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,16 +100,16 @@ export const DEFAULT_SUBSCRIPTIONS = {
status: true,
},
file: {
status: true,
selection: true,
ot: true,
status: false,
selection: false,
ot: false,
},
fs: {
operations: true,
operations: false,
},
git: {
status: true,
operations: true,
status: false,
operations: false,
},
port: {
status: true,
Expand All @@ -121,7 +121,7 @@ export const DEFAULT_SUBSCRIPTIONS = {
status: true,
},
system: {
metrics: true,
metrics: false,
},
};

Expand Down