feat(aws-lambda): publish plan v2 directly to S3

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<PlanV2Result> => {
+    const planV2WithPublisherMock = mock(
+      async (
+        projectDir: string,
+        _config: unknown,
+        publisher: PlanV2ArtifactPublisher,
+        options: Readonly<{ stagingParentDir?: string }>,
+      ): Promise<PlanV2Manifest> => {
         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(

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<PlanLambdaResult> {
   if (event.PlanProtocol === "v2") {
     return handlePlanV2(event, deps);
@@ -350,7 +351,7 @@ async function handlePlanV2(
 ): Promise<Extract<PlanLambdaResult, { PlanProtocol: "v2" }>> {
   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<T>(
       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(

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";

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<string, StoredObject>();
+  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<unknown> {
+    // 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<string, string>;
+        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",
+    });
+  });
+});

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<string, unknown> {
+  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<string> {
+  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<string>();
+  #state: "open" | "committed" | "aborted" = "open";
+
+  constructor(options: Readonly<S3PlanV2ArtifactPublisherOptions>) {
+    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<PlanV2PublishBlob>): Promise<void> {
+    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<void> {
+    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<void> {
+    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}`);
+    }
+  }
+}

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<string, { bytes: Buffer; sha256: string }>();
   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);

packages/aws-lambda/src/s3Transport.ts

@@ -159,34 +159,42 @@ export async function uploadContentAddressedFileToS3(
 
   const { bucket, key } = parseS3Uri(uri);
   const size = statSync(localPath).size;
-  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`,
-    );
-    error.name = "PLAN_ARTIFACT_DIGEST_MISMATCH";
-    throw error;
-  } catch (error) {
-    if (!isS3NotFound(error)) throw error;
-  }
+  const existing = await inspectContentAddressedObject(client, bucket, key, size, expectedSha256);
+  if (existing === "matching") return "reused";
+  if (existing === "conflict") throwImmutableObjectConflict(uri);
 
-  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";
+  const body = createReadStream(localPath);
+  try {
+    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: "*",
+      }),
+    );
+    return "uploaded";
+  } catch (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();
+  }
 }
 
 export async function sha256File(path: string): Promise<string> {
@@ -205,17 +213,52 @@ function assertSha256(value: string): void {
   }
 }
 
-function isS3NotFound(error: unknown): boolean {
-  if (!error || typeof error !== "object") return false;
-  const candidate = error as {
-    name?: string;
-    $metadata?: { httpStatusCode?: number };
-  };
-  return (
-    candidate.name === "NotFound" ||
-    candidate.name === "NoSuchKey" ||
-    candidate.$metadata?.httpStatusCode === 404
+type ContentAddressedObjectState = "missing" | "matching" | "conflict";
+
+async function inspectContentAddressedObject(
+  client: S3Client,
+  bucket: string,
+  key: string,
+  expectedSize: number,
+  expectedSha256: string,
+): Promise<ContentAddressedObjectState> {
+  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 (!isRecord(error)) return false;
+  const metadata = isRecord(error.$metadata) ? error.$metadata : undefined;
+  return (
+    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<string, unknown> {
+  return value !== null && typeof value === "object" && !Array.isArray(value);
 }
 
 /**