acyclic_objects/

provider.rs

//! Transport-neutral provider contract and deterministic process-local reference implementation.

use std::{
    collections::BTreeMap,
    future::Future,
    sync::{
        Arc,
        atomic::{AtomicU64, Ordering},
    },
    time::{Duration, Instant},
};

#[cfg(feature = "local")]
use std::collections::BTreeSet;
#[cfg(feature = "local")]
use std::path::PathBuf;

use async_trait::async_trait;
use futures::StreamExt as _;
use futures::future::BoxFuture;
use imbl::OrdMap;
use tokio::sync::Mutex;

use crate::{limits, wire};

const MEMORY_BYTES: usize = 64 * 1_024 * 1_024;
const SYSTEM_METADATA_BYTES: usize = 2 * 1_024;
const MAX_LISTING_VIEWS: usize = 1_024;
const MAX_CONCURRENT_BATCH_READS: usize = 16;
static PROVIDER_SEQUENCE: AtomicU64 = AtomicU64::new(1);

async fn indexed<F: Future>(index: usize, future: F) -> (usize, F::Output) {
    (index, future.await)
}

async fn collect_ordered_bounded<I, F, T>(items: I, concurrency: usize) -> Vec<T>
where
    I: IntoIterator<Item = F>,
    F: Future<Output = T>,
{
    let mut pending = items.into_iter().enumerate();
    let mut active = futures::stream::FuturesUnordered::new();
    for _ in 0..concurrency {
        let Some((index, future)) = pending.next() else {
            break;
        };
        active.push(indexed(index, future));
    }
    let mut results = Vec::new();
    while let Some(result) = active.next().await {
        results.push(result);
        if let Some((index, future)) = pending.next() {
            active.push(indexed(index, future));
        }
    }
    results.sort_unstable_by_key(|(index, _)| *index);
    results.into_iter().map(|(_, result)| result).collect()
}

/// Exactly one current-version mutation condition shared by every provider transport.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum Condition {
    /// Require no live current version.
    IfAbsent,
    /// Require the current opaque `ETag`.
    IfMatch(String),
    /// Require the current opaque version identity.
    IfVersion(String),
}

#[cfg(any(feature = "grpc", feature = "local"))]
impl Condition {
    pub(crate) fn wire(self) -> wire::Preconditions {
        let condition = match self {
            Self::IfAbsent => wire::preconditions::Condition::IfAbsent(true),
            Self::IfMatch(etag) => wire::preconditions::Condition::IfMatch(etag),
            Self::IfVersion(version_id) => wire::preconditions::Condition::IfVersion(version_id),
        };
        wire::Preconditions {
            condition: Some(condition),
        }
    }
}

/// A bucket or immutable snapshot selected for a read.
#[derive(Clone, Debug, PartialEq)]
pub enum ReadTarget {
    /// Read the current state of this bucket identity.
    Bucket(wire::BucketRef),
    /// Read the captured state of this snapshot identity.
    Snapshot(wire::SnapshotRef),
}

/// One transport-independent put request.
#[derive(Clone, Debug)]
pub struct PutRequest {
    /// Exact bucket identity and display name.
    pub bucket: wire::BucketRef,
    /// UTF-8 object key.
    pub object_key: String,
    /// Complete body bytes.
    pub body: bytes::Bytes,
    /// Immutable metadata for the new version.
    pub metadata: wire::ObjectMetadata,
    /// Optional current-version precondition.
    pub condition: Option<Condition>,
    /// Optional original-request idempotency key.
    pub idempotency_key: Option<String>,
}

/// One transport-independent object read.
#[derive(Clone, Debug)]
pub struct GetRequest {
    /// Current bucket or immutable snapshot.
    pub target: ReadTarget,
    /// UTF-8 object key.
    pub object_key: String,
    /// Exact retained version, or the visible current version when absent.
    pub version_id: Option<String>,
    /// One inclusive byte range, or the complete body when absent.
    pub range: Option<(u64, Option<u64>)>,
    /// Optional representation validator that must match.
    pub if_match: Option<String>,
    /// Optional representation validator that must not match.
    pub if_none_match: Option<String>,
    /// Maximum selected body bytes the provider may allocate or return.
    pub maximum_bytes: u64,
}

/// One immutable descriptor and its selected body bytes.
#[derive(Clone, Debug, PartialEq)]
pub struct BufferedObject {
    /// Immutable version descriptor.
    pub version: wire::ObjectVersion,
    /// Complete body or requested range.
    pub body: bytes::Bytes,
}

/// Result of an unqualified marker creation or exact-version deletion.
#[derive(Clone, Debug, PartialEq)]
pub struct DeleteResult {
    /// Whether an exact addressed version existed. Marker creation is always `true`.
    pub existed: bool,
    /// Newly created marker for an unqualified delete.
    pub marker: Option<wire::ObjectVersion>,
}

/// One stable page over a captured listing view.
#[derive(Clone, Debug, PartialEq)]
pub struct ProviderListPage {
    /// Object/version entries in contract order.
    pub entries: Vec<wire::ListEntry>,
    /// Delimiter-grouped prefixes in lexical order.
    pub common_prefixes: Vec<String>,
    /// Opaque continuation for the same immutable view.
    pub continuation: Option<String>,
}

/// Provider-visible semantic failure. Transport errors remain adapter-specific.
#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
pub enum ObjectsError {
    /// Input violates the public contract.
    #[error("invalid objects request: {0}")]
    Invalid(&'static str),
    /// The addressed bucket, key, version, or snapshot does not exist.
    #[error("objects resource not found")]
    NotFound,
    /// A bucket name or fork destination already exists.
    #[error("objects resource already exists")]
    AlreadyExists,
    /// A current-version condition did not hold and no mutation occurred.
    #[error("objects precondition failed")]
    PreconditionFailed,
    /// An idempotency key was reused with different arguments.
    #[error("objects idempotency mismatch")]
    IdempotencyMismatch,
    /// A fixed or provider-specific bound was exhausted.
    #[error("objects capacity exhausted")]
    Capacity,
    /// The provider does not implement the requested capability.
    #[error("objects capability is unsupported")]
    Unsupported,
    /// Authentication or authorization was rejected.
    #[error("objects provider rejected authorization")]
    Unauthorized,
    /// The provider could not be reached or returned an invalid response.
    #[error("objects provider is unavailable")]
    Unavailable,
}

/// Public semantic provider implemented by memory, customer-hosted, and Acyclic adapters.
#[async_trait]
pub trait ObjectsProvider: Send + Sync {
    /// Create one permanently versioned bucket.
    async fn create_bucket(
        &self,
        name: String,
        idempotency_key: Option<String>,
    ) -> Result<wire::Bucket, ObjectsError>;
    /// Resolve one exact bucket identity.
    async fn head_bucket(&self, bucket: &wire::BucketRef) -> Result<wire::Bucket, ObjectsError>;
    /// Delete one empty exact bucket identity.
    async fn delete_bucket(
        &self,
        bucket: &wire::BucketRef,
        idempotency_key: Option<String>,
    ) -> Result<bool, ObjectsError>;
    /// Create one immutable version.
    async fn put(&self, request: PutRequest) -> Result<wire::ObjectVersion, ObjectsError>;
    /// Creates an ordered bounded group of immutable versions.
    ///
    /// Providers without a native durable batch use the exact sequential
    /// semantics of [`Self::put`]. Each result corresponds to the request at
    /// the same position.
    async fn put_batch(
        &self,
        requests: Vec<PutRequest>,
    ) -> Vec<Result<wire::ObjectVersion, ObjectsError>> {
        let mut results = Vec::with_capacity(requests.len());
        for request in requests {
            results.push(self.put(request).await);
        }
        results
    }
    /// Read one immutable version or its visible current selection.
    async fn get(&self, request: GetRequest) -> Result<BufferedObject, ObjectsError>;
    /// Reads an ordered bounded group of immutable versions.
    ///
    /// Providers without a native multi-get use the exact sequential semantics
    /// of [`Self::get`]. Each result corresponds to the request at the same
    /// position.
    async fn get_batch(
        &self,
        requests: Vec<GetRequest>,
    ) -> Vec<Result<BufferedObject, ObjectsError>> {
        let mut results = Vec::with_capacity(requests.len());
        for request in requests {
            results.push(self.get(request).await);
        }
        results
    }
    /// Create a delete marker or delete one exact version.
    async fn delete(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        version_id: Option<String>,
        condition: Option<Condition>,
        idempotency_key: Option<String>,
    ) -> Result<DeleteResult, ObjectsError>;
    /// Capture or continue one stable listing.
    async fn list(
        &self,
        target: ReadTarget,
        prefix: String,
        delimiter: Option<String>,
        versions: bool,
        page_size: u32,
        continuation: Option<String>,
    ) -> Result<ProviderListPage, ObjectsError>;
    /// Atomically capture an immutable whole-bucket snapshot.
    async fn snapshot(
        &self,
        bucket: wire::BucketRef,
        idempotency_key: Option<String>,
    ) -> Result<wire::Snapshot, ObjectsError>;
    /// Destroy one exact snapshot identity.
    async fn destroy_snapshot(
        &self,
        snapshot: wire::SnapshotRef,
        idempotency_key: Option<String>,
    ) -> Result<bool, ObjectsError>;
    /// Atomically fork a whole bucket or snapshot into an independent bucket.
    async fn fork(
        &self,
        source: ReadTarget,
        destination_name: String,
        idempotency_key: Option<String>,
    ) -> Result<wire::Bucket, ObjectsError>;
    /// Create one multipart upload without publishing a version.
    async fn create_multipart(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        metadata: wire::ObjectMetadata,
        condition: Option<Condition>,
        idempotency_key: Option<String>,
    ) -> Result<wire::MultipartUpload, ObjectsError>;
    /// Upload or replace one exact staged part.
    async fn upload_part(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
        part_number: u32,
        body: bytes::Bytes,
        idempotency_key: Option<String>,
    ) -> Result<wire::UploadedPart, ObjectsError>;
    /// List staged parts in ascending part-number order.
    async fn list_parts(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
    ) -> Result<Vec<wire::UploadedPart>, ObjectsError>;
    /// Atomically publish a version from exact staged part receipts.
    async fn complete_multipart(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
        parts: Vec<wire::UploadedPart>,
        idempotency_key: Option<String>,
    ) -> Result<wire::ObjectVersion, ObjectsError>;
    /// Abort one exact staged upload.
    async fn abort_multipart(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
        idempotency_key: Option<String>,
    ) -> Result<bool, ObjectsError>;
}

#[derive(Clone)]
enum StoredBody {
    Memory(bytes::Bytes),
    Composite {
        parts: Arc<[StoredBody]>,
        length: usize,
    },
    #[cfg(feature = "local")]
    Local {
        root: Arc<PathBuf>,
        digest: [u8; 32],
        length: usize,
        location: LocalBodyLocation,
    },
}

#[derive(Clone, Eq, Ord, PartialEq, PartialOrd)]
enum BodyIdentity {
    Memory {
        address: usize,
        length: usize,
    },
    #[cfg(feature = "local")]
    Local([u8; 32]),
}

impl StoredBody {
    fn memory(body: bytes::Bytes) -> Self {
        Self::Memory(body)
    }

