commonware_storage/metadata/

mod.rs

//! A key-value store optimized for atomically committing a small collection of metadata.
//!
//! [Metadata] is a key-value store optimized for tracking a small collection of metadata
//! that allows multiple updates to be committed in a single batch. It is commonly used with
//! a variety of other underlying storage systems to persist application state across restarts.
//!
//! # Format
//!
//! Data stored in [Metadata] is serialized as a sequence of key-value pairs in either a
//! "left" or "right" blob:
//!
//! ```text
//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
//! | 0 | 1 |    ...    | 8 | 9 |10 |11 |12 |13 |14 |15 |16 |  ...  |50 |...|90 |91 |92 |93 |
//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
//! |    Version (u64)  |      Key1     |              Value1           |...|  CRC32(u32)   |
//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
//! ```
//!
//! _To ensure the integrity of the data, a CRC32 checksum is appended to the end of the blob.
//! This ensures that partial writes are detected before any data is relied on._
//!
//! # Atomic Updates
//!
//! To provide support for atomic updates, [Metadata] maintains two blobs: a "left" and a "right"
//! blob. When a new update is committed, it is written to the "older" of the two blobs (indicated
//! by the version persisted). Writes to [commonware_runtime::Blob] are not atomic and may only
//! complete partially, so we only overwrite the "newer" blob once the "older" blob has been synced
//! (otherwise, we would not be guaranteed to recover the latest complete state from disk on
//! restart as half of a blob could be old data and half new data).
//!
//! # Delta Writes
//!
//! If the set of keys and the length of values are stable, [Metadata] will only write an update's
//! delta to disk (rather than rewriting the entire metadata). This makes [Metadata] a great choice
//! for maintaining even large collections of data (with the majority rarely modified).
//!
//! # Example
//!
//! ```rust
//! use commonware_runtime::{Spawner, Runner, deterministic};
//! use commonware_storage::metadata::{Metadata, Config};
//! use commonware_utils::sequence::U64;
//!
//! let executor = deterministic::Runner::default();
//! executor.start(|context| async move {
//!     // Create a store
//!     let mut metadata = Metadata::init(context, Config {
//!         partition: "partition".into(),
//!         codec_config: ((0..).into(), ()),
//!     }).await.unwrap();
//!
//!     // Store metadata
//!     metadata.put(U64::new(1), b"hello".to_vec());
//!     metadata.put(U64::new(2), b"world".to_vec());
//!
//!     // Sync the metadata store (batch write changes)
//!     metadata = metadata.sync().await.unwrap();
//!
//!     // Retrieve some metadata
//!     let value = metadata.get(&U64::new(1)).unwrap();
//!
//! });
//! ```

#[cfg(all(test, feature = "arbitrary"))]
mod conformance;
mod storage;
pub use storage::Metadata;
use thiserror::Error;

/// Errors that can occur when interacting with [Metadata].
#[derive(Debug, Error)]
pub enum Error {
    #[error("runtime error: {0}")]
    Runtime(#[from] commonware_runtime::Error),
    #[error("corruption: {0}")]
    Corruption(String),
}

/// Configuration for [Metadata] storage.
#[derive(Clone)]
pub struct Config<C> {
    /// The [commonware_runtime::Storage] partition to use for storing metadata.
    pub partition: String,

    /// The codec configuration to use for the value stored in the metadata.
    pub codec_config: C,
}

#[cfg(test)]
mod tests {
    use super::*;
    use commonware_formatting::hex;
    use commonware_macros::{test_group, test_traced};
    use commonware_runtime::{
        Blob, Metrics as _, ReadOptions, Runner, Storage, Supervisor as _, WriteOptions,
        deterministic,
        mocks::{
            DelayedSyncContext, PendingSyncs, RecordingContext, Recordings, WriteFaultContext,
            WriteFaults, drive_pending_syncs, fail_pending_syncs, release_pending_syncs,
        },
    };
    use commonware_utils::sequence::U64;
    use futures::FutureExt as _;
    use rand::{Rng, RngExt as _};

    fn assert_options(recordings: &Recordings, reads: &[ReadOptions], writes: &[WriteOptions]) {
        let snapshot = recordings.snapshot();
        assert_eq!(snapshot.reads.as_slice(), reads);
        assert_eq!(snapshot.writes.as_slice(), writes);
        recordings.clear();
    }

    fn assert_durability(pending: &PendingSyncs, calls: usize, starts: usize, completions: usize) {
        assert_eq!(pending.calls(), calls);
        assert_eq!(pending.starts(), starts);
        assert_eq!(pending.completions(), completions);
    }

    #[test_traced]
    fn test_io_options_and_durability() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let (recording, recordings) = RecordingContext::new(DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            });
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let key = U64::new(1);
            let extra_key = U64::new(2);
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(recording.child("first"), cfg.clone())
                    .await
                    .unwrap();

            // Seed both mirrors so equal-size updates take the incremental branch.
            metadata.put(key.clone(), vec![1; 8]);
            metadata = metadata.sync().await.unwrap();
            metadata = metadata.sync().await.unwrap();
            recordings.clear();
            pending.arm();

