feat(device-plane): add core-initiated edge channel
This commit is contained in:
Codex
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"name": "@nodedc/device-edge-channel",
|
||||
"version": "0.1.0",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"exports": {
|
||||
".": "./src/runtime.mjs"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=20"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,495 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createSecureServer } from "node:http2";
|
||||
|
||||
import {
|
||||
DEVICE_EDGE_CHANNEL_LIMITS,
|
||||
DEVICE_EDGE_CHANNEL_PATH,
|
||||
createChannelEnvelope,
|
||||
createChannelEnvelopeDecoder,
|
||||
encodeChannelEnvelope,
|
||||
normalizeCertificateFingerprint,
|
||||
} from "../../../packages/device-edge-channel-contract/src/index.mjs";
|
||||
import {
|
||||
normalizeAdapterAcceptance,
|
||||
normalizeAdapterMessage,
|
||||
normalizeDiscoverySignal,
|
||||
} from "../../../packages/device-protocol-contract/src/index.mjs";
|
||||
|
||||
const CHANNEL_TRACKER_SESSION_ID = "channel:control";
|
||||
const CHANNEL_PROFILE_REF = "channel.control.v1";
|
||||
|
||||
export function createDeviceEdgeChannelServer(options = {}) {
|
||||
const config = normalizeConfig(options);
|
||||
const pending = new Map();
|
||||
let active = null;
|
||||
let started = false;
|
||||
let heartbeatTimer = null;
|
||||
let totalChannelsAccepted = 0;
|
||||
let totalChannelsRejected = 0;
|
||||
let totalEventsSubmitted = 0;
|
||||
let totalEventsAccepted = 0;
|
||||
let totalEventsRejected = 0;
|
||||
let totalProtocolFailures = 0;
|
||||
|
||||
const server = createSecureServer({
|
||||
key: config.tls.key,
|
||||
cert: config.tls.cert,
|
||||
ca: config.tls.ca,
|
||||
minVersion: "TLSv1.3",
|
||||
maxVersion: "TLSv1.3",
|
||||
allowHTTP1: false,
|
||||
requestCert: true,
|
||||
rejectUnauthorized: true,
|
||||
settings: {
|
||||
enablePush: false,
|
||||
maxConcurrentStreams: 1,
|
||||
initialWindowSize: 1024 * 1024,
|
||||
},
|
||||
});
|
||||
|
||||
server.on("stream", (stream, headers) => {
|
||||
if (headers[":method"] !== "POST" || headers[":path"] !== DEVICE_EDGE_CHANNEL_PATH) {
|
||||
stream.respond({ ":status": 404, "cache-control": "no-store" });
|
||||
stream.end();
|
||||
return;
|
||||
}
|
||||
if (!isApprovedCorePeer(stream)) {
|
||||
totalChannelsRejected += 1;
|
||||
stream.respond({ ":status": 403, "cache-control": "no-store" });
|
||||
stream.end();
|
||||
return;
|
||||
}
|
||||
if (active) {
|
||||
totalChannelsRejected += 1;
|
||||
stream.respond({ ":status": 409, "cache-control": "no-store" });
|
||||
stream.end();
|
||||
return;
|
||||
}
|
||||
|
||||
const state = {
|
||||
stream,
|
||||
decoder: createChannelEnvelopeDecoder({
|
||||
direction: "core-to-edge",
|
||||
maxEnvelopeBytes: config.maxEnvelopeBytes,
|
||||
}),
|
||||
accepted: false,
|
||||
edgeSequence: 0,
|
||||
coreSequence: 0,
|
||||
lastCoreActivityAt: config.clock(),
|
||||
closed: false,
|
||||
};
|
||||
active = state;
|
||||
stream.respond({
|
||||
":status": 200,
|
||||
"content-type": "application/x-ndjson",
|
||||
"cache-control": "no-store",
|
||||
"x-content-type-options": "nosniff",
|
||||
});
|
||||
send(state, "channel.hello", {
|
||||
status: "ready",
|
||||
transport: "http2-mtls",
|
||||
commandTransport: "disabled",
|
||||
}, {
|
||||
trackerSessionId: CHANNEL_TRACKER_SESSION_ID,
|
||||
adapterProfileRef: CHANNEL_PROFILE_REF,
|
||||
correlationId: `correlation:${randomUUID()}`,
|
||||
});
|
||||
|
||||
let processing = Promise.resolve();
|
||||
stream.on("data", (chunk) => {
|
||||
stream.pause();
|
||||
processing = processing
|
||||
.then(async () => {
|
||||
const envelopes = state.decoder.push(chunk);
|
||||
for (const envelope of envelopes) {
|
||||
await handleCoreEnvelope(state, envelope);
|
||||
}
|
||||
})
|
||||
.catch(() => protocolFailure(state))
|
||||
.finally(() => {
|
||||
if (!state.closed) stream.resume();
|
||||
});
|
||||
});
|
||||
stream.on("aborted", () => closeActive(state));
|
||||
stream.on("close", () => closeActive(state));
|
||||
stream.on("error", () => closeActive(state));
|
||||
});
|
||||
server.on("tlsClientError", () => {
|
||||
totalChannelsRejected += 1;
|
||||
});
|
||||
server.on("sessionError", () => {
|
||||
totalProtocolFailures += 1;
|
||||
});
|
||||
|
||||
return Object.freeze({
|
||||
async start() {
|
||||
if (started) return server.address();
|
||||
await listen(server, config.port, config.host);
|
||||
started = true;
|
||||
heartbeatTimer = setInterval(checkChannelHealth, config.keepaliveMs);
|
||||
heartbeatTimer.unref?.();
|
||||
return server.address();
|
||||
},
|
||||
async stop() {
|
||||
clearInterval(heartbeatTimer);
|
||||
heartbeatTimer = null;
|
||||
const current = active;
|
||||
if (current) {
|
||||
current.stream.close();
|
||||
closeActive(current);
|
||||
}
|
||||
rejectAllPending("device_edge_channel_stopped");
|
||||
if (started) await closeServer(server);
|
||||
started = false;
|
||||
},
|
||||
async submitDiscovery(signal) {
|
||||
const normalized = normalizeDiscoverySignal(signal);
|
||||
const result = await submitEvent("discovery.observed", {
|
||||
signal: normalized,
|
||||
}, {
|
||||
trackerSessionId: normalized.sessionRef,
|
||||
adapterProfileRef: normalized.modelProfileRef,
|
||||
eventAt: normalized.observedAt,
|
||||
});
|
||||
if (!result?.discovery) {
|
||||
throw new Error("device_edge_channel_discovery_acceptance_invalid");
|
||||
}
|
||||
return result.discovery;
|
||||
},
|
||||
async submitAdapterMessage(message) {
|
||||
const normalized = normalizeAdapterMessage(message, {
|
||||
maxBytes: config.maxEnvelopeBytes,
|
||||
});
|
||||
const result = await submitEvent("adapter.message", {
|
||||
message: normalized,
|
||||
}, {
|
||||
trackerSessionId: normalized.sessionRef,
|
||||
adapterProfileRef: normalized.protocolProfileRef,
|
||||
eventAt: normalized.observedAt,
|
||||
});
|
||||
const acceptance = normalizeAdapterAcceptance(result?.acceptance);
|
||||
if (acceptance.idempotencyKey !== normalized.idempotencyKey) {
|
||||
throw new Error("device_edge_channel_acceptance_mismatch");
|
||||
}
|
||||
return acceptance;
|
||||
},
|
||||
status() {
|
||||
return Object.freeze({
|
||||
listening: started,
|
||||
channel: active?.accepted ? "accepted" : active ? "negotiating" : "absent",
|
||||
edgeRegistrationId: config.edgeRegistrationId,
|
||||
channelGeneration: config.channelGeneration,
|
||||
pendingAcceptances: pending.size,
|
||||
channelsAccepted: totalChannelsAccepted,
|
||||
channelsRejected: totalChannelsRejected,
|
||||
eventsSubmitted: totalEventsSubmitted,
|
||||
eventsAccepted: totalEventsAccepted,
|
||||
eventsRejected: totalEventsRejected,
|
||||
protocolFailures: totalProtocolFailures,
|
||||
trackerIngress: "disabled",
|
||||
commandTransport: "disabled",
|
||||
});
|
||||
},
|
||||
disconnectActiveChannel() {
|
||||
active?.stream.close();
|
||||
},
|
||||
});
|
||||
|
||||
async function submitEvent(messageKind, payload, metadata) {
|
||||
const state = active;
|
||||
if (!state?.accepted || state.closed) {
|
||||
throw new Error("device_edge_channel_unavailable");
|
||||
}
|
||||
if (pending.size >= config.maxPendingAcceptances) {
|
||||
throw new Error("device_edge_channel_acceptance_window_full");
|
||||
}
|
||||
const correlationId = `correlation:${randomUUID()}`;
|
||||
totalEventsSubmitted += 1;
|
||||
return new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
pending.delete(correlationId);
|
||||
totalEventsRejected += 1;
|
||||
reject(new Error("device_edge_channel_acceptance_timeout"));
|
||||
}, config.acceptanceTimeoutMs);
|
||||
timer.unref?.();
|
||||
pending.set(correlationId, { resolve, reject, timer });
|
||||
try {
|
||||
send(state, messageKind, payload, {
|
||||
...metadata,
|
||||
correlationId,
|
||||
});
|
||||
} catch (error) {
|
||||
clearTimeout(timer);
|
||||
pending.delete(correlationId);
|
||||
totalEventsRejected += 1;
|
||||
reject(error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function handleCoreEnvelope(state, envelope) {
|
||||
assertActiveState(state);
|
||||
if (
|
||||
envelope.edgeRegistrationId !== config.edgeRegistrationId
|
||||
|| envelope.channelGeneration !== config.channelGeneration
|
||||
|| envelope.sequence !== state.coreSequence + 1
|
||||
) {
|
||||
throw new Error("device_edge_channel_core_envelope_mismatch");
|
||||
}
|
||||
state.coreSequence = envelope.sequence;
|
||||
state.lastCoreActivityAt = config.clock();
|
||||
|
||||
if (!state.accepted) {
|
||||
if (envelope.messageKind !== "channel.accepted") {
|
||||
throw new Error("device_edge_channel_acceptance_required");
|
||||
}
|
||||
if (envelope.payload?.status !== "accepted") {
|
||||
throw new Error("device_edge_channel_acceptance_invalid");
|
||||
}
|
||||
state.accepted = true;
|
||||
totalChannelsAccepted += 1;
|
||||
return;
|
||||
}
|
||||
if (envelope.messageKind === "channel.heartbeat") return;
|
||||
if (!["event.accepted", "event.rejected"].includes(envelope.messageKind)) {
|
||||
throw new Error("device_edge_channel_core_message_unhandled");
|
||||
}
|
||||
const receipt = pending.get(envelope.correlationId);
|
||||
if (!receipt) {
|
||||
throw new Error("device_edge_channel_correlation_unknown");
|
||||
}
|
||||
clearTimeout(receipt.timer);
|
||||
pending.delete(envelope.correlationId);
|
||||
if (envelope.messageKind === "event.accepted") {
|
||||
totalEventsAccepted += 1;
|
||||
receipt.resolve(envelope.payload?.result);
|
||||
return;
|
||||
}
|
||||
totalEventsRejected += 1;
|
||||
receipt.reject(new Error(normalizeRejectionCode(envelope.payload?.errorCode)));
|
||||
}
|
||||
|
||||
function send(state, messageKind, payload, metadata) {
|
||||
assertActiveState(state);
|
||||
state.edgeSequence += 1;
|
||||
const now = config.now();
|
||||
const envelope = createChannelEnvelope({
|
||||
edgeRegistrationId: config.edgeRegistrationId,
|
||||
channelGeneration: config.channelGeneration,
|
||||
trackerSessionId: metadata.trackerSessionId,
|
||||
adapterProfileRef: metadata.adapterProfileRef,
|
||||
sequence: state.edgeSequence,
|
||||
eventAt: metadata.eventAt ?? now,
|
||||
receivedAt: now,
|
||||
messageKind,
|
||||
correlationId: metadata.correlationId,
|
||||
payload,
|
||||
}, {
|
||||
direction: "edge-to-core",
|
||||
maxEnvelopeBytes: config.maxEnvelopeBytes,
|
||||
});
|
||||
const encoded = encodeChannelEnvelope(envelope, {
|
||||
direction: "edge-to-core",
|
||||
maxEnvelopeBytes: config.maxEnvelopeBytes,
|
||||
});
|
||||
if (!state.stream.write(encoded)) {
|
||||
state.stream.once("drain", () => {});
|
||||
}
|
||||
}
|
||||
|
||||
function checkChannelHealth() {
|
||||
const state = active;
|
||||
if (!state || state.closed) return;
|
||||
if (config.clock() - state.lastCoreActivityAt >= config.deadPeerMs) {
|
||||
protocolFailure(state);
|
||||
return;
|
||||
}
|
||||
if (state.accepted) {
|
||||
try {
|
||||
send(state, "channel.heartbeat", { status: "alive" }, {
|
||||
trackerSessionId: CHANNEL_TRACKER_SESSION_ID,
|
||||
adapterProfileRef: CHANNEL_PROFILE_REF,
|
||||
correlationId: `correlation:${randomUUID()}`,
|
||||
});
|
||||
} catch {
|
||||
protocolFailure(state);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function isApprovedCorePeer(stream) {
|
||||
const socket = stream.session?.socket;
|
||||
if (!socket?.authorized) return false;
|
||||
let fingerprint;
|
||||
try {
|
||||
fingerprint = normalizeCertificateFingerprint(
|
||||
socket.getPeerCertificate()?.fingerprint256,
|
||||
);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
return config.tls.allowedCoreFingerprints.has(fingerprint);
|
||||
}
|
||||
|
||||
function protocolFailure(state) {
|
||||
totalProtocolFailures += 1;
|
||||
state.stream.close();
|
||||
closeActive(state);
|
||||
}
|
||||
|
||||
function closeActive(state) {
|
||||
if (state.closed) return;
|
||||
state.closed = true;
|
||||
if (active === state) active = null;
|
||||
rejectAllPending("device_edge_channel_disconnected");
|
||||
}
|
||||
|
||||
function rejectAllPending(code) {
|
||||
for (const receipt of pending.values()) {
|
||||
clearTimeout(receipt.timer);
|
||||
receipt.reject(new Error(code));
|
||||
}
|
||||
pending.clear();
|
||||
}
|
||||
|
||||
function assertActiveState(state) {
|
||||
if (!state || state.closed || active !== state) {
|
||||
throw new Error("device_edge_channel_unavailable");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeConfig(options) {
|
||||
const edgeRegistrationId = normalizeRef(
|
||||
options.edgeRegistrationId,
|
||||
"edge_registration_id",
|
||||
);
|
||||
const channelGeneration = normalizeRef(
|
||||
options.channelGeneration,
|
||||
"channel_generation",
|
||||
);
|
||||
const tls = normalizeTls(options.tls);
|
||||
const keepaliveMs = normalizeDuration(options.keepaliveMs, 10,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.keepaliveMs, "keepalive");
|
||||
const deadPeerMs = normalizeDuration(options.deadPeerMs, keepaliveMs * 2,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.deadPeerMs, "dead_peer");
|
||||
if (deadPeerMs < keepaliveMs * 2) {
|
||||
throw new TypeError("device_edge_channel_dead_peer_invalid");
|
||||
}
|
||||
return Object.freeze({
|
||||
edgeRegistrationId,
|
||||
channelGeneration,
|
||||
tls,
|
||||
host: normalizeHost(options.host ?? "127.0.0.1"),
|
||||
port: normalizePort(options.port ?? 8443),
|
||||
keepaliveMs,
|
||||
deadPeerMs,
|
||||
acceptanceTimeoutMs: normalizeDuration(
|
||||
options.acceptanceTimeoutMs,
|
||||
10,
|
||||
5_000,
|
||||
"acceptance_timeout",
|
||||
),
|
||||
maxEnvelopeBytes: normalizeInteger(
|
||||
options.maxEnvelopeBytes,
|
||||
256,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.maxEnvelopeBytes,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.maxEnvelopeBytes,
|
||||
"max_envelope_bytes",
|
||||
),
|
||||
maxPendingAcceptances: normalizeInteger(
|
||||
options.maxPendingAcceptances,
|
||||
1,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.maxPendingAcceptances,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.maxPendingAcceptances,
|
||||
"max_pending_acceptances",
|
||||
),
|
||||
clock: typeof options.clock === "function" ? options.clock : Date.now,
|
||||
now: typeof options.now === "function"
|
||||
? () => new Date(options.now()).toISOString()
|
||||
: () => new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeTls(value) {
|
||||
if (!value || typeof value !== "object") {
|
||||
throw new TypeError("device_edge_channel_tls_invalid");
|
||||
}
|
||||
for (const key of ["key", "cert", "ca"]) {
|
||||
if (!(typeof value[key] === "string" || Buffer.isBuffer(value[key]))) {
|
||||
throw new TypeError(`device_edge_channel_tls_${key}_invalid`);
|
||||
}
|
||||
}
|
||||
if (!Array.isArray(value.allowedCoreFingerprints) || value.allowedCoreFingerprints.length < 1) {
|
||||
throw new TypeError("device_edge_channel_core_identity_allowlist_invalid");
|
||||
}
|
||||
return Object.freeze({
|
||||
key: value.key,
|
||||
cert: value.cert,
|
||||
ca: value.ca,
|
||||
allowedCoreFingerprints: new Set(
|
||||
value.allowedCoreFingerprints.map(normalizeCertificateFingerprint),
|
||||
),
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeRejectionCode(value) {
|
||||
if (typeof value !== "string" || !/^[a-z][a-z0-9._:-]{2,127}$/.test(value)) {
|
||||
return "device_edge_channel_event_rejected";
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function normalizeRef(value, field) {
|
||||
if (typeof value !== "string" || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(value)) {
|
||||
throw new TypeError(`device_edge_channel_${field}_invalid`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function normalizeHost(value) {
|
||||
if (typeof value !== "string" || value.length < 1 || value.length > 253) {
|
||||
throw new TypeError("device_edge_channel_host_invalid");
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function normalizePort(value) {
|
||||
const number = Number(value);
|
||||
if (!Number.isSafeInteger(number) || number < 0 || number > 65_535) {
|
||||
throw new TypeError("device_edge_channel_port_invalid");
|
||||
}
|
||||
return number;
|
||||
}
|
||||
|
||||
function normalizeDuration(value, minimum, fallback, field) {
|
||||
return normalizeInteger(value, minimum, 120_000, fallback, field);
|
||||
}
|
||||
|
||||
function normalizeInteger(value, minimum, maximum, fallback, field) {
|
||||
const number = value == null ? fallback : Number(value);
|
||||
if (!Number.isSafeInteger(number) || number < minimum || number > maximum) {
|
||||
throw new TypeError(`device_edge_channel_${field}_invalid`);
|
||||
}
|
||||
return number;
|
||||
}
|
||||
|
||||
function listen(server, port, host) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const onError = (error) => {
|
||||
server.off("listening", onListening);
|
||||
reject(error);
|
||||
};
|
||||
const onListening = () => {
|
||||
server.off("error", onError);
|
||||
resolve();
|
||||
};
|
||||
server.once("error", onError);
|
||||
server.once("listening", onListening);
|
||||
server.listen(port, host);
|
||||
});
|
||||
}
|
||||
|
||||
function closeServer(server) {
|
||||
return new Promise((resolve) => server.close(() => resolve()));
|
||||
}
|
||||
@@ -0,0 +1,449 @@
|
||||
import assert from "node:assert/strict";
|
||||
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 { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { after, before, test } from "node:test";
|
||||
|
||||
import {
|
||||
DEVICE_ADAPTER_ACCEPTANCE_SCHEMA,
|
||||
DEVICE_ADAPTER_MESSAGE_SCHEMA,
|
||||
DEVICE_DISCOVERY_SIGNAL_SCHEMA,
|
||||
} from "../../../packages/device-protocol-contract/src/index.mjs";
|
||||
import { createDeviceGatewayCoreChannelClient } from
|
||||
"../../device-gateway-core/src/runtime.mjs";
|
||||
import { createDeviceEdgeChannelServer } from "../src/runtime.mjs";
|
||||
|
||||
let fixtureDirectory;
|
||||
let certificates;
|
||||
|
||||
before(async () => {
|
||||
fixtureDirectory = await mkdtemp(join(tmpdir(), "nodedc-edge-channel-"));
|
||||
certificates = await generateCertificateFixture(fixtureDirectory);
|
||||
});
|
||||
|
||||
after(async () => {
|
||||
await rm(fixtureDirectory, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
test("accepts synthetic discovery and durable message results over Core-initiated mTLS", async () => {
|
||||
const acceptedMessages = new Map();
|
||||
let messageCalls = 0;
|
||||
const pair = await startPair({
|
||||
acceptMessage: async (message) => {
|
||||
messageCalls += 1;
|
||||
const previous = acceptedMessages.get(message.idempotencyKey);
|
||||
if (previous) return { ...previous, replayed: true };
|
||||
const acceptance = acceptanceFor(message, false);
|
||||
acceptedMessages.set(message.idempotencyKey, acceptance);
|
||||
return acceptance;
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
const discovery = await pair.edge.submitDiscovery(discoverySignal());
|
||||
assert.equal(discovery.lifecycleState, "quarantine");
|
||||
assert.equal(discovery.identifier.masked, "***********1088");
|
||||
|
||||
const message = adapterMessage();
|
||||
const first = await pair.edge.submitAdapterMessage(message);
|
||||
const replay = await pair.edge.submitAdapterMessage(message);
|
||||
assert.equal(first.replayed, false);
|
||||
assert.equal(replay.replayed, true);
|
||||
assert.equal(messageCalls, 2);
|
||||
assert.equal(pair.edge.status().eventsAccepted, 3);
|
||||
assert.equal(pair.core.status().eventsAccepted, 3);
|
||||
assert.equal(pair.edge.status().trackerIngress, "disabled");
|
||||
assert.equal(pair.core.status().commandTransport, "disabled");
|
||||
} finally {
|
||||
await stopPair(pair);
|
||||
}
|
||||
});
|
||||
|
||||
test("keeps the channel alive and reconnects without losing idempotency", async () => {
|
||||
const acceptedMessages = new Map();
|
||||
const pair = await startPair({
|
||||
keepaliveMs: 20,
|
||||
deadPeerMs: 60,
|
||||
reconnectMinimumMs: 20,
|
||||
reconnectMaximumMs: 40,
|
||||
acceptMessage: async (message) => {
|
||||
const previous = acceptedMessages.get(message.idempotencyKey);
|
||||
if (previous) return { ...previous, replayed: true };
|
||||
const acceptance = acceptanceFor(message, false);
|
||||
acceptedMessages.set(message.idempotencyKey, acceptance);
|
||||
return acceptance;
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
await delay(120);
|
||||
assert.equal(pair.edge.status().channel, "accepted");
|
||||
const message = adapterMessage();
|
||||
assert.equal((await pair.edge.submitAdapterMessage(message)).replayed, false);
|
||||
|
||||
pair.edge.disconnectActiveChannel();
|
||||
await waitFor(() => pair.core.status().connectionAttempts >= 2
|
||||
&& pair.edge.status().channel === "accepted", 2_000);
|
||||
assert.equal((await pair.edge.submitAdapterMessage(message)).replayed, true);
|
||||
assert.ok(pair.core.status().reconnects >= 1);
|
||||
} finally {
|
||||
await stopPair(pair);
|
||||
}
|
||||
});
|
||||
|
||||
test("rejects a revoked Edge registration before opening a channel", async () => {
|
||||
const edge = createEdgeServer();
|
||||
const address = await edge.start();
|
||||
const core = createCoreClient({
|
||||
address,
|
||||
lifecycleState: "revoked",
|
||||
});
|
||||
try {
|
||||
await core.start();
|
||||
await waitFor(() => core.status().connectionAttempts >= 1, 500);
|
||||
assert.equal(core.status().channel, "absent");
|
||||
assert.equal(edge.status().channelsAccepted, 0);
|
||||
assert.match(core.status().lastErrorCode, /registration_inactive/);
|
||||
} finally {
|
||||
await core.stop();
|
||||
await edge.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("rejects an authenticated but non-allowlisted Core certificate", async () => {
|
||||
const edge = createEdgeServer();
|
||||
const address = await edge.start();
|
||||
const core = createCoreClient({
|
||||
address,
|
||||
clientCertificate: certificates.intruder,
|
||||
});
|
||||
try {
|
||||
await core.start();
|
||||
await waitFor(() => edge.status().channelsRejected >= 1, 2_000);
|
||||
assert.equal(edge.status().channelsAccepted, 0);
|
||||
assert.equal(edge.status().channel, "absent");
|
||||
} finally {
|
||||
await core.stop();
|
||||
await edge.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("rejects an unknown Edge certificate fingerprint", async () => {
|
||||
const edge = createEdgeServer();
|
||||
const address = await edge.start();
|
||||
const core = createCoreClient({
|
||||
address,
|
||||
expectedEdgeFingerprint: certificates.intruder.fingerprint,
|
||||
});
|
||||
try {
|
||||
await core.start();
|
||||
await waitFor(() => core.status().protocolFailures >= 1, 2_000);
|
||||
assert.equal(core.status().channelsAccepted, 0);
|
||||
assert.match(core.status().lastErrorCode, /edge_identity_mismatch/);
|
||||
} finally {
|
||||
await core.stop();
|
||||
await edge.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("returns a conclusive rejection when Core cannot accept a package", async () => {
|
||||
const pair = await startPair({
|
||||
acceptMessage: async () => {
|
||||
throw new Error("device_control_core_unavailable");
|
||||
},
|
||||
});
|
||||
try {
|
||||
await assert.rejects(
|
||||
pair.edge.submitAdapterMessage(adapterMessage()),
|
||||
/device_control_core_unavailable/,
|
||||
);
|
||||
assert.equal(pair.edge.status().eventsRejected, 1);
|
||||
assert.equal(pair.core.status().eventsRejected, 1);
|
||||
} finally {
|
||||
await stopPair(pair);
|
||||
}
|
||||
});
|
||||
|
||||
test("closes the logical session on an unknown message kind", async () => {
|
||||
const edge = createEdgeServer();
|
||||
const address = await edge.start();
|
||||
const session = connectHttp2(`https://127.0.0.1:${address.port}`, {
|
||||
key: certificates.core.key,
|
||||
cert: certificates.core.cert,
|
||||
ca: certificates.ca,
|
||||
servername: "localhost",
|
||||
minVersion: "TLSv1.3",
|
||||
maxVersion: "TLSv1.3",
|
||||
rejectUnauthorized: true,
|
||||
});
|
||||
try {
|
||||
await onceEvent(session, "connect");
|
||||
const request = session.request({
|
||||
":method": "POST",
|
||||
":path": "/internal/v1/device-edge/channel",
|
||||
}, { endStream: false });
|
||||
const hello = await readFirstEnvelope(request);
|
||||
const invalid = {
|
||||
schemaVersion: hello.schemaVersion,
|
||||
edgeRegistrationId: hello.edgeRegistrationId,
|
||||
channelGeneration: hello.channelGeneration,
|
||||
trackerSessionId: "channel:control",
|
||||
adapterProfileRef: "channel.control.v1",
|
||||
sequence: 1,
|
||||
eventAt: new Date().toISOString(),
|
||||
receivedAt: new Date().toISOString(),
|
||||
payloadBytes: 2,
|
||||
messageKind: "tcp.forward",
|
||||
correlationId: `correlation:${randomUUID()}`,
|
||||
payload: {},
|
||||
};
|
||||
request.write(`${JSON.stringify(invalid)}\n`);
|
||||
await waitFor(() => edge.status().protocolFailures >= 1, 1_000);
|
||||
assert.equal(edge.status().channel, "absent");
|
||||
request.close();
|
||||
} finally {
|
||||
session.close();
|
||||
await edge.stop();
|
||||
}
|
||||
});
|
||||
|
||||
async function startPair(options = {}) {
|
||||
const edge = createEdgeServer(options);
|
||||
const address = await edge.start();
|
||||
const core = createCoreClient({ address, ...options });
|
||||
await core.start();
|
||||
await core.waitForReady(2_000);
|
||||
await waitFor(() => edge.status().channel === "accepted", 2_000);
|
||||
return { edge, core };
|
||||
}
|
||||
|
||||
function createEdgeServer(options = {}) {
|
||||
return createDeviceEdgeChannelServer({
|
||||
edgeRegistrationId: "edge:pilot-1",
|
||||
channelGeneration: "generation:pilot-1",
|
||||
host: "127.0.0.1",
|
||||
port: 0,
|
||||
tls: {
|
||||
key: certificates.edge.key,
|
||||
cert: certificates.edge.cert,
|
||||
ca: certificates.ca,
|
||||
allowedCoreFingerprints: [certificates.core.fingerprint],
|
||||
},
|
||||
keepaliveMs: options.keepaliveMs ?? 50,
|
||||
deadPeerMs: options.deadPeerMs ?? 150,
|
||||
acceptanceTimeoutMs: 500,
|
||||
});
|
||||
}
|
||||
|
||||
function createCoreClient(options) {
|
||||
const clientCertificate = options.clientCertificate ?? certificates.core;
|
||||
return createDeviceGatewayCoreChannelClient({
|
||||
registration: {
|
||||
edgeRegistrationId: "edge:pilot-1",
|
||||
endpoint: `https://127.0.0.1:${options.address.port}/`,
|
||||
servername: "localhost",
|
||||
certificateFingerprint: options.expectedEdgeFingerprint
|
||||
?? certificates.edge.fingerprint,
|
||||
lifecycleState: options.lifecycleState ?? "active",
|
||||
},
|
||||
tls: {
|
||||
key: clientCertificate.key,
|
||||
cert: clientCertificate.cert,
|
||||
ca: certificates.ca,
|
||||
},
|
||||
coreIdentity: "workload:device-gateway-core",
|
||||
keepaliveMs: options.keepaliveMs ?? 50,
|
||||
deadPeerMs: options.deadPeerMs ?? 150,
|
||||
reconnectMinimumMs: options.reconnectMinimumMs ?? 20,
|
||||
reconnectMaximumMs: options.reconnectMaximumMs ?? 80,
|
||||
random: () => 0,
|
||||
observeDiscovery: options.observeDiscovery ?? (async () => ({
|
||||
schemaVersion: "nodedc.device.discovery-view.v1",
|
||||
lifecycleState: "quarantine",
|
||||
commandTransport: "disabled",
|
||||
identifier: { kind: "imei", masked: "***********1088" },
|
||||
})),
|
||||
acceptMessage: options.acceptMessage ?? (async (message) =>
|
||||
acceptanceFor(message, false)),
|
||||
});
|
||||
}
|
||||
|
||||
async function stopPair(pair) {
|
||||
await pair.core.stop();
|
||||
await pair.edge.stop();
|
||||
}
|
||||
|
||||
function discoverySignal() {
|
||||
return {
|
||||
schemaVersion: DEVICE_DISCOVERY_SIGNAL_SCHEMA,
|
||||
sessionRef: "session:pilot-1",
|
||||
modelProfileRef: "arusnavi.internal.b2.v1",
|
||||
protocol: "INTERNAL",
|
||||
observedAt: new Date().toISOString(),
|
||||
identifier: { kind: "imei", value: "863151070211088" },
|
||||
evidence: {
|
||||
transport: "tcp",
|
||||
bytesObserved: 10,
|
||||
framingStatus: "verified",
|
||||
specificationRef: "arusnavi.internal.protocol.v1",
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function adapterMessage() {
|
||||
return {
|
||||
schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA,
|
||||
edgeRef: "edge:pilot-1",
|
||||
adapterRef: "arusnavi-b2",
|
||||
protocolProfileRef: "arusnavi.internal.b2.v1",
|
||||
protocol: "INTERNAL",
|
||||
sessionRef: "session:pilot-1",
|
||||
messageRef: "message:pilot-1",
|
||||
messageType: "telemetry.package",
|
||||
sequence: 1,
|
||||
observedAt: new Date().toISOString(),
|
||||
idempotencyKey: `sha256:${"a".repeat(64)}`,
|
||||
identifier: { kind: "imei", value: "863151070211088" },
|
||||
payloadSchemaRef: "arusnavi.internal.package-metadata.v1",
|
||||
payload: {
|
||||
packageNumber: 1,
|
||||
packetCount: 1,
|
||||
byteLength: 11,
|
||||
packageDigest: `sha256:${"b".repeat(64)}`,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function acceptanceFor(message, replayed) {
|
||||
return {
|
||||
schemaVersion: DEVICE_ADAPTER_ACCEPTANCE_SCHEMA,
|
||||
acceptanceRef: "acceptance:pilot-1",
|
||||
idempotencyKey: message.idempotencyKey,
|
||||
status: "accepted",
|
||||
replayed,
|
||||
acceptedAt: "2026-08-11T12:00:00.000Z",
|
||||
};
|
||||
}
|
||||
|
||||
async function generateCertificateFixture(directory) {
|
||||
runOpenSsl(directory, [
|
||||
"req", "-x509", "-newkey", "rsa:2048", "-nodes", "-sha256",
|
||||
"-days", "1", "-subj", "/CN=NODEDC Test Device Edge CA",
|
||||
"-addext", "basicConstraints=critical,CA:TRUE",
|
||||
"-addext", "keyUsage=critical,keyCertSign,cRLSign",
|
||||
"-keyout", "ca.key", "-out", "ca.crt",
|
||||
]);
|
||||
await issueCertificate(directory, "edge", "localhost", [
|
||||
"subjectAltName=DNS:localhost,IP:127.0.0.1",
|
||||
"extendedKeyUsage=serverAuth",
|
||||
]);
|
||||
await issueCertificate(directory, "core", "nodedc-device-gateway-core", [
|
||||
"extendedKeyUsage=clientAuth",
|
||||
]);
|
||||
await issueCertificate(directory, "intruder", "unapproved-core", [
|
||||
"extendedKeyUsage=clientAuth",
|
||||
]);
|
||||
const ca = await readFile(join(directory, "ca.crt"));
|
||||
return {
|
||||
ca,
|
||||
edge: await readCertificate(directory, "edge"),
|
||||
core: await readCertificate(directory, "core"),
|
||||
intruder: await readCertificate(directory, "intruder"),
|
||||
};
|
||||
}
|
||||
|
||||
async function issueCertificate(directory, name, commonName, extensions) {
|
||||
runOpenSsl(directory, [
|
||||
"req", "-newkey", "rsa:2048", "-nodes", "-sha256",
|
||||
"-subj", `/CN=${commonName}`,
|
||||
"-keyout", `${name}.key`, "-out", `${name}.csr`,
|
||||
]);
|
||||
const extensionFile = `${name}.ext`;
|
||||
await writeFile(join(directory, extensionFile), [
|
||||
"basicConstraints=critical,CA:FALSE",
|
||||
"keyUsage=critical,digitalSignature,keyEncipherment",
|
||||
...extensions,
|
||||
].join("\n"));
|
||||
runOpenSsl(directory, [
|
||||
"x509", "-req", "-sha256", "-days", "1",
|
||||
"-in", `${name}.csr`, "-CA", "ca.crt", "-CAkey", "ca.key",
|
||||
"-CAcreateserial", "-extfile", extensionFile, "-out", `${name}.crt`,
|
||||
]);
|
||||
}
|
||||
|
||||
async function readCertificate(directory, name) {
|
||||
const cert = await readFile(join(directory, `${name}.crt`));
|
||||
return {
|
||||
key: await readFile(join(directory, `${name}.key`)),
|
||||
cert,
|
||||
fingerprint: new X509Certificate(cert).fingerprint256,
|
||||
};
|
||||
}
|
||||
|
||||
function runOpenSsl(directory, arguments_) {
|
||||
const result = spawnSync("openssl", arguments_, {
|
||||
cwd: directory,
|
||||
encoding: "utf8",
|
||||
});
|
||||
if (result.status !== 0) {
|
||||
throw new Error(`openssl_failed:${result.stderr}`);
|
||||
}
|
||||
}
|
||||
|
||||
function readFirstEnvelope(stream) {
|
||||
return new Promise((resolve, reject) => {
|
||||
let buffered = "";
|
||||
const onData = (chunk) => {
|
||||
buffered += chunk.toString("utf8");
|
||||
const newline = buffered.indexOf("\n");
|
||||
if (newline < 0) return;
|
||||
cleanup();
|
||||
resolve(JSON.parse(buffered.slice(0, newline)));
|
||||
};
|
||||
const onError = (error) => {
|
||||
cleanup();
|
||||
reject(error);
|
||||
};
|
||||
const cleanup = () => {
|
||||
stream.off("data", onData);
|
||||
stream.off("error", onError);
|
||||
};
|
||||
stream.on("data", onData);
|
||||
stream.once("error", onError);
|
||||
});
|
||||
}
|
||||
|
||||
function onceEvent(emitter, event) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const onEvent = (...args) => {
|
||||
cleanup();
|
||||
resolve(args);
|
||||
};
|
||||
const onError = (error) => {
|
||||
cleanup();
|
||||
reject(error);
|
||||
};
|
||||
const cleanup = () => {
|
||||
emitter.off(event, onEvent);
|
||||
emitter.off("error", onError);
|
||||
};
|
||||
emitter.once(event, onEvent);
|
||||
emitter.once("error", onError);
|
||||
});
|
||||
}
|
||||
|
||||
async function waitFor(predicate, timeoutMs) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (predicate()) return;
|
||||
await delay(10);
|
||||
}
|
||||
assert.fail("condition_timeout");
|
||||
}
|
||||
|
||||
function delay(milliseconds) {
|
||||
return new Promise((resolve) => setTimeout(resolve, milliseconds));
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"name": "@nodedc/device-gateway-core",
|
||||
"version": "0.1.0",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"exports": {
|
||||
".": "./src/runtime.mjs"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=20"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,565 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { connect as connectHttp2 } from "node:http2";
|
||||
|
||||
import {
|
||||
DEVICE_EDGE_CHANNEL_LIMITS,
|
||||
DEVICE_EDGE_CHANNEL_PATH,
|
||||
createChannelEnvelope,
|
||||
createChannelEnvelopeDecoder,
|
||||
encodeChannelEnvelope,
|
||||
nextReconnectDelay,
|
||||
normalizeCertificateFingerprint,
|
||||
} from "../../../packages/device-edge-channel-contract/src/index.mjs";
|
||||
import {
|
||||
normalizeAdapterMessage,
|
||||
normalizeDiscoverySignal,
|
||||
} from "../../../packages/device-protocol-contract/src/index.mjs";
|
||||
|
||||
const CHANNEL_TRACKER_SESSION_ID = "channel:control";
|
||||
const CHANNEL_PROFILE_REF = "channel.control.v1";
|
||||
|
||||
export function createDeviceGatewayCoreChannelClient(options = {}) {
|
||||
const config = normalizeConfig(options);
|
||||
const readyWaiters = new Set();
|
||||
let running = false;
|
||||
let state = null;
|
||||
let reconnectTimer = null;
|
||||
let reconnectAttempt = 0;
|
||||
let connectionSerial = 0;
|
||||
let totalConnectionAttempts = 0;
|
||||
let totalChannelsAccepted = 0;
|
||||
let totalReconnects = 0;
|
||||
let totalEventsAccepted = 0;
|
||||
let totalEventsRejected = 0;
|
||||
let totalProtocolFailures = 0;
|
||||
let lastErrorCode = null;
|
||||
|
||||
return Object.freeze({
|
||||
async start() {
|
||||
if (running) return;
|
||||
running = true;
|
||||
void connectNow();
|
||||
},
|
||||
async stop() {
|
||||
running = false;
|
||||
clearTimeout(reconnectTimer);
|
||||
reconnectTimer = null;
|
||||
const current = state;
|
||||
state = null;
|
||||
if (current) closeConnection(current, false);
|
||||
rejectReadyWaiters("device_gateway_core_channel_stopped");
|
||||
},
|
||||
waitForReady(timeoutMs = 5_000) {
|
||||
if (state?.ready && !state.closed) return Promise.resolve(status());
|
||||
const normalizedTimeout = normalizeInteger(
|
||||
timeoutMs,
|
||||
10,
|
||||
120_000,
|
||||
5_000,
|
||||
"ready_timeout",
|
||||
);
|
||||
return new Promise((resolve, reject) => {
|
||||
const waiter = { resolve, reject, timer: null };
|
||||
waiter.timer = setTimeout(() => {
|
||||
readyWaiters.delete(waiter);
|
||||
reject(new Error("device_gateway_core_channel_ready_timeout"));
|
||||
}, normalizedTimeout);
|
||||
waiter.timer.unref?.();
|
||||
readyWaiters.add(waiter);
|
||||
});
|
||||
},
|
||||
status,
|
||||
disconnect() {
|
||||
if (state) closeConnection(state, true);
|
||||
},
|
||||
});
|
||||
|
||||
function status() {
|
||||
return Object.freeze({
|
||||
running,
|
||||
channel: state?.ready ? "accepted" : state ? "connecting" : "absent",
|
||||
edgeRegistrationId: state?.registration?.edgeRegistrationId ?? null,
|
||||
channelGeneration: state?.channelGeneration ?? null,
|
||||
connectionAttempts: totalConnectionAttempts,
|
||||
channelsAccepted: totalChannelsAccepted,
|
||||
reconnects: totalReconnects,
|
||||
eventsAccepted: totalEventsAccepted,
|
||||
eventsRejected: totalEventsRejected,
|
||||
protocolFailures: totalProtocolFailures,
|
||||
lastErrorCode,
|
||||
trackerIngress: "remote-edge-only",
|
||||
commandTransport: "disabled",
|
||||
});
|
||||
}
|
||||
|
||||
async function connectNow() {
|
||||
if (!running || state) return;
|
||||
totalConnectionAttempts += 1;
|
||||
const serial = ++connectionSerial;
|
||||
let registration;
|
||||
try {
|
||||
registration = normalizeRegistration(await config.registrationProvider());
|
||||
if (registration.lifecycleState !== "active") {
|
||||
throw new Error("device_gateway_core_edge_registration_inactive");
|
||||
}
|
||||
} catch (error) {
|
||||
lastErrorCode = safeErrorCode(error);
|
||||
scheduleReconnect();
|
||||
return;
|
||||
}
|
||||
|
||||
const connection = {
|
||||
serial,
|
||||
registration,
|
||||
session: null,
|
||||
request: null,
|
||||
decoder: createChannelEnvelopeDecoder({
|
||||
direction: "edge-to-core",
|
||||
maxEnvelopeBytes: config.maxEnvelopeBytes,
|
||||
}),
|
||||
channelGeneration: null,
|
||||
edgeSequence: 0,
|
||||
coreSequence: 0,
|
||||
lastEdgeActivityAt: config.clock(),
|
||||
heartbeatTimer: null,
|
||||
ready: false,
|
||||
closed: false,
|
||||
};
|
||||
state = connection;
|
||||
const endpoint = new URL(registration.endpoint);
|
||||
const authority = `${endpoint.protocol}//${endpoint.host}`;
|
||||
const session = connectHttp2(authority, {
|
||||
key: config.tls.key,
|
||||
cert: config.tls.cert,
|
||||
ca: config.tls.ca,
|
||||
minVersion: "TLSv1.3",
|
||||
maxVersion: "TLSv1.3",
|
||||
rejectUnauthorized: true,
|
||||
servername: registration.servername,
|
||||
ALPNProtocols: ["h2"],
|
||||
settings: {
|
||||
enablePush: false,
|
||||
initialWindowSize: 1024 * 1024,
|
||||
},
|
||||
});
|
||||
connection.session = session;
|
||||
session.once("error", (error) => failConnection(connection, error));
|
||||
session.once("close", () => closeConnection(connection, true));
|
||||
session.once("connect", () => {
|
||||
try {
|
||||
verifyEdgePeer(connection);
|
||||
openChannelStream(connection);
|
||||
} catch (error) {
|
||||
failConnection(connection, error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function openChannelStream(connection) {
|
||||
assertCurrent(connection);
|
||||
const request = connection.session.request({
|
||||
":method": "POST",
|
||||
":path": DEVICE_EDGE_CHANNEL_PATH,
|
||||
"content-type": "application/x-ndjson",
|
||||
"cache-control": "no-store",
|
||||
}, { endStream: false });
|
||||
connection.request = request;
|
||||
request.once("response", (headers) => {
|
||||
if (Number(headers[":status"]) !== 200) {
|
||||
failConnection(connection, new Error(
|
||||
`device_gateway_core_channel_http_status_${headers[":status"]}`,
|
||||
));
|
||||
}
|
||||
});
|
||||
let processing = Promise.resolve();
|
||||
request.on("data", (chunk) => {
|
||||
request.pause();
|
||||
processing = processing
|
||||
.then(async () => {
|
||||
const envelopes = connection.decoder.push(chunk);
|
||||
for (const envelope of envelopes) {
|
||||
await handleEdgeEnvelope(connection, envelope);
|
||||
}
|
||||
})
|
||||
.catch((error) => failConnection(connection, error))
|
||||
.finally(() => {
|
||||
if (!connection.closed) request.resume();
|
||||
});
|
||||
});
|
||||
request.once("aborted", () => closeConnection(connection, true));
|
||||
request.once("close", () => closeConnection(connection, true));
|
||||
request.once("error", (error) => failConnection(connection, error));
|
||||
connection.heartbeatTimer = setInterval(
|
||||
() => checkChannelHealth(connection),
|
||||
config.keepaliveMs,
|
||||
);
|
||||
connection.heartbeatTimer.unref?.();
|
||||
}
|
||||
|
||||
async function handleEdgeEnvelope(connection, envelope) {
|
||||
assertCurrent(connection);
|
||||
if (
|
||||
envelope.edgeRegistrationId !== connection.registration.edgeRegistrationId
|
||||
|| envelope.sequence !== connection.edgeSequence + 1
|
||||
) {
|
||||
throw new Error("device_gateway_core_edge_envelope_mismatch");
|
||||
}
|
||||
if (
|
||||
connection.channelGeneration
|
||||
&& envelope.channelGeneration !== connection.channelGeneration
|
||||
) {
|
||||
throw new Error("device_gateway_core_channel_generation_mismatch");
|
||||
}
|
||||
connection.edgeSequence = envelope.sequence;
|
||||
connection.lastEdgeActivityAt = config.clock();
|
||||
|
||||
if (!connection.ready) {
|
||||
if (envelope.messageKind !== "channel.hello") {
|
||||
throw new Error("device_gateway_core_channel_hello_required");
|
||||
}
|
||||
if (
|
||||
envelope.payload?.status !== "ready"
|
||||
|| envelope.payload?.transport !== "http2-mtls"
|
||||
|| envelope.payload?.commandTransport !== "disabled"
|
||||
) {
|
||||
throw new Error("device_gateway_core_channel_hello_invalid");
|
||||
}
|
||||
connection.channelGeneration = envelope.channelGeneration;
|
||||
send(connection, "channel.accepted", {
|
||||
status: "accepted",
|
||||
coreIdentity: config.coreIdentity,
|
||||
commandTransport: "disabled",
|
||||
}, {
|
||||
trackerSessionId: CHANNEL_TRACKER_SESSION_ID,
|
||||
adapterProfileRef: CHANNEL_PROFILE_REF,
|
||||
correlationId: envelope.correlationId,
|
||||
});
|
||||
connection.ready = true;
|
||||
reconnectAttempt = 0;
|
||||
totalChannelsAccepted += 1;
|
||||
lastErrorCode = null;
|
||||
resolveReadyWaiters();
|
||||
return;
|
||||
}
|
||||
if (envelope.messageKind === "channel.heartbeat") return;
|
||||
if (envelope.messageKind === "discovery.observed") {
|
||||
await acceptDiscovery(connection, envelope);
|
||||
return;
|
||||
}
|
||||
if (envelope.messageKind === "adapter.message") {
|
||||
await acceptAdapterMessage(connection, envelope);
|
||||
return;
|
||||
}
|
||||
throw new Error("device_gateway_core_edge_message_unhandled");
|
||||
}
|
||||
|
||||
async function acceptDiscovery(connection, envelope) {
|
||||
try {
|
||||
const signal = normalizeDiscoverySignal(envelope.payload?.signal);
|
||||
const discovery = await config.observeDiscovery(signal);
|
||||
sendEventResult(connection, envelope, { discovery });
|
||||
totalEventsAccepted += 1;
|
||||
} catch (error) {
|
||||
sendEventRejection(connection, envelope, error);
|
||||
totalEventsRejected += 1;
|
||||
}
|
||||
}
|
||||
|
||||
async function acceptAdapterMessage(connection, envelope) {
|
||||
try {
|
||||
const message = normalizeAdapterMessage(envelope.payload?.message, {
|
||||
maxBytes: config.maxEnvelopeBytes,
|
||||
});
|
||||
const acceptance = await config.acceptMessage(message);
|
||||
sendEventResult(connection, envelope, { acceptance });
|
||||
totalEventsAccepted += 1;
|
||||
} catch (error) {
|
||||
sendEventRejection(connection, envelope, error);
|
||||
totalEventsRejected += 1;
|
||||
}
|
||||
}
|
||||
|
||||
function sendEventResult(connection, envelope, result) {
|
||||
send(connection, "event.accepted", { result }, {
|
||||
trackerSessionId: envelope.trackerSessionId,
|
||||
adapterProfileRef: envelope.adapterProfileRef,
|
||||
correlationId: envelope.correlationId,
|
||||
});
|
||||
}
|
||||
|
||||
function sendEventRejection(connection, envelope, error) {
|
||||
send(connection, "event.rejected", {
|
||||
errorCode: safeErrorCode(error),
|
||||
}, {
|
||||
trackerSessionId: envelope.trackerSessionId,
|
||||
adapterProfileRef: envelope.adapterProfileRef,
|
||||
correlationId: envelope.correlationId,
|
||||
});
|
||||
}
|
||||
|
||||
function send(connection, messageKind, payload, metadata) {
|
||||
assertCurrent(connection);
|
||||
if (!connection.channelGeneration) {
|
||||
throw new Error("device_gateway_core_channel_generation_absent");
|
||||
}
|
||||
connection.coreSequence += 1;
|
||||
const now = config.now();
|
||||
const envelope = createChannelEnvelope({
|
||||
edgeRegistrationId: connection.registration.edgeRegistrationId,
|
||||
channelGeneration: connection.channelGeneration,
|
||||
trackerSessionId: metadata.trackerSessionId,
|
||||
adapterProfileRef: metadata.adapterProfileRef,
|
||||
sequence: connection.coreSequence,
|
||||
eventAt: metadata.eventAt ?? now,
|
||||
receivedAt: now,
|
||||
messageKind,
|
||||
correlationId: metadata.correlationId,
|
||||
payload,
|
||||
}, {
|
||||
direction: "core-to-edge",
|
||||
maxEnvelopeBytes: config.maxEnvelopeBytes,
|
||||
});
|
||||
connection.request.write(encodeChannelEnvelope(envelope, {
|
||||
direction: "core-to-edge",
|
||||
maxEnvelopeBytes: config.maxEnvelopeBytes,
|
||||
}));
|
||||
}
|
||||
|
||||
function checkChannelHealth(connection) {
|
||||
if (connection.closed || state !== connection) return;
|
||||
if (config.clock() - connection.lastEdgeActivityAt >= config.deadPeerMs) {
|
||||
failConnection(connection, new Error("device_gateway_core_edge_dead_peer"));
|
||||
return;
|
||||
}
|
||||
if (connection.ready) {
|
||||
try {
|
||||
send(connection, "channel.heartbeat", { status: "alive" }, {
|
||||
trackerSessionId: CHANNEL_TRACKER_SESSION_ID,
|
||||
adapterProfileRef: CHANNEL_PROFILE_REF,
|
||||
correlationId: `correlation:${randomUUID()}`,
|
||||
});
|
||||
} catch (error) {
|
||||
failConnection(connection, error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function verifyEdgePeer(connection) {
|
||||
const socket = connection.session.socket;
|
||||
if (!socket?.authorized || socket.alpnProtocol !== "h2") {
|
||||
throw new Error("device_gateway_core_edge_tls_unauthorized");
|
||||
}
|
||||
const observed = normalizeCertificateFingerprint(
|
||||
socket.getPeerCertificate()?.fingerprint256,
|
||||
);
|
||||
if (observed !== connection.registration.certificateFingerprint) {
|
||||
throw new Error("device_gateway_core_edge_identity_mismatch");
|
||||
}
|
||||
}
|
||||
|
||||
function failConnection(connection, error) {
|
||||
if (connection.closed) return;
|
||||
totalProtocolFailures += 1;
|
||||
lastErrorCode = safeErrorCode(error);
|
||||
closeConnection(connection, true);
|
||||
}
|
||||
|
||||
function closeConnection(connection, reconnect) {
|
||||
if (connection.closed) return;
|
||||
connection.closed = true;
|
||||
clearInterval(connection.heartbeatTimer);
|
||||
connection.heartbeatTimer = null;
|
||||
try {
|
||||
connection.request?.close();
|
||||
} catch {}
|
||||
try {
|
||||
connection.session?.close();
|
||||
} catch {}
|
||||
if (state === connection) state = null;
|
||||
if (reconnect && running) scheduleReconnect();
|
||||
}
|
||||
|
||||
function scheduleReconnect() {
|
||||
if (!running || reconnectTimer || state) return;
|
||||
const delay = nextReconnectDelay(reconnectAttempt, {
|
||||
minimumMs: config.reconnectMinimumMs,
|
||||
maximumMs: config.reconnectMaximumMs,
|
||||
random: config.random,
|
||||
});
|
||||
reconnectAttempt += 1;
|
||||
totalReconnects += 1;
|
||||
reconnectTimer = setTimeout(() => {
|
||||
reconnectTimer = null;
|
||||
void connectNow();
|
||||
}, delay);
|
||||
reconnectTimer.unref?.();
|
||||
}
|
||||
|
||||
function assertCurrent(connection) {
|
||||
if (!running || connection.closed || state !== connection) {
|
||||
throw new Error("device_gateway_core_channel_unavailable");
|
||||
}
|
||||
}
|
||||
|
||||
function resolveReadyWaiters() {
|
||||
const value = status();
|
||||
for (const waiter of readyWaiters) {
|
||||
clearTimeout(waiter.timer);
|
||||
waiter.resolve(value);
|
||||
}
|
||||
readyWaiters.clear();
|
||||
}
|
||||
|
||||
function rejectReadyWaiters(code) {
|
||||
for (const waiter of readyWaiters) {
|
||||
clearTimeout(waiter.timer);
|
||||
waiter.reject(new Error(code));
|
||||
}
|
||||
readyWaiters.clear();
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeConfig(options) {
|
||||
if (typeof options.observeDiscovery !== "function") {
|
||||
throw new TypeError("device_gateway_core_observe_discovery_invalid");
|
||||
}
|
||||
if (typeof options.acceptMessage !== "function") {
|
||||
throw new TypeError("device_gateway_core_accept_message_invalid");
|
||||
}
|
||||
const registrationProvider = typeof options.registrationProvider === "function"
|
||||
? options.registrationProvider
|
||||
: async () => options.registration;
|
||||
const tls = normalizeTls(options.tls);
|
||||
const keepaliveMs = normalizeInteger(
|
||||
options.keepaliveMs,
|
||||
10,
|
||||
120_000,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.keepaliveMs,
|
||||
"keepalive",
|
||||
);
|
||||
const deadPeerMs = normalizeInteger(
|
||||
options.deadPeerMs,
|
||||
keepaliveMs * 2,
|
||||
120_000,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.deadPeerMs,
|
||||
"dead_peer",
|
||||
);
|
||||
const reconnectMinimumMs = normalizeInteger(
|
||||
options.reconnectMinimumMs,
|
||||
10,
|
||||
120_000,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.reconnectMinimumMs,
|
||||
"reconnect_minimum",
|
||||
);
|
||||
const reconnectMaximumMs = normalizeInteger(
|
||||
options.reconnectMaximumMs,
|
||||
10,
|
||||
120_000,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.reconnectMaximumMs,
|
||||
"reconnect_maximum",
|
||||
);
|
||||
if (reconnectMaximumMs < reconnectMinimumMs) {
|
||||
throw new TypeError("device_gateway_core_reconnect_range_invalid");
|
||||
}
|
||||
return Object.freeze({
|
||||
registrationProvider,
|
||||
tls,
|
||||
coreIdentity: normalizeRef(options.coreIdentity, "core_identity"),
|
||||
observeDiscovery: options.observeDiscovery,
|
||||
acceptMessage: options.acceptMessage,
|
||||
keepaliveMs,
|
||||
deadPeerMs,
|
||||
reconnectMinimumMs,
|
||||
reconnectMaximumMs,
|
||||
maxEnvelopeBytes: normalizeInteger(
|
||||
options.maxEnvelopeBytes,
|
||||
256,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.maxEnvelopeBytes,
|
||||
DEVICE_EDGE_CHANNEL_LIMITS.maxEnvelopeBytes,
|
||||
"max_envelope_bytes",
|
||||
),
|
||||
random: typeof options.random === "function" ? options.random : Math.random,
|
||||
clock: typeof options.clock === "function" ? options.clock : Date.now,
|
||||
now: typeof options.now === "function"
|
||||
? () => new Date(options.now()).toISOString()
|
||||
: () => new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeRegistration(value) {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) {
|
||||
throw new TypeError("device_gateway_core_edge_registration_invalid");
|
||||
}
|
||||
let endpoint;
|
||||
try {
|
||||
endpoint = new URL(String(value.endpoint || ""));
|
||||
} catch {
|
||||
throw new TypeError("device_gateway_core_edge_endpoint_invalid");
|
||||
}
|
||||
if (
|
||||
endpoint.protocol !== "https:"
|
||||
|| endpoint.username
|
||||
|| endpoint.password
|
||||
|| endpoint.pathname !== "/"
|
||||
|| endpoint.search
|
||||
|| endpoint.hash
|
||||
) {
|
||||
throw new TypeError("device_gateway_core_edge_endpoint_invalid");
|
||||
}
|
||||
if (!["active", "revoked", "disabled"].includes(value.lifecycleState)) {
|
||||
throw new TypeError("device_gateway_core_edge_lifecycle_invalid");
|
||||
}
|
||||
const servername = String(value.servername || "");
|
||||
if (!/^[A-Za-z0-9.-]{1,253}$/.test(servername)) {
|
||||
throw new TypeError("device_gateway_core_edge_servername_invalid");
|
||||
}
|
||||
return Object.freeze({
|
||||
edgeRegistrationId: normalizeRef(
|
||||
value.edgeRegistrationId,
|
||||
"edge_registration_id",
|
||||
),
|
||||
endpoint: endpoint.toString(),
|
||||
servername,
|
||||
certificateFingerprint: normalizeCertificateFingerprint(
|
||||
value.certificateFingerprint,
|
||||
),
|
||||
lifecycleState: value.lifecycleState,
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeTls(value) {
|
||||
if (!value || typeof value !== "object") {
|
||||
throw new TypeError("device_gateway_core_channel_tls_invalid");
|
||||
}
|
||||
for (const key of ["key", "cert", "ca"]) {
|
||||
if (!(typeof value[key] === "string" || Buffer.isBuffer(value[key]))) {
|
||||
throw new TypeError(`device_gateway_core_channel_tls_${key}_invalid`);
|
||||
}
|
||||
}
|
||||
return Object.freeze({ key: value.key, cert: value.cert, ca: value.ca });
|
||||
}
|
||||
|
||||
function safeErrorCode(error) {
|
||||
const value = String(error?.message || error || "device_gateway_core_error")
|
||||
.toLowerCase()
|
||||
.replaceAll(/[^a-z0-9._:-]/g, "_")
|
||||
.slice(0, 128);
|
||||
return /^[a-z][a-z0-9._:-]{2,127}$/.test(value)
|
||||
? value
|
||||
: "device_gateway_core_event_rejected";
|
||||
}
|
||||
|
||||
function normalizeRef(value, field) {
|
||||
if (typeof value !== "string" || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(value)) {
|
||||
throw new TypeError(`device_gateway_core_${field}_invalid`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function normalizeInteger(value, minimum, maximum, fallback, field) {
|
||||
const number = value == null ? fallback : Number(value);
|
||||
if (!Number.isSafeInteger(number) || number < minimum || number > maximum) {
|
||||
throw new TypeError(`device_gateway_core_${field}_invalid`);
|
||||
}
|
||||
return number;
|
||||
}
|
||||
Reference in New Issue
Block a user