Skip to content

Commit 5c5179b

Browse files
jrusso1020jrusso1020
authored
feat(gcp-cloud-run): publish plan v2 directly to GCS (#2799)
## What - publishes Plan v2 content-addressed artifacts directly from the Cloud Run planner's private staging directory to GCS - commits the Plan v2 manifest only after every referenced artifact is durable - removes the second local Plan v2 CAS directory and preserves the existing Cloud Workflows wire contract - exports a reusable GCS publisher adapter ## Why Cloud Run planner, chunk, and assembler requests are independent containers and do not share a filesystem. The distributed contract must contain only durable GCS locators. This layer makes the Plan v2 publication path object-store-native. Like the AWS parent PR, it still relies on one planner-local frozen v1 tree inside the producer; eliminating that remaining staging tree requires direct Plan v2 emission in a later layer. ## Design invariants - local paths never cross a request or worker boundary - manifest and artifact locators derive from one validated GCS output prefix - immutable objects use generation-zero conditional creation and exact digest/size verification - retries may reuse exact objects but never overwrite conflicts - the manifest is the final publication commit point - chunk and assembler requests independently download and verify only their target artifacts ## Test plan - [x] all 95 GCP Cloud Run package tests pass - [x] package typecheck passes - [x] package build passes - [x] changed-file lint, format, fallow, and repository commit gates pass - [x] end-to-end adapter test covers plan, target-scoped chunk, and assemble through Fake GCS with no shared local directory
2 parents a069730 + 74d7bfd commit 5c5179b

5 files changed

Lines changed: 356 additions & 57 deletions

File tree

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

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
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("trims an arbitrary trailing-slash run in linear time", () => {
44+
const publisher = new GcsPlanV2ArtifactPublisher({
45+
storage: asStorage(new FakeGcs()),
46+
planOutputGcsPrefix: `gs://bucket/render${"/".repeat(10_000)}`,
47+
});
48+
49+
expect(publisher.artifactPrefix).toBe("gs://bucket/render/v2/artifacts/sha256");
50+
expect(publisher.manifestUri).toBe("gs://bucket/render/v2/manifest.json");
51+
});
52+
53+
it("publishes immutable blobs before the fixed-key manifest", async () => {
54+
const source = makeSource("hello");
55+
const gcs = new FakeGcs();
56+
const artifactPrefix = "gs://bucket/render/v2/artifacts/sha256";
57+
const manifestUri = "gs://bucket/render/v2/manifest.json";
58+
const publisher = new GcsPlanV2ArtifactPublisher({
59+
storage: asStorage(gcs),
60+
planOutputGcsPrefix: "gs://bucket/render",
61+
temporaryRoot: source.root,
62+
});
63+
64+
await publisher.putBlob({
65+
sourcePath: source.path,
66+
sha256: source.digest,
67+
sizeBytes: source.sizeBytes,
68+
});
69+
const manifest = manifestFor(source.digest);
70+
await publisher.commitManifest(manifest);
71+
72+
const blobUri = `${artifactPrefix}/${source.digest.slice(0, 2)}/${source.digest}`;
73+
expect(gcs.ops.filter((operation) => operation.kind === "upload").map((op) => op.uri)).toEqual([
74+
blobUri,
75+
manifestUri,
76+
]);
77+
expect(gcs.objects.get(manifestUri)?.toString("utf8")).toBe(manifest);
78+
});
79+
80+
it("refuses to expose a manifest that references an unpublished digest", async () => {
81+
const source = makeSource("hello");
82+
const gcs = new FakeGcs();
83+
const manifestUri = "gs://bucket/render/v2/manifest.json";
84+
const publisher = new GcsPlanV2ArtifactPublisher({
85+
storage: asStorage(gcs),
86+
planOutputGcsPrefix: "gs://bucket/render",
87+
temporaryRoot: source.root,
88+
});
89+
90+
await expect(publisher.commitManifest(manifestFor(source.digest))).rejects.toMatchObject({
91+
name: "PlanV2IntegrityError",
92+
});
93+
expect(gcs.objects.has(manifestUri)).toBe(false);
94+
});
95+
96+
it("rejects malformed digests before constructing a GCS object key", async () => {
97+
const source = makeSource("hello");
98+
const gcs = new FakeGcs();
99+
const publisher = new GcsPlanV2ArtifactPublisher({
100+
storage: asStorage(gcs),
101+
planOutputGcsPrefix: "gs://bucket/render",
102+
temporaryRoot: source.root,
103+
});
104+
105+
await expect(
106+
publisher.putBlob({
107+
sourcePath: source.path,
108+
sha256: "../outside-prefix",
109+
sizeBytes: source.sizeBytes,
110+
}),
111+
).rejects.toMatchObject({ name: "PlanV2IntegrityError" });
112+
expect(gcs.ops.filter((operation) => operation.kind === "upload")).toHaveLength(0);
113+
});
114+
115+
it("reuses matching objects and rejects a conflicting fixed-key manifest", async () => {
116+
const source = makeSource("hello");
117+
const gcs = new FakeGcs();
118+
const options = {
119+
storage: asStorage(gcs),
120+
planOutputGcsPrefix: "gs://bucket/render",
121+
temporaryRoot: source.root,
122+
};
123+
const blob = {
124+
sourcePath: source.path,
125+
sha256: source.digest,
126+
sizeBytes: source.sizeBytes,
127+
};
128+
const first = new GcsPlanV2ArtifactPublisher(options);
129+
await first.putBlob(blob);
130+
await first.commitManifest(manifestFor(source.digest, "one"));
131+
132+
const retry = new GcsPlanV2ArtifactPublisher(options);
133+
await retry.putBlob(blob);
134+
await retry.commitManifest(manifestFor(source.digest, "one"));
135+
expect(gcs.ops.filter((operation) => operation.kind === "upload")).toHaveLength(2);
136+
137+
const conflict = new GcsPlanV2ArtifactPublisher(options);
138+
await conflict.putBlob(blob);
139+
await expect(conflict.commitManifest(manifestFor(source.digest, "two"))).rejects.toMatchObject({
140+
name: "PLAN_ARTIFACT_DIGEST_MISMATCH",
141+
});
142+
expect(gcs.ops.filter((operation) => operation.kind === "upload")).toHaveLength(2);
143+
});
144+
145+
it("leaves durable remote CAS blobs intact when publication aborts", async () => {
146+
const source = makeSource("hello");
147+
const gcs = new FakeGcs();
148+
const publisher = new GcsPlanV2ArtifactPublisher({
149+
storage: asStorage(gcs),
150+
planOutputGcsPrefix: "gs://bucket/render",
151+
temporaryRoot: source.root,
152+
});
153+
const blob = {
154+
sourcePath: source.path,
155+
sha256: source.digest,
156+
sizeBytes: source.sizeBytes,
157+
};
158+
159+
await publisher.putBlob(blob);
160+
await publisher.abort();
161+
await publisher.abort();
162+
expect(gcs.objects.size).toBe(1);
163+
await expect(publisher.putBlob(blob)).rejects.toMatchObject({
164+
name: "PlanV2IntegrityError",
165+
});
166+
});
167+
});

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

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
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+
let end = value.length;
55+
while (end > 0 && value.charCodeAt(end - 1) === 47) end -= 1;
56+
return value.slice(0, end);
57+
}
58+
59+
/**
60+
* Manifest-last GCS implementation of the producer's plan-v2 publication seam.
61+
*
62+
* Every path remains private to the planner container. Remote workers receive
63+
* only the manifest URI and artifact prefix and materialize their own target.
64+
*/
65+
export class GcsPlanV2ArtifactPublisher implements PlanV2ArtifactPublisher {
66+
readonly artifactPrefix: string;
67+
readonly manifestUri: string;
68+
readonly #storage: Storage;
69+
readonly #temporaryRoot: string;
70+
readonly #publishedDigests = new Set<string>();
71+
#state: "open" | "committed" | "aborted" = "open";
72+
73+
constructor(options: Readonly<GcsPlanV2ArtifactPublisherOptions>) {
74+
const outputPrefix = `${trimTrailingSlash(options.planOutputGcsPrefix)}/v2`;
75+
parseGcsUri(outputPrefix);
76+
this.#storage = options.storage;
77+
this.artifactPrefix = `${outputPrefix}/artifacts/sha256`;
78+
this.manifestUri = `${outputPrefix}/manifest.json`;
79+
this.#temporaryRoot = options.temporaryRoot ?? tmpdir();
80+
mkdirSync(this.#temporaryRoot, { recursive: true });
81+
}
82+
83+
async putBlob(blob: Readonly<PlanV2PublishBlob>): Promise<void> {
84+
this.#assertOpen("publish a blob");
85+
const digest = assertSha256(blob.sha256, "GCS published blob sha256");
86+
const sourceSize = statSync(blob.sourcePath).size;
87+
if (sourceSize !== blob.sizeBytes) {
88+
throw new PlanV2IntegrityError(
89+
`GCS published blob size changed for ${digest}: expected ${blob.sizeBytes}, got ${sourceSize}`,
90+
);
91+
}
92+
const uri = `${this.artifactPrefix}/${digest.slice(0, 2)}/${digest}`;
93+
await uploadContentAddressedFileToGcs(this.#storage, blob.sourcePath, uri, digest);
94+
this.#publishedDigests.add(digest);
95+
}
96+
97+
async commitManifest(manifestBytes: string): Promise<void> {
98+
this.#assertOpen("commit a manifest");
99+
for (const digest of manifestDigests(manifestBytes)) {
100+
if (!this.#publishedDigests.has(digest)) {
101+
throw new PlanV2IntegrityError(
102+
`cannot commit GCS manifest before referenced blob is durable: ${digest}`,
103+
);
104+
}
105+
}
106+
107+
const manifestDigest = createHash("sha256").update(manifestBytes, "utf8").digest("hex");
108+
const stagingDir = mkdtempSync(join(this.#temporaryRoot, "hf-plan-v2-manifest-"));
109+
const manifestPath = join(stagingDir, "manifest.json");
110+
try {
111+
writeFileSync(manifestPath, manifestBytes, "utf8");
112+
await uploadContentAddressedFileToGcs(
113+
this.#storage,
114+
manifestPath,
115+
this.manifestUri,
116+
manifestDigest,
117+
"application/json",
118+
);
119+
this.#state = "committed";
120+
} finally {
121+
rmSync(stagingDir, { recursive: true, force: true });
122+
}
123+
}
124+
125+
async abort(): Promise<void> {
126+
if (this.#state === "open") this.#state = "aborted";
127+
// Immutable CAS blobs may be shared with or reused by another retry.
128+
// Unreferenced blobs expire under the bucket's intermediate lifecycle.
129+
}
130+
131+
#assertOpen(operation: string): void {
132+
if (this.#state !== "open") {
133+
throw new PlanV2IntegrityError(`cannot ${operation} after publisher is ${this.#state}`);
134+
}
135+
}
136+
}

‎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)