fix(node): handle brotli compression end chunk sizes (#22322)

Fixes https://github.com/denoland/deno/issues/22259

3 files changed, 30 insertions, 11 deletions

diff --git a/cli/tests/unit_node/zlib_test.ts b/cli/tests/unit_node/zlib_test.ts
index fa94493c1..b61cd4d4f 100644
--- a/cli/tests/unit_node/zlib_test.ts
+++ b/cli/tests/unit_node/zlib_test.ts
@@ -14,6 +14,8 @@ import {
 } from "node:zlib";
 import { Buffer } from "node:buffer";
 import { createReadStream, createWriteStream } from "node:fs";
+import { Readable } from "node:stream";
+import { buffer } from "node:stream/consumers";
 
 Deno.test("brotli compression sync", () => {
   const buf = Buffer.from("hello world");
@@ -155,3 +157,16 @@ Deno.test("zlib compression with an encoded string", {
   const decompressed = unzipSync(compressed);
   assertEquals(decompressed.toString(), "hello world");
 });
+
+Deno.test("brotli large chunk size", async () => {
+  const input = new Uint8Array(1000000);
+  for (let i = 0; i < input.length; i++) {
+    input[i] = Math.random() * 256;
+  }
+  const output = await buffer(
+    Readable.from([input])
+      .pipe(createBrotliCompress())
+      .pipe(createBrotliDecompress()),
+  );
+  assertEquals(output.length, input.length);
+});
diff --git a/ext/node/ops/zlib/brotli.rs b/ext/node/ops/zlib/brotli.rs
index bc8aa9a25..f906e8827 100644
--- a/ext/node/ops/zlib/brotli.rs
+++ b/ext/node/ops/zlib/brotli.rs
@@ -167,7 +167,6 @@ pub fn op_brotli_compress_stream(
     let mut next_in = input.as_ptr();
     let mut available_out = output.len();
     let mut next_out = output.as_mut_ptr();
-    let mut total_out = 0;
 
     if BrotliEncoderCompressStream(
       ctx.inst,
@@ -176,7 +175,7 @@ pub fn op_brotli_compress_stream(
       &mut next_in,
       &mut available_out,
       &mut next_out,
-      &mut total_out,
+      std::ptr::null_mut(),
     ) != 1
     {
       return Err(type_error("Failed to compress"));
@@ -194,7 +193,7 @@ pub fn op_brotli_compress_stream_end(
   #[smi] rid: u32,
   #[buffer] output: &mut [u8],
 ) -> Result<usize, AnyError> {
-  let ctx = state.resource_table.take::<BrotliCompressCtx>(rid)?;
+  let ctx = state.resource_table.get::<BrotliCompressCtx>(rid)?;
 
   // SAFETY: TODO(littledivy)
   unsafe {
@@ -281,7 +280,6 @@ pub fn op_brotli_decompress_stream(
     let mut next_in = input.as_ptr();
     let mut available_out = output.len();
     let mut next_out = output.as_mut_ptr();
-    let mut total_out = 0;
 
     if matches!(
       CBrotliDecoderDecompressStream(
@@ -290,11 +288,12 @@ pub fn op_brotli_decompress_stream(
         &mut next_in,
         &mut available_out,
         &mut next_out,
-        &mut total_out,
+        std::ptr::null_mut(),
       ),
       BrotliDecoderResult::BROTLI_DECODER_RESULT_ERROR
     ) {
-      return Err(type_error("Failed to decompress"));
+      let ec = CBrotliDecoderGetErrorCode(ctx.inst) as i32;
+      return Err(type_error(format!("Failed to decompress, error {ec}")));
     }
 
     // On progress, next_out is advanced and available_out is reduced.
diff --git a/ext/node/polyfills/_brotli.js b/ext/node/polyfills/_brotli.js
index c815add22..b019836cf 100644
--- a/ext/node/polyfills/_brotli.js
+++ b/ext/node/polyfills/_brotli.js
@@ -51,7 +51,7 @@ export class BrotliDecompress extends Transform {
       // TODO(littledivy): use `encoding` argument
       transform(chunk, _encoding, callback) {
         const input = toU8(chunk);
-        const output = new Uint8Array(1024);
+        const output = new Uint8Array(chunk.byteLength);
         const avail = op_brotli_decompress_stream(context, input, output);
         this.push(output.slice(0, avail));
         callback();
@@ -76,14 +76,19 @@ export class BrotliCompress extends Transform {
       transform(chunk, _encoding, callback) {
         const input = toU8(chunk);
         const output = new Uint8Array(brotliMaxCompressedSize(input.length));
-        const avail = op_brotli_compress_stream(context, input, output);
-        this.push(output.slice(0, avail));
+        const written = op_brotli_compress_stream(context, input, output);
+        if (written > 0) {
+          this.push(output.slice(0, written));
+        }
         callback();
       },
       flush(callback) {
         const output = new Uint8Array(1024);
-        const avail = op_brotli_compress_stream_end(context, output);
-        this.push(output.slice(0, avail));
+        let avail;
+        while ((avail = op_brotli_compress_stream_end(context, output)) > 0) {
+          this.push(output.slice(0, avail));
+        }
+        core.close(context);
         callback();
       },
     });