node:zlib: don't abort when the native handle is driven outside the zlib.ts lifecycle

src/runtime/node/node_zlib_binding.rs

@@ -279,6 +279,34 @@ pub(crate) trait CompressionStreamImpl: Sized + Taskable + 'static {
 }
 
 impl<T: CompressionStreamImpl> CompressionStream<T> {
+    /// Rejects a call on a handle that cannot accept a new operation: an async
+    /// write still holds `&mut Context` on a worker thread, a pending close is
+    /// about to tear it down, or a closed one already did (`mode` is `NONE`).
+    pub(crate) fn throw_unless_idle(this: &T, global_this: &JSGlobalObject) -> JsResult<()> {
+        if this.write_in_progress().get() {
+            return Err(global_this
+                .err(
+                    ErrorCode::INVALID_STATE,
+                    format_args!("Write already in progress"),
+                )
+                .throw());
+        }
+        if this.pending_close().get() {
+            return Err(global_this
+                .err(ErrorCode::INVALID_STATE, format_args!("Pending close"))
+                .throw());
+        }
+        if this.closed().get() {
+            return Err(global_this
+                .err(
+                    ErrorCode::INVALID_STATE,
+                    format_args!("zlib binding closed"),
+                )
+                .throw());
+        }
+        Ok(())
+    }
+
     pub(crate) fn write(
         this: &T,
         global_this: &JSGlobalObject,
@@ -375,19 +403,7 @@ impl<T: CompressionStreamImpl> CompressionStream<T> {
         }
         let _ = (in_off, in_len, out_off, out_len);
 
-        if this.write_in_progress().get() {
-            return Err(global_this
-                .err(
-                    ErrorCode::INVALID_STATE,
-                    format_args!("Write already in progress"),
-                )
-                .throw());
-        }
-        if this.pending_close().get() {
-            return Err(global_this
-                .err(ErrorCode::INVALID_STATE, format_args!("Pending close"))
-                .throw());
-        }
+        Self::throw_unless_idle(this, global_this)?;
         // Pin both buffers before mutating any state: materializing a
         // FastTypedArray's backing store can fail on OOM, and failing here
         // leaves nothing to unwind.
@@ -540,10 +556,12 @@ impl<T: CompressionStreamImpl> CompressionStream<T> {
         this.flush_write_result(global, this_value);
         this_value.ensure_still_alive();
 
-        let write_callback: JSValue = T::write_callback_get_cached(this_value).unwrap();
-
-        vm.event_loop_ref()
-            .run_callback(write_callback, global, this_value, &[]);
+        // `init()` caches the JS write callback; a handle whose `init()` was
+        // never called has none, so there is nothing to notify.
+        if let Some(write_callback) = T::write_callback_get_cached(this_value) {
+            vm.event_loop_ref()
+                .run_callback(write_callback, global, this_value, &[]);
+        }
 
         if this.pending_reset().get() {
             Self::reset_internal(&this, global, this_value);
@@ -664,19 +682,7 @@ impl<T: CompressionStreamImpl> CompressionStream<T> {
         );
         let _ = (in_off, in_len, out_off, out_len);
 
-        if this.write_in_progress().get() {
-            return Err(global_this
-                .err(
-                    ErrorCode::INVALID_STATE,
-                    format_args!("Write already in progress"),
-                )
-                .throw());
-        }
-        if this.pending_close().get() {
-            return Err(global_this
-                .err(ErrorCode::INVALID_STATE, format_args!("Pending close"))
-                .throw());
-        }
+        Self::throw_unless_idle(this, global_this)?;
         this.write_in_progress().set(true);
         this.ref_();
 
@@ -827,20 +833,19 @@ impl<T: CompressionStreamImpl> CompressionStream<T> {
             Err(_) => return,
         };
 
-        let callback: JSValue = T::error_callback_get_cached(this_value).unwrap_or_else(|| {
-            bun_core::Output::panic(format_args!(
-                "Assertion failure: cachedErrorCallback is null in node:zlib binding",
-            ))
-        });
-
-        // SAFETY: `bun_vm()` and `event_loop()` are non-null for a Bun-owned global.
-        let vm = global_this.bun_vm();
-        vm.event_loop_ref().run_callback(
-            callback,
-            global_this,
-            this_value,
-            &[msg_value, err_value, code_value],
-        );
+        // The `zlib.ts` wrapper installs `onerror` right after construction,
+        // but a handle driven directly has none; there is nobody to notify.
+        // The pending reset/close handling below still runs either way.
+        if let Some(callback) = T::error_callback_get_cached(this_value) {
+            // SAFETY: `bun_vm()` and `event_loop()` are non-null for a Bun-owned global.
+            let vm = global_this.bun_vm();
+            vm.event_loop_ref().run_callback(
+                callback,
+                global_this,
+                this_value,
+                &[msg_value, err_value, code_value],
+            );
+        }
 
         if this.pending_reset().get() {
             Self::reset_internal(this, global_this, this_value);

src/runtime/node/zlib/NativeBrotli.rs

@@ -189,6 +189,9 @@ mod _impl {
                     )
                     .throw());
             }
+            // Racing an in-flight async write would alias `&mut Context`
+            // across threads; a closed `Context` cannot be re-initialized.
+            CompressionStream::<Self>::throw_unless_idle(self, global_this)?;
 
             // `flush_write_result` writes two u32s into this array, so the
             // caller-supplied array must hold at least 2 elements.
@@ -265,6 +268,9 @@ mod _impl {
                 if err.is_error() {
                     // impl.emitError(this, globalThis, this_value, err); //XXX: onerror isn't set yet
                     self.stream.with_mut(|s| s.close());
+                    // The Context is torn down (`mode` is `NONE`); reject any
+                    // further operation the way `close()` does.
+                    self.closed.set(true);
                     return Ok(JSValue::FALSE);
                 }
             }
@@ -429,6 +435,11 @@ mod _impl {
         }
 
         pub fn do_work(&mut self) {
+            // A handle driven before `init()` has no encoder/decoder state;
+            // brotli dereferences the state pointer unconditionally.
+            if self.state.is_none() {
+                return;
+            }
             match self.mode {
                 bun_zlib::NodeMode::BROTLI_ENCODE => {
                     let mut next_in = self.next_in;
@@ -521,7 +532,11 @@ mod _impl {
         }
 
         pub fn close(&mut self) {
-            self.deinit_state();
+            // Idempotent: a handle that was never (successfully) initialized,
+            // or that was already closed, has no encoder/decoder to free.
+            if self.state.is_some() {
+                self.deinit_state();
+            }
             self.mode = bun_zlib::NodeMode::NONE;
         }
 

src/runtime/node/zlib/NativeZlib.rs

@@ -130,6 +130,9 @@ mod _impl {
                 )
                 .throw());
             }
+            // Racing an in-flight async write would alias `&mut Context`
+            // across threads; a closed `Context` cannot be re-initialized.
+            CompressionStream::<Self>::throw_unless_idle(self, global)?;
 
             let window_bits =
                 validators::validate_int32(global, arguments.ptr[0], "windowBits", None, None)?;
@@ -183,6 +186,18 @@ mod _impl {
                         ));
                     }
                 };
+                // `deflateSetDictionary`/`inflateSetDictionary` take a `uInt`
+                // length; reject anything larger before `Context::init` copies it.
+                if dictionary_buf.byte_len > u32::MAX as usize {
+                    return Err(global.throw_range_error(
+                        dictionary_buf.byte_len as i64,
+                        bun_jsc::RangeErrorOptions {
+                            field_name: b"dictionary.byteLength",
+                            max: i64::from(u32::MAX),
+                            ..Default::default()
+                        },
+                    ));
+                }
                 Some(dictionary_buf.byte_slice())
             };
 
