rollback to a consistent state after fsync error

.github/workflows/rust.yml

@@ -43,7 +43,9 @@ jobs:
           RUST_BACKTRACE: 1
       - name: Run asan tests
         if: ${{ matrix.os == 'ubuntu-latest' }}
-        run: cargo test -Zbuild-std --target x86_64-unknown-linux-gnu --features failpoints --verbose -- --nocapture
+        run: |
+          cargo test -Zbuild-std --target x86_64-unknown-linux-gnu --all --verbose -- --nocapture
+          cargo test -Zbuild-std --target x86_64-unknown-linux-gnu --test failpoints --features failpoints --verbose -- --test-threads 1 --nocapture
         env:
           RUST_BACKTRACE: 1
           RUSTFLAGS: '-Zsanitizer=address'

src/engine.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 use std::time::Instant;
 use std::u64;
 
+use fail::fail_point;
 use log::{error, info};
 use protobuf::{parse_from_bytes, Message};
 
@@ -20,7 +21,7 @@ use crate::pipe_log::{FileBlockHandle, FileId, LogQueue, PipeLog};
 use crate::purge::{PurgeHook, PurgeManager};
 use crate::util::InstantExt;
 use crate::write_barrier::{WriteBarrier, Writer};
-use crate::{Error, Result};
+use crate::Result;
 
 pub struct Engine<B = DefaultFileBuilder, P = FilePipeLog<B>>
 where
