mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bound provider egress tunnel lifecycle

Browse Source
This commit is contained in:
Codex
2026-05-17 08:39:03 +00:00
parent 0ed5c31509
commit dd448c7218
6 changed files with 154 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/backend-core/src/egress-tcp.ts
+47 -2
View File
@@ -23,6 +23,9 @@ function isValidEgressPort(port: number): boolean {
return Number.isInteger(port) && port > 0 && port <= 65_535;
}
const egressTcpConnectTimeoutMs = 15_000;
const egressTcpIdleTimeoutMs = 600_000;
function sendEgressClose(provider: ProviderSocket, connectionId: string, error?: string): void {
const message: CoreEgressTcpCloseMessage = error === undefined
? { type: "egress_tcp_close", connectionId }
@@ -30,13 +33,31 @@ function sendEgressClose(provider: ProviderSocket, connectionId: string, error?:
wsSendJson(provider, message);
}
function clearConnectionTimers(connection: EgressTcpConnection): void {
if (connection.connectTimer !== null) clearTimeout(connection.connectTimer);
if (connection.idleTimer !== null) clearTimeout(connection.idleTimer);
connection.connectTimer = null;
connection.idleTimer = null;
}
function closeEgressTcpConnection(providerId: string, connectionId: string, error?: string): void {
const key = egressTcpKey(providerId, connectionId);
const connection = ctx.activeEgressTcpConnections.get(key);
if (connection === undefined) return;
ctx.activeEgressTcpConnections.delete(key);
clearConnectionTimers(connection);
connection.socket.destroy();
if (error !== undefined) sendEgressClose(connection.provider, connectionId, error);
if (error !== undefined) {
logger("warn", "egress_tcp_connection_closed", { providerId, connectionId, error });
sendEgressClose(connection.provider, connectionId, error);
}
}
function refreshConnectionIdle(connection: EgressTcpConnection): void {
if (connection.idleTimer !== null) clearTimeout(connection.idleTimer);
connection.idleTimer = setTimeout(() => {
closeEgressTcpConnection(connection.providerId, connection.connectionId, "egress tcp idle timeout");
}, egressTcpIdleTimeoutMs);
}
export function handleEgressTcpOpen(ws: ProviderSocket, message: ProviderEgressTcpOpenMessage): void {
@@ -49,13 +70,32 @@ export function handleEgressTcpOpen(ws: ProviderSocket, message: ProviderEgressT
const key = egressTcpKey(message.providerId, message.connectionId);
closeEgressTcpConnection(message.providerId, message.connectionId);
const socket = connectTcp({ host, port });
const connection: EgressTcpConnection = { providerId: message.providerId, connectionId: message.connectionId, socket, provider: ws };
const connection: EgressTcpConnection = {
providerId: message.providerId,
connectionId: message.connectionId,
socket,
provider: ws,
connectTimer: null,
idleTimer: null,
};
ctx.activeEgressTcpConnections.set(key, connection);
connection.connectTimer = setTimeout(() => {
closeEgressTcpConnection(message.providerId, message.connectionId, "egress tcp connect timeout");
}, egressTcpConnectTimeoutMs);
refreshConnectionIdle(connection);
socket.on("connect", () => {
if (ctx.activeEgressTcpConnections.get(key) !== connection) return;
if (connection.connectTimer !== null) {
clearTimeout(connection.connectTimer);
connection.connectTimer = null;
}
refreshConnectionIdle(connection);
const opened: CoreEgressTcpOpenedMessage = { type: "egress_tcp_opened", connectionId: message.connectionId };
wsSendJson(ws, opened);
});
socket.on("data", (chunk) => {
if (ctx.activeEgressTcpConnections.get(key) !== connection) return;
refreshConnectionIdle(connection);
const data: CoreEgressTcpDataMessage = {
type: "egress_tcp_data",
connectionId: message.connectionId,
@@ -67,11 +107,13 @@ export function handleEgressTcpOpen(ws: ProviderSocket, message: ProviderEgressT
socket.on("close", () => {
if (ctx.activeEgressTcpConnections.get(key) !== connection) return;
ctx.activeEgressTcpConnections.delete(key);
clearConnectionTimers(connection);
sendEgressClose(ws, message.connectionId);
});
socket.on("error", (error) => {
if (ctx.activeEgressTcpConnections.get(key) !== connection) return;
ctx.activeEgressTcpConnections.delete(key);
clearConnectionTimers(connection);
sendEgressClose(ws, message.connectionId, error.message);
});
}
@@ -79,6 +121,7 @@ export function handleEgressTcpOpen(ws: ProviderSocket, message: ProviderEgressT
export function handleEgressTcpData(message: ProviderEgressTcpDataMessage): void {
const connection = ctx.activeEgressTcpConnections.get(egressTcpKey(message.providerId, message.connectionId));
if (connection === undefined) return;
refreshConnectionIdle(connection);
connection.socket.write(Buffer.from(message.data, message.encoding === "base64" ? "base64" : "utf8"));
}
@@ -90,6 +133,7 @@ export function closeEgressTcpConnectionsForProvider(providerId: string): void {
for (const [key, connection] of ctx.activeEgressTcpConnections) {
if (connection.providerId !== providerId) continue;
ctx.activeEgressTcpConnections.delete(key);
clearConnectionTimers(connection);
connection.socket.destroy();
}
}
@@ -98,6 +142,7 @@ export function closeEgressTcpConnectionsForSocket(provider: ProviderSocket): vo
for (const [key, connection] of ctx.activeEgressTcpConnections) {
if (connection.provider !== provider) continue;
ctx.activeEgressTcpConnections.delete(key);
clearConnectionTimers(connection);
connection.socket.destroy();
}
}
src/components/backend-core/src/types.ts
+2
View File
@@ -166,6 +166,8 @@ export interface EgressTcpConnection {
connectionId: string;
socket: Socket;
provider: ProviderSocket;
connectTimer: ReturnType<typeof setTimeout> | null;
idleTimer: ReturnType<typeof setTimeout> | null;
}
export type HttpTunnelFailureReason =
src/components/provider-gateway/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@unidesk/provider-gateway",
"version": "0.2.21",
"version": "0.2.22",
"private": true,
"type": "module",
"scripts": {
src/components/provider-gateway/src/egress-proxy.ts
+100 -21
View File
@@ -29,7 +29,12 @@ interface Tunnel {
method: string;
opened: boolean;
pending: Buffer[];
pendingBytes: number;
closed: boolean;
createdAt: number;
lastActivityAt: number;
openTimer: ReturnType<typeof setTimeout> | null;
idleTimer: ReturnType<typeof setTimeout> | null;
}
export interface ProviderEgressProxyHandle {
@@ -111,6 +116,10 @@ function proxyUrlFor(host: string, port: number): string {
return `http://${host}:${port}`;
}
const tunnelOpenTimeoutMs = 15_000;
const tunnelIdleTimeoutMs = 600_000;
const maxPendingBytes = 4 * 1024 * 1024;
export function startProviderEgressProxy(options: ProviderEgressProxyOptions): ProviderEgressProxyHandle {
const proxyUrl = proxyUrlFor(options.listenHost, options.listenPort);
const tunnels = new Map<string, Tunnel>();
@@ -118,25 +127,73 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
const send = (message: EgressToCoreMessage): boolean => options.sendToCore(message);
const status = (): Record<string, JsonValue> => ({
enabled: true,
providerId: options.providerId,
connected: options.isCoreConnected(),
proxyUrl,
listenHost: options.listenHost,
listenPort: options.listenPort,
activeTunnels: tunnels.size,
channel: "provider-gateway",
});
const status = (): Record<string, JsonValue> => {
const now = Date.now();
const tunnelList = Array.from(tunnels.values());
const ages = tunnelList.map((tunnel) => now - tunnel.createdAt);
return {
enabled: true,
providerId: options.providerId,
connected: options.isCoreConnected(),
proxyUrl,
listenHost: options.listenHost,
listenPort: options.listenPort,
activeTunnels: tunnels.size,
pendingTunnels: tunnelList.filter((tunnel) => !tunnel.opened).length,
oldestTunnelAgeMs: ages.length > 0 ? Math.max(...ages) : 0,
openTimeoutMs: tunnelOpenTimeoutMs,
idleTimeoutMs: tunnelIdleTimeoutMs,
maxPendingBytes,
channel: "provider-gateway",
};
};
const clearTunnelTimers = (tunnel: Tunnel): void => {
if (tunnel.openTimer !== null) clearTimeout(tunnel.openTimer);
if (tunnel.idleTimer !== null) clearTimeout(tunnel.idleTimer);
tunnel.openTimer = null;
tunnel.idleTimer = null;
};
const destroyTunnel = (tunnel: Tunnel, notifyCore: boolean, error?: string): void => {
tunnels.delete(tunnel.id);
tunnel.closed = true;
clearTunnelTimers(tunnel);
tunnel.pending.splice(0);
tunnel.pendingBytes = 0;
if (!tunnel.client.destroyed) tunnel.client.destroy();
if (notifyCore) send({ type: "egress_tcp_close", providerId: options.providerId, connectionId: tunnel.id, at: nowIso() });
if (error !== undefined) {
options.logger("warn", "egress_proxy_tunnel_closed", {
connectionId: tunnel.id,
opened: tunnel.opened,
ageMs: Date.now() - tunnel.createdAt,
error,
});
}
};
const closeTunnel = (id: string, error?: string): void => {
const tunnel = tunnels.get(id);
if (tunnel === undefined) return;
tunnels.delete(id);
tunnel.closed = true;
if (!tunnel.client.destroyed) tunnel.client.destroy();
send({ type: "egress_tcp_close", providerId: options.providerId, connectionId: id, at: nowIso() });
if (error !== undefined) options.logger("warn", "egress_proxy_tunnel_closed", { connectionId: id, error });
destroyTunnel(tunnel, true, error);
};
const refreshTunnelIdle = (tunnel: Tunnel): void => {
if (tunnel.closed) return;
tunnel.lastActivityAt = Date.now();
if (tunnel.idleTimer !== null) clearTimeout(tunnel.idleTimer);
tunnel.idleTimer = setTimeout(() => closeTunnel(tunnel.id, "egress proxy idle timeout"), tunnelIdleTimeoutMs);
};
const queuePendingChunk = (tunnel: Tunnel, chunk: Buffer): boolean => {
if (tunnel.closed) return false;
tunnel.pending.push(chunk);
tunnel.pendingBytes += chunk.byteLength;
refreshTunnelIdle(tunnel);
if (tunnel.pendingBytes <= maxPendingBytes) return true;
closeTunnel(tunnel.id, "egress proxy pending buffer exceeded");
return false;
};
const handleCoreMessage = (message: EgressFromCoreMessage): boolean => {
@@ -144,24 +201,29 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
if (message.type === "egress_tcp_opened") {
if (tunnel === undefined || tunnel.closed) return true;
tunnel.opened = true;
if (tunnel.openTimer !== null) {
clearTimeout(tunnel.openTimer);
tunnel.openTimer = null;
}
refreshTunnelIdle(tunnel);
if (tunnel.method === "CONNECT") {
tunnel.client.write("HTTP/1.1 200 Connection Established\r\nProxy-Agent: UniDesk-ProviderGateway\r\n\r\n");
}
for (const chunk of tunnel.pending.splice(0)) {
send({ type: "egress_tcp_data", providerId: options.providerId, connectionId: tunnel.id, data: chunk.toString("base64"), encoding: "base64", at: nowIso() });
}
tunnel.pendingBytes = 0;
return true;
}
if (message.type === "egress_tcp_data") {
if (tunnel === undefined || tunnel.client.destroyed) return true;
refreshTunnelIdle(tunnel);
tunnel.client.write(Buffer.from(message.data, message.encoding === "base64" ? "base64" : "utf8"));
return true;
}
if (message.type === "egress_tcp_close") {
if (tunnel !== undefined) {
tunnels.delete(message.connectionId);
tunnel.closed = true;
if (!tunnel.client.destroyed) tunnel.client.destroy();
destroyTunnel(tunnel, false, message.error);
}
if (message.error !== undefined && message.error.length > 0) {
options.logger("warn", "egress_proxy_remote_close", { connectionId: message.connectionId, error: message.error });
@@ -222,14 +284,28 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
fail("400 Bad Request", "unsupported proxy target\n");
return;
}
const tunnel: Tunnel = { id, client, method: parsed.method, opened: false, pending: [], closed: false };
const createdAt = Date.now();
const tunnel: Tunnel = {
id,
client,
method: parsed.method,
opened: false,
pending: [],
pendingBytes: 0,
closed: false,
createdAt,
lastActivityAt: createdAt,
openTimer: null,
idleTimer: null,
};
tunnels.set(id, tunnel);
client.on("data", (nextChunk) => {
const nextBuffer = Buffer.isBuffer(nextChunk) ? nextChunk : Buffer.from(nextChunk);
if (!tunnel.opened) {
tunnel.pending.push(nextBuffer);
queuePendingChunk(tunnel, nextBuffer);
return;
}
refreshTunnelIdle(tunnel);
send({ type: "egress_tcp_data", providerId: options.providerId, connectionId: id, data: nextBuffer.toString("base64"), encoding: "base64", at: nowIso() });
});
client.on("close", () => closeTunnel(id));
@@ -238,10 +314,13 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
if (!opened) {
tunnels.delete(id);
tunnel.closed = true;
clearTunnelTimers(tunnel);
fail("503 Service Unavailable", "provider-gateway core channel is not connected\n");
return;
}
if (firstPayload !== null) tunnel.pending.push(firstPayload);
tunnel.openTimer = setTimeout(() => closeTunnel(id, "egress proxy open timeout"), tunnelOpenTimeoutMs);
refreshTunnelIdle(tunnel);
if (firstPayload !== null) queuePendingChunk(tunnel, firstPayload);
});
client.on("error", () => undefined);
});