Skip to content

node:http2: don't run an earlier write's callback mid-payload #46823

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dylan-conway wants to merge 2 commits into
base: main
Choose a base branch
from
Open

node:http2: don't run an earlier write's callback mid-payload #46823

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fccbc18
node:http2: call a displaced write callback after the write that disp…
claude Oct 24, 2026
1324cde
node:http2: keep a queued END_STREAM on the last queued frame
claude Oct 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
129 changes: 54 additions & 75 deletions src/runtime/api/bun/h2_frame_parser.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -901,6 +901,16 @@ struct SendDataOptions {
defer_write_callback: bool,
}

/// What one `H2FrameParser::send_data` call leaves to its caller.
struct SentData {
/// The state the close tail settled on (5 = HALF_CLOSED_LOCAL, 7 = CLOSED, 0 = none).
settled_state: u8,
/// `callback` was left to the caller.
callback_deferred: bool,
/// An earlier write's callback, to call once the payload and `&mut Stream` borrows have ended.
displaced_callback: StrongOptional,
}

struct DispatchGuard<'a>(&'a Cell<u32>);

impl Drop for DispatchGuard<'_> {
Expand Down Expand Up @@ -1660,75 +1670,39 @@ impl Stream {
}
}

/// Returns the write callback that `callback` displaced from the tail frame, not yet called.
#[must_use]
pub(crate) fn queue_frame(
&mut self,
client: &H2FrameParser,
bytes: &[u8],
callback: JSValue,
end_stream: bool,
) {
) -> StrongOptional {
let global_this = client.global_this;

// Note: `dispatch_write_callback()` below re-enters JS, which can
// call back into `H2FrameParser` host-fns (e.g. `writeStream`) that
// look this `Stream` up by id from `client.streams` and reach
// `queue_frame()` again with a fresh `&mut Stream` aliasing this one.
// R-2: `client` is now `&H2FrameParser` (UnsafeCell-backed fields), so
// the parser-side noalias miscompile is structurally impossible. The
// `Stream`-side `&mut self` alias across re-entry remains; keep the
// `black_box` launder on `self`/`last_frame` as defense-in-depth until
// `Stream` itself is celled.
let this: *mut Self = core::hint::black_box(core::ptr::from_mut(self));
// SAFETY: `this` is the live `&mut self` payload; no other `&` to
// `*this` exists between here and the dispatch call.
if let Some(last_frame_ref) = unsafe { (*this).data_frame_queue.peek_last() } {
// Raw, opaque-provenance pointer for post-dispatch accesses.
let last_frame: *mut PendingFrame =
core::hint::black_box(core::ptr::from_mut(last_frame_ref));
// SAFETY: helper for the pre-dispatch accesses below; `last_frame`
// is the unique tail slot in `self.data_frame_queue.data`, valid
// until the dispatch call (after which we re-`black_box` before
// every access — see note above).
macro_rules! lf {
() => {
// SAFETY: `last_frame` points at the live tail slot of
// `self.data_frame_queue`; provenance is re-laundered via
// `black_box` before each post-dispatch expansion so no
// other `&mut` to the slot is live here (see note).
unsafe { &mut *last_frame }
};
}
if let Some(last_frame) = self.data_frame_queue.peek_last() {
if bytes.is_empty() {
// just merge the end_stream
lf!().end_stream = end_stream;
// we can only hold 1 callback at a time so we conclude the last one, and keep the last one as pending
// this is fine is like a per-stream CORKING in a frame level
let old_callback = core::mem::replace(
&mut lf!().callback,
last_frame.end_stream = end_stream;
// a frame holds 1 callback: like a per-stream CORKING in a frame level
return core::mem::replace(
&mut last_frame.callback,
StrongOptional::create(callback, &global_this),
);
if let Some(old_callback_value) = old_callback.get() {
// Escape `this` so a self-derived address is observable
// across the opaque JS call (belt-and-suspenders; either
// launder alone defeats the caching).
core::hint::black_box(this);
client.dispatch_write_callback(old_callback_value);
}
drop(old_callback);
return;
}
if lf!().len == 0 {
if last_frame.len == 0 {
// we have an empty frame with means we can just use this frame with a new buffer
lf!().buffer = Vec::with_capacity(MAX_PAYLOAD_SIZE_WITHOUT_FRAME);
last_frame.buffer = Vec::with_capacity(MAX_PAYLOAD_SIZE_WITHOUT_FRAME);
}
let max_size = MAX_PAYLOAD_SIZE_WITHOUT_FRAME as u32;
let remaining = max_size - lf!().len;
let remaining = max_size - last_frame.len;
if remaining > 0 {
// ok we can cork frames
let consumed_len = (remaining as usize).min(bytes.len());
let merge = &bytes[0..consumed_len];
lf!().buffer.extend_from_slice(merge);
lf!().len += u32::try_from(consumed_len).expect("int cast");
last_frame.buffer.extend_from_slice(merge);
last_frame.len += u32::try_from(consumed_len).expect("int cast");
bun_output::scoped_log!(H2FrameParser, "dataFrame merged {}", consumed_len);

client
Expand All @@ -1737,25 +1711,15 @@ impl Stream {
// lets fallthrough if we still have some data
let more_data = &bytes[consumed_len..];
if more_data.is_empty() {
lf!().end_stream = end_stream;
// we can only hold 1 callback at a time so we conclude the last one, and keep the last one as pending
// this is fine is like a per-stream CORKING in a frame level
let old_callback = core::mem::replace(
&mut lf!().callback,
last_frame.end_stream = end_stream;
Comment thread
dylan-conway marked this conversation as resolved.
Resolved
// a frame holds 1 callback: like a per-stream CORKING in a frame level
return core::mem::replace(
&mut last_frame.callback,
StrongOptional::create(callback, &global_this),
);
if let Some(old_callback_value) = old_callback.get() {
core::hint::black_box(this);
client.dispatch_write_callback(old_callback_value);
}
drop(old_callback);
return;
}
// we keep the old callback because the new will be part of another frame
// SAFETY: `this` is the live `&mut self`; no borrow of `*this`
// is held here (the `last_frame` raw pointer is unused past
// this point).
return unsafe { (*this).queue_frame(client, more_data, callback, end_stream) };
return self.queue_frame(client, more_data, callback, end_stream);
}
}
bun_output::scoped_log!(
Expand Down Expand Up @@ -1800,6 +1764,7 @@ impl Stream {
client
.queued_data_size
.set(client.queued_data_size.get() + bytes.len() as u64);
StrongOptional::empty()
}

pub(crate) fn init(
Expand Down Expand Up @@ -5130,15 +5095,13 @@ impl H2FrameParser {
(self.write_buffer.get().len_u32() as u64 + self.queued_data_size.get()) as f64
}

/// Returns `(settled_state, callback_deferred)`: the state the close tail settled on (5 =
/// HALF_CLOSED_LOCAL, 7 = CLOSED, 0 = none) and whether `callback` was left to the caller.
fn send_data(
&self,
stream: &mut Stream,
payload: &[u8],
callback: JSValue,
options: SendDataOptions,
) -> (u8, bool) {
) -> SentData {
let SendDataOptions {
close,
suppress_half_closed_local_dispatch,
Expand All @@ -5159,6 +5122,7 @@ impl H2FrameParser {

let stream_id = stream.id;
let mut enqueued = false;
let mut displaced_callback = StrongOptional::empty();
let _keepalive = self.keepalive();

let can_close = close && !stream.wait_for_trailers;
Expand All @@ -5176,7 +5140,7 @@ impl H2FrameParser {
};
if self.has_backpressure() || self.outbound_queue_size.get() > 0 {
enqueued = true;
stream.queue_frame(self, b"", callback, close);
displaced_callback = stream.queue_frame(self, b"", callback, close);
} else {
let mut writer = self.to_writer();
let _ = data_header.write(&mut writer, &self.frames_sent_legacy);
Expand Down Expand Up @@ -5219,10 +5183,9 @@ impl H2FrameParser {
// Preserve wire order: anything already batched goes out before the
// queued remainder is flushed later by the drain path.
self.flush_batch_buffer();
enqueued = true;
// write the full frame in memory and queue the frame
// the callback will only be called after the last frame is sended
stream.queue_frame(
let displaced = stream.queue_frame(
self,
slice,
if offset >= payload.len() {
Expand All @@ -5232,6 +5195,11 @@ impl H2FrameParser {
},
offset >= payload.len() && close,
);
// Later slices only land on frames this write queued or took over.
if !enqueued {
displaced_callback = displaced;
}
enqueued = true;
} else {
let padding = stream.get_padding(size, max_size - 1);
let payload_size = size
Expand Down Expand Up @@ -5373,7 +5341,11 @@ impl H2FrameParser {
}
}
}
(settled_state, callback_deferred)
SentData {
settled_state,
callback_deferred,
displaced_callback,
}
}

fn stream_from_js(
Expand Down Expand Up @@ -5402,7 +5374,7 @@ impl H2FrameParser {
let stream = unsafe { &mut *stream };

stream.wait_for_trailers = false;
let _ = self.send_data(
let sent = self.send_data(
stream,
b"",
JSValue::UNDEFINED,
Expand All @@ -5412,6 +5384,9 @@ impl H2FrameParser {
defer_write_callback: false,
},
);
if let Some(displaced_callback) = sent.displaced_callback.get() {
self.dispatch_write_callback(displaced_callback);
}
Ok(())
}

Expand Down Expand Up @@ -5905,7 +5880,7 @@ impl H2FrameParser {
};

let payload = self.stable_payload(buffer.slice());
let (settled_state, callback_deferred) = self.send_data(
let sent = self.send_data(
&mut stream,
&payload,
callback,
Expand All @@ -5915,14 +5890,18 @@ impl H2FrameParser {
defer_write_callback: defer_callback.to_boolean(),
},
);
drop(stream);
if let Some(displaced_callback) = sent.displaced_callback.get() {
self.dispatch_write_callback(displaced_callback);
}

// 5 = HALF_CLOSED_LOCAL: the JS caller runs markWritableDone itself instead of
// the engine re-entering the VM with an onStreamEnd(5) dispatch.
// WRITE_FLUSHED_WITHOUT_CALLBACK: the data was handed to the socket synchronously and
// the write callback was not (and will not be) invoked by the engine; the JS caller
// completes the Writable callback asynchronously.
let mut result = settled_state as u32;
if callback_deferred {
let mut result = sent.settled_state as u32;
if sent.callback_deferred {
result |= WRITE_FLUSHED_WITHOUT_CALLBACK;
}
Ok(JSValue::js_number(result as f64))
Expand Down
56 changes: 55 additions & 1 deletion test/js/node/http2/node-http2.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4106,7 +4106,7 @@ it("http2 client keeps parsing a socket chunk whose ArrayBuffer is transferred b
});

describe.concurrent(
"http2 DATA payload survives its ArrayBuffer being detached/resized by transport JS mid-send",
"http2 DATA payload survives its ArrayBuffer being detached/resized by JS that runs mid-send",
() => {
// When the transport under an Http2Session is user JS (a createConnection Duplex, or a
// TLSSocket upgraded from a JS Duplex), that JS runs synchronously while the native side is
Expand Down Expand Up @@ -4367,6 +4367,60 @@ describe.concurrent(
},
);

// No transport JS at all: the parser writes to a native TCP socket. A frame holds one write
// callback, so a write that tops up a queued frame concludes the callback the frame had, and
// that callback is the JS. node:http2 keeps one write in flight per stream, so this drives
// the native parser itself. Oracle: the DATA bytes the peer receives.
it("an earlier write's callback concluded by a multi-frame write, native writer", async () => {
const result = await run(/* js */ `
const session = http2.connect("http://localhost:1", { createConnection: () => wireDuplex() });
session.on("error", die("session error"));
const H2FrameParser = session[Symbol.for("::bunhttp2native::")].constructor;
const fail = { error: (_, e) => die("socket error")(e), close: () => die("socket")("closed early") };
const results = {};
for (const mode of ["transfer", "resize0", "overwrite", "write"]) {
const wire = wireDuplex();
const server = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { ...fail, data(_, chunk) { wire.write(chunk); } } });
const socket = await Bun.connect({ hostname: "127.0.0.1", port: server.port, socket: { ...fail, data() {} } });
// A client parser queues its DATA until the socket is attached again below.
const parser = new H2FrameParser({ context: {}, handlers: { write() { return true; } }, native: socket });
const id = parser.getNextStream();
parser.request(id, {}, { ":method": "POST", ":path": "/", ":scheme": "http", ":authority": "localhost" }, {});
const holder = { src: payload(40000, mode === "resize0") };
const expected = Buffer.concat([Buffer.from(holder.src), Buffer.from(mode === "write" ? "tail" : "")]);
const calls = [];
parser.writeStream(id, "", undefined, false, () => {
calls.push("empty");
if (mode === "overwrite") holder.src.fill(0x5a);
else if (mode === "write") parser.writeStream(id, "tail", undefined, false, () => calls.push("tail"));
else yank(holder, mode);
});
parser.writeStream(id, holder.src, undefined, false, () => calls.push("payload"));
calls.push("payload queued");
// An empty frame concludes the callback of the frame it lands on too, from either entry point.
parser.writeStream(id, "", undefined, false, () => calls.push("empty again"));
calls.push("empty queued");
parser.noTrailers(id);
calls.push("end queued");
parser.setNativeSocket(socket);
await wire.dataSeen(expected.length);
results[mode] = { calls, foreign: foreign(wire.data(), expected) };
}
console.log(JSON.stringify(results));
process.exit(0);
`);
const calls = ["empty", "payload queued", "payload", "empty queued", "empty again", "end queued"];
expect(result).toEqual({
transfer: { calls, foreign: 0 },
resize0: { calls, foreign: 0 },
overwrite: { calls, foreign: 0 },
write: {
calls: ["empty", "payload", "payload queued", "tail", "empty queued", "empty again", "end queued"],
foreign: 0,
},
});
});

// The session's socket is a native TLSSocket, but one upgraded from a JS Duplex
// (tls.connect({ socket })), so every TLS record is written through that Duplex's JS.
// Oracle: the request body a real secure server receives.
Expand Down
Loading