Compressed MongoDB and MySQL commands allocate nothing on WASI (I03)

protocols/turnloop-mongodb/Cargo.toml

@@ -38,9 +38,9 @@ workspace = true
 
 [package.metadata.turnloop-ci]
 role = "protocol"
-wasi-tests = ["asynchronous"]
+wasi-tests = ["asynchronous", "allocations"]
 # Pinned p3 compiler: custom allocator traps before main in debug (wasm2 report).
-wasi-p3-release-tests = ["asynchronous"]
+wasi-p3-release-tests = ["asynchronous", "allocations"]
 wasi-integration-tests = ["async_server"]
 wasi-setup-tests = ["real_mongodb"]
 integration-tests = ["real_mongodb", "async_server"]

protocols/turnloop-mongodb/src/connection.rs

@@ -64,7 +64,8 @@ impl Connection {
             decoder: Decoder::new(wire::DEFAULT_MAX_MESSAGE),
             expanded: Vec::with_capacity(8192),
             compressed: Vec::with_capacity(8192),
-            compressor: flate2::Compress::new(flate2::Compression::default(), true),
+            // Raw deflate: zlib.rs writes the RFC 1950 wrapper per message.
+            compressor: flate2::Compress::new(flate2::Compression::default(), false),
             decompressor: flate2::Decompress::new(true),
             zlib: false,
             events: VecDeque::with_capacity(8),
@@ -175,20 +176,9 @@ impl Connection {
             self.compressed
                 .extend_from_slice(&((self.tx.len() - 16) as i32).to_le_bytes());
             self.compressed.push(2);
-            self.compressor.reset();
-            self.compressed
-                .reserve(self.tx.len() + self.tx.len() / 100 + 256);
-            let status = self
-                .compressor
-                .compress_vec(
-                    &self.tx[16..],
-                    &mut self.compressed,
-                    flate2::FlushCompress::Finish,
-                )
-                .map_err(|_| Error::protocol("Compression failed"))?;
-            if status != flate2::Status::StreamEnd {
-                return Err(Error::protocol("Compression buffer exhausted"));
-            }
+            // One retained deflate state, one zlib stream per message; see zlib.rs.
+            crate::zlib::append_stream(&mut self.compressor, &self.tx[16..], &mut self.compressed)
+                .map_err(Error::protocol)?;
             let n = self.compressed.len() as i32;
             self.compressed[..4].copy_from_slice(&n.to_le_bytes());
             std::mem::swap(&mut self.tx, &mut self.compressed);

protocols/turnloop-mongodb/src/lib.rs

@@ -29,6 +29,7 @@ pub mod topology;
 pub use time::Instant;
 pub mod uri;
 pub mod wire;
+mod zlib;
 pub use connection::{Connection, ConnectionEvent};
 pub use error::{Error, ErrorKind, Result};
 

protocols/turnloop-mongodb/src/zlib.rs

@@ -0,0 +1,165 @@
+//! One retained deflate state per connection, one independent zlib stream per
+//! message, and no allocation after the first.
+//!
+//! `OP_COMPRESSED` payloads are separate RFC 1950 streams, which normally means
+//! `flate2::Compress::reset` between messages. That is free natively and is not
+//! free on wasm32: miniz_oxide keeps the LZ code buffer on the heap there
+//! (`deflate/core.rs`, `LZOxide::new`) and `CompressorOxide::reset` replaces it
+//! wholesale, so one reset per message is one allocation per message on WASI and
+//! on the web. DESIGN §3.3 forbids per-operation allocation in every
+//! configuration, so nothing here resets.
+//!
+//! Each message is flushed with `FlushCompress::Full` instead. Its documented
+//! contract writes every pending byte, ends on a byte boundary and clears the
+//! match dictionary, so the bytes that follow are decodable by a decompressor
+//! that starts there. The RFC 1950 wrapper is then written directly: the
+//! two-byte header ahead of the deflate data, and behind it a final empty
+//! fixed-Huffman block — valid precisely because the full flush left the stream
+//! byte-aligned — followed by the big-endian Adler-32 of the message. The result
+//! is a complete, independently decompressible stream per message, while the
+//! compressor keeps its storage for the life of the connection.
+use flate2::{Compress, FlushCompress, Status};
+
+/// CM 8 (deflate) and CINFO 7 (32 KiB window), FLEVEL 2, with FCHECK chosen so
+/// the pair is a multiple of 31 as RFC 1950 § 2.2 requires.
+const HEADER: [u8; 2] = [0x78, 0x9c];
+/// BFINAL 1 and BTYPE 01 (fixed Huffman) followed by the seven-bit end-of-block
+/// symbol: the shortest terminator for an already byte-aligned deflate stream.
+const FINAL_BLOCK: [u8; 2] = [0x03, 0x00];
+/// The four-byte Adler-32 plus `FINAL_BLOCK`.
+const TRAILER: usize = FINAL_BLOCK.len() + 4;
+
+/// Bytes [`append_stream`] can append for `len` input bytes. Comfortably above
+/// zlib's own `deflateBound` (`len + len/4096 + len/16384 + 13`, the stored-block
+/// worst case) plus the two-byte header, the five-byte full-flush marker and the
+/// trailer, so a complete flush always leaves unused capacity behind.
+fn bound(len: usize) -> usize {
+    len + len / 1000 + 64
+}
+
+/// Appends one complete zlib stream of `input` to `out`, reusing `compressor`'s
+/// storage. Allocates only if `out` lacks [`bound`] spare bytes.
+pub(crate) fn append_stream(
+    compressor: &mut Compress,
+    input: &[u8],
+    out: &mut Vec<u8>,
+) -> Result<(), &'static str> {
+    out.reserve(bound(input.len()));
+    out.extend_from_slice(&HEADER);
+    let consumed = compressor.total_in();
+    let start = out.len();
+    let status = compressor
+        .compress_vec(input, out, FlushCompress::Full)
+        .map_err(|_| "Compression failed")?;
+    if status != Status::Ok || compressor.total_in() - consumed != input.len() as u64 {
+        return Err("Compression did not consume the message");
+    }
+    // compress_vec writes into spare capacity only. Capacity left over proves the
+    // flush was not cut short by a full output buffer, and holds the trailer.
+    if out.capacity() - out.len() < TRAILER {
+        return Err("Compression buffer exhausted");
+    }
+    debug_assert!(out.len() - start <= bound(input.len()) - HEADER.len() - TRAILER);
+    out.extend_from_slice(&FINAL_BLOCK);
+    out.extend_from_slice(&adler32(input).to_be_bytes());
+    Ok(())
+}
+
+/// RFC 1950 § 9. 5552 is the longest run that cannot overflow before reduction.
+fn adler32(input: &[u8]) -> u32 {
+    const BASE: u32 = 65521;
+    let (mut low, mut high) = (1_u32, 0_u32);
+    for chunk in input.chunks(5552) {
+        for &byte in chunk {
+            low += u32::from(byte);
+            high += low;
+        }
+        low %= BASE;
+        high %= BASE;
+    }
+    (high << 16) | low
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use flate2::{Compression, Decompress, FlushDecompress};
+
+    fn inflate(stream: &[u8], expected: usize) -> Vec<u8> {
+        // A fresh decompressor per message: exactly what the server does.
+        let mut decoder = Decompress::new(true);
+        let mut out = vec![0; expected + 1];
+        let status = decoder
+            .decompress(stream, &mut out, FlushDecompress::Finish)
+            .expect("fixture operation must succeed");
+        assert_eq!(status, Status::StreamEnd, "stream must terminate");
+        assert_eq!(decoder.total_in(), stream.len() as u64, "trailing bytes");
+        out.truncate(decoder.total_out() as usize);
+        out
+    }
+
+    #[test]
+    fn adler32_matches_rfc1950_examples() {
+        assert_eq!(adler32(b""), 1);
+        assert_eq!(adler32(b"a"), 0x0062_0062);
+        assert_eq!(adler32(b"Wikipedia"), 0x11E6_0398);
+        // Two full reduction windows, so the chunked accumulation is exercised.
+        assert_eq!(adler32(&[0xff; 11_104]), 0xFF6F_3726);
+    }
+
+    #[test]
+    fn consecutive_messages_decompress_independently() {
+        let mut compressor = Compress::new(Compression::default(), false);
+        let mut out = Vec::new();
+        let mut decompressed = 0;
+        for round in 0..64_u32 {
+            let mut message = Vec::new();
+            for index in 0..300_u32 {
+                message.extend_from_slice(&(round * index).to_le_bytes());
+            }
+            out.clear();
+            append_stream(&mut compressor, &message, &mut out)
+                .expect("fixture operation must succeed");
+            assert!(out.len() < message.len(), "round {round} must compress");
+            assert_eq!(inflate(&out, message.len()), message, "round {round}");
+            decompressed += 1;
+        }
+        assert_eq!(decompressed, 64);
+    }
+
+    #[test]
+    fn steady_state_appends_without_allocating_or_exceeding_the_bound() {
+        let mut compressor = Compress::new(Compression::default(), false);
+        // Incompressible input is deflate's worst case for the bound.
+        let mut message = Vec::new();
+        let mut state = 0x2545_F491_4F6C_DD1D_u64;
+        for _ in 0..8192 {
+            state ^= state << 13;
+            state ^= state >> 7;
+            state ^= state << 17;
+            message.extend_from_slice(&state.to_le_bytes());
+        }
+        let mut out = Vec::with_capacity(bound(message.len()));
+        let capacity = out.capacity();
+        for round in 0..16 {
+            out.clear();
+            append_stream(&mut compressor, &message, &mut out)
+                .expect("fixture operation must succeed");
+            assert_eq!(out.capacity(), capacity, "round {round} reallocated");
+            assert!(out.len() <= bound(message.len()), "round {round} bound");
+            assert_eq!(inflate(&out, message.len()), message, "round {round}");
+        }
+    }
+
+    #[test]
+    fn empty_and_single_byte_messages_round_trip() {
+        let mut compressor = Compress::new(Compression::default(), false);
+        let mut out = Vec::new();
+        for message in [b"".as_slice(), b"x".as_slice(), &[0_u8; 1]] {
+            out.clear();
+            append_stream(&mut compressor, message, &mut out)
+                .expect("fixture operation must succeed");
+            assert_eq!(inflate(&out, message.len().max(1)), message);
+        }
+    }
+}

protocols/turnloop-mongodb/tests/allocations.rs

@@ -63,6 +63,7 @@ fn measure(compressed: bool, coordinator: bool) {
     use turnloop_mongodb::operation::*;
     let mut operation = Operation::new();
     let mut rows = 0;
+    let mut commands = 0;
     let mut run = |round| {
         cmd.find("db", "items", &filter, None).unwrap();
         if coordinator {
@@ -100,6 +101,16 @@ fn measure(compressed: bool, coordinator: bool) {
             c.command(round, cmd.raw(), &[], now).unwrap();
         }
         let req = wire::i32_at(c.transmit(), 4).unwrap();
+        // The measured subject must be the compressed encoder in compressed mode.
+        assert_eq!(
+            wire::i32_at(c.transmit(), 12).unwrap(),
+            if compressed {
+                wire::OP_COMPRESSED
+            } else {
+                wire::OP_MSG
+            }
+        );
+        commands += 1;
         let n = c.transmit().len();
         c.consume_transmit(n).unwrap();
         wire::encode(&mut reply, 77, req, 0, &result, &[], 10000).unwrap();
@@ -136,6 +147,13 @@ fn measure(compressed: bool, coordinator: bool) {
         }
     });
     assert_eq!(rows, 2004);
+    assert_eq!(
+        commands, 1002,
+        "two warm-ups plus every measured command ran"
+    );
+    println!(
+        "zlib={compressed} operation={coordinator}: 1000 measured commands, 2000 measured rows, {allocations} allocations"
+    );
     assert_eq!(
         allocations, 0,
         "1000 warmed commands/2000 rows allocated {allocations} times (zlib={compressed}, operation={coordinator})"

protocols/turnloop-mongodb/tests/protocol.rs

@@ -777,3 +777,93 @@ fn discarded_lease_preserves_other_connections_and_generation() {
         "stale lease rejected"
     );
 }
+/// One retained deflate state must still emit one independently decodable RFC 1950
+/// stream per command (`src/zlib.rs`). Every round is decoded by a fresh upstream
+/// decoder that never saw the previous message, exactly as a server does.
+#[test]
+fn compressed_commands_are_independent_zlib_streams() {
+    use std::io::Read;
+    let mut compressed =
+        Connection::new(Options::parse("mongodb://localhost/?compressors=zlib").unwrap());
+    let mut plain = ready();
+    compressed.connected(clock::now(), "").unwrap();
+    let req = Message::parse(compressed.transmit(), wire::DEFAULT_MAX_MESSAGE)
+        .unwrap()
+        .request_id;
+    let n = compressed.transmit().len();
+    compressed.consume_transmit(n).unwrap();
+    let mut hello = Vec::new();
+    wire::encode(
+        &mut hello,
+        2,
+        req,
+        0,
+        &raw(&doc! {"ok":1,"maxWireVersion":27,"compression":["zlib"]}),
+        &[],
+        wire::DEFAULT_MAX_MESSAGE,
+    )
+    .unwrap();
+    feed(&mut compressed, &hello);
+    assert!(matches!(
+        compressed.poll_event(),
+        Some(ConnectionEvent::Ready)
+    ));
+    let mut decoded_rounds = 0;
+    let mut shrank = 0;
+    for round in 0..64_u64 {
+        // Repetitive filters make real deflate output smaller than the input.
+        let body = raw(&doc! {
+            "find":"items","$db":"db","filter":{"tag":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
+            "comment":format!("round {round} aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
+        });
+        compressed.command(round, &body, &[], clock::now()).unwrap();
+        plain.command(round, &body, &[], clock::now()).unwrap();
+        let framed = compressed.transmit().to_vec();
+        let expected = plain.transmit().to_vec();
+        assert_eq!(wire::i32_at(&framed, 12).unwrap(), wire::OP_COMPRESSED);
+        assert_eq!(wire::i32_at(&framed, 0).unwrap() as usize, framed.len());
+        assert_eq!(wire::i32_at(&framed, 16).unwrap(), wire::OP_MSG);
+        let size = wire::i32_at(&framed, 20).unwrap() as usize;
+        assert_eq!(size, expected.len() - 16);
+        assert_eq!(framed[24], 2, "zlib compressor id");
+        if framed.len() < expected.len() {
+            shrank += 1;
+        }
+        let mut body_out = Vec::new();
+        let read = flate2::read::ZlibDecoder::new(&framed[25..])
+            .read_to_end(&mut body_out)
+            .unwrap();
+        assert_eq!(read, size, "round {round} declared size");
+        // The decompressed payload must be the exact OP_MSG an uncompressed
+        // connection sends, header excluded.
+        assert_eq!(body_out, expected[16..], "round {round} payload");
+        decoded_rounds += 1;
+        for (c, reply_to) in [
+            (&mut compressed, wire::i32_at(&framed, 4).unwrap()),
+            (&mut plain, wire::i32_at(&expected, 4).unwrap()),
+        ] {
+            let n = c.transmit().len();
+            c.consume_transmit(n).unwrap();
+            let mut reply = Vec::new();
+            wire::encode(
+                &mut reply,
+                900,
+                reply_to,
+                0,
+                &raw(&doc! {"ok":1}),
+                &[],
+                wire::DEFAULT_MAX_MESSAGE,
+            )
+            .unwrap();
+            feed(c, &reply);
+            assert!(matches!(
+                c.poll_event(),
+                Some(ConnectionEvent::Reply { token }) if token == round
+            ));
+            assert_eq!(c.reply().unwrap().get_i32("ok").unwrap(), 1);
+            c.release_reply().unwrap();
+        }
+    }
+    assert_eq!(decoded_rounds, 64, "every round must be decoded");
+    assert_eq!(shrank, 64, "every round must actually compress");
+}

protocols/turnloop-mysql/Cargo.toml

@@ -30,9 +30,9 @@ workspace = true
 
 [package.metadata.turnloop-ci]
 role = "protocol"
-wasi-tests = ["asynchronous"]
+wasi-tests = ["asynchronous", "allocations"]
 # Pinned p3 compiler: custom allocator traps before main in debug (wasm2 report).
-wasi-p3-release-tests = ["asynchronous"]
+wasi-p3-release-tests = ["asynchronous", "allocations"]
 wasi-integration-tests = ["async_server"]
 integration-tests = ["server", "async_server"]
 

protocols/turnloop-mysql/README.md

@@ -84,8 +84,11 @@ terminates input. To reject an enabled request safely, abort the connection.
 
 Compression negotiates CLIENT_COMPRESS. Plain and compressed framing handle
 24-bit lengths, sequence wrapping, continuation and exact-boundary empty frames.
-The outer zlib encoder/decoder states are retained and reset; upstream's
-per-frame zlib construction did not satisfy this lane's allocation contract.
+The outer zlib encoder/decoder states are retained for the connection; upstream's
+per-frame zlib construction did not satisfy this lane's allocation contract. The
+encoder is never reset either — each frame is a full-flushed segment wrapped in
+its own RFC 1950 header, final block and Adler-32 (`src/zlib.rs`), because
+resetting a deflate state reallocates its LZ buffer on wasm32.
 Zstd and MariaDB extensions are out of scope.
 
 ## mysql2 conversions and hooks