fix(device-plane): unwrap durable discovery receipts
This commit is contained in:
@@ -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");
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user