Address review comments

Cargo.lock

@@ -2288,6 +2288,7 @@ dependencies = [
  "indoc",
  "itertools 0.14.0",
  "kcp-sys",
+ "lzokay-native",
  "machine-uid",
  "maplit",
  "mimalloc",
@@ -4874,6 +4875,16 @@ version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
 
+[[package]]
+name = "lzokay-native"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "792ba667add2798c6c3e988e630f4eb921b5cbc735044825b7111ef1582c8730"
+dependencies = [
+ "byteorder",
+ "thiserror 1.0.63",
+]
+
 [[package]]
 name = "mac"
 version = "0.1.1"

easytier/Cargo.toml

@@ -221,6 +221,7 @@ async-ringbuf = "0.3.1"
 service-manager = { git = "https://github.com/EasyTier/service-manager-rs.git", branch = "main" }
 
 zstd = { version = "0.13", optional = true }
+lzokay-native = { version = "0.1", optional = true }
 
 kcp-sys = { git = "https://github.com/EasyTier/kcp-sys", rev = "94964794caaed5d388463137da59b97499619e5f", optional = true }
 
@@ -358,6 +359,7 @@ default = [
     "faketcp",
     "magic-dns",
     "zstd",
+    "lzo",
 ]
 full = [
     "websocket",
@@ -372,6 +374,7 @@ full = [
     "faketcp",
     "magic-dns",
     "zstd",
+    "lzo",
 ]
 wireguard = ["dep:boringtun", "dep:ring"]
 quic = ["dep:quinn", "dep:quinn-plaintext", "dep:rustls", "dep:rcgen"]
@@ -402,5 +405,6 @@ tracing = ["tokio/tracing", "dep:console-subscriber"]
 magic-dns = ["dep:hickory-client", "dep:hickory-server"]
 faketcp = ["dep:flume"]
 zstd = ["dep:zstd"]
+lzo = ["dep:lzokay-native"]
 # For Network Extension on macOS
 macos-ne = []

easytier/build.rs

@@ -191,6 +191,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         )
         .type_attribute("peer_rpc.RouteForeignNetworkSummary", "#[derive(Hash, Eq)]")
         .type_attribute("common.RpcDescriptor", "#[derive(Hash, Eq)]")
+        .type_attribute("acl.Acl", "#[serde(default)]")
+        .type_attribute("acl.AclV1", "#[serde(default)]")
+        .type_attribute("acl.Chain", "#[serde(default)]")
+        .type_attribute("acl.Rule", "#[serde(default)]")
+        .type_attribute("acl.GroupInfo", "#[serde(default)]")
         .field_attribute(".api.manage.NetworkConfig", "#[serde(default)]")
         .service_generator(Box::new(easytier_rpc_build::ServiceGenerator::default()))
         .btree_map(["."])

easytier/locales/app.yml

@@ -194,8 +194,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, supported: %{algorithms}. default is none"
+    zh-CN: "要使用的压缩算法，支持：%{algorithms}。默认为 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，可以指定多个。"

easytier/src/common/acl_processor.rs

@@ -39,7 +39,6 @@ pub struct RateLimitValue {
 pub enum RuleId {
     Priority(u32),
     Stateful(u32),
-    StatefulReverse,
     Default,
 }
 
@@ -49,7 +48,6 @@ impl RuleId {
         match self {
             RuleId::Priority(p) => p.to_string(),
             RuleId::Stateful(p) => format!("stateful-{}", p),
-            RuleId::StatefulReverse => "stateful-reverse".to_string(),
             RuleId::Default => "default".to_string(),
         }
     }
@@ -484,30 +482,20 @@ impl AclProcessor {
         stats
     }
 
-    /// Process a packet through ACL rules.
+    /// Process a packet through ACL rules - Now lock-free!
     pub fn process_packet(&self, packet_info: &PacketInfo, chain_type: ChainType) -> AclResult {
         // Check cache first for performance
         let cache_key = AclCacheKey::from_packet_info(packet_info, chain_type);
 
-        // If cache hit and can skip checks, return cached result. Cached drops may be
-        // overridden by a stateful reverse connection that was created after caching.
+        // If cache hit and can skip checks, return cached result
         if let Some(mut cached) = self.rule_cache.get_mut(&cache_key) {
             // Update last access time for LRU
             cached.last_access = Instant::now();
 
             self.increment_stat(AclStatKey::CacheHits);
-            if cached.acl_result.as_ref().map(|r| r.action) == Some(Action::Drop)
-                && let Some(result) = self.check_reverse_connection(packet_info)
-            {
-                return result;
-            }
             return self.process_packet_with_cache_entry(packet_info, &cached);
         }
 
-        if let Some(result) = self.check_reverse_connection(packet_info) {
-            return result;
-        }
-
         // Direct access to rules - no locks needed!
         let rules = match chain_type {
             ChainType::Inbound => &self.inbound_rules,
@@ -742,68 +730,28 @@ impl AclProcessor {
     }
 
     fn conn_track_key(&self, packet_info: &PacketInfo) -> String {
-        Self::make_conn_track_key(
-            packet_info.src_ip,
-            packet_info.src_port,
-            packet_info.dst_ip,
-            packet_info.dst_port,
-        )
-    }
-
-    fn reverse_conn_track_key(&self, packet_info: &PacketInfo) -> String {
-        Self::make_conn_track_key(
-            packet_info.dst_ip,
-            packet_info.dst_port,
-            packet_info.src_ip,
-            packet_info.src_port,
-        )
-    }
-
-    fn make_conn_track_key(
-        src_ip: IpAddr,
-        src_port: Option<u16>,
-        dst_ip: IpAddr,
-        dst_port: Option<u16>,
-    ) -> String {
         format!(
             "{}:{}->{}:{}",
-            src_ip,
-            src_port.unwrap_or(0),
-            dst_ip,
-            dst_port.unwrap_or(0)
+            packet_info.src_ip,
+            packet_info.src_port.unwrap_or(0),
+            packet_info.dst_ip,
+            packet_info.dst_port.unwrap_or(0)
         )
     }
 
-    fn check_reverse_connection(&self, packet_info: &PacketInfo) -> Option<AclResult> {
-        let reverse_key = self.reverse_conn_track_key(packet_info);
-        let mut entry = self.conn_track.get_mut(&reverse_key)?;
-        Self::update_conn_track_entry(entry.value_mut(), packet_info);
-        Some(AclResult {
-            action: Action::Allow,
-            matched_rule: Some(RuleId::StatefulReverse),
-            should_log: false,
-            log_context: Some(AclLogContext::StatefulMatch {
-                src_ip: packet_info.src_ip,
-                dst_ip: packet_info.dst_ip,
-            }),
-        })
-    }
-
-    fn update_conn_track_entry(entry: &mut ConnTrackEntry, packet_info: &PacketInfo) {
-        entry.last_seen = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_secs();
-        entry.packet_count += 1;
-        entry.byte_count += packet_info.packet_size as u64;
-        entry.state = ConnState::Established as i32;
-    }
-
     /// Check connection state for stateful rules
     fn check_connection_state(&self, conn_track_key: &str, packet_info: &PacketInfo) {
         self.conn_track
             .entry(conn_track_key.to_string())
-            .and_modify(|x| Self::update_conn_track_entry(x, packet_info))
+            .and_modify(|x| {
+                x.last_seen = SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap()
+                    .as_secs();
+                x.packet_count += 1;
+                x.byte_count += packet_info.packet_size as u64;
+                x.state = ConnState::Established as i32;
+            })
             .or_insert_with(|| ConnTrackEntry {
                 src_addr: Some(
                     SocketAddr::new(packet_info.src_ip, packet_info.src_port.unwrap_or(0)).into(),
@@ -1391,6 +1339,45 @@ mod tests {
         assert_eq!(result.matched_rule, Some(RuleId::Priority(70)));
     }
 
+    #[tokio::test]
+    async fn test_forward_acl_source_ip_whitelist() {
+        let mut acl_config = Acl::default();
+        let mut acl_v1 = AclV1::default();
+        let mut chain = Chain {
+            name: "subnet_proxy_protect".to_string(),
+            chain_type: ChainType::Forward as i32,
+            enabled: true,
+            default_action: Action::Drop as i32,
+            ..Default::default()
+        };
+
+        chain.rules.push(Rule {
+            name: "allow_my_devices".to_string(),
+            priority: 1000,
+            enabled: true,
+            action: Action::Allow as i32,
+            protocol: Protocol::Any as i32,
+            source_ips: vec!["10.172.192.2/32".to_string()],
+            ..Default::default()
+        });
+        acl_v1.chains.push(chain);
+        acl_config.acl_v1 = Some(acl_v1);
+
+        let processor = AclProcessor::new(acl_config);
+        let mut packet_info = create_test_packet_info();
+        packet_info.dst_ip = "192.168.1.10".parse().unwrap();
+
+        packet_info.src_ip = "10.172.192.2".parse().unwrap();
+        let result = processor.process_packet(&packet_info, ChainType::Forward);
+        assert_eq!(result.action, Action::Allow);
+        assert_eq!(result.matched_rule, Some(RuleId::Priority(1000)));
+
+        packet_info.src_ip = "10.172.192.3".parse().unwrap();
+        let result = processor.process_packet(&packet_info, ChainType::Forward);
+        assert_eq!(result.action, Action::Drop);
+        assert_eq!(result.matched_rule, Some(RuleId::Default));
+    }
+
     fn create_test_acl_config() -> Acl {
         let mut acl_config = Acl::default();
 
@@ -1434,73 +1421,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_stateful_allows_reverse_traffic_before_default_drop() {
-        let runtime = tokio::runtime::Runtime::new().unwrap();
-        let _runtime_guard = runtime.enter();
-
-        let mut acl_config = Acl::default();
-        let mut acl_v1 = AclV1::default();
-
-        let mut outbound_chain = Chain {
-            name: "outbound_stateful".to_string(),
-            chain_type: ChainType::Outbound as i32,
-            enabled: true,
-            default_action: Action::Drop as i32,
-            ..Default::default()
-        };
-        outbound_chain.rules.push(Rule {
-            name: "allow_out_stateful".to_string(),
-            priority: 100,
-            enabled: true,
-            action: Action::Allow as i32,
-            protocol: Protocol::Tcp as i32,
-            stateful: true,
-            ..Default::default()
-        });
-
-        let inbound_chain = Chain {
-            name: "inbound_default_drop".to_string(),
-            chain_type: ChainType::Inbound as i32,
-            enabled: true,
-            default_action: Action::Drop as i32,
-            ..Default::default()
-        };
-
-        acl_v1.chains.push(outbound_chain);
-        acl_v1.chains.push(inbound_chain);
-        acl_config.acl_v1 = Some(acl_v1);
-
-        let processor = AclProcessor::new(acl_config);
-        let outbound_packet = PacketInfo {
-            src_ip: IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)),
-            dst_ip: IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2)),
-            src_port: Some(12345),
-            dst_port: Some(22),
-            protocol: Protocol::Tcp,
-            packet_size: 64,
-            src_groups: Arc::new(vec![]),
-            dst_groups: Arc::new(vec![]),
-        };
-        let inbound_reply = PacketInfo {
-            src_ip: outbound_packet.dst_ip,
-            dst_ip: outbound_packet.src_ip,
-            src_port: outbound_packet.dst_port,
-            dst_port: outbound_packet.src_port,
-            protocol: outbound_packet.protocol,
-            packet_size: 64,
-            src_groups: Arc::new(vec![]),
-            dst_groups: Arc::new(vec![]),
-        };
-
-        let outbound_result = processor.process_packet(&outbound_packet, ChainType::Outbound);
-        assert_eq!(outbound_result.action, Action::Allow);
-
-        let inbound_result = processor.process_packet(&inbound_reply, ChainType::Inbound);
-        assert_eq!(inbound_result.action, Action::Allow);
-        assert_eq!(inbound_result.matched_rule, Some(RuleId::StatefulReverse));
-    }
-
     #[test]
     fn test_acl_cache_key_creation() {
         let packet_info = create_test_packet_info();

easytier/src/common/compressor.rs

@@ -1,4 +1,4 @@
-#[cfg(feature = "zstd")]
+#[cfg(any(feature = "zstd", feature = "lzo"))]
 use anyhow::Context;
 #[cfg(feature = "zstd")]
 use dashmap::DashMap;
@@ -53,6 +53,13 @@ impl DefaultCompressor {
                     )
                 })
             }),
+            #[cfg(feature = "lzo")]
+            CompressorAlgo::Lzo => lzokay_native::compress(data).with_context(|| {
+                format!(
+                    "Failed to compress data with algorithm: {:?}",
+                    compress_algo
+                )
+            }),
             CompressorAlgo::None => Ok(data.to_vec()),
         }
     }
