Address comments

Signed-off-by: Yang Zhang <[email protected]>

src/file_pipe_log/log_file.rs

@@ -74,19 +74,21 @@ impl<F: FileSystem> LogFileWriter<F> {
         self.write(&buf, 0)
     }
 
-    pub fn close(&mut self) {
+    pub fn close(&mut self) -> Result<()> {
         // Necessary to truncate extra zeros from fallocate().
-        self.truncate();
+        self.truncate()?;
         self.sync();
+        Ok(())
     }
 
-    pub fn truncate(&mut self) {
+    pub fn truncate(&mut self) -> Result<()> {
         if self.written < self.capacity {
             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.writer.truncate(self.written)?;
             self.capacity = self.written;
         }
+        Ok(())
     }
 
     pub fn write(&mut self, buf: &[u8], target_size_hint: usize) -> IoResult<()> {

src/file_pipe_log/mod.rs

@@ -231,7 +231,7 @@ pub mod debug {
                         len,
                     });
                 }
-                writer.close();
+                writer.close().unwrap();
                 // 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();
+            writer.close().unwrap();
 
             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();
+                        writer.close().unwrap();
                         if shorter {
                             f.set_len(len as u64 - 1).unwrap();
                         }
@@ -341,7 +341,7 @@ pub mod debug {
                             false, /* create */
                         )
                         .unwrap();
-                        writer.close();
+                        writer.close().unwrap();
                         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,7 +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();
-        writable_file.writer.close();
+        writable_file.writer.close().unwrap();
         let mut recycled_files = self.recycled_files.write();
         let mut next_reserved_seq = recycled_files
             .iter()
@@ -246,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().unwrap();
 
         let (path_id, handle) = self
             .recycle_file(new_seq)
@@ -352,7 +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) {
-            writer.truncate();
+            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,

src/filter.rs

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