fix(device-edge): bound stalled Core channel connects

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(

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",