fix(aws-lambda): lambda sites create uploads source files over 2 GiB and 5 GB (#5964)

* fix(aws-lambda): lambda sites create uploads source files over 2 GiB and 5 GB

Site ids are hashed in fixed-size reads instead of whole-file reads, so files
over 2 GiB no longer fail; every existing site keeps its id.

Files over S3's 5 GB single-PUT limit upload in parts that S3 checks one by
one, and the object is created only after the whole file's SHA-256 matches.
Content-store refresh and render-time verification handle those objects too.

* fix(aws-lambda): keep small-site hashing as fast as before

Reads use a 256 KiB buffer: the 8 MiB one made hashing small projects slower,
and big files hash no slower with the smaller buffer.

* perf(aws-lambda): stream big uploads from disk with 8 parts in flight

Each part is hashed from disk and then streamed from disk instead of held in
memory, so a 5 GB upload peaks near 150 MB instead of 500 MB. Eight parts go
up at once, and a part whose connection drops is resent from a fresh read.

packages/aws-lambda/src/s3Stitch.ts

@@ -2,16 +2,14 @@
 // chunk's mdat payload with UploadPartCopy, so almost no video byte passes through the function.
 
 import {
-  AbortMultipartUploadCommand,
-  CompleteMultipartUploadCommand,
-  CreateMultipartUploadCommand,
+  type CompletedPart,
   GetObjectCommand,
   type S3Client,
   UploadPartCommand,
   UploadPartCopyCommand,
 } from "@aws-sdk/client-s3";
 import { buildStitchedHeader, readMp4Layout } from "@hyperframes/producer/distributed";
-import { parseS3Uri } from "./s3Transport.js";
+import { multipartUpload, type MultipartTarget, parseS3Uri } from "./s3Transport.js";
 
 const MIN_PART_BYTES = 5 * 1024 * 1024;
 
@@ -64,9 +62,7 @@ async function getRange(s3: S3Client, uri: string, start: number, end: number):
   return Buffer.from(await response.Body.transformToByteArray());
 }
 
