refactor: share registry runtime configuration leases
This commit is contained in:
@@ -0,0 +1,322 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import {
|
||||
chmodSync, closeSync, constants as fsConstants, fchmodSync, fstatSync, fsyncSync, lstatSync,
|
||||
existsSync, mkdirSync, openSync, readFileSync, renameSync, unlinkSync,
|
||||
writeSync,
|
||||
} from "node:fs";
|
||||
import { dirname, isAbsolute, join, relative, resolve } from "node:path";
|
||||
import { parseAllDocuments } from "yaml";
|
||||
import { resolveRuntimeBindings } from "./bindings.js";
|
||||
import {
|
||||
renderRuntimeConfig,
|
||||
type RuntimeInstallationOverlay,
|
||||
type RuntimePaths,
|
||||
type SemanticRuntimeConfig,
|
||||
} from "./runtime-renderer.js";
|
||||
import { parseWorkspaceYaml, validateOperationalWorkspace, type WorkspaceDescriptor } from "./schema.js";
|
||||
|
||||
export interface RuntimeConfigLease {
|
||||
path: string;
|
||||
manifestPath: string;
|
||||
workspaceId: string;
|
||||
workspaceRevision: string;
|
||||
release(): void;
|
||||
}
|
||||
|
||||
export interface MaintenanceRuntimeInput {
|
||||
/** The immutable registry snapshot. `workspaceConfigPath` is accepted for callers using that name. */
|
||||
snapshotPath?: string;
|
||||
workspaceConfigPath?: string;
|
||||
}
|
||||
|
||||
export interface WorkspaceRuntimeConfigLeaseFactoryInput {
|
||||
dataRoot: string;
|
||||
runtimeSnapshotRoot: string;
|
||||
harnessDir: string;
|
||||
configPath: string;
|
||||
secretRoots?: readonly string[];
|
||||
env?: NodeJS.ProcessEnv;
|
||||
installationOverlay?: RuntimeInstallationOverlay;
|
||||
semanticRuntime: SemanticRuntimeConfig;
|
||||
}
|
||||
|
||||
interface SnapshotIdentity {
|
||||
workspace: WorkspaceDescriptor;
|
||||
workspaceId: string;
|
||||
workspaceRevision: string;
|
||||
revisionContentRoot: string;
|
||||
digest: string;
|
||||
}
|
||||
interface PublishedIdentity {
|
||||
path: string;
|
||||
manifestPath: string;
|
||||
workspaceId: string;
|
||||
workspaceRevision: string;
|
||||
digest: string;
|
||||
content: string;
|
||||
manifest: string;
|
||||
refs: number;
|
||||
}
|
||||
|
||||
const DEFAULT_SEMANTIC_RUNTIME: SemanticRuntimeConfig = {
|
||||
internalQdrantUrl: "http://qdrant:6333", internalEmbeddingUrl: "http://embedding:11434",
|
||||
internalEmbeddingModel: "qwen3-embedding:0.6b", internalEmbeddingDimensions: 1024,
|
||||
};
|
||||
|
||||
/** Normalize the installation HTTP private-host policy once, before rendering. */
|
||||
export function normalizePrivateHostAllowlist(value: string | readonly string[] | undefined): string[] {
|
||||
const values: readonly string[] = value === undefined ? [] : typeof value === "string" ? (value === "" ? [] : value.split(",")) : [...value];
|
||||
if (values.length > 32) throw new Error("HTTP private host allowlist has too many entries");
|
||||
const result: string[] = [];
|
||||
for (const host of values) {
|
||||
if (
|
||||
typeof host !== "string" || host.length === 0 || host.length > 253 || host !== host.toLowerCase()
|
||||
|| host.endsWith(".") || host.includes(" ") || host.includes("\t")
|
||||
|| host.includes("*") || host.includes("/") || host.includes("_")
|
||||
|| /^[0-9.]+$/.test(host) || host.includes(":")
|
||||
) throw new Error("HTTP private host allowlist contains an invalid hostname");
|
||||
const labels = host.split(".");
|
||||
if (labels.some((label) => label.length === 0 || label.length > 63 || !/^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$/.test(label))) {
|
||||
throw new Error("HTTP private host allowlist contains an invalid hostname");
|
||||
}
|
||||
if (result.includes(host)) throw new Error("HTTP private host allowlist contains a duplicate hostname");
|
||||
result.push(host);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
export const normalizeHttpPrivateHostAllowlist = normalizePrivateHostAllowlist;
|
||||
|
||||
function parseInstallationOverlay(path: string): RuntimeInstallationOverlay {
|
||||
if (!isAbsolute(path) || !existsSync(path)) return {};
|
||||
const docs = parseAllDocuments(readFileSync(path, "utf8"), { uniqueKeys: true });
|
||||
if (docs.length !== 1 || docs[0].errors.length || docs[0].warnings.length) {
|
||||
throw new Error("installation config contains invalid YAML");
|
||||
}
|
||||
const value = docs[0].toJSON();
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) throw new Error("installation config must be a YAML mapping");
|
||||
const source = value as Record<string, unknown>;
|
||||
return {
|
||||
...(source.session_storage === undefined ? {} : { session_storage: source.session_storage }),
|
||||
...(source.profile === undefined ? {} : { profile: source.profile }),
|
||||
};
|
||||
}
|
||||
|
||||
function isSafeMode(mode: number, expected: number): boolean {
|
||||
return (mode & 0o777) === expected;
|
||||
}
|
||||
function digest(data: string | Buffer): string { return createHash("sha256").update(data).digest("hex"); }
|
||||
function regularNoLink(path: string, mode?: number): ReturnType<typeof lstatSync> {
|
||||
const entry = lstatSync(path);
|
||||
if (!entry.isFile() || entry.isSymbolicLink() || entry.nlink !== 1 || (mode !== undefined && !isSafeMode(entry.mode, mode))) {
|
||||
throw new Error("runtime config destination is not trusted");
|
||||
}
|
||||
return entry;
|
||||
}
|
||||
function fsyncDirectory(path: string): void {
|
||||
const fd = openSync(path, fsConstants.O_RDONLY | fsConstants.O_DIRECTORY);
|
||||
try { fsyncSync(fd); } finally { closeSync(fd); }
|
||||
}
|
||||
|
||||
export class WorkspaceRuntimeConfigLeaseFactory {
|
||||
private readonly env: NodeJS.ProcessEnv;
|
||||
private readonly secretRoots: readonly string[];
|
||||
private readonly installation: RuntimeInstallationOverlay;
|
||||
private readonly published = new Map<string, PublishedIdentity>();
|
||||
|
||||
constructor(private readonly input: WorkspaceRuntimeConfigLeaseFactoryInput) {
|
||||
if (!isAbsolute(input.dataRoot) || !isAbsolute(input.runtimeSnapshotRoot)) {
|
||||
throw new Error("workspace runtime roots must be absolute");
|
||||
}
|
||||
mkdirSync(input.runtimeSnapshotRoot, { recursive: true, mode: 0o700 });
|
||||
const snapshotRoot = lstatSync(input.runtimeSnapshotRoot);
|
||||
if (!snapshotRoot.isDirectory() || snapshotRoot.isSymbolicLink() || (snapshotRoot.mode & 0o077) !== 0) {
|
||||
throw new Error("runtime snapshot root is not trusted");
|
||||
}
|
||||
this.env = { ...(input.env ?? process.env) };
|
||||
this.secretRoots = [...(input.secretRoots ?? [])];
|
||||
const configPath = isAbsolute(input.configPath) ? input.configPath : resolve(input.harnessDir, input.configPath);
|
||||
const suppliedAllowlist = input.installationOverlay?.egress?.http_private_host_allowlist;
|
||||
this.installation = {
|
||||
...parseInstallationOverlay(configPath),
|
||||
...(input.installationOverlay ?? {}),
|
||||
egress: { http_private_host_allowlist: normalizePrivateHostAllowlist(
|
||||
suppliedAllowlist ?? this.env.THT_HTTP_PRIVATE_HOST_ALLOWLIST,
|
||||
) },
|
||||
} as RuntimeInstallationOverlay;
|
||||
}
|
||||
|
||||
acquireSession(snapshotPath: string): RuntimeConfigLease { return this.acquire(snapshotPath); }
|
||||
acquireMaintenance(input: MaintenanceRuntimeInput): RuntimeConfigLease {
|
||||
const snapshotPath = input.snapshotPath ?? input.workspaceConfigPath;
|
||||
if (!snapshotPath) throw new Error("maintenance runtime snapshot is required");
|
||||
return this.acquire(snapshotPath);
|
||||
}
|
||||
|
||||
private acquire(snapshotPath: string): RuntimeConfigLease {
|
||||
const snapshot = this.readSnapshot(snapshotPath);
|
||||
const paths = this.runtimePaths(snapshot.workspaceId);
|
||||
const rendered = renderRuntimeConfig(
|
||||
snapshot.workspace,
|
||||
resolveRuntimeBindings(snapshot.workspace, this.env, this.secretRoots),
|
||||
paths,
|
||||
snapshot,
|
||||
this.installation,
|
||||
this.input.semanticRuntime ?? DEFAULT_SEMANTIC_RUNTIME,
|
||||
);
|
||||
const configPath = join(this.input.dataRoot, "sessions", snapshot.workspaceId, "preprocessing", "runtime-config", `${snapshot.workspaceRevision}.yaml`);
|
||||
const manifestPath = join(dirname(configPath), `${snapshot.workspaceRevision}.manifest.json`);
|
||||
const renderedDigest = digest(rendered);
|
||||
const existing = this.published.get(configPath);
|
||||
if (existing) {
|
||||
if (existing.digest !== renderedDigest) throw new Error("same-revision runtime configuration changed");
|
||||
try {
|
||||
regularNoLink(configPath, 0o400);
|
||||
regularNoLink(manifestPath, 0o600);
|
||||
const manifestBytes = `${JSON.stringify({
|
||||
workspace_id: snapshot.workspaceId,
|
||||
workspace_revision: snapshot.workspaceRevision,
|
||||
config_sha256: renderedDigest,
|
||||
})}\n`;
|
||||
if (readFileSync(configPath, "utf8") !== rendered || readFileSync(manifestPath, "utf8") !== manifestBytes) {
|
||||
throw new Error("same-revision runtime configuration changed");
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof Error && /same-revision/.test(error.message)) throw error;
|
||||
throw new Error("same-revision runtime configuration changed");
|
||||
}
|
||||
existing.refs += 1;
|
||||
return this.lease(existing);
|
||||
}
|
||||
this.ensureDestinationDirectory(dirname(configPath));
|
||||
this.publish(configPath, manifestPath, rendered, {
|
||||
workspace_id: snapshot.workspaceId,
|
||||
workspace_revision: snapshot.workspaceRevision,
|
||||
config_sha256: renderedDigest,
|
||||
});
|
||||
const identity: PublishedIdentity = {
|
||||
path: configPath, manifestPath, workspaceId: snapshot.workspaceId,
|
||||
workspaceRevision: snapshot.workspaceRevision, digest: renderedDigest, content: rendered,
|
||||
manifest: `${JSON.stringify({ workspace_id: snapshot.workspaceId, workspace_revision: snapshot.workspaceRevision, config_sha256: renderedDigest })}\n`, refs: 1,
|
||||
};
|
||||
this.published.set(configPath, identity);
|
||||
return this.lease(identity);
|
||||
}
|
||||
|
||||
private lease(identity: PublishedIdentity): RuntimeConfigLease {
|
||||
let released = false;
|
||||
return {
|
||||
path: identity.path, manifestPath: identity.manifestPath,
|
||||
workspaceId: identity.workspaceId, workspaceRevision: identity.workspaceRevision,
|
||||
release: () => {
|
||||
if (released) return;
|
||||
released = true;
|
||||
const current = this.published.get(identity.path);
|
||||
if (!current) return;
|
||||
current.refs -= 1;
|
||||
if (current.refs > 0) return;
|
||||
this.published.delete(identity.path);
|
||||
this.removeIfUnchanged(identity.manifestPath, identity.manifest, 0o600);
|
||||
this.removeIfUnchanged(identity.path, identity.content, 0o400);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private removeIfUnchanged(path: string, expected: string, mode: number): void {
|
||||
try {
|
||||
regularNoLink(path, mode);
|
||||
if (readFileSync(path, "utf8") === expected) unlinkSync(path);
|
||||
} catch { /* never remove a replaced or untrusted destination */ }
|
||||
}
|
||||
|
||||
private runtimePaths(workspaceId: string): RuntimePaths {
|
||||
const root = join(this.input.dataRoot, "sessions", workspaceId);
|
||||
return { sessions: join(root, "sessions"), artifacts: join(root, "artifacts"), indexes: join(root, "indexes") };
|
||||
}
|
||||
|
||||
private snapshotRoots(): string[] {
|
||||
return [this.input.runtimeSnapshotRoot, dirname(this.input.runtimeSnapshotRoot)];
|
||||
}
|
||||
|
||||
private readSnapshot(path: string): SnapshotIdentity {
|
||||
if (!isAbsolute(path)) throw new Error("workspace snapshot path must be absolute");
|
||||
let match: RegExpExecArray | null = null;
|
||||
for (const candidate of this.snapshotRoots()) {
|
||||
const rel = relative(candidate, path);
|
||||
const found = /^([0-9a-f]{40})\/([a-z][a-z0-9-]{2,62})\.yaml$/.exec(rel);
|
||||
if (found && !rel.startsWith("..") && !isAbsolute(rel)) { match = found; break; }
|
||||
}
|
||||
if (!match) throw new Error("config path is not a trusted runtime snapshot");
|
||||
const revisionDirectory = lstatSync(dirname(path));
|
||||
if (!revisionDirectory.isDirectory() || revisionDirectory.isSymbolicLink()) {
|
||||
throw new Error("workspace snapshot parent is not trusted");
|
||||
}
|
||||
const fd = openSync(path, fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW);
|
||||
try {
|
||||
const before = fstatSync(fd);
|
||||
if (!before.isFile() || before.nlink !== 1) throw new Error("workspace snapshot is not a trusted file");
|
||||
const source = readFileSync(fd, "utf8");
|
||||
const after = fstatSync(fd);
|
||||
if (before.dev !== after.dev || before.ino !== after.ino || before.size !== after.size) throw new Error("workspace snapshot changed while reading");
|
||||
const workspace = validateOperationalWorkspace(parseWorkspaceYaml(source));
|
||||
if (workspace.workspace.id !== match[2]) throw new Error("workspace snapshot identity does not match its path");
|
||||
return { workspace, workspaceId: match[2], workspaceRevision: match[1], revisionContentRoot: dirname(path), digest: digest(source) };
|
||||
} finally { closeSync(fd); }
|
||||
}
|
||||
|
||||
private ensureDestinationDirectory(path: string): void {
|
||||
const root = this.input.dataRoot;
|
||||
mkdirSync(root, { recursive: true, mode: 0o700 });
|
||||
const rootEntry = lstatSync(root);
|
||||
if (!rootEntry.isDirectory() || rootEntry.isSymbolicLink()) {
|
||||
throw new Error("runtime config destination is not trusted");
|
||||
}
|
||||
chmodSync(root, 0o700);
|
||||
const components = relative(root, path).split("/").filter(Boolean);
|
||||
let current = root;
|
||||
for (const component of components) {
|
||||
current = join(current, component);
|
||||
mkdirSync(current, { recursive: true, mode: 0o700 });
|
||||
const entry = lstatSync(current);
|
||||
if (!entry.isDirectory() || entry.isSymbolicLink() || (entry.mode & 0o077) !== 0) throw new Error("runtime config destination is not trusted");
|
||||
}
|
||||
}
|
||||
|
||||
private publish(path: string, manifestPath: string, content: string, manifest: Record<string, unknown>): void {
|
||||
const manifestBytes = `${JSON.stringify(manifest)}\n`;
|
||||
const configExists = (() => { try { regularNoLink(path, 0o400); return true; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; throw error; } })();
|
||||
const manifestExists = (() => { try { regularNoLink(manifestPath, 0o600); return true; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; throw error; } })();
|
||||
if (configExists || manifestExists) {
|
||||
if (!configExists || !manifestExists || readFileSync(path, "utf8") !== content || readFileSync(manifestPath, "utf8") !== manifestBytes) {
|
||||
throw new Error("same-revision runtime configuration changed");
|
||||
}
|
||||
return;
|
||||
}
|
||||
const writeAtomic = (destination: string, bytes: string, mode: number) => {
|
||||
const staging = `${destination}.staging-${randomUUID()}`;
|
||||
const fd = openSync(staging, fsConstants.O_WRONLY | fsConstants.O_CREAT | fsConstants.O_EXCL | fsConstants.O_NOFOLLOW, 0o600);
|
||||
try {
|
||||
try {
|
||||
writeSync(fd, bytes, undefined, "utf8");
|
||||
fchmodSync(fd, mode); fsyncSync(fd);
|
||||
} catch (error) {
|
||||
try { unlinkSync(staging); } catch { /* retain the original durability error */ }
|
||||
throw error;
|
||||
} finally { closeSync(fd); }
|
||||
} catch (error) {
|
||||
try { unlinkSync(staging); } catch { /* retain the original durability error */ }
|
||||
throw error;
|
||||
}
|
||||
try { renameSync(staging, destination); fsyncDirectory(dirname(destination)); }
|
||||
catch (error) { try { unlinkSync(staging); } catch { /* preserve original failure */ } throw error; }
|
||||
regularNoLink(destination, mode);
|
||||
};
|
||||
try { writeAtomic(path, content, 0o400); writeAtomic(manifestPath, manifestBytes, 0o600); }
|
||||
catch (error) {
|
||||
this.removeIfUnchanged(path, content, 0o400);
|
||||
this.removeIfUnchanged(manifestPath, manifestBytes, 0o600);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -24,6 +24,7 @@ export interface RuntimeRenderContext extends RuntimeIdentity {
|
||||
export interface RuntimeInstallationOverlay {
|
||||
session_storage?: unknown;
|
||||
profile?: unknown;
|
||||
egress?: { http_private_host_allowlist: readonly string[] };
|
||||
}
|
||||
|
||||
export interface SemanticRuntimeConfig {
|
||||
@@ -253,6 +254,7 @@ export function renderRuntimeConfig(
|
||||
...(installation.session_storage === undefined
|
||||
? {} : { session_storage: installation.session_storage }),
|
||||
...(installation.profile === undefined ? {} : { profile: installation.profile }),
|
||||
...(installation.egress === undefined ? {} : { egress: installation.egress }),
|
||||
language: descriptor.workspace.language,
|
||||
database,
|
||||
vectors: {
|
||||
|
||||
Reference in New Issue
Block a user