Return error instead of panicking if rewriting fails

.github/workflows/rust.yml

@@ -60,7 +60,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: 1.66.0
+          toolchain: 1.67.1
           override: true
           components: rustfmt, clippy, rust-src
       - uses: Swatinem/rust-cache@v1

Cargo.toml

@@ -3,7 +3,7 @@ name = "raft-engine"
 version = "0.4.1"
 authors = ["The TiKV Project Developers"]
 edition = "2018"
-rust-version = "1.66.0"
+rust-version = "1.67.1"
 description = "A persistent storage engine for Multi-Raft logs"
 readme = "README.md"
 repository = "https://github.com/tikv/raft-engine"
@@ -95,8 +95,6 @@ nightly_group = ["nightly", "swap"]
 raft-proto = { git = "https://github.com/tikv/raft-rs", branch = "master" }
 protobuf = { git = "https://github.com/pingcap/rust-protobuf", branch = "v2.8" }
 protobuf-codegen = { git = "https://github.com/pingcap/rust-protobuf", branch = "v2.8" }
-# TODO: Use official grpc-rs once https://github.com/tikv/grpc-rs/pull/622 is merged.
-grpcio = { git = "https://github.com/tabokie/grpc-rs", branch = "v0.10.x-win" }
 
 [workspace]
 members = ["stress", "ctl"]

src/engine.rs

@@ -172,9 +172,9 @@ where
                 }
                 perf_context!(log_write_duration).observe_since(now);
                 if sync {
-                    // As per trait protocol, this error should be retriable. But we panic anyway to
+                    // As per trait protocol, sync error should be retriable. But we panic anyway to
                     // save the trouble of propagating it to other group members.
-                    self.pipe_log.sync(LogQueue::Append).expect("pipe::sync()");
+                    self.pipe_log.sync(LogQueue::Append);
                 }
                 // Pass the perf context diff to all the writers.
                 let diff = get_perf_context();
@@ -2576,7 +2576,7 @@ pub(crate) mod tests {
             flush(&mut log_batch);
             engine.pipe_log.rotate(LogQueue::Rewrite).unwrap();
         }
