//! This crate defines tools to implement datastructures that can live
//! in main memory or on-disk, meaning that their natural habitat is
//! memory-mapped files, but if that environment is threatened, they
//! might seek refuge in lower-level environments.
//!
//! One core building block of this library is the notion of virtual
//! memory pages, which are allocated and freed by an
//! externally-provided allocator (see how the `sanakirja` crate does
//! this). The particular implementation used here is meant to allow a
//! transactional system with readers reading the structures
//! concurrently with one writer at a time.
//!
//! At the moment, only B trees are implemented, as well as the
//! following general traits:
//!
//! - [`LoadPage`] is a trait used to get a pointer to a page. In the
//! most basic version, this may just return a pointer to the file,
//! offset by the requested offset. In more sophisticated versions,
//! this can be used to encrypt and compress pages.
//! - [`AllocPage`] allocates and frees pages, because as
//! datastructures need to be persisted on disk, we can't rely on
//! Rust's memory management to do it for us. Users of this crate
//! don't have to worry about this though.
//!
//! Moreover, two other traits can be used to store things on pages:
//! [`Storable`] is a simple trait that all `Sized + Ord` types
//! without references can readily implement (the [`direct_repr!`]
//! macro does that). For types containing references to pages
//! allocated in the database, the comparison function can be
//! customised. Moreover, these types must supply an iterator over
//! these references, in order for reference-counting to work properly
//! when the datastructures referencing these types are forked.
//!
//! Dynamically-sized types, or types that need to be represented in a
//! dynamically-sized way, can use the [`UnsizedStorable`] format.
use async_trait::async_trait;
pub mod btree;
/// There's a hard-coded assumption that pages have 4K bytes. This is
/// true for normal memory pages on almost all platforms.
pub const PAGE_SIZE: usize = 4096;
/// Types that can be stored on disk. This trait may be used in
/// conjunction with `Sized` in order to determine the on-disk size,
/// or with [`UnsizedStorable`] when special arrangements are needed.
#[async_trait(?Send)]
pub trait Storable: Send + Sync + core::fmt::Debug {
    /// This is required for B trees, not necessarily for other
    /// structures. The default implementation panics.
    fn compare(&self, _b: &Self) -> core::cmp::Ordering {
        unimplemented!()
    }
    /// If this value is an offset to another page at offset `offset`,
    /// return `Some(offset)`. Return `None` else.
    fn page_references(&self) -> Self::PageReferences;
    /// If this value is an offset to another page at offset `offset`,
    /// return `Some(offset)`. Return `None` else.
    async fn drop<T: AllocPage>(&self, txn: &mut T) -> Result<(), T::Error> {
        for p in self.page_references() {
            txn.decr_rc(p).await?;
        }
        Ok(())
    }
    /// An iterator over the offsets to pages contained in this
    /// value. Only values from this crate can generate non-empty
    /// iterators, but combined values (like tuples) must chain the
    /// iterators returned by method `page_offsets`.
    type PageReferences: Iterator<Item = u64> + Send;
}
/// A macro to implement [`Storable`] on "plain" types,
/// i.e. fixed-sized types that are `repr(C)` and don't hold
/// references.
#[macro_export]
macro_rules! direct_repr {
    ($t: ty) => {
        impl Storable for $t {
            type PageReferences = core::iter::Empty<u64>;
            fn page_references(&self) -> Self::PageReferences {
                core::iter::empty()
            }
            fn compare(&self, b: &Self) -> core::cmp::Ordering {
                self.cmp(b)
            }
        }
        impl UnsizedStorable for $t {
            const ALIGN: usize = core::mem::align_of::<$t>();
            /// If `Self::SIZE.is_some()`, this must return the same
            /// value. The default implementation is `Self;:SIZE.unwrap()`.
            fn size(&self) -> usize {
                core::mem::size_of::<Self>()
            }
            /// Read the size from an on-page entry.
            unsafe fn onpage_size(_: *const u8) -> usize {
                core::mem::size_of::<Self>()
            }
            /// Write to a page. Must not overwrite the allocated size, but
            /// this isn't checked (which is why it's unsafe).
            unsafe fn write_to_page(&self, p: *mut u8) {
                core::ptr::copy_nonoverlapping(self, p as *mut Self, 1)
            }
            unsafe fn from_raw_ptr<'a>(p: *const u8) -> &'a Self {
                &*(p as *const Self)
            }
        }
    };
}
direct_repr!(());
direct_repr!(u8);
direct_repr!(i8);
direct_repr!(u16);
direct_repr!(i16);
direct_repr!(u32);
direct_repr!(i32);
direct_repr!(u64);
direct_repr!(i64);
direct_repr!([u8; 16]);
#[cfg(feature = "std")]
extern crate std;
#[cfg(feature = "std")]
direct_repr!(std::net::Ipv4Addr);
#[cfg(feature = "std")]
direct_repr!(std::net::Ipv6Addr);
#[cfg(feature = "std")]
direct_repr!(std::net::IpAddr);
#[cfg(feature = "std")]
direct_repr!(std::net::SocketAddr);
#[cfg(feature = "std")]
direct_repr!(std::time::SystemTime);
#[cfg(feature = "std")]
direct_repr!(std::time::Duration);
#[cfg(feature = "uuid")]
direct_repr!(uuid::Uuid);
#[cfg(feature = "ed25519")]
direct_repr!(ed25519_zebra::VerificationKeyBytes);
/// Types that can be stored on disk.
pub trait UnsizedStorable: Storable {
    const ALIGN: usize;
    /// If `Self::SIZE.is_some()`, this must return the same
    /// value. The default implementation is `Self;:SIZE.unwrap()`.
    fn size(&self) -> usize;
    /// Read the size from an on-page entry. If `Self::SIZE.is_some()`
    /// this must be the same value.
    unsafe fn onpage_size(_: *const u8) -> usize;
    /// Write to a page. Must not overwrite the allocated size, but
    /// this isn't checked (which is why it's unsafe).
    unsafe fn write_to_page(&self, p: *mut u8);
    unsafe fn from_raw_ptr<'a>(p: *const u8) -> &'a Self;
}
impl Storable for [u8] {
    type PageReferences = core::iter::Empty<u64>;
    fn page_references(&self) -> Self::PageReferences {
        core::iter::empty()
    }
    fn compare(&self, b: &Self) -> core::cmp::Ordering {
        self.cmp(b)
    }
}
impl UnsizedStorable for [u8] {
    const ALIGN: usize = 2;
    fn size(&self) -> usize {
        2 + self.len()
    }
    unsafe fn from_raw_ptr<'a>(p: *const u8) -> &'a Self {
        let len = u16::from_le(*(p as *const u16));
        assert_ne!(len, 0);
        assert_eq!(len & 0xf000, 0);
        core::slice::from_raw_parts(p.add(2), len as usize)
    }
    unsafe fn onpage_size(p: *const u8) -> usize {
        let len = u16::from_le(*(p as *const u16));
        2 + len as usize
    }
    unsafe fn write_to_page(&self, p: *mut u8) {
        assert!(self.len() <= 510);
        *(p as *mut u16) = (self.len() as u16).to_le();
        core::ptr::copy_nonoverlapping(self.as_ptr(), p.add(2), self.len())
    }
}
unsafe fn read<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
    k: *const u8,
) -> (*const u8, *const u8) {
    let s = K::onpage_size(k);
    let v = k.add(s);
    let al = v.align_offset(V::ALIGN);
    let v = v.add(al);
    (k, v)
}
unsafe fn entry_size<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
    k: *const u8,
) -> usize {
    assert_eq!(k.align_offset(K::ALIGN), 0);
    let ks = K::onpage_size(k);
    // next multiple of va, assuming va is a power of 2.
    let v_off = (ks + V::ALIGN - 1) & !(V::ALIGN - 1);
    let v_ptr = k.add(v_off);
    let vs = V::onpage_size(v_ptr);
    let ka = K::ALIGN.max(V::ALIGN);
    let size = v_off + vs;
    (size + ka - 1) & !(ka - 1)
}
/// Representation of a mutable or shared page. This is an owned page
/// (like `Vec` in Rust's std), but we do not know whether we can
/// mutate it or not.
///
/// The least-significant bit of the first byte of each page is 1 if
/// and only if the page was allocated by the current transaction (and
/// hence isn't visible to any other transaction, meaning we can write
/// on it).
#[derive(Debug)]
#[repr(C)]
pub struct CowPage {
    pub data: *mut u8,
    pub offset: u64,
}
/// Representation of a borrowed, or immutable page, like a slice in
/// Rust.
#[derive(Debug, Clone, Copy)]
#[repr(C)]
pub struct Page<'a> {
    pub data: &'a [u8; PAGE_SIZE],
    pub offset: u64,
}
impl CowPage {
    /// Borrows the page.
    pub fn as_page(&self) -> Page {
        Page {
            data: unsafe { &*(self.data as *const [u8; PAGE_SIZE]) },
            offset: self.offset,
        }
    }
    #[cfg(feature = "crc32")]
    pub fn crc(&self, hasher: &crc32fast::Hasher) -> u32 {
        let mut hasher = hasher.clone();
        hasher.reset();
        // Hash the beginning and the end of the page (i.e. remove
        // the CRC).
        unsafe {
            // Remove the dirty bit.
            let x = [(*self.data) & 0xfe];
            hasher.update(&x[..]);
            hasher.update(core::slice::from_raw_parts(self.data.add(1), 3));
            hasher.update(core::slice::from_raw_parts(
                self.data.add(8),
                PAGE_SIZE - 8,
            ));
        }
        hasher.finalize()
    }
    #[cfg(feature = "crc32")]
    pub fn crc_check(&self, hasher: &crc32fast::Hasher) -> bool {
        let crc = unsafe { u32::from_le(*(self.data as *const u32).add(1)) };
        self.crc(hasher) == crc
    }
}
/// An owned page on which we can write. This is just a wrapper around
/// `CowPage` to avoid checking the "dirty" bit at runtime.
#[derive(Debug)]
pub struct MutPage(pub CowPage);
#[cfg(not(feature = "crc32"))]
pub fn clear_dirty(data: &mut [u8]) {
    data[0] &= 0xfe
}
#[cfg(feature = "crc32")]
pub fn clear_dirty(data: &mut [u8], hasher: &crc32fast::Hasher) {
    unsafe {
        data[0] &= 0xfe;
        let crc = (self.0.data.as_mut_ptr() as *mut u32).add(1);
        *crc = self.0.crc(hasher)
    }
}
impl MutPage {
    #[cfg(not(feature = "crc32"))]
    pub fn clear_dirty(&mut self) {
        unsafe { *self.0.data &= 0xfe }
    }
    #[cfg(feature = "crc32")]
    pub fn clear_dirty(&mut self, hasher: &crc32fast::Hasher) {
        unsafe {
            *self.0.data &= 0xfe;
            let crc = (self.0.data as *mut u32).add(1);
            *crc = self.0.crc(hasher)
        }
    }
}
unsafe impl Sync for CowPage {}
unsafe impl Send for CowPage {}
impl CowPage {
    /// Checks the dirty bit of a page.
    pub fn is_dirty(&self) -> bool {
        unsafe { (*self.data) & 1 != 0 }
    }
}
/// Trait for loading a page.
#[async_trait(?Send)]
pub trait LoadPage: Send + Sync {
    type Error;
    /// Loading a page.
    async fn load_page(&self, off: u64) -> Result<CowPage, Self::Error>;
    /// Reference-counting. Since reference-counts are designed to be
    /// storable into B trees by external allocators, pages referenced
    /// once aren't stored, and hence are indistinguishable from pages
    /// that are never referenced. The default implementation returns
    /// 0.
    ///
    /// This has the extra benefit of requiring less disk space, and
    /// isn't more unsafe than storing the reference count, since we
    /// aren't supposed to hold a reference to a page with "logical
    /// RC" 0, so storing "1" for that page would be redundant anyway.
    async fn rc(&self, _off: u64) -> Result<u64, Self::Error> {
        Ok(0)
    }
}
/// Trait for allocating and freeing pages.
#[async_trait(?Send)]
pub trait AllocPage: LoadPage {
    /// Allocate a new page.
    async fn alloc_page(&mut self) -> Result<MutPage, Self::Error>;
    /// Increment the page's reference count.
    async fn incr_rc(&mut self, off: u64) -> Result<usize, Self::Error>;
    /// Decrement the page's reference count, assuming the page was
    /// first allocated by another transaction. If the RC reaches 0,
    /// free the page. Must return the new RC (0 if freed).
    async fn decr_rc(&mut self, off: u64) -> Result<usize, Self::Error>;
    /// Same as [`Self::decr_rc`], but for pages allocated by the current
    /// transaction. This is an important distinction, as pages
    /// allocated by the current transaction can be reused immediately
    /// after being freed.
    async fn decr_rc_owned(&mut self, off: u64) -> Result<usize, Self::Error>;
}

//! Insertions into a B tree.
use super::cursor::*;
use super::*;
/// The representation of the update to a page.
#[derive(Debug)]
pub struct Ok {
    /// The new page, possibly the same as the previous version.
    pub page: MutPage,
    /// The freed page, or 0 if no page was freed.
    pub freed: u64,
}
/// The result of an insertion, i.e. either an update or a split.
#[derive(Debug)]
pub enum Put<'a, K: ?Sized, V: ?Sized> {
    Ok(Ok),
    Split {
        split_key: &'a K,
        split_value: &'a V,
        left: MutPage,
        right: MutPage,
        freed: u64,
    },
}
/// Insert an entry into a database, returning `false` if and only if
/// the exact same entry (key *and* value) was already in the
/// database.
pub async fn put_async<
    T: AsyncAllocPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreeMutPage<K, V>,
>(
    txn: &mut T,
    db: &mut Db_<K, V, P>,
    key: &K,
    value: &V,
) -> Result<bool, T::Error> {
    // Set the cursor to the first element larger than or equal to
    // `(key, value)`. We will insert at that position (where "insert"
    // is meant in the sense of Rust's `Vec::insert`.
    let mut cursor = Cursor::async_new(txn, db).await?;
    if cursor.async_set(txn, key, Some(value)).await?.is_some() {
        // If `(key, value)` already exists in the tree, return.
        return Ok(false);
    }
    // Else, we are at a leaf, since cursors that don't find an
    // element go all the way to the leaves.
    //
    // We put on the leaf page, resulting in a `Put`, i.e. either
    // `Put::Ok` or `Put::Split`.
    let p = cursor.len(); // Save the position of the leaf cursor.
    let is_owned = p < cursor.first_rc_len();
    // Insert the key and value at the leaf, i.e. pop the top level of
    // the stack (the leaf) and insert in there.
    let cur = cursor.pop().unwrap();
    let put = P::put(
        txn,
        cur.page,
        is_owned,
        false,
        &cur.cursor,
        key,
        value,
        None,
        0,
        0,
    )?;
    // In both cases (`Ok` and `Split`), we need to walk up the tree
    // and update each node.
    // Moreover, since the middle elements of the splits may be on
    // pages that must be freed at the end of this call, we collect
    // them in the `free` array below, and free them only at the end
    // of this function.
    //
    // If we didn't hold on to these pages, they could be reallocated
    // in subsequent splits, and the middle element could be
    // lost. This is important when the middle element climbs up more
    // than one level (i.e. the middle element of the split of a leaf
    // is also the middle element of the split of the leaf's parent,
    // etc).
    let mut free = [0; N_CURSORS];
    db.db = put_cascade(txn, &mut cursor, put, &mut free).await?.0.offset;
    for f in &free[..p] {
        if *f & 1 != 0 {
            txn.decr_rc_owned((*f) ^ 1)?;
        } else if *f > 0 {
            txn.decr_rc(*f)?;
        }
    }
    Ok(true)
}
async fn put_cascade<
        'a,
    T: AsyncAllocPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreeMutPage<K, V>,
>(
    txn: &mut T,
    cursor: &mut Cursor<K, V, P>,
    mut put: Put<'a, K, V>,
    free: &mut [u64; N_CURSORS],
) -> Result<MutPage, T::Error> {
    loop {
        match put {
            Put::Split {
                split_key,
                split_value,
                left,
                right,
                freed,
            } => {
                // Here, we are copying all children of the freed
                // page, except possibly the last freed page (which is
                // a child of the current node, if we aren't at a
                // leaf). This includes the split key/value, also
                // incremented in the following call:
                incr_descendants::<T, K, V, P>(txn, cursor, free, freed)?;
                // Then, insert the split key/value in the relevant page:
                let is_owned = cursor.len() < cursor.first_rc_len();
                if let Some(cur) = cursor.pop() {
                    // In this case, there's a page above the page
                    // that just split (since we can pop the stack),
                    // so the page that just split isn't the root (but
                    // `cur` might be).
                    put = P::put(
                        txn,
                        cur.page,
                        is_owned,
                        false,
                        &cur.cursor,
                        split_key,
                        split_value,
                        None,
                        left.0.offset,
                        right.0.offset,
                    )?;
                } else {
                    // No page above the split, so the root has just
                    // split. Insert the split key/value into a new
                    // page above the entire tree.
                    let mut p = txn.alloc_page()?;
                    let cursor = P::cursor_first(&p.0);
                    P::init(&mut p);
                    if let Put::Ok(p) = P::put(
                        txn,
                        p.0,
                        true,
                        false,
                        &cursor,
                        split_key,
                        split_value,
                        None,
                        left.0.offset,
                        right.0.offset,
                    )? {
                        // Return the new root.
                        return Ok(p.page);
                    } else {
                        unreachable!()
                    }
                }
            }
            Put::Ok(Ok { page, freed }) => {
                // Same as above: increment the relevant reference
                // counters.
                incr_descendants::<T, K, V, P>(txn, cursor, free, freed)?;
                // And update the left child of the current cursor,
                // since the main invariant of cursors is that we're
                // always visiting the left child (if we're visiting
                // the last child of a page, the cursor is set to the
                // position strictly after the entries).
                let is_owned = cursor.len() < cursor.first_rc_len();
                if let Some(curs) = cursor.pop() {
                    // If we aren't at the root, just update the child.
                    put = Put::Ok(P::update_left_child(
                        txn,
                        curs.page,
                        is_owned,
                        &curs.cursor,
                        page.0.offset,
                    )?)
                } else {
                    // If we are at the root, `page` is the updated root.
                    return Ok(page);
                }
            }
        }
    }
}
/// This function does all the memory management for `put`.
fn incr_descendants<
    T: AllocPage + LoadPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreePage<K, V>,
