feat(sdk): implement vsr header framing

core/common/Cargo.toml

@@ -27,6 +27,9 @@ documentation = "https://iggy.apache.org/docs"
 repository = "https://github.com/apache/iggy"
 readme = "README.md"
 
+[features]
+vsr = []
+
 [dependencies]
 aes-gcm = { workspace = true }
 aligned-vec = { workspace = true }

core/common/src/traits/binary_impls/personal_access_tokens.rs

@@ -25,17 +25,30 @@ use crate::{
 };
 use iggy_binary_protocol::WireName;
 use iggy_binary_protocol::codec::WireEncode;
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::codes::LOGIN_REGISTER_WITH_PAT_CODE;
+#[cfg(not(feature = "vsr"))]
+use iggy_binary_protocol::codes::LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE;
 use iggy_binary_protocol::codes::{
     CREATE_PERSONAL_ACCESS_TOKEN_CODE, DELETE_PERSONAL_ACCESS_TOKEN_CODE,
-    GET_PERSONAL_ACCESS_TOKENS_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE,
+    GET_PERSONAL_ACCESS_TOKENS_CODE,
 };
+#[cfg(not(feature = "vsr"))]
+use iggy_binary_protocol::requests::personal_access_tokens::LoginWithPersonalAccessTokenRequest;
 use iggy_binary_protocol::requests::personal_access_tokens::{
     CreatePersonalAccessTokenRequest, DeletePersonalAccessTokenRequest,
-    GetPersonalAccessTokensRequest, LoginWithPersonalAccessTokenRequest,
+    GetPersonalAccessTokensRequest,
 };
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::requests::users::LoginRegisterWithPatRequest;
 use iggy_binary_protocol::responses::personal_access_tokens::create_personal_access_token::RawPersonalAccessTokenResponse;
 use iggy_binary_protocol::responses::personal_access_tokens::get_personal_access_tokens::GetPersonalAccessTokensResponse;
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::responses::users::LoginRegisterResponse;
+#[cfg(not(feature = "vsr"))]
 use iggy_binary_protocol::responses::users::login_user::IdentityResponse;
+#[cfg(feature = "vsr")]
+use secrecy::SecretString;
 
 #[async_trait::async_trait]
 impl<B: BinaryClient> PersonalAccessTokenClient for B {
@@ -90,16 +103,63 @@ impl<B: BinaryClient> PersonalAccessTokenClient for B {
         &self,
         token: &str,
     ) -> Result<IdentityInfo, IggyError> {
+        #[cfg(feature = "vsr")]
+        {
+            let client_id = self.get_vsr_client_id().await?;
+            let response = match self
+                .send_raw_with_response(
+                    LOGIN_REGISTER_WITH_PAT_CODE,
+                    LoginRegisterWithPatRequest {
+                        client_id,
+                        token: SecretString::from(token.to_string()),
+                        version: Some(env!("CARGO_PKG_VERSION").to_string()),
+                        client_context: Some(String::new()),
+                    }
+                    .to_bytes(),
+                )
+                .await
+            {
+                Ok(response) => response,
+                Err(error) => {
+                    self.reset_vsr_session().await?;
+                    return Err(error);
+                }
+            };
+            let wire_resp = match super::decode_response::<LoginRegisterResponse>(&response) {
+                Ok(wire_resp) => wire_resp,
+                Err(error) => {
+                    self.reset_vsr_session().await?;
+                    return Err(error);
+                }
+            };
+            if let Err(error) = self.bind_vsr_session(wire_resp.session).await {
+                self.reset_vsr_session().await?;
+                return Err(error);
+            }
+            self.set_state(ClientState::Authenticated).await;
+            self.publish_event(DiagnosticEvent::SignedIn).await;
+            return Ok(IdentityInfo {
+                user_id: wire_resp.user_id,
+                access_token: None,
+            });
+        }
+
+        #[cfg(not(feature = "vsr"))]
         let wire_token = WireName::new(token).map_err(|_| IggyError::InvalidFormat)?;
+        #[cfg(not(feature = "vsr"))]
         let response = self
             .send_raw_with_response(
                 LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE,
                 LoginWithPersonalAccessTokenRequest { token: wire_token }.to_bytes(),
             )
             .await?;
+        #[cfg(not(feature = "vsr"))]
         self.set_state(ClientState::Authenticated).await;
+        #[cfg(not(feature = "vsr"))]
         self.publish_event(DiagnosticEvent::SignedIn).await;
+        #[cfg(not(feature = "vsr"))]
         let wire_resp = super::decode_response::<IdentityResponse>(&response)?;
+        #[cfg(not(feature = "vsr"))]
         Ok(IdentityInfo::from(wire_resp))
     }
 }

