From 3b884dcc220fedf1aa0289466c464cc762d4b5c4 Mon Sep 17 00:00:00 2001 From: Kauri Date: Tue, 13 Oct 2026 14:03:20 +0300 Subject: [PATCH] perf: fewer syscalls, copies and allocations on I/O paths (2.4.0) - VarInt/VarLong encode on the stack and write in one call; add `encode()` - RawPacket reads no longer zero the buffer first - uncompress reuses the inflated buffer; add consuming `into_uncompressed` - zlib compress/decompress reuse per-thread state, no 32 KiB BufReader - add `ByteArray` for bulk VarInt-prefixed bytes - cap Vec preallocation by bytes instead of element count - CFB8 `encrypt_in_place`/`decrypt_in_place`; sync write half reuses its buffer Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 3 +- Cargo.toml | 5 +- src/compression.rs | 83 +++++++++++++++++++----- src/encryption.rs | 29 +++++++-- src/packet.rs | 88 +++++++++++++++++++------- src/prelude/mod.rs | 2 +- src/ser.rs | 43 ++++++++++++- src/varint.rs | 84 ++++++++++++------------ tests/bulk_io.rs | 81 ++++++++++++++++++++++++ tests/compression.rs | 39 +++++++++++- tests/ui/non_serializable_field.stderr | 4 +- 11 files changed, 368 insertions(+), 93 deletions(-) create mode 100644 tests/bulk_io.rs diff --git a/Cargo.lock b/Cargo.lock index 5ecbe21..1a1a348 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -282,9 +282,10 @@ dependencies = [ [[package]] name = "mc_protocol" -version = "2.3.0" +version = "2.4.0" dependencies = [ "aes", + "bytes", "cfb8", "flate2", "mc_protocol_derive", diff --git a/Cargo.toml b/Cargo.toml index e4c1ddc..e56339f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,7 +26,7 @@ print_stderr = "warn" [package] name = "mc_protocol" -version = "2.3.0" +version = "2.4.0" edition.workspace = true rust-version.workspace = true authors.workspace = true @@ -47,7 +47,7 @@ workspace = true [features] default = ["async", "compression"] # Enable async I/O via tokio -async = ["dep:tokio"] +async = ["dep:tokio", "dep:bytes"] # Enable AES-128-CFB8 encryption support (pure Rust, RustCrypto) encryption = ["dep:aes", "dep:cfb8"] # Enable zlib packet compression support @@ -59,6 +59,7 @@ thiserror = "2.0.21" uuid = "1.27.0" tokio = { version = "1.53.2", features = ["io-util", "net"], optional = true } +bytes = { version = "1.12.1", optional = true } aes = { version = "0.9.3", optional = true } cfb8 = { version = "0.9.1", optional = true } flate2 = { version = "1.1.10", optional = true } diff --git a/src/compression.rs b/src/compression.rs index 599d3ce..fcbd441 100644 --- a/src/compression.rs +++ b/src/compression.rs @@ -16,10 +16,27 @@ //! assert_eq!(&decompressed, data); //! ``` -use flate2::{Compression, read::ZlibDecoder, write::ZlibEncoder}; -use std::io::{Read, Write}; +use flate2::{ + Compress, Compression, Decompress, FlushCompress, FlushDecompress, Status, + bufread::ZlibDecoder, write::ZlibEncoder, +}; +use std::cell::RefCell; +use std::io::{self, Read, Write}; use thiserror::Error; +/// Upper bound on the output buffer reserved up front, before any data is inflated. +const MAX_PREALLOC: usize = 16 * 1024 * 1024; + +thread_local! { + // Deflate/inflate state is large and costly to initialise, so reuse it per thread. + static DEFLATE: RefCell = RefCell::new(Compress::new(Compression::default(), true)); + static INFLATE: RefCell = RefCell::new(Decompress::new(true)); +} + +fn invalid_data(msg: impl Into) -> CompressionError { + CompressionError::Io(io::Error::new(io::ErrorKind::InvalidData, msg.into())) +} + /// Errors that can occur during compression or decompression. #[derive(Debug, Error)] pub enum CompressionError { @@ -39,9 +56,20 @@ impl From for crate::packet::PacketError { /// /// Returns the compressed bytes. pub fn compress_zlib(data: &[u8]) -> Result, CompressionError> { - let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); - encoder.write_all(data)?; - Ok(encoder.finish()?) + DEFLATE.with_borrow_mut(|deflate| { + deflate.reset(); + let mut output = Vec::with_capacity(data.len() / 2 + 64); + loop { + let consumed = deflate.total_in() as usize; + let status = deflate + .compress_vec(&data[consumed..], &mut output, FlushCompress::Finish) + .map_err(|e| invalid_data(e.to_string()))?; + if status == Status::StreamEnd { + return Ok(output); + } + output.reserve(output.capacity().max(64)); + } + }) } /// Compress `data` using zlib deflate at a specific compression level. @@ -49,7 +77,10 @@ pub fn compress_zlib(data: &[u8]) -> Result, CompressionError> { /// Valid levels are 0 (no compression) through 9 (best compression); /// values above 9 are clamped to 9. pub fn compress_zlib_level(data: &[u8], level: u32) -> Result, CompressionError> { - let mut encoder = ZlibEncoder::new(Vec::new(), Compression::new(level.min(9))); + let mut encoder = ZlibEncoder::new( + Vec::with_capacity(data.len() / 2 + 64), + Compression::new(level.min(9)), + ); encoder.write_all(data)?; Ok(encoder.finish()?) } @@ -72,15 +103,33 @@ pub fn decompress_zlib(data: &[u8]) -> Result, CompressionError> { /// exceed `limit` — this guards against zip bombs, where a tiny compressed /// input expands to a huge output. pub fn decompress_zlib_limited(data: &[u8], limit: usize) -> Result, CompressionError> { - // Read at most limit + 1 bytes: anything past `limit` means the input lied. - let mut decoder = ZlibDecoder::new(data).take(limit as u64 + 1); - let mut output = Vec::new(); - decoder.read_to_end(&mut output)?; - if output.len() > limit { - return Err(CompressionError::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!("decompressed size exceeds declared limit of {limit} bytes"), - ))); - } - Ok(output) + // Room for limit + 1 bytes: producing more than `limit` means the input lied. + let max_len = limit.saturating_add(1); + INFLATE.with_borrow_mut(|inflate| { + inflate.reset(true); + let mut output = Vec::with_capacity(max_len.min(MAX_PREALLOC)); + loop { + let (consumed, produced) = (inflate.total_in(), inflate.total_out()); + let status = inflate + .decompress_vec( + &data[consumed as usize..], + &mut output, + FlushDecompress::Finish, + ) + .map_err(|e| invalid_data(e.to_string()))?; + if output.len() > limit { + return Err(invalid_data(format!( + "decompressed size exceeds declared limit of {limit} bytes" + ))); + } + if status == Status::StreamEnd { + return Ok(output); + } + if output.len() == output.capacity() { + output.reserve(output.len().max(64).min(max_len - output.len())); + } else if (consumed, produced) == (inflate.total_in(), inflate.total_out()) { + return Err(CompressionError::Io(io::ErrorKind::UnexpectedEof.into())); + } + } + }) } diff --git a/src/encryption.rs b/src/encryption.rs index af8b4d2..72164d3 100644 --- a/src/encryption.rs +++ b/src/encryption.rs @@ -70,9 +70,14 @@ impl Cfb8Encryptor { /// Encrypt a slice of plaintext bytes, returning the ciphertext. pub fn encrypt(&mut self, plaintext: &[u8]) -> io::Result> { let mut output = plaintext.to_vec(); - self.cipher.encrypt(&mut output); + self.encrypt_in_place(&mut output); Ok(output) } + + /// Encrypt `data` in place, without allocating. + pub fn encrypt_in_place(&mut self, data: &mut [u8]) { + self.cipher.encrypt(data); + } } /// Stateful AES-128-CFB8 decryptor. @@ -92,9 +97,14 @@ impl Cfb8Decryptor { /// Decrypt a slice of ciphertext bytes, returning the plaintext. pub fn decrypt(&mut self, ciphertext: &[u8]) -> io::Result> { let mut output = ciphertext.to_vec(); - self.cipher.decrypt(&mut output); + self.decrypt_in_place(&mut output); Ok(output) } + + /// Decrypt `data` in place, without allocating. + pub fn decrypt_in_place(&mut self, data: &mut [u8]) { + self.cipher.decrypt(data); + } } // --------------------------------------------------------------------------- @@ -128,7 +138,7 @@ impl Read for Cfb8ReadHalf { if n == 0 { return Ok(0); } - self.decryptor.cipher.decrypt(&mut buf[..n]); + self.decryptor.decrypt_in_place(&mut buf[..n]); Ok(n) } } @@ -137,6 +147,8 @@ impl Read for Cfb8ReadHalf { pub struct Cfb8WriteHalf { inner: W, encryptor: Cfb8Encryptor, + /// Scratch space for ciphertext, reused across writes. + buffer: Vec, } impl Cfb8WriteHalf { @@ -145,6 +157,7 @@ impl Cfb8WriteHalf { Ok(Self { inner, encryptor: Cfb8Encryptor::new(key)?, + buffer: Vec::new(), }) } @@ -156,8 +169,10 @@ impl Cfb8WriteHalf { impl Write for Cfb8WriteHalf { fn write(&mut self, buf: &[u8]) -> io::Result { - let encrypted = self.encryptor.encrypt(buf)?; - self.inner.write_all(&encrypted)?; + self.buffer.clear(); + self.buffer.extend_from_slice(buf); + self.encryptor.encrypt_in_place(&mut self.buffer); + self.inner.write_all(&self.buffer)?; Ok(buf.len()) } @@ -298,9 +313,11 @@ impl AsyncWrite for AsyncCfb8WriteHalf { return Poll::Ready(Ok(0)); } + // The drain above emptied the buffer, so only fresh plaintext is encrypted here. let this = &mut *self; + debug_assert!(this.buffer.is_empty()); this.buffer.extend_from_slice(buf); - this.encryptor.cipher.encrypt(&mut this.buffer); + this.encryptor.encrypt_in_place(&mut this.buffer); // Opportunistically write; whatever the inner writer doesn't take now // stays buffered and is drained on the next write/flush/shutdown. diff --git a/src/packet.rs b/src/packet.rs index 16567f6..4459438 100644 --- a/src/packet.rs +++ b/src/packet.rs @@ -27,7 +27,7 @@ use crate::ser::{Deserialize, SerializationError, Serialize}; use crate::varint::{VarInt, VarIntError}; #[cfg(feature = "async")] -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; // --------------------------------------------------------------------------- // Errors @@ -74,6 +74,25 @@ fn active_threshold(threshold: Option) -> Option { threshold.and_then(|t| usize::try_from(t).ok()) } +fn checked_read_len(len: VarInt) -> Result { + if len.0 < 0 || len.0 as usize > MAX_PACKET_LENGTH { + return Err(PacketError::InvalidLength(len.0 as i64)); + } + Ok(len.0 as usize) +} + +/// Splits `[packet_id][payload]` in place, reusing the buffer for the payload. +fn split_packet_id(mut data: Vec, offset: usize) -> Result { + let mut cursor = Cursor::new(&data[offset..]); + let packet_id = VarInt::read_sync(&mut cursor)?.0; + let end = offset + cursor.position() as usize; + data.drain(..end); + Ok(UncompressedPacket { + packet_id, + payload: data, + }) +} + fn checked_frame_len(len: usize) -> Result { if len > MAX_PACKET_LENGTH { return Err(PacketError::InvalidLength(len as i64)); @@ -118,13 +137,16 @@ impl RawPacket { /// /// Fails with [`PacketError::InvalidLength`] if the declared length is /// negative or exceeds [`MAX_PACKET_LENGTH`]. + /// + /// The length prefix is read byte by byte, so wrap unbuffered sources such + /// as sockets in a `BufReader`. pub fn read_sync(reader: &mut R) -> Result { - let len = VarInt::read_sync(reader)?; - if len.0 < 0 || len.0 as usize > MAX_PACKET_LENGTH { - return Err(PacketError::InvalidLength(len.0 as i64)); + let len = checked_read_len(VarInt::read_sync(reader)?)?; + let mut data = Vec::with_capacity(len); + reader.take(len as u64).read_to_end(&mut data)?; + if data.len() != len { + return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into()); } - let mut data = vec![0u8; len.0 as usize]; - reader.read_exact(&mut data)?; Ok(RawPacket { data }) } @@ -141,20 +163,28 @@ impl RawPacket { /// /// Fails with [`PacketError::InvalidLength`] if the declared length is /// negative or exceeds [`MAX_PACKET_LENGTH`]. + /// + /// The length prefix is read byte by byte, so wrap unbuffered sources such + /// as sockets in a `tokio::io::BufReader`. Not cancellation safe: a frame + /// interrupted midway is lost and the stream is left desynchronized. #[cfg(feature = "async")] - pub async fn read_async(reader: &mut R) -> Result { - let len = VarInt::read_async(reader).await?; - if len.0 < 0 || len.0 as usize > MAX_PACKET_LENGTH { - return Err(PacketError::InvalidLength(len.0 as i64)); + pub async fn read_async(reader: &mut R) -> Result { + use bytes::BufMut; + + let len = checked_read_len(VarInt::read_async(reader).await?)?; + let mut data = Vec::with_capacity(len); + while data.len() < len { + let remaining = len - data.len(); + if reader.read_buf(&mut (&mut data).limit(remaining)).await? == 0 { + return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into()); + } } - let mut data = vec![0u8; len.0 as usize]; - reader.read_exact(&mut data).await?; Ok(RawPacket { data }) } /// Write this packet frame to an async writer; rejects frames over [`MAX_PACKET_LENGTH`]. #[cfg(feature = "async")] - pub async fn write_async( + pub async fn write_async( &self, writer: &mut W, ) -> Result<(), PacketError> { @@ -226,14 +256,28 @@ impl RawPacket { if uncompressed.len() != data_length.0 as usize { return Err(PacketError::InvalidLength(uncompressed.len() as i64)); } - let mut inner = Cursor::new(&uncompressed); - let packet_id = VarInt::read_sync(&mut inner)?; - let pos = inner.position() as usize; - let payload = uncompressed[pos..].to_vec(); - Ok(UncompressedPacket { - packet_id: packet_id.0, - payload, - }) + split_packet_id(uncompressed, 0) + } + + /// Like [`uncompress`](Self::uncompress), but consumes the frame so an + /// uncompressed body is reused in place instead of copied. + pub fn into_uncompressed( + self, + threshold: Option, + ) -> Result { + #[cfg(feature = "compression")] + if active_threshold(threshold).is_some() { + let mut cursor = Cursor::new(&self.data); + let data_length = VarInt::read_sync(&mut cursor)?.0; + if data_length != 0 { + return self.uncompress(threshold); + } + let offset = cursor.position() as usize; + return split_packet_id(self.data, offset); + } + #[cfg(not(feature = "compression"))] + let _ = threshold; + split_packet_id(self.data, 0) } /// Interpret frame without feature-gating (no compression). @@ -352,7 +396,7 @@ impl UncompressedPacket { /// Write this packet to an async writer (uncompressed framing). #[cfg(feature = "async")] - pub async fn write_async( + pub async fn write_async( &self, writer: &mut W, ) -> Result<(), PacketError> { diff --git a/src/prelude/mod.rs b/src/prelude/mod.rs index 0fc801a..2be481a 100644 --- a/src/prelude/mod.rs +++ b/src/prelude/mod.rs @@ -25,7 +25,7 @@ //! AES-128-CFB8 types. // --- Core serialization --- -pub use crate::ser::{Deserialize, RawBytes, SerializationError, Serialize}; +pub use crate::ser::{ByteArray, Deserialize, RawBytes, SerializationError, Serialize}; // --- Variable-length integers --- pub use crate::varint::{VarInt, VarLong}; diff --git a/src/ser.rs b/src/ser.rs index 182e2e1..d264db8 100644 --- a/src/ser.rs +++ b/src/ser.rs @@ -323,7 +323,8 @@ impl Deserialize for Vec { // Cap the pre-allocation: the length prefix is attacker-controlled, // so don't reserve gigabytes up front. The vector still grows to the // full length as elements are actually read. - let mut result = Vec::with_capacity((len as usize).min(4096)); + let cap = MAX_PREALLOC_BYTES / std::mem::size_of::().max(1); + let mut result = Vec::with_capacity((len as usize).min(cap)); for _ in 0..len { result.push(T::deserialize(reader)?); } @@ -331,6 +332,46 @@ impl Deserialize for Vec { } } +/// Upper bound on memory reserved from an untrusted length prefix before data arrives. +const MAX_PREALLOC_BYTES: usize = 64 * 1024; + +// --------------------------------------------------------------------------- +// Prefixed byte array +// --------------------------------------------------------------------------- + +/// A VarInt-length-prefixed byte array, copied in bulk. +/// +/// Wire-compatible with `Vec`, which (de)serializes one element at a time. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct ByteArray(pub Vec); + +impl Serialize for ByteArray { + fn serialize(&self, writer: &mut W) -> Result<(), SerializationError> { + let len = i32::try_from(self.0.len()) + .map_err(|_| SerializationError::InvalidLength(self.0.len() as i64))?; + VarInt(len).write_sync(writer)?; + writer.write_all(&self.0)?; + Ok(()) + } +} + +impl Deserialize for ByteArray { + fn deserialize(reader: &mut R) -> Result { + let len = VarInt::read_sync(reader)?.0; + if len < 0 || len as usize > crate::packet::MAX_UNCOMPRESSED_LENGTH { + return Err(SerializationError::InvalidLength(len as i64)); + } + let len = len as usize; + // Grows only as bytes actually arrive, so a lying prefix costs nothing. + let mut buf = Vec::with_capacity(len.min(MAX_PREALLOC_BYTES)); + reader.take(len as u64).read_to_end(&mut buf)?; + if buf.len() != len { + return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into()); + } + Ok(Self(buf)) + } +} + // --------------------------------------------------------------------------- // Raw byte array (no length prefix) // --------------------------------------------------------------------------- diff --git a/src/varint.rs b/src/varint.rs index eb3791e..f47df3b 100644 --- a/src/varint.rs +++ b/src/varint.rs @@ -91,6 +91,22 @@ impl VarInt { count } + /// Encode into a stack buffer; returns the buffer and the number of bytes used. + pub fn encode(&self) -> ([u8; 5], usize) { + let mut buf = [0u8; 5]; + let mut value = self.0 as u32; + let mut len = 0; + loop { + if value & !(SEGMENT_BITS as u32) == 0 { + buf[len] = value as u8; + return (buf, len + 1); + } + buf[len] = ((value as u8) & SEGMENT_BITS) | CONTINUE_BIT; + value >>= 7; + len += 1; + } + } + // --- Synchronous I/O --- /// Decode a `VarInt` from a synchronous reader. @@ -122,15 +138,9 @@ impl VarInt { /// Encode this `VarInt` into a synchronous writer. pub fn write_sync(&self, writer: &mut W) -> Result<(), VarIntError> { - let mut value = self.0 as u32; - loop { - if value & !(SEGMENT_BITS as u32) == 0 { - writer.write_all(&[value as u8])?; - return Ok(()); - } - writer.write_all(&[((value as u8) & SEGMENT_BITS) | CONTINUE_BIT])?; - value >>= 7; - } + let (buf, len) = self.encode(); + writer.write_all(&buf[..len])?; + Ok(()) } // --- Async I/O --- @@ -164,17 +174,9 @@ impl VarInt { &self, writer: &mut W, ) -> Result<(), VarIntError> { - let mut value = self.0 as u32; - loop { - if value & !(SEGMENT_BITS as u32) == 0 { - writer.write_u8(value as u8).await?; - return Ok(()); - } - writer - .write_u8(((value as u8) & SEGMENT_BITS) | CONTINUE_BIT) - .await?; - value >>= 7; - } + let (buf, len) = self.encode(); + writer.write_all(&buf[..len]).await?; + Ok(()) } } @@ -221,6 +223,22 @@ impl VarLong { count } + /// Encode into a stack buffer; returns the buffer and the number of bytes used. + pub fn encode(&self) -> ([u8; 10], usize) { + let mut buf = [0u8; 10]; + let mut value = self.0 as u64; + let mut len = 0; + loop { + if value & !(SEGMENT_BITS as u64) == 0 { + buf[len] = value as u8; + return (buf, len + 1); + } + buf[len] = ((value as u8) & SEGMENT_BITS) | CONTINUE_BIT; + value >>= 7; + len += 1; + } + } + // --- Synchronous I/O --- /// Decode a `VarLong` from a synchronous reader. @@ -252,15 +270,9 @@ impl VarLong { /// Encode this `VarLong` into a synchronous writer. pub fn write_sync(&self, writer: &mut W) -> Result<(), VarIntError> { - let mut value = self.0 as u64; - loop { - if value & !(SEGMENT_BITS as u64) == 0 { - writer.write_all(&[value as u8])?; - return Ok(()); - } - writer.write_all(&[((value as u8) & SEGMENT_BITS) | CONTINUE_BIT])?; - value >>= 7; - } + let (buf, len) = self.encode(); + writer.write_all(&buf[..len])?; + Ok(()) } // --- Async I/O --- @@ -294,17 +306,9 @@ impl VarLong { &self, writer: &mut W, ) -> Result<(), VarIntError> { - let mut value = self.0 as u64; - loop { - if value & !(SEGMENT_BITS as u64) == 0 { - writer.write_u8(value as u8).await?; - return Ok(()); - } - writer - .write_u8(((value as u8) & SEGMENT_BITS) | CONTINUE_BIT) - .await?; - value >>= 7; - } + let (buf, len) = self.encode(); + writer.write_all(&buf[..len]).await?; + Ok(()) } } diff --git a/tests/bulk_io.rs b/tests/bulk_io.rs new file mode 100644 index 0000000..fb8f0fd --- /dev/null +++ b/tests/bulk_io.rs @@ -0,0 +1,81 @@ +//! Tests for bulk, allocation-light I/O paths. + +use std::io::Cursor; + +use mc_protocol::packet::RawPacket; +use mc_protocol::ser::{ByteArray, Deserialize, Serialize}; +use mc_protocol::varint::{VarInt, VarLong}; + +#[test] +fn byte_array_is_wire_compatible_with_vec_u8() { + let bytes: Vec = (0..300u32).map(|i| i as u8).collect(); + let mut from_vec = Vec::new(); + bytes.serialize(&mut from_vec).unwrap(); + let mut from_array = Vec::new(); + ByteArray(bytes.clone()).serialize(&mut from_array).unwrap(); + assert_eq!(from_vec, from_array); + assert_eq!( + ByteArray::deserialize(&mut Cursor::new(&from_vec)) + .unwrap() + .0, + bytes + ); +} + +#[test] +fn byte_array_rejects_truncated_and_lying_prefix() { + let mut buf = Vec::new(); + VarInt(1_000_000).write_sync(&mut buf).unwrap(); + buf.extend_from_slice(&[1, 2, 3]); + assert!(ByteArray::deserialize(&mut Cursor::new(&buf)).is_err()); + + let mut buf = Vec::new(); + VarInt(-1).write_sync(&mut buf).unwrap(); + assert!(ByteArray::deserialize(&mut Cursor::new(&buf)).is_err()); +} + +#[test] +fn encode_matches_byte_length() { + for v in [0, 1, 127, 128, 300, i32::MAX, -1, i32::MIN] { + let (buf, len) = VarInt(v).encode(); + assert_eq!(len, VarInt(v).encoded_len()); + assert_eq!(VarInt::read_sync(&mut &buf[..len]).unwrap().0, v); + } + for v in [0, 1, i64::MAX, -1, i64::MIN] { + let (buf, len) = VarLong(v).encode(); + assert_eq!(len, VarLong(v).encoded_len()); + assert_eq!(VarLong::read_sync(&mut &buf[..len]).unwrap().0, v); + } +} + +#[test] +fn read_sync_rejects_truncated_frame() { + let mut buf = Vec::new(); + VarInt(10).write_sync(&mut buf).unwrap(); + buf.extend_from_slice(&[0; 4]); + assert!(RawPacket::read_sync(&mut Cursor::new(&buf)).is_err()); +} + +#[cfg(feature = "async")] +#[tokio::test] +async fn read_async_reads_exact_frames_and_rejects_truncation() { + let first = RawPacket::new(vec![1; 70_000]); + let second = RawPacket::new(vec![2; 3]); + let mut buf = Vec::new(); + first.write_sync(&mut buf).unwrap(); + second.write_sync(&mut buf).unwrap(); + + let mut reader = &buf[..]; + assert_eq!( + RawPacket::read_async(&mut reader).await.unwrap().data, + first.data + ); + assert_eq!( + RawPacket::read_async(&mut reader).await.unwrap().data, + second.data + ); + + let mut truncated = &buf[..buf.len() - 1]; + RawPacket::read_async(&mut truncated).await.unwrap(); + assert!(RawPacket::read_async(&mut truncated).await.is_err()); +} diff --git a/tests/compression.rs b/tests/compression.rs index c1696cb..0931291 100644 --- a/tests/compression.rs +++ b/tests/compression.rs @@ -1,6 +1,8 @@ //! Tests for zlib packet compression. -use mc_protocol::compression::{compress_zlib, compress_zlib_level, decompress_zlib}; +use mc_protocol::compression::{ + compress_zlib, compress_zlib_level, decompress_zlib, decompress_zlib_limited, +}; use mc_protocol::packet::UncompressedPacket; // --------------------------------------------------------------------------- @@ -197,3 +199,38 @@ fn large_packet_round_trip_preserves_exact_content() { assert_eq!(decoded.payload, payload); assert_eq!(decoded.payload.len(), 2000); } + +#[test] +fn into_uncompressed_matches_uncompress() { + let packet = UncompressedPacket::new(0x27, (0..5000u32).map(|i| i as u8).collect()); + for threshold in [None, Some(-1), Some(0), Some(64), Some(100_000)] { + let raw = packet.to_raw_packet_compressed(threshold).unwrap(); + let borrowed = raw.uncompress(threshold).unwrap(); + let owned = raw.into_uncompressed(threshold).unwrap(); + assert_eq!(borrowed.packet_id, 0x27); + assert_eq!(owned.packet_id, 0x27); + assert_eq!(owned.payload, packet.payload); + assert_eq!(borrowed.payload, packet.payload); + } +} + +#[test] +fn decompress_zlib_limited_rejects_truncated_stream() { + let data: Vec = (0..10_000u32).map(|i| (i * 7) as u8).collect(); + let compressed = compress_zlib(&data).unwrap(); + let truncated = &compressed[..compressed.len() / 2]; + assert!(decompress_zlib_limited(truncated, data.len()).is_err()); +} + +#[test] +fn reused_zlib_state_round_trips_repeatedly() { + for i in 0..50u32 { + let data: Vec = (0..(i * 997)).map(|j| (j ^ i) as u8).collect(); + let compressed = compress_zlib(&data).unwrap(); + assert_eq!(decompress_zlib(&compressed).unwrap(), data); + assert_eq!( + decompress_zlib_limited(&compressed, data.len()).unwrap(), + data + ); + } +} diff --git a/tests/ui/non_serializable_field.stderr b/tests/ui/non_serializable_field.stderr index 33f787b..9c39168 100644 --- a/tests/ui/non_serializable_field.stderr +++ b/tests/ui/non_serializable_field.stderr @@ -10,6 +10,7 @@ help: the trait `Serialize` is not implemented for `Opaque` 1 | struct Opaque; | ^^^^^^^^^^^^^ = help: the following other types implement trait `Serialize`: + ByteArray HasOpaque Option RawBytes @@ -17,7 +18,6 @@ help: the trait `Serialize` is not implemented for `Opaque` VarLong Vec bool - f32 and $N others error[E0277]: the trait bound `Opaque: Deserialize` is not satisfied @@ -32,6 +32,7 @@ help: the trait `Deserialize` is not implemented for `Opaque` 1 | struct Opaque; | ^^^^^^^^^^^^^ = help: the following other types implement trait `Deserialize`: + ByteArray HasOpaque Option String @@ -39,5 +40,4 @@ help: the trait `Deserialize` is not implemented for `Opaque` Vec bool f32 - f64 and $N others