From a4cb9785df84ed98dede2ead2878c868013e1cd9 Mon Sep 17 00:00:00 2001 From: fanyang Date: Sat, 7 Nov 2026 23:10:56 +0800 Subject: [PATCH] feat: support LZO compression (#2204) Add optional lzo compression support using lzokay-native Wire LZO through compression proto/config mappings and packet compression metadata --- Cargo.lock | 11 +++ easytier-core/Cargo.toml | 2 + easytier-core/src/packet/compressor.rs | 90 +++++++++++++++++++--- easytier-core/src/packet/compressor/lzo.rs | 36 +++++++++ easytier-core/src/packet/mod.rs | 22 ++++++ easytier-proto/Cargo.toml | 2 + easytier-proto/proto/common.proto | 1 + easytier/Cargo.toml | 3 + easytier/locales/app.yml | 4 +- 9 files changed, 159 insertions(+), 12 deletions(-) create mode 100644 easytier-core/src/packet/compressor/lzo.rs diff --git a/Cargo.lock b/Cargo.lock index dfd3a3be..3e77d58c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2680,6 +2680,7 @@ dependencies = [ "hyper", "hyper-util", "idna", + "lzokay-native", "openssl", "optionize", "ordered_hash_map", @@ -5322,6 +5323,16 @@ version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" +[[package]] +name = "lzokay-native" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae83dc0af9b591a1c6a5d5ff799a27668710e8866f7d6239fe6f3c7f30282752" +dependencies = [ + "byteorder", + "thiserror 1.0.69", +] + [[package]] name = "lzma-rust2" version = "0.16.5" diff --git a/easytier-core/Cargo.toml b/easytier-core/Cargo.toml index 8d23387c..563a281f 100644 --- a/easytier-core/Cargo.toml +++ b/easytier-core/Cargo.toml @@ -89,6 +89,7 @@ webpki-roots = { version = "1.0", optional = true } x25519-dalek = { workspace = true, features = ["static_secrets"] } zerocopy = { workspace = true, features = ["derive", "simd"] } zstd = { version = "0.14", optional = true } +lzokay-native = { version = "0.2", optional = true } # Match snow 0.10's AEAD generation to share aead, cipher and crypto-common. # Upgrade together with snow rather than pulling in a second crypto stack. aes-gcm = { version = "0.10.3", optional = true } @@ -171,6 +172,7 @@ wasm-host-tunnel-outbound = ["wasm-host-tunnel"] test-utils = [] tracing-log = ["tracing/log"] zstd = ["dep:zstd"] +lzo = ["dep:lzokay-native"] [dev-dependencies] futures = { workspace = true, features = ["executor"] } diff --git a/easytier-core/src/packet/compressor.rs b/easytier-core/src/packet/compressor.rs index 68db8d66..fde0d595 100644 --- a/easytier-core/src/packet/compressor.rs +++ b/easytier-core/src/packet/compressor.rs @@ -2,6 +2,7 @@ use zerocopy::{AsBytes as _, FromBytes as _}; use super::{COMPRESSOR_TAIL_SIZE, CompressorAlgo, CompressorTail, ZCPacket}; +mod lzo; mod zstd; type Error = anyhow::Error; @@ -36,6 +37,7 @@ impl DefaultCompressor { ) -> Result, Error> { match compress_algo { CompressorAlgo::ZstdDefault => zstd::compress(data, compress_algo), + CompressorAlgo::Lzo => lzo::compress(data, compress_algo), CompressorAlgo::None => Ok(data.to_vec()), } } @@ -47,6 +49,7 @@ impl DefaultCompressor { ) -> Result, Error> { match compress_algo { CompressorAlgo::ZstdDefault => zstd::decompress(data, compress_algo), + CompressorAlgo::Lzo => lzo::decompress(data, compress_algo), CompressorAlgo::None => Ok(data.to_vec()), } } @@ -145,11 +148,10 @@ pub(super) fn zstd_available() -> bool { pub mod tests { use super::*; - #[cfg(feature = "zstd")] - #[tokio::test] - async fn test_compress() { - let text = b"12345670000000000000000000"; - let mut packet = ZCPacket::new_with_payload(text); + #[cfg(any(feature = "zstd", feature = "lzo"))] + async fn test_compress_algo(compress_algo: CompressorAlgo) { + let text = vec![b'a'; 4096]; + let mut packet = ZCPacket::new_with_payload(&text); packet.fill_peer_manager_hdr(0, 0, 0); let compressor = DefaultCompressor {}; @@ -161,7 +163,7 @@ pub mod tests { ); compressor - .compress(&mut packet, CompressorAlgo::ZstdDefault) + .compress(&mut packet, compress_algo) .await .unwrap(); println!( @@ -176,9 +178,8 @@ pub mod tests { assert!(!packet.peer_manager_header().unwrap().is_compressed()); } - #[cfg(feature = "zstd")] - #[tokio::test] - async fn test_short_text_compress() { + #[cfg(any(feature = "zstd", feature = "lzo"))] + async fn test_short_text_compress_algo(compress_algo: CompressorAlgo) { let text = b"1234"; let mut packet = ZCPacket::new_with_payload(text); packet.fill_peer_manager_hdr(0, 0, 0); @@ -187,7 +188,7 @@ pub mod tests { // short text can't be compressed compressor - .compress(&mut packet, CompressorAlgo::ZstdDefault) + .compress(&mut packet, compress_algo) .await .unwrap(); assert!(!packet.peer_manager_header().unwrap().is_compressed()); @@ -197,6 +198,30 @@ pub mod tests { assert!(!packet.peer_manager_header().unwrap().is_compressed()); } + #[cfg(feature = "zstd")] + #[tokio::test] + async fn test_zstd_compress() { + test_compress_algo(CompressorAlgo::ZstdDefault).await; + } + + #[cfg(feature = "zstd")] + #[tokio::test] + async fn test_zstd_short_text_compress() { + test_short_text_compress_algo(CompressorAlgo::ZstdDefault).await; + } + + #[cfg(feature = "lzo")] + #[tokio::test] + async fn test_lzo_compress() { + test_compress_algo(CompressorAlgo::Lzo).await; + } + + #[cfg(feature = "lzo")] + #[tokio::test] + async fn test_lzo_short_text_compress() { + test_short_text_compress_algo(CompressorAlgo::Lzo).await; + } + #[cfg(not(feature = "zstd"))] #[tokio::test] async fn unavailable_zstd_returns_an_explicit_error() { @@ -210,4 +235,49 @@ pub mod tests { "compression algorithm is unavailable in this build: ZstdDefault" ); } + + #[cfg(not(feature = "lzo"))] + #[tokio::test] + async fn unavailable_lzo_returns_an_explicit_error() { + let compressor = DefaultCompressor::new(); + for result in [ + compressor + .compress_raw(b"payload", CompressorAlgo::Lzo) + .await, + compressor + .decompress_raw(b"payload", CompressorAlgo::Lzo) + .await, + ] { + assert_eq!( + result.unwrap_err().to_string(), + "compression algorithm is unavailable in this build: Lzo" + ); + } + } + + #[cfg(feature = "lzo")] + #[tokio::test] + async fn test_lzo_decompress_output_limit() { + let compressor = DefaultCompressor::new(); + for size in [64 * 1024, 64 * 1024 + 1] { + let text = vec![b'a'; size]; + let compressed = compressor + .compress_raw(&text, CompressorAlgo::Lzo) + .await + .unwrap(); + assert!(compressed.len() * 32 < 64 * 1024); + + let result = compressor + .decompress_raw(&compressed, CompressorAlgo::Lzo) + .await; + if size == 64 * 1024 { + assert_eq!(result.unwrap(), text); + } else { + assert!(matches!( + result.unwrap_err().downcast_ref::(), + Some(lzokay_native::Error::OutputOverrun) + )); + } + } + } } diff --git a/easytier-core/src/packet/compressor/lzo.rs b/easytier-core/src/packet/compressor/lzo.rs new file mode 100644 index 00000000..f0d9ec5c --- /dev/null +++ b/easytier-core/src/packet/compressor/lzo.rs @@ -0,0 +1,36 @@ +#[cfg(feature = "lzo")] +use anyhow::Context as _; + +use super::CompressorAlgo; + +#[cfg(feature = "lzo")] +pub(super) fn compress(data: &[u8], compress_algo: CompressorAlgo) -> anyhow::Result> { + lzokay_native::compress(data).with_context(|| { + format!( + "Failed to compress data with algorithm: {:?}", + compress_algo + ) + }) +} + +#[cfg(not(feature = "lzo"))] +pub(super) fn compress(_data: &[u8], compress_algo: CompressorAlgo) -> anyhow::Result> { + Err(super::super::CompressionUnavailableError(compress_algo).into()) +} + +#[cfg(feature = "lzo")] +pub(super) fn decompress(data: &[u8], compress_algo: CompressorAlgo) -> anyhow::Result> { + // Match the maximum output size allowed by the Zstd path. + let max_size = data.len().saturating_mul(32).max(64 * 1024); + lzokay_native::decompress_all_bounded(data, max_size).with_context(|| { + format!( + "Failed to decompress data with algorithm: {:?}", + compress_algo + ) + }) +} + +#[cfg(not(feature = "lzo"))] +pub(super) fn decompress(_data: &[u8], compress_algo: CompressorAlgo) -> anyhow::Result> { + Err(super::super::CompressionUnavailableError(compress_algo).into()) +} diff --git a/easytier-core/src/packet/mod.rs b/easytier-core/src/packet/mod.rs index f627d5b2..5b16a0f9 100644 --- a/easytier-core/src/packet/mod.rs +++ b/easytier-core/src/packet/mod.rs @@ -354,6 +354,7 @@ pub type StandardAeadTail = AeadTail<16, 12>; pub enum CompressorAlgo { None = 0, ZstdDefault = 1, + Lzo = 2, } impl CompressorAlgo { @@ -361,6 +362,7 @@ impl CompressorAlgo { match self { Self::None => true, Self::ZstdDefault => compressor::zstd_available(), + Self::Lzo => cfg!(feature = "lzo"), } } @@ -385,6 +387,7 @@ impl TryFrom for CompressorAlgo { fn try_from(value: CompressionAlgoPb) -> Result { match value { CompressionAlgoPb::Zstd => Ok(CompressorAlgo::ZstdDefault), + CompressionAlgoPb::Lzo => Ok(CompressorAlgo::Lzo), CompressionAlgoPb::None => Ok(CompressorAlgo::None), _ => Err(CompressionAlgoError(value)), } @@ -397,6 +400,7 @@ impl TryFrom for CompressionAlgoPb { fn try_from(value: CompressorAlgo) -> Result { match value { CompressorAlgo::ZstdDefault => Ok(CompressionAlgoPb::Zstd), + CompressorAlgo::Lzo => Ok(CompressionAlgoPb::Lzo), CompressorAlgo::None => Ok(CompressionAlgoPb::None), } } @@ -413,6 +417,7 @@ impl CompressorTail { pub fn get_algo(&self) -> Option { match self.algo { 1 => Some(CompressorAlgo::ZstdDefault), + 2 => Some(CompressorAlgo::Lzo), _ => None, } } @@ -970,6 +975,23 @@ mod tests { ); } + #[test] + fn converts_lzo_compression_algo_in_every_profile() { + assert_eq!( + CompressorAlgo::Lzo, + CompressorAlgo::try_from(CompressionAlgoPb::Lzo).unwrap() + ); + assert_eq!( + CompressionAlgoPb::Lzo, + CompressionAlgoPb::try_from(CompressorAlgo::Lzo).unwrap() + ); + assert_eq!( + Some(CompressorAlgo::Lzo), + CompressorTail { algo: 2 }.get_algo() + ); + assert_eq!(CompressorAlgo::Lzo.is_available(), cfg!(feature = "lzo")); + } + #[cfg(not(feature = "zstd"))] #[test] fn reports_zstd_as_unavailable_without_changing_vocabulary() { diff --git a/easytier-proto/Cargo.toml b/easytier-proto/Cargo.toml index 1e74dc7f..85557960 100644 --- a/easytier-proto/Cargo.toml +++ b/easytier-proto/Cargo.toml @@ -77,6 +77,7 @@ full = [ "websocket", "wireguard", "zstd", + "lzo", "json-rpc", ] api = ["core"] @@ -113,6 +114,7 @@ utils = [] websocket = [] wireguard = [] zstd = [] +lzo = [] json-rpc = ["dep:prost-wkt-types"] [package.metadata.cargo-machete] diff --git a/easytier-proto/proto/common.proto b/easytier-proto/proto/common.proto index 5cbeb9fa..3a1ae9c7 100644 --- a/easytier-proto/proto/common.proto +++ b/easytier-proto/proto/common.proto @@ -137,6 +137,7 @@ enum CompressionAlgoPb { Invalid = 0; None = 1; Zstd = 2; + Lzo = 3; } message RpcCompressionInfo { diff --git a/easytier/Cargo.toml b/easytier/Cargo.toml index a468e6f6..f674ed73 100644 --- a/easytier/Cargo.toml +++ b/easytier/Cargo.toml @@ -333,6 +333,7 @@ default = [ "faketcp", "magic-dns", "zstd", + "lzo", "upnp", "icmp-proxy", "management", @@ -355,6 +356,7 @@ full = [ "faketcp", "magic-dns", "zstd", + "lzo", "upnp", "icmp-proxy", "management", @@ -414,6 +416,7 @@ faketcp = [ "easytier-proto/faketcp", ] zstd = ["easytier-core/zstd", "easytier-proto/zstd"] +lzo = ["easytier-core/lzo", "easytier-proto/lzo"] upnp = [ "dep:http-body-util", "dep:hyper", diff --git a/easytier/locales/app.yml b/easytier/locales/app.yml index 69d9704e..d168138c 100644 --- a/easytier/locales/app.yml +++ b/easytier/locales/app.yml @@ -213,8 +213,8 @@ core_clap: en: "the url of the ipv6 listener, e.g.: tcp://[::]:11010, if not set, will listen on random udp port" zh-CN: "IPv6 监听器的URL，例如：tcp://[::]:11010，如果未设置，将在随机UDP端口上监听" compression: - en: "compression algorithm to use, support None, Zstd. default is None" - zh-CN: "要使用的压缩算法，支持 None、Zstd。默认为 None" + en: "compression algorithm to use, support None, Zstd, Lzo. default is None" + zh-CN: "要使用的压缩算法，支持 None、Zstd、Lzo。默认为 None" mapped_listeners: en: "manually specify the public address of the listener, other nodes can use this address to connect to this node. e.g.: tcp://123.123.123.123:11223, can specify multiple." zh-CN: "手动指定监听器的公网地址，其他节点可以使用该地址连接到本节点。例如：tcp://123.123.123.123:11223，可以指定多个。"