Repository: denoland/deno
Pull request: fix(node/ext): make zlib write callback async to match node behavior

Deno's zlib `Zlib::write` was calling `processCallback` synchronously, while Node.js runs compression on the libuv threadpool and calls it asynchronously. This caused several issues:

- destroy() couldn't propagate before subsequent write callbacks fired
- stream buffering didn't work correctly (needDrain was never set)
- flush operations didn't produce separate compressed chunks

This change removes the synchronous callback from Rust's `Zlib::write` and schedules `processCallback` via `process.nextTick` from JavaScript. A `_callbackPending` flag ensures backward compatibility with Brotli/Zstd handles which still call the callback synchronously.

Enables 3 node_compat tests:
- test-zlib-destroy-pipe
- test-zlib-flush-drain
- test-zlib-flush-write-sync-interleaved

Changed files:
filename: ext/node/ops/zlib/mod.rs
patch: @@ -473,7 +473,7 @@ impl Zlib {
     #[smi] out_off: u32,
     #[smi] out_len: u32,
   ) -> Result<(), ZlibError> {
-    let (err_info, callback) = {
+    let err_info = {
       let mut zlib = self.inner.borrow_mut();
       let zlib = zlib.as_mut().ok_or(ZlibError::NotInitialized)?;
 
@@ -487,21 +487,14 @@ impl Zlib {
       };
       result[0] = zlib.strm.avail_out;
       result[1] = zlib.strm.avail_in;
-      (
-        zlib.get_error_info(),
-        v8::Local::new(
-          scope,
-          zlib.callback.as_ref().expect("callback not set"),
-        ),
-      )
+      zlib.get_error_info()
     };
 
-    if !ZlibInner::check_error(err_info, scope, &this) {
-      return Ok(());
-    }
-
-    let this = v8::Local::new(scope, &this);
-    let _ = callback.call(scope, this.into(), &[]);
+    // Report errors via onerror callback (which defers destroy via
+    // process.nextTick). The processCallback is NOT called here — it is
+    // scheduled asynchronously from JavaScript to match Node.js behavior
+    // where compression runs on the libuv threadpool.
+    ZlibInner::check_error(err_info, scope, &this);
 
     Ok(())
   }

filename: ext/node/polyfills/zlib.js
patch: @@ -563,6 +563,8 @@ function processChunk(self, chunk, flushFlag, cb) {
   handle.inOff = 0;
   handle.flushFlag = flushFlag;
 
+  handle._callbackPending = true;
+
   try {
     handle.write(
       flushFlag,
@@ -579,6 +581,14 @@ function processChunk(self, chunk, flushFlag, cb) {
       err.code = "ZSTD_error_srcSize_wrong";
     }
     self.destroy(err);
+    return;
+  }
+
+  // If the native binding did not call processCallback synchronously
+  // (Zlib defers it to match Node.js where compression runs on the libuv
+  // threadpool), schedule it asynchronously.
+  if (handle._callbackPending && !self[kError]) {
+    process.nextTick(processCallback.bind(handle));
   }
 }
 
@@ -587,6 +597,7 @@ function processCallback() {
   // important to null out the values once they are no longer needed since
   // `_handle` can stay in memory long after the buffer is needed.
   const handle = this;
+  handle._callbackPending = false;
   const self = this[owner_symbol];
   const state = self._writeState;
 
@@ -634,6 +645,12 @@ function processCallback() {
 
     if (!streamBufferIsFull) {
       process.nextTick(() => {
+        if (self.destroyed) {
+          this.buffer = null;
+          this.cb();
+          return;
+        }
+        handle._callbackPending = true;
         this.write(
           handle.flushFlag,
           this.buffer, // in
@@ -643,12 +660,21 @@ function processCallback() {
           self._outOffset, // out_off
           self._chunkSize,
         ); // out_len
+        if (handle._callbackPending && !self[kError]) {
+          processCallback.call(this);
+        }
       });
     } else {
       const oldRead = self._read;
       self._read = (n) => {
         self._read = oldRead;
         process.nextTick(() => {
+          if (self.destroyed) {
+            this.buffer = null;
+            this.cb();
+            return;
+          }
+          handle._callbackPending = true;
           this.write(
             handle.flushFlag,
             this.buffer, // in
@@ -658,6 +684,9 @@ function processCallback() {
             self._outOffset, // out_off
             self._chunkSize,
           ); // out_len
+          if (handle._callbackPending && !self[kError]) {
+            processCallback.call(this);
+          }
         });
         self._read(n);
       };

filename: tests/node_compat/config.jsonc
patch: @@ -1602,16 +1602,16 @@
     "parallel/test-zlib-create-raw.js": {},
     "parallel/test-zlib-deflate-constructors.js": {},
     "parallel/test-zlib-deflate-raw-inherits.js": {},
-    // "parallel/test-zlib-destroy-pipe.js": {},
+    "parallel/test-zlib-destroy-pipe.js": {},
     "parallel/test-zlib-destroy.js": {},
     "parallel/test-zlib-dictionary-fail.js": {},
     "parallel/test-zlib-dictionary.js": {},
     "parallel/test-zlib-empty-buffer.js": {},
     "parallel/test-zlib-failed-init.js": {},
     "parallel/test-zlib-flush-drain-longblock.js": {},
-    // "parallel/test-zlib-flush-drain.js": {},
+    "parallel/test-zlib-flush-drain.js": {},
     "parallel/test-zlib-flush-flags.js": {},
-    // "parallel/test-zlib-flush-write-sync-interleaved.js": {},
+    "parallel/test-zlib-flush-write-sync-interleaved.js": {},
     "parallel/test-zlib-flush.js": {},
     "parallel/test-zlib-from-concatenated-gzip.js": {},
     "parallel/test-zlib-from-gzip-with-trailing-garbage.js": {},