perf(v8-runtime): wake loopback peer on FIN/close

crates/bridge/bridge-contract.json

@@ -122,6 +122,7 @@
 				"_fsWriteFile",
 				"_fsReadFileBinary",
 				"_fsWriteFileBinary",
+				"_fsWriteFileBinaryRaw",
 				"_fsReadDir",
 				"_fsMkdir",
 				"_fsRmdir",
@@ -141,7 +142,9 @@
 				"fs.openSync",
 				"fs.closeSync",
 				"fs.readSync",
+				"_fsReadRaw",
 				"fs.writeSync",
+				"_fsWriteRaw",
 				"fs.fstatSync",
 				"fs.futimesSync"
 			]
@@ -697,6 +700,10 @@
 			"method": "fs.writeFileSync",
 			"translateArgs": true
 		},
+		"_fsWriteFileBinaryRaw": {
+			"method": "fs.writeFileSync",
+			"translateArgs": false
+		},
 		"_fsWriteFileBinaryAsync": {
 			"method": "fs.promises.writeFile",
 			"translateArgs": true
@@ -1061,9 +1068,17 @@
 			"method": "fs.readSync",
 			"translateArgs": false
 		},
+		"_fsReadRaw": {
+			"method": "fs.readSync",
+			"translateArgs": false
+		},
 		"fs.writeSync": {
 			"method": "fs.writeSync",
 			"translateArgs": false
+		},
+		"_fsWriteRaw": {
+			"method": "fs.writeSync",
+			"translateArgs": false
 		}
 	}
 }

crates/execution/src/javascript.rs

