add down object heal

Signed-off-by: junxiang Mu <1948535941@qq.com>

ecstore/src/heal/background_heal_ops.rs

@@ -1,6 +1,7 @@
 use madmin::heal_commands::HealResultItem;
 use std::{cmp::Ordering, env, path::PathBuf, sync::Arc, time::Duration};
 use tokio::{
+    spawn,
     sync::{
         mpsc::{self, Receiver, Sender},
         RwLock,
@@ -14,7 +15,10 @@ use super::{
     heal_commands::HealOpts,
     heal_ops::{new_bg_heal_sequence, HealSequence},
 };
+use crate::global::GLOBAL_MRFState;
 use crate::heal::error::ERR_RETRY_HEALING;
+use crate::heal::heal_commands::{HealScanMode, HEAL_ITEM_BUCKET};
+use crate::heal::heal_ops::{HealSource, BG_HEALING_UUID};
 use crate::{
     config::RUSTFS_CONFIG_PREFIX,
     disk::{endpoint::Endpoint, error::DiskError, DiskAPI, DiskInfoOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
@@ -42,18 +46,21 @@ pub async fn init_auto_heal() {
             GLOBAL_BackgroundHealState
                 .push_heal_local_disks(&get_local_disks_to_heal().await)
                 .await;
-            tokio::spawn(async {
+            spawn(async {
                 monitor_local_disks_and_heal().await;
             });
         }
     }
+    spawn(async {
+        GLOBAL_MRFState.heal_routine().await;
+    });
 }
 
 async fn init_background_healing() {
     let bg_seq = Arc::new(new_bg_heal_sequence());
     for _ in 0..GLOBAL_BackgroundHealRoutine.workers {
         let bg_seq_clone = bg_seq.clone();
-        tokio::spawn(async {
+        spawn(async {
             GLOBAL_BackgroundHealRoutine.add_worker(bg_seq_clone).await;
         });
     }
@@ -92,13 +99,14 @@ async fn monitor_local_disks_and_heal() {
         interval.tick().await;
         let heal_disks = GLOBAL_BackgroundHealState.get_heal_local_disk_endpoints().await;
         if heal_disks.is_empty() {
+            info!("heal local disks is empty");
             interval.reset();
             continue;
         }
 
         let store = new_object_layer_fn().expect("errServerNotInitialized");
         if let (_, Some(err)) = store.heal_format(false).await.expect("heal format failed") {
-            if let Some(DiskError::NoHealRequired) = err.downcast_ref() {
+            if let Some(DiskError::NoHealRequired) = err.downcast_ref::<DiskError>() {
             } else {
                 info!("heal format err: {}", err.to_string());
                 interval.reset();
@@ -108,7 +116,7 @@ async fn monitor_local_disks_and_heal() {
 
         for disk in heal_disks.into_ref().iter() {
             let disk_clone = disk.clone();
-            tokio::spawn(async move {
+            spawn(async move {
                 GLOBAL_BackgroundHealState
                     .set_disk_healing_status(disk_clone.clone(), true)
                     .await;
@@ -420,3 +428,30 @@ async fn heal_disk_format(opts: HealOpts) -> Result<(HealResultItem, Option<Erro
     }
     Ok((res, err))
 }
+
+pub(crate) async fn heal_bucket(bucket: &str) -> Result<()> {
+    let (bg_seq, ok) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await;
+    if ok {
+        // bg_seq must be Some when ok is true
+        return bg_seq
+            .unwrap()
+            .queue_heal_task(
+                HealSource {
+                    bucket: bucket.to_string(),
+                    ..Default::default()
+                },
+                HEAL_ITEM_BUCKET.to_string(),
+            )
+            .await;
+    }
+    Ok(())
+}
+
+pub(crate) async fn heal_object(bucket: &str, object: &str, version_id: &str, scan_mode: HealScanMode) -> Result<()> {
+    let (bg_seq, ok) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await;
+    if ok {
+        // bg_seq must be Some when ok is true
+        return HealSequence::heal_object(bg_seq.unwrap(), bucket, object, version_id, scan_mode).await;
+    }
+    Ok(())
+}

ecstore/src/heal/mod.rs

@@ -6,3 +6,4 @@ pub mod data_usage_cache;
 pub mod error;
 pub mod heal_commands;
 pub mod heal_ops;
+pub mod mrf;

ecstore/src/heal/mrf.rs

@@ -0,0 +1,124 @@
+use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
+use crate::heal::background_heal_ops::{heal_bucket, heal_object};
+use crate::heal::heal_commands::{HEAL_DEEP_SCAN, HEAL_NORMAL_SCAN};
+use crate::utils::path::SLASH_SEPARATOR;
+use chrono::{DateTime, TimeDelta, Utc};
+use lazy_static::lazy_static;
+use regex::Regex;
+use std::ops::Sub;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::sync::mpsc::{Receiver, Sender};
+use tokio::sync::RwLock;
+use tokio::time::sleep;
+use tracing::error;
+use uuid::Uuid;
+
+pub const MRF_OPS_QUEUE_SIZE: u64 = 100000;
+pub const HEAL_DIR: &'static str = ".heal";
+pub const HEAL_MRFMETA_FORMAT: u64 = 1;
+pub const HEAL_MRFMETA_VERSION_V1: u64 = 1;
+
+lazy_static! {
+    pub static ref HEAL_MRF_DIR: String =
+        format!("{}{}{}{}{}", BUCKET_META_PREFIX, SLASH_SEPARATOR, HEAL_DIR, SLASH_SEPARATOR, "mrf");
+    static ref PATTERNS: Vec<Regex> = vec![
+        Regex::new(r"^buckets/.*/.metacache/.*").unwrap(),
+        Regex::new(r"^tmp/.*").unwrap(),
+        Regex::new(r"^multipart/.*").unwrap(),
+        Regex::new(r"^tmp-old/.*").unwrap(),
+    ];
+}
+
+#[derive(Default)]
+pub struct PartialOperation {
+    pub bucket: String,
+    pub object: String,
+    pub version_id: Option<String>,
+    pub versions: Vec<u8>,
+    pub set_index: usize,
+    pub pool_index: usize,
+    pub queued: DateTime<Utc>,
+    pub bitrot_scan: bool,
+}
+
+pub struct MRFState {
+    tx: Sender<PartialOperation>,
+    rx: RwLock<Receiver<PartialOperation>>,
+    closed: AtomicBool,
+    closing: AtomicBool,
+}
+
+impl MRFState {
+    pub fn new() -> MRFState {
+        let (tx, rx) = tokio::sync::mpsc::channel(MRF_OPS_QUEUE_SIZE as usize);
+        MRFState {
+            tx,
+            rx: RwLock::new(rx),
+            closed: Default::default(),
+            closing: Default::default(),
+        }
+    }
+
+    pub async fn add_partial(&self, op: PartialOperation) {
+        if self.closed.load(Ordering::SeqCst) || self.closing.load(Ordering::SeqCst) {
+            return;
+        }
+        let _ = self.tx.send(op).await;
+    }
+
+    pub async fn heal_routine(&self) {
+        loop {
+            // rx used only there,
+            if let Some(op) = self.rx.write().await.recv().await {
+                if op.bucket == RUSTFS_META_BUCKET {
+                    for pattern in &*PATTERNS {
+                        if pattern.is_match(&op.object) {
+                            return;
+                        }
+                    }
+                }
+
+                let now = Utc::now();
+                if now.sub(op.queued).num_seconds() < 1 {
+                    sleep(Duration::from_secs(1)).await;
+                }
+
+                let scan_mode = if op.bitrot_scan { HEAL_DEEP_SCAN } else { HEAL_NORMAL_SCAN };
+                if op.object.is_empty() {
+                    if let Err(err) = heal_bucket(&op.bucket).await {
+                        error!("heal bucket failed, bucket: {}, err: {:?}", op.bucket, err);
+                    }
+                } else {
+                    if op.versions.is_empty() {
+                        if let Err(err) =
+                            heal_object(&op.bucket, &op.object, &op.version_id.clone().unwrap_or_default(), scan_mode).await
+                        {
+                            error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err);
+                        }
+                    } else {
+                        let vers = op.versions.len() / 16;
+                        if vers > 0 {
+                            for i in 0..vers {
+                                let start = i * 16;
+                                let end = start + 16;
+                                if let Err(err) = heal_object(
+                                    &op.bucket,
+                                    &op.object,
+                                    &Uuid::from_slice(&op.versions[start..end]).expect("").to_string(),
+                                    scan_mode,
+                                )
+                                .await
+                                {
+                                    error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err);
+                                }
+                            }
+                        }
+                    }
+                }
+            } else {
+                return;
+            }
+        }
+    }
+}