node:http2: send a PUSH_PROMISE ahead of the parent's corked END_STREAM

src/event_loop/DeferredTaskQueue.rs

@@ -50,6 +50,8 @@ pub type DeferredRepeatingTask = unsafe extern "C" fn(*mut c_void) -> bool;
 #[derive(Default)]
 pub struct DeferredTaskQueue {
     pub(crate) map: ArrayHashMap<Option<NonNull<c_void>>, DeferredRepeatingTask>,
+    /// An entry was posted since the last pass started.
+    unrun: bool,
 }
 
 impl DeferredTaskQueue {
@@ -60,18 +62,27 @@ impl DeferredTaskQueue {
             bun_collections::hash_map::Entry::Occupied(_) => true,
             bun_collections::hash_map::Entry::Vacant(v) => {
                 v.insert(task);
+                self.unrun = true;
                 false
             }
         }
     }
 
+    /// Whether an entry may still be waiting for its first run. Clears the mark.
+    pub fn take_unrun(&mut self) -> bool {
+        let unrun = self.unrun && !self.map.is_empty();
+        self.unrun = false;
+        unrun
+    }
+
     pub fn unregister_task(&mut self, ctx: Option<NonNull<c_void>>) -> bool {
         // Order is irrelevant for this map's contract (see file doc), so
         // swap-remove is fine and O(1).
         self.map.swap_remove(&ctx)
     }
 
     pub fn run(&mut self) {
+        self.unrun = false;
         // Callbacks may re-entrantly mutate `self.map` (see the re-entrancy
         // note in the file doc), so re-read `len()` every iteration and
         // re-check slot `i` after each callback. `remaining` is a livelock bound.

src/js/node/http2.ts

@@ -2719,11 +2719,6 @@ class Http2Stream extends Duplex {
             } else {
               this.emit("wantTrailers");
             }
-            // Hand the END_STREAM (or trailer) frame to the socket now: with deferred write
-            // completion, _final can run on the program's last live turn and a frame left in
-            // the cork for the auto-flusher would strand a generic-streams (duplexPair) peer
-            // waiting for 'end'.
-            native.flush();
             if (this[bunHTTP2StreamFinal] !== callback) {
               // markWritableDone already consumed and invoked the callback during the
               // synchronous streamEnd dispatch above.
@@ -2735,9 +2730,6 @@ class Http2Stream extends Duplex {
           return;
         }
         const settled = native.writeStream(this.#id, "", "ascii", true, callback);
-        // Same as above: don't leave the END_STREAM frame in the cork on what may be the
-        // program's last live turn.
-        native.flush();
         if (settled === 5) {
           // HALF_CLOSED_LOCAL settled synchronously; the dispatch was suppressed.
           markWritableDone(this);

src/jsc/bindings/ZigGlobalObject.cpp

@@ -3163,6 +3163,15 @@ extern "C" uint8_t JSC__JSGlobalObject__drainMicrotasks(Zig::GlobalObject* globa
     return globalObject->drainMicrotasks();
 }
 
+extern "C" bool JSC__JSGlobalObject__hasPendingMicrotasks(Zig::GlobalObject* globalObject)
+{
+    auto& vm = globalObject->vm();
+    if (!vm.defaultMicrotaskQueue().isEmpty())
+        return true;
+    auto* nextTickQueue = globalObject->m_nextTickQueue.get();
+    return nextTickQueue && !nextTickQueue->isEmpty();
+}
+
 template<class Visitor, class T> static void visitGlobalObjectMember(Visitor& visitor, T& anything)
 {
     anything.visit(visitor);

src/jsc/event_loop.rs

@@ -154,6 +154,12 @@ mod drain_result {
 // the microtask queue through it is interior mutation invisible to Rust.
 unsafe extern "C" {
     safe fn JSC__JSGlobalObject__drainMicrotasks(global: &JSGlobalObject) -> u8;
+    safe fn JSC__JSGlobalObject__hasPendingMicrotasks(global: &JSGlobalObject) -> bool;
+}
+
+/// No-op yield task: the tick that runs it ends in a checkpoint, which runs the deferred queue.
+fn deferred_task_checkpoint(_queue: *mut DeferredTaskQueue::DeferredTaskQueue) -> JsResult<()> {
+    Ok(())
 }
 
 impl JSGlobalObject {
@@ -439,6 +445,25 @@ impl EventLoop {
         self.deferred_tasks.run();
         vm.is_inside_deferred_task_queue.set(false);
 
+        // A deferred flush into a JS transport runs user code: drain what it queued.
+        if JSC__JSGlobalObject__hasPendingMicrotasks(global_object) {
+            match JSC__JSGlobalObject__drainMicrotasks(global_object) {
+                drain_result::SUCCESS => {}
+                drain_result::STOPPED => return Err(Stopped),
+                drain_result::PENDING_EXCEPTION => return Ok(()),
+                _ => unreachable!(),
+            }
+        }
+
+        // A deferred task posted too late for the pass: keep the loop alive for one more.
+        if self.deferred_tasks.take_unrun() {
+            let task = ManagedTask::ManagedTask::new(
+                core::ptr::from_mut(&mut self.deferred_tasks),
+                deferred_task_checkpoint,
+            );
+            self.enqueue_task_after_yield(task);
+        }
+
         // Guard on `event_loop_handle` being set, but drain via `uws_loop_mut()`:
         // on Windows the uSockets loop (`uws::Loop::get()`) is NOT
         // `event_loop_handle` (which is the libuv loop).