@@ -3119,12 +3119,18 @@ fn spawn_v8_event_bridge(
                     let phase_start = Instant::now();
                     let request_args = translate_request_args_for_legacy(sidecar_method, &args);
                     let mut raw_bytes_args = HashMap::new();
-                    if sidecar_method == "net.write" {
+                    if sidecar_method == "net.write"
+                        || sidecar_method == "fs.writeSync"
+                        || sidecar_method == "fs.writeFileSync"
+                    {
                         if let Ok(Some(bytes)) = v8_runtime::cbor_payload_raw_byte_arg(&payload, 1)
                         {
                             raw_bytes_args.insert(1, bytes);
                         }
                     }
+                    if method == "_fsReadRaw" {
+                        raw_bytes_args.insert(usize::MAX, Vec::new());
+                    }
                     record_sync_bridge_phase(
                         &method,
                         "event_translate_args",

crates/execution/tests/javascript_v8.rs

@@ -2165,7 +2165,7 @@ if (summary.rinfo.address !== "127.0.0.1" || summary.rinfo.port !== 7) {
 
     let request = expect_next_sync_rpc(&mut execution, "poll dgram.poll request");
     assert_eq!(request.method, "dgram.poll");
-    assert_eq!(request.args, vec![json!("udp-1"), json!(10)]);
+    assert_eq!(request.args, vec![json!("udp-1"), json!(0)]);
     execution
         .respond_sync_rpc_success(request.id, json!(null))
         .expect("respond to initial dgram.poll");
@@ -2462,7 +2462,7 @@ fn javascript_execution_v8_timer_callbacks_fire_and_clear_correctly() {
             limits: Default::default(),
             guest_runtime: Default::default(),
             vm_id: String::from("vm-js"),
-            context_id: context.context_id,
+            context_id: context.context_id.clone(),
             argv: vec![String::from("./entry.js")],
             env: BTreeMap::new(),
             cwd: temp.path().to_path_buf(),
@@ -2494,6 +2494,62 @@ fn javascript_execution_v8_timer_callbacks_fire_and_clear_correctly() {
   if (intervalTicks !== 2) {
     throw new Error(`interval tick count mismatch: ${intervalTicks}`);
   }
+
+  let immediateFired = false;
+  await new Promise((resolve) => {
+    setImmediate(() => {
+      immediateFired = true;
+      resolve();
+    });
+  });
+  if (!immediateFired) {
+    throw new Error("setImmediate callback did not fire");
+  }
+
+  const order = [];
+  setImmediate(() => order.push("immediate"));
+  queueMicrotask(() => order.push("microtask"));
+  await new Promise((resolve) => setImmediate(resolve));
+  if (order.join(",") !== "microtask,immediate") {
+    throw new Error(`unexpected immediate order: ${order.join(",")}`);
+  }
+
+  for (let i = 0; i < 100; i += 1) {
+    await new Promise((resolve) => setImmediate(resolve));
+  }
+
+  let clearedImmediateFired = false;
+  const clearedImmediate = setImmediate(() => {
+    clearedImmediateFired = true;
+  });
+  clearImmediate(clearedImmediate);
+  await new Promise((resolve) => setImmediate(resolve));
+  if (clearedImmediateFired) {
+    throw new Error("cleared immediate fired");
+  }
+
+  const { setImmediate: promiseImmediate } = await import("node:timers/promises");
+  const promiseImmediateValue = await promiseImmediate("promise-value");
+  if (promiseImmediateValue !== "promise-value") {
+    throw new Error(`timers/promises setImmediate mismatch: ${promiseImmediateValue}`);
+  }
+
+  const t0 = Date.now();
+  let chainCount = 0;
+  const immediateChainMs = await new Promise((resolve) => {
+    function tick() {
+      if (++chainCount < 1000) {
+        setImmediate(tick);
+      } else {
+        resolve(Date.now() - t0);
+      }
+    }
+    setImmediate(tick);
+  });
+  if (immediateChainMs >= 500) {
+    throw new Error(`setImmediate chain too slow: ${immediateChainMs}ms`);
+  }
+  console.log(`setImmediate-chain-ms=${immediateChainMs}`);
 })().catch((error) => {
   process.exitCode = 1;
   throw error;
@@ -2508,6 +2564,58 @@ fn javascript_execution_v8_timer_callbacks_fire_and_clear_correctly() {
     let stderr = String::from_utf8(result.stderr.clone()).expect("stderr utf8");
     assert_eq!(result.exit_code, 0, "stdout:\n{stdout}\nstderr:\n{stderr}");
     assert!(stderr.is_empty(), "unexpected stderr: {stderr}");
+    let chain_ms_line = stdout
+        .lines()
+        .find(|line| line.starts_with("setImmediate-chain-ms="))
+        .expect("setImmediate timing line");
+    let chain_ms: u64 = chain_ms_line
+        .trim_start_matches("setImmediate-chain-ms=")
+        .parse()
+        .expect("parse setImmediate timing");
+    println!("setImmediate 1000-chain elapsed ms: {chain_ms}");
+    assert!(
+        chain_ms < 500,
+        "setImmediate 1000-chain elapsed too high: {chain_ms}ms"
+    );
+
+    let only_immediate_execution = engine
+        .start_execution(StartJavascriptExecutionRequest {
+            limits: Default::default(),
+            guest_runtime: Default::default(),
+            vm_id: String::from("vm-js"),
+            context_id: context.context_id.clone(),
+            argv: vec![String::from("./entry.js")],
+            env: BTreeMap::new(),
+            cwd: temp.path().to_path_buf(),
+            inline_code: Some(String::from(
+                r#"
+setImmediate(() => {
+  console.log("only-immediate-fired");
+});
+"#,
+            )),
+        })
+        .expect("start only-immediate JavaScript execution");
+
+    let only_immediate_result = only_immediate_execution
+        .wait()
+        .expect("wait for only-immediate JavaScript execution");
+    let only_immediate_stdout =
+        String::from_utf8(only_immediate_result.stdout.clone()).expect("stdout utf8");
+    let only_immediate_stderr =
+        String::from_utf8(only_immediate_result.stderr.clone()).expect("stderr utf8");
+    assert_eq!(
+        only_immediate_result.exit_code, 0,
+        "stdout:\n{only_immediate_stdout}\nstderr:\n{only_immediate_stderr}"
+    );
+    assert!(
+        only_immediate_stdout.contains("only-immediate-fired"),
+        "only pending setImmediate did not fire; stdout:\n{only_immediate_stdout}"
+    );
+    assert!(
+        only_immediate_stderr.is_empty(),
+        "unexpected stderr: {only_immediate_stderr}"
+    );
 }
 
 fn javascript_execution_v8_readline_polyfill_emits_lines() {

crates/sidecar/src/execution.rs

@@ -4,7 +4,8 @@ use secure_exec_vm_config as vm_config;
 
 use crate::filesystem::{
     handle_python_vfs_rpc_request as filesystem_handle_python_vfs_rpc_request,
-    service_javascript_fs_sync_rpc, service_javascript_module_sync_rpc,
+    service_javascript_fs_read_sync_rpc, service_javascript_fs_sync_rpc,
+    service_javascript_module_sync_rpc,
 };
 use crate::protocol::{
     CloseStdinRequest, EventFrame, EventPayload, ExecuteRequest, FindBoundUdpRequest,
@@ -15082,6 +15083,11 @@ where
         resource_limits,
         network_counts,
     } = request;
+    if request.raw_bytes_args.contains_key(&usize::MAX) && request.method == "fs.readSync" {
+        let kernel_pid = process.kernel_pid;
+        let bytes = service_javascript_fs_read_sync_rpc(kernel, process, kernel_pid, request)?;
+        return Ok(JavascriptSyncRpcServiceResponse::Raw(bytes));
+    }
     let response = match request.method.as_str() {
         "__bench.noop" => Ok(Value::Null),
         "__bench.net_tcp_metrics_reset" => {

crates/sidecar/src/filesystem.rs

@@ -984,6 +984,38 @@ fn fs_sync_request_marks_host_write_dirty(
     })
 }
 
+pub(crate) fn service_javascript_fs_read_sync_rpc(
+    kernel: &mut SidecarKernel,
+    process: &mut ActiveProcess,
+    kernel_pid: u32,
+    request: &JavascriptSyncRpcRequest,
+) -> Result<Vec<u8>, SidecarError> {
+    let fd = javascript_sync_rpc_arg_u32(&request.args, 0, "filesystem read fd")?;
+    let length = usize::try_from(javascript_sync_rpc_arg_u64(
+        &request.args,
+        1,
+        "filesystem read length",
+    )?)
+    .map_err(|_| {
+        SidecarError::InvalidState("filesystem read length must fit within usize".to_string())
+    })?;
+    let position =
+        javascript_sync_rpc_arg_u64_optional(&request.args, 2, "filesystem read position")?;
+    if let Some(mapped) = process.mapped_host_fd_mut(fd) {
+        let value = read_mapped_host_fd(mapped, fd, length, position)?;
+        return javascript_sync_rpc_bytes_arg(
+            std::slice::from_ref(&value),
+            0,
+            "filesystem mapped read response",
+        );
+    }
+    match position {
+        Some(offset) => kernel.fd_pread(EXECUTION_DRIVER_NAME, kernel_pid, fd, length, offset),
+        None => kernel.fd_read(EXECUTION_DRIVER_NAME, kernel_pid, fd, length),
+    }
+    .map_err(kernel_error)
+}
+
 pub(crate) fn service_javascript_fs_sync_rpc(
     kernel: &mut SidecarKernel,
     process: &mut ActiveProcess,
@@ -1034,36 +1066,17 @@ pub(crate) fn service_javascript_fs_sync_rpc(
                 .map(|fd| json!(fd))
                 .map_err(|error| kernel_path_error("fs.open", path, error))
         }
-        "fs.read" | "fs.readSync" => {
-            let fd = javascript_sync_rpc_arg_u32(&request.args, 0, "filesystem read fd")?;
-            let length = usize::try_from(javascript_sync_rpc_arg_u64(
-                &request.args,
-                1,
-                "filesystem read length",
-            )?)
-            .map_err(|_| {
-                SidecarError::InvalidState(
-                    "filesystem read length must fit within usize".to_string(),
-                )
-            })?;
-            let position =
-                javascript_sync_rpc_arg_u64_optional(&request.args, 2, "filesystem read position")?;
-            if let Some(mapped) = process.mapped_host_fd_mut(fd) {
-                return read_mapped_host_fd(mapped, fd, length, position);
-            }
-            let bytes = match position {
-                Some(offset) => {
-                    kernel.fd_pread(EXECUTION_DRIVER_NAME, kernel_pid, fd, length, offset)
-                }
-                None => kernel.fd_read(EXECUTION_DRIVER_NAME, kernel_pid, fd, length),
-            }
-            .map_err(kernel_error)?;
-            Ok(javascript_sync_rpc_bytes_value(&bytes))
-        }
+        "fs.read" | "fs.readSync" => service_javascript_fs_read_sync_rpc(
+            kernel, process, kernel_pid, request,
+        )
+        .map(|bytes| javascript_sync_rpc_bytes_value(&bytes)),
         "fs.write" | "fs.writeSync" => {
             let fd = javascript_sync_rpc_arg_u32(&request.args, 0, "filesystem write fd")?;
-            let contents =
-                javascript_sync_rpc_bytes_arg(&request.args, 1, "filesystem write contents")?;
+            let contents = if let Some(bytes) = request.raw_bytes_args.get(&1) {
+                bytes.clone()
+            } else {
+                javascript_sync_rpc_bytes_arg(&request.args, 1, "filesystem write contents")?
+            };
             let position = javascript_sync_rpc_arg_u64_optional(
                 &request.args,
                 2,
@@ -1226,8 +1239,11 @@ pub(crate) fn service_javascript_fs_sync_rpc(
                 "filesystem writeFile path",
             )?;
             let path = path.as_str();
-            let contents =
-                javascript_sync_rpc_bytes_arg(&request.args, 1, "filesystem writeFile contents")?;
+            let contents = if let Some(bytes) = request.raw_bytes_args.get(&1) {
+                bytes.clone()
+            } else {
+                javascript_sync_rpc_bytes_arg(&request.args, 1, "filesystem writeFile contents")?
+            };
             match mapped_runtime_host_path(process, path, true) {
                 Some(MappedRuntimeHostAccess::Writable(mapped_host)) => {
                     let opened = open_mapped_runtime_beneath(