    fn len(&self) -> usize {
        match self {
            Self::Memory(body) => body.len(),
            Self::Composite { length, .. } => *length,
            #[cfg(feature = "local")]
            Self::Local { length, .. } => *length,
        }
    }

    fn update_hash(&self, hasher: &mut blake3::Hasher) -> Result<(), ObjectsError> {
        match self {
            Self::Memory(body) => {
                hasher.update(body);
                Ok(())
            }
            Self::Composite { parts, .. } => {
                for part in parts.iter() {
                    part.update_hash(hasher)?;
                }
                Ok(())
            }
            #[cfg(feature = "local")]
            Self::Local {
                root,
                digest,
                length,
                location,
            } => crate::local::hash_body_at(root, digest, *length, location, hasher),
        }
    }

    fn leaves(&self, output: &mut Vec<(BodyIdentity, usize)>) {
        match self {
            Self::Memory(body) => output.push((
                BodyIdentity::Memory {
                    address: body.as_ptr() as usize,
                    length: body.len(),
                },
                body.len(),
            )),
            Self::Composite { parts, .. } => {
                for part in parts.iter() {
                    part.leaves(output);
                }
            }
            #[cfg(feature = "local")]
            Self::Local { digest, length, .. } => {
                output.push((BodyIdentity::Local(*digest), *length));
            }
        }
    }

    #[cfg(feature = "local")]
    fn local_references(&self, output: &mut BTreeSet<LocalBodyReference>) {
        match self {
            Self::Memory(_) => {}
            Self::Composite { parts, .. } => {
                for part in parts.iter() {
                    part.local_references(output);
                }
            }
            Self::Local {
                digest,
                length,
                location,
                ..
            } => {
                output.insert(LocalBodyReference {
                    digest: *digest,
                    length: *length,
                    location: location.clone(),
                });
            }
        }
    }

    fn read_async(
        &self,
        start: usize,
        end: usize,
    ) -> BoxFuture<'_, Result<bytes::Bytes, ObjectsError>> {
        Box::pin(async move {
            match self {
                Self::Memory(body) => Ok(body.slice(start..end)),
                Self::Composite { parts, .. } => {
                    let mut output = Vec::with_capacity(end.saturating_sub(start));
                    let mut offset = 0usize;
                    for part in parts.iter() {
                        let part_end = offset
                            .checked_add(part.len())
                            .ok_or(ObjectsError::Unavailable)?;
                        if part_end > start && offset < end {
                            let selected_start = start.saturating_sub(offset).min(part.len());
                            let selected_end = end.saturating_sub(offset).min(part.len());
                            output.extend_from_slice(
                                &part.read_async(selected_start, selected_end).await?,
                            );
                        }
                        offset = part_end;
                    }
                    if output.len() != end.saturating_sub(start) {
                        return Err(ObjectsError::Unavailable);
                    }
                    Ok(output.into())
                }
                #[cfg(feature = "local")]
                Self::Local {
                    root,
                    digest,
                    length,
                    location,
                } => {
                    crate::local::read_body_at_async(root, digest, *length, location, start, end)
                        .await
                }
            }
        })
    }
}

#[derive(Clone)]
struct Version {
    descriptor: wire::ObjectVersion,
    body: Option<StoredBody>,
}

struct ResolvedGet {
    descriptor: wire::ObjectVersion,
    body: StoredBody,
    start: usize,
    end: usize,
}

#[derive(Clone)]
struct BucketState {
    reference: wire::BucketRef,
    created_at: prost_types::Timestamp,
    objects: OrdMap<String, Vec<Version>>,
    listing: OrdMap<String, Vec<wire::ObjectVersion>>,
}

impl BucketState {
    fn append(&mut self, key: String, version: Version) {
        self.listing
            .entry(key.clone())
            .or_default()
            .push(version.descriptor.clone());
        self.objects.entry(key).or_default().push(version);
    }

    fn remove_exact(&mut self, key: &str, identity: &str) -> Result<Option<Version>, ObjectsError> {
        let Some(history) = self.objects.get_mut(key) else {
            return Ok(None);
        };
        let Some(position) = history
            .iter()
            .position(|version| version.descriptor.version_id == identity)
        else {
            return Ok(None);
        };
        let removed = history.remove(position);
        let descriptors = self.listing.get_mut(key).ok_or(ObjectsError::Unavailable)?;
        if descriptors.get(position) != Some(&removed.descriptor) {
            return Err(ObjectsError::Unavailable);
        }
        descriptors.remove(position);
        if history.is_empty() {
            self.objects.remove(key);
            self.listing.remove(key);
        }
        Ok(Some(removed))
    }
}

#[derive(Clone)]
struct SnapshotState {
    reference: wire::SnapshotRef,
    bucket: BucketState,
}

struct MultipartState {
    bucket: wire::BucketRef,
    object_key: String,
    metadata: wire::ObjectMetadata,
    condition: Option<Condition>,
    parts: BTreeMap<u32, (wire::UploadedPart, StoredBody)>,
}

struct StoredPartRequest {
    bucket: wire::BucketRef,
    object_key: String,
    upload_id: String,
    part_number: u32,
    body: StoredBody,
    body_digest: [u8; 32],
    idempotency_key: Option<String>,
}

#[cfg(feature = "local")]
pub(crate) struct ExternalBody {
    pub(crate) root: PathBuf,
    pub(crate) digest: [u8; 32],
    pub(crate) length: usize,
    pub(crate) location: LocalBodyLocation,
}

#[cfg(feature = "local")]
#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub(crate) enum LocalBodyLocation {
    Segment { id: [u8; 32], offset: u64 },
}

#[cfg(feature = "local")]
#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub(crate) struct LocalBodyReference {
    pub(crate) digest: [u8; 32],
    pub(crate) length: usize,
    pub(crate) location: LocalBodyLocation,
}

enum ListingItem {
    Entry(Box<wire::ListEntry>),
    Prefix(String),
}

struct ListingView {
    binding: ListingBinding,
    objects: OrdMap<String, Vec<wire::ObjectVersion>>,
    expires_at: Instant,
    prefetched: Option<PrefetchedListingItem>,
}

struct PrefetchedListingItem {
    offset: usize,
    cursor: ListingCursor,
    item: ListingItem,
    after: ListingCursor,
}

#[derive(Clone, PartialEq)]
struct ListingBinding {
    target: ReadTarget,
    prefix: String,
    delimiter: Option<String>,
    versions: bool,
}

#[derive(Clone, Eq, PartialEq)]
struct ListingCursor {
    key: Option<String>,
    version_index: usize,
}

#[derive(Clone)]
enum MutationOutcome {
    Bucket(Result<wire::Bucket, ObjectsError>),
    Boolean(Result<bool, ObjectsError>),
    Version(Result<wire::ObjectVersion, ObjectsError>),
    Delete(Result<DeleteResult, ObjectsError>),
    Snapshot(Result<wire::Snapshot, ObjectsError>),
    Multipart(Result<wire::MultipartUpload, ObjectsError>),
    Part(Result<wire::UploadedPart, ObjectsError>),
}

struct IdempotencyRecord {
    fingerprint: blake3::Hash,
    outcome: MutationOutcome,
}

struct State {
    maximum_bytes: usize,
    maximum_object_bytes: usize,
    sequence: u64,
    listing_sequence: u64,
    listing_epoch: u64,
    bytes: usize,
    body_references: BTreeMap<BodyIdentity, (usize, usize)>,
    names: BTreeMap<String, String>,
    buckets: BTreeMap<String, BucketState>,
    snapshots: BTreeMap<String, SnapshotState>,
    multiparts: BTreeMap<String, MultipartState>,
    listings: BTreeMap<String, ListingView>,
    idempotency: BTreeMap<String, IdempotencyRecord>,
}

impl Default for State {
    fn default() -> Self {
        Self {
            maximum_bytes: MEMORY_BYTES,
            maximum_object_bytes: MEMORY_BYTES,
            sequence: 0,
            listing_sequence: 0,
            listing_epoch: PROVIDER_SEQUENCE.fetch_add(1, Ordering::Relaxed),
            bytes: 0,
            body_references: BTreeMap::new(),
            names: BTreeMap::new(),
            buckets: BTreeMap::new(),
            snapshots: BTreeMap::new(),
            multiparts: BTreeMap::new(),
            listings: BTreeMap::new(),
            idempotency: BTreeMap::new(),
        }
    }
}

impl State {
    fn additional_body_bytes(&self, body: &StoredBody) -> Result<usize, ObjectsError> {
        let mut leaves = Vec::new();
        body.leaves(&mut leaves);
        leaves
            .into_iter()
            .try_fold(0usize, |total, (identity, length)| {
                if self.body_references.contains_key(&identity) {
                    Ok(total)
                } else {
                    total.checked_add(length).ok_or(ObjectsError::Capacity)
                }
            })
    }

    fn retain_body(&mut self, body: &StoredBody) -> Result<(), ObjectsError> {
        let mut leaves = Vec::new();
        body.leaves(&mut leaves);
        for (identity, length) in leaves {
            match self.body_references.entry(identity) {
                std::collections::btree_map::Entry::Vacant(entry) => {
                    self.bytes = self
                        .bytes
                        .checked_add(length)
                        .ok_or(ObjectsError::Capacity)?;
                    entry.insert((length, 1));
                }
                std::collections::btree_map::Entry::Occupied(mut entry) => {
                    if entry.get().0 != length {
                        return Err(ObjectsError::Unavailable);
                    }
                    entry.get_mut().1 =
                        entry.get().1.checked_add(1).ok_or(ObjectsError::Capacity)?;
                }
            }
        }
        Ok(())
    }

    fn release_body(&mut self, body: &StoredBody) -> Result<(), ObjectsError> {
        let mut leaves = Vec::new();
        body.leaves(&mut leaves);
        for (identity, length) in leaves {
            let references = self
                .body_references
                .get_mut(&identity)
                .ok_or(ObjectsError::Unavailable)?;
            if references.0 != length || references.1 == 0 {
                return Err(ObjectsError::Unavailable);
            }
            references.1 -= 1;
            if references.1 == 0 {
                self.body_references.remove(&identity);
                self.bytes = self
                    .bytes
                    .checked_sub(length)
                    .ok_or(ObjectsError::Unavailable)?;
            }
        }
        Ok(())
    }

    fn retain_bucket(&mut self, bucket: &BucketState) -> Result<(), ObjectsError> {
        for body in bucket
            .objects
            .values()
            .flatten()
            .filter_map(|version| version.body.as_ref())
        {
            self.retain_body(body)?;
        }
        Ok(())
    }