core/common/src/traits/binary_impls/users.rs

@@ -24,16 +24,29 @@ use crate::{
 };
 use iggy_binary_protocol::WireName;
 use iggy_binary_protocol::codec::WireEncode;
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::codes::LOGIN_REGISTER_CODE;
 use iggy_binary_protocol::codes::{
     CHANGE_PASSWORD_CODE, CREATE_USER_CODE, DELETE_USER_CODE, GET_USER_CODE, GET_USERS_CODE,
-    LOGIN_USER_CODE, LOGOUT_USER_CODE, UPDATE_PERMISSIONS_CODE, UPDATE_USER_CODE,
+    UPDATE_PERMISSIONS_CODE, UPDATE_USER_CODE,
 };
+#[cfg(not(feature = "vsr"))]
+use iggy_binary_protocol::codes::{LOGIN_USER_CODE, LOGOUT_USER_CODE};
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::requests::users::LoginRegisterRequest;
 use iggy_binary_protocol::requests::users::{
     ChangePasswordRequest, CreateUserRequest, DeleteUserRequest, GetUserRequest, GetUsersRequest,
-    LoginUserRequest, LogoutUserRequest, UpdatePermissionsRequest, UpdateUserRequest,
+    UpdatePermissionsRequest, UpdateUserRequest,
 };
+#[cfg(not(feature = "vsr"))]
+use iggy_binary_protocol::requests::users::{LoginUserRequest, LogoutUserRequest};
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::responses::users::LoginRegisterResponse;
+#[cfg(not(feature = "vsr"))]
 use iggy_binary_protocol::responses::users::login_user::IdentityResponse;
 use iggy_binary_protocol::responses::users::{GetUsersResponse, UserDetailsResponse};
