841 multiple local replicas dc fix

CHANGELOG.md

@@ -10,7 +10,7 @@ Bob versions changelog
 
 
 #### Fixed
-
+- Fix local replicas error (#841)
 
 #### Updated
 

bob-backend/src/core.rs

@@ -33,6 +33,10 @@ impl Operation {
     pub fn remote_node_name(&self) -> Option<&NodeName> {
         self.remote_node_name.as_ref()
     }
+
+    pub fn disk_path(&self) -> Option<DiskPath> {
+        self.disk_path.clone()
+    }
 }
 
 impl Debug for Operation {

bob-backend/src/pearl/disk_controller.rs

@@ -435,11 +435,13 @@ impl DiskController {
     }
 
     pub(crate) fn can_process_operation(&self, op: &Operation) -> bool {
-        self.is_alien && op.is_data_alien()
-            || self
-                .vdisks
-                .iter()
-                .any(|&vdisk_id| vdisk_id == op.vdisk_id())
+        if self.is_alien && op.is_data_alien() {
+            return true;
+        }
+
+        let path_match = op.disk_path().map(|p| p == self.disk).unwrap_or(true);
+        let vdisk_match = self.vdisks.iter().any(|&vdisk_id| vdisk_id == op.vdisk_id());
+        path_match && vdisk_match
     }
 
     pub(crate) async fn get(&self, op: Operation, key: BobKey) -> Result<BobData, Error> {

bob-backend/src/pearl/tests.rs

@@ -72,7 +72,7 @@ async fn test_write_multiple_read() {
     let vdisk_id = 0;
     let backend = backend().await;
     backend.run().await.unwrap();
-    let path = DiskPath::new(DISK_NAME.into(), "");
+    let path = DiskPath::new(DISK_NAME.into(), "/tmp/d1");
     let operation = Operation::new_local(vdisk_id, path);
     let data = BobData::new(vec![].into(), BobMeta::new(TIMESTAMP));
     let write = backend
@@ -91,3 +91,79 @@ async fn test_write_multiple_read() {
     assert_eq!(TIMESTAMP, res.unwrap().meta().timestamp());
     drop_pearl().await;
 }
+
+async fn multiple_local_replicas_backend() -> PearlBackend {
+    let node_config = "
+log_config: logger.yaml
+users_config: users.yaml
+name: local_node
+quorum: 2
+operation_timeout: 3sec
+check_interval: 5000ms
+cluster_policy: quorum             # quorum
+backend_type: pearl                # in_memory, stub, pearl
+cleanup_interval: 1d
+pearl:                             # used only for 'backend_type: pearl'
+  max_blob_size: 10000000          # size in bytes. required for 'pearl'
+  max_data_in_blob: 10000          # optional
+  blob_file_name_prefix: bob       # optional
+  fail_retry_timeout: 100ms
+  alien_disk: disk1                # required for 'pearl'
+  settings:                        # describes how create and manage bob directories. required for 'pearl'
+    root_dir_name: bob             # root dir for bob storage. required for 'pearl'
+    alien_root_dir_name: alien     # root dir for alien storage in 'alien_disk'. required for 'pearl'
+    timestamp_period: 1d           # period when new pearl directory created. required for 'pearl'
+    create_pearl_wait_delay: 100ms
+";
+    let cluster_config = "
+nodes:
+    - name: local_node
+      address: 127.0.0.1:20000
+      disks:
+        - name: disk1
+          path: /tmp/multiple-d1
+        - name: disk2
+          path: /tmp/multiple-d2
+vdisks:
+    - id: 0
+      replicas:
+        - node: local_node
+          disk: disk1
+        - node: local_node
+          disk: disk2
+";
+    debug!("node_config: {}", node_config);
+    debug!("cluster_config: {}", cluster_config);
+    create_backend(node_config, cluster_config).await.unwrap()
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_write_multiple_local_replicas() {
+    let disk1_dir = "/tmp/multiple-d1";
+    let vdisk_id = 0;
+    let backend = multiple_local_replicas_backend().await;
+    backend.run().await.unwrap();
+
+    let ops = [("disk1", disk1_dir), ("disk2", "/tmp/multiple-d2")];
+    for (disk, path) in ops { 
+        let path = DiskPath::new(disk.into(), path);
+        let operation = Operation::new_local(vdisk_id, path);
+        let data = BobData::new(vec![].into(), BobMeta::new(TIMESTAMP));
+        let write = backend
+            .put(operation.clone(), BobKey::from(KEY_ID), &data)
+            .await;
+        assert!(write.is_ok());
+
+        let read = backend.get(operation.clone(), BobKey::from(KEY_ID)).await;
+        assert_eq!(TIMESTAMP, read.unwrap().meta().timestamp());
+    }
+
+    let mut alien_path = PathBuf::from(disk1_dir);
+    alien_path.push("alien");
+    let content = std::fs::read_dir(alien_path).unwrap();
+    assert_eq!(0, content.count(), "alien path should be empty");
+
+    for (_, path) in ops {
+        remove_dir_all(path).await.unwrap();
+    }
+}