bytes.rs

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 * Copyright (c) Jan-Paul Bultmann
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
*/

//! Core byte container types.
//!
//! [`Bytes`] provides cheap, zero-copy access to immutable bytes from a
//! variety of sources. Implement the [`ByteSource`] trait for your type to
//! integrate it with the container. Each `Bytes` keeps its backing storage
//! alive through a reference-counted [`ByteOwner`], ensuring the data stays
//! valid for as long as needed.
//!
//! `Bytes` decouples data access from ownership so that callers can obtain a
//! slice and then release any external locks. After reading the bytes from a
//! [`ByteSource`], convert it into its [`ByteOwner`] to keep the data alive
//! without retaining the original guard. This is especially useful for Python
//! integration where acquiring the raw pointer to a `bytes` object requires
//! holding the GIL, but once the slice is acquired, only the owner needs to be
//! kept alive.
//!
//! ## Weak references
//!
//! A `Bytes` can be downgraded to a [`WeakBytes`] to hold a non-owning reference
//! without keeping the underlying data alive:
//!
//! ```
//! use anybytes::Bytes;
//!
//! let bytes = Bytes::from(vec![1u8, 2, 3]);
//! let weak = bytes.downgrade();
//! assert!(weak.upgrade().is_some());
//! drop(bytes);
//! assert!(weak.upgrade().is_none());
//! ```
//!
//! ## Downcasting owners
//!
//! When the backing type is known, [`Bytes::downcast_to_owner`] retrieves it
//! again:
//!
//! ```
//! use anybytes::Bytes;
//! use std::sync::Arc;
//!
//! let bytes = Bytes::from_source(vec![1u8, 2, 3, 4]);
//! let owner: Arc<Vec<u8>> = bytes
//!     .downcast_to_owner()
//!     .expect("Downcast of known type.");
//! assert_eq!(&*owner, &[1, 2, 3, 4]);
//! ```

use std::any::Any;
use std::ascii::escape_default;
use std::borrow::Borrow;
use std::cmp;
use std::fmt;
use std::hash;
use std::ops::Deref;
use std::slice::SliceIndex;
use std::sync::{Arc, Weak};

pub(crate) fn is_subslice(slice: &[u8], subslice: &[u8]) -> bool {
    let slice_start = slice.as_ptr() as usize;
    let slice_end = slice_start + slice.len();
    let subslice_start = subslice.as_ptr() as usize;
    let subslice_end = subslice_start + subslice.len();
    subslice_start >= slice_start && subslice_end <= slice_end
}

/// A type that can provide its bytes and yield an owner for them.
///
/// Implementors of this trait serve as sources for [`Bytes`].  The slice
/// returned by [`ByteSource::as_bytes`] must remain valid for as long as the
/// source itself is alive **or** the [`ByteOwner`] obtained from
/// [`ByteSource::get_owner`] is kept alive.  Splitting these capabilities lets
/// callers obtain a byte slice and then drop any locks or guards by converting
/// the source into its owner while still keeping the data valid.
///
/// The returned owner keeps the underlying data alive as long as any [`Bytes`]
/// derived from it are in scope.
pub unsafe trait ByteSource {
    /// The type that owns the bytes.
    type Owner: ByteOwner;

    /// Returns a view of the contained bytes.
    fn as_bytes(&self) -> &[u8];

    /// Consumes the source and returns the owning value.
    fn get_owner(self) -> Self::Owner;
}

/// A trait for types that keep the backing bytes of [`Bytes`] alive.
///
/// Implementors must guarantee that the returned `Arc<dyn ByteOwner>` keeps
/// the underlying storage valid for as long as that `Arc` is held.  Dropping
/// the original value must not invalidate any [`Bytes`] or [`View`](crate::view::View)
/// instances that cloned the owner.
///
/// This trait extends [`Any`] so that owners can be downcast directly via
/// [`Arc::downcast`].  Callers can upcast the owner to
/// `Arc<dyn Any + Send + Sync>` and then attempt a downcast to reclaim the
/// concrete type.
pub trait ByteOwner: Any + Sync + Send {}

