5 files changed, 117 insertions, 4 deletions
diff --git a/cli/tests/integration/inspector_tests.rs b/cli/tests/integration/inspector_tests.rs
index b4b1f1678..526efbdf2 100644
--- a/cli/tests/integration/inspector_tests.rs
+++ b/cli/tests/integration/inspector_tests.rs
@@ -34,7 +34,7 @@ where
   Fut::Output: Send + 'static,
 {
   fn execute(&self, fut: Fut) {
-    deno_core::task::spawn(fut);
+    deno_core::unsync::spawn(fut);
   }
 }
 
diff --git a/cli/tests/integration/run_tests.rs b/cli/tests/integration/run_tests.rs
index 0c39c2b72..3a385f7cd 100644
--- a/cli/tests/integration/run_tests.rs
+++ b/cli/tests/integration/run_tests.rs
@@ -4220,7 +4220,7 @@ where
   Fut::Output: Send + 'static,
 {
   fn execute(&self, fut: Fut) {
-    deno_core::task::spawn(fut);
+    deno_core::unsync::spawn(fut);
   }
 }
 
diff --git a/cli/tests/testdata/run/heapstats.js.out b/cli/tests/testdata/run/heapstats.js.out
index b75a755f8..954266333 100644
--- a/cli/tests/testdata/run/heapstats.js.out
+++ b/cli/tests/testdata/run/heapstats.js.out
@@ -1,2 +1,2 @@
-Allocated: 4MB
-Freed: -4MB
+Allocated: 8MB
+Freed: -8MB
diff --git a/cli/tests/unit/serve_test.ts b/cli/tests/unit/serve_test.ts
index f412a9071..05affc4f8 100644
--- a/cli/tests/unit/serve_test.ts
+++ b/cli/tests/unit/serve_test.ts
@@ -1483,6 +1483,48 @@ Deno.test(
   },
 );
 
+// Make sure that the chunks of a large response aren't repeated or corrupted in some other way by
+// scatterning sentinels throughout.
+// https://github.com/denoland/fresh/issues/1699
+Deno.test(
+  { permissions: { net: true } },
+  async function httpLargeReadableStreamChunk() {
+    const ac = new AbortController();
+    const server = Deno.serve({
+      handler() {
+        return new Response(
+          new ReadableStream({
+            start(controller) {
+              const buffer = new Uint8Array(1024 * 1024);
+              // Mark the buffer with sentinels
+              for (let i = 0; i < 256; i++) {
+                buffer[i * 4096] = i;
+              }
+              controller.enqueue(buffer);
+              controller.close();
+            },
+          }),
+        );
+      },
+      port: servePort,
+      signal: ac.signal,
+    });
+    const response = await fetch(`http://localhost:${servePort}/`);
+    const body = await response.arrayBuffer();
+    assertEquals(1024 * 1024, body.byteLength);
+    const buffer = new Uint8Array(body);
+    for (let i = 0; i < 256; i++) {
+      assertEquals(
+        i,
+        buffer[i * 4096],
+        `sentinel mismatch at index ${i * 4096}`,
+      );
+    }
+    ac.abort();
+    await server.finished;
+  },
+);
+
 Deno.test(
   { permissions: { net: true } },
   async function httpRequestLatin1Headers() {
diff --git a/cli/tests/unit/streams_test.ts b/cli/tests/unit/streams_test.ts
index c62c48469..4ecea39da 100644
--- a/cli/tests/unit/streams_test.ts
+++ b/cli/tests/unit/streams_test.ts
@@ -77,6 +77,22 @@ function emptyStream(onPull: boolean) {
   }).pipeThrough(new TextEncoderStream());
 }
 
+function largePacketStream(packetSize: number, count: number) {
+  return new ReadableStream({
+    pull(controller) {
+      if (count-- > 0) {
+        const buffer = new Uint8Array(packetSize);
+        for (let i = 0; i < 256; i++) {
+          buffer[i * (packetSize / 256)] = i;
+        }
+        controller.enqueue(buffer);
+      } else {
+        controller.close();
+      }
+    },
+  });
+}
+
 // Include an empty chunk
 function emptyChunkStream() {
   return new ReadableStream({
@@ -260,6 +276,61 @@ Deno.test(async function readableStreamWithEmptyChunkOneByOne() {
   core.ops.op_close(rid);
 });
 
+// Ensure that we correctly transmit all the sub-chunks of the larger chunks.
+Deno.test(async function readableStreamReadSmallerChunks() {
+  const packetSize = 16 * 1024;
+  const rid = resourceForReadableStream(largePacketStream(packetSize, 1));
+  const buffer = new Uint8Array(packetSize);
+  for (let i = 0; i < packetSize / 1024; i++) {
+    await core.ops.op_read(rid, buffer.subarray(i * 1024, i * 1024 + 1024));
+  }
+  for (let i = 0; i < 256; i++) {
+    assertEquals(
+      i,
+      buffer[i * (packetSize / 256)],
+      `at index ${i * (packetSize / 256)}`,
+    );
+  }
+  core.ops.op_close(rid);
+});
+
+Deno.test(async function readableStreamLargePackets() {
+  const packetSize = 128 * 1024;
+  const rid = resourceForReadableStream(largePacketStream(packetSize, 1024));
+  for (let i = 0; i < 1024; i++) {
+    const buffer = new Uint8Array(packetSize);
+    assertEquals(packetSize, await core.ops.op_read(rid, buffer));
+    for (let i = 0; i < 256; i++) {
+      assertEquals(
+        i,
+        buffer[i * (packetSize / 256)],
+        `at index ${i * (packetSize / 256)}`,
+      );
+    }
+  }
+  assertEquals(0, await core.ops.op_read(rid, new Uint8Array(1)));
+  core.ops.op_close(rid);
+});
+
+Deno.test(async function readableStreamVeryLargePackets() {
+  // 1024 packets of 1MB
+  const rid = resourceForReadableStream(largePacketStream(1024 * 1024, 1024));
+  let total = 0;
+  // Read 96kB up to 12,288 times (96kB is not an even multiple of the 1MB packet size to test this)
+  const readCounts: Record<number, number> = {};
+  for (let i = 0; i < 12 * 1024; i++) {
+    const nread = await core.ops.op_read(rid, new Uint8Array(96 * 1024));
+    total += nread;
+    readCounts[nread] = (readCounts[nread] || 0) + 1;
+    if (nread == 0) {
+      break;
+    }
+  }
+  assertEquals({ 0: 1, 65536: 1024, 98304: 10 * 1024 }, readCounts);
+  assertEquals(total, 1024 * 1024 * 1024);
+  core.ops.op_close(rid);
+});
+
 for (const count of [0, 1, 2, 3]) {
   for (const delay of [0, 1, 10]) {
     // Creating a stream that errors in start will throw