commonware_runtime/utils/buffer/

mod.rs

//! Buffers for reading and writing to [crate::Blob]s.

use crate::WriteOptions;
use futures::future::{BoxFuture, FutureExt as _, Shared};

pub mod paged;
mod read;
mod tip;
mod write;

pub use read::Read;
pub use write::Write;

/// A shared sync result.
///
/// Handles returned by [Completion::handle] are detached observers of the same shared result:
/// dropping one neither cancels the underlying sync nor consumes its result, and every
/// observer sees the same outcome.
#[derive(Clone)]
struct Completion(Shared<BoxFuture<'static, Result<(), crate::Error>>>);

impl Completion {
    /// Return a handle for the sync result.
    fn handle(&self) -> crate::Handle<()> {
        crate::Handle::from_future(self.0.clone())
    }

    /// Wait for the sync result.
    async fn wait(&self) -> Result<(), crate::Error> {
        self.0.clone().await
    }
}

impl From<crate::Handle<()>> for Completion {
    fn from(handle: crate::Handle<()>) -> Self {
        Self(handle.boxed().shared())
    }
}

/// Tracks whether blob mutations still need a sync.
///
/// Callers rely on three properties:
/// - Every operation that mutates the blob first waits for an in-flight sync, so a started
///   sync's coverage is never disturbed by later writes.
/// - [SyncState::start_sync] on a [SyncState::Pending] state returns the in-flight sync's
///   handle (completed syncs resolve immediately), so re-requesting a sync is a cheap way to
///   observe outstanding work.
/// - A failure is never lost: every handle cloned from the shared completion reports it, and
///   an unobserved failure surfaces on the next [SyncState::wait_for_pending] call. Callers
///   propagate that error through the consuming writer operation.
///
/// The owner must be dropped after any failed operation.
enum SyncState {
    // No unsynced mutations.
    Clean,
    // Unsynced mutations need a sync.
    Dirty,
    // A started sync is in flight or awaits observation.
    Pending(Completion),
}

impl SyncState {
    /// Whether no mutation still needs a sync.
    const fn is_clean(&self) -> bool {
        matches!(self, Self::Clean)
    }

    /// Wait for an in-flight sync before reusing or mutating the blob.
    async fn wait_for_pending(&mut self) -> Result<(), crate::Error> {
        let Self::Pending(pending) = self else {
            return Ok(());
        };
        pending.wait().await?;
        *self = Self::Clean;
        Ok(())
    }

    /// Write data with the provided options while tracking durability.
    async fn write_at(
        &mut self,
        blob: &impl crate::Blob,
        offset: u64,
        bufs: impl Into<crate::IoBufs> + Send,
        options: WriteOptions,
    ) -> Result<(), crate::Error> {
        self.wait_for_pending().await?;
        let bufs = bufs.into();
        if !options.contains(WriteOptions::SYNC) {
            blob.write_at(offset, bufs, options).await?;
            *self = Self::Dirty;
            return Ok(());
        }

        match self {
            Self::Dirty => {
                // Earlier mutations need a full durability barrier too.
                blob.write_at(offset, bufs, options.without(WriteOptions::SYNC))
                    .await?;
                blob.sync().await?;
                *self = Self::Clean;
                Ok(())
            }
            Self::Clean => blob.write_at(offset, bufs, options).await,
            Self::Pending(_) => unreachable!("pending sync waited above"),
        }
    }

    /// Resize the blob and require a later sync.
    async fn resize(&mut self, blob: &impl crate::Blob, len: u64) -> Result<(), crate::Error> {
        self.wait_for_pending().await?;
        blob.resize(len).await?;
        *self = Self::Dirty;
        Ok(())
    }

    /// Make all pending mutations durable before returning.
    async fn sync(&mut self, blob: &impl crate::Blob) -> Result<(), crate::Error> {
        self.wait_for_pending().await?;
        if matches!(self, Self::Clean) {
            return Ok(());
        }
        blob.sync().await?;
        *self = Self::Clean;
        Ok(())
    }

    /// Start making pending mutations durable and return a handle for completion.
    async fn start_sync(&mut self, blob: &impl crate::Blob) -> crate::Handle<()> {
        match self {
            Self::Clean => crate::Handle::ready(Ok(())),
            Self::Dirty => {
                // Store a shared completion so repeated calls observe the same sync.
                let pending = Completion::from(blob.start_sync().await);
                let handle = pending.handle();
                *self = Self::Pending(pending);
                handle
            }
            Self::Pending(pending) => pending.handle(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        Blob as _, Error, Handle, IoBufMut, IoBufs, IoBufsMut, ReadOptions, Runner, Storage,
        WriteOptions, deterministic,
        mocks::{DelayedSyncBlob, WriteFaultContext, WriteFaults, next_pending_sync},
    };
    use commonware_macros::test_traced;
    use commonware_utils::{NZUsize, sync::Mutex};
    use futures::FutureExt;
    use std::sync::Arc;

    #[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 write operations.
        writes: usize,

        /// Number of full sync barriers.
        full_syncs: usize,

        /// Number of range-scoped write syncs.
        range_syncs: usize,

        /// Number of writes carrying the uncached hint.
        uncached_writes: usize,

        /// Number of range-scoped write syncs carrying the uncached hint.
        uncached_range_syncs: usize,
    }

    /// Test blob with separate visible and durable state.
    ///
    /// 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.
    pub struct SyncTrackingBlob {
        state: Mutex<RangeSyncState>,
    }

    impl SyncTrackingBlob {
        pub fn new() -> Self {
            Self {
                state: Mutex::new(RangeSyncState::default()),
            }
        }

        pub fn snapshot(&self) -> (Vec<u8>, usize, usize, usize) {
            let state = self.state.lock();
            (
                state.durable.clone(),
                state.writes,
                state.full_syncs,
                state.range_syncs,
            )
        }

        pub fn size(&self) -> u64 {
            self.state.lock().data.len() as u64
        }

        pub fn uncached_snapshot(&self) -> (usize, usize) {
            let state = self.state.lock();
            (state.uncached_writes, state.uncached_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 SyncTrackingBlob {
        async fn read_at(
            &self,
            offset: u64,
            len: usize,
            options: ReadOptions,
        ) -> Result<IoBufsMut, Error> {
            self.read_at_buf(offset, len, IoBufMut::with_capacity(len), options)
                .await
        }

        async fn read_at_buf(
            &self,
            offset: u64,
            len: usize,
            buf: impl Into<IoBufsMut> + Send,
            _options: ReadOptions,
        ) -> 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();
            assert!(out.capacity() >= len);
            // SAFETY: `len` bytes are filled by copy_from_slice below.
            unsafe { out.set_len(len) };
            out.copy_from_slice(&state.data[start..end]);
            Ok(out)
        }

        async fn write_at(
            &self,
            offset: u64,
            buf: impl Into<IoBufs> + Send,
            options: WriteOptions,
        ) -> Result<(), Error> {
            let buf = buf.into().coalesce();
            let mut state = self.state.lock();
            Self::write(&mut state.data, offset, buf.as_ref())?;
            state.writes += 1;

            let sync = options.contains(WriteOptions::SYNC);
            if sync {
                Self::write(&mut state.durable, offset, buf.as_ref())?;
                state.range_syncs += 1;
            }
            if options.contains(WriteOptions::DONT_CACHE) {
                if sync {
                    state.uncached_range_syncs += 1;
                } else {
                    state.uncached_writes += 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(())
        }

        async fn start_sync(&self) -> Handle<()> {
            Handle::ready(self.sync().await)
        }
    }

    #[test_traced]
    fn test_read_basic() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test basic buffered reading functionality with sequential reads
            let data = b"Hello, world! This is a test.";
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();
            let size = data.len() as u64;

            // Create a buffered reader with small buffer to test refilling
            let mut reader = Read::new(blob, size, NZUsize!(10));

            // Read some data
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"Hello");

            // Read more data that requires a buffer refill
            let read = reader.read(14).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b", world! This ");

            // Verify position tracking
            assert_eq!(reader.position(), 19);

            // Read the remaining data
            let read = reader.read(7).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"is a te");

            // Attempt to read beyond the end should fail
            let result = reader.read(5).await;
            assert!(matches!(result, Err(Error::BlobInsufficientLength)));
        });
    }

    #[test_traced]
    fn test_read_cross_boundary() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test reading data that spans multiple buffer refills
            let data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ";
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();
            let size = data.len() as u64;

            // Use a buffer smaller than the total data size
            let mut reader = Read::new(blob, size, NZUsize!(10));

            // Read data that crosses buffer boundaries
            let read = reader.read(15).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ABCDEFGHIJKLMNO");

            // Verify position tracking
            assert_eq!(reader.position(), 15);

            // Read the remaining data
            let read = reader.read(11).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"PQRSTUVWXYZ");

            // Verify we're at the end
            assert_eq!(reader.position(), 26);
            assert_eq!(reader.blob_remaining(), 0);
        });
    }

    // Regression test for https://github.com/commonwarexyz/monorepo/issues/1348
    #[test_traced]
    fn test_read_to_end_then_rewind_and_read_again() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ";
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();
            let size = data.len() as u64;

            let mut reader = Read::new(blob, size, NZUsize!(20));

            // Read data that crosses buffer boundaries
            let read = reader.read(21).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ABCDEFGHIJKLMNOPQRSTU");

            // Verify position tracking
            assert_eq!(reader.position(), 21);

            // Read the remaining data
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"VWXYZ");

            // Rewind and read again
            reader.seek_to(0).unwrap();
            let read = reader.read(21).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ABCDEFGHIJKLMNOPQRSTU");
        });
    }

    #[test_traced]
    fn test_read_with_known_size() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test reader behavior with known blob size limits
            let data = b"This is a test with known size limitations.";
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();
            let size = data.len() as u64;

            // Create a buffered reader with buffer smaller than total data
            let mut reader = Read::new(blob, size, NZUsize!(10));

            // Check initial remaining bytes
            assert_eq!(reader.blob_remaining(), size);

            // Read partial data
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"This ");

            // Check remaining bytes after partial read
            assert_eq!(reader.blob_remaining(), size - 5);

            // Read exactly up to the size limit
            let read = reader.read((size - 5) as usize).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"is a test with known size limitations.");

            // Verify we're at the end
            assert_eq!(reader.blob_remaining(), 0);

            // Reading beyond the end should fail
            let result = reader.read(1).await;
            assert!(matches!(result, Err(Error::BlobInsufficientLength)));
        });
    }

    #[test_traced]
    fn test_read_oversized_request_does_not_consume_buffered_bytes() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let data = b"abcdefghij";
            let (blob, size) = context
                .open("partition", b"double-count-regression")
                .await
                .unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();

            let mut reader = Read::new(blob, data.len() as u64, NZUsize!(8));

            // Fill the internal buffer and consume most of it (2 bytes remain buffered).
            let first = reader.read(6).await.unwrap().coalesce();
            assert_eq!(first.as_ref(), b"abcdef");
            assert_eq!(reader.position(), 6);

            // Only 4 bytes remain total, so this must fail without consuming anything.
            let err = reader.read(5).await.unwrap_err();
            assert!(matches!(err, Error::BlobInsufficientLength));
            assert_eq!(reader.position(), 6);

            // Remaining bytes should still be readable in full.
            let tail = reader.read(4).await.unwrap().coalesce();
            assert_eq!(tail.as_ref(), b"ghij");
            assert_eq!(reader.position(), 10);
        });
    }

    #[test_traced]
    fn test_read_large_data() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test reading large amounts of data in chunks
            let data_size = 1024 * 256; // 256KB of data
            let data = vec![0x42; data_size];
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data.clone(), WriteOptions::default())
                .await
                .unwrap();
            let size = data.len() as u64;

            // Use a buffer much smaller than the total data
            let mut reader = Read::new(blob, size, NZUsize!(64 * 1024));

            // Read all data in smaller chunks
            let mut total_read = 0;
            let chunk_size = 8 * 1024; // 8KB chunks

            while total_read < data_size {
                let to_read = std::cmp::min(chunk_size, data_size - total_read);
                let read = reader.read(to_read).await.unwrap().coalesce();

                // Verify data integrity
                assert!(
                    read.as_ref().iter().all(|&b| b == 0x42),
                    "Data at position {total_read} is not correct"
                );

                total_read += to_read;
            }

            // Verify we read everything
            assert_eq!(total_read, data_size);

            // Reading beyond the end should fail
            let result = reader.read(1).await;
            assert!(matches!(result, Err(Error::BlobInsufficientLength)));
        });
    }

    #[test_traced]
    fn test_read_exact_size_reads() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a blob with exactly 2.5 buffer sizes of data
            let buffer_size = 1024;
            let data_size = buffer_size * 5 / 2; // 2.5 buffers
            let data = vec![0x37; data_size];

            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data.clone(), WriteOptions::default())
                .await
                .unwrap();
            let size = data.len() as u64;

            let mut reader = Read::new(blob, size, NZUsize!(buffer_size));

            // Read exactly one buffer size
            let read = reader.read(buffer_size).await.unwrap().coalesce();
            assert!(read.as_ref().iter().all(|&b| b == 0x37));

            // Read exactly one buffer size more
            let read = reader.read(buffer_size).await.unwrap().coalesce();
            assert!(read.as_ref().iter().all(|&b| b == 0x37));

            // Read the remaining half buffer
            let half_buffer = buffer_size / 2;
            let read = reader.read(half_buffer).await.unwrap().coalesce();
            assert!(read.as_ref().iter().all(|&b| b == 0x37));

            // Verify we're at the end
            assert_eq!(reader.blob_remaining(), 0);
            assert_eq!(reader.position(), size);
        });
    }

    #[test_traced]
    fn test_read_structure_single_vs_chunked() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let data = b"ABCDEFGHIJKL";
            let (blob, size) = context.open("partition", b"structural").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();

            let mut reader = Read::new(blob, data.len() as u64, NZUsize!(5));

            // First read fits in one fetched chunk.
            let first = reader.read(3).await.unwrap();
            assert!(first.is_single());
            assert_eq!(first.coalesce().as_ref(), b"ABC");

            // This read spans refill boundaries and should be represented as multiple chunks.
            let second = reader.read(7).await.unwrap();
            assert!(!second.is_single());
            assert_eq!(second.coalesce().as_ref(), b"DEFGHIJ");
        });
    }

    #[test_traced]
    fn test_read_seek_to() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a memory blob with some test data
            let data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ";
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();
            let size = data.len() as u64;

            // Create a buffer reader
            let mut reader = Read::new(blob, size, NZUsize!(10));

            // Read some data to advance the position
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ABCDE");
            assert_eq!(reader.position(), 5);

            // Seek to a specific position
            reader.seek_to(10).unwrap();
            assert_eq!(reader.position(), 10);

            // Read data from the new position
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"KLMNO");

            // Seek to beginning
            reader.seek_to(0).unwrap();
            assert_eq!(reader.position(), 0);

            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ABCDE");

            // Seek to end
            reader.seek_to(size).unwrap();
            assert_eq!(reader.position(), size);

            // Trying to read should fail
            let result = reader.read(1).await;
            assert!(matches!(result, Err(Error::BlobInsufficientLength)));

            // Seek beyond end should fail
            let result = reader.seek_to(size + 10);
            assert!(matches!(result, Err(Error::BlobInsufficientLength)));
        });
    }

    #[test_traced]
    fn test_read_seek_with_refill() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a memory blob with longer data
            let data = vec![0x41; 1000]; // 1000 'A' characters
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data.clone(), WriteOptions::default())
                .await
                .unwrap();
            let size = data.len() as u64;

            // Create a buffer reader with small buffer
            let mut reader = Read::new(blob, size, NZUsize!(10));

            // Read some data
            let _ = reader.read(5).await.unwrap().coalesce();

            // Seek far ahead, past the current buffer
            reader.seek_to(500).unwrap();

            // Read data - should get data from position 500
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"AAAAA"); // Should still be 'A's);
            assert_eq!(reader.position(), 505);

            // Seek backwards
            reader.seek_to(100).unwrap();

            // Read again - should be at position 100
            let _ = reader.read(5).await.unwrap().coalesce();
            assert_eq!(reader.position(), 105);
        });
    }

    #[test_traced]
    fn test_read_seek_within_buffered_range() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ";
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();

            let mut reader = Read::new(blob, data.len() as u64, NZUsize!(10));

            // Reads 0..=5, while the internal fetch cursor advances to 10.
            let read = reader.read(6).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ABCDEF");
            assert_eq!(reader.position(), 6);
            assert_eq!(reader.buffer_remaining(), 4);

            // Seek back within [buffer_start, fetch_position).
            reader.seek_to(3).unwrap();
            assert_eq!(reader.position(), 3);
            assert_eq!(reader.buffer_remaining(), 7);

            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"DEFGH");
            assert_eq!(reader.position(), 8);
            assert_eq!(reader.buffer_remaining(), 2);
        });
    }

    #[test_traced]
    fn test_read_seek_within_unread_buffer_does_not_refill() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ";
            let (blob, size) = context
                .open("partition", b"seek_unread_no_refill")
                .await
                .unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();

            let mut reader = Read::new(blob, data.len() as u64, NZUsize!(10));

            // First read triggers a single refill of 10 bytes.
            let first = reader.read(6).await.unwrap();
            assert_eq!(first.coalesce().as_ref(), b"ABCDEF");
            assert_eq!(reader.position(), 6);
            assert_eq!(reader.buffer_remaining(), 4);

            // Seek within the unread buffered window [6, 10).
            reader.seek_to(7).unwrap();
            assert_eq!(reader.position(), 7);
            assert_eq!(reader.buffer_remaining(), 3);

            // Consume only from the already buffered window.
            let second = reader.read(3).await.unwrap();
            assert_eq!(second.coalesce().as_ref(), b"HIJ");
            assert_eq!(reader.position(), 10);
            assert_eq!(reader.buffer_remaining(), 0);

            // Refill should happen only now (at exhaustion), not at seek/read above.
            let third = reader.read(1).await.unwrap();
            assert_eq!(third.coalesce().as_ref(), b"K");
            assert_eq!(reader.position(), 11);
            assert_eq!(reader.buffer_remaining(), 9);
        });
    }

    #[test_traced]
    fn test_read_resize() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a memory blob with some test data
            let data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ";
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();
            let data_len = data.len() as u64;

            // Create a buffer reader
            let reader = Read::new(blob, data_len, NZUsize!(10));

            // Resize the blob to half its size
            let resize_len = data_len / 2;
            reader.resize(resize_len).await.unwrap();

            // Reopen to check truncation
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, resize_len, "Blob should be resized to half size");

            // Create a new buffer and read to verify truncation
            let mut new_reader = Read::new(blob, size, NZUsize!(10));

            // Read the content
            let read = new_reader.read(size as usize).await.unwrap().coalesce();
            assert_eq!(
                read.as_ref(),
                b"ABCDEFGHIJKLM",
                "Resized content should match"
            );

            // Reading beyond resized size should fail
            let result = new_reader.read(1).await;
            assert!(matches!(result, Err(Error::BlobInsufficientLength)));

            // Test resize to larger size
            new_reader.resize(data_len * 2).await.unwrap();

            // Reopen to check resize
            let (blob, new_size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(new_size, data_len * 2);

            // Create a new buffer and read to verify resize
            let mut new_reader = Read::new(blob, new_size, NZUsize!(10));
            let read = new_reader.read(new_size as usize).await.unwrap().coalesce();
            assert_eq!(&read.as_ref()[..size as usize], b"ABCDEFGHIJKLM");
            assert_eq!(
                &read.as_ref()[size as usize..],
                vec![0u8; new_size as usize - size as usize]
            );
        });
    }

    #[test_traced]
    fn test_read_resize_to_zero() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a memory blob with some test data
            let data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ";
            let data_len = data.len() as u64;
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0);
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();

            // Create a buffer reader
            let reader = Read::new(blob, data_len, NZUsize!(10));

            // Resize the blob to zero
            reader.resize(0).await.unwrap();

            // Reopen to check truncation
            let (blob, size) = context.open("partition", b"test").await.unwrap();
            assert_eq!(size, 0, "Blob should be resized to zero");

            // Create a new buffer and try to read (should fail)
            let mut new_reader = Read::new(blob, size, NZUsize!(10));

            // Reading from resized blob should fail
            let result = new_reader.read(1).await;
            assert!(matches!(result, Err(Error::BlobInsufficientLength)));
        });
    }

    #[test_traced]
    fn test_write_basic() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test basic buffered write and sync functionality
            let (blob, size) = context.open("partition", b"write_basic").await.unwrap();
            assert_eq!(size, 0);

            let mut writer = Write::new(blob, size, NZUsize!(8));
            writer = writer.write_at(0, b"hello").await.unwrap();
            assert_eq!(writer.size(), 5);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 5);

            // Verify data was written correctly
            drop(writer);
            let (blob, size) = context.open("partition", b"write_basic").await.unwrap();
            assert_eq!(size, 5);
            let mut reader = Read::new(blob, size, NZUsize!(8));
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"hello");
        });
    }

    #[test_traced]
    fn test_write_multiple_flushes() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test writes that cause buffer flushes due to capacity limits
            let (blob, size) = context.open("partition", b"write_multi").await.unwrap();
            assert_eq!(size, 0);

            let mut writer = Write::new(blob, size, NZUsize!(4));
            writer = writer.write_at(0, b"abc").await.unwrap();
            assert_eq!(writer.size(), 3);
            writer = writer.write_at(3, b"defg").await.unwrap();
            assert_eq!(writer.size(), 7);
            writer = writer.sync().await.unwrap();

            // Verify the final result
            drop(writer);
            let (blob, size) = context.open("partition", b"write_multi").await.unwrap();
            assert_eq!(size, 7);
            let mut reader = Read::new(blob, size, NZUsize!(4));
            let read = reader.read(7).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"abcdefg");
        });
    }

    #[test_traced]
    fn test_write_large_data() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test writing data larger than buffer capacity (direct write)
            let (blob, size) = context.open("partition", b"write_large").await.unwrap();
            assert_eq!(size, 0);

            let mut writer = Write::new(blob, size, NZUsize!(4));
            writer = writer.write_at(0, b"abc").await.unwrap();
            assert_eq!(writer.size(), 3);
            writer = writer
                .write_at(3, b"defghijklmnopqrstuvwxyz")
                .await
                .unwrap();
            assert_eq!(writer.size(), 26);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 26);

            // Verify the complete data
            drop(writer);
            let (blob, size) = context.open("partition", b"write_large").await.unwrap();
            assert_eq!(size, 26);
            let mut reader = Read::new(blob, size, NZUsize!(4));
            let read = reader.read(26).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"abcdefghijklmnopqrstuvwxyz");
        });
    }

    #[test_traced]
    fn test_write_append_to_buffer() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test sequential appends that exceed buffer capacity
            let (blob, size) = context.open("partition", b"append_buf").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Write data that fits in buffer
            writer = writer.write_at(0, b"hello").await.unwrap();
            assert_eq!(writer.size(), 5);

            // Append data that causes buffer flush
            writer = writer.write_at(5, b" world").await.unwrap();
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 11);

            // Verify the complete result
            drop(writer);
            let (blob, size) = context.open("partition", b"append_buf").await.unwrap();
            assert_eq!(size, 11);
            let mut reader = Read::new(blob, size, NZUsize!(10));
            let read = reader.read(11).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"hello world");
        });
    }

    #[test_traced]
    fn test_write_into_middle_of_buffer() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test overwriting data within the buffer and extending it
            let (blob, size) = context.open("partition", b"middle_buf").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(20));

            // Initial write
            writer = writer.write_at(0, b"abcdefghij").await.unwrap();
            assert_eq!(writer.size(), 10);

            // Overwrite middle section
            writer = writer.write_at(2, b"01234").await.unwrap();
            assert_eq!(writer.size(), 10);
            writer = writer.sync().await.unwrap();

            // Verify overwrite result
            drop(writer);
            let (blob, size) = context.open("partition", b"middle_buf").await.unwrap();
            let blob = Arc::new(blob);
            assert_eq!(size, 10);
            let mut reader = Read::new(blob.clone(), size, NZUsize!(10));
            let read = reader.read(10).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ab01234hij");
            drop(reader);
            let mut writer = Write::new(blob, size, NZUsize!(20));

            // Extend buffer and do partial overwrite
            writer = writer.write_at(10, b"klmnopqrst").await.unwrap();
            assert_eq!(writer.size(), 20);
            writer = writer.write_at(9, b"wxyz").await.unwrap();
            assert_eq!(writer.size(), 20);
            writer = writer.sync().await.unwrap();

            // Verify final result
            drop(writer);
            let (blob, size) = context.open("partition", b"middle_buf").await.unwrap();
            assert_eq!(size, 20);
            let mut reader = Read::new(blob, size, NZUsize!(20));
            let read = reader.read(20).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ab01234hiwxyznopqrst");
        });
    }

    #[test_traced]
    fn test_write_before_buffer() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test writing at offsets before the current buffer position
            let (blob, size) = context.open("partition", b"before_buf").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Write data at a later offset first
            writer = writer.write_at(10, b"0123456789").await.unwrap();
            assert_eq!(writer.size(), 20);

            // Write at an earlier offset (should flush buffer first)
            writer = writer.write_at(0, b"abcde").await.unwrap();
            assert_eq!(writer.size(), 20);
            writer = writer.sync().await.unwrap();

            // Verify data placement with gap
            drop(writer);
            let (blob, size) = context.open("partition", b"before_buf").await.unwrap();
            let blob = Arc::new(blob);
            assert_eq!(size, 20);
            let mut reader = Read::new(blob.clone(), size, NZUsize!(20));
            let read = reader.read(20).await.unwrap().coalesce();
            let mut expected = vec![0u8; 20];
            expected[0..5].copy_from_slice("abcde".as_bytes());
            expected[10..20].copy_from_slice("0123456789".as_bytes());
            assert_eq!(read.as_ref(), expected.as_slice());
            drop(reader);
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Fill the gap between existing data
            writer = writer.write_at(5, b"fghij").await.unwrap();
            assert_eq!(writer.size(), 20);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 20);

            // Verify gap is filled
            drop(writer);
            let (blob, size) = context.open("partition", b"before_buf").await.unwrap();
            assert_eq!(size, 20);
            let mut reader = Read::new(blob, size, NZUsize!(20));
            let read = reader.read(20).await.unwrap().coalesce();
            expected[0..10].copy_from_slice("abcdefghij".as_bytes());
            assert_eq!(read.as_ref(), expected.as_slice());
        });
    }

    #[test_traced]
    fn test_write_resize() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test blob resize functionality and subsequent writes
            let (blob, size) = context.open("partition", b"resize_write").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Write initial data
            writer = writer.write_at(0, b"hello world").await.unwrap();
            assert_eq!(writer.size(), 11);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 11);

            drop(writer);
            let (blob_check, size_check) =
                context.open("partition", b"resize_write").await.unwrap();
            assert_eq!(size_check, 11);
            let mut writer = Write::new(blob_check, size_check, NZUsize!(10));

            // Resize to smaller size
            writer = writer.resize(5).await.unwrap();
            assert_eq!(writer.size(), 5);
            writer = writer.sync().await.unwrap();

            // Verify resize
            drop(writer);
            let (blob, size) = context.open("partition", b"resize_write").await.unwrap();
            let blob = Arc::new(blob);
            assert_eq!(size, 5);
            let mut reader = Read::new(blob.clone(), size, NZUsize!(5));
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"hello");
            drop(reader);
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Write to resized blob
            writer = writer.write_at(0, b"X").await.unwrap();
            assert_eq!(writer.size(), 5);
            writer = writer.sync().await.unwrap();

            // Verify overwrite
            drop(writer);
            let (blob, size) = context.open("partition", b"resize_write").await.unwrap();
            let blob = Arc::new(blob);
            assert_eq!(size, 5);
            let mut reader = Read::new(blob.clone(), size, NZUsize!(5));
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"Xello");
            drop(reader);
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Test resize to larger size
            writer = writer.resize(10).await.unwrap();
            assert_eq!(writer.size(), 10);
            writer = writer.sync().await.unwrap();

            // Verify resize
            drop(writer);
            let (blob, size) = context.open("partition", b"resize_write").await.unwrap();
            assert_eq!(size, 10);
            let mut reader = Read::new(blob, size, NZUsize!(10));
            let read = reader.read(10).await.unwrap().coalesce();
            assert_eq!(&read.as_ref()[0..5], b"Xello");
            assert_eq!(&read.as_ref()[5..10], [0u8; 5]);

            // Test resize to zero
            let (blob_zero, size) = context.open("partition", b"resize_zero").await.unwrap();
            let mut writer_zero = Write::new(blob_zero, size, NZUsize!(10));
            writer_zero = writer_zero.write_at(0, b"some data").await.unwrap();
            assert_eq!(writer_zero.size(), 9);
            writer_zero = writer_zero.sync().await.unwrap();
            assert_eq!(writer_zero.size(), 9);
            writer_zero = writer_zero.resize(0).await.unwrap();
            assert_eq!(writer_zero.size(), 0);
            writer_zero = writer_zero.sync().await.unwrap();
            assert_eq!(writer_zero.size(), 0);

            // Ensure the blob is empty
            drop(writer_zero);
            let (_, size_z) = context.open("partition", b"resize_zero").await.unwrap();
            assert_eq!(size_z, 0);
        });
    }

    #[test_traced]
    fn test_write_resize_grow_flushes_buffered_data() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let (blob, size) = context.open("partition", b"resize_grow").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(10));

            writer = writer.write_at(0, b"hello").await.unwrap();
            writer = writer.resize(10).await.unwrap();
            writer = writer.sync().await.unwrap();

            let read = writer.read_at(0, 5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"hello");
        });
    }

    #[test_traced]
    fn test_write_read_at_on_writer() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test reading through writer's read_at method (buffer + blob reads)
            let (blob, size) = context.open("partition", b"read_at_writer").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Write data that stays in buffer
            writer = writer.write_at(0, b"buffered").await.unwrap();
            assert_eq!(writer.size(), 8);

            // Read from buffer via writer
            let read_buf_vec = writer.read_at(0, 4).await.unwrap().coalesce();
            assert_eq!(read_buf_vec, b"buff");

            let read_buf_vec = writer.read_at(4, 4).await.unwrap().coalesce();
            assert_eq!(read_buf_vec, b"ered");

            // Reading past buffer end should fail
            assert!(writer.read_at(8, 1).await.is_err());

            // Write large data that flushes buffer
            writer = writer.write_at(8, b" and flushed").await.unwrap();
            assert_eq!(writer.size(), 20);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 20);

            // Read from underlying blob through writer
            let read_buf_vec_2 = writer.read_at(0, 4).await.unwrap().coalesce();
            assert_eq!(read_buf_vec_2, b"buff");

            let read_buf_7_vec = writer.read_at(13, 7).await.unwrap().coalesce();
            assert_eq!(read_buf_7_vec, b"flushed");

            // Buffer new data at the end
            writer = writer.write_at(20, b" more data").await.unwrap();
            assert_eq!(writer.size(), 30);

            // Read newly buffered data
            let read_buf_vec_3 = writer.read_at(20, 5).await.unwrap().coalesce();
            assert_eq!(read_buf_vec_3, b" more");

            // Read spanning both blob and buffer
            let combo_read_buf_vec = writer.read_at(16, 12).await.unwrap();
            assert_eq!(combo_read_buf_vec.coalesce(), b"shed more da");

            // Verify complete content by reopening
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 30);
            drop(writer);
            let (final_blob, final_size) =
                context.open("partition", b"read_at_writer").await.unwrap();
            assert_eq!(final_size, 30);
            let mut final_reader = Read::new(final_blob, final_size, NZUsize!(30));
            let read = final_reader.read(30).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"buffered and flushed more data");
        });
    }

    #[test_traced]
    fn test_write_zero_length_read_past_eof_errors() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let (blob, size) = context.open("partition", b"zero_len_probe").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(8));
            writer = writer.write_at(0, b"abc").await.unwrap();

            let empty = writer.read_at(3, 0).await.unwrap();
            assert!(empty.is_empty());

            let err = writer.read_at(4, 0).await.unwrap_err();
            assert!(matches!(err, Error::BlobInsufficientLength));
        });
    }

    #[test_traced]
    fn test_write_straddling_non_mergeable() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test writes that cannot be merged into buffer (non-contiguous/too large)
            let (blob, size) = context.open("partition", b"write_straddle").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Fill buffer completely
            writer = writer.write_at(0, b"0123456789").await.unwrap();
            assert_eq!(writer.size(), 10);

            // Write at non-contiguous offset (should flush then write directly)
            writer = writer.write_at(15, b"abc").await.unwrap();
            assert_eq!(writer.size(), 18);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 18);

            // Verify data with gap
            drop(writer);
            let (blob_check, size_check) =
                context.open("partition", b"write_straddle").await.unwrap();
            assert_eq!(size_check, 18);
            let mut reader = Read::new(blob_check, size_check, NZUsize!(20));
            let read = reader.read(18).await.unwrap().coalesce();

            let mut expected = vec![0u8; 18];
            expected[0..10].copy_from_slice(b"0123456789");
            expected[15..18].copy_from_slice(b"abc");
            assert_eq!(read.as_ref(), expected.as_slice());

            // Test write that exceeds buffer capacity
            let (blob2, size) = context.open("partition", b"write_straddle2").await.unwrap();
            let mut writer2 = Write::new(blob2, size, NZUsize!(10));
            writer2 = writer2.write_at(0, b"0123456789").await.unwrap();
            assert_eq!(writer2.size(), 10);

            // Write large data that exceeds capacity
            writer2 = writer2.write_at(5, b"ABCDEFGHIJKL").await.unwrap();
            assert_eq!(writer2.size(), 17);
            writer2 = writer2.sync().await.unwrap();
            assert_eq!(writer2.size(), 17);

            // Verify overwrite result
            drop(writer2);
            let (blob_check2, size_check2) =
                context.open("partition", b"write_straddle2").await.unwrap();
            assert_eq!(size_check2, 17);
            let mut reader2 = Read::new(blob_check2, size_check2, NZUsize!(20));
            let read = reader2.read(17).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"01234ABCDEFGHIJKL");
        });
    }

    #[test_traced]
    fn test_write_close() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Buffer bytes to persist before releasing the writer.
            let (blob_orig, size) = context.open("partition", b"write_close").await.unwrap();
            let mut writer = Write::new(blob_orig, size, NZUsize!(8));
            writer = writer.write_at(0, b"pending").await.unwrap();
            assert_eq!(writer.size(), 7);

            // Sync writer to persist data
            writer = writer.sync().await.unwrap();

            // Verify data persistence
            drop(writer);
            let (blob_check, size_check) = context.open("partition", b"write_close").await.unwrap();
            assert_eq!(size_check, 7);
            let mut reader = Read::new(blob_check, size_check, NZUsize!(8));
            let read = reader.read(7).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"pending");
        });
    }

    #[test_traced]
    fn test_write_direct_due_to_size() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test direct writes when data exceeds buffer capacity
            let (blob, size) = context
                .open("partition", b"write_direct_size")
                .await
                .unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(5));

            // Write data larger than buffer capacity (should write directly)
            let data_large = b"0123456789";
            writer = writer.write_at(0, data_large).await.unwrap();
            assert_eq!(writer.size(), 10);

            // Sync to ensure data is persisted
            writer = writer.sync().await.unwrap();

            // Verify direct write worked
            drop(writer);
            let (blob_check, size_check) = context
                .open("partition", b"write_direct_size")
                .await
                .unwrap();
            let blob_check = Arc::new(blob_check);
            assert_eq!(size_check, 10);
            let mut reader = Read::new(blob_check.clone(), size_check, NZUsize!(10));
            let read = reader.read(10).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), data_large.as_slice());
            drop(reader);
            let mut writer = Write::new(blob_check, size_check, NZUsize!(5));

            // Now write small data that should be buffered
            writer = writer.write_at(10, b"abc").await.unwrap();
            assert_eq!(writer.size(), 13);

            // Verify it's in buffer by reading through writer
            let read_small_buf_vec = writer.read_at(10, 3).await.unwrap().coalesce();
            assert_eq!(read_small_buf_vec, b"abc");

            writer = writer.sync().await.unwrap();

            // Verify final state
            drop(writer);
            let (blob_check2, size_check2) = context
                .open("partition", b"write_direct_size")
                .await
                .unwrap();
            assert_eq!(size_check2, 13);
            let mut reader2 = Read::new(blob_check2, size_check2, NZUsize!(13));
            let read = reader2.read(13).await.unwrap().coalesce();
            assert_eq!(&read.as_ref()[10..], b"abc".as_slice());
        });
    }

    #[test_traced]
    fn test_write_overwrite_and_extend_in_buffer() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test complex buffer operations: overwrite and extend within capacity
            let (blob, size) = context
                .open("partition", b"overwrite_extend_buf")
                .await
                .unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(15));

            // Write initial data
            writer = writer.write_at(0, b"0123456789").await.unwrap();
            assert_eq!(writer.size(), 10);

            // Overwrite and extend within buffer capacity
            writer = writer.write_at(5, b"ABCDEFGHIJ").await.unwrap();
            assert_eq!(writer.size(), 15);

            // Verify buffer content through writer
            let read_buf_vec = writer.read_at(0, 15).await.unwrap().coalesce();
            assert_eq!(read_buf_vec, b"01234ABCDEFGHIJ");

            writer = writer.sync().await.unwrap();

            // Verify persisted result
            drop(writer);
            let (blob_check, size_check) = context
                .open("partition", b"overwrite_extend_buf")
                .await
                .unwrap();
            assert_eq!(size_check, 15);
            let mut reader = Read::new(blob_check, size_check, NZUsize!(15));
            let read = reader.read(15).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"01234ABCDEFGHIJ".as_slice());
        });
    }

    #[test_traced]
    fn test_write_at_size() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test writing at the current logical end of the blob
            let (blob, size) = context.open("partition", b"write_end").await.unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(20));

            // Write initial data
            writer = writer.write_at(0, b"0123456789").await.unwrap();
            assert_eq!(writer.size(), 10);
            writer = writer.sync().await.unwrap();

            // Append at the current size (logical end)
            let end = writer.size();
            writer = writer.write_at(end, b"abc").await.unwrap();
            assert_eq!(writer.size(), 13);
            writer = writer.sync().await.unwrap();

            // Verify complete result
            drop(writer);
            let (blob_check, size_check) = context.open("partition", b"write_end").await.unwrap();
            assert_eq!(size_check, 13);
            let mut reader = Read::new(blob_check, size_check, NZUsize!(13));
            let read = reader.read(13).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"0123456789abc");
        });
    }

    #[test_traced]
    fn test_write_at_size_multiple_appends() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test multiple appends using writer.size()
            let (blob, size) = context
                .open("partition", b"write_multiple_appends_at_size")
                .await
                .unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(5));

            // First write
            writer = writer.write_at(0, b"AAA").await.unwrap();
            assert_eq!(writer.size(), 3);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 3);

            // Append using size()
            let end = writer.size();
            writer = writer.write_at(end, b"BBB").await.unwrap();
            assert_eq!(writer.size(), 6); // 3 (AAA) + 3 (BBB)
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 6);

            // Append again using size()
            let end = writer.size();
            writer = writer.write_at(end, b"CCC").await.unwrap();
            assert_eq!(writer.size(), 9); // 6 + 3 (CCC)
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 9);

            // Verify final content
            drop(writer);
            let (blob_check, size_check) = context
                .open("partition", b"write_multiple_appends_at_size")
                .await
                .unwrap();
            assert_eq!(size_check, 9);
            let mut reader = Read::new(blob_check, size_check, NZUsize!(9));
            let read = reader.read(9).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"AAABBBCCC");
        });
    }

    #[test_traced]
    fn test_write_non_contiguous_then_append_at_size() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Test writing non-contiguously, then appending at the new size
            let (blob, size) = context
                .open("partition", b"write_non_contiguous_then_append")
                .await
                .unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Buffer the initial seven bytes at the logical tip.
            writer = writer.write_at(0, b"INITIAL").await.unwrap();
            assert_eq!(writer.size(), 7);

            // Non-contiguous write, forces flush of "INITIAL" and direct write of "NONCONTIG"
            writer = writer.write_at(20, b"NONCONTIG").await.unwrap();
            assert_eq!(writer.size(), 29);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 29);

            // Append at the new size
            let end = writer.size();
            writer = writer.write_at(end, b"APPEND").await.unwrap();
            assert_eq!(writer.size(), 35);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 35);

            // Verify final content
            drop(writer);
            let (blob_check, size_check) = context
                .open("partition", b"write_non_contiguous_then_append")
                .await
                .unwrap();
            assert_eq!(size_check, 35);
            let mut reader = Read::new(blob_check, size_check, NZUsize!(35));
            let read = reader.read(35).await.unwrap().coalesce();

            let mut expected = vec![0u8; 35];
            expected[0..7].copy_from_slice(b"INITIAL");
            expected[20..29].copy_from_slice(b"NONCONTIG");
            expected[29..35].copy_from_slice(b"APPEND");
            assert_eq!(read.as_ref(), expected.as_slice());
        });
    }

    #[test_traced]
    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
            let (blob, size) = context
                .open("partition", b"resize_then_append_at_size")
                .await
                .unwrap();
            let mut writer = Write::new(blob, size, NZUsize!(10));

            // Persist the initial 16 bytes, leaving the tip buffer empty.
            writer = writer.write_at(0, b"0123456789ABCDEF").await.unwrap();
            assert_eq!(writer.size(), 16);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 16);

            // Durably shorten the blob and its logical tip to five bytes.
            let resize_to = 5;
            writer = writer.resize(resize_to).await.unwrap();
            assert_eq!(writer.size(), resize_to);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), resize_to);

            // Buffer five more bytes at the resized tip, then persist the combined contents.
            let end = writer.size();
            writer = writer.write_at(end, b"XXXXX").await.unwrap();
            assert_eq!(writer.size(), 10);
            writer = writer.sync().await.unwrap();
            assert_eq!(writer.size(), 10);

            // Verify final content
            drop(writer);
            let (blob_check, size_check) = context
                .open("partition", b"resize_then_append_at_size")
                .await
                .unwrap();
            assert_eq!(size_check, 10);
            let mut reader = Read::new(blob_check, size_check, NZUsize!(10));
            let read = reader.read(10).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"01234XXXXX");
        });
    }

    /// A failed flush inside `start_sync` returns an error.
    #[test_traced]
    fn test_write_start_sync_flush_failure_returns_err() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let faults = WriteFaults::default();
            let context = WriteFaultContext {
                inner: context,
                faults: faults.clone(),
            };
            let (blob, size) = context
                .open("partition", b"start_sync_flush_failure")
                .await
                .unwrap();
            let writer = Write::new(blob, size, NZUsize!(8));

            // Buffer a write so start_sync must flush it.
            let writer = writer.write_at(0, b"abc").await.unwrap();

            // The flush inside start_sync fails.
            faults.arm();
            assert!(writer.start_sync().await.is_err());
        });
    }

    /// `try_write_at` merges a write that fits the tip buffer and leaves the writer unchanged
    /// otherwise.
    #[test_traced]
    fn test_write_try_write_at_merges_only_when_it_fits() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let (blob, _) = context.open("partition", b"try_write_at").await.unwrap();
            let mut writer = Write::new(blob, 0, NZUsize!(8));

            // A write within the buffer's capacity merges without I/O.
            assert!(writer.try_write_at(0, b"abc"));
            assert_eq!(writer.size(), 3);

            // A write past the buffer's capacity is refused and changes nothing.
            assert!(!writer.try_write_at(3, b"0123456789"));
            assert_eq!(writer.size(), 3);

            // An overflowing offset is refused, leaving write_at to report the overflow.
            assert!(!writer.try_write_at(u64::MAX, b"x"));
            assert_eq!(writer.size(), 3);

            // An empty write leaves the writer unchanged, like write_at.
            assert!(writer.try_write_at(6, &[]));
            assert_eq!(writer.size(), 3);

            // The merged bytes read back and persist through a sync.
            let writer = writer.sync().await.unwrap();
            let read = writer.read_at(0, 3).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"abc");
        });
    }

    /// `needs_sync` reports whether `sync` has work: buffered bytes, unsynced mutations, or a
    /// started sync that the writer has not observed.
    #[test_traced]
    fn test_write_needs_sync_tracks_sync_work() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let (blob, pending) = DelayedSyncBlob::new(SyncTrackingBlob::new());
            let writer = Write::new(blob, 0, NZUsize!(8));

            // A fresh writer may wrap unsynced bytes, so it starts dirty. Syncing cleans it.
            assert!(writer.needs_sync());
            let mut writer = writer.sync().await.unwrap();
            assert!(!writer.needs_sync());

            // A buffered write is flush work even though the blob is clean.
            writer = writer.write_at(0, b"abc").await.unwrap();
            assert!(writer.needs_sync());
            writer = writer.sync().await.unwrap();
            assert!(!writer.needs_sync());

            // A started sync needs observation while it is in flight and after it completes.
            writer = writer.write_at(3, b"d").await.unwrap();
            let (writer, handle) = writer.start_sync().await.unwrap();
            assert!(writer.needs_sync());
            next_pending_sync(&pending).release.send(Ok(())).unwrap();
            handle.await.unwrap();
            assert!(writer.needs_sync());
            let writer = writer.wait_for_sync().await.unwrap();
            assert!(!writer.needs_sync());

            // A failed started sync keeps the writer in need of a sync. The handle reports the
            // failure, and so does the next sync.
            let writer = writer.write_at(4, b"e").await.unwrap();
            let (writer, handle) = writer.start_sync().await.unwrap();
            next_pending_sync(&pending)
                .release
                .send(Err(Error::Closed))
                .unwrap();
            assert!(handle.await.is_err());
            assert!(writer.needs_sync());
            assert!(writer.sync().await.is_err());
        });
    }

    /// Observing a completed started sync lets the next write use range sync.
    #[test_traced]
    fn test_write_start_sync_persists_and_marks_clean() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let blob = SyncTrackingBlob::new();
            let blob = Arc::new(blob);
            let mut writer = Write::new(blob.clone(), 0, NZUsize!(8));

            // Start a sync for buffered bytes and wait for the returned handle.
            writer = writer.write_at(0, b"abc").await.unwrap();
            let (mut writer, handle) = writer.start_sync().await.unwrap();
            handle.await.unwrap();

            // The buffered write required a full sync because the fresh writer starts dirty.
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abc");
            assert_eq!(writes, 1);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 0);

            // The next mutation observes the completed sync and can sync just its new range.
            writer = writer.write_at(3, b"d").await.unwrap();
            writer = writer.sync().await.unwrap();
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abcd");
            assert_eq!(writes, 2);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 1);

            // Nothing left to sync.
            let (_, handle) = writer.start_sync().await.unwrap();
            handle.await.unwrap();
            let (_, _, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 1);
        });
    }

    // Verifies sync waits for an outstanding start_sync instead of starting new disk work.
    #[test_traced]
    fn test_write_sync_waits_for_outstanding_start_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let inner = SyncTrackingBlob::new();
            let inner = Arc::new(inner);
            let (blob, pending) = DelayedSyncBlob::new(inner.clone());
            let writer = Write::new(blob, 0, NZUsize!(8));

            // Hold the started sync open so a later sync cannot finish right away.
            let (writer, handle) = writer.start_sync().await.unwrap();
            let deferred = next_pending_sync(&pending);

            // The attempted sync reaches the pending handle and cannot complete yet.
            let mut sync = Box::pin(writer.sync());
            assert!(
                sync.as_mut().now_or_never().is_none(),
                "sync must wait for the outstanding start_sync handle"
            );
            deferred
                .blocked
                .await
                .expect("sync never waited on start_sync");

            let (_, _, full_syncs, range_syncs) = inner.snapshot();
            assert_eq!(full_syncs, 0);
            assert_eq!(range_syncs, 0);

            // Releasing the original handle lets sync observe the completed disk sync.
            deferred.release.send(Ok(())).unwrap();
            sync.await.unwrap();
            handle.await.unwrap();
            let (_, _, full_syncs, range_syncs) = inner.snapshot();
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 0);
        });
    }

    // Verifies writes made after start_sync wait before they are flushed.
    #[test_traced]
    fn test_write_sync_after_start_sync_and_small_write_waits_before_range_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let inner = SyncTrackingBlob::new();
            let inner = Arc::new(inner);
            let (blob, pending) = DelayedSyncBlob::new(inner.clone());
            let writer = Write::new(blob, 0, NZUsize!(8));

            // Begin syncing the initial dirty state and keep that sync blocked.
            let (mut writer, handle) = writer.start_sync().await.unwrap();
            let deferred = next_pending_sync(&pending);

            // The tip must not reach the blob while the earlier sync is pending.
            writer = writer.write_at(0, b"abc").await.unwrap();
            let mut sync = Box::pin(writer.sync());
            assert!(
                sync.as_mut().now_or_never().is_none(),
                "sync must wait for the outstanding start_sync before flushing the small write"
            );
            deferred
                .blocked
                .await
                .expect("sync never waited on start_sync");

            let (_, writes, full_syncs, range_syncs) = inner.snapshot();
            assert_eq!(writes, 0);
            assert_eq!(full_syncs, 0);
            assert_eq!(range_syncs, 0);

            // After the earlier sync completes, the buffered write can be persisted.
            deferred.release.send(Ok(())).unwrap();
            sync.await.unwrap();
            handle.await.unwrap();

            let (durable, writes, full_syncs, range_syncs) = inner.snapshot();
            assert_eq!(durable.as_slice(), b"abc");
            assert_eq!(writes, 1);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 1);
        });
    }

    // Verifies overlapping writes wait before flushing buffered bytes while start_sync is pending.
    #[test_traced]
    fn test_write_at_overlap_flush_waits_for_outstanding_start_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let inner = SyncTrackingBlob::new();
            let inner = Arc::new(inner);
            inner
                .write_at(0, b"xxx", WriteOptions::default())
                .await
                .unwrap();

            let (blob, pending) = DelayedSyncBlob::new(inner.clone());
            let writer = Write::new(blob, inner.size(), NZUsize!(8));

            let (mut writer, handle) = writer.start_sync().await.unwrap();
            let deferred = next_pending_sync(&pending);

            // This append is local while the earlier sync is pending.
            writer = writer.write_at(3, b"abc").await.unwrap();

            // The drained tip must not reach the blob while the earlier sync is pending.
            let mut write = Box::pin(writer.write_at(2, b"ZZ"));
            assert!(
                write.as_mut().now_or_never().is_none(),
                "overlapping write must wait for the outstanding start_sync before flushing"
            );
            deferred
                .blocked
                .await
                .expect("write never waited on start_sync");

            let (_, writes, full_syncs, range_syncs) = inner.snapshot();
            assert_eq!(writes, 1);
            assert_eq!(full_syncs, 0);
            assert_eq!(range_syncs, 0);

            // Releasing the sync lets the parked write reach the blob.
            deferred.release.send(Ok(())).unwrap();
            write.await.unwrap();
            handle.await.unwrap();

            let (_, writes, full_syncs, range_syncs) = inner.snapshot();
            assert_eq!(writes, 3);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 0);
        });
    }

    // Verifies resize does not mutate the blob before an outstanding start_sync completes.
    #[test_traced]
    fn test_write_resize_waits_for_outstanding_start_sync_before_resizing() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let inner = SyncTrackingBlob::new();
            let inner = Arc::new(inner);
            inner
                .write_at(0, b"abcdef", WriteOptions::default())
                .await
                .unwrap();

            let (blob, pending) = DelayedSyncBlob::new(inner.clone());
            let writer = Write::new(blob, inner.size(), NZUsize!(8));

            let (writer, handle) = writer.start_sync().await.unwrap();
            let deferred = next_pending_sync(&pending);
            let original_size = inner.size();

            // Resize must not reach the blob while the earlier sync is pending.
            let mut resize = Box::pin(writer.resize(3));
            assert!(
                resize.as_mut().now_or_never().is_none(),
                "resize must wait for the outstanding start_sync handle"
            );
            deferred
                .blocked
                .await
                .expect("resize never waited on start_sync");
            assert_eq!(
                inner.size(),
                original_size,
                "resize must not mutate the blob before the pending sync finishes"
            );

            // Releasing the sync lets the resize apply.
            deferred.release.send(Ok(())).unwrap();
            let writer = resize.await.unwrap();
            handle.await.unwrap();
            assert_eq!(writer.size(), 3);
            assert_eq!(inner.size(), 3);
        });
    }

    #[test_traced]
    fn test_write_sync_uses_range_sync_for_buffer_only_write() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let blob = SyncTrackingBlob::new();
            let blob = Arc::new(blob);
            let mut writer = Write::new(blob.clone(), 0, NZUsize!(8));

            // A fresh writer preserves one sync barrier for mutations that predate wrapping.
            writer = writer.sync().await.unwrap();
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert!(durable.is_empty());
            assert_eq!(writes, 0);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 0);

            // The write remains entirely buffered, so sync can make just this range durable.
            writer = writer.write_at(0, b"abc").await.unwrap();
            writer = writer.sync().await.unwrap();

            // No prior plain blob mutation required another full sync barrier.
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abc");
            assert_eq!(writes, 1);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 1);

            // The prior sync used a range-scoped write, so there is no pending full-sync barrier.
            writer.sync().await.unwrap();
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abc");
            assert_eq!(writes, 1);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 1);
        });
    }

    #[test_traced]
    fn test_write_sync_persists_pre_wrapped_blob_mutation() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let blob = SyncTrackingBlob::new();
            let blob = Arc::new(blob);

            // Simulate a plain blob mutation before the writer wraps it.
            blob.write_at(0, b"abc", WriteOptions::default())
                .await
                .unwrap();

            let mut writer = Write::new(blob.clone(), 3, NZUsize!(8));
            writer = writer.sync().await.unwrap();

            // The first sync must use a full barrier to make the pre-wrapped write durable.
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abc");
            assert_eq!(writes, 1);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 0);

            // After the barrier is clear, a buffered tip-only write can use range sync again.
            writer = writer.write_at(3, b"d").await.unwrap();
            writer.sync().await.unwrap();

            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abcd");
            assert_eq!(writes, 2);
            assert_eq!(full_syncs, 1);
            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(|_| async move {
            let blob = SyncTrackingBlob::new();
            let blob = Arc::new(blob);
            let mut writer = Write::new(blob.clone(), 0, NZUsize!(4));

            // This exceeds the buffer and forces a plain write before the final buffered tip.
            writer = writer.write_at(0, b"abcdef").await.unwrap();
            writer = writer.write_at(6, b"g").await.unwrap();
            writer = writer.sync().await.unwrap();

            // The final sync must cover both the prior plain write and the buffered tip.
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abcdefg");
            assert_eq!(writes, 2);
            assert_eq!(full_syncs, 1);
            assert_eq!(range_syncs, 0);

            // With no new writes, sync has no work left.
            writer = writer.sync().await.unwrap();
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abcdefg");
            assert_eq!(writes, 2);
            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 = writer.write_at(7, b"h").await.unwrap();
            writer.sync().await.unwrap();

            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abcdefgh");
            assert_eq!(writes, 3);
            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(|_| async move {
            let blob = SyncTrackingBlob::new();
            let blob = Arc::new(blob);
            let mut writer = Write::new(blob.clone(), 0, NZUsize!(8));
            writer = writer.sync().await.unwrap();

            // Establish already-durable data with a range sync.
            writer = writer.write_at(0, b"abcdef").await.unwrap();
            writer = writer.sync().await.unwrap();

            // Resize alone is an unsynced blob mutation.
            writer = writer.resize(4).await.unwrap();
            writer.sync().await.unwrap();

            // The resized contents require a full sync barrier to become durable.
            let (durable, writes, full_syncs, range_syncs) = blob.snapshot();
            assert_eq!(durable.as_slice(), b"abcd");
            assert_eq!(writes, 1);
            assert_eq!(full_syncs, 2);
            assert_eq!(range_syncs, 1);
        });
    }
}