impl<T: ByteSource + Sync + Send + 'static> ByteOwner for T {}

/// Immutable bytes with zero-copy slicing and cloning.
///
/// Access itself is extremely cheap via no-op conversion to a `&[u8]`.
///
/// The storage mechanism backing the bytes can be extended
/// and is implemented for a variety of sources already,
/// including other byte handling crates `Bytes`, mmap-ed files,
/// `String`s and `Zerocopy` types.
///
/// See [ByteOwner] for an exhaustive list and more details.
pub struct Bytes {
    // Raw pointer instead of a reference to avoid Stacked/Tree Borrows
    // violations when `Bytes` is passed by value to a function (including
    // `std::mem::drop`) and this happens to be the last strong reference.
    // A `&'static [u8]` would be "protected" as a function argument while
    // the `Arc` owner is deallocated, which is UB under both borrow models.
    data: *const [u8],
    // Actual owner of the bytes.
    owner: Arc<dyn ByteOwner>,
}

/// Weak variant of [Bytes] that doesn't retain the data
/// unless a strong [Bytes] is referencing it.
///
/// The referenced subrange of the [Bytes] is reconstructed
/// on [WeakBytes::upgrade].
#[derive(Clone, Debug)]
pub struct WeakBytes {
    data: *const [u8],
    owner: Weak<dyn ByteOwner>,
}

// ByteOwner is Send + Sync and Bytes is immutable.
// Raw pointers are !Send + !Sync by default, but the owner guarantees
// the backing data is accessible from any thread.
unsafe impl Send for Bytes {}
unsafe impl Sync for Bytes {}

impl Clone for Bytes {
    fn clone(&self) -> Self {
        Self {
            data: self.data,
            owner: self.owner.clone(),
        }
    }
}

// Core implementation of Bytes.
impl Bytes {
    #[inline]
    pub(crate) fn data_ptr(&self) -> *const [u8] {
        self.data
    }

    #[inline]
    pub(crate) unsafe fn get_data(&self) -> &[u8] {
        unsafe { &*self.data }
    }

    #[inline]
    pub(crate) unsafe fn set_data(&mut self, data: &[u8]) {
        self.data = data as *const [u8];
    }

    #[inline]
    pub(crate) fn get_owner(&self) -> Arc<dyn ByteOwner> {
        self.owner.clone()
    }

    #[inline]
    pub(crate) fn take_owner(self) -> Arc<dyn ByteOwner> {
        self.owner
    }

    /// Creates an empty `Bytes`.
    #[inline]
    pub fn empty() -> Self {
        Self::from_source(&[0u8; 0][..])
    }

    /// Creates `Bytes` from an arbitrary slice and its owner.
    ///
    /// # Safety
    /// The caller must ensure that `data` remains valid for the lifetime of
    /// `owner`. No lifetime checks are performed.
    pub unsafe fn from_raw_parts(data: &[u8], owner: Arc<dyn ByteOwner>) -> Self {
        Self {
            data: data as *const [u8],
            owner,
        }
    }

    /// Creates `Bytes` from a [`ByteSource`] (for example, `Vec<u8>`).
    pub fn from_source(source: impl ByteSource) -> Self {
        let data = source.as_bytes() as *const [u8];
        let owner = source.get_owner();
        let owner = Arc::new(owner);
        Self { data, owner }
    }

    /// Creates `Bytes` from an `Arc<ByteSource + ByteOwner>`.
    ///
    /// This provides a potentially faster path for `Bytes` creation
    /// as it can forgoe an additional allocation for the wrapping Arc.
    /// For example when you implement `ByteOwner` for a `zerocopy` type,
    /// sadly we can't provide a blanked implementation for those types
    /// because of the orphane rule.
    pub fn from_owning_source_arc(arc: Arc<impl ByteSource + ByteOwner>) -> Self {
        let data = arc.as_bytes() as *const [u8];
        Self { data, owner: arc }
    }

