diff --git a/packages/aws-lambda/src/handler.test.ts b/packages/aws-lambda/src/handler.test.ts index f67314eb30..0b1787a11b 100644 --- a/packages/aws-lambda/src/handler.test.ts +++ b/packages/aws-lambda/src/handler.test.ts @@ -1,3 +1,4 @@ +// fallow-ignore-file code-duplication complexity /** * Handler dispatch unit tests. * @@ -18,14 +19,15 @@ import { afterEach, beforeEach, describe, expect, it, mock } from "bun:test"; import { createHash } from "node:crypto"; import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { dirname, join } from "node:path"; import { CURRENT_PLAN_PROTOCOL, - createPlanV2FromV1, type AssembleResult, type ChunkResult, type PlanResult, - type PlanV2Result, + type PlanV2ArtifactPublisher, + type PlanV2Manifest, + publishPlanV2FromV1, } from "@hyperframes/producer/distributed"; import { recomputePlanHashFromPlanDir } from "../../producer/src/services/render/stages/freezePlan.js"; import type { AssembleEvent, LambdaEvent, PlanEvent, RenderChunkEvent } from "./events.js"; @@ -528,11 +530,19 @@ describe("handler dispatch", () => { const s3 = new FakeS3Client(); s3.objects.set("s3://bucket/project.tar.gz", await makeMinimalProjectTar()); - const planV2Mock = mock( - async (_projectDir: string, _config: unknown, planV2Dir: string): Promise => { + const planV2WithPublisherMock = mock( + async ( + projectDir: string, + _config: unknown, + publisher: PlanV2ArtifactPublisher, + options: Readonly<{ stagingParentDir?: string }>, + ): Promise => { const v1Dir = join(tmpRoot, `v1-${Date.now()}`); makeMinimalV1PlanDir(v1Dir, true); - return createPlanV2FromV1(v1Dir, planV2Dir); + const manifest = await publishPlanV2FromV1(v1Dir, publisher); + expect(options.stagingParentDir).toBe(dirname(projectDir)); + expect(existsSync(join(dirname(projectDir), "plan-v2"))).toBe(false); + return manifest; }, ); const renderChunkMock = mock( @@ -568,7 +578,8 @@ describe("handler dispatch", () => { plan: mock(async () => { throw new Error("v1 plan should not be called"); }) as unknown as typeof import("@hyperframes/producer/distributed").plan, - planV2: planV2Mock as unknown as typeof import("@hyperframes/producer/distributed").planV2, + planV2WithPublisher: + planV2WithPublisherMock as unknown as typeof import("@hyperframes/producer/distributed").planV2WithPublisher, renderChunk: renderChunkMock as unknown as typeof import("@hyperframes/producer/distributed").renderChunk, assemble: @@ -597,6 +608,13 @@ describe("handler dispatch", () => { if (!("PlanProtocol" in planned) || planned.PlanProtocol !== "v2") { throw new Error("expected v2 plan result"); } + const planUploads = s3.ops.filter((operation) => operation.kind === "upload"); + expect(planUploads.at(-1)?.uri).toBe(planned.PlanV2ManifestS3Uri); + expect( + planUploads + .slice(0, -1) + .every((operation) => operation.uri.startsWith(`${planned.PlanV2ArtifactS3Prefix}/`)), + ).toBe(true); const beforeChunk = s3.ops.length; const chunk = await handler( diff --git a/packages/aws-lambda/src/handler.ts b/packages/aws-lambda/src/handler.ts index 62476255dd..9f2d6b97bf 100644 --- a/packages/aws-lambda/src/handler.ts +++ b/packages/aws-lambda/src/handler.ts @@ -23,11 +23,11 @@ import { listPlanV2ArtifactsForTarget, materializePlanV2Target, plan, - planV2, + planV2WithPublisher, type PlanResult, type PlanV2Artifact, + type PlanV2Manifest, type PlanV2MaterializationTarget, - type PlanV2Result, readPlanV2Manifest, renderChunk, } from "@hyperframes/producer/distributed"; @@ -48,12 +48,11 @@ import { downloadS3ObjectToFile, downloadS3ObjectToFileVerified, parseS3Uri, - sha256File, tarDirectory, untarDirectory, - uploadContentAddressedFileToS3, uploadFileToS3, } from "./s3Transport.js"; +import { S3PlanV2ArtifactPublisher } from "./s3PlanV2Publisher.js"; /** * Lazily-constructed S3 client. Cached at module scope so warm Lambda @@ -77,7 +76,7 @@ export interface HandlerDeps { s3?: S3Client; primitives?: { plan: typeof plan; - planV2?: typeof planV2; + planV2WithPublisher?: typeof planV2WithPublisher; renderChunk: typeof renderChunk; assemble: typeof assemble; }; @@ -266,6 +265,8 @@ function primeRuntimeEnv(): void { // ── Plan ──────────────────────────────────────────────────────────────────── +// The v1 handler owns one transactional download, plan, archive, upload, and cleanup lifecycle. +// fallow-ignore-next-line complexity async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise { if (event.PlanProtocol === "v2") { return handlePlanV2(event, deps); @@ -350,7 +351,7 @@ async function handlePlanV2( ): Promise> { const started = Date.now(); const s3 = deps?.s3 ?? getS3Client(); - const primitive = deps?.primitives?.planV2 ?? planV2; + const primitive = deps?.primitives?.planV2WithPublisher ?? planV2WithPublisher; if (!deps?.skipChromeResolution && !process.env.PRODUCER_HEADLESS_SHELL_PATH) { process.env.PRODUCER_HEADLESS_SHELL_PATH = await resolveChromeExecutablePath(); } @@ -358,57 +359,34 @@ async function handlePlanV2( const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-lambda-plan-v2-")); const projectArchive = join(work, "project.tar.gz"); const projectDir = join(work, "project"); - const planV2Dir = join(work, "plan-v2"); try { await downloadS3ObjectToFile(s3, event.ProjectS3Uri, projectArchive); await untarDirectory(projectArchive, projectDir); - const result: PlanV2Result = await primitive(projectDir, { ...event.Config }, planV2Dir); - const manifest = readPlanV2Manifest(planV2Dir); - if (manifest.planHash !== result.planHash) { - throwPlanHashMismatch(result.planHash, manifest.planHash); - } - - const outputPrefix = `${trimTrailingSlash(event.PlanOutputS3Prefix)}/v2`; - const artifactPrefix = `${outputPrefix}/artifacts/sha256`; - const uniqueArtifacts = [ - ...new Map(manifest.artifacts.map((artifact) => [artifact.sha256, artifact])).values(), - ]; - await mapConcurrent(uniqueArtifacts, 16, async (artifact) => { - const localPath = planV2BlobPath(planV2Dir, artifact.sha256); - await uploadContentAddressedFileToS3( - s3, - localPath, - planV2BlobUri(artifactPrefix, artifact.sha256), - artifact.sha256, - ); - }); - - // Publish the manifest only after every referenced blob is durable. - const manifestUri = `${outputPrefix}/manifest.json`; - await uploadContentAddressedFileToS3( + const publisher = new S3PlanV2ArtifactPublisher({ s3, - result.manifestPath, - manifestUri, - await sha256File(result.manifestPath), - "application/json", - ); + planOutputS3Prefix: event.PlanOutputS3Prefix, + temporaryRoot: work, + }); + const manifest: PlanV2Manifest = await primitive(projectDir, { ...event.Config }, publisher, { + stagingParentDir: work, + }); return { Action: "plan", PlanProtocol: "v2", - PlanV2ManifestS3Uri: manifestUri, - PlanV2ArtifactS3Prefix: artifactPrefix, - PlanHash: result.planHash, - ChunkCount: result.chunkCount, - TotalFrames: result.totalFrames, - Fps: result.fps, - Width: result.width, - Height: result.height, - Format: result.format, + PlanV2ManifestS3Uri: publisher.manifestUri, + PlanV2ArtifactS3Prefix: publisher.artifactPrefix, + PlanHash: manifest.planHash, + ChunkCount: manifest.chunkCount, + TotalFrames: manifest.totalFrames, + Fps: manifest.fps, + Width: manifest.width, + Height: manifest.height, + Format: manifest.format, HasAudio: manifest.artifacts.some((artifact) => artifact.path === "audio.aac"), AudioS3Uri: null, - FfmpegVersion: result.ffmpegVersion, - ProducerVersion: result.producerVersion, + FfmpegVersion: manifest.ffmpegVersion, + ProducerVersion: manifest.producerVersion, DurationMs: Date.now() - started, }; } finally { @@ -728,7 +706,16 @@ async function mapConcurrent( await fn(values[index]!); } } - await Promise.all(Array.from({ length: Math.min(concurrency, values.length) }, () => worker())); + const results = await Promise.allSettled( + Array.from({ length: Math.min(concurrency, values.length) }, () => worker()), + ); + const failure = results.find( + (result): result is PromiseRejectedResult => result.status === "rejected", + ); + // Do not reject while sibling workers may still be writing into invocation + // scratch. The caller removes that directory in `finally`; draining the pool + // first prevents late S3 streams from racing cleanup after another GET fails. + if (failure) throw failure.reason; } async function downloadChunkObjects( diff --git a/packages/aws-lambda/src/index.ts b/packages/aws-lambda/src/index.ts index eb035249dc..144e861a69 100644 --- a/packages/aws-lambda/src/index.ts +++ b/packages/aws-lambda/src/index.ts @@ -65,6 +65,10 @@ export { uploadContentAddressedFileToS3, uploadFileToS3, } from "./s3Transport.js"; +export { + S3PlanV2ArtifactPublisher, + type S3PlanV2ArtifactPublisherOptions, +} from "./s3PlanV2Publisher.js"; // ── Client-side SDK ───────────────────────────────────────────────────────── export { deploySite, type DeploySiteOptions, type SiteHandle } from "./sdk/deploySite.js"; diff --git a/packages/aws-lambda/src/s3PlanV2Publisher.test.ts b/packages/aws-lambda/src/s3PlanV2Publisher.test.ts new file mode 100644 index 0000000000..94a35a6e68 --- /dev/null +++ b/packages/aws-lambda/src/s3PlanV2Publisher.test.ts @@ -0,0 +1,229 @@ +// fallow-ignore-file code-duplication complexity +import { afterEach, describe, expect, it } from "bun:test"; +import { createHash } from "node:crypto"; +import { mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { S3PlanV2ArtifactPublisher } from "./s3PlanV2Publisher.js"; + +interface StoredObject { + readonly bytes: Buffer; + readonly sha256: string; +} + +interface PutOperation { + readonly uri: string; + readonly ifNoneMatch?: string; +} + +class FakeS3 { + readonly objects = new Map(); + readonly puts: PutOperation[] = []; + + asClient(): import("@aws-sdk/client-s3").S3Client { + return this as unknown as import("@aws-sdk/client-s3").S3Client; + } + + async send(command: unknown): Promise { + // AWS command inputs are the runtime boundary exercised by this fake. + const value = command as unknown as { + readonly constructor: { readonly name: string }; + readonly input: { + readonly Bucket: string; + readonly Key: string; + readonly Body?: NodeJS.ReadableStream; + readonly Metadata?: Record; + readonly IfNoneMatch?: string; + }; + }; + const uri = `s3://${value.input.Bucket}/${value.input.Key}`; + if (value.constructor.name === "HeadObjectCommand") { + const object = this.objects.get(uri); + if (!object) { + const error = new Error("not found"); + error.name = "NotFound"; + Object.assign(error, { $metadata: { httpStatusCode: 404 } }); + throw error; + } + return { + ContentLength: object.bytes.length, + Metadata: { sha256: object.sha256 }, + }; + } + if (value.constructor.name === "PutObjectCommand") { + if (value.input.IfNoneMatch === "*" && this.objects.has(uri)) { + const error = new Error("precondition failed"); + error.name = "PreconditionFailed"; + Object.assign(error, { $metadata: { httpStatusCode: 412 } }); + throw error; + } + const chunks: Buffer[] = []; + for await (const chunk of value.input.Body ?? []) chunks.push(Buffer.from(chunk)); + const bytes = Buffer.concat(chunks); + this.objects.set(uri, { + bytes, + sha256: value.input.Metadata?.sha256 ?? "", + }); + this.puts.push({ uri, ifNoneMatch: value.input.IfNoneMatch }); + return {}; + } + throw new Error(`unexpected command ${value.constructor.name}`); + } +} + +const roots: string[] = []; + +afterEach(() => { + for (const root of roots) rmSync(root, { recursive: true, force: true }); + roots.length = 0; +}); + +function makeSource(contents: string): { + readonly root: string; + readonly path: string; + readonly digest: string; + readonly sizeBytes: number; +} { + const root = mkdtempSync(join(tmpdir(), "hf-s3-plan-v2-publisher-")); + roots.push(root); + const path = join(root, "artifact.bin"); + writeFileSync(path, contents); + return { + root, + path, + digest: createHash("sha256").update(contents).digest("hex"), + sizeBytes: statSync(path).size, + }; +} + +function manifestFor(digest: string, marker = "one"): string { + return JSON.stringify({ + planHash: marker, + artifacts: [{ path: "compiled/index.html", sha256: digest, sizeBytes: 5 }], + }); +} + +describe("S3PlanV2ArtifactPublisher", () => { + it("trims an arbitrary trailing-slash run in linear time", () => { + const publisher = new S3PlanV2ArtifactPublisher({ + s3: new FakeS3().asClient(), + planOutputS3Prefix: `s3://bucket/render${"/".repeat(10_000)}`, + }); + + expect(publisher.artifactPrefix).toBe("s3://bucket/render/v2/artifacts/sha256"); + expect(publisher.manifestUri).toBe("s3://bucket/render/v2/manifest.json"); + }); + + it("publishes immutable blobs before the fixed-key manifest", async () => { + const source = makeSource("hello"); + const s3 = new FakeS3(); + const artifactPrefix = "s3://bucket/render/v2/artifacts/sha256"; + const manifestUri = "s3://bucket/render/v2/manifest.json"; + const publisher = new S3PlanV2ArtifactPublisher({ + s3: s3.asClient(), + planOutputS3Prefix: "s3://bucket/render", + temporaryRoot: source.root, + }); + + await publisher.putBlob({ + sourcePath: source.path, + sha256: source.digest, + sizeBytes: source.sizeBytes, + }); + const manifest = manifestFor(source.digest); + await publisher.commitManifest(manifest); + + const blobUri = `${artifactPrefix}/${source.digest.slice(0, 2)}/${source.digest}`; + expect(s3.puts.map((operation) => operation.uri)).toEqual([blobUri, manifestUri]); + expect(s3.puts.every((operation) => operation.ifNoneMatch === "*")).toBe(true); + expect(s3.objects.get(manifestUri)?.bytes.toString("utf8")).toBe(manifest); + }); + + it("refuses to expose a manifest that references an unpublished digest", async () => { + const source = makeSource("hello"); + const s3 = new FakeS3(); + const manifestUri = "s3://bucket/render/v2/manifest.json"; + const publisher = new S3PlanV2ArtifactPublisher({ + s3: s3.asClient(), + planOutputS3Prefix: "s3://bucket/render", + temporaryRoot: source.root, + }); + + await expect(publisher.commitManifest(manifestFor(source.digest))).rejects.toMatchObject({ + name: "PlanV2IntegrityError", + }); + expect(s3.objects.has(manifestUri)).toBe(false); + }); + + it("rejects malformed digests before constructing an S3 object key", async () => { + const source = makeSource("hello"); + const s3 = new FakeS3(); + const publisher = new S3PlanV2ArtifactPublisher({ + s3: s3.asClient(), + planOutputS3Prefix: "s3://bucket/render", + temporaryRoot: source.root, + }); + + await expect( + publisher.putBlob({ + sourcePath: source.path, + sha256: "../outside-prefix", + sizeBytes: source.sizeBytes, + }), + ).rejects.toMatchObject({ name: "PlanV2IntegrityError" }); + expect(s3.puts).toHaveLength(0); + }); + + it("reuses matching objects and rejects a conflicting fixed-key manifest", async () => { + const source = makeSource("hello"); + const s3 = new FakeS3(); + const options = { + s3: s3.asClient(), + planOutputS3Prefix: "s3://bucket/render", + temporaryRoot: source.root, + }; + const blob = { + sourcePath: source.path, + sha256: source.digest, + sizeBytes: source.sizeBytes, + }; + const first = new S3PlanV2ArtifactPublisher(options); + await first.putBlob(blob); + await first.commitManifest(manifestFor(source.digest, "one")); + + const retry = new S3PlanV2ArtifactPublisher(options); + await retry.putBlob(blob); + await retry.commitManifest(manifestFor(source.digest, "one")); + expect(s3.puts).toHaveLength(2); + + const conflict = new S3PlanV2ArtifactPublisher(options); + await conflict.putBlob(blob); + await expect(conflict.commitManifest(manifestFor(source.digest, "two"))).rejects.toMatchObject({ + name: "PLAN_ARTIFACT_DIGEST_MISMATCH", + }); + expect(s3.puts).toHaveLength(2); + }); + + it("leaves durable remote CAS blobs intact when publication aborts", async () => { + const source = makeSource("hello"); + const s3 = new FakeS3(); + const publisher = new S3PlanV2ArtifactPublisher({ + s3: s3.asClient(), + planOutputS3Prefix: "s3://bucket/render", + temporaryRoot: source.root, + }); + const blob = { + sourcePath: source.path, + sha256: source.digest, + sizeBytes: source.sizeBytes, + }; + + await publisher.putBlob(blob); + await publisher.abort(); + await publisher.abort(); + expect(s3.objects.size).toBe(1); + await expect(publisher.putBlob(blob)).rejects.toMatchObject({ + name: "PlanV2IntegrityError", + }); + }); +}); diff --git a/packages/aws-lambda/src/s3PlanV2Publisher.ts b/packages/aws-lambda/src/s3PlanV2Publisher.ts new file mode 100644 index 0000000000..afcb45928e --- /dev/null +++ b/packages/aws-lambda/src/s3PlanV2Publisher.ts @@ -0,0 +1,137 @@ +import { createHash } from "node:crypto"; +import { mkdirSync, mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { S3Client } from "@aws-sdk/client-s3"; +import { + PlanV2IntegrityError, + type PlanV2ArtifactPublisher, + type PlanV2PublishBlob, +} from "@hyperframes/producer/distributed"; +import { parseS3Uri, uploadContentAddressedFileToS3 } from "./s3Transport.js"; + +export interface S3PlanV2ArtifactPublisherOptions { + readonly s3: S3Client; + /** Validated render output prefix from which all v2 object keys are derived. */ + readonly planOutputS3Prefix: string; + /** Planner-local scratch parent for the small manifest upload file. */ + readonly temporaryRoot?: string; +} + +function isRecord(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +function assertSha256(value: unknown, label: string): string { + if (typeof value !== "string" || !/^[a-f0-9]{64}$/.test(value)) { + throw new PlanV2IntegrityError(`${label} must be a lowercase SHA-256 digest`); + } + return value; +} + +function manifestDigests(manifestBytes: string): ReadonlySet { + let value: unknown; + try { + value = JSON.parse(manifestBytes); + } catch { + throw new PlanV2IntegrityError("S3 publisher received invalid manifest JSON"); + } + if (!isRecord(value) || !Array.isArray(value.artifacts)) { + throw new PlanV2IntegrityError("S3 publisher manifest requires an artifacts array"); + } + return new Set( + value.artifacts.map((artifact, index) => { + if (!isRecord(artifact)) { + throw new PlanV2IntegrityError(`S3 publisher artifacts[${index}] must be an object`); + } + return assertSha256(artifact.sha256, `S3 publisher artifacts[${index}].sha256`); + }), + ); +} + +function trimTrailingSlash(value: string): string { + let end = value.length; + while (end > 0 && value.charCodeAt(end - 1) === 47) end -= 1; + return value.slice(0, end); +} + +/** + * Manifest-last S3 implementation of the producer's plan-v2 publication seam. + * + * Blobs stream from the planner's private frozen directory directly to S3. + * Successfully uploaded or safely reused digests are tracked so a manifest + * cannot become visible before all of its references are durable. + */ +export class S3PlanV2ArtifactPublisher implements PlanV2ArtifactPublisher { + readonly artifactPrefix: string; + readonly manifestUri: string; + readonly #s3: S3Client; + readonly #temporaryRoot: string; + readonly #publishedDigests = new Set(); + #state: "open" | "committed" | "aborted" = "open"; + + constructor(options: Readonly) { + const outputPrefix = `${trimTrailingSlash(options.planOutputS3Prefix)}/v2`; + parseS3Uri(outputPrefix); + this.#s3 = options.s3; + this.artifactPrefix = `${outputPrefix}/artifacts/sha256`; + this.manifestUri = `${outputPrefix}/manifest.json`; + this.#temporaryRoot = options.temporaryRoot ?? tmpdir(); + mkdirSync(this.#temporaryRoot, { recursive: true }); + } + + async putBlob(blob: Readonly): Promise { + this.#assertOpen("publish a blob"); + const digest = assertSha256(blob.sha256, "S3 published blob sha256"); + const sourceSize = statSync(blob.sourcePath).size; + if (sourceSize !== blob.sizeBytes) { + throw new PlanV2IntegrityError( + `S3 published blob size changed for ${digest}: expected ${blob.sizeBytes}, got ${sourceSize}`, + ); + } + const uri = `${this.artifactPrefix}/${digest.slice(0, 2)}/${digest}`; + await uploadContentAddressedFileToS3(this.#s3, blob.sourcePath, uri, digest); + this.#publishedDigests.add(digest); + } + + async commitManifest(manifestBytes: string): Promise { + this.#assertOpen("commit a manifest"); + for (const digest of manifestDigests(manifestBytes)) { + if (!this.#publishedDigests.has(digest)) { + throw new PlanV2IntegrityError( + `cannot commit S3 manifest before referenced blob is durable: ${digest}`, + ); + } + } + + const manifestDigest = createHash("sha256").update(manifestBytes, "utf8").digest("hex"); + const stagingDir = mkdtempSync(join(this.#temporaryRoot, "hf-plan-v2-manifest-")); + const manifestPath = join(stagingDir, "manifest.json"); + try { + writeFileSync(manifestPath, manifestBytes, "utf8"); + await uploadContentAddressedFileToS3( + this.#s3, + manifestPath, + this.manifestUri, + manifestDigest, + "application/json", + ); + this.#state = "committed"; + } finally { + rmSync(stagingDir, { recursive: true, force: true }); + } + } + + async abort(): Promise { + if (this.#state === "open") this.#state = "aborted"; + // Remote CAS blobs are immutable and may already be reused by a retry. + // Without a committed manifest they are unreachable and expire under the + // render bucket's intermediate-object lifecycle policy. + } + + #assertOpen(operation: string): void { + if (this.#state !== "open") { + throw new PlanV2IntegrityError(`cannot ${operation} after publisher is ${this.#state}`); + } + } +} diff --git a/packages/aws-lambda/src/s3Transport.test.ts b/packages/aws-lambda/src/s3Transport.test.ts index ea03aea097..cd101ad090 100644 --- a/packages/aws-lambda/src/s3Transport.test.ts +++ b/packages/aws-lambda/src/s3Transport.test.ts @@ -1,3 +1,4 @@ +// fallow-ignore-file code-duplication complexity /** * Unit tests for the S3 URI parser + tar helpers. Real S3 network calls * are covered by the dispatch tests in `handler.test.ts` via a fake @@ -132,6 +133,35 @@ describe("content-addressed v2 artifacts", () => { expect(s3.putCount).toBe(0); }); + it("reuses a matching object won by a concurrent conditional create", async () => { + const source = join(scratchRoot, "artifact-race.bin"); + writeFileSync(source, "race-safe bytes"); + const digest = await sha256File(source); + const uri = `s3://bucket/v2/artifacts/sha256/${digest.slice(0, 2)}/${digest}`; + const s3 = new ContentAddressedFakeS3(); + s3.raceOnNextPut = { bytes: Buffer.from("race-safe bytes"), sha256: digest }; + + expect(await uploadContentAddressedFileToS3(s3.asClient(), source, uri, digest)).toBe("reused"); + expect(s3.putCount).toBe(0); + }); + + it("rejects a conflicting object won by a concurrent conditional create", async () => { + const source = join(scratchRoot, "artifact-race-conflict.bin"); + writeFileSync(source, "race-safe bytes"); + const digest = await sha256File(source); + const uri = `s3://bucket/v2/artifacts/sha256/${digest.slice(0, 2)}/${digest}`; + const s3 = new ContentAddressedFakeS3(); + s3.raceOnNextPut = { + bytes: Buffer.alloc(Buffer.byteLength("race-safe bytes"), "x"), + sha256: "0".repeat(64), + }; + + await expect( + uploadContentAddressedFileToS3(s3.asClient(), source, uri, digest), + ).rejects.toMatchObject({ name: "PLAN_ARTIFACT_DIGEST_MISMATCH" }); + expect(s3.putCount).toBe(0); + }); + it("deletes a downloaded artifact when digest verification fails", async () => { const expectedSource = join(scratchRoot, "artifact-expected.bin"); const destination = join(scratchRoot, "artifact-download.bin"); @@ -152,6 +182,7 @@ describe("content-addressed v2 artifacts", () => { class ContentAddressedFakeS3 { readonly objects = new Map(); putCount = 0; + raceOnNextPut: { bytes: Buffer; sha256: string } | undefined; asClient(): import("@aws-sdk/client-s3").S3Client { return this as unknown as import("@aws-sdk/client-s3").S3Client; @@ -193,6 +224,17 @@ class ContentAddressedFakeS3 { return { Body: Readable.from([object.bytes]) }; } if (value.constructor.name === "PutObjectCommand") { + if (this.raceOnNextPut) { + this.objects.set(uri, this.raceOnNextPut); + this.raceOnNextPut = undefined; + if (value.input.Body && "destroy" in value.input.Body) { + value.input.Body.destroy(); + } + const error = new Error("precondition failed"); + error.name = "PreconditionFailed"; + Object.assign(error, { $metadata: { httpStatusCode: 412 } }); + throw error; + } const chunks: Buffer[] = []; for await (const chunk of value.input.Body ?? []) chunks.push(Buffer.from(chunk)); const bytes = Buffer.concat(chunks); diff --git a/packages/aws-lambda/src/s3Transport.ts b/packages/aws-lambda/src/s3Transport.ts index becfe3828e..5efbac7be1 100644 --- a/packages/aws-lambda/src/s3Transport.ts +++ b/packages/aws-lambda/src/s3Transport.ts @@ -159,34 +159,42 @@ export async function uploadContentAddressedFileToS3( const { bucket, key } = parseS3Uri(uri); const size = statSync(localPath).size; + const existing = await inspectContentAddressedObject(client, bucket, key, size, expectedSha256); + if (existing === "matching") return "reused"; + if (existing === "conflict") throwImmutableObjectConflict(uri); + + const body = createReadStream(localPath); try { - const existing = await client.send( - new HeadObjectCommand({ Bucket: bucket, Key: key, ChecksumMode: "ENABLED" }), - ); - if (existing.ContentLength === size && existing.Metadata?.sha256 === expectedSha256) { - return "reused"; - } - const error = new Error( - `[s3Transport] PLAN_ARTIFACT_DIGEST_MISMATCH: immutable object ${uri} already exists with different digest metadata or size`, + await client.send( + new PutObjectCommand({ + Bucket: bucket, + Key: key, + Body: body, + ContentType: contentType, + ContentLength: size, + Metadata: { sha256: expectedSha256 }, + ChecksumSHA256: Buffer.from(expectedSha256, "hex").toString("base64"), + // HEAD followed by an unconditional PUT can overwrite a conflicting + // object published by a concurrent planner. Conditional create makes + // immutable CAS and fixed-key manifest publication race-safe. + IfNoneMatch: "*", + }), ); - error.name = "PLAN_ARTIFACT_DIGEST_MISMATCH"; - throw error; + return "uploaded"; } catch (error) { - if (!isS3NotFound(error)) throw error; + if (!isS3PreconditionFailed(error)) throw error; + const raced = await inspectContentAddressedObject(client, bucket, key, size, expectedSha256); + if (raced === "matching") return "reused"; + if (raced === "conflict") throwImmutableObjectConflict(uri); + // The winning object was deleted between the conditional failure and + // verification. Preserve the service error so the orchestrator may retry. + throw error; + } finally { + // A failed conditional request may reject before consuming the stream. + // Explicit teardown avoids retaining the source descriptor on a warm + // Lambda planner. + body.destroy(); } - - await client.send( - new PutObjectCommand({ - Bucket: bucket, - Key: key, - Body: createReadStream(localPath), - ContentType: contentType, - ContentLength: size, - Metadata: { sha256: expectedSha256 }, - ChecksumSHA256: Buffer.from(expectedSha256, "hex").toString("base64"), - }), - ); - return "uploaded"; } export async function sha256File(path: string): Promise { @@ -205,19 +213,54 @@ function assertSha256(value: string): void { } } +type ContentAddressedObjectState = "missing" | "matching" | "conflict"; + +async function inspectContentAddressedObject( + client: S3Client, + bucket: string, + key: string, + expectedSize: number, + expectedSha256: string, +): Promise { + try { + const existing = await client.send( + new HeadObjectCommand({ Bucket: bucket, Key: key, ChecksumMode: "ENABLED" }), + ); + return existing.ContentLength === expectedSize && existing.Metadata?.sha256 === expectedSha256 + ? "matching" + : "conflict"; + } catch (error) { + if (isS3NotFound(error)) return "missing"; + throw error; + } +} + +function throwImmutableObjectConflict(uri: string): never { + const error = new Error( + `[s3Transport] PLAN_ARTIFACT_DIGEST_MISMATCH: immutable object ${uri} already exists with different digest metadata or size`, + ); + error.name = "PLAN_ARTIFACT_DIGEST_MISMATCH"; + throw error; +} + function isS3NotFound(error: unknown): boolean { - if (!error || typeof error !== "object") return false; - const candidate = error as { - name?: string; - $metadata?: { httpStatusCode?: number }; - }; + if (!isRecord(error)) return false; + const metadata = isRecord(error.$metadata) ? error.$metadata : undefined; return ( - candidate.name === "NotFound" || - candidate.name === "NoSuchKey" || - candidate.$metadata?.httpStatusCode === 404 + error.name === "NotFound" || error.name === "NoSuchKey" || metadata?.httpStatusCode === 404 ); } +function isS3PreconditionFailed(error: unknown): boolean { + if (!isRecord(error)) return false; + const metadata = isRecord(error.$metadata) ? error.$metadata : undefined; + return error.name === "PreconditionFailed" || metadata?.httpStatusCode === 412; +} + +function isRecord(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + /** * Pack a directory into a `.tar.gz` at `destTarball`. Uses the `tar` npm * package (pure JS over `node:zlib`) rather than spawning a system tar