    fn release_bucket(&mut self, bucket: &BucketState) -> Result<(), ObjectsError> {
        for body in bucket
            .objects
            .values()
            .flatten()
            .filter_map(|version| version.body.as_ref())
        {
            self.release_body(body)?;
        }
        Ok(())
    }
}

/// Bounded deterministic process-local reference provider.
///
/// Acknowledged state is ephemeral and has no isolation, durability, distribution, or production
/// availability claim.
#[derive(Clone, Default)]
pub struct MemoryObjects {
    state: Arc<Mutex<State>>,
}

impl MemoryObjects {
    #[cfg(feature = "local")]
    pub(crate) async fn local_body_references(&self) -> BTreeSet<LocalBodyReference> {
        let state = self.state.lock().await;
        let mut references = BTreeSet::new();
        for bucket in state
            .buckets
            .values()
            .chain(state.snapshots.values().map(|snapshot| &snapshot.bucket))
        {
            for body in bucket
                .objects
                .values()
                .flatten()
                .filter_map(|version| version.body.as_ref())
            {
                body.local_references(&mut references);
            }
        }
        for body in state
            .multiparts
            .values()
            .flat_map(|upload| upload.parts.values().map(|(_, body)| body))
        {
            body.local_references(&mut references);
        }
        references
    }

    /// Resolves a bucket by its canonical name without mutating provider state.
    pub async fn bucket_named(&self, name: &str) -> Result<Option<wire::BucketRef>, ObjectsError> {
        Self::validate_bucket_name(name)?;
        let state = self.state.lock().await;
        let Some(bucket_id) = state.names.get(name) else {
            return Ok(None);
        };
        let bucket = state
            .buckets
            .get(bucket_id)
            .ok_or(ObjectsError::Unavailable)?;
        Ok(Some(bucket.reference.clone()))
    }

    /// Creates the default bounded reference provider with one deterministic bucket.
    #[must_use]
    pub fn with_default_bucket() -> (Self, wire::BucketRef) {
        Self::from_bucket("memory-bucket".to_owned(), MEMORY_BYTES)
    }

    /// Creates an empty reference provider with an explicit aggregate byte ceiling.
    ///
    /// # Errors
    ///
    /// Rejects zero or process-unrepresentable limits before allocating provider state.
    pub fn new(maximum_bytes: u64) -> Result<Self, ObjectsError> {
        if maximum_bytes == 0 {
            return Err(ObjectsError::Invalid("memory byte limit must be positive"));
        }
        Self::new_with_limits(maximum_bytes, maximum_bytes)
    }

    pub(crate) fn new_with_limits(
        maximum_object_bytes: u64,
        maximum_bytes: u64,
    ) -> Result<Self, ObjectsError> {
        let maximum_object_bytes = usize::try_from(maximum_object_bytes)
            .map_err(|_| ObjectsError::Invalid("memory object limit is not representable"))?;
        let maximum_bytes = usize::try_from(maximum_bytes)
            .map_err(|_| ObjectsError::Invalid("memory byte limit is not representable"))?;
        if maximum_object_bytes == 0 || maximum_bytes == 0 || maximum_object_bytes > maximum_bytes {
            return Err(ObjectsError::Invalid("invalid memory object limits"));
        }
        Ok(Self {
            state: Arc::new(Mutex::new(State {
                maximum_bytes,
                maximum_object_bytes,
                ..State::default()
            })),
        })
    }

    /// Creates a bounded provider with one deterministic bucket already admitted.
    ///
    /// This synchronous constructor lets another in-process SDK family compose the exact public
    /// reference provider without running an executor merely to establish its private bucket.
    ///
    /// # Errors
    ///
    /// Rejects an invalid bucket name or memory ceiling.
    pub fn with_bucket(
        name: impl Into<String>,
        maximum_bytes: u64,
    ) -> Result<(Self, wire::BucketRef), ObjectsError> {
        let maximum_bytes = usize::try_from(maximum_bytes)
            .map_err(|_| ObjectsError::Invalid("memory byte limit is not representable"))?;
        if maximum_bytes == 0 {
            return Err(ObjectsError::Invalid("memory byte limit must be positive"));
        }
        let name = name.into();
        Self::validate_bucket_name(&name)?;
        Ok(Self::from_bucket(name, maximum_bytes))
    }

    /// Creates one deterministic bucket with independent per-object and aggregate bounds.
    ///
    /// # Errors
    ///
    /// Rejects an invalid bucket name, zero limits, a per-object limit larger than the aggregate
    /// limit, or limits that cannot be represented by this process.
    pub fn with_bucket_limits(
        name: impl Into<String>,
        maximum_object_bytes: u64,
        maximum_bytes: u64,
    ) -> Result<(Self, wire::BucketRef), ObjectsError> {
        let maximum_object_bytes = usize::try_from(maximum_object_bytes)
            .map_err(|_| ObjectsError::Invalid("memory object limit is not representable"))?;
        let maximum_bytes = usize::try_from(maximum_bytes)
            .map_err(|_| ObjectsError::Invalid("memory byte limit is not representable"))?;
        if maximum_object_bytes == 0 || maximum_bytes == 0 || maximum_object_bytes > maximum_bytes {
            return Err(ObjectsError::Invalid("invalid memory object limits"));
        }
        let name = name.into();
        Self::validate_bucket_name(&name)?;
        Ok(Self::from_bucket_limits(
            name,
            maximum_object_bytes,
            maximum_bytes,
        ))
    }

    fn from_bucket(name: String, maximum_bytes: usize) -> (Self, wire::BucketRef) {
        Self::from_bucket_limits(name, maximum_bytes, maximum_bytes)
    }

    fn from_bucket_limits(
        name: String,
        maximum_object_bytes: usize,
        maximum_bytes: usize,
    ) -> (Self, wire::BucketRef) {
        let bucket_id = "bucket-0000000000000001".to_owned();
        let reference = wire::BucketRef {
            bucket_id: bucket_id.clone(),
            name: name.clone(),
        };
        let mut state = State {
            maximum_bytes,
            maximum_object_bytes,
            sequence: 1,
            ..State::default()
        };
        state.names.insert(name, bucket_id.clone());
        state.buckets.insert(
            bucket_id,
            BucketState {
                reference: reference.clone(),
                created_at: Self::timestamp(&state),
                objects: OrdMap::new(),
                listing: OrdMap::new(),
            },
        );
        (
            Self {
                state: Arc::new(Mutex::new(state)),
            },
            reference,
        )
    }

    fn fingerprint(kind: &str, fields: &[&[u8]]) -> blake3::Hash {
        let mut hasher = blake3::Hasher::new();
        for field in std::iter::once(kind.as_bytes()).chain(fields.iter().copied()) {
            hasher.update(&(field.len() as u64).to_le_bytes());
            hasher.update(field);
        }
        hasher.finalize()
    }

    fn execute<T: Clone>(
        state: &mut State,
        idempotency_key: Option<&str>,
        fingerprint: blake3::Hash,
        unpack: impl FnOnce(&MutationOutcome) -> Option<Result<T, ObjectsError>>,
        pack: impl FnOnce(Result<T, ObjectsError>) -> MutationOutcome,
        operation: impl FnOnce(&mut State) -> Result<T, ObjectsError>,
    ) -> Result<T, ObjectsError> {
        let Some(idempotency_key) = idempotency_key else {
            return operation(state);
        };
        if idempotency_key.is_empty() || idempotency_key.len() > 256 {
            return Err(ObjectsError::Invalid("invalid idempotency key"));
        }
        if let Some(record) = state.idempotency.get(idempotency_key) {
            if record.fingerprint != fingerprint {
                return Err(ObjectsError::IdempotencyMismatch);
            }
            return unpack(&record.outcome).ok_or(ObjectsError::IdempotencyMismatch)?;
        }
        let outcome = operation(state);
        state.idempotency.insert(
            idempotency_key.to_owned(),
            IdempotencyRecord {
                fingerprint,
                outcome: pack(outcome.clone()),
            },
        );
        outcome
    }

    fn reference_fields(reference: &wire::BucketRef) -> [&[u8]; 2] {
        [reference.bucket_id.as_bytes(), reference.name.as_bytes()]
    }

    fn condition_bytes(condition: &Option<Condition>) -> Vec<u8> {
        match condition {
            None => b"none".to_vec(),
            Some(Condition::IfAbsent) => b"absent".to_vec(),
            Some(Condition::IfMatch(value)) => [b"etag:".as_slice(), value.as_bytes()].concat(),
            Some(Condition::IfVersion(value)) => {
                [b"version:".as_slice(), value.as_bytes()].concat()
            }
        }
    }

    fn metadata_bytes(metadata: &wire::ObjectMetadata) -> Vec<u8> {
        let mut fields = vec![
            metadata.content_type.as_str(),
            metadata.content_encoding.as_str(),
            metadata.cache_control.as_str(),
            metadata.content_disposition.as_str(),
            metadata.content_language.as_str(),
        ];
        fields.push(if metadata.expires_unix_seconds.is_some() {
            "expires"
        } else {
            "no-expires"
        });
        let mut encoded = Vec::new();
        for field in fields {
            encoded.extend_from_slice(&(field.len() as u64).to_le_bytes());
            encoded.extend_from_slice(field.as_bytes());
        }
        if let Some(expires) = metadata.expires_unix_seconds {
            encoded.extend_from_slice(&expires.to_le_bytes());
        }
        let mut user = metadata.user.iter().collect::<Vec<_>>();
        user.sort_unstable_by(|left, right| left.0.cmp(right.0));
        for (name, value) in user {
            for field in [name.as_bytes(), value.as_bytes()] {
                encoded.extend_from_slice(&(field.len() as u64).to_le_bytes());
                encoded.extend_from_slice(field);
            }
        }
        encoded
    }

    fn next(state: &mut State, kind: &str) -> Result<String, ObjectsError> {
        let sequence = if kind == "listing" {
            &mut state.listing_sequence
        } else {
            &mut state.sequence
        };
        *sequence = sequence.checked_add(1).ok_or(ObjectsError::Capacity)?;
        Ok(format!("{kind}-{sequence:016x}"))
    }

    fn timestamp(state: &State) -> prost_types::Timestamp {
        prost_types::Timestamp {
            seconds: i64::try_from(state.sequence).unwrap_or(i64::MAX),
            nanos: 0,
        }
    }

