mptyl/ThothII
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: harden workspace registry refresh and snapshots

Browse Source
This commit is contained in:
marcopan
2026-08-03 22:33:39 +02:00
parent 2087fbb0c9
commit 553bb41138
4 changed files with 312 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/workspaces/git-repository.ts
+89 -14
View File
@@ -1,6 +1,8 @@
import { execFile } from "node:child_process";
import { constants, lstatSync, mkdirSync, openSync, closeSync, unlinkSync } from "node:fs";
import { access, lstat, mkdir } from "node:fs/promises";
import {
closeSync, constants, lstatSync, mkdirSync, openSync, readFileSync, unlinkSync, writeFileSync,
} from "node:fs";
import { mkdir } from "node:fs/promises";
import { basename, isAbsolute, join } from "node:path";
import { promisify } from "node:util";
import type { WorkspaceErrorCode, WorkspaceRegistryConfig } from "./types.js";
@@ -83,9 +85,14 @@ export class GitWorkspaceRepository {
}
async ensureLayout(): Promise<void> {
for (const path of [this.root, this.snapshotsPath, this.statePath, this.locksPath, this.hooksPath]) {
await mkdir(path, { recursive: true, mode: 0o700 });
assertDirectory(path);
try {
for (const path of [this.root, this.snapshotsPath, this.statePath, this.locksPath, this.hooksPath]) {
await mkdir(path, { recursive: true, mode: 0o700 });
assertDirectory(path);
}
} catch (error) {
if (error instanceof WorkspaceRegistryError) throw error;
throw new WorkspaceRegistryError("git_unavailable", "Workspace registry storage is unavailable");
}
}
@@ -162,11 +169,25 @@ export class GitWorkspaceRepository {
}
private async refresh(): Promise<void> {
if ((await this.git(["status", "--porcelain"])).trim() !== "") {
throw new WorkspaceRegistryError("workspace_stale", "Workspace checkout has local changes");
}
if (this.config.remoteUrl) {
await this.git(["remote", "set-url", "origin", "--", this.config.remoteUrl]);
}
await this.git(["fetch", "--no-tags", "origin", this.config.branch]);
await this.git(["merge", "--ff-only", "FETCH_HEAD"]);
const remoteHead = (await this.git(["rev-parse", "FETCH_HEAD"])).trim();
const localHead = (await this.git(["rev-parse", "HEAD"])).trim();
if (localHead !== remoteHead) {
const commonAncestor = (await this.git(["merge-base", "HEAD", "FETCH_HEAD"])).trim();
if (commonAncestor !== localHead) {
throw new WorkspaceRegistryError("git_non_fast_forward", "Workspace checkout diverged from remote");
}
await this.git(["merge", "--ff-only", "FETCH_HEAD"]);
}
if ((await this.git(["rev-parse", "HEAD"])).trim() !== remoteHead) {
throw new WorkspaceRegistryError("git_non_fast_forward", "Workspace checkout does not match remote");
}
}
private async git(args: string[]): Promise<string> {
@@ -207,18 +228,21 @@ export class WorkspaceRepositoryLock {
this.queue = new Promise<void>((resolve) => { releaseQueue = resolve; });
await previous;
mkdirSync(this.locksPath, { recursive: true, mode: 0o700 });
assertDirectory(this.locksPath);
let descriptor: number | undefined;
const lockPath = join(this.locksPath, "repository.lock");
try {
descriptor = openSync(lockPath, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY, 0o600);
return await operation();
mkdirSync(this.locksPath, { recursive: true, mode: 0o700 });
assertDirectory(this.locksPath);
} catch (error) {
if (typeof error === "object" && error !== null && "code" in error && error.code === "EEXIST") {
throw new WorkspaceRegistryError("workspace_stale", "Workspace registry is busy");
}
throw error;
throw new WorkspaceRegistryError("git_unavailable", "Workspace registry lock is unavailable");
}
try {
descriptor = this.acquire(lockPath);
} catch (error) {
throw this.lockError(error);
}
try {
return await operation();
} finally {
if (descriptor !== undefined) closeSync(descriptor);
if (descriptor !== undefined) {
@@ -227,4 +251,55 @@ export class WorkspaceRepositoryLock {
releaseQueue();
}
}
private acquire(lockPath: string): number {
try {
return this.createProcessLock(lockPath);
} catch (error) {
if (!this.recoverDeadProcessLock(lockPath, error)) throw error;
return this.createProcessLock(lockPath);
}
}
private createProcessLock(lockPath: string): number {
const descriptor = openSync(lockPath, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY, 0o600);
try {
writeFileSync(descriptor, JSON.stringify({ pid: process.pid }), "utf8");
return descriptor;
} catch (error) {
closeSync(descriptor);
try { unlinkSync(lockPath); } catch { /* incomplete lock is never treated as recoverable */ }
throw error;
}
}
private recoverDeadProcessLock(lockPath: string, error: unknown): boolean {
if (!(typeof error === "object" && error !== null && "code" in error && error.code === "EEXIST")) {
return false;
}
try {
const record = JSON.parse(readFileSync(lockPath, "utf8")) as { pid?: unknown };
const pid = record.pid;
if (typeof pid !== "number" || !Number.isSafeInteger(pid) || pid <= 0 || pid === process.pid) return false;
try {
process.kill(pid, 0);
return false;
} catch (probeError) {
if (!(typeof probeError === "object" && probeError !== null && "code" in probeError && probeError.code === "ESRCH")) {
return false;
}
}
unlinkSync(lockPath);
return true;
} catch {
return false;
}
}
private lockError(error: unknown): WorkspaceRegistryError {
if (typeof error === "object" && error !== null && "code" in error && error.code === "EEXIST") {
return new WorkspaceRegistryError("workspace_stale", "Workspace registry is busy");
}
return new WorkspaceRegistryError("git_unavailable", "Workspace registry lock is unavailable");
}
}
backend/src/workspaces/registry.ts
+108 -26
View File
@@ -1,4 +1,4 @@
import { randomUUID } from "node:crypto";
import { createHash, randomUUID } from "node:crypto";
import { lstatSync } from "node:fs";
import { mkdir, readFile, rename, rm, writeFile } from "node:fs/promises";
import { isAbsolute, join } from "node:path";
@@ -31,6 +31,10 @@ interface ActiveState {
revisions: WorkspaceRevision[];
}
interface SnapshotManifest extends ActiveState {
files: Record<string, string>;
}
function workspacePath(id: string): string {
if (!/^[a-z][a-z0-9-]{2,62}$/.test(id)) {
throw new WorkspaceRegistryError("workspace_invalid", "Workspace ID is invalid");
@@ -45,6 +49,17 @@ function safeCommit(commit: string): string {
return commit;
}
function safeBlob(blob: string): string {
if (!/^[0-9a-f]{40}$/.test(blob)) {
throw new WorkspaceRegistryError("workspace_invalid", "Workspace snapshot blob is invalid");
}
return blob;
}
function digest(contents: string | Buffer): string {
return createHash("sha256").update(contents).digest("hex");
}
function workspaceError(error: unknown): WorkspaceRegistryError {
if (error instanceof WorkspaceRegistryError) return error;
return new WorkspaceRegistryError("workspace_invalid", "Workspace repository content is invalid");
@@ -136,20 +151,32 @@ export class WorkspaceRegistry {
}
const snapshotDirectory = join(this.repository.snapshotsPath, safeHead);
if (!this.pathExists(snapshotDirectory)) {
const revisions = snapshots.map((snapshot) => ({
id: snapshot.id,
commit: safeHead,
blob: snapshot.blob,
snapshotPath: this.snapshotPath(safeHead, snapshot.id),
}));
if (this.pathExists(snapshotDirectory)) {
await this.assertSnapshotIntegrity({ head: safeHead, revisions });
} else {
const staging = join(this.repository.snapshotsPath, `.staging-${randomUUID()}`);
await mkdir(staging, { mode: 0o700 });
try {
const revisions: WorkspaceRevision[] = [];
const files: Record<string, string> = {};
for (const snapshot of snapshots) {
const path = join(staging, `${snapshot.id}.yaml`);
const yamlName = `${snapshot.id}.yaml`;
const envName = `${snapshot.id}.env.example`;
const docsName = `${snapshot.id}.md`;
const docs = renderWorkspaceDocs(snapshot.workspace);
await writeFile(path, snapshot.source, { encoding: "utf8", mode: 0o400 });
await writeFile(join(staging, `${snapshot.id}.env.example`), docs.envExample, { encoding: "utf8", mode: 0o400 });
await writeFile(join(staging, `${snapshot.id}.md`), docs.markdown, { encoding: "utf8", mode: 0o400 });
revisions.push({ id: snapshot.id, commit: safeHead, blob: snapshot.blob, snapshotPath: this.snapshotPath(safeHead, snapshot.id) });
await writeFile(join(staging, yamlName), snapshot.source, { encoding: "utf8", mode: 0o400 });
await writeFile(join(staging, envName), docs.envExample, { encoding: "utf8", mode: 0o400 });
await writeFile(join(staging, docsName), docs.markdown, { encoding: "utf8", mode: 0o400 });
files[yamlName] = digest(snapshot.source);
files[envName] = digest(docs.envExample);
files[docsName] = digest(docs.markdown);
}
await writeFile(join(staging, "snapshot.json"), JSON.stringify({ head: safeHead, revisions }), {
await writeFile(join(staging, "snapshot.json"), JSON.stringify({ head: safeHead, revisions, files }), {
encoding: "utf8", mode: 0o400,
});
await rename(staging, snapshotDirectory);
@@ -159,12 +186,6 @@ export class WorkspaceRegistry {
}
}
const revisions = snapshots.map((snapshot) => ({
id: snapshot.id,
commit: safeHead,
blob: snapshot.blob,
snapshotPath: this.snapshotPath(safeHead, snapshot.id),
}));
await this.writeActiveState({ head: safeHead, revisions });
}
@@ -193,18 +214,13 @@ export class WorkspaceRegistry {
const file = join(this.repository.statePath, "active.json");
try {
const state = JSON.parse(await readFile(file, "utf8")) as ActiveState;
safeCommit(state.head);
if (!Array.isArray(state.revisions) || state.revisions.length === 0) throw new Error("bad state");
for (const revision of state.revisions) {
safeCommit(revision.commit);
workspacePath(revision.id);
if (!isAbsolute(revision.snapshotPath) || revision.snapshotPath !== this.snapshotPath(revision.commit, revision.id)) {
throw new Error("bad snapshot path");
}
}
this.assertActiveState(state);
await this.assertSnapshotIntegrity(state);
return state;
} catch {
return undefined;
} catch (error) {
if (this.pathIsMissing(file)) return undefined;
if (error instanceof WorkspaceRegistryError) throw error;
throw new WorkspaceRegistryError("workspace_invalid", "Workspace active snapshot is invalid");
}
}
@@ -215,6 +231,63 @@ export class WorkspaceRegistry {
await rename(staging, target);
}
private assertActiveState(state: ActiveState): void {
safeCommit(state.head);
if (!Array.isArray(state.revisions) || state.revisions.length === 0) throw new Error("bad state");
const ids = new Set<string>();
for (const revision of state.revisions) {
safeCommit(revision.commit);
safeBlob(revision.blob);
if (revision.commit !== state.head || ids.has(revision.id)) throw new Error("bad revision");
ids.add(revision.id);
workspacePath(revision.id);
if (!isAbsolute(revision.snapshotPath) || revision.snapshotPath !== this.snapshotPath(revision.commit, revision.id)) {
throw new Error("bad snapshot path");
}
}
}
private async assertSnapshotIntegrity(state: ActiveState): Promise<void> {
const directory = join(this.repository.snapshotsPath, state.head);
const manifestPath = join(directory, "snapshot.json");
try {
const manifest = JSON.parse(await readFile(manifestPath, "utf8")) as SnapshotManifest;
this.assertActiveState(manifest);
if (manifest.head !== state.head || !this.sameRevisions(manifest.revisions, state.revisions)) {
throw new Error("manifest revisions do not match active state");
}
const expected = state.revisions.flatMap((revision) => [
`${revision.id}.yaml`, `${revision.id}.env.example`, `${revision.id}.md`,
]);
if (Object.keys(manifest.files).length !== expected.length || !expected.every((name) => (
/^[0-9a-f]{64}$/.test(manifest.files[name] ?? "")
))) throw new Error("manifest files are invalid");
for (const name of expected) {
const path = join(directory, name);
const entry = lstatSync(path);
if (!entry.isFile() || entry.isSymbolicLink()) throw new Error("snapshot file is invalid");
const contents = await readFile(path);
if (digest(contents) !== manifest.files[name]) throw new Error("snapshot file does not match manifest");
if (name.endsWith(".yaml")) {
const workspace = parseWorkspaceYaml(contents.toString("utf8"));
if (workspace.workspace.id !== name.slice(0, -".yaml".length)) throw new Error("snapshot workspace is invalid");
}
}
} catch (error) {
if (error instanceof WorkspaceRegistryError) throw error;
throw new WorkspaceRegistryError("workspace_invalid", "Workspace snapshot integrity check failed");
}
}
private sameRevisions(left: WorkspaceRevision[], right: WorkspaceRevision[]): boolean {
return left.length === right.length && left.every((revision, index) => {
const candidate = right[index];
return candidate !== undefined
&& candidate.id === revision.id && candidate.commit === revision.commit
&& candidate.blob === revision.blob && candidate.snapshotPath === revision.snapshotPath;
});
}
private pathExists(path: string): boolean {
try {
const entry = lstatSync(path);
@@ -227,4 +300,13 @@ export class WorkspaceRegistry {
return false;
}
}
private pathIsMissing(path: string): boolean {
try {
lstatSync(path);
return false;
} catch {
return true;
}
}
}