diff --git a/packages/producer/src/distributed.ts b/packages/producer/src/distributed.ts index a280906002..97a0661931 100644 --- a/packages/producer/src/distributed.ts +++ b/packages/producer/src/distributed.ts @@ -60,6 +60,8 @@ export { listPlanV2ArtifactsForTarget, materializePlanV2Target, planV2, + planV2WithPublisher, + publishPlanV2FromV1, readPlanV2Manifest, validatePlanV2MaterializedTarget, PLAN_V2_INTEGRITY_UNRECOVERABLE, @@ -71,7 +73,14 @@ export { type PlanV2MaterializationResult, type PlanV2MaterializationTarget, type PlanV2Result, + type PlanV2WithPublisherOptions, } from "./services/distributed/planV2.js"; +export { + LocalPlanV2ArtifactPublisher, + type LocalPlanV2ArtifactPublisherOptions, + type PlanV2ArtifactPublisher, + type PlanV2PublishBlob, +} from "./services/distributed/planV2Publisher.js"; export { assembleV2, renderChunkV2 } from "./services/distributed/planV2Execution.js"; // ── RenderChunk (Activity B) ──────────────────────────────────────────────── diff --git a/packages/producer/src/index.ts b/packages/producer/src/index.ts index e8001cd2ec..f3e2bc9203 100644 --- a/packages/producer/src/index.ts +++ b/packages/producer/src/index.ts @@ -152,11 +152,13 @@ export { materializePlanV2Target, plan, planV2, + planV2WithPublisher, PlanV2IntegrityError, PlanProtocolUnsupportedError, readPlanProtocol, readPlanProtocolV1, readPlanV2Manifest, + publishPlanV2FromV1, renderChunk, renderChunkV2, validatePlanV2MaterializedTarget, @@ -175,5 +177,8 @@ export { type PlanV2MaterializationResult, type PlanV2MaterializationTarget, type PlanV2Result, + type PlanV2WithPublisherOptions, + type PlanV2ArtifactPublisher, + type PlanV2PublishBlob, type SupportedPlanProtocolDescriptor, } from "./distributed.js"; diff --git a/packages/producer/src/services/distributed/planV2.test.ts b/packages/producer/src/services/distributed/planV2.test.ts index d7b77346ea..6cf1088d26 100644 --- a/packages/producer/src/services/distributed/planV2.test.ts +++ b/packages/producer/src/services/distributed/planV2.test.ts @@ -5,6 +5,7 @@ import { mkdtempSync, readFileSync, rmSync, + statSync, symlinkSync, writeFileSync, } from "node:fs"; @@ -20,9 +21,11 @@ import { materializePlanV2Target, PLAN_V2_INTEGRITY_UNRECOVERABLE, PlanV2IntegrityError, + publishPlanV2FromV1, readPlanV2Manifest, validatePlanV2MaterializedTarget, } from "./planV2.js"; +import { LocalPlanV2ArtifactPublisher, type PlanV2ArtifactPublisher } from "./planV2Publisher.js"; const tempDirs: string[] = []; @@ -399,6 +402,149 @@ describe("Plan v2 manifest", () => { }); }); +describe("Plan v2 artifact publisher", () => { + it("publishes the manifest last and hard-links local immutable blobs", async () => { + const root = tempPath("hf-plan-v2-publisher-"); + const v1 = createV1Plan(root, { audio: true }); + const destination = join(root, "v2"); + const publisher = new LocalPlanV2ArtifactPublisher(destination); + const manifest = await publishPlanV2FromV1(v1, publisher); + const artifact = manifest.artifacts.find( + (candidate) => candidate.path === "compiled/asset.txt", + ); + if (artifact === undefined) throw new Error("test fixture is missing compiled/asset.txt"); + const sourceStat = statSync(join(v1, artifact.path)); + const blobStat = statSync( + join(destination, "artifacts", "sha256", artifact.sha256.slice(0, 2), artifact.sha256), + ); + + expect(readPlanV2Manifest(destination)).toEqual(manifest); + expect({ dev: blobStat.dev, ino: blobStat.ino }).toEqual({ + dev: sourceStat.dev, + ino: sourceStat.ino, + }); + }); + + it("falls back to an atomic copy when hard-linking is unavailable", async () => { + const root = tempPath("hf-plan-v2-publisher-copy-"); + const v1 = createV1Plan(root); + const destination = join(root, "v2"); + const publisher = new LocalPlanV2ArtifactPublisher(destination, { + linkFile() { + throw Object.assign(new Error("cross-device link"), { code: "EXDEV" }); + }, + }); + const manifest = await publishPlanV2FromV1(v1, publisher); + const artifact = manifest.artifacts.find( + (candidate) => candidate.path === "compiled/asset.txt", + ); + if (artifact === undefined) throw new Error("test fixture is missing compiled/asset.txt"); + const sourcePath = join(v1, artifact.path); + const blobPath = join( + destination, + "artifacts", + "sha256", + artifact.sha256.slice(0, 2), + artifact.sha256, + ); + + expect(readFileSync(blobPath)).toEqual(readFileSync(sourcePath)); + expect({ dev: statSync(blobPath).dev, ino: statSync(blobPath).ino }).not.toEqual({ + dev: statSync(sourcePath).dev, + ino: statSync(sourcePath).ino, + }); + }); + + it("produces byte-identical local CAS output through both publication paths", async () => { + const root = tempPath("hf-plan-v2-publisher-parity-"); + const v1 = createV1Plan(root, { audio: true }); + const directDir = join(root, "direct"); + const publishedDir = join(root, "published"); + createPlanV2FromV1(v1, directDir); + const publisher = new LocalPlanV2ArtifactPublisher(publishedDir); + const manifest = await publishPlanV2FromV1(v1, publisher); + + expect(readFileSync(join(publishedDir, "plan.json"))).toEqual( + readFileSync(join(directDir, "plan.json")), + ); + for (const artifact of manifest.artifacts) { + const suffix = join("artifacts", "sha256", artifact.sha256.slice(0, 2), artifact.sha256); + expect(readFileSync(join(publishedDir, suffix))).toEqual( + readFileSync(join(directDir, suffix)), + ); + } + }); + + it("supports a remote publisher contract with no shared destination filesystem", async () => { + const root = tempPath("hf-plan-v2-remote-publisher-"); + const v1 = createV1Plan(root, { audio: true }); + const blobs = new Map(); + let committedManifest: string | undefined; + const publisher: PlanV2ArtifactPublisher = { + async putBlob(blob) { + blobs.set(blob.sha256, readFileSync(blob.sourcePath)); + }, + async commitManifest(manifestBytes) { + committedManifest = manifestBytes; + }, + async abort() {}, + }; + + const manifest = await publishPlanV2FromV1(v1, publisher); + expect(committedManifest).toBe(canonicalJsonStringify(manifest)); + expect(blobs.size).toBe(new Set(manifest.artifacts.map((artifact) => artifact.sha256)).size); + for (const artifact of manifest.artifacts) { + expect(blobs.get(artifact.sha256)?.byteLength).toBe(artifact.sizeBytes); + } + }); + + it("rejects malformed digests before constructing a local CAS path", async () => { + const root = tempPath("hf-plan-v2-publisher-digest-"); + const sourcePath = join(root, "source"); + writeFileSync(sourcePath, "bytes"); + const publisher = new LocalPlanV2ArtifactPublisher(join(root, "v2")); + + await expect( + publisher.putBlob({ sourcePath, sha256: "../escape", sizeBytes: 5 }), + ).rejects.toThrow("must be a lowercase sha256 digest"); + await publisher.abort(); + }); + + it("refuses to commit a manifest until every referenced blob is durable", async () => { + const root = tempPath("hf-plan-v2-publisher-incomplete-"); + const publisher = new LocalPlanV2ArtifactPublisher(join(root, "v2")); + const digest = "a".repeat(64); + + await expect( + publisher.commitManifest(JSON.stringify({ artifacts: [{ sha256: digest }] })), + ).rejects.toThrow("cannot commit manifest before referenced blob is durable"); + await publisher.abort(); + }); + + it("aborts without committing a manifest when a blob publish fails", async () => { + const root = tempPath("hf-plan-v2-publisher-failure-"); + const calls: string[] = []; + const publisher: PlanV2ArtifactPublisher = { + async putBlob(blob) { + calls.push(`blob:${blob.sha256}`); + throw new Error("injected blob failure"); + }, + async commitManifest() { + calls.push("manifest"); + }, + async abort() { + calls.push("abort"); + }, + }; + + await expect(publishPlanV2FromV1(createV1Plan(root), publisher)).rejects.toThrow( + "injected blob failure", + ); + expect(calls.at(-1)).toBe("abort"); + expect(calls).not.toContain("manifest"); + }); +}); + describe("Plan v2 hash schema", () => { it("does not reuse a raw artifact digest as its manifest hash", () => { const root = tempPath("hf-plan-v2-hash-"); diff --git a/packages/producer/src/services/distributed/planV2.ts b/packages/producer/src/services/distributed/planV2.ts index 91bbff309a..d1e68adfdb 100644 --- a/packages/producer/src/services/distributed/planV2.ts +++ b/packages/producer/src/services/distributed/planV2.ts @@ -27,6 +27,7 @@ import { writeFileSync, } from "node:fs"; import { createHash } from "node:crypto"; +import { tmpdir } from "node:os"; import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; import { createFrameLookupTable, type ExtractedFrames } from "@hyperframes/engine"; import { recomputePlanHashFromPlanDir, type ChunkSliceJson } from "../render/stages/freezePlan.js"; @@ -37,6 +38,13 @@ import { readPlanProtocolV1, type PlanProtocolV2Descriptor, } from "./planProtocol.js"; +import { + LocalPlanV2ArtifactPublisher, + type PlanV2ArtifactPublisher, + type PlanV2PublishBlob, +} from "./planV2Publisher.js"; +import { PLAN_V2_INTEGRITY_UNRECOVERABLE, PlanV2IntegrityError } from "./planV2Errors.js"; +import { planV2BlobPath } from "./planV2Layout.js"; import { PLAN_AUDIO_RELATIVE_PATH, PLAN_VIDEOS_META_RELATIVE_PATH, @@ -46,23 +54,7 @@ import { const PLAN_V2_HASH_PREFIX = "hyperframes-plan-manifest-hash-v2\x00"; export const PLAN_V2_MATERIALIZATION_MARKER = ".hyperframes-plan-v2.json"; -export const PLAN_V2_INTEGRITY_UNRECOVERABLE = "PLAN_V2_INTEGRITY_UNRECOVERABLE" as const; - -/** - * A deterministic plan-v2 validation or integrity failure. Distributed - * adapters classify both the class name and code as terminal so immutable - * corruption does not consume the retry budget. - */ -export class PlanV2IntegrityError extends Error { - // Public adapters inspect this typed code even though OSS producer does not. - // fallow-ignore-next-line unused-class-member - readonly code: typeof PLAN_V2_INTEGRITY_UNRECOVERABLE = PLAN_V2_INTEGRITY_UNRECOVERABLE; - - constructor(message: string) { - super(`[planV2] ${message}`); - this.name = "PlanV2IntegrityError"; - } -} +export { PLAN_V2_INTEGRITY_UNRECOVERABLE, PlanV2IntegrityError }; export type PlanV2MaterializationTarget = | Readonly<{ role: "chunk"; chunkIndex: number }> @@ -117,6 +109,16 @@ export interface PlanV2Result { readonly limitations: Readonly; } +export interface PlanV2WithPublisherOptions { + /** + * Parent for the planner-private v1 staging directory. Remote adapters + * normally leave this unset and use the OS temp directory. The local + * compatibility wrapper keeps staging beside its destination so hard links + * can avoid a second set of data blocks. + */ + readonly stagingParentDir?: string; +} + export interface PlanV2MaterializationResult { readonly planDir: string; readonly target: PlanV2MaterializationTarget; @@ -188,10 +190,6 @@ function listFiles(root: string): Array<{ path: string; absolutePath: string }> return files.sort((a, b) => a.path.localeCompare(b.path)); } -function blobPath(planV2Dir: string, sha256: string): string { - return join(planV2Dir, "artifacts", "sha256", sha256.slice(0, 2), sha256); -} - /** Hash large plan artifacts with bounded memory (important above the v1 2 GiB cap). */ function sha256File(path: string): string { const hash = createHash("sha256"); @@ -447,11 +445,14 @@ function writeBlob(sourcePath: string, destinationPath: string): void { } } -/** Convert a frozen v1 execution directory into the immutable v2 transport. */ -export function createPlanV2FromV1(planV1Dir: string, planV2Dir: string): PlanV2Result { - if (existsSync(planV2Dir)) { - throw new PlanV2IntegrityError(`output directory already exists: ${planV2Dir}`); - } +interface PlanV2Publication { + readonly manifest: PlanV2Manifest; + readonly blobs: readonly Readonly[]; +} + +// Artifact classification intentionally keeps every fail-safe branch together. +// fallow-ignore-next-line complexity +function buildPlanV2Publication(planV1Dir: string): PlanV2Publication { const v1PlanPath = join(planV1Dir, "plan.json"); if (!existsSync(v1PlanPath)) { throw new PlanV2IntegrityError(`v1 plan is missing plan.json: ${v1PlanPath}`); @@ -474,63 +475,142 @@ export function createPlanV2FromV1(planV1Dir: string, planV2Dir: string): PlanV2 ); } - mkdirSync(dirname(planV2Dir), { recursive: true }); - const tempDir = mkdtempSync(join(dirname(planV2Dir), ".plan-v2-build-")); - try { - const artifacts: PlanV2Artifact[] = []; - const dimensions = v1Plan.dimensions; - if (!isRecord(dimensions)) { - throw new PlanV2IntegrityError("v1 plan.json.dimensions must be an object"); + const artifacts: PlanV2Artifact[] = []; + const blobs = new Map(); + const dimensions = v1Plan.dimensions; + if (!isRecord(dimensions)) { + throw new PlanV2IntegrityError("v1 plan.json.dimensions must be an object"); + } + const videoDependencyPlan = buildVideoChunkDependencies(planV1Dir, dimensions); + for (const file of listFiles(planV1Dir)) { + const targets = artifactTargets(file.path, videoDependencyPlan.dependencies); + if ( + file.path.startsWith("video-frames/") && + targets.chunks !== "all" && + targets.chunks.length === 0 && + !targets.assembler + ) { + continue; } - const videoDependencyPlan = buildVideoChunkDependencies(planV1Dir, dimensions); - for (const file of listFiles(planV1Dir)) { - const targets = artifactTargets(file.path, videoDependencyPlan.dependencies); - if ( - file.path.startsWith("video-frames/") && - targets.chunks !== "all" && - targets.chunks.length === 0 && - !targets.assembler - ) { - continue; - } - const sha256 = sha256File(file.absolutePath); - const sizeBytes = statSync(file.absolutePath).size; - writeBlob(file.absolutePath, blobPath(tempDir, sha256)); - artifacts.push({ - path: file.path, - sha256, - sizeBytes, - ...targets, - }); + const sha256 = sha256File(file.absolutePath); + const sizeBytes = statSync(file.absolutePath).size; + if (!blobs.has(sha256)) { + blobs.set(sha256, { sourcePath: file.absolutePath, sha256, sizeBytes }); } + artifacts.push({ + path: file.path, + sha256, + sizeBytes, + ...targets, + }); + } - const base: Omit = { - protocol: PLAN_PROTOCOL_V2, - sourcePlanV1Hash, - chunkCount: readPositiveInteger(v1Plan.chunkCount, "chunkCount"), - totalFrames: readPositiveInteger(v1Plan.totalFrames, "totalFrames"), - fps: readV1PlanFps(dimensions), - width: readPositiveInteger(dimensions.width, "dimensions.width"), - height: readPositiveInteger(dimensions.height, "dimensions.height"), - format: readDistributedFormat(dimensions.format), - ffmpegVersion: readString(v1Plan.ffmpegVersion, "ffmpegVersion"), - producerVersion: readString(v1Plan.producerVersion, "producerVersion"), - limitations: { videoDependencyMode: videoDependencyPlan.mode }, - artifacts, - }; - const manifest: PlanV2Manifest = { + const base: Omit = { + protocol: PLAN_PROTOCOL_V2, + sourcePlanV1Hash, + chunkCount: readPositiveInteger(v1Plan.chunkCount, "chunkCount"), + totalFrames: readPositiveInteger(v1Plan.totalFrames, "totalFrames"), + fps: readV1PlanFps(dimensions), + width: readPositiveInteger(dimensions.width, "dimensions.width"), + height: readPositiveInteger(dimensions.height, "dimensions.height"), + format: readDistributedFormat(dimensions.format), + ffmpegVersion: readString(v1Plan.ffmpegVersion, "ffmpegVersion"), + producerVersion: readString(v1Plan.producerVersion, "producerVersion"), + limitations: { videoDependencyMode: videoDependencyPlan.mode }, + artifacts, + }; + return { + manifest: { ...base, planHash: computeManifestHash(base), - }; - writeFileSync(join(tempDir, "plan.json"), canonicalJsonStringify(manifest), "utf-8"); + }, + blobs: [...blobs.values()], + }; +} + +/** Convert a frozen v1 execution directory into the immutable v2 transport. */ +export function createPlanV2FromV1(planV1Dir: string, planV2Dir: string): PlanV2Result { + if (existsSync(planV2Dir)) { + throw new PlanV2IntegrityError(`output directory already exists: ${planV2Dir}`); + } + const publication = buildPlanV2Publication(planV1Dir); + mkdirSync(dirname(planV2Dir), { recursive: true }); + const tempDir = mkdtempSync(join(dirname(planV2Dir), ".plan-v2-build-")); + try { + for (const blob of publication.blobs) { + writeBlob(blob.sourcePath, planV2BlobPath(tempDir, blob.sha256)); + } + writeFileSync( + join(tempDir, "plan.json"), + canonicalJsonStringify(publication.manifest), + "utf-8", + ); renameSync(tempDir, planV2Dir); - return resultFromManifest(planV2Dir, manifest); + return resultFromManifest(planV2Dir, publication.manifest); } catch (error) { rmSync(tempDir, { recursive: true, force: true }); throw error; } } +export async function publishPlanV2FromV1( + planV1Dir: string, + publisher: PlanV2ArtifactPublisher, +): Promise { + try { + const publication = buildPlanV2Publication(planV1Dir); + const concurrency = 16; + for (let offset = 0; offset < publication.blobs.length; offset += concurrency) { + const batch = publication.blobs.slice(offset, offset + concurrency); + const results = await Promise.allSettled(batch.map((blob) => publisher.putBlob(blob))); + for (const result of results) { + if (result.status === "rejected") throw result.reason; + } + } + await publisher.commitManifest(canonicalJsonStringify(publication.manifest)); + return publication.manifest; + } catch (error) { + try { + await publisher.abort(); + } catch { + // Preserve the publication failure; cleanup is best-effort. + } + throw error; + } +} + +/** + * Plan into a storage-neutral publisher. Implementations may write to a local + * directory, S3, GCS, or another durable CAS. Only the planner's private + * staging directory is local; distributed workers receive adapter-owned + * manifest and artifact locators. + */ +export async function planV2WithPublisher( + projectDir: string, + config: DistributedRenderConfig, + publisher: PlanV2ArtifactPublisher, + options: Readonly = {}, +): Promise { + const stagingRoot = mkdtempSync(join(options.stagingParentDir ?? tmpdir(), ".plan-v2-source-")); + try { + await plan( + projectDir, + { ...config, planDirSizeLimitBytes: Number.MAX_SAFE_INTEGER }, + stagingRoot, + ); + return await publishPlanV2FromV1(stagingRoot, publisher); + } catch (error) { + try { + await publisher.abort(); + } catch { + // Preserve the planning/publication failure; cleanup is best-effort. + } + throw error; + } finally { + rmSync(stagingRoot, { recursive: true, force: true }); + } +} + /** * Plan directly into v2. The large v1 directory exists only as local staging; * its historical 2 GiB transport cap is disabled because no monolithic @@ -544,18 +624,11 @@ export async function planV2( if (existsSync(planV2Dir)) { throw new PlanV2IntegrityError(`output directory already exists: ${planV2Dir}`); } - mkdirSync(dirname(planV2Dir), { recursive: true }); - const stagingDir = mkdtempSync(join(dirname(planV2Dir), ".plan-v2-source-")); - try { - await plan( - projectDir, - { ...config, planDirSizeLimitBytes: Number.MAX_SAFE_INTEGER }, - stagingDir, - ); - return createPlanV2FromV1(stagingDir, planV2Dir); - } finally { - rmSync(stagingDir, { recursive: true, force: true }); - } + const publisher = new LocalPlanV2ArtifactPublisher(planV2Dir); + const manifest = await planV2WithPublisher(projectDir, config, publisher, { + stagingParentDir: dirname(planV2Dir), + }); + return resultFromManifest(planV2Dir, manifest); } function resultFromManifest(planV2Dir: string, manifest: PlanV2Manifest): PlanV2Result { @@ -747,7 +820,7 @@ export function listPlanV2ArtifactsForTarget( } function verifyBlob(planV2Dir: string, artifact: Readonly): string { - const sourcePath = blobPath(planV2Dir, artifact.sha256); + const sourcePath = planV2BlobPath(planV2Dir, artifact.sha256); if (!existsSync(sourcePath)) { throw new PlanV2IntegrityError(`missing content-addressed artifact ${artifact.sha256}`); } diff --git a/packages/producer/src/services/distributed/planV2Errors.ts b/packages/producer/src/services/distributed/planV2Errors.ts new file mode 100644 index 0000000000..276b217fbc --- /dev/null +++ b/packages/producer/src/services/distributed/planV2Errors.ts @@ -0,0 +1,17 @@ +export const PLAN_V2_INTEGRITY_UNRECOVERABLE = "PLAN_V2_INTEGRITY_UNRECOVERABLE" as const; + +/** + * A deterministic plan-v2 validation or integrity failure. Distributed + * adapters classify both the class name and code as terminal so immutable + * corruption does not consume the retry budget. + */ +export class PlanV2IntegrityError extends Error { + // Public adapters inspect this typed code even though OSS producer does not. + // fallow-ignore-next-line unused-class-member + readonly code: typeof PLAN_V2_INTEGRITY_UNRECOVERABLE = PLAN_V2_INTEGRITY_UNRECOVERABLE; + + constructor(message: string) { + super(`[planV2] ${message}`); + this.name = "PlanV2IntegrityError"; + } +} diff --git a/packages/producer/src/services/distributed/planV2Layout.ts b/packages/producer/src/services/distributed/planV2Layout.ts new file mode 100644 index 0000000000..980c6865d8 --- /dev/null +++ b/packages/producer/src/services/distributed/planV2Layout.ts @@ -0,0 +1,16 @@ +import { join } from "node:path"; +import { PlanV2IntegrityError } from "./planV2Errors.js"; + +const SHA256_PATTERN = /^[0-9a-f]{64}$/; + +export function assertPlanV2Sha256(value: unknown, field: string): string { + if (typeof value !== "string" || !SHA256_PATTERN.test(value)) { + throw new PlanV2IntegrityError(`${field} must be a lowercase sha256 digest`); + } + return value; +} + +export function planV2BlobPath(root: string, sha256: string): string { + const digest = assertPlanV2Sha256(sha256, "artifact sha256"); + return join(root, "artifacts", "sha256", digest.slice(0, 2), digest); +} diff --git a/packages/producer/src/services/distributed/planV2Publisher.ts b/packages/producer/src/services/distributed/planV2Publisher.ts new file mode 100644 index 0000000000..f6edb8522d --- /dev/null +++ b/packages/producer/src/services/distributed/planV2Publisher.ts @@ -0,0 +1,137 @@ +import { + copyFileSync, + existsSync, + linkSync, + mkdirSync, + mkdtempSync, + renameSync, + rmSync, + statSync, + writeFileSync, +} from "node:fs"; +import { dirname, join } from "node:path"; +import { PlanV2IntegrityError } from "./planV2Errors.js"; +import { assertPlanV2Sha256, planV2BlobPath } from "./planV2Layout.js"; + +export interface PlanV2PublishBlob { + readonly sourcePath: string; + readonly sha256: string; + readonly sizeBytes: number; +} + +export interface PlanV2ArtifactPublisher { + /** + * Durably publish one immutable digest before resolving. `sourcePath` is + * planner-local and valid only for the lifetime of this call; remote + * adapters must finish reading/uploading it before the promise resolves. + */ + putBlob(blob: Readonly): Promise; + /** Commit the manifest only after every referenced blob is durable. */ + commitManifest(manifestBytes: string): Promise; + /** Idempotent best-effort cleanup for an unpublished or partial plan. */ + abort(): Promise; +} + +function canFallbackToCopy(error: unknown): boolean { + if (!(error instanceof Error) || !("code" in error)) return false; + const code = error.code; + return code === "EXDEV" || code === "EPERM" || code === "EACCES" || code === "ENOTSUP"; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function manifestDigests(manifestBytes: string): readonly string[] { + let value: unknown; + try { + value = JSON.parse(manifestBytes); + } catch { + throw new PlanV2IntegrityError("manifest publisher received invalid JSON"); + } + if (!isRecord(value) || !Array.isArray(value.artifacts)) { + throw new PlanV2IntegrityError("manifest publisher requires an artifacts array"); + } + return value.artifacts.map((artifact, index) => { + if (!isRecord(artifact)) { + throw new PlanV2IntegrityError(`manifest artifacts[${index}] must be an object`); + } + return assertPlanV2Sha256(artifact.sha256, `manifest artifacts[${index}].sha256`); + }); +} + +export interface LocalPlanV2ArtifactPublisherOptions { + /** Test seam for exercising cross-device/restricted-filesystem fallback. */ + readonly linkFile?: typeof linkSync; +} + +/** + * Planner-local manifest-last publisher. + * + * Same-filesystem blobs are hard-linked from the frozen source plan, so the + * staging tree and local CAS do not consume a second set of data blocks. + * Cross-device and restricted filesystems fall back to an atomic copy. This + * implementation is a compatibility adapter for local callers; distributed + * nodes exchange remote manifest/CAS locators through their cloud adapters. + */ +export class LocalPlanV2ArtifactPublisher implements PlanV2ArtifactPublisher { + readonly destinationDir: string; + readonly temporaryDir: string; + readonly #linkFile: typeof linkSync; + #committed = false; + + constructor(destinationDir: string, options: Readonly = {}) { + if (existsSync(destinationDir)) { + throw new PlanV2IntegrityError(`output directory already exists: ${destinationDir}`); + } + this.destinationDir = destinationDir; + this.#linkFile = options.linkFile ?? linkSync; + mkdirSync(dirname(destinationDir), { recursive: true }); + this.temporaryDir = mkdtempSync(join(dirname(destinationDir), ".plan-v2-publish-")); + } + + async putBlob(blob: Readonly): Promise { + const digest = assertPlanV2Sha256(blob.sha256, "published blob sha256"); + const sourceSize = statSync(blob.sourcePath).size; + if (sourceSize !== blob.sizeBytes) { + throw new PlanV2IntegrityError( + `published blob size changed for ${digest}: expected ${blob.sizeBytes}, got ${sourceSize}`, + ); + } + const destinationPath = planV2BlobPath(this.temporaryDir, digest); + if (existsSync(destinationPath)) return; + mkdirSync(dirname(destinationPath), { recursive: true }); + const stagingDir = mkdtempSync(join(dirname(destinationPath), ".plan-v2-blob-")); + const temporaryPath = join(stagingDir, "blob"); + try { + try { + this.#linkFile(blob.sourcePath, temporaryPath); + } catch (error) { + if (!canFallbackToCopy(error)) throw error; + copyFileSync(blob.sourcePath, temporaryPath); + } + renameSync(temporaryPath, destinationPath); + } finally { + rmSync(stagingDir, { recursive: true, force: true }); + } + } + + async commitManifest(manifestBytes: string): Promise { + for (const digest of new Set(manifestDigests(manifestBytes))) { + if (!existsSync(planV2BlobPath(this.temporaryDir, digest))) { + throw new PlanV2IntegrityError( + `cannot commit manifest before referenced blob is durable: ${digest}`, + ); + } + } + writeFileSync(join(this.temporaryDir, "plan.json"), manifestBytes, "utf-8"); + renameSync(this.temporaryDir, this.destinationDir); + this.#committed = true; + } + + async abort(): Promise { + if (!this.#committed) { + rmSync(this.temporaryDir, { recursive: true, force: true }); + } + } +} diff --git a/packages/producer/src/services/distributed/publicExports.test.ts b/packages/producer/src/services/distributed/publicExports.test.ts index 97c73f516f..742af19986 100644 --- a/packages/producer/src/services/distributed/publicExports.test.ts +++ b/packages/producer/src/services/distributed/publicExports.test.ts @@ -98,6 +98,9 @@ describe("@hyperframes/producer/distributed (subpath)", () => { expect(typeof distributedSubpath.getDistributedRenderCapabilities).toBe("function"); expect(typeof distributedSubpath.readPlanProtocol).toBe("function"); expect(typeof distributedSubpath.planV2).toBe("function"); + expect(typeof distributedSubpath.planV2WithPublisher).toBe("function"); + expect(typeof distributedSubpath.publishPlanV2FromV1).toBe("function"); + expect(typeof distributedSubpath.LocalPlanV2ArtifactPublisher).toBe("function"); expect(typeof distributedSubpath.renderChunkV2).toBe("function"); expect(typeof distributedSubpath.assembleV2).toBe("function"); expect(typeof distributedSubpath.readPlanV2Manifest).toBe("function"); @@ -122,6 +125,8 @@ describe("@hyperframes/producer (main entry)", () => { expect(producerIndex.PLAN_PROTOCOL_UNSUPPORTED).toBe("PLAN_PROTOCOL_UNSUPPORTED"); expect(producerIndex.PLAN_V2_INTEGRITY_UNRECOVERABLE).toBe("PLAN_V2_INTEGRITY_UNRECOVERABLE"); expect(typeof producerIndex.readPlanProtocol).toBe("function"); + expect(typeof producerIndex.planV2WithPublisher).toBe("function"); + expect(typeof producerIndex.publishPlanV2FromV1).toBe("function"); expect(typeof producerIndex.PlanV2IntegrityError).toBe("function"); expect(typeof producerIndex.PlanProtocolUnsupportedError).toBe("function"); });