[pull] master from clockworklabs:master

crates/commitlog/src/commitlog.rs

@@ -1,6 +1,6 @@
 use std::{
     fmt::Debug,
-    io,
+    io::{self, Seek},
     marker::PhantomData,
     mem,
     ops::{Range, RangeBounds},
@@ -748,6 +748,8 @@ fn reset_to_internal(repo: &impl Repo, segments: &[u64], offset: u64) -> io::Res
                 }
 
                 file.ftruncate(offset, byte_offset)?;
+                // We should be reopening the log anyway, but just in case.
+                file.seek(io::SeekFrom::End(0))?;
                 // Some filesystems require fsync after ftruncate.
                 file.fsync()?;
                 break;

crates/commitlog/src/segment.rs

@@ -498,7 +498,7 @@ pub fn seek_to_offset<R: io::Read + io::Seek>(
     debug_assert!(index_key <= start_tx_offset);
 
     // Check if the offset index is pointing to the right commit.
-    let hdr = validate_commit_header(&mut segment, byte_offset)?;
+    let hdr = validate_commit_at_byte_offset(&mut segment, byte_offset)?;
     if hdr.min_tx_offset == index_key {
         // Advance the segment Seek if expected commit is found.
         segment.seek(SeekFrom::Start(byte_offset))
@@ -513,20 +513,39 @@ pub fn seek_to_offset<R: io::Read + io::Seek>(
 
 /// Try to extract the commit header from the asked position without advancing seek.
 /// `IndexFileMut` fsync asynchoronously, which makes it important for reader to verify its entry
-pub fn validate_commit_header<Reader: io::Read + io::Seek>(
+fn validate_commit_at_byte_offset<Reader: io::Read + io::Seek>(
     mut reader: &mut Reader,
     byte_offset: u64,
 ) -> io::Result<commit::Header> {
     let pos = reader.stream_position()?;
     reader.seek(SeekFrom::Start(byte_offset))?;
 
-    let hdr = commit::Header::decode(&mut reader)
-        .and_then(|hdr| hdr.ok_or_else(|| io::Error::new(ErrorKind::UnexpectedEof, "unexpected EOF")));
+    let hdr_or_error = StoredCommit::decode(&mut reader).and_then(|maybe_commit| {
+        let StoredCommit {
+            min_tx_offset,
+            epoch,
+            n,
+            records,
+            ..
+        } = maybe_commit.ok_or_else(|| {
+            io::Error::new(
+                ErrorKind::UnexpectedEof,
+                format!("unexpected EOF while validating commit at byte offset {byte_offset}"),
+            )
+        })?;
+
+        Ok(commit::Header {
+            min_tx_offset,
+            epoch,
+            n,
+            len: records.len() as u32,
+        })
+    });
 
     // Restore the original position
     reader.seek(SeekFrom::Start(pos))?;
 
-    hdr
+    hdr_or_error
 }
 
 /// Pair of transaction offset and payload.

crates/commitlog/tests/random_payload/mod.rs

@@ -78,6 +78,37 @@ fn resets() {
     }
 }
 
+#[test]
+fn reset_then_resume() {
+    enable_logging();
+
+    let root = tempdir().unwrap();
+    let mut clog = Commitlog::open(CommitLogDir::from_path_unchecked(root.path()), <_>::default(), None).unwrap();
+
+    let payload = gen_payload();
+    for i in 0..10 {
+        clog.commit([(i, payload)]).unwrap();
+    }
+    clog.flush_and_sync().unwrap();
+
+    clog = clog.reset_to(4).unwrap();
+
+    let payload2 = gen_payload();
+    for i in 5..10 {
+        clog.commit([(i, payload2)]).unwrap();
+    }
+    clog.flush_and_sync().unwrap();
+
+    let txs = clog
+        .transactions(&payload::ArrayDecoder)
+        .map(Result::unwrap)
+        .collect::<Vec<_>>();
+
+    assert_eq!(txs.len(), 10);
+    assert!(txs[..5].iter().all(|tx| tx.txdata == payload));
+    assert!(txs[5..].iter().all(|tx| tx.txdata == payload2));
+}
+
 /// Try to generate commitlogs that will be amenable to compression -
 /// random data doesn't compress well, so try and have there be repetition
 fn compressible_payloads() -> impl Iterator<Item = [u8; 256]> {
@@ -221,21 +252,25 @@ fn resume_small_trailing_garbage() {
         }
     }
 
+    let last_segment_offset = {
+        let segments = repo.existing_offsets().unwrap();
+        segments.last().copied().unwrap()
+    };
+
     // Add some extra bytes, less than the commit header length.
     let last_segment_size = {
-        let segments = repo.existing_offsets().unwrap();
-        let mut last_segment = repo.open_segment_writer(segments.last().copied().unwrap()).unwrap();
+        let mut last_segment = repo.open_segment_writer(last_segment_offset).unwrap();
         last_segment.write_all(&[67u8; commit::Header::LEN - 1]).unwrap();
         last_segment.flush().unwrap();
         last_segment.sync_all().unwrap();
         last_segment.segment_len().unwrap()
     };
     {
         let mut clog = commitlog::Generic::open(&repo, <_>::default()).unwrap();
-
+        // The segment list should be unchanged.
+        assert_eq!(&last_segment_offset, repo.existing_offsets().unwrap().last().unwrap());
         // The extra bytes should have been truncated away.
-        let segments = repo.existing_offsets().unwrap();
-        let mut last_segment = repo.open_segment_writer(segments.last().copied().unwrap()).unwrap();
+        let mut last_segment = repo.open_segment_writer(last_segment_offset).unwrap();
         assert_eq!(
             last_segment.segment_len().unwrap(),
             last_segment_size - (commit::Header::LEN - 1) as u64