Merge branch 'master' into feat/cli-login-logout

Author: atharvalade

Cargo.lock

@@ -135,7 +135,6 @@ dirs = "6.0.0"
 dlopen2 = "0.8.2"
 dotenvy = "0.15.7"
 elasticsearch = { version = "9.1.0-alpha.1", features = ["rustls-tls"], default-features = false }
-enum_dispatch = "0.3.13"
 enumset = "1.1"
 env_logger = "0.11.9"
 err_trail = { version = "0.11.0", features = ["tracing"] }

Cargo.toml

@@ -297,7 +297,6 @@ embedded-io: 0.6.1, "Apache-2.0 OR MIT",
 encode_unicode: 1.0.0, "Apache-2.0 OR MIT",
 encoding_rs: 0.8.35, "(Apache-2.0 OR MIT) AND BSD-3-Clause",
 enum-as-inner: 0.6.1, "Apache-2.0 OR MIT",
-enum_dispatch: 0.3.13, "Apache-2.0 OR MIT",
 enumset: 1.1.10, "Apache-2.0 OR MIT",
 enumset_derive: 0.14.0, "Apache-2.0 OR MIT",
 equator: 0.4.2, "MIT",

DEPENDENCIES.md

@@ -48,19 +48,40 @@ impl<'a> RequestFrame<'a> {
     /// Size of the frame header: `[length:4][code:4]`.
     pub const HEADER_SIZE: usize = 8;
 
+    /// Validate a frame length field and return the payload size.
+    ///
+    /// Transport layers that read the length and code fields incrementally
+    /// (e.g. compio completion-based I/O) can use this to validate the length
+    /// before reading the payload, without buffering the entire frame.
+    ///
+    /// # Errors
+    /// Returns `WireError::Validation` if `frame_length < 4` (too small to
+    /// contain even the command code).
+    pub fn payload_length(frame_length: u32) -> Result<u32, WireError> {
+        frame_length
+            .checked_sub(4)
+            .ok_or(WireError::Validation(Cow::Borrowed(
+                "request frame length must be at least 4 (code size)",
+            )))
+    }
+
+    /// Construct a frame from pre-parsed header fields and a payload slice.
+    ///
+    /// Used by transport layers that read the header incrementally (e.g.
+    /// compio completion-based I/O) and then read the payload separately.
+    #[must_use]
+    pub const fn from_parts(code: u32, payload: &'a [u8]) -> Self {
+        Self { code, payload }
+    }
+
     /// Decode a request frame from a complete buffer.
     ///
     /// # Errors
     /// Returns `WireError::UnexpectedEof` if the buffer is too short.
     pub fn decode(buf: &'a [u8]) -> Result<(Self, usize), WireError> {
-        let length = read_u32_le(buf, 0)? as usize;
-        if length < 4 {
-            return Err(WireError::Validation(Cow::Borrowed(
-                "request frame length must be at least 4 (code size)",
-            )));
-        }
+        let frame_length = read_u32_le(buf, 0)?;
+        let payload_len = Self::payload_length(frame_length)? as usize;
         let code = read_u32_le(buf, 4)?;
-        let payload_len = length - 4;
         let payload = read_bytes(buf, Self::HEADER_SIZE, payload_len)?;
         let total = Self::HEADER_SIZE + payload_len;
         Ok((Self { code, payload }, total))
@@ -222,6 +243,23 @@ mod tests {
         assert!(RequestFrame::decode(&buf).is_err());
     }
 
+    #[test]
+    fn payload_length_valid() {
+        assert_eq!(RequestFrame::payload_length(4).unwrap(), 0);
+        assert_eq!(RequestFrame::payload_length(104).unwrap(), 100);
+        assert_eq!(
+            RequestFrame::payload_length(u32::MAX).unwrap(),
+            u32::MAX - 4
+        );
+    }
+
+    #[test]
+    fn payload_length_too_small() {
+        assert!(RequestFrame::payload_length(0).is_err());
+        assert!(RequestFrame::payload_length(1).is_err());
+        assert!(RequestFrame::payload_length(3).is_err());
+    }
+
     #[test]
     fn request_encoded_size() {
         assert_eq!(RequestFrame::encoded_size(0), Some(8));

core/binary_protocol/src/framing.rs

@@ -16,24 +16,57 @@
 // under the License.
 
 use crate::WireError;
-use crate::codec::{WireDecode, WireEncode};
-use bytes::BytesMut;
+use crate::codec::{WireDecode, WireEncode, read_u8};
+use bytes::{BufMut, BytesMut};
 
-/// `GetSnapshot` request. Wire format: empty.
+/// `GetSnapshot` request.
+///
+/// Wire format:
+/// ```text
+/// [compression:1][types_count:1][snapshot_type:1]*
+/// ```
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub struct GetSnapshotRequest;
+pub struct GetSnapshotRequest {
+    /// Compression method code (1=Stored, 2=Deflated, 3=Bzip2, 4=Zstd, 5=Lzma, 6=Xz).
+    pub compression: u8,
+    /// Snapshot type codes (1=FilesystemOverview, 2=ProcessList, 3=ResourceUsage,
+    /// 4=Test, 5=ServerLogs, 6=ServerConfig, 100=All).
+    pub snapshot_types: Vec<u8>,
+}
 
 impl WireEncode for GetSnapshotRequest {
     fn encoded_size(&self) -> usize {
-        0
+        1 + 1 + self.snapshot_types.len()
     }
 
-    fn encode(&self, _buf: &mut BytesMut) {}
+    fn encode(&self, buf: &mut BytesMut) {
+        buf.put_u8(self.compression);
+        let count =
+            u8::try_from(self.snapshot_types.len()).expect("snapshot_types count exceeds u8::MAX");
+        buf.put_u8(count);
+        for &code in &self.snapshot_types {
+            buf.put_u8(code);
+        }
+    }
 }
 
 impl WireDecode for GetSnapshotRequest {
-    fn decode(_buf: &[u8]) -> Result<(Self, usize), WireError> {
-        Ok((Self, 0))
+    fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
+        let compression = read_u8(buf, 0)?;
+        let types_count = read_u8(buf, 1)? as usize;
+        let mut pos = 2;
+        let mut snapshot_types = Vec::with_capacity(types_count);
+        for _ in 0..types_count {
+            snapshot_types.push(read_u8(buf, pos)?);
+            pos += 1;
+        }
+        Ok((
+            Self {
+                compression,
+                snapshot_types,
+            },
+            pos,
+        ))
     }
 }
 
@@ -43,11 +76,54 @@ mod tests {
 
     #[test]
     fn roundtrip() {
-        let req = GetSnapshotRequest;
+        let req = GetSnapshotRequest {
+            compression: 2,
+            snapshot_types: vec![1, 5],
+        };
+        let bytes = req.to_bytes();
+        assert_eq!(bytes.len(), 4); // 1 + 1 + 2
+        let (decoded, consumed) = GetSnapshotRequest::decode(&bytes).unwrap();
+        assert_eq!(consumed, bytes.len());
+        assert_eq!(decoded, req);
+    }
+
+    #[test]
+    fn roundtrip_empty_types() {
+        let req = GetSnapshotRequest {
+            compression: 1,
+            snapshot_types: vec![],
+        };
+        let bytes = req.to_bytes();
+        assert_eq!(bytes.len(), 2);
+        let (decoded, consumed) = GetSnapshotRequest::decode(&bytes).unwrap();
+        assert_eq!(consumed, 2);
+        assert_eq!(decoded, req);
+    }
+
+    #[test]
+    fn roundtrip_all_type() {
+        let req = GetSnapshotRequest {
+            compression: 2,
+            snapshot_types: vec![100],
+        };
         let bytes = req.to_bytes();
-        assert!(bytes.is_empty());
         let (decoded, consumed) = GetSnapshotRequest::decode(&bytes).unwrap();
-        assert_eq!(consumed, 0);
+        assert_eq!(consumed, bytes.len());
         assert_eq!(decoded, req);
     }
+
+    #[test]
+    fn truncated_returns_error() {
+        let req = GetSnapshotRequest {
+            compression: 2,
+            snapshot_types: vec![1, 2, 3],
+        };
+        let bytes = req.to_bytes();
+        for i in 0..bytes.len() {
+            assert!(
+                GetSnapshotRequest::decode(&bytes[..i]).is_err(),
+                "expected error for truncation at byte {i}"
+            );
+        }
+    }
 }

core/binary_protocol/src/requests/system/get_snapshot.rs

@@ -42,8 +42,10 @@ impl WireEncode for CreateUserRequest {
             + self.password.len()
             + 1 // status
             + 1 // has_permissions
-            + 4 // permissions_len (always present)
-            + self.permissions.as_ref().map_or(0, WireEncode::encoded_size)
+            + self
+                .permissions
+                .as_ref()
+                .map_or(0, |p| 4 + p.encoded_size())
     }
 
     fn encode(&self, buf: &mut BytesMut) {
@@ -60,7 +62,6 @@ impl WireEncode for CreateUserRequest {
             buf.put_slice(&perm_bytes);
         } else {
             buf.put_u8(0);
-            buf.put_u32_le(0);
         }
     }
 }
@@ -80,10 +81,9 @@ impl WireDecode for CreateUserRequest {
         let has_permissions = read_u8(buf, pos)?;
         pos += 1;
 
-        let perm_len = read_u32_le(buf, pos)? as usize;
-        pos += 4;
-
-        let permissions = if has_permissions == 1 && perm_len > 0 {
+        let permissions = if has_permissions == 1 {
+            let perm_len = read_u32_le(buf, pos)? as usize;
+            pos += 4;
             let (perms, consumed) = WirePermissions::decode(&buf[pos..])?;
             if consumed != perm_len {
                 return Err(WireError::Validation(Cow::Owned(format!(
@@ -197,8 +197,8 @@ mod tests {
         };
         let bytes = req.to_bytes();
         // username: [1, b'u'] + password: [1, b'p'] + status: [0]
-        // + has_perm: [0] + perm_len: [0,0,0,0]
-        let expected: &[u8] = &[1, b'u', 1, b'p', 0, 0, 0, 0, 0, 0];
+        // + has_perm: [0]
+        let expected: &[u8] = &[1, b'u', 1, b'p', 0, 0];
         assert_eq!(&bytes[..], expected);
     }
 }

core/binary_protocol/src/requests/users/create_user.rs

@@ -36,8 +36,10 @@ impl WireEncode for UpdatePermissionsRequest {
     fn encoded_size(&self) -> usize {
         self.user_id.encoded_size()
             + 1 // has_permissions
-            + 4 // permissions_len (always present)
-            + self.permissions.as_ref().map_or(0, WireEncode::encoded_size)
+            + self
+                .permissions
+                .as_ref()
+                .map_or(0, |p| 4 + p.encoded_size())
     }
 
     fn encode(&self, buf: &mut BytesMut) {
@@ -50,7 +52,6 @@ impl WireEncode for UpdatePermissionsRequest {
             buf.put_slice(&perm_bytes);
         } else {
             buf.put_u8(0);
-            buf.put_u32_le(0);
         }
     }
 }
@@ -62,10 +63,9 @@ impl WireDecode for UpdatePermissionsRequest {
         let has_permissions = read_u8(buf, pos)?;
         pos += 1;
 
-        let perm_len = read_u32_le(buf, pos)? as usize;
-        pos += 4;
-
-        let permissions = if has_permissions == 1 && perm_len > 0 {
+        let permissions = if has_permissions == 1 {
+            let perm_len = read_u32_le(buf, pos)? as usize;
+            pos += 4;
             let (perms, consumed) = WirePermissions::decode(&buf[pos..])?;
             if consumed != perm_len {
                 return Err(WireError::Validation(Cow::Owned(format!(

core/binary_protocol/src/requests/users/update_permissions.rs

@@ -42,6 +42,7 @@ harness_derive = { workspace = true }
 humantime = { workspace = true }
 iggy = { workspace = true }
 iggy-cli = { workspace = true }
+iggy_binary_protocol = { workspace = true }
 iggy_common = { workspace = true }
 iggy_connector_sdk = { workspace = true, features = ["api"] }
 keyring = { workspace = true }
@@ -68,7 +69,6 @@ serial_test = { workspace = true }
 server = { workspace = true }
 socket2 = { workspace = true }
 sqlx = { workspace = true }
-strum = { workspace = true }
 sysinfo = { workspace = true }
 tempfile = { workspace = true }
 test-case = { workspace = true }