[runtime/utils/buffer/write] optimistically use Blob::write_at_sync

andresilva · andresilva · commit 7e7fccac9384 · 2026-05-21T15:56:15.000+01:00
diff --git a/runtime/src/utils/buffer/mod.rs b/runtime/src/utils/buffer/mod.rs
@@ -128,6 +128,114 @@ mod tests {
         }
     }
 
+    #[derive(Default)]
+    struct RangeSyncState {
+        /// All data currently stored in the blob.
+        ///
+        /// This includes every durable byte plus any newer bytes that have not
+        /// been made durable yet.
+        data: Vec<u8>,
+
+        /// Prefix/ranges of `data` that would survive a crash.
+        durable: Vec<u8>,
+
+        /// Number of full sync barriers.
+        full_syncs: usize,
+
+        /// Number of range-scoped write syncs.
+        range_syncs: usize,
+    }
+
+    /// Test blob with separate visible and durable state.
+    ///
+    /// Plain writes and resizes only update `data`. `write_at_sync` updates `data`
+    /// and then copies only that submitted range into `durable`. `sync` copies all
+    /// of `data` to `durable`. This lets tests assert that `Write::sync` uses range
+    /// sync only when no earlier unsynced mutation needs a full durability barrier.
+    #[derive(Clone)]
+    struct RangeSyncBlob {
+        state: Arc<Mutex<RangeSyncState>>,
+    }
+
+    impl RangeSyncBlob {
+        fn new() -> Self {
+            Self {
+                state: Arc::new(Mutex::new(RangeSyncState::default())),
+            }
+        }
+
+        fn snapshot(&self) -> (Vec<u8>, usize, usize) {
+            let state = self.state.lock();
+            (state.durable.clone(), state.full_syncs, state.range_syncs)
+        }
+
+        fn write(data: &mut Vec<u8>, offset: u64, buf: &[u8]) -> Result<(), Error> {
+            let start = usize::try_from(offset).map_err(|_| Error::OffsetOverflow)?;
+            let end = start.checked_add(buf.len()).ok_or(Error::OffsetOverflow)?;
+            if end > data.len() {
+                data.resize(end, 0);
+            }
+            data[start..end].copy_from_slice(buf);
+            Ok(())
+        }
+    }
+
+    impl crate::Blob for RangeSyncBlob {
+        async fn read_at(&self, offset: u64, len: usize) -> Result<IoBufsMut, Error> {
+            self.read_at_buf(offset, len, IoBufMut::default()).await
+        }
+
+        async fn read_at_buf(
+            &self,
+            offset: u64,
+            len: usize,
+            buf: impl Into<IoBufsMut> + Send,
+        ) -> Result<IoBufsMut, Error> {
+            let start = usize::try_from(offset).map_err(|_| Error::OffsetOverflow)?;
+            let end = start.checked_add(len).ok_or(Error::OffsetOverflow)?;
+            let state = self.state.lock();
+            if end > state.data.len() {
+                return Err(Error::BlobInsufficientLength);
+            }
+
+            let mut out = buf.into();
+            out.put_slice(&state.data[start..end]);
+            Ok(out)
+        }
+
+        async fn write_at(&self, offset: u64, buf: impl Into<IoBufs> + Send) -> Result<(), Error> {
+            let buf = buf.into().coalesce();
+            let mut state = self.state.lock();
+            Self::write(&mut state.data, offset, buf.as_ref())
+        }
+
+        async fn write_at_sync(
+            &self,
+            offset: u64,
+            buf: impl Into<IoBufs> + Send,
+        ) -> Result<(), Error> {
+            let buf = buf.into().coalesce();
+            let mut state = self.state.lock();
+            Self::write(&mut state.data, offset, buf.as_ref())?;
+            Self::write(&mut state.durable, offset, buf.as_ref())?;
+            state.range_syncs += 1;
+            Ok(())
+        }
+
+        async fn resize(&self, len: u64) -> Result<(), Error> {
+            let len = usize::try_from(len).map_err(|_| Error::OffsetOverflow)?;
+            self.state.lock().data.resize(len, 0);
+            Ok(())
+        }
+
+        async fn sync(&self) -> Result<(), Error> {
+            let mut state = self.state.lock();
+            state.durable = state.data.clone();
+            state.full_syncs += 1;
+            Ok(())
+        }
+    }
+
     #[test_traced]
     fn test_read_basic() {
         let executor = deterministic::Runner::default();
@@ -1329,7 +1437,7 @@ mod tests {
     }
 
     #[test_traced]
-    fn test_resize_then_append_at_size() {
+    fn test_write_resize_then_append_at_size() {
         let executor = deterministic::Runner::default();
         executor.start(|context| async move {
             // Test truncating, then appending at the new size
@@ -1375,4 +1483,75 @@ mod tests {
             assert_eq!(read.as_ref(), b"01234XXXXX");
         });
     }
+
+    #[test_traced]
+    fn test_write_sync_uses_range_sync_for_buffer_only_write() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let blob = RangeSyncBlob::new();
+            let writer = Write::from_pooler(&context, blob.clone(), 0, NZUsize!(8));
+
+            // The write remains entirely buffered, so sync can make just this range durable.
+            writer.write_at(0, b"abc").await.unwrap();
+            writer.sync().await.unwrap();
+
+            // No prior plain blob mutation required a full sync barrier.
+            let (durable, full_syncs, range_syncs) = blob.snapshot();
+            assert_eq!(durable.as_slice(), b"abc");
+            assert_eq!(full_syncs, 0);
+            assert_eq!(range_syncs, 1);
+        });
+    }
+
+    #[test_traced]
+    fn test_write_sync_persists_prior_direct_flushes_with_buffered_tip() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let blob = RangeSyncBlob::new();
+            let writer = Write::from_pooler(&context, blob.clone(), 0, NZUsize!(4));
+
+            // This exceeds the buffer and forces a plain write before the final buffered tip.
+            writer.write_at(0, b"abcdef").await.unwrap();
+            writer.write_at(6, b"g").await.unwrap();
+            writer.sync().await.unwrap();
+
+            // The final sync must cover both the prior plain write and the buffered tip.
+            let (durable, full_syncs, range_syncs) = blob.snapshot();
+            assert_eq!(durable.as_slice(), b"abcdefg");
+            assert_eq!(full_syncs, 1);
+            assert_eq!(range_syncs, 0);
+
+            // After the full sync, the next buffer-only write can use range sync again.
+            writer.write_at(7, b"h").await.unwrap();
+            writer.sync().await.unwrap();
+
+            let (durable, full_syncs, range_syncs) = blob.snapshot();
+            assert_eq!(durable.as_slice(), b"abcdefgh");
+            assert_eq!(full_syncs, 1);
+            assert_eq!(range_syncs, 1);
+        });
+    }
+
+    #[test_traced]
+    fn test_write_sync_uses_full_sync_after_resize() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let blob = RangeSyncBlob::new();
+            let writer = Write::from_pooler(&context, blob.clone(), 0, NZUsize!(8));
+
+            // Establish already-durable data with a range sync.
+            writer.write_at(0, b"abcdef").await.unwrap();
+            writer.sync().await.unwrap();
+
+            // Resize alone is an unsynced blob mutation.
+            writer.resize(4).await.unwrap();
+            writer.sync().await.unwrap();
+
+            // The resized contents require a full sync barrier to become durable.
+            let (durable, full_syncs, range_syncs) = blob.snapshot();
+            assert_eq!(durable.as_slice(), b"abcd");
+            assert_eq!(full_syncs, 1);
+            assert_eq!(range_syncs, 1);
+        });
+    }
 }