@@ -85,6 +92,13 @@ impl DefaultCompressor {
                     compress_algo
                 ))
             }),
+            #[cfg(feature = "lzo")]
+            CompressorAlgo::Lzo => lzokay_native::decompress_all(data, None).with_context(|| {
+                format!(
+                    "Failed to decompress data with algorithm: {:?}",
+                    compress_algo
+                )
+            }),
             CompressorAlgo::None => Ok(data.to_vec()),
         }
     }
@@ -181,14 +195,13 @@ thread_local! {
     static DCTX_MAP: RefCell<DashMap<CompressorAlgo, bulk::Decompressor<'static>>> = RefCell::new(DashMap::new());
 }
 
-#[cfg(all(test, feature = "zstd"))]
+#[cfg(all(test, any(feature = "zstd", feature = "lzo")))]
 pub mod tests {
     use super::*;
 
-    #[tokio::test]
-    async fn test_compress() {
-        let text = b"12345670000000000000000000";
-        let mut packet = ZCPacket::new_with_payload(text);
+    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 {};
@@ -200,7 +213,7 @@ pub mod tests {
         );
 
         compressor
-            .compress(&mut packet, CompressorAlgo::ZstdDefault)
+            .compress(&mut packet, compress_algo)
             .await
             .unwrap();
         println!(
@@ -215,8 +228,7 @@ pub mod tests {
         assert!(!packet.peer_manager_header().unwrap().is_compressed());
     }
 
-    #[tokio::test]
-    async fn test_short_text_compress() {
+    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);
@@ -225,7 +237,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());
@@ -234,4 +246,28 @@ pub mod tests {
         assert_eq!(packet.payload(), text);
         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;
+    }
 }

easytier/src/common/config.rs

@@ -1337,6 +1337,71 @@ stun_servers = [
         assert!(err.to_string().contains("mapped listener port is missing"));
     }
 
+    #[test]
+    fn test_acl_toml_rule_uses_defaults_for_omitted_fields() {
+        use crate::proto::acl::{Action, ChainType, Protocol};
+
+        let config_str = r#"
+[[acl.acl_v1.chains]]
+name = "subnet_proxy_protect"
+chain_type = 3
+enabled = true
+default_action = 2
+
+[[acl.acl_v1.chains.rules]]
+name = "allow_my_devices"
+priority = 1000
+action = 1
+source_ips = ["10.172.192.2/32"]
+protocol = 5
+enabled = true
+"#;
+
+        let config = TomlConfigLoader::new_from_str(config_str).unwrap();
+        let acl = config.get_acl().unwrap();
+        let acl_v1 = acl.acl_v1.unwrap();
+        let chain = &acl_v1.chains[0];
+        let rule = &chain.rules[0];
+
+        assert_eq!(chain.chain_type, ChainType::Forward as i32);
+        assert_eq!(chain.default_action, Action::Drop as i32);
+        assert_eq!(rule.action, Action::Allow as i32);
+        assert_eq!(rule.protocol, Protocol::Any as i32);
+        assert_eq!(rule.source_ips, vec!["10.172.192.2/32"]);
+        assert!(rule.ports.is_empty());
+        assert!(rule.source_ports.is_empty());
+        assert!(rule.destination_ips.is_empty());
+        assert!(rule.source_groups.is_empty());
+        assert!(rule.destination_groups.is_empty());
+        assert_eq!(rule.rate_limit, 0);
+        assert_eq!(rule.burst_limit, 0);
+        assert!(!rule.stateful);
+    }
+
+    #[test]
+    fn test_acl_toml_group_can_omit_declares_or_members() {
+        let declares_only = r#"
+[acl.acl_v1.group]
+
+[[acl.acl_v1.group.declares]]
+group_name = "admin"
+group_secret = "admin-pw"
+"#;
+        let config = TomlConfigLoader::new_from_str(declares_only).unwrap();
+        let group = config.get_acl().unwrap().acl_v1.unwrap().group.unwrap();
+        assert_eq!(group.declares.len(), 1);
+        assert!(group.members.is_empty());
+
+        let members_only = r#"
+[acl.acl_v1.group]
+members = ["admin"]
+"#;
+        let config = TomlConfigLoader::new_from_str(members_only).unwrap();
+        let group = config.get_acl().unwrap().acl_v1.unwrap().group.unwrap();
+        assert!(group.declares.is_empty());
+        assert_eq!(group.members, vec!["admin"]);
+    }
+
     #[test]
     fn test_network_config_source_user_is_implicit() {
         let config = TomlConfigLoader::default();

easytier/src/core.rs

@@ -37,6 +37,38 @@ use crate::tunnel::IpScheme;
 #[cfg(feature = "jemalloc-prof")]
 use jemalloc_ctl::{Access as _, AsName as _, epoch, stats};
 
+fn supported_compression_algorithms() -> &'static str {
+    cfg_select! {
+        all(feature = "zstd", feature = "lzo") => "none, zstd, lzo",
+        feature = "zstd" => "none, zstd",
+        feature = "lzo" => "none, lzo",
+        _ => "none",
+    }
+}
+
+fn compression_help() -> String {
+    t!(
+        "core_clap.compression",
+        algorithms = supported_compression_algorithms()
+    )
+    .to_string()
+}
+
+fn parse_compression_algorithm(compression: &str) -> anyhow::Result<CompressionAlgoPb> {
+    match compression {
+        "none" => Ok(CompressionAlgoPb::None),
+        #[cfg(feature = "zstd")]
+        "zstd" => Ok(CompressionAlgoPb::Zstd),
+        #[cfg(feature = "lzo")]
+        "lzo" => Ok(CompressionAlgoPb::Lzo),
+        _ => anyhow::bail!(
+            "unknown compression algorithm: {}, supported: {}",
+            compression,
+            supported_compression_algorithms()
+        ),
+    }
+}
+
 #[cfg(target_os = "windows")]
 windows_service::define_windows_service!(ffi_service_main, win_service_main);
 
@@ -513,7 +545,7 @@ struct NetworkOptions {
     #[arg(
         long,
         env = "ET_COMPRESSION",
-        help = t!("core_clap.compression").to_string(),
+        help = compression_help(),
     )]
     compression: Option<String>,
 