>(
    txn: &mut T,
    cursor: &mut Cursor<K, V, P>,
    free: &mut [u64; N_CURSORS],
    mut freed: u64,
) -> Result<(), T::Error> {
    // The freed page is on the page below.
    if cursor.len() < cursor.first_rc_len() {
        // If a page has split or was immutable (allocated by a
        // previous transaction) and has been updated, we need to free
        // the old page.
        free[cursor.len()] = freed;
    } else {
        // This means that the new version of the page (either split
        // or not) contains references to all the children of the
        // freed page, except the last freed page (because the new
        // version of that page isn't shared).
        let cur = cursor.cur();
        // Start a cursor at the leftmost position and increase the
        // leftmost child page's RC (if we aren't at a leaf, and if
        // that child isn't the last freed page).
        let mut c = P::cursor_first(&cur.page);
        let left = P::left_child(cur.page.as_page(), &c);
        if left != 0 {
            if left != (freed & !1) {
                txn.incr_rc(left)?;
            } else if cursor.len() == cursor.first_rc_len() {
                freed = 0
            }
        }
        // Then, for each entry of the page, increase the RCs of
        // references contained in the keys and values, and increase
        // the RC of the right child.
        while let Some((k, v, r)) = P::next(txn, cur.page.as_page(), &mut c) {
            for o in k.page_references().chain(v.page_references()) {
                txn.incr_rc(o)?;
            }
            if r != 0 {
                if r != (freed & !1) {
                    txn.incr_rc(r)?;
                } else if cursor.len() == cursor.first_rc_len() {
                    freed = 0
                }
            }
        }
        // Else, the "freed" page is shared with another tree, and
        // hence we just need to decrement its RC.
        if freed > 0 && cursor.len() == cursor.first_rc_len() {
            free[cursor.len()] = freed;
        }
    }
    Ok(())
}

//! Insertions into a B tree.
use super::cursor::*;
use super::*;
/// The representation of the update to a page.
#[derive(Debug)]
pub struct Ok {
    /// The new page, possibly the same as the previous version.
    pub page: MutPage,
    /// The freed page, or 0 if no page was freed.
    pub freed: u64,
}
/// The result of an insertion, i.e. either an update or a split.
#[derive(Debug)]
pub enum Put<'a, K: ?Sized, V: ?Sized> {
    Ok(Ok),
    Split {
        split_key: &'a K,
        split_value: &'a V,
        left: MutPage,
        right: MutPage,
        freed: u64,
    },
}
/// Insert an entry into a database, returning `false` if and only if
/// the exact same entry (key *and* value) was already in the
/// database.
pub async fn put<
    T: AllocPage,
    K: Storable + ?Sized + Sync,
    V: Storable + ?Sized + Sync,
    P: BTreeMutPage<K, V>,
>(
    txn: &mut T,
    db: &mut Db_<K, V, P>,
    key: &K,
    value: &V,
) -> Result<bool, T::Error> {
    // Set the cursor to the first element larger than or equal to
    // `(key, value)`. We will insert at that position (where "insert"
    // is meant in the sense of Rust's `Vec::insert`.
    let mut cursor = Cursor::new(txn, db).await?;
    if cursor.set(txn, key, Some(value)).await?.is_some() {
        // If `(key, value)` already exists in the tree, return.
        return Ok(false);
    }
    // Else, we are at a leaf, since cursors that don't find an
    // element go all the way to the leaves.
    //
    // We put on the leaf page, resulting in a `Put`, i.e. either
    // `Put::Ok` or `Put::Split`.
    let p = cursor.len(); // Save the position of the leaf cursor.
    let is_owned = p < cursor.first_rc_len();
    // Insert the key and value at the leaf, i.e. pop the top level of
    // the stack (the leaf) and insert in there.
    let cur = cursor.pop().unwrap();
    let put = P::put(
        txn,
        cur.page,
        is_owned,
        false,
        &cur.cursor,
        key,
        value,
        None,
        0,
        0,
    ).await?;
    // In both cases (`Ok` and `Split`), we need to walk up the tree
    // and update each node.
    // Moreover, since the middle elements of the splits may be on
    // pages that must be freed at the end of this call, we collect
    // them in the `free` array below, and free them only at the end
    // of this function.
    //
    // If we didn't hold on to these pages, they could be reallocated
    // in subsequent splits, and the middle element could be
    // lost. This is important when the middle element climbs up more
    // than one level (i.e. the middle element of the split of a leaf
    // is also the middle element of the split of the leaf's parent,
    // etc).
    let mut free = [0; N_CURSORS];
    db.db = put_cascade(txn, &mut cursor, put, &mut free).await?.0.offset;
    for f in &free[..p] {
        if *f & 1 != 0 {
            txn.decr_rc_owned((*f) ^ 1).await?;
        } else if *f > 0 {
            txn.decr_rc(*f).await?;
        }
    }
    Ok(true)
}
async fn put_cascade<
        'a,
    T: AllocPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreeMutPage<K, V>,
>(
    txn: &mut T,
    cursor: &mut Cursor<K, V, P>,
    mut put: Put<'a, K, V>,
    free: &mut [u64; N_CURSORS],
) -> Result<MutPage, T::Error> {
    loop {
        match put {
            Put::Split {
                split_key,
                split_value,
                left,
                right,
                freed,
            } => {
                // Here, we are copying all children of the freed
                // page, except possibly the last freed page (which is
                // a child of the current node, if we aren't at a
                // leaf). This includes the split key/value, also
                // incremented in the following call:
                incr_descendants::<T, K, V, P>(txn, cursor, free, freed).await?;
                // Then, insert the split key/value in the relevant page:
                let is_owned = cursor.len() < cursor.first_rc_len();
                if let Some(cur) = cursor.pop() {
                    // In this case, there's a page above the page
                    // that just split (since we can pop the stack),
                    // so the page that just split isn't the root (but
                    // `cur` might be).
                    put = P::put(
                        txn,
                        cur.page,
                        is_owned,
                        false,
                        &cur.cursor,
                        split_key,
                        split_value,
                        None,
                        left.0.offset,
                        right.0.offset,
                    ).await?;
                } else {
                    // No page above the split, so the root has just
                    // split. Insert the split key/value into a new
                    // page above the entire tree.
                    let mut p = txn.alloc_page().await?;
                    let cursor = P::cursor_first(&p.0);
                    P::init(&mut p);
                    if let Put::Ok(p) = P::put(
                        txn,
                        p.0,
                        true,
                        false,
                        &cursor,
                        split_key,
                        split_value,
                        None,
                        left.0.offset,
                        right.0.offset,
                    ).await? {
                        // Return the new root.
                        return Ok(p.page);
                    } else {
                        unreachable!()
                    }
                }
            }
            Put::Ok(Ok { page, freed }) => {
                // Same as above: increment the relevant reference
                // counters.
                incr_descendants::<T, K, V, P>(txn, cursor, free, freed).await?;
                // And update the left child of the current cursor,
                // since the main invariant of cursors is that we're
                // always visiting the left child (if we're visiting
                // the last child of a page, the cursor is set to the
                // position strictly after the entries).
                let is_owned = cursor.len() < cursor.first_rc_len();
                if let Some(curs) = cursor.pop() {
                    // If we aren't at the root, just update the child.
                    put = Put::Ok(P::update_left_child(
                        txn,
                        curs.page,
                        is_owned,
                        &curs.cursor,
                        page.0.offset,
                    ).await?)
                } else {
                    // If we are at the root, `page` is the updated root.
                    return Ok(page);
                }
            }
        }
    }
}
/// This function does all the memory management for `put`.
async fn incr_descendants<
    T: AllocPage + LoadPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreePage<K, V>,
>(
    txn: &mut T,
    cursor: &mut Cursor<K, V, P>,
    free: &mut [u64; N_CURSORS],
    mut freed: u64,
) -> Result<(), T::Error> {
    // The freed page is on the page below.
    if cursor.len() < cursor.first_rc_len() {
        // If a page has split or was immutable (allocated by a
        // previous transaction) and has been updated, we need to free
        // the old page.
        free[cursor.len()] = freed;
    } else {
        // This means that the new version of the page (either split
        // or not) contains references to all the children of the
        // freed page, except the last freed page (because the new
        // version of that page isn't shared).
        let cur = cursor.cur();
        // Start a cursor at the leftmost position and increase the
        // leftmost child page's RC (if we aren't at a leaf, and if
        // that child isn't the last freed page).
        let mut c = P::cursor_first(&cur.page);
        let left = P::left_child(cur.page.as_page(), &c);
        if left != 0 {
            if left != (freed & !1) {
                txn.incr_rc(left).await?;
            } else if cursor.len() == cursor.first_rc_len() {
                freed = 0
            }
        }
        // Then, for each entry of the page, increase the RCs of
        // references contained in the keys and values, and increase
        // the RC of the right child.
        while let Some((k, v, r)) = P::next(cur.page.as_page(), &mut c) {
            for o in k.page_references().chain(v.page_references()) {
                txn.incr_rc(o).await?;
            }
            if r != 0 {
                if r != (freed & !1) {
                    txn.incr_rc(r).await?;
                } else if cursor.len() == cursor.first_rc_len() {
                    freed = 0
                }
            }
        }
        // Else, the "freed" page is shared with another tree, and
        // hence we just need to decrement its RC.
        if freed > 0 && cursor.len() == cursor.first_rc_len() {
            free[cursor.len()] = freed;
        }
    }
    Ok(())
}

