fix errors/delete_object

ecstore/src/store.rs

@@ -6,6 +6,7 @@ use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_stri
 use crate::config::GLOBAL_StorageClass;
 use crate::config::{self, storageclass, GLOBAL_ConfigSys};
 use crate::disk::endpoint::EndpointType;
+use crate::disk::error::is_err_file_not_found;
 use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions, MetaCacheEntry};
 use crate::global::{
     is_dist_erasure, is_erasure_sd, set_global_deployment_id, set_object_layer, DISK_ASSUME_UNKNOWN_SIZE, DISK_FILL_FRACTION,
@@ -17,7 +18,8 @@ use crate::new_object_layer_fn;
 use crate::pools::PoolMeta;
 use crate::store_api::{BackendByte, BackendDisks, BackendInfo, ListMultipartsInfo, ObjectIO, StorageInfo};
 use crate::store_err::{
-    is_err_bucket_exists, is_err_invalid_upload_id, is_err_object_not_found, is_err_version_not_found, StorageError,
+    is_err_bucket_exists, is_err_invalid_upload_id, is_err_object_not_found, is_err_read_quorum, is_err_version_not_found,
+    to_object_err, StorageError,
 };
 use crate::store_init::ec_drives_no_config;
 use crate::utils::crypto::base64_decode;
@@ -490,8 +492,123 @@ impl ECStore {
         bucket: &str,
         object: &str,
         opts: &ObjectOptions,
-    ) -> Result<(PoolObjInfo, Vec<Error>)> {
-        internal_get_pool_info_existing_with_opts(&self.pools, bucket, object, opts).await
+    ) -> Result<(PoolObjInfo, Vec<PoolErr>)> {
+        self.internal_get_pool_info_existing_with_opts(bucket, object, opts).await
+    }
+
+    async fn internal_get_pool_info_existing_with_opts(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+    ) -> Result<(PoolObjInfo, Vec<PoolErr>)> {
+        let mut futures = Vec::new();
+
+        for pool in self.pools.iter() {
+            futures.push(pool.get_object_info(bucket, object, opts));
+        }
+
+        let results = join_all(futures).await;
+
+        let mut ress = Vec::new();
+
+        // join_all结果跟输入顺序一致
+        for (i, res) in results.into_iter().enumerate() {
+            let index = i;
+
+            match res {
+                Ok(r) => {
+                    ress.push(PoolObjInfo {
+                        index,
+                        object_info: r,
+                        err: None,
+                    });
+                }
+                Err(e) => {
+                    ress.push(PoolObjInfo {
+                        index,
+                        err: Some(e),
+                        ..Default::default()
+                    });
+                }
+            }
+        }
+
+        ress.sort_by(|a, b| {
+            let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
+            let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
+
+            at.cmp(&bt)
+        });
+
+        let mut def_pool = PoolObjInfo::default();
+        let mut has_def_pool = false;
+
+        for pinfo in ress.iter() {
+            if opts.skip_decommissioned && self.pool_meta.is_suspended(pinfo.index) {
+                continue;
+            }
+
+            // TODO:SkipRebalancing
+            // if opts.SkipRebalancing && z.IsPoolRebalancing(pinfo.Index) {
+            //     continue
+            // }
+
+            if pinfo.err.is_none() {
+                return Ok((pinfo.clone(), self.pools_with_object(&ress, opts)));
+            }
+
+            let err = pinfo.err.as_ref().unwrap();
+
+            if is_err_read_quorum(err) && !opts.metadata_chg {
+                return Ok((pinfo.clone(), self.pools_with_object(&ress, opts)));
+            }
+
+            def_pool = pinfo.clone();
+            has_def_pool = true;
+
+            if !is_err_object_not_found(err) && !is_err_version_not_found(err) {
+                return Err(err.clone());
+            }
+
+            if pinfo.object_info.delete_marker && !pinfo.object_info.name.is_empty() {
+                return Ok((pinfo.clone(), Vec::new()));
+            }
+        }
+
+        if opts.replication_request && opts.delete_marker && has_def_pool {
+            return Ok((def_pool, Vec::new()));
+        }
+
+        Err(to_object_err(Error::new(DiskError::FileNotFound), vec![bucket, object]))
+    }
+
+    fn pools_with_object(&self, pools: &Vec<PoolObjInfo>, opts: &ObjectOptions) -> Vec<PoolErr> {
+        let mut errs = Vec::new();
+        for pool in pools.iter() {
+            if opts.skip_decommissioned && self.pool_meta.is_suspended(pool.index) {
+                continue;
+            }
+            // TODO:SkipRebalancing
+            // if opts.SkipRebalancing && z.IsPoolRebalancing(pinfo.Index) {
+            //     continue
+            // }
+
+            if let Some(err) = &pool.err {
+                if is_err_read_quorum(err) {
+                    errs.push(PoolErr {
+                        index: Some(pool.index),
+                        err: Some(Error::new(StorageError::InsufficientReadQuorum)),
+                    });
+                }
+            } else {
+                errs.push(PoolErr {
+                    index: Some(pool.index),
+                    err: None,
+                });
+            }
+        }
+        errs
     }
 
     async fn get_latest_object_info_with_idx(
@@ -587,6 +704,47 @@ impl ECStore {
             )))
         }
     }