    /// Memory-map the given file descriptor and return the bytes.
    ///
    /// # Safety
    ///
    /// The caller must ensure that the mapped file is not modified for the
    /// lifetime of the returned [`Bytes`]. Changing the file contents while it
    /// is mapped can lead to undefined behavior. The file handle does not need
    /// to remain open after mapping.
    ///
    /// The argument may be any type that implements [`memmap2::MmapAsRawDesc`],
    /// such as [`&std::fs::File`] or [`&tempfile::NamedTempFile`].
    #[cfg(feature = "mmap")]
    pub unsafe fn map_file<F>(file: F) -> std::io::Result<Self>
    where
        F: memmap2::MmapAsRawDesc,
    {
        let map = memmap2::MmapOptions::new().map(file)?;
        Ok(Self::from_source(map))
    }

    /// Memory-map a specific region of the given file descriptor and return the
    /// bytes.
    ///
    /// # Safety
    ///
    /// The caller must ensure that the mapped file is not modified for the
    /// lifetime of the returned [`Bytes`]. Changing the file contents while it
    /// is mapped can lead to undefined behavior. The file handle does not need
    /// to remain open after mapping. The `offset` must meet the underlying
    /// platform's alignment requirements (typically page size).
    ///
    /// The argument may be any type that implements [`memmap2::MmapAsRawDesc`],
    /// such as [`&std::fs::File`] or [`&tempfile::NamedTempFile`].
    #[cfg(feature = "mmap")]
    pub unsafe fn map_file_region<F>(file: F, offset: u64, len: usize) -> std::io::Result<Self>
    where
        F: memmap2::MmapAsRawDesc,
    {
        let map = memmap2::MmapOptions::new()
            .offset(offset)
            .len(len)
            .map(file)?;
        Ok(Self::from_source(map))
    }

    #[inline]
    pub(crate) fn as_slice(&self) -> &[u8] {
        // SAFETY: The owner keeps the data alive for the lifetime of self.
        unsafe { &*self.data }
    }

    /// Returns the owner of the Bytes as a `Arc<T>`.
    ///
    /// # Examples
    ///
    /// ```
    /// use anybytes::Bytes;
    /// use std::sync::Arc;
    /// let owner: Vec<u8> = vec![0,1,2,3];
    /// let bytes = Bytes::from_source(owner);
    /// let owner: Arc<Vec<u8>> = bytes
    ///     .downcast_to_owner()
    ///     .expect("Downcast of known type.");
    /// ```
    pub fn downcast_to_owner<T>(self) -> Result<Arc<T>, Bytes>
    where
        T: Send + Sync + 'static,
    {
        let owner_any: Arc<dyn Any + Send + Sync> = self.owner.clone();
        match owner_any.downcast::<T>() {
            Ok(owner) => Ok(owner),
            Err(_) => Err(self),
        }
    }

    /// Attempt to reclaim the owner as `T` if this is the
    /// last strong reference.
    ///
    /// Returns `Ok(T)` on success, otherwise returns the
    /// original `Bytes` if the owner is of a different type or
    /// still referenced elsewhere.
    pub fn try_unwrap_owner<T>(self) -> Result<T, Self>
    where
        T: ByteOwner + Send + Sync + 'static,
    {
        let Self { data, owner } = self;

        // Verify the concrete type without releasing the owner yet.
        let any: Arc<dyn Any + Send + Sync> = owner.clone();
        let arc_t = match Arc::downcast::<T>(any) {
            Ok(arc_t) => arc_t,
            Err(_) => return Err(Self { data, owner }),
        };

        // Drop our dynamic reference so the downcasted `Arc` becomes unique.
        drop(owner);

        match Arc::try_unwrap(arc_t) {
            Ok(t) => Ok(t),
            Err(arc_t) => {
                // Another strong reference exists; rebuild the `Bytes`.
                let owner: Arc<dyn ByteOwner> = arc_t;
                Err(Self { data, owner })
            }
        }
    }

    /// Returns a slice of self for the provided range.
    /// This operation is `O(1)`.
    pub fn slice(&self, range: impl SliceIndex<[u8], Output = [u8]>) -> Self {
        let sliced = &self.as_slice()[range];
        Self {
            data: sliced as *const [u8],
            owner: self.owner.clone(),
        }
    }