@@ -195,6 +210,12 @@ mod _impl {
 
             self.stream
                 .with_mut(|s| s.init(level, window_bits, mem_level, strategy, dictionary));
+            // `Context::init` leaves `mode` at `NONE` when `deflateInit2_` /
+            // `inflateInit2_` rejects its arguments; mark the wrapper closed so
+            // the next operation is rejected instead of re-entering `NONE`.
+            if self.stream.with_mut(|s| s.mode == c::NodeMode::NONE) {
+                self.closed.set(true);
+            }
 
             Ok(JSValue::UNDEFINED)
         }
@@ -211,6 +232,9 @@ mod _impl {
                     )
                     .throw());
             }
+            // `set_params` calls `deflateParams` on the same `z_stream` an
+            // in-flight async write's `deflate()` is using.
+            CompressionStream::<Self>::throw_unless_idle(self, global)?;
 
             let level = validators::validate_int32(global, arguments.ptr[0], "level", None, None)?;
             let strategy =
@@ -609,7 +633,14 @@ impl Context {
             BROTLI_ENCODE | BROTLI_DECODE => {}
             ZSTD_COMPRESS | ZSTD_DECOMPRESS => {}
         }
-        debug_assert!(status == c::ReturnCode::Ok || status == c::ReturnCode::DataError);
+        // Ok: normal. DataError: pending output discarded by inflateEnd.
+        // StreamError: a handle whose init() threw before deflateInit2_/
+        // inflateInit2_ ran, so the zeroed z_stream has nothing to free.
+        debug_assert!(
+            status == c::ReturnCode::Ok
+                || status == c::ReturnCode::DataError
+                || status == c::ReturnCode::StreamError
+        );
         self.mode = NONE;
     }
 }

