wasm-dbms-memory 0.10.1

// Rust guideline compliant 2026-02-28

mod autoincrement_ledger;
mod free_segments_ledger;
mod index_ledger;
mod page_ledger;
mod raw_record;
mod raw_table_reader;
mod record_address;
mod schema_snapshot_ledger;
mod sized_read;
mod table_reader;
mod write_at;

use std::borrow::Cow;

use wasm_dbms_api::prelude::{
    DataSize, DecodeError, Encode, MSize, MemoryError, MemoryResult, Page, PageOffset, Value,
};

pub use self::autoincrement_ledger::AutoincrementLedger;
use self::free_segments_ledger::FreeSegmentsLedger;
pub use self::index_ledger::{IndexLedger, IndexTreeWalker};
use self::page_ledger::PageLedger;
use self::raw_record::{RAW_RECORD_HEADER_SIZE, RawRecord};
pub use self::raw_table_reader::{RawRecordBytes, RawTableReader};
pub use self::record_address::RecordAddress;
pub use self::schema_snapshot_ledger::SchemaSnapshotLedger;
pub use self::table_reader::{NextRecord, TableReader};
use self::write_at::WriteAt;
use crate::{MemoryAccess, TableRegistryPage, align_up};

/// The table registry takes care of storing the records for each table,
/// using the `FreeSegmentsLedger` and `PageLedger` to derive exactly where to read/write.
///
/// A registry is generic over a record, which must implement [`Encode`].
///
/// The CRUD operations provided by the table registry do NOT perform any logical checks,
/// but just allow to read/write records from/to memory.
/// So CRUD checks must be performed by a higher layer, prior to calling these methods.
pub struct TableRegistry {
    /// Page holding the schema snapshot, loaded on demand since only migrations read it.
    schema_snapshot_page: Page,
    pub(crate) page_ledger: PageLedger,
    free_segments_ledger: FreeSegmentsLedger,
    index_ledger: IndexLedger,
    auto_increment_ledger: Option<AutoincrementLedger>,
}

impl TableRegistry {
    /// Loads the table registry from memory.
    pub fn load(table_pages: TableRegistryPage, mm: &mut impl MemoryAccess) -> MemoryResult<Self> {
        Ok(Self {
            schema_snapshot_page: table_pages.schema_snapshot_page,
            page_ledger: PageLedger::load(table_pages.pages_list_page, mm)?,
            free_segments_ledger: FreeSegmentsLedger::load(table_pages.free_segments_page, mm)?,
            index_ledger: IndexLedger::load(table_pages.index_registry_page, mm)?,
            auto_increment_ledger: if let Some(page) = table_pages.autoincrement_registry_page {
                Some(AutoincrementLedger::load(page, mm)?)
            } else {
                None
            },
        })
    }

