fix(ext/web): add stream tests to detect v8slice split bug (#20253)

Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>

22 files changed, 143 insertions, 30 deletions

diff --git a/cli/cache/cache_db.rs b/cli/cache/cache_db.rs
index cdab27f4c..9b591f58a 100644
--- a/cli/cache/cache_db.rs
+++ b/cli/cache/cache_db.rs
@@ -3,7 +3,7 @@
 use deno_core::error::AnyError;
 use deno_core::parking_lot::Mutex;
 use deno_core::parking_lot::MutexGuard;
-use deno_core::task::spawn_blocking;
+use deno_core::unsync::spawn_blocking;
 use deno_runtime::deno_webstorage::rusqlite;
 use deno_runtime::deno_webstorage::rusqlite::Connection;
 use deno_runtime::deno_webstorage::rusqlite::OptionalExtension;
diff --git a/cli/cache/incremental.rs b/cli/cache/incremental.rs
index 04ac4243b..65490a7cd 100644
--- a/cli/cache/incremental.rs
+++ b/cli/cache/incremental.rs
@@ -7,8 +7,8 @@ use std::path::PathBuf;
 use deno_core::error::AnyError;
 use deno_core::parking_lot::Mutex;
 use deno_core::serde_json;
-use deno_core::task::spawn;
-use deno_core::task::JoinHandle;
+use deno_core::unsync::spawn;
+use deno_core::unsync::JoinHandle;
 use deno_runtime::deno_webstorage::rusqlite::params;
 use serde::Serialize;
 
diff --git a/cli/lsp/client.rs b/cli/lsp/client.rs
index 13347a772..45fc88376 100644
--- a/cli/lsp/client.rs
+++ b/cli/lsp/client.rs
@@ -8,7 +8,7 @@ use deno_core::anyhow::bail;
 use deno_core::error::AnyError;
 use deno_core::serde_json;
 use deno_core::serde_json::Value;
-use deno_core::task::spawn;
+use deno_core::unsync::spawn;
 use tower_lsp::lsp_types as lsp;
 use tower_lsp::lsp_types::ConfigurationItem;
 
diff --git a/cli/lsp/diagnostics.rs b/cli/lsp/diagnostics.rs
index 817d2fbc4..dea43ad87 100644
--- a/cli/lsp/diagnostics.rs
+++ b/cli/lsp/diagnostics.rs
@@ -28,9 +28,9 @@ use deno_core::resolve_url;
 use deno_core::serde::Deserialize;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
-use deno_core::task::spawn;
-use deno_core::task::spawn_blocking;
-use deno_core::task::JoinHandle;
+use deno_core::unsync::spawn;
+use deno_core::unsync::spawn_blocking;
+use deno_core::unsync::JoinHandle;
 use deno_core::ModuleSpecifier;
 use deno_graph::Resolution;
 use deno_graph::ResolutionError;
diff --git a/cli/lsp/language_server.rs b/cli/lsp/language_server.rs
index f532430c8..c29817919 100644
--- a/cli/lsp/language_server.rs
+++ b/cli/lsp/language_server.rs
@@ -8,7 +8,7 @@ use deno_core::resolve_url;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
-use deno_core::task::spawn;
+use deno_core::unsync::spawn;
 use deno_core::ModuleSpecifier;
 use deno_graph::GraphKind;
 use deno_lockfile::Lockfile;
@@ -1705,7 +1705,7 @@ impl Inner {
     }
 
     // spawn a blocking task to allow doing other work while this is occurring
-    let format_result = deno_core::task::spawn_blocking({
+    let format_result = deno_core::unsync::spawn_blocking({
       let fmt_options = self.fmt_options.options.clone();
       let document = document.clone();
       move || {
diff --git a/cli/lsp/parent_process_checker.rs b/cli/lsp/parent_process_checker.rs
index f83543c04..07bfbb648 100644
--- a/cli/lsp/parent_process_checker.rs
+++ b/cli/lsp/parent_process_checker.rs
@@ -1,6 +1,6 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 
-use deno_core::task::spawn;
+use deno_core::unsync::spawn;
 use tokio::time::sleep;
 use tokio::time::Duration;
 
diff --git a/cli/lsp/testing/execution.rs b/cli/lsp/testing/execution.rs
index 6a5474ea7..6474d7354 100644
--- a/cli/lsp/testing/execution.rs
+++ b/cli/lsp/testing/execution.rs
@@ -24,8 +24,8 @@ use deno_core::futures::stream;
 use deno_core::futures::StreamExt;
 use deno_core::parking_lot::Mutex;
 use deno_core::parking_lot::RwLock;
-use deno_core::task::spawn;
-use deno_core::task::spawn_blocking;
+use deno_core::unsync::spawn;
+use deno_core::unsync::spawn_blocking;
 use deno_core::ModuleSpecifier;
 use deno_runtime::permissions::Permissions;
 use deno_runtime::tokio_util::create_and_run_current_thread;
diff --git a/cli/main.rs b/cli/main.rs
index cdbd878d6..9044a14da 100644
--- a/cli/main.rs
+++ b/cli/main.rs
@@ -36,7 +36,7 @@ use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
 use deno_core::error::JsError;
 use deno_core::futures::FutureExt;
-use deno_core::task::JoinHandle;
+use deno_core::unsync::JoinHandle;
 use deno_runtime::colors;
 use deno_runtime::fmt_errors::format_js_error;
 use deno_runtime::tokio_util::create_and_run_current_thread_with_maybe_metrics;
@@ -76,7 +76,7 @@ impl SubcommandOutput for Result<(), std::io::Error> {
 fn spawn_subcommand<F: Future<Output = T> + 'static, T: SubcommandOutput>(
   f: F,
 ) -> JoinHandle<Result<i32, AnyError>> {
-  deno_core::task::spawn(f.map(|r| r.output()))
+  deno_core::unsync::spawn(f.map(|r| r.output()))
 }
 
 async fn run_subcommand(flags: Flags) -> Result<i32, AnyError> {
diff --git a/cli/npm/resolvers/common.rs b/cli/npm/resolvers/common.rs
index a705052df..fec96738e 100644
--- a/cli/npm/resolvers/common.rs
+++ b/cli/npm/resolvers/common.rs
@@ -11,7 +11,7 @@ use async_trait::async_trait;
 use deno_ast::ModuleSpecifier;
 use deno_core::error::AnyError;
 use deno_core::futures;
-use deno_core::task::spawn;
+use deno_core::unsync::spawn;
 use deno_core::url::Url;
 use deno_npm::NpmPackageCacheFolderId;
 use deno_npm::NpmPackageId;
diff --git a/cli/npm/resolvers/local.rs b/cli/npm/resolvers/local.rs
index 71b91c452..74b797309 100644
--- a/cli/npm/resolvers/local.rs
+++ b/cli/npm/resolvers/local.rs
@@ -24,8 +24,8 @@ use deno_ast::ModuleSpecifier;
 use deno_core::anyhow::bail;
 use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
-use deno_core::task::spawn;
-use deno_core::task::JoinHandle;
+use deno_core::unsync::spawn;
+use deno_core::unsync::JoinHandle;
 use deno_core::url::Url;
 use deno_npm::resolution::NpmResolutionSnapshot;
 use deno_npm::NpmPackageCacheFolderId;
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
diff --git a/cli/tools/bench/mod.rs b/cli/tools/bench/mod.rs
index 34ce46bc3..247422f71 100644
--- a/cli/tools/bench/mod.rs
+++ b/cli/tools/bench/mod.rs
@@ -27,8 +27,8 @@ use deno_core::futures::stream;
 use deno_core::futures::StreamExt;
 use deno_core::located_script_name;
 use deno_core::serde_v8;
-use deno_core::task::spawn;
-use deno_core::task::spawn_blocking;
+use deno_core::unsync::spawn;
+use deno_core::unsync::spawn_blocking;
 use deno_core::v8;
 use deno_core::ModuleSpecifier;
 use deno_runtime::permissions::Permissions;
diff --git a/cli/tools/fmt.rs b/cli/tools/fmt.rs
index fc1df3281..5ab82795c 100644
--- a/cli/tools/fmt.rs
+++ b/cli/tools/fmt.rs
@@ -29,7 +29,7 @@ use deno_core::error::generic_error;
 use deno_core::error::AnyError;
 use deno_core::futures;
 use deno_core::parking_lot::Mutex;
-use deno_core::task::spawn_blocking;
+use deno_core::unsync::spawn_blocking;
 use log::debug;
 use log::info;
 use log::warn;
diff --git a/cli/tools/repl/mod.rs b/cli/tools/repl/mod.rs
index 34acb8a4e..05c822e11 100644
--- a/cli/tools/repl/mod.rs
+++ b/cli/tools/repl/mod.rs
@@ -8,7 +8,7 @@ use crate::factory::CliFactory;
 use crate::file_fetcher::FileFetcher;
 use deno_core::error::AnyError;
 use deno_core::futures::StreamExt;
-use deno_core::task::spawn_blocking;
+use deno_core::unsync::spawn_blocking;
 use deno_runtime::permissions::Permissions;
 use deno_runtime::permissions::PermissionsContainer;
 use rustyline::error::ReadlineError;
diff --git a/cli/tools/test/mod.rs b/cli/tools/test/mod.rs
index 101817ac9..365b83268 100644
--- a/cli/tools/test/mod.rs
+++ b/cli/tools/test/mod.rs
@@ -40,8 +40,8 @@ use deno_core::futures::StreamExt;
 use deno_core::located_script_name;
 use deno_core::parking_lot::Mutex;
 use deno_core::serde_v8;
-use deno_core::task::spawn;
-use deno_core::task::spawn_blocking;
+use deno_core::unsync::spawn;
+use deno_core::unsync::spawn_blocking;
 use deno_core::url::Url;
 use deno_core::v8;
 use deno_core::ModuleSpecifier;
diff --git a/cli/tools/upgrade.rs b/cli/tools/upgrade.rs
index f37ee8455..bef4538b3 100644
--- a/cli/tools/upgrade.rs
+++ b/cli/tools/upgrade.rs
@@ -17,7 +17,7 @@ use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
 use deno_core::futures::future::BoxFuture;
 use deno_core::futures::FutureExt;
-use deno_core::task::spawn;
+use deno_core::unsync::spawn;
 use deno_semver::Version;
 use once_cell::sync::Lazy;
 use std::borrow::Cow;
diff --git a/cli/util/draw_thread.rs b/cli/util/draw_thread.rs
index 2fd81a78a..832c7f74b 100644
--- a/cli/util/draw_thread.rs
+++ b/cli/util/draw_thread.rs
@@ -2,7 +2,7 @@
 
 use console_static_text::ConsoleStaticText;
 use deno_core::parking_lot::Mutex;
-use deno_core::task::spawn_blocking;
+use deno_core::unsync::spawn_blocking;
 use deno_runtime::ops::tty::ConsoleSize;
 use once_cell::sync::Lazy;
 use std::sync::Arc;
diff --git a/cli/util/fs.rs b/cli/util/fs.rs
index ff5ed8553..2ccd70b6d 100644
--- a/cli/util/fs.rs
+++ b/cli/util/fs.rs
@@ -3,7 +3,7 @@
 use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
 pub use deno_core::normalize_path;
-use deno_core::task::spawn_blocking;
+use deno_core::unsync::spawn_blocking;
 use deno_core::ModuleSpecifier;
 use deno_runtime::deno_crypto::rand;
 use deno_runtime::deno_node::PathClean;