Skip to content

Commit f899d6e

Browse files
jrusso1020jrusso1020
committed
feat(gcp-cloud-run): publish plan v2 directly to GCS
1 parent c881f23 commit f899d6e

5 files changed

Lines changed: 334 additions & 56 deletions

File tree

‎packages/gcp-cloud-run/src/gcsPlanV2Publisher.test.ts‎

Lines changed: 157 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,157 @@
1+
// fallow-ignore-file code-duplication complexity
2+
import { afterEach, describe, expect, it } from "bun:test";
3+
import { createHash } from "node:crypto";
4+
import { mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs";
5+
import { tmpdir } from "node:os";
6+
import { join } from "node:path";
7+
import { asStorage, FakeGcs } from "./__fixtures__/fakeGcs.js";
8+
import { GcsPlanV2ArtifactPublisher } from "./gcsPlanV2Publisher.js";
9+
10+
const roots: string[] = [];
11+
12+
afterEach(() => {
13+
for (const root of roots) rmSync(root, { recursive: true, force: true });
14+
roots.length = 0;
15+
});
16+
17+
function makeSource(contents: string): {
18+
readonly root: string;
19+
readonly path: string;
20+
readonly digest: string;
21+
readonly sizeBytes: number;
22+
} {
23+
const root = mkdtempSync(join(tmpdir(), "hf-gcs-plan-v2-publisher-"));
24+
roots.push(root);
25+
const path = join(root, "artifact.bin");
26+
writeFileSync(path, contents);
27+
return {
28+
root,
29+
path,
30+
digest: createHash("sha256").update(contents).digest("hex"),
31+
sizeBytes: statSync(path).size,
32+
};
33+
}
34+
35+
function manifestFor(digest: string, marker = "one"): string {
36+
return JSON.stringify({
37+
planHash: marker,
38+
artifacts: [{ path: "compiled/index.html", sha256: digest, sizeBytes: 5 }],
39+
});
40+
}
41+
42+
describe("GcsPlanV2ArtifactPublisher", () => {
43+
it("publishes immutable blobs before the fixed-key manifest", async () => {
44+
const source = makeSource("hello");
45+
const gcs = new FakeGcs();
46+
const artifactPrefix = "gs://bucket/render/v2/artifacts/sha256";
47+
const manifestUri = "gs://bucket/render/v2/manifest.json";
48+
const publisher = new GcsPlanV2ArtifactPublisher({
49+
storage: asStorage(gcs),
50+
planOutputGcsPrefix: "gs://bucket/render",
51+
temporaryRoot: source.root,
52+
});
53+
54+
await publisher.putBlob({
55+
sourcePath: source.path,
56+
sha256: source.digest,
57+
sizeBytes: source.sizeBytes,
58+
});
59+
const manifest = manifestFor(source.digest);
60+
await publisher.commitManifest(manifest);
61+
62+
const blobUri = `${artifactPrefix}/${source.digest.slice(0, 2)}/${source.digest}`;
63+
expect(gcs.ops.filter((operation) => operation.kind === "upload").map((op) => op.uri)).toEqual([
64+
blobUri,
65+
manifestUri,
66+
]);
67+
expect(gcs.objects.get(manifestUri)?.toString("utf8")).toBe(manifest);
68+
});
69+
70+
it("refuses to expose a manifest that references an unpublished digest", async () => {
71+
const source = makeSource("hello");
72+
const gcs = new FakeGcs();
73+
const manifestUri = "gs://bucket/render/v2/manifest.json";
74+
const publisher = new GcsPlanV2ArtifactPublisher({
75+
storage: asStorage(gcs),
76+
planOutputGcsPrefix: "gs://bucket/render",
77+
temporaryRoot: source.root,
78+
});
79+
80+
await expect(publisher.commitManifest(manifestFor(source.digest))).rejects.toMatchObject({
81+
name: "PlanV2IntegrityError",
82+
});
83+
expect(gcs.objects.has(manifestUri)).toBe(false);
84+
});
85+
86+
it("rejects malformed digests before constructing a GCS object key", async () => {
87+
const source = makeSource("hello");
88+
const gcs = new FakeGcs();
89+
const publisher = new GcsPlanV2ArtifactPublisher({
90+
storage: asStorage(gcs),
91+
planOutputGcsPrefix: "gs://bucket/render",
92+
temporaryRoot: source.root,
93+
});
94+
95+
await expect(
96+
publisher.putBlob({
97+
sourcePath: source.path,
98+
sha256: "../outside-prefix",
99+
sizeBytes: source.sizeBytes,
100+
}),
101+
).rejects.toMatchObject({ name: "PlanV2IntegrityError" });
102+
expect(gcs.ops.filter((operation) => operation.kind === "upload")).toHaveLength(0);
103+
});
104+
105+
it("reuses matching objects and rejects a conflicting fixed-key manifest", async () => {
106+
const source = makeSource("hello");
107+
const gcs = new FakeGcs();
108+
const options = {
109+
storage: asStorage(gcs),
110+
planOutputGcsPrefix: "gs://bucket/render",
111+
temporaryRoot: source.root,
112+
};
113+
const blob = {
114+
sourcePath: source.path,
115+
sha256: source.digest,
116+
sizeBytes: source.sizeBytes,
117+
};
118+
const first = new GcsPlanV2ArtifactPublisher(options);
119+
await first.putBlob(blob);
120+
await first.commitManifest(manifestFor(source.digest, "one"));
121+
122+
const retry = new GcsPlanV2ArtifactPublisher(options);
123+
await retry.putBlob(blob);
124+
await retry.commitManifest(manifestFor(source.digest, "one"));
125+
expect(gcs.ops.filter((operation) => operation.kind === "upload")).toHaveLength(2);
126+
127+
const conflict = new GcsPlanV2ArtifactPublisher(options);
128+
await conflict.putBlob(blob);
129+
await expect(conflict.commitManifest(manifestFor(source.digest, "two"))).rejects.toMatchObject({
130+
name: "PLAN_ARTIFACT_DIGEST_MISMATCH",
131+
});
132+
expect(gcs.ops.filter((operation) => operation.kind === "upload")).toHaveLength(2);
133+
});
134+
135+
it("leaves durable remote CAS blobs intact when publication aborts", async () => {
136+
const source = makeSource("hello");
137+
const gcs = new FakeGcs();
138+
const publisher = new GcsPlanV2ArtifactPublisher({
139+
storage: asStorage(gcs),
140+
planOutputGcsPrefix: "gs://bucket/render",
141+
temporaryRoot: source.root,
142+
});
143+
const blob = {
144+
sourcePath: source.path,
145+
sha256: source.digest,
146+
sizeBytes: source.sizeBytes,
147+
};
148+
149+
await publisher.putBlob(blob);
150+
await publisher.abort();
151+
await publisher.abort();
152+
expect(gcs.objects.size).toBe(1);
153+
await expect(publisher.putBlob(blob)).rejects.toMatchObject({
154+
name: "PlanV2IntegrityError",
155+
});
156+
});
157+
});

‎packages/gcp-cloud-run/src/gcsPlanV2Publisher.ts‎

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
// fallow-ignore-file code-duplication
2+
import { createHash } from "node:crypto";
3+
import { mkdirSync, mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs";
4+
import { tmpdir } from "node:os";
5+
import { join } from "node:path";
6+
import type { Storage } from "@google-cloud/storage";
7+
import {
8+
PlanV2IntegrityError,
9+
type PlanV2ArtifactPublisher,
10+
type PlanV2PublishBlob,
11+
} from "@hyperframes/producer/distributed";
12+
import { parseGcsUri, uploadContentAddressedFileToGcs } from "./gcsTransport.js";
13+
14+
export interface GcsPlanV2ArtifactPublisherOptions {
15+
readonly storage: Storage;
16+
/** Validated render output prefix from which all v2 object keys are derived. */
17+
readonly planOutputGcsPrefix: string;
18+
/** Planner-local scratch parent for the small manifest upload file. */
19+
readonly temporaryRoot?: string;
20+
}
21+
22+
function isRecord(value: unknown): value is Record<string, unknown> {
23+
return value !== null && typeof value === "object" && !Array.isArray(value);
24+
}
25+
26+
function assertSha256(value: unknown, label: string): string {
27+
if (typeof value !== "string" || !/^[a-f0-9]{64}$/.test(value)) {
28+
throw new PlanV2IntegrityError(`${label} must be a lowercase SHA-256 digest`);
29+
}
30+
return value;
31+
}
32+
33+
function manifestDigests(manifestBytes: string): ReadonlySet<string> {
34+
let value: unknown;
35+
try {
36+
value = JSON.parse(manifestBytes);
37+
} catch {
38+
throw new PlanV2IntegrityError("GCS publisher received invalid manifest JSON");
39+
}
40+
if (!isRecord(value) || !Array.isArray(value.artifacts)) {
41+
throw new PlanV2IntegrityError("GCS publisher manifest requires an artifacts array");
42+
}
43+
return new Set(
44+
value.artifacts.map((artifact, index) => {
45+
if (!isRecord(artifact)) {
46+
throw new PlanV2IntegrityError(`GCS publisher artifacts[${index}] must be an object`);
47+
}
48+
return assertSha256(artifact.sha256, `GCS publisher artifacts[${index}].sha256`);
49+
}),
50+
);
51+
}
52+
53+
function trimTrailingSlash(value: string): string {
54+
return value.replace(/\/+$/, "");
55+
}
56+
57+
/**
58+
* Manifest-last GCS implementation of the producer's plan-v2 publication seam.
59+
*
60+
* Every path remains private to the planner container. Remote workers receive
61+
* only the manifest URI and artifact prefix and materialize their own target.
62+
*/
63+
export class GcsPlanV2ArtifactPublisher implements PlanV2ArtifactPublisher {
64+
readonly artifactPrefix: string;
65+
readonly manifestUri: string;
66+
readonly #storage: Storage;
67+
readonly #temporaryRoot: string;
68+
readonly #publishedDigests = new Set<string>();
69+
#state: "open" | "committed" | "aborted" = "open";
70+
71+
constructor(options: Readonly<GcsPlanV2ArtifactPublisherOptions>) {
72+
const outputPrefix = `${trimTrailingSlash(options.planOutputGcsPrefix)}/v2`;
73+
parseGcsUri(outputPrefix);
74+
this.#storage = options.storage;
75+
this.artifactPrefix = `${outputPrefix}/artifacts/sha256`;
76+
this.manifestUri = `${outputPrefix}/manifest.json`;
77+
this.#temporaryRoot = options.temporaryRoot ?? tmpdir();
78+
mkdirSync(this.#temporaryRoot, { recursive: true });
79+
}
80+
81+
async putBlob(blob: Readonly<PlanV2PublishBlob>): Promise<void> {
82+
this.#assertOpen("publish a blob");
83+
const digest = assertSha256(blob.sha256, "GCS published blob sha256");
84+
const sourceSize = statSync(blob.sourcePath).size;
85+
if (sourceSize !== blob.sizeBytes) {
86+
throw new PlanV2IntegrityError(
87+
`GCS published blob size changed for ${digest}: expected ${blob.sizeBytes}, got ${sourceSize}`,
88+
);
89+
}
90+
const uri = `${this.artifactPrefix}/${digest.slice(0, 2)}/${digest}`;
91+
await uploadContentAddressedFileToGcs(this.#storage, blob.sourcePath, uri, digest);
92+
this.#publishedDigests.add(digest);
93+
}
94+
95+
async commitManifest(manifestBytes: string): Promise<void> {
96+
this.#assertOpen("commit a manifest");
97+
for (const digest of manifestDigests(manifestBytes)) {
98+
if (!this.#publishedDigests.has(digest)) {
99+
throw new PlanV2IntegrityError(
100+
`cannot commit GCS manifest before referenced blob is durable: ${digest}`,
101+
);
102+
}
103+
}
104+
105+
const manifestDigest = createHash("sha256").update(manifestBytes, "utf8").digest("hex");
106+
const stagingDir = mkdtempSync(join(this.#temporaryRoot, "hf-plan-v2-manifest-"));
107+
const manifestPath = join(stagingDir, "manifest.json");
108+
try {
109+
writeFileSync(manifestPath, manifestBytes, "utf8");
110+
await uploadContentAddressedFileToGcs(
111+
this.#storage,
112+
manifestPath,
113+
this.manifestUri,
114+
manifestDigest,
115+
"application/json",
116+
);
117+
this.#state = "committed";
118+
} finally {
119+
rmSync(stagingDir, { recursive: true, force: true });
120+
}
121+
}
122+
123+
async abort(): Promise<void> {
124+
if (this.#state === "open") this.#state = "aborted";
125+
// Immutable CAS blobs may be shared with or reused by another retry.
126+
// Unreferenced blobs expire under the bucket's intermediate lifecycle.
127+
}
128+
129+
#assertOpen(operation: string): void {
130+
if (this.#state !== "open") {
131+
throw new PlanV2IntegrityError(`cannot ${operation} after publisher is ${this.#state}`);
132+
}
133+
}
134+
}

‎packages/gcp-cloud-run/src/index.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,10 @@ export {
5353
uploadContentAddressedFileToGcs,
5454
uploadFileToGcs,
5555
} from "./gcsTransport.js";
56+
export {
57+
GcsPlanV2ArtifactPublisher,
58+
type GcsPlanV2ArtifactPublisherOptions,
59+
} from "./gcsPlanV2Publisher.js";
5660

5761
// ── Client-side SDK ─────────────────────────────────────────────────────────
5862
export { deploySite, type DeploySiteOptions, type SiteHandle } from "./sdk/deploySite.js";

‎packages/gcp-cloud-run/src/server.test.ts‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -18,17 +18,18 @@ import { afterEach, describe, expect, it } from "bun:test";
1818
import { createHash } from "node:crypto";
1919
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
2020
import { tmpdir } from "node:os";
21-
import { join } from "node:path";
21+
import { dirname, join } from "node:path";
2222
import {
2323
CURRENT_PLAN_PROTOCOL,
24-
createPlanV2FromV1,
2524
PLAN_V2_INTEGRITY_UNRECOVERABLE,
2625
PlanV2IntegrityError,
2726
PlanProtocolUnsupportedError,
2827
type AssembleResult,
2928
type ChunkResult,
3029
type PlanResult,
31-
type PlanV2Result,
30+
type PlanV2ArtifactPublisher,
31+
type PlanV2Manifest,
32+
publishPlanV2FromV1,
3233
} from "@hyperframes/producer/distributed";
3334
import { recomputePlanHashFromPlanDir } from "../../producer/src/services/render/stages/freezePlan.js";
3435
import { asStorage, FakeGcs } from "./__fixtures__/fakeGcs.js";
@@ -244,14 +245,18 @@ describe("dispatch", () => {
244245
const gcs = new FakeGcs();
245246
await seedProjectTar(gcs, "gs://b/sites/v2/project.tar.gz");
246247
const root = mkTmp("hf-v2-e2e-");
247-
const planV2 = async (
248-
_projectDir: string,
248+
const planV2WithPublisher = async (
249+
projectDir: string,
249250
_config: unknown,
250-
planV2Dir: string,
251-
): Promise<PlanV2Result> => {
251+
publisher: PlanV2ArtifactPublisher,
252+
options: Readonly<{ stagingParentDir?: string }>,
253+
): Promise<PlanV2Manifest> => {
252254
const v1Dir = join(root, "v1");
253255
makeMinimalV1PlanDir(v1Dir, true);
254-
return createPlanV2FromV1(v1Dir, planV2Dir);
256+
const manifest = await publishPlanV2FromV1(v1Dir, publisher);
257+
expect(options.stagingParentDir).toBe(dirname(projectDir));
258+
expect(existsSync(join(dirname(projectDir), "plan-v2"))).toBe(false);
259+
return manifest;
255260
};
256261
const renderChunk = async (
257262
planDir: string,
@@ -278,7 +283,7 @@ describe("dispatch", () => {
278283
writeFileSync(finalOutput, "v2-output");
279284
return { framesEncoded: 30, fileSize: 9 };
280285
};
281-
const deps = depsWith(gcs, { planV2, renderChunk, assemble });
286+
const deps = depsWith(gcs, { planV2WithPublisher, renderChunk, assemble });
282287

283288
const planned = await dispatch(
284289
{

0 commit comments

Comments
 (0)