quixos/camino-package-runtime
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Format authored monorepo code with pinned language formatters

Browse Source
This commit is contained in:
Timothy J. Aveni
2026-09-15 15:23:24 -07:00
parent bee4452852
commit 0d3c013c07
12 changed files with 1223 additions and 674 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flake.nix
+35 -9
View File
@@ -11,7 +11,13 @@
};
};
outputs = inputs@{ nixpkgs, flake-utils, quixosNixHelpers, ... }:
outputs =
inputs@{
nixpkgs,
flake-utils,
quixosNixHelpers,
...
}:
let
quixosHelpers = import "${quixosNixHelpers}/quixos-package-helpers.nix";
packageOutputs = quixosHelpers.mkCaminoTsYarnNixifyFlake {
@@ -22,18 +28,37 @@
nativeBuildInputs = { pkgs, ... }: [
pkgs.protobuf
];
buildEnv = { inputs, pkgs, system }: {
QUIXOS_PROTO_PATH = "${pkgs.protobuf}/include:${inputs.quixos-protocol.packages.${system}.default}/proto";
};
buildEnv =
{
inputs,
pkgs,
system,
}:
{
QUIXOS_PROTO_PATH = "${pkgs.protobuf}/include:${
inputs.quixos-protocol.packages.${system}.default
}/proto";
};
devShellPackages = { pkgs, ... }: [
pkgs.protobuf
];
devShellHook = { inputs, pkgs, system, ... }: ''
export QUIXOS_PROTO_PATH="${pkgs.protobuf}/include:${inputs.quixos-protocol.packages.${system}.default}/proto''${QUIXOS_PROTO_PATH:+:$QUIXOS_PROTO_PATH}"
'';
devShellHook =
{
inputs,
pkgs,
system,
...
}:
''
export QUIXOS_PROTO_PATH="${pkgs.protobuf}/include:${
inputs.quixos-protocol.packages.${system}.default
}/proto''${QUIXOS_PROTO_PATH:+:$QUIXOS_PROTO_PATH}"
'';
};
in
packageOutputs // flake-utils.lib.eachDefaultSystem (system:
packageOutputs
// flake-utils.lib.eachDefaultSystem (
system:
let
pkgs = import nixpkgs { inherit system; };
builtPackage = packageOutputs.packages.${system}.default;
@@ -45,5 +70,6 @@
${builtPackage}/libexec/-quixos-camino-package-runtime/dist
touch "$out"
'';
});
}
);
}
src/bindings.ts
+144 -55
View File
@@ -1,7 +1,14 @@
import { create } from "@bufbuild/protobuf";
import { ValueSchema, ObjectValueSchema, type Value } from "./camino/api_pb.js";
import { derived, jsToProtoValue, liveValue, protoValueToJs,
type RuntimeContext, type RuntimeHandler, type DerivedHandler } from "./index.js";
import {
derived,
jsToProtoValue,
liveValue,
protoValueToJs,
type RuntimeContext,
type RuntimeHandler,
type DerivedHandler,
} from "./index.js";
export type { QxObjectRef } from "./references.js";
import { assertReferenceFree, referenceToWire } from "./references.js";
@@ -25,8 +32,8 @@ export const opaqueReactPropsBinding: MessageBinding<Record<string, unknown>> =
export type BindingValue<B> = B extends MessageBinding<infer T> ? T : never;
export type QxHandler<C, O> = (context: C) => O | Promise<O>;
export type QxDerived<C, O> = { kind: "derived"; get: QxHandler<C, O> };
export type QxSession<C> = {id: string; run<T>(work: (context: C) => Promise<T>): Promise<T>; close(): Promise<void>};
export type QxContextLifecycle<C> = {signal?: AbortSignal; openSession?: () => Promise<QxSession<C>>};
export type QxSession<C> = { id: string; run<T>(work: (context: C) => Promise<T>): Promise<T>; close(): Promise<void> };
export type QxContextLifecycle<C> = { signal?: AbortSignal; openSession?: () => Promise<QxSession<C>> };
export const qxDerived = <C, O>(get: QxHandler<C, O>): QxDerived<C, O> => ({ kind: "derived", get });
/** Versioned binding ABI. This mirrors the language-neutral value IR. */
@@ -44,7 +51,9 @@ export type QxPortSpec =
| { kind: "interface"; id: string; operations: Record<string, QxOperationSpec> }
| { kind: "constructor"; id: string; inputType: QxValueType };
export type QxHandlerSpec = {
inputType: QxValueType; outputType: QxValueType; eventType?: QxValueType;
inputType: QxValueType;
outputType: QxValueType;
eventType?: QxValueType;
ports: Record<string, QxPortSpec>;
};
export type QxMessages = Record<string, MessageBinding<any>>;
@@ -57,10 +66,14 @@ export const decodeQxValue = (type: QxValueType, value: Value | undefined, messa
if (type.kind === "record") {
if (value.kind.case !== "objectValue") throw new Error("Expected QX record");
const fields = value.kind.value.fields;
if (Object.keys(fields).some((name) => !Object.hasOwn(type.fields, name))) throw new Error("Unexpected QX record field");
return Object.fromEntries(Object.entries(type.fields).map(([name, field]) => [name, decodeQxValue(field, fields[name], messages)]));
if (Object.keys(fields).some((name) => !Object.hasOwn(type.fields, name)))
throw new Error("Unexpected QX record field");
return Object.fromEntries(
Object.entries(type.fields).map(([name, field]) => [name, decodeQxValue(field, fields[name], messages)]),
);
}
if (type.kind === "optional") return value.kind.case === "nullValue" ? null : decodeQxValue(type.value, value, messages);
if (type.kind === "optional")
return value.kind.case === "nullValue" ? null : decodeQxValue(type.value, value, messages);
if (type.kind === "list") {
if (value.kind.case !== "listValue") throw new Error("Expected QX list");
return value.kind.value.values.map((entry) => decodeQxValue(type.value, entry, messages));
@@ -98,16 +111,25 @@ export const encodeQxValue = (type: QxValueType, value: any, messages: QxMessage
if (type.kind === "builtin" && type.name === "unit") return jsToProtoValue(null);
if (type.kind === "record") {
if (!value || typeof value !== "object" || Array.isArray(value)) throw new Error("Expected QX record");
if (Object.keys(value).some((name) => !Object.hasOwn(type.fields, name))) throw new Error("Unexpected QX record field");
const fields = Object.fromEntries(Object.entries(type.fields).map(([name, field]) => {
if (!Object.hasOwn(value, name) && field.kind !== "optional") throw new Error(`Missing QX record field ${name}`);
return [name, encodeQxValue(field, value[name] ?? (field.kind === "optional" ? null : value[name]), messages)];
}));
return create(ValueSchema, {kind: {case: "objectValue", value: create(ObjectValueSchema, {fields})}});
if (Object.keys(value).some((name) => !Object.hasOwn(type.fields, name)))
throw new Error("Unexpected QX record field");
const fields = Object.fromEntries(
Object.entries(type.fields).map(([name, field]) => {
if (!Object.hasOwn(value, name) && field.kind !== "optional")
throw new Error(`Missing QX record field ${name}`);
return [name, encodeQxValue(field, value[name] ?? (field.kind === "optional" ? null : value[name]), messages)];
}),
);
return create(ValueSchema, { kind: { case: "objectValue", value: create(ObjectValueSchema, { fields }) } });
}
if (type.kind === "optional")
return value === null ? jsToProtoValue(null) : encodeQxValue(type.value, value, messages);
if (type.kind === "list")
return jsToProtoValue(value.map((entry: unknown) => liveValue(encodeQxValue(type.value, entry, messages))));
if (type.kind === "object-ref") {
referenceToWire(value);
return jsToProtoValue(value);
}
if (type.kind === "optional") return value === null ? jsToProtoValue(null) : encodeQxValue(type.value, value, messages);
if (type.kind === "list") return jsToProtoValue(value.map((entry: unknown) => liveValue(encodeQxValue(type.value, entry, messages))));
if (type.kind === "object-ref") { referenceToWire(value); return jsToProtoValue(value); }
if (type.kind !== "message" || type.descriptorId !== reactPropsDescriptor) assertReferenceFree(value);
if (type.kind === "message") {
const encoded = requireMessage(messages, type.descriptorId).encode(value);
@@ -118,8 +140,10 @@ export const encodeQxValue = (type: QxValueType, value: any, messages: QxMessage
};
const inputValue = (context: RuntimeContext, type: QxValueType) => {
if (type.kind === "message" || type.kind === "record") return create(ValueSchema, { kind: { case: "objectValue",
value: create(ObjectValueSchema, { fields: context.inputProto }) } });
if (type.kind === "message" || type.kind === "record")
return create(ValueSchema, {
kind: { case: "objectValue", value: create(ObjectValueSchema, { fields: context.inputProto }) },
});
return context.inputProto.value;
};
const inputFields = (type: QxValueType, value: unknown, messages: QxMessages): Record<string, unknown> => {
@@ -134,45 +158,110 @@ const inputFields = (type: QxValueType, value: unknown, messages: QxMessages): R
/** The sole unchecked cast connects generated contracts to the dynamic RPC runtime. */
export const bindQxHandler = <C, O>(
spec: QxHandlerSpec, handler: QxHandler<C, O> | QxDerived<C, O>, messages: QxMessages,
spec: QxHandlerSpec,
handler: QxHandler<C, O> | QxDerived<C, O>,
messages: QxMessages,
): RuntimeHandler | DerivedHandler => {
const bindContext = (raw: RuntimeContext): C => {
const ports = Object.fromEntries(Object.entries(spec.ports).map(([name, port]) => {
switch (port.kind) {
case "state": {
const state = raw.state(port.id);
return [name, {
...(port.primitives.includes("read") ? { get: async () => decodeQxValue(port.valueType, (await state.live()).$quixosValue, messages), live: () => state.live() } : {}),
...(port.primitives.includes("write") ? { set: async (value: unknown) => state.set(liveValue(encodeQxValue(port.valueType, value, messages))) } : {}),
}];
const ports = Object.fromEntries(
Object.entries(spec.ports).map(([name, port]) => {
switch (port.kind) {
case "state": {
const state = raw.state(port.id);
return [
name,
{
...(port.primitives.includes("read")
? {
get: async () => decodeQxValue(port.valueType, (await state.live()).$quixosValue, messages),
live: () => state.live(),
}
: {}),
...(port.primitives.includes("write")
? {
set: async (value: unknown) =>
state.set(liveValue(encodeQxValue(port.valueType, value, messages))),
}
: {}),
},
];
}
case "edge": {
const edge = raw.edge(port.id);
return [
name,
{
...Object.fromEntries(
port.primitives.map((primitive) => [
primitive,
edge[primitive as "resolve" | "connect" | "disconnect"],
]),
),
...(port.primitives.includes("resolve") ? { collection: edge.collection } : {}),
...(port.primitives.includes("resolve") &&
port.primitives.includes("connect") &&
port.primitives.includes("disconnect")
? { replace: edge.replace }
: {}),
},
];
}
case "interface": {
const target = raw.interface(port.id);
return [
name,
{
objectId: target.objectId,
live: Object.fromEntries(
Object.entries(port.operations).map(([name, operation]) => [
name,
(input: unknown) => target.live(operation.id, inputFields(operation.inputType, input, messages)),
]),
),
...Object.fromEntries(
Object.entries(port.operations).map(([name, operation]) => [
name,
async (input: unknown) =>
decodeQxValue(
operation.outputType,
(await target.live(operation.id, inputFields(operation.inputType, input, messages)))
.$quixosValue,
messages,
),
]),
),
},
];
}
case "constructor":
return [
name,
{
construct: (input: unknown) =>
raw.constructor(port.id).construct(inputFields(port.inputType, input, messages)),
},
];
}
case "edge": {
const edge = raw.edge(port.id);
return [name, {...Object.fromEntries(port.primitives.map((primitive) => [primitive, edge[primitive as "resolve" | "connect" | "disconnect"]])),
...(port.primitives.includes("resolve") ? {collection: edge.collection} : {}),
...(port.primitives.includes("resolve") && port.primitives.includes("connect") && port.primitives.includes("disconnect") ? {replace: edge.replace} : {})}];
}
case "interface": {
const target = raw.interface(port.id);
return [name, {objectId: target.objectId,
live: Object.fromEntries(Object.entries(port.operations).map(([name, operation]) => [name,
(input: unknown) => target.live(operation.id, inputFields(operation.inputType, input, messages)),
])),
...Object.fromEntries(Object.entries(port.operations).map(([name, operation]) => [name,
async (input: unknown) => decodeQxValue(operation.outputType,
(await target.live(operation.id, inputFields(operation.inputType, input, messages))).$quixosValue, messages),
]))}];
}
case "constructor": return [name, { construct: (input: unknown) => raw.constructor(port.id).construct(inputFields(port.inputType, input, messages)) }];
}
}));
return { objectId: raw.objectId, signal: raw.signal,
...(raw.openSession ? {openSession: async () => {
const session = await raw.openSession!();
return {id: session.id, close: () => session.close(),
run: <T>(work: (context: C) => Promise<T>) => session.run((next) => work(bindContext(next)))};
}} : {}),
input: decodeQxValue(spec.inputType, inputValue(raw, spec.inputType), messages), ports } as C;
}),
);
return {
objectId: raw.objectId,
signal: raw.signal,
...(raw.openSession
? {
openSession: async () => {
const session = await raw.openSession!();
return {
id: session.id,
close: () => session.close(),
run: <T>(work: (context: C) => Promise<T>) => session.run((next) => work(bindContext(next))),
};
},
}
: {}),
input: decodeQxValue(spec.inputType, inputValue(raw, spec.inputType), messages),
ports,
} as C;
};
const execute = async (raw: RuntimeContext) => {
const context = bindContext(raw);
src/index.ts
+435 -271
View File
@@ -1,12 +1,26 @@
import http from "node:http";
import { readFileSync } from "node:fs";
import { createInvocationRegistry } from "./invocations.js";
import { isObjectReference, referenceFromWire, referenceToWire, assertReferenceFree, type QxObjectRef } from "./references.js";
import {
isObjectReference,
referenceFromWire,
referenceToWire,
assertReferenceFree,
type QxObjectRef,
} from "./references.js";
export * from "./bindings.js";
export {relationshipMap, relationshipList, relationshipSet} from "./relationships.js";
export { relationshipMap, relationshipList, relationshipSet } from "./relationships.js";
import { AsyncLocalStorage } from "node:async_hooks";
import { createHmac, randomBytes, randomUUID, timingSafeEqual } from "node:crypto";
export {createMigrationContext, migrationObjectId, serveMigration, type MigrationContext, type MigrationInput, type MigrationOutput, type MigrationEdge} from "./migration.js";
export {
createMigrationContext,
migrationObjectId,
serveMigration,
type MigrationContext,
type MigrationInput,
type MigrationOutput,
type MigrationEdge,
} from "./migration.js";
import { create, equals } from "@bufbuild/protobuf";
import { Code, ConnectError, createClient, type Client, type ConnectRouter } from "@connectrpc/connect";
import { connectNodeAdapter, createConnectTransport } from "@connectrpc/connect-node";
@@ -63,14 +77,22 @@ const base64ToBytes = (value: string) => Buffer.from(value, "base64");
const isRecord = (value: unknown): value is Record<string, unknown> =>
Boolean(value) && typeof value === "object" && !Array.isArray(value);
const isWrappedValue = (value: unknown): value is { $quixosValue: Value } =>
isRecord(value) && "$quixosValue" in value &&
isRecord(value.$quixosValue) && value.$quixosValue.$typeName === "camino.Value";
isRecord(value) &&
"$quixosValue" in value &&
isRecord(value.$quixosValue) &&
value.$quixosValue.$typeName === "camino.Value";
export const objectRef = (reference: QxObjectRef) => { referenceToWire(reference); return reference; };
export const objectRef = (reference: QxObjectRef) => {
referenceToWire(reference);
return reference;
};
export const liveValue = (value: Value) => ({ $quixosValue: value });
export const jsToProtoValue = (value: unknown): Value => {
if (isObjectReference(value)) return create(ValueSchema, {kind: {case: "refValue", value: create(RefValueSchema, {objectId: referenceToWire(value)})}});
if (isObjectReference(value))
return create(ValueSchema, {
kind: { case: "refValue", value: create(RefValueSchema, { objectId: referenceToWire(value) }) },
});
if (isWrappedValue(value)) return value.$quixosValue;
if (value === null || value === undefined) {
return create(ValueSchema, { kind: { case: "nullValue", value: create(NullValueSchema, {}) } });
@@ -86,43 +108,55 @@ export const jsToProtoValue = (value: unknown): Value => {
});
}
if (isRecord(value) && "$quixosRef" in value) throw new Error("Raw ID wrappers are not object references");
if (isRecord(value) && typeof value.$quixosCrdtType === "string" &&
typeof value.$quixosCrdtPayload === "string") {
if (isRecord(value) && typeof value.$quixosCrdtType === "string" && typeof value.$quixosCrdtPayload === "string") {
return create(ValueSchema, {
kind: { case: "crdtValue", value: create(CrdtValueSchema, {
type: value.$quixosCrdtType,
encoding: typeof value.$quixosCrdtEncoding === "string" ? value.$quixosCrdtEncoding : "base64",
payload: base64ToBytes(value.$quixosCrdtPayload),
}) },
kind: {
case: "crdtValue",
value: create(CrdtValueSchema, {
type: value.$quixosCrdtType,
encoding: typeof value.$quixosCrdtEncoding === "string" ? value.$quixosCrdtEncoding : "base64",
payload: base64ToBytes(value.$quixosCrdtPayload),
}),
},
});
}
if (!isRecord(value)) throw new Error(`Unsupported runtime value ${typeof value}`);
return create(ValueSchema, {
kind: { case: "objectValue", value: create(ObjectValueSchema, {
fields: Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, jsToProtoValue(entry)])),
}) },
kind: {
case: "objectValue",
value: create(ObjectValueSchema, {
fields: Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, jsToProtoValue(entry)])),
}),
},
});
};
export const protoValueToJs = (value: Value | undefined): unknown => {
switch (value?.kind.case) {
case "nullValue":
case undefined: return null;
case undefined:
return null;
case "boolValue":
case "numberValue":
case "stringValue":
case "integerValue": return value.kind.value;
case "bytesValue": return bytesToBase64(value.kind.value);
case "refValue": return referenceFromWire(value.kind.value.objectId);
case "listValue": return value.kind.value.values.map(protoValueToJs);
case "objectValue": return Object.fromEntries(
Object.entries(value.kind.value.fields).map(([key, entry]) => [key, protoValueToJs(entry)]),
);
case "crdtValue": return {
$quixosCrdtType: value.kind.value.type,
$quixosCrdtEncoding: value.kind.value.encoding,
$quixosCrdtPayload: bytesToBase64(value.kind.value.payload),
};
case "integerValue":
return value.kind.value;
case "bytesValue":
return bytesToBase64(value.kind.value);
case "refValue":
return referenceFromWire(value.kind.value.objectId);
case "listValue":
return value.kind.value.values.map(protoValueToJs);
case "objectValue":
return Object.fromEntries(
Object.entries(value.kind.value.fields).map(([key, entry]) => [key, protoValueToJs(entry)]),
);
case "crdtValue":
return {
$quixosCrdtType: value.kind.value.type,
$quixosCrdtEncoding: value.kind.value.encoding,
$quixosCrdtPayload: bytesToBase64(value.kind.value.payload),
};
}
};
@@ -144,8 +178,15 @@ export type EdgePort = {
collection(): Promise<RelationshipCollection>;
replace(entries: RelationshipEntry[], expectedRevision: bigint): Promise<RelationshipCollection>;
};
export type RelationshipEntry<T extends QxObjectRef = QxObjectRef> = {edgeId?: string; target: T; key?: string | boolean | bigint};
export type RelationshipCollection<T extends QxObjectRef = QxObjectRef> = {revision: bigint; entries: RelationshipEntry<T>[]};
export type RelationshipEntry<T extends QxObjectRef = QxObjectRef> = {
edgeId?: string;
target: T;
key?: string | boolean | bigint;
};
export type RelationshipCollection<T extends QxObjectRef = QxObjectRef> = {
revision: bigint;
entries: RelationshipEntry<T>[];
};
export type InterfacePort = {
objectId: QxObjectRef;
interfaceRevisionId: string;
@@ -180,13 +221,17 @@ export type RuntimeSession = {
};
export class RuntimeAuthorityError extends Error {
readonly retryable: boolean;
constructor(message: string) { super(message); this.name = "RuntimeAuthorityError"; this.retryable = /WORKSPACE_FENCED|STALE_EPOCH/.test(message); }
constructor(message: string) {
super(message);
this.name = "RuntimeAuthorityError";
this.retryable = /WORKSPACE_FENCED|STALE_EPOCH/.test(message);
}
}
const targetForEdge = (
edge: { firstObjectId: string; secondObjectId: string; firstProjectionId: string },
projectionId: string,
) => edge.firstProjectionId === projectionId ? edge.secondObjectId : edge.firstObjectId;
) => (edge.firstProjectionId === projectionId ? edge.secondObjectId : edge.firstObjectId);
export const createRuntimeContext = (
camino: CaminoClient,
@@ -222,25 +267,63 @@ export const createRuntimeContext = (
case "edge": {
const { edgeTypeId, projectionId } = dependency.binding.value;
const dependencyObjectId = dependency.objectId || request.objectId;
const collectionResult = (response: {revision: bigint; entries: {edgeId: string; targetObjectId: string; key?: Value}[]}): RelationshipCollection => ({revision: response.revision,
entries: response.entries.map((entry) => ({edgeId: entry.edgeId, target: referenceFromWire(entry.targetObjectId),
...(entry.key ? {key: entry.key.kind.case === "integerValue" ? BigInt(entry.key.kind.value) : protoValueToJs(entry.key) as string | boolean} : {})}))});
const collectionResult = (response: {
revision: bigint;
entries: { edgeId: string; targetObjectId: string; key?: Value }[];
}): RelationshipCollection => ({
revision: response.revision,
entries: response.entries.map((entry) => ({
edgeId: entry.edgeId,
target: referenceFromWire(entry.targetObjectId),
...(entry.key
? {
key:
entry.key.kind.case === "integerValue"
? BigInt(entry.key.kind.value)
: (protoValueToJs(entry.key) as string | boolean),
}
: {}),
})),
});
const edge: EdgePort = {
edgeTypeId,
projectionId,
async collection() {
await recordDependency({kind: "edge", objectId: dependencyObjectId, attachmentId: edgeTypeId, projectionId});
return collectionResult(await camino.readCollection({objectId: dependencyObjectId, edgeTypeId, projectionId}));
await recordDependency({
kind: "edge",
objectId: dependencyObjectId,
attachmentId: edgeTypeId,
projectionId,
});
return collectionResult(
await camino.readCollection({ objectId: dependencyObjectId, edgeTypeId, projectionId }),
);
},
async replace(entries, expectedRevision) {
return collectionResult(await camino.replaceCollection({objectId: dependencyObjectId, edgeTypeId, projectionId, expectedRevision,
entries: entries.map((entry) => {
assertReferenceFree(entry.key);
return {edgeId: entry.edgeId ?? "", targetObjectId: referenceToWire(entry.target), key: entry.key === undefined ? undefined : jsToProtoValue(entry.key)};
})}));
return collectionResult(
await camino.replaceCollection({
objectId: dependencyObjectId,
edgeTypeId,
projectionId,
expectedRevision,
entries: entries.map((entry) => {
assertReferenceFree(entry.key);
return {
edgeId: entry.edgeId ?? "",
targetObjectId: referenceToWire(entry.target),
key: entry.key === undefined ? undefined : jsToProtoValue(entry.key),
};
}),
}),
);
},
async resolve() {
await recordDependency({ kind: "edge", objectId: dependencyObjectId, attachmentId: edgeTypeId, projectionId });
await recordDependency({
kind: "edge",
objectId: dependencyObjectId,
attachmentId: edgeTypeId,
projectionId,
});
const result = await camino.resolveEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId });
return result.edges.map((entry) => referenceFromWire(targetForEdge(entry, projectionId)));
},
@@ -252,7 +335,8 @@ export const createRuntimeContext = (
const targetObjectId = referenceToWire(target);
const result = await camino.resolveEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId });
for (const entry of result.edges) {
if (targetForEdge(entry, projectionId) === targetObjectId) await camino.disconnectEdge({ edgeId: entry.id });
if (targetForEdge(entry, projectionId) === targetObjectId)
await camino.disconnectEdge({ edgeId: entry.id });
}
},
};
@@ -348,20 +432,21 @@ const evaluate = async (
observe?: (dependency: RuntimeDependency) => Promise<void>,
) => {
const dependencies = new Map<string, RuntimeDependency>();
const result = await dependencyScope.run(
{ dependencies, observe },
() => isDerived(handler) ? handler.get(context) : handler(context),
const result = await dependencyScope.run({ dependencies, observe }, () =>
isDerived(handler) ? handler.get(context) : handler(context),
);
return { value: jsToProtoValue(result), dependencies: [...dependencies.values()] };
};
const protoDependencies = (dependencies: RuntimeDependency[]) => dependencies.map((entry) =>
create(DerivedDependencySchema, {
kind: entry.kind,
objectId: entry.objectId,
attachmentId: entry.attachmentId,
projectionId: entry.kind === "edge" ? entry.projectionId : "",
}));
const protoDependencies = (dependencies: RuntimeDependency[]) =>
dependencies.map((entry) =>
create(DerivedDependencySchema, {
kind: entry.kind,
objectId: entry.objectId,
attachmentId: entry.attachmentId,
projectionId: entry.kind === "edge" ? entry.projectionId : "",
}),
);
export const createPackageRuntimeRoutes = (config: {
packageRevisionId: string;
@@ -371,27 +456,38 @@ export const createPackageRuntimeRoutes = (config: {
}) => {
const invocations = createInvocationRegistry();
const headers: Record<string, string> = {};
const processToken = process.env.CAMINO_RUNTIME_AUTH_TOKEN ?? (process.env.CAMINO_RUNTIME_AUTH_TOKEN_FILE
? readFileSync(process.env.CAMINO_RUNTIME_AUTH_TOKEN_FILE, "utf8").trim() : "");
const processToken =
process.env.CAMINO_RUNTIME_AUTH_TOKEN ??
(process.env.CAMINO_RUNTIME_AUTH_TOKEN_FILE
? readFileSync(process.env.CAMINO_RUNTIME_AUTH_TOKEN_FILE, "utf8").trim()
: "");
if (processToken) {
headers["x-camino-runtime-token"] = processToken;
} else if (process.env.CAMINO_RUNTIME_AUTH_REQUIRED === "1") {
throw new Error("CAMINO_RUNTIME_AUTH_TOKEN is required");
}
const camino = createClient(CaminoService, createConnectTransport({
baseUrl: config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310",
httpVersion: "1.1",
interceptors: headers["x-camino-runtime-token"] ? [
(next) => async (request) => {
request.header.set("x-camino-runtime-token", headers["x-camino-runtime-token"]!);
return await next(request);
},
] : [],
}));
const orch = createClient(OrchestratorRuntime, createConnectTransport({
baseUrl: config.orchUrl ?? process.env.QUIXOS_ORCH_URL ?? "http://127.0.0.1:7311",
httpVersion: "1.1",
}));
const camino = createClient(
CaminoService,
createConnectTransport({
baseUrl: config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310",
httpVersion: "1.1",
interceptors: headers["x-camino-runtime-token"]
? [
(next) => async (request) => {
request.header.set("x-camino-runtime-token", headers["x-camino-runtime-token"]!);
return await next(request);
},
]
: [],
}),
);
const orch = createClient(
OrchestratorRuntime,
createConnectTransport({
baseUrl: config.orchUrl ?? process.env.QUIXOS_ORCH_URL ?? "http://127.0.0.1:7311",
httpVersion: "1.1",
}),
);
const authenticateInstance = (header: Headers) => {
if (!process.env.QUIXOS_RUNTIME_INSTANCE_ID) return; // standalone development ABI
@@ -403,35 +499,67 @@ export const createPackageRuntimeRoutes = (config: {
};
const clientsFor = (request: { context?: { grant: string; instanceId: string; workspaceEpoch: string } }) => {
const context = request.context;
if (process.env.QUIXOS_RUNTIME_INSTANCE_ID && (!context?.grant || context.instanceId !== process.env.QUIXOS_RUNTIME_INSTANCE_ID || !context.workspaceEpoch)) {
if (
process.env.QUIXOS_RUNTIME_INSTANCE_ID &&
(!context?.grant || context.instanceId !== process.env.QUIXOS_RUNTIME_INSTANCE_ID || !context.workspaceEpoch)
) {
throw new ConnectError("Managed invocation requires an exact instance and epoch grant", Code.Unauthenticated);
}
if (!context?.grant) return { camino, orch };
const transport = (url: string) => createConnectTransport({ baseUrl: url, httpVersion: "1.1", interceptors: [(next) => async (call) => {
call.header.set("x-quixos-invocation-grant", context.grant);
call.header.set("x-camino-runtime-token", processToken);
return next(call);
}] });
const transport = (url: string) =>
createConnectTransport({
baseUrl: url,
httpVersion: "1.1",
interceptors: [
(next) => async (call) => {
call.header.set("x-quixos-invocation-grant", context.grant);
call.header.set("x-camino-runtime-token", processToken);
return next(call);
},
],
});
return {
camino: createClient(CaminoService, transport(config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310")),
orch: createClient(OrchestratorRuntime, transport(config.orchUrl ?? process.env.QUIXOS_ORCH_URL ?? "http://127.0.0.1:7311")),
camino: createClient(
CaminoService,
transport(config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310"),
),
orch: createClient(
OrchestratorRuntime,
transport(config.orchUrl ?? process.env.QUIXOS_ORCH_URL ?? "http://127.0.0.1:7311"),
),
};
};
const runtimeControl = async <T>(operation: string, input: unknown): Promise<T> => {
const response = await fetch(`${config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310"}/__runtime/${operation}`, {
method: "POST", headers: { "content-type": "application/json", "x-camino-runtime-token": processToken }, body: JSON.stringify(input), signal: AbortSignal.timeout(10_000),
});
const value = await response.json() as T & {error?: string};
const response = await fetch(
`${config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310"}/__runtime/${operation}`,
{
method: "POST",
headers: { "content-type": "application/json", "x-camino-runtime-token": processToken },
body: JSON.stringify(input),
signal: AbortSignal.timeout(10_000),
},
);
const value = (await response.json()) as T & { error?: string };
if (!response.ok) throw new RuntimeAuthorityError(value.error ?? "Runtime authority request failed");
return value;
};
const attachSessions = (runtimeContext: RuntimeContext, request: RuntimeRequest & {context?: {grant: string; instanceId: string; workspaceEpoch: string; ownerConformanceId?: string}}) => {
const attachSessions = (
runtimeContext: RuntimeContext,
request: RuntimeRequest & {
context?: { grant: string; instanceId: string; workspaceEpoch: string; ownerConformanceId?: string };
},
) => {
if (!request.context?.grant || !request.context.ownerConformanceId) return;
runtimeContext.openSession = async () => {
const ownerId = request.context!.ownerConformanceId!;
const registration = { grant: request.context!.grant, objectId: request.objectId, ownerId,
sessionId: `session:${randomBytes(16).toString("hex")}`, token: randomBytes(32).toString("base64url") };
const register = () => runtimeControl<{sessionId: string; token: string}>("register-session", registration);
const registration = {
grant: request.context!.grant,
objectId: request.objectId,
ownerId,
sessionId: `session:${randomBytes(16).toString("hex")}`,
token: randomBytes(32).toString("base64url"),
};
const register = () => runtimeControl<{ sessionId: string; token: string }>("register-session", registration);
const registered = await register().catch((error) => {
// Retry a transport/lost-response failure with exactly the same identity.
// Admission/authority errors are definitive and must not be retried here.
@@ -445,210 +573,246 @@ export const createPackageRuntimeRoutes = (config: {
if (closed) throw new RuntimeAuthorityError("SESSION_CLOSED");
// Acquisition happens before user code. A fence failure can be retried
// by the caller without replaying a side-effecting callback.
const grant = await runtimeControl<{grant: string; epoch: string; instanceId: string; invocationId: string; bindingDigest: string}>("acquire-session", registered);
const grant = await runtimeControl<{
grant: string;
epoch: string;
instanceId: string;
invocationId: string;
bindingDigest: string;
}>("acquire-session", registered);
const execution = invocations.begin(grant.invocationId);
const sessionRequest = { ...request, context: { grant: grant.grant, instanceId: grant.instanceId, workspaceEpoch: grant.epoch } };
const sessionRequest = {
...request,
context: { grant: grant.grant, instanceId: grant.instanceId, workspaceEpoch: grant.epoch },
};
const clients = clientsFor(sessionRequest);
const context = createRuntimeContext(clients.camino, clients.orch, sessionRequest);
context.signal = execution.signal;
try { return await work(context); }
finally {
try {
return await work(context);
} finally {
execution.finish();
await runtimeControl("complete-invocation", { invocationId: grant.invocationId }).catch((error) => console.error("Session completion will be reconciled by the host", error));
await runtimeControl("complete-invocation", { invocationId: grant.invocationId }).catch((error) =>
console.error("Session completion will be reconciled by the host", error),
);
}
},
async close() { await runtimeControl("close-session", registered); closed = true; },
async close() {
await runtimeControl("close-session", registered);
closed = true;
},
};
};
};
return (router: ConnectRouter) => router.service(PackageRuntime, {
handshake: (request) => create(HandshakeResponseSchema, {
packageRevisionId: config.packageRevisionId,
runtimeProtocolVersion: "quixos-capabilities-v1",
exportIds: Object.keys(config.exports),
capabilities: ["invocation-completion-v1", "instance-authentication-v1", "epoch-grants-v1"],
instanceId: process.env.QUIXOS_RUNTIME_INSTANCE_ID ?? "",
authenticationProof: request.nonce && processToken ? createHmac("sha256", processToken)
.update(JSON.stringify([request.nonce, process.env.QUIXOS_RUNTIME_INSTANCE_ID ?? "", config.packageRevisionId]))
.digest("hex") : "",
}),
getInvocationStatus: (request, context) => {
authenticateInstance(context.requestHeader);
return invocations.status(request.invocationId);
},
cancelInvocation: (request, context) => {
authenticateInstance(context.requestHeader);
return invocations.cancel(request.invocationId);
},
invoke: async (request, context) => {
authenticateInstance(context.requestHeader);
const { camino, orch } = clientsFor(request);
const exportId = request.export?.exportId;
const handler = exportId ? config.exports[exportId] : undefined;
if (!handler) throw new ConnectError(`Unknown export ${exportId ?? ""}`, Code.NotFound);
const execution = invocations.begin(request.invocationId || (process.env.QUIXOS_RUNTIME_INSTANCE_ID ? "" : randomUUID()));
try {
const runtimeContext = createRuntimeContext(camino, orch, request);
attachSessions(runtimeContext, request);
runtimeContext.signal = AbortSignal.any([context.signal, execution.signal]);
const result = await evaluate(handler, runtimeContext);
execution.finish();
return create(InvokeResponseSchema, {
ok: true,
result: result.value,
dependencies: protoDependencies(result.dependencies),
});
} catch (error) {
execution.finish(true);
return create(InvokeResponseSchema, {
ok: false,
error: error instanceof Error ? error.message : String(error),
});
}
},
watch: async function* (request, context) {
authenticateInstance(context.requestHeader);
const { camino, orch } = clientsFor(request);
const exportId = request.export?.exportId;
const handler = exportId ? config.exports[exportId] : undefined;
if (!handler || !isDerived(handler)) {
throw new ConnectError(`Export ${exportId ?? ""} is not derived`, Code.FailedPrecondition);
}
const execution = invocations.begin(request.invocationId || (process.env.QUIXOS_RUNTIME_INSTANCE_ID ? "" : randomUUID()));
const signal = AbortSignal.any([context.signal, execution.signal]);
try {
const watchId = `watch:${randomUUID()}`;
const runtimeContext = createRuntimeContext(camino, orch, request);
attachSessions(runtimeContext, request);
runtimeContext.signal = signal;
type WatchOutcome = { key: string; done: boolean; error?: unknown };
type Subscription = {
dependency: RuntimeDependency;
controller: AbortController;
next: Promise<WatchOutcome>;
waitNext: () => Promise<WatchOutcome>;
};
const subscriptions = new Map<string, Subscription>();
const establishing = new Map<string, Promise<void>>();
let subscriptionEpoch = 0;
const ensureSubscription = async (dependency: RuntimeDependency) => {
const key = dependencyKey(dependency);
if (subscriptions.has(key)) return;
const pending = establishing.get(key);
if (pending) return await pending;
const establish = (async () => {
const controller = new AbortController();
const stream = camino.watchObject(
{ objectId: dependency.objectId, includeSnapshot: true, attachmentIds: request.context?.grant ? [dependency.attachmentId] : [] },
{ signal: controller.signal },
)[Symbol.asyncIterator]();
try {
// Camino subscribes before producing the snapshot, so once this
// resolves the following state/edge read cannot race the stream.
const snapshot = await stream.next();
if (snapshot.done) throw new Error(`Dependency stream ${key} ended during setup`);
const waitNext = () => stream.next().then(
(result) => ({ key, done: Boolean(result.done) }),
(error: unknown) => ({ key, done: true, error }),
);
const subscription: Subscription = {
dependency,
controller,
waitNext,
next: Promise.resolve({ key, done: false }),
};
subscription.next = waitNext();
subscriptions.set(key, subscription);
subscriptionEpoch += 1;
} catch (error) {
controller.abort();
throw error;
}
})();
establishing.set(key, establish);
return (router: ConnectRouter) =>
router.service(PackageRuntime, {
handshake: (request) =>
create(HandshakeResponseSchema, {
packageRevisionId: config.packageRevisionId,
runtimeProtocolVersion: "quixos-capabilities-v1",
exportIds: Object.keys(config.exports),
capabilities: ["invocation-completion-v1", "instance-authentication-v1", "epoch-grants-v1"],
instanceId: process.env.QUIXOS_RUNTIME_INSTANCE_ID ?? "",
authenticationProof:
request.nonce && processToken
? createHmac("sha256", processToken)
.update(
JSON.stringify([
request.nonce,
process.env.QUIXOS_RUNTIME_INSTANCE_ID ?? "",
config.packageRevisionId,
]),
)
.digest("hex")
: "",
}),
getInvocationStatus: (request, context) => {
authenticateInstance(context.requestHeader);
return invocations.status(request.invocationId);
},
cancelInvocation: (request, context) => {
authenticateInstance(context.requestHeader);
return invocations.cancel(request.invocationId);
},
invoke: async (request, context) => {
authenticateInstance(context.requestHeader);
const { camino, orch } = clientsFor(request);
const exportId = request.export?.exportId;
const handler = exportId ? config.exports[exportId] : undefined;
if (!handler) throw new ConnectError(`Unknown export ${exportId ?? ""}`, Code.NotFound);
const execution = invocations.begin(
request.invocationId || (process.env.QUIXOS_RUNTIME_INSTANCE_ID ? "" : randomUUID()),
);
try {
await establish;
} finally {
establishing.delete(key);
const runtimeContext = createRuntimeContext(camino, orch, request);
attachSessions(runtimeContext, request);
runtimeContext.signal = AbortSignal.any([context.signal, execution.signal]);
const result = await evaluate(handler, runtimeContext);
execution.finish();
return create(InvokeResponseSchema, {
ok: true,
result: result.value,
dependencies: protoDependencies(result.dependencies),
});
} catch (error) {
execution.finish(true);
return create(InvokeResponseSchema, {
ok: false,
error: error instanceof Error ? error.message : String(error),
});
}
};
const abortAll = () => {
for (const subscription of subscriptions.values()) {
subscription.controller.abort();
},
watch: async function* (request, context) {
authenticateInstance(context.requestHeader);
const { camino, orch } = clientsFor(request);
const exportId = request.export?.exportId;
const handler = exportId ? config.exports[exportId] : undefined;
if (!handler || !isDerived(handler)) {
throw new ConnectError(`Export ${exportId ?? ""} is not derived`, Code.FailedPrecondition);
}
};
signal.addEventListener("abort", abortAll, { once: true });
const execution = invocations.begin(
request.invocationId || (process.env.QUIXOS_RUNTIME_INSTANCE_ID ? "" : randomUUID()),
);
const signal = AbortSignal.any([context.signal, execution.signal]);
try {
const watchId = `watch:${randomUUID()}`;
const runtimeContext = createRuntimeContext(camino, orch, request);
attachSessions(runtimeContext, request);
runtimeContext.signal = signal;
type WatchOutcome = { key: string; done: boolean; error?: unknown };
type Subscription = {
dependency: RuntimeDependency;
controller: AbortController;
next: Promise<WatchOutcome>;
waitNext: () => Promise<WatchOutcome>;
};
const subscriptions = new Map<string, Subscription>();
const establishing = new Map<string, Promise<void>>();
let subscriptionEpoch = 0;
const evaluateWithStableSubscriptions = async () => {
// A direct state/edge port records its dependency before reading it,
// but a nested interface invocation can only report its transitive
// dependencies after that invocation returns. Once a new dependency
// stream is established, evaluate again so every read contributing to
// the emitted value happened after its stream became live.
for (let pass = 0; pass < 32; pass += 1) {
const before = subscriptionEpoch;
const result = await evaluate(handler, runtimeContext, ensureSubscription);
if (subscriptionEpoch === before) return result;
}
throw new Error("Derived dependency discovery did not stabilize after 32 passes");
};
try {
let current = await evaluateWithStableSubscriptions();
yield create(WatchEventSchema, {
watchId,
value: current.value,
dependencies: protoDependencies(current.dependencies),
initial: true,
});
const abort = new Promise<"abort">((resolve) => {
if (signal.aborted) resolve("abort");
else signal.addEventListener("abort", () => resolve("abort"), { once: true });
});
while (!signal.aborted) {
if (subscriptions.size === 0) {
await abort;
break;
}
const outcome = await Promise.race([
...[...subscriptions.values()].map((entry) => entry.next),
abort,
]);
if (outcome === "abort") break;
const subscription = subscriptions.get(outcome.key);
if (!subscription) continue;
if (outcome.error) throw outcome.error;
if (outcome.done) throw new Error(`Dependency stream ${outcome.key} ended unexpectedly`);
subscription.next = subscription.waitNext();
const updated = await evaluateWithStableSubscriptions();
const active = new Set(updated.dependencies.map(dependencyKey));
for (const [key, entry] of subscriptions) {
if (!active.has(key)) {
entry.controller.abort();
subscriptions.delete(key);
const ensureSubscription = async (dependency: RuntimeDependency) => {
const key = dependencyKey(dependency);
if (subscriptions.has(key)) return;
const pending = establishing.get(key);
if (pending) return await pending;
const establish = (async () => {
const controller = new AbortController();
const stream = camino
.watchObject(
{
objectId: dependency.objectId,
includeSnapshot: true,
attachmentIds: request.context?.grant ? [dependency.attachmentId] : [],
},
{ signal: controller.signal },
)
[Symbol.asyncIterator]();
try {
// Camino subscribes before producing the snapshot, so once this
// resolves the following state/edge read cannot race the stream.
const snapshot = await stream.next();
if (snapshot.done) throw new Error(`Dependency stream ${key} ended during setup`);
const waitNext = () =>
stream.next().then(
(result) => ({ key, done: Boolean(result.done) }),
(error: unknown) => ({ key, done: true, error }),
);
const subscription: Subscription = {
dependency,
controller,
waitNext,
next: Promise.resolve({ key, done: false }),
};
subscription.next = waitNext();
subscriptions.set(key, subscription);
subscriptionEpoch += 1;
} catch (error) {
controller.abort();
throw error;
}
})();
establishing.set(key, establish);
try {
await establish;
} finally {
establishing.delete(key);
}
}
if (!equals(ValueSchema, current.value, updated.value)) {
};
const abortAll = () => {
for (const subscription of subscriptions.values()) {
subscription.controller.abort();
}
};
signal.addEventListener("abort", abortAll, { once: true });
const evaluateWithStableSubscriptions = async () => {
// A direct state/edge port records its dependency before reading it,
// but a nested interface invocation can only report its transitive
// dependencies after that invocation returns. Once a new dependency
// stream is established, evaluate again so every read contributing to
// the emitted value happened after its stream became live.
for (let pass = 0; pass < 32; pass += 1) {
const before = subscriptionEpoch;
const result = await evaluate(handler, runtimeContext, ensureSubscription);
if (subscriptionEpoch === before) return result;
}
throw new Error("Derived dependency discovery did not stabilize after 32 passes");
};
try {
let current = await evaluateWithStableSubscriptions();
yield create(WatchEventSchema, {
watchId,
value: updated.value,
dependencies: protoDependencies(updated.dependencies),
value: current.value,
dependencies: protoDependencies(current.dependencies),
initial: true,
});
const abort = new Promise<"abort">((resolve) => {
if (signal.aborted) resolve("abort");
else signal.addEventListener("abort", () => resolve("abort"), { once: true });
});
while (!signal.aborted) {
if (subscriptions.size === 0) {
await abort;
break;
}
const outcome = await Promise.race([...[...subscriptions.values()].map((entry) => entry.next), abort]);
if (outcome === "abort") break;
const subscription = subscriptions.get(outcome.key);
if (!subscription) continue;
if (outcome.error) throw outcome.error;
if (outcome.done) throw new Error(`Dependency stream ${outcome.key} ended unexpectedly`);
subscription.next = subscription.waitNext();
const updated = await evaluateWithStableSubscriptions();
const active = new Set(updated.dependencies.map(dependencyKey));
for (const [key, entry] of subscriptions) {
if (!active.has(key)) {
entry.controller.abort();
subscriptions.delete(key);
}
}
if (!equals(ValueSchema, current.value, updated.value)) {
yield create(WatchEventSchema, {
watchId,
value: updated.value,
dependencies: protoDependencies(updated.dependencies),
});
}
current = updated;
}
} finally {
signal.removeEventListener("abort", abortAll);
abortAll();
}
current = updated;
} finally {
execution.finish();
}
} finally {
signal.removeEventListener("abort", abortAll);
abortAll();
}
} finally { execution.finish(); }
},
});
},
});
};
export const servePackageRuntime = (config: {
src/invocations.ts
+11 -3
View File
@@ -3,15 +3,23 @@ export const createInvocationRegistry = () => {
const entries = new Map<string, { state: string; controller: AbortController }>();
return {
begin(id: string) {
if (!id || entries.has(id)) throw new Error("Invocation ID is missing or already used; invocations are never replayed implicitly");
if (!id || entries.has(id))
throw new Error("Invocation ID is missing or already used; invocations are never replayed implicitly");
// Completed identities remain until process retirement. A bounded process
// may reject new work; it must not evict and accidentally replay a call.
if (entries.size >= 100_000) throw new Error("Invocation registry full; explicit runtime retirement required");
const entry = { state: "running", controller: new AbortController() };
entries.set(id, entry);
return { signal: entry.controller.signal, finish(failed = false) { entry.state = failed ? "failed" : "completed"; } };
return {
signal: entry.controller.signal,
finish(failed = false) {
entry.state = failed ? "failed" : "completed";
},
};
},
status(id: string) {
return { invocationId: id, state: entries.get(id)?.state ?? "unknown" };
},
status(id: string) { return { invocationId: id, state: entries.get(id)?.state ?? "unknown" }; },
cancel(id: string) {
const entry = entries.get(id);
if (entry && ["running", "cancellation-requested"].includes(entry.state)) {
src/migration.ts
+94 -35
View File
@@ -1,18 +1,42 @@
import {createHash} from "node:crypto";
import { createHash } from "node:crypto";
export type MigrationInput = {
schemaVersion: 1; executionId: string; exportId: string;
ports: {name: string; binding: string; view: "old" | "new"; access: ("read" | "write" | "create" | "edge")[]; atomId?: string;
attachedAtomId?: string; defaultValue?: unknown;
states?: {objectId: string; value: unknown}[]; edges?: MigrationEdge[]}[];
schemaVersion: 1;
executionId: string;
exportId: string;
ports: {
name: string;
binding: string;
view: "old" | "new";
access: ("read" | "write" | "create" | "edge")[];
atomId?: string;
attachedAtomId?: string;
defaultValue?: unknown;
states?: { objectId: string; value: unknown }[];
edges?: MigrationEdge[];
}[];
};
export type MigrationEdge = {
id: string;
edgeTypeId: string;
firstObjectId: string;
secondObjectId: string;
firstProjectionId: string;
secondProjectionId: string;
firstOrdinal?: number;
secondOrdinal?: number;
firstKeyJson?: string;
secondKeyJson?: string;
};
export type MigrationOutput = {
schemaVersion: 1;
executionId: string;
writes: { port: string; objectId: string; value: unknown }[];
creates: { port: string; logicalKey: string; objectId: string }[];
edgeReplacements: { port: string; edges: MigrationEdge[] }[];
};
export type MigrationEdge = {id: string; edgeTypeId: string; firstObjectId: string; secondObjectId: string; firstProjectionId: string; secondProjectionId: string; firstOrdinal?: number; secondOrdinal?: number; firstKeyJson?: string; secondKeyJson?: string};
export type MigrationOutput = {schemaVersion: 1; executionId: string;
writes: {port: string; objectId: string; value: unknown}[];
creates: {port: string; logicalKey: string; objectId: string}[];
edgeReplacements: {port: string; edges: MigrationEdge[]}[]};
export type MigrationContext = {
enumerate(port: string): {objectId: string; value: unknown}[];
enumerate(port: string): { objectId: string; value: unknown }[];
read(port: string, objectId: string): unknown;
write(port: string, objectId: string, value: unknown): void;
create(port: string, logicalKey: string): string;
@@ -20,65 +44,100 @@ export type MigrationContext = {
replaceEdges(port: string, edges: MigrationEdge[]): void;
};
export const migrationObjectId = (executionId: string, port: string, logicalKey: string) =>
`obj:migration:${createHash("sha256").update(JSON.stringify([executionId, port, logicalKey])).digest("hex")}`;
`obj:migration:${createHash("sha256")
.update(JSON.stringify([executionId, port, logicalKey]))
.digest("hex")}`;
/** No ordinary RuntimeContext or network/database clients are supplied here.
* Process isolation belongs to the host, not this convenience API. */
export const createMigrationContext = (input: MigrationInput) => {
if (input.schemaVersion !== 1 || !input.executionId || new Set(input.ports.map((entry) => entry.name)).size !== input.ports.length) throw new Error("Invalid migration input");
const output: MigrationOutput = {schemaVersion: 1, executionId: input.executionId, writes: [], creates: [], edgeReplacements: []};
if (
input.schemaVersion !== 1 ||
!input.executionId ||
new Set(input.ports.map((entry) => entry.name)).size !== input.ports.length
)
throw new Error("Invalid migration input");
const output: MigrationOutput = {
schemaVersion: 1,
executionId: input.executionId,
writes: [],
creates: [],
edgeReplacements: [],
};
const port = (name: string, access: string) => {
const selected = input.ports.find((entry) => entry.name === name);
if (!selected?.access.includes(access as "read") || (selected.view === "old" && access !== "read")) throw new Error(`Migration port ${name} does not grant ${access}`);
if (!selected?.access.includes(access as "read") || (selected.view === "old" && access !== "read"))
throw new Error(`Migration port ${name} does not grant ${access}`);
return selected;
};
const context: MigrationContext = {
enumerate(name) {
const selected = port(name, "read"), states = structuredClone(selected.states ?? []);
if (selected.view === "new" && Object.hasOwn(selected, "defaultValue")) for (const helper of output.creates) {
if (input.ports.find((entry) => entry.name === helper.port)?.atomId === selected.attachedAtomId && !states.some((entry) => entry.objectId === helper.objectId)) states.push({objectId: helper.objectId, value: structuredClone(selected.defaultValue)});
}
if (selected.view === "new") for (const write of output.writes) {
if (input.ports.find((entry) => entry.name === write.port)?.binding !== selected.binding) continue;
const existing = states.findIndex((entry) => entry.objectId === write.objectId), entry = {objectId: write.objectId, value: structuredClone(write.value)};
if (existing < 0) states.push(entry); else states[existing] = entry;
}
return states.sort((a, b) => a.objectId < b.objectId ? -1 : a.objectId > b.objectId ? 1 : 0);
const selected = port(name, "read"),
states = structuredClone(selected.states ?? []);
if (selected.view === "new" && Object.hasOwn(selected, "defaultValue"))
for (const helper of output.creates) {
if (
input.ports.find((entry) => entry.name === helper.port)?.atomId === selected.attachedAtomId &&
!states.some((entry) => entry.objectId === helper.objectId)
)
states.push({ objectId: helper.objectId, value: structuredClone(selected.defaultValue) });
}
if (selected.view === "new")
for (const write of output.writes) {
if (input.ports.find((entry) => entry.name === write.port)?.binding !== selected.binding) continue;
const existing = states.findIndex((entry) => entry.objectId === write.objectId),
entry = { objectId: write.objectId, value: structuredClone(write.value) };
if (existing < 0) states.push(entry);
else states[existing] = entry;
}
return states.sort((a, b) => (a.objectId < b.objectId ? -1 : a.objectId > b.objectId ? 1 : 0));
},
read(name, objectId) {
return context.enumerate(name).find((entry) => entry.objectId === objectId)?.value;
},
read(name, objectId) {return context.enumerate(name).find((entry) => entry.objectId === objectId)?.value;},
write(name, objectId, value) {
port(name, "write");
const previous = output.writes.findIndex((entry) => entry.port === name && entry.objectId === objectId);
const entry = {port: name, objectId, value: structuredClone(value)};
if (previous < 0) output.writes.push(entry); else output.writes[previous] = entry;
const entry = { port: name, objectId, value: structuredClone(value) };
if (previous < 0) output.writes.push(entry);
else output.writes[previous] = entry;
},
create(name, logicalKey) {
port(name, "create");
if (!logicalKey || logicalKey.length > 1024) throw new Error("Migration creation requires a bounded stable logical key");
if (!logicalKey || logicalKey.length > 1024)
throw new Error("Migration creation requires a bounded stable logical key");
const objectId = migrationObjectId(input.executionId, name, logicalKey);
if (!output.creates.some((entry) => entry.objectId === objectId)) output.creates.push({port: name, logicalKey, objectId});
if (!output.creates.some((entry) => entry.objectId === objectId))
output.creates.push({ port: name, logicalKey, objectId });
return objectId;
},
edges(name) {
const selected = port(name, "read");
const replacement = selected.view === "new" ? output.edgeReplacements.find((entry) => input.ports.find((candidate) => candidate.name === entry.port)?.binding === selected.binding) : undefined;
const replacement =
selected.view === "new"
? output.edgeReplacements.find(
(entry) => input.ports.find((candidate) => candidate.name === entry.port)?.binding === selected.binding,
)
: undefined;
return structuredClone(replacement?.edges ?? selected.edges ?? []);
},
replaceEdges(name, edges) {
port(name, "edge");
const previous = output.edgeReplacements.findIndex((entry) => entry.port === name);
const entry = {port: name, edges: structuredClone(edges)};
if (previous < 0) output.edgeReplacements.push(entry); else output.edgeReplacements[previous] = entry;
const entry = { port: name, edges: structuredClone(edges) };
if (previous < 0) output.edgeReplacements.push(entry);
else output.edgeReplacements[previous] = entry;
},
};
return {context, result: () => structuredClone(output)};
return { context, result: () => structuredClone(output) };
};
/** Entrypoint for an immutable package's dedicated bin/migrate executable.
* stdout is protocol-only; send diagnostics to stderr. The host independently
* validates every write, helper identity, contract, and completion receipt. */
export const serveMigration = async (exports: Record<string, (context: MigrationContext) => void | Promise<void>>) => {
const chunks: Buffer[] = []; let bytes = 0;
const chunks: Buffer[] = [];
let bytes = 0;
for await (const chunk of process.stdin) {
bytes += chunk.length;
if (bytes > 16 * 1024 * 1024) throw new Error("Migration input exceeds 16 MiB");
src/references.ts
+21 -7
View File
@@ -3,15 +3,26 @@
const identities = new WeakMap<object, string>();
declare const referenceBrand: unique symbol;
export interface QxObjectRef<Identity extends string = string> {
readonly [referenceBrand]: {readonly [K in Identity]: true};
readonly [referenceBrand]: { readonly [K in Identity]: true };
equals(other: QxObjectRef<string>): boolean;
}
class Reference {
constructor(id: string) { identities.set(this, id); Object.freeze(this); }
equals(other: unknown) { return isObjectReference(other) && identities.get(this) === identities.get(other); }
toJSON(): never { throw new Error("Object references cannot be serialized into ordinary data"); }
toString(): never { throw new Error("Object references cannot be coerced to strings"); }
[Symbol.toPrimitive](): never { throw new Error("Object references cannot be coerced to scalar values"); }
constructor(id: string) {
identities.set(this, id);
Object.freeze(this);
}
equals(other: unknown) {
return isObjectReference(other) && identities.get(this) === identities.get(other);
}
toJSON(): never {
throw new Error("Object references cannot be serialized into ordinary data");
}
toString(): never {
throw new Error("Object references cannot be coerced to strings");
}
[Symbol.toPrimitive](): never {
throw new Error("Object references cannot be coerced to scalar values");
}
}
export const isObjectReference = (value: unknown): value is QxObjectRef =>
typeof value === "object" && value !== null && identities.has(value);
@@ -27,7 +38,10 @@ export const referenceToWire = (value: unknown): string => {
};
export const assertReferenceFree = (value: unknown, seen = new Set<object>()): void => {
if (!value || typeof value !== "object") return;
if (isObjectReference(value)) throw new Error("Managed references belong in declared RPC references or graph relationships, not ordinary state/messages");
if (isObjectReference(value))
throw new Error(
"Managed references belong in declared RPC references or graph relationships, not ordinary state/messages",
);
if (seen.has(value)) throw new Error("Cyclic ordinary data");
seen.add(value);
if (!(value instanceof Uint8Array)) for (const child of Object.values(value)) assertReferenceFree(child, seen);
src/relationships.ts
+29 -12
View File
@@ -1,7 +1,10 @@
import type {QxObjectRef} from "./references.js";
import type {RelationshipCollection, RelationshipEntry} from "./index.js";
import type { QxObjectRef } from "./references.js";
import type { RelationshipCollection, RelationshipEntry } from "./index.js";
type Key = string | boolean | bigint;
type Port<T extends QxObjectRef> = {collection(): Promise<RelationshipCollection<T>>; replace(entries: RelationshipEntry<T>[], expectedRevision: bigint): Promise<RelationshipCollection<T>>};
type Port<T extends QxObjectRef> = {
collection(): Promise<RelationshipCollection<T>>;
replace(entries: RelationshipEntry<T>[], expectedRevision: bigint): Promise<RelationshipCollection<T>>;
};
const checked = async <T extends QxObjectRef>(port: Port<T>, revision: bigint) => {
const snapshot = await port.collection();
if (snapshot.revision !== revision) throw new Error("STALE_COLLECTION_REVISION");
@@ -10,31 +13,39 @@ const checked = async <T extends QxObjectRef>(port: Port<T>, revision: bigint) =
/** Helpers never retry a failed CAS or silently overwrite concurrent edits. */
export const relationshipMap = <T extends QxObjectRef, K extends Key = Key>(port: Port<T>) => ({
read: () => port.collection(),
async get(key: K) {const snapshot = await port.collection(); return {revision: snapshot.revision, value: snapshot.entries.find((entry) => entry.key === key)?.target};},
async get(key: K) {
const snapshot = await port.collection();
return { revision: snapshot.revision, value: snapshot.entries.find((entry) => entry.key === key)?.target };
},
async set(key: K, target: T, expectedRevision: bigint) {
const snapshot = await checked(port, expectedRevision);
const entries = snapshot.entries.filter((entry) => entry.key !== key);
const existing = snapshot.entries.find((entry) => entry.key === key && entry.target.equals(target));
entries.push(existing ?? {key, target});
entries.push(existing ?? { key, target });
return port.replace(entries, expectedRevision);
},
async delete(key: K, expectedRevision: bigint) {
const snapshot = await checked(port, expectedRevision);
return port.replace(snapshot.entries.filter((entry) => entry.key !== key), expectedRevision);
return port.replace(
snapshot.entries.filter((entry) => entry.key !== key),
expectedRevision,
);
},
});
export const relationshipList = <T extends QxObjectRef>(port: Port<T>) => ({
read: () => port.collection(),
async insert(index: number, target: T, expectedRevision: bigint) {
const snapshot = await checked(port, expectedRevision);
if (!Number.isSafeInteger(index) || index < 0 || index > snapshot.entries.length) throw new Error("List index out of bounds");
snapshot.entries.splice(index, 0, {target});
if (!Number.isSafeInteger(index) || index < 0 || index > snapshot.entries.length)
throw new Error("List index out of bounds");
snapshot.entries.splice(index, 0, { target });
return port.replace(snapshot.entries, expectedRevision);
},
async move(edgeId: string, index: number, expectedRevision: bigint) {
const snapshot = await checked(port, expectedRevision);
const prior = snapshot.entries.findIndex((entry) => entry.edgeId === edgeId);
if (prior < 0 || !Number.isSafeInteger(index) || index < 0 || index >= snapshot.entries.length) throw new Error("Unknown list entry or invalid index");
if (prior < 0 || !Number.isSafeInteger(index) || index < 0 || index >= snapshot.entries.length)
throw new Error("Unknown list entry or invalid index");
const [entry] = snapshot.entries.splice(prior, 1);
snapshot.entries.splice(index, 0, entry);
return port.replace(snapshot.entries, expectedRevision);
@@ -42,7 +53,10 @@ export const relationshipList = <T extends QxObjectRef>(port: Port<T>) => ({
async delete(edgeId: string, expectedRevision: bigint) {
const snapshot = await checked(port, expectedRevision);
if (!snapshot.entries.some((entry) => entry.edgeId === edgeId)) throw new Error("Unknown list entry");
return port.replace(snapshot.entries.filter((entry) => entry.edgeId !== edgeId), expectedRevision);
return port.replace(
snapshot.entries.filter((entry) => entry.edgeId !== edgeId),
expectedRevision,
);
},
});
export const relationshipSet = <T extends QxObjectRef>(port: Port<T>) => ({
@@ -50,10 +64,13 @@ export const relationshipSet = <T extends QxObjectRef>(port: Port<T>) => ({
async add(target: T, expectedRevision: bigint) {
const snapshot = await checked(port, expectedRevision);
if (snapshot.entries.some((entry) => entry.target.equals(target))) return snapshot;
return port.replace([...snapshot.entries, {target}], expectedRevision);
return port.replace([...snapshot.entries, { target }], expectedRevision);
},
async delete(target: T, expectedRevision: bigint) {
const snapshot = await checked(port, expectedRevision);
return port.replace(snapshot.entries.filter((entry) => !entry.target.equals(target)), expectedRevision);
return port.replace(
snapshot.entries.filter((entry) => !entry.target.equals(target)),
expectedRevision,
);
},
});
test/bindings.test.mjs
+151 -64
View File
@@ -1,108 +1,192 @@
import assert from "node:assert/strict";
import test from "node:test";
import { referenceFromWire } from "../dist/references.js";
import { bindQxHandler, decodeQxValue, encodeQxValue, jsToProtoValue, liveValue, protoValueToJs, opaqueReactPropsBinding } from "../dist/index.js";
import {
bindQxHandler,
decodeQxValue,
encodeQxValue,
jsToProtoValue,
liveValue,
protoValueToJs,
opaqueReactPropsBinding,
} from "../dist/index.js";
const scalar = (name) => ({ kind: "scalar", name });
const unit = { kind: "builtin", name: "unit" };
test("opaque React props retain managed references while ordinary messages remain closed", () => {
const descriptorId = "org.quixos.web-studio.ReactProps";
const reference = referenceFromWire("obj:task");
const type = {kind: "message", descriptorId};
const messages = {[descriptorId]: opaqueReactPropsBinding};
const decoded = decodeQxValue(type, encodeQxValue(type, {subject: reference, title: "Task"}, messages), messages);
const type = { kind: "message", descriptorId };
const messages = { [descriptorId]: opaqueReactPropsBinding };
const decoded = decodeQxValue(type, encodeQxValue(type, { subject: reference, title: "Task" }, messages), messages);
assert.ok(decoded.subject.equals(reference));
assert.equal(decoded.title, "Task");
assert.throws(() => opaqueReactPropsBinding.encode(null), /object/);
assert.throws(() => opaqueReactPropsBinding.encode([]), /object/);
assert.throws(() => encodeQxValue({kind: "message", descriptorId: "Ordinary"}, {subject: reference},
{Ordinary: opaqueReactPropsBinding}), /Managed references/);
assert.throws(
() =>
encodeQxValue(
{ kind: "message", descriptorId: "Ordinary" },
{ subject: reference },
{ Ordinary: opaqueReactPropsBinding },
),
/Managed references/,
);
});
test("declared record inputs preserve opaque references without opening message codecs", async () => {
const target = referenceFromWire("obj:board");
const type = {kind: "record", fields: {
target: {kind: "object-ref", expectation: {kind: "atom", atomId: "board"}},
x: scalar("double"), note: {kind: "optional", value: scalar("string")},
}};
const encoded = encodeQxValue(type, {target, x: 12}, {});
const type = {
kind: "record",
fields: {
target: { kind: "object-ref", expectation: { kind: "atom", atomId: "board" } },
x: scalar("double"),
note: { kind: "optional", value: scalar("string") },
},
};
const encoded = encodeQxValue(type, { target, x: 12 }, {});
const decoded = decodeQxValue(type, encoded, {});
assert.ok(decoded.target.equals(target));
assert.equal(decoded.x, 12);
assert.equal(decoded.note, null);
assert.throws(() => JSON.stringify(decoded), /cannot be serialized/);
assert.throws(() => encodeQxValue(type, {target, x: 12, hidden: target}, {}), /Unexpected/);
assert.throws(() => encodeQxValue(type, {x: 12}, {}), /Missing/);
assert.throws(() => encodeQxValue(type, {target: "obj:board", x: 12}, {}), /opaque object reference/);
const handler = bindQxHandler({inputType: type, outputType: unit, ports: {}}, async (context) => {
assert.ok(context.input.target.equals(target));
assert.equal(context.input.x, 12);
}, {});
await handler({objectId: target, inputProto: encoded.kind.value.fields});
assert.throws(() => encodeQxValue(type, { target, x: 12, hidden: target }, {}), /Unexpected/);
assert.throws(() => encodeQxValue(type, { x: 12 }, {}), /Missing/);
assert.throws(() => encodeQxValue(type, { target: "obj:board", x: 12 }, {}), /opaque object reference/);
const handler = bindQxHandler(
{ inputType: type, outputType: unit, ports: {} },
async (context) => {
assert.ok(context.input.target.equals(target));
assert.equal(context.input.x, 12);
},
{},
);
await handler({ objectId: target, inputProto: encoded.kind.value.fields });
});
test("typed sessions rebind ports and cancellation to each acquired invocation", async () => {
const first = new AbortController(), second = new AbortController();
const first = new AbortController(),
second = new AbortController();
let closed = false;
const context = (value, signal) => ({objectId: referenceFromWire("obj:owner"), inputProto: {}, signal,
state: () => ({live: async () => liveValue(jsToProtoValue(value))})});
const raw = {...context(1n, first.signal), openSession: async () => ({id: "session:test",
run: (work) => work(context(2n, second.signal)), close: async () => {closed = true;}})};
const handler = bindQxHandler({inputType: unit, outputType: unit, ports: {counter: {kind: "state", id: "counter", valueType: scalar("int64"), primitives: ["read"]}}}, async (bound) => {
assert.equal(bound.signal, first.signal);
const session = await bound.openSession();
await session.run(async (next) => {
assert.equal(next.signal, second.signal);
assert.equal(await next.ports.counter.get(), 2n);
assert.equal(next.openSession, undefined);
});
await session.close();
}, {});
const context = (value, signal) => ({
objectId: referenceFromWire("obj:owner"),
inputProto: {},
signal,
state: () => ({ live: async () => liveValue(jsToProtoValue(value)) }),
});
const raw = {
...context(1n, first.signal),
openSession: async () => ({
id: "session:test",
run: (work) => work(context(2n, second.signal)),
close: async () => {
closed = true;
},
}),
};
const handler = bindQxHandler(
{
inputType: unit,
outputType: unit,
ports: { counter: { kind: "state", id: "counter", valueType: scalar("int64"), primitives: ["read"] } },
},
async (bound) => {
assert.equal(bound.signal, first.signal);
const session = await bound.openSession();
await session.run(async (next) => {
assert.equal(next.signal, second.signal);
assert.equal(await next.ports.counter.get(), 2n);
assert.equal(next.openSession, undefined);
});
await session.close();
},
{},
);
await handler(raw);
assert.equal(closed, true);
});
test("binding codecs round trip nested bytes, 64-bit integers, nulls, and references", () => {
const values = [[scalar("int64"), -(2n ** 63n)], [scalar("uint64"), 2n ** 64n - 1n],
const values = [
[scalar("int64"), -(2n ** 63n)],
[scalar("uint64"), 2n ** 64n - 1n],
[scalar("bytes"), new Uint8Array([0, 255])],
[{ kind: "list", value: { kind: "optional", value: scalar("int64") } }, [null, 2n ** 60n]],
[{ kind: "object-ref", expectation: { kind: "atom", atomId: "thing" } }, referenceFromWire("obj:thing")]];
[{ kind: "object-ref", expectation: { kind: "atom", atomId: "thing" } }, referenceFromWire("obj:thing")],
];
for (const [type, value] of values) assert.deepEqual(decodeQxValue(type, encodeQxValue(type, value, {}), {}), value);
});
test("opaque references pass declared RPC boundaries but cannot enter ordinary data", () => {
const reference = referenceFromWire("obj:thing");
const type = {kind: "object-ref", expectation: {kind: "atom", atomId: "thing"}};
const type = { kind: "object-ref", expectation: { kind: "atom", atomId: "thing" } };
const roundtrip = decodeQxValue(type, encodeQxValue(type, reference, {}), {});
assert.equal(reference.equals(roundtrip), true);
assert.equal(reference.equals(referenceFromWire("obj:other")), false);
assert.throws(() => JSON.stringify({nested: [reference]}), /cannot be serialized/);
assert.throws(() => JSON.stringify({ nested: [reference] }), /cannot be serialized/);
assert.throws(() => String(reference), /cannot be coerced/);
assert.throws(() => encodeQxValue(type, "obj:thing", {}), /opaque/);
assert.throws(() => encodeQxValue(scalar("string"), reference, {}), /Managed references/);
const message = {kind: "message", descriptorId: "Payload"};
assert.throws(() => encodeQxValue(message, {nested: reference}, {Payload: {encode: jsToProtoValue}}), /Managed references/);
assert.throws(() => encodeQxValue(message, {}, {Payload: {encode: () => jsToProtoValue(reference)}}), /Managed references/);
assert.throws(() => decodeQxValue(message, jsToProtoValue(reference), {Payload: {decode: () => ({})}}), /Managed references/);
const message = { kind: "message", descriptorId: "Payload" };
assert.throws(
() => encodeQxValue(message, { nested: reference }, { Payload: { encode: jsToProtoValue } }),
/Managed references/,
);
assert.throws(
() => encodeQxValue(message, {}, { Payload: { encode: () => jsToProtoValue(reference) } }),
/Managed references/,
);
assert.throws(
() => decodeQxValue(message, jsToProtoValue(reference), { Payload: { decode: () => ({}) } }),
/Managed references/,
);
});
test("typed state and interface ports preserve declared values and exact operation IDs", async () => {
const spec = { inputType: scalar("int64"), outputType: scalar("bytes"), ports: {
data: { kind: "state", id: "state-id", valueType: scalar("int64"), primitives: ["read", "write"] },
reader: { kind: "interface", id: "interface-id", operations: { "payload.get": { id: "get-id", inputType: unit, outputType: scalar("bytes") } } },
} };
const spec = {
inputType: scalar("int64"),
outputType: scalar("bytes"),
ports: {
data: { kind: "state", id: "state-id", valueType: scalar("int64"), primitives: ["read", "write"] },
reader: {
kind: "interface",
id: "interface-id",
operations: { "payload.get": { id: "get-id", inputType: unit, outputType: scalar("bytes") } },
},
},
};
let written;
const handler = bindQxHandler(spec, async ({ input, ports }) => {
assert.equal(input, 2n ** 60n);
assert.equal(await ports.data.get(), 9n);
assert.deepEqual((await ports.data.live()).$quixosValue, jsToProtoValue(9n));
assert.ok(ports.reader.objectId.equals(referenceFromWire("obj:reader")));
assert.deepEqual((await ports.reader.live["payload.get"]()).$quixosValue, jsToProtoValue(new Uint8Array([7])));
await ports.data.set(input);
return ports.reader["payload.get"]();
}, {});
const result = await handler({ inputProto: { value: jsToProtoValue(2n ** 60n) }, objectId: "obj",
state: (id) => { assert.equal(id, "state-id"); return {
live: async () => liveValue(jsToProtoValue(9n)), set: async (value) => { written = value; },
}; },
interface: (id) => { assert.equal(id, "interface-id"); return { objectId: referenceFromWire("obj:reader"), live: async (operation, input) => {
assert.equal(operation, "get-id"); assert.deepEqual(input, {}); return liveValue(jsToProtoValue(new Uint8Array([7])));
} }; },
const handler = bindQxHandler(
spec,
async ({ input, ports }) => {
assert.equal(input, 2n ** 60n);
assert.equal(await ports.data.get(), 9n);
assert.deepEqual((await ports.data.live()).$quixosValue, jsToProtoValue(9n));
assert.ok(ports.reader.objectId.equals(referenceFromWire("obj:reader")));
assert.deepEqual((await ports.reader.live["payload.get"]()).$quixosValue, jsToProtoValue(new Uint8Array([7])));
await ports.data.set(input);
return ports.reader["payload.get"]();
},
{},
);
const result = await handler({
inputProto: { value: jsToProtoValue(2n ** 60n) },
objectId: "obj",
state: (id) => {
assert.equal(id, "state-id");
return {
live: async () => liveValue(jsToProtoValue(9n)),
set: async (value) => {
written = value;
},
};
},
interface: (id) => {
assert.equal(id, "interface-id");
return {
objectId: referenceFromWire("obj:reader"),
live: async (operation, input) => {
assert.equal(operation, "get-id");
assert.deepEqual(input, {});
return liveValue(jsToProtoValue(new Uint8Array([7])));
},
};
},
});
assert.equal(written.$quixosValue.kind.value, String(2n ** 60n));
assert.deepEqual(result.$quixosValue.kind.value, new Uint8Array([7]));
@@ -110,8 +194,11 @@ test("typed state and interface ports preserve declared values and exact operati
test("external message bindings and derived event types are used at the boundary", async () => {
const message = { kind: "message", descriptorId: "Payload" };
const messages = { Payload: { encode: jsToProtoValue, decode: protoValueToJs } };
const handler = bindQxHandler({ inputType: message, outputType: { kind: "builtin", name: "watch-handle" }, eventType: message, ports: {} },
{ kind: "derived", get: ({ input }) => ({ value: input.title }) }, messages);
const handler = bindQxHandler(
{ inputType: message, outputType: { kind: "builtin", name: "watch-handle" }, eventType: message, ports: {} },
{ kind: "derived", get: ({ input }) => ({ value: input.title }) },
messages,
);
assert.equal(handler.kind, "derived");
const result = await handler.get({ objectId: "obj", inputProto: { title: jsToProtoValue("hello") } });
assert.deepEqual(protoValueToJs(result.$quixosValue), { value: "hello" });
test/credential-file.test.mjs
+38 -21
View File
@@ -1,30 +1,47 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import crypto from 'node:crypto';
import {spawnSync} from 'node:child_process';
import test from "node:test";
import assert from "node:assert/strict";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import crypto from "node:crypto";
import { spawnSync } from "node:child_process";
test('supervised SDK reads a credential file and proves instance identity without an environment token', async t => {
const directory = await fs.mkdtemp(path.join(os.tmpdir(), 'qx-sdk-credential-'));
t.after(() => fs.rm(directory, {recursive:true, force:true}));
const token = crypto.randomBytes(32).toString('hex');
const filename = path.join(directory, 'process-token');
await fs.writeFile(filename, token, {mode:0o600});
const env = {...process.env, CAMINO_RUNTIME_AUTH_TOKEN_FILE: filename,
CAMINO_RUNTIME_AUTH_REQUIRED:'1', QUIXOS_RUNTIME_INSTANCE_ID:'instance:test'};
test("supervised SDK reads a credential file and proves instance identity without an environment token", async (t) => {
const directory = await fs.mkdtemp(path.join(os.tmpdir(), "qx-sdk-credential-"));
t.after(() => fs.rm(directory, { recursive: true, force: true }));
const token = crypto.randomBytes(32).toString("hex");
const filename = path.join(directory, "process-token");
await fs.writeFile(filename, token, { mode: 0o600 });
const env = {
...process.env,
CAMINO_RUNTIME_AUTH_TOKEN_FILE: filename,
CAMINO_RUNTIME_AUTH_REQUIRED: "1",
QUIXOS_RUNTIME_INSTANCE_ID: "instance:test",
};
delete env.CAMINO_RUNTIME_AUTH_TOKEN;
const result = spawnSync(process.execPath, ['--input-type=module', '-e', `
import {createPackageRuntimeRoutes} from ${JSON.stringify(new URL('../dist/index.js', import.meta.url).href)};
const result = spawnSync(
process.execPath,
[
"--input-type=module",
"-e",
`
import {createPackageRuntimeRoutes} from ${JSON.stringify(new URL("../dist/index.js", import.meta.url).href)};
createPackageRuntimeRoutes({packageRevisionId:'package:test',exports:{}})({
service(_type, implementation) { console.log(JSON.stringify(implementation.handshake({nonce:'challenge'}))); }
});
`], {env, encoding:'utf8'});
`,
],
{ env, encoding: "utf8" },
);
assert.equal(result.status, 0, result.stderr);
const handshake = JSON.parse(result.stdout);
assert.equal(handshake.authenticationProof, crypto.createHmac('sha256', token)
.update(JSON.stringify(['challenge','instance:test','package:test'])).digest('hex'));
assert.ok(handshake.capabilities.includes('epoch-grants-v1'));
assert.equal(
handshake.authenticationProof,
crypto
.createHmac("sha256", token)
.update(JSON.stringify(["challenge", "instance:test", "package:test"]))
.digest("hex"),
);
assert.ok(handshake.capabilities.includes("epoch-grants-v1"));
assert.ok(!result.stdout.includes(token));
});
test/migration.test.mjs
+20 -8
View File
@@ -1,13 +1,25 @@
import test from "node:test";
import assert from "node:assert/strict";
import {createMigrationContext, migrationObjectId} from "../dist/migration.js";
import { createMigrationContext, migrationObjectId } from "../dist/migration.js";
test("migration context offers restricted ports and deterministic helper identities", () => {
const input = {schemaVersion: 1, executionId: "operation:transition", exportId: "migrate", ports: [
{name: "old", binding: "slot:name", view: "old", access: ["read"], states: [{objectId: "obj:one", value: "Alice"}]},
{name: "new", binding: "slot:name", view: "new", access: ["write"]}, {name: "newRead", binding: "slot:name", view: "new", access: ["read"]},
{name: "helpers", binding: "atom:helper", view: "new", access: ["create"]},
]};
const {context, result} = createMigrationContext(input);
const input = {
schemaVersion: 1,
executionId: "operation:transition",
exportId: "migrate",
ports: [
{
name: "old",
binding: "slot:name",
view: "old",
access: ["read"],
states: [{ objectId: "obj:one", value: "Alice" }],
},
{ name: "new", binding: "slot:name", view: "new", access: ["write"] },
{ name: "newRead", binding: "slot:name", view: "new", access: ["read"] },
{ name: "helpers", binding: "atom:helper", view: "new", access: ["create"] },
],
};
const { context, result } = createMigrationContext(input);
assert.equal(context.read("old", "obj:one"), "Alice");
assert.throws(() => context.write("old", "obj:one", "Bob"), /does not grant/);
assert.throws(() => context.read("new", "obj:one"), /does not grant/);
@@ -20,5 +32,5 @@ test("migration context offers restricted ports and deterministic helper identit
assert.equal(helper, migrationObjectId(input.executionId, "helpers", "one"));
assert.equal(context.create("helpers", "one"), helper);
assert.equal(result().creates.length, 1);
assert.deepEqual(input.ports[0].states, [{objectId: "obj:one", value: "Alice"}]);
assert.deepEqual(input.ports[0].states, [{ objectId: "obj:one", value: "Alice" }]);
});
test/relationships.test.mjs
+14 -7
View File
@@ -1,16 +1,23 @@
import test from "node:test";
import assert from "node:assert/strict";
import {relationshipMap, relationshipList, relationshipSet} from "../dist/index.js";
import {referenceFromWire} from "../dist/references.js";
import { relationshipMap, relationshipList, relationshipSet } from "../dist/index.js";
import { referenceFromWire } from "../dist/references.js";
test("relationship helpers preserve handles and require explicit collection revisions", async () => {
let current = {revision: 0n, entries: []}, serial = 0;
const port = {collection: async () => ({revision: current.revision, entries: current.entries.map((entry) => ({...entry}))}),
let current = { revision: 0n, entries: [] },
serial = 0;
const port = {
collection: async () => ({ revision: current.revision, entries: current.entries.map((entry) => ({ ...entry })) }),
replace: async (entries, revision) => {
assert.equal(revision, current.revision);
current = {revision: revision + 1n, entries: entries.map((entry) => ({...entry, edgeId: entry.edgeId ?? `edge:${++serial}`}))};
current = {
revision: revision + 1n,
entries: entries.map((entry) => ({ ...entry, edgeId: entry.edgeId ?? `edge:${++serial}` })),
};
return port.collection();
}};
const a = referenceFromWire("obj:a"), b = referenceFromWire("obj:b");
},
};
const a = referenceFromWire("obj:a"),
b = referenceFromWire("obj:b");
const map = relationshipMap(port);
await map.set("a1", a, 0n);
assert.equal((await map.get("a1")).value.equals(a), true);
test/runtime.test.mjs
+231 -182
View File
@@ -21,11 +21,7 @@ import {
ValueSchema,
ValueSourceSchema,
} from "../dist/camino/api_pb.js";
import {
EdgeDependencySchema,
InjectedDependencySchema,
PackageExportRefSchema,
} from "../dist/quixos/refs_pb.js";
import { EdgeDependencySchema, InjectedDependencySchema, PackageExportRefSchema } from "../dist/quixos/refs_pb.js";
import { InvokeCapabilityResponseSchema, OrchestratorRuntime } from "../dist/quixos/orch_pb.js";
import { DerivedDependencySchema, PackageRuntime } from "../dist/quixos/runtime_pb.js";
@@ -93,26 +89,30 @@ test("runtime context exposes only explicitly injected ports", async () => {
resolveEdge: async () => ({ edges: [] }),
connectEdge: async () => ({}),
};
const context = createRuntimeContext(camino, {}, {
objectId: "obj:task",
input: {},
dependencies: [
create(InjectedDependencySchema, {
portId: "port:title",
binding: { case: "stateSlotId", value: "slot:task:title" },
}),
create(InjectedDependencySchema, {
portId: "port:children",
binding: {
case: "edge",
value: create(EdgeDependencySchema, {
edgeTypeId: "edge:task:children",
projectionId: "projection:task:children",
}),
},
}),
],
});
const context = createRuntimeContext(
camino,
{},
{
objectId: "obj:task",
input: {},
dependencies: [
create(InjectedDependencySchema, {
portId: "port:title",
binding: { case: "stateSlotId", value: "slot:task:title" },
}),
create(InjectedDependencySchema, {
portId: "port:children",
binding: {
case: "edge",
value: create(EdgeDependencySchema, {
edgeTypeId: "edge:task:children",
projectionId: "projection:task:children",
}),
},
}),
],
},
);
assert.equal(await context.state("port:title").get(), "obj:task/slot:task:title");
await context.state("port:title").set("Changed");
@@ -126,23 +126,30 @@ test("runtime context exposes only explicitly injected ports", async () => {
test("injected ports preserve an explicitly traversed object through PackageRuntime RPC", async () => {
const reads = [];
const caminoServer = await listen((router) => router.service(CaminoService, {
readState: (request) => {
reads.push(request);
return { value: jsToProtoValue("Project name") };
},
}));
const runtimeServer = await listen(createPackageRuntimeRoutes({
packageRevisionId: "package:test@1",
caminoUrl: caminoServer.url,
exports: {
"export:test:value": (context) => context.state("port:value").get(),
},
}));
const client = createClient(PackageRuntime, createConnectTransport({
baseUrl: runtimeServer.url,
httpVersion: "1.1",
}));
const caminoServer = await listen((router) =>
router.service(CaminoService, {
readState: (request) => {
reads.push(request);
return { value: jsToProtoValue("Project name") };
},
}),
);
const runtimeServer = await listen(
createPackageRuntimeRoutes({
packageRevisionId: "package:test@1",
caminoUrl: caminoServer.url,
exports: {
"export:test:value": (context) => context.state("port:value").get(),
},
}),
);
const client = createClient(
PackageRuntime,
createConnectTransport({
baseUrl: runtimeServer.url,
httpVersion: "1.1",
}),
);
try {
const response = await client.invoke({
export: create(PackageExportRefSchema, {
@@ -150,11 +157,13 @@ test("injected ports preserve an explicitly traversed object through PackageRunt
exportId: "export:test:value",
}),
objectId: "obj:component",
dependencies: [create(InjectedDependencySchema, {
portId: "port:value",
objectId: "obj:project",
binding: { case: "stateSlotId", value: "slot:project:name" },
})],
dependencies: [
create(InjectedDependencySchema, {
portId: "port:value",
objectId: "obj:project",
binding: { case: "stateSlotId", value: "slot:project:name" },
}),
],
});
assert.equal(response.ok, true);
assert.equal(protoValueToJs(response.result), "Project name");
@@ -178,32 +187,40 @@ test("interface views invoke a related object and propagate transitive live depe
}),
}),
});
const orchServer = await listen((router) => router.service(OrchestratorRuntime, {
invokeCapability: (request) => {
invocations.push(request);
return create(InvokeCapabilityResponseSchema, {
ok: true,
result: sourceValue,
dependencies: [create(DerivedDependencySchema, {
kind: "state",
objectId: "obj:project",
attachmentId: "slot:project:name",
})],
});
},
}));
const runtimeServer = await listen(createPackageRuntimeRoutes({
packageRevisionId: "package:test@1",
orchUrl: orchServer.url,
exports: {
"export:test:value": derived((context) =>
context.interface("port:named").live("operation:named:name:get")),
},
}));
const client = createClient(PackageRuntime, createConnectTransport({
baseUrl: runtimeServer.url,
httpVersion: "1.1",
}));
const orchServer = await listen((router) =>
router.service(OrchestratorRuntime, {
invokeCapability: (request) => {
invocations.push(request);
return create(InvokeCapabilityResponseSchema, {
ok: true,
result: sourceValue,
dependencies: [
create(DerivedDependencySchema, {
kind: "state",
objectId: "obj:project",
attachmentId: "slot:project:name",
}),
],
});
},
}),
);
const runtimeServer = await listen(
createPackageRuntimeRoutes({
packageRevisionId: "package:test@1",
orchUrl: orchServer.url,
exports: {
"export:test:value": derived((context) => context.interface("port:named").live("operation:named:name:get")),
},
}),
);
const client = createClient(
PackageRuntime,
createConnectTransport({
baseUrl: runtimeServer.url,
httpVersion: "1.1",
}),
);
try {
const response = await client.invoke({
export: create(PackageExportRefSchema, {
@@ -211,11 +228,13 @@ test("interface views invoke a related object and propagate transitive live depe
exportId: "export:test:value",
}),
objectId: "obj:component",
dependencies: [create(InjectedDependencySchema, {
portId: "port:named",
objectId: "obj:project",
binding: { case: "interfaceRevisionId", value: "interface:named@1" },
})],
dependencies: [
create(InjectedDependencySchema, {
portId: "port:named",
objectId: "obj:project",
binding: { case: "interfaceRevisionId", value: "interface:named@1" },
}),
],
});
assert.equal(response.ok, true);
assert.equal(invocations[0]?.objectId, "obj:project");
@@ -232,64 +251,80 @@ test("interface views invoke a related object and propagate transitive live depe
test("derived interface views reread after their transitive subscriptions become live", async () => {
let current = "before subscription";
let invocationCount = 0;
const caminoServer = await listen((router) => router.service(CaminoService, {
watchObject: async function* (request, context) {
// Model a write racing the first nested capability read. Camino makes
// the subscription live before yielding this snapshot.
current = "after subscription";
yield {
objectId: request.objectId,
snapshot: create(CaminoObjectSchema, {
id: request.objectId,
atomId: "atom:project",
workspaceRevisionId: "workspace:test@1",
}),
};
await new Promise((resolve) =>
context.signal.addEventListener("abort", resolve, { once: true }));
},
}));
const orchServer = await listen((router) => router.service(OrchestratorRuntime, {
invokeCapability: () => {
invocationCount += 1;
return create(InvokeCapabilityResponseSchema, {
ok: true,
result: jsToProtoValue(current),
dependencies: [create(DerivedDependencySchema, {
kind: "state",
objectId: "obj:project",
attachmentId: "slot:project:name",
})],
});
},
}));
const runtimeServer = await listen(createPackageRuntimeRoutes({
packageRevisionId: "package:test@1",
caminoUrl: caminoServer.url,
orchUrl: orchServer.url,
exports: {
"export:test:value": derived((context) =>
context.interface("port:named").invoke("operation:named:name:get")),
},
}));
const client = createClient(PackageRuntime, createConnectTransport({
baseUrl: runtimeServer.url,
httpVersion: "1.1",
}));
const caminoServer = await listen((router) =>
router.service(CaminoService, {
watchObject: async function* (request, context) {
// Model a write racing the first nested capability read. Camino makes
// the subscription live before yielding this snapshot.
current = "after subscription";
yield {
objectId: request.objectId,
snapshot: create(CaminoObjectSchema, {
id: request.objectId,
atomId: "atom:project",
workspaceRevisionId: "workspace:test@1",
}),
};
await new Promise((resolve) => context.signal.addEventListener("abort", resolve, { once: true }));
},
}),
);
const orchServer = await listen((router) =>
router.service(OrchestratorRuntime, {
invokeCapability: () => {
invocationCount += 1;
return create(InvokeCapabilityResponseSchema, {
ok: true,
result: jsToProtoValue(current),
dependencies: [
create(DerivedDependencySchema, {
kind: "state",
objectId: "obj:project",
attachmentId: "slot:project:name",
}),
],
});
},
}),
);
const runtimeServer = await listen(
createPackageRuntimeRoutes({
packageRevisionId: "package:test@1",
caminoUrl: caminoServer.url,
orchUrl: orchServer.url,
exports: {
"export:test:value": derived((context) => context.interface("port:named").invoke("operation:named:name:get")),
},
}),
);
const client = createClient(
PackageRuntime,
createConnectTransport({
baseUrl: runtimeServer.url,
httpVersion: "1.1",
}),
);
const controller = new AbortController();
try {
const stream = client.watch({
export: create(PackageExportRefSchema, {
packageRevisionId: "package:test@1",
exportId: "export:test:value",
}),
objectId: "obj:component",
dependencies: [create(InjectedDependencySchema, {
portId: "port:named",
objectId: "obj:project",
binding: { case: "interfaceRevisionId", value: "interface:named@1" },
})],
}, { signal: controller.signal })[Symbol.asyncIterator]();
const stream = client
.watch(
{
export: create(PackageExportRefSchema, {
packageRevisionId: "package:test@1",
exportId: "export:test:value",
}),
objectId: "obj:component",
dependencies: [
create(InjectedDependencySchema, {
portId: "port:named",
objectId: "obj:project",
binding: { case: "interfaceRevisionId", value: "interface:named@1" },
}),
],
},
{ signal: controller.signal },
)
[Symbol.asyncIterator]();
const initial = await stream.next();
assert.equal(protoValueToJs(initial.value?.value), "after subscription");
assert.equal(invocationCount, 2);
@@ -305,56 +340,70 @@ test("derived watches subscribe before reading and emit only real changes", asyn
let current = "before";
let subscribed = false;
const changes = [];
const caminoServer = await listen((router) => router.service(CaminoService, {
readState: () => {
assert.equal(subscribed, true, "the dependency stream must be live before the state read");
return { value: jsToProtoValue(current) };
},
watchObject: async function* (request, context) {
subscribed = true;
yield {
objectId: request.objectId,
snapshot: create(CaminoObjectSchema, {
id: request.objectId,
atomId: "atom:test",
workspaceRevisionId: "workspace:test@1",
}),
};
while (!context.signal.aborted) {
const changed = await new Promise((resolve) => {
const finish = () => resolve(true);
changes.push(finish);
context.signal.addEventListener("abort", () => resolve(false), { once: true });
});
if (!changed) return;
yield { objectId: request.objectId };
}
},
}));
const runtimeServer = await listen(createPackageRuntimeRoutes({
packageRevisionId: "package:test@1",
caminoUrl: caminoServer.url,
exports: {
"export:test:value": derived((context) => context.state("port:value").get()),
},
}));
const client = createClient(PackageRuntime, createConnectTransport({
baseUrl: runtimeServer.url,
httpVersion: "1.1",
}));
const caminoServer = await listen((router) =>
router.service(CaminoService, {
readState: () => {
assert.equal(subscribed, true, "the dependency stream must be live before the state read");
return { value: jsToProtoValue(current) };
},
watchObject: async function* (request, context) {
subscribed = true;
yield {
objectId: request.objectId,
snapshot: create(CaminoObjectSchema, {
id: request.objectId,
atomId: "atom:test",
workspaceRevisionId: "workspace:test@1",
}),
};
while (!context.signal.aborted) {
const changed = await new Promise((resolve) => {
const finish = () => resolve(true);
changes.push(finish);
context.signal.addEventListener("abort", () => resolve(false), { once: true });
});
if (!changed) return;
yield { objectId: request.objectId };
}
},
}),
);
const runtimeServer = await listen(
createPackageRuntimeRoutes({
packageRevisionId: "package:test@1",
caminoUrl: caminoServer.url,
exports: {
"export:test:value": derived((context) => context.state("port:value").get()),
},
}),
);
const client = createClient(
PackageRuntime,
createConnectTransport({
baseUrl: runtimeServer.url,
httpVersion: "1.1",
}),
);
const controller = new AbortController();
try {
const stream = client.watch({
export: create(PackageExportRefSchema, {
packageRevisionId: "package:test@1",
exportId: "export:test:value",
}),
objectId: "obj:test",
dependencies: [create(InjectedDependencySchema, {
portId: "port:value",
binding: { case: "stateSlotId", value: "slot:test:value" },
})],
}, { signal: controller.signal })[Symbol.asyncIterator]();
const stream = client
.watch(
{
export: create(PackageExportRefSchema, {
packageRevisionId: "package:test@1",
exportId: "export:test:value",
}),
objectId: "obj:test",
dependencies: [
create(InjectedDependencySchema, {
portId: "port:value",
binding: { case: "stateSlotId", value: "slot:test:value" },
}),
],
},
{ signal: controller.signal },
)
[Symbol.asyncIterator]();
const initial = await stream.next();
assert.equal(protoValueToJs(initial.value?.value), "before");
assert.equal(initial.value?.initial, true);