            // Non-pipelined incremental writes request cache bypass and retain a trailing sync.
            metadata.put(key.clone(), vec![2; 8]);
            metadata = drive_pending_syncs(&pending, metadata.sync())
                .await
                .unwrap();
            assert_options(
                &recordings,
                &[],
                &[
                    WriteOptions::DONT_CACHE,
                    WriteOptions::DONT_CACHE,
                    WriteOptions::DONT_CACHE,
                ],
            );
            assert_durability(&pending, 1, 0, 0);

            // Pipelined incremental writes request cache bypass and retain a started sync.
            metadata.put(key.clone(), vec![3; 8]);
            let (next, handle) = metadata.start_sync().await.unwrap();
            metadata = next;
            drive_pending_syncs(&pending, handle).await.unwrap();
            assert_options(
                &recordings,
                &[],
                &[
                    WriteOptions::DONT_CACHE,
                    WriteOptions::DONT_CACHE,
                    WriteOptions::DONT_CACHE,
                ],
            );
            assert_durability(&pending, 2, 1, 1);

            // A growing pipelined rewrite requests cache bypass and retains a started sync.
            metadata.put(extra_key.clone(), vec![4; 16]);
            let (next, handle) = metadata.start_sync().await.unwrap();
            metadata = next;
            drive_pending_syncs(&pending, handle).await.unwrap();
            assert_options(&recordings, &[], &[WriteOptions::DONT_CACHE]);
            assert_durability(&pending, 3, 2, 2);

            // A growing non-pipelined rewrite requests cache bypass and retains durability.
            metadata = drive_pending_syncs(&pending, metadata.sync())
                .await
                .unwrap();
            assert_options(
                &recordings,
                &[],
                &[WriteOptions::SYNC | WriteOptions::DONT_CACHE],
            );
            assert_durability(&pending, 4, 2, 2);

            // A shrinking pipelined rewrite requests cache bypass and retains a started sync.
            metadata.remove(&extra_key);
            let (next, handle) = metadata.start_sync().await.unwrap();
            metadata = next;
            drive_pending_syncs(&pending, handle).await.unwrap();
            assert_options(&recordings, &[], &[WriteOptions::DONT_CACHE]);
            assert_durability(&pending, 5, 3, 3);

            // A shrinking non-pipelined rewrite requests cache bypass and retains a trailing sync.
            metadata = drive_pending_syncs(&pending, metadata.sync())
                .await
                .unwrap();
            assert_options(&recordings, &[], &[WriteOptions::DONT_CACHE]);
            assert_durability(&pending, 6, 3, 3);

            // Both populated mirrors request cache bypass when reloaded.
            drop(metadata);
            let metadata = Metadata::<_, U64, Vec<u8>>::init(recording.child("second"), cfg)
                .await
                .unwrap();
            assert_options(
                &recordings,
                &[ReadOptions::DONT_CACHE, ReadOptions::DONT_CACHE],
                &[],
            );
            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_start_sync_pipelined_destroy() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();

            // Two pipelined syncs back to back: the second drains the first before targeting
            // the copy it left as last-known-durable.
            let key = U64::new(1);
            metadata.put(key.clone(), vec![3]);
            let (mut metadata, h1) = metadata.start_sync().await.unwrap();
            metadata.put(key.clone(), vec![4]);
            let (metadata, h2) = metadata.start_sync().await.unwrap();
            h1.await.unwrap();
            h2.await.unwrap();
            metadata.destroy().await.unwrap();