//! Implementation of B tree pages for Unsized types, or types with an
//! dynamically-sized representation (for example enums with widely
//! different sizes).
//!
//! This module follows the same organisation as the sized
//! implementation, and contains types shared between the two
//! implementations.
//!
//! The types that can be used with this implementation must implement
//! the [`UnsizedStorable`] trait, which essentially replaces the
//! [`core::mem`] functions for determining the size and alignment of
//! values.
//!
//! One key difference is the implementation of leaves (internal nodes
//! have the same format): indeed, in this implementation, leaves have
//! almost the same format as internal nodes, except that their
//! offsets are written on the page as little-endian-encoded [`u16`]
//! (with only the 12 LSBs used, i.e. 4 bits unused).
use super::*;
use crate::btree::del::*;
use crate::btree::put::*;
use core::cmp::Ordering;
// The header is the same as for the sized implementation, so we share
// it here.
pub(super) mod header;
// Like in the sized implementation, we have the same three submodules.
mod alloc;
// This is a common module with the sized implementation.
pub(super) mod cursor;
mod put;
pub(super) mod rebalance;
use alloc::*;
use cursor::*;
use header::*;
use rebalance::*;
#[derive(Debug)]
pub struct Page<K: ?Sized, V: ?Sized> {
    k: core::marker::PhantomData<K>,
    v: core::marker::PhantomData<V>,
}
// Many of these functions are the same as in the Sized implementations.
#[async_trait(?Send)]
impl<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized> super::BTreePage<K, V>
    for Page<K, V>
{
    fn is_empty(c: &Self::Cursor) -> bool {
        c.cur >= c.total as isize
    }
    fn is_init(c: &Self::Cursor) -> bool {
        c.cur < 0
    }
    type Cursor = PageCursor;
    fn cursor_before(p: &crate::CowPage) -> Self::Cursor {
        PageCursor::new(p, -1)
    }
    fn cursor_after(p: &crate::CowPage) -> Self::Cursor {
        PageCursor::after(p)
    }
    // Split a cursor, returning two cursors `(a, b)` where b is the
    // same as `c`, and `a` is a cursor running from the first element
    // of the page to `c` (excluding `c`).
    fn split_at(c: &Self::Cursor) -> (Self::Cursor, Self::Cursor) {
        (
            PageCursor {
                cur: 0,
                total: c.cur.max(0) as usize,
                is_leaf: c.is_leaf,
            },
            *c,
        )
    }
    fn move_next(c: &mut Self::Cursor) -> bool {
        if c.cur < c.total as isize {
            c.cur += 1;
            true
        } else {
            false
        }
    }
    fn move_prev(c: &mut Self::Cursor) -> bool {
        if c.cur > 0 {
            c.cur -= 1;
            true
        } else {
            c.cur = -1;
            false
        }
    }
    // This function is the same as the sized implementation for
    // internal nodes, since the only difference between leaves and
    // internal nodes in this implementation is the size of offsets (2
    // bytes for leaves, 8 bytes for internal nodes).
    fn current<'a>(
        page: crate::Page<'a>,
        c: &Self::Cursor,
    ) -> Option<(&'a K, &'a V, u64)> {
        if c.cur < 0 || c.cur >= c.total as isize {
            None
        } else if c.is_leaf {
            unsafe {
                let off =
                    u16::from_le(*(page.data.as_ptr().add(HDR + c.cur as usize * 2) as *const u16));
                let (k, v) = read::<K, V>(page.data.as_ptr().add(off as usize));
                Some((
                    K::from_raw_ptr(k as *const u8),
                    V::from_raw_ptr(v as *const u8),
                    0,
                ))
            }
        } else {
            unsafe {
                let off =
                    u64::from_le(*(page.data.as_ptr().add(HDR + c.cur as usize * 8) as *const u64));
                let (k, v) = read::<K, V>(page.data.as_ptr().add((off & 0xfff) as usize));
                Some((
                    K::from_raw_ptr(k as *const u8),
                    V::from_raw_ptr(v as *const u8),
                    off & !0xfff,
                ))
            }
        }
    }
    // These methods are the same as in the sized implementation.
    fn left_child(page: crate::Page, c: &Self::Cursor) -> u64 {
        assert!(c.cur >= 0);
        if c.is_leaf {
            0
        } else {
            assert!(c.cur >= 0 && HDR as isize + c.cur * 8 - 8 <= 4088);
            let off =
                unsafe { *(page.data.as_ptr().offset(HDR as isize + c.cur * 8 - 8) as *const u64) };
            u64::from_le(off) & !0xfff
        }
    }
    fn right_child(page: crate::Page, c: &Self::Cursor) -> u64 {
        assert!(c.cur < c.total as isize);
        if c.is_leaf {
            0
        } else {
            assert!(c.cur < c.total as isize && HDR as isize + c.cur * 8 - 8 <= 4088);
            let off = unsafe { *(page.data.as_ptr().offset(HDR as isize + c.cur * 8) as *const u64) };
            u64::from_le(off) & !0xfff
        }
    }
    async fn set_cursor<'a, 'b, T: LoadPage>(
        _txn: &'a T,
        page: crate::Page<'b>,
        c: &mut PageCursor,
        k0: &K,
        v0: Option<&V>,
    ) -> Result<(&'a K, &'a V, u64), usize> {
        unsafe {
            // `lookup` has the same semantic as
            // `core::slice::binary_search`, i.e. it returns either
            // `Ok(n)`, where `n` is the position where `(k0, v0)` was
            // found, or `Err(n)` where `n` is the position where
            // `(k0, v0)` can be inserted to preserve the order.
            match lookup(page, c, k0, v0).await {
                Ok(n) => {
                    c.cur = n as isize;
                    if c.is_leaf {
                        let off =
                            u16::from_le(*(page.data.as_ptr().add(HDR + n * 2) as *const u16));
                        let (k, v) = read::<K, V>(page.data.as_ptr().add(off as usize));
                        Ok((K::from_raw_ptr(k), V::from_raw_ptr(v), 0))
                    } else {
                        let off =
                            u64::from_le(*(page.data.as_ptr().add(HDR + n * 8) as *const u64));
                        let (k, v) =
                            read::<K, V>(page.data.as_ptr().add(off as usize & 0xfff));
                        Ok((
                            K::from_raw_ptr(k),
                            V::from_raw_ptr(v),
                            off & !0xfff,
                        ))
                    }
                }
                Err(n) => {
                    c.cur = n as isize;
                    Err(n)
                }
            }
        }
    }
}
// There quite some duplicated code in the following function, because
// we're forming a slice of offsets, and the using the core library's
// `binary_search_by` method on slices.
async unsafe fn lookup<'a, K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
    page: crate::Page<'a>,
    c: &mut PageCursor,
    k0: &K,
    v0: Option<&V>,
) -> Result<usize, usize> {
    let hdr = header(page);
    c.total = hdr.n() as usize;
    c.is_leaf = hdr.is_leaf();
    if c.is_leaf {
        lookup_leaf(page, k0, v0, hdr)
    } else {
        lookup_internal(page, k0, v0, hdr)
    }
}
unsafe fn lookup_internal<'a, K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
    page: crate::Page<'a>,
    k0: &K,
    v0: Option<&V>,
    hdr: &header::Header,
) -> Result<usize, usize> {
    let s = core::slice::from_raw_parts(
        page.data.as_ptr().add(HDR) as *const u64,
        hdr.n() as usize,
    );
    if let Some(v0) = v0 {
        s.binary_search_by(|&off| {
            let off = u64::from_le(off) & 0xfff;
            let (k, v) = read::<K, V>(page.data.as_ptr().offset(off as isize & 0xfff));
            let k = K::from_raw_ptr(k);
            match k.compare(k0) {
                Ordering::Equal => {
                    let v = V::from_raw_ptr(v);
                    v.compare(v0)
                }
                cmp => cmp,
            }
        })
    } else {
        match s.binary_search_by(|&off| {
            let off = u64::from_le(off) & 0xfff;
            let (k, _) = read::<K, V>(page.data.as_ptr().offset(off as isize & 0xfff));
            let k = K::from_raw_ptr(k);
            k.compare(k0)
        }) {
            Err(i) => Err(i),
            Ok(mut i) => {
                // Rewind if there are multiple matching keys.
                while i > 0 {
                    let off = u64::from_le(s[i-1]) & 0xfff;
                    let (k, _) = read::<K, V>(page.data.as_ptr().offset(off as isize));
                    let k = K::from_raw_ptr(k);
                    if let Ordering::Equal = k.compare(k0) {
                        i -= 1
                    } else {
                        break
                    }
                }
                Ok(i)
            }
        }
    }
}
unsafe fn lookup_leaf<'a, K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
    page: crate::Page<'a>,
    k0: &K,
    v0: Option<&V>,
    hdr: &header::Header,
) -> Result<usize, usize> {
    let s = core::slice::from_raw_parts(
        page.data.as_ptr().add(HDR) as *const u16,
        hdr.n() as usize,
    );
    if let Some(v0) = v0 {
        s.binary_search_by(|&off| {
            let off = u16::from_le(off);
            let (k, v) = read::<K, V>(page.data.as_ptr().offset(off as isize));
            let k = K::from_raw_ptr(k as *const u8);
            match k.compare(k0) {
                Ordering::Equal => {
                    let v = V::from_raw_ptr(v as *const u8);
                    v.compare(v0)
                }
                cmp => cmp,
            }
        })
    } else {
        match s.binary_search_by(|&off| {
            let off = u16::from_le(off);
            let (k, _) = read::<K, V>(page.data.as_ptr().offset(off as isize));
            let k = K::from_raw_ptr(k);
            k.compare(k0)
        }) {
            Err(e) => Err(e),
            Ok(mut i) => {
                // Rewind if there are multiple matching keys.
                while i > 0 {
                    let off = u16::from_le(s[i-1]);
                    let (k, _) = read::<K, V>(page.data.as_ptr().offset(off as isize));
                    let k = K::from_raw_ptr(k);
                    if let Ordering::Equal = k.compare(k0){
                        i -= 1
                    } else {
                        break
                    }
                }
                Ok(i)
            }
        }
    }
}
#[async_trait(?Send)]
impl<
        K: UnsizedStorable + ?Sized + core::fmt::Debug,
        V: UnsizedStorable + ?Sized + core::fmt::Debug,
    > super::BTreeMutPage<K, V> for Page<K, V>
{
    // The init function is straightforward.
    fn init(page: &mut MutPage) {
        let h = header_mut(page);
        h.init();
    }
    // Deletions in internal nodes of a B tree need to replace the
    // deleted value with a value from a leaf.
    //
    // In this implementation of pages, we never actually wipe any
    // data from pages, we're even only appending data, or cloning the
    // pages to compact them. Therefore, raw pointers to leaves are
    // always valid for as long as the page isn't freed, which can
    // only happen at the very last step of an insertion or deletion.
    type Saved = (*const K, *const V);
    fn save_deleted_leaf_entry(k: &K, v: &V) -> Self::Saved {
        (k as *const K, v as *const V)
    }
    unsafe fn from_saved<'a>(s: &Self::Saved) -> (&'a K, &'a V) {
        (&*s.0, &*s.1)
    }
    // As in the sized implementation, `put` is split into its own submodule.
    async fn put<'a, T: AllocPage>(
        txn: &mut T,
        page: CowPage,
        mutable: bool,
        replace: bool,
        c: &PageCursor,
        k0: &'a K,
        v0: &'a V,
        k1v1: Option<(&'a K, &'a V)>,
        l: u64,
        r: u64,
    ) -> Result<crate::btree::put::Put<'a, K, V>, T::Error> {
        debug_assert!(c.cur >= 0);
        debug_assert!(k1v1.is_none() || replace);
        if r == 0 {
            put::put::<_, _, _, Leaf>(
                txn,
                page,
                mutable,
                replace,
                c.cur as usize,
                k0,
                v0,
                k1v1,
                0,
                0,
            ).await
        } else {
            put::put::<_, _, _, Internal>(
                txn,
                page,
                mutable,
                replace,
                c.cur as usize,
                k0,
                v0,
                k1v1,
                l,
                r,
            ).await
        }
    }
    unsafe fn put_mut(
        page: &mut MutPage,
        c: &mut Self::Cursor,
        k0: &K,
        v0: &V,
        r: u64,
    ) {
        let mut n = c.cur;
        if r == 0 {
            Leaf::alloc_write(page, k0, v0, 0, r, &mut n);
        } else {
            Internal::alloc_write(page, k0, v0, 0, r, &mut n);
        }
        c.total += 1;
    }
    unsafe fn set_left_child(
        page: &mut MutPage,
        c: &Self::Cursor,
        l: u64
    ) {
        let off = (page.0.data.add(HDR) as *mut u64).offset(c.cur - 1);
        *off = (l | (u64::from_le(*off) & 0xfff)).to_le();
    }
    // Update the left child of the entry pointed to by cursor `c`.
    async fn update_left_child<T: AllocPage>(
        txn: &mut T,
        page: CowPage,
        mutable: bool,
        c: &Self::Cursor,
        l: u64,
    ) -> Result<crate::btree::put::Ok, T::Error> {
        assert!(!c.is_leaf);
        let freed;
        let page = if mutable && page.is_dirty() {
            // If the page is dirty (allocated by this transaction)
            // and isn't shared, just make it mutable.
            freed = 0;
            MutPage(page)
        } else {
            // Else, clone the page:
            let mut new = txn.alloc_page().await?;
            <Page<K, V> as BTreeMutPage<K, V>>::init(&mut new);
            // Copy the left child
            let l = header(page.as_page()).left_page() & !0xfff;
            let hdr = header_mut(&mut new);
            hdr.set_left_page(l);
            // Copy all the entries
            let s = Internal::offset_slice(page.as_page());
            clone::<K, V, Internal>(page.as_page(), &mut new, s, &mut 0);
            // Mark the old version for freeing.
            freed = page.offset | if page.is_dirty() { 1 } else { 0 };
            new
        };
        // Finally, update the left children of the cursor. We know
        // that all valid positions of a cursor except the leftmost
        // one (-1) have a left child.
        assert!(c.cur >= 0);
        unsafe {
            let off = (page.0.data.add(HDR) as *mut u64).offset(c.cur as isize - 1);
            *off = (l | (u64::from_le(*off) & 0xfff)).to_le();
        }
        Ok(Ok { page, freed })
    }
    // Here is how deletions work: if the page is dirty and mutable,
    // we "unlink" the value by moving the end of the offset array to
    // the left by one offset (2 bytes in leaves, 8 bytes in internal
    // nodes).
    async fn del<T: AllocPage>(
        txn: &mut T,
        page: crate::CowPage,
        mutable: bool,
        c: &PageCursor,
        l: u64,
    ) -> Result<(MutPage, u64), T::Error> {
        // Check that the cursor is at a valid position for a deletion.
        debug_assert!(c.cur >= 0 && c.cur < c.total as isize);
        if mutable && page.is_dirty() {
            let p = page.data;
            let mut page = MutPage(page);
            let hdr = header_mut(&mut page);
            let n = hdr.n();
            if c.is_leaf {
                unsafe {
                    let ptr = p.add(HDR + c.cur as usize * 2) as *mut u16;
                    // Get the offset in the page of the key/value data.
                    let off = u16::from_le(*ptr);
                    assert_eq!(off & 0xf000, 0);
                    // Erase the offset by shifting the last (`n -
                    // c.cur - 1`) offsets. The reasoning against
                    // "off-by-one errors" is that if the cursor is
                    // positioned on the first element (`c.cur == 0`),
                    // there are `n-1` elements to shift.
                    core::ptr::copy(ptr.offset(1), ptr, n as usize - c.cur as usize - 1);
                    // Remove the size of the actualy key/value, plus
                    // 2 bytes (the offset).
                    hdr.decr(2 + entry_size::<K, V>(p.add(off as usize)));
                }
            } else {
                unsafe {
                    let ptr = p.add(HDR + c.cur as usize * 8) as *mut u64;
                    // Offset to the key/value data.
                    let off = u64::from_le(*ptr);
                    // Shift the offsets like in the leaf case above.
                    core::ptr::copy(ptr.offset(1), ptr, n as usize - c.cur as usize - 1);
                    if l > 0 {
                        // In an internal node, we may have to update
                        // the left child of the current
                        // position. After the move, the current
                        // offset is at `ptr`, so we need to find the
                        // offset one step to the left.
                        let p = ptr.offset(-1);
                        *p = (l | (u64::from_le(*p) & 0xfff)).to_le();
                    }
                    // Remove the size of the key/value, plus 8 bytes
                    // (the offset).
                    hdr.decr(8 + entry_size::<K, V>(p.add((off & 0xfff) as usize)));
                }
            };
            hdr.set_n(n - 1);
            // Return the page, and 0 for "nothing was freed".
            Ok((page, 0))
        } else {
            // If the page cannot be mutated, we allocate a new page and clone.
            let mut new = txn.alloc_page().await?;
            <Page<K, V> as BTreeMutPage<K, V>>::init(&mut new);
            if c.is_leaf {
                // In a leaf, this is just a matter of getting the
                // offset slice, removing the current position and
                // cloning.
                let s = Leaf::offset_slice(page.as_page());
                let (s0, s1) = s.split_at(c.cur as usize);
                let (_, s1) = s1.split_at(1);
                let mut n = 0;
                clone::<K, V, Leaf>(page.as_page(), &mut new, s0, &mut n);
                clone::<K, V, Leaf>(page.as_page(), &mut new, s1, &mut n);
            } else {
                // In an internal node, things are a bit trickier,
                // since we might need to update the left child.
                //
                // First, clone the leftmost child of the page.
                let hdr = header(page.as_page());
                let left = hdr.left_page() & !0xfff;
                unsafe {
                    *(new.0.data.add(HDR) as *mut u64).offset(-1) = left.to_le();
                }
                // Then, clone the first half of the page.
                let s = Internal::offset_slice(page.as_page());
                let (s0, s1) = s.split_at(c.cur as usize);
                let (_, s1) = s1.split_at(1);
                let mut n = 0;
                clone::<K, V, Internal>(page.as_page(), &mut new, s0, &mut n);
                // Update the left child, which was written by the
                // call to `clone` as the right child of the last
                // entry in `s0`.
                if l > 0 {
                    unsafe {
                        let off = (new.0.data.add(HDR) as *mut u64).offset(n - 1);
                        *off = (l | (u64::from_le(*off) & 0xfff)).to_le();
                    }
                }
                // Then, clone the second half of the page.
                clone::<K, V, Internal>(page.as_page(), &mut new, s1, &mut n);
            }
            // Return the new page, and the offset of the freed page.
            Ok((new, page.offset))
        }
    }
    // Decide what to do with the concatenation of two neighbouring
    // pages, with a middle element.
    //
    // This is highly similar to the sized case.
    async fn merge_or_rebalance<'a, T: AllocPage>(
        txn: &mut T,
        m: Concat<'a, K, V, Self>,
    ) -> Result<Op<'a, T, K, V>, T::Error> {
        // First evaluate the size of the middle element on a
        // page. Contrarily to the sized case, the offsets are
        // aligned, so the header is always the same size (no
        // padding).
        let mid_size = if m.modified.c0.is_leaf {
            2 + alloc_size::<K, V>(m.mid.0, m.mid.1)
        } else {
            8 + alloc_size::<K, V>(m.mid.0, m.mid.1)
        };
        // Evaluate the size of the modified page of the concatenation
        // (which includes the header).
        let mod_size = size::<K, V>(&m.modified);
        // Add the "occupied" size (which excludes the header).
        let occupied = {
            let hdr = header(m.other.as_page());
            (hdr.left_page() & 0xfff) as usize
        };
        if mod_size + mid_size + occupied <= PAGE_SIZE {
            // If the concatenation fits on a page, merge.
            return if m.modified.c0.is_leaf {
                merge::<_, _, _, _, Leaf>(txn, m).await
            } else {
                merge::<_, _, _, _, Internal>(txn, m).await
            };
        }
        // If we can't merge, evaluate the size of the first binding
        // on the other page, to see if we can rebalance.
        let first_other = PageCursor::new(&m.other, 0);
        let first_other_size = current_size::<K, V>(m.other.as_page(), &first_other);
        // If the modified page is at least half-full, or if removing
        // the first entry on the other page would make that other
        // page less than half-full, don't rebalance. See the Sized
        // implementation to see cases where this happens.
        if mod_size >= PAGE_SIZE / 2 || HDR + occupied - first_other_size < PAGE_SIZE / 2 {
            if let Some((k, v)) = m.modified.ins {
                return Ok(Op::Put(Self::put(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    m.modified.skip_first,
                    &m.modified.c1,
                    k,
                    v,
                    m.modified.ins2,
                    m.modified.l,
                    m.modified.r,
                ).await?));
            } else if m.modified.skip_first {
                debug_assert!(m.modified.ins2.is_none());
                let (page, freed) = Self::del(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    &m.modified.c1,
                    m.modified.l,
                ).await?;
                return Ok(Op::Put(Put::Ok(Ok { page, freed })));
            } else {
                let mut c1 = m.modified.c1.clone();
                let mut l = m.modified.l;
                if l == 0 {
                    Self::move_next(&mut c1);
                    l = m.modified.r;
                }
                return Ok(Op::Put(Put::Ok(Self::update_left_child(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    &c1,
                    l,
                ).await?)));
            }
        }
        // Finally, if we're here, we can rebalance. There are four
        // (relatively explicit) cases, see the `rebalance` submodule
        // to see how this is done.
        if m.mod_is_left {
            if m.modified.c0.is_leaf {
                rebalance_left::<_, _, _, Leaf>(txn, m).await
            } else {
                rebalance_left::<_, _, _, Internal>(txn, m).await
            }
        } else {
            rebalance_right::<_, _, _, Self>(txn, m).await
        }
    }
}
/// Size of a modified page (including the header).
fn size<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
    m: &ModifiedPage<K, V, Page<K, V>>,
) -> usize {
    let mut total = {
        let hdr = header(m.page.as_page());
        (hdr.left_page() & 0xfff) as usize
    };
    total += HDR;
    // Extra size for the offsets.
    let extra = if m.c1.is_leaf { 2 } else { 8 };
    if let Some((k, v)) = m.ins {
        total += extra + alloc_size(k, v) as usize;
        if let Some((k, v)) = m.ins2 {
            total += extra + alloc_size(k, v) as usize;
        }
    }
    if m.skip_first {
        total -= current_size::<K, V>(m.page.as_page(), &m.c1) as usize;
    }
    total
}
// Size of a pair of type `(K, V)`. This is computed in the same way
// as a struct with fields of type `K` and `V` in C.
fn alloc_size<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(k: &K, v: &V) -> usize {
    let s = ((k.size() + V::ALIGN - 1) & !(V::ALIGN - 1)) + v.size();
    let al = K::ALIGN.max(V::ALIGN);
    (s + al - 1) & !(al - 1)
}
// Total size of the entry for that cursor position, including the
// offset size.
fn current_size<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
    page: crate::Page,
    c: &PageCursor,
) -> usize {
    if c.is_leaf {
        Leaf::current_size::<K, V>(page, c.cur)
    } else {
        Internal::current_size::<K, V>(page, c.cur)
    }
}
pub(super) trait AllocWrite<K: ?Sized, V: ?Sized> {
    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize);
    fn set_left_child(new: &mut MutPage, n: isize, l: u64);
}
/// Perform the modifications on a page, by copying it onto page `new`.
fn modify<K: core::fmt::Debug + ?Sized, V: core::fmt::Debug + ?Sized, P: BTreeMutPage<K, V>, L: AllocWrite<K, V>>(
    new: &mut MutPage,
    m: &mut ModifiedPage<K, V, P>,
    n: &mut isize,
) {
    let mut l = P::left_child(m.page.as_page(), &m.c0);
    while let Some((k, v, r)) = P::next(m.page.as_page(), &mut m.c0) {
        L::alloc_write(new, k, v, l, r, n);
        l = 0;
    }
    let mut rr = m.r;
    if let Some((k, v)) = m.ins {
        if let Some((k2, v2)) = m.ins2 {
            L::alloc_write(new, k, v, l, m.l, n);
            L::alloc_write(new, k2, v2, 0, m.r, n);
        } else if m.l > 0 {
            L::alloc_write(new, k, v, m.l, m.r, n);
        } else {
            L::alloc_write(new, k, v, l, m.r, n);
        }
        l = 0;
        rr = 0;
    } else if m.l > 0 {
        l = m.l
    }
    let mut c1 = m.c1.clone();
    if m.skip_first {
        P::move_next(&mut c1);
    }
    // Here's a confusing thing: if the first element of `c1` is the
    // last element of a page, we may be updating its right child (in
    // which case m.r > 0) rather than its left child like for all
    // other elements.
    //
    // This case only ever happens for this function when we're
    // updating the last child of a page p, and then merging p with
    // its right neighbour.
    while let Some((k, v, r)) = P::next(m.page.as_page(), &mut c1) {
        if rr > 0 {
            L::alloc_write(new, k, v, l, rr, n);
            rr = 0;
        } else {
            L::alloc_write(new, k, v, l, r, n);
        }
        l = 0;
    }
    if l != 0 {
        // The while loop above didn't run, i.e. the insertion
        // happened at the end of the page. In this case, we haven't
        // had a chance to update the left page, so do it now.
        L::set_left_child(new, *n, l)
    }
}
/// The very unsurprising `merge` function
pub(super) async fn merge<
    'a,
    T: AllocPage + LoadPage,
    K: ?Sized + core::fmt::Debug,
    V: ?Sized + core::fmt::Debug,
    P: BTreeMutPage<K, V>,
    L: AllocWrite<K, V>,