@@ -122,14 +123,12 @@ where
             let mut writer = Writer::new(log_batch as &_, sync);
             if let Some(mut group) = self.write_barrier.enter(&mut writer) {
                 for writer in group.iter_mut() {
+                    fail_point!("engine::write::pre");
                     sync |= writer.is_sync();
                     let log_batch = writer.get_payload();
                     let res = if !log_batch.is_empty() {
-                        self.pipe_log.append(
-                            LogQueue::Append,
-                            log_batch.encoded_bytes(),
-                            false, /*sync*/
-                        )
+                        self.pipe_log
+                            .append(LogQueue::Append, log_batch.encoded_bytes())
                     } else {
                         // TODO(tabokie)
                         Ok(FileBlockHandle {
@@ -140,16 +139,12 @@ where
                     };
                     writer.set_output(res);
                 }
-                if sync {
-                    // fsync() is not retryable, a failed attempt could result in
-                    // unrecoverable loss of data written after last successful
-                    // fsync(). See [PostgreSQL's fsync() surprise]
-                    // (https://lwn.net/Articles/752063/) for more details.
-                    if let Err(e) = self.pipe_log.sync(LogQueue::Append) {
-                        for writer in group.iter_mut() {
-                            writer.set_output(Err(Error::Fsync(e.to_string())));
-                        }
-                    }
+                if let Err(e) = self.pipe_log.maybe_sync(LogQueue::Append, sync) {
+                    panic!(
+                        "Cannot sync queue: {:?}, for there is an IO error raised: {}",
+                        LogQueue::Append,
+                        e
+                    );
                 }
             }
             writer.finish()?
@@ -171,7 +166,7 @@ where
     /// Synchronize the Raft engine.
     pub fn sync(&self) -> Result<()> {
         // TODO(tabokie): use writer.
-        self.pipe_log.sync(LogQueue::Append)
+        self.pipe_log.maybe_sync(LogQueue::Append, true)
     }
 
     pub fn put_message<S: Message>(&self, region_id: u64, key: &[u8], m: &S) -> Result<()> {

src/errors.rs

@@ -15,8 +15,6 @@ pub enum Error {
     Corruption(String),
     #[error("IO Error: {0}")]
     Io(#[from] IoError),
-    #[error("Fsync Error: {0}")]
-    Fsync(String),
     #[error("Codec Error: {0}")]
     Codec(#[from] CodecError),
     #[error("Protobuf Error: {0}")]

src/file_pipe_log.rs

@@ -9,6 +9,7 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Instant;
 
+use fail::fail_point;
 use fs2::FileExt;
 use log::{debug, info, warn};
 use num_derive::{FromPrimitive, ToPrimitive};
@@ -160,9 +161,6 @@ impl<W: Seek + Write> ActiveFile<W> {
     }
 
     fn rotate(&mut self, fd: Arc<LogFd>, writer: W) -> Result<()> {
-        if self.last_sync < self.written {
-            self.fd.sync()?;
-        }
         self.writer = writer;
         self.written = 0;
         self.capacity = 0;
@@ -172,6 +170,10 @@ impl<W: Seek + Write> ActiveFile<W> {
     }
 
     fn truncate(&mut self) -> Result<()> {
+        fail_point!("active_file::truncate::force", |_| {
+            self.fd.truncate(self.written)?;
+            Ok(())
+        });
         if self.written < self.capacity {
             self.fd.truncate(self.written)?;
             self.capacity = self.written;
@@ -184,10 +186,10 @@ impl<W: Seek + Write> ActiveFile<W> {
         self.written = 0;
         let mut buf = Vec::with_capacity(LOG_FILE_HEADER_LEN);
         LogFileHeader::new().encode(&mut buf)?;
-        self.write(&buf, true, 0)
+        self.write(&buf, 0)
     }
 
-    fn write(&mut self, buf: &[u8], sync: bool, target_file_size: usize) -> Result<()> {
+    fn write(&mut self, buf: &[u8], target_file_size: usize) -> Result<()> {
         if self.written + buf.len() > self.capacity {
             // Use fallocate to pre-allocate disk space for active file.
             let alloc = std::cmp::max(
@@ -204,8 +206,15 @@ impl<W: Seek + Write> ActiveFile<W> {
         }
         self.writer.write_all(buf)?;
         self.written += buf.len();
-        if sync {
+        Ok(())
+    }
+
+    fn sync(&mut self) -> Result<()> {
+        if self.last_sync < self.written {
+            let start = Instant::now();
+            self.fd.sync()?;
             self.last_sync = self.written;
+            LOG_SYNC_TIME_HISTOGRAM.observe(start.saturating_elapsed().as_secs_f64());
         }
         Ok(())
     }
@@ -305,19 +314,20 @@ impl<B: FileBuilder> LogManager<B> {
         Ok(manager)
     }
 
-    fn new_log_file(&mut self) -> Result<()> {
+    fn rotate(&mut self) -> Result<()> {
         debug_assert!(self.active_file_seq >= INIT_FILE_ID);
         // Necessary to truncate extra zeros from fallocate().
-        self.truncate_active_log()?;
+        self.active_file.truncate()?;
+        self.active_file.sync()?;
         self.active_file_seq += 1;
-
         let path = build_file_path(
             &self.dir,
             FileId {
                 queue: self.queue,
                 seq: self.active_file_seq,
             },
         );
+
         let fd = Arc::new(LogFd::create(&path)?);
         self.all_files.push_back(fd.clone());
         self.active_file.rotate(
@@ -326,29 +336,26 @@ impl<B: FileBuilder> LogManager<B> {
                 .build_writer(&path, LogFile::new(fd), true /*create*/)?,
         )?;
         self.sync_dir()?;
-
         for listener in &self.listeners {
             listener.post_new_log_file(FileId {
                 queue: self.queue,
                 seq: self.active_file_seq,
             });
         }
-
         self.update_metrics();
-
         Ok(())
     }
 
+    fn sync(&mut self) -> Result<()> {
+        self.active_file.sync()
+    }
+
     fn sync_dir(&self) -> Result<()> {
         let path = PathBuf::from(&self.dir);
         std::fs::File::open(path).and_then(|d| d.sync_all())?;
         Ok(())
     }
 
-    fn truncate_active_log(&mut self) -> Result<()> {
-        self.active_file.truncate()
-    }
-
     fn get_fd(&self, file_seq: FileSeq) -> Result<Arc<LogFd>> {
         if file_seq < self.first_file_seq || file_seq > self.active_file_seq {
             return Err(Error::Io(IoError::new(
@@ -359,10 +366,6 @@ impl<B: FileBuilder> LogManager<B> {
         Ok(self.all_files[(file_seq - self.first_file_seq) as usize].clone())
     }
 
-    fn get_active_fd(&self) -> Option<Arc<LogFd>> {
-        self.all_files.back().cloned()
-    }
-
     fn purge_to(&mut self, file_seq: FileSeq) -> Result<usize> {
         if file_seq > self.active_file_seq {
             return Err(box_err!("Purge active or newer files"));
@@ -374,15 +377,17 @@ impl<B: FileBuilder> LogManager<B> {
         Ok(end_offset)
     }
 
-    fn append(&mut self, content: &[u8], sync: &mut bool) -> Result<(FileBlockHandle, Arc<LogFd>)> {
-        if self.active_file.written >= self.rotate_size {
-            self.new_log_file()?;
-        }
-        if self.active_file.since_last_sync() >= self.bytes_per_sync {
-            *sync = true;
-        }
+    fn append(&mut self, content: &[u8]) -> Result<FileBlockHandle> {
         let offset = self.active_file.written as u64;
-        self.active_file.write(content, *sync, self.rotate_size)?;
+        if let Err(e) = self.active_file.write(content, self.rotate_size) {
+            if let Err(te) = self.active_file.truncate() {
+                panic!(
+                    "error when truncate {} after error: {}, get: {}",
+                    self.active_file_seq, e, te
+                );
+            }
+            return Err(e);
+        }
         let handle = FileBlockHandle {
             id: FileId {
                 queue: self.queue,
@@ -394,7 +399,7 @@ impl<B: FileBuilder> LogManager<B> {
         for listener in &self.listeners {
             listener.on_append_log_file(handle);
         }
-        Ok((handle, self.active_file.fd.clone()))
+        Ok(handle)
     }
 
     fn update_metrics(&self) {
@@ -638,7 +643,6 @@ impl<B: FileBuilder> FilePipeLog<B> {
                         }
                     }
                 }
-                debug!("Recover queue:{:?} finish.", queue);
                 Ok(sequential_replay_machine)
             })
             .try_reduce(
@@ -648,24 +652,10 @@ impl<B: FileBuilder> FilePipeLog<B> {
                     Ok(sequential_replay_machine_left)
                 },
             )?;
+        debug!("Recover queue:{:?} finish.", queue);
         Ok(sequential_replay_machine)
     }
 
-    fn append_bytes(
-        &self,
-        queue: LogQueue,
-        content: &[u8],
-        sync: &mut bool,
-    ) -> Result<FileBlockHandle> {
-        let (block_handle, fd) = self.mut_queue(queue).append(content, sync)?;
-        if *sync {
-            let start = Instant::now();
-            fd.sync()?;
-            LOG_SYNC_TIME_HISTOGRAM.observe(start.saturating_elapsed().as_secs_f64());
-        }
-        Ok(block_handle)
-    }
-
     fn get_queue(&self, queue: LogQueue) -> RwLockReadGuard<LogManager<B>> {
         match queue {
             LogQueue::Append => self.appender.read(),
@@ -694,9 +684,9 @@ impl<B: FileBuilder> PipeLog for FilePipeLog<B> {
         Ok(buf)
     }
 
-    fn append(&self, queue: LogQueue, bytes: &[u8], mut sync: bool) -> Result<FileBlockHandle> {
+    fn append(&self, queue: LogQueue, bytes: &[u8]) -> Result<FileBlockHandle> {
         let start = Instant::now();
-        let block_handle = self.append_bytes(queue, bytes, &mut sync)?;
+        let block_handle = self.mut_queue(queue).append(bytes)?;
         match queue {
             LogQueue::Rewrite => {
                 LOG_APPEND_TIME_HISTOGRAM_VEC
@@ -712,10 +702,28 @@ impl<B: FileBuilder> PipeLog for FilePipeLog<B> {
         Ok(block_handle)
     }
 
-    fn sync(&self, queue: LogQueue) -> Result<()> {
-        if let Some(fd) = self.get_queue(queue).get_active_fd() {
-            fd.sync()?;
+    fn maybe_sync(&self, queue: LogQueue, force: bool) -> Result<()> {
+        let mut manager = match queue {
+            LogQueue::Append => self.appender.write(),
+            LogQueue::Rewrite => self.rewriter.write(),
+        };
+
+        if manager.active_file.written >= manager.rotate_size {
+            if let Err(e) = manager.rotate() {
+                panic!(
+                    "error when rotate [{:?}:{}]: {}",
+                    queue, manager.active_file_seq, e
+                );
+            }
+        } else if manager.active_file.since_last_sync() >= manager.bytes_per_sync || force {
+            if let Err(e) = manager.sync() {
+                panic!(
+                    "error when sync [{:?}:{}]: {}",
+                    queue, manager.active_file_seq, e,
+                );
+            }
         }
+
         Ok(())
     }
 
@@ -728,8 +736,8 @@ impl<B: FileBuilder> PipeLog for FilePipeLog<B> {
         self.get_queue(queue).size()
     }
 
-    fn new_log_file(&self, queue: LogQueue) -> Result<()> {
-        self.mut_queue(queue).new_log_file()
+    fn rotate(&self, queue: LogQueue) -> Result<()> {
+        self.mut_queue(queue).rotate()
     }
 
     fn purge_to(&self, file_id: FileId) -> Result<usize> {
@@ -860,34 +868,34 @@ mod tests {
 
         // generate file 1, 2, 3
         let content: Vec<u8> = vec![b'a'; 1024];
-        let file_handle = pipe_log.append_bytes(queue, &content, &mut false).unwrap();
+        let file_handle = pipe_log.mut_queue(queue).append(&content).unwrap();
+        pipe_log.maybe_sync(queue, false).unwrap();
         assert_eq!(file_handle.id.seq, 1);
         assert_eq!(file_handle.offset, header_size);
-        assert_eq!(pipe_log.file_span(queue).1, 1);
+        assert_eq!(pipe_log.file_span(queue).1, 2);
 
-        let file_handle = pipe_log.append_bytes(queue, &content, &mut false).unwrap();
+        let file_handle = pipe_log.mut_queue(queue).append(&content).unwrap();
+        pipe_log.maybe_sync(queue, false).unwrap();
         assert_eq!(file_handle.id.seq, 2);
         assert_eq!(file_handle.offset, header_size);
-        assert_eq!(pipe_log.file_span(queue).1, 2);
+        assert_eq!(pipe_log.file_span(queue).1, 3);
 
         // purge file 1
         assert_eq!(pipe_log.purge_to(FileId { queue, seq: 2 }).unwrap(), 1);
         assert_eq!(pipe_log.file_span(queue).0, 2);
 
         // cannot purge active file
-        assert!(pipe_log.purge_to(FileId { queue, seq: 3 }).is_err());
+        assert!(pipe_log.purge_to(FileId { queue, seq: 4 }).is_err());
 
         // append position
         let s_content = b"short content".to_vec();
-        let file_handle = pipe_log
-            .append_bytes(queue, &s_content, &mut false)
-            .unwrap();
+        let file_handle = pipe_log.mut_queue(queue).append(&s_content).unwrap();
+        pipe_log.maybe_sync(queue, false).unwrap();
         assert_eq!(file_handle.id.seq, 3);
         assert_eq!(file_handle.offset, header_size);
 
-        let file_handle = pipe_log
-            .append_bytes(queue, &s_content, &mut false)
-            .unwrap();
+        let file_handle = pipe_log.mut_queue(queue).append(&s_content).unwrap();
+        pipe_log.maybe_sync(queue, false).unwrap();
         assert_eq!(file_handle.id.seq, 3);
         assert_eq!(
             file_handle.offset,

src/log_file.rs

@@ -122,6 +122,9 @@ impl LogFd {
             written += bytes;
             offset += bytes;
         }
+        fail_point!("log_fd::write::post_err", |_| {
+            Err(from_nix_error(nix::Error::invalid_argument(), "fp"))
+        });
         Ok(written)
     }