Merge pull request #2467 from Kingtous/master

fix: infinite loop when skipping files

libs/hbb_common/src/fs.rs

@@ -572,6 +572,22 @@ impl TransferJob {
         self.file_skipped() && self.files.len() == 1
     }
 
+    /// Check whether the job is completed after `read` returns `None`
+    /// This is a helper function which gives additional lifecycle when the job reads `None`.
+    /// If returns `true`, it means we can delete the job automatically. `False` otherwise.
+    ///
+    /// [`Note`]
+    /// Conditions:
+    /// 1. Files are not waiting for comfirmation by peers.
+    #[inline]
+    pub fn job_completed(&self) -> bool {
+        // has no error, Condition 2
+        if !self.enable_overwrite_detection || (!self.file_confirmed && !self.file_is_waiting) {
+            return true;
+        }
+        return false;
+    }
+
     /// Get job error message, useful for getting status when job had finished
     pub fn job_error(&self) -> Option<String> {
         if self.job_skipped() {
@@ -597,7 +613,6 @@ impl TransferJob {
             match r.union {
                 Some(file_transfer_send_confirm_request::Union::Skip(s)) => {
                     if s {
-                        log::debug!("skip file id:{}, file_num:{}", r.id, r.file_num);
                         self.set_file_skipped();
                     } else {
                         self.set_file_confirmed(true);
@@ -744,13 +759,16 @@ pub async fn handle_read_jobs(
                 stream.send(&new_block(block)).await?;
             }
             Ok(None) => {
-                if !job.enable_overwrite_detection || (!job.file_confirmed && !job.file_is_waiting)
+                if job.job_completed() {
-                {
+                    finished.push(job.id());
-                    // for getting error detail, we do not remove this job, we will handle it in io loop
+                    let err = job.job_error();
-                    if job.job_error().is_none() {
+                    if err.is_some() {
-                        finished.push(job.id());
+                        stream
+                            .send(&new_error(job.id(), err.unwrap(), job.file_num()))
+                            .await?;
+                    } else {
+                        stream.send(&new_done(job.id(), job.file_num())).await?;
                     }
-                    stream.send(&new_done(job.id(), job.file_num())).await?;
                 } else {
                     // waiting confirmation.
                 }

src/client/io_loop.rs

@@ -933,14 +933,12 @@ impl<T: InvokeUiSession> Remote<T> {
                                 err = job.job_error();
                                 fs::remove_job(d.id, &mut self.write_jobs);
                             }
-                            if let Some(job) = fs::get_job(d.id, &mut self.read_jobs) {
-                                job.modify_time();
-                                err = job.job_error();
-                                fs::remove_job(d.id, &mut self.read_jobs);
-                            }
                             self.handle_job_status(d.id, d.file_num, err);
                         }
                         Some(file_response::Union::Error(e)) => {
+                            if let Some(job) = fs::get_job(e.id, &mut self.write_jobs) {
+                                fs::remove_job(e.id, &mut self.write_jobs);
+                            }
                             self.handle_job_status(e.id, e.file_num, Some(e.error));
                         }
                         _ => {}

src/ipc.rs

@@ -75,6 +75,11 @@ pub enum FS {
         id: i32,
         file_num: i32,
     },
+    WriteError {
+        id: i32,
+        file_num: i32,
+        err: String
+    },
     WriteOffset {
         id: i32,
         file_num: i32,

src/server/connection.rs

@@ -1304,6 +1304,13 @@ impl Connection {
                         last_modified: d.last_modified,
                         is_upload: true,
                     }),
+                    Some(file_response::Union::Error(e)) => {
+                        self.send_fs(ipc::FS::WriteError {
+                            id: e.id,
+                            file_num: e.file_num,
+                            err: e.error,
+                        });
+                    }
                     _ => {}
                 },
                 Some(message::Union::Misc(misc)) => match misc.union {

src/ui_cm_interface.rs

@@ -596,6 +596,12 @@ async fn handle_fs(fs: ipc::FS, write_jobs: &mut Vec<fs::TransferJob>, tx: &Unbo
                 fs::remove_job(id, write_jobs);
             }
         }
+        ipc::FS::WriteError { id, file_num, err } => {
+            if let Some(job) = fs::get_job(id, write_jobs) {
+                send_raw(fs::new_error(job.id(), err, file_num), tx);
+                fs::remove_job(job.id(), write_jobs);
+            }
+        }
         ipc::FS::WriteBlock {
             id,
             file_num,