fix(p2p): address PR CurvineIO#671 review findings

Author: 80347547

Cargo.lock

@@ -28,7 +28,9 @@ use orpc::runtime::{RpcRuntime, Runtime};
 use orpc::sys::DataSlice;
 use orpc::{err_box, CommonResult};
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::{Mutex as AsyncMutex, OwnedMutexGuard};
+use tokio::time::timeout;
 
 enum ReaderAdapter {
     Local(BlockReaderLocal),
@@ -303,7 +305,10 @@ impl BlockReader {
             return None;
         }
         let lock = self.fs_context.read_chunk_flight_lock(read_key.clone());
-        let guard = lock.clone().lock_owned().await;
+        let timeout_ms = self.fs_context.conf.client.p2p.transfer_timeout_ms.max(1);
+        let guard = timeout(Duration::from_millis(timeout_ms), lock.clone().lock_owned())
+            .await
+            .ok()?;
         Some((lock, guard))
     }
 

curvine-client/src/block/block_reader.rs

@@ -351,12 +351,9 @@ impl CurvineFileSystem {
 
         let (version, peer_whitelist, tenant_whitelist, signature) =
             self.fs_client.get_p2p_runtime_policy().await?;
-        if !p2p_service
+        let _ = p2p_service
             .sync_runtime_policy_from_master(version, peer_whitelist, tenant_whitelist, signature)
-            .await
-        {
-            return err_box!("failed to apply p2p runtime policy from master");
-        }
+            .await;
         Ok(())
     }
 

curvine-client/src/file/curvine_filesystem.rs