    /// Attempt to convert `slice` to a zero-copy slice of this `Bytes`.
    ///
    /// Returns `None` if `slice` is outside the memory range of this
    /// `Bytes`.
    ///
    /// This is similar to `bytes::Bytes::slice_ref` from `bytes 0.5.4`,
    /// but does not panic.
    pub fn slice_to_bytes(&self, slice: &[u8]) -> Option<Self> {
        if is_subslice(self.as_slice(), slice) {
            let owner = self.owner.clone();
            Some(Self {
                data: slice as *const [u8],
                owner,
            })
        } else {
            None
        }
    }

    /// Returns a `Bytes` with the first `len` bytes of `self`.
    /// Modifies `self` to contain the remaining bytes.
    /// Returns `None` if `len` is greater than the length of `self`.
    /// This operation is `O(1)`.
    pub fn take_prefix(&mut self, len: usize) -> Option<Self> {
        // Copy the pointer to avoid borrowing self.
        let slice = unsafe { &*self.data };
        if len > slice.len() {
            return None;
        }
        let (data, rest) = slice.split_at(len);
        self.data = rest as *const [u8];
        Some(Self {
            data: data as *const [u8],
            owner: self.owner.clone(),
        })
    }

    /// Returns a `Bytes` with the last `len` bytes of `self`.
    /// Modifies `self` to contain the remaining bytes.
    /// Returns `None` if `len` is greater than the length of `self`.
    /// This operation is `O(1)`.
    pub fn take_suffix(&mut self, len: usize) -> Option<Self> {
        let slice = unsafe { &*self.data };
        if len > slice.len() {
            return None;
        }
        let (rest, data) = slice.split_at(slice.len() - len);
        self.data = rest as *const [u8];
        Some(Self {
            data: data as *const [u8],
            owner: self.owner.clone(),
        })
    }

    /// Removes and returns the first byte of `self`.
    pub fn pop_front(&mut self) -> Option<u8> {
        let slice = unsafe { &*self.data };
        let (&b, rest) = slice.split_first()?;
        self.data = rest as *const [u8];
        Some(b)
    }

    /// Removes and returns the last byte of `self`.
    pub fn pop_back(&mut self) -> Option<u8> {
        let slice = unsafe { &*self.data };
        let (last, rest) = slice.split_last()?;
        self.data = rest as *const [u8];
        Some(*last)
    }

    /// Create a weak pointer.
    pub fn downgrade(&self) -> WeakBytes {
        WeakBytes {
            data: self.data,
            owner: Arc::downgrade(&self.owner),
        }
    }
}

impl WeakBytes {
    /// The reverse of `downgrade`. Returns `None` if the value was dropped.
    pub fn upgrade(&self) -> Option<Bytes> {
        let arc = self.owner.upgrade()?;
        Some(Bytes {
            data: self.data,
            owner: arc,
        })
    }
}

impl<T: ByteSource> From<T> for Bytes {
    fn from(value: T) -> Self {
        Self::from_source(value)
    }
}

/// Creates `Bytes` directly from an [`Arc`] containing the source.
///
/// This reuses the provided `Arc` as the owner so there is no extra
/// allocation. Common containers like `Arc<Vec<u8>>` therefore work
/// out of the box without needing a special wrapper type or an
/// additional `ByteSource` implementation.
impl<T: ByteSource + ByteOwner> From<Arc<T>> for Bytes {
    fn from(arc: Arc<T>) -> Self {
        Self::from_owning_source_arc(arc)
    }
}

#[cfg(feature = "bytes")]
impl From<Bytes> for bytes::Bytes {
    fn from(bytes: Bytes) -> Self {
        bytes::Bytes::from_owner(bytes)
    }
}

impl Deref for Bytes {
    type Target = [u8];
    #[inline]
    fn deref(&self) -> &Self::Target {
        self.as_slice()
    }
}

#[cfg(feature = "ownedbytes")]
unsafe impl ownedbytes::StableDeref for Bytes {}

impl Borrow<[u8]> for Bytes {
    fn borrow(&self) -> &[u8] {
        self
    }
}

impl AsRef<[u8]> for Bytes {
    #[inline]
    fn as_ref(&self) -> &[u8] {
        self
    }
}

