Support recycling log files

src/engine.rs

@@ -1860,8 +1860,14 @@ mod tests {
             Ok(())
         }
 
-        fn rename<P: AsRef<Path>>(&self, src_path: P, dst_path: P) -> std::io::Result<()> {
-            self.inner.rename(src_path.as_ref(), dst_path.as_ref())?;
+        fn rename<P: AsRef<Path>>(
+            &self,
+            src_path: P,
+            dst_path: P,
+            keep_data: bool,
+        ) -> std::io::Result<()> {
+            self.inner
+                .rename(src_path.as_ref(), dst_path.as_ref(), keep_data)?;
             self.update_metadata(src_path.as_ref(), true);
             self.update_metadata(dst_path.as_ref(), false);
             Ok(())
@@ -2093,14 +2099,17 @@ mod tests {
             let engine = RaftLogEngine::open_with_file_system(cfg, fs).unwrap();
             let (start, end) = engine.file_span(LogQueue::Append);
             assert_eq!((start, end), (1, 11));
-            for rid in 1..=5 {
+            for rid in 6..=10 {
+                engine.append(rid, 11, 20, Some(&entry_data));
+            }
+            // Mark region_id -> 6 obsolete.
+            for rid in 6..=6 {
                 engine.clean(rid);
             }
-            // the [1, 11] files are recycled
+            // the [1, 12] files are recycled
             engine.purge_expired_files().unwrap();
-            assert!(start < engine.file_span(LogQueue::Append).0);
-            assert_eq!(start + 15, engine.file_span(LogQueue::Append).0);
-            assert_eq!(engine.file_count(Some(LogQueue::Append)), 1);
+            assert_eq!(engine.file_count(Some(LogQueue::Append)), 5);
+            assert_eq!(start + 12, engine.file_span(LogQueue::Append).0);
         }
     }
 }

src/env/default.rs

@@ -279,7 +279,8 @@ impl FileSystem for DefaultFileSystem {
         std::fs::remove_file(path)
     }
 
-    fn rename<P: AsRef<Path>>(&self, src_path: P, dst_path: P) -> IoResult<()> {
+    fn rename<P: AsRef<Path>>(&self, src_path: P, dst_path: P, keep_data: bool) -> IoResult<()> {
+        debug_assert!(!keep_data);
         std::fs::rename(src_path, dst_path)
     }
 

src/env/mod.rs

@@ -22,7 +22,13 @@ pub trait FileSystem: Send + Sync {
 
     fn delete<P: AsRef<Path>>(&self, path: P) -> Result<()>;
 
-    fn rename<P: AsRef<Path>>(&self, src_path: P, dst_path: P) -> Result<()>;
+    /// Rename the file from `src_path` to `dst_path`.
+    ///
+    /// `keep_data` is introduced to be compatible with user-defined
+    /// `FileSystem`, which means whether the data are allowed to be
+    /// modified by user-defined implementation on `rename` or not.
+    /// Defautly, `keep_data` is `false`.
+    fn rename<P: AsRef<Path>>(&self, src_path: P, dst_path: P, keep_data: bool) -> Result<()>;
 
     /// Deletes user implemented metadata associated with `path`. Returns
     /// `true` if any metadata is deleted.

src/env/obfuscated.rs

@@ -114,8 +114,9 @@ impl FileSystem for ObfuscatedFileSystem {
         r
     }
 
-    fn rename<P: AsRef<Path>>(&self, src_path: P, dst_path: P) -> IoResult<()> {
-        self.inner.rename(src_path, dst_path)
+    fn rename<P: AsRef<Path>>(&self, src_path: P, dst_path: P, keep_data: bool) -> IoResult<()> {
+        debug_assert!(!keep_data);
+        self.inner.rename(src_path, dst_path, keep_data)
     }
 
     fn new_reader(&self, handle: Arc<Self::Handle>) -> IoResult<Self::Reader> {

src/file_pipe_log/pipe.rs

@@ -62,16 +62,13 @@ impl<F: FileSystem> FileCollection<F> {
         if self.fds.pop_front().is_some() {
             let src_path = first_file_id.build_file_path(dir_path); // src filepath
             let dst_path = dst_fd.build_file_path(dir_path); // dst filepath
-            match file_system.rename(&src_path, &dst_path) {
-                Ok(_) => {
-                    // Update the first_seq
-                    self.first_seq += 1;
-                    ret = true;
-                }
-                Err(e) => {
-                    ret = false;
-                    error!("error while trying to recycle one expired file: {}", e);
-                }
+            if let Err(e) = file_system.rename(&src_path, &dst_path, false) {
+                error!("error while trying to recycle one expired file: {}", e);
+                ret = false;
+            } else {
+                // Update the first_seq
+                self.first_seq += 1;
+                ret = true;
             }
         }
         // Only if the `rename` made sense, we could return success.
@@ -385,11 +382,13 @@ impl<F: FileSystem> SinglePipe<F> {
         self.rotate_imp(&mut self.active_file.lock())
     }
 
+    /// Purge obsolete log files to the specific `FileSeq`.
+    ///
+    /// Return the actual removed count of purged files.
     fn purge_to(&self, file_seq: FileSeq) -> Result<usize> {
         let (
             first_purge_seq, /* first seq for purging */
             purged,          /* count of purged files */
-            recycled,        /* count of recycled files */
             remained,        /* count of remained files */
         ) = {
             let mut files = self.files.write();
@@ -399,39 +398,33 @@ impl<F: FileSystem> SinglePipe<F> {
                 return Ok(0);
             }
 
-            // If capacity == 0, `files` not support to `recycle`.
-            let logically_purged = if files.capacity == 0 {
-                0
-            } else {
-                (file_seq - files.first_seq_in_use) as usize
-            };
             // Remove some obsolete files if capacity is exceeded.
             let obsolete_files = (file_seq - files.first_seq) as usize;
             // When capacity is zero, always remove logically deleted files.
             let capacity_exceeded = files.fds.len().saturating_sub(files.capacity);
             let purged = std::cmp::min(capacity_exceeded, obsolete_files);
 
-            // The files marked with `recycle` but with format_version V1 should also
-            // be removed.
-            files.first_seq += purged as u64;
             let extra_purged = {
+                // The files with format_version `V1` cannot be chosen as recycle
+                // candidates, which should also be removed.
                 let mut count = 0;
-                for recycle_idx in 0..(file_seq - files.first_seq) as usize {
+                for recycle_idx in purged..obsolete_files {
                     if !files.fds[recycle_idx].context.version.has_log_signing() {
                         count += 1;
+                    } else {
+                        break;
                     }
                 }
                 count
             };
             let final_purge_count = purged + extra_purged;
             // Update metadata of files
-            files.first_seq += extra_purged as u64;
+            files.first_seq += final_purge_count as u64;
             files.first_seq_in_use = file_seq;
             files.fds.drain(..final_purge_count);
             (
                 files.first_seq - final_purge_count as u64,
                 final_purge_count,
-                logically_purged.saturating_sub(extra_purged),
                 files.fds.len(),
             )
         };
@@ -454,15 +447,13 @@ impl<F: FileSystem> SinglePipe<F> {
             }
             self.file_system.delete(&path)?;
         }
-        Ok(purged + recycled)
+        Ok(purged)
     }
 
     fn fetch_active_file(&self) -> LogFileContext {
         let files = self.files.read();
-        match files.fds.back() {
-            Some(active_fd) => LogFileContext::new(active_fd.context.id, active_fd.context.version),
-            _ => unreachable!(),
-        }
+        let active_fd = files.fds.back().unwrap();
+        LogFileContext::new(active_fd.context.id, active_fd.context.version)
     }
 }
 
@@ -875,8 +866,8 @@ mod tests {
         assert_eq!(file_handle.offset, header_size);
         assert_eq!(pipe_log.file_span(queue).1, 3);
 
-        // purge file 1
-        assert_eq!(pipe_log.purge_to(FileId { queue, seq: 2 }).unwrap(), 1);
+        // Purge to file 1, this file would be recycled
+        assert_eq!(pipe_log.purge_to(FileId { queue, seq: 2 }).unwrap(), 0);
         assert_eq!(pipe_log.file_span(queue).0, 2);
 
         // cannot purge active file

src/filter.rs

@@ -255,14 +255,14 @@ impl RhaiFilterMachine {
                 // Backup file and set up a guard to recover on exit.
                 let target_path = f.file_id.build_file_path(path);
                 let bak_path = target_path.with_extension("bak");
-                system.rename(&target_path, &bak_path)?;
+                system.rename(&target_path, &bak_path, false)?;
                 guards.push((
                     bak_path.clone(),
                     guard(f.file_id, |f| {
                         let original = f.build_file_path(path);
                         let bak = original.with_extension("bak");
                         if bak.exists() {
-                            system.rename(&bak, &original).unwrap_or_else(|e| {
+                            system.rename(&bak, &original, false).unwrap_or_else(|e| {
                                 panic!(
                                     "Failed to recover original log file {} ({}),
                                 you should manually replace it with {}.bak.",

src/log_batch.rs

@@ -421,21 +421,11 @@ impl LogItemBatch {
                 // Insert the signature into the encoded bytes. Rewrite checksum of
                 // `LogItemBatch` in `LogBatch`.
                 let footer_checksum_offset = buf.len() - LOG_BATCH_CHECKSUM_LEN;
-                match codec::decode_u32_le(&mut &buf[footer_checksum_offset..]) {
-                    // The final checksum is generated by `signature` ***XOR***
-                    // `original checksum of buf`.
-                    Ok(original_checksum) => {
-                        let _ = (&mut buf[footer_checksum_offset..])
-                            .write_u32::<LittleEndian>(original_checksum ^ signature);
-                        return Ok(());
-                    }
-                    _ => {
-                        return Err(Error::Corruption(format!(
-                            "Checksum in this LogBatch is corrputed, footer_checksum_offset: {}",
-                            footer_checksum_offset
-                        )));
-                    }
-                }
+                let original_checksum = codec::decode_u32_le(&mut &buf[footer_checksum_offset..])?;
+                // The final checksum is generated by `signature` ***XOR***
+                // `original checksum of buf`.
+                (&mut buf[footer_checksum_offset..])
+                    .write_u32::<LittleEndian>(original_checksum ^ signature)?;
             }
         }
         Ok(())
@@ -551,6 +541,12 @@ enum BufState {
     /// # Invariants
     /// LOG_BATCH_HEADER_LEN <= buf.len()
     Open,
+    /// Buffer contains header, entries and footer; ready to be written. The
+    /// footer may be signed with extra information depending on the format
+    /// version.
+    /// # Content
+    /// (header_offset, entries_len)
+    Encoded(usize, usize),
     /// Buffer contains header, entries and footer; ready to be written. This
     /// state only briefly exists between encoding and writing, user operation
     /// will panic under this state.
@@ -559,12 +555,6 @@ enum BufState {
     /// # Invariants
     /// LOG_BATCH_HEADER_LEN <= buf.len()
     Sealed(usize, usize),
-    /// Buffer contains header, entries and footer; ready to be written. The
-    /// footer may be signed with extra information depending on the format
-    /// version.
-    /// # Content
-    /// (header_offset, entries_len)
-    Encoded(usize, usize),
     /// Buffer is undergoing writes. User operation will panic under this state.
     Incomplete,
 }
@@ -739,7 +729,7 @@ impl LogBatch {
         let _t = StopWatch::new(perf_context!(log_populating_duration));
         debug_assert!(self.buf_state == BufState::Open);
         if self.is_empty() {
-            self.buf_state = BufState::Sealed(self.buf.len(), 0);
+            self.buf_state = BufState::Encoded(self.buf.len(), 0);
             return Ok(0);
         }
         self.buf_state = BufState::Incomplete;
@@ -792,19 +782,17 @@ impl LogBatch {
             }
         }
 
-        self.buf_state = BufState::Sealed(header_offset, footer_roffset - LOG_BATCH_HEADER_LEN);
+        self.buf_state = BufState::Encoded(header_offset, footer_roffset - LOG_BATCH_HEADER_LEN);
         Ok(self.buf.len() - header_offset)
     }
 
     /// Make preparations for the write of `LogBatch`.
     #[inline]
     pub(crate) fn prepare_write(&mut self, file_context: &LogFileContext) -> Result<()> {
         match self.buf_state {
-            BufState::Sealed(header_offset, entries_len) => {
-                if let Err(e) = LogItemBatch::prepare_write(&mut self.buf, file_context) {
-                    return Err(e);
-                }
-                self.buf_state = BufState::Encoded(header_offset, entries_len);
+            BufState::Encoded(header_offset, entries_len) => {
+                LogItemBatch::prepare_write(&mut self.buf, file_context)?;
+                self.buf_state = BufState::Sealed(header_offset, entries_len);
             }
             _ => unreachable!(),
         }
@@ -815,7 +803,7 @@ impl LogBatch {
     /// Assumes called after a successful call of [`prepare_write`].
     pub(crate) fn encoded_bytes(&self) -> &[u8] {
         match self.buf_state {
-            BufState::Encoded(header_offset, _) => &self.buf[header_offset..],
+            BufState::Sealed(header_offset, _) => &self.buf[header_offset..],
             _ => unreachable!(),
         }
     }
@@ -824,12 +812,12 @@ impl LogBatch {
     ///
     /// Internally sets the file locations of each log entry indexes.
     pub(crate) fn finish_write(&mut self, mut handle: FileBlockHandle) {
-        debug_assert!(matches!(self.buf_state, BufState::Encoded(_, _)));
+        debug_assert!(matches!(self.buf_state, BufState::Sealed(_, _)));
         if !self.is_empty() {
             // adjust log batch handle to log entries handle.
             handle.offset += LOG_BATCH_HEADER_LEN as u64;
             match self.buf_state {
-                BufState::Encoded(_, entries_len) => {
+                BufState::Sealed(_, entries_len) => {
                     debug_assert!(LOG_BATCH_HEADER_LEN + entries_len < handle.len as usize);
                     handle.len = entries_len;
                 }
@@ -859,8 +847,8 @@ impl LogBatch {
                 BufState::Open => {
                     self.buf.len() + LOG_BATCH_CHECKSUM_LEN + self.item_batch.approximate_size()
                 }
-                BufState::Sealed(header_offset, _) => self.buf.len() - header_offset,
                 BufState::Encoded(header_offset, _) => self.buf.len() - header_offset,
+                BufState::Sealed(header_offset, _) => self.buf.len() - header_offset,
                 s => {
                     error!("querying incomplete log batch with state {:?}", s);
                     0