Revert "Append commit instead of individual transactions to commitlog (#4140)" (#4292)

Reverts #4140 per @kim's request — was not ready to merge yet.

Co-authored-by: clockwork-labs-bot <clockwork-labs-bot@users.noreply.github.com>

Author: clockwork-labs-bot

crates/commitlog/src/commitlog.rs

@@ -1,6 +1,6 @@
 use std::{
     io,
-    num::NonZeroU64,
+    num::{NonZeroU16, NonZeroU64},
     ops::RangeBounds,
     sync::{Arc, RwLock},
 };
@@ -17,11 +17,10 @@ pub mod segment;
 mod varchar;
 mod varint;
 
-use crate::segment::Committed;
 pub use crate::{
     commit::{Commit, StoredCommit},
     payload::{Decoder, Encode},
-    repo::{fs::SizeOnDisk, TxOffset},
+    repo::fs::SizeOnDisk,
     segment::{Transaction, DEFAULT_LOG_FORMAT_VERSION},
     varchar::Varchar,
 };
@@ -58,6 +57,14 @@ pub struct Options {
     /// Default: 1GiB
     #[cfg_attr(feature = "serde", serde(default = "Options::default_max_segment_size"))]
     pub max_segment_size: u64,
+    /// The maximum number of records in a commit.
+    ///
+    /// If this number is exceeded, the commit is flushed to disk even without
+    /// explicitly calling [`Commitlog::flush`].
+    ///
+    /// Default: 1
+    #[cfg_attr(feature = "serde", serde(default = "Options::default_max_records_in_commit"))]
+    pub max_records_in_commit: NonZeroU16,
     /// Whenever at least this many bytes have been written to the currently
     /// active segment, an entry is added to its offset index.
     ///
@@ -89,12 +96,6 @@ pub struct Options {
     /// Has no effect if the `fallocate` feature is not enabled.
     #[cfg_attr(feature = "serde", serde(default = "Options::default_preallocate_segments"))]
     pub preallocate_segments: bool,
-    /// Size in bytes of the memory buffer holding commit data before flushing
-    /// to storage.
-    ///
-    /// Default: 4KiB
-    #[cfg_attr(feature = "serde", serde(default = "Options::default_write_buffer_size"))]
-    pub write_buffer_size: usize,
 }
 
 impl Default for Options {
@@ -105,18 +106,18 @@ impl Default for Options {
 
 impl Options {
     pub const DEFAULT_MAX_SEGMENT_SIZE: u64 = 1024 * 1024 * 1024;
+    pub const DEFAULT_MAX_RECORDS_IN_COMMIT: NonZeroU16 = NonZeroU16::new(1).expect("1 > 0, qed");
     pub const DEFAULT_OFFSET_INDEX_INTERVAL_BYTES: NonZeroU64 = NonZeroU64::new(4096).expect("4096 > 0, qed");
     pub const DEFAULT_OFFSET_INDEX_REQUIRE_SEGMENT_FSYNC: bool = false;
     pub const DEFAULT_PREALLOCATE_SEGMENTS: bool = false;
-    pub const DEFAULT_WRITE_BUFFER_SIZE: usize = 4 * 1024;
 
     pub const DEFAULT: Self = Self {
         log_format_version: DEFAULT_LOG_FORMAT_VERSION,
         max_segment_size: Self::default_max_segment_size(),
+        max_records_in_commit: Self::default_max_records_in_commit(),
         offset_index_interval_bytes: Self::default_offset_index_interval_bytes(),
         offset_index_require_segment_fsync: Self::default_offset_index_require_segment_fsync(),
         preallocate_segments: Self::default_preallocate_segments(),
-        write_buffer_size: Self::default_write_buffer_size(),
     };
 
     pub const fn default_log_format_version() -> u8 {
@@ -127,6 +128,10 @@ impl Options {
         Self::DEFAULT_MAX_SEGMENT_SIZE
     }
 
+    pub const fn default_max_records_in_commit() -> NonZeroU16 {
+        Self::DEFAULT_MAX_RECORDS_IN_COMMIT
+    }
+
     pub const fn default_offset_index_interval_bytes() -> NonZeroU64 {
         Self::DEFAULT_OFFSET_INDEX_INTERVAL_BYTES
     }
@@ -139,10 +144,6 @@ impl Options {
         Self::DEFAULT_PREALLOCATE_SEGMENTS
     }
 
-    pub const fn default_write_buffer_size() -> usize {
-        Self::DEFAULT_WRITE_BUFFER_SIZE
-    }
-
     /// Compute the length in bytes of an offset index based on the settings in
     /// `self`.
     pub fn offset_index_len(&self) -> u64 {
@@ -261,7 +262,7 @@ impl<T> Commitlog<T> {
     pub fn flush(&self) -> io::Result<Option<u64>> {
         let mut inner = self.inner.write().unwrap();
         trace!("flush commitlog");
-        inner.flush()?;
+        inner.commit()?;
 
         Ok(inner.max_committed_offset())
     }
@@ -281,7 +282,7 @@ impl<T> Commitlog<T> {
     pub fn flush_and_sync(&self) -> io::Result<Option<u64>> {
         let mut inner = self.inner.write().unwrap();
         trace!("flush and sync commitlog");
-        inner.flush()?;
+        inner.commit()?;
         inner.sync();
 
         Ok(inner.max_committed_offset())
@@ -382,47 +383,57 @@ impl<T> Commitlog<T> {
 }
 
 impl<T: Encode> Commitlog<T> {
-    /// Write `transactions` to the log.
-    ///
-    /// This will store all `transactions` as a single [Commit]
-    /// (note that `transactions` must not yield more than [u16::MAX] elements).
-    ///
-    /// Data is buffered internally, call [Self::flush] to force flushing to
-    /// the underlying storage.
-    ///
-    /// Returns `Ok(None)` if `transactions` was empty, otherwise [Committed],
-    /// which contains the offset range and checksum of the commit.
-    ///
-    /// # Errors
+    /// Append the record `txdata` to the log.
     ///
-    /// An `Err` value is returned in the following cases:
+    /// If the internal buffer exceeds [`Options::max_records_in_commit`], the
+    /// argument is returned in an `Err`. The caller should [`Self::flush`] the
+    /// log and try again.
     ///
-    /// - if the transaction sequence is invalid, e.g. because the transaction
-    ///   offsets are not contiguous.
+    /// In case the log is appended to from multiple threads, this may result in
+    /// a busy loop trying to acquire a slot in the buffer. In such scenarios,
+    /// [`Self::append_maybe_flush`] is preferable.
+    pub fn append(&self, txdata: T) -> Result<(), T> {
+        let mut inner = self.inner.write().unwrap();
+        inner.append(txdata)
+    }
+
+    /// Append the record `txdata` to the log.
     ///
-    ///   In this case, **none** of the `transactions` will be written.
+    /// The `txdata` payload is buffered in memory until either:
     ///
-    /// - if creating the new segment fails due to an I/O error.
+    /// - [`Self::flush`] is called explicitly, or
+    /// - [`Options::max_records_in_commit`] is exceeded
     ///
-    /// # Panics
+    /// In the latter case, [`Self::append`] flushes implicitly, _before_
+    /// appending the `txdata` argument.
     ///
-    /// The method panics if:
+    /// I.e. the argument is not guaranteed to be flushed after the method
+    /// returns. If that is desired, [`Self::flush`] must be called explicitly.
     ///
-    /// - `transactions` exceeds [u16::MAX] elements
+    /// If writing `txdata` to the commitlog results in a new segment file being opened,
+    /// we will send a message down `on_new_segment`.
+    /// This will be hooked up to the `request_snapshot` channel of a `SnapshotWorker`.
     ///
-    /// - [Self::flush] or writing to the underlying buffered writer fails
+    /// # Errors
     ///
-    ///   This is likely caused by some storage issue. As we cannot tell with
-    ///   certainty how much data (if any) has been written, the internal state
-    ///   becomes invalid and thus a panic is raised.
+    /// If the log needs to be flushed, but an I/O error occurs, ownership of
+    /// `txdata` is returned back to the caller alongside the [`io::Error`].
     ///
-    /// - [Self::sync] panics (called when rotating segments)
-    pub fn commit<U: Into<Transaction<T>>>(
-        &self,
-        transactions: impl IntoIterator<Item = U>,
-    ) -> io::Result<Option<Committed>> {
+    /// The value can then be used to retry appending.
+    pub fn append_maybe_flush(&self, txdata: T) -> Result<(), error::Append<T>> {
         let mut inner = self.inner.write().unwrap();
-        inner.commit(transactions)
+
+        if let Err(txdata) = inner.append(txdata) {
+            if let Err(source) = inner.commit() {
+                return Err(error::Append { txdata, source });
+            }
+
+            // `inner.commit.n` must be zero at this point
+            let res = inner.append(txdata);
+            debug_assert!(res.is_ok(), "failed to append while holding write lock");
+        }
+
+        Ok(())
     }
 
     /// Obtain an iterator which traverses the log from the start, yielding

crates/commitlog/src/lib.rs

@@ -214,11 +214,13 @@ pub fn create_segment_writer<R: Repo>(
             records: Vec::new(),
             epoch,
         },
-        inner: io::BufWriter::with_capacity(opts.write_buffer_size, storage),
+        inner: io::BufWriter::new(storage),
 
         min_tx_offset: offset,
         bytes_written: Header::LEN as u64,
 
+        max_records_in_commit: opts.max_records_in_commit,
+
         offset_index_head: create_offset_index_writer(repo, offset, opts),
     })
 }
@@ -291,6 +293,8 @@ pub fn resume_segment_writer<R: Repo>(
         min_tx_offset: tx_range.start,
         bytes_written: size_in_bytes,
 
+        max_records_in_commit: opts.max_records_in_commit,
+
         offset_index_head: create_offset_index_writer(repo, offset, opts),
     }))
 }