update re-uploader

Author: eliezhao

Cargo.lock

@@ -23,6 +23,7 @@ backon = { workspace = true }
 rand = { workspace = true }
 tracing = { workspace = true }
 futures = { workspace = true }
+regex = "1.10.5"
 
 
 [features]

icda-core/Cargo.toml

@@ -0,0 +1,120 @@
+use crate::canister_interface::storage::BlobChunk;
+use crate::icda::ICDA;
+use candid::Principal;
+use regex::Regex;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tokio::fs::{File, ReadDir};
+use tokio::io::AsyncReadExt;
+
+pub async fn cycle_monitor() {
+    unimplemented!()
+}
+
+const BACKUP_PATH: &str = "backup";
+
+pub struct ReUploader {
+    backup: ReadDir,
+    icda: Arc<ICDA>,
+}
+
+impl ReUploader {
+    pub async fn new(icda: ICDA) -> Self {
+        // create backup dir
+        tokio::fs::create_dir(BACKUP_PATH)
+            .await
+            .expect("failed to create backup dir");
+
+        let backup = tokio::fs::read_dir(BACKUP_PATH)
+            .await
+            .expect("failed to read backup dir");
+
+        let icda = Arc::new(icda);
+
+        Self { backup, icda }
+    }
+
+    // cycling monitor backup file and re-upload the failed chunks
+    pub async fn monitor(&mut self) {
+        loop {
+            match self.backup.next_entry().await {
+                Ok(entry) => match entry {
+                    Some(entry) => {
+                        tracing::warn!("ICDA ReUploader: monitor: catch file {:?}", entry.path());
+                        let path = entry.path();
+                        let fut = ReUploader::reupload(self.icda.clone(), path);
+                        tokio::spawn(fut);
+                    }
+                    None => {
+                        tracing::info!("ICDA ReUploader: monitor: no files to reupload");
+                    }
+                },
+                Err(e) => {
+                    tracing::error!(
+                        "ICDA ReUploader: monitor: failed to read backup dir: {:?}",
+                        e
+                    );
+                }
+            }
+            tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
+        }
+    }
+
+    //todo: 可以不用每次都传path, 以及将这个模块放到icda里面
+    // 以及将对chunk的deserialize以及命名放到这个模块
+    async fn reupload(icda: Arc<ICDA>, path: PathBuf) {
+        let mut file = File::open(&path).await.expect("failed to open file");
+        let mut buffer = Vec::new();
+        file.read_to_end(&mut buffer)
+            .await
+            .expect("failed to read file");
+
+        // file's name is : chunk-{canister_id}-{chunk.index}.bin
+        let canister_id = parse_canister_id_from_file_name(path);
+        let sc = icda
+            .storage_canisters_map
+            .get(&canister_id)
+            .expect("failed to get canister")
+            .clone();
+
+        let chunk: BlobChunk = bincode::deserialize(&buffer).expect("failed to deserialize");
+
+        loop {
+            match sc.save_blob(&chunk).await {
+                Ok(_) => {
+                    tracing::info!(
+                        "ICDA ReUploader: reupload success, canister id: {}, chunk index: {}",
+                        canister_id.to_text(),
+                        chunk.index
+                    );
+                    break;
+                }
+                Err(e) => {
+                    tracing::error!(
+                        "ICDA ReUploader: reupload failed: canister id: {}, chunk index: {}.  error: {:?}, retry after 60s",
+                        canister_id.to_text(),
+                        chunk.index,
+                        e
+                    );
+                    tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
+                }
+            }
+        }
+    }
+}
+
+fn parse_canister_id_from_file_name(path: PathBuf) -> Principal {
+    let re = Regex::new(r"chunk_(\d+)_\d+.bin").expect("failed to compile regex");
+    let canister_id = re
+        .captures(
+            path.file_name()
+                .expect("failed to get file name")
+                .to_str()
+                .unwrap(),
+        )
+        .expect("failed to get canister id")
+        .get(1)
+        .expect("failed to get canister id")
+        .as_str();
+    Principal::from_text(canister_id).expect("failed to parse canister id")
+}

icda-core/src/backup.rs

@@ -146,15 +146,16 @@ impl StorageCanister {
     }
 
     pub async fn save_blob(&self, chunk: &BlobChunk) -> anyhow::Result<()> {
-        let arg = Encode!(&chunk)?;
+        let arg = Encode!(chunk)?;
         let raw_response = self
             .agent
             .update_call(&self.canister_id, "save_blob", arg)
             .await?;
         let response = Decode!(&raw_response, Result<(), String>)?;
         if let Err(e) = response {
-            bail!("failed to save blob: {}", e)
+            bail!("storage canister: save blob: failed to save blob: {}", e)
         }
+
         Ok(())
     }
 