-type UploadTarget = { Bucket: string; Key: string; UploadId: string };
-
-async function sendPart(s3: S3Client, part: Part, partNumber: number, target: UploadTarget) {
+async function sendPart(s3: S3Client, part: Part, partNumber: number, target: MultipartTarget) {
   if ("copy" in part) {
     const { uri, start, end } = part.copy;
     const source = parseS3Uri(uri);
@@ -90,7 +86,11 @@ async function sendPart(s3: S3Client, part: Part, partNumber: number, target: Up
 }
 
 /** Settles every part before failing, so an abort never races a part still in flight. */
-async function sendParts(s3: S3Client, parts: readonly Part[], target: UploadTarget) {
+async function sendParts(
+  s3: S3Client,
+  parts: readonly Part[],
+  target: MultipartTarget,
+): Promise<CompletedPart[]> {
   const settled = await Promise.allSettled(
     parts.map(async (part, index) => {
       const etag = await sendPart(s3, part, index + 1, target);
@@ -102,7 +102,10 @@ async function sendParts(s3: S3Client, parts: readonly Part[], target: UploadTar
     (result): result is PromiseRejectedResult => result.status === "rejected",
   );
   if (failure) throw failure.reason;
-  return settled.map((result) => (result as PromiseFulfilledResult<string>).value);
+  return settled.map((result, index) => ({
+    ETag: (result as PromiseFulfilledResult<string>).value,
+    PartNumber: index + 1,
+  }));
 }
 
 /** Writes `parts` in order to `destUri`; a failure aborts the upload so no parts linger. */
@@ -113,28 +116,9 @@ export async function uploadParts(
   contentType: string,
 ): Promise<void> {
   const { bucket, key } = parseS3Uri(destUri);
-  const created = await s3.send(
-    new CreateMultipartUploadCommand({ Bucket: bucket, Key: key, ContentType: contentType }),
+  await multipartUpload(s3, { Bucket: bucket, Key: key, ContentType: contentType }, (target) =>
+    sendParts(s3, parts, target),
   );
-  if (!created.UploadId) throw new Error(`[s3Stitch] no upload id for ${destUri}`);
-  const target = { Bucket: bucket, Key: key, UploadId: created.UploadId };
-  try {
-    const etags = await sendParts(s3, parts, target);
-    await s3.send(
-      new CompleteMultipartUploadCommand({
-        ...target,
-        MultipartUpload: { Parts: etags.map((ETag, index) => ({ ETag, PartNumber: index + 1 })) },
-      }),
-    );
-  } catch (error) {
-    await s3.send(new AbortMultipartUploadCommand(target)).catch((abortError: unknown) => {
-      console.error(
-        `[s3Stitch] abort failed for ${destUri}; parts remain until a lifecycle rule removes them`,
-        abortError,
-      );
-    });
-    throw error;
-  }
 }
 
 export interface StitchResult {

packages/aws-lambda/src/s3Transport.test.ts

@@ -7,13 +7,15 @@
 
 import { afterAll, beforeAll, describe, expect, it } from "bun:test";
 import { createHash } from "node:crypto";
-import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { mkdtempSync, readFileSync, rmSync, truncateSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { CONTENT_STORE_REFRESH_AFTER_DAYS, contentStoreUri } from "./contentStore.js";
 import { HeadObjectCommand, S3Client } from "@aws-sdk/client-s3";
 import {
   downloadS3ObjectToFileVerified,
+  ensureContentStoreObject,
+  MAX_SINGLE_PUT_BYTES,
   formatS3Uri,
   parseS3Uri,
   presignVerifiedS3GetUrl,
@@ -276,6 +278,184 @@ describe("presignVerifiedS3GetUrl", () => {
   });
 });
 
+describe("objects over the single-PUT limit", () => {
+  const bigSize = MAX_SINGLE_PUT_BYTES + 1;
+  // sha256 of bigSize zero bytes, computed outside this code.
+  const bigDigest = "edcddf01fc829bf06be2b5393a9793cdd43598a0fd483c57f41a9b58183f6e33";
+  function sparseFile(name: string): string {
+    const path = join(scratchRoot, name);
+    writeFileSync(path, "");
+    truncateSync(path, bigSize);
+    return path;
+  }
+
+  it("stores a file over 5 GB in checked parts, created only if the key is free", async () => {
+    const source = sparseFile("big-upload.bin");
+    const s3 = new MultipartFakeS3();
+    try {
+      expect(await putContentStoreFile(s3.asClient(), source, "bucket", bigDigest)).toBe(
+        "uploaded",
+      );
+    } finally {
+      rmSync(source);
+    }
+    const [create, complete] = [s3.calls[1]!, s3.calls.at(-1)!];
+    expect(s3.calls.map((c) => c.name)).not.toContain("PutObjectCommand");
+    expect(create).toMatchObject({
+      name: "CreateMultipartUploadCommand",
+      input: { Metadata: { sha256: bigDigest }, ChecksumAlgorithm: "SHA256" },
+    });
+    expect(complete).toMatchObject({
+      name: "CompleteMultipartUploadCommand",
+      input: { IfNoneMatch: "*", MpuObjectSize: bigSize },
+    });
+    expect(s3.partSizes.reduce((a, b) => a + b, 0)).toBe(bigSize);
+    expect(s3.partSizes.length).toBeLessThanOrEqual(10_000);
+    expect(s3.maxPartsInFlight).toBeGreaterThan(1);
+    expect(s3.maxPartsInFlight).toBeLessThanOrEqual(8);
+  }, 120_000);
+
+  it("resends a part from a fresh read after a dropped connection", async () => {
+    const source = sparseFile("big-retry.bin");
+    const s3 = new MultipartFakeS3();
+    s3.failNextPart = true;
+    try {
+      expect(await putContentStoreFile(s3.asClient(), source, "bucket", bigDigest)).toBe(
+        "uploaded",
+      );
+    } finally {
+      rmSync(source);
+    }
+    expect(s3.partSizes.reduce((a, b) => a + b, 0)).toBe(bigSize);
+    expect(s3.calls.at(-1)).toMatchObject({ name: "CompleteMultipartUploadCommand" });
+  }, 120_000);
+
+  it("aborts instead of completing when the file no longer has the digest it was stored under", async () => {
+    const source = sparseFile("big-changed.bin");
+    const s3 = new MultipartFakeS3();
+    try {
+      await expect(
+        putContentStoreFile(s3.asClient(), source, "bucket", "0".repeat(64)),
+      ).rejects.toMatchObject({ name: "PLAN_ARTIFACT_DIGEST_MISMATCH" });
+    } finally {
+      rmSync(source);
+    }
+    const names = s3.calls.map((c) => c.name);
+    expect(names).toContain("AbortMultipartUploadCommand");
+    expect(names).not.toContain("CompleteMultipartUploadCommand");
+  }, 120_000);
+
+  it("refreshes an aging object over 5 GB by copying it onto itself in parts", async () => {
+    const s3 = new MultipartFakeS3();
+    s3.head = {
+      ContentLength: bigSize,
+      Metadata: { sha256: bigDigest },
+      LastModified: new Date(Date.now() - CONTENT_STORE_REFRESH_AFTER_DAYS * 86_400_000),
+    };
+    await ensureContentStoreObject(s3.asClient(), "bucket", bigDigest, bigSize);
+    const copies = s3.calls.filter((c) => c.name === "UploadPartCopyCommand");
+    expect(copies.map((c) => c.input.CopySourceRange)).toEqual([
+      `bytes=0-${MAX_SINGLE_PUT_BYTES - 1}`,
+      `bytes=${MAX_SINGLE_PUT_BYTES}-${MAX_SINGLE_PUT_BYTES}`,
+    ]);
+    expect(s3.calls.at(-1)).toMatchObject({ name: "CompleteMultipartUploadCommand" });
+    expect(s3.calls[1]).toMatchObject({ input: { Metadata: { sha256: bigDigest } } });
+  });
+
+  it("signs a multipart object by its recorded digest only above the single-PUT limit", async () => {
+    const uri = contentStoreUri("bucket", bigDigest);
+    const head = {
+      ChecksumSHA256: "Y29tcG9zaXRl-2",
+      ChecksumType: "COMPOSITE",
+      Metadata: { sha256: bigDigest },
+    };
+    const holding = (ContentLength: number) => {
+      const s3 = new MultipartFakeS3();
+      s3.head = { ...head, ContentLength };
+      return s3.asClient();
+    };
+    expect(await presignVerifiedS3GetUrl(holding(bigSize), uri, bigDigest, bigSize, 60)).toContain(
+      bigDigest,
+    );
+    await expect(
+      presignVerifiedS3GetUrl(holding(42), uri, bigDigest, 42, 60),
+    ).rejects.toMatchObject({ name: "PLAN_ARTIFACT_DIGEST_MISMATCH" });
+  });
+});
+
+/** Behaves like S3 at the size limits and records part sizes instead of holding the bytes. */
+class MultipartFakeS3 {
+  readonly calls: { name: string; input: Record<string, unknown> }[] = [];
+  readonly partSizes: number[] = [];
+  head: Record<string, unknown> | undefined;
+  failNextPart = false;
+  maxPartsInFlight = 0;
+  #inFlight = 0;
+
+  asClient(): S3Client {
+    const client = new S3Client({
+      region: "us-east-2",
+      credentials: { accessKeyId: "AKIDTEST", secretAccessKey: "test-secret" },
+    });
+    client.send = ((command: { constructor: { name: string }; input: Record<string, unknown> }) =>
+      this.send(command)) as S3Client["send"];
+    return client;
+  }
+
+  async send(command: {
+    constructor: { name: string };
+    input: Record<string, unknown>;
+  }): Promise<unknown> {
+    const { name } = command.constructor;
+    const { Body, ...input } = command.input;
+    this.calls.push({ name, input });
+    if (name === "HeadObjectCommand") {
+      if (this.head) return this.head;
+      throw Object.assign(new Error("not found"), {
+        name: "NotFound",
+        $metadata: { httpStatusCode: 404 },
+      });
+    }
+    if (name === "PutObjectCommand" && Number(input.ContentLength) > MAX_SINGLE_PUT_BYTES) {
+      (Body as NodeJS.ReadableStream & { destroy(): void }).destroy();
+      throw Object.assign(new Error("EntityTooLarge"), { name: "EntityTooLarge" });
+    }
+    if (name === "CopyObjectCommand" && Number(this.head?.ContentLength) > MAX_SINGLE_PUT_BYTES) {
+      throw Object.assign(new Error("copy source too large"), { name: "InvalidRequest" });
+    }
+    if (name === "CreateMultipartUploadCommand") return { UploadId: "upload-1" };
+    if (name === "UploadPartCommand") {
+      this.maxPartsInFlight = Math.max(this.maxPartsInFlight, ++this.#inFlight);
+      try {
+        const hash = createHash("sha256");
+        let length = 0;
+        for await (const chunk of Body as AsyncIterable<Buffer>) {
+          hash.update(chunk);
+          length += chunk.length;
+        }
+        if (this.failNextPart) {
+          this.failNextPart = false;
+          throw Object.assign(new Error("connection reset"), { name: "ECONNRESET" });
+        }
+        if (length !== input.ContentLength || hash.digest("base64") !== input.ChecksumSHA256) {
+          throw Object.assign(new Error("part checksum mismatch"), { name: "BadDigest" });
+        }
+        this.partSizes.push(length);
+        return { ETag: `"part-${String(input.PartNumber)}"` };
+      } finally {
+        this.#inFlight -= 1;
+      }
+    }
+    if (name === "UploadPartCopyCommand") {
+      return { CopyPartResult: { ETag: `"copy-${String(input.PartNumber)}"` } };
+    }
+    if (name === "CompleteMultipartUploadCommand" || name === "AbortMultipartUploadCommand") {
+      return {};
+    }
+    throw new Error(`unexpected command ${name}`);
+  }
+}
+
 class ContentAddressedFakeS3 {
   readonly objects = new Map<string, { bytes: Buffer; sha256: string; lastModified?: Date }>();
   readonly copies: Record<string, unknown>[] = [];

packages/aws-lambda/src/s3Transport.ts

@@ -25,20 +25,35 @@ import {
   rmSync,
   statSync,
 } from "node:fs";
-import { createHash } from "node:crypto";
+import { createHash, type Hash } from "node:crypto";
 import { dirname } from "node:path";
 import { pipeline } from "node:stream/promises";
 import {
+  AbortMultipartUploadCommand,
+  CompleteMultipartUploadCommand,
+  type CompletedPart,
   CopyObjectCommand,
+  CreateMultipartUploadCommand,
+  type CreateMultipartUploadCommandInput,
   GetObjectCommand,
   HeadObjectCommand,
+  type HeadObjectCommandOutput,
   PutObjectCommand,
   type S3Client,
+  UploadPartCommand,
+  UploadPartCopyCommand,
 } from "@aws-sdk/client-s3";
 import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
 import * as tar from "tar";
 import { CONTENT_STORE_REFRESH_AFTER_DAYS, contentStoreUri } from "./contentStore.js";
 
+/** S3's largest PutObject or CopyObject; bigger objects go multipart. */
+export const MAX_SINGLE_PUT_BYTES = 5 * 1024 ** 3;
+const MAX_PARTS = 10_000;
+const MIN_FILE_PART_BYTES = 64 * 1024 * 1024;
+const FILE_PARTS_IN_FLIGHT = 8;
+const PART_ATTEMPTS = 3;
+
 /** Parsed `s3://bucket/key` URI. */
 export interface S3Location {
   bucket: string;
@@ -121,7 +136,9 @@ export async function presignVerifiedS3GetUrl(
     new HeadObjectCommand({ Bucket: bucket, Key: key, ChecksumMode: "ENABLED" }),
   );
   const expectedChecksum = Buffer.from(expectedSha256, "hex").toString("base64");
-  if (head.ChecksumSHA256 !== expectedChecksum || head.ContentLength !== expectedSize) {
+  const attested =
+    head.ChecksumSHA256 === expectedChecksum || isDigestCheckedMultipart(head, expectedSha256);
+  if (!attested || head.ContentLength !== expectedSize) {
     const error = new Error(
       `[s3Transport] PLAN_ARTIFACT_DIGEST_MISMATCH: ${uri} expected ${expectedSha256} (${expectedSize} bytes), S3 holds checksum ${String(head.ChecksumSHA256)} (${String(head.ContentLength)} bytes)`,
     );
@@ -133,13 +150,17 @@ export async function presignVerifiedS3GetUrl(
   });
 }
 
-/**
- * Upload a local file's contents to an S3 URI using a streaming
- * `PutObjectCommand`. PutObject's 5 GB cap comfortably exceeds the
- * distributed pipeline's 2 GB planDir limit and the typical
- * chunk size (≤ 200 MB), so a single PUT works for every artifact this
- * adapter handles.
- */
+// S3 keeps only a checksum of part checksums for a multipart object; its uploader
+// checked the whole-file digest it records before completing (uploadFileInParts).
+function isDigestCheckedMultipart(head: HeadObjectCommandOutput, expectedSha256: string): boolean {
+  return (
+    head.ChecksumType === "COMPOSITE" &&
+    (head.ContentLength ?? 0) > MAX_SINGLE_PUT_BYTES &&
+    head.Metadata?.sha256 === expectedSha256
+  );
+}
+
+/** Upload a local file to an S3 URI, streamed: one PutObject up to S3's 5 GB cap, multipart above. */
 export async function uploadFileToS3(
   client: S3Client,
   localPath: string,
@@ -151,6 +172,14 @@ export async function uploadFileToS3(
   }
   const { bucket, key } = parseS3Uri(uri);
   const size = statSync(localPath).size;
+  if (size > MAX_SINGLE_PUT_BYTES) {
+    await uploadFileInParts(client, localPath, size, {
+      Bucket: bucket,
+      Key: key,
+      ContentType: contentType,
+    });
+    return;
+  }
   await client.send(
     new PutObjectCommand({
       Bucket: bucket,
@@ -183,7 +212,7 @@ async function putImmutableFile(
   uri: string,
   expectedSha256: string,
   contentType?: string,
-): Promise<{ outcome: "uploaded" | "reused"; lastModified?: Date }> {
+): Promise<{ outcome: "uploaded" | "reused"; lastModified?: Date; size?: number }> {
   assertSha256(expectedSha256);
   if (!existsSync(localPath)) {
     throw new Error(`[s3Transport] upload source missing: ${localPath}`);
@@ -192,9 +221,39 @@ async function putImmutableFile(
   const size = statSync(localPath).size;
   const existing = await inspectContentAddressedObject(client, bucket, key, size, expectedSha256);
   if (existing.state === "matching")
-    return { outcome: "reused", lastModified: existing.lastModified };
+    return { outcome: "reused", lastModified: existing.lastModified, size };
   if (existing.state === "conflict") throwImmutableObjectConflict(uri);
 
+  // HEAD followed by an unconditional PUT can overwrite a conflicting object published by a
+  // concurrent planner. Conditional create (IfNoneMatch) makes immutable CAS and fixed-key
+  // manifest publication race-safe.
+  try {
+    if (size > MAX_SINGLE_PUT_BYTES) {
+      const create = { Bucket: bucket, Key: key, ContentType: contentType };
+      await uploadFileInParts(client, localPath, size, create, expectedSha256);
+    } else {
+      await putImmutableBody(client, localPath, bucket, key, size, expectedSha256, contentType);
+    }
+    return { outcome: "uploaded" };
+  } catch (error) {
+    if (!isS3PreconditionFailed(error)) throw error;
+    const raced = await inspectContentAddressedObject(client, bucket, key, size, expectedSha256);
+    if (raced.state === "matching") return { outcome: "reused" };
+    if (raced.state === "conflict") throwImmutableObjectConflict(uri);
+    // No winner left to verify: rethrow the service error so the orchestrator may retry.
+    throw error;
+  }
+}
+
+async function putImmutableBody(
+  client: S3Client,
+  localPath: string,
+  bucket: string,
+  key: string,
+  size: number,
+  expectedSha256: string,
+  contentType?: string,
+): Promise<void> {
   const body = createReadStream(localPath);
   try {
     await client.send(
@@ -206,21 +265,9 @@ async function putImmutableFile(
         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 { outcome: "uploaded" };
-  } catch (error) {
-    if (!isS3PreconditionFailed(error)) throw error;
-    const raced = await inspectContentAddressedObject(client, bucket, key, size, expectedSha256);
-    if (raced.state === "matching") return { outcome: "reused" };
-    if (raced.state === "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
@@ -229,6 +276,147 @@ async function putImmutableFile(
   }
 }
 
+export type MultipartTarget = { Bucket: string; Key: string; UploadId: string };
+
+/**
+ * Creates a multipart upload, lets `sendParts` fill it, then completes it. A failure aborts the
+ * upload so no parts linger; `sendParts` must settle every part before it rejects.
+ */
+export async function multipartUpload(
+  client: S3Client,
+  create: CreateMultipartUploadCommandInput & { Bucket: string; Key: string },
+  sendParts: (target: MultipartTarget) => Promise<CompletedPart[]>,
+  complete: { ifNoneMatch?: "*"; objectSize?: number } = {},
+): Promise<void> {
+  const created = await client.send(new CreateMultipartUploadCommand(create));
+  const uri = formatS3Uri({ bucket: create.Bucket, key: create.Key });
+  if (!created.UploadId) throw new Error(`[s3Transport] no multipart upload id for ${uri}`);
+  const target = { Bucket: create.Bucket, Key: create.Key, UploadId: created.UploadId };
+  try {
+    const parts = await sendParts(target);
+    await client.send(
+      new CompleteMultipartUploadCommand({
+        ...target,
+        MultipartUpload: { Parts: parts },
+        IfNoneMatch: complete.ifNoneMatch,
+        MpuObjectSize: complete.objectSize,
+      }),
+    );
+  } catch (error) {
+    await client.send(new AbortMultipartUploadCommand(target)).catch((abortError: unknown) => {
+      console.error(
+        `[s3Transport] abort failed for ${uri}; parts remain until a lifecycle rule removes them`,
+        abortError,
+      );
+    });
+    throw error;
+  }
+}
+
+// Hashes each part from disk in order, then streams it from disk, so memory stays flat; S3 checks
+// each part's SHA-256. With
+// `expectedSha256`, the object exists only if the whole file matches it and the key is free.
+async function uploadFileInParts(
+  client: S3Client,
+  localPath: string,
+  size: number,
+  create: CreateMultipartUploadCommandInput & { Bucket: string; Key: string },
+  expectedSha256?: string,
+): Promise<void> {
+  const partBytes = Math.max(MIN_FILE_PART_BYTES, Math.ceil(size / MAX_PARTS));
+  const sendParts = async (target: MultipartTarget): Promise<CompletedPart[]> => {
+    const whole = createHash("sha256");
+    const parts: Promise<CompletedPart>[] = [];
+    const inFlight = new Set<Promise<void>>();
+    let failure: { error: unknown } | undefined;
+    try {
+      for (let start = 0; start < size && !failure; start += partBytes) {
+        if (inFlight.size >= FILE_PARTS_IN_FLIGHT) await Promise.race(inFlight);
+        const range = { start, end: Math.min(start + partBytes, size) };
+        const checksum = await hashRange(localPath, range, whole);
+        const sent = sendFilePart(client, target, parts.length + 1, localPath, range, checksum);
+        parts.push(sent);
+        const slot: Promise<void> = sent.then(
+          () => void inFlight.delete(slot),
+          (error: unknown) => {
+            failure ??= { error };
+            inFlight.delete(slot);
+          },
+        );
+        inFlight.add(slot);
+      }
+    } finally {
+      await Promise.all(inFlight);
+    }
+    if (failure) throw failure.error;
+    const actual = whole.digest("hex");
+    if (expectedSha256 && actual !== expectedSha256) {
+      const error = new Error(
+        `[s3Transport] PLAN_ARTIFACT_DIGEST_MISMATCH: ${localPath} changed: expected ${expectedSha256}, read ${actual}`,
+      );
+      error.name = "PLAN_ARTIFACT_DIGEST_MISMATCH";
+      throw error;
+    }
+    return Promise.all(parts);
+  };
+  await multipartUpload(
+    client,
+    {
+      ...create,
+      Metadata: expectedSha256 ? { sha256: expectedSha256 } : undefined,
+      ChecksumAlgorithm: "SHA256",
+    },
+    sendParts,
+    { ifNoneMatch: expectedSha256 ? "*" : undefined, objectSize: size },
+  );
+}
+
+type ByteRange = { start: number; end: number };
+
+/** SHA-256 (base64) of `[start, end)` of the file, also fed to `whole`. */
+async function hashRange(path: string, { start, end }: ByteRange, whole: Hash): Promise<string> {
+  const part = createHash("sha256");
+  let read = 0;
+  for await (const chunk of createReadStream(path, { start, end: end - 1 })) {
+    whole.update(chunk as Buffer);
+    part.update(chunk as Buffer);
+    read += (chunk as Buffer).length;
+  }
+  if (read !== end - start) throw new Error(`[s3Transport] ${path} shrank while uploading`);
+  return part.digest("base64");
+}
+
+// A streamed body cannot be replayed by the SDK's own retry, so each attempt opens a fresh stream.
+async function sendFilePart(
+  client: S3Client,
+  target: MultipartTarget,
+  partNumber: number,
+  path: string,
+  { start, end }: ByteRange,
+  checksum: string,
+): Promise<CompletedPart> {
+  for (let attempt = 1; ; attempt++) {
+    const body = createReadStream(path, { start, end: end - 1 });
+    try {
+      const result = await client.send(
+        new UploadPartCommand({
+          ...target,
+          PartNumber: partNumber,
+          Body: body,
+          ContentLength: end - start,
+          ChecksumSHA256: checksum,
+        }),
+      );
+      if (!result.ETag) throw new Error(`[s3Transport] part ${partNumber} returned no ETag`);
+      return { ETag: result.ETag, PartNumber: partNumber, ChecksumSHA256: checksum };
+    } catch (error) {
+      if (attempt >= PART_ATTEMPTS) throw error;
+    } finally {
+      body.destroy();
+    }
+  }
+}
+
 export async function mapConcurrent<T>(
   values: readonly T[],
   concurrency: number,
@@ -301,8 +489,8 @@ export async function putContentStoreFile(
   sha256: string,
 ): Promise<"uploaded" | "reused"> {
   const uri = contentStoreUri(bucket, sha256);
-  const { outcome, lastModified } = await putImmutableFile(client, localPath, uri, sha256);
-  if (lastModified) await refreshAgingObject(client, uri, sha256, lastModified);
+  const { outcome, lastModified, size } = await putImmutableFile(client, localPath, uri, sha256);
+  if (lastModified) await refreshAgingObject(client, uri, sha256, lastModified, size ?? 0);
   return outcome;
 }
 
@@ -317,7 +505,9 @@ export async function ensureContentStoreObject(
   const { key } = parseS3Uri(uri);
   const existing = await inspectContentAddressedObject(client, bucket, key, sizeBytes, sha256);
   if (existing.state === "matching") {
-    if (existing.lastModified) await refreshAgingObject(client, uri, sha256, existing.lastModified);
+    if (existing.lastModified) {
+      await refreshAgingObject(client, uri, sha256, existing.lastModified, sizeBytes);
+    }
     return;
   }
   if (existing.state === "conflict") throwImmutableObjectConflict(uri);
@@ -345,10 +535,44 @@ async function refreshAgingObject(
   uri: string,
   sha256: string,
   lastModified: Date,
+  size: number,
 ): Promise<void> {
   const { bucket, key } = parseS3Uri(uri);
   const ageMs = Date.now() - lastModified.getTime();
   if (ageMs < CONTENT_STORE_REFRESH_AFTER_DAYS * 86_400_000) return;
+  if (size > MAX_SINGLE_PUT_BYTES) {
+    const create = {
+      Bucket: bucket,
+      Key: key,
+      Metadata: { sha256 },
+      ChecksumAlgorithm: "SHA256" as const,
+    };
+    await multipartUpload(client, create, async (target) => {
+      // ponytail: serial copies (a 6 GB object is two parts); parallel if refresh time matters.
+      const parts: CompletedPart[] = [];
+      for (let start = 0; start < size; start += MAX_SINGLE_PUT_BYTES) {
+        const end = Math.min(start + MAX_SINGLE_PUT_BYTES, size);
+        const PartNumber = parts.length + 1;
+        const { CopyPartResult } = await client.send(
+          new UploadPartCopyCommand({
+            ...target,
+            PartNumber,
+            CopySource: `${bucket}/${key}`,
+            CopySourceRange: `bytes=${start}-${end - 1}`,
+          }),
+        );
+        if (!CopyPartResult?.ETag)
+          throw new Error(`[s3Transport] copy part ${PartNumber} of ${uri} returned no ETag`);
+        parts.push({
+          ETag: CopyPartResult.ETag,
+          PartNumber,
+          ChecksumSHA256: CopyPartResult.ChecksumSHA256,
+        });
+      }
+      return parts;
+    });
+    return;
+  }
   await client.send(
     new CopyObjectCommand({
       Bucket: bucket,

packages/producer/src/services/distributed/projectHash.test.ts

@@ -1,6 +1,6 @@
 import { afterEach, describe, expect, it } from "bun:test";
 import { createHash } from "node:crypto";
-import { mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs";
+import { mkdirSync, mkdtempSync, rmSync, symlinkSync, truncateSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { hashProjectDir, listSiteFiles } from "./projectHash.js";
@@ -71,4 +71,19 @@ describe("listSiteFiles / hashProjectDir", () => {
       .slice(0, 16);
     expect(hashProjectDir(dir)).not.toBe(unversioned);
   });
+
+  it("hashes a file larger than a runtime can read in one call", () => {
+    const dir = project();
+    writeFileSync(join(dir, "big.mov"), "");
+    truncateSync(join(dir, "big.mov"), 2 ** 32 + 1);
+    // Same bytes hashed outside this code: "site-v2\0big.mov\0" + 4 GiB + 1 zero bytes + index.html.
+    expect(hashProjectDir(dir)).toBe("014f8f69f7633edb");
+  }, 60_000);
+
+  it("keeps the site id of a small project unchanged", () => {
+    const dir = project();
+    mkdirSync(join(dir, "assets"));
+    writeFileSync(join(dir, "assets", "clip.mp4"), Buffer.from([0, 1, 2, 255]));
+    expect(hashProjectDir(dir)).toBe("12972d522bd6829d");
+  });
 });

packages/producer/src/services/distributed/projectHash.ts

@@ -9,13 +9,14 @@
  * exactly why it lives here rather than being copy-pasted per adapter.
  */
 
-import { readdirSync, readFileSync, realpathSync, statSync } from "node:fs";
+import { closeSync, openSync, readdirSync, readSync, realpathSync, statSync } from "node:fs";
 import { createHash } from "node:crypto";
 import { join } from "node:path";
 import { PLAN_PROJECT_DIR_SKIP_SEGMENTS } from "./plan.js";
 
 // Bump when the hashed set changes so uploads cached under an older scheme are never reused.
 const SITE_ID_SCHEME = "site-v2\0";
+const READ_CHUNK_BYTES = 256 * 1024;
 
 /**
  * The files a site upload ships, as `/`-separated paths under `projectDir`. Symlinks are followed
@@ -62,20 +63,26 @@ export function listSiteFiles(projectDir: string): string[] {
  * same-bytes rename still yields a fresh id. We trim to 16 chars because the
  * full 64 isn't useful in an object key for legibility. Pass `files` when the
  * caller already listed them, so the hashed set is the shipped set.
- *
- * Reads are synchronous: project trees are typically tens of MB at most
- * (HTML/CSS/JS plus a few composition assets), so the simpler shape wins over
- * a streaming pipeline.
+ * Files are read in fixed-size chunks, so no file is ever held whole in memory.
  */
 export function hashProjectDir(
   projectDir: string,
   files: readonly string[] = listSiteFiles(projectDir),
 ): string {
   const hash = createHash("sha256").update(SITE_ID_SCHEME);
+  const chunk = Buffer.allocUnsafe(READ_CHUNK_BYTES);
   for (const rel of files) {
     hash.update(rel);
     hash.update("\0");
-    hash.update(readFileSync(join(projectDir, rel)));
+    const fd = openSync(join(projectDir, rel), "r");
+    try {
+      let read: number;
+      while ((read = readSync(fd, chunk, 0, chunk.length, null)) > 0) {
+        hash.update(chunk.subarray(0, read));
+      }
+    } finally {
+      closeSync(fd);
+    }
   }
   return hash.digest("hex").slice(0, 16);
 }