    /// Inserts a new record into the table registry.
    ///
    /// Returns the address where the record was inserted, which can be used to read it back or to update/delete it.
    ///
    /// NOTE: this function does NOT make any logical checks on the record being inserted.
    ///
    /// # Errors
    ///
    /// Returns [`MemoryError::DataTooLarge`] before writing anything when the
    /// record cannot be stored on a single page.
    pub fn insert<E>(
        &mut self,
        record: E,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<RecordAddress>
    where
        E: Encode,
    {
        ensure_record_fits(&record, mm.page_size())?;

        // get position to write the record
        let raw_record = RawRecord::new(record);
        let write_at = self.get_write_position(&raw_record, mm)?;

        // align insert to RawRecord<E> alignment (includes the 2-byte header)
        let aligned_offset = align_up::<RawRecord<E>>(write_at.offset() as usize) as PageOffset;

        // write record
        mm.write_at(write_at.page(), aligned_offset, &raw_record)?;

        let pointer = RecordAddress {
            page: write_at.page(),
            offset: aligned_offset,
        };

        // commit post-write actions
        self.post_write(write_at, &raw_record, mm)?;

        Ok(pointer)
    }

    /// Creates a [`TableReader`] to read records from the table registry.
    ///
    /// Use [`TableReader::try_next`] to read records one by one.
    pub fn read<'a, E, MA>(&'a self, mm: &'a mut MA) -> TableReader<'a, E, MA>
    where
        E: Encode,
        MA: MemoryAccess,
    {
        TableReader::new(&self.page_ledger, mm)
    }

    /// Reads a single record at the given address.
    ///
    /// Only the record bytes are copied out of memory: the length header is
    /// read first, then exactly that many body bytes.
    ///
    /// # Errors
    ///
    /// Returns [`MemoryError::OffsetNotAligned`] when `address` is not aligned
    /// for `E`, and [`DecodeError::TooShort`] when the stored length runs past
    /// the end of the page.
    pub fn read_at<E, MA>(&self, address: RecordAddress, mm: &mut MA) -> MemoryResult<E>
    where
        E: Encode,
        MA: MemoryAccess,
    {
        if let DataSize::Fixed(_) = E::SIZE {
            let raw_record: RawRecord<E> = mm.read_at(address.page, address.offset)?;
            return Ok(raw_record.data);
        }

        let alignment = <RawRecord<E> as Encode>::ALIGNMENT;
        if alignment != 0 && !address.offset.is_multiple_of(alignment) {
            return Err(MemoryError::OffsetNotAligned {
                offset: address.offset,
                alignment,
            });
        }

        let mut header = [0u8; RAW_RECORD_HEADER_SIZE as usize];
        if mm.read_at_raw(address.page, address.offset, &mut header)? < header.len() {
            return Err(MemoryError::DecodeError(DecodeError::TooShort));
        }
        let length = MSize::from_le_bytes(header) as usize;

        let body_offset = address.offset as usize + RAW_RECORD_HEADER_SIZE as usize;
        if body_offset + length > mm.page_size() as usize {
            return Err(MemoryError::DecodeError(DecodeError::TooShort));
        }
        let mut body = vec![0u8; length];
        mm.read_at_raw(address.page, body_offset as PageOffset, &mut body)?;

        E::decode(Cow::Owned(body))
    }

    /// Deletes a record at the given page and offset.
    ///
    /// The space occupied by the record is marked as free and zeroed.
    pub fn delete(
        &mut self,
        record: impl Encode,
        address: RecordAddress,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<()> {
        let raw_record = RawRecord::new(record);

        // zero the record in memory
        mm.zero(address.page, address.offset, &raw_record)?;

        // insert a free segment for the deleted record
        self.free_segments_ledger
            .insert_free_segment(address.page, address.offset, &raw_record, mm)
    }

    /// Updates a record at the given page and offset.
    ///
    /// The [`RecordAddress`] of the new record is returned, which can be different from the old one if the record was reallocated.
    ///
    /// The logic is the following:
    ///
    /// 1. If the new record has exactly the same size of the old record, overwrite it in place.
    /// 2. If the new record does not fit, delete the old record and insert the new record.
    ///
    /// # Errors
    ///
    /// Returns [`MemoryError::DataTooLarge`] before touching the old record
    /// when the new record cannot be stored on a single page.
    pub fn update(
        &mut self,
        new_record: impl Encode,
        old_record: impl Encode,
        old_address: RecordAddress,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<RecordAddress> {
        ensure_record_fits(&new_record, mm.page_size())?;

        if new_record.size() == old_record.size() {
            self.update_in_place(new_record, old_address, mm)
        } else {
            self.update_by_realloc(new_record, old_record, old_address, mm)
        }
    }

    /// Iterate the table's records as raw bytes under the given alignment.
    ///
    /// Used by the migration apply pipeline to read records under the stored
    /// snapshot. `alignment` must match the on-disk layout (i.e. the
    /// `TableSchemaSnapshot::alignment` of the snapshot the records were
    /// inserted under).
    pub fn iter_raw<'a, MA>(
        &'a self,
        alignment: PageOffset,
        mm: &'a mut MA,
    ) -> RawTableReader<'a, MA>
    where
        MA: MemoryAccess,
    {
        RawTableReader::new(&self.page_ledger, alignment, mm)
    }

    /// Insert pre-encoded record bytes under the given alignment.
    ///
    /// Used by the migration apply pipeline. `bytes` is the body of the
    /// record (without the 2-byte length header — the registry writes the
    /// header). `alignment` comes from the target snapshot.
    pub fn insert_raw(
        &mut self,
        bytes: &[u8],
        alignment: PageOffset,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<RecordAddress> {
        let physical_size_msize = raw_record_footprint(bytes.len(), alignment, mm.page_size())?;
        // The footprint fits in an `MSize`, so header + body does too.
        let length = bytes.len() as MSize;
        let total = (RAW_RECORD_HEADER_SIZE + length) as u64;
        let physical_size_u64 = physical_size_msize as u64;

        // Reuse a free segment only if the whole aligned record (header,
        // body and padding) fits; comparing the body length alone lets the
        // write spill into the next live record.
        let (page, offset) = if let Some(segment) = self
            .free_segments_ledger
            .find_reusable_segment_raw(physical_size_msize, mm)?
        {
            let page = segment.segment.page;
            let offset = segment.segment.offset;
            self.free_segments_ledger
                .commit_reused_space_raw(physical_size_msize, segment, mm)?;
            (page, offset)
        } else {
            let (page, offset) =
                self.page_ledger
                    .get_page_and_offset_raw(physical_size_u64, alignment, mm)?;
            self.page_ledger.commit_raw(page, total, alignment, mm)?;
            (page, offset)
        };
        let mut full = Vec::with_capacity(total as usize);
        full.extend_from_slice(&length.to_le_bytes());
        full.extend_from_slice(bytes);
        mm.write_at_raw(page, offset, &full)?;
        let padding = physical_size_msize - (RAW_RECORD_HEADER_SIZE + length);
        if padding > 0 {
            let padding_offset = offset + RAW_RECORD_HEADER_SIZE + length;
            mm.zero_raw(page, padding_offset, padding)?;
        }

        Ok(RecordAddress { page, offset })
    }

    /// Read raw record bytes at the given address (header stripped).
    pub fn read_raw_at(
        &self,
        address: RecordAddress,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<Vec<u8>> {
        let mut header = [0u8; RAW_RECORD_HEADER_SIZE as usize];
        mm.read_at_raw(address.page, address.offset, &mut header)?;
        let length = u16::from_le_bytes(header) as usize;
        let mut body = vec![0u8; length];
        mm.read_at_raw(
            address.page,
            address.offset + RAW_RECORD_HEADER_SIZE,
            &mut body,
        )?;
        Ok(body)
    }

    /// Delete a raw record at the given address. Mirrors [`Self::delete`]
    /// but parameterises size + alignment rather than reading them from a
    /// compile-time `Encode` impl.
    pub fn delete_raw(
        &mut self,
        address: RecordAddress,
        body_len: MSize,
        alignment: PageOffset,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<()> {
        let physical_size = raw_record_footprint(body_len as usize, alignment, mm.page_size())?;
        mm.zero_raw(address.page, address.offset, physical_size)?;
        self.free_segments_ledger.insert_free_segment_raw(
            address.page,
            address.offset,
            physical_size,
            mm,
        )
    }

    /// Releases every page owned by this table back to the unclaimed-pages
    /// ledger.
    ///
    /// Walks the page ledger, free-segments ledger, every B-tree in the
    /// index ledger, plus the dedicated schema-snapshot and
    /// autoincrement-registry pages, and hands each one to
    /// [`MemoryAccess::unclaim_page`]. Used by `MigrationOp::DropTable`.
    ///
    /// `table_pages` must be the [`TableRegistryPage`] this registry was
    /// loaded from — the schema-snapshot and autoincrement pages are not
    /// stored inside the registry itself.
    ///
    /// After this call the table's pages are reclaimable by future
    /// [`MemoryAccess::claim_page`] calls. The caller is responsible for
    /// removing the table's entry from the schema registry.
    ///
    /// # Errors
    ///
    /// Propagates any [`wasm_dbms_api::prelude::MemoryError`] surfaced
    /// while walking ledgers or releasing pages.
    pub fn release_pages(
        self,
        table_pages: TableRegistryPage,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<()> {
        self.page_ledger.release_pages(mm)?;
        self.free_segments_ledger.release_pages(mm)?;
        self.index_ledger.release_pages(mm)?;
        mm.unclaim_page(table_pages.schema_snapshot_page)?;
        if let Some(page) = table_pages.autoincrement_registry_page {
            mm.unclaim_page(page)?;
        }
        Ok(())
    }

    /// Returns how many pages dropping this table would release.
    pub fn releasable_pages_count(
        &self,
        table_pages: TableRegistryPage,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<usize> {
        let mut count = self.page_ledger.releasable_pages_count();
        count += self.free_segments_ledger.releasable_pages_count();
        count += self.index_ledger.releasable_pages_count(mm)?;
        count += 1; // schema snapshot page
        if table_pages.autoincrement_registry_page.is_some() {
            count += 1;
        }
        Ok(count)
    }

    /// Get a reference to the index ledger, allowing to read the indexes.
    pub fn index_ledger(&self) -> &IndexLedger {
        &self.index_ledger
    }

    /// Get a mutable reference to the index ledger, allowing to modify the indexes.
    pub fn index_ledger_mut(&mut self) -> &mut IndexLedger {
        &mut self.index_ledger
    }

    /// Loads the [`SchemaSnapshotLedger`] of this table from memory.
    ///
    /// The snapshot is only needed by migrations, so it is read on demand
    /// instead of on every [`Self::load`].
    ///
    /// # Errors
    ///
    /// Returns a [`MemoryError`] if the snapshot page cannot be read or does
    /// not decode into a valid snapshot.
    pub fn schema_snapshot_ledger(
        &self,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<SchemaSnapshotLedger> {
        SchemaSnapshotLedger::load(self.schema_snapshot_page, mm)
    }

    /// Get next value for an autoincrement column of the given type, and increment it in the ledger.
    pub fn next_autoincrement(
        &mut self,
        column_name: &str,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<Option<Value>> {
        if let Some(ledger) = &mut self.auto_increment_ledger {
            ledger.next(column_name, mm).map(Some)
        } else {
            Ok(None)
        }
    }

    /// Update a [`RawRecord`] in place at the given page and offset.
    ///
    /// The [`RecordAddress`] of the record is returned, which is the same as the old one.
    ///
    /// This must be used IF AND ONLY if the new record has the SAME size as the old record.
    fn update_in_place(
        &mut self,
        record: impl Encode,
        address: RecordAddress,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<RecordAddress> {
        let raw_record = RawRecord::new(record);
        mm.write_at(address.page, address.offset, &raw_record)?;

        Ok(address)
    }

    /// Updates a record by reallocating it.
    ///
    /// The old record is deleted and the new record is inserted.
    ///
    /// The [`RecordAddress`] of the new record is returned, which can be different from the old one.
    fn update_by_realloc(
        &mut self,
        new_record: impl Encode,
        old_record: impl Encode,
        old_address: RecordAddress,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<RecordAddress> {
        // delete old record
        self.delete(old_record, old_address, mm)?;

        // insert new record
        self.insert(new_record, mm)
    }

    /// Gets the position where to write a record of the given size.
    fn get_write_position<E>(
        &mut self,
        record: &RawRecord<E>,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<WriteAt>
    where
        E: Encode,
    {
        // check if there is a free segment that can hold the record
        if let Some(segment) = self
            .free_segments_ledger
            .find_reusable_segment(record, mm)?
        {
            return Ok(WriteAt::ReusedSegment(segment));
        }

        // otherwise, write at the end of the table
        self.page_ledger
            .get_page_and_offset_for_record(record, mm)
            .map(|(page, offset)| WriteAt::End(page, offset))
    }

    /// Commits the post-write actions after writing a record at the given position.
    ///
    /// - If the record was a [`WriteAt::ReusedSegment`], the free segment is marked as used.
    /// - If the record was a [`WriteAt::End`], the page ledger is updated.
    fn post_write<E>(
        &mut self,
        write_at: WriteAt,
        record: &RawRecord<E>,
        mm: &mut impl MemoryAccess,
    ) -> MemoryResult<()>
    where
        E: Encode,
    {
        match write_at {
            WriteAt::ReusedSegment(free_segment) => {
                // mark segment as used
                self.free_segments_ledger
                    .commit_reused_space(record, free_segment, mm)
            }
            WriteAt::End(page, ..) => {
                // update page ledger
                self.page_ledger.commit(page, record, mm)
            }
        }
    }
}

/// Rejects a typed record whose raw-record footprint does not fit an
/// [`MSize`], using the same limit as the raw insert path.
///
/// A record whose [`Encode::size`] saturated at [`MSize::MAX`] always fails
/// this check, so oversized values never reach [`Encode::encode`].
///
/// # Errors
///
/// Returns [`MemoryError::DataTooLarge`] when the record cannot be stored on
/// a single page.
fn ensure_record_fits<E>(record: &E, page_size: u64) -> MemoryResult<()>
where
    E: Encode,
{
    raw_record_footprint(
        record.size() as usize,
        <RawRecord<E> as Encode>::ALIGNMENT,
        page_size,
    )
    .map(|_| ())
}

/// Computes the on-page footprint of a raw record with a `body_len`-byte
/// body: the 2-byte length header plus the body, rounded up to `alignment`.
/// Used by the raw insert/delete path where alignment comes from a stored
/// snapshot.
///
/// # Errors
///
/// Returns [`MemoryError::DataTooLarge`] when the footprint does not fit in
/// an [`MSize`], i.e. the record cannot be stored on a single page.
fn raw_record_footprint(
    body_len: usize,
    alignment: PageOffset,
    page_size: u64,
) -> MemoryResult<MSize> {
    let unaligned = (body_len as u64).saturating_add(RAW_RECORD_HEADER_SIZE as u64);
    let aligned = if alignment == 0 {
        unaligned
    } else {
        unaligned
            .div_ceil(alignment as u64)
            .saturating_mul(alignment as u64)
    };
    MSize::try_from(aligned).map_err(|_| MemoryError::DataTooLarge {
        page_size,
        requested: aligned,
    })
}

/// Test utilities shared across the table_registry submodules.
#[cfg(test)]
pub(crate) mod test_utils {
    use wasm_dbms_api::prelude::{
        DEFAULT_ALIGNMENT, DataSize, DecodeError, Encode, MSize, MemoryError, MemoryResult, Page,
        PageOffset, TableSchemaSnapshot,
    };

    use crate::MemoryAccess;

    /// Writes a minimal valid [`TableSchemaSnapshot`] to `page` so subsequent
    /// `TableRegistry::load` calls can decode it.
    ///
    /// Required because the table-registry unit tests allocate raw pages and
    /// load the registry directly, bypassing the
    /// [`SchemaRegistry::register_table`](crate::SchemaRegistry::register_table)
    /// path that normally writes the snapshot for them.
    pub fn write_dummy_schema_snapshot(page: Page, mm: &mut impl MemoryAccess) {
        let dummy = TableSchemaSnapshot {
            version: TableSchemaSnapshot::latest_version(),
            name: "dummy".to_string(),
            primary_key: "id".to_string(),
            alignment: 8,
            columns: vec![],
            indexes: vec![],
        };
        mm.write_at(page, 0, &dummy)
            .expect("failed to write dummy snapshot");
    }

    /// A simple user struct for testing purposes (no macro dependencies).
    #[derive(Debug, Clone, PartialEq, Eq)]
    pub struct User {
        pub id: u32,
        pub name: String,
        pub email: String,
        pub age: u32,
    }

    impl Encode for User {
        const SIZE: DataSize = DataSize::Dynamic;

        const ALIGNMENT: PageOffset = DEFAULT_ALIGNMENT;

        fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
            let mut buf = Vec::new();
            // id: 4 bytes
            buf.extend_from_slice(&self.id.to_le_bytes());
            // name length: 2 bytes + name bytes
            buf.extend_from_slice(&(self.name.len() as u16).to_le_bytes());
            buf.extend_from_slice(self.name.as_bytes());
            // email length: 2 bytes + email bytes
            buf.extend_from_slice(&(self.email.len() as u16).to_le_bytes());
            buf.extend_from_slice(self.email.as_bytes());
            // age: 4 bytes
            buf.extend_from_slice(&self.age.to_le_bytes());
            std::borrow::Cow::Owned(buf)
        }

        fn decode(data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
        where
            Self: Sized,
        {
            if data.len() < 12 {
                return Err(MemoryError::DecodeError(DecodeError::TooShort));
            }
            let mut offset = 0;
            // id
            let id = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap());
            offset += 4;
            // name
            let name_len =
                u16::from_le_bytes(data[offset..offset + 2].try_into().unwrap()) as usize;
            offset += 2;
            let name = String::from_utf8_lossy(&data[offset..offset + name_len]).to_string();
            offset += name_len;
            // email
            let email_len =
                u16::from_le_bytes(data[offset..offset + 2].try_into().unwrap()) as usize;
            offset += 2;
            let email = String::from_utf8_lossy(&data[offset..offset + email_len]).to_string();
            offset += email_len;
            // age
            let age = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap());

            Ok(User {
                id,
                name,
                email,
                age,
            })
        }

        fn size(&self) -> MSize {
            (4 + 2 + self.name.len() + 2 + self.email.len() + 4) as MSize
        }
    }
}

#[cfg(test)]
mod tests {

    use self::test_utils::User;
    use super::free_segments_ledger::FreeSegment;
    use super::table_reader::NextRecord;
    use super::*;
    use crate::{HeapMemoryProvider, MemoryManager};

    #[test]
    fn test_should_create_table_registry() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let schema_snapshot_page = mm.claim_page().expect("failed to get page");
        let page_ledger_page = mm.claim_page().expect("failed to get page");
        let free_segments_page = mm.claim_page().expect("failed to get page");
        let index_registry_page = mm.claim_page().expect("failed to get page");
        let autoincrement_page = mm.claim_page().expect("failed to get page");
        super::test_utils::write_dummy_schema_snapshot(schema_snapshot_page, &mut mm);
        let table_pages = TableRegistryPage {
            schema_snapshot_page,
            pages_list_page: page_ledger_page,
            free_segments_page,
            index_registry_page,
            autoincrement_registry_page: Some(autoincrement_page),
        };

        let registry: MemoryResult<TableRegistry> = TableRegistry::load(table_pages, &mut mm);
        assert!(registry.is_ok());
    }

    #[test]
    fn test_should_get_write_at_end() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let record = RawRecord::new(User {
            id: 1,
            name: "Test".to_string(),
            email: "new_user@example.com".to_string(),
            age: 25,
        });
        let write_at = registry
            .get_write_position(&record, &mut mm)
            .expect("failed to get write at");

        assert!(matches!(write_at, WriteAt::End(_, 0)));
    }

    #[test]
    fn test_should_get_write_at_free_segment() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let record = RawRecord::new(User {
            id: 1,
            name: "Test".to_string(),
            email: "new_user@example.com".to_string(),
            age: 25,
        });
        // allocate a page to insert a free segment
        let (page, _) = registry
            .page_ledger
            .get_page_and_offset_for_record(&record, &mut mm)
            .expect("failed to get page and offset");
        registry
            .page_ledger
            .commit(page, &record, &mut mm)
            .expect("failed to commit page ledger");
        // insert data about a free segment
        registry
            .free_segments_ledger
            .insert_free_segment(page, 256, &record, &mut mm)
            .expect("failed to insert free segment");

        let write_at = registry
            .get_write_position(&record, &mut mm)
            .expect("failed to get write at");

        let reused_segment = match write_at {
            WriteAt::ReusedSegment(segment) => segment.segment,
            _ => panic!("expected reused segment"),
        };

        assert_eq!(
            reused_segment,
            FreeSegment {
                page,
                offset: 256,
                size: 64, // padded size
            }
        );
    }

    #[test]
    fn test_should_insert_record_into_table_registry() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let record = User {
            id: 1,
            name: "Test".to_string(),
            email: "new_user@example.com".to_string(),
            age: 25,
        };

        // insert record
        assert!(registry.insert(record, &mut mm).is_ok());
    }

    #[test]
    fn test_should_manage_to_insert_users_to_exceed_one_page() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        for id in 0..4000 {
            let record = User {
                id,
                name: format!("User {}", id),
                email: "new_user@example.com".to_string(),
                age: 20 + id,
            };
            registry
                .insert(record, &mut mm)
                .expect("failed to insert record");
        }
    }

    #[test]
    fn test_should_delete_record() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let record = User {
            id: 1,
            name: "Test".to_string(),
            email: "new_user@example.com".to_string(),
            age: 25,
        };

        // insert record
        registry
            .insert(record.clone(), &mut mm)
            .expect("failed to insert");

        // find where it was written
        let mut reader = registry.read(&mut mm);
        let next_record: NextRecord<User> = reader
            .try_next()
            .expect("failed to read")
            .expect("no record");
        let page = next_record.page;
        let offset = next_record.offset;
        let record = next_record.record;
        let raw_user = RawRecord::new(record.clone());
        let raw_user_size = raw_user.size();

        // delete record
        assert!(
            registry
                .delete(record, RecordAddress { page, offset }, &mut mm)
                .is_ok()
        );

        // should have been deleted
        let mut reader = registry.read::<User, _>(&mut mm);
        assert!(reader.try_next().expect("failed to read").is_none());

        // should have a free segment
        let free_segment = registry
            .free_segments_ledger
            .find_reusable_segment(
                &User {
                    id: 2,
                    name: "Test".to_string(),
                    email: "new_user@example.com".to_string(),
                    age: 25,
                },
                &mut mm,
            )
            .expect("failed to find free segment")
            .expect("could not find the free segment after free")
            .segment;
        assert_eq!(free_segment.page, page);
        assert_eq!(free_segment.offset, offset);
        assert_eq!(free_segment.size, 64); // padded

        // should have zeroed the memory
        let mut buffer = vec![0u8; raw_user_size as usize];
        mm.read_at_raw(page, offset, &mut buffer)
            .expect("failed to read memory");
        assert!(buffer.iter().all(|&b| b == 0));
    }

    #[test]
    fn test_read_at_returns_record_at_address() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);
        let record = User {
            id: 1,
            name: "Alice".to_string(),
            email: "alice@example.com".to_string(),
            age: 30,
        };

        let address = registry
            .insert(record.clone(), &mut mm)
            .expect("failed to insert record");

        let stored: User = registry
            .read_at(address, &mut mm)
            .expect("failed to read record");
        assert_eq!(stored, record);
    }

    #[test]
    fn test_read_at_rejects_record_length_past_page_end() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);
        let address = registry
            .insert(
                User {
                    id: 1,
                    name: "Alice".to_string(),
                    email: "alice@example.com".to_string(),
                    age: 30,
                },
                &mut mm,
            )
            .expect("failed to insert record");

        // Corrupt the length header so the record would run past the page end.
        mm.write_at_raw(address.page, address.offset, &MSize::MAX.to_le_bytes())
            .expect("failed to corrupt header");

        let result: MemoryResult<User> = registry.read_at(address, &mut mm);
        assert!(
            matches!(result, Err(MemoryError::DecodeError(DecodeError::TooShort))),
            "expected TooShort, got {result:?}"
        );
    }

