Address codec review: fix gaps and add missing functionality

- LZW: propagate errors from Deflater::encode() in compress() instead
  of silently swallowing them and returning truncated output
- checksum: add Hasher trait implementation for XXH64 (CRC32, Adler32,
  and XXH32 already had it), with checksum() returning lower 32 bits
- flate: add compress_with_dict() convenience function to match the
  existing decompress_with_dict()
- gzip: always write FHCRC (header CRC-16) in encoder output; add
  compress_members() for multi-member archive encoding to match
  decompress_members()
- snappy: add streaming FramedInflater for framed format decoding,
  symmetric with the existing FramedDeflater
- async: add async compress/decompress wrappers for zstd, brotli, lz4,
  and snappy (flate, gzip, zlib, bzip2, lzw already had them)

Author: bikallem

brotli/async/compress.mbt

@@ -0,0 +1,29 @@
+///|
+/// Reads raw data from `reader`, compresses it using Brotli, and writes
+/// the compressed stream to `writer`.
+pub async fn compress(
+  reader : &@io.Reader,
+  writer : &@io.Writer,
+  level? : @brotli.CompressionLevel = Default,
+) -> Unit {
+  let def = @brotli.Deflater::new(level~)
+  let buf = FixedArray::make(32768, b'\x00')
+  while true {
+    let n = reader.read(buf)
+    if n == 0 {
+      break
+    }
+    let chunk = Bytes::makei(n, fn(i) { buf[i] })
+    match def.encode(Some(chunk[:])) {
+      Data(out) => writer.write(out)
+      Ok => ()
+      End => return
+      Error(e) => raise e
+    }
+  }
+  match def.encode(None) {
+    Data(out) => writer.write(out)
+    End | Ok => ()
+    Error(e) => raise e
+  }
+}

brotli/async/decompress.mbt

@@ -0,0 +1,24 @@
+///|
+/// Reads Brotli-compressed data from `reader`, decompresses it, and writes
+/// the decompressed stream to `writer`.
+pub async fn decompress(reader : &@io.Reader, writer : &@io.Writer) -> Unit {
+  let inf = @brotli.Inflater::new()
+  let buf = FixedArray::make(32768, b'\x00')
+  loop inf.decode() {
+    Await => {
+      let n = reader.read(buf)
+      if n == 0 {
+        raise @brotli.BrotliError::UnexpectedEOF
+      }
+      let chunk = Bytes::makei(n, fn(i) { buf[i] })
+      inf.src(chunk[:])
+      continue inf.decode()
+    }
+    Data(out) => {
+      writer.write(out)
+      continue inf.decode()
+    }
+    End => ()
+    Error(e) => raise e
+  }
+}

brotli/async/moon.pkg

@@ -0,0 +1,4 @@
+import {
+  "bikallem/compress/brotli",
+  "moonbitlang/async/io",
+}

checksum/xxh64.mbt

@@ -193,3 +193,23 @@ pub fn XXH64::checksum64(self : XXH64) -> UInt64 {
   }
   xxh64_avalanche(h)
 }
+
+///|
+pub impl Hasher for XXH64 with size(self) -> Int {
+  self.size()
+}
+
+///|
+pub impl Hasher for XXH64 with reset(self) -> Unit {
+  self.reset()
+}
+
+///|
+pub impl Hasher for XXH64 with update(self, data) -> Unit {
+  self.update(data)
+}
+
+///|
+pub impl Hasher for XXH64 with checksum(self) -> UInt {
+  self.checksum64().to_uint()
+}

flate/convenience.mbt

@@ -44,6 +44,20 @@ pub fn compress(
   buf.to_bytes()
 }
 
+///|
+/// Compress data using DEFLATE with a preset dictionary.
+pub fn compress_with_dict(
+  data : Bytes,
+  dict : Bytes,
+  level? : CompressionLevel = DefaultCompression,
+) -> Bytes raise CompressError {
+  let e = Deflater::new(level~, dict~)
+  let buf = @blit.Buffer::new()
+  drain(e, buf, Some(data[:]))
+  drain(e, buf, None)
+  buf.to_bytes()
+}
+
 ///|
 fn drain(
   e : Deflater,

gzip/compress.mbt

@@ -30,7 +30,7 @@ fn build_gzip_header(header : Header, level : @flate.CompressionLevel) -> Bytes
   // Compression method: deflate
   buf.write_byte(GZIP_DEFLATE)
   // Build FLG byte
-  let mut flg = 0
+  let mut flg = GZIP_FHCRC
   if header.name.length() > 0 {
     flg = flg | GZIP_FNAME
   }
@@ -73,6 +73,26 @@ fn build_gzip_header(header : Header, level : @flate.CompressionLevel) -> Bytes
     }
     buf.write_byte(b'\x00')
   }
+  // FHCRC: CRC-32 of header so far, truncated to 16 bits (LE)
+  let header_bytes = buf.to_bytes()
+  let hcrc = @checksum.crc32(header_bytes[:])
+  buf.write_byte((hcrc & 0xFFU).to_byte())
+  buf.write_byte(((hcrc >> 8) & 0xFFU).to_byte())
+  buf.to_bytes()
+}
+
+///|
+/// Compresses multiple data segments as a multi-member gzip archive (RFC 1952).
+/// Each entry produces a separate gzip member; concatenated members form a valid archive.
+pub fn compress_members(
+  members : Array[(Bytes, Header)],
+  level? : @flate.CompressionLevel = DefaultCompression,
+) -> Bytes raise GzipError {
+  let buf = @blit.Buffer::new()
+  for entry in members {
+    let (data, header) = entry
+    buf.write_bytes(compress(data, level~, header~))
+  }
   buf.to_bytes()
 }
 