+#[cfg(feature = "vsr")]
+use secrecy::SecretString;
 
 #[async_trait::async_trait]
 impl<B: BinaryClient> UserClient for B {
@@ -169,7 +182,52 @@ impl<B: BinaryClient> UserClient for B {
     }
 
     async fn login_user(&self, username: &str, password: &str) -> Result<IdentityInfo, IggyError> {
+        #[cfg(feature = "vsr")]
+        {
+            let wire_name = WireName::new(username).map_err(|_| IggyError::InvalidFormat)?;
+            let client_id = self.get_vsr_client_id().await?;
+            let response = match self
+                .send_raw_with_response(
+                    LOGIN_REGISTER_CODE,
+                    LoginRegisterRequest {
+                        client_id,
+                        username: wire_name,
+                        password: SecretString::from(password.to_string()),
+                        version: Some(env!("CARGO_PKG_VERSION").to_string()),
+                        client_context: Some(String::new()),
+                    }
+                    .to_bytes(),
+                )
+                .await
+            {
+                Ok(response) => response,
+                Err(error) => {
+                    self.reset_vsr_session().await?;
+                    return Err(error);
+                }
+            };
+            let wire_resp = match super::decode_response::<LoginRegisterResponse>(&response) {
+                Ok(wire_resp) => wire_resp,
+                Err(error) => {
+                    self.reset_vsr_session().await?;
+                    return Err(error);
+                }
+            };
+            if let Err(error) = self.bind_vsr_session(wire_resp.session).await {
+                self.reset_vsr_session().await?;
+                return Err(error);
+            }
+            self.set_state(ClientState::Authenticated).await;
+            self.publish_event(DiagnosticEvent::SignedIn).await;
+            return Ok(IdentityInfo {
+                user_id: wire_resp.user_id,
+                access_token: None,
+            });
+        }
+
+        #[cfg(not(feature = "vsr"))]
         let wire_name = WireName::new(username).map_err(|_| IggyError::InvalidFormat)?;
+        #[cfg(not(feature = "vsr"))]
         let response = self
             .send_raw_with_response(
                 LOGIN_USER_CODE,
@@ -182,18 +240,32 @@ impl<B: BinaryClient> UserClient for B {
                 .to_bytes(),
             )
             .await?;
+        #[cfg(not(feature = "vsr"))]
         self.set_state(ClientState::Authenticated).await;
+        #[cfg(not(feature = "vsr"))]
         self.publish_event(DiagnosticEvent::SignedIn).await;
+        #[cfg(not(feature = "vsr"))]
         let wire_resp = super::decode_response::<IdentityResponse>(&response)?;
+        #[cfg(not(feature = "vsr"))]
         Ok(IdentityInfo::from(wire_resp))
     }
 
     async fn logout_user(&self) -> Result<(), IggyError> {
+        #[cfg(feature = "vsr")]
+        {
+            return Err(IggyError::FeatureUnavailable);
+        }
+
+        #[cfg(not(feature = "vsr"))]
         fail_if_not_authenticated(self).await?;
+        #[cfg(not(feature = "vsr"))]
         self.send_raw_with_response(LOGOUT_USER_CODE, LogoutUserRequest.to_bytes())
             .await?;
+        #[cfg(not(feature = "vsr"))]
         self.set_state(ClientState::Connected).await;
+        #[cfg(not(feature = "vsr"))]
         self.publish_event(DiagnosticEvent::SignedOut).await;
+        #[cfg(not(feature = "vsr"))]
         Ok(())
     }
 }

core/common/src/traits/binary_transport.rs

@@ -29,4 +29,19 @@ pub trait BinaryTransport {
     async fn publish_event(&self, event: DiagnosticEvent);
     async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError>;
     fn get_heartbeat_interval(&self) -> IggyDuration;
+
+    #[cfg(feature = "vsr")]
+    async fn get_vsr_client_id(&self) -> Result<u128, IggyError> {
+        Err(IggyError::FeatureUnavailable)
+    }
+
+    #[cfg(feature = "vsr")]
+    async fn bind_vsr_session(&self, _session: u64) -> Result<(), IggyError> {
+        Err(IggyError::FeatureUnavailable)
+    }
+
+    #[cfg(feature = "vsr")]
+    async fn reset_vsr_session(&self) -> Result<(), IggyError> {
+        Err(IggyError::FeatureUnavailable)
+    }
 }

core/harness_derive/src/attrs.rs

@@ -182,6 +182,9 @@ pub struct ServerAttrs {
     /// Dynamic config overrides using dot-notation paths.
     pub config_overrides: Vec<ConfigOverride>,
 
+    /// Executable path or bare cargo binary name to launch for the test server.
+    pub executable_path: Option<String>,
+
     /// Path to a TOML config file for the server.
     pub config_path: Option<String>,
 
@@ -440,6 +443,11 @@ fn parse_server_attrs(input: ParseStream) -> syn::Result<ServerAttrs> {
                 let lit: LitStr = input.parse()?;
                 server.config_path = Some(lit.value());
             }
+            "executable_path" => {
+                input.parse::<Token![=]>()?;
+                let lit: LitStr = input.parse()?;
+                server.executable_path = Some(lit.value());
+            }
             _ => {
                 input.parse::<Token![=]>()?;
                 let value = parse_config_value(input)?;
@@ -627,6 +635,15 @@ mod tests {
         assert!(matches!(&segment_size.value, ConfigValue::Static(s) if s == "1MiB"));
     }
 
+    #[test]
+    fn parse_server_executable_path() {
+        let attrs: IggyTestAttrs = syn::parse_quote!(server(executable_path = "iggy-server-ng"));
+        assert_eq!(
+            attrs.server.executable_path.as_deref(),
+            Some("iggy-server-ng")
+        );
+    }
+
     #[test]
     fn parse_server_matrix() {
         let attrs: IggyTestAttrs = syn::parse_quote!(server(segment.size = ["512B", "1MiB"]));

core/harness_derive/src/codegen.rs

@@ -424,6 +424,10 @@ fn generate_harness_setup(
     // Always add extra_envs (may be empty)
     server_builder_calls.push(quote!(.extra_envs(__extra_envs)));
 
+    if let Some(ref executable_path) = attrs.server.executable_path {
+        server_builder_calls.push(quote!(.executable_path(#executable_path)));
+    }
+
     // If a config_path is specified, inject IGGY_CONFIG_PATH into extra_envs
     let config_path_setup = if let Some(ref config_path) = attrs.server.config_path {
         quote! {

core/integration-vsr/src/lib.rs

@@ -0,0 +1,17 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */

core/integration/Cargo.toml

@@ -26,6 +26,7 @@ publish = false
 # inside the docker containers. This is a temporary workaround (hopefully).
 [features]
 ci-qemu = []
+vsr = ["iggy/vsr"]
 
 [dependencies]
 assert_cmd = { workspace = true }

core/integration/src/harness/config/server.rs

@@ -56,10 +56,12 @@ mod tests {
     fn test_server_config_builder() {
         let config = TestServerConfig::builder()
             .quic_enabled(false)
+            .executable_path("iggy-server-ng")
             .extra_envs(HashMap::from([("FOO".to_string(), "BAR".to_string())]))
             .build();
 
         assert!(!config.quic_enabled);
+        assert_eq!(config.executable_path.as_deref(), Some("iggy-server-ng"));
         assert_eq!(config.extra_envs.get("FOO"), Some(&"BAR".to_string()));
     }