diff --git a/crates/commitlog/src/repo/mod.rs b/crates/commitlog/src/repo/mod.rs index 49e674c2d4b..3ad26da70a4 100644 --- a/crates/commitlog/src/repo/mod.rs +++ b/crates/commitlog/src/repo/mod.rs @@ -424,7 +424,7 @@ pub fn resume_segment_writer( 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, diff --git a/crates/commitlog/src/segment.rs b/crates/commitlog/src/segment.rs index b29adf156c6..c810e2399ae 100644 --- a/crates/commitlog/src/segment.rs +++ b/crates/commitlog/src/segment.rs @@ -144,10 +144,19 @@ impl Writer { 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() { diff --git a/crates/commitlog/src/tests/partial.rs b/crates/commitlog/src/tests/partial.rs index 34671599a5f..24b204f346f 100644 --- a/crates/commitlog/src/tests/partial.rs +++ b/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, range: Range) { debug!("writing range {range:?}"); @@ -167,14 +186,17 @@ fn first_commit_in_last_segment_corrupt() { } fn open_log(repo: ShortMem) -> commitlog::Generic { - commitlog::Generic::open( + open_log_with_options( repo, Options { max_segment_size: 1024, ..Options::default() }, ) - .unwrap() +} + +fn open_log_with_options(repo: ShortMem, options: Options) -> commitlog::Generic { + commitlog::Generic::open(repo, options).unwrap() } const ENOSPC: i32 = 28;