    #[test]
    fn test_read_at_rejects_misaligned_address() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);
        let address = registry
            .insert(
                User {
                    id: 1,
                    name: "Alice".to_string(),
                    email: "alice@example.com".to_string(),
                    age: 30,
                },
                &mut mm,
            )
            .expect("failed to insert record");

        let misaligned = RecordAddress {
            page: address.page,
            offset: address.offset + 1,
        };
        let result: MemoryResult<User> = registry.read_at(misaligned, &mut mm);
        assert!(
            matches!(result, Err(MemoryError::OffsetNotAligned { .. })),
            "expected OffsetNotAligned, got {result:?}"
        );
    }

    #[test]
    fn test_schema_snapshot_ledger_is_read_on_demand() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let registry = registry_with_autoincrement(&mut mm);

        let ledger = registry
            .schema_snapshot_ledger(&mut mm)
            .expect("failed to load snapshot ledger");
        assert_eq!(ledger.get(), &AutoincUser::schema_snapshot());
    }

    #[test]
    fn test_read_at_after_update_returns_updated_record() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);
        let old_record = User {
            id: 1,
            name: "Alice".to_string(),
            email: "alice@example.com".to_string(),
            age: 30,
        };
        let new_record = User {
            id: 1,
            name: "Alice Updated".to_string(),
            email: "alice.updated@example.com".to_string(),
            age: 31,
        };

        let old_address = registry
            .insert(old_record.clone(), &mut mm)
            .expect("failed to insert record");
        let new_address = registry
            .update(new_record.clone(), old_record, old_address, &mut mm)
            .expect("failed to update record");

        let stored: User = registry
            .read_at(new_address, &mut mm)
            .expect("failed to read updated record");
        assert_eq!(stored, new_record);
    }

    #[test]
    fn test_should_update_record_in_place() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let old_record = User {
            id: 1,
            name: "John".to_string(),
            email: "new_user@example.com".to_string(),
            age: 28,
        };
        let new_record = User {
            id: 1,
            name: "Mark".to_string(), // same length as "John"
            email: "new_user@example.com".to_string(),
            age: 30,
        };

        // insert old record
        registry
            .insert(old_record.clone(), &mut mm)
            .expect("failed to insert");

        // find where it was written
        let mut reader = registry.read::<User, _>(&mut mm);
        let next_record = reader
            .try_next()
            .expect("failed to read")
            .expect("no record");
        let page = next_record.page;
        let offset = next_record.offset;

        // update in place
        let old_address = RecordAddress { page, offset };
        let new_location = registry
            .update(
                new_record.clone(),
                next_record.record.clone(),
                old_address,
                &mut mm,
            )
            .expect("failed to update record");
        assert_eq!(new_location, old_address); // should be same address

        // read back the record
        let mut reader = registry.read::<User, _>(&mut mm);
        let next_record = reader
            .try_next()
            .expect("failed to read")
            .expect("no record");
        assert_eq!(next_record.page, page); // should be same page
        assert_eq!(next_record.offset, offset); // should be same offset
        assert_eq!(next_record.record, new_record);
    }

    #[test]
    fn test_should_update_record_reallocating() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let old_record = User {
            id: 1,
            name: "John".to_string(),
            email: "new_user@example.com".to_string(),
            age: 28,
        };
        // this user creates a record with same size as old_record to avoid reusing the free segment
        let extra_user = User {
            id: 2,
            name: "Extra".to_string(),
            email: "new_user@example.com".to_string(),
            age: 25,
        };
        let new_record = User {
            id: 1,
            name: "Alexanderejruwgjowergjioewrgjioewrigjewriogjweoirgjiowerjgoiwerjiogewirogjowejrgiwer".to_string(), // must exceed padding
            email: "new_user@example.com".to_string(),
            age: 30,
        };

        // insert old record
        registry
            .insert(old_record.clone(), &mut mm)
            .expect("failed to insert");
        // insert extra record to avoid reusing the free segment
        registry
            .insert(extra_user.clone(), &mut mm)
            .expect("failed to insert extra user");

        // find where it was written
        let mut reader = registry.read::<User, _>(&mut mm);
        let old_record_from_db = reader
            .try_next()
            .expect("failed to read")
            .expect("no record");
        assert_eq!(old_record_from_db.record, old_record);
        let page = old_record_from_db.page;
        let offset = old_record_from_db.offset;

        // update by reallocating
        let old_address = RecordAddress { page, offset };
        let new_location = registry
            .update(
                new_record.clone(),
                old_record_from_db.record.clone(),
                old_address,
                &mut mm,
            )
            .expect("failed to update record");
        assert_ne!(new_location, old_address); // should be different page

        // read back the record
        let mut reader = registry.read::<User, _>(&mut mm);

        // find extra record first
        let _ = reader
            .try_next()
            .expect("failed to read")
            .expect("no record");

        let updated_record = reader
            .try_next()
            .expect("failed to read")
            .expect("no record");
        assert_ne!(updated_record.offset, offset); // should be different offset
        assert_eq!(updated_record.record, new_record);
    }

    #[test]
    fn test_should_insert_delete_insert_many() {
        const COUNT: u32 = 1_000;
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);
        for id in 0..COUNT {
            let record = User {
                id,
                name: format!("User {id}"),
                email: format!("user_{id}@example.com"),
                age: 20,
            };

            // insert record
            registry
                .insert(record.clone(), &mut mm)
                .expect("failed to insert");
        }

        // delete odd records
        for id in (0..COUNT).filter(|id| id % 2 == 1) {
            let record = User {
                id,
                name: format!("User {id}"),
                email: format!("user_{id}@example.com"),
                age: 20,
            };
            // find where it was written
            let mut reader = registry.read::<User, _>(&mut mm);
            let mut deleted = false;
            while let Some(next_record) = reader.try_next().expect("failed to read") {
                if next_record.record.id == id {
                    registry
                        .delete(
                            record.clone(),
                            RecordAddress {
                                page: next_record.page,
                                offset: next_record.offset,
                            },
                            &mut mm,
                        )
                        .expect("failed to delete");
                    deleted = true;
                    break;
                }
            }
            assert!(deleted, "record with id {} was not found", id);
        }

        // now delete also the others
        for id in (0..COUNT).filter(|id| id % 2 == 0) {
            let record = User {
                id,
                name: format!("User {id}"),
                email: format!("user_{id}@example.com"),
                age: 20,
            };
            // find where it was written
            let mut reader = registry.read::<User, _>(&mut mm);
            let mut deleted = false;
            while let Some(next_record) = reader.try_next().expect("failed to read") {
                if next_record.record.id == id {
                    registry
                        .delete(
                            record.clone(),
                            RecordAddress {
                                page: next_record.page,
                                offset: next_record.offset,
                            },
                            &mut mm,
                        )
                        .expect("failed to delete");
                    deleted = true;
                    break;
                }
            }
            assert!(deleted, "record with id {} was not found", id);
        }

        // insert back
        for id in 0..COUNT {
            let record = User {
                id,
                name: format!("User {id}"),
                email: format!("user_{id}@example.com"),
                age: 20,
            };

            // insert record
            registry
                .insert(record.clone(), &mut mm)
                .expect("failed to insert");
        }
    }

    #[test]
    fn test_should_reduce_free_segment_size_with_padding() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        // first insert a user
        let long_name = vec!['A'; 1024].into_iter().collect::<String>();
        let record = User {
            id: 1,
            name: "Test User".to_string(),
            email: long_name,
            age: 30,
        };
        registry
            .insert(record.clone(), &mut mm)
            .expect("failed to insert");
        // get record page
        let mut reader = registry.read::<User, _>(&mut mm);
        let next_record = reader
            .try_next()
            .expect("failed to read")
            .expect("no record");
        // delete user
        registry
            .delete(
                next_record.record,
                RecordAddress {
                    page: next_record.page,
                    offset: next_record.offset,
                },
                &mut mm,
            )
            .expect("failed to delete");

        // get the free segment
        let raw_record = RawRecord::new(record.clone());
        let free_segment = registry
            .free_segments_ledger
            .find_reusable_segment(&raw_record, &mut mm)
            .expect("failed to find reusable segment")
            .expect("could not find the free segment after free")
            .segment;
        // size should be at least 1024
        assert!(free_segment.size >= 1024);
        let previous_size = free_segment.size;

        // now insert a small user at 0
        let small_record = User {
            id: 2,
            name: "Bob The Builder".to_string(),
            email: "bob@hotmail.com".to_string(),
            age: 22,
        };
        registry
            .insert(small_record.clone(), &mut mm)
            .expect("failed to insert small user");

        // get free segment
        let free_segment_after = registry
            .free_segments_ledger
            .find_reusable_segment(&small_record, &mut mm)
            .expect("failed to find reusable segment")
            .expect("could not find the free segment after inserting small user")
            .segment;

        // size should be reduced
        assert_eq!(
            free_segment_after.offset, 64,
            "expected offset to be 64, but had: {}",
            free_segment_after.offset
        ); // which is the padding
        assert_eq!(
            free_segment_after.size,
            previous_size - 64,
            "Expected free segment to have size: {} but got: {}",
            previous_size - 64,
            free_segment_after.size
        );
    }

    #[test]
    fn test_raw_reuse_preserves_the_next_live_record() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        // 1-byte body: 2 + 1 = 3 bytes, aligned to a 32-byte slot.
        let deleted = registry
            .insert_raw(&[1], 32, &mut mm)
            .expect("insert deleted");
        let live = registry
            .insert_raw(&[2; 10], 32, &mut mm)
            .expect("insert live");
        registry
            .delete_raw(deleted, 1, 32, &mut mm)
            .expect("delete");

        // 31-byte body fits the 32-byte segment by body length only:
        // with the header it needs 33 bytes (64 once aligned).
        registry
            .insert_raw(&[9; 31], 32, &mut mm)
            .expect("insert replacement");

        assert_eq!(
            registry.read_raw_at(live, &mut mm).expect("read live"),
            vec![2; 10]
        );
    }

    #[test]
    fn test_raw_reuse_selects_segment_when_full_footprint_fits() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        // 30-byte body: 2 + 30 = 32 bytes, exactly one aligned slot.
        let deleted = registry
            .insert_raw(&[1; 30], 32, &mut mm)
            .expect("insert deleted");
        let live = registry
            .insert_raw(&[2; 10], 32, &mut mm)
            .expect("insert live");
        registry
            .delete_raw(deleted, 30, 32, &mut mm)
            .expect("delete");

        let reused = registry
            .insert_raw(&[9; 30], 32, &mut mm)
            .expect("insert replacement");

        assert_eq!(reused, deleted);
        assert_eq!(
            registry.read_raw_at(live, &mut mm).expect("read live"),
            vec![2; 10]
        );
    }

    fn registry(mm: &mut MemoryManager<HeapMemoryProvider>) -> TableRegistry {
        let schema_snapshot_page = mm.claim_page().expect("failed to get page");
        let page_ledger_page = mm.claim_page().expect("failed to get page");
        let free_segments_page = mm.claim_page().expect("failed to get page");
        let index_registry_page = mm.claim_page().expect("failed to get page");
        let autoincrement_page = mm.claim_page().expect("failed to get page");
        super::test_utils::write_dummy_schema_snapshot(schema_snapshot_page, mm);
        let table_pages = TableRegistryPage {
            schema_snapshot_page,
            pages_list_page: page_ledger_page,
            free_segments_page,
            index_registry_page,
            autoincrement_registry_page: Some(autoincrement_page),
        };

        TableRegistry::load(table_pages, mm).expect("failed to load")
    }

    /// Creates a [`TableRegistry`] with a properly initialized autoincrement ledger
    /// via [`SchemaRegistry::register_table`].
    fn registry_with_autoincrement(mm: &mut MemoryManager<HeapMemoryProvider>) -> TableRegistry {
        use crate::SchemaRegistry;

        let mut schema = SchemaRegistry::load(mm).expect("failed to load schema");
        let pages = schema
            .register_table::<AutoincUser>(mm)
            .expect("failed to register table");
        TableRegistry::load(pages, mm).expect("failed to load")
    }

    /// Creates a [`TableRegistry`] without an autoincrement ledger.
    fn registry_without_autoincrement(mm: &mut MemoryManager<HeapMemoryProvider>) -> TableRegistry {
        let schema_snapshot_page = mm.claim_page().expect("failed to get page");
        let page_ledger_page = mm.claim_page().expect("failed to get page");
        let free_segments_page = mm.claim_page().expect("failed to get page");
        let index_registry_page = mm.claim_page().expect("failed to get page");
        super::test_utils::write_dummy_schema_snapshot(schema_snapshot_page, mm);
        let table_pages = TableRegistryPage {
            schema_snapshot_page,
            pages_list_page: page_ledger_page,
            free_segments_page,
            index_registry_page,
            autoincrement_registry_page: None,
        };

        TableRegistry::load(table_pages, mm).expect("failed to load")
    }

    // -- AutoincUser mock: a table with an autoincrement Uint32 column --

    use candid::CandidType;
    use serde::{Deserialize, Serialize};
    use wasm_dbms_api::prelude::{
        ColumnDef, DbmsResult, IndexDef, InsertRecord, NoForeignFetcher, TableColumns, TableRecord,
        TableSchema, UpdateRecord,
    };

    #[derive(Clone, CandidType)]
    struct AutoincUser;

    impl Encode for AutoincUser {
        const SIZE: wasm_dbms_api::prelude::DataSize = wasm_dbms_api::prelude::DataSize::Dynamic;
        const ALIGNMENT: PageOffset = wasm_dbms_api::prelude::DEFAULT_ALIGNMENT;

        fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
            std::borrow::Cow::Owned(vec![])
        }

        fn decode(_data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
        where
            Self: Sized,
        {
            Ok(Self)
        }

        fn size(&self) -> wasm_dbms_api::prelude::MSize {
            0
        }
    }

    #[derive(Clone, CandidType, Deserialize)]
    struct AutoincUserRecord;

    impl TableRecord for AutoincUserRecord {
        type Schema = AutoincUser;

        fn from_values(_values: TableColumns) -> Self {
            Self
        }

        fn to_values(&self) -> Vec<(ColumnDef, Value)> {
            vec![]
        }
    }

    #[derive(Clone, CandidType, Serialize)]
    struct AutoincUserInsert;

    impl InsertRecord for AutoincUserInsert {
        type Record = AutoincUserRecord;
        type Schema = AutoincUser;

        fn from_values(_values: &[(ColumnDef, Value)]) -> DbmsResult<Self> {
            Ok(Self)
        }

        fn into_values(self) -> Vec<(ColumnDef, Value)> {
            vec![]
        }

        fn into_record(self) -> Self::Schema {
            AutoincUser
        }
    }

    #[derive(Clone, CandidType, Serialize)]
    struct AutoincUserUpdate;

    impl UpdateRecord for AutoincUserUpdate {
        type Record = AutoincUserRecord;
        type Schema = AutoincUser;

        fn from_values(
            _values: &[(ColumnDef, Value)],
            _where_clause: Option<wasm_dbms_api::prelude::Filter>,
        ) -> wasm_dbms_api::prelude::DbmsResult<Self> {
            Ok(Self)
        }

        fn update_values(&self) -> Vec<(ColumnDef, Value)> {
            vec![]
        }

        fn where_clause(&self) -> Option<wasm_dbms_api::prelude::Filter> {
            None
        }
    }

    impl TableSchema for AutoincUser {
        type Record = AutoincUserRecord;
        type Insert = AutoincUserInsert;
        type Update = AutoincUserUpdate;
        type ForeignFetcher = NoForeignFetcher;

        fn table_name() -> &'static str {
            "autoinc_users"
        }

        fn columns() -> &'static [ColumnDef] {
            use wasm_dbms_api::prelude::DataTypeKind;

            &[ColumnDef {
                name: "id",
                data_type: DataTypeKind::Uint32,
                auto_increment: true,
                nullable: false,
                primary_key: true,
                unique: true,
                foreign_key: None,
                default: None,
                renamed_from: &[],
            }]
        }

        fn primary_key() -> &'static str {
            "id"
        }

        fn indexes() -> &'static [IndexDef] {
            &[IndexDef(&["id"])]
        }

        fn to_values(self) -> Vec<(ColumnDef, Value)> {
            vec![]
        }

        fn sanitizer(
            _column_name: &'static str,
        ) -> Option<Box<dyn wasm_dbms_api::prelude::Sanitize>> {
            None
        }

        fn validator(
            _column_name: &'static str,
        ) -> Option<Box<dyn wasm_dbms_api::prelude::Validate>> {
            None
        }
    }

    // -- next_autoincrement tests --

    #[test]
    fn test_next_autoincrement_returns_sequential_values() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry_with_autoincrement(&mut mm);

        let v1 = registry
            .next_autoincrement("id", &mut mm)
            .expect("failed")
            .expect("expected Some");
        let v2 = registry
            .next_autoincrement("id", &mut mm)
            .expect("failed")
            .expect("expected Some");
        let v3 = registry
            .next_autoincrement("id", &mut mm)
            .expect("failed")
            .expect("expected Some");

        assert_eq!(v1, Value::Uint32(1u32.into()));
        assert_eq!(v2, Value::Uint32(2u32.into()));
        assert_eq!(v3, Value::Uint32(3u32.into()));
    }

    #[test]
    fn test_next_autoincrement_returns_none_without_ledger() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry_without_autoincrement(&mut mm);

        let result = registry.next_autoincrement("id", &mut mm).expect("failed");
        assert!(result.is_none());
    }

    #[test]
    fn test_next_autoincrement_persists_across_reload() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());

        use crate::SchemaRegistry;
        let mut schema = SchemaRegistry::load(&mut mm).expect("failed to load schema");
        let pages = schema
            .register_table::<AutoincUser>(&mut mm)
            .expect("failed to register table");

        // advance 5 times
        let mut registry = TableRegistry::load(pages, &mut mm).expect("failed to load");
        for _ in 0..5 {
            let _ = registry
                .next_autoincrement("id", &mut mm)
                .expect("next failed");
        }

        // reload the registry from the same pages
        let mut reloaded = TableRegistry::load(pages, &mut mm).expect("failed to reload");
        let value = reloaded
            .next_autoincrement("id", &mut mm)
            .expect("failed")
            .expect("expected Some");
        assert_eq!(value, Value::Uint32(6u32.into()));
    }

    #[test]
    fn test_next_autoincrement_overflow_returns_error() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());

        // manually set up a Uint8 autoincrement to hit overflow quickly
        let schema_snapshot_page = mm.claim_page().expect("failed to get page");
        let page_ledger_page = mm.claim_page().expect("failed to get page");
        let free_segments_page = mm.claim_page().expect("failed to get page");
        let index_registry_page = mm.claim_page().expect("failed to get page");
        let autoinc_page = mm.claim_page().expect("failed to get page");

        // Use the Uint8AutoincTable from the autoincrement_ledger tests — we replicate the
        // TableSchema inline since it's in a sibling test module.
        // Instead, just init the ledger page directly with a Uint8 value.
        {
            let mut registry_data = super::autoincrement_ledger::AutoincrementLedger::init::<
                Uint8AutoincSchema,
            >(autoinc_page, &mut mm)
            .expect("failed to init autoinc ledger");

            // advance to 255
            for _ in 0..255 {
                let _ = registry_data.next("val", &mut mm).expect("next failed");
            }
        }

        // init index ledger
        IndexLedger::init(index_registry_page, &[], &mut mm).expect("failed to init index ledger");

        super::test_utils::write_dummy_schema_snapshot(schema_snapshot_page, &mut mm);

        let table_pages = TableRegistryPage {
            schema_snapshot_page,
            pages_list_page: page_ledger_page,
            free_segments_page,
            index_registry_page,
            autoincrement_registry_page: Some(autoinc_page),
        };

        let mut registry = TableRegistry::load(table_pages, &mut mm).expect("failed to load");
        let result = registry.next_autoincrement("val", &mut mm);
        assert!(result.is_err());
        assert!(matches!(
            result.unwrap_err(),
            wasm_dbms_api::prelude::MemoryError::AutoincrementOverflow(_)
        ));
    }

    // Minimal Uint8 autoincrement schema for overflow test

    #[derive(Clone, CandidType)]
    struct Uint8AutoincSchema;

    impl Encode for Uint8AutoincSchema {
        const SIZE: wasm_dbms_api::prelude::DataSize = wasm_dbms_api::prelude::DataSize::Dynamic;
        const ALIGNMENT: PageOffset = wasm_dbms_api::prelude::DEFAULT_ALIGNMENT;

        fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
            std::borrow::Cow::Owned(vec![])
        }

        fn decode(_data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
        where
            Self: Sized,
        {
            Ok(Self)
        }

        fn size(&self) -> wasm_dbms_api::prelude::MSize {
            0
        }
    }

    #[derive(Clone, CandidType, Deserialize)]
    struct Uint8AutoincSchemaRecord;

    impl TableRecord for Uint8AutoincSchemaRecord {
        type Schema = Uint8AutoincSchema;

        fn from_values(_values: TableColumns) -> Self {
            Self
        }

        fn to_values(&self) -> Vec<(ColumnDef, Value)> {
            vec![]
        }
    }

    #[derive(Clone, CandidType, Serialize)]
    struct Uint8AutoincSchemaInsert;

    impl InsertRecord for Uint8AutoincSchemaInsert {
        type Record = Uint8AutoincSchemaRecord;
        type Schema = Uint8AutoincSchema;

        fn from_values(_values: &[(ColumnDef, Value)]) -> DbmsResult<Self> {
            Ok(Self)
        }

        fn into_values(self) -> Vec<(ColumnDef, Value)> {
            vec![]
        }

        fn into_record(self) -> Self::Schema {
            Uint8AutoincSchema
        }
    }

    #[derive(Clone, CandidType, Serialize)]
    struct Uint8AutoincSchemaUpdate;

    impl UpdateRecord for Uint8AutoincSchemaUpdate {
        type Record = Uint8AutoincSchemaRecord;
        type Schema = Uint8AutoincSchema;

        fn from_values(
            _values: &[(ColumnDef, Value)],
            _where_clause: Option<wasm_dbms_api::prelude::Filter>,
        ) -> wasm_dbms_api::prelude::DbmsResult<Self> {
            Ok(Self)
        }

        fn update_values(&self) -> Vec<(ColumnDef, Value)> {
            vec![]
        }

        fn where_clause(&self) -> Option<wasm_dbms_api::prelude::Filter> {
            None
        }
    }

    impl TableSchema for Uint8AutoincSchema {
        type Record = Uint8AutoincSchemaRecord;
        type Insert = Uint8AutoincSchemaInsert;
        type Update = Uint8AutoincSchemaUpdate;
        type ForeignFetcher = NoForeignFetcher;

        fn table_name() -> &'static str {
            "uint8_autoinc_schema"
        }

        fn columns() -> &'static [ColumnDef] {
            use wasm_dbms_api::prelude::DataTypeKind;

            &[ColumnDef {
                name: "val",
                data_type: DataTypeKind::Uint8,
                auto_increment: true,
                nullable: false,
                primary_key: true,
                unique: true,
                foreign_key: None,
                default: None,
                renamed_from: &[],
            }]
        }

        fn primary_key() -> &'static str {
            "val"
        }

        fn indexes() -> &'static [IndexDef] {
            &[IndexDef(&["val"])]
        }

        fn to_values(self) -> Vec<(ColumnDef, Value)> {
            vec![]
        }

        fn sanitizer(
            _column_name: &'static str,
        ) -> Option<Box<dyn wasm_dbms_api::prelude::Sanitize>> {
            None
        }

        fn validator(
            _column_name: &'static str,
        ) -> Option<Box<dyn wasm_dbms_api::prelude::Validate>> {
            None
        }
    }

    use wasm_dbms_api::prelude::{DataSize, DecodeError, MSize, MemoryError};

    /// A fixed-size record for regression testing (issue #80).
    ///
    /// Layout: u64 (8) + u64 (8) + u8 (1) = 17 bytes, all fixed-size.
    #[derive(Debug, Clone, PartialEq, Eq)]
    struct FixedSizeRecord {
        id: u64,
        timestamp: u64,
        tag: u8,
    }

    impl Encode for FixedSizeRecord {
        const SIZE: DataSize = DataSize::Fixed(17);
        const ALIGNMENT: PageOffset = 17;

        fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
            let mut buf = Vec::with_capacity(17);
            buf.extend_from_slice(&self.id.to_le_bytes());
            buf.extend_from_slice(&self.timestamp.to_le_bytes());
            buf.push(self.tag);
            std::borrow::Cow::Owned(buf)
        }

        fn decode(data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
        where
            Self: Sized,
        {
            if data.len() < 17 {
                return Err(MemoryError::DecodeError(DecodeError::TooShort));
            }
            let id = u64::from_le_bytes(data[0..8].try_into().unwrap());
            let timestamp = u64::from_le_bytes(data[8..16].try_into().unwrap());
            let tag = data[16];
            Ok(Self { id, timestamp, tag })
        }

        fn size(&self) -> MSize {
            17
        }
    }

    #[test]
    fn test_should_insert_multiple_fixed_size_records() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        for i in 0..10 {
            let record = FixedSizeRecord {
                id: i,
                timestamp: 1000 + i,
                tag: (i % 2) as u8,
            };
            registry
                .insert(record, &mut mm)
                .unwrap_or_else(|e| panic!("failed to insert record {i}: {e}"));
        }

        // verify all records can be read back
        let mut reader = registry.read::<FixedSizeRecord, _>(&mut mm);
        let mut count = 0;
        while let Some(next) = reader.try_next().expect("failed to read") {
            assert_eq!(next.record.id, count);
            count += 1;
        }
        assert_eq!(count, 10);
    }

    #[test]
    fn test_reader_skips_deleted_fixed_size_record() {
        use wasm_dbms_api::prelude::Uint32;

        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let first = registry.insert(Uint32(11), &mut mm).expect("insert 11");
        registry.insert(Uint32(22), &mut mm).expect("insert 22");
        registry
            .delete(Uint32(11), first, &mut mm)
            .expect("delete 11");

        let mut reader = registry.read::<Uint32, _>(&mut mm);
        let mut rows = Vec::new();
        while let Some(next) = reader.try_next().expect("scan") {
            rows.push(next.record);
        }
        assert_eq!(rows, vec![Uint32(22)]);
    }

    /// Dynamic 4-byte record with alignment 10 (raw footprint 6, padded
    /// to 10), so the last slot of a page fits unpadded but not padded.
    #[derive(Debug, Clone, PartialEq, Eq)]
    struct TenAlignedRecord(u32);

    impl Encode for TenAlignedRecord {
        const SIZE: DataSize = DataSize::Dynamic;
        const ALIGNMENT: PageOffset = 10;

        fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
            std::borrow::Cow::Owned(self.0.to_le_bytes().to_vec())
        }

        fn decode(data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
        where
            Self: Sized,
        {
            if data.len() < 4 {
                return Err(MemoryError::DecodeError(DecodeError::TooShort));
            }
            Ok(Self(u32::from_le_bytes(data[0..4].try_into().unwrap())))
        }

        fn size(&self) -> MSize {
            4
        }
    }

    #[test]
    fn test_insert_moves_to_new_page_when_padding_would_cross_page_end() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        // 6_553 padded slots fill bytes 0..65_530; record 6_554 fits there
        // unpadded (65_536) but not padded (65_540).
        const COUNT: u32 = 6_554;
        for id in 0..COUNT {
            registry
                .insert(TenAlignedRecord(id), &mut mm)
                .unwrap_or_else(|e| panic!("failed to insert record {id}: {e}"));
        }
        assert_eq!(registry.page_ledger.pages().len(), 2);

        let mut reader = registry.read::<TenAlignedRecord, _>(&mut mm);
        let mut count = 0;
        while let Some(next) = reader.try_next().expect("scan") {
            assert_eq!(next.record, TenAlignedRecord(count));
            count += 1;
        }
        assert_eq!(count, COUNT);
    }

    #[test]
    fn test_insert_raw_rejects_footprint_of_a_full_page() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        // 2 + 65_503 = 65_505 bytes, aligned to 32 = 65_536: not an MSize.
        let result = registry.insert_raw(&vec![0u8; 65_503], 32, &mut mm);
        assert!(matches!(
            result,
            Err(MemoryError::DataTooLarge {
                requested: 65_536,
                ..
            })
        ));
    }

    #[test]
    fn test_insert_raw_accepts_largest_representable_footprint() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        // 2 + 65_502 = 65_504 bytes, already a multiple of 32.
        let payload = vec![7u8; 65_502];
        let address = registry
            .insert_raw(&payload, 32, &mut mm)
            .expect("65_504-byte footprint fits");
        assert_eq!(
            registry.read_raw_at(address, &mut mm).expect("read back"),
            payload
        );
    }

    #[test]
    fn test_delete_raw_rejects_unrepresentable_footprint() {
        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let result = registry.delete_raw(RecordAddress::new(0, 0), MSize::MAX, 32, &mut mm);
        assert!(matches!(result, Err(MemoryError::DataTooLarge { .. })));
    }

    #[test]
    fn test_insert_rejects_record_that_does_not_fit_a_page() {
        use wasm_dbms_api::prelude::Text;

        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        // 65_533 fits the Text prefix; the others saturate `size()`.
        for len in [65_533, 65_534, 65_536, 70_000] {
            let result = registry.insert(Text("x".repeat(len)), &mut mm);
            assert!(
                matches!(result, Err(MemoryError::DataTooLarge { .. })),
                "{len}-byte text: expected DataTooLarge, got {result:?}"
            );
        }
        assert!(registry.page_ledger.pages().is_empty());
    }

    #[test]
    fn test_insert_rejects_footprint_of_a_full_page() {
        use wasm_dbms_api::prelude::Text;

        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        // header(2) + prefix(2) + 65_501 = 65_505, aligned to 32 = 65_536:
        // not an MSize, so the free-segment size would wrap on delete.
        let result = registry.insert(Text("x".repeat(65_501)), &mut mm);
        assert!(matches!(
            result,
            Err(MemoryError::DataTooLarge {
                requested: 65_536,
                ..
            })
        ));

        // header(2) + prefix(2) + 65_500 = 65_504: largest storable record.
        let text = Text("x".repeat(65_500));
        let address = registry
            .insert(text.clone(), &mut mm)
            .expect("65_504-byte footprint fits");
        let read: Text = registry.read_at(address, &mut mm).expect("read back");
        assert_eq!(read, text);
    }

    #[test]
    fn test_update_rejects_record_that_does_not_fit_a_page() {
        use wasm_dbms_api::prelude::Text;

        let mut mm = MemoryManager::init(HeapMemoryProvider::default());
        let mut registry = registry(&mut mm);

        let old = Text("small".to_string());
        let address = registry.insert(old.clone(), &mut mm).expect("insert");

        let result = registry.update(Text("x".repeat(70_000)), old.clone(), address, &mut mm);
        assert!(
            matches!(result, Err(MemoryError::DataTooLarge { .. })),
            "expected DataTooLarge, got {result:?}"
        );

        let read: Text = registry.read_at(address, &mut mm).expect("read back");
        assert_eq!(read, old);
    }
}