Repository: denoland/deno
Pull request: fix(ext/node): defer uv_write callbacks to prevent re-entrancy panic

## Summary

- `uv_write` in the uv compat layer was firing write callbacks synchronously when `try_write` succeeded fully or failed. This caused a `RefCell` double-borrow panic ("RefCell already borrowed") when the caller (e.g. `StreamWrap.write_buffer`) held an `OpState` borrow during the `uv_write` call, and the synchronous callback re-entered JS which called another op borrowing `OpState`.
- Changed both TCP and TTY `uv_write` implementations to always queue completed writes with a pre-determined status, deferring callbacks to `poll_tcp_handle`/`poll_tty_handle` in the event loop. This matches real libuv behavior where `uv_write` never fires callbacks synchronously.
- Added a `status` field to `WritePending` to support pre-completed entries that skip I/O in the drain loop.

Closes #32891

## Test plan

- [x] Added regression test `uv_write_callback_is_deferred` verifying callback does not fire synchronously from `uv_write`
- [x] All 58 uv_compat tests pass
- [x] Node.js unit tests pass (buffer, net, http, http2, tty)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Changed files:
filename: libs/core/uv_compat/stream.rs
patch: @@ -254,10 +254,14 @@ pub unsafe fn uv_write(
         data: write_data,
         offset: 0,
         cb,
+        status: None,
       });
       return 0;
     }
 
+    // Never fire callbacks synchronously from uv_write — always queue.
+    // This matches real libuv behavior and prevents re-entrancy panics
+    // when callers (e.g. StreamWrap ops) hold OpState borrows.
     if nbufs == 1 {
       let buf = &*bufs;
       if !buf.base.is_null() && buf.len > 0 {
@@ -274,9 +278,15 @@ pub unsafe fn uv_write(
             Ok(n) => {
               offset += n;
               if offset >= data.len() {
-                if let Some(cb) = cb {
-                  cb(req, 0);
-                }
+                // Fully written — queue a completed entry for the
+                // callback to fire from poll_tcp_handle/run_io.
+                (*tcp).internal_write_queue.push_back(WritePending {
+                  req,
+                  data: Vec::new(),
+                  offset: 0,
+                  cb,
+                  status: Some(0),
+                });
                 return 0;
               }
             }
@@ -286,21 +296,32 @@ pub unsafe fn uv_write(
                 data: data[offset..].to_vec(),
                 offset: 0,
                 cb,
+                status: None,
               });
               return 0;
             }
             Err(_) => {
-              if let Some(cb) = cb {
-                cb(req, UV_EPIPE);
-              }
+              // Queue a completed-with-error entry.
+              (*tcp).internal_write_queue.push_back(WritePending {
+                req,
+                data: Vec::new(),
+                offset: 0,
+                cb,
+                status: Some(UV_EPIPE),
+              });
               return 0;
             }
           }
         }
       }
-      if let Some(cb) = cb {
-        cb(req, 0);
-      }
+      // Empty buffer — queue a completed entry.
+      (*tcp).internal_write_queue.push_back(WritePending {
+        req,
+        data: Vec::new(),
+        offset: 0,
+        cb,
+        status: Some(0),
+      });
       return 0;
     }
 
@@ -321,9 +342,13 @@ pub unsafe fn uv_write(
 
     let total_len: usize = iovecs.iter().map(|s| s.len()).sum();
     if total_len == 0 {
-      if let Some(cb) = cb {
-        cb(req, 0);
-      }
+      (*tcp).internal_write_queue.push_back(WritePending {
+        req,
+        data: Vec::new(),
+        offset: 0,
+        cb,
+        status: Some(0),
+      });
       return 0;
     }
 
@@ -335,10 +360,13 @@ pub unsafe fn uv_write(
       .try_write_vectored(&iovecs);
     match write_result {
       Ok(n) if n >= total_len => {
-        if let Some(cb) = cb {
-          cb(req, 0);
-        }
-        return 0;
+        (*tcp).internal_write_queue.push_back(WritePending {
+          req,
+          data: Vec::new(),
+          offset: 0,
+          cb,
+          status: Some(0),
+        });
       }
       Ok(n) => {
         let mut write_data = Vec::with_capacity(total_len - n);
@@ -356,6 +384,7 @@ pub unsafe fn uv_write(
           data: write_data,
           offset: 0,
           cb,
+          status: None,
         });
       }
       Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
@@ -365,12 +394,17 @@ pub unsafe fn uv_write(
           data: write_data,
           offset: 0,
           cb,
+          status: None,
         });
       }
       Err(_) => {
-        if let Some(cb) = cb {
-          cb(req, UV_EPIPE);
-        }
+        (*tcp).internal_write_queue.push_back(WritePending {
+          req,
+          data: Vec::new(),
+          offset: 0,
+          cb,
+          status: Some(UV_EPIPE),
+        });
       }
     }
   }

