Keep checked execution groups and public catalog preparation cooperative

This commit is contained in:
Timothy J. Aveni
2026-10-12 19:21:34 -07:00
parent a4ba6169a6
commit 47b316180d
6 changed files with 243 additions and 228 deletions
+172 -175
View File
@@ -6,7 +6,7 @@ import type {
PersistentAttachment,
} from "../capability-model/types.js";
import { type ExecutionInstallation, type ExecutionGroupIdentity } from "./installation.js";
import { type ExecutionPlan, hexDigest, ExecutionPlanError } from "./plan.js";
import { type ExecutionPlan, hexDigest, hexDigestAsync, ExecutionPlanError } from "./plan.js";
export interface ExecutionEnvironment {
/** Explicit host placement. Merely including an optional module does not select it. */
@@ -23,12 +23,12 @@ export interface InstalledExecutionGroup extends ExecutionGroupIdentity {
}
/** Conservative outbound closure. Runtime graph values, reverse callers, source
* provenance and the workspace epoch are deliberately absent from this identity. */
export function compileExecutionGroups(
export async function compileExecutionGroups(
plan: ExecutionPlan,
installation: ExecutionInstallation,
environments: Readonly<Record<string, ExecutionEnvironment>>,
): InstalledExecutionGroup[] {
if (installation.plan !== plan.digest || hexDigest(plan.definition) !== plan.digest)
): Promise<InstalledExecutionGroup[]> {
if (installation.plan !== plan.digest || (await hexDigestAsync(plan.definition)) !== plan.digest)
throw new ExecutionPlanError("GROUP_PLAN_MISMATCH", plan.digest, "Expected one checked installation");
const workspace = plan.definition.workspace;
const packages = new Map(workspace.packageImports.map((p) => [p.revisionId as string, p]));
@@ -82,178 +82,175 @@ export function compileExecutionGroups(
};
const bindingContract = (value: DependencyBinding) =>
value.kind === "query" ? { ...value, packageRevisionId: logical(value.packageRevisionId) } : value;
return installation.packages
.filter((p) => !!implementations.get(p.revision))
.map((root) => {
const consumed = new Map<string, unknown>(),
visited = new Set<string>(),
visitedQueries = new Set<string>(),
consumedPackages = new Set<string>();
const remember = (key: string, value: unknown) => consumed.set(key, value);
const storage = (id: string) => {
const value = attachments.get(id);
if (!value) throw new ExecutionPlanError("GROUP_STORAGE_MISSING", id, "Uninstalled consumed storage");
remember(`storage:${id}`, attachmentContract(value));
};
// Deletion may visit related objects and their owned state. Conservatively
// include the connected schema component, including possible future edges.
const lifecycle = (atom: string) => {
const visited = new Set<string>(),
pending = [atom];
const atoms = (
constraint: Extract<PersistentAttachment, { kind: "edge" }>["endpoints"][number]["constraint"],
) =>
constraint.kind === "atom"
? [constraint.atomId as string]
: workspace.conformances
.filter((c) => c.interfaceRevisionId === constraint.interfaceRevisionId)
.map((c) => c.atomId as string);
while (pending.length) {
const current = pending.pop()!;
if (visited.has(current)) continue;
visited.add(current);
for (const owner of workspace.conformances.filter((c) => c.atomId === current))
for (const attachment of owner.privateAttachments) storage(attachment.id);
for (const attachment of attachments.values())
if (attachment.kind === "edge" && attachment.endpoints.some((e) => atoms(e.constraint).includes(current))) {
storage(attachment.id);
pending.push(...attachment.endpoints.flatMap((e) => atoms(e.constraint)));
}
}
};
const targets = (binding: DependencyBinding, atom: string): string[] => {
if (binding.kind === "interface" && binding.receiver === "input")
return workspace.conformances
.filter((c) => c.interfaceRevisionId === binding.interfaceRevisionId)
.map((c) => c.atomId);
if (!("via" in binding) || !binding.via) return [atom];
storage(binding.via.edgeTypeId);
const edge = attachments.get(binding.via.edgeTypeId);
if (!edge || edge.kind !== "edge")
throw new ExecutionPlanError("GROUP_EDGE_MISSING", atom, "Missing traversal edge");
const endpoint = edge.endpoints.find((e) => e.projectionId !== binding.via!.projectionId)!;
const constraint = endpoint.constraint;
return constraint.kind === "atom"
? [constraint.atomId]
const groups: InstalledExecutionGroup[] = [];
for (const root of installation.packages.filter((p) => !!implementations.get(p.revision))) {
const consumed = new Map<string, unknown>(),
visited = new Set<string>(),
visitedQueries = new Set<string>(),
consumedPackages = new Set<string>();
const remember = (key: string, value: unknown) => consumed.set(key, value);
const storage = (id: string) => {
const value = attachments.get(id);
if (!value) throw new ExecutionPlanError("GROUP_STORAGE_MISSING", id, "Uninstalled consumed storage");
remember(`storage:${id}`, attachmentContract(value));
};
// Deletion may visit related objects and their owned state. Conservatively
// include the connected schema component, including possible future edges.
const lifecycle = (atom: string) => {
const visited = new Set<string>(),
pending = [atom];
const atoms = (constraint: Extract<PersistentAttachment, { kind: "edge" }>["endpoints"][number]["constraint"]) =>
constraint.kind === "atom"
? [constraint.atomId as string]
: workspace.conformances
.filter((c) => c.interfaceRevisionId === constraint.interfaceRevisionId)
.map((c) => c.atomId);
};
const query = (revision: string, id: string) => {
const key = JSON.stringify([revision, id]);
if (visitedQueries.has(key)) return;
visitedQueries.add(key);
const value = plan.definition.queries.find((q) => q.id === key);
if (!value) throw new ExecutionPlanError("GROUP_QUERY_MISSING", key, "Missing consumed query");
remember(`query:${JSON.stringify([logical(revision), id])}`, value.query.checked.definitionDigest);
for (const operation of value.operations) visit(operation);
};
const dependencies = (entry: PackageExport, bindings: readonly BoundDependency[], atom: string) => {
for (const bound of bindings) {
const target = bound.binding,
port = entry.dependencyPorts.find((p) => p.id === bound.portId)!;
if ("via" in target && target.via) storage(target.via.edgeTypeId);
if (target.kind === "reflection") {
for (const selector of target.selectors)
remember("reflection:" + selector, portableReflectionValue(workspace, selector));
} else if (target.kind === "state") storage(target.slotId);
else if (target.kind === "edge") storage(target.edgeTypeId);
else if (target.kind === "query") query(target.packageRevisionId, target.queryId);
else if (target.kind === "interface" && port.requirement.kind === "interface") {
const iface = workspace.interfaceImports.find((i) => i.revisionId === target.interfaceRevisionId)!;
for (const targetAtom of targets(target, atom))
for (const op of port.requirement.operations ??
iface.members.flatMap((m) => m.operations.map((o) => o.id)))
visit(JSON.stringify([targetAtom, target.interfaceRevisionId, op]));
} else if (target.kind === "creation") {
for (const seed of target.owners) {
const owner = workspace.conformances.find(
(c) => c.atomId === target.atomId && c.interfaceRevisionId === seed.interfaceRevisionId,
)!;
remember(`birth:${owner.id}`, {
id: owner.id,
atom: owner.atomId,
interface: owner.interfaceRevisionId,
initialization: owner.initialization ?? [],
});
for (const attachment of owner.privateAttachments) storage(attachment.id);
}
} else
throw new ExecutionPlanError(
"GROUP_DEPENDENCY_UNSUPPORTED",
bound.portId,
"Cannot establish execution reuse for this dependency",
);
}
};
const visit = (key: string) => {
if (visited.has(key)) return;
visited.add(key);
const operation = operations.get(key);
if (!operation) throw new ExecutionPlanError("GROUP_OPERATION_MISSING", key, "Missing consumed operation");
const binding = operation.binding;
if (binding.kind === "package") {
const pkg = packages.get(binding.packageRevisionId)!;
const implementation = implementations.get(pkg.revisionId);
if (!implementation)
throw new ExecutionPlanError(
"GROUP_IMPLEMENTATION_MISSING",
pkg.revisionId,
"Consumed package has no selected server implementation",
);
consumedPackages.add(pkg.packageId);
remember(`implementation:${pkg.packageId}`, {
digest: implementation.contentDigest,
environment: environment(pkg.revisionId),
});
const entry = pkg.exports.find((e) => e.id === binding.exportId)!;
const bindings = binding.dependencies.map((d) => ({ portId: d.portId, binding: bindingContract(d.binding) }));
const acceptance = (binding.acceptanceDependencies ?? []).map((d) => ({
portId: d.portId,
binding: bindingContract(d.binding),
}));
const contract = {
input: operation.input,
output: operation.output,
execution: operation.execution,
scope: operation.scope,
binding: {
package: pkg.packageId,
export: binding.exportId,
dependencies: bindings,
acceptanceDependencies: acceptance,
},
};
// Deduplicate aliases of the same bound export, rather than keying its owner by reverse callers.
remember(`binding:${hexDigest(contract)}`, contract);
dependencies(entry, binding.dependencies, operation.atomId);
if (entry.acceptanceExportId)
dependencies(
pkg.exports.find((e) => e.id === entry.acceptanceExportId)!,
binding.acceptanceDependencies ?? [],
operation.atomId,
);
} else {
remember(`operation:${key}`, operation);
if (binding.kind === "state") storage(binding.slotId);
if (binding.kind === "edge") storage(binding.edgeTypeId);
if (binding.kind === "lifecycle") lifecycle(operation.atomId);
}
};
for (const operation of plan.definition.operations)
if (operation.binding.kind === "package" && operation.binding.packageRevisionId === root.revision)
visit(operation.key);
const own = implementations.get(root.revision)!;
const exports = packages.get(root.revision)!.exports.map(({ displayName: _name, dependencyPorts, ...entry }) => ({
...entry,
dependencyPorts: dependencyPorts.map(({ displayName: _name, ...port }) => port),
}));
const entries = [...consumed].sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0));
return {
logicalId: logical(root.revision),
packageRevision: root.revision,
executionDigest: `sha256:${hexDigest({ implementation: own.contentDigest, exports, environment: environment(root.revision), versions: plan.definition.versions, consumed: entries })}`,
consumedPackages: [...consumedPackages].sort(),
};
.map((c) => c.atomId as string);
while (pending.length) {
const current = pending.pop()!;
if (visited.has(current)) continue;
visited.add(current);
for (const owner of workspace.conformances.filter((c) => c.atomId === current))
for (const attachment of owner.privateAttachments) storage(attachment.id);
for (const attachment of attachments.values())
if (attachment.kind === "edge" && attachment.endpoints.some((e) => atoms(e.constraint).includes(current))) {
storage(attachment.id);
pending.push(...attachment.endpoints.flatMap((e) => atoms(e.constraint)));
}
}
};
const targets = (binding: DependencyBinding, atom: string): string[] => {
if (binding.kind === "interface" && binding.receiver === "input")
return workspace.conformances
.filter((c) => c.interfaceRevisionId === binding.interfaceRevisionId)
.map((c) => c.atomId);
if (!("via" in binding) || !binding.via) return [atom];
storage(binding.via.edgeTypeId);
const edge = attachments.get(binding.via.edgeTypeId);
if (!edge || edge.kind !== "edge")
throw new ExecutionPlanError("GROUP_EDGE_MISSING", atom, "Missing traversal edge");
const endpoint = edge.endpoints.find((e) => e.projectionId !== binding.via!.projectionId)!;
const constraint = endpoint.constraint;
return constraint.kind === "atom"
? [constraint.atomId]
: workspace.conformances
.filter((c) => c.interfaceRevisionId === constraint.interfaceRevisionId)
.map((c) => c.atomId);
};
const query = (revision: string, id: string) => {
const key = JSON.stringify([revision, id]);
if (visitedQueries.has(key)) return;
visitedQueries.add(key);
const value = plan.definition.queries.find((q) => q.id === key);
if (!value) throw new ExecutionPlanError("GROUP_QUERY_MISSING", key, "Missing consumed query");
remember(`query:${JSON.stringify([logical(revision), id])}`, value.query.checked.definitionDigest);
for (const operation of value.operations) visit(operation);
};
const dependencies = (entry: PackageExport, bindings: readonly BoundDependency[], atom: string) => {
for (const bound of bindings) {
const target = bound.binding,
port = entry.dependencyPorts.find((p) => p.id === bound.portId)!;
if ("via" in target && target.via) storage(target.via.edgeTypeId);
if (target.kind === "reflection") {
for (const selector of target.selectors)
remember("reflection:" + selector, portableReflectionValue(workspace, selector));
} else if (target.kind === "state") storage(target.slotId);
else if (target.kind === "edge") storage(target.edgeTypeId);
else if (target.kind === "query") query(target.packageRevisionId, target.queryId);
else if (target.kind === "interface" && port.requirement.kind === "interface") {
const iface = workspace.interfaceImports.find((i) => i.revisionId === target.interfaceRevisionId)!;
for (const targetAtom of targets(target, atom))
for (const op of port.requirement.operations ?? iface.members.flatMap((m) => m.operations.map((o) => o.id)))
visit(JSON.stringify([targetAtom, target.interfaceRevisionId, op]));
} else if (target.kind === "creation") {
for (const seed of target.owners) {
const owner = workspace.conformances.find(
(c) => c.atomId === target.atomId && c.interfaceRevisionId === seed.interfaceRevisionId,
)!;
remember(`birth:${owner.id}`, {
id: owner.id,
atom: owner.atomId,
interface: owner.interfaceRevisionId,
initialization: owner.initialization ?? [],
});
for (const attachment of owner.privateAttachments) storage(attachment.id);
}
} else
throw new ExecutionPlanError(
"GROUP_DEPENDENCY_UNSUPPORTED",
bound.portId,
"Cannot establish execution reuse for this dependency",
);
}
};
const visit = (key: string) => {
if (visited.has(key)) return;
visited.add(key);
const operation = operations.get(key);
if (!operation) throw new ExecutionPlanError("GROUP_OPERATION_MISSING", key, "Missing consumed operation");
const binding = operation.binding;
if (binding.kind === "package") {
const pkg = packages.get(binding.packageRevisionId)!;
const implementation = implementations.get(pkg.revisionId);
if (!implementation)
throw new ExecutionPlanError(
"GROUP_IMPLEMENTATION_MISSING",
pkg.revisionId,
"Consumed package has no selected server implementation",
);
consumedPackages.add(pkg.packageId);
remember(`implementation:${pkg.packageId}`, {
digest: implementation.contentDigest,
environment: environment(pkg.revisionId),
});
const entry = pkg.exports.find((e) => e.id === binding.exportId)!;
const bindings = binding.dependencies.map((d) => ({ portId: d.portId, binding: bindingContract(d.binding) }));
const acceptance = (binding.acceptanceDependencies ?? []).map((d) => ({
portId: d.portId,
binding: bindingContract(d.binding),
}));
const contract = {
input: operation.input,
output: operation.output,
execution: operation.execution,
scope: operation.scope,
binding: {
package: pkg.packageId,
export: binding.exportId,
dependencies: bindings,
acceptanceDependencies: acceptance,
},
};
// Deduplicate aliases of the same bound export, rather than keying its owner by reverse callers.
remember(`binding:${hexDigest(contract)}`, contract);
dependencies(entry, binding.dependencies, operation.atomId);
if (entry.acceptanceExportId)
dependencies(
pkg.exports.find((e) => e.id === entry.acceptanceExportId)!,
binding.acceptanceDependencies ?? [],
operation.atomId,
);
} else {
remember(`operation:${key}`, operation);
if (binding.kind === "state") storage(binding.slotId);
if (binding.kind === "edge") storage(binding.edgeTypeId);
if (binding.kind === "lifecycle") lifecycle(operation.atomId);
}
};
for (const operation of plan.definition.operations)
if (operation.binding.kind === "package" && operation.binding.packageRevisionId === root.revision)
visit(operation.key);
const own = implementations.get(root.revision)!;
const exports = packages.get(root.revision)!.exports.map(({ displayName: _name, dependencyPorts, ...entry }) => ({
...entry,
dependencyPorts: dependencyPorts.map(({ displayName: _name, ...port }) => port),
}));
const entries = [...consumed].sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0));
groups.push({
logicalId: logical(root.revision),
packageRevision: root.revision,
executionDigest: `sha256:${await hexDigestAsync({ implementation: own.contentDigest, exports, environment: environment(root.revision), versions: plan.definition.versions, consumed: entries })}`,
consumedPackages: [...consumedPackages].sort(),
});
}
return groups;
}
+28 -20
View File
@@ -3,7 +3,7 @@ import path from "node:path";
import { loadCheckedArtifactWorld } from "./receipts.js";
import { contentArtifactUrl, loadArtifactWorld, type LoadedArtifactWorld } from "./files.js";
import { selectArtifactImplementation, type ArtifactImplementation, type ComponentArtifact } from "./artifacts.js";
import { ExecutionPlanError, hexDigest, type ExecutionPlan } from "./plan.js";
import { ExecutionPlanError, hexDigestAsync, type ExecutionPlan } from "./plan.js";
import {
publicOperationContract,
publicQueryContract,
@@ -41,13 +41,13 @@ export interface ClientContractCatalog {
}[];
}
/** Explicit projection prevents accidental spreading of filesystem roots or private bindings. */
export function clientContractCatalog(
export async function clientContractCatalog(
plan: ExecutionPlan,
installation: ExecutionInstallation,
requestedPackages: readonly string[],
checkedKernel?: InspectionKernel,
): ClientContractCatalog {
if (installation.plan !== plan.digest || hexDigest(plan.definition) !== plan.digest)
): Promise<ClientContractCatalog> {
if (installation.plan !== plan.digest || (await hexDigestAsync(plan.definition)) !== plan.digest)
throw new ExecutionPlanError(
"CATALOG_PLAN_MISMATCH",
plan.digest,
@@ -112,6 +112,7 @@ export function createClientContractCatalog(
const revisions = plan.definition.workspace.packageImports.map((pkg) => pkg.revisionId);
const installed = new Set<string>(revisions);
let catalog: ClientContractCatalog | undefined;
let preparing: Promise<void> | undefined;
let release: (() => void) | undefined;
let closed = false;
const freeze = (value: unknown, seen = new WeakSet<object>()) => {
@@ -121,7 +122,7 @@ export function createClientContractCatalog(
Object.freeze(value);
};
return {
select(requestedPackages: readonly string[]): ClientContractCatalog {
async select(requestedPackages: readonly string[]): Promise<ClientContractCatalog> {
if (closed) throw Error("CANDIDATE_METADATA_CLOSED");
for (const revision of requestedPackages)
if (!installed.has(revision))
@@ -129,22 +130,29 @@ export function createClientContractCatalog(
if (!catalog) {
// Like other candidate admission charges, this is an allocation estimate,
// not a V8 heap limit. The process/cgroup ceiling bounds transient work.
const reservation = metadata.reserve(32 * 1024 * 1024);
try {
const compiled = clientContractCatalog(plan, installation, revisions, checkedKernel);
const bytes = Buffer.byteLength(JSON.stringify(compiled));
if (bytes > 8 * 1024 * 1024) throw Error("CANDIDATE_CATALOG_LIMIT");
freeze(compiled);
// No await between reservations: replacing the transient upper bound
// with a smaller retained estimate cannot race another admission.
reservation();
release = metadata.reserve(Math.max(64 * 1024, bytes * 4));
catalog = compiled;
} catch (error) {
reservation();
throw error;
}
preparing ??= (async () => {
const reservation = metadata.reserve(32 * 1024 * 1024);
try {
const compiled = await clientContractCatalog(plan, installation, revisions, checkedKernel);
if (closed) throw Error("CANDIDATE_METADATA_CLOSED");
const bytes = Buffer.byteLength(JSON.stringify(compiled));
if (bytes > 8 * 1024 * 1024) throw Error("CANDIDATE_CATALOG_LIMIT");
freeze(compiled);
// No await between reservations: replacing the transient upper bound
// with a smaller retained estimate cannot race another admission.
reservation();
release = metadata.reserve(Math.max(64 * 1024, bytes * 4));
catalog = compiled;
} catch (error) {
reservation();
throw error;
}
})().finally(() => {
preparing = undefined;
});
await preparing;
}
if (closed || !catalog) throw Error("CANDIDATE_METADATA_CLOSED");
const requested = new Set(requestedPackages);
return {
inspections: catalog.inspections,
+7 -12
View File
@@ -1,22 +1,17 @@
import { compileInspectionTemplates, type InspectionKernel } from "./inspection.js";
import { hexDigest, type ExecutionPlan } from "./plan.js";
import { clientContractCatalog, type ExecutionInstallation } from "./installation.js";
import { hexDigestAsync, type ExecutionPlan } from "./plan.js";
import type { ClientContractCatalog } from "./installation.js";
import { publicOperationContract, publicQueryContract } from "./client-contracts.js";
/** Explicit public projection of checked inputs. Never serialize conformance
* storage or package ports, or recompile a historical Candidate at runtime. */
export function publicDefinitionSnapshot(
* storage or package ports, or recompile a historical Candidate at runtime.
* The caller supplies that Candidate's checked, cached public catalog. */
export async function publicDefinitionSnapshot(
plan: ExecutionPlan,
installation: ExecutionInstallation,
kernel: InspectionKernel,
catalog: ClientContractCatalog,
) {
const workspace = plan.definition.workspace;
const catalog = clientContractCatalog(
plan,
installation,
installation.packages.map((pkg) => pkg.revision),
kernel,
);
const rows = {
atoms: workspace.atoms.map((atom) => ({ id: atom.id, name: atom.displayName, documentation: atom.documentation })),
interfaces: workspace.interfaceImports.map((iface) => ({
@@ -95,5 +90,5 @@ export function publicDefinitionSnapshot(
{ id: string; [key: string]: unknown }[]
>;
for (const entries of Object.values(publicRows)) entries.sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0));
return { codec: "camino-definitions-v1" as const, id: hexDigest(publicRows), rows: publicRows };
return { codec: "camino-definitions-v1" as const, id: await hexDigestAsync(publicRows), rows: publicRows };
}