    fn bucket<'a>(
        state: &'a State,
        reference: &wire::BucketRef,
    ) -> Result<&'a BucketState, ObjectsError> {
        state
            .buckets
            .get(&reference.bucket_id)
            .filter(|bucket| bucket.reference == *reference)
            .ok_or(ObjectsError::NotFound)
    }

    fn bucket_mut<'a>(
        state: &'a mut State,
        reference: &wire::BucketRef,
    ) -> Result<&'a mut BucketState, ObjectsError> {
        state
            .buckets
            .get_mut(&reference.bucket_id)
            .filter(|bucket| bucket.reference == *reference)
            .ok_or(ObjectsError::NotFound)
    }

    fn validate_key(value: &str) -> Result<(), ObjectsError> {
        if value.is_empty() || value.len() > limits::KEY_BYTES || value.contains('\0') {
            Err(ObjectsError::Invalid("invalid object key"))
        } else {
            Ok(())
        }
    }

    fn validate_bucket_name(name: &str) -> Result<(), ObjectsError> {
        let valid_edges = name
            .as_bytes()
            .first()
            .zip(name.as_bytes().last())
            .is_some_and(|(first, last)| {
                first.is_ascii_alphanumeric() && last.is_ascii_alphanumeric()
            });
        let valid_body = name
            .bytes()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-');
        if !(3..=63).contains(&name.len()) || !valid_edges || !valid_body {
            Err(ObjectsError::Invalid("invalid bucket name"))
        } else {
            Ok(())
        }
    }

    fn validate_metadata(metadata: &wire::ObjectMetadata) -> Result<(), ObjectsError> {
        let fixed = [
            &metadata.content_type,
            &metadata.content_encoding,
            &metadata.cache_control,
            &metadata.content_disposition,
            &metadata.content_language,
        ]
        .into_iter()
        .map(String::len)
        .sum::<usize>();
        let user = metadata
            .user
            .iter()
            .map(|(name, value)| name.len() + value.len())
            .sum::<usize>();
        if fixed > SYSTEM_METADATA_BYTES
            || user > limits::USER_METADATA_BYTES
            || metadata.user.iter().any(|(name, value)| {
                !name.bytes().all(|byte| {
                    byte.is_ascii_alphanumeric()
                        || matches!(
                            byte,
                            b'!' | b'#'
                                | b'$'
                                | b'%'
                                | b'&'
                                | b'\''
                                | b'*'
                                | b'+'
                                | b'-'
                                | b'.'
                                | b'^'
                                | b'_'
                                | b'`'
                                | b'|'
                                | b'~'
                        )
                }) || name.is_empty()
                    || !Self::valid_header_value(value)
            })
            || [
                &metadata.content_type,
                &metadata.content_encoding,
                &metadata.cache_control,
                &metadata.content_disposition,
                &metadata.content_language,
            ]
            .into_iter()
            .any(|value| !Self::valid_header_value(value))
            || metadata
                .expires_unix_seconds
                .is_some_and(|seconds| !(-62_167_219_200..=253_402_300_799).contains(&seconds))
        {
            Err(ObjectsError::Invalid("invalid object metadata"))
        } else {
            Ok(())
        }
    }

    fn valid_header_value(value: &str) -> bool {
        value
            .bytes()
            .all(|byte| byte == b'\t' || (0x20..=0x7e).contains(&byte))
    }

    fn target(state: &State, target: &ReadTarget) -> Result<BucketState, ObjectsError> {
        match target {
            ReadTarget::Bucket(reference) => Ok(Self::bucket(state, reference)?.clone()),
            ReadTarget::Snapshot(reference) => state
                .snapshots
                .get(&reference.snapshot_id)
                .filter(|snapshot| snapshot.reference == *reference)
                .map(|snapshot| snapshot.bucket.clone())
                .ok_or(ObjectsError::NotFound),
        }
    }

    fn target_ref<'a>(
        state: &'a State,
        target: &ReadTarget,
    ) -> Result<&'a BucketState, ObjectsError> {
        match target {
            ReadTarget::Bucket(reference) => Self::bucket(state, reference),
            ReadTarget::Snapshot(reference) => state
                .snapshots
                .get(&reference.snapshot_id)
                .filter(|snapshot| snapshot.reference == *reference)
                .map(|snapshot| &snapshot.bucket)
                .ok_or(ObjectsError::NotFound),
        }
    }

    fn visible<'a>(
        bucket: &'a BucketState,
        object_key: &str,
        version_id: Option<&str>,
    ) -> Result<&'a Version, ObjectsError> {
        let versions = bucket
            .objects
            .get(object_key)
            .ok_or(ObjectsError::NotFound)?;
        let version = match version_id {
            Some(identity) => versions
                .iter()
                .rev()
                .find(|version| version.descriptor.version_id == identity),
            None => versions.last(),
        }
        .ok_or(ObjectsError::NotFound)?;
        if version.descriptor.delete_marker {
            Err(ObjectsError::NotFound)
        } else {
            Ok(version)
        }
    }

    fn resolve_get(state: &State, request: &GetRequest) -> Result<ResolvedGet, ObjectsError> {
        Self::validate_key(&request.object_key)?;
        let bucket = Self::target_ref(state, &request.target)?;
        let version = Self::visible(bucket, &request.object_key, request.version_id.as_deref())?;
        if request
            .if_match
            .as_ref()
            .is_some_and(|etag| *etag != version.descriptor.etag)
            || request
                .if_none_match
                .as_ref()
                .is_some_and(|etag| *etag == version.descriptor.etag)
        {
            return Err(ObjectsError::PreconditionFailed);
        }
        let descriptor = version.descriptor.clone();
        let body = version.body.clone().ok_or(ObjectsError::NotFound)?;
        let (start, end) = match request.range {
            None => (0, body.len()),
            Some((start, end)) => {
                let start =
                    usize::try_from(start).map_err(|_| ObjectsError::Invalid("invalid range"))?;
                let end = usize::try_from(end.unwrap_or(body.len().saturating_sub(1) as u64))
                    .map_err(|_| ObjectsError::Invalid("invalid range"))?;
                if start > end || end >= body.len() {
                    return Err(ObjectsError::Invalid("invalid range"));
                }
                (start, end.saturating_add(1))
            }
        };
        if end.saturating_sub(start) as u64 > request.maximum_bytes {
            return Err(ObjectsError::Capacity);
        }
        Ok(ResolvedGet {
            descriptor,
            body,
            start,
            end,
        })
    }

    async fn read_resolved_get(resolved: ResolvedGet) -> Result<BufferedObject, ObjectsError> {
        let body = resolved
            .body
            .read_async(resolved.start, resolved.end)
            .await?;
        Ok(BufferedObject {
            version: resolved.descriptor,
            body,
        })
    }

    fn condition(current: Option<&Version>, condition: &Option<Condition>) -> bool {
        match condition {
            None => true,
            Some(Condition::IfAbsent) => current.is_none_or(|value| value.descriptor.delete_marker),
            Some(Condition::IfMatch(etag)) => current.is_some_and(|value| {
                !value.descriptor.delete_marker && value.descriptor.etag == *etag
            }),
            Some(Condition::IfVersion(identity)) => {
                current.is_some_and(|value| value.descriptor.version_id == *identity)
            }
        }
    }

    fn descriptor(
        state: &mut State,
        body_digest: &[u8; 32],
        body_length: usize,
        metadata: wire::ObjectMetadata,
        marker: bool,
    ) -> Result<wire::ObjectVersion, ObjectsError> {
        let version_id = Self::next(state, "version")?;
        let etag = if marker {
            String::new()
        } else {
            format!("\"{}\"", blake3::Hash::from_bytes(*body_digest).to_hex())
        };
        Ok(wire::ObjectVersion {
            version_id,
            etag,
            size: body_length as u64,
            delete_marker: marker,
            metadata: Some(metadata),
            created_at: Some(Self::timestamp(state)),
        })
    }

    fn prefix_successor(prefix: &str) -> Option<String> {
        let mut chars: Vec<char> = prefix.chars().collect();
        while let Some(last) = chars.pop() {
            let next = u32::from(last) + 1;
            let next = char::from_u32(if next == 0xd800 { 0xe000 } else { next });
            if let Some(next) = next {
                chars.push(next);
                return Some(chars.into_iter().collect());
            }
        }
        None
    }

    fn listing_next(
        objects: &OrdMap<String, Vec<wire::ObjectVersion>>,
        prefix: &str,
        delimiter: Option<&str>,
        versions: bool,
        cursor: &mut ListingCursor,
    ) -> Option<ListingItem> {
        while let Some(key) = cursor.key.as_ref() {
            let Some((object_key, history)) = objects.range(key.clone()..).next() else {
                cursor.key = None;
                return None;
            };
            if !object_key.starts_with(prefix) {
                cursor.key = None;
                return None;
            }
            if let Some(delimiter) = delimiter
                && let Some(remainder) = object_key.get(prefix.len()..)
                && let Some(position) = remainder.find(delimiter)
                && let Some(group) = object_key.get(..prefix.len() + position + delimiter.len())
            {
                cursor.key = Self::prefix_successor(group);
                cursor.version_index = 0;
                return Some(ListingItem::Prefix(group.to_owned()));
            }
            if versions {
                if let Some(version) = history
                    .len()
                    .checked_sub(cursor.version_index.saturating_add(1))
                    .and_then(|index| history.get(index))
                {
                    cursor.version_index += 1;
                    if cursor.version_index == history.len() {
                        cursor.key = Some(format!("{object_key}\0"));
                        cursor.version_index = 0;
                    }
                    return Some(ListingItem::Entry(Box::new(wire::ListEntry {
                        object_key: object_key.clone(),
                        version: Some(version.clone()),
                    })));
                }
            } else if let Some(version) = history.last()
                && !version.delete_marker
            {
                cursor.key = Some(format!("{object_key}\0"));
                cursor.version_index = 0;
                return Some(ListingItem::Entry(Box::new(wire::ListEntry {
                    object_key: object_key.clone(),
                    version: Some(version.clone()),
                })));
            }
            cursor.key = Some(format!("{object_key}\0"));
            cursor.version_index = 0;
        }
        None
    }

    fn listing_token(
        view_id: &str,
        offset: usize,
        cursor: &ListingCursor,
    ) -> Result<String, ObjectsError> {
        let key = cursor.key.as_deref().ok_or(ObjectsError::Unavailable)?;
        let payload = format!(
            "{view_id}:{offset}:{}:{}",
            cursor.version_index,
            hex::encode(key.as_bytes())
        );
        Ok(payload)
    }

    fn parse_listing_token(token: &str) -> Result<(String, usize, ListingCursor), ObjectsError> {
        let mut fields = token.split(':');
        let view_id = fields
            .next()
            .ok_or(ObjectsError::Invalid("invalid continuation"))?;
        let offset = fields
            .next()
            .ok_or(ObjectsError::Invalid("invalid continuation"))?
            .parse()
            .map_err(|_| ObjectsError::Invalid("invalid continuation"))?;
        let version_index = fields
            .next()
            .ok_or(ObjectsError::Invalid("invalid continuation"))?
            .parse()
            .map_err(|_| ObjectsError::Invalid("invalid continuation"))?;
        let key = fields
            .next()
            .ok_or(ObjectsError::Invalid("invalid continuation"))?;
        if fields.next().is_some() {
            return Err(ObjectsError::Invalid("invalid continuation"));
        }
        let key = String::from_utf8(
            hex::decode(key).map_err(|_| ObjectsError::Invalid("invalid continuation"))?,
        )
        .map_err(|_| ObjectsError::Invalid("invalid continuation"))?;
        Ok((
            view_id.to_owned(),
            offset,
            ListingCursor {
                key: Some(key),
                version_index,
            },
        ))
    }

    fn listing_page(
        view: &mut ListingView,
        view_id: &str,
        offset: usize,
        page_size: u32,
        mut cursor: ListingCursor,
    ) -> Result<ProviderListPage, ObjectsError> {
        let mut prefetched = view
            .prefetched
            .take()
            .filter(|item| item.offset == offset && item.cursor == cursor);
        let mut entries = Vec::new();
        let mut common_prefixes = Vec::new();
        for _ in 0..page_size {
            let item = if let Some(item) = prefetched.take() {
                cursor = item.after;
                Some(item.item)
            } else {
                Self::listing_next(
                    &view.objects,
                    &view.binding.prefix,
                    view.binding.delimiter.as_deref(),
                    view.binding.versions,
                    &mut cursor,
                )
            };
            let Some(item) = item else {
                break;
            };
            match item {
                ListingItem::Entry(value) => entries.push(*value),
                ListingItem::Prefix(value) => common_prefixes.push(value),
            }
        }
        let end = offset + entries.len() + common_prefixes.len();
        let mut lookahead = cursor.clone();
        let continuation = Self::listing_next(
            &view.objects,
            &view.binding.prefix,
            view.binding.delimiter.as_deref(),
            view.binding.versions,
            &mut lookahead,
        )
        .map(|item| {
            view.prefetched = Some(PrefetchedListingItem {
                offset: end,
                cursor: cursor.clone(),
                item,
                after: lookahead,
            });
            Self::listing_token(view_id, end, &cursor)
        })
        .transpose()?;
        Ok(ProviderListPage {
            entries,
            common_prefixes,
            continuation,
        })
    }

    fn put_fingerprint(request: &PutRequest, body_digest: &[u8; 32]) -> blake3::Hash {
        let bucket_fields = Self::reference_fields(&request.bucket);
        let condition_bytes = Self::condition_bytes(&request.condition);
        let metadata_bytes = Self::metadata_bytes(&request.metadata);
        Self::fingerprint(
            "put",
            &[
                bucket_fields[0],
                bucket_fields[1],
                request.object_key.as_bytes(),
                &condition_bytes,
                &metadata_bytes,
                body_digest,
            ],
        )
    }

    #[cfg(feature = "local")]
    pub(crate) async fn recorded_put(
        &self,
        request: &PutRequest,
        body_digest: &[u8; 32],
    ) -> Option<Result<wire::ObjectVersion, ObjectsError>> {
        let key = request.idempotency_key.as_deref()?;
        // Only an exact recorded retry is free of a new journal intent.
        // Invalid and mismatched requests retain the original append path.
        if key.is_empty()
            || key.len() > 256
            || Self::validate_key(&request.object_key).is_err()
            || Self::validate_metadata(&request.metadata).is_err()
        {
            return None;
        }
        let fingerprint = Self::put_fingerprint(request, body_digest);
        let state = self.state.lock().await;
        let prior = state.idempotency.get(key)?;
        if prior.fingerprint != fingerprint {
            return None;
        }
        match &prior.outcome {
            MutationOutcome::Version(result) => Some(result.clone()),
            _ => None,
        }
    }

    async fn put_stored(
        &self,
        request: PutRequest,
        body: StoredBody,
        body_digest: [u8; 32],
    ) -> Result<wire::ObjectVersion, ObjectsError> {
        Self::validate_key(&request.object_key)?;
        Self::validate_metadata(&request.metadata)?;
        let fingerprint = Self::put_fingerprint(&request, &body_digest);
        let PutRequest {
            bucket,
            object_key,
            body: _,
            metadata,
            condition,
            idempotency_key,
        } = request;
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Version(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Version,
            |state| {
                let current = Self::bucket(state, &bucket)?
                    .objects
                    .get(&object_key)
                    .and_then(|history| history.last())
                    .cloned();
                if !Self::condition(current.as_ref(), &condition) {
                    return Err(ObjectsError::PreconditionFailed);
                }
                let size = body.len();
                let additional = state.additional_body_bytes(&body)?;
                if size > state.maximum_object_bytes
                    || state
                        .bytes
                        .checked_add(additional)
                        .is_none_or(|bytes| bytes > state.maximum_bytes)
                {
                    return Err(ObjectsError::Capacity);
                }
                let descriptor = Self::descriptor(state, &body_digest, size, metadata, false)?;
                state.retain_body(&body)?;
                Self::bucket_mut(state, &bucket)?.append(
                    object_key,
                    Version {
                        descriptor: descriptor.clone(),
                        body: Some(body),
                    },
                );
                Ok(descriptor)
            },
        )
    }

    #[cfg(feature = "local")]
    pub(crate) async fn put_external(
        &self,
        request: PutRequest,
        external: ExternalBody,
    ) -> Result<wire::ObjectVersion, ObjectsError> {
        let ExternalBody {
            root,
            digest,
            length,
            location,
        } = external;
        self.put_stored(
            request,
            StoredBody::Local {
                root: Arc::new(root),
                digest,
                length,
                location,
            },
            digest,
        )
        .await
    }

    async fn upload_part_stored(
        &self,
        request: StoredPartRequest,
    ) -> Result<wire::UploadedPart, ObjectsError> {
        let StoredPartRequest {
            bucket,
            object_key,
            upload_id,
            part_number,
            body,
            body_digest,
            idempotency_key,
        } = request;
        if !(1..=limits::MULTIPART_PARTS).contains(&part_number)
            || body.len() as u64 > limits::MAX_MULTIPART_PART_BYTES
        {
            return Err(ObjectsError::Invalid("invalid multipart part"));
        }
        let bucket_fields = Self::reference_fields(&bucket);
        let part_number_bytes = part_number.to_le_bytes();
        let fingerprint = Self::fingerprint(
            "upload-part",
            &[
                bucket_fields[0],
                bucket_fields[1],
                object_key.as_bytes(),
                upload_id.as_bytes(),
                &part_number_bytes,
                &body_digest,
            ],
        );
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Part(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Part,
            |state| {
                let existing_body = state
                    .multiparts
                    .get(&upload_id)
                    .filter(|upload| upload.bucket == bucket && upload.object_key == object_key)
                    .ok_or(ObjectsError::NotFound)?
                    .parts
                    .get(&part_number)
                    .map(|(_, body)| body.clone());
                if let Some(existing) = &existing_body {
                    state.release_body(existing)?;
                }
                let additional = state.additional_body_bytes(&body)?;
                if state
                    .bytes
                    .checked_add(additional)
                    .is_none_or(|bytes| bytes > state.maximum_bytes)
                {
                    if let Some(existing) = &existing_body {
                        state.retain_body(existing)?;
                    }
                    return Err(ObjectsError::Capacity);
                }
                let part = wire::UploadedPart {
                    part_number,
                    etag: format!("\"{}\"", blake3::Hash::from_bytes(body_digest).to_hex()),
                    size: body.len() as u64,
                };
                state
                    .multiparts
                    .get_mut(&upload_id)
                    .ok_or(ObjectsError::NotFound)?
                    .parts
                    .insert(part_number, (part.clone(), body));
                let admitted = state
                    .multiparts
                    .get(&upload_id)
                    .and_then(|upload| upload.parts.get(&part_number))
                    .map(|(_, body)| body.clone())
                    .ok_or(ObjectsError::Unavailable)?;
                state.retain_body(&admitted)?;
                Ok(part)
            },
        )
    }

    #[cfg(feature = "local")]
    pub(crate) async fn upload_part_external(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
        part_number: u32,
        external: ExternalBody,
        idempotency_key: Option<String>,
    ) -> Result<wire::UploadedPart, ObjectsError> {
        let ExternalBody {
            root,
            digest,
            length,
            location,
        } = external;
        self.upload_part_stored(StoredPartRequest {
            bucket,
            object_key,
            upload_id,
            part_number,
            body: StoredBody::Local {
                root: Arc::new(root),
                digest,
                length,
                location,
            },
            body_digest: digest,
            idempotency_key,
        })
        .await
    }
}

