SILVER/NODEDC_PLATFORM
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(device-plane): add universal adapter acceptance boundary

Browse Source
This commit is contained in:
Codex
2026-08-11 19:11:48 +03:00
parent 227c7c26c1
commit 393741f1bd
26 changed files with 1887 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
device-plane/services/device-gateway/src/runtime.mjs
+214 -38
View File
@@ -1,16 +1,15 @@
import { randomUUID } from "node:crypto";
import { createHash, randomUUID } from "node:crypto";
import { createServer as createHttpServer } from "node:http";
import { createServer as createTcpServer } from "node:net";
import {
ARUSNAVI_B2_MODEL_PROFILE,
buildB2HeaderAcknowledgement,
buildB2PackageAcknowledgement,
tryParseB2Header2,
tryParseB2Package,
} from "../../../packages/arusnavi-b2-adapter/src/index.mjs";
assertDeviceAdapterSession,
} from "../../../packages/device-adapter-runtime/src/index.mjs";
import {
DEVICE_ADAPTER_MESSAGE_SCHEMA,
DEVICE_DISCOVERY_SIGNAL_SCHEMA,
normalizeAdapterAcceptance,
normalizeAdapterMessage,
} from "../../../packages/device-protocol-contract/src/index.mjs";
export function createDeviceGatewayRuntime(options = {}) {
@@ -21,7 +20,9 @@ export function createDeviceGatewayRuntime(options = {}) {
let totalAccepted = 0;
let totalRejected = 0;
let totalDiscoveries = 0;
let totalMessagesAccepted = 0;
let totalPackagesAcknowledged = 0;
let totalBufferedBytes = 0;
const tcpServer = createTcpServer((socket) => {
const remoteAddress = normalizeRemoteAddress(socket.remoteAddress);
@@ -40,6 +41,13 @@ export function createDeviceGatewayRuntime(options = {}) {
sessionRef,
remoteAddress,
buffer: Buffer.alloc(0),
adapterSession: assertDeviceAdapterSession(
config.adapter.createSession({
profileRef: config.profile.profileRef,
}),
),
identifier: null,
sequence: 0,
state: "awaiting-header",
processing: false,
rejected: false,
@@ -54,11 +62,11 @@ export function createDeviceGatewayRuntime(options = {}) {
socket.on("data", (chunk) => {
if (session.closed || session.rejected) return;
socket.pause();
session.buffer = Buffer.concat(
[session.buffer, chunk],
session.buffer.length + chunk.length,
);
if (session.buffer.length > config.maxBufferedBytes) {
appendBuffer(session, chunk);
if (
session.buffer.length > config.maxBufferedBytes
|| totalBufferedBytes > config.maxAggregateBufferedBytes
) {
rejectSession(socket, session);
return;
}
@@ -88,11 +96,12 @@ export function createDeviceGatewayRuntime(options = {}) {
return response.end(`${JSON.stringify({
ok: true,
service: "nodedc-device-gateway",
protocolProfile: ARUSNAVI_B2_MODEL_PROFILE.profileRef,
framing: ARUSNAVI_B2_MODEL_PROFILE.framing.status,
tcpListener: config.listenEnabled ? "discovery-only" : "disabled",
adapter: config.adapter?.adapterRef ?? "disabled",
protocolProfile: config.profile?.profileRef ?? "disabled",
framing: config.profile?.framing?.status ?? "disabled",
tcpListener: config.listenEnabled ? "telemetry-ingest" : "disabled",
publicIngress: config.publicIngressEnabled
? "discovery-only"
? "telemetry-ingest"
: "disabled",
commandTransport: "disabled",
sessions: {
@@ -100,7 +109,9 @@ export function createDeviceGatewayRuntime(options = {}) {
accepted: totalAccepted,
rejected: totalRejected,
discoveries: totalDiscoveries,
messagesAccepted: totalMessagesAccepted,
packagesAcknowledged: totalPackagesAcknowledged,
bufferedBytes: totalBufferedBytes,
},
})}\n`);
});
@@ -118,7 +129,7 @@ export function createDeviceGatewayRuntime(options = {}) {
},
async stop() {
for (const [socket, session] of sessions) {
session.closed = true;
closeSession(socket, session);
socket.destroy();
}
await Promise.all([
@@ -132,10 +143,12 @@ export function createDeviceGatewayRuntime(options = {}) {
totalAccepted,
totalRejected,
totalDiscoveries,
totalMessagesAccepted,
totalPackagesAcknowledged,
totalBufferedBytes,
commandTransport: "disabled",
publicIngress: config.publicIngressEnabled
? "discovery-only"
? "telemetry-ingest"
: "disabled",
};
},
@@ -144,36 +157,74 @@ export function createDeviceGatewayRuntime(options = {}) {
async function processSession(socket, session) {
while (!session.closed && !session.rejected) {
if (session.state === "awaiting-header") {
const parsed = tryParseB2Header2(session.buffer);
const parsed = session.adapterSession.parseHeader(session.buffer);
if (parsed.status === "incomplete") return;
const observedAt = config.now().toISOString();
await config.onDiscovery?.({
const discovery = {
schemaVersion: DEVICE_DISCOVERY_SIGNAL_SCHEMA,
sessionRef: session.sessionRef,
modelProfileRef: ARUSNAVI_B2_MODEL_PROFILE.profileRef,
protocol: ARUSNAVI_B2_MODEL_PROFILE.protocol,
...(config.routeRef ? { routeRef: config.routeRef } : {}),
modelProfileRef: config.profile.profileRef,
protocol: config.profile.protocol,
observedAt,
identifier: {
kind: parsed.identifier.kind,
value: parsed.identifier.value,
},
evidence: parsed.evidence,
});
session.buffer = session.buffer.subarray(parsed.bytesConsumed);
};
const acceptedDiscovery = await config.onDiscovery(discovery);
assertDiscoveryAccepted(acceptedDiscovery);
session.identifier = discovery.identifier;
consumeBuffer(session, parsed.bytesConsumed);
session.state = "packages";
totalDiscoveries += 1;
socket.write(buildB2HeaderAcknowledgement(
await writeWithBackpressure(socket, session.adapterSession.buildHeaderAcknowledgement(
Math.floor(new Date(observedAt).getTime() / 1000),
));
continue;
}
const parsed = tryParseB2Package(session.buffer);
const parsed = session.adapterSession.parseMessage(session.buffer);
if (parsed.status === "incomplete") return;
session.buffer = session.buffer.subarray(parsed.bytesConsumed);
const observedAt = config.now().toISOString();
session.sequence += 1;
const message = normalizeAdapterMessage({
schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA,
edgeRef: config.edgeRef,
adapterRef: config.adapter.adapterRef,
protocolProfileRef: config.profile.profileRef,
protocol: config.profile.protocol,
sessionRef: session.sessionRef,
...(config.routeRef ? { routeRef: config.routeRef } : {}),
messageRef: messageRef(parsed),
messageType: parsed.messageType,
sequence: session.sequence,
observedAt,
idempotencyKey: messageIdempotencyKey({
adapterRef: config.adapter.adapterRef,
profileRef: config.profile.profileRef,
identifier: session.identifier,
payload: parsed.payload,
}),
identifier: session.identifier,
payloadSchemaRef: parsed.payloadSchemaRef,
payload: parsed.payload,
});
const acceptance = normalizeAdapterAcceptance(
await config.onMessage(message),
);
if (acceptance.idempotencyKey !== message.idempotencyKey) {
throw new TypeError("device_gateway_core_acceptance_mismatch");
}
consumeBuffer(session, parsed.bytesConsumed);
totalMessagesAccepted += 1;
totalPackagesAcknowledged += 1;
socket.write(buildB2PackageAcknowledgement(parsed.packageNumber));
await writeWithBackpressure(
socket,
session.adapterSession.buildMessageAcknowledgement(parsed),
);
}
}
@@ -199,8 +250,16 @@ export function createDeviceGatewayRuntime(options = {}) {
function consumeConnectionPermit(remoteAddress) {
const nowMs = config.now().getTime();
for (const [address, window] of connectionWindows) {
if (nowMs - window.startedAt >= 60_000) {
connectionWindows.delete(address);
}
}
const current = connectionWindows.get(remoteAddress);
if (!current || nowMs - current.startedAt >= 60_000) {
if (connectionWindows.size >= config.maxTrackedSourceAddresses) {
return false;
}
connectionWindows.set(remoteAddress, {
startedAt: nowMs,
count: 1,
@@ -225,9 +284,24 @@ export function createDeviceGatewayRuntime(options = {}) {
function closeSession(socket, session) {
if (session.closed) return;
session.closed = true;
totalBufferedBytes -= session.buffer.length;
session.buffer = Buffer.alloc(0);
sessions.delete(socket);
decrementAddressSessions(session.remoteAddress);
}
function appendBuffer(session, chunk) {
session.buffer = Buffer.concat(
[session.buffer, chunk],
session.buffer.length + chunk.length,
);
totalBufferedBytes += chunk.length;
}
function consumeBuffer(session, bytesConsumed) {
session.buffer = session.buffer.subarray(bytesConsumed);
totalBufferedBytes -= bytesConsumed;
}
}
function normalizeConfig(input) {
@@ -236,15 +310,30 @@ function normalizeConfig(input) {
if (publicIngressEnabled && !listenEnabled) {
throw new TypeError("device_gateway_public_ingress_listener_required");
}
if (
publicIngressEnabled
&& typeof input.onDiscovery !== "function"
) {
throw new TypeError("device_gateway_discovery_sink_required");
if (publicIngressEnabled && input.coreChannelAuthenticated !== true) {
throw new TypeError("device_gateway_authenticated_core_channel_required");
}
if (
listenEnabled
&& (typeof input.onDiscovery !== "function"
|| typeof input.onMessage !== "function")
) {
throw new TypeError("device_gateway_core_acceptance_sink_required");
}
const registration = listenEnabled
? resolveAdapterRegistration(input.adapterRegistry, input.protocolProfileRef)
: null;
return {
listenEnabled,
publicIngressEnabled,
adapter: registration?.adapter ?? null,
profile: registration?.profile ?? null,
edgeRef: listenEnabled
? normalizeOpaqueRef(input.edgeRef, "device_gateway_edge_ref_invalid")
: "edge:disabled",
routeRef: input.routeRef == null || input.routeRef === ""
? undefined
: normalizeRouteRef(input.routeRef),
healthHost: normalizeHealthHost(input.healthHost, "127.0.0.1"),
healthPort: parseInteger(
input.healthPort,
@@ -267,32 +356,46 @@ function normalizeConfig(input) {
),
maxBufferedBytes: parseInteger(
input.maxBufferedBytes,
ARUSNAVI_B2_MODEL_PROFILE.framing.maxBufferedBytes,
registration?.profile?.framing?.maxBufferedBytes ?? 256 * 1024,
1024,
ARUSNAVI_B2_MODEL_PROFILE.framing.maxBufferedBytes,
registration?.profile?.framing?.maxBufferedBytes ?? 256 * 1024,
"device_gateway_buffer_limit_invalid",
),
maxAggregateBufferedBytes: parseInteger(
input.maxAggregateBufferedBytes,
32 * 1024 * 1024,
1024,
32 * 1024 * 1024,
"device_gateway_aggregate_buffer_limit_invalid",
),
maxConcurrentSessions: parseInteger(
input.maxConcurrentSessions,
100,
128,
1,
10000,
"device_gateway_session_limit_invalid",
),
maxSessionsPerAddress: parseInteger(
input.maxSessionsPerAddress,
10,
16,
1,
1000,
"device_gateway_address_session_limit_invalid",
),
maxConnectionsPerMinutePerAddress: parseInteger(
input.maxConnectionsPerMinutePerAddress,
30,
60,
1,
10000,
"device_gateway_address_rate_limit_invalid",
),
maxTrackedSourceAddresses: parseInteger(
input.maxTrackedSourceAddresses,
2048,
1,
65536,
"device_gateway_source_tracking_limit_invalid",
),
sessionTimeoutMs: parseInteger(
input.sessionTimeoutMs,
10000,
@@ -303,10 +406,83 @@ function normalizeConfig(input) {
onDiscovery: typeof input.onDiscovery === "function"
? input.onDiscovery
: undefined,
onMessage: typeof input.onMessage === "function"
? input.onMessage
: undefined,
now: typeof input.now === "function" ? input.now : () => new Date(),
};
}
function resolveAdapterRegistration(registry, profileRef) {
if (!registry || typeof registry.resolveProfile !== "function") {
throw new TypeError("device_gateway_adapter_registry_required");
}
return registry.resolveProfile(profileRef);
}
function normalizeOpaqueRef(value, errorCode) {
if (
typeof value !== "string"
|| !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(value)
) {
throw new TypeError(errorCode);
}
return value;
}
function normalizeRouteRef(value) {
if (
typeof value !== "string"
|| !/^route:[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value)
) {
throw new TypeError("device_gateway_route_ref_invalid");
}
return value.toLowerCase();
}
function assertDiscoveryAccepted(value) {
if (
!value
|| typeof value !== "object"
|| !["quarantine", "claimed"].includes(value.lifecycleState)
) {
throw new TypeError("device_gateway_core_discovery_not_accepted");
}
}
function messageRef(parsed) {
const digest = String(parsed?.payload?.packageDigest || "");
if (!/^sha256:[a-f0-9]{64}$/.test(digest)) {
throw new TypeError("device_gateway_adapter_message_digest_invalid");
}
return `package:${parsed.packageNumber}:${digest.slice("sha256:".length)}`;
}
function messageIdempotencyKey({ adapterRef, profileRef, identifier, payload }) {
return `sha256:${createHash("sha256")
.update(JSON.stringify({ adapterRef, profileRef, identifier, payload }), "utf8")
.digest("hex")}`;
}
function writeWithBackpressure(socket, bytes) {
if (!Buffer.isBuffer(bytes) || bytes.length === 0 || bytes.length > 4096) {
throw new TypeError("device_gateway_adapter_ack_invalid");
}
if (socket.write(bytes)) return Promise.resolve();
return new Promise((resolve, reject) => {
const onDrain = () => {
socket.off("error", onError);
resolve();
};
const onError = (error) => {
socket.off("drain", onDrain);
reject(error);
};
socket.once("drain", onDrain);
socket.once("error", onError);
});
}
function normalizeHealthHost(value, fallback) {
const normalized = String(value || fallback).trim();
if (!["127.0.0.1", "::1", "0.0.0.0", "::"].includes(normalized)) {