|
|
|
@@ -1,52 +1,121 @@
|
|
|
|
|
import { createHash, randomBytes } from "node:crypto";
|
|
|
|
|
import { mkdir, readFile, rename, writeFile } from "node:fs/promises";
|
|
|
|
|
import { join } from "node:path";
|
|
|
|
|
import { constants as fsConstants } from "node:fs";
|
|
|
|
|
import { lstat, mkdir, open, readFile, readdir, rename, rm, stat, unlink, writeFile } from "node:fs/promises";
|
|
|
|
|
import { dirname, join } from "node:path";
|
|
|
|
|
import type { CanonicalWorkspaceId } from "./workspace-lock-root-lease.js";
|
|
|
|
|
import type { BorrowedWorkspaceSessionReadersExclusiveLockLease, OrderedWorkspaceWriterCapabilitySet } from "./preprocessing-state.js";
|
|
|
|
|
export interface BorrowedOrderedWorkspaceWriterLeaseV1 { readonly workspaceId: CanonicalWorkspaceId; readonly rootLease: unknown; readonly writerCapability: unknown; }
|
|
|
|
|
|
|
|
|
|
/** Addressed registry publication is deliberately data-only: callers provide identities and the
|
|
|
|
|
* executor persists every transition before invoking a side effect. */
|
|
|
|
|
export type Revision40 = string & { readonly __revision40: unique symbol };
|
|
|
|
|
export type Sha256Hex = string & { readonly __sha256: unique symbol };
|
|
|
|
|
export type RegistryRunId32 = string & { readonly __registryRunId32: unique symbol };
|
|
|
|
|
export type RegistryAddressedOperationV1 = "registry_bootstrap" | "registry_pull";
|
|
|
|
|
export type RegistryAddressedPublicationPhaseV1 = "request_claimed" | "target_advertised" | "target_fetched" | "planned" | "participants_prepared" | "publication_intent_durable" | "target_published" | "terminal_durable";
|
|
|
|
|
export type RegistryAddressedJobArtifactPathV1 = `addressed-publication-jobs/${RegistryRunId32}.json`;
|
|
|
|
|
export interface RegistryWorkspaceManifestIdentityV1 { readonly workspaceId: CanonicalWorkspaceId; readonly revision: Revision40; readonly descriptorBlob: Revision40; readonly manifestSha256: Sha256Hex; }
|
|
|
|
|
export interface RegistryActiveSnapshotV1 { readonly schemaVersion: 1; readonly commit: Revision40; readonly manifestSha256: Sha256Hex; readonly workspaces: readonly RegistryWorkspaceManifestIdentityV1[]; }
|
|
|
|
|
interface PlanFields { readonly schemaVersion: 1; readonly installationIdentitySha256: Sha256Hex; readonly repositoryIdentitySha256: Sha256Hex; readonly remoteRefIdentitySha256: Sha256Hex; readonly jobArtifactPath: RegistryAddressedJobArtifactPathV1; readonly advertisedTargetCommit: Revision40; readonly immutableTargetRef: `refs/thoth/addressed-runs/${RegistryRunId32}/target`; readonly fetchedTargetCommit: Revision40; readonly targetCommit: Revision40; readonly targetManifestSha256: Sha256Hex; readonly targetWorkspaces: readonly RegistryWorkspaceManifestIdentityV1[]; readonly changedWorkspaceIds: readonly CanonicalWorkspaceId[]; readonly changedSetSha256: Sha256Hex; }
|
|
|
|
|
export interface RegistryBootstrapAddressedPlanV1 extends PlanFields { readonly operation: "registry_bootstrap"; readonly changedSetRule: "all_target_workspace_ids"; readonly baseCommit: null; readonly baseManifestSha256: null; readonly baseWorkspaces: readonly []; }
|
|
|
|
|
export interface RegistryPullAddressedPlanV1 extends PlanFields { readonly operation: "registry_pull"; readonly changedSetRule: "symmetric_base_target_workspace_difference"; readonly baseCommit: Revision40; readonly baseManifestSha256: Sha256Hex; readonly baseWorkspaces: readonly RegistryWorkspaceManifestIdentityV1[]; }
|
|
|
|
|
export type RegistryAddressedPlanV1 = RegistryBootstrapAddressedPlanV1 | RegistryPullAddressedPlanV1;
|
|
|
|
|
interface StateFields { readonly schemaVersion: 1; readonly runId: RegistryRunId32; readonly requestSha256: Sha256Hex; readonly jobArtifactPath: RegistryAddressedJobArtifactPathV1; readonly phase: RegistryAddressedPublicationPhaseV1; readonly installationIdentitySha256: Sha256Hex; readonly repositoryIdentitySha256: Sha256Hex; readonly remoteRefIdentitySha256: Sha256Hex; readonly advertisedTargetCommit: Revision40 | null; readonly immutableTargetRef: `refs/thoth/addressed-runs/${RegistryRunId32}/target` | null; readonly fetchedTargetCommit: Revision40 | null; readonly targetCommit: Revision40 | null; readonly targetManifestSha256: Sha256Hex | null; readonly targetWorkspaces: readonly RegistryWorkspaceManifestIdentityV1[] | null; readonly changedWorkspaceIds: readonly CanonicalWorkspaceId[] | null; readonly planSha256: Sha256Hex | null; readonly changedSetSha256: Sha256Hex | null; readonly participantsSha256: Sha256Hex | null; readonly synchronizersSha256: Sha256Hex | null; readonly publicationIntentSha256: Sha256Hex | null; readonly publishedActiveStateSha256: Sha256Hex | null; readonly terminalResultSha256: Sha256Hex | null; readonly priorStateSha256: Sha256Hex | null; readonly baseCommit: Revision40 | null; readonly baseManifestSha256: Sha256Hex | null; readonly baseWorkspaces: readonly RegistryWorkspaceManifestIdentityV1[]; readonly changedSetRule: "all_target_workspace_ids" | "symmetric_base_target_workspace_difference" | null; }
|
|
|
|
|
export interface RegistryAddressedSnapshotV1 extends RegistryActiveSnapshotV1 { readonly requestDigest: string; }
|
|
|
|
|
export interface RegistryInstallationIdentityV1 { readonly installationId: string; readonly digest: string; }
|
|
|
|
|
export interface RegistryRepositoryIdentityV1 { readonly remote: string; readonly branch: string; readonly head: string; readonly digest: string; }
|
|
|
|
|
export interface RegistryRemoteIdentityV1 { readonly remote: string; readonly head: string; readonly digest: string; }
|
|
|
|
|
export interface RegistryRecoveryIdentityV1 { readonly runId: string; readonly requestDigest: string; readonly digest: string; }
|
|
|
|
|
interface RegistryAddressedRequestFieldsV1 { readonly installation: RegistryInstallationIdentityV1; readonly repository: RegistryRepositoryIdentityV1; readonly remote: RegistryRemoteIdentityV1; readonly workspaceIds: readonly string[]; readonly requestDigest: string; }
|
|
|
|
|
export interface RegistryBootstrapAddressedRequestV1 extends RegistryAddressedRequestFieldsV1 { readonly kind: "bootstrap"; }
|
|
|
|
|
export interface RegistryPublishAddressedRequestV1 extends RegistryAddressedRequestFieldsV1 { readonly kind: "publish"; readonly target: string; }
|
|
|
|
|
export type RegistryAddressedRequestV1 = RegistryBootstrapAddressedRequestV1 | RegistryPublishAddressedRequestV1;
|
|
|
|
|
export interface RegistryAddressedSnapshotV1 { readonly schemaVersion: 1; readonly commit: string | null; readonly baseCommit: string | null; readonly baseDigest: string | null; readonly workspaceIds: readonly string[]; readonly changedWorkspaceIds: readonly string[]; readonly installation: RegistryInstallationIdentityV1; readonly repository: RegistryRepositoryIdentityV1; readonly remote: RegistryRemoteIdentityV1; readonly requestDigest: string; readonly recovery?: RegistryRecoveryIdentityV1; }
|
|
|
|
|
export interface RegistryAddressedPublicationStateV1 extends RegistryAddressedSnapshotV1 { readonly runId: string; readonly phase: "request_claimed" | "target_advertised" | "target_fetched" | "planned" | "terminal"; readonly result?: unknown; readonly updatedAt: string; }
|
|
|
|
|
export interface RegistryBootstrapAddressedRequestV1 { readonly kind: "bootstrap"; readonly installation: RegistryInstallationIdentityV1; readonly repository: RegistryRepositoryIdentityV1; readonly remote: RegistryRemoteIdentityV1; readonly workspaceIds: readonly string[]; readonly requestDigest: string; }
|
|
|
|
|
export interface RegistryPublishAddressedRequestV1 { readonly kind: "publish"; readonly installation: RegistryInstallationIdentityV1; readonly repository: RegistryRepositoryIdentityV1; readonly remote: RegistryRemoteIdentityV1; readonly workspaceIds: readonly string[]; readonly requestDigest: string; readonly target?: string; }
|
|
|
|
|
export interface RegistryAddressedPublicationResultV1 { readonly runId: string; readonly snapshot: RegistryAddressedSnapshotV1; readonly result?: unknown; }
|
|
|
|
|
export type RegistryEnsureBootstrapAddressedResultV1 =
|
|
|
|
|
| { readonly kind: "already_active"; readonly snapshot: RegistryAddressedSnapshotV1 }
|
|
|
|
|
| { readonly kind: "bootstrap_terminal"; readonly result: RegistryAddressedPublicationResultV1; readonly snapshot: RegistryAddressedSnapshotV1 };
|
|
|
|
|
|
|
|
|
|
/** Explicit request variants used by callers that already own a durable run. */
|
|
|
|
|
export interface RegistryAddressedBootstrapCreateRequestV1 extends RegistryBootstrapAddressedRequestV1 { readonly mode: "create"; }
|
|
|
|
|
export interface RegistryAddressedBootstrapResumeRequestV1 extends RegistryBootstrapAddressedRequestV1 { readonly mode: "resume"; readonly runId: string; }
|
|
|
|
|
export interface RegistryAddressedPublishCreateRequestV1 extends RegistryPublishAddressedRequestV1 { readonly mode: "create"; }
|
|
|
|
|
export interface RegistryAddressedPublishResumeRequestV1 extends RegistryPublishAddressedRequestV1 { readonly mode: "resume"; readonly runId: string; }
|
|
|
|
|
export type RegistryAddressedCreateRequestV1 = RegistryAddressedBootstrapCreateRequestV1 | RegistryAddressedPublishCreateRequestV1;
|
|
|
|
|
export type RegistryAddressedResumeRequestV1 = RegistryAddressedBootstrapResumeRequestV1 | RegistryAddressedPublishResumeRequestV1;
|
|
|
|
|
export type RegistryAddressedCreateResultV1 = RegistryAddressedPublicationResultV1;
|
|
|
|
|
export type RegistryAddressedResumeResultV1 = RegistryAddressedPublicationResultV1;
|
|
|
|
|
export const REGISTRY_SCAN_LIMITS_V1 = Object.freeze({ entries: 128, bytes: 1024 * 1024, fileBytes: 128 * 1024 });
|
|
|
|
|
export const registryDigest = (value: unknown): string => createHash("sha256").update(JSON.stringify(value)).digest("hex");
|
|
|
|
|
export function canonicalBootstrapRequestDigest(request: { readonly kind: "bootstrap" | "publish"; readonly installation: RegistryInstallationIdentityV1; readonly repository: RegistryRepositoryIdentityV1; readonly remote: RegistryRemoteIdentityV1; readonly workspaceIds: readonly string[] }): string { return registryDigest({ kind: request.kind, installation: request.installation, repository: request.repository, remote: request.remote, workspaceIds: [...request.workspaceIds].sort() }); }
|
|
|
|
|
export function addressedRunId(): string { return randomBytes(16).toString("hex"); }
|
|
|
|
|
function safeRunId(v: string): void { if (!/^[0-9a-f]{32}$/.test(v)) throw new Error("preprocessing_conflict"); }
|
|
|
|
|
export class RegistryAddressedPublicationStore {
|
|
|
|
|
constructor(readonly root: string) {}
|
|
|
|
|
private path(runId: string): string { safeRunId(runId); return join(this.root, "addressed-publication-jobs", `${runId}.json`); }
|
|
|
|
|
async claim(request: RegistryAddressedRequestV1, runId = addressedRunId()): Promise<RegistryAddressedPublicationStateV1> {
|
|
|
|
|
safeRunId(runId); await mkdir(join(this.root, "addressed-publication-jobs"), { recursive: true, mode: 0o700 });
|
|
|
|
|
const now = new Date().toISOString(); const snapshot = this.snapshot(request); const state: RegistryAddressedPublicationStateV1 = { ...snapshot, runId, phase: "request_claimed", updatedAt: now };
|
|
|
|
|
try { await writeFile(this.path(runId), `${JSON.stringify(state)}\n`, { flag: "wx", mode: 0o600 }); } catch { throw new Error("preprocessing_conflict"); }
|
|
|
|
|
return state;
|
|
|
|
|
export interface RegistryBootstrapAddressedPublicationStateV1 extends StateFields { readonly operation: "registry_bootstrap"; readonly baseCommit: null; readonly baseManifestSha256: null; readonly baseWorkspaces: readonly []; readonly changedSetRule: "all_target_workspace_ids" | null; }
|
|
|
|
|
export interface RegistryPullAddressedPublicationStateV1 extends StateFields { readonly operation: "registry_pull"; readonly baseCommit: Revision40; readonly baseManifestSha256: Sha256Hex; readonly baseWorkspaces: readonly RegistryWorkspaceManifestIdentityV1[]; readonly changedSetRule: "symmetric_base_target_workspace_difference" | null; }
|
|
|
|
|
export type RegistryAddressedPublicationStateV1 = RegistryBootstrapAddressedPublicationStateV1 | RegistryPullAddressedPublicationStateV1;
|
|
|
|
|
export class BorrowedWorkspaceMaintenanceQuiescenceLease { private constructor(readonly workspaceId: CanonicalWorkspaceId) {} static create(id: CanonicalWorkspaceId) { return new BorrowedWorkspaceMaintenanceQuiescenceLease(id); } }
|
|
|
|
|
export interface AddressedWorkspacePublicationLeaseV1 extends BorrowedOrderedWorkspaceWriterLeaseV1 { readonly quiescence: BorrowedWorkspaceMaintenanceQuiescenceLease; readonly readers: BorrowedWorkspaceSessionReadersExclusiveLockLease; }
|
|
|
|
|
export interface CapabilityAwareRegistryPublicationParticipant<T> { readonly participantId: string; prepare(plan: RegistryAddressedPlanV1, workspace: AddressedWorkspacePublicationLeaseV1): Promise<T>; reconcile(plan: RegistryAddressedPlanV1, workspace: AddressedWorkspacePublicationLeaseV1, prepared: T, phase: RegistryAddressedPublicationPhaseV1): Promise<void>; }
|
|
|
|
|
export interface RegistrySynchronizerPreparedV1 { readonly synchronizerId: string; readonly preparedSha256: Sha256Hex; }
|
|
|
|
|
export interface CapabilityAwareRegistryPublicationSynchronizer { readonly synchronizerId: string; ensureForPublication(plan: RegistryAddressedPlanV1, capabilities: OrderedWorkspaceWriterCapabilitySet, phase: "planned" | "participants_prepared" | "publication_intent_durable" | "target_published"): Promise<RegistrySynchronizerPreparedV1>; }
|
|
|
|
|
export class CapabilityAwareRegistryPublicationLifecycleOwner {
|
|
|
|
|
async run<T>(input: { readonly plan: RegistryAddressedPlanV1; readonly capabilities: OrderedWorkspaceWriterCapabilitySet; readonly participants: readonly CapabilityAwareRegistryPublicationParticipant<unknown>[]; readonly synchronizers: readonly CapabilityAwareRegistryPublicationSynchronizer[]; readonly action: () => Promise<T>; }): Promise<T> {
|
|
|
|
|
// Enter reader gates recursively: every changed workspace remains quiescent and reader-exclusive
|
|
|
|
|
// until publication, reconciliation, and terminal durability complete.
|
|
|
|
|
const prepared: Array<{ participant: CapabilityAwareRegistryPublicationParticipant<unknown>; value: unknown; lease: AddressedWorkspacePublicationLeaseV1 }> = [];
|
|
|
|
|
const enter = async (index: number): Promise<T> => {
|
|
|
|
|
if (index < input.capabilities.workspaceIds.length) {
|
|
|
|
|
const id = input.capabilities.workspaceIds[index] as CanonicalWorkspaceId;
|
|
|
|
|
return input.capabilities.forWorkspace(id, ({ writerCapability }) => writerCapability.runUnderSessionReadersExclusive(async readers => {
|
|
|
|
|
const lease = { workspaceId: id, rootLease: undefined, writerCapability, quiescence: BorrowedWorkspaceMaintenanceQuiescenceLease.create(id), readers } as unknown as AddressedWorkspacePublicationLeaseV1;
|
|
|
|
|
for (const participant of input.participants) prepared.push({ participant, value: await participant.prepare(input.plan, lease), lease });
|
|
|
|
|
return enter(index + 1);
|
|
|
|
|
}));
|
|
|
|
|
}
|
|
|
|
|
for (const synchronizer of input.synchronizers) await synchronizer.ensureForPublication(input.plan, input.capabilities, "planned");
|
|
|
|
|
const result = await input.action();
|
|
|
|
|
for (const item of prepared) await item.participant.reconcile(input.plan, item.lease, item.value, "target_published");
|
|
|
|
|
return result;
|
|
|
|
|
};
|
|
|
|
|
return enter(0);
|
|
|
|
|
}
|
|
|
|
|
async read(runId: string): Promise<RegistryAddressedPublicationStateV1> { try { return JSON.parse(await readFile(this.path(runId), "utf8")) as RegistryAddressedPublicationStateV1; } catch { throw new Error("preprocessing_conflict"); } }
|
|
|
|
|
async transition(runId: string, phase: RegistryAddressedPublicationStateV1["phase"], patch: Partial<RegistryAddressedPublicationStateV1> = {}): Promise<RegistryAddressedPublicationStateV1> {
|
|
|
|
|
const old = await this.read(runId); const order = ["request_claimed", "target_advertised", "target_fetched", "planned", "terminal"] as const; if (order.indexOf(phase) < order.indexOf(old.phase)) throw new Error("preprocessing_conflict");
|
|
|
|
|
const next = { ...old, ...patch, phase, updatedAt: new Date().toISOString() }; await writeFile(this.path(runId), `${JSON.stringify(next)}\n`, { mode: 0o600 }); return next;
|
|
|
|
|
}
|
|
|
|
|
export type RegistryAddressedRequestV1 = RegistryBootstrapAddressedRequestV1 | RegistryPublishAddressedRequestV1
|
|
|
|
|
| { readonly mode: "create"; readonly operation: "registry_bootstrap"; readonly runId: RegistryRunId32; readonly requestSha256: Sha256Hex; readonly installationIdentitySha256: Sha256Hex; readonly repositoryIdentitySha256: Sha256Hex; readonly expectedBaseCommit: null; readonly remoteRefIdentitySha256: Sha256Hex }
|
|
|
|
|
| { readonly mode: "resume"; readonly operation: "registry_bootstrap"; readonly runId: RegistryRunId32; readonly requestSha256: Sha256Hex; readonly installationIdentitySha256: Sha256Hex; readonly repositoryIdentitySha256: Sha256Hex; readonly remoteRefIdentitySha256: Sha256Hex }
|
|
|
|
|
| { readonly mode: "create"; readonly operation: "registry_pull"; readonly runId: RegistryRunId32; readonly requestSha256: Sha256Hex; readonly installationIdentitySha256: Sha256Hex; readonly repositoryIdentitySha256: Sha256Hex; readonly expectedBaseCommit: Revision40; readonly remoteRefIdentitySha256: Sha256Hex }
|
|
|
|
|
| { readonly mode: "resume"; readonly operation: "registry_pull"; readonly runId: RegistryRunId32; readonly requestSha256: Sha256Hex; readonly installationIdentitySha256: Sha256Hex; readonly repositoryIdentitySha256: Sha256Hex; readonly remoteRefIdentitySha256: Sha256Hex };
|
|
|
|
|
export interface RegistryBootstrapAddressedResultV1 { readonly operation: "registry_bootstrap"; readonly runId: RegistryRunId32; readonly jobArtifactPath: RegistryAddressedJobArtifactPathV1; readonly plan: RegistryBootstrapAddressedPlanV1; readonly planSha256: Sha256Hex; readonly phase: "terminal_durable"; readonly publication: "target" | "reconciled_target" | "unchanged"; }
|
|
|
|
|
export interface RegistryPullAddressedResultV1 { readonly operation: "registry_pull"; readonly runId: RegistryRunId32; readonly jobArtifactPath: RegistryAddressedJobArtifactPathV1; readonly plan: RegistryPullAddressedPlanV1; readonly planSha256: Sha256Hex; readonly phase: "terminal_durable"; readonly publication: "target" | "reconciled_target" | "unchanged"; }
|
|
|
|
|
export type RegistryAddressedResultV1 = RegistryBootstrapAddressedResultV1 | RegistryPullAddressedResultV1;
|
|
|
|
|
export interface RegistryBootstrapRecoveryIdentityV1 { readonly operation: "registry_bootstrap"; readonly requestSha256: Sha256Hex; readonly installationIdentitySha256: Sha256Hex; readonly repositoryIdentitySha256: Sha256Hex; readonly remoteRefIdentitySha256: Sha256Hex; }
|
|
|
|
|
export type RegistryEnsureBootstrapAddressedResultV1 = { readonly kind: "already_active"; readonly snapshot: RegistryActiveSnapshotV1 } | { readonly kind: "bootstrap_terminal"; readonly snapshot: RegistryActiveSnapshotV1; readonly result: RegistryBootstrapAddressedResultV1 };
|
|
|
|
|
export interface RegistryBootstrapRecoveryScanLimitsV1 { readonly maximumDirectoryEntries: 4096; readonly maximumArtifactBytes: 1048576; readonly maximumTotalArtifactBytes: 67108864; }
|
|
|
|
|
export const REGISTRY_SCAN_LIMITS_V1: RegistryBootstrapRecoveryScanLimitsV1 = Object.freeze({ maximumDirectoryEntries: 4096, maximumArtifactBytes: 1048576, maximumTotalArtifactBytes: 67108864 });
|
|
|
|
|
const RUN = /^[0-9a-f]{32}$/; const SHA = /^[0-9a-f]{64}$/; const REV = /^[0-9a-f]{40}$/;
|
|
|
|
|
const PHASES: readonly RegistryAddressedPublicationPhaseV1[] = ["request_claimed", "target_advertised", "target_fetched", "planned", "participants_prepared", "publication_intent_durable", "target_published", "terminal_durable"];
|
|
|
|
|
const CONFLICT = () => Object.assign(new Error("preprocessing_conflict"), { code: "preprocessing_conflict" });
|
|
|
|
|
const canonical = (v: unknown): string => JSON.stringify(v, (_k, x) => x && typeof x === "object" && !Array.isArray(x) ? Object.fromEntries(Object.keys(x).sort().map(k => [k, x[k]])) : x);
|
|
|
|
|
export const registryDigest = (value: unknown): string => createHash("sha256").update(canonical(value)).digest("hex");
|
|
|
|
|
export function canonicalBootstrapRequestDigest(request: { readonly kind?: "bootstrap" | "publish"; readonly operation?: RegistryAddressedOperationV1; readonly installation?: unknown; readonly repository?: unknown; readonly remote?: unknown; readonly workspaceIds?: readonly string[] }): string { return registryDigest({ kind: request.kind ?? (request.operation === "registry_pull" ? "publish" : "bootstrap"), operation: request.operation, installation: request.installation, repository: request.repository, remote: request.remote, workspaceIds: [...(request.workspaceIds ?? [])].sort() }); }
|
|
|
|
|
export function addressedRunId(): RegistryRunId32 { return randomBytes(16).toString("hex") as RegistryRunId32; }
|
|
|
|
|
function failIfBadIdentity(s: StateFields, runId: string): void { if (s.schemaVersion !== 1 || s.runId !== runId || !RUN.test(s.runId) || s.jobArtifactPath !== `addressed-publication-jobs/${s.runId}.json` || !SHA.test(s.requestSha256) || !SHA.test(s.installationIdentitySha256) || !SHA.test(s.repositoryIdentitySha256) || !SHA.test(s.remoteRefIdentitySha256) || !PHASES.includes(s.phase)) throw CONFLICT(); }
|
|
|
|
|
function immutable(s: StateFields): unknown { const { phase: _p, priorStateSha256: _h, ...rest } = s; return rest; }
|
|
|
|
|
async function fsync(path: string): Promise<void> { const h = await open(path, "r"); try { await h.sync(); } finally { await h.close(); } }
|
|
|
|
|
async function fsyncParent(path: string): Promise<void> { await fsync(dirname(path)); }
|
|
|
|
|
function ownerMode(st: Awaited<ReturnType<typeof stat>>, mode: number): boolean { const x = st as any; return x.isFile() && (Number(x.mode) & 0o777) === mode && Number(x.nlink) === 1 && Number(x.uid) === (process.getuid?.() ?? Number(x.uid)); }
|
|
|
|
|
async function strictRead(path: string, max = REGISTRY_SCAN_LIMITS_V1.maximumArtifactBytes): Promise<{ text: string; identity: { size: number; mtimeMs: number; ino: bigint } }> {
|
|
|
|
|
const h = await open(path, fsConstants.O_RDONLY | (fsConstants.O_NOFOLLOW ?? 0));
|
|
|
|
|
try { const before = await h.stat(); if (!ownerMode(before, 0o600) || before.size > max) throw CONFLICT(); const text = await h.readFile({ encoding: "utf8" }); const after = await h.stat(); if (before.ino !== after.ino || before.size !== after.size || text.length > max) throw CONFLICT(); return { text, identity: { size: Number(after.size), mtimeMs: Number(after.mtimeMs), ino: BigInt(after.ino) } }; } finally { await h.close(); }
|
|
|
|
|
}
|
|
|
|
|
export class RegistryAddressedPublicationStore {
|
|
|
|
|
readonly jobsDirectory: string;
|
|
|
|
|
constructor(readonly root: string) { this.jobsDirectory = join(root, "addressed-publication-jobs"); }
|
|
|
|
|
private path(runId: string): string { if (!RUN.test(runId)) throw CONFLICT(); return join(this.jobsDirectory, `${runId}.json`); }
|
|
|
|
|
private async dirs(): Promise<void> { await mkdir(this.jobsDirectory, { recursive: true, mode: 0o700 }); const st = await lstat(this.jobsDirectory); if (st.isSymbolicLink() || !st.isDirectory() || (Number((st as any).mode) & 0o777) !== 0o700 || Number((st as any).uid) !== (process.getuid?.() ?? Number((st as any).uid))) throw CONFLICT(); }
|
|
|
|
|
private async durable(path: string, value: unknown, exclusive = false): Promise<void> { const name = path.split("/").pop()!; const tmp = join(this.jobsDirectory, `.${name}.tmp`); if (exclusive) { try { await stat(path); throw CONFLICT(); } catch (e) { if ((e as NodeJS.ErrnoException).code !== "ENOENT") throw CONFLICT(); } } const bytes = `${canonical(value)}\n`; try { const h = await open(tmp, fsConstants.O_WRONLY | fsConstants.O_CREAT | fsConstants.O_EXCL | (fsConstants.O_NOFOLLOW ?? 0), 0o600); try { await h.writeFile(bytes); await h.sync(); } finally { await h.close(); } const st = await stat(tmp); if (!ownerMode(st, 0o600)) throw CONFLICT(); if (!exclusive) { const current = await lstat(path); if (current.isSymbolicLink() || !ownerMode(current, 0o600)) throw CONFLICT(); } await rename(tmp, path); await fsyncParent(path); } catch (e) { await rm(tmp, { force: true }).catch(() => undefined); if (exclusive && (e as NodeJS.ErrnoException)?.code === "EEXIST") throw CONFLICT(); throw CONFLICT(); } }
|
|
|
|
|
async claim(request: RegistryAddressedRequestV1, runId: RegistryRunId32 = ("runId" in request ? request.runId : addressedRunId())): Promise<RegistryAddressedPublicationStateV1> { await this.dirs(); const legacy = !(("operation" in request) && ("requestSha256" in request)); const operation = (legacy ? (request as RegistryBootstrapAddressedRequestV1 | RegistryPublishAddressedRequestV1).kind === "publish" ? "registry_pull" : "registry_bootstrap" : (request as any).operation) as RegistryAddressedOperationV1; const requestSha256 = (legacy ? (request as any).requestDigest : (request as any).requestSha256) as Sha256Hex; const installationIdentitySha256 = (legacy ? (request as any).installation.digest : (request as any).installationIdentitySha256) as Sha256Hex; const repositoryIdentitySha256 = (legacy ? (request as any).repository.digest : (request as any).repositoryIdentitySha256) as Sha256Hex; const remoteRefIdentitySha256 = (legacy ? (request as any).remote.digest : (request as any).remoteRefIdentitySha256) as Sha256Hex; if (!RUN.test(runId)) throw CONFLICT(); const now = new Date().toISOString(); const baseCommit = operation === "registry_pull" && "expectedBaseCommit" in request ? request.expectedBaseCommit : null; const state = { schemaVersion: 1, runId, requestSha256: requestSha256, jobArtifactPath: `addressed-publication-jobs/${runId}.json`, phase: "request_claimed" as const, operation, installationIdentitySha256, repositoryIdentitySha256, remoteRefIdentitySha256, advertisedTargetCommit: null, immutableTargetRef: null, fetchedTargetCommit: null, targetCommit: null, targetManifestSha256: null, targetWorkspaces: null, changedWorkspaceIds: null, planSha256: null, changedSetSha256: null, participantsSha256: null, synchronizersSha256: null, publicationIntentSha256: null, publishedActiveStateSha256: null, terminalResultSha256: null, priorStateSha256: null, baseCommit, baseManifestSha256: operation === "registry_pull" ? (null as Sha256Hex | null) : null, baseWorkspaces: [], changedSetRule: null } as unknown as RegistryAddressedPublicationStateV1 & { readonly updatedAt: string; readonly operation: RegistryAddressedOperationV1 };
|
|
|
|
|
await this.durable(this.path(runId), state, true); return state;
|
|
|
|
|
}
|
|
|
|
|
snapshotFor(state: RegistryAddressedPublicationStateV1): RegistryAddressedSnapshotV1 { const { runId: _r, phase: _p, updatedAt: _u, result: _x, ...snapshot } = state; return snapshot; }
|
|
|
|
|
private snapshot(request: RegistryAddressedRequestV1): RegistryAddressedSnapshotV1 { return { schemaVersion: 1, commit: null, baseCommit: null, baseDigest: null, workspaceIds: [...request.workspaceIds].sort(), changedWorkspaceIds: [...request.workspaceIds].sort(), installation: request.installation, repository: request.repository, remote: request.remote, requestDigest: request.requestDigest }; }
|
|
|
|
|
async read(runId: RegistryRunId32): Promise<RegistryAddressedPublicationStateV1> {
|
|
|
|
|
const path = this.path(runId); let parsed: unknown;
|
|
|
|
|
try { parsed = JSON.parse((await strictRead(path)).text); } catch { throw CONFLICT(); }
|
|
|
|
|
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) throw CONFLICT();
|
|
|
|
|
const s = parsed as StateFields & { operation?: unknown };
|
|
|
|
|
const keys = Object.keys(parsed).sort();
|
|
|
|
|
const expected = ["schemaVersion","runId","requestSha256","jobArtifactPath","phase","operation","installationIdentitySha256","repositoryIdentitySha256","remoteRefIdentitySha256","advertisedTargetCommit","immutableTargetRef","fetchedTargetCommit","targetCommit","targetManifestSha256","targetWorkspaces","changedWorkspaceIds","planSha256","changedSetSha256","participantsSha256","synchronizersSha256","publicationIntentSha256","publishedActiveStateSha256","terminalResultSha256","priorStateSha256","baseCommit","baseManifestSha256","baseWorkspaces","changedSetRule"].sort();
|
|
|
|
|
if (keys.length !== expected.length || keys.some((v, i) => v !== expected[i])) throw CONFLICT();
|
|
|
|
|
failIfBadIdentity(s, runId);
|
|
|
|
|
if (s.operation !== "registry_bootstrap" && s.operation !== "registry_pull") throw CONFLICT();
|
|
|
|
|
for (const key of ["advertisedTargetCommit","fetchedTargetCommit","targetCommit"] as const) if (s[key] !== null && !REV.test(s[key])) throw CONFLICT();
|
|
|
|
|
for (const key of ["requestSha256","installationIdentitySha256","repositoryIdentitySha256","remoteRefIdentitySha256"] as const) if (!SHA.test(s[key])) throw CONFLICT();
|
|
|
|
|
for (const key of ["targetManifestSha256","changedSetSha256","planSha256","participantsSha256","synchronizersSha256","publicationIntentSha256","publishedActiveStateSha256","terminalResultSha256","priorStateSha256"] as const) if (s[key] !== null && !SHA.test(s[key])) throw CONFLICT();
|
|
|
|
|
if (s.immutableTargetRef !== null && s.immutableTargetRef !== `refs/thoth/addressed-runs/${runId}/target`) throw CONFLICT();
|
|
|
|
|
if (s.operation === "registry_bootstrap" && (s.baseCommit !== null || s.baseManifestSha256 !== null || s.baseWorkspaces.length !== 0 || (s.changedSetRule !== null && s.changedSetRule !== "all_target_workspace_ids"))) throw CONFLICT();
|
|
|
|
|
if (s.operation === "registry_pull" && (s.baseCommit === null || !REV.test(s.baseCommit) || (s.baseManifestSha256 !== null && !SHA.test(s.baseManifestSha256)) || s.changedSetRule !== null && s.changedSetRule !== "symmetric_base_target_workspace_difference")) throw CONFLICT();
|
|
|
|
|
if (s.targetWorkspaces !== null && !Array.isArray(s.targetWorkspaces) || s.changedWorkspaceIds !== null && !Array.isArray(s.changedWorkspaceIds)) throw CONFLICT();
|
|
|
|
|
return s as RegistryAddressedPublicationStateV1;
|
|
|
|
|
}
|
|
|
|
|
async transition(runId: RegistryRunId32, phase: RegistryAddressedPublicationPhaseV1, patch: Partial<RegistryAddressedPublicationStateV1> = {}): Promise<RegistryAddressedPublicationStateV1> { const old = await this.read(runId); if (PHASES.indexOf(phase) < PHASES.indexOf(old.phase) || PHASES.indexOf(phase) > PHASES.indexOf(old.phase) + 1) throw CONFLICT(); const allowed = new Set(["advertisedTargetCommit","immutableTargetRef","fetchedTargetCommit","targetCommit","targetManifestSha256","targetWorkspaces","changedWorkspaceIds","planSha256","changedSetSha256","participantsSha256","synchronizersSha256","publicationIntentSha256","publishedActiveStateSha256","terminalResultSha256","changedSetRule","baseManifestSha256","baseWorkspaces"]); for (const key of Object.keys(patch)) if (!allowed.has(key)) throw CONFLICT(); for (const key of ["schemaVersion","runId","requestSha256","jobArtifactPath","installationIdentitySha256","repositoryIdentitySha256","remoteRefIdentitySha256","operation","baseCommit"] as const) if (key in patch && (patch as unknown as Record<string, unknown>)[key] !== (old as unknown as Record<string, unknown>)[key]) throw CONFLICT(); const next = { ...old, ...patch, phase, priorStateSha256: registryDigest(old) as Sha256Hex }; await this.durable(this.path(runId), next); return next as RegistryAddressedPublicationStateV1; }
|
|
|
|
|
snapshotFor(state: RegistryAddressedPublicationStateV1): RegistryActiveSnapshotV1 { if (!state.targetCommit || !state.targetManifestSha256 || !state.targetWorkspaces) throw CONFLICT(); return { schemaVersion: 1, commit: state.targetCommit, manifestSha256: state.targetManifestSha256, workspaces: state.targetWorkspaces }; }
|
|
|
|
|
async scan(): Promise<readonly RegistryAddressedPublicationStateV1[]> { await this.dirs(); let entries: string[]; try { entries = (await readdir(this.jobsDirectory)).sort(); } catch { throw CONFLICT(); } if (entries.length > REGISTRY_SCAN_LIMITS_V1.maximumDirectoryEntries) throw CONFLICT();
|
|
|
|
|
for (const name of entries.filter(x => /^\.[0-9a-f]{32}\.json\.tmp$/.test(x))) { const path = join(this.jobsDirectory, name); const st = await lstat(path); if (st.isSymbolicLink() || !ownerMode(st, 0o600) || st.size > REGISTRY_SCAN_LIMITS_V1.maximumArtifactBytes) throw CONFLICT(); await unlink(path); await fsyncParent(path); }
|
|
|
|
|
entries = (await readdir(this.jobsDirectory)).sort(); if (entries.length > REGISTRY_SCAN_LIMITS_V1.maximumDirectoryEntries) throw CONFLICT(); const names = entries.filter(x => /^[0-9a-f]{32}\.json$/.test(x)); if (names.length !== entries.length) throw CONFLICT(); let total = 0; const identities = new Map<string, string>(); const out: RegistryAddressedPublicationStateV1[] = []; for (const name of names) { const st = await lstat(join(this.jobsDirectory, name)); if (st.isSymbolicLink() || !ownerMode(st, 0o600) || st.size > REGISTRY_SCAN_LIMITS_V1.maximumArtifactBytes || (total += st.size) > REGISTRY_SCAN_LIMITS_V1.maximumTotalArtifactBytes) throw CONFLICT(); identities.set(name, `${String((st as any).dev)}:${String((st as any).ino)}:${String((st as any).size)}:${String((st as any).mtimeMs)}`); out.push(await this.read(name.slice(0, -5) as RegistryRunId32)); } const verify = (await readdir(this.jobsDirectory)).sort(); if (verify.length !== names.length || verify.some((name, i) => name !== names[i])) throw CONFLICT(); for (const name of names) { const st = await lstat(join(this.jobsDirectory, name)); const key = `${String((st as any).dev)}:${String((st as any).ino)}:${String((st as any).size)}:${String((st as any).mtimeMs)}`; if (identities.get(name) !== key) throw CONFLICT(); } return out; }
|
|
|
|
|
async removeSibling(runId: RegistryRunId32): Promise<void> { const path = join(this.jobsDirectory, `.${runId}.json.tmp`); try { const st = await lstat(path); if (st.isSymbolicLink() || !ownerMode(st, 0o600)) throw CONFLICT(); await unlink(path); await fsyncParent(path); } catch (e) { if ((e as NodeJS.ErrnoException).code !== "ENOENT") throw CONFLICT(); } }
|
|
|
|
|
}
|
|
|
|
|