#[async_trait]
impl ObjectsProvider for MemoryObjects {
    async fn create_bucket(
        &self,
        name: String,
        idempotency_key: Option<String>,
    ) -> Result<wire::Bucket, ObjectsError> {
        Self::validate_bucket_name(&name)?;
        let fingerprint = Self::fingerprint("create-bucket", &[name.as_bytes()]);
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Bucket(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Bucket,
            |state| {
                if state.names.contains_key(&name) {
                    return Err(ObjectsError::AlreadyExists);
                }
                let bucket_id = Self::next(state, "bucket")?;
                let reference = wire::BucketRef {
                    bucket_id: bucket_id.clone(),
                    name: name.clone(),
                };
                let bucket = BucketState {
                    reference: reference.clone(),
                    created_at: Self::timestamp(state),
                    objects: OrdMap::new(),
                    listing: OrdMap::new(),
                };
                state.names.insert(name, bucket_id.clone());
                state.buckets.insert(bucket_id, bucket.clone());
                Ok(wire::Bucket {
                    bucket: Some(reference),
                    created_at: Some(bucket.created_at),
                })
            },
        )
    }

    async fn head_bucket(&self, bucket: &wire::BucketRef) -> Result<wire::Bucket, ObjectsError> {
        let state = self.state.lock().await;
        let bucket = Self::bucket(&state, bucket)?;
        Ok(wire::Bucket {
            bucket: Some(bucket.reference.clone()),
            created_at: Some(bucket.created_at),
        })
    }

    async fn delete_bucket(
        &self,
        bucket: &wire::BucketRef,
        idempotency_key: Option<String>,
    ) -> Result<bool, ObjectsError> {
        let fields = Self::reference_fields(bucket);
        let fingerprint = Self::fingerprint("delete-bucket", &fields);
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Boolean(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Boolean,
            |state| {
                let existing = Self::bucket(state, bucket)?.clone();
                if existing.objects.values().any(|history| !history.is_empty())
                    || state
                        .multiparts
                        .values()
                        .any(|upload| upload.bucket == *bucket)
                {
                    return Err(ObjectsError::PreconditionFailed);
                }
                state.buckets.remove(&bucket.bucket_id);
                state.names.remove(&bucket.name);
                Ok(true)
            },
        )
    }

    async fn put(&self, request: PutRequest) -> Result<wire::ObjectVersion, ObjectsError> {
        let digest = *blake3::hash(&request.body).as_bytes();
        let body = StoredBody::memory(request.body.clone());
        self.put_stored(request, body, digest).await
    }

    async fn get(&self, request: GetRequest) -> Result<BufferedObject, ObjectsError> {
        let state = self.state.lock().await;
        let resolved = Self::resolve_get(&state, &request)?;
        drop(state);
        Self::read_resolved_get(resolved).await
    }

