commitlog: Include latest commit offset in segment metadata (#2733)

Author: kim

crates/commitlog/src/repo/mod.rs

@@ -243,6 +243,7 @@ pub fn resume_segment_writer<R: Repo>(
         tx_range,
         size_in_bytes,
         max_epoch,
+        max_commit_offset: _,
     } = match Metadata::extract(offset, &mut storage, offset_index.as_ref()) {
         Err(error::SegmentMetadata::InvalidCommit { sofar, source }) => {
             warn!("invalid commit in segment {offset}: {source}");

crates/commitlog/src/segment.rs

@@ -526,14 +526,24 @@ impl<R: io::BufRead> Iterator for CommitsWithVersion<R> {
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct Metadata {
+    /// The segment header.
     pub header: Header,
+    /// The range of transactions contained in the segment.
     pub tx_range: Range<u64>,
+    /// The size of the segment.
     pub size_in_bytes: u64,
+    /// The largest epoch found in the segment.
     pub max_epoch: u64,
+    /// The latest commit found in the segment.
+    ///
+    /// The value is the `min_tx_offset` of the commit, i.e.
+    /// `max_commit_offset..tx_range.end` is the range of
+    /// transactions contained in it.
+    pub max_commit_offset: u64,
 }
 
 impl Metadata {
-    /// Reads and validates metadata from a segment.  
+    /// Reads and validates metadata from a segment.
     /// It will look for last commit index offset and then traverse the segment
     ///
     /// Determines `max_tx_offset`, `size_in_bytes`, and `max_epoch` from the segment.
@@ -562,6 +572,7 @@ impl Metadata {
                 },
                 size_in_bytes: Header::LEN as u64,
                 max_epoch: u64::default(),
+                max_commit_offset: min_tx_offset,
             });
 
         reader.seek(SeekFrom::Start(sofar.size_in_bytes))?;
@@ -597,6 +608,7 @@ impl Metadata {
             sofar.size_in_bytes += commit.size_in_bytes;
             // TODO: Should it be an error to encounter an epoch going backwards?
             sofar.max_epoch = commit.epoch.max(sofar.max_epoch);
+            sofar.max_commit_offset = commit.tx_range.start;
         }
 
         Ok(sofar)
@@ -628,6 +640,7 @@ impl Metadata {
                         },
                         size_in_bytes: byte_offset + commit.size_in_bytes,
                         max_epoch: commit.epoch,
+                        max_commit_offset: commit.tx_range.start,
                     });
                 }
 
@@ -734,28 +747,31 @@ mod tests {
         let repo = repo::Memory::default();
 
         let mut writer = repo::create_segment_writer(&repo, Options::default(), Commit::DEFAULT_EPOCH, 0).unwrap();
+        // Commit 0..2
         writer.append([0; 32]).unwrap();
         writer.append([0; 32]).unwrap();
         writer.commit().unwrap();
+        // Commit 2..3
         writer.append([1; 32]).unwrap();
         writer.commit().unwrap();
+        // Commit 3..5
         writer.append([2; 32]).unwrap();
         writer.append([2; 32]).unwrap();
         writer.commit().unwrap();
 
         let reader = repo::open_segment_reader(&repo, DEFAULT_LOG_FORMAT_VERSION, 0).unwrap();
-        let Metadata {
-            header: _,
-            tx_range,
-            size_in_bytes,
-            max_epoch: _,
-        } = reader.metadata().unwrap();
-
-        assert_eq!(tx_range.start, 0);
-        assert_eq!(tx_range.end, 5);
+        let metadata = reader.metadata().unwrap();
+
         assert_eq!(
-            size_in_bytes,
-            (Header::LEN + (5 * 32) + (3 * Commit::FRAMING_LEN)) as u64
+            metadata,
+            Metadata {
+                header: Header::default(),
+                tx_range: Range { start: 0, end: 5 },
+                // header + 5 txs + 3 commits
+                size_in_bytes: (Header::LEN + (5 * 32) + (3 * Commit::FRAMING_LEN)) as u64,
+                max_epoch: Commit::DEFAULT_EPOCH,
+                max_commit_offset: 3
+            }
         );
     }