impl hash::Hash for Bytes {
    fn hash<H: hash::Hasher>(&self, state: &mut H) {
        self.as_slice().hash(state);
    }
}

impl Default for Bytes {
    fn default() -> Self {
        Self::empty()
    }
}

impl PartialEq for Bytes {
    fn eq(&self, other: &Self) -> bool {
        self.as_slice() == other.as_slice()
    }
}

impl Eq for Bytes {}

impl PartialOrd for Bytes {
    fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
        self.as_slice().partial_cmp(other.as_slice())
    }
}

impl Ord for Bytes {
    fn cmp(&self, other: &Bytes) -> cmp::Ordering {
        self.as_slice().cmp(other.as_slice())
    }
}

impl fmt::Debug for Bytes {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        // Use `[u8]::escape_ascii` when inherent_ascii_escape is stabilized.
        f.write_str("b\"")?;
        for &byte in self.as_slice() {
            fmt::Display::fmt(&escape_default(byte), f)?;
        }
        f.write_str("\"")?;
        Ok(())
    }
}

#[cfg(feature = "bytes")]
impl bytes::Buf for Bytes {
    #[inline]
    fn remaining(&self) -> usize {
        self.as_slice().len()
    }

    #[inline]
    fn chunk(&self) -> &[u8] {
        self.as_slice()
    }

    #[inline]
    fn advance(&mut self, cnt: usize) {
        let slice = unsafe { &*self.data };
        if cnt > slice.len() {
            panic!("advance out of bounds: {} > {}", cnt, slice.len());
        }
        self.data = &slice[cnt..] as *const [u8];
    }
}

#[cfg(test)]
mod tests {
    use super::Bytes;

    #[test]
    fn niche_optimisation() {
        assert_eq!(size_of::<Bytes>(), size_of::<Option<Bytes>>());
    }
}