>(
    txn: &mut T,
    mut m: Concat<'a, K, V, P>,
) -> Result<Op<'a, T, K, V>, T::Error> {
    // Here, we first allocate a page, then clone both pages onto it,
    // in a different order depending on whether the modified page is
    // the left or the right child.
    //
    // Note that in the case that this merge happens immediately after
    // a put that reallocated the two sides of the merge in order to
    // split (not all splits do that), we could be slightly more
    // efficient, but with considerably more code.
    let mut new = txn.alloc_page().await?;
    P::init(&mut new);
    let mut n = 0;
    if m.mod_is_left {
        modify::<_, _, _, L>(&mut new, &mut m.modified, &mut n);
        let mut rc = P::cursor_first(&m.other);
        let l = P::left_child(m.other.as_page(), &rc);
        L::alloc_write(&mut new, m.mid.0, m.mid.1, 0, l, &mut n);
        while let Some((k, v, r)) = P::next(m.other.as_page(), &mut rc) {
            L::alloc_write(&mut new, k, v, 0, r, &mut n);
        }
    } else {
        let mut rc = P::cursor_first(&m.other);
        let mut l = P::left_child(m.other.as_page(), &rc);
        while let Some((k, v, r)) = P::next(m.other.as_page(), &mut rc) {
            L::alloc_write(&mut new, k, v, l, r, &mut n);
            l = 0;
        }
        L::alloc_write(&mut new, m.mid.0, m.mid.1, 0, 0, &mut n);
        modify::<_, _, _, L>(&mut new, &mut m.modified, &mut n);
    }
    let b0 = if m.modified.page.is_dirty() { 1 } else { 0 };
    let b1 = if m.other.is_dirty() { 1 } else { 0 };
    Ok(Op::Merged {
        page: new,
        freed: [m.modified.page.offset | b0, m.other.offset | b1],
        marker: core::marker::PhantomData,
    })
}
fn clone<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized, L: Alloc>(
    page: crate::Page,
    new: &mut MutPage,
    s: Offsets<L::Offset>,
    n: &mut isize,
) {
    for off in s.0.iter() {
        let (r, off): (u64, usize) = (*off).into();
        unsafe {
            let ptr = page.data.as_ptr().add(off);
            let size = entry_size::<K, V>(ptr);
            // Reserve the space on the page
            let hdr = header_mut(new);
            let data = hdr.data() as u16;
            let off_new = data - size as u16;
            hdr.set_data(off_new);
            hdr.set_n(hdr.n() + 1);
            if hdr.is_leaf() {
                hdr.incr(2 + size);
                let ptr = new.0.data.offset(HDR as isize + *n * 2) as *mut u16;
                *ptr = (off_new as u16).to_le();
            } else {
                hdr.incr(8 + size);
                // Set the offset to this new entry in the offset
                // array, along with the right child page.
                let ptr = new.0.data.offset(HDR as isize + *n * 8) as *mut u64;
                *ptr = (r | off_new as u64).to_le();
            }
            core::ptr::copy_nonoverlapping(ptr, new.0.data.offset(off_new as isize), size);
        }
        *n += 1;
    }
}

use super::cursor::*;
use super::*;
// The implementation here is mostly the same as in the sized case,
// except for leaves, which makes it hard to refactor.
pub(crate) async fn rebalance_left<
    'a,
    T: AllocPage,
    K: UnsizedStorable + ?Sized + core::fmt::Debug,
    V: UnsizedStorable + ?Sized + core::fmt::Debug,
    L: Alloc,
>(
    txn: &mut T,
    m: Concat<'a, K, V, Page<K, V>>,
) -> Result<Op<'a, T, K, V>, T::Error> {
    assert!(m.mod_is_left);
    // First element of the right page. We'll rotate it to the left
    // page, i.e. return a middle element, and append the current
    // middle element to the left page.
    let rc = super::cursor::PageCursor::new(&m.other, 0);
    let rl = <Page<K, V>>::left_child(m.other.as_page(), &rc);
    let (k, v, r) = <Page<K, V>>::current(m.other.as_page(), &rc).unwrap();
    let mut freed_ = [0, 0];
    // First, perform the modification on the modified page, which we
    // know is the left page, and return the resulting mutable page
    // (the modified page may or may not be mutable before we do this).
    let new_left = if let Some((k, v)) = m.modified.ins {
        let is_dirty = m.modified.page.is_dirty();
        let new_left = if let Put::Ok(Ok { page, freed }) = <Page<K, V>>::put(
            txn,
            m.modified.page,
            m.modified.mutable,
            m.modified.skip_first,
            &m.modified.c1,
            k,
            v,
            m.modified.ins2,
            m.modified.l,
            m.modified.r,
        ).await? {
            if freed > 0 {
                let b = if is_dirty { 1 } else { 0 };
                freed_[0] = freed | b;
            }
            page
        } else {
            unreachable!()
        };
        // Append the middle element of the concatenation at the end
        // of the left page. We know the left page to be mutable by
        // now, and we also know there's enough space to do this.
        let lc = PageCursor::after(&new_left.0);
        if let Put::Ok(Ok { freed, page }) = <Page<K, V>>::put(
            txn, new_left.0, true, false, &lc, m.mid.0, m.mid.1, None, 0, rl,
        ).await? {
            if freed > 0 {
                let b = if is_dirty { 1 } else { 0 };
                freed_[0] = freed | b;
            }
            page
        } else {
            unreachable!()
        }
    } else if m.modified.skip_first {
        let is_dirty = m.modified.page.is_dirty();
        let (page, freed) = <Page<K, V>>::del(
            txn,
            m.modified.page,
            m.modified.mutable,
            &m.modified.c1,
            m.modified.l,
        ).await?;
        if freed > 0 {
            let b = if is_dirty { 1 } else { 0 };
            freed_[0] = freed | b;
        }
        // Append the middle element of the concatenation at the end
        // of the left page. We know the left page to be mutable by
        // now, and we also know there's enough space to do this.
        let lc = PageCursor::after(&page.0);
        if let Put::Ok(Ok { freed, page }) =
            <Page<K, V>>::put(txn, page.0, true, false, &lc, m.mid.0, m.mid.1, None, 0, rl).await?
        {
            if freed > 0 {
                let b = if is_dirty { 1 } else { 0 };
                freed_[0] = freed | b;
            }
            page
        } else {
            unreachable!()
        }
    } else {
        // Append the middle element of the concatenation at the end
        // of the left page. We know the left page to be mutable by
        // now, and we also know there's enough space to do this.
        let is_dirty = m.modified.page.is_dirty();
        let lc = PageCursor::after(&m.modified.page);
        if let Put::Ok(Ok { freed, page }) = <Page<K, V>>::put(
            txn,
            m.modified.page,
            m.modified.mutable,
            false,
            &lc,
            m.mid.0,
            m.mid.1,
            None,
            0,
            rl,
        ).await? {
            if m.modified.l > 0 {
                assert_eq!(m.modified.r, 0);
                unsafe {
                    let off = (page.0.data.add(HDR) as *mut u64).offset(m.modified.c1.cur - 1);
                    *off = (m.modified.l | (u64::from_le(*off) & 0xfff)).to_le();
                }
            } else if m.modified.r > 0 {
                unsafe {
                    let off = (page.0.data.add(HDR) as *mut u64).offset(m.modified.c1.cur);
                    *off = (m.modified.r | (u64::from_le(*off) & 0xfff)).to_le();
                }
            }
            if freed > 0 {
                let b = if is_dirty { 1 } else { 0 };
                freed_[0] = freed | b;
            }
            page
        } else {
            unreachable!()
        }
    };
    // We extend the pointer's lifetime here, because we know the
    // current deletion (we only rebalance during deletions) won't
    // touch this page anymore after this.
    let k = unsafe { core::mem::transmute(k) };
    let v = unsafe { core::mem::transmute(v) };
    // If this frees the old "other" page, add it to the "freed"
    // array.
    let is_dirty = m.other.is_dirty();
    let (new_right, freed) = <Page<K, V>>::del(txn, m.other, m.other_is_mutable, &rc, r).await?;
    if freed > 0 {
        freed_[1] = if is_dirty { freed | 1 } else { freed }
    }
    Ok(Op::Rebalanced {
        l: new_left.0.offset,
        r: new_right.0.offset,
        k,
        v,
        freed: freed_,
    })
}
// Surprisingly, the `rebalance_right` function is simpler,
// since:
//
// - if we call it to rebalance two internal nodes, we're in the easy
// case of rebalance_left.
//
// - Else, the middle element is the last one on the left page, and
// isn't erased be leaf deletions, because these just move entries to
// the left.
//
// This implementation is shared with the sized one.
pub(crate) async fn rebalance_right<
    'a,
    T: AllocPage,
    K: ?Sized,
    V: ?Sized,
    P: BTreeMutPage<K, V, Cursor = super::PageCursor>,
>(
    txn: &mut T,
    m: Concat<'a, K, V, P>,
) -> Result<Op<'a, T, K, V>, T::Error> {
    assert!(!m.mod_is_left);
    // Take the last element of the left page.
    let lc = P::cursor_last(&m.other);
    let (k0, v0, r0) = P::current(m.other.as_page(), &lc).unwrap();
    let mut freed_ = [0, 0];
    // Perform the modification on the modified page.
    let new_right = if let Some((k, v)) = m.modified.ins {
        let is_dirty = m.modified.page.is_dirty();
        let new_right = if let Put::Ok(Ok { page, freed }) = P::put(
            txn,
            m.modified.page,
            m.modified.mutable,
            m.modified.skip_first,
            &m.modified.c1,
            k,
            v,
            m.modified.ins2,
            m.modified.l,
            m.modified.r,
        ).await? {
            if freed > 0 {
                freed_[0] = if is_dirty { freed | 1 } else { freed };
            }
            page
        } else {
            unreachable!()
        };
        // Add the middle element of the concatenation as the first
        // element of the right page. We know the right page is
        // mutable, since we just modified it (hence the
        // `assert_eq!(freed, 0)`.
        // First element of the right page (after potential
        // modification by `put` above).
        let rc = P::cursor_first(&new_right.0);
        let rl = P::left_child(new_right.0.as_page(), &rc);
        if let Put::Ok(Ok { freed, page }) = P::put(
            txn,
            new_right.0,
            true,
            false,
            &rc,
            m.mid.0,
            m.mid.1,
            None,
            r0,
            rl,
        ).await? {
            debug_assert_eq!(freed, 0);
            page
        } else {
            unreachable!()
        }
    } else if m.modified.skip_first {
        let is_dirty = m.modified.page.is_dirty();
        let (page, freed) = P::del(
            txn,
            m.modified.page,
            m.modified.mutable,
            &m.modified.c1,
            m.modified.l,
        ).await?;
        if freed > 0 {
            freed_[0] = if is_dirty { freed | 1 } else { freed };
        }
        // Add the middle element of the concatenation as the first
        // element of the right page. We know the right page is
        // mutable, since we just modified it. Moreover, if it is
        // compacted by the `put` below, we know that the `del` didn't
        // free anything, hence we can reuse the slot 0..
        // First element of the right page (after potential
        // modification by `del` above).
        let rc = P::cursor_first(&page.0);
        let rl = P::left_child(page.0.as_page(), &rc);
        if let Put::Ok(Ok { freed, page }) = P::put(
            txn, page.0, true, false, &rc, m.mid.0, m.mid.1, None, r0, rl,
        ).await? {
            if freed > 0 {
                freed_[0] = if is_dirty { freed | 1 } else { freed };
            }
            page
        } else {
            unreachable!()
        }
    } else {
        let is_dirty = m.modified.page.is_dirty();
        let rc = P::cursor_first(&m.modified.page);
        let rl = P::left_child(m.modified.page.as_page(), &rc);
        if let Put::Ok(Ok { freed, page }) = P::put(
            txn,
            m.modified.page,
            m.modified.mutable,
            false,
            &rc,
            m.mid.0,
            m.mid.1,
            None,
            r0,
            rl,
        ).await? {
            // Update the left and right offsets. We know that at
            // least one of them is 0, but it can be either one (or
            // both), depending on what happened on the page below.
            //
            // Since we inserted an entry at the beginning of the
            // page, we need to add 1 to the index given by
            // `m.modified.c1.cur`.
            if m.modified.l > 0 {
                assert_eq!(m.modified.r, 0);
                unsafe {
                    let off = (page.0.data.add(HDR) as *mut u64).offset(m.modified.c1.cur);
                    *off = (m.modified.l | (u64::from_le(*off) & 0xfff)).to_le();
                }
            } else if m.modified.r > 0 {
                unsafe {
                    let off = (page.0.data.add(HDR) as *mut u64).offset(m.modified.c1.cur + 1);
                    *off = (m.modified.r | (u64::from_le(*off) & 0xfff)).to_le();
                }
            }
            if freed > 0 {
                freed_[0] = if is_dirty { freed | 1 } else { freed };
            }
            page
        } else {
            unreachable!()
        }
    };
    // As explained in the general comment on this function, this
    // entry isn't erased by the deletion in `m.other` below, so we
    // can safely extend its lifetime.
    let k = unsafe { core::mem::transmute(k0) };
    let v = unsafe { core::mem::transmute(v0) };
    let is_dirty = m.other.is_dirty();
    let (new_left, freed) = P::del(txn, m.other, m.other_is_mutable, &lc, 0).await?;
    if freed > 0 {
        freed_[1] = if is_dirty { freed | 1 } else { freed }
    }
    Ok(Op::Rebalanced {
        l: new_left.0.offset,
        r: new_right.0.offset,
        k,
        v,
        freed: freed_,
    })
}

use super::header::{header, header_mut, HDR};
use super::*;
pub(super) async fn put<
    'a,
    T: AllocPage,
    K: UnsizedStorable + ?Sized + core::fmt::Debug,
    V: UnsizedStorable + ?Sized + core::fmt::Debug,
    L: Alloc + AllocWrite<K, V>,
>(
    txn: &mut T,
    page: CowPage,
    mutable: bool,
    replace: bool,
    u: usize,
    k0: &'a K,
    v0: &'a V,
    k1v1: Option<(&'a K, &'a V)>,
    l: u64,
    r: u64,
) -> Result<crate::btree::put::Put<'a, K, V>, T::Error> {
    // Size of the new insertions, not counting the offsets.
    let size = alloc_size(k0, v0)
        + if let Some((k1, v1)) = k1v1 {
            alloc_size(k1, v1)
        } else {
            0
        };
    // Sized occupied by the data in these insertions (not counting
    // the offsets), if we need to compact the page.
    let size_replaced = if replace {
        // We're always in an internal node if we replace.
        let cur_size = Internal::current_size::<K, V>(page.as_page(), u as isize);
        // Since `size` isn't counting the size of the 64-bit offset,
        // and `cur_size` is, we need to add 8.
        8 + size as isize - cur_size as isize
    } else {
        size as isize
    };
    // Number of extra offsets.
    let n_ins = {
        let n_ins = if k1v1.is_some() { 2 } else { 1 };
        if replace {
            n_ins - 1
        } else {
            n_ins
        }
    };
    let hdr = header(page.as_page());
    if mutable && page.is_dirty() && L::can_alloc(header(page.as_page()), size, n_ins) {
        // First, if the page is dirty and mutable, mark it mutable.
        let mut page = MutPage(page);
        let mut n = u as isize;
        // If we replace, we first need to "unlink" the previous
        // value, by erasing its offset.
        if replace {
            let p = page.0.data;
            let hdr = header_mut(&mut page);
            // In this case, we know we're not in a leaf, so offsets
            // are of size 8.
            assert!(!hdr.is_leaf());
            unsafe {
                let ptr = p.add(HDR + n as usize * 8) as *mut u64;
                let off = (u64::from_le(*ptr) & 0xfff) as usize;
                let kv_ptr = p.add(off);
                let size = entry_size::<K, V>(kv_ptr);
                core::ptr::copy(ptr.offset(1), ptr, hdr.n() as usize - n as usize - 1);
                // Decreasing these figures here, we'll increase them
                // again in the calls to `alloc_write` below.
                hdr.decr(8 + size);
                hdr.set_n(hdr.n() - 1);
            }
        }
        // Do the actual insertions.
        if let Some((k1, v1)) = k1v1 {
            L::alloc_write(&mut page, k0, v0, 0, 0, &mut n);
            L::alloc_write(&mut page, k1, v1, l, r, &mut n);
        } else {
            L::alloc_write(&mut page, k0, v0, l, r, &mut n);
        }
        // Return: no page was freed.
        Ok(Put::Ok(Ok { page, freed: 0 }))
    } else if L::can_compact(hdr, size_replaced, n_ins) {
        // Allocate a new page, split the offsets at the position of
        // the insertion, and each side of the split onto the new
        // page, with the insertion between them.
        let mut new = txn.alloc_page().await?;
        <Page<K, V> as BTreeMutPage<K, V>>::init(&mut new);
        // Clone the leftmost child.
        L::set_right_child(&mut new, -1, hdr.left_page() & !0xfff);
        // Split the offsets.
        let s = L::offset_slice(page.as_page());
        let (s0, s1) = s.split_at(u as usize);
        // Drop the first element of `s1` if we're replacing it.
        let s1 = if replace { s1.split_at(1).1 } else { s1 };
        // Finally, clone and insert.
        let mut n = 0;
        clone::<K, V, L>(page.as_page(), &mut new, s0, &mut n);
        if let Some((k1, v1)) = k1v1 {
            L::alloc_write(&mut new, k0, v0, 0, l, &mut n);
            L::alloc_write(&mut new, k1, v1, 0, r, &mut n);
        } else {
            L::alloc_write(&mut new, k0, v0, l, r, &mut n);
        }
        clone::<K, V, L>(page.as_page(), &mut new, s1, &mut n);
        // And return, freeing the old version of the page.
        Ok(Put::Ok(Ok {
            page: new,
            freed: page.offset | if page.is_dirty() { 1 } else { 0 },
        }))
    } else {
        split_unsized::<_, _, _, L>(txn, page, replace, u, k0, v0, k1v1, l, r).await
    }
}
// Surprisingly, the following function is simpler than in the sized
// case, because we can't be too efficient in this case, since we have
// to go through all the elements linearly anyway to get their sizes.
async fn split_unsized<
    'a,
    T: AllocPage,
    K: UnsizedStorable + ?Sized + core::fmt::Debug,
    V: UnsizedStorable + ?Sized + core::fmt::Debug,
    L: Alloc + AllocWrite<K, V>,
