diff --git a/device-plane/services/device-control-core/src/device-gateway-core-runtime.mjs b/device-plane/services/device-control-core/src/device-gateway-core-runtime.mjs index bf855e3..ef2574a 100644 --- a/device-plane/services/device-control-core/src/device-gateway-core-runtime.mjs +++ b/device-plane/services/device-control-core/src/device-gateway-core-runtime.mjs @@ -19,6 +19,7 @@ import { // Runtime-owned transport implementation; kept inside the deployable Core context. const CHANNEL_TRACKER_SESSION_ID = "channel:control"; const CHANNEL_PROFILE_REF = "channel.control.v1"; +const DEFAULT_CONNECT_TIMEOUT_MS = 10_000; export function createDeviceGatewayCoreChannelClient(options = {}) { const config = normalizeConfig(options); @@ -129,6 +130,7 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { edgeSequence: 0, coreSequence: 0, lastEdgeActivityAt: config.clock(), + connectTimer: null, heartbeatTimer: null, ready: false, closed: false, @@ -151,6 +153,12 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { }, }); connection.session = session; + connection.connectTimer = setTimeout(() => { + failConnection(connection, new Error( + "device_gateway_core_channel_connect_timeout", + )); + }, config.connectTimeoutMs); + connection.connectTimer.unref?.(); session.once("error", (error) => failConnection(connection, error)); session.once("close", () => closeConnection(connection, true)); session.once("connect", () => { @@ -250,6 +258,8 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { correlationId: envelope.correlationId, }); connection.ready = true; + clearTimeout(connection.connectTimer); + connection.connectTimer = null; reconnectAttempt = 0; totalChannelsAccepted += 1; lastErrorCode = null; @@ -403,6 +413,8 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { function closeConnection(connection, reconnect) { if (connection.closed) return; connection.closed = true; + clearTimeout(connection.connectTimer); + connection.connectTimer = null; clearInterval(connection.heartbeatTimer); connection.heartbeatTimer = null; connection.sessionChains.clear(); @@ -506,6 +518,13 @@ function normalizeConfig(options) { acceptMessage: options.acceptMessage, keepaliveMs, deadPeerMs, + connectTimeoutMs: normalizeInteger( + options.connectTimeoutMs, + 10, + 120_000, + DEFAULT_CONNECT_TIMEOUT_MS, + "connect_timeout", + ), reconnectMinimumMs, reconnectMaximumMs, maxEnvelopeBytes: normalizeInteger( diff --git a/device-plane/services/device-edge-channel/test/channel-integration.test.mjs b/device-plane/services/device-edge-channel/test/channel-integration.test.mjs index 25881e5..c1a2eaa 100644 --- a/device-plane/services/device-edge-channel/test/channel-integration.test.mjs +++ b/device-plane/services/device-edge-channel/test/channel-integration.test.mjs @@ -3,6 +3,7 @@ import { spawnSync } from "node:child_process"; import { randomUUID, X509Certificate } from "node:crypto"; import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { connect as connectHttp2 } from "node:http2"; +import { createServer as createTcpServer } from "node:net"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { after, before, test } from "node:test"; @@ -94,6 +95,34 @@ test("keeps the channel alive and reconnects without losing idempotency", async } }); +test("bounds a stalled TCP/TLS handshake and reconnects", async () => { + const sockets = new Set(); + const server = createTcpServer((socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const core = createCoreClient({ + address: server.address(), + connectTimeoutMs: 30, + reconnectMinimumMs: 20, + reconnectMaximumMs: 20, + }); + try { + await core.start(); + await waitFor(() => core.status().connectionAttempts >= 2, 1_000); + assert.match(core.status().lastErrorCode, /connect_timeout/); + assert.ok(core.status().reconnects >= 1); + } finally { + await core.stop(); + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => server.close(resolve)); + } +}); + test("rotates the Edge certificate through staged overlap and rejects retired identity", async () => { const edge = createEdgeServer(); const address = await edge.start(); @@ -430,6 +459,7 @@ function createCoreClient(options) { deadPeerMs: options.deadPeerMs ?? 150, reconnectMinimumMs: options.reconnectMinimumMs ?? 20, reconnectMaximumMs: options.reconnectMaximumMs ?? 80, + connectTimeoutMs: options.connectTimeoutMs, random: () => 0, observeDiscovery: options.observeDiscovery ?? (async () => ({ schemaVersion: "nodedc.device.discovery-view.v1",