Timothy J. Aveni
6a17cd1ae0
Implement scoped interface extents across QX, generated bindings, SDK, Orch, Camino and browser queries. Preserve native projection, aggregation, pagination and watch invalidation without synthetic persisted objects. Validate exact query grants and scope freshness on execution and replay. Add local PostgreSQL, transport, browser and code-generation coverage. C1 implementation slice only: full-text indexes and Studio search adoption (C2/C3) remain tracked in the implementation handoff. No publication or deployment.
668 lines
34 KiB
JavaScript
668 lines
34 KiB
JavaScript
import http from "node:http";
|
|
import { readFileSync } from "node:fs";
|
|
import { createInvocationRegistry } from "./invocations.js";
|
|
import { isObjectReference, referenceFromWire, referenceToWire, assertReferenceFree, } from "./references.js";
|
|
export * from "./bindings.js";
|
|
export * from "./queries.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, } from "./migration.js";
|
|
import { create, equals } from "@bufbuild/protobuf";
|
|
import { Code, ConnectError, createClient } from "@connectrpc/connect";
|
|
import { connectNodeAdapter, createConnectTransport } from "@connectrpc/connect-node";
|
|
import { CaminoService, CrdtValueSchema, ListValueSchema, NullValueSchema, ObjectValueSchema, RefValueSchema, ValueSchema, } from "./camino/api_pb.js";
|
|
import { OrchestratorRuntime } from "./quixos/orch_pb.js";
|
|
import { DerivedDependencySchema, HandshakeResponseSchema, InvokeResponseSchema, PackageRuntime, WatchEventSchema, } from "./quixos/runtime_pb.js";
|
|
import { CapabilityRefSchema } from "./quixos/refs_pb.js";
|
|
const dependencyKey = (dependency) => `${dependency.kind}:${dependency.objectId}:${dependency.attachmentId}:${dependency.kind === "edge" ? dependency.projectionId : ""}`;
|
|
const dependencyScope = new AsyncLocalStorage();
|
|
const recordDependency = async (dependency) => {
|
|
const scope = dependencyScope.getStore();
|
|
if (!scope)
|
|
return;
|
|
const key = `${dependency.kind}:${dependency.objectId}:${dependency.attachmentId}:${dependency.kind === "edge" ? dependency.projectionId : ""}`;
|
|
scope.dependencies.set(key, dependency);
|
|
await scope.observe?.(dependency);
|
|
};
|
|
const bytesToBase64 = (value) => Buffer.from(value).toString("base64");
|
|
const base64ToBytes = (value) => Buffer.from(value, "base64");
|
|
const isRecord = (value) => Boolean(value) && typeof value === "object" && !Array.isArray(value);
|
|
const isWrappedValue = (value) => isRecord(value) &&
|
|
"$quixosValue" in value &&
|
|
isRecord(value.$quixosValue) &&
|
|
value.$quixosValue.$typeName === "camino.Value";
|
|
export const objectRef = (reference) => {
|
|
referenceToWire(reference);
|
|
return reference;
|
|
};
|
|
export const liveValue = (value) => ({ $quixosValue: value });
|
|
export const jsToProtoValue = (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, {}) } });
|
|
}
|
|
if (typeof value === "boolean")
|
|
return create(ValueSchema, { kind: { case: "boolValue", value } });
|
|
if (typeof value === "number")
|
|
return create(ValueSchema, { kind: { case: "numberValue", value } });
|
|
if (typeof value === "bigint")
|
|
return create(ValueSchema, { kind: { case: "integerValue", value: String(value) } });
|
|
if (typeof value === "string")
|
|
return create(ValueSchema, { kind: { case: "stringValue", value } });
|
|
if (value instanceof Uint8Array)
|
|
return create(ValueSchema, { kind: { case: "bytesValue", value } });
|
|
if (Array.isArray(value)) {
|
|
return create(ValueSchema, {
|
|
kind: { case: "listValue", value: create(ListValueSchema, { values: value.map(jsToProtoValue) }) },
|
|
});
|
|
}
|
|
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") {
|
|
return create(ValueSchema, {
|
|
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)])),
|
|
}),
|
|
},
|
|
});
|
|
};
|
|
export const protoValueToJs = (value) => {
|
|
switch (value?.kind.case) {
|
|
case "nullValue":
|
|
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),
|
|
};
|
|
}
|
|
};
|
|
export const protoFieldsToJs = (fields) => Object.fromEntries(Object.entries(fields).map(([key, value]) => [key, protoValueToJs(value)]));
|
|
export class RuntimeAuthorityError extends Error {
|
|
retryable;
|
|
constructor(message) {
|
|
super(message);
|
|
this.name = "RuntimeAuthorityError";
|
|
this.retryable = /WORKSPACE_FENCED|STALE_EPOCH/.test(message);
|
|
}
|
|
}
|
|
const targetForEdge = (edge, projectionId) => (edge.firstProjectionId === projectionId ? edge.secondObjectId : edge.firstObjectId);
|
|
export const createRuntimeContext = (camino, orch, request) => {
|
|
const acquiredPort = (objectId, interfaceRevisionId, conformance) => {
|
|
const invoke = async (operationId, input = {}) => {
|
|
const response = await orch.invokeCapability({
|
|
objectId,
|
|
capability: create(CapabilityRefSchema, { interfaceRevisionId, operationId, conformance }),
|
|
input: Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsToProtoValue(value)])),
|
|
});
|
|
if (!response.ok)
|
|
throw new Error(response.error || "Capability invocation failed");
|
|
for (const dependency of response.dependencies) {
|
|
if (dependency.kind === "state" || dependency.kind === "edge")
|
|
await recordDependency({
|
|
kind: dependency.kind,
|
|
objectId: dependency.objectId,
|
|
attachmentId: dependency.attachmentId,
|
|
...(dependency.kind === "edge" ? { projectionId: dependency.projectionId } : {}),
|
|
});
|
|
}
|
|
if (!response.result)
|
|
throw new Error("Capability returned no value");
|
|
return response.result;
|
|
};
|
|
return {
|
|
objectId: referenceFromWire(objectId),
|
|
interfaceRevisionId,
|
|
invoke: async (operation, input) => protoValueToJs(await invoke(operation, input)),
|
|
live: async (operation, input) => liveValue(await invoke(operation, input)),
|
|
};
|
|
};
|
|
const ports = new Map();
|
|
for (const dependency of request.dependencies) {
|
|
switch (dependency.binding.case) {
|
|
case "queryId":
|
|
case "workspaceQuery": {
|
|
if (dependency.binding.case === "workspaceQuery" && !dependency.binding.value.scope)
|
|
throw new Error(`Workspace query port ${dependency.portId} has no scope`);
|
|
const queryId = dependency.binding.case === "queryId" ? dependency.binding.value : dependency.binding.value.queryId;
|
|
const root = dependency.binding.case === "queryId"
|
|
? { case: "objectId", value: dependency.objectId || request.objectId }
|
|
: { case: "workspaceScope", value: dependency.binding.value.scope };
|
|
ports.set(dependency.portId, {
|
|
queryId,
|
|
execute: (variables, expectedDefinitionDigest) => orch.executeQuery({ queryId, root, variables, expectedDefinitionDigest }),
|
|
watch: (variables, signal, expectedDefinitionDigest) => orch.watchQuery({ queryId, root, variables, expectedDefinitionDigest }, { signal }),
|
|
});
|
|
break;
|
|
}
|
|
case "stateSlotId": {
|
|
const slotId = dependency.binding.value;
|
|
const dependencyObjectId = dependency.objectId || request.objectId;
|
|
const state = {
|
|
slotId,
|
|
async get() {
|
|
await recordDependency({ kind: "state", objectId: dependencyObjectId, attachmentId: slotId });
|
|
return protoValueToJs((await camino.readState({ objectId: dependencyObjectId, slotId })).value);
|
|
},
|
|
async live() {
|
|
await recordDependency({ kind: "state", objectId: dependencyObjectId, attachmentId: slotId });
|
|
const value = await camino.readState({ objectId: dependencyObjectId, slotId });
|
|
if (!value.value)
|
|
throw new Error(`State ${slotId} returned no value`);
|
|
return liveValue(value.value);
|
|
},
|
|
async set(value) {
|
|
assertReferenceFree(isWrappedValue(value) ? protoValueToJs(value.$quixosValue) : value);
|
|
await camino.writeState({ objectId: dependencyObjectId, slotId, value: jsToProtoValue(value) });
|
|
},
|
|
};
|
|
ports.set(dependency.portId, state);
|
|
break;
|
|
}
|
|
case "edge": {
|
|
const { edgeTypeId, projectionId } = dependency.binding.value;
|
|
const dependencyObjectId = dependency.objectId || request.objectId;
|
|
const collectionResult = (response) => ({
|
|
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),
|
|
}
|
|
: {}),
|
|
})),
|
|
});
|
|
const edge = {
|
|
edgeTypeId,
|
|
projectionId,
|
|
async collection() {
|
|
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),
|
|
};
|
|
}),
|
|
}));
|
|
},
|
|
async resolve() {
|
|
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)));
|
|
},
|
|
async connect(target) {
|
|
const targetObjectId = referenceToWire(target);
|
|
await camino.connectEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId, targetObjectId });
|
|
},
|
|
async disconnect(target) {
|
|
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 });
|
|
}
|
|
},
|
|
};
|
|
ports.set(dependency.portId, edge);
|
|
break;
|
|
}
|
|
case "interfaceRevisionId": {
|
|
const interfaceRevisionId = dependency.binding.value;
|
|
const dependencyObjectId = dependency.objectId || request.objectId;
|
|
ports.set(dependency.portId, acquiredPort(dependencyObjectId, interfaceRevisionId));
|
|
break;
|
|
}
|
|
case "constructorAtomId": {
|
|
const atomId = dependency.binding.value;
|
|
const constructor = {
|
|
atomId,
|
|
async construct(input = {}) {
|
|
const response = await orch.constructObject({
|
|
atomId,
|
|
input: Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsToProtoValue(value)])),
|
|
});
|
|
if (!response.object)
|
|
throw new Error(`Constructor ${atomId} returned no object`);
|
|
return referenceFromWire(response.object.id);
|
|
},
|
|
};
|
|
ports.set(dependency.portId, constructor);
|
|
}
|
|
}
|
|
}
|
|
const requirePort = (portId, kind) => {
|
|
const port = ports.get(portId);
|
|
if (!port || !(kind in port))
|
|
throw new Error(`Missing ${kind} dependency port ${portId}`);
|
|
return port;
|
|
};
|
|
return {
|
|
async tryConform(object, interfaceRevisionId) {
|
|
const objectId = referenceToWire(object);
|
|
const { conformance } = await orch.tryConform({ objectId, interfaceRevisionId });
|
|
if (conformance && (conformance.objectId !== objectId || conformance.interfaceRevisionId !== interfaceRevisionId))
|
|
throw new Error("Conformance response does not match the requested view");
|
|
return conformance ? acquiredPort(objectId, interfaceRevisionId, conformance) : undefined;
|
|
},
|
|
get objectId() {
|
|
return referenceFromWire(request.objectId);
|
|
},
|
|
input: protoFieldsToJs(request.input),
|
|
inputProto: request.input,
|
|
ports,
|
|
state: (portId) => requirePort(portId, "slotId"),
|
|
edge: (portId) => requirePort(portId, "edgeTypeId"),
|
|
interface: (portId) => requirePort(portId, "interfaceRevisionId"),
|
|
constructor: (portId) => requirePort(portId, "atomId"),
|
|
query: (portId) => requirePort(portId, "queryId"),
|
|
};
|
|
};
|
|
export const derived = (get) => ({ kind: "derived", get });
|
|
const isDerived = (handler) => typeof handler === "object" && handler.kind === "derived";
|
|
const evaluate = async (handler, context, observe) => {
|
|
const dependencies = new Map();
|
|
const result = await dependencyScope.run({ dependencies, observe }, () => isDerived(handler) ? handler.get(context) : handler(context));
|
|
return { value: jsToProtoValue(result), dependencies: [...dependencies.values()] };
|
|
};
|
|
const protoDependencies = (dependencies) => dependencies.map((entry) => create(DerivedDependencySchema, {
|
|
kind: entry.kind,
|
|
objectId: entry.objectId,
|
|
attachmentId: entry.attachmentId,
|
|
projectionId: entry.kind === "edge" ? entry.projectionId : "",
|
|
}));
|
|
export const createPackageRuntimeRoutes = (config) => {
|
|
const invocations = createInvocationRegistry();
|
|
const headers = {};
|
|
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 authenticateInstance = (header) => {
|
|
if (!process.env.QUIXOS_RUNTIME_INSTANCE_ID)
|
|
return; // standalone development ABI
|
|
const supplied = Buffer.from(header.get("x-quixos-instance-token") ?? "");
|
|
const expected = Buffer.from(processToken);
|
|
if (!expected.length || supplied.length !== expected.length || !timingSafeEqual(supplied, expected)) {
|
|
throw new ConnectError("Invalid runtime instance credential", Code.Unauthenticated);
|
|
}
|
|
};
|
|
const clientsFor = (request) => {
|
|
const context = request.context;
|
|
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) => 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")),
|
|
};
|
|
};
|
|
const runtimeControl = async (operation, input) => {
|
|
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());
|
|
if (!response.ok)
|
|
throw new RuntimeAuthorityError(value.error ?? "Runtime authority request failed");
|
|
return value;
|
|
};
|
|
const attachSessions = (runtimeContext, request) => {
|
|
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("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.
|
|
if (error instanceof RuntimeAuthorityError)
|
|
throw error;
|
|
return register();
|
|
});
|
|
let closed = false;
|
|
return {
|
|
id: registered.sessionId,
|
|
async run(work) {
|
|
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("acquire-session", registered);
|
|
const execution = invocations.begin(grant.invocationId);
|
|
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 {
|
|
execution.finish();
|
|
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;
|
|
},
|
|
};
|
|
};
|
|
};
|
|
return (router) => 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;
|
|
const subscriptions = new Map();
|
|
const establishing = new Map();
|
|
let subscriptionEpoch = 0;
|
|
const ensureSubscription = async (dependency) => {
|
|
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) => ({ key, done: true, error }));
|
|
const 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);
|
|
}
|
|
};
|
|
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: current.value,
|
|
dependencies: protoDependencies(current.dependencies),
|
|
initial: true,
|
|
});
|
|
const abort = new Promise((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();
|
|
}
|
|
}
|
|
finally {
|
|
execution.finish();
|
|
}
|
|
},
|
|
});
|
|
};
|
|
export const servePackageRuntime = (config) => {
|
|
const host = process.env.QUIXOS_RUNTIME_HOST ?? "127.0.0.1";
|
|
const port = Number(process.env.QUIXOS_RUNTIME_PORT ?? "0");
|
|
const handler = connectNodeAdapter({ routes: createPackageRuntimeRoutes(config) });
|
|
const server = http.createServer((request, response) => void handler(request, response));
|
|
server.listen(port, host, () => console.log(`${config.packageRevisionId} listening on ${host}:${port}`));
|
|
const shutdown = () => {
|
|
server.close(() => process.exit(0));
|
|
server.closeAllConnections();
|
|
};
|
|
process.on("SIGTERM", shutdown);
|
|
process.on("SIGINT", shutdown);
|
|
return server;
|
|
};
|