From 66e0c6245186805d1a41702f4d1e5b5e1645151f Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 12 Aug 2026 18:08:44 +0300 Subject: [PATCH] fix(device-plane): unwrap durable discovery receipts --- .../src/device-gateway-core-runtime.mjs | 26 ++++++++- .../test/channel-integration.test.mjs | 53 ++++++++++++++++--- 2 files changed, 72 insertions(+), 7 deletions(-) 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 ef2574a..a8686e0 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 @@ -12,6 +12,8 @@ import { normalizeCertificateFingerprint, } from "../../../packages/device-edge-channel-contract/src/index.mjs"; import { + DEVICE_DISCOVERY_VIEW_SCHEMA, + assertSafeProjection, normalizeAdapterMessage, normalizeDiscoverySignal, } from "../../../packages/device-protocol-contract/src/index.mjs"; @@ -298,7 +300,9 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { async function acceptDiscovery(connection, envelope) { try { const signal = normalizeDiscoverySignal(envelope.payload?.signal); - const discovery = await config.observeDiscovery(signal); + const discovery = normalizeDiscoveryReceipt( + await config.observeDiscovery(signal), + ); sendEventResult(connection, envelope, { discovery }); totalEventsAccepted += 1; } catch (error) { @@ -468,6 +472,26 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { } } +function normalizeDiscoveryReceipt(input) { + if ( + !input + || typeof input !== "object" + || Array.isArray(input) + || typeof input.created !== "boolean" + ) { + throw new TypeError("device_gateway_core_discovery_receipt_invalid"); + } + const discovery = assertSafeProjection(input.value); + if ( + discovery.schemaVersion !== DEVICE_DISCOVERY_VIEW_SCHEMA + || !["quarantine", "claimed"].includes(discovery.lifecycleState) + || discovery.commandTransport !== "disabled" + ) { + throw new TypeError("device_gateway_core_discovery_receipt_invalid"); + } + return discovery; +} + function normalizeConfig(options) { if (typeof options.observeDiscovery !== "function") { throw new TypeError("device_gateway_core_observe_discovery_invalid"); 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 c1a2eaa..1143ff4 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 @@ -15,6 +15,8 @@ import { } from "../../../packages/device-protocol-contract/src/index.mjs"; import { createDeviceGatewayCoreChannelClient } from "../../device-gateway-core/src/runtime.mjs"; +import { createDeviceGatewayIngest } from + "../../device-control-core/src/gateway-ingest.mjs"; import { createDeviceEdgeChannelServer } from "../src/runtime.mjs"; let fixtureDirectory; @@ -33,6 +35,7 @@ test("accepts synthetic discovery and durable message results over Core-initiate const acceptedMessages = new Map(); let messageCalls = 0; const pair = await startPair({ + observeDiscovery: productionDiscoveryObserver(), acceptMessage: async (message) => { messageCalls += 1; const previous = acceptedMessages.get(message.idempotencyKey); @@ -47,6 +50,8 @@ test("accepts synthetic discovery and durable message results over Core-initiate const discovery = await pair.edge.submitDiscovery(discoverySignal()); assert.equal(discovery.lifecycleState, "quarantine"); assert.equal(discovery.identifier.masked, "***********1088"); + assert.equal("created" in discovery, false); + assert.equal("value" in discovery, false); const message = adapterMessage(); const first = await pair.edge.submitAdapterMessage(message); @@ -274,6 +279,27 @@ test("returns a conclusive rejection when Core cannot accept a package", async ( } }); +test("rejects a discovery when Core does not return its durable receipt", async () => { + const pair = await startPair({ + observeDiscovery: async () => ({ + schemaVersion: "nodedc.device.discovery-view.v1", + lifecycleState: "quarantine", + commandTransport: "disabled", + identifier: { kind: "imei", masked: "***********1088" }, + }), + }); + try { + await assert.rejects( + pair.edge.submitDiscovery(discoverySignal()), + /device_gateway_core_discovery_receipt_invalid/, + ); + assert.equal(pair.edge.status().eventsRejected, 1); + assert.equal(pair.core.status().eventsRejected, 1); + } finally { + await stopPair(pair); + } +}); + test("isolates tracker session ordering while allowing cross-session progress", async () => { let releaseSlow; const slowGate = new Promise((resolve) => { @@ -461,17 +487,32 @@ function createCoreClient(options) { reconnectMaximumMs: options.reconnectMaximumMs ?? 80, connectTimeoutMs: options.connectTimeoutMs, random: () => 0, - observeDiscovery: options.observeDiscovery ?? (async () => ({ - schemaVersion: "nodedc.device.discovery-view.v1", - lifecycleState: "quarantine", - commandTransport: "disabled", - identifier: { kind: "imei", masked: "***********1088" }, - })), + observeDiscovery: options.observeDiscovery ?? productionDiscoveryObserver(), acceptMessage: options.acceptMessage ?? (async (message) => acceptanceFor(message, false)), }); } +function productionDiscoveryObserver() { + return createDeviceGatewayIngest({ + identifierPepper: "test-only-device-edge-channel-identifier-pepper", + repository: { + async upsertQuarantineDiscovery(value) { + return { + created: true, + value: { + ...value.safeView, + discoveryRef: "discovery:pilot-1", + }, + }; + }, + async acceptAdapterMessage() { + throw new Error("device_edge_channel_test_unexpected_ingest_message"); + }, + }, + }).observeDiscovery; +} + function edgeRegistration(address, certificateIdentities, lifecycleState = "active") { return { edgeRegistrationId: "edge:pilot-1",