gzip/deflater_inflater_test.mbt

@@ -271,7 +271,8 @@ test "streaming gzip inflater preserves repeat state across byte chunks" {
 ///|
 test "streaming gzip inflater validates FHCRC across header chunks" {
   let original = b"header crc fixture"
-  let compressed = gzip_with_fhcrc(@gzip.compress(original))
+  // Encoder now always writes FHCRC — use directly.
+  let compressed = @gzip.compress(original)
   let d = @gzip.Inflater::new()
   let buf = @blit.Buffer::new()
   let mut pos = 0

gzip/gzip_test.mbt

@@ -126,35 +126,6 @@ fn gzip_named_header(name : String) -> @gzip.Header {
   { name, comment: "", extra: Bytes::default(), mod_time: 0U, os: b'\xFF' }
 }
 
-///|
-fn gzip_with_fhcrc(data : Bytes) -> Bytes {
-  let header = Bytes::makei(10, fn(i) {
-    if i == 3 {
-      (data[i].to_int() | 0x02).to_byte()
-    } else {
-      data[i]
-    }
-  })
-  let hcrc = @checksum.crc32(header[:]) & 0xFFFFU
-  let out = @blit.Buffer::new(size_hint=data.length() + 2)
-  out.write_bytes(header)
-  out.write_byte((hcrc & 0xFFU).to_byte())
-  out.write_byte(((hcrc >> 8) & 0xFFU).to_byte())
-  out.write_bytesview(data[10:])
-  out.to_bytes()
-}
-
-///|
-fn gzip_with_bad_fhcrc(data : Bytes) -> Bytes {
-  let with_crc = gzip_with_fhcrc(data)
-  Bytes::makei(with_crc.length(), fn(i) {
-    if i == 10 {
-      (with_crc[i].to_int() ^ 0xFF).to_byte()
-    } else {
-      with_crc[i]
-    }
-  })
-}
 
 ///|
 fn gzip_hex_digit(c : UInt16) -> Int {
@@ -217,16 +188,26 @@ test "gzip checksum mismatch" {
 ///|
 test "gzip accepts valid FHCRC" {
   let data = b"header crc fixture"
-  let compressed = gzip_with_fhcrc(@gzip.compress(data))
+  // Encoder now always writes FHCRC — just round-trip directly.
+  let compressed = @gzip.compress(data)
   let (decompressed, _header) = @gzip.decompress(compressed)
   assert_eq(decompressed, data)
 }
 
 ///|
 test "gzip rejects invalid FHCRC" {
   let data = b"header crc fixture"
-  let compressed = gzip_with_bad_fhcrc(@gzip.compress(data))
-  let result = try? @gzip.decompress(compressed)
+  // Encoder writes FHCRC as last 2 bytes of header (bytes 10-11 for a default header).
+  // Corrupt byte 10 to invalidate the header CRC.
+  let compressed = @gzip.compress(data)
+  let corrupted = Bytes::makei(compressed.length(), fn(i) {
+    if i == 10 {
+      (compressed[i].to_int() ^ 0xFF).to_byte()
+    } else {
+      compressed[i]
+    }
+  })
+  let result = try? @gzip.decompress(corrupted)
   guard result is Err(_) else { fail("expected error for bad header checksum") }
 }
 

lz4/async/compress.mbt

@@ -0,0 +1,29 @@
+///|
+/// Reads raw data from `reader`, compresses it using LZ4 frame format, and writes
+/// the compressed stream to `writer`.
+pub async fn compress(
+  reader : &@io.Reader,
+  writer : &@io.Writer,
+  options? : @lz4.FrameOptions = { ..@lz4.FrameOptions::default() },
+) -> Unit {
+  let def = @lz4.Deflater::new(options~)
+  let buf = FixedArray::make(32768, b'\x00')
+  while true {
+    let n = reader.read(buf)
+    if n == 0 {
+      break
+    }
+    let chunk = Bytes::makei(n, fn(i) { buf[i] })
+    match def.encode(Some(chunk[:])) {
+      Data(out) => writer.write(out)
+      Ok => ()
+      End => return
+      Error(e) => raise e
+    }
+  }
+  match def.encode(None) {
+    Data(out) => writer.write(out)
+    End | Ok => ()
+    Error(e) => raise e
+  }
+}

lz4/async/decompress.mbt

@@ -0,0 +1,25 @@
+///|
+/// Reads LZ4-compressed data from `reader`, decompresses it, and writes
+/// the decompressed stream to `writer`.
+pub async fn decompress(reader : &@io.Reader, writer : &@io.Writer) -> Unit {
+  let inf = @lz4.Inflater::new()
+  let buf = FixedArray::make(32768, b'\x00')
+  loop inf.decode() {
+    Await => {
+      let n = reader.read(buf)
+      if n == 0 {
+        inf.finish()
+        continue inf.decode()
+      }
+      let chunk = Bytes::makei(n, fn(i) { buf[i] })
+      inf.src(chunk[:])
+      continue inf.decode()
+    }
+    Data(out) => {
+      writer.write(out)
+      continue inf.decode()
+    }
+    End => ()
+    Error(e) => raise e
+  }
+}