+
+    async fn delete_object_from_all_pools(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        errs: Vec<PoolErr>,
+    ) -> Result<ObjectInfo> {
+        let mut objs = Vec::new();
+        let mut derrs = Vec::new();
+
+        for pe in errs.iter() {
+            if let Some(err) = &pe.err {
+                if is_err_read_quorum(err) {
+                    objs.push(None);
+                    derrs.push(Some(Error::new(StorageError::InsufficientWriteQuorum)));
+                    continue;
+                }
+            }
+
+            if let Some(idx) = pe.index {
+                match self.pools[idx].delete_object(bucket, object, opts.clone()).await {
+                    Ok(res) => {
+                        objs.push(Some(res));
+
+                        derrs.push(None);
+                    }
+                    Err(err) => {
+                        objs.push(None);
+                        derrs.push(Some(err));
+                    }
+                }
+            }
+        }
+
+        if derrs[0].is_some() {
+            return Err(derrs[0].as_ref().unwrap().clone());
+        }
+
+        Ok(objs[0].as_ref().unwrap().clone())
+    }
 }
 
 pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
@@ -661,65 +819,13 @@ pub async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()>
     Ok(())
 }
 
-async fn internal_get_pool_info_existing_with_opts(
-    pools: &[Arc<Sets>],
-    bucket: &str,
-    object: &str,
-    opts: &ObjectOptions,
-) -> Result<(PoolObjInfo, Vec<Error>)> {
-    let mut futures = Vec::new();
-
-    for pool in pools.iter() {
-        futures.push(pool.get_object_info(bucket, object, opts));
-    }
-
-    let results = join_all(futures).await;
-
-    let mut ress = Vec::new();
-
-    // join_all结果跟输入顺序一致
-    for (i, res) in results.into_iter().enumerate() {
-        let index = i;
-
-        match res {
-            Ok(r) => {
-                ress.push(PoolObjInfo {
-                    index,
-                    object_info: r,
-                    err: None,
-                });
-            }
-            Err(e) => {
-                ress.push(PoolObjInfo {
-                    index,
-                    err: Some(e),
-                    ..Default::default()
-                });
-            }
-        }
-    }
-
-    ress.sort_by(|a, b| {
-        let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
-        let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
-
-        at.cmp(&bt)
-    });
-
-    for res in ress {
-        // check
-        if res.err.is_none() {
-            // TODO: let errs = self.poolsWithObject()
-            return Ok((res, Vec::new()));
-        }
-    }
-
-    let ret = PoolObjInfo::default();
-
-    Ok((ret, Vec::new()))
+#[derive(Debug, Default)]
+struct PoolErr {
+    index: Option<usize>,
+    err: Option<Error>,
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct PoolObjInfo {
     pub index: usize,
     pub object_info: ObjectInfo,
@@ -764,6 +870,8 @@ impl ObjectIO for ECStore {
         h: HeaderMap,
         opts: &ObjectOptions,
     ) -> Result<GetObjectReader> {
+        check_get_obj_args(bucket, object)?;
+
         let object = utils::path::encode_dir_object(object);
 
         if self.single_pool() {
@@ -776,6 +884,7 @@ impl ObjectIO for ECStore {
 
         opts.no_lock = true;
 
+        // TODO: check if DeleteMarker
         let (_oi, idx) = self.get_latest_object_info_with_idx(bucket, &object, &opts).await?;
 
         self.pools[idx]
@@ -863,6 +972,7 @@ impl StorageAPI for ECStore {
         }
     }
     async fn storage_info(&self) -> StorageInfo {
+        // FIXME: globalNotificationSys.StorageInfo
         unimplemented!()
     }
     async fn local_storage_info(&self) -> StorageInfo {
@@ -912,11 +1022,14 @@ impl StorageAPI for ECStore {
 
         let mut opts = opts.clone();
         if !opts.force {
-            // TODO: check bucket exists
+            // FIXME: check bucket exists
             opts.force = true
         }
 
-        self.peer_sys.delete_bucket(bucket, &opts).await?;
+        self.peer_sys
+            .delete_bucket(bucket, &opts)
+            .await
+            .map_err(|e| to_object_err(e, vec![bucket]))?;
 
         // TODO: replication opts.srdelete_op
 
@@ -964,14 +1077,18 @@ impl StorageAPI for ECStore {
             meta.versioning_config_xml = xml::serialize::<VersioningConfiguration>(&enableVersioningConfig)?;
         }
 
-        meta.save(self).await?;
+        meta.save(self).await.map_err(|e| to_object_err(e, vec![bucket]))?;
 
         set_bucket_metadata(bucket.to_string(), meta).await;
 
         Ok(())
     }
     async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
-        let mut info = self.peer_sys.get_bucket_info(bucket, opts).await?;
+        let mut info = self
+            .peer_sys
+            .get_bucket_info(bucket, opts)
+            .await
+            .map_err(|e| to_object_err(e, vec![bucket]))?;
 
         if let Ok(sys) = metadata_sys::get(bucket).await {
             info.created = Some(sys.created);
@@ -1009,32 +1126,52 @@ impl StorageAPI for ECStore {
 
         // TODO: nslock
 
-        let mut jhs = Vec::new();
-        let semaphore = Arc::new(Semaphore::new(num_cpus::get()));
-        let pools = Arc::new(self.pools.clone());
+        let mut futures = Vec::with_capacity(objects.len());
 
         for obj in objects.iter() {
-            let (semaphore, pools, bucket, object_name, opt) = (
-                semaphore.clone(),
-                pools.clone(),
-                bucket.to_string(),
-                obj.object_name.to_string(),
-                ObjectOptions::default(),
-            );
-
-            let jh = tokio::spawn(async move {
-                let _permit = semaphore.acquire().await.unwrap();
-                internal_get_pool_info_existing_with_opts(pools.as_ref(), &bucket, &object_name, &opt).await
+            futures.push(async move {
+                self.internal_get_pool_info_existing_with_opts(
+                    &bucket,
+                    &obj.object_name,
+                    &ObjectOptions {
+                        no_lock: true,
+                        ..Default::default()
+                    },
+                )
+                .await
             });
-            jhs.push(jh);
-        }
-        let mut results = Vec::new();
-        for jh in jhs {
-            results.push(jh.await.unwrap());
         }
 
+        let results = join_all(futures).await;
+
+        // let mut jhs = Vec::new();
+        // let semaphore = Arc::new(Semaphore::new(num_cpus::get()));
+        // let pools = Arc::new(self.pools.clone());
+
+        // for obj in objects.iter() {
+        //     let (semaphore, pools, bucket, object_name, opt) = (
+        //         semaphore.clone(),
+        //         pools.clone(),
+        //         bucket.to_string(),
+        //         obj.object_name.to_string(),
+        //         ObjectOptions::default(),
+        //     );
+
+        //     let jh = tokio::spawn(async move {
+        //         let _permit = semaphore.acquire().await.unwrap();
+        //         self.internal_get_pool_info_existing_with_opts(pools.as_ref(), &bucket, &object_name, &opt)
+        //             .await
+        //     });
+        //     jhs.push(jh);
+        // }
+        // let mut results = Vec::new();
+        // for jh in jhs {
+        //     results.push(jh.await.unwrap());
+        // }
+
         // 记录pool Index 对应的objects pool_idx -> objects idx
-        let mut pool_index_objects = HashMap::new();
+        let mut pool_obj_idx_map = HashMap::new();
+        let mut orig_index_map = HashMap::new();
 
         for (i, res) in results.into_iter().enumerate() {
             match res {
@@ -1049,77 +1186,122 @@ impl StorageAPI for ECStore {
                         };
                     }
 
-                    if !pool_index_objects.contains_key(&pinfo.index) {
-                        pool_index_objects.insert(pinfo.index, vec![i]);
-                    } else {
-                        // let mut vals = pool_index_objects.
-                        if let Some(val) = pool_index_objects.get_mut(&pinfo.index) {
-                            val.push(i);
+                    if let Some(obj) = objects.get(i) {
+                        if !pool_obj_idx_map.contains_key(&pinfo.index) {
+                            pool_obj_idx_map.insert(pinfo.index, vec![obj.clone()]);
+                        } else {
+                            if let Some(val) = pool_obj_idx_map.get_mut(&pinfo.index) {
+                                val.push(obj.clone());
+                            }
+                        }
+
+                        if !orig_index_map.contains_key(&pinfo.index) {
+                            orig_index_map.insert(pinfo.index, vec![i]);
+                        } else {
+                            if let Some(val) = orig_index_map.get_mut(&pinfo.index) {
+                                val.push(i);
+                            }
                         }
                     }
                 }
                 Err(e) => {
-                    //TODO: check not found
+                    if !is_err_object_not_found(&e) && is_err_version_not_found(&e) {
+                        del_errs[i] = Some(e)
+                    }
 
-                    del_errs[i] = Some(e)
+                    if let Some(obj) = objects.get(i) {
+                        del_objects[i] = DeletedObject {
+                            object_name: utils::path::decode_dir_object(&obj.object_name),
+                            version_id: obj.version_id.map(|v| v.to_string()),
+                            ..Default::default()
+                        }
+                    }
                 }
             }
         }
 
-        if !pool_index_objects.is_empty() {
-            for sets in self.pools.iter() {
+        if !pool_obj_idx_map.is_empty() {
+            for (i, sets) in self.pools.iter().enumerate() {
                 //  取pool idx 对应的 objects index
-                let vals = pool_index_objects.get(&sets.pool_idx);
-                if vals.is_none() {
-                    continue;
-                }
+                if let Some(objs) = pool_obj_idx_map.get(&i) {
+                    //  取对应obj,理论上不会none
+                    // let objs: Vec<ObjectToDelete> = obj_idxs.iter().filter_map(|&idx| objects.get(idx).cloned()).collect();
 
-                let obj_idxs = vals.unwrap();
-                //  取对应obj,理论上不会none
-                let objs: Vec<ObjectToDelete> = obj_idxs.iter().filter_map(|&idx| objects.get(idx).cloned()).collect();
-
-                if objs.is_empty() {
-                    continue;
-                }
-
-                let (pdel_objs, perrs) = sets.delete_objects(bucket, objs, opts.clone()).await?;
-
-                // perrs的顺序理论上跟obj_idxs顺序一致
-                for (i, err) in perrs.into_iter().enumerate() {
-                    let obj_idx = obj_idxs[i];
-
-                    if err.is_some() {
-                        del_errs[obj_idx] = err;
+                    if objs.is_empty() {
+                        continue;
                     }
 
-                    let mut dobj = pdel_objs.get(i).unwrap().clone();
-                    dobj.object_name = utils::path::decode_dir_object(&dobj.object_name);
+                    let (pdel_objs, perrs) = sets.delete_objects(bucket, objs.clone(), opts.clone()).await?;
 
-                    del_objects[obj_idx] = dobj;
+                    // 同时存入不可能为none
+                    let org_indexes = orig_index_map.get(&i).unwrap();
+
+                    // perrs的顺序理论上跟obj_idxs顺序一致
+                    for (i, err) in perrs.into_iter().enumerate() {
+                        let obj_idx = org_indexes[i];
+
+                        if err.is_some() {
+                            del_errs[obj_idx] = err;
+                        }
+
+                        let mut dobj = pdel_objs.get(i).unwrap().clone();
+                        dobj.object_name = utils::path::decode_dir_object(&dobj.object_name);
+
+                        del_objects[obj_idx] = dobj;
+                    }
                 }
             }
         }
 
         Ok((del_objects, del_errs))
     }
+
     async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
+        check_del_obj_args(bucket, object)?;
+
         if opts.delete_prefix {
             self.delete_prefix(bucket, object).await?;
             return Ok(ObjectInfo::default());
         }
 
+        // TODO: nslock
+
         let object = utils::path::encode_dir_object(object);
         let object = object.as_str();
 
         // 查询在哪个pool
-        let (mut pinfo, errs) = self.get_pool_info_existing_with_opts(bucket, object, &opts).await?;
+        let (mut pinfo, errs) = self
+            .get_pool_info_existing_with_opts(bucket, object, &opts)
+            .await
+            .map_err(|e| {
+                if is_err_read_quorum(&e) {
+                    Error::new(StorageError::InsufficientWriteQuorum)
+                } else {
+                    e
+                }
+            })?;
+
         if pinfo.object_info.delete_marker && opts.version_id.is_none() {
             pinfo.object_info.name = utils::path::decode_dir_object(object);
             return Ok(pinfo.object_info);
         }
 
-        if !errs.is_empty() {
-            // TODO: deleteObjectFromAllPools
+        if opts.data_movement && opts.src_pool_idx == pinfo.index {
+            return Err(Error::new(StorageError::DataMovementOverwriteErr(
+                bucket.to_owned(),
+                object.to_owned(),
+                opts.version_id.unwrap_or_default(),
+            )));
+        }
+
+        if opts.data_movement {
+            let mut obj = self.pools[pinfo.index].delete_object(bucket, object, opts).await?;
+            obj.name = decode_dir_object(obj.name.as_str());
+            return Ok(obj);
+        }
+
+        if !errs.is_empty() && !opts.versioned && !opts.version_suspended {
+            return self.delete_object_from_all_pools(bucket, object, &opts, errs).await;
         }
 
         let mut obj = self.pools[pinfo.index].delete_object(bucket, object, opts.clone()).await?;
@@ -1610,18 +1792,18 @@ fn check_object_name_for_length_and_slash(bucket: &str, object: &str) -> Result<
 }
 
 fn _check_copy_obj_args(bucket: &str, object: &str) -> Result<()> {
-    _check_bucket_and_object_names(bucket, object)
+    check_bucket_and_object_names(bucket, object)
 }
 
-fn _check_get_obj_args(bucket: &str, object: &str) -> Result<()> {
-    _check_bucket_and_object_names(bucket, object)
+fn check_get_obj_args(bucket: &str, object: &str) -> Result<()> {
+    check_bucket_and_object_names(bucket, object)
 }
 
-fn _check_del_obj_args(bucket: &str, object: &str) -> Result<()> {
-    _check_bucket_and_object_names(bucket, object)
+fn check_del_obj_args(bucket: &str, object: &str) -> Result<()> {
+    check_bucket_and_object_names(bucket, object)
 }
 
-fn _check_bucket_and_object_names(bucket: &str, object: &str) -> Result<()> {
+fn check_bucket_and_object_names(bucket: &str, object: &str) -> Result<()> {
     if !is_meta_bucketname(bucket) && check_valid_bucket_name_strict(bucket).is_err() {
         return Err(Error::new(StorageError::BucketNameInvalid(bucket.to_string())));
     }