icda-core/src/canister_interface/storage.rs

@@ -1,9 +1,11 @@
+use crate::backup::ReUploader;
 use crate::canister_interface::rr_agent::RoundRobinAgent;
 use crate::canister_interface::signature::{ConfirmationStatus, SignatureCanister};
 use crate::canister_interface::storage::{BlobChunk, RoutingInfo, StorageCanister};
 use anyhow::bail;
 use anyhow::Result;
 use candid::{Deserialize, Principal};
+use futures::executor::block_on;
 use ic_agent::identity::BasicIdentity;
 use rand::random;
 use serde::Serialize;
@@ -44,6 +46,8 @@ pub const CANISTER_COLLECTIONS: [[&str; REPLICA_NUM]; COLLECTION_SIZE] = [
     ["r2xtu-uiaaa-aaaag-alf6q-cai"], // lspz2-jx4pu-k3e7p-znm7j-q4yum-ork6e-6w4q6-pijwq-znehu-4jabe-kqe
 ];
 
+const RETRY_TIMES: usize = 3;
+
 #[derive(Serialize, Deserialize, Clone, Default)]
 pub struct BlobKey {
     pub digest: [u8; 32],
@@ -95,23 +99,34 @@ impl ICDA {
                         info!("ICDA::new(): signature canister init success");
                     }
                     Err(e) => {
-                        warn!("ICDA::new(): signature canister init failed, error: {:?}, retry after 5 seconds",e);
-                        tokio::time::sleep(Duration::from_secs(5)).await;
+                        bail!(
+                            "ICDA::new(): signature canister init failed, error: {:?}",
+                            e
+                        );
                     }
                 }
             }
         }
 
         let canister_collection_index = Arc::new(Mutex::new(random::<usize>() % COLLECTION_SIZE));
 
-        // create backup dir
-        let _ = tokio::fs::create_dir("backup").await;
-
-        Ok(Self {
+        let _self = Self {
             canister_collection_index,
             storage_canisters_map,
             signature_canister,
-        })
+        };
+
+        // create backup thread
+        let _icda = _self.clone();
+        let backup_thread = async move {
+            let mut reuploader = ReUploader::new(_icda).await;
+            reuploader.monitor().await;
+        };
+        tokio::task::spawn_blocking(move || {
+            block_on(backup_thread);
+        });
+
+        Ok(_self)
     }
 
     pub async fn push_blob_to_canisters(&self, blob: Vec<u8>) -> Result<BlobKey> {
@@ -299,13 +314,13 @@ impl ICDA {
 
 impl ICDA {
     // push chunks to a single canister
-    async fn push_chunks_to_canister(
+    pub(crate) async fn push_chunks_to_canister(
         sc: StorageCanister,
         chunks: Arc<Vec<BlobChunk>>,
     ) -> Result<()> {
         for chunk in chunks.iter() {
             // simple re-upload
-            for i in 0..3 {
+            for i in 0..RETRY_TIMES {
                 if let Err(e) = sc.save_blob(chunk).await {
                     warn!(
                         "ICDA::save_blob_chunk(): cid: {}, error: {:?}, retry after 5 seconds",
@@ -314,7 +329,9 @@ impl ICDA {
                     );
                     if i == 2 {
                         // save chunks into local storage
-                        let serialized = bincode::serialize(&chunk).unwrap();
+                        let serialized = bincode::serialize(chunk).unwrap();
+
+                        // 放到icda 的 reuploader中
                         let _ = tokio::fs::write(
                             format!(
                                 "backup/chunk_{}_{}.bin",

icda-core/src/icda.rs

@@ -1,2 +1,3 @@
+mod backup;
 pub mod canister_interface;
 pub mod icda;

icda-core/src/lib.rs

@@ -1,4 +1,3 @@
-pub mod canister_service;
 pub mod storage;
 pub mod disperser {
     #![allow(clippy::all)]

server/src/canister_service.rs

@@ -231,7 +231,7 @@ async fn main() -> Result<()> {
 
     // check if cmd is IC
     if is_icda {
-        tokio::time::sleep(Duration::from_secs(10)).await
+        tokio::time::sleep(Duration::from_secs(20)).await
     }
 
     let _ret = storage.get_blob(id).await.unwrap();