diff --git a/runtime/src/utils/buffer/write.rs b/runtime/src/utils/buffer/write.rs
@@ -2,6 +2,20 @@ use crate::{buffer::tip::Buffer, Blob, Buf, BufferPool, BufferPooler, Error, IoB
 use commonware_utils::sync::AsyncRwLock;
 use std::{num::NonZeroUsize, sync::Arc};
 
+/// Buffered tip data and durability bookkeeping.
+struct State {
+    /// The buffer containing the data yet to be appended to the tip of the
+    /// underlying blob.
+    buffer: Buffer,
+
+    /// Whether prior blob mutation requires a full sync barrier.
+    ///
+    /// Set after plain [Blob::write_at] or [Blob::resize]. While set, [Write::sync]
+    /// cannot use [Blob::write_at_sync] because that only makes the current write
+    /// durable.
+    needs_sync: bool,
+}
+
 /// A writer that buffers the raw content of a [Blob] to optimize the performance of appending or
 /// updating data.
 ///
@@ -48,8 +62,8 @@ pub struct Write<B: Blob> {
     /// The underlying blob to write to.
     blob: B,
 
-    /// The buffer containing the data yet to be appended to the tip of the underlying blob.
-    buffer: Arc<AsyncRwLock<Buffer>>,
+    /// Shared tip buffer and durability state.
+    state: Arc<AsyncRwLock<State>>,
 }
 
 impl<B: Blob> Write<B> {
@@ -58,7 +72,10 @@ impl<B: Blob> Write<B> {
     pub fn new(blob: B, size: u64, capacity: NonZeroUsize, pool: BufferPool) -> Self {
         Self {
             blob,
-            buffer: Arc::new(AsyncRwLock::new(Buffer::new(size, capacity.get(), pool))),
+            state: Arc::new(AsyncRwLock::new(State {
+                buffer: Buffer::new(size, capacity.get(), pool),
+                needs_sync: false,
+            })),
         }
     }
 
@@ -75,10 +92,9 @@ impl<B: Blob> Write<B> {
     /// Returns the current logical size of the blob including any buffered data.
     ///
     /// This represents the total size of data that would be present after flushing.
-    #[allow(clippy::len_without_is_empty)]
     pub async fn size(&self) -> u64 {
-        let buffer = self.buffer.read().await;
-        buffer.size()
+        let state = self.state.read().await;
+        state.buffer.size()
     }
 
     /// Read exactly `len` immutable bytes starting at `offset`.
@@ -88,8 +104,9 @@ impl<B: Blob> Write<B> {
             .checked_add(len as u64)
             .ok_or(Error::OffsetOverflow)?;
 
-        // Acquire a read lock on the buffer.
-        let buffer = self.buffer.read().await;
+        // Acquire a read lock on the buffer state.
+        let state = self.state.read().await;
+        let buffer = &state.buffer;
 
         // If the data required is beyond the size of the blob, return an error.
         if end_offset > buffer.size() {
@@ -138,8 +155,8 @@ impl<B: Blob> Write<B> {
             .checked_add(bufs.remaining() as u64)
             .ok_or(Error::OffsetOverflow)?;
 
-        // Acquire a write lock on the buffer.
-        let mut buffer = self.buffer.write().await;
+        // Acquire a write lock on the buffer state.
+        let mut state = self.state.write().await;
 
         // Process each chunk of the input buffer, attempting to merge into the tip buffer
         // or writing directly to the underlying blob.
@@ -149,7 +166,7 @@ impl<B: Blob> Write<B> {
             let chunk_len = chunk.len();
 
             // Chunk falls entirely within the buffer's current range and can be merged.
-            if buffer.merge(chunk, current_offset) {
+            if state.buffer.merge(chunk, current_offset) {
                 bufs.advance(chunk_len);
                 current_offset += chunk_len as u64;
                 continue;
@@ -158,10 +175,11 @@ impl<B: Blob> Write<B> {
             // Chunk cannot be merged, so flush the buffer if the range overlaps, and check
             // if merge is possible after.
             let chunk_end = current_offset + chunk_len as u64;
-            if buffer.offset < chunk_end {
-                if let Some((old_buf, old_offset)) = buffer.take() {
+            if state.buffer.offset < chunk_end {
+                if let Some((old_buf, old_offset)) = state.buffer.take() {
                     self.blob.write_at(old_offset, old_buf).await?;
-                    if buffer.merge(chunk, current_offset) {
+                    state.needs_sync = true;
+                    if state.buffer.merge(chunk, current_offset) {
                         bufs.advance(chunk_len);
                         current_offset += chunk_len as u64;
                         continue;
@@ -175,11 +193,12 @@ impl<B: Blob> Write<B> {
             // below. Removing this inefficiency may not be worth the additional complexity.
             let direct = bufs.split_to(chunk_len);
             self.blob.write_at(current_offset, direct).await?;
+            state.needs_sync = true;
             current_offset += chunk_len as u64;
 
             // Maintain the "buffer at tip" invariant by advancing offset to the end of this
             // write if it extended the underlying blob.
-            buffer.offset = buffer.offset.max(current_offset);
+            state.buffer.offset = state.buffer.offset.max(current_offset);
         }
 
         Ok(())
@@ -191,27 +210,38 @@ impl<B: Blob> Write<B> {
     /// before resizing the underlying blob.
     pub async fn resize(&self, len: u64) -> Result<(), Error> {
         // Acquire a write lock on the buffer.
-        let mut buffer = self.buffer.write().await;
+        let mut state = self.state.write().await;
 
         // Flush buffered data to the underlying blob.
         //
         // This can only happen if the new size is greater than the current size.
-        if let Some((buf, offset)) = buffer.resize(len) {
+        if let Some((buf, offset)) = state.buffer.resize(len) {
             self.blob.write_at(offset, buf).await?;
+            state.needs_sync = true;
         }
 
         // Resize the underlying blob.
         self.blob.resize(len).await?;
+        state.needs_sync = true;
 
         Ok(())
     }
 
     /// Flush buffered bytes and durably sync the underlying blob.
     pub async fn sync(&self) -> Result<(), Error> {
-        let mut buffer = self.buffer.write().await;
-        if let Some((buf, offset)) = buffer.take() {
+        let mut state = self.state.write().await;
+        if let Some((buf, offset)) = state.buffer.take() {
+            if !state.needs_sync {
+                return self.blob.write_at_sync(offset, buf).await;
+            }
+
             self.blob.write_at(offset, buf).await?;
+            self.blob.sync().await?;
+            state.needs_sync = false;
+            return Ok(());
         }
-        self.blob.sync().await
+        self.blob.sync().await?;
+        state.needs_sync = false;
+        Ok(())
     }
 }
