socket: drain node:net's pending-write queue with a cursor

src/codegen/generate-js2native.ts

@@ -88,6 +88,7 @@ const rustIdentifierPaths: Record<string, string> = {
   "postgres.rs": "sql_jsc/postgres.rs",
   "runtime/dns_jsc/dns.rs": "runtime/dns_jsc/dns.rs",
   "runtime/node/types.rs": "runtime/node/types.rs",
+  "runtime/socket/pending_writes.rs": "runtime/socket/pending_writes.rs",
   "runtime/socket/socket.rs": "runtime/socket/socket.rs",
   "runtime/timer/Timer.rs": "runtime/timer/Timer.rs",
   "runtime/webcore/ByteStream.rs": "runtime/webcore/ByteStream.rs",

src/js/internal-for-testing.ts

@@ -656,6 +656,24 @@ export const namedPipeInternals = {
    */
   liveCount: $newRustFunction("runtime/socket/socket.rs", "TestingAPIs.jsNamedPipeContextLiveCount", 0) as () => number,
 };
+
+/** Replays appends (`n >= 0`) and drains (`n < 0`) on a fresh queue of unsent node:net bytes. */
+export const pendingWritesReplayProbe = $newRustFunction("runtime/socket/pending_writes.rs", "replayProbe", 1) as (
+  schedule: number[],
+) => {
+  /** Every byte sent was the next byte of the appended stream. */
+  intact: boolean;
+  sent: number;
+  unsent: number;
+  /** Times the unsent bytes changed address in a drain, and in an append. */
+  movesOnDrain: number;
+  movesOnAppend: number;
+  /** Unsent bytes that changed address, summed over all moves. */
+  bytesMoved: number;
+  /** The allocation at the end of the schedule, and its peak. */
+  capacity: number;
+  peakCapacity: number;
+};
 type SerializationContext = "worker" | "window" | "postMessage" | "default";
 export const structuredCloneAdvanced: (
   value: any,

src/runtime/socket/mod.rs

@@ -21,6 +21,8 @@ pub(crate) mod js_socket_handlers;
 #[path = "Listener.rs"]
 pub(crate) mod listener;
 
+pub(crate) mod pending_writes;
+
 #[path = "UpgradedDuplex.rs"]
 pub(crate) mod upgraded_duplex;
 

src/runtime/socket/pending_writes.rs

@@ -0,0 +1,110 @@
+use bun_io::StreamBuffer;
+use bun_jsc::{CallFrame, JSGlobalObject, JSValue, JsResult};
+
+/// Unsent bytes of node:net writes. A drain moves no bytes and an empty queue holds no allocation.
+#[derive(Default)]
+pub(crate) struct PendingWrites(StreamBuffer);
+
+impl PendingWrites {
+    #[inline]
+    pub(crate) fn len(&self) -> usize {
+        self.0.size()
+    }
+
+    #[inline]
+    pub(crate) fn slice(&self) -> &[u8] {
+        self.0.slice()
+    }
+
+    /// The allocation, sent prefix included.
+    #[inline]
+    pub(crate) fn capacity(&self) -> usize {
+        self.0.memory_cost()
+    }
+
+    #[inline]
+    pub(crate) fn append_slice(&mut self, bytes: &[u8]) {
+        bun_core::handle_oom(self.0.write(bytes));
+    }
+
+    #[inline]
+    pub(crate) fn clear_and_free(&mut self) {
+        self.0 = StreamBuffer::default();
+    }
+
+    /// `n` can exceed `len()`: one writev sends the queue and the chunk behind it.
+    #[inline]
+    pub(crate) fn consume(&mut self, n: usize) {
+        if n >= self.len() {
+            self.clear_and_free();
+        } else {
+            self.0.wrote(n);
+        }
+    }
+}
+
+/// The byte at `position` of the stream that `replay_probe` appends.
+fn probe_byte(position: usize) -> u8 {
+    (position ^ (position >> 8) ^ (position >> 16)) as u8
+}
+
+/// `bun:internal-for-testing`: replays appends (`n >= 0`) and drains (`n < 0`) on a fresh queue.
+pub(crate) fn replay_probe(global: &JSGlobalObject, frame: &CallFrame) -> JsResult<JSValue> {
+    let mut schedule = frame.argument(0).array_iterator(global)?;
+
+    let mut queue = PendingWrites::default();
+    let mut appended: usize = 0;
+    let mut sent: usize = 0;
+    let mut intact = true;
+    let mut moves_on_drain: usize = 0;
+    let mut moves_on_append: usize = 0;
+    let mut bytes_moved: usize = 0;
+    let mut peak_capacity: usize = 0;
+
+    while let Some(entry) = schedule.next()? {
+        let n = entry.coerce_to_i32(global)?;
+        let count = n.unsigned_abs() as usize;
+        let unsent_before = queue.len();
+        let front_before = queue.slice().as_ptr();
+
+        let advanced = if n >= 0 {
+            let bytes: Vec<u8> = (appended..appended + count).map(probe_byte).collect();
+            queue.append_slice(&bytes);
+            appended += bytes.len();
+            0
+        } else {
+            let take = count.min(unsent_before);
+            intact &= queue.slice()[..take]
+                .iter()
+                .zip(sent..)
+                .all(|(byte, position)| *byte == probe_byte(position));
+            sent += take;
+            queue.consume(count);
+            take
+        };
+
+        let kept = unsent_before - advanced;
+        if kept > 0 && queue.slice().as_ptr() != front_before.wrapping_add(advanced) {
+            if n >= 0 {
+                moves_on_append += 1;
+            } else {
+                moves_on_drain += 1;
+            }
+            bytes_moved += kept;
+        }
+        intact &= queue.len() == appended - sent;
+        peak_capacity = peak_capacity.max(queue.capacity());
+    }
+
+    let result = JSValue::create_empty_object(global, 8);
+    let number = |value: usize| JSValue::js_number(value as f64);
+    result.put(global, b"intact", JSValue::from(intact));
+    result.put(global, b"sent", number(sent));
+    result.put(global, b"unsent", number(queue.len()));
+    result.put(global, b"movesOnDrain", number(moves_on_drain));
+    result.put(global, b"movesOnAppend", number(moves_on_append));
+    result.put(global, b"bytesMoved", number(bytes_moved));
+    result.put(global, b"capacity", number(queue.capacity()));
+    result.put(global, b"peakCapacity", number(peak_capacity));
+    Ok(result)
+}

src/runtime/socket/socket_body.rs

@@ -14,7 +14,6 @@ use bun_ptr::RefPtr;
 // `const SSL: bool` generic param in `NewSocket<SSL>` below, making rustc
 // resolve `<SSL>` as a type arg (E0747). Use the qualified path instead.
 use bun_boringssl_sys::SSL_CTX;
-use bun_collections::VecExt;
 use bun_core::{self, fmt as bun_fmt};
 use bun_jsc::{self as jsc, CallFrame, JSGlobalObject, JSValue, JsRef, JsResult, SystemError};
 // `err.to_js(global)` on `sys::Error` (the `SysErrorJsc` trait method) is only
@@ -95,6 +94,7 @@ pub(crate) extern "C" fn Bun__socketReadErrorFromCloseCode(
 // ──────────────────────────────────────────────────────────────────────────
 
 pub(super) use super::handlers::Handlers;
+use super::pending_writes::PendingWrites;
 use std::rc::Rc;
 
 mod tls_socket_functions;
@@ -333,7 +333,7 @@ pub(crate) struct NewSocket<const SSL: bool> {
     pub(crate) local_binding: JsCell<Option<(Box<[u8]>, u16)>>,
     pub(crate) protos: JsCell<Option<Box<[u8]>>>,
     pub(crate) server_name: JsCell<Option<Box<[u8]>>>,
-    pub(crate) buffered_data_for_node_net: JsCell<Vec<u8>>,
+    pub(crate) buffered_data_for_node_net: JsCell<PendingWrites>,
     pub(crate) bytes_written: Cell<u64>,
 
     pub(crate) native_callback: JsCell<NativeCallbacks>,
@@ -2819,7 +2819,7 @@ impl<const SSL: bool> NewSocket<SSL> {
                         let remaining_in_input_data = &buffer.slice()[input_written..];
 
                         self.buffered_data_for_node_net
-                            .with_mut(|b| b.drain_front(written));
+                            .with_mut(|b| b.consume(written));
 
                         if !remaining_in_input_data.is_empty() {
                             // Result intentionally discarded
@@ -2850,21 +2850,8 @@ impl<const SSL: bool> NewSocket<SSL> {
                     .with_mut(|b| b.clear_and_free());
             } else if rc > 0 {
                 let wrote_u: usize = usize::try_from(rc.max(0)).expect("int cast");
-                self.buffered_data_for_node_net.with_mut(|b| {
-                    // did we write everything?
-                    // we can free this temporary buffer.
-                    if wrote_u == b.len() as usize {
-                        b.clear_and_free();
-                    } else {
-                        // Otherwise, let's move the temporary buffer back.
-                        let len = b.len() as usize - wrote_u;
-                        debug_assert!(len <= b.len() as usize);
-                        debug_assert!(len <= b.capacity() as usize);
-                        // `wrote_u < b.len()` (else branch) — safe overlapping memmove.
-                        b.copy_within(wrote_u.., 0);
-                        b.truncate(len);
-                    }
-                });
+                self.buffered_data_for_node_net
+                    .with_mut(|b| b.consume(wrote_u));
             }
 
             rc
@@ -3149,16 +3136,8 @@ impl<const SSL: bool> NewSocket<SSL> {
             self.bytes_written
                 .set(self.bytes_written.get() + written as u64);
             if written > 0 {
-                self.buffered_data_for_node_net.with_mut(|b| {
-                    if b.len() as usize > written {
-                        let remaining_len = b.len() as usize - written;
-                        // `written < b.len()` — safe overlapping memmove.
-                        b.copy_within(written.., 0);
-                        b.truncate(remaining_len);
-                    } else {
-                        b.clear_and_free();
-                    }
-                });
+                self.buffered_data_for_node_net
+                    .with_mut(|b| b.consume(written));
             }
         }
 
@@ -3624,7 +3603,7 @@ impl<const SSL: bool> NewSocket<SSL> {
             this_value: JsCell::new(JsRef::empty()),
             poll_ref: JsCell::new(KeepAlive::init()),
             ref_pollref_on_connect: Cell::new(true),
-            buffered_data_for_node_net: JsCell::new(Vec::new()),
+            buffered_data_for_node_net: JsCell::new(PendingWrites::default()),
             bytes_written: Cell::new(0),
             native_callback: JsCell::new(NativeCallbacks::None),
             twin: JsCell::new(None),
@@ -3749,7 +3728,7 @@ impl<const SSL: bool> NewSocket<SSL> {
             this_value: JsCell::new(JsRef::empty()),
             poll_ref: JsCell::new(KeepAlive::init()),
             ref_pollref_on_connect: Cell::new(true),
-            buffered_data_for_node_net: JsCell::new(Vec::new()),
+            buffered_data_for_node_net: JsCell::new(PendingWrites::default()),
             bytes_written: Cell::new(0),
             native_callback: JsCell::new(NativeCallbacks::None),
             twin: JsCell::new(None),
@@ -4805,7 +4784,7 @@ pub(crate) fn js_upgrade_duplex_to_tls(
         this_value: JsCell::new(JsRef::empty()),
         poll_ref: JsCell::new(KeepAlive::init()),
         ref_pollref_on_connect: Cell::new(true),
-        buffered_data_for_node_net: JsCell::new(Vec::new()),
+        buffered_data_for_node_net: JsCell::new(PendingWrites::default()),
         bytes_written: Cell::new(0),
         native_callback: JsCell::new(NativeCallbacks::None),
         twin: JsCell::new(None),

test/internal/socket-pending-writes.test.ts

@@ -0,0 +1,108 @@
+/**
+ * `PendingWrites` (src/runtime/socket/pending_writes.rs) holds the bytes of a
+ * node:net write that a send did not take. It has no JS surface of its own, so
+ * `pendingWritesReplayProbe` replays a schedule on a fresh queue: an entry
+ * `n >= 0` appends `n` bytes, an entry `n < 0` sends up to `-n` bytes from the
+ * front and drains `-n`. Unsent bytes that change address count as moved.
+ *
+ * test/js/node/net/net-syscall-fault.test.ts drives the same queue through a
+ * socket, in builds that can clamp a send.
+ */
+import { pendingWritesReplayProbe as probe } from "bun:internal-for-testing";
+import { describe, expect, test } from "bun:test";
+
+const step = 4096;
+const tail = 255 * step;
+
+describe("the queue of unsent node:net bytes", () => {
+  // Moving the unsent bytes to the front after every partial send cost
+  // tail * tail / (2 * step) bytes: 132,648,960 for the first row.
+  test.each([
+    [tail, step],
+    [2 * tail, step],
+    [1023 * 1024, 1024],
+  ])("a drain of %i bytes in %i-byte steps moves no bytes and frees the allocation", (bytes, step) => {
+    expect(probe([bytes, ...new Array(bytes / step).fill(-step)])).toEqual({
+      intact: true,
+      sent: bytes,
+      unsent: 0,
+      movesOnDrain: 0,
+      movesOnAppend: 0,
+      bytesMoved: 0,
+      capacity: 0,
+      peakCapacity: bytes,
+    });
+  });
+
+  // One writev sends the queue and the chunk behind it, so the count it
+  // reports can be larger than the queue.
+  test("a drain count past the end empties the queue", () => {
+    expect(probe([1000, -3000, 5000, -100, -4900])).toEqual({
+      intact: true,
+      sent: 6000,
+      unsent: 0,
+      movesOnDrain: 0,
+      movesOnAppend: 0,
+      bytesMoved: 0,
+      capacity: 0,
+      peakCapacity: 5000,
+    });
+    expect(probe([1000, -999])).toEqual({
+      intact: true,
+      sent: 999,
+      unsent: 1,
+      movesOnDrain: 0,
+      movesOnAppend: 0,
+      bytesMoved: 0,
+      capacity: 1000,
+      peakCapacity: 1000,
+    });
+  });
+
+  // An append drops the sent prefix once it is at least as large as the unsent
+  // bytes. So an append moves at most the bytes sent since the move before.
+  test("an empty write behind a tail moves the unsent bytes once per halving", () => {
+    const schedule = [tail];
+    for (let i = 0; i < tail / step; i++) schedule.push(0, -step);
+    expect(probe(schedule)).toEqual({
+      intact: true,
+      sent: tail,
+      unsent: 0,
+      movesOnDrain: 0,
+      movesOnAppend: 7,
+      bytesMoved: (127 + 63 + 31 + 15 + 7 + 3 + 1) * step,
+      capacity: 0,
+      peakCapacity: tail,
+    });
+  });
+
+  test("writes behind a tail that is half sent fit in its allocation", () => {
+    const schedule = [tail, ...new Array(128).fill(-step)];
+    for (let i = 0; i < 400; i++) schedule.push(1024, -step);
+    expect(probe(schedule)).toEqual({
+      intact: true,
+      sent: tail + 400 * 1024,
+      unsent: 0,
+      movesOnDrain: 0,
+      movesOnAppend: 10,
+      bytesMoved: (508 + 289 + 163 + 91 + 52 + 28 + 16 + 7 + 4 + 1) * 1024,
+      capacity: 0,
+      peakCapacity: tail,
+    });
+  });
+
+  // Whether the bytes move when the allocation grows is the allocator's choice.
+  test("a write behind a tail that is less than half sent doubles the allocation", () => {
+    const { movesOnAppend, bytesMoved, ...result } = probe([tail, ...new Array(127).fill(-step), 1024]);
+    expect(result).toEqual({
+      intact: true,
+      sent: 127 * step,
+      unsent: 128 * step + 1024,
+      movesOnDrain: 0,
+      capacity: 2 * tail,
+      peakCapacity: 2 * tail,
+    });
+    expect([0, 128 * step]).toContain(bytesMoved);
+    expect(movesOnAppend).toBe(bytesMoved === 0 ? 0 : 1);
+  });
+});