    async fn get_batch(
        &self,
        requests: Vec<GetRequest>,
    ) -> Vec<Result<BufferedObject, ObjectsError>> {
        let state = self.state.lock().await;
        let resolved = requests
            .iter()
            .map(|request| Self::resolve_get(&state, request))
            .collect::<Vec<_>>();
        drop(state);
        collect_ordered_bounded(
            resolved.into_iter().map(|result| async move {
                match result {
                    Ok(value) => Self::read_resolved_get(value).await,
                    Err(error) => Err(error),
                }
            }),
            MAX_CONCURRENT_BATCH_READS,
        )
        .await
    }

    async fn delete(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        version_id: Option<String>,
        condition: Option<Condition>,
        idempotency_key: Option<String>,
    ) -> Result<DeleteResult, ObjectsError> {
        Self::validate_key(&object_key)?;
        let bucket_fields = Self::reference_fields(&bucket);
        let condition_bytes = Self::condition_bytes(&condition);
        let fingerprint = Self::fingerprint(
            "delete",
            &[
                bucket_fields[0],
                bucket_fields[1],
                object_key.as_bytes(),
                version_id.as_deref().unwrap_or_default().as_bytes(),
                &condition_bytes,
            ],
        );
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Delete(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Delete,
            |state| {
                let current = Self::bucket(state, &bucket)?
                    .objects
                    .get(&object_key)
                    .and_then(|history| history.last())
                    .cloned();
                if !Self::condition(current.as_ref(), &condition) {
                    return Err(ObjectsError::PreconditionFailed);
                }
                if let Some(identity) = version_id {
                    let Some(removed) =
                        Self::bucket_mut(state, &bucket)?.remove_exact(&object_key, &identity)?
                    else {
                        return Ok(DeleteResult {
                            existed: false,
                            marker: None,
                        });
                    };
                    if let Some(body) = removed.body {
                        state.release_body(&body)?;
                    }
                    Ok(DeleteResult {
                        existed: true,
                        marker: None,
                    })
                } else {
                    let marker = Self::descriptor(
                        state,
                        blake3::hash(&[]).as_bytes(),
                        0,
                        wire::ObjectMetadata::default(),
                        true,
                    )?;
                    Self::bucket_mut(state, &bucket)?.append(
                        object_key,
                        Version {
                            descriptor: marker.clone(),
                            body: None,
                        },
                    );
                    Ok(DeleteResult {
                        existed: true,
                        marker: Some(marker),
                    })
                }
            },
        )
    }

    async fn list(
        &self,
        target: ReadTarget,
        prefix: String,
        delimiter: Option<String>,
        versions: bool,
        page_size: u32,
        continuation: Option<String>,
    ) -> Result<ProviderListPage, ObjectsError> {
        if page_size == 0
            || page_size > limits::LIST_PAGE_ENTRIES
            || delimiter.as_ref().is_some_and(String::is_empty)
        {
            return Err(ObjectsError::Invalid("invalid listing"));
        }
        let mut state = self.state.lock().await;
        if continuation.is_none() {
            let now = Instant::now();
            state.listings.retain(|_, view| now < view.expires_at);
        }
        let binding = ListingBinding {
            target: target.clone(),
            prefix: prefix.clone(),
            delimiter: delimiter.clone(),
            versions,
        };
        let fresh = continuation.is_none();
        let (view_id, offset, cursor) = if let Some(token) = continuation {
            Self::parse_listing_token(&token)?
        } else {
            let objects = Self::target_ref(&state, &target)?.listing.clone();
            let sequence = Self::next(&mut state, "listing")?;
            let view_id = format!("{}-{sequence}", state.listing_epoch);
            state.listings.insert(
                view_id.clone(),
                ListingView {
                    binding: binding.clone(),
                    objects,
                    expires_at: Instant::now() + Duration::from_secs(limits::LISTING_VIEW_SECONDS),
                    prefetched: None,
                },
            );
            (
                view_id,
                0,
                ListingCursor {
                    key: Some(prefix.clone()),
                    version_index: 0,
                },
            )
        };
        if state
            .listings
            .get(&view_id)
            .is_some_and(|view| Instant::now() >= view.expires_at)
        {
            state.listings.remove(&view_id);
            return Err(ObjectsError::Invalid("invalid continuation"));
        }
        let view = state
            .listings
            .get_mut(&view_id)
            .filter(|view| view.binding == binding)
            .ok_or(ObjectsError::Invalid("invalid continuation"))?;
        let page = Self::listing_page(view, &view_id, offset, page_size, cursor)?;
        if fresh {
            if page.continuation.is_none() {
                state.listings.remove(&view_id);
            } else if state.listings.len() > MAX_LISTING_VIEWS {
                state.listings.remove(&view_id);
                return Err(ObjectsError::Capacity);
            }
        }
        Ok(page)
    }

    async fn snapshot(
        &self,
        bucket: wire::BucketRef,
        idempotency_key: Option<String>,
    ) -> Result<wire::Snapshot, ObjectsError> {
        let fields = Self::reference_fields(&bucket);
        let fingerprint = Self::fingerprint("snapshot", &fields);
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Snapshot(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Snapshot,
            |state| {
                let source = Self::bucket(state, &bucket)?.clone();
                state.retain_bucket(&source)?;
                let snapshot_id = Self::next(state, "snapshot")?;
                let reference = wire::SnapshotRef {
                    snapshot_id: snapshot_id.clone(),
                    source_bucket_id: bucket.bucket_id,
                };
                let created_at = Self::timestamp(state);
                state.snapshots.insert(
                    snapshot_id,
                    SnapshotState {
                        reference: reference.clone(),
                        bucket: source,
                    },
                );
                Ok(wire::Snapshot {
                    snapshot: Some(reference),
                    created_at: Some(created_at),
                })
            },
        )
    }

    async fn destroy_snapshot(
        &self,
        snapshot: wire::SnapshotRef,
        idempotency_key: Option<String>,
    ) -> Result<bool, ObjectsError> {
        let fingerprint = Self::fingerprint(
            "destroy-snapshot",
            &[
                snapshot.snapshot_id.as_bytes(),
                snapshot.source_bucket_id.as_bytes(),
            ],
        );
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Boolean(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Boolean,
            |state| {
                let exists = state
                    .snapshots
                    .get(&snapshot.snapshot_id)
                    .is_some_and(|value| value.reference == snapshot);
                if !exists {
                    return Ok(false);
                }
                let removed = state
                    .snapshots
                    .remove(&snapshot.snapshot_id)
                    .ok_or(ObjectsError::Unavailable)?;
                state.release_bucket(&removed.bucket)?;
                Ok(true)
            },
        )
    }

    async fn fork(
        &self,
        source: ReadTarget,
        destination_name: String,
        idempotency_key: Option<String>,
    ) -> Result<wire::Bucket, ObjectsError> {
        let source_identity = match &source {
            ReadTarget::Bucket(value) => format!("bucket:{}:{}", value.bucket_id, value.name),
            ReadTarget::Snapshot(value) => {
                format!("snapshot:{}:{}", value.snapshot_id, value.source_bucket_id)
            }
        };
        let fingerprint = Self::fingerprint(
            "fork",
            &[source_identity.as_bytes(), destination_name.as_bytes()],
        );
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Bucket(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Bucket,
            |state| {
                if state.names.contains_key(&destination_name) {
                    return Err(ObjectsError::AlreadyExists);
                }
                let source = Self::target(state, &source)?;
                state.retain_bucket(&source)?;
                let bucket_id = Self::next(state, "bucket")?;
                let reference = wire::BucketRef {
                    bucket_id: bucket_id.clone(),
                    name: destination_name.clone(),
                };
                let created_at = Self::timestamp(state);
                state.names.insert(destination_name, bucket_id.clone());
                state.buckets.insert(
                    bucket_id,
                    BucketState {
                        reference: reference.clone(),
                        created_at,
                        objects: source.objects,
                        listing: source.listing,
                    },
                );
                Ok(wire::Bucket {
                    bucket: Some(reference),
                    created_at: Some(created_at),
                })
            },
        )
    }

    async fn create_multipart(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        metadata: wire::ObjectMetadata,
        condition: Option<Condition>,
        idempotency_key: Option<String>,
    ) -> Result<wire::MultipartUpload, ObjectsError> {
        Self::validate_key(&object_key)?;
        Self::validate_metadata(&metadata)?;
        let bucket_fields = Self::reference_fields(&bucket);
        let metadata_bytes = Self::metadata_bytes(&metadata);
        let condition_bytes = Self::condition_bytes(&condition);
        let fingerprint = Self::fingerprint(
            "create-multipart",
            &[
                bucket_fields[0],
                bucket_fields[1],
                object_key.as_bytes(),
                &metadata_bytes,
                &condition_bytes,
            ],
        );
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Multipart(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Multipart,
            |state| {
                Self::bucket(state, &bucket)?;
                let upload_id = Self::next(state, "upload")?;
                state.multiparts.insert(
                    upload_id.clone(),
                    MultipartState {
                        bucket,
                        object_key,
                        metadata,
                        condition,
                        parts: BTreeMap::new(),
                    },
                );
                Ok(wire::MultipartUpload { upload_id })
            },
        )
    }

    async fn upload_part(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
        part_number: u32,
        body: bytes::Bytes,
        idempotency_key: Option<String>,
    ) -> Result<wire::UploadedPart, ObjectsError> {
        let digest = *blake3::hash(&body).as_bytes();
        self.upload_part_stored(StoredPartRequest {
            bucket,
            object_key,
            upload_id,
            part_number,
            body: StoredBody::memory(body),
            body_digest: digest,
            idempotency_key,
        })
        .await
    }

    async fn list_parts(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
    ) -> Result<Vec<wire::UploadedPart>, ObjectsError> {
        let state = self.state.lock().await;
        let upload = state
            .multiparts
            .get(&upload_id)
            .filter(|upload| upload.bucket == bucket && upload.object_key == object_key)
            .ok_or(ObjectsError::NotFound)?;
        Ok(upload
            .parts
            .values()
            .map(|(part, _)| part.clone())
            .collect())
    }