>(
    txn: &mut T,
    page: CowPage,
    replace: bool,
    u: usize,
    k0: &'a K,
    v0: &'a V,
    k1v1: Option<(&'a K, &'a V)>,
    l0: u64,
    r0: u64,
) -> Result<crate::btree::put::Put<'a, K, V>, T::Error> {
    let hdr = header(page.as_page());
    let mut left = txn.alloc_page().await?;
    <Page<K, V> as BTreeMutPage<K, V>>::init(&mut left);
    let mut right = txn.alloc_page().await?;
    <Page<K, V> as BTreeMutPage<K, V>>::init(&mut right);
    // Clone the leftmost child. If we're inserting at position 0,
    // this means this leftmost child must, and will be replaced.
    let left_child = hdr.left_page() & !0xfff;
    L::set_right_child(&mut left, -1, left_child);
    // Pointer to the split key and value.
    let mut split = None;
    // We start filling the left page, and we will change to filling
    // the right page when the left one is at least half-full.
    let mut current_page = &mut left;
    // There are two synchronised counters here: `hdr.n()` and
    // `n`. The reason for this is that operations on `hdr.n()` are
    // somewhat cumbersome, as they might involve extra operations to
    // convert to/from the little-endian encoding of `hdr.n()`.
    let mut n = 0;
    let s = L::offset_slice(page.as_page());
    let mut total = HDR;
    for (uu, off) in s.0.iter().enumerate() {
        // If the current position of the cursor is the insertion,
        // (i.e. `uu == u`), insert.
        if uu == u {
            // The following is tricky and makes assumptions on the
            // rest of the code. If we are inserting two elements
            // (i.e. if `k1v1.is_some()`), this means we're replacing
            // one on the page.
            header_mut(current_page).set_n(n as u16);
            if let Some((k1, v1)) = k1v1 {
                // As explained in the documentation for `put` in the
                // definition of `BTreeMutPage`, `l0` and `r0` are the
                // left and right children of k1v1, since this means
                // the right child of a deleted entry has split, and
                // we must replace the deleted entry with `(k0, v0)`.
                L::alloc_write(current_page, k0, v0, 0, l0, &mut n);
                total += alloc_size(k0, v0) + L::OFFSET_SIZE;
                L::alloc_write(current_page, k1, v1, 0, r0, &mut n);
                total += alloc_size(k1, v1) + L::OFFSET_SIZE;
                // Replacing the next element:
                debug_assert!(replace);
                continue;
            } else {
                L::alloc_write(current_page, k0, v0, l0, r0, &mut n);
                total += alloc_size(k0, v0) + L::OFFSET_SIZE;
                if replace {
                    continue;
                }
            }
        }
        // Then, i.e. either if we're not at the position of an
        // insertion, or else if we're not replacing the current
        // position, just copy the current entry to the appropriate
        // page (left or right).
        let (r, off): (u64, usize) = (*off).into();
        assert_ne!(off, 0);
        unsafe {
            let ptr = page.data.add(off);
            let size = entry_size::<K, V>(ptr);
            // If the left side of the split is at least half-full, we
            // know that we can do all the insertions we need on the
            // right-hand side (even if there are two of them, and the
            // replaced value is of size 0), so we switch.
            if split.is_none() && total >= PAGE_SIZE / 2 {
                // Don't write the next entry onto any page, keep it
                // as the split entry.
                let (k, v) = read::<K, V>(ptr);
                split = Some((K::from_raw_ptr(k), V::from_raw_ptr(v)));
                // Set the entry count for the current page, before
                // switching and resetting the counter.
                header_mut(current_page).set_n(n as u16);
                // And set the leftmost child of the right page to
                // `r`.
                L::set_right_child(&mut right, -1, r);
                // Then, switch page and reset the counter.
                current_page = &mut right;
                n = 0;
            } else {
                // Else, we're simply allocating a new entry on the
                // current page.
                let off_new = {
                    let hdr = header_mut(current_page);
                    let data = hdr.data();
                    assert!(
                        HDR + hdr.n() as usize * L::OFFSET_SIZE + L::OFFSET_SIZE + size
                            < data as usize
                    );
                    // Move the data pointer `size` bytes to the left.
                    hdr.set_data(data - size as u16);
                    // And increase the number of entries, and
                    // occupied size of the page.
                    hdr.incr(size + L::OFFSET_SIZE);
                    data - size as u16
                };
                // Copy the entry.
                core::ptr::copy_nonoverlapping(
                    ptr,
                    current_page.0.data.offset(off_new as isize),
                    size,
                );
                // And set the offset.
                L::set_offset(current_page, n, r | (off_new as u64));
                n += 1;
            }
            total += size + L::OFFSET_SIZE;
        }
    }
    header_mut(current_page).set_n(n as u16);
    // Finally, it is possible that we haven't had a chance to do the
    // insertions yet: this is the case iff we're inserting at the end
    // of all the entries, so handle this now.
    if u == s.0.len() {
        if let Some((k1, v1)) = k1v1 {
            L::alloc_write(current_page, k0, v0, 0, l0, &mut n);
            L::alloc_write(current_page, k1, v1, 0, r0, &mut n);
        } else {
            L::alloc_write(current_page, k0, v0, l0, r0, &mut n);
        }
    }
    let (split_key, split_value) = split.unwrap();
    Ok(Put::Split {
        split_key,
        split_value,
        left,
        right,
        freed: if page.is_dirty() {
            page.offset | 1
        } else {
            page.offset
        },
    })
}

use crate::PAGE_SIZE;
#[derive(Debug)]
#[repr(C)]
pub struct Header {
    data: u16,
    n: u16,
    crc: u32,
    left_page: u64,
}
impl Header {
    pub(crate) fn init(&mut self) {
        self.data = (((PAGE_SIZE as u16) << 3) | 1).to_le();
        self.n = 0;
        self.crc = 0;
        self.left_page = 0;
    }
    pub(crate) fn n(&self) -> u16 {
        u16::from_le(self.n)
    }
    pub(crate) fn set_n(&mut self, n: u16) {
        self.n = n.to_le();
    }
    pub fn left_page(&self) -> u64 {
        u64::from_le(self.left_page)
    }
    pub fn set_left_page(&mut self, l: u64) {
        self.left_page = l.to_le()
    }
    pub fn data(&self) -> u16 {
        u16::from_le(self.data) >> 3
    }
    pub fn set_data(&mut self, d: u16) {
        let dirty = u16::from_le(self.data) & 7;
        self.data = ((d << 3) | dirty).to_le()
    }
    pub fn decr(&mut self, s: usize) {
        self.left_page = (self.left_page() - s as u64).to_le();
    }
    pub fn set_occupied(&mut self, size: u64) {
        self.left_page = ((self.left_page() & !0xfff) | size).to_le()
    }
    pub fn incr(&mut self, s: usize) {
        self.left_page = (self.left_page() + s as u64).to_le();
    }
    pub fn is_leaf(&self) -> bool {
        u64::from_le(self.left_page) <= 0xfff
    }
}
pub const HDR: usize = core::mem::size_of::<Header>();
pub(crate) fn header<'a>(page: crate::Page<'a>) -> &'a Header {
    unsafe { &*(page.data.as_ptr() as *const Header) }
}
pub fn header_mut(page: &mut crate::MutPage) -> &mut Header {
    unsafe { &mut *(page.0.data as *mut Header) }
}

use super::{header, Header};
#[derive(Debug, Clone, Copy)]
pub struct PageCursor {
    pub cur: isize,
    pub total: usize,
    pub is_leaf: bool,
}
impl PageCursor {
    /// Initialise a cursor on page `p` at position `cur`, checking
    /// that `cur` is a valid position.
    pub fn new(p: &crate::CowPage, cur: isize) -> Self {
        let hdr = unsafe { &*(p.data as *const Header) };
        assert!(cur >= -1 && cur < hdr.n() as isize);
        PageCursor {
            cur,
            total: hdr.n() as usize,
            is_leaf: hdr.is_leaf(),
        }
    }
    /// Initialise a cursor set after the last entry of page `p`.
    pub fn after(p: &crate::CowPage) -> Self {
        let hdr = unsafe { &*(p.data as *const Header) };
        let total = hdr.n();
        PageCursor {
            cur: total as isize,
            total: total as usize,
            is_leaf: hdr.is_leaf(),
        }
    }
    /// Initialise a cursor set on the last entry of page `p`.
    pub fn last(p: crate::Page) -> Self {
        let hdr = header(p);
        let total = hdr.n();
        PageCursor {
            cur: (total - 1) as isize,
            total: total as usize,
            is_leaf: hdr.is_leaf(),
        }
    }
}

use super::header::{header, header_mut, Header, HDR};
use super::*;
/// We'd love to share this trait with the sized implementation, but
/// unfortunately, the type parameters of almost all methods are
/// different.
pub(crate) trait Alloc {
    const OFFSET_SIZE: usize;
    /// Size (including the offset size) of the entry at position
    /// `cur`.
    fn current_size<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
        page: crate::Page,
        cur: isize,
    ) -> usize;
    /// Can we allocate an entry of `size` bytes, where `size` doesn't
    /// include the offset bytes?
    fn can_alloc(hdr: &Header, size: usize, n: usize) -> bool {
        (HDR as usize) + (hdr.n() as usize) * Self::OFFSET_SIZE + n * Self::OFFSET_SIZE + size
            <= hdr.data() as usize
    }
    /// If we cloned the page, could we allocate an entry of `size`
    /// bytes, where `size` doesn't include the offset bytes?
    fn can_compact(hdr: &Header, size: isize, n: usize) -> bool {
        (HDR as isize)
            + ((hdr.left_page() & 0xfff) as isize)
            + (n * Self::OFFSET_SIZE) as isize
            + size
            <= PAGE_SIZE as isize
    }
    /// Set the right child of the `n`th entry to `r`. If `n == -1`,
    /// this method can be used to set the leftmost child of a page.
    fn set_right_child(_new: &mut MutPage, _n: isize, _r: u64) {}
    /// Set the n^th offset on the page to `r`, which encodes a page
    /// offset in its 52 MSBs, and an offset on the page in its 12 LSBs.
    fn set_offset(new: &mut MutPage, n: isize, r: u64);
    /// The type of offsets, u64 in internal nodes, u16 in leaves.
    type Offset: Into<(u64, usize)> + Copy + core::fmt::Debug;
    fn offset_slice<'a>(page: crate::Page<'a>) -> Offsets<'a, Self::Offset>;
}
#[derive(Debug, Clone, Copy)]
pub struct Offsets<'a, A>(pub &'a [A]);
impl<'a, A: Copy> Offsets<'a, A> {
    pub fn split_at(self, mid: usize) -> (Self, Self) {
        if mid < self.0.len() {
            let (a, b) = self.0.split_at(mid);
            (Offsets(a), Offsets(b))
        } else {
            debug_assert_eq!(mid, self.0.len());
            (self, Offsets(&[][..]))
        }
    }
}
pub struct Leaf {}
pub struct Internal {}
impl Alloc for Leaf {
    const OFFSET_SIZE: usize = 2;
    // Note: the size returned by this function includes the offset.
    fn current_size<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
        page: crate::Page,
        cur: isize,
    ) -> usize {
        // Find the offset of the current position, get its size.
        unsafe {
            debug_assert!(cur >= 0);
            let off = *(page.data.as_ptr().add(HDR + 2 * cur as usize) as *const u16);
            Self::OFFSET_SIZE
                + entry_size::<K, V>(page.data.as_ptr().add(u16::from_le(off) as usize))
        }
    }
    fn set_offset(new: &mut MutPage, n: isize, off: u64) {
        unsafe {
            let ptr = new.0.data.offset(HDR as isize + n * 2) as *mut u16;
            *ptr = (off as u16).to_le();
        }
    }
    type Offset = LeafOffset;
    fn offset_slice<'a>(page: crate::Page<'a>) -> Offsets<'a, Self::Offset> {
        let hdr_n = header(page).n();
        unsafe {
            Offsets(core::slice::from_raw_parts(
                page.data.as_ptr().add(HDR) as *const LeafOffset,
                hdr_n as usize,
            ))
        }
    }
}
impl Alloc for Internal {
    const OFFSET_SIZE: usize = 8;
    fn current_size<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(
        page: crate::Page,
        cur: isize,
    ) -> usize {
        unsafe {
            8 + entry_size::<K, V>(page.data.as_ptr().add(
                (u64::from_le(*(page.data.as_ptr().add(HDR) as *const u64).offset(cur)) & 0xfff)
                    as usize,
            ))
        }
    }
    /// Set the right-hand child in the offset array, preserving the
    /// current offset.
    fn set_right_child(page: &mut MutPage, n: isize, r: u64) {
        unsafe {
            debug_assert_eq!(r & 0xfff, 0);
            let ptr = page.0.data.offset(HDR as isize + n * 8) as *mut u64;
            let off = u64::from_le(*ptr) & 0xfff;
            *ptr = (r | off as u64).to_le();
        }
    }
    fn set_offset(new: &mut MutPage, n: isize, off: u64) {
        unsafe {
            let ptr = new.0.data.offset(HDR as isize + n * 8) as *mut u64;
            *ptr = off.to_le();
        }
    }
    type Offset = InternalOffset;
    fn offset_slice<'a>(page: crate::Page<'a>) -> Offsets<'a, Self::Offset> {
        let hdr_n = header(page).n() as usize;
        unsafe {
            Offsets(core::slice::from_raw_parts(
                page.data.as_ptr().add(HDR) as *const InternalOffset,
                hdr_n,
            ))
        }
    }
}
#[derive(Debug, Clone, Copy)]
#[repr(C)]
pub struct LeafOffset(u16);
impl Into<(u64, usize)> for LeafOffset {
    fn into(self) -> (u64, usize) {
        (0, self.0 as usize)
    }
}
impl Into<usize> for LeafOffset {
    fn into(self) -> usize {
        self.0 as usize
    }
}
#[derive(Debug, Clone, Copy)]
#[repr(C)]
pub struct InternalOffset(u64);
impl Into<(u64, usize)> for InternalOffset {
    fn into(self) -> (u64, usize) {
        (self.0 & !0xfff, (self.0 & 0xfff) as usize)
    }
}
impl Into<usize> for InternalOffset {
    fn into(self) -> usize {
        self.0 as usize
    }
}
impl<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized> AllocWrite<K, V> for Leaf {
    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize) {
        alloc_write::<K, V, Self>(new, k0, v0, l, r, n)
    }
    fn set_left_child(_new: &mut MutPage, _n: isize, l: u64) {
        debug_assert_eq!(l, 0);
    }
}
impl<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized> AllocWrite<K, V> for Internal {
    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize) {
        alloc_write::<K, V, Self>(new, k0, v0, l, r, n)
    }
    fn set_left_child(new: &mut MutPage, n: isize, l: u64) {
        if l > 0 {
            Internal::set_right_child(new, n - 1, l);
        }
    }
}
// Allocate a new entry and copy (k0, v0) into the new slot.
fn alloc_write<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized, L: Alloc>(
    new: &mut MutPage,
    k0: &K,
    v0: &V,
    l: u64,
    r: u64,
    n: &mut isize,
) {
    let size = alloc_size(k0, v0);
    let off_new = alloc_insert::<K, V, L>(new, n, size, r);
    unsafe {
        let new_ptr = new.0.data.add(off_new);
        k0.write_to_page(new_ptr);
        let ks = k0.size();
        let v_ptr = new_ptr.add((ks + V::ALIGN - 1) & !(V::ALIGN - 1));
        v0.write_to_page(v_ptr);
    }
    if l > 0 {
        L::set_right_child(new, *n - 1, l);
    }
    *n += 1;
}
/// Reserve space on the page for `size` bytes of data (i.e. not
/// counting the offset), set the right child of the new position
/// to `r`, add the offset to the new data to the offset array,
/// and return the new offset.
fn alloc_insert<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized, L: Alloc>(
    new: &mut MutPage,
    n: &mut isize,
    size: usize,
    r: u64,
) -> usize {
    let hdr = header_mut(new);
    let data = hdr.data();
    // If we're here, the following assertion has been checked by the
    // `can_alloc` method.
    debug_assert!(HDR + (hdr.n() as usize + 1) * L::OFFSET_SIZE + size <= data as usize);
    hdr.set_data(data - size as u16);
    hdr.set_n(hdr.n() + 1);
    hdr.incr(L::OFFSET_SIZE + size);
    let off_new = data - size as u16;
    let hdr_n = hdr.n();
    // Making space for the new offset
    unsafe {
        core::ptr::copy(
            new.0.data.add(HDR + (*n as usize) * L::OFFSET_SIZE),
            new.0
                .data
                .add(HDR + (*n as usize) * L::OFFSET_SIZE + L::OFFSET_SIZE),
            (hdr_n as usize - (*n as usize)) * L::OFFSET_SIZE,
        );
    }
    L::set_offset(new, *n, r | (off_new as u64));
    off_new as usize
}

