add client version check

Author: lzjqsdd

curvine-client/src/file/curvine_filesystem.rs

@@ -47,6 +47,10 @@ impl CurvineFileSystem {
     pub fn with_rt(conf: ClusterConf, rt: Arc<Runtime>) -> FsResult<Self> {
         let fs_context = Arc::new(FsContext::with_rt(conf, rt.clone())?);
         let fs_client = FsClient::new(fs_context.clone());
+
+        // Perform version handshake with master
+        rt.block_on(fs_client.handshake())?;
+
         let fs = Self {
             fs_context,
             fs_client: Arc::new(fs_client),

curvine-client/src/file/fs_client.rs

@@ -45,6 +45,21 @@ impl FsClient {
         self.context.clone()
     }
 
+    /// Perform version handshake with master
+    pub async fn handshake(&self) -> FsResult<()> {
+        let client_version = env!("CARGO_PKG_VERSION");
+
+        let req = GetMasterInfoRequest {
+            client_version: Some(client_version.to_string()),
+        };
+
+        // This will trigger version check on master side
+        let _: GetMasterInfoResponse = self.rpc(RpcCode::GetMasterInfo, req).await?;
+
+        log::info!("Client version {} handshake successful", client_version);
+        Ok(())
+    }
+
     pub async fn mkdir(&self, path: &Path, opts: MkdirOpts) -> FsResult<FileStatus> {
         let header = MkdirRequest {
             path: path.encode(),

curvine-common/proto/master.proto

@@ -175,7 +175,8 @@ message GetBlockLocationsResponse {
 
 // Request master profile information.
 message GetMasterInfoRequest {
-
+    // Client version (format: "major.minor.patch")
+    optional string client_version = 1 [default = ""];
 }
 
 message GetMasterInfoResponse {

curvine-common/src/conf/master_conf.rs

@@ -90,6 +90,7 @@ pub struct MasterConf {
 
     pub log: LogConf,
     pub min_worker_version: String,
+    pub min_client_version: String,
 
     pub ttl_checker_retry_attempts: u32,
 
@@ -267,6 +268,7 @@ impl Default for MasterConf {
             block_replication_concurrency_limit: 1000,
             log: Default::default(),
             min_worker_version: "0.1.0".to_string(),
+            min_client_version: "0.1.0".to_string(),
 
             ttl_checker_retry_attempts: 3,
 

curvine-server/src/master/fs/worker_manager.rs

@@ -20,56 +20,35 @@ use curvine_common::state::{
     BlockLocation, ExtendedBlock, HeartbeatStatus, LocatedBlock, StorageInfo, WorkerAddress,
     WorkerCommand, WorkerInfo, WorkerStatus,
 };
-use curvine_common::version::{CompatibilityPolicy, CompatibilityResult, Version, VersionChecker};
+use curvine_common::version::Version;
 use curvine_common::FsResult;
 use log::{info, warn};
 use orpc::common::ByteUnit;
 use orpc::{err_box, CommonResult};
 use std::collections::HashSet;
 use std::fmt::{Display, Formatter};
-use std::str::FromStr;
 
 pub struct WorkerManager {
     pub(crate) worker_map: WorkerMap,
     pub(crate) block_map: BlockMap,
     pub(crate) worker_policy: WorkerPolicyAdapter,
     pub(crate) cluster_id: String,
     pub(crate) conf: ClusterConf,
-    pub(crate) version_checker: VersionChecker,
 }
 
 impl WorkerManager {
     pub fn new(conf: &ClusterConf) -> Self {
         let worker_policy = WorkerPolicyAdapter::from_conf(conf).unwrap();
 
-        // Parse master version and compatibility policy
-        let master_version = Version::current();
-        let policy = Self::parse_compatibility_policy(conf);
-        let version_checker = VersionChecker::new(master_version, policy);
-
         Self {
             worker_map: WorkerMap::new(),
             block_map: BlockMap::new(),
             worker_policy,
             cluster_id: conf.cluster_id.to_string(),
             conf: conf.clone(),
-            version_checker,
         }
     }
 
-    fn parse_compatibility_policy(conf: &ClusterConf) -> CompatibilityPolicy {
-        // Parse min_worker_version from configuration
-        let min_version = Version::from_str(&conf.master.min_worker_version).unwrap_or_else(|e| {
-            warn!(
-                "Failed to parse min_worker_version '{}': {}, using default 0.1.0",
-                conf.master.min_worker_version, e
-            );
-            Version::new(0, 1, 0)
-        });
-
-        CompatibilityPolicy::new(min_version)
-    }
-
     pub fn heartbeat(
         &mut self,
         cluster_id: &str,
@@ -87,20 +66,8 @@ impl WorkerManager {
             );
         }
 
-        // Check version compatibility
-        let compat_result = self.version_checker.check_compatibility(&version);
-        if let CompatibilityResult::Incompatible(reason) = compat_result {
-            warn!(
-                "Worker {} registration rejected due to version incompatibility: {}",
-                addr, reason
-            );
-            return Err(curvine_common::error::FsError::version_incompatible(reason));
-        }
-
-        info!(
-            "Worker {} registered successfully with version {}",
-            addr, version
-        );
+        // Version check is performed in MasterHandler layer
+        info!("Worker {} registered with version {}", addr, version);
 
         let cmds = match status {
             HeartbeatStatus::Start => {

curvine-server/src/master/master_handler.rs

@@ -28,7 +28,7 @@ use curvine_common::state::{
     CreateFileOpts, FileBlocks, FileStatus, HeartbeatStatus, OpenFlags, RenameFlags,
 };
 use curvine_common::utils::ProtoUtils;
-use curvine_common::version::Version;
+use curvine_common::version::{CompatibilityPolicy, CompatibilityResult, Version, VersionChecker};
 use curvine_common::FsResult;
 use orpc::err_box;
 use orpc::handler::MessageHandler;
@@ -46,6 +46,8 @@ pub struct MasterHandler {
     pub(crate) job_handler: JobHandler,
     pub(crate) mount_manager: Arc<MountManager>,
     pub(crate) replication_handler: Option<MasterReplicationHandler>,
+    pub(crate) worker_version_checker: VersionChecker,
+    pub(crate) client_version_checker: VersionChecker,
 }
 
 impl MasterHandler {
@@ -59,6 +61,34 @@ impl MasterHandler {
         job_handler: JobHandler,
         replication_manager: Arc<MasterReplicationManager>,
     ) -> Self {
+        let master_version = Version::current();
+
+        // Initialize Worker version checker
+        let worker_min_version =
+            Version::from_str(&conf.master.min_worker_version).unwrap_or_else(|e| {
+                warn!(
+                    "Failed to parse min_worker_version '{}': {}, using default 0.1.0",
+                    conf.master.min_worker_version, e
+                );
+                Version::new(0, 1, 0)
+            });
+        let worker_policy = CompatibilityPolicy::new(worker_min_version);
+        let worker_version_checker = VersionChecker::new(master_version, worker_policy);
+
+        // Initialize Client version checker (looser policy - no upper bound)
+        let client_min_version =
+            Version::from_str(&conf.master.min_client_version).unwrap_or_else(|e| {
+                warn!(
+                    "Failed to parse min_client_version '{}': {}, using default 0.1.0",
+                    conf.master.min_client_version, e
+                );
+                Version::new(0, 1, 0)
+            });
+        let client_policy = CompatibilityPolicy::new(client_min_version);
+        // Client uses a very large upper bound for compatibility
+        let client_version_checker =
+            VersionChecker::new(Version::new(u32::MAX, u32::MAX, u32::MAX), client_policy);
+
         Self {
             fs,
             retry_cache,
@@ -68,6 +98,8 @@ impl MasterHandler {
             mount_manager,
             job_handler,
             replication_handler: Some(MasterReplicationHandler::new(replication_manager)),
+            worker_version_checker,
+            client_version_checker,
         }
     }
 
@@ -384,7 +416,36 @@ impl MasterHandler {
     }
 
     pub fn get_master_info(&self, ctx: &mut RpcContext<'_>) -> FsResult<Message> {
-        let _: GetMasterInfoRequest = ctx.parse_header()?;
+        let header: GetMasterInfoRequest = ctx.parse_header()?;
+
+        // Check client version if provided
+        if let Some(client_version_str) = &header.client_version {
+            if !client_version_str.is_empty() {
+                let client_version = Version::from_str(client_version_str).map_err(|e| {
+                    FsError::common(format!(
+                        "Failed to parse client version '{}': {}",
+                        client_version_str, e
+                    ))
+                })?;
+
+                let compat_result = self
+                    .client_version_checker
+                    .check_compatibility(&client_version);
+                if let CompatibilityResult::Incompatible(reason) = compat_result {
+                    warn!("Client version {} rejected: {}", client_version, reason);
+                    return Err(FsError::version_incompatible(reason));
+                }
+
+                // Version check passed
+                log::info!("Client version {} accepted", client_version);
+            } else {
+                // Empty version string - backward compatibility
+                warn!("Client connected without version information (backward compatibility mode)");
+            }
+        } else {
+            // No version field - old client (backward compatibility)
+            warn!("Client connected without version information (backward compatibility mode)");
+        }
 
         let info = self.fs.master_info()?;
         let rep_header = ProtoUtils::master_info_to_pb(info);
@@ -393,7 +454,6 @@ impl MasterHandler {
 
     pub fn worker_heartbeat(&self, ctx: &mut RpcContext<'_>) -> FsResult<Message> {
         let header: WorkerHeartbeatRequest = ctx.parse_header()?;
-        let mut wm = self.fs.worker_manager.write();
 
         // Parse worker version from request
         let worker_version_str = &header.version;
@@ -410,10 +470,25 @@ impl MasterHandler {
             Version::new(0, 1, 0)
         };
 
+        // Check version compatibility at Handler layer
+        let worker_addr = ProtoUtils::worker_address_from_pb(&header.address);
+        let compat_result = self
+            .worker_version_checker
+            .check_compatibility(&worker_version);
+        if let CompatibilityResult::Incompatible(reason) = compat_result {
+            warn!(
+                "Worker {} registration rejected due to version incompatibility: {}",
+                worker_addr, reason
+            );
+            return Err(FsError::version_incompatible(reason));
+        }
+
+        // Version check passed, proceed with heartbeat
+        let mut wm = self.fs.worker_manager.write();
         let cmds = wm.heartbeat(
             &header.cluster_id,
             HeartbeatStatus::from(header.status),
-            ProtoUtils::worker_address_from_pb(&header.address),
+            worker_addr,
             ProtoUtils::storage_info_list_from_pb(header.storages),
             worker_version,
         )?;

etc/curvine-cluster.toml

@@ -10,6 +10,8 @@ format_worker = false
 [master]
 # Minimum supported worker version (backward compatibility)
 min_worker_version = "0.1.0"
+# Minimum supported client version (backward compatibility)
+min_client_version = "0.1.0"
 meta_dir = "testing/meta"
 log = { level = "info", log_dir = "stdout", file_name = "master.log" }