fix(upload): return 4xx for upload errors and handle empty/final-chunk blobs instead of 500

src/registry/r2.ts

@@ -11,8 +11,8 @@ import {
 } from "../chunk";
 import { InternalError, ManifestError, RangeError, ServerError } from "../errors";
 import { SHA256_PREFIX_LEN, getSHA256, hexToDigest, isValidDigest } from "../user";
-import { readableToBlob, readerToBlob, wrap } from "../utils";
-import { BlobUnknownError, ManifestUnknownError } from "../v2-errors";
+import { errorString, jsonHeaders, readableToBlob, readerToBlob, wrap } from "../utils";
+import { BlobUnknownError, DigestInvalidError, ManifestUnknownError } from "../v2-errors";
 import {
   CheckLayerResponse,
   CheckManifestResponse,
@@ -975,35 +975,56 @@ export class R2Registry implements Registry {
     const state = hashedState.state;
 
     const uuid = state.registryUploadId;
-    if (state.parts.length === 0) {
-      if (!stream) {
-        console.error("There has been an upload with zero parts and the body is null");
+
+    // Commit the finished content under the client-claimed digest. R2 verifies the sha256 we
+    // hand it against the bytes it stored, so a digest the client got wrong surfaces here as a
+    // checksum-mismatch rejection — translate that into a 400 DIGEST_INVALID rather than letting
+    // it bubble up as an opaque 500. Any other failure is a genuine server error.
+    const putBlob = async (body: ReadableStream | Uint8Array | null): Promise<RegistryError | null> => {
+      const [, err] = await wrap(
+        this.env.REGISTRY.put(`${namespace}/blobs/${expectedSha}`, body, {
+          sha256: (expectedSha as string).slice(SHA256_PREFIX_LEN),
+        }),
+      );
+      if (err === null) return null;
+      const message = errorString(err);
+      // Matches the wording R2 uses when the stored bytes don't hash to the requested sha256.
+      // This couples to the runtime's error text; the unit test asserts the 400 body so a future
+      // wording change surfaces as a test failure rather than a silent regression to 500.
+      if (/checksum|did not match/i.test(message)) {
         return {
-          response: new InternalError(),
+          response: new Response(JSON.stringify(DigestInvalidError()), { status: 400, headers: jsonHeaders() }),
         };
       }
+      console.error("finishUpload put failed:", message);
+      return { response: new InternalError() };
+    };
 
+    if (state.parts.length === 0) {
+      // No multipart parts were staged: the whole blob arrives in this request body (a monolithic
+      // PUT), or it is a zero-byte blob. An absent body is a valid empty blob — store empty bytes
+      // (R2 requires a known-length body, so a length-less empty stream cannot be used here) and
+      // let the checksum check confirm the client really claimed the empty digest.
       if (length && length > MAXIMUM_CHUNK) {
         console.error("Surpasses MAXIMUM_CHUNK");
         return {
           response: new InternalError(),
         };
       }
 
-      await this.env.REGISTRY.put(`${namespace}/blobs/${expectedSha}`, stream, {
-        sha256: (expectedSha as string).slice(SHA256_PREFIX_LEN),
-      });
+      // With bytes to store, the request body carries its own (known) length. With none, it is a
+      // zero-byte blob — hand R2 empty bytes rather than a length-less empty stream, which it rejects.
+      const putErr = await putBlob(length && length > 0 ? stream! : new Uint8Array(0));
+      if (putErr) return putErr;
     } else {
       const upload = this.env.REGISTRY.resumeMultipartUpload(uuid, state.uploadId);
-      // TODO: Handle one last buffer here
+      // A final chunk carried by the finalizing PUT is appended beforehand via uploadChunk (the
+      // same path a PATCH uses), so the staged parts are complete here. See the PUT handler.
       await upload.complete(state.parts);
       const obj = await this.env.REGISTRY.get(uuid);
-      const put = this.env.REGISTRY.put(`${namespace}/blobs/${expectedSha}`, obj!.body, {
-        sha256: (expectedSha as string).slice(SHA256_PREFIX_LEN),
-      });
-
-      await put;
+      const putErr = await putBlob(obj!.body);
       await this.env.REGISTRY.delete(uuid);
+      if (putErr) return putErr;
     }
 
     await this.env.REGISTRY.delete(getRegistryUploadsPath(state));

src/router.ts

@@ -558,15 +558,36 @@ v2Router.put("/:name+/blobs/uploads/:uuid", async (req, env: Env) => {
   const { digest } = req.query;
 
   const url = new URL(req.url);
+  let location = url.pathname + "?" + url.searchParams.toString();
+  const contentLength = +(req.headers.get("Content-Length") ?? "0");
+
+  // A finalizing PUT may carry the last chunk. Append it through the same path a PATCH uses, so
+  // small chunks are combined into a valid part and an out-of-order chunk is rejected with 416
+  // (instead of corrupting the assembled blob). finishUpload then completes the staged parts.
+  if (req.body && contentLength > 0) {
+    const contentRange = req.headers.get("Content-Range");
+    const [start, end] = contentRange?.split("-") ?? [undefined, undefined];
+    const [chunk, chunkErr] = await wrap<UploadObject | RegistryError, Error>(
+      env.REGISTRY_CLIENT.uploadChunk(
+        name,
+        uuid,
+        location,
+        req.body,
+        contentLength,
+        end !== undefined && start !== undefined ? [+start, +end] : undefined,
+      ),
+    );
+    if (chunkErr) {
+      return new InternalError();
+    }
+    if ("response" in chunk) {
+      return chunk.response;
+    }
+    location = chunk.location;
+  }
+
   const [res, err] = await wrap<FinishedUploadObject | RegistryError, Error>(
-    env.REGISTRY_CLIENT.finishUpload(
-      name,
-      uuid,
-      url.pathname + "?" + url.searchParams.toString(),
-      digest! as string,
-      req.body ?? undefined,
-      +(req.headers.get("Content-Length") ?? "0"),
-    ),
+    env.REGISTRY_CLIENT.finishUpload(name, uuid, location, digest! as string),
   );
 
   if (err) {

src/v2-errors.ts

@@ -22,3 +22,16 @@ export const BlobUnknownError = {
     },
   ],
 };
+
+export const DigestInvalidError = (message = "provided digest did not match uploaded content") =>
+  ({
+    errors: [
+      {
+        code: "DIGEST_INVALID",
+        message,
+        detail: {
+          message: "The provided digest did not match the content received by the registry.",
+        },
+      },
+    ],
+  }) as const;

test/index.test.ts

@@ -2380,3 +2380,96 @@ test("docker.io", () => {
     }
   }
 });
+
+// Multi-chunk uploads (a PATCH chunk followed by a chunk in the finalizing PUT) exercise the
+// small-chunk reconstruction path, which the registry only performs under full push-compatibility
+// mode (PUSH_COMPATIBILITY_MODE=full); these finalize tests enable that mode.
+async function fetchFullCompat(r: Request): Promise<Response> {
+  r.headers.append("Authorization", usernamePasswordToAuth(username, "world"));
+  const ctx = createExecutionContext();
+  const res = await worker.fetch(r, { ...env, PUSH_COMPATIBILITY_MODE: "full" } as Env, ctx);
+  await waitOnExecutionContext(ctx);
+  return res as Response;
+}
+
+describe("blob upload finalization error handling", () => {
+  test("PUT finalize with a wrong digest is rejected 400 DIGEST_INVALID (not 500)", async () => {
+    const name = "uploaderr/baddigest";
+    const data = "the-real-content";
+    const post = await fetch(createRequest("POST", `/v2/${name}/blobs/uploads/`, null, {}));
+    const patch = await fetch(
+      createRequest("PATCH", post.headers.get("location")!, limit(new Blob([data]).stream(), data.length), {}),
+    );
+    expect(patch.status).toBe(202);
+    const wrongDigest = "sha256:" + "0".repeat(64);
+    const put = await fetch(createRequest("PUT", patch.headers.get("location")! + "&digest=" + wrongDigest, null, {}));
+    expect(put.status).toBe(400);
+    // Assert the mapped body, which guards the coupling to R2's checksum-mismatch wording: if the
+    // runtime changes that text, putBlob would fall through to 500 and this assertion would fail.
+    const body = (await put.json()) as { errors: { code: string }[] };
+    expect(body.errors[0].code).toBe("DIGEST_INVALID");
+  });
+
+  test("PUT finalize with an out-of-order final chunk is rejected 416", async () => {
+    const name = "uploaderr/outoforder";
+    const first = "0123456789"; // 10 bytes staged at 0-9
+    const post = await fetch(createRequest("POST", `/v2/${name}/blobs/uploads/`, null, {}));
+    const patch = await fetch(
+      createRequest("PATCH", post.headers.get("location")!, limit(new Blob([first]).stream(), first.length), {
+        "Content-Range": "0-9",
+      }),
+    );
+    expect(patch.status).toBe(202);
+    // A final chunk whose range does not continue at byte 10 is out of order.
+    const finalData = "abcdefghij";
+    const digest = await getSHA256(first + finalData);
+    const put = await fetch(
+      createRequest(
+        "PUT",
+        patch.headers.get("location")! + "&digest=" + digest,
+        limit(new Blob([finalData]).stream(), finalData.length),
+        { "Content-Range": "50-59", "Content-Length": `${finalData.length}` },
+      ),
+    );
+    expect(put.status).toBe(416);
+  });
+
+  test("PUT finalize carrying the final chunk assembles the blob (201) and round-trips", async () => {
+    const name = "uploaderr/finalchunk";
+    const first = "first-part-bytes";
+    const finalData = "final-chunk-bytes";
+    const full = first + finalData;
+    const digest = await getSHA256(full);
+    const post = await fetchFullCompat(createRequest("POST", `/v2/${name}/blobs/uploads/`, null, {}));
+    const patch = await fetchFullCompat(
+      createRequest("PATCH", post.headers.get("location")!, limit(new Blob([first]).stream(), first.length), {
+        "Content-Range": `0-${first.length - 1}`,
+      }),
+    );
+    expect(patch.status).toBe(202);
+    const put = await fetchFullCompat(
+      createRequest(
+        "PUT",
+        patch.headers.get("location")! + "&digest=" + digest,
+        limit(new Blob([finalData]).stream(), finalData.length),
+        { "Content-Range": `${first.length}-${full.length - 1}`, "Content-Length": `${finalData.length}` },
+      ),
+    );
+    expect(put.status).toBe(201);
+    // The PUT-carried final chunk must actually be stored (previously it was silently dropped).
+    const get = await fetchFullCompat(createRequest("GET", `/v2/${name}/blobs/${digest}`, null));
+    expect(get.status).toBe(200);
+    expect(await get.text()).toEqual(full);
+  });
+
+  test("PUT finalize of an empty (zero-byte) blob succeeds 201", async () => {
+    const name = "uploaderr/empty";
+    const emptyDigest = await getSHA256("");
+    const post = await fetch(createRequest("POST", `/v2/${name}/blobs/uploads/`, null, {}));
+    const put = await fetch(createRequest("PUT", post.headers.get("location")! + "&digest=" + emptyDigest, null, {}));
+    expect(put.status).toBe(201);
+    const get = await fetch(createRequest("GET", `/v2/${name}/blobs/${emptyDigest}`, null));
+    expect(get.status).toBe(200);
+    expect(await get.text()).toEqual("");
+  });
+});