Fix #64, add range header forwarding

src/registry/http.ts

@@ -17,6 +17,7 @@ import {
   RegistryError,
   UploadId,
   UploadObject,
+  BlobRangeRequest,
 } from "./registry";
 import { ociImageIndexContentType } from "./r2";
 
@@ -156,18 +157,50 @@ function ctxIntoHeaders(ctx: HTTPContext): Headers {
   return headers;
 }
 
-function ctxIntoRequest(ctx: HTTPContext, url: URL, method: string, path: string, body?: BodyInit): Request {
+function ctxIntoRequest(
+  ctx: HTTPContext,
+  url: URL,
+  method: string,
+  path: string,
+  body?: BodyInit,
+  extraHeaders?: HeadersInit,
+): Request {
   const urlReq = `${url.protocol}//${url.host}/v2${
     ctx.repository === "" || ctx.repository === "/" ? "/" : ctx.repository + "/"
   }${path}`;
+  const headers = ctxIntoHeaders(ctx);
+  if (extraHeaders !== undefined) {
+    new Headers(extraHeaders).forEach((value, key) => headers.set(key, value));
+  }
   return new Request(urlReq, {
     method,
     body,
     redirect: "follow",
-    headers: ctxIntoHeaders(ctx),
+    headers,
   });
 }
 
+// Serializes a blob range request into an HTTP "Range" request header value.
+function rangeRequestHeader(range: BlobRangeRequest): string {
+  if ("suffix" in range) {
+    return `bytes=-${range.suffix}`;
+  }
+
+  return `bytes=${range.offset}-${range.end === undefined ? "" : range.end}`;
+}
+
+// Parses an HTTP "Content-Range: bytes <start>-<end>/<size>" response header.
+function parseContentRange(header: string | null): { start: number; end: number; size: number } | null {
+  if (header === null) return null;
+  const match = /^bytes (\d+)-(\d+)\/(\d+)$/.exec(header.trim());
+  if (match === null) return null;
+  const start = Number(match[1]);
+  const end = Number(match[2]);
+  const size = Number(match[3]);
+  if (!Number.isInteger(start) || !Number.isInteger(end) || !Number.isInteger(size)) return null;
+  return { start, end, size };
+}
+
 function authHeaderIntoAuthContext(urlObject: URL, authenticateHeader: string): AuthContext {
   const url = urlObject.toString();
   const parts = authenticateHeader.split(" ");
@@ -519,18 +552,28 @@ export class RegistryHTTPClient implements Registry {
     }
   }
 
-  async getLayer(name: string, digest: string): Promise<GetLayerResponse | RegistryError> {
+  async getLayer(name: string, digest: string, range?: BlobRangeRequest): Promise<GetLayerResponse | RegistryError> {
     const namespace = name.includes("/") || !isDockerDotIO(this.url) ? name : `library/${name}`;
     try {
       const ctx = await this.authenticate(namespace);
-      const req = ctxIntoRequest(ctx, this.url, "GET", `${namespace}/blobs/${digest}`);
+      const rangeHeader = range === undefined ? undefined : rangeRequestHeader(range);
+      const req = ctxIntoRequest(
+        ctx,
+        this.url,
+        "GET",
+        `${namespace}/blobs/${digest}`,
+        undefined,
+        rangeHeader !== undefined ? { Range: rangeHeader } : undefined,
+      );
       let res = await fetch(req);
       if (!res.ok) {
         // This means we got a redirect, so let's try again this URL but
         // without any headers. Services like S3 reject authorization headers altogether
         // if the authentication is included in the URL.
         if (res.url !== req.url) {
-          const redirectResponse = await fetch(new Request(res.url));
+          const redirectResponse = await fetch(
+            new Request(res.url, rangeHeader !== undefined ? { headers: { Range: rangeHeader } } : undefined),
+          );
           if (!redirectResponse.ok) {
             return {
               response: res,
@@ -549,11 +592,27 @@ export class RegistryHTTPClient implements Registry {
         throw new Error("returned body is null");
       }
 
-      return {
+      const layer: GetLayerResponse = {
         stream: res.body,
         size: +(res.headers.get("Content-Length") ?? "0"),
         digest: res.headers.get("Digest-Content-Digest") ?? digest,
       };
+
+      // If we asked for a range and the upstream honored it, surface the partial-content metadata so
+      // the caller can reply with 206. A 200 here means the upstream ignored the range and we serve
+      // the full blob (best-effort). Serving a partial body as if it were complete would corrupt it,
+      // so a 206 without a parseable Content-Range is treated as an error.
+      if (range !== undefined && res.status === 206) {
+        const contentRange = parseContentRange(res.headers.get("Content-Range"));
+        if (contentRange === null) {
+          throw new Error("upstream returned 206 without a parseable Content-Range header");
+        }
+
+        layer.size = contentRange.size;
+        layer.contentRange = contentRange;
+      }
+
+      return layer;
     } catch (err) {
       console.error(`Error doing get layer with ${namespace} and ${digest}: ` + errorString(err));
       return {

src/registry/r2.ts

@@ -28,12 +28,23 @@ import {
   UploadId,
   UploadObject,
   wrapError,
+  BlobRangeRequest,
 } from "./registry";
 import { GarbageCollectionMode, GarbageCollector } from "./garbage-collector";
 import { ManifestSchema, manifestSchema } from "../manifest";
 
 export const ociImageIndexContentType = "application/vnd.oci.image.index.v1+json";
 
+function rangeNotSatisfiableResponse(size: number): Response {
+  return new Response(null, {
+    status: 416,
+    headers: {
+      "Content-Range": `bytes */${size}`,
+      "Accept-Ranges": "bytes",
+    },
+  });
+}
+
 function referrersPrefix(name: string, digest: string): string {
   return `${name}/_referrers/${digest}/`;
 }
@@ -690,38 +701,122 @@ export class R2Registry implements Registry {
     };
   }
 
-  async getLayer(name: string, digest: string): Promise<RegistryError | GetLayerResponse> {
-    const [res, err] = await wrap(this.env.REGISTRY.get(`${name}/blobs/${digest}`));
-    if (err) {
-      return wrapError("getLayer", err);
+  async getLayer(name: string, digest: string, range?: BlobRangeRequest): Promise<RegistryError | GetLayerResponse> {
+    const key = `${name}/blobs/${digest}`;
+
+    if (range === undefined) {
+      const [res, err] = await wrap(this.env.REGISTRY.get(key));
+      if (err) {
+        return wrapError("getLayer", err);
+      }
+
+      if (!res) {
+        return {
+          response: new Response(JSON.stringify(BlobUnknownError), { status: 404 }),
+        };
+      }
+
+      // Handle R2 symlink
+      if (res.customMetadata && symlinkHeader in res.customMetadata) {
+        return await this.followLayerSymlink(name, digest, res, undefined);
+      }
+
+      return {
+        stream: res.body!,
+        digest: hexToDigest(res.checksums.sha256!),
+        size: res.size,
+      };
     }
 
-    if (!res) {
+    // Ranged read: inspect object metadata first so we can validate the requested range and
+    // resolve symlinks without streaming the full object.
+    const [head, headErr] = await wrap(this.env.REGISTRY.head(key));
+    if (headErr) {
+      return wrapError("getLayer", headErr);
+    }
+
+    if (!head) {
       return {
         response: new Response(JSON.stringify(BlobUnknownError), { status: 404 }),
       };
     }
 
-    // Handle R2 symlink
-    if (res.customMetadata && symlinkHeader in res.customMetadata) {
-      const layerPath = await res.text();
-      // Symlink detected! Will download layer from "layerPath"
-      const [linkName, linkDigest] = layerPath.split("/blobs/");
-      if (linkName == name && linkDigest == digest) {
+    if (head.customMetadata && symlinkHeader in head.customMetadata) {
+      const [link, linkErr] = await wrap(this.env.REGISTRY.get(key));
+      if (linkErr) {
+        return wrapError("getLayer", linkErr);
+      }
+
+      if (!link) {
         return {
           response: new Response(JSON.stringify(BlobUnknownError), { status: 404 }),
         };
       }
-      return await this.env.REGISTRY_CLIENT.getLayer(linkName, linkDigest);
+
+      return await this.followLayerSymlink(name, digest, link, range);
+    }
+
+    const totalSize = head.size;
+    let start: number;
+    let end: number;
+    if ("suffix" in range) {
+      // A suffix longer than the object is satisfied by the whole object, but a zero-length suffix
+      // selects no bytes at all and cannot be satisfied.
+      if (range.suffix <= 0 || totalSize === 0) {
+        return { response: rangeNotSatisfiableResponse(totalSize) };
+      }
+
+      start = Math.max(totalSize - range.suffix, 0);
+      end = totalSize - 1;
+    } else {
+      start = range.offset;
+      if (start < 0 || start >= totalSize) {
+        return { response: rangeNotSatisfiableResponse(totalSize) };
+      }
+
+      end = range.end === undefined ? totalSize - 1 : Math.min(range.end, totalSize - 1);
+      if (end < start) {
+        return { response: rangeNotSatisfiableResponse(totalSize) };
+      }
+    }
+
+    const length = end - start + 1;
+    const [res, err] = await wrap(this.env.REGISTRY.get(key, { range: { offset: start, length } }));
+    if (err) {
+      return wrapError("getLayer", err);
+    }
+
+    if (!res) {
+      return {
+        response: new Response(JSON.stringify(BlobUnknownError), { status: 404 }),
+      };
     }
 
     return {
       stream: res.body!,
-      digest: hexToDigest(res.checksums.sha256!),
-      size: res.size,
+      digest: hexToDigest(head.checksums.sha256!),
+      size: totalSize,
+      contentRange: { start, end, size: totalSize },
     };
   }
 
+  private async followLayerSymlink(
+    name: string,
+    digest: string,
+    object: R2ObjectBody,
+    range: BlobRangeRequest | undefined,
+  ): Promise<RegistryError | GetLayerResponse> {
+    const layerPath = await object.text();
+    // Symlink detected! Will download layer from "layerPath"
+    const [linkName, linkDigest] = layerPath.split("/blobs/");
+    if (linkName == name && linkDigest == digest) {
+      return {
+        response: new Response(JSON.stringify(BlobUnknownError), { status: 404 }),
+      };
+    }
+    return await this.env.REGISTRY_CLIENT.getLayer(linkName, linkDigest, range);
+  }
+
   async startUpload(namespace: string): Promise<RegistryError | UploadObject> {
     // Generate a unique ID for this upload
     const uuid = crypto.randomUUID();

src/registry/registry.ts

@@ -99,11 +99,17 @@ export type GetManifestResponse = {
   contentType: string;
 };
 
+// requested byte range for a layer read. Either an offset with an inclusive end (end omitted means
+// until the end of the object), or a suffix asking for the last N bytes of the object.
+export type BlobRangeRequest = { offset: number; end?: number } | { suffix: number };
+
 // returned by getLayer when it successfully retrieves a layer
 export type GetLayerResponse = {
   stream: ReadableStream;
   digest: string;
   size: number;
+  // present when the response is a partial (ranged) read, drives the 206 Partial Content response
+  contentRange?: { start: number; end: number; size: number };
 };
 
 export type ReferrerDescriptor = {
@@ -150,7 +156,7 @@ export interface Registry {
   layerExists(namespace: string, digest: string): Promise<CheckLayerResponse | RegistryError>;
 
   // get a layer stream from the registry
-  getLayer(namespace: string, digest: string): Promise<GetLayerResponse | RegistryError>;
+  getLayer(namespace: string, digest: string, range?: BlobRangeRequest): Promise<GetLayerResponse | RegistryError>;
 
   // list referrers for a subject digest
   listReferrers(