@@ -121,8 +121,7 @@ impl FsContext {
             .eviction_policy(EvictionPolicy::lru())
             .build_with_hasher(BuildHasherDefault::<FxHasher>::default());
 
-        let read_chunk_flights =
-            FastDashMap::with_capacity(conf.client.read_chunk_cache_capacity.min(65536) as usize);
+        let read_chunk_flights = FastDashMap::default();
 
         let p2p_service = if conf.client.p2p.enable {
             let service = Arc::new(P2pService::new_with_runtime(

curvine-client/src/file/fs_context.rs

@@ -186,9 +186,10 @@ impl CacheManager {
         let (path, stored_len, mtime, entry_checksum) = match self.entries.get_mut(&chunk_id) {
             Some(mut entry) => {
                 if entry.expire_at_ms <= now_ms {
-                    self.expired_chunks.fetch_add(1, Ordering::Relaxed);
                     drop(entry);
-                    self.invalidate(chunk_id);
+                    if self.invalidate(chunk_id) {
+                        self.expired_chunks.fetch_add(1, Ordering::Relaxed);
+                    }
                     return CacheGetResult {
                         tag: CacheGetResultTag::Miss,
                         chunk: None,
@@ -280,10 +281,15 @@ impl CacheManager {
             .filter_map(|v| (v.value().expire_at_ms <= now_ms).then_some(*v.key()))
             .collect();
         if !expired.is_empty() {
-            self.expired_chunks
-                .fetch_add(expired.len() as u64, Ordering::Relaxed);
+            let mut newly_expired = 0u64;
             for chunk_id in expired {
-                self.invalidate(chunk_id);
+                if self.invalidate(chunk_id) {
+                    newly_expired += 1;
+                }
+            }
+            if newly_expired > 0 {
+                self.expired_chunks
+                    .fetch_add(newly_expired, Ordering::Relaxed);
             }
         }
     }
@@ -310,12 +316,14 @@ impl CacheManager {
         }
     }
 
-    fn invalidate(&self, chunk_id: ChunkId) {
+    fn invalidate(&self, chunk_id: ChunkId) -> bool {
         if let Some((_, entry)) = self.entries.remove(&chunk_id) {
             self.usage_bytes.fetch_sub(entry.len, Ordering::Relaxed);
             self.invalidations.fetch_add(1, Ordering::Relaxed);
             let _ = fs::remove_file(entry.path);
+            return true;
         }
+        false
     }
 
     fn evict_until_fit(&self) {

curvine-client/src/p2p/cache_manager.rs

@@ -41,6 +41,7 @@ slog-stdlog  = { workspace = true }
 chrono = { workspace = true }
 trait-variant = { workspace = true }
 sha2 = "0.10"
+hmac = "0.12"
 
 anyhow = { workspace = true }
 tracing = { workspace = true }

curvine-common/Cargo.toml

@@ -113,6 +113,13 @@ impl ClientP2pConf {
             )
             .into());
         }
+        if self.provider_ttl.is_zero() {
+            return Err(Error::new(
+                ErrorKind::InvalidInput,
+                "client.p2p.provider_ttl must be greater than 0",
+            )
+            .into());
+        }
         if self.max_inflight_per_peer > 0 && self.min_inflight_per_peer > self.max_inflight_per_peer
         {
             return Err(Error::new(
@@ -653,6 +660,16 @@ mod tests {
         assert!(conf.init().is_err());
     }
 
+    #[test]
+    fn client_p2p_conf_rejects_zero_provider_ttl() {
+        let mut conf = ClientP2pConf {
+            provider_ttl_str: "0s".to_string(),
+            provider_publish_interval_str: "1s".to_string(),
+            ..ClientP2pConf::default()
+        };
+        assert!(conf.init().is_err());
+    }
+
     #[test]
     fn client_p2p_conf_rejects_zero_layered_timeout() {
         let mut conf = ClientP2pConf {

curvine-common/src/conf/client_conf.rs

@@ -12,13 +12,16 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use hmac::{Hmac, Mac};
 use log::info;
 use orpc::common::Utils;
 use orpc::{err_msg, CommonResult};
-use sha2::{Digest, Sha256};
+use sha2::Sha256;
 use std::collections::HashMap;
 use std::process::{Command, Stdio};
 
+type PolicyHmac = Hmac<Sha256>;
+
 pub struct CommonUtils;
 
 impl CommonUtils {
@@ -67,18 +70,19 @@ impl CommonUtils {
         if secret.trim().is_empty() {
             return String::new();
         }
-        let mut hasher = Sha256::new();
-        Self::update_signing_segment(&mut hasher, secret);
-        Self::update_signing_segment(&mut hasher, &version.to_string());
-        Self::update_signing_segment(&mut hasher, &peer_whitelist.len().to_string());
+        let Ok(mut mac) = PolicyHmac::new_from_slice(secret.as_bytes()) else {
+            return String::new();
+        };
+        Self::update_signing_segment(&mut mac, &version.to_string());
+        Self::update_signing_segment(&mut mac, &peer_whitelist.len().to_string());
         for peer in peer_whitelist {
-            Self::update_signing_segment(&mut hasher, peer);
+            Self::update_signing_segment(&mut mac, peer);
         }
-        Self::update_signing_segment(&mut hasher, &tenant_whitelist.len().to_string());
+        Self::update_signing_segment(&mut mac, &tenant_whitelist.len().to_string());
         for tenant in tenant_whitelist {
-            Self::update_signing_segment(&mut hasher, tenant);
+            Self::update_signing_segment(&mut mac, tenant);
         }
-        let digest = hasher.finalize();
+        let digest = mac.finalize().into_bytes();
         digest.iter().map(|v| format!("{:02x}", v)).collect()
     }
 
@@ -121,17 +125,48 @@ impl CommonUtils {
             })
     }
 
-    fn update_signing_segment(hasher: &mut Sha256, value: &str) {
-        hasher.update(value.len().to_string().as_bytes());
-        hasher.update(b":");
-        hasher.update(value.as_bytes());
-        hasher.update(b"\n");
+    fn update_signing_segment(mac: &mut PolicyHmac, value: &str) {
+        mac.update(value.len().to_string().as_bytes());
+        mac.update(b":");
+        mac.update(value.as_bytes());
+        mac.update(b"\n");
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::CommonUtils;
+    use sha2::{Digest, Sha256};
+
+    fn legacy_prefix_signature(
+        secret: &str,
+        version: u64,
+        peer_whitelist: &[String],
+        tenant_whitelist: &[String],
+    ) -> String {
+        fn update_segment(hasher: &mut Sha256, value: &str) {
+            hasher.update(value.len().to_string().as_bytes());
+            hasher.update(b":");
+            hasher.update(value.as_bytes());
+            hasher.update(b"\n");
+        }
+        let mut hasher = Sha256::new();
+        update_segment(&mut hasher, secret);
+        update_segment(&mut hasher, &version.to_string());
+        update_segment(&mut hasher, &peer_whitelist.len().to_string());
+        for peer in peer_whitelist {
+            update_segment(&mut hasher, peer);
+        }
+        update_segment(&mut hasher, &tenant_whitelist.len().to_string());
+        for tenant in tenant_whitelist {
+            update_segment(&mut hasher, tenant);
+        }
+        hasher
+            .finalize()
+            .iter()
+            .map(|v| format!("{:02x}", v))
+            .collect()
+    }
 
     #[test]
     fn p2p_policy_signature_roundtrip() {
@@ -169,4 +204,13 @@ mod tests {
             &signatures
         ));
     }
+
+    #[test]
+    fn p2p_policy_signature_should_not_use_legacy_prefix_hash() {
+        let peers = vec!["peer-a".to_string()];
+        let tenants = vec!["tenant-a".to_string(), "tenant-b".to_string()];
+        let hmac_signature = CommonUtils::sign_p2p_policy("secret", 7, &peers, &tenants);
+        let legacy_signature = legacy_prefix_signature("secret", 7, &peers, &tenants);
+        assert_ne!(hmac_signature, legacy_signature);
+    }
 }

curvine-common/src/utils/common_utils.rs

@@ -390,7 +390,11 @@ impl InodeFile {
     }
 
     fn bump_version_epoch(&mut self) {
-        self.version_epoch = self.version_epoch.max(1).saturating_add(1);
+        self.version_epoch = if self.version_epoch <= 0 {
+            1
+        } else {
+            self.version_epoch.saturating_add(1)
+        };
     }
 
     /// Extend the file to the specified length by allocating new blocks or extending existing ones.
@@ -515,3 +519,23 @@ impl PartialEq for InodeFile {
         self.id == other.id
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::InodeFile;
+
+    #[test]
+    fn bump_version_epoch_should_increment_normal_epoch() {
+        let mut inode = InodeFile::new(1, 1);
+        inode.bump_version_epoch();
+        assert_eq!(inode.version_epoch, 2);
+    }
+
+    #[test]
+    fn bump_version_epoch_should_recover_legacy_zero_epoch() {
+        let mut inode = InodeFile::new(1, 1);
+        inode.version_epoch = 0;
+        inode.bump_version_epoch();
+        assert_eq!(inode.version_epoch, 1);
+    }
+}