#[cfg(kani)]
mod verification {
    use super::*;
    use kani::BoundedArbitrary;

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_take_prefix_ok() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        kani::assume(data.len() >= 5);
        let mut bytes = Bytes::from_source(data.clone());
        let original = bytes.clone();
        let prefix = bytes.take_prefix(5).expect("prefix exists");
        assert_eq!(prefix.as_ref(), &original.as_ref()[..5]);
        assert_eq!(bytes.as_ref(), &original.as_ref()[5..]);
    }

    #[kani::proof]
    #[kani::unwind(32)]
    pub fn check_take_prefix_too_large() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let mut bytes = Bytes::from_source(data.clone());
        let copy = bytes.clone();
        let res = bytes.take_prefix(32);
        assert!(res.is_none());
        assert_eq!(bytes.as_ref(), copy.as_ref());
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_take_suffix_ok() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        kani::assume(data.len() >= 4);
        let mut bytes = Bytes::from_source(data.clone());
        let original = bytes.clone();
        let suffix = bytes.take_suffix(4).expect("suffix exists");
        assert_eq!(suffix.as_ref(), &original.as_ref()[original.len() - 4..]);
        assert_eq!(bytes.as_ref(), &original.as_ref()[..original.len() - 4]);
    }

    #[kani::proof]
    #[kani::unwind(32)]
    pub fn check_take_suffix_too_large() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let mut bytes = Bytes::from_source(data.clone());
        let copy = bytes.clone();
        let res = bytes.take_suffix(64);
        assert!(res.is_none());
        assert_eq!(bytes.as_ref(), copy.as_ref());
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_pop_front_behaviour() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let mut bytes = Bytes::from_source(data.clone());
        let snapshot = bytes.clone();

        if let Some((expected, remainder)) = data.split_first() {
            let popped = bytes.pop_front().expect("non-empty slice");
            assert_eq!(popped, *expected);
            assert_eq!(bytes.as_ref(), remainder);
        } else {
            assert!(bytes.pop_front().is_none());
            assert_eq!(bytes.as_ref(), snapshot.as_ref());
        }
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_pop_back_behaviour() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let mut bytes = Bytes::from_source(data.clone());
        let snapshot = bytes.clone();

        if let Some((expected, remainder)) = data.split_last() {
            let popped = bytes.pop_back().expect("non-empty slice");
            assert_eq!(popped, *expected);
            assert_eq!(bytes.as_ref(), remainder);
        } else {
            assert!(bytes.pop_back().is_none());
            assert_eq!(bytes.as_ref(), snapshot.as_ref());
        }
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_slice_to_bytes_ok() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        kani::assume(data.len() >= 8);
        let bytes = Bytes::from_source(data.clone());
        let slice = &bytes.as_ref()[3..8];
        let sub = bytes.slice_to_bytes(slice).expect("slice from same bytes");
        assert_eq!(sub.as_ref(), slice);
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_slice_to_bytes_unrelated() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let bytes = Bytes::from_source(data.clone());
        let other: [u8; 4] = kani::any();
        assert!(bytes.slice_to_bytes(&other).is_none());
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_try_unwrap_owner_unique() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let bytes = Bytes::from_source(data.clone());
        let recovered = bytes.try_unwrap_owner::<Vec<u8>>().expect("unwrap owner");
        assert_eq!(recovered, data);
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_try_unwrap_owner_shared() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let bytes = Bytes::from_source(data.clone());
        let _clone = bytes.clone();
        let result = bytes.try_unwrap_owner::<Vec<u8>>();
        assert!(result.is_err());
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_try_unwrap_owner_wrong_type() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let bytes = Bytes::from_source(data);
        assert!(bytes.try_unwrap_owner::<String>().is_err());
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_weakbytes_upgrade_some() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let bytes = Bytes::from_source(data.clone());
        let weak = bytes.downgrade();
        let upgraded = weak.upgrade().expect("upgrade");
        assert_eq!(upgraded.as_ref(), data.as_slice());
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_weakbytes_upgrade_none() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let bytes = Bytes::from_source(data);
        let weak = bytes.downgrade();
        drop(bytes);
        assert!(weak.upgrade().is_none());
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_downcast_to_owner_preserves_data() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        kani::assume(data.len() <= 16);

        let bytes = Bytes::from_source(data.clone());

        // Invariant: when the owner really is a Vec<u8>, downcasting the Bytes
        // owner should succeed and recover the same Arc<Vec<u8>> that backs the
        // original allocation.  This ensures Bytes keeps the concrete owner
        // type intact through cloning and slicing operations.
        let arc_vec = bytes
            .clone()
            .downcast_to_owner::<Vec<u8>>()
            .expect("downcast to Vec<u8>");
        assert_eq!(&*arc_vec, &data);

        // Invariant: attempting to downcast to the wrong owner type must fail
        // without mutating the Bytes value.  The returned Bytes should still
        // expose the same data slice so callers can continue using it.
        let result = bytes.downcast_to_owner::<String>();
        let Err(returned) = result else {
            panic!("downcast to String should fail");
        };
        assert_eq!(returned.as_ref(), data.as_slice());
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_is_subslice_accepts_ranges() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let bytes = Bytes::from_source(data.clone());

        let start: usize = kani::any();
        let end: usize = kani::any();
        kani::assume(start <= end);
        kani::assume(end <= data.len());

        let slice = &bytes.as_ref()[start..end];
        assert!(is_subslice(bytes.as_ref(), slice));
    }

    #[kani::proof]
    #[kani::unwind(16)]
    pub fn check_is_subslice_rejects_disjoint_ranges() {
        let data: Vec<u8> = Vec::bounded_any::<16>();
        let other: Vec<u8> = Vec::bounded_any::<8>();

        let base_len = data.len();
        kani::assume(base_len > 0);
        let base_ptr = data.as_ptr();

        let other_len = other.len();
        kani::assume(other_len > 0);
        let other_ptr = other.as_ptr();

        let base_start = base_ptr as usize;
        let base_end = base_start + base_len;
        let other_start = other_ptr as usize;
        let other_end = other_start + other_len;
        kani::assume(other_end <= base_start || other_start >= base_end);

        let bytes = Bytes::from_source(data);
        assert!(!is_subslice(bytes.as_ref(), other.as_slice()));
    }
}