Add Content-Length and FixedLengthStream to R2 get response (#712)

* Add Content-Length and FixedLengthStream to R2 get response

Before this, we would see that the following code wouldn't work:
```
  const res = await env.BUCKET.get(name);
  await env.BUCKET.put(name, res!.body);
```

As it would throw:
	TypeError: Provided readable stream must have a known length (request/response body or readable half of FixedLengthStream)

* Test fixed length R2 get responses

---------

Co-authored-by: bcoll <[email protected]>

Author: gabivlj

packages/miniflare/src/workers/r2/bucket.worker.ts

@@ -184,6 +184,7 @@ function encodeResult(
     headers: {
       [R2Headers.METADATA_SIZE]: `${encoded.metadataSize}`,
       "Content-Type": "application/json",
+      "Content-Length": `${encoded.size}`,
     },
   });
 }

packages/miniflare/src/workers/r2/r2Object.worker.ts

@@ -9,6 +9,7 @@ import {
 export interface EncodedMetadata {
   metadataSize: number;
   value: ReadableStream<Uint8Array>;
+  size: number;
 }
 
 export class InternalR2Object {
@@ -68,7 +69,7 @@ export class InternalR2Object {
   encode(): EncodedMetadata {
     const json = JSON.stringify(this.#rawProperties());
     const blob = new Blob([json]);
-    return { metadataSize: blob.size, value: blob.stream() };
+    return { metadataSize: blob.size, value: blob.stream(), size: blob.size };
   }
 
   static encodeMultiple(objects: InternalR2Objects): EncodedMetadata {
@@ -77,7 +78,7 @@ export class InternalR2Object {
       objects: objects.objects.map((o) => o.#rawProperties()),
     });
     const blob = new Blob([json]);
-    return { metadataSize: blob.size, value: blob.stream() };
+    return { metadataSize: blob.size, value: blob.stream(), size: blob.size };
   }
 }
 
@@ -92,13 +93,15 @@ export class InternalR2ObjectBody extends InternalR2Object {
 
   encode(): EncodedMetadata {
     const { metadataSize, value: metadata } = super.encode();
-    const identity = new IdentityTransformStream();
+    const size = this.range?.length ?? this.size;
+    const identity = new FixedLengthStream(size + metadataSize);
     void metadata
       .pipeTo(identity.writable, { preventClose: true })
       .then(() => this.body.pipeTo(identity.writable));
     return {
       metadataSize: metadataSize,
       value: identity.readable,
+      size,
     };
   }
 }

packages/miniflare/test/plugins/r2/index.spec.ts

@@ -549,6 +549,50 @@ test("put: validates metadata size", async (t) => {
     expectations
   );
 });
+test("put: can copy values", async (t) => {
+  const mf = new Miniflare({
+    r2Buckets: ["BUCKET"],
+    modules: true,
+    script: `export default {
+      async fetch(request, env, ctx) {
+        await env.BUCKET.put("key", "0123456789");
+        
+        let object = await env.BUCKET.get("key");
+        await env.BUCKET.put("key-copy", object.body);
+        const copy = await (await env.BUCKET.get("key-copy"))?.text();
+        
+        object = await env.BUCKET.get("key", { range: { offset: 1, length: 4 } });
+        await env.BUCKET.put("key-copy-range-1", object.body);
+        const copyRange1 = await (await env.BUCKET.get("key-copy-range-1"))?.text();
+        
+        object = await env.BUCKET.get("key", { range: { length: 3 } });
+        await env.BUCKET.put("key-copy-range-2", object.body);
+        const copyRange2 = await (await env.BUCKET.get("key-copy-range-2"))?.text();
+        
+        object = await env.BUCKET.get("key", { range: { suffix: 5 } });
+        await env.BUCKET.put("key-copy-range-3", object.body);
+        const copyRange3 = await (await env.BUCKET.get("key-copy-range-3"))?.text();
+        
+        const range = new Headers();
+        range.set("Range", "bytes=0-5");
+        object = await env.BUCKET.get("key", { range });
+        await env.BUCKET.put("key-copy-range-4", object.body);
+        const copyRange4 = await (await env.BUCKET.get("key-copy-range-4"))?.text();
+        
+        return Response.json({ copy, copyRange1, copyRange2, copyRange3, copyRange4 });
+      }
+    }`,
+  });
+  t.teardown(() => mf.dispose());
+  const res = await mf.dispatchFetch("http://localhost");
+  t.deepEqual(await res.json(), {
+    copy: "0123456789",
+    copyRange1: "1234",
+    copyRange2: "012",
+    copyRange3: "56789",
+    copyRange4: "012345",
+  });
+});
 
 test("delete: deletes existing keys", async (t) => {
   const { r2, ns, object } = t.context;