//! Implementation of B tree pages for Sized types, i.e. types whose
//! representation has a size known at compile time (and the same as
//! [`core::mem::size_of()`]).
//!
//! The details of the implementation are as follows:
//!
//! - The page starts with a 16 bytes header of the following form
//! (where all the fields are encoded in Little-Endian):
//!
//!     ```
//!     #[repr(C)]
//!     pub struct Header {
//!         /// Offset to the first entry in the page, shifted 3 bits
//!         /// to the right to allow for the dirty bit (plus two
//!         /// extra bits, zero for now), as explained in the
//!         /// documentation of CowPage, at the root of this
//!         /// crate. This is 4096 for empty pages, and it is unused
//!         /// for leaves. Moreover, this field can't be increased:
//!         /// when it reaches the bottom, the page must be cloned.
//!         data: u16,
//!         /// Number of entries on the page.
//!         n: u16,
//!         /// CRC (if used).
//!         crc: u32,
//!         /// The 52 most significant bits are the left child of
//!         /// this page (0 for leaves), while the 12 LSBs represent
//!         /// the space this page would take when cloned from scratch,
//!         /// minus the header. The reason for this is that entries
//!         /// in internal nodes aren't really removed when deleted,
//!         /// they're only "unlinked" from the array of offsets (see
//!         /// below). Therefore, we must have a way to tell when a
//!         /// page can be "compacted" to reclaim space.
//!         left_page: u64,
//!     }
//!     ```
//! - For leaves, the rest of the page has the same representation as
//! an array of length `n`, of elements of type `Tuple<K, V>`:
//!   ```
//!   #[repr(C)]
//!   struct Tuple<K, V> {
//!       k: K,
//!       v: V,
//!   }
//!   ```
//!   If the alignment of that structure is more than 16 bytes, we
//!   need to add some padding between the header and that array.
//! - For internal nodes, the rest of the page starts with an array of
//! length `n` of Little-Endian-encoded [`u64`], where the 12 least
//! significant bits of each [`u64`] are an offset to a `Tuple<K, V>` in
//! the page, and the 52 other bits are an offset in the file to the
//! right child of the entry.
//!
//!   Moreover, the offset represented by the 12 LSBs is after (or at)
//!   `header.data`.
//!   We say we can "allocate" in the page if the `data` of the header
//!   is greater than or equal to the position of the last "offset",
//!   plus the size we want to allocate (note that since we allocate
//!   from the end of the page, `data` is always a multiple of the
//!   alignment of `Tuple<K, V>`).
use super::*;
use crate::btree::del::*;
use crate::btree::put::*;
use core::cmp::Ordering;
mod alloc; // The "alloc" trait, to provide common methods for leaves and internal nodes.
mod put; // Inserting a new value onto a page (possibly splitting the page).
mod rebalance; // Rebalance two sibling pages to the left or to the right.
use super::page_unsized::{cursor::PageCursor, header};
use alloc::*;
use header::*;
use rebalance::*;
/// This struct is used to allocate a pair `(K, V)` on the stack, and
/// copy it from a C pointer from a page.
///
/// This is also used to form slices of pairs in order to use
/// functions from the core library.
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
#[repr(C)]
struct Tuple<K, V> {
    k: K,
    v: V,
}
/// Empty type implementing `BTreePage` and `BTreeMutPage`.
#[derive(Debug)]
pub struct Page<K, V> {
    k: core::marker::PhantomData<K>,
    v: core::marker::PhantomData<V>,
}
#[async_trait(?Send)]
impl<K: Storable, V: Storable> super::BTreePage<K, V> for Page<K, V> {
    // Cursors are quite straightforward. One non-trivial thing is
    // that they represent both a position on a page and the interval
    // from that position to the end of the page, as is apparent in
    // their `split_at` method.
    //
    // Another thing to note is that -1 and `c.total` are valid
    // positions for a cursor `c`. The reason for this is that
    // position `-1` has a right child (which is the first element's
    // left child).
    type Cursor = PageCursor;
    fn is_empty(c: &Self::Cursor) -> bool {
        c.cur >= c.total as isize
    }
    fn is_init(c: &Self::Cursor) -> bool {
        c.cur < 0
    }
    fn cursor_before(p: &crate::CowPage) -> Self::Cursor {
        PageCursor::new(p, -1)
    }
    fn cursor_after(p: &crate::CowPage) -> Self::Cursor {
        PageCursor::after(p)
    }
    fn split_at(c: &Self::Cursor) -> (Self::Cursor, Self::Cursor) {
        (
            PageCursor {
                cur: 0,
                total: c.cur.max(0) as usize,
                is_leaf: c.is_leaf,
            },
            *c,
        )
    }
    fn move_next(c: &mut Self::Cursor) -> bool {
        if c.cur >= c.total as isize {
            return false;
        }
        c.cur += 1;
        true
    }
    fn move_prev(c: &mut Self::Cursor) -> bool {
        if c.cur < 0 {
            return false;
        }
        c.cur -= 1;
        true
    }
    fn current<'a>(
        page: crate::Page<'a>,
        c: &Self::Cursor,
    ) -> Option<(&'a K, &'a V, u64)> {
        // First, there's no current entry if the cursor is outside
        // the range of entries.
        if c.cur < 0 || c.cur >= c.total as isize {
            None
        } else if c.is_leaf {
            // Else, if this is a leaf, the elements are packed
            // together in a contiguous array.
            //
            // This means that the header may be followed by padding
            // (in order to align the entries). These are constants
            // known at compile-time, so `al` and `hdr` are optimised
            // away by the compiler.
            let al = core::mem::align_of::<Tuple<K, V>>();
            // The following is a way to compute the first multiple of
            // `al` after `HDR`, assuming `al` is a power of 2 (which
            // is always the case since we get it from `align_of`).
            let hdr = (HDR + al - 1) & !(al - 1);
            // The position of the `Tuple<K, V>` we're looking for is
            // `f * cur` bytes after the padded header (because
            // `size_of` includes alignment padding).
            let f = core::mem::size_of::<Tuple<K, V>>();
            let kv = unsafe {
                &*(page.data.as_ptr().add(hdr + c.cur as usize * f) as *const Tuple<K, V>)
            };
            Some((&kv.k, &kv.v, 0))
        } else {
            // Internal nodes have an extra level of indirection: we
            // first need to find `off`, the offset in the page, in
            // the initial array of offsets. Since these offsets are
            // `u64`, and the header is of size 16 bytes, the array is
            // already aligned.
            unsafe {
                let off =
                    u64::from_le(*(page.data.as_ptr().add(HDR + 8 * c.cur as usize) as *const u64));
                // Check that we aren't reading outside of the page
                // (for example because of a malformed offset).
                assert!((off as usize & 0xfff) + core::mem::size_of::<Tuple<K, V>>() <= 4096);
                // Once we have the offset, cast its 12 LSBs to a
                // position in the page, and read the `Tuple<K, V>` at
                // that position.
                let kv = &*(page.data.as_ptr().add((off as usize) & 0xfff) as *const Tuple<K, V>);
                Some((&kv.k, &kv.v, off & !0xfff))
            }
        }
    }
    // The left and right child methods aren't really surprising. One
    // thing to note is that cursors are always in positions between
    // `-1` and `c.total` (bounds included), so we only have to check
    // one side of the bound in the assertions.
    //
    // We also check, before entering the `unsafe` sections, that
    // we're only reading data that is on a page.
    fn left_child(page: crate::Page, c: &Self::Cursor) -> u64 {
        if c.is_leaf {
            0
        } else {
            assert!(c.cur >= 0 && HDR as isize + c.cur * 8 - 8 <= 4088);
            let off =
                unsafe { *(page.data.as_ptr().offset((HDR as isize + c.cur * 8) - 8) as *const u64) };
            u64::from_le(off) & !0xfff
        }
    }
    fn right_child(page: crate::Page, c: &Self::Cursor) -> u64 {
        if c.is_leaf {
            0
        } else {
            assert!(c.cur < c.total as isize && HDR as isize + c.cur * 8 <= 4088);
            let off =
                unsafe { *(page.data.as_ptr().offset(HDR as isize + c.cur * 8) as *const u64) };
            u64::from_le(off) & !0xfff
        }
    }
    async fn set_cursor<'a, 'b, T: LoadPage>(
        _txn: &'a T,
        page: crate::Page<'b>,
        c: &mut PageCursor,
        k0: &K,
        v0: Option<&V>,
    ) -> Result<(&'a K, &'a V, u64), usize> {
        unsafe {
            // `lookup` has the same semantic as
            // `core::slice::binary_search`, i.e. it returns either
            // `Ok(n)`, where `n` is the position where `(k0, v0)` was
            // found, or `Err(n)` where `n` is the position where
            // `(k0, v0)` can be inserted to preserve the order.
            match lookup(page, c, k0, v0).await {
                Ok(n) => {
                    c.cur = n as isize;
                    // Just read the tuple and return it.
                    if c.is_leaf {
                        let f = core::mem::size_of::<Tuple<K, V>>();
                        let al = core::mem::align_of::<Tuple<K, V>>();
                        let hdr_size = (HDR + al - 1) & !(al - 1);
                        let tup =
                            &*(page.data.as_ptr().add(hdr_size + f * n) as *const Tuple<K, V>);
                        Ok((&tup.k, &tup.v, 0))
                    } else {
                        let off =
                            u64::from_le(*(page.data.as_ptr().add(HDR + n * 8) as *const u64));
                        let tup =
                            &*(page.data.as_ptr().add(off as usize & 0xfff) as *const Tuple<K, V>);
                        Ok((&tup.k, &tup.v, off & !0xfff))
                    }
                }
                Err(n) => {
                    c.cur = n as isize;
                    Err(n)
                }
            }
        }
    }
}
#[async_trait(?Send)]
impl<K: Storable + core::fmt::Debug, V: Storable + core::fmt::Debug> super::BTreeMutPage<K, V>
    for Page<K, V>
{
    // Once again, this is quite straightforward.
    fn init(page: &mut MutPage) {
        let h = header_mut(page);
        h.init();
    }
    // When deleting from internal nodes, we take a replacement from
    // one of the leaves (in our current implementation, the leftmost
    // entry of the right subtree). This method copies an entry from
    // the leaf onto the program stack, which is necessary since
    // deletions in leaves overwrites entries.
    //
    // Another design choice would have been to do the same as for the
    // unsized implementation, but in this case this would have meant
    // copying the saved value to the end of the leaf, potentially
    // preventing merges, and not even saving a memory copy.
    type Saved = (K, V);
    fn save_deleted_leaf_entry(k: &K, v: &V) -> Self::Saved {
        unsafe {
            let mut k0 = core::mem::MaybeUninit::uninit();
            let mut v0 = core::mem::MaybeUninit::uninit();
            core::ptr::copy_nonoverlapping(k, k0.as_mut_ptr(), 1);
            core::ptr::copy_nonoverlapping(v, v0.as_mut_ptr(), 1);
            (k0.assume_init(), v0.assume_init())
        }
    }
    unsafe fn from_saved<'a>(s: &Self::Saved) -> (&'a K, &'a V) {
        (core::mem::transmute(&s.0), core::mem::transmute(&s.1))
    }
    // `put` inserts one or two entries onto a node (internal or
    // leaf). This is implemented in the `put` module.
    async fn put<'a, T: AllocPage>(
        txn: &mut T,
        page: CowPage,
        mutable: bool,
        replace: bool,
        c: &PageCursor,
        k0: &'a K,
        v0: &'a V,
        k1v1: Option<(&'a K, &'a V)>,
        l: u64,
        r: u64,
    ) -> Result<super::put::Put<'a, K, V>, T::Error> {
        assert!(c.cur >= 0);
        // In the sized case, deletions can never cause a split, so we
        // never have to insert two elements at the same position.
        assert!(k1v1.is_none());
        if r == 0 {
            put::put::<_, _, _, Leaf>(txn, page, mutable, replace, c.cur as usize, k0, v0, 0, 0).await
        } else {
            put::put::<_, _, _, Internal>(txn, page, mutable, replace, c.cur as usize, k0, v0, l, r).await
        }
    }
    unsafe fn put_mut(page: &mut MutPage, c: &mut Self::Cursor, k0: &K, v0: &V, r: u64) {
        use super::page_unsized::AllocWrite;
        let mut n = c.cur;
        if r == 0 {
            Leaf::alloc_write(page, k0, v0, 0, r, &mut n);
        } else {
            Internal::alloc_write(page, k0, v0, 0, r, &mut n);
        }
        c.total += 1;
    }
    unsafe fn set_left_child(page: &mut MutPage, c: &Self::Cursor, l: u64) {
        let off = (page.0.data.add(HDR) as *mut u64).offset(c.cur - 1);
        *off = (l | (u64::from_le(*off) & 0xfff)).to_le();
    }
    // This function updates an internal node, setting the left child
    // of the cursor to `l`.
    async fn update_left_child<T: AllocPage>(
        txn: &mut T,
        page: CowPage,
        mutable: bool,
        c: &Self::Cursor,
        l: u64,
    ) -> Result<crate::btree::put::Ok, T::Error> {
        assert!(!c.is_leaf && c.cur >= 0 && (c.cur as usize) <= c.total);
        let freed;
        let page = if mutable && page.is_dirty() {
            // If the page is mutable (dirty), just convert it to a
            // mutable page, and update.
            freed = 0;
            MutPage(page)
        } else {
            // Else, clone the page.
            let mut new = txn.alloc_page().await?;
            <Page<K, V> as BTreeMutPage<K, V>>::init(&mut new);
            // First clone the left child of the page.
            let l = header(page.as_page()).left_page() & !0xfff;
            let hdr = header_mut(&mut new);
            hdr.set_left_page(l);
            // And then the rest of the page.
            let s = Internal::offset_slice::<T, K, V>(page.as_page());
            clone::<K, V, Internal>(page.as_page(), &mut new, s, &mut 0);
            // Mark the former version of the page as free.
            freed = page.offset | if page.is_dirty() { 1 } else { 0 };
            new
        };
        // Finally, update the left child of the cursor.
        unsafe {
            let off = (page.0.data.add(HDR) as *mut u64).offset(c.cur - 1);
            *off = (l | (u64::from_le(*off) & 0xfff)).to_le();
        }
        Ok(Ok { page, freed })
    }
    async fn del<T: AllocPage>(
        txn: &mut T,
        page: crate::CowPage,
        mutable: bool,
        c: &PageCursor,
        l: u64,
    ) -> Result<(MutPage, u64), T::Error> {
        assert!(c.cur >= 0 && (c.cur as usize) < c.total);
        if mutable && page.is_dirty() {
            // In the mutable case, we just need to move some memory
            // around.
            let p = page.data;
            let mut page = MutPage(page);
            let hdr = header_mut(&mut page);
            let f = core::mem::size_of::<Tuple<K, V>>();
            if c.is_leaf {
                // In leaves, we need to move the n - c - 1 elements
                // that are strictly after the cursor, by `f` (the
                // size of an entry).
                //
                // Here's the reasoning to avoid off-by-one errors: if
                // `c` is 0 (i.e. we're deleting the first element on
                // the page), we remove one entry, so there are n - 1
                // remaining entries.
                let n = hdr.n() as usize;
                let hdr_size = {
                    // As usual, header + padding
                    let al = core::mem::align_of::<Tuple<K, V>>();
                    (HDR + al - 1) & !(al - 1)
                };
                let off = hdr_size + c.cur as usize * f;
                unsafe {
                    core::ptr::copy(p.add(off + f), p.add(off), f * (n - c.cur as usize - 1));
                }
                // Removing `f` bytes from the page.
                hdr.decr(f);
            } else {
                // Internal nodes are easier to deal with, as we just
                // have to move the offsets.
                unsafe {
                    let ptr = p.add(HDR + c.cur as usize * 8) as *mut u64;
                    core::ptr::copy(ptr.offset(1), ptr, hdr.n() as usize - c.cur as usize - 1);
                }
                // Removing `f` bytes from the page (the tuple), plus
                // one 8-bytes offset.
                hdr.decr(f + 8);
                // Updating the left page if necessary.
                if l > 0 {
                    unsafe {
                        let off = (p.add(HDR) as *mut u64).offset(c.cur as isize - 1);
                        *off = (l | (u64::from_le(*off) & 0xfff)).to_le();
                    }
                }
            }
            hdr.set_n(hdr.n() - 1);
            // Returning the mutable page itself, and 0 (for "no page freed")
            Ok((page, 0))
        } else {
            // Immutable pages need to be cloned. The strategy is the
            // same in both cases: get an "offset slice", split it at
            // the cursor, remove the first entry of the second half,
            // and clone.
            let mut new = txn.alloc_page().await?;
            <Page<K, V> as BTreeMutPage<K, V>>::init(&mut new);
            if c.is_leaf {
                let s = Leaf::offset_slice::<T, K, V>(page.as_page());
                let (s0, s1) = s.split_at(c.cur as usize);
                let (_, s1) = s1.split_at(1);
                let mut n = 0;
                clone::<K, V, Leaf>(page.as_page(), &mut new, s0, &mut n);
                clone::<K, V, Leaf>(page.as_page(), &mut new, s1, &mut n);
            } else {
                // Internal nodes a bit trickier, since the left child
                // is not counted in the "offset slice", so we need to
                // clone it separately. Also, the `l` argument to this
                // function might be non-zero in this case.
                let s = Internal::offset_slice::<T, K, V>(page.as_page());
                let (s0, s1) = s.split_at(c.cur as usize);
                let (_, s1) = s1.split_at(1);
                // First, clone the left child of the page.
                let hdr = header(page.as_page());
                let left = hdr.left_page() & !0xfff;
                unsafe { *(new.0.data.add(HDR - 8) as *mut u64) = left.to_le() };
                // Then, clone the entries strictly before the cursor
                // (i.e. clone `s0`).
                let mut n = 0;
                clone::<K, V, Internal>(page.as_page(), &mut new, s0, &mut n);
                // If we need to update the left child of the deleted
                // item, do it.
                if l > 0 {
                    unsafe {
                        let off = new.0.data.offset(HDR as isize + (n - 1) * 8) as *mut u64;
                        *off = (l | (u64::from_le(*off) & 0xfff)).to_le();
                    }
                }
                // Finally, clone the right half of the page (`s1`).
                clone::<K, V, Internal>(page.as_page(), &mut new, s1, &mut n);
            }
            Ok((new, page.offset))
        }
    }
    // Decide what to do with the concatenation of two neighbouring
    // pages, with a middle element.
    async fn merge_or_rebalance<'a, T: AllocPage>(
        txn: &mut T,
        m: Concat<'a, K, V, Self>,
    ) -> Result<Op<'a, T, K, V>, T::Error> {
        // First evaluate the size of the middle element on a page.
        let (hdr_size, mid_size) = if m.modified.c0.is_leaf {
            let al = core::mem::align_of::<Tuple<K, V>>();
            (
                (HDR + al - 1) & !(al - 1),
                core::mem::size_of::<Tuple<K, V>>(),
            )
        } else {
            (HDR, 8 + core::mem::size_of::<Tuple<K, V>>())
        };
        // Evaluate the size of the modified half of the concatenation
        // (which includes the header).
        let mod_size = size::<K, V>(&m.modified);
        // Add the "occupied" size (which excludes the header).
        let occupied = {
            let hdr = header(m.other.as_page());
            (hdr.left_page() & 0xfff) as usize
        };
        // One surprising observation here is that adding the sizes
        // works. This is surprising because of alignment and
        // padding. It works because we can split the sizes into an
        // offset part (always 8 bytes) and a data part (of a constant
        // alignment), and thus we never need any padding anywhere on
        // the page.
        if mod_size + mid_size + occupied <= PAGE_SIZE {
            // If the concatenation fits on a page, merge.
            return if m.modified.c0.is_leaf {
                super::page_unsized::merge::<_, _, _, _, Leaf>(txn, m).await
            } else {
                super::page_unsized::merge::<_, _, _, _, Internal>(txn, m).await
            };
        }
        // If the modified page is large enough, or if the other page
        // has only just barely the minimum number of elements to be
        // at least half-full, return.
        //
        // The situation where the other page isn't full enough might
        // happen for example if elements occupy exactly 1/5th of a
        // page + 1 byte, and the modified page has 2 of them after
        // the deletion, while the other page has 3.
        if mod_size >= PAGE_SIZE / 2 || hdr_size + occupied - mid_size < PAGE_SIZE / 2 {
            if let Some((k, v)) = m.modified.ins {
                // Perform the required modification and return.
                return Ok(Op::Put(Self::put(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    m.modified.skip_first,
                    &m.modified.c1,
                    k,
                    v,
                    m.modified.ins2,
                    m.modified.l,
                    m.modified.r,
                ).await?));
            } else if m.modified.skip_first {
                // `ins2` is only ever used when the page immediately
                // below a deletion inside an internal node has split,
                // and we need to replace the deleted value, *and*
                // insert the middle entry of the split.
                debug_assert!(m.modified.ins2.is_none());
                let (page, freed) = Self::del(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    &m.modified.c1,
                    m.modified.l,
                ).await?;
                return Ok(Op::Put(Put::Ok(Ok { page, freed })));
            } else {
                let mut c1 = m.modified.c1.clone();
                let mut l = m.modified.l;
                if l == 0 {
                    Self::move_next(&mut c1);
                    l = m.modified.r;
                }
                return Ok(Op::Put(Put::Ok(Self::update_left_child(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    &c1,
                    l,
                ).await?)));
            }
        }
        // Finally, if we're here, we can rebalance. There are four
        // (relatively explicit) cases, see the `rebalance` submodule
        // to see how this is done.
        if m.mod_is_left {
            if m.modified.c0.is_leaf {
                rebalance_left::<_, _, _, Leaf>(txn, m).await
            } else {
                rebalance_left::<_, _, _, Internal>(txn, m).await
            }
        } else {
            super::page_unsized::rebalance::rebalance_right::<_, _, _, Self>(txn, m).await
        }
    }
}
/// Size of a modified page (including the header).
fn size<K: Storable, V: Storable>(m: &ModifiedPage<K, V, Page<K, V>>) -> usize {
    let mut total = {
        let hdr = header(m.page.as_page());
        (hdr.left_page() & 0xfff) as usize
    };
    if m.c1.is_leaf {
        let al = core::mem::align_of::<Tuple<K, V>>();
        total += (HDR + al - 1) & (!al - 1);
    } else {
        total += HDR
    };
    // Extra size for the offsets.
    let extra = if m.c1.is_leaf { 0 } else { 8 };
    if m.ins.is_some() {
        total += extra + core::mem::size_of::<Tuple<K, V>>();
        if m.ins2.is_some() {
            total += extra + core::mem::size_of::<Tuple<K, V>>()
        }
    }
    // If we skip the first entry of `m.c1`, remove its size.
    if m.skip_first {
        total -= extra + core::mem::size_of::<Tuple<K, V>>()
    }
    total
}
/// Linear search, leaf version. This is relatively straightforward.
fn leaf_linear_search<K: Storable, V: Storable>(
    k0: &K,
    v0: Option<&V>,
    s: &[Tuple<K, V>],
) -> Result<usize, usize> {
    let mut n = 0;
    for sm in s.iter() {
        match sm.k.compare(k0) {
            Ordering::Less => n += 1,
            Ordering::Greater => return Err(n),
            Ordering::Equal => {
                if let Some(v0) = v0 {
                    match sm.v.compare(v0) {
                        Ordering::Less => n += 1,
                        Ordering::Greater => return Err(n),
                        Ordering::Equal => return Ok(n),
                    }
                } else {
                    return Ok(n);
                }
            }
        }
    }
    Err(n)
}
/// An equivalent of `Ord::cmp` on `Tuple<K, V>` but using
/// `Storable::compare` instead of `Ord::cmp` on `K` and `V`.
fn cmp<K: Storable, V: Storable>(
    k0: &K,
    v0: Option<&V>,
    p: &[u8; 4096],
    off: u64,
) -> Ordering {
    let off = u64::from_le(off) & 0xfff;
    if off as usize + core::mem::size_of::<Tuple<K, V>>() > PAGE_SIZE {
        panic!(
            "off = {:?}, size = {:?}",
            off,
            core::mem::size_of::<Tuple<K, V>>()
        );
    }
    let tup = unsafe { &*(p.as_ptr().offset(off as isize & 0xfff) as *const Tuple<K, V>) };
    match tup.k.compare(k0) {
        Ordering::Equal => {
            if let Some(v0) = v0 {
                tup.v.compare(v0)
            } else {
                Ordering::Equal
            }
        }
        o => o,
    }
}
/// Linear search for internal nodes. Does what it says.
unsafe fn internal_search<K: Storable, V: Storable>(
    k0: &K,
    v0: Option<&V>,
    s: &[u64],
    p: &[u8; 4096],
) -> Result<usize, usize> {
    for (n, off) in s.iter().enumerate() {
        match cmp(k0, v0, p, *off) {
            Ordering::Less => {}
            Ordering::Greater => return Err(n),
            Ordering::Equal => return Ok(n),
        }
    }
    Err(s.len())
}
/// Lookup just forms slices of offsets (for internal nodes) or values
/// (for leaves), and runs an internal search on them.
async unsafe fn lookup<'a, K: Storable, V: Storable>(
    page: crate::Page<'a>,
    c: &mut PageCursor,
    k0: &K,
    v0: Option<&V>,
) -> Result<usize, usize> {
    let hdr = header(page);
    c.total = hdr.n() as usize;
    c.is_leaf = hdr.is_leaf();
    if c.is_leaf {
        let al = core::mem::align_of::<Tuple<K, V>>();
        let hdr_size = (HDR + al - 1) & !(al - 1);
        let s = core::slice::from_raw_parts(
            page.data.as_ptr().add(hdr_size) as *const Tuple<K, V>,
            hdr.n() as usize,
        );
        leaf_linear_search(k0, v0, s)
    } else {
        let s = core::slice::from_raw_parts(
            page.data.as_ptr().add(HDR) as *const u64,
            hdr.n() as usize,
        );
        internal_search(k0, v0, s, page.data)
    }
}
/// Clone a slice of offsets onto a page. This essentially does what
/// it says. Note that the leftmost child of a page is not included in
/// the offset slices, so we don't have to handle it here.
///
/// This should really be in the `Alloc` trait, but Rust doesn't have
/// associated type constructors, so we can't have an associated type
/// that's sometimes a slice and sometimes a "Range".
fn clone<K: Storable, V: Storable, L: Alloc>(
    page: crate::Page,
    new: &mut MutPage,
    s: Offsets,
    n: &mut isize,
) {
    match s {
        Offsets::Slice(s) => {
            // We know we're in an internal node here.
            let size = core::mem::size_of::<Tuple<K, V>>();
            for off in s.iter() {
                let off = u64::from_le(*off);
                let r = off & !0xfff;
                let off = off & 0xfff;
                unsafe {
                    let ptr = page.data.as_ptr().add(off as usize);
                    // Reserve the space on the page
                    let hdr = header_mut(new);
                    let data = hdr.data() as u16;
                    let off_new = data - size as u16;
                    hdr.set_data(off_new);
                    // Copy the entry from the original page to its
                    // position on the new page.
                    core::ptr::copy_nonoverlapping(ptr, new.0.data.add(off_new as usize), size);
                    // Set the offset to this new entry in the offset
                    // array, along with the right child page.
                    let ptr = new.0.data.offset(HDR as isize + *n * 8) as *mut u64;
                    *ptr = (r | off_new as u64).to_le();
                }
                *n += 1;
            }
            let hdr = header_mut(new);
            hdr.set_n(hdr.n() + s.len() as u16);
            hdr.incr((8 + size) * s.len());
        }
        Offsets::Range(r) => {
            let size = core::mem::size_of::<Tuple<K, V>>();
            let a = core::mem::align_of::<Tuple<K, V>>();
            let header_size = (HDR + a - 1) & !(a - 1);
            let len = r.len();
            for off in r {
                unsafe {
                    let ptr = page.data.as_ptr().add(header_size + off * size);
                    let new_ptr = new.0.data.add(header_size + (*n as usize) * size);
                    core::ptr::copy_nonoverlapping(ptr, new_ptr, size);
                }
                *n += 1;
            }
            // On leaves, we do have to update everything manually,
            // because we're simply copying stuff.
            let hdr = header_mut(new);
            hdr.set_n(hdr.n() + len as u16);
            hdr.incr(size * len);
        }
    }
}