filename: libs/core/uv_compat/tcp.rs
patch: @@ -134,6 +134,11 @@ pub(crate) struct WritePending {
   pub(crate) data: Vec<u8>,
   pub(crate) offset: usize,
   pub(crate) cb: Option<uv_write_cb>,
+  /// Pre-determined completion status. When `Some`, the write is already
+  /// complete and the callback should be fired with this status without
+  /// attempting any I/O. This is used to defer synchronous write
+  /// completions to the event loop, preventing re-entrancy issues.
+  pub(crate) status: Option<c_int>,
 }
 
 /// Pending shutdown request, deferred until the write queue drains.
@@ -710,6 +715,19 @@ pub(crate) unsafe fn poll_tcp_handle(
           break;
         }
 
+        // Check if the front entry has a pre-determined status
+        // (deferred from uv_write).
+        let pre_status =
+          (*tcp_ptr).internal_write_queue.front().unwrap().status;
+        if let Some(status) = pre_status {
+          let pw = (*tcp_ptr).internal_write_queue.pop_front().unwrap();
+          if let Some(cb) = pw.cb {
+            cb(pw.req, status);
+          }
+          any_work = true;
+          continue;
+        }
+
         // Try writing in a limited scope.
         let (done, error) = {
           let stream = (*tcp_ptr).internal_stream.as_ref().unwrap();

filename: libs/core/uv_compat/tests.rs
patch: @@ -1,6 +1,7 @@
 // Copyright 2018-2026 the Deno authors. MIT license.
 
 use std::cell::Cell;
+use std::cell::RefCell;
 use std::ffi::c_void;
 use std::future::poll_fn;
 use std::rc::Rc;
@@ -785,8 +786,6 @@ async fn phase_ordering_idle_prepare_check() {
     // by recording the order callbacks fire in.
     let order = Rc::new(RefCell::new(Vec::<&'static str>::new()));
 
-    use std::cell::RefCell;
-
     let order_idle = Rc::into_raw(order.clone());
     let order_prepare = Rc::into_raw(order.clone());
     let order_check = Rc::into_raw(order.clone());
@@ -2176,3 +2175,168 @@ async fn tty_read_stop() {
   })
   .await;
 }
+
+// ========== Regression: uv_write must not fire callbacks synchronously ==========
+
+/// Regression test for https://github.com/denoland/deno/issues/32891
+///
+/// uv_write must never fire the write callback synchronously. Doing so causes
+/// re-entrancy panics when callers (e.g. StreamWrap ops) hold an OpState borrow
+/// during the call to uv_write. This test verifies that the callback is deferred
+/// to the event loop (poll_tcp_handle/run_io) rather than firing inline.
+#[tokio::test(flavor = "current_thread")]
+async fn uv_write_callback_is_deferred() {
+  run_test(async |runtime, uv_loop| {
+    let fired = Rc::new(Cell::new(false));
+    let fired_ptr = Rc::into_raw(fired.clone());
+
+    unsafe extern "C" fn write_cb(req: *mut uv_write_t, status: i32) {
+      assert_eq!(status, 0);
+      let fired = unsafe { Rc::from_raw((*req).data as *const Cell<bool>) };
+      fired.set(true);
+      let _ = Rc::into_raw(fired);
+    }
+
+    // --- Set up a server ---
+    let mut server = std::mem::MaybeUninit::<uv_tcp_t>::uninit();
+    let server_ptr = server.as_mut_ptr();
+
+    unsafe extern "C" fn on_connection(server: *mut uv_stream_t, _status: i32) {
+      let _ = server;
+    }
+
+    let server_port: u16;
+    unsafe {
+      uv_tcp_init(uv_loop, server_ptr);
+
+      let mut addr = std::mem::MaybeUninit::<sockaddr_in>::uninit();
+      let ip = std::ffi::CString::new("127.0.0.1").unwrap();
+      uv_ip4_addr(ip.as_ptr(), 0, addr.as_mut_ptr());
+
+      assert_ok(uv_tcp_bind(
+        server_ptr,
+        addr.as_ptr() as *const c_void,
+        0,
+        0,
+      ));
+      assert_ok(uv_listen(
+        server_ptr as *mut uv_stream_t,
+        1,
+        Some(on_connection),
+      ));
+
+      let mut name = std::mem::MaybeUninit::<sockaddr_in>::zeroed();
+      let mut namelen = std::mem::size_of::<sockaddr_in>() as i32;
+      uv_tcp_getsockname(
+        server_ptr,
+        name.as_mut_ptr() as *mut c_void,
+        &mut namelen,
+      );
+      server_port = u16::from_be(name.assume_init_ref().sin_port);
+    }
+
+    // --- Connect a client ---
+    let connected = Rc::new(Cell::new(false));
+    let connected_ptr = Rc::into_raw(connected.clone());
+
+    unsafe extern "C" fn on_connect(req: *mut uv_connect_t, status: i32) {
+      assert_eq!(status, 0);
+      let connected = unsafe { Rc::from_raw((*req).data as *const Cell<bool>) };
+      connected.set(true);
+      let _ = Rc::into_raw(connected);
+    }
+
+    let mut client = std::mem::MaybeUninit::<uv_tcp_t>::uninit();
+    let client_ptr = client.as_mut_ptr();
+    let mut connect_req = std::mem::MaybeUninit::<uv_connect_t>::uninit();
+    let connect_req_ptr = connect_req.as_mut_ptr();
+
+    unsafe {
+      uv_tcp_init(uv_loop, client_ptr);
+      (*connect_req_ptr).data = connected_ptr as *mut c_void;
+
+      let mut addr = std::mem::MaybeUninit::<sockaddr_in>::uninit();
+      let ip = std::ffi::CString::new("127.0.0.1").unwrap();
+      uv_ip4_addr(ip.as_ptr(), server_port as i32, addr.as_mut_ptr());
+
+      assert_ok(uv_tcp_connect(
+        connect_req_ptr,
+        client_ptr,
+        addr.as_ptr() as *const c_void,
+        Some(on_connect),
+      ));
+    }
+
+    // Poll until connected.
+    for _ in 0..100 {
+      tick(runtime).await;
+      if connected.get() {
+        break;
+      }
+      tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+    }
+    assert!(connected.get(), "Client should have connected");
+
+    // Accept the connection on the server side.
+    let mut accepted = std::mem::MaybeUninit::<uv_tcp_t>::uninit();
+    let accepted_ptr = accepted.as_mut_ptr();
+    unsafe {
+      uv_tcp_init(uv_loop, accepted_ptr);
+      assert_ok(uv_accept(
+        server_ptr as *mut uv_stream_t,
+        accepted_ptr as *mut uv_stream_t,
+      ));
+    }
+
+    // Now write a small buffer — should succeed via try_write but the
+    // callback must NOT fire synchronously.
+    let write_data = b"hello";
+    let mut write_req = std::mem::MaybeUninit::<uv_write_t>::uninit();
+    let write_req_ptr = write_req.as_mut_ptr();
+
+    unsafe {
+      (*write_req_ptr).data = fired_ptr as *mut c_void;
+      let buf = uv_buf_t {
+        base: write_data.as_ptr() as *mut _,
+        len: write_data.len(),
+      };
+      assert_ok(uv_write(
+        write_req_ptr,
+        client_ptr as *mut uv_stream_t,
+        &buf,
+        1,
+        Some(write_cb),
+      ));
+    }
+
+    // The callback must NOT have fired yet (it should be deferred).
+    assert!(
+      !fired.get(),
+      "uv_write callback must not fire synchronously"
+    );
+
+    // Tick the event loop — now the callback should fire.
+    for _ in 0..100 {
+      tick(runtime).await;
+      if fired.get() {
+        break;
+      }
+      tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+    }
+    assert!(
+      fired.get(),
+      "write callback should fire after event loop tick"
+    );
+
+    // Cleanup.
+    unsafe {
+      uv_close(accepted_ptr as *mut uv_handle_t, None);
+      uv_close(client_ptr as *mut uv_handle_t, None);
+      uv_close(server_ptr as *mut uv_handle_t, None);
+      Rc::from_raw(connected_ptr);
+      Rc::from_raw(fired_ptr);
+    }
+    tick(runtime).await;
+  })
+  .await;
+}

filename: libs/core/uv_compat/tty.rs
patch: @@ -1245,11 +1245,13 @@ pub(crate) unsafe fn write_tty(
         data,
         offset: 0,
         cb,
+        status: None,
       });
       return 0;
     }
 
