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

@@ -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"

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"] }

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<Vec<u8>, 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<Vec<u8>, 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::<lzokay_native::Error>(),
+                    Some(lzokay_native::Error::OutputOverrun)
+                ));
+            }
+        }
+    }
 }

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<Vec<u8>> {
+    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<Vec<u8>> {
+    Err(super::super::CompressionUnavailableError(compress_algo).into())
+}
+
+#[cfg(feature = "lzo")]
+pub(super) fn decompress(data: &[u8], compress_algo: CompressorAlgo) -> anyhow::Result<Vec<u8>> {
+    // 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<Vec<u8>> {
+    Err(super::super::CompressionUnavailableError(compress_algo).into())
+}

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<CompressionAlgoPb> for CompressorAlgo {
     fn try_from(value: CompressionAlgoPb) -> Result<Self, Self::Error> {
         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<CompressorAlgo> for CompressionAlgoPb {
     fn try_from(value: CompressorAlgo) -> Result<Self, Self::Error> {
         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<CompressorAlgo> {
         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() {

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]

easytier-proto/proto/common.proto

@@ -137,6 +137,7 @@ enum CompressionAlgoPb {
   Invalid = 0;
   None = 1;
   Zstd = 2;
+  Lzo = 3;
 }
 
 message RpcCompressionInfo {

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",

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，可以指定多个。"