use super::*;
use core::mem::MaybeUninit;
pub(crate) async fn rebalance_left<
    'a,
    T: AllocPage,
    K: Storable + core::fmt::Debug,
    V: Storable + core::fmt::Debug,
    L: Alloc,
>(
    txn: &mut T,
    m: Concat<'a, K, V, Page<K, V>>,
) -> Result<Op<'a, T, K, V>, T::Error> {
    assert!(m.mod_is_left);
    // First element of the right page. We'll rotate it to the left
    // page, i.e. return a middle element, and append the current
    // middle element to the left page.
    let rc = super::PageCursor::new(&m.other, 0);
    let rl = <Page<K, V>>::left_child(m.other.as_page(), &rc);
    let (k, v, r) = <Page<K, V>>::current(m.other.as_page(), &rc).unwrap();
    let mut freed_ = [0, 0];
    // First, perform the modification on the modified page, which we
    // know is the left page, and return the resulting mutable page
    // (the modified page may or may not be mutable before we do this).
    let new_left = if let Some((k, v)) = m.modified.ins {
        let is_dirty = m.modified.page.is_dirty();
        let page = if let Put::Ok(Ok { page, freed }) = <Page<K, V>>::put(
            txn,
            m.modified.page,
            m.modified.mutable,
            m.modified.skip_first,
            &m.modified.c1,
            k,
            v,
            m.modified.ins2,
            m.modified.l,
            m.modified.r,
        ).await? {
            if freed > 0 {
                let b = if is_dirty { 1 } else { 0 };
                freed_[0] = freed | b;
            }
            page
        } else {
            unreachable!()
        };
        // Append the middle element of the concatenation at the end
        // of the left page. We know the left page to be mutable by
        // now, and we also know there's enough space to do this.
        let lc = PageCursor::after(&page.0);
        if let Put::Ok(Ok { page, freed }) = <Page<K, V>>::put(
            txn, page.0, true, false, &lc, m.mid.0, m.mid.1, None, 0, rl,
        ).await? {
            if freed > 0 {
                let b = if is_dirty { 1 } else { 0 };
                // index 0 is ok here: if a page is freed, it means we
                // just compacted a page, which implies that the call
                // to `put` above didn't free.
                freed_[0] = freed | b
            }
            page
        } else {
            unreachable!()
        }
    } else if m.modified.skip_first {
        // Looking at module `super::del`, the only way we can be in
        // this case
        let is_dirty = m.modified.page.is_dirty();
        let (page, freed) = <Page<K, V>>::del(
            txn,
            m.modified.page,
            m.modified.mutable,
            &m.modified.c1,
            m.modified.l,
        ).await?;
        if freed > 0 {
            let b = if is_dirty { 1 } else { 0 };
            freed_[0] = freed | b
        }
        // Append the middle element of the concatenation at the end
        // of the left page. We know the left page to be mutable by
        // now, and we also know there's enough space to do this.
        let lc = PageCursor::after(&page.0);
        if let Put::Ok(Ok { page, freed }) = <Page<K, V>>::put(
            txn, page.0, true, false, &lc, m.mid.0, m.mid.1, None, 0, rl,
        ).await? {
            if freed > 0 {
                let b = if is_dirty { 1 } else { 0 };
                // index 0 is ok here: if we freed a page in the call
                // to `del` above, it is mutable and we can allocate
                // on it. Else, we just compacted a page, but that
                // also means `del` above didn't free.
                freed_[0] = freed | b
            }
            page
        } else {
            unreachable!()
        }
    } else {
        let is_dirty = m.modified.page.is_dirty();
        let lc = PageCursor::after(&m.modified.page);
        if let Put::Ok(Ok { page, freed }) = <Page<K, V>>::put(
            txn, m.modified.page, m.modified.mutable, false, &lc, m.mid.0, m.mid.1, None, 0, rl,
        ).await? {
            if m.modified.l > 0 {
                assert_eq!(m.modified.r, 0);
                unsafe {
                    let off = (page.0.data.add(HDR) as *mut u64).offset(m.modified.c1.cur - 1);
                    *off = (m.modified.l | (u64::from_le(*off) & 0xfff)).to_le();
                }
            } else if m.modified.r > 0 {
                unsafe {
                    let off = (page.0.data.add(HDR) as *mut u64).offset(m.modified.c1.cur);
                    *off = (m.modified.r | (u64::from_le(*off) & 0xfff)).to_le();
                }
            }
            if freed > 0 {
                let b = if is_dirty { 1 } else { 0 };
                freed_[0] = freed | b
            }
            page
        } else {
            unreachable!()
        }
    };
    let f = core::mem::size_of::<Tuple<K, V>>();
    let (new_right, k, v) = if r == 0 && m.other_is_mutable && m.other.is_dirty() {
        // Since `r == 0`, we know this is a leaf. Therefore, we need
        // to rewrite the deleted element to the end of the page, so
        // that the pointer to the new middle entry is valid when this
        // function returns.
        let data = m.other.data;
        let mut other = MutPage(m.other);
        let right_hdr = header_mut(&mut other);
        // Remove the space for one entry.
        let n = right_hdr.n() as usize;
        right_hdr.decr(f);
        right_hdr.set_n((n - 1) as u16);
        let al = core::mem::align_of::<Tuple<K, V>>();
        let hdr_size = (HDR + al - 1) & !(al - 1);
        let mut t: MaybeUninit<Tuple<K, V>> = MaybeUninit::uninit();
        unsafe {
            // Save the leftmost element (we can't directly copy it to
            // the end, because the page may be full).
            core::ptr::copy_nonoverlapping(
                data.add(hdr_size) as *mut Tuple<K, V>,
                t.as_mut_ptr(),
                1,
            );
            // Move the n - 1 last elements one entry to the left.
            core::ptr::copy(
                data.add(hdr_size + f) as *const Tuple<K, V>,
                data.add(hdr_size) as *mut Tuple<K, V>,
                n - 1,
            );
            // Copy the last element to the end of the page.
            let last = data.add(hdr_size + f * (n - 1)) as *mut Tuple<K, V>;
            core::ptr::copy_nonoverlapping(t.as_ptr(), last, 1);
            (other, &(&*last).k, &(&*last).v)
        }
    } else {
        // If this isn't a leaf, or isn't mutable, use the standard
        // deletion function, because we know this will leave the
        // pointer to the current entry valid.
        // We extend the pointer's lifetime here, because we know the
        // current deletion (we only rebalance during deletions) won't
        // touch this page anymore after this.
        let k = unsafe { core::mem::transmute(k) };
        let v = unsafe { core::mem::transmute(v) };
        // If this frees the old "other" page, add it to the "freed"
        // array.
        let is_dirty = m.other.is_dirty();
        let (page, freed) = <Page<K, V>>::del(txn, m.other, m.other_is_mutable, &rc, r).await?;
        if freed > 0 {
            freed_[1] = if is_dirty { freed | 1 } else { freed };
        }
        (page, k, v)
    };
    Ok(Op::Rebalanced {
        l: new_left.0.offset,
        r: new_right.0.offset,
        k,
        v,
        freed: freed_,
    })
}

use super::*;
pub(super) async fn put<
    'a,
    T: AllocPage,
    K: Storable + core::fmt::Debug,
    V: Storable + core::fmt::Debug,
    L: Alloc + super::super::page_unsized::AllocWrite<K, V>,
>(
    txn: &mut T,
    page: CowPage,
    mutable: bool,
    replace: bool,
    u: usize,
    k0: &'a K,
    v0: &'a V,
    l: u64,
    r: u64,
) -> Result<crate::btree::put::Put<'a, K, V>, T::Error> {
    let hdr = header(page.as_page());
    let is_dirty = page.is_dirty();
    if mutable && is_dirty && L::can_alloc::<K, V>(hdr) {
        // If the page is mutable (i.e. not shared with another tree)
        // and dirty, here's what we do:
        let mut page = MutPage(page);
        let p = page.0.data;
        let hdr = header_mut(&mut page);
        let mut n = u as isize;
        if replace {
            // If we're replacing a value, this can't be in a leaf,
            // hence the only thing that needs to be done is erasing
            // the offset in the offset array. This is a bit naive,
            // since we're moving that array back and forth.
            assert!(!hdr.is_leaf());
            unsafe {
                let ptr = p.add(HDR + n as usize * 8) as *mut u64;
                core::ptr::copy(ptr.offset(1), ptr, hdr.n() as usize - n as usize - 1);
                hdr.decr(8 + core::mem::size_of::<Tuple<K, V>>());
                hdr.set_n(hdr.n() - 1);
            }
        }
        // Do the actual insertions.
        L::alloc_write(&mut page, k0, v0, l, r, &mut n);
        // No page was freed, return the page.
        Ok(Put::Ok(Ok { page, freed: 0 }))
    } else if replace || L::can_compact::<K, V>(hdr) {
        // Here, we could allocate if we cloned, so let's clone:
        let mut new = txn.alloc_page().await?;
        <Page<K, V> as BTreeMutPage<K, V>>::init(&mut new);
        L::set_right_child(&mut new, -1, hdr.left_page() & !0xfff);
        // Take the offsets and split it at the cursor position.
        let s = L::offset_slice::<T, K, V>(page.as_page());
        let (s0, s1) = s.split_at(u as usize);
        // If we're replacing, remove the offset that needs to go.
        let s1 = if replace { s1.split_at(1).1 } else { s1 };
        // And then clone the page, inserting the new elements between
        // the two halves of the split offsets.
        let mut n = 0;
        clone::<K, V, L>(page.as_page(), &mut new, s0, &mut n);
        L::alloc_write(&mut new, k0, v0, l, r, &mut n);
        clone::<K, V, L>(page.as_page(), &mut new, s1, &mut n);
        Ok(Put::Ok(Ok {
            page: new,
            freed: if is_dirty {
                page.offset | 1
            } else {
                page.offset
            },
        }))
    } else {
        // Else, split the page.
        split::<_, _, _, L>(txn, page, mutable, u, k0, v0, l, r).await
    }
}
async fn split<
    'a,
    T: AllocPage,
    K: Storable + core::fmt::Debug,
    V: Storable + core::fmt::Debug,
    L: Alloc + super::super::page_unsized::AllocWrite<K, V>,