-    // Fast path: single buffer.
+    // Never fire callbacks synchronously — always queue.
+    // This matches real libuv behavior and prevents re-entrancy panics.
     if nbufs == 1 {
       let buf = &*bufs;
       if !buf.base.is_null() && buf.len > 0 {
@@ -1260,9 +1262,14 @@ pub(crate) unsafe fn write_tty(
             Ok(n) => {
               offset += n;
               if offset >= data.len() {
-                if let Some(cb) = cb {
-                  cb(req, 0);
-                }
+                (*tty).internal_write_queue.push_back(WritePending {
+                  req,
+                  data: Vec::new(),
+                  offset: 0,
+                  cb,
+                  status: Some(0),
+                });
+                ensure_tty_registered(tty);
                 return 0;
               }
             }
@@ -1272,31 +1279,47 @@ pub(crate) unsafe fn write_tty(
                 data: data[offset..].to_vec(),
                 offset: 0,
                 cb,
+                status: None,
               });
               ensure_tty_registered(tty);
               return 0;
             }
             Err(_) => {
-              if let Some(cb) = cb {
-                cb(req, UV_EPIPE);
-              }
+              (*tty).internal_write_queue.push_back(WritePending {
+                req,
+                data: Vec::new(),
+                offset: 0,
+                cb,
+                status: Some(UV_EPIPE),
+              });
+              ensure_tty_registered(tty);
               return 0;
             }
           }
         }
       }