            // Destroy drained the pending sync, so nothing survives the reopen.
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(metadata.get(&key), None, "destroyed store must be empty");
        });
    }

    #[test_traced]
    fn test_start_sync_failure_fails_next_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(
                DelayedSyncContext {
                    inner: context.child("first"),
                    pending: pending.clone(),
                },
                cfg,
            )
            .await
            .unwrap();

            metadata.put(U64::new(1), vec![3]);
            let (mut metadata, handle) = metadata.start_sync().await.unwrap();
            fail_pending_syncs(&pending);
            assert!(handle.await.is_err());

            // The failed copy's on-disk state is unknown: the next sync observes the failure
            // and fails, consuming the store, without writing the only durable copy.
            metadata.put(U64::new(1), vec![4]);
            assert!(metadata.start_sync().await.is_err());
        });
    }

    #[test_traced]
    fn test_start_sync_dropped_handle_does_not_cancel() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(
                DelayedSyncContext {
                    inner: context.child("first"),
                    pending: pending.clone(),
                },
                cfg.clone(),
            )
            .await
            .unwrap();

            // Drop the handle while its sync is still parked: the sync must proceed anyway.
            let key = U64::new(1);
            metadata.put(key.clone(), vec![3]);
            let (metadata, handle) = metadata.start_sync().await.unwrap();
            drop(handle);
            release_pending_syncs(&pending);
            let metadata = drive_pending_syncs(&pending, metadata.sync())
                .await
                .unwrap();
            drop(metadata);

            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(metadata.get(&key), Some(&vec![3]));
        });
    }

    #[test_traced]
    fn test_start_sync_dropped_handle_fails_next_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(
                DelayedSyncContext {
                    inner: context.child("first"),
                    pending: pending.clone(),
                },
                cfg,
            )
            .await
            .unwrap();

            // Drop the handle before its sync fails: nobody observes the failure directly,
            // but the next sync does, failing without writing the only durable copy.
            metadata.put(U64::new(1), vec![3]);
            let (metadata, handle) = metadata.start_sync().await.unwrap();
            drop(handle);
            fail_pending_syncs(&pending);

            assert!(metadata.sync().await.is_err());
        });
    }

    #[test_traced]
    fn test_start_sync_newest_copy_wins_on_reopen() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();

            let key = U64::new(1);
            metadata.put(key.clone(), vec![3]);
            let (mut metadata, h1) = metadata.start_sync().await.unwrap();
            metadata.put(key.clone(), vec![4]);
            let (metadata, h2) = metadata.start_sync().await.unwrap();
            h1.await.unwrap();
            h2.await.unwrap();
            drop(metadata);

            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(metadata.get(&key), Some(&vec![4]));
        });
    }

    #[test_traced]
    fn test_start_sync_write_failure_consumes() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let faults = WriteFaults::default();
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(
                WriteFaultContext {
                    inner: context.child("first"),
                    faults: faults.clone(),
                },
                cfg.clone(),
            )
            .await
            .unwrap();

            // Establish a stable key order with equal-size values so later syncs take the
            // overwrite path, which mutates the target copy's state before writing.
            let key = U64::new(1);
            metadata.put(key.clone(), vec![1; 8]);
            metadata = metadata.sync().await.unwrap();
            metadata.put(key.clone(), vec![2; 8]);
            metadata = metadata.sync().await.unwrap();

            // The injected failure hits the inline writes, after the cursor has rotated onto
            // the target copy: the call fails, consuming the store.
            faults.arm();
            metadata.put(key.clone(), vec![3; 8]);
            assert!(metadata.start_sync().await.is_err());
            faults.disarm();

            // The store died with the target copy in an unknown state: a reopen recovers the
            // last durable value.
            let metadata = Metadata::<_, U64, Vec<u8>>::init(
                WriteFaultContext {
                    inner: context.child("second"),
                    faults,
                },
                cfg,
            )
            .await
            .unwrap();
            assert_eq!(metadata.get(&key), Some(&vec![2; 8]));
        });
    }

    #[test_traced]
    fn test_start_sync_second_sync_waits_for_first() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let faults = WriteFaults::default();
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(
                WriteFaultContext {
                    inner: DelayedSyncContext {
                        inner: context.child("first"),
                        pending: pending.clone(),
                    },
                    faults: faults.clone(),
                },
                cfg,
            )
            .await
            .unwrap();

            // Park the first sync, then start a second: until the first sync's fsync completes,
            // its target copy's on-disk state is unknown, so the second sync must not write a
            // single byte to the other (only durable) copy.
            let key = U64::new(1);
            metadata.put(key.clone(), vec![3]);
            let (mut metadata, handle) = metadata.start_sync().await.unwrap();
            metadata.put(key.clone(), vec![4]);
            let writes_before = faults.writes();
            let mut second = Box::pin(metadata.sync());
            for _ in 0..8 {
                assert!((&mut second).now_or_never().is_none());
            }
            assert_eq!(faults.writes(), writes_before, "second sync must not write");
            assert_eq!(
                pending.starts(),
                1,
                "second sync must not start a blob sync"
            );

            release_pending_syncs(&pending);
            handle.await.unwrap();
            let metadata = drive_pending_syncs(&pending, second).await.unwrap();
            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_put_get_clear() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
                .await
                .unwrap();

            // Get a key that doesn't exist
            let key = U64::new(42);
            let value = metadata.get(&key);
            assert!(value.is_none());

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 0"));
            assert!(buffer.contains("first_sync_overwrites_total 0"));
            assert!(buffer.contains("first_keys 0"));

            // Put a key
            let hello = b"hello".to_vec();
            metadata.put(key.clone(), hello.clone());

            // Get the key
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, &hello);

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 0"));
            assert!(buffer.contains("first_sync_overwrites_total 0"));
            assert!(buffer.contains("first_keys 1"));

            // Sync the metadata store
            metadata = metadata.sync().await.unwrap();

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 1"));
            assert!(buffer.contains("first_sync_overwrites_total 0"));
            assert!(buffer.contains("first_keys 1"));

            // Reopen the metadata store
            drop(metadata);
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("second_sync_rewrites_total 0"));
            assert!(buffer.contains("second_sync_overwrites_total 0"));
            assert!(buffer.contains("second_keys 1"));

            // Get the key
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, &hello);

            // Test clearing the metadata store
            metadata.clear();
            let value = metadata.get(&key);
            assert!(value.is_none());

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("second_sync_rewrites_total 0"));
            assert!(buffer.contains("second_sync_overwrites_total 0"));
            assert!(buffer.contains("second_keys 0"));

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_put_returns_previous_value() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
                .await
                .unwrap();

            let key = U64::new(42);

            // First put returns None (no previous value)
            let previous = metadata.put(key.clone(), b"first".to_vec());
            assert!(previous.is_none());

            // Second put returns the previous value
            let previous = metadata.put(key.clone(), b"second".to_vec());
            assert_eq!(previous, Some(b"first".to_vec()));

            // Third put returns the previous value
            let previous = metadata.put(key.clone(), b"third".to_vec());
            assert_eq!(previous, Some(b"second".to_vec()));

            // Current value is the latest
            assert_eq!(metadata.get(&key), Some(&b"third".to_vec()));

            // Different key returns None
            let other_key = U64::new(99);
            let previous = metadata.put(other_key.clone(), b"other".to_vec());
            assert!(previous.is_none());

            // Sync and verify persistence
            metadata.sync().await.unwrap();

            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // After restart, put still returns previous value
            let previous = metadata.put(key.clone(), b"fourth".to_vec());
            assert_eq!(previous, Some(b"third".to_vec()));

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_multi_sync() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
                .await
                .unwrap();

            // Put a key
            let key = U64::new(42);
            let hello = b"hello".to_vec();
            metadata.put(key.clone(), hello.clone());

            // Sync the metadata store
            metadata = metadata.sync().await.unwrap();

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 1"));
            assert!(buffer.contains("first_sync_overwrites_total 0"));
            assert!(buffer.contains("first_keys 1"));

            // Put an overlapping key and a new key
            let world = b"world".to_vec();
            metadata.put(key.clone(), world.clone());
            let key2 = U64::new(43);
            let foo = b"foo".to_vec();
            metadata.put(key2.clone(), foo.clone());

            // Sync the metadata store
            metadata = metadata.sync().await.unwrap();

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 2"));
            assert!(buffer.contains("first_sync_overwrites_total 0"));
            assert!(buffer.contains("first_keys 2"));

            // Reopen the metadata store
            drop(metadata);
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("second_sync_rewrites_total 0"));
            assert!(buffer.contains("second_sync_overwrites_total 0"));
            assert!(buffer.contains("second_keys 2"));

            // Get the key
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, &world);
            let value = metadata.get(&key2).unwrap();
            assert_eq!(value, &foo);

            // Remove the key
            metadata.remove(&key);

            // Sync the metadata store
            metadata = metadata.sync().await.unwrap();

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("second_sync_rewrites_total 1"));
            assert!(buffer.contains("second_sync_overwrites_total 0"));
            assert!(buffer.contains("second_keys 1"));

            // Reopen the metadata store
            drop(metadata);
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("third"), cfg)
                .await
                .unwrap();

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("third_sync_rewrites_total 0"));
            assert!(buffer.contains("third_sync_overwrites_total 0"));
            assert!(buffer.contains("third_keys 1"));

            // Get the key
            let value = metadata.get(&key);
            assert!(value.is_none());
            let value = metadata.get(&key2).unwrap();
            assert_eq!(value, &foo);

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_recover_corrupted_one() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
                .await
                .unwrap();

            // Put a key
            let key = U64::new(42);
            let hello = b"hello".to_vec();
            metadata.put(key.clone(), hello.clone());

            // Sync the metadata store
            metadata = metadata.sync().await.unwrap();

            // Put an overlapping key and a new key
            let world = b"world".to_vec();
            metadata.put(key.clone(), world.clone());
            let key2 = U64::new(43);
            let foo = b"foo".to_vec();
            metadata.put(key2, foo.clone());

            // Sync the metadata store
            metadata.sync().await.unwrap();

            // Corrupt the metadata store
            let (blob, _) = context.open("test", b"left").await.unwrap();
            blob.write_at(0, b"corrupted".to_vec(), WriteOptions::SYNC)
                .await
                .unwrap();

            // Reopen the metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Get the key (falls back to non-corrupt)
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, &hello);

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_recovered_mirror_supports_shrinking_rewrite() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();
            let key = U64::new(42);
            let hello = b"hello".to_vec();
            metadata.put(key.clone(), hello.clone());
            metadata = metadata.sync().await.unwrap();
            metadata.put(key.clone(), b"world".to_vec());
            metadata.put(U64::new(43), b"foo".to_vec());
            metadata.sync().await.unwrap();

            // Corrupt the newer copy so the next initialization must repair it.
            let (blob, _) = context.open("test", b"left").await.unwrap();
            blob.write_at(0, b"corrupted".to_vec(), WriteOptions::SYNC)
                .await
                .unwrap();

            // The repaired copy must support a shrinking rewrite: a stale tail left behind by
            // recovery would survive the smaller write and poison the next reopen.
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg.clone())
                    .await
                    .unwrap();
            assert_eq!(metadata.get(&key).unwrap(), &hello);
            metadata.clear();
            metadata.sync().await.unwrap();

            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("third"), cfg)
                .await
                .unwrap();
            assert!(metadata.get(&key).is_none());
            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_recover_corrupted_both() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
                .await
                .unwrap();

            // Put a key
            let key = U64::new(42);
            let hello = b"hello".to_vec();
            metadata.put(key.clone(), hello.clone());

            // Sync the metadata store
            metadata = metadata.sync().await.unwrap();

            // Put an overlapping key and a new key
            let world = b"world".to_vec();
            metadata.put(key.clone(), world.clone());
            let key2 = U64::new(43);
            let foo = b"foo".to_vec();
            metadata.put(key2, foo.clone());

            // Sync the metadata store
            metadata.sync().await.unwrap();

            // Corrupt the metadata store
            let (blob, _) = context.open("test", b"left").await.unwrap();
            blob.write_at(0, b"corrupted".to_vec(), WriteOptions::SYNC)
                .await
                .unwrap();
            let (blob, _) = context.open("test", b"right").await.unwrap();
            blob.write_at(0, b"corrupted".to_vec(), WriteOptions::SYNC)
                .await
                .unwrap();

            // Both copies failing validation is impossible under a crash (syncs alternate and
            // drain), so reopening must fail loudly rather than adopt a fresh store.
            for child in ["second", "third"] {
                let cfg = Config {
                    partition: "test".into(),
                    codec_config: ((0..).into(), ()),
                };
                let result = Metadata::<_, U64, Vec<u8>>::init(context.child(child), cfg).await;
                assert!(matches!(result, Err(Error::Corruption(_))));
            }
        });
    }

    #[test_traced]
    fn test_recover_corrupted_truncate() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::init(context.child("first"), cfg).await.unwrap();

            // Put a key
            let key = U64::new(42);
            let hello = b"hello".to_vec();
            metadata.put(key.clone(), hello.clone());

            // Sync the metadata store
            metadata = metadata.sync().await.unwrap();

            // Put an overlapping key and a new key
            let world = b"world".to_vec();
            metadata.put(key.clone(), world.clone());
            let key2 = U64::new(43);
            let foo = b"foo".to_vec();
            metadata.put(key2, foo.clone());

            // Sync the metadata store
            metadata.sync().await.unwrap();

            // Corrupt the metadata store
            let (blob, len) = context.open("test", b"left").await.unwrap();
            blob.resize(len - 8).await.unwrap();
            blob.sync().await.unwrap();

            // Reopen the metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Get the key (falls back to non-corrupt)
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, &hello);

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_recover_corrupted_short() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::init(context.child("first"), cfg).await.unwrap();

            // Put a key
            let key = U64::new(42);
            let hello = b"hello".to_vec();
            metadata.put(key.clone(), hello.clone());

            // Sync the metadata store
            metadata = metadata.sync().await.unwrap();

            // Put an overlapping key and a new key
            let world = b"world".to_vec();
            metadata.put(key.clone(), world.clone());
            let key2 = U64::new(43);
            let foo = b"foo".to_vec();
            metadata.put(key2, foo.clone());

            // Sync the metadata store
            metadata.sync().await.unwrap();

            // Corrupt the metadata store
            let (blob, _) = context.open("test", b"left").await.unwrap();
            blob.resize(5).await.unwrap();
            blob.sync().await.unwrap();

            // Reopen the metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Get the key (falls back to non-corrupt)
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, &hello);

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_unclean_shutdown() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let key = U64::new(42);
            let hello = b"hello".to_vec();
            {
                // Create a metadata store
                let cfg = Config {
                    partition: "test".into(),
                    codec_config: ((0..).into(), ()),
                };
                let mut metadata = Metadata::init(context.child("first"), cfg).await.unwrap();

                // Put a key
                metadata.put(key.clone(), hello.clone());

                // Drop metadata before sync
            }

            // Reopen the metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Get the key
            let value = metadata.get(&key);
            assert!(value.is_none());

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("second_sync_rewrites_total 0"));
            assert!(buffer.contains("second_sync_overwrites_total 0"));
            assert!(buffer.contains("second_keys 0"));

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    #[should_panic(expected = "usize value is larger than u32")]
    fn test_value_too_big_error() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::init(context.child("storage"), cfg).await.unwrap();

            // Create a value that exceeds u32::MAX bytes
            let value = vec![0u8; (u32::MAX as usize) + 1];
            metadata.put(U64::new(1), value);

            // Assert
            metadata.sync().await.unwrap();
        });
    }

    #[test_traced]
    fn test_delta_writes() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::init(context.child("storage"), cfg).await.unwrap();

            // Put initial keys
            for i in 0..100 {
                metadata.put(U64::new(i), vec![i as u8; 100]);
            }

            // First sync - should write everything to the first blob
            //
            // 100 keys * (8 bytes for key + 1 byte for len + 100 bytes for value) + 8 bytes for version + 4 bytes for checksum
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 1"), "{buffer}");
            assert!(buffer.contains("sync_overwrites_total 0"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 10912"),
                "{buffer}",
            );

            // Modify just one key
            metadata.put(U64::new(51), vec![0xff; 100]);

            // Sync again - should write everything to the second blob
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"), "{buffer}");
            assert!(buffer.contains("sync_overwrites_total 0"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 21824"),
                "{buffer}",
            );

            // Sync again - should write only diff from the first blob
            //
            // 1 byte for len + 100 bytes for value + 8 byte for version + 4 bytes for checksum
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"), "{buffer}");
            assert!(buffer.contains("sync_overwrites_total 1"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 21937"),
                "{buffer}",
            );

            // Sync again - both blobs already contain the latest state
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"), "{buffer}");
            assert!(buffer.contains("sync_overwrites_total 1"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 21937"),
                "{buffer}",
            );

            // Remove a key - should rewrite everything
            //
            // 99 keys * (8 bytes for key + 1 bytes for len + 100 bytes for value) + 8 bytes for version + 4 bytes for checksum
            metadata.remove(&U64::new(51));
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 3"), "{buffer}");
            assert!(buffer.contains("sync_overwrites_total 1"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 32740"),
                "{buffer}"
            );

            // Sync again - should also rewrite
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 4"), "{buffer}");
            assert!(buffer.contains("sync_overwrites_total 1"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 43543"),
                "{buffer}"
            );

            // Modify in-place - should overwrite
            //
            // 1 byte for len + 100 bytes for value + 8 byte for version + 4 bytes for checksum
            metadata.put(U64::new(50), vec![0xff; 100]);
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 4"), "{buffer}");
            assert!(buffer.contains("sync_overwrites_total 2"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 43656"),
                "{buffer}"
            );

            // Clean up
            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_multi_key_overwrites() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();

            // Put initial keys and populate both blobs
            for i in 0..100 {
                metadata.put(U64::new(i), vec![i as u8; 100]);
            }
            metadata = metadata.sync().await.unwrap();
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 2"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 21824"),
                "{buffer}",
            );

            // Modify several keys with same-size values
            for i in [10u64, 11, 12, 50, 98, 99] {
                metadata.put(U64::new(i), vec![0xAA; 100]);
            }

            // Sync writes one delta per modified value.
            //
            // 6 * (1 byte for len + 100 bytes for value) + 8 bytes for version
            // + 4 bytes for checksum.
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 2"), "{buffer}");
            assert!(buffer.contains("first_sync_overwrites_total 1"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 22442"),
                "{buffer}",
            );

            // Sync again - the same deltas propagate to the other blob
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_overwrites_total 2"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 23060"),
                "{buffer}",
            );

            // Sync again - both blobs already contain the latest state
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 2"), "{buffer}");
            assert!(buffer.contains("first_sync_overwrites_total 2"), "{buffer}");
            assert!(
                buffer.contains("runtime_storage_write_bytes_total 23060"),
                "{buffer}",
            );

            // Mix a same-size update with a size-changing update. The overwrite
            // scan updates the mirror for the smaller key before the size change
            // forces a rewrite, which must discard that partial mutation.
            metadata.put(U64::new(20), vec![0xBB; 100]);
            metadata.put(U64::new(30), vec![0xCC; 150]);
            metadata = metadata.sync().await.unwrap();
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 4"), "{buffer}");
            assert!(buffer.contains("first_sync_overwrites_total 2"), "{buffer}");

            // Restart the metadata store
            drop(metadata);
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Verify every value survived exactly
            for i in 0..100u64 {
                let expected = match i {
                    10 | 11 | 12 | 50 | 98 | 99 => vec![0xAA; 100],
                    20 => vec![0xBB; 100],
                    30 => vec![0xCC; 150],
                    _ => vec![i as u8; 100],
                };
                assert_eq!(metadata.get(&U64::new(i)).unwrap(), &expected, "key {i}");
            }

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_sync_with_no_changes() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();

            // Put initial data
            metadata = metadata
                .put_sync(U64::new(1), b"hello".to_vec())
                .await
                .unwrap();

            // Sync again with no changes. This still rewrites because only one blob
            // has the new key order.
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"));
            assert!(buffer.contains("sync_overwrites_total 0"));

            // Sync again - both blobs already contain the latest state
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"));
            assert!(buffer.contains("sync_overwrites_total 0"));

            // Sync again - should remain a no-op
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"));
            assert!(buffer.contains("sync_overwrites_total 0"));

            // Restart the metadata store and verify the no-op left durable state
            drop(metadata);
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(metadata.get(&U64::new(1)).unwrap(), b"hello");

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_get_mut_marks_modified() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();

            // Put initial data
            metadata = metadata
                .put_sync(U64::new(1), b"hello".to_vec())
                .await
                .unwrap();

            // Sync again to ensure both blobs are populated
            metadata = metadata.sync().await.unwrap();

            // Use get_mut to modify value
            let value = metadata.get_mut(&U64::new(1)).unwrap();
            value[0] = b'H';

            // Sync should detect the modification and do a rewrite (due to recent key_order_changed)
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 2"));
            assert!(buffer.contains("first_sync_overwrites_total 1"));

            // Restart the metadata store
            drop(metadata);
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Verify the change persisted
            let value = metadata.get(&U64::new(1)).unwrap();
            assert_eq!(value[0], b'H');

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_mixed_operation_sequences() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();

            let key = U64::new(1);

            // Test: put -> remove -> put same key
            metadata.put(key.clone(), b"first".to_vec());
            metadata.remove(&key);
            metadata = metadata
                .put_sync(key.clone(), b"second".to_vec())
                .await
                .unwrap();
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, b"second");

            // Test: put -> get_mut -> remove -> put
            metadata.put(key.clone(), b"third".to_vec());
            let value = metadata.get_mut(&key).unwrap();
            value[0] = b'T';
            metadata.remove(&key);
            metadata = metadata
                .put_sync(key.clone(), b"fourth".to_vec())
                .await
                .unwrap();
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, b"fourth");

            // Restart the metadata store
            drop(metadata);
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Verify the changes persisted
            let value = metadata.get(&key).unwrap();
            assert_eq!(value, b"fourth");

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_overwrite_vs_rewrite() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("storage"), cfg)
                .await
                .unwrap();

            // Set up initial data
            metadata.put(U64::new(1), vec![1; 10]);
            metadata.put(U64::new(2), vec![2; 10]);
            metadata = metadata.sync().await.unwrap();

            // Same size modification before both blobs are populated
            metadata.put(U64::new(1), vec![0xFF; 10]);
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"));
            assert!(buffer.contains("sync_overwrites_total 0"));

            // Let key order stabilize with another sync
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"));
            assert!(buffer.contains("sync_overwrites_total 1"));

            // Same size modification after both blobs are populated - should overwrite
            metadata.put(U64::new(1), vec![0xAA; 10]);
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 2"));
            assert!(buffer.contains("sync_overwrites_total 2"));

            // Different size modification - should rewrite
            metadata.put(U64::new(1), vec![0xFF; 20]);
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 3"));
            assert!(buffer.contains("sync_overwrites_total 2"));

            // Add new key - should rewrite (key order changed)
            metadata.put(U64::new(3), vec![3; 10]);
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 4"));
            assert!(buffer.contains("sync_overwrites_total 2"));

            // Stabilize key order
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 5"));
            assert!(buffer.contains("sync_overwrites_total 2"));

            // Modify existing key with same size - should overwrite after stabilized
            metadata.put(U64::new(2), vec![0xAA; 10]);
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("sync_rewrites_total 5"));
            assert!(buffer.contains("sync_overwrites_total 3"));

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_blob_resize() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();

            // Start with large data
            for i in 0..10 {
                metadata.put(U64::new(i), vec![i as u8; 100]);
            }
            metadata = metadata.sync().await.unwrap();

            // Stabilize key order
            metadata = metadata.sync().await.unwrap();
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 2"));
            assert!(buffer.contains("first_sync_overwrites_total 0"));

            // Remove most data to make blob smaller
            for i in 1..10 {
                metadata.remove(&U64::new(i));
            }
            metadata = metadata.sync().await.unwrap();

            // Verify the remaining data is still accessible
            let value = metadata.get(&U64::new(0)).unwrap();
            assert_eq!(value.len(), 100);
            assert_eq!(value[0], 0);

            // Check that sync properly handles blob resizing
            let buffer = context.encode();
            assert!(buffer.contains("first_sync_rewrites_total 3"));
            assert!(buffer.contains("first_sync_overwrites_total 0"));

            // Restart the metadata store
            drop(metadata);
            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Verify the changes persisted
            let value = metadata.get(&U64::new(0)).unwrap();
            assert_eq!(value.len(), 100);
            assert_eq!(value[0], 0);

            // Verify the removed keys are not present
            for i in 1..10 {
                assert!(metadata.get(&U64::new(i)).is_none());
            }

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_clear_and_repopulate() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();

            // Initial data
            metadata.put(U64::new(1), b"first".to_vec());
            metadata = metadata
                .put_sync(U64::new(2), b"second".to_vec())
                .await
                .unwrap();

            // Clear everything
            metadata.clear();
            metadata = metadata.sync().await.unwrap();

            // Verify empty
            assert!(metadata.get(&U64::new(1)).is_none());
            assert!(metadata.get(&U64::new(2)).is_none());

            // Restart the metadata store
            drop(metadata);
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Verify the changes persisted
            assert!(metadata.get(&U64::new(1)).is_none());
            assert!(metadata.get(&U64::new(2)).is_none());

            // Repopulate with different data
            metadata.put(U64::new(3), b"third".to_vec());
            metadata = metadata
                .put_sync(U64::new(4), b"fourth".to_vec())
                .await
                .unwrap();

            // Verify new data
            assert_eq!(metadata.get(&U64::new(3)).unwrap(), b"third");
            assert_eq!(metadata.get(&U64::new(4)).unwrap(), b"fourth");
            assert!(metadata.get(&U64::new(1)).is_none());
            assert!(metadata.get(&U64::new(2)).is_none());

            metadata.destroy().await.unwrap();
        });
    }

    fn test_metadata_operations_and_restart(num_operations: usize) -> String {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let cfg = Config {
                partition: "test-determinism".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata =
                Metadata::<_, U64, Vec<u8>>::init(context.child("storage"), cfg.clone())
                    .await
                    .unwrap();

            // Perform a series of deterministic operations
            for i in 0..num_operations {
                let key = U64::new(i as u64);
                let mut value = vec![0u8; 64];
                context.fill_bytes(&mut value);
                metadata.put(key, value);

                // Sync occasionally
                if context.random_bool(0.1) {
                    metadata = metadata.sync().await.unwrap();
                }

                // Update some existing keys
                if context.random_bool(0.1) {
                    let selected_index = context.random_range(0..=i);
                    let update_key = U64::new(selected_index as u64);
                    let mut new_value = vec![0u8; 64];
                    context.fill_bytes(&mut new_value);
                    metadata.put(update_key, new_value);
                }

                // Remove some keys
                if context.random_bool(0.1) {
                    let selected_index = context.random_range(0..=i);
                    let remove_key = U64::new(selected_index as u64);
                    metadata.remove(&remove_key);
                }

                // Use get_mut occasionally
                if context.random_bool(0.1) {
                    let selected_index = context.random_range(0..=i);
                    let mut_key = U64::new(selected_index as u64);
                    if let Some(value) = metadata.get_mut(&mut_key)
                        && !value.is_empty()
                    {
                        value[0] = value[0].wrapping_add(1);
                    }
                }
            }
            metadata = metadata.sync().await.unwrap();

            // Destroy the metadata store
            metadata.destroy().await.unwrap();

            context.auditor().state()
        })
    }

    #[test_group("slow")]
    #[test_traced]
    fn test_determinism() {
        let state1 = test_metadata_operations_and_restart(1_000);
        let state2 = test_metadata_operations_and_restart(1_000);
        assert_eq!(state1, state2);
    }

    #[test_traced]
    fn test_keys_iterator() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("storage"), cfg)
                .await
                .unwrap();

            // Add some keys with different prefixes
            metadata.put(U64::new(0x1000), b"value1".to_vec());
            metadata.put(U64::new(0x1001), b"value2".to_vec());
            metadata.put(U64::new(0x1002), b"value3".to_vec());
            metadata.put(U64::new(0x2000), b"value4".to_vec());
            metadata.put(U64::new(0x2001), b"value5".to_vec());
            metadata.put(U64::new(0x3000), b"value6".to_vec());

            // Test iterating over all keys
            let all_keys: Vec<_> = metadata.keys().cloned().collect();
            assert_eq!(all_keys.len(), 6);
            assert!(all_keys.contains(&U64::new(0x1000)));
            assert!(all_keys.contains(&U64::new(0x3000)));

            // Test iterating with prefix 0x10
            let prefix = hex!("0x00000000000010");
            let prefix_keys: Vec<_> = metadata
                .keys()
                .filter(|k| k.as_ref().starts_with(&prefix))
                .cloned()
                .collect();
            assert_eq!(prefix_keys.len(), 3);
            assert!(prefix_keys.contains(&U64::new(0x1000)));
            assert!(prefix_keys.contains(&U64::new(0x1001)));
            assert!(prefix_keys.contains(&U64::new(0x1002)));
            assert!(!prefix_keys.contains(&U64::new(0x2000)));

            // Test iterating with prefix 0x20
            let prefix = hex!("0x00000000000020");
            let prefix_keys: Vec<_> = metadata
                .keys()
                .filter(|k| k.as_ref().starts_with(&prefix))
                .cloned()
                .collect();
            assert_eq!(prefix_keys.len(), 2);
            assert!(prefix_keys.contains(&U64::new(0x2000)));
            assert!(prefix_keys.contains(&U64::new(0x2001)));

            // Test with non-matching prefix
            let prefix = hex!("0x00000000000040");
            let prefix_keys: Vec<_> = metadata
                .keys()
                .filter(|k| k.as_ref().starts_with(&prefix))
                .cloned()
                .collect();
            assert_eq!(prefix_keys.len(), 0);

            metadata.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_retain() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create a metadata store
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
                .await
                .unwrap();

            // Add some keys with different prefixes
            metadata.put(U64::new(0x1000), b"value1".to_vec());
            metadata.put(U64::new(0x1001), b"value2".to_vec());
            metadata.put(U64::new(0x1002), b"value3".to_vec());
            metadata.put(U64::new(0x2000), b"value4".to_vec());
            metadata.put(U64::new(0x2001), b"value5".to_vec());
            metadata.put(U64::new(0x3000), b"value6".to_vec());

            // Check initial metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_keys 6"));

            // Remove keys with prefix 0x10
            let prefix = hex!("0x00000000000010");
            metadata.retain(|k, _| !k.as_ref().starts_with(&prefix));

            // Check metrics after removal
            let buffer = context.encode();
            assert!(buffer.contains("first_keys 3"));

            // Verify remaining keys
            assert!(metadata.get(&U64::new(0x1000)).is_none());
            assert!(metadata.get(&U64::new(0x1001)).is_none());
            assert!(metadata.get(&U64::new(0x1002)).is_none());
            assert!(metadata.get(&U64::new(0x2000)).is_some());
            assert!(metadata.get(&U64::new(0x2001)).is_some());
            assert!(metadata.get(&U64::new(0x3000)).is_some());

            // Sync and reopen to ensure persistence
            metadata.sync().await.unwrap();
            let cfg = Config {
                partition: "test".into(),
                codec_config: ((0..).into(), ()),
            };
            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();

            // Verify keys are still removed after restart
            assert!(metadata.get(&U64::new(0x1000)).is_none());
            assert!(metadata.get(&U64::new(0x2000)).is_some());
            assert_eq!(metadata.keys().count(), 3);

            // Remove non-existing prefix
            let prefix = hex!("0x00000000000040");
            metadata.retain(|k, _| !k.as_ref().starts_with(&prefix));

            // Remove all remaining keys
            metadata.retain(|_, _| false);
            assert_eq!(metadata.keys().count(), 0);

            metadata.destroy().await.unwrap();
        });
    }
}