@@ -1106,15 +1138,7 @@ impl NetworkOptions {
         f.need_p2p = self.need_p2p.unwrap_or(f.need_p2p);
         f.multi_thread = self.multi_thread.unwrap_or(f.multi_thread);
         if let Some(compression) = &self.compression {
-            f.data_compress_algo = match compression.as_str() {
-                "none" => CompressionAlgoPb::None,
-                "zstd" => CompressionAlgoPb::Zstd,
-                _ => panic!(
-                    "unknown compression algorithm: {}, supported: none, zstd",
-                    compression
-                ),
-            }
-            .into();
+            f.data_compress_algo = parse_compression_algorithm(compression)?.into();
         }
         f.bind_device = self.bind_device.unwrap_or(f.bind_device);
         f.enable_kcp_proxy = self.enable_kcp_proxy.unwrap_or(f.enable_kcp_proxy);
@@ -1627,6 +1651,21 @@ async fn validate_config(cli: &Cli) -> anyhow::Result<()> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn test_compression_help_uses_supported_algorithms() {
+        assert!(compression_help().contains(supported_compression_algorithms()));
+    }
+
+    #[test]
+    fn test_parse_compression_algorithm_rejects_unknown() {
+        let err = parse_compression_algorithm("snappy")
+            .unwrap_err()
+            .to_string();
+
+        assert!(err.contains("snappy"));
+        assert!(err.contains(supported_compression_algorithms()));
+    }
+
     #[test]
     fn test_parse_listeners() {
         type IpSchemeMap = fn(&IpScheme) -> String;

easytier/src/instance/listeners.rs

@@ -25,7 +25,7 @@ use crate::{
 
 pub fn create_listener_by_url(
     l: &url::Url,
-    global_ctx: ArcGlobalCtx,
+    _global_ctx: ArcGlobalCtx,
 ) -> Result<Box<dyn TunnelListener>, Error> {
     Ok(match l.try_into()? {
         TunnelScheme::Ip(scheme) => match scheme {
@@ -34,7 +34,7 @@ pub fn create_listener_by_url(
             #[cfg(feature = "wireguard")]
             IpScheme::Wg => {
                 use crate::tunnel::wireguard::{WgConfig, WgTunnelListener};
-                let nid = global_ctx.get_network_identity();
+                let nid = _global_ctx.get_network_identity();
                 let wg_config = WgConfig::new_from_network_identity(
                     &nid.network_name,
                     &nid.network_secret.unwrap_or_default(),
@@ -43,7 +43,7 @@ pub fn create_listener_by_url(
             }
             #[cfg(feature = "quic")]
             IpScheme::Quic => {
-                tunnel::quic::QuicTunnelListener::new(l.clone(), global_ctx.clone()).boxed()
+                tunnel::quic::QuicTunnelListener::new(l.clone(), _global_ctx.clone()).boxed()
             }
             #[cfg(feature = "websocket")]
             IpScheme::Ws | IpScheme::Wss => {

easytier/src/proto/common.proto

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

easytier/src/proto/common.rs

@@ -467,6 +467,8 @@ impl TryFrom<CompressionAlgoPb> for CompressorAlgo {
         match value {
             #[cfg(feature = "zstd")]
             CompressionAlgoPb::Zstd => Ok(CompressorAlgo::ZstdDefault),
+            #[cfg(feature = "lzo")]
+            CompressionAlgoPb::Lzo => Ok(CompressorAlgo::Lzo),
             CompressionAlgoPb::None => Ok(CompressorAlgo::None),
             _ => Err(anyhow::anyhow!("Invalid CompressionAlgoPb")),
         }
@@ -480,6 +482,8 @@ impl TryFrom<CompressorAlgo> for CompressionAlgoPb {
         match value {
             #[cfg(feature = "zstd")]
             CompressorAlgo::ZstdDefault => Ok(CompressionAlgoPb::Zstd),
+            #[cfg(feature = "lzo")]
+            CompressorAlgo::Lzo => Ok(CompressionAlgoPb::Lzo),
             CompressorAlgo::None => Ok(CompressionAlgoPb::None),
         }
     }

easytier/src/tunnel/packet_def.rs

@@ -309,6 +309,8 @@ pub enum CompressorAlgo {
     None = 0,
     #[cfg(feature = "zstd")]
     ZstdDefault = 1,
+    #[cfg(feature = "lzo")]
+    Lzo = 2,
 }
 
 #[repr(C, packed)]
@@ -323,6 +325,8 @@ impl CompressorTail {
         match self.algo {
             #[cfg(feature = "zstd")]
             1 => Some(CompressorAlgo::ZstdDefault),
+            #[cfg(feature = "lzo")]
+            2 => Some(CompressorAlgo::Lzo),
             _ => None,
         }
     }