>(
    txn: &mut T,
    page: CowPage,
    mutable: bool,
    u: usize,
    k0: &'a K,
    v0: &'a V,
    l: u64,
    r: u64,
) -> Result<crate::btree::put::Put<'a, K, V>, T::Error> {
    let mut left;
    let hdr = header(page.as_page());
    let page_is_dirty = if page.is_dirty() { 1 } else { 0 };
    let mut n = hdr.n();
    let k = n / 2;
    n += 1;
    let s = L::offset_slice::<T, K, V>(page.as_page());
    let (s0, s1) = s.split_at(k as usize);
    // Find the split entry (i.e. the entry going up in the B
    // tree). Then, mid_child is the right child of the split
    // key/value, and `s1` is the offsets of the right side of the
    // split.
    let (mut split_key, mut split_value, mid_child, s1) = if u == k as usize {
        // The inserted element is exactly in the middle.
        (k0, v0, r, s1)
    } else {
        // Else, the split key is the first element of `s1`: set the
        // new, updated `s1`.
        let (s1a, s1b) = s1.split_at(1);
        let (k, v, r) = L::kv(page.as_page(), L::first::<K, V>(&s1a));
        (k, v, r, s1b)
    };
    let mut freed = 0;
    if u >= k as usize {
        // If we are here, u >= k, i.e. the insertion is in the
        // right-hand side of the split, or is the split entry itself.
        let mut right = txn.alloc_page().await?;
        <Page<K, V> as BTreeMutPage<K, V>>::init(&mut right);
        if u > k as usize {
            // if we're inserting in the right-hand side, clone to the
            // newly allocated `right` page, by
            let mut n = 0;
            // Splitting the offsets to make space for an insertion,
            // and insert in the middle.
            //
            // Off-by-one error? Nope: since `k` is the split entry,
            // the right page starts at index `k + 1`, hence if `u ==
            // k + 1`, `kk` must be 0.
            let kk = u as usize - k as usize - 1;
            let (s1a, s1b) = s1.split_at(kk);
            L::set_right_child(&mut right, -1, mid_child);
            clone::<K, V, L>(page.as_page(), &mut right, s1a, &mut n);
            L::alloc_write(&mut right, k0, v0, l, r, &mut n);
            clone::<K, V, L>(page.as_page(), &mut right, s1b, &mut n);
        } else {
            // Else, just clone the page, we'll take care of returning
            // the split entry afterwards.
            clone::<K, V, L>(page.as_page(), &mut right, s1, &mut 0);
        }
        if mutable && page.is_dirty() {
            // (k0, v0) is to be inserted on the right-hand side of
            // the split, hence we don't have to clone the left-hand
            // side, we can just truncate it.
            left = MutPage(page);
            let hdr = header_mut(&mut left);
            hdr.set_n(k);
            hdr.decr((n - 1 - k) as usize * core::mem::size_of::<Tuple<K, V>>());
        } else {
            // Else, we need to clone the first `k-1` entries,
            // i.e. `s0`, onto a new left page.
            left = txn.alloc_page().await?;
            <Page<K, V> as BTreeMutPage<K, V>>::init(&mut left);
            let mut n = 0;
            L::set_right_child(&mut left, -1, hdr.left_page() & !0xfff);
            clone::<K, V, L>(page.as_page(), &mut left, s0, &mut n);
            freed = page.offset | page_is_dirty
        }
        // Finally, if `u` is the middle element, its left and right
        // children become the leftmost child of the right page, and
        // the rightmost child of the left page, respectively.
        if u == k as usize {
            // Insertion in the middle:
            // - `l` becomes the right child of the last element on `left`.
            L::set_right_child(&mut left, k as isize - 1, l);
            // - `r` (i.e. `mid_child`) becomes the left child of `right`.
            L::set_right_child(&mut right, -1, mid_child);
        }
        Ok(Put::Split {
            split_key,
            split_value,
            left,
            right,
            freed,
        })
    } else {
        // Else, the insertion is in the new left page. We first clone
        // the left page, inserting (k,v) at its position:
        left = txn.alloc_page().await?;
        <Page<K, V> as BTreeMutPage<K, V>>::init(&mut left);
        // Clone the leftmost page
        let ll = header(page.as_page()).left_page() & !0xfff;
        L::set_right_child(&mut left, -1, ll);
        // Clone the two sides, with an entry in between.
        let (s0a, s0b) = s0.split_at(u as usize);
        let mut n = 0;
        clone::<K, V, L>(page.as_page(), &mut left, s0a, &mut n);
        L::alloc_write(&mut left, k0, v0, l, r, &mut n);
        clone::<K, V, L>(page.as_page(), &mut left, s0b, &mut n);
        let mut right;
        let freed;
        // Then, for the right page:
        if mutable && page.is_dirty() {
            // If we can mutate the original page, remove the first
            // `k` entries, and return a pointer to the split key (at
            // position `k` on the page)
            right = MutPage(page);
            if let Some((k, v)) = L::truncate_left::<K, V>(&mut right, k as usize + 1) {
                unsafe {
                    split_key = &*k;
                    split_value = &*v;
                }
            }
            freed = 0;
        } else {
            // Else, clone the right page.
            right = txn.alloc_page().await?;
            <Page<K, V> as BTreeMutPage<K, V>>::init(&mut right);
            // The left child of the right page is `mid_child`,
            // i.e. the right child of the split entry.
            L::set_right_child(&mut right, -1, mid_child);
            // Clone the rest of the page.
            let mut n = 0;
            clone::<K, V, L>(page.as_page(), &mut right, s1, &mut n);
            freed = page.offset | page_is_dirty
        }
        Ok(Put::Split {
            split_key,
            split_value,
            left,
            right,
            freed,
        })
    }
}

use super::*;
/// This trait contains methods common to leaves and internal
/// nodes. These methods are meant to be just a few lines long each,
/// and avoid duplicating code in higher-level functions where just a
/// few constants change between internal nodes and leaves.
pub(crate) trait Alloc {
    /// Is there enough room to add one entry into this page (without cloning)?
    fn can_alloc<K: Storable, V: Storable>(hdr: &Header) -> bool;
    /// If we clone the page, will there be enough room for one entry?
    fn can_compact<K: Storable, V: Storable>(hdr: &Header) -> bool;
    /// Remove the leftmost `n` elements from the page. On leaves,
    /// this moves the last truncated element to the end of the page
    /// and returns a pointer to that element (this function is only
    /// used in `crate::btree::put`, and the pointer becomes invalid
    /// at the end of the `crate::btree::put`).
    ///
    /// Returning the last deleted element isn't required for internal
    /// nodes, since the entries are still valid there.
    fn truncate_left<K: Storable, V: Storable>(
        page: &mut MutPage,
        n: usize,
    ) -> Option<(*const K, *const V)>;
    /// Allocate a new entry (size inferred), and return the offset on
    /// the page where the new entry can be copied.
    fn alloc_insert<K: Storable, V: Storable>(new: &mut MutPage, n: &mut isize, r: u64) -> usize;
    /// Set the right child of the `n`th entry to `r`. If `n == -1`,
    /// this method can be used to set the leftmost child of a page.
    fn set_right_child(new: &mut MutPage, n: isize, r: u64);
    /// "Slices" of offsets, which aren't really slices in the case of
    /// leaves, but just intervals or "ranges".
    fn offset_slice<'a, T: LoadPage, K: Storable, V: Storable>(
        page: crate::Page<'a>,
    ) -> Offsets<'a>;
    /// First element of a slice offset. For the sake of code
    /// simplicity, we return a `u64` even for leaves. In internal
    /// nodes, the 52 MSBs encode a child page, and the 12 LSBs encode
    /// an offset in the page.
    fn first<'a, K, V>(off: &Offsets<'a>) -> u64;
    /// The tuple and right child at offset `off`.
    fn kv<'a, K: Storable, V: Storable>(page: crate::Page, off: u64) -> (&'a K, &'a V, u64);
}
/// The type of offsets.
#[derive(Debug, Clone)]
pub enum Offsets<'a> {
    /// Slices of child pages and offset-in-page for internal nodes,
    Slice(&'a [u64]),
    /// Ranges for leaves.
    Range(core::ops::Range<usize>),
}
impl<'a> Offsets<'a> {
    /// Splitting an offset slice, with the same semantics as
    /// `split_at` for slices, except if `mid` is equal to the length,
    /// in which case we return `(self, &[])`.
    pub fn split_at(&self, mid: usize) -> (Self, Self) {
        match self {
            Offsets::Slice(s) if mid < s.len() => {
                let (a, b) = s.split_at(mid);
                (Offsets::Slice(a), Offsets::Slice(b))
            }
            Offsets::Slice(s) => {
                debug_assert_eq!(mid, s.len());
                (Offsets::Slice(s), Offsets::Slice(&[][..]))
            }
            Offsets::Range(r) => (
                Offsets::Range(r.start..r.start + mid),
                Offsets::Range(r.start + mid..r.end),
            ),
        }
    }
}
pub struct Leaf {}
pub struct Internal {}
impl Alloc for Leaf {
    // Checking if there's enough room is just bookkeeping.
    fn can_alloc<K: Storable, V: Storable>(hdr: &Header) -> bool {
        let f = core::mem::size_of::<Tuple<K, V>>();
        let al = core::mem::align_of::<Tuple<K, V>>();
        let header_size = (HDR + al - 1) & !(al - 1);
        header_size + (hdr.n() as usize + 1) * f <= PAGE_SIZE
    }
    /// There's no possible compaction on leaves, it's the same as allocating.
    fn can_compact<K: Storable, V: Storable>(hdr: &Header) -> bool {
        Self::can_alloc::<K, V>(hdr)
    }
    fn set_right_child(_: &mut MutPage, _: isize, _: u64) {}
    fn offset_slice<'a, T: LoadPage, K: Storable, V: Storable>(
        page: crate::Page<'a>,
    ) -> Offsets<'a> {
        let hdr = header(page);
        Offsets::Range(0..(hdr.n() as usize))
    }
    /// This returns an offset on the page, so we need to compute that.
    fn first<'a, K, V>(off: &Offsets<'a>) -> u64 {
        match off {
            Offsets::Slice(_) => unreachable!(),
            Offsets::Range(r) => {
                let size = core::mem::size_of::<Tuple<K, V>>();
                let al = core::mem::align_of::<Tuple<K, V>>();
                let hdr_size = (HDR + al - 1) & !(al - 1);
                (hdr_size + r.start * size) as u64
            }
        }
    }
    fn kv<'a, K: Storable, V: Storable>(page: crate::Page, off: u64) -> (&'a K, &'a V, u64) {
        unsafe {
            let tup = &*(page.data.as_ptr().add(off as usize) as *const Tuple<K, V>);
            (&tup.k, &tup.v, 0)
        }
    }
    /// Here, we just move `total - *n` elements to the right, and
    /// increase the bookkeeping values (occupied bytes and number of
    /// elements).
    fn alloc_insert<K: Storable, V: Storable>(new: &mut MutPage, n: &mut isize, _: u64) -> usize {
        let f = core::mem::size_of::<Tuple<K, V>>();
        let al = core::mem::align_of::<Tuple<K, V>>();
        let hdr_size = (HDR + al - 1) & !(al - 1);
        let hdr_n = header_mut(new).n();
        unsafe {
            core::ptr::copy(
                new.0.data.add(hdr_size + (*n as usize) * f),
                new.0.data.add(hdr_size + (*n as usize) * f + f),
                (hdr_n as usize - (*n as usize)) * f,
            );
        }
        let hdr = header_mut(new);
        hdr.set_n(hdr.n() + 1);
        hdr.incr(f);
        hdr_size + (*n as usize) * f
    }
    fn truncate_left<K: Storable, V: Storable>(
        page: &mut MutPage,
        n: usize,
    ) -> Option<(*const K, *const V)> {
        let f = core::mem::size_of::<Tuple<K, V>>();
        let al = core::mem::align_of::<Tuple<K, V>>();
        let hdr_size = (HDR + al - 1) & !(al - 1);
        let hdr = header_mut(page);
        let hdr_n = hdr.n();
        hdr.set_n(hdr_n - n as u16);
        hdr.decr(n * f);
        // Now, this is a bit trickier. This performs a rotation by 1
        // without all the rotation machinery from the Rust core
        // library.
        //
        // Indeed, since we're in a leaf, we are effectively erasing
        // the split entry. There is a choice to be made here, between
        // passing the entry by value or by reference. Because splits
        // might cascade with the same middle entry, passing it by
        // value may be significantly worse if the tree is deep, and
        // is never significantly better (we'll end up copying this
        // entry twice anyway).
        unsafe {
            let mut swap: core::mem::MaybeUninit<Tuple<K, V>> = core::mem::MaybeUninit::uninit();
            core::ptr::copy(
                page.0.data.add(hdr_size + (n - 1) * f),
                swap.as_mut_ptr() as *mut u8,
                f,
            );
            core::ptr::copy(
                page.0.data.add(hdr_size + n * f),
                page.0.data.add(hdr_size),
                (hdr_n as usize - n) * f,
            );
            core::ptr::copy(
                swap.as_ptr() as *mut u8,
                page.0.data.add(hdr_size + (hdr_n as usize - n) * f),
                f,
            );
            let tup =
                &*(page.0.data.add(hdr_size + (hdr_n as usize - n) * f) as *const Tuple<K, V>);
            Some((&tup.k, &tup.v))
        }
    }
}
impl Alloc for Internal {
    fn can_alloc<K: Storable, V: Storable>(hdr: &Header) -> bool {
        (HDR as usize) + (hdr.n() as usize + 1) * 8 + core::mem::size_of::<Tuple<K, V>>()
            < hdr.data() as usize
    }
    fn can_compact<K: Storable, V: Storable>(hdr: &Header) -> bool {
        (HDR as usize)
            + ((hdr.left_page() & 0xfff) as usize)
            + 8
            + core::mem::size_of::<Tuple<K, V>>()
            <= PAGE_SIZE
    }
    /// Set the right-hand child in the offset array, preserving the
    /// current offset.
    fn set_right_child(page: &mut MutPage, n: isize, r: u64) {
        unsafe {
            debug_assert_eq!(r & 0xfff, 0);
            let ptr = page.0.data.offset(HDR as isize + n * 8) as *mut u64;
            let off = u64::from_le(*ptr) & 0xfff;
            *ptr = (r | off as u64).to_le();
        }
    }
    fn offset_slice<'a, T, K: Storable, V: Storable>(page: crate::Page<'a>) -> Offsets<'a> {
        let hdr = header(page);
        unsafe {
            Offsets::Slice(core::slice::from_raw_parts(
                page.data.as_ptr().add(HDR) as *const u64,
                hdr.n() as usize,
            ))
        }
    }
    fn first<'a, K, V>(off: &Offsets<'a>) -> u64 {
        match off {
            Offsets::Slice(s) => s[0],
            Offsets::Range(_) => unreachable!(),
        }
    }
    fn kv<'a, K: Storable, V: Storable>(page: crate::Page, off: u64) -> (&'a K, &'a V, u64) {
        unsafe {
            let tup = &*(page.data.as_ptr().add((off & 0xfff) as usize) as *const Tuple<K, V>);
            (&tup.k, &tup.v, off & !0xfff)
        }
    }
    // Much simpler than in leaves, because we just need to move the
    // offsets. There's a little bit of bookkeeping involved though.
    fn truncate_left<K: Storable, V: Storable>(
        page: &mut MutPage,
        n: usize,
    ) -> Option<(*const K, *const V)> {
        let hdr_n = header_mut(page).n();
        unsafe {
            // We want to copy the left child of the first entry that
            // will be kept alive as the left child of the page.
            core::ptr::copy(
                page.0.data.add(HDR + (n - 1) * 8),
                page.0.data.add(HDR - 8),
                // We're removing n entries, so we need to copy the
                // remaining `hdr_n - n`, plus the leftmost child
                // (hence the + 1).
                (hdr_n as usize - n + 1) * 8,
            );
        }
        // Remaining occupied size on the page (minus the header).
        let size = (8 + core::mem::size_of::<Tuple<K, V>>()) * (hdr_n as usize - n);
        let hdr = header_mut(page);
        hdr.set_n(hdr_n - n as u16);
        // Because the leftmost child has been copied from an entry
        // containing an offset, its 12 LBSs are still enconding an
        // offset on the page, whereas it should encode the occupied
        // bytes on the page. Reset it.
        hdr.set_occupied(size as u64);
        None
    }
    fn alloc_insert<K: Storable, V: Storable>(new: &mut MutPage, n: &mut isize, r: u64) -> usize {
        let hdr = header_mut(new);
        // Move the `data` field one entry to the left.
        let data = hdr.data() as u16;
        let off_new = data - core::mem::size_of::<Tuple<K, V>>() as u16;
        hdr.set_data(off_new);
        // Increment the number of entries, add the newly occupied size.
        hdr.set_n(hdr.n() + 1);
        hdr.incr(8 + core::mem::size_of::<Tuple<K, V>>());
        let hdr_n = hdr.n();
        unsafe {
            // Make space in the offset array by moving the last
            // `hdr_n - *n` elements.
            core::ptr::copy(
                new.0.data.add(HDR + (*n as usize) * 8),
                new.0.data.add(HDR + (*n as usize) * 8 + 8),
                (hdr_n as usize - (*n as usize)) * 8,
            );
            // Add the offset.
            let ptr = new.0.data.offset(HDR as isize + *n * 8) as *mut u64;
            *ptr = (r | off_new as u64).to_le();
        }
        off_new as usize
    }
}
impl<K: Storable, V: Storable> crate::btree::page_unsized::AllocWrite<K, V> for Internal {
    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize) {
        alloc_write::<K, V, Self>(new, k0, v0, l, r, n)
    }
    fn set_left_child(new: &mut MutPage, n: isize, l: u64) {
        if l > 0 {
            Internal::set_right_child(new, n - 1, l);
        }
    }
}
impl<K: Storable, V: Storable> crate::btree::page_unsized::AllocWrite<K, V> for Leaf {
    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize) {
        alloc_write::<K, V, Self>(new, k0, v0, l, r, n)
    }
    fn set_left_child(_new: &mut MutPage, _n: isize, l: u64) {
        debug_assert_eq!(l, 0);
    }
}
/// Simply allocate an entry in the page, copy it, and set its right
/// and left children.
fn alloc_write<K: Storable, V: Storable, L: Alloc>(
    new: &mut MutPage,
    k0: &K,
    v0: &V,
    l: u64,
    r: u64,
    n: &mut isize,
) {
    let off_new = L::alloc_insert::<K, V>(new, n, r);
    unsafe {
        let new_ptr = &mut *(new.0.data.add(off_new as usize) as *mut Tuple<K, V>);
        core::ptr::copy_nonoverlapping(k0, &mut new_ptr.k, 1);
        core::ptr::copy_nonoverlapping(v0, &mut new_ptr.v, 1);
    }
    if l > 0 {
        L::set_right_child(new, *n - 1, l);
    }
    *n += 1;
}