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 from [SyncState::wait_for_pending] on the next operation,
///   which also marks the state [SyncState::Dirty] since the mutations still need durability.
enum SyncState {
    // No unsynced mutations.
    Clean,
    // Unsynced mutations need a sync.
    Dirty,
    // A started sync is in flight.
    Pending(Completion),
}

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

    /// Mark a new unsynced mutation.
    fn mark_dirty(&mut self) {
        assert!(
            !matches!(self, Self::Pending(_)),
            "pending sync must be joined before marking dirty"
        );
        *self = Self::Dirty;
    }

    /// 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(());
        };
        match pending.wait().await {
            Ok(()) => {
                *self = Self::Clean;
                Ok(())
            }
            Err(err) => {
                // The sync failed, so the pending mutations still need durability.
                *self = Self::Dirty;
                Err(err)
            }
        }
    }

    /// 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.mark_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 => {
                // If this fails, a later sync must still cover the attempted write.
                self.mark_dirty();
                blob.write_at(offset, bufs, options).await?;
                *self = Self::Clean;
                Ok(())
            }
            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.mark_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 _, BufferPool, BufferPoolConfig, Error, Handle, IoBufMut, IoBufs, IoBufsMut,
        ReadOptions, Runner, Storage, WriteOptions, deterministic,
        mocks::{DelayedSyncBlob, next_pending_sync},
        telemetry::metrics::Registry,
    };
    use commonware_macros::test_traced;
    use commonware_utils::{NZU32, 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.
    #[derive(Clone)]
    pub struct SyncTrackingBlob {
        state: Arc<Mutex<RangeSyncState>>,
    }

    impl SyncTrackingBlob {
        pub fn new() -> Self {
            Self {
                state: Arc::new(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::from_pooler(&context, 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_allocates_lazily_on_first_refill() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let data = b"hello world";
            let (blob, _) = context.open("partition", b"lazy").await.unwrap();
            blob.write_at(0, data, WriteOptions::default())
                .await
                .unwrap();

            // A dedicated single-slot pool: reader construction must not touch
            // it (an eagerly-allocated unwritten buffer would be checked out of
            // the pool and, under empty-freeze semantics, released straight
            // back).
            let mut registry = Registry::default();
            let config = BufferPoolConfig::for_storage().with_max_per_class(NZU32!(1));
            let pool = BufferPool::new(config, &mut registry);
            let mut reader = Read::new(blob, data.len() as u64, NZUsize!(10), pool.clone());

            // No size class may report a created buffer: construction must not
            // check one out of the pool even transiently. The created gauge
            // registers per-class label sets lazily, so a clean pool encodes no
            // member lines at all.
            let encoded = registry.encode();
            assert!(
                !encoded
                    .lines()
                    .any(|line| line.starts_with("buffer_pool_created{")),
                "reader construction created a pool buffer: {encoded}"
            );

            // The single slot is also unclaimed at read time.
            let probe = pool
                .try_alloc(1)
                .expect("reader construction must not retain a buffer");
            drop(probe);

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

    #[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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, blob.clone(), 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::from_pooler(&context, 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::from_pooler(&context, 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::from_pooler(&context, blob.clone(), 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::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(8));
            writer.write_at(0, b"hello").await.unwrap();
            assert_eq!(writer.size(), 5);
            writer.sync().await.unwrap();
            assert_eq!(writer.size(), 5);

            // Verify data was written correctly
            let (blob, size) = context.open("partition", b"write_basic").await.unwrap();
            assert_eq!(size, 5);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(4));
            writer.write_at(0, b"abc").await.unwrap();
            assert_eq!(writer.size(), 3);
            writer.write_at(3, b"defg").await.unwrap();
            assert_eq!(writer.size(), 7);
            writer.sync().await.unwrap();

            // Verify the final result
            let (blob, size) = context.open("partition", b"write_multi").await.unwrap();
            assert_eq!(size, 7);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(4));
            writer.write_at(0, b"abc").await.unwrap();
            assert_eq!(writer.size(), 3);
            writer
                .write_at(3, b"defghijklmnopqrstuvwxyz")
                .await
                .unwrap();
            assert_eq!(writer.size(), 26);
            writer.sync().await.unwrap();
            assert_eq!(writer.size(), 26);

            // Verify the complete data
            let (blob, size) = context.open("partition", b"write_large").await.unwrap();
            assert_eq!(size, 26);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(10));

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

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

            // Verify the complete result
            let (blob, size) = context.open("partition", b"append_buf").await.unwrap();
            assert_eq!(size, 11);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(20));

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

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

            // Verify overwrite result
            let (blob, size) = context.open("partition", b"middle_buf").await.unwrap();
            assert_eq!(size, 10);
            let mut reader = Read::from_pooler(&context, blob, size, NZUsize!(10));
            let read = reader.read(10).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"ab01234hij");

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

            // Verify final result
            let (blob, size) = context.open("partition", b"middle_buf").await.unwrap();
            assert_eq!(size, 20);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(10));

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

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

            // Verify data placement with gap
            let (blob, size) = context.open("partition", b"before_buf").await.unwrap();
            assert_eq!(size, 20);
            let mut reader = Read::from_pooler(&context, blob, 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());

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

            // Verify gap is filled
            let (blob, size) = context.open("partition", b"before_buf").await.unwrap();
            assert_eq!(size, 20);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob, size, NZUsize!(10));

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

            let (blob_check, size_check) =
                context.open("partition", b"resize_write").await.unwrap();
            assert_eq!(size_check, 11);
            drop(blob_check);

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

            // Verify resize
            let (blob, size) = context.open("partition", b"resize_write").await.unwrap();
            assert_eq!(size, 5);
            let mut reader = Read::from_pooler(&context, blob, size, NZUsize!(5));
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"hello");

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

            // Verify overwrite
            let (blob, size) = context.open("partition", b"resize_write").await.unwrap();
            assert_eq!(size, 5);
            let mut reader = Read::from_pooler(&context, blob, size, NZUsize!(5));
            let read = reader.read(5).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"Xello");

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

            // Verify resize
            let (blob, size) = context.open("partition", b"resize_write").await.unwrap();
            assert_eq!(size, 10);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob_zero.clone(), size, NZUsize!(10));
            writer_zero.write_at(0, b"some data").await.unwrap();
            assert_eq!(writer_zero.size(), 9);
            writer_zero.sync().await.unwrap();
            assert_eq!(writer_zero.size(), 9);
            writer_zero.resize(0).await.unwrap();
            assert_eq!(writer_zero.size(), 0);
            writer_zero.sync().await.unwrap();
            assert_eq!(writer_zero.size(), 0);

            // Ensure the blob is empty
            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::from_pooler(&context, blob, size, NZUsize!(10));

            writer.write_at(0, b"hello").await.unwrap();
            writer.resize(10).await.unwrap();
            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::from_pooler(&context, blob.clone(), size, NZUsize!(10));

            // Write data that stays in buffer
            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.write_at(8, b" and flushed").await.unwrap();
            assert_eq!(writer.size(), 20);
            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.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.sync().await.unwrap();
            assert_eq!(writer.size(), 30);
            let (final_blob, final_size) =
                context.open("partition", b"read_at_writer").await.unwrap();
            assert_eq!(final_size, 30);
            let mut final_reader =
                Read::from_pooler(&context, 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::from_pooler(&context, blob, size, NZUsize!(8));
            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::from_pooler(&context, blob.clone(), size, NZUsize!(10));

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

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

            // Verify data with gap
            let (blob_check, size_check) =
                context.open("partition", b"write_straddle").await.unwrap();
            assert_eq!(size_check, 18);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob2.clone(), size, NZUsize!(10));
            writer2.write_at(0, b"0123456789").await.unwrap();
            assert_eq!(writer2.size(), 10);

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

            // Verify overwrite result
            let (blob_check2, size_check2) =
                context.open("partition", b"write_straddle2").await.unwrap();
            assert_eq!(size_check2, 17);
            let mut reader2 = Read::from_pooler(&context, 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 {
            // Test that closing writer flushes and persists buffered data
            let (blob_orig, size) = context.open("partition", b"write_close").await.unwrap();
            let mut writer = Write::from_pooler(&context, blob_orig.clone(), size, NZUsize!(8));
            writer.write_at(0, b"pending").await.unwrap();
            assert_eq!(writer.size(), 7);

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

            // Verify data persistence
            let (blob_check, size_check) = context.open("partition", b"write_close").await.unwrap();
            assert_eq!(size_check, 7);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(5));

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

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

            // Verify direct write worked
            let (blob_check, size_check) = context
                .open("partition", b"write_direct_size")
                .await
                .unwrap();
            assert_eq!(size_check, 10);
            let mut reader = Read::from_pooler(&context, blob_check, size_check, NZUsize!(10));
            let read = reader.read(10).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), data_large.as_slice());

            // Now write small data that should be buffered
            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.sync().await.unwrap();

            // Verify final state
            let (blob_check2, size_check2) = context
                .open("partition", b"write_direct_size")
                .await
                .unwrap();
            assert_eq!(size_check2, 13);
            let mut reader2 = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(15));

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

            // Overwrite and extend within buffer capacity
            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.sync().await.unwrap();

            // Verify persisted result
            let (blob_check, size_check) = context
                .open("partition", b"overwrite_extend_buf")
                .await
                .unwrap();
            assert_eq!(size_check, 15);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(20));

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

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

            // Verify complete result
            let (blob_check, size_check) = context.open("partition", b"write_end").await.unwrap();
            assert_eq!(size_check, 13);
            let mut reader = Read::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(5));

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

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

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

            // Verify final content
            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::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(10));

            // Initial buffered write
            writer.write_at(0, b"INITIAL").await.unwrap(); // 7 bytes
            assert_eq!(writer.size(), 7);
            // Buffer contains "INITIAL", inner.position = 0

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

            // Append at the new size
            writer.write_at(writer.size(), b"APPEND").await.unwrap();
            assert_eq!(writer.size(), 35); // 29 + 6
            writer.sync().await.unwrap();
            assert_eq!(writer.size(), 35);

            // Verify final content
            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::from_pooler(&context, 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::from_pooler(&context, blob.clone(), size, NZUsize!(10));

            // Write initial data and sync
            writer.write_at(0, b"0123456789ABCDEF").await.unwrap(); // 16 bytes
            assert_eq!(writer.size(), 16);
            writer.sync().await.unwrap(); // inner.position = 16, buffer empty
            assert_eq!(writer.size(), 16);

            // Resize
            let resize_to = 5;
            writer.resize(resize_to).await.unwrap();
            // after resize, inner.position should be `resize_to` (5)
            // buffer should be empty
            assert_eq!(writer.size(), resize_to);
            writer.sync().await.unwrap(); // Ensure truncation is persisted for verify step
            assert_eq!(writer.size(), resize_to);

            // Append at the new (resized) size
            writer.write_at(writer.size(), b"XXXXX").await.unwrap(); // 5 bytes
            // inner.buffer = "XXXXX", inner.position = 5
            assert_eq!(writer.size(), 10); // 5 (resized) + 5 (XXXXX)
            writer.sync().await.unwrap();
            assert_eq!(writer.size(), 10);

            // Verify final content
            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::from_pooler(&context, blob_check, size_check, NZUsize!(10));
            let read = reader.read(10).await.unwrap().coalesce();
            assert_eq!(read.as_ref(), b"01234XXXXX");
        });
    }

    // Verifies start_sync flushes current bytes, completes durability, and marks the writer clean.
    #[test_traced]
    fn test_write_start_sync_persists_and_marks_clean() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let blob = SyncTrackingBlob::new();
            let mut writer = Write::from_pooler(&context, blob.clone(), 0, NZUsize!(8));

            // Start a sync for buffered bytes and wait for the returned handle.
            writer.write_at(0, b"abc").await.unwrap();
            let handle = writer.start_sync().await;
            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 started sync marked the writer clean, so the next buffered write can use a
            // range-scoped sync.
            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);

            // Nothing left to sync.
            let handle = writer.start_sync().await;
            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(|context| async move {
            let inner = SyncTrackingBlob::new();
            let (blob, pending) = DelayedSyncBlob::new(inner.clone());
            let mut writer = Write::from_pooler(&context, blob, 0, NZUsize!(8));

            // Hold the started sync open so a later sync cannot finish right away.
            let handle = writer.start_sync().await;
            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(|context| async move {
            let inner = SyncTrackingBlob::new();
            let (blob, pending) = DelayedSyncBlob::new(inner.clone());
            let mut writer = Write::from_pooler(&context, blob, 0, NZUsize!(8));

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

            // The tip must not reach the blob while the earlier sync is pending.
            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(|context| async move {
            let inner = SyncTrackingBlob::new();
            inner
                .write_at(0, b"xxx", WriteOptions::default())
                .await
                .unwrap();

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

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

            // This append is local while the earlier sync is pending.
            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(|context| async move {
            let inner = SyncTrackingBlob::new();
            inner
                .write_at(0, b"abcdef", WriteOptions::default())
                .await
                .unwrap();

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

            let handle = writer.start_sync().await;
            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();
            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(|context| async move {
            let blob = SyncTrackingBlob::new();
            let mut writer = Write::from_pooler(&context, blob.clone(), 0, NZUsize!(8));

            // A fresh writer preserves one sync barrier for mutations that predate wrapping.
            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.write_at(0, b"abc").await.unwrap();
            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(|context| async move {
            let blob = SyncTrackingBlob::new();

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

            let mut writer = Write::from_pooler(&context, blob.clone(), 3, NZUsize!(8));
            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.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_failed_range_sync_does_not_mark_clean() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let name = b"failed_range_sync";
            let (blob, size) = context.open("partition", name).await.unwrap();
            let mut writer = Write::from_pooler(&context, blob, size, NZUsize!(8));
            writer.sync().await.unwrap();

            // Keep the write buffered so sync attempts the clean range-scoped write path.
            writer.write_at(0, b"abc").await.unwrap();

            // Removing the blob makes the range-sync flush fail.
            context.remove("partition", Some(name)).await.unwrap();
            assert!(writer.sync().await.is_err());

            // The failed range-scoped write must leave a pending full-sync barrier, so a
            // later sync cannot report success.
            assert!(writer.sync().await.is_err());
        });
    }

    #[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 = SyncTrackingBlob::new();
            let mut 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, 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.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.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(|context| async move {
            let blob = SyncTrackingBlob::new();
            let mut writer = Write::from_pooler(&context, blob.clone(), 0, NZUsize!(8));
            writer.sync().await.unwrap();

            // 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, 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);
        });
    }
}