src/runtime/node/zlib/NativeZstd.rs

@@ -152,6 +152,9 @@ mod _impl {
                     )
                     .throw());
             }
+            // Racing an in-flight async write would alias `&mut Context`
+            // across threads; a closed `Context` cannot be re-initialized.
+            CompressionStream::<Self>::throw_unless_idle(self, global)?;
 
             let init_params_array_value = arguments[0];
             let pledged_src_size_value = arguments[1];
@@ -233,6 +236,9 @@ mod _impl {
                     .with_mut(|s| s.set_params(c_uint::try_from(i).expect("int cast"), x));
                 if err_.is_error() {
                     self.stream.with_mut(|s| s.close());
+                    // The Context is torn down (`mode` is `NONE`); reject any
+                    // further operation the way `close()` does.
+                    self.closed.set(true);
                     // SAFETY: is_error() ⇔ msg is non-null; it points at a NUL-terminated C string.
                     let msg = unsafe { bun_core::ffi::cstr(err_.msg) }.to_bytes();
                     return Err(global
@@ -420,6 +426,11 @@ mod _impl {
         }
 
         pub fn do_work(&mut self) {
+            // A handle driven before `init()` has no CCtx/DCtx; zstd
+            // dereferences the context pointer unconditionally.
+            if self.state.is_none() {
+                return;
+            }
             self.remaining = match self.mode {
                 // SAFETY: state is a valid CCtx; input/output point to caller-kept-alive buffers (set_buffers).
                 NodeMode::ZSTD_COMPRESS => unsafe {
@@ -514,6 +525,12 @@ mod _impl {
         }
 
         pub fn close(&mut self) {
+            // Idempotent: a handle that was never (successfully) initialized,
+            // or that was already closed, has no CCtx/DCtx to reset or free.
+            if self.state.is_none() {
+                self.mode = NodeMode::NONE;
+                return;
+            }
             let _ = match self.mode {
                 // SAFETY: state is a valid CCtx/DCtx for this mode.
                 NodeMode::ZSTD_COMPRESS => unsafe {