    async fn complete_multipart(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
        parts: Vec<wire::UploadedPart>,
        idempotency_key: Option<String>,
    ) -> Result<wire::ObjectVersion, ObjectsError> {
        if parts.is_empty() || parts.len() > limits::MULTIPART_PARTS as usize {
            return Err(ObjectsError::Invalid("invalid multipart completion"));
        }
        let bucket_fields = Self::reference_fields(&bucket);
        let mut part_bytes = Vec::new();
        for part in &parts {
            part_bytes.extend_from_slice(&part.part_number.to_le_bytes());
            part_bytes.extend_from_slice(&(part.etag.len() as u64).to_le_bytes());
            part_bytes.extend_from_slice(part.etag.as_bytes());
            part_bytes.extend_from_slice(&part.size.to_le_bytes());
        }
        let fingerprint = Self::fingerprint(
            "complete-multipart",
            &[
                bucket_fields[0],
                bucket_fields[1],
                object_key.as_bytes(),
                upload_id.as_bytes(),
                &part_bytes,
            ],
        );
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Version(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Version,
            |state| {
                let upload = state
                    .multiparts
                    .remove(&upload_id)
                    .filter(|upload| upload.bucket == bucket && upload.object_key == object_key)
                    .ok_or(ObjectsError::NotFound)?;
                let exact = upload.parts.values().map(|(part, _)| part).eq(parts.iter());
                let sizes_valid = parts.iter().enumerate().all(|(index, part)| {
                    index + 1 == parts.len() || part.size >= limits::MIN_MULTIPART_PART_BYTES
                });
                if !exact || !sizes_valid {
                    state.multiparts.insert(upload_id, upload);
                    return Err(ObjectsError::Invalid("multipart receipts do not match"));
                }
                let current = Self::bucket(state, &bucket)?
                    .objects
                    .get(&object_key)
                    .and_then(|history| history.last())
                    .cloned();
                if !Self::condition(current.as_ref(), &upload.condition) {
                    state.multiparts.insert(upload_id, upload);
                    return Err(ObjectsError::PreconditionFailed);
                }
                let body_length =
                    match upload.parts.values().try_fold(0usize, |total, (_, body)| {
                        total.checked_add(body.len()).ok_or(ObjectsError::Capacity)
                    }) {
                        Ok(body_length) => body_length,
                        Err(error) => {
                            state.multiparts.insert(upload_id, upload);
                            return Err(error);
                        }
                    };
                if body_length > state.maximum_object_bytes {
                    state.multiparts.insert(upload_id, upload);
                    return Err(ObjectsError::Capacity);
                }
                let bodies = upload
                    .parts
                    .into_values()
                    .map(|(_, body)| body)
                    .collect::<Vec<_>>();
                let mut hasher = blake3::Hasher::new();
                for body in &bodies {
                    body.update_hash(&mut hasher)?;
                }
                let body_digest = *hasher.finalize().as_bytes();
                let descriptor =
                    Self::descriptor(state, &body_digest, body_length, upload.metadata, false)?;
                Self::bucket_mut(state, &bucket)?.append(
                    object_key,
                    Version {
                        descriptor: descriptor.clone(),
                        body: Some(StoredBody::Composite {
                            parts: bodies.into(),
                            length: body_length,
                        }),
                    },
                );
                Ok(descriptor)
            },
        )
    }

    async fn abort_multipart(
        &self,
        bucket: wire::BucketRef,
        object_key: String,
        upload_id: String,
        idempotency_key: Option<String>,
    ) -> Result<bool, ObjectsError> {
        let bucket_fields = Self::reference_fields(&bucket);
        let fingerprint = Self::fingerprint(
            "abort-multipart",
            &[
                bucket_fields[0],
                bucket_fields[1],
                object_key.as_bytes(),
                upload_id.as_bytes(),
            ],
        );
        let mut state = self.state.lock().await;
        Self::execute(
            &mut state,
            idempotency_key.as_deref(),
            fingerprint,
            |outcome| match outcome {
                MutationOutcome::Boolean(value) => Some(value.clone()),
                _ => None,
            },
            MutationOutcome::Boolean,
            |state| {
                let matches = state.multiparts.get(&upload_id).is_some_and(|upload| {
                    upload.bucket == bucket && upload.object_key == object_key
                });
                if !matches {
                    return Ok(false);
                }
                if let Some(upload) = state.multiparts.remove(&upload_id) {
                    for (_, body) in upload.parts.values() {
                        state.release_body(body)?;
                    }
                }
                Ok(true)
            },
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeSet;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn bounded_collection_limits_concurrency_and_restores_order() {
        const LIMIT: usize = 3;
        let active = Arc::new(AtomicUsize::new(0));
        let peak = Arc::new(AtomicUsize::new(0));
        let barrier = Arc::new(tokio::sync::Barrier::new(LIMIT));
        let futures = (0..11).map(|index| {
            let active = Arc::clone(&active);
            let peak = Arc::clone(&peak);
            let barrier = Arc::clone(&barrier);
            async move {
                let now = active.fetch_add(1, Ordering::SeqCst) + 1;
                peak.fetch_max(now, Ordering::SeqCst);
                if index < LIMIT {
                    barrier.wait().await;
                }
                tokio::task::yield_now().await;
                active.fetch_sub(1, Ordering::SeqCst);
                10 - index
            }
        });

        let results = collect_ordered_bounded(futures, LIMIT).await;

        assert_eq!(peak.load(Ordering::SeqCst), LIMIT);
        assert_eq!(active.load(Ordering::SeqCst), 0);
        assert_eq!(results, (0..11).map(|index| 10 - index).collect::<Vec<_>>());
    }

    #[tokio::test]
    async fn get_batch_preserves_order_and_independent_failures() {
        let (store, bucket) = MemoryObjects::with_default_bucket();
        put(&store, &bucket, "first", b"abc", None).await;
        put(&store, &bucket, "second", b"defg", None).await;
        let requests = ["second", "missing", "first"]
            .into_iter()
            .map(|object_key| GetRequest {
                target: ReadTarget::Bucket(bucket.clone()),
                object_key: object_key.to_owned(),
                version_id: None,
                range: None,
                if_match: None,
                if_none_match: None,
                maximum_bytes: 4,
            })
            .collect();
        let results = store.get_batch(requests).await;
        assert_eq!(results.len(), 3);
        assert_eq!(
            results
                .first()
                .and_then(|value| value.as_ref().ok())
                .map(|value| &value.body[..]),
            Some(b"defg".as_slice())
        );
        assert!(matches!(results.get(1), Some(Err(ObjectsError::NotFound))));
        assert_eq!(
            results
                .get(2)
                .and_then(|value| value.as_ref().ok())
                .map(|value| &value.body[..]),
            Some(b"abc".as_slice())
        );
    }

    #[tokio::test]
    async fn listing_continuation_is_bound_to_exact_query() {
        let (store, bucket) = MemoryObjects::with_default_bucket();
        put(&store, &bucket, "a|b1", b"first", None).await;
        put(&store, &bucket, "a|b2", b"second", None).await;
        let first = store
            .list(
                ReadTarget::Bucket(bucket.clone()),
                "a|b".into(),
                Some("c".into()),
                false,
                1,
                None,
            )
            .await
            .unwrap_or_else(|_| unreachable!());
        assert_eq!(
            store
                .list(
                    ReadTarget::Bucket(bucket),
                    "a".into(),
                    Some("b|c".into()),
                    false,
                    1,
                    first.continuation,
                )
                .await,
            Err(ObjectsError::Invalid("invalid continuation"))
        );
    }

    #[tokio::test]
    async fn listing_views_are_bounded_and_do_not_hold_bodies() {
        let (store, bucket) = MemoryObjects::with_default_bucket();
        let first = put(&store, &bucket, "a", b"payload", None).await;
        put(&store, &bucket, "b", b"other", None).await;
        let mut oldest = None;
        for index in 0..MAX_LISTING_VIEWS {
            let page = store
                .list(
                    ReadTarget::Bucket(bucket.clone()),
                    String::new(),
                    None,
                    false,
                    1,
                    None,
                )
                .await
                .unwrap_or_else(|_| unreachable!());
            if index == 0 {
                oldest = page.continuation;
            }
        }
        assert_eq!(store.state.lock().await.listings.len(), MAX_LISTING_VIEWS);
        assert_eq!(
            store
                .list(
                    ReadTarget::Bucket(bucket.clone()),
                    String::new(),
                    None,
                    false,
                    1,
                    None,
                )
                .await,
            Err(ObjectsError::Capacity)
        );
        assert_eq!(store.state.lock().await.listings.len(), MAX_LISTING_VIEWS);
        let continuation = store
            .list(
                ReadTarget::Bucket(bucket.clone()),
                String::new(),
                None,
                false,
                1,
                oldest,
            )
            .await
            .unwrap_or_else(|_| unreachable!());
        assert_eq!(
            continuation
                .entries
                .first()
                .map(|entry| entry.object_key.as_str()),
            Some("b")
        );
        store
            .delete(
                bucket.clone(),
                "a".into(),
                Some(first.version_id),
                None,
                None,
            )
            .await
            .unwrap_or_else(|_| unreachable!());
        assert_eq!(store.state.lock().await.bytes, b"other".len());
        let one_page = store
            .list(
                ReadTarget::Bucket(bucket),
                String::new(),
                None,
                false,
                2,
                None,
            )
            .await
            .unwrap_or_else(|_| unreachable!());
        assert!(one_page.continuation.is_none());
        assert_eq!(store.state.lock().await.listings.len(), MAX_LISTING_VIEWS);
    }

    #[tokio::test]
    async fn expired_listing_continuation_is_rejected_and_reaped() {
        let (store, bucket) = MemoryObjects::with_default_bucket();
        put(&store, &bucket, "a", b"a", None).await;
        put(&store, &bucket, "b", b"b", None).await;
        let token = store
            .list(
                ReadTarget::Bucket(bucket.clone()),
                String::new(),
                None,
                false,
                1,
                None,
            )
            .await
            .unwrap_or_else(|_| unreachable!())
            .continuation;
        {
            let mut state = store.state.lock().await;
            let view = state
                .listings
                .values_mut()
                .next()
                .unwrap_or_else(|| unreachable!());
            view.expires_at = Instant::now();
        }
        assert_eq!(
            store
                .list(
                    ReadTarget::Bucket(bucket),
                    String::new(),
                    None,
                    false,
                    1,
                    token
                )
                .await,
            Err(ObjectsError::Invalid("invalid continuation"))
        );
        assert!(store.state.lock().await.listings.is_empty());
    }

    #[tokio::test]
    async fn long_single_entry_pagination_keeps_one_view_and_bounded_tokens() {
        let (store, bucket) = MemoryObjects::with_default_bucket();
        for index in 0..2_000 {
            put(&store, &bucket, &format!("key-{index:04}"), b"x", None).await;
        }
        let mut continuation = None;
        for expected in 0..2_000 {
            let page = store
                .list(
                    ReadTarget::Bucket(bucket.clone()),
                    String::new(),
                    None,
                    false,
                    1,
                    continuation,
                )
                .await
                .unwrap_or_else(|_| unreachable!());
            assert_eq!(
                page.entries.first().map(|entry| entry.object_key.as_str()),
                Some(format!("key-{expected:04}").as_str())
            );
            continuation = page.continuation;
            let state = store.state.lock().await;
            assert_eq!(state.listings.len(), 1);
            let view = state
                .listings
                .values()
                .next()
                .unwrap_or_else(|| unreachable!());
            assert_eq!(view.objects.len(), 2_000);
            assert_eq!(view.prefetched.is_some(), expected < 1_999);
            assert!(continuation.as_ref().is_none_or(|token| token.len() < 256));
        }
        assert!(continuation.is_none());
    }

    #[tokio::test]
    async fn listing_prefetch_preserves_replay_and_page_size_changes() {
        let (store, bucket) = MemoryObjects::with_default_bucket();
        for key in ["a", "b", "c", "d"] {
            put(&store, &bucket, key, b"x", None).await;
        }
        let first = store
            .list(
                ReadTarget::Bucket(bucket.clone()),
                String::new(),
                None,
                false,
                1,
                None,
            )
            .await
            .unwrap_or_else(|_| unreachable!());
        let token = first.continuation.unwrap_or_else(|| unreachable!());
        let two = store
            .list(
                ReadTarget::Bucket(bucket.clone()),
                String::new(),
                None,
                false,
                2,
                Some(token.clone()),
            )
            .await
            .unwrap_or_else(|_| unreachable!());
        let replay = store
            .list(
                ReadTarget::Bucket(bucket),
                String::new(),
                None,
                false,
                1,
                Some(token),
            )
            .await
            .unwrap_or_else(|_| unreachable!());
        assert_eq!(
            two.entries
                .iter()
                .map(|entry| entry.object_key.as_str())
                .collect::<Vec<_>>(),
            ["b", "c"]
        );
        assert_eq!(
            replay
                .entries
                .first()
                .map(|entry| entry.object_key.as_str()),
            Some("b")
        );
    }

    #[allow(
        clippy::too_many_lines,
        reason = "one deterministic matrix compares every paginated mode with an independent model"
    )]
    #[tokio::test]
    async fn paged_listing_matches_independent_sorted_model() {
        let (store, bucket) = MemoryObjects::with_default_bucket();
        let keys = [
            "a",
            "a/",
            "a/child",
            "a//child",
            "a/child/next",
            "a:one",
            "a::two",
            "b",
            "é/child",
            "\u{d7ff}/child",
            "\u{e000}/child",
            "\u{10ffff}/child",
        ];
        let mut history = BTreeMap::<String, Vec<wire::ObjectVersion>>::new();
        for key in keys {
            let first = put(&store, &bucket, key, b"first", None).await;
            let second = put(&store, &bucket, key, b"second", None).await;
            history.insert(key.to_owned(), vec![first, second]);
        }
        let deleted = store
            .delete(bucket.clone(), "b".into(), None, None, None)
            .await
            .unwrap_or_else(|_| unreachable!())
            .marker
            .unwrap_or_else(|| unreachable!());
        history
            .get_mut("b")
            .unwrap_or_else(|| unreachable!())
            .push(deleted);
        for prefix in ["", "a", "a/", "é", "\u{d7ff}", "\u{10ffff}"] {
            for delimiter in [None, Some("/"), Some("::")] {
                for versions in [false, true] {
                    let mut model = Vec::<(String, Option<String>)>::new();
                    let mut groups = BTreeSet::new();
                    for (key, versions_for_key) in &history {
                        let Some(remainder) = key.strip_prefix(prefix) else {
                            continue;
                        };
                        if let Some(delimiter) = delimiter
                            && let Some(position) = remainder.find(delimiter)
                        {
                            groups.insert(
                                key.get(..prefix.len() + position + delimiter.len())
                                    .unwrap_or_else(|| unreachable!())
                                    .to_owned(),
                            );
                            continue;
                        }
                        if versions {
                            model.extend(
                                versions_for_key
                                    .iter()
                                    .rev()
                                    .map(|version| (key.clone(), Some(version.version_id.clone()))),
                            );
                        } else if let Some(version) = versions_for_key.last()
                            && !version.delete_marker
                        {
                            model.push((key.clone(), Some(version.version_id.clone())));
                        }
                    }
                    model.extend(groups.into_iter().map(|group| (group, None)));
                    model.sort_by(|left, right| left.0.cmp(&right.0));
                    for page_size in [1, 2, 5, 100] {
                        let mut continuation = None;
                        let mut offset = 0;
                        loop {
                            let page = store
                                .list(
                                    ReadTarget::Bucket(bucket.clone()),
                                    prefix.to_owned(),
                                    delimiter.map(str::to_owned),
                                    versions,
                                    page_size,
                                    continuation,
                                )
                                .await
                                .unwrap_or_else(|_| unreachable!());
                            let expected = model
                                .get(offset..model.len().min(offset + page_size as usize))
                                .unwrap_or_else(|| unreachable!());
                            let entries: Vec<_> = expected
                                .iter()
                                .filter_map(|(key, version)| {
                                    version.as_ref().map(|id| (key.clone(), id.clone()))
                                })
                                .collect();
                            let prefixes: Vec<_> = expected
                                .iter()
                                .filter(|(_, version)| version.is_none())
                                .map(|(key, _)| key.clone())
                                .collect();
                            assert_eq!(
                                page.entries
                                    .iter()
                                    .map(|entry| (
                                        entry.object_key.clone(),
                                        entry
                                            .version
                                            .as_ref()
                                            .unwrap_or_else(|| unreachable!())
                                            .version_id
                                            .clone(),
                                    ))
                                    .collect::<Vec<_>>(),
                                entries,
                                "entries: prefix={prefix:?} delimiter={delimiter:?} versions={versions} page_size={page_size} offset={offset}"
                            );
                            assert_eq!(page.common_prefixes, prefixes);
                            offset += expected.len();
                            assert_eq!(page.continuation.is_some(), offset < model.len());
                            let Some(token) = page.continuation else {
                                break;
                            };
                            continuation = Some(token);
                        }
                        assert_eq!(offset, model.len());
                    }
                }
            }
        }
    }

