commitlog: Add a test that triggers the panic when commit.write fails

crates/commitlog/src/repo/mod.rs

@@ -424,7 +424,7 @@ pub fn resume_segment_writer<R: Repo>(
                 records: Vec::new(),
                 epoch: meta.max_epoch,
             },
-            inner: io::BufWriter::new(writer),
+            inner: io::BufWriter::with_capacity(opts.write_buffer_size, writer),
 
             min_tx_offset: meta.tx_range.start,
             bytes_written: meta.size_in_bytes,

crates/commitlog/src/segment.rs

@@ -144,10 +144,19 @@ impl<W: io::Write> Writer<W> {
         let checksum = self
             .commit
             .write(&mut self.inner)
-            // Panic here as we don't know how much of the commit has been
-            // written (if anything). Further commits would leave corrupted data
-            // in the log.
-            .unwrap_or_else(|e| panic!("failed to write commit {}: {:#}", self.commit.min_tx_offset, e));
+            // Panic here as the data remaining in the `BufWriter`'s buffer is
+            // probably not on a commit boundary. If the caller rotates segments
+            // and continues to commit, we'd write a torn commit.
+            //
+            // Panicking ensures that the buffer is discarded, and that the
+            // commitlog is reopened to recover the last durable commit.
+            .unwrap_or_else(|e| {
+                let buffered = self.inner.buffer().len();
+                panic!(
+                    "failed to write commit {} with {} bytes remaining buffered: {:#}",
+                    self.commit.min_tx_offset, buffered, e
+                )
+            });
         let commit_len = self.commit.encoded_len() as u64;
 
         if let Some(index) = self.offset_index_head.as_mut() {

crates/commitlog/src/tests/partial.rs

@@ -30,6 +30,25 @@ fn panics_on_partial_write() {
     }
 }
 
+#[test]
+#[should_panic(expected = "failed to write commit")]
+fn panics_on_partial_buffer_write() {
+    enable_logging();
+
+    let mut log = open_log_with_options::<[u8; 32]>(
+        ShortMem::new(800),
+        Options {
+            max_segment_size: 1024,
+            write_buffer_size: 64,
+            ..<_>::default()
+        },
+    );
+    for i in 0..20 {
+        info!("commit {i}");
+        log.commit([(i, [b'z'; 32])]).expect("unexpected `Err` result");
+    }
+}
+
 fn fill_log(mut log: commitlog::Generic<ShortMem, [u8; 32]>, range: Range<TxOffset>) {
     debug!("writing range {range:?}");
 
@@ -167,14 +186,17 @@ fn first_commit_in_last_segment_corrupt() {
 }
 
 fn open_log<T>(repo: ShortMem) -> commitlog::Generic<ShortMem, T> {
-    commitlog::Generic::open(
+    open_log_with_options(
         repo,
         Options {
             max_segment_size: 1024,
             ..Options::default()
         },
     )
-    .unwrap()
+}
+
+fn open_log_with_options<T>(repo: ShortMem, options: Options) -> commitlog::Generic<ShortMem, T> {
+    commitlog::Generic::open(repo, options).unwrap()
 }
 
 const ENOSPC: i32 = 28;