-        engine.pipe_log.sync(LogQueue::Rewrite).unwrap();
+        engine.pipe_log.sync(LogQueue::Rewrite);
 
         let engine = engine.reopen();
         for rid in engine.raft_groups() {

src/file_pipe_log/log_file.rs

@@ -67,28 +67,26 @@ impl<F: FileSystem> LogFileWriter<F> {
     }
 
     fn write_header(&mut self, format: LogFileFormat) -> IoResult<()> {
-        self.writer.seek(SeekFrom::Start(0))?;
+        self.writer.rewind()?;
         self.written = 0;
         let mut buf = Vec::with_capacity(LogFileFormat::encoded_len(format.version));
         format.encode(&mut buf).unwrap();
         self.write(&buf, 0)
     }
 
-    pub fn close(&mut self) -> IoResult<()> {
+    pub fn close(&mut self) {
         // Necessary to truncate extra zeros from fallocate().
-        self.truncate()?;
-        self.sync()
+        self.truncate();
+        self.sync();
     }
 
-    pub fn truncate(&mut self) -> IoResult<()> {
+    pub fn truncate(&mut self) {
         if self.written < self.capacity {
-            fail_point!("file_pipe_log::log_file_writer::skip_truncate", |_| {
-                Ok(())
-            });
-            self.writer.truncate(self.written)?;
+            fail_point!("file_pipe_log::log_file_writer::skip_truncate", |_| {});
+            // Panic if truncate fails, in case of data loss.
+            self.writer.truncate(self.written).unwrap();
             self.capacity = self.written;
         }
-        Ok(())
     }
 
     pub fn write(&mut self, buf: &[u8], target_size_hint: usize) -> IoResult<()> {
@@ -117,10 +115,10 @@ impl<F: FileSystem> LogFileWriter<F> {
         Ok(())
     }
 
-    pub fn sync(&mut self) -> IoResult<()> {
+    pub fn sync(&mut self) {
         let _t = StopWatch::new(&*LOG_SYNC_DURATION_HISTOGRAM);
-        self.handle.sync()?;
-        Ok(())
+        // Panic if sync fails, in case of data loss.
+        self.handle.sync().unwrap();
     }
 
     #[inline]

src/file_pipe_log/mod.rs

@@ -231,7 +231,7 @@ pub mod debug {
                         len,
                     });
                 }
-                writer.close().unwrap();
+                writer.close();
                 // Read and verify.
                 let mut reader =
                     LogItemReader::new_file_reader(file_system.clone(), &file_path).unwrap();
@@ -280,7 +280,7 @@ pub mod debug {
                 true, /* create */
             )
             .unwrap();
-            writer.close().unwrap();
+            writer.close();
 
             assert!(LogItemReader::new_file_reader(file_system.clone(), dir.path()).is_err());
             assert!(
@@ -330,7 +330,7 @@ pub mod debug {
                         .unwrap();
                         let f = std::fs::OpenOptions::new().write(true).open(&path).unwrap();
                         let len = writer.offset();
-                        writer.close().unwrap();
+                        writer.close();
                         if shorter {
                             f.set_len(len as u64 - 1).unwrap();
                         }
@@ -341,7 +341,7 @@ pub mod debug {
                             false, /* create */
                         )
                         .unwrap();
-                        writer.close().unwrap();
+                        writer.close();
                         let mut reader = build_file_reader(file_system.as_ref(), &path).unwrap();
                         assert_eq!(reader.parse_format().unwrap(), to);
                         std::fs::remove_file(&path).unwrap();

src/file_pipe_log/pipe.rs

@@ -72,9 +72,7 @@ pub(super) struct SinglePipe<F: FileSystem> {
 impl<F: FileSystem> Drop for SinglePipe<F> {
     fn drop(&mut self) {
         let mut writable_file = self.writable_file.lock();
-        if let Err(e) = writable_file.writer.close() {
-            error!("error while closing the active writer: {e}");
-        }
+        writable_file.writer.close();
         let mut recycled_files = self.recycled_files.write();
         let mut next_reserved_seq = recycled_files
             .iter()
@@ -248,7 +246,7 @@ impl<F: FileSystem> SinglePipe<F> {
         let new_seq = writable_file.seq + 1;
         debug_assert!(new_seq > DEFAULT_FIRST_FILE_SEQ);
 
-        writable_file.writer.close()?;
+        writable_file.writer.close();
 
         let (path_id, handle) = self
             .recycle_file(new_seq)
@@ -272,7 +270,7 @@ impl<F: FileSystem> SinglePipe<F> {
         };
         // File header must be persisted. This way we can recover gracefully if power
         // loss before a new entry is written.
-        new_file.writer.sync()?;
+        new_file.writer.sync();
         self.sync_dir(path_id)?;
 
         **writable_file = new_file;
@@ -321,12 +319,7 @@ impl<F: FileSystem> SinglePipe<F> {
         fail_point!("file_pipe_log::append");
         let mut writable_file = self.writable_file.lock();
         if writable_file.writer.offset() >= self.target_file_size {
-            if let Err(e) = self.rotate_imp(&mut writable_file) {
-                panic!(
-                    "error when rotate [{:?}:{}]: {e}",
-                    self.queue, writable_file.seq,
-                );
-            }
+            self.rotate_imp(&mut writable_file)?;
         }
 
         let seq = writable_file.seq;
@@ -359,9 +352,7 @@ impl<F: FileSystem> SinglePipe<F> {
         }
         let start_offset = writer.offset();
         if let Err(e) = writer.write(bytes.as_bytes(&ctx), self.target_file_size) {
-            if let Err(te) = writer.truncate() {
-                panic!("error when truncate {seq} after error: {e}, get: {}", te);
-            }
+            writer.truncate();
             if is_no_space_err(&e) {
                 // TODO: There exists several corner cases should be tackled if
                 // `bytes.len()` > `target_file_size`. For example,
@@ -372,12 +363,7 @@ impl<F: FileSystem> SinglePipe<F> {
                 // - [3] Both main-dir and spill-dir have several recycled logs.
                 // But as `bytes.len()` is always smaller than `target_file_size` in common
                 // cases, this issue will be ignored temprorarily.
-                if let Err(e) = self.rotate_imp(&mut writable_file) {
-                    panic!(
-                        "error when rotate [{:?}:{}]: {e}",
-                        self.queue, writable_file.seq
-                    );
-                }
+                self.rotate_imp(&mut writable_file)?;
                 // If there still exists free space for this record, rotate the file
                 // and return a special TryAgain Err (for retry) to the caller.
                 return Err(Error::TryAgain(format!(
@@ -401,18 +387,11 @@ impl<F: FileSystem> SinglePipe<F> {
         Ok(handle)
     }
 
-    fn sync(&self) -> Result<()> {
+    fn sync(&self) {
         let mut writable_file = self.writable_file.lock();
-        let seq = writable_file.seq;
         let writer = &mut writable_file.writer;
-        {
-            let _t = StopWatch::new(perf_context!(log_sync_duration));
-            if let Err(e) = writer.sync() {
-                panic!("error when sync [{:?}:{seq}]: {e}", self.queue);
-            }
-        }
-
-        Ok(())
+        let _t = StopWatch::new(perf_context!(log_sync_duration));
+        writer.sync();
     }
 
     fn file_span(&self) -> (FileSeq, FileSeq) {
@@ -520,8 +499,8 @@ impl<F: FileSystem> PipeLog for DualPipes<F> {
     }
 
     #[inline]
-    fn sync(&self, queue: LogQueue) -> Result<()> {
-        self.pipes[queue as usize].sync()
+    fn sync(&self, queue: LogQueue) {
+        self.pipes[queue as usize].sync();
     }
 
     #[inline]
@@ -716,7 +695,7 @@ mod tests {
         let mut handles = Vec::new();
         for i in 0..10 {
             handles.push(pipe_log.append(&mut &content(i)).unwrap());
-            pipe_log.sync().unwrap();
+            pipe_log.sync();
         }
         pipe_log.rotate().unwrap();
         let (first, last) = pipe_log.file_span();
@@ -733,7 +712,7 @@ mod tests {
         let mut handles = Vec::new();
         for i in 0..10 {
             handles.push(pipe_log.append(&mut &content(i + 1)).unwrap());
-            pipe_log.sync().unwrap();
+            pipe_log.sync();
         }
         // Verify the data.
         for (i, handle) in handles.into_iter().enumerate() {

src/filter.rs

@@ -333,7 +333,7 @@ impl RhaiFilterMachine {
                     )?;
                     log_batch.drain();
                 }
-                writer.close()?;
+                writer.close();
             }
         }
         // Delete backup file and defuse the guard.

src/pipe_log.rs

@@ -179,7 +179,7 @@ pub trait PipeLog: Sized {
     ///
     /// This operation might incurs a great latency overhead. It's advised to
     /// call it once every batch of writes.
-    fn sync(&self, queue: LogQueue) -> Result<()>;
+    fn sync(&self, queue: LogQueue);
 
     /// Returns the smallest and largest file sequence number, still in use,
     /// of the specified log queue.

src/purge.rs

@@ -273,7 +273,7 @@ where
     // Rewrites the entire rewrite queue into new log files.
     fn rewrite_rewrite_queue(&self) -> Result<Vec<u64>> {
         let _t = StopWatch::new(&*ENGINE_REWRITE_REWRITE_DURATION_HISTOGRAM);
-        self.pipe_log.rotate(LogQueue::Rewrite)?;
+        self.pipe_log.rotate(LogQueue::Rewrite).unwrap();
 
         let mut force_compact_regions = vec![];
         let memtables = self.memtables.collect(|t| {
@@ -430,7 +430,7 @@ where
     ) -> Result<Option<FileBlockHandle>> {
         if log_batch.is_empty() {
             debug_assert!(sync);
-            self.pipe_log.sync(LogQueue::Rewrite)?;
+            self.pipe_log.sync(LogQueue::Rewrite);
             return Ok(None);
         }
         log_batch.finish_populate(
@@ -439,7 +439,7 @@ where
         )?;
         let file_handle = self.pipe_log.append(LogQueue::Rewrite, log_batch)?;
         if sync {
-            self.pipe_log.sync(LogQueue::Rewrite)?
+            self.pipe_log.sync(LogQueue::Rewrite);
         }
         log_batch.finish_write(file_handle);
         self.memtables.apply_rewrite_writes(

tests/failpoints/test_io_error.rs

@@ -165,19 +165,17 @@ fn test_file_rotate_error() {
     {
         // Fail to create new log file.
         let _f = FailGuard::new("default_fs::create::err", "return");
-        assert!(catch_unwind_silent(|| {
-            let _ = engine.write(&mut generate_batch(1, 4, 5, Some(&entry)), false);
-        })
-        .is_err());
+        assert!(engine
+            .write(&mut generate_batch(1, 4, 5, Some(&entry)), false)
+            .is_err());
         assert_eq!(engine.file_span(LogQueue::Append).1, 1);
     }
     {
         // Fail to write header of new log file.
         let _f = FailGuard::new("log_file::write::err", "1*off->return");
-        assert!(catch_unwind_silent(|| {
-            let _ = engine.write(&mut generate_batch(1, 4, 5, Some(&entry)), false);
-        })
-        .is_err());
+        assert!(engine
+            .write(&mut generate_batch(1, 4, 5, Some(&entry)), false)
+            .is_err());
         assert_eq!(engine.file_span(LogQueue::Append).1, 1);
     }
     {
@@ -527,20 +525,17 @@ fn test_no_space_write_error() {
         cfg.dir = dir.path().to_str().unwrap().to_owned();
         cfg.spill_dir = Some(spill_dir.path().to_str().unwrap().to_owned());
         {
-            // Case 1: `Write` is abnormal for no space left, Engine should panic at
+            // Case 1: `Write` is abnormal for no space left, Engine should fail at
             // `rotate`.
             let cfg_err = Config {
                 target_file_size: ReadableSize(1),
                 ..cfg.clone()
             };
             let engine = Engine::open(cfg_err).unwrap();
             let _f = FailGuard::new("log_fd::write::no_space_err", "return");
-            assert!(catch_unwind_silent(|| {
-                engine
-                    .write(&mut generate_batch(2, 11, 21, Some(&entry)), true)
-                    .unwrap_err();
-            })
-            .is_err());
+            assert!(engine
+                .write(&mut generate_batch(2, 11, 21, Some(&entry)), true)
+                .is_err());
             assert_eq!(
                 0,
                 engine
@@ -554,12 +549,9 @@ fn test_no_space_write_error() {
             let _f1 = FailGuard::new("log_fd::write::no_space_err", "2*return->off");
             let _f2 = FailGuard::new("file_pipe_log::force_choose_dir", "return");
             // The first write should fail, because all dirs run out of space for writing.
-            assert!(catch_unwind_silent(|| {
-                engine
-                    .write(&mut generate_batch(2, 11, 21, Some(&entry)), true)
-                    .unwrap_err();
-            })
-            .is_err());
+            assert!(engine
+                .write(&mut generate_batch(2, 11, 21, Some(&entry)), true)
+                .is_err());
             assert_eq!(
                 0,
                 engine