-      if let Some(cb) = cb {
-        cb(req, 0);
-      }
+      (*tty).internal_write_queue.push_back(WritePending {
+        req,
+        data: Vec::new(),
+        offset: 0,
+        cb,
+        status: Some(0),
+      });
+      ensure_tty_registered(tty);
       return 0;
     }
 
     // Multi-buffer: collect and write.
     let data = collect_bufs(bufs, nbufs);
     if data.is_empty() {
-      if let Some(cb) = cb {
-        cb(req, 0);
-      }
+      (*tty).internal_write_queue.push_back(WritePending {
+        req,
+        data: Vec::new(),
+        offset: 0,
+        cb,
+        status: Some(0),
+      });
+      ensure_tty_registered(tty);
       return 0;
     }
 
@@ -1306,9 +1329,14 @@ pub(crate) unsafe fn write_tty(
         Ok(n) => {
           offset += n;
           if offset >= data.len() {
-            if let Some(cb) = cb {
-              cb(req, 0);
-            }
+            (*tty).internal_write_queue.push_back(WritePending {
+              req,
+              data: Vec::new(),
+              offset: 0,
+              cb,
+              status: Some(0),
+            });
+            ensure_tty_registered(tty);
             return 0;
           }
         }
@@ -1318,14 +1346,20 @@ pub(crate) unsafe fn write_tty(
             data: data[offset..].to_vec(),
             offset: 0,
             cb,
+            status: None,
           });
           ensure_tty_registered(tty);
           return 0;
         }
         Err(_) => {
-          if let Some(cb) = cb {
-            cb(req, UV_EPIPE);
-          }
+          (*tty).internal_write_queue.push_back(WritePending {
+            req,
+            data: Vec::new(),
+            offset: 0,
+            cb,
+            status: Some(UV_EPIPE),
+          });
+          ensure_tty_registered(tty);
           return 0;
         }
       }
@@ -1572,6 +1606,19 @@ pub(crate) unsafe fn poll_tty_handle(
           break;
         }
 
+        // Check if the front entry has a pre-determined status
+        // (deferred from write_tty).
+        let pre_status =
+          (*tty_ptr).internal_write_queue.front().unwrap().status;
+        if let Some(status) = pre_status {
+          let pw = (*tty_ptr).internal_write_queue.pop_front().unwrap();
+          if let Some(cb) = pw.cb {
+            cb(pw.req, status);
+          }
+          any_work = true;
+          continue;
+        }
+
         let (done, error) = {
           let pw = (*tty_ptr).internal_write_queue.front_mut().unwrap();
           let mut done = false;