    fn metadata() -> wire::ObjectMetadata {
        wire::ObjectMetadata {
            content_type: "text/plain".into(),
            ..Default::default()
        }
    }

    async fn put(
        store: &MemoryObjects,
        bucket: &wire::BucketRef,
        object_key: &str,
        body: &[u8],
        condition: Option<Condition>,
    ) -> wire::ObjectVersion {
        store
            .put(PutRequest {
                bucket: bucket.clone(),
                object_key: object_key.into(),
                body: bytes::Bytes::copy_from_slice(body),
                metadata: metadata(),
                condition,
                idempotency_key: None,
            })
            .await
            .unwrap_or_else(|_| unreachable!())
    }

    #[tokio::test]
    async fn synchronous_bucket_composition_is_exact_and_capacity_bounded() {
        assert!(matches!(
            MemoryObjects::new(0),
            Err(ObjectsError::Invalid("memory byte limit must be positive"))
        ));
        assert!(MemoryObjects::with_bucket("INVALID", 4).is_err());
        assert!(MemoryObjects::with_bucket_limits("embedded", 5, 4).is_err());

        let (independent, independent_bucket) =
            MemoryObjects::with_bucket_limits("independent", 3, 6)
                .unwrap_or_else(|_| unreachable!());
        put(&independent, &independent_bucket, "first", b"one", None).await;
        put(&independent, &independent_bucket, "second", b"two", None).await;
        assert_eq!(
            independent
                .put(PutRequest {
                    bucket: independent_bucket.clone(),
                    object_key: "aggregate-overflow".into(),
                    body: bytes::Bytes::from_static(b"x"),
                    metadata: metadata(),
                    condition: None,
                    idempotency_key: None,
                })
                .await,
            Err(ObjectsError::Capacity)
        );
        let (per_object, per_object_bucket) = MemoryObjects::with_bucket_limits("per-object", 3, 6)
            .unwrap_or_else(|_| unreachable!());
        assert_eq!(
            per_object
                .put(PutRequest {
                    bucket: per_object_bucket,
                    object_key: "too-large".into(),
                    body: bytes::Bytes::from_static(b"four"),
                    metadata: metadata(),
                    condition: None,
                    idempotency_key: None,
                })
                .await,
            Err(ObjectsError::Capacity)
        );

        let (store, bucket) =
            MemoryObjects::with_bucket("embedded", 4).unwrap_or_else(|_| unreachable!());
        assert_eq!(
            store
                .head_bucket(&bucket)
                .await
                .unwrap_or_else(|_| unreachable!())
                .bucket,
            Some(bucket.clone())
        );
        put(&store, &bucket, "full", b"four", None).await;
        assert_eq!(
            store
                .put(PutRequest {
                    bucket,
                    object_key: "overflow".into(),
                    body: bytes::Bytes::from_static(b"x"),
                    metadata: metadata(),
                    condition: None,
                    idempotency_key: None,
                })
                .await,
            Err(ObjectsError::Capacity)
        );
    }

    #[tokio::test]
    async fn memory_provider_passes_the_public_conformance_suite() -> Result<(), ObjectsError> {
        crate::conformance::verify(&MemoryObjects::default(), "conformance").await
    }

    #[tokio::test]
    async fn shared_snapshot_and_fork_bodies_retain_exact_capacity() {
        let store = MemoryObjects::new(3).unwrap_or_else(|_| unreachable!());
        let source = store
            .create_bucket("source-bucket".into(), Some("create-source".into()))
            .await
            .unwrap_or_else(|_| unreachable!())
            .bucket
            .unwrap_or_else(|| unreachable!());
        let version = put(&store, &source, "shared", b"abc", None).await;
        let version_id = version.version_id.clone();
        let snapshot = store
            .snapshot(source.clone(), Some("snapshot".into()))
            .await
            .unwrap_or_else(|_| unreachable!())
            .snapshot
            .unwrap_or_else(|| unreachable!());
        assert!(
            store
                .delete(
                    source.clone(),
                    "shared".into(),
                    Some(version.version_id),
                    None,
                    Some("delete-source".into()),
                )
                .await
                .is_ok()
        );
        assert!(matches!(
            store
                .put(PutRequest {
                    bucket: source.clone(),
                    object_key: "replacement".into(),
                    body: bytes::Bytes::from_static(b"def"),
                    metadata: metadata(),
                    condition: None,
                    idempotency_key: None,
                })
                .await,
            Err(ObjectsError::Capacity)
        ));
        let fork = store
            .fork(
                ReadTarget::Snapshot(snapshot.clone()),
                "fork-bucket".into(),
                Some("fork".into()),
            )
            .await
            .unwrap_or_else(|_| unreachable!())
            .bucket
            .unwrap_or_else(|| unreachable!());
        assert!(
            store
                .destroy_snapshot(snapshot, Some("destroy-snapshot".into()))
                .await
                .unwrap_or(false)
        );
        assert!(matches!(
            store
                .put(PutRequest {
                    bucket: source.clone(),
                    object_key: "replacement".into(),
                    body: bytes::Bytes::from_static(b"def"),
                    metadata: metadata(),
                    condition: None,
                    idempotency_key: None,
                })
                .await,
            Err(ObjectsError::Capacity)
        ));
        assert!(
            store
                .delete(
                    fork,
                    "shared".into(),
                    Some(version_id),
                    None,
                    Some("delete-fork".into()),
                )
                .await
                .is_ok()
        );
        assert!(
            store
                .put(PutRequest {
                    bucket: source,
                    object_key: "replacement".into(),
                    body: bytes::Bytes::from_static(b"def"),
                    metadata: metadata(),
                    condition: None,
                    idempotency_key: None,
                })
                .await
                .is_ok()
        );
    }
}