wasm_dbms_memory/
schema_registry.rs

1// Rust guideline compliant 2026-02-28
2
3use std::collections::HashMap;
4
5use wasm_dbms_api::memory::MemoryError;
6use wasm_dbms_api::prelude::{
7    DEFAULT_ALIGNMENT, DataSize, DecodeError, Encode, MSize, MemoryResult, Page, PageOffset,
8    TableFingerprint, TableSchema, TableSchemaSnapshot, fingerprint_for_name,
9};
10use xxhash_rust::xxh3::Xxh3;
11
12use crate::memory_manager::{SCHEMA_PAGE, UNCLAIMED_PAGES_PAGE};
13use crate::table_registry::{AutoincrementLedger, IndexLedger, SchemaSnapshotLedger};
14use crate::{MemoryAccess, TableRegistry, UnclaimedPages};
15
16/// The dictionary of tables, mapping the table schema fingerprint to the pages where the table data and metadata are stored.
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub struct TableRegistryPage {
19    /// The page where the Schema Snapshot for this table is stored.
20    pub schema_snapshot_page: Page,
21    /// The page where the list of pages for this table is stored.
22    pub pages_list_page: Page,
23    /// The page where the free segments for this table are stored.
24    pub free_segments_page: Page,
25    /// The page where the index registry for this table is stored.
26    pub index_registry_page: Page,
27    /// The page where the autoincrement registry for this table is stored.
28    /// Only used if the table has an autoincrement column.
29    pub autoincrement_registry_page: Option<Page>,
30}
31
32/// The schema registry takes care of storing and retrieving table schemas from memory.
33#[derive(Debug, Default, Clone, PartialEq, Eq)]
34pub struct SchemaRegistry {
35    schema_hash: u64,
36    tables: HashMap<TableFingerprint, TableRegistryPage>,
37}
38
39impl SchemaRegistry {
40    /// Load the schema registry from memory.
41    pub fn load(mm: &mut impl MemoryAccess) -> MemoryResult<Self> {
42        let registry: Self = mm.read_at(SCHEMA_PAGE, 0)?;
43        Ok(registry)
44    }
45
46    /// Returns the cached hash of every persisted schema snapshot.
47    pub const fn schema_hash(&self) -> u64 {
48        self.schema_hash
49    }
50
51    /// Registers a table and allocates it registry page.
52    ///
53    /// The [`TableSchema`] type parameter is used to get the [`TableSchema::fingerprint`] of the
54    /// table schema. The fingerprint is derived from the table name, so two distinct names that
55    /// hash to the same value are detected eagerly: when the slot is already occupied by a table
56    /// with a different name, [`MemoryError::NameCollision`] is returned and no allocation is
57    /// performed.
58    ///
59    /// # Errors
60    ///
61    /// - [`MemoryError::NameCollision`] when the fingerprint slot is occupied by a table whose
62    ///   persisted snapshot carries a different name.
63    /// - [`MemoryError::ConstraintViolation`] when the table's schema snapshot exceeds a limit of
64    ///   the snapshot format (see [`TableSchemaSnapshot::validate_encoding`]); no page is claimed.
65    /// - Any [`MemoryError`] propagated from page allocation, snapshot init, or the registry
66    ///   write-back.
67    pub fn register_table<TS>(
68        &mut self,
69        mm: &mut impl MemoryAccess,
70    ) -> MemoryResult<TableRegistryPage>
71    where
72        TS: TableSchema,
73    {
74        // check if already registered, and detect name-hash collisions eagerly
75        let fingerprint = TS::fingerprint();
76        let candidate_name = TS::table_name();
77        if let Some(pages) = self.tables.get(&fingerprint).copied() {
78            let existing = SchemaSnapshotLedger::load(pages.schema_snapshot_page, mm)?;
79            if existing.get().name != candidate_name {
80                return Err(MemoryError::NameCollision {
81                    candidate: candidate_name.to_string(),
82                    existing: existing.get().name.clone(),
83                });
84            }
85            return Ok(pages);
86        }
87
88        // Reject metadata the snapshot and ledger encodings cannot hold before
89        // any page is claimed or the registry is touched.
90        TS::schema_snapshot().validate_encoding()?;
91        for index in TS::indexes() {
92            IndexLedger::validate_columns(index.0)?;
93        }
94        AutoincrementLedger::validate::<TS>()?;
95
96        // allocate table registry page
97        let schema_snapshot_page = mm.claim_page()?;
98        let pages_list_page = mm.claim_page()?;
99        let free_segments_page = mm.claim_page()?;
100        let index_registry_page = mm.claim_page()?;
101        // allocate autoincrement registry page if needed
102        let has_autoincrement = TS::columns().iter().any(|col| col.auto_increment);
103        let autoincrement_registry_page = if has_autoincrement {
104            Some(mm.claim_page()?)
105        } else {
106            None
107        };
108
109        // insert into tables map
110        let pages = TableRegistryPage {
111            schema_snapshot_page,
112            pages_list_page,
113            free_segments_page,
114            index_registry_page,
115            autoincrement_registry_page,
116        };
117        self.tables.insert(fingerprint, pages);
118
119        // init snapshot ledger for this table
120        SchemaSnapshotLedger::init::<TS>(pages.schema_snapshot_page, mm)?;
121        // init index ledger for this table
122        IndexLedger::init(pages.index_registry_page, TS::indexes(), mm)?;
123        // init autoincrement ledger for this table if needed
124        if let Some(autoinc_page) = pages.autoincrement_registry_page {
125            AutoincrementLedger::init::<TS>(autoinc_page, mm)?;
126        }
127
128        self.refresh_schema_hash(mm)?;
129        self.save(mm)?;
130
131        Ok(pages)
132    }
133
134    /// Save the schema registry to memory.
135    pub fn save(&self, mm: &mut impl MemoryAccess) -> MemoryResult<()> {
136        mm.write_at(SCHEMA_PAGE, 0, self)
137    }
138
139    /// Returns the table registry page for a given table schema.
140    pub fn table_registry_page<TS>(&self) -> Option<TableRegistryPage>
141    where
142        TS: TableSchema,
143    {
144        self.tables.get(&TS::fingerprint()).copied()
145    }
146
147    /// Returns the table registry page for the table with the given name.
148    ///
149    /// Used by the migration engine, which knows tables only by name when
150    /// applying ops decoded from snapshots.
151    pub fn table_registry_page_by_name(&self, name: &str) -> Option<TableRegistryPage> {
152        self.tables.get(&fingerprint_for_name(name)).copied()
153    }
154
155    /// Registers a table from a snapshot, allocating its registry pages.
156    ///
157    /// The migration engine uses this entry point when applying a
158    /// `MigrationOp::CreateTable`: the source of truth is a
159    /// [`TableSchemaSnapshot`], not a `TableSchema` type.
160    ///
161    /// # Errors
162    ///
163    /// - [`MemoryError::NameCollision`] when the fingerprint slot is occupied
164    ///   by a table with a different name.
165    /// - [`MemoryError::ConstraintViolation`] when the snapshot exceeds a limit
166    ///   of the snapshot format (see
167    ///   [`TableSchemaSnapshot::validate_encoding`]) or declares an
168    ///   unsupported autoincrement column; no page is claimed.
169    /// - Any [`MemoryError`] propagated from page allocation, snapshot init,
170    ///   index init, or registry persistence.
171    pub fn register_table_from_snapshot(
172        &mut self,
173        snapshot: &TableSchemaSnapshot,
174        mm: &mut impl MemoryAccess,
175    ) -> MemoryResult<TableRegistryPage> {
176        let fingerprint = fingerprint_for_name(&snapshot.name);
177        let candidate_name = snapshot.name.as_str();
178        if let Some(pages) = self.tables.get(&fingerprint).copied() {
179            let existing = SchemaSnapshotLedger::load(pages.schema_snapshot_page, mm)?;
180            if existing.get().name != candidate_name {
181                return Err(MemoryError::NameCollision {
182                    candidate: candidate_name.to_string(),
183                    existing: existing.get().name.clone(),
184                });
185            }
186            return Ok(pages);
187        }
188
189        // Reject unsupported autoincrement columns and metadata the ledger
190        // encodings cannot hold before any page is claimed or the registry is
191        // touched.
192        snapshot.validate_encoding()?;
193        AutoincrementLedger::validate_snapshot(snapshot)?;
194        for index in &snapshot.indexes {
195            IndexLedger::validate_columns(&index.columns)?;
196        }
197
198        let schema_snapshot_page = mm.claim_page()?;
199        let pages_list_page = mm.claim_page()?;
200        let free_segments_page = mm.claim_page()?;
201        let index_registry_page = mm.claim_page()?;
202        let has_autoincrement = snapshot.columns.iter().any(|col| col.auto_increment);
203        let autoincrement_registry_page = if has_autoincrement {
204            Some(mm.claim_page()?)
205        } else {
206            None
207        };
208
209        let pages = TableRegistryPage {
210            schema_snapshot_page,
211            pages_list_page,
212            free_segments_page,
213            index_registry_page,
214            autoincrement_registry_page,
215        };
216        self.tables.insert(fingerprint, pages);
217
218        mm.write_at(pages.schema_snapshot_page, 0, snapshot)?;
219        IndexLedger::init_from_keys(
220            pages.index_registry_page,
221            snapshot.indexes.iter().map(|idx| idx.columns.clone()),
222            mm,
223        )?;
224        if let Some(autoinc_page) = pages.autoincrement_registry_page {
225            AutoincrementLedger::init_from_snapshot(autoinc_page, snapshot, mm)?;
226        }
227        self.refresh_schema_hash(mm)?;
228        self.save(mm)?;
229
230        Ok(pages)
231    }
232
233    /// Removes the table identified by `name` from the registry, releases
234    /// every page it owned back to the unclaimed-pages ledger, and persists
235    /// the updated registry.
236    ///
237    /// Used by the migration engine when applying a `MigrationOp::DropTable`.
238    ///
239    /// Returns the [`TableRegistryPage`] previously associated with the table,
240    /// or `None` if no such table was registered.
241    ///
242    /// # Errors
243    ///
244    /// Returns a [`MemoryError`] if loading the table registry, releasing
245    /// its pages, or persisting the updated schema registry fails.
246    pub fn unregister_table(
247        &mut self,
248        name: &str,
249        mm: &mut impl MemoryAccess,
250    ) -> MemoryResult<Option<TableRegistryPage>> {
251        let fingerprint = fingerprint_for_name(name);
252        let pages = self.tables.get(&fingerprint).copied();
253        if let Some(pages) = pages {
254            // Release every page owned by the dropped table before persisting
255            // the updated schema registry — once the schema page is rewritten
256            // there is no way back to the table's `TableRegistryPage`.
257            let registry = TableRegistry::load(pages, mm)?;
258            let pages_to_release = registry.releasable_pages_count(pages, mm)?;
259            let ledger: UnclaimedPages = mm.read_at(UNCLAIMED_PAGES_PAGE, 0)?;
260            if ledger.remaining_capacity() < pages_to_release as u32 {
261                return Err(MemoryError::UnclaimedPagesFull {
262                    capacity: crate::UNCLAIMED_PAGES_CAPACITY,
263                });
264            }
265
266            let removed = self.tables.remove(&fingerprint);
267            debug_assert_eq!(removed, Some(pages));
268            registry.release_pages(pages, mm)?;
269            self.refresh_schema_hash(mm)?;
270            self.save(mm)?;
271            return Ok(removed);
272        }
273        Ok(None)
274    }
275
276    /// Returns the persisted [`TableSchemaSnapshot`] for every registered table.
277    ///
278    /// The order is unspecified. Callers that need a stable order (e.g. for
279    /// drift hashing) must sort by [`TableSchemaSnapshot::name`].
280    ///
281    /// # Errors
282    ///
283    /// Returns the first [`MemoryError`] encountered while loading any
284    /// snapshot page.
285    pub fn stored_snapshots(
286        &self,
287        mm: &mut impl MemoryAccess,
288    ) -> MemoryResult<Vec<TableSchemaSnapshot>> {
289        self.tables
290            .values()
291            .map(|pages| {
292                SchemaSnapshotLedger::load(pages.schema_snapshot_page, mm)
293                    .map(|ledger| ledger.get().clone())
294            })
295            .collect()
296    }
297
298    /// Recomputes the cached schema hash from the currently registered tables.
299    pub fn refresh_schema_hash(&mut self, mm: &mut impl MemoryAccess) -> MemoryResult<()> {
300        self.schema_hash = compute_hash(self.stored_snapshots(mm)?);
301        Ok(())
302    }
303}
304
305fn compute_hash(mut snapshots: Vec<TableSchemaSnapshot>) -> u64 {
306    snapshots.sort_by(|a, b| a.name.cmp(&b.name));
307
308    let mut hasher = Xxh3::new();
309    hasher.update(&[TableSchemaSnapshot::latest_version()]);
310    for snapshot in &snapshots {
311        let bytes = snapshot.encode();
312        hasher.update(&(bytes.len() as u64).to_le_bytes());
313        hasher.update(&bytes);
314    }
315    hasher.digest()
316}
317
318/// Smallest encoded registry entry: fingerprint (8), four pages (16) and the
319/// autoincrement flag (1).
320const MIN_ENTRY_SIZE: usize = 25;
321
322/// Reads `N` bytes at `offset` and advances it.
323///
324/// # Errors
325///
326/// Returns [`DecodeError::TooShort`] if fewer than `N` bytes remain.
327fn read_array<const N: usize>(data: &[u8], offset: &mut usize) -> MemoryResult<[u8; N]> {
328    let end = offset
329        .checked_add(N)
330        .filter(|end| *end <= data.len())
331        .ok_or(MemoryError::DecodeError(DecodeError::TooShort))?;
332    let bytes = data[*offset..end].try_into()?;
333    *offset = end;
334    Ok(bytes)
335}
336
337impl Encode for SchemaRegistry {
338    const SIZE: DataSize = DataSize::Dynamic;
339
340    const ALIGNMENT: PageOffset = DEFAULT_ALIGNMENT;
341
342    fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
343        // prepare buffer; size is 8 bytes for schema_hash + 8 bytes for len
344        // + fixed-size fields for each entry.
345        let mut buffer = Vec::with_capacity(self.size() as usize);
346        buffer.extend_from_slice(&self.schema_hash.to_le_bytes());
347        // write 8 bytes len of map
348        buffer.extend_from_slice(&(self.tables.len() as u64).to_le_bytes());
349        // write each entry
350        for (fingerprint, page) in &self.tables {
351            buffer.extend_from_slice(&fingerprint.to_le_bytes());
352            buffer.extend_from_slice(&page.schema_snapshot_page.to_le_bytes());
353            buffer.extend_from_slice(&page.pages_list_page.to_le_bytes());
354            buffer.extend_from_slice(&page.free_segments_page.to_le_bytes());
355            buffer.extend_from_slice(&page.index_registry_page.to_le_bytes());
356            // autoincrement registry page is optional, so we write a flag and then the page if it exists
357            if let Some(autoinc_page) = page.autoincrement_registry_page {
358                buffer.push(1); // flag for presence of autoincrement registry page
359                buffer.extend_from_slice(&autoinc_page.to_le_bytes());
360            } else {
361                buffer.push(0); // flag for absence of autoincrement registry page
362            }
363        }
364        std::borrow::Cow::Owned(buffer)
365    }
366
367    fn decode(data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
368    where
369        Self: Sized,
370    {
371        let mut offset = 0;
372        let schema_hash = u64::from_le_bytes(read_array(&data, &mut offset)?);
373        let len = u64::from_le_bytes(read_array(&data, &mut offset)?);
374        // Reject counts the payload cannot hold before allocating for them.
375        let max_entries = (data.len() - offset) / MIN_ENTRY_SIZE;
376        let len = usize::try_from(len)
377            .ok()
378            .filter(|len| *len <= max_entries)
379            .ok_or(MemoryError::DecodeError(DecodeError::TooShort))?;
380        let mut tables = HashMap::with_capacity(len);
381        // read each entry
382        for _ in 0..len {
383            let fingerprint = u64::from_le_bytes(read_array(&data, &mut offset)?);
384            let schema_snapshot_page = Page::from_le_bytes(read_array(&data, &mut offset)?);
385            let pages_list_page = Page::from_le_bytes(read_array(&data, &mut offset)?);
386            let free_segments_page = Page::from_le_bytes(read_array(&data, &mut offset)?);
387            let index_registry_page = Page::from_le_bytes(read_array(&data, &mut offset)?);
388            let [has_autoincrement] = read_array::<1>(&data, &mut offset)?;
389            let autoincrement_registry_page = if has_autoincrement == 1 {
390                Some(Page::from_le_bytes(read_array(&data, &mut offset)?))
391            } else {
392                None
393            };
394            tables.insert(
395                fingerprint,
396                TableRegistryPage {
397                    schema_snapshot_page,
398                    pages_list_page,
399                    free_segments_page,
400                    index_registry_page,
401                    autoincrement_registry_page,
402                },
403            );
404        }
405        Ok(Self {
406            schema_hash,
407            tables,
408        })
409    }
410
411    fn size(&self) -> MSize {
412        // - 8 bytes for `self.schema_hash`
413        // - 8 bytes for `self.tables.len()`
414        // - for each entry:
415        //  - 8 bytes for the fingerprint
416        //  - 4 bytes for the schema_snapshot_page
417        //  - 4 bytes for the pages_list_page
418        //  - 4 bytes for the free_segments_page
419        //  - 4 bytes for the index_registry_page
420        //  - 1 byte for the autoincrement registry page flag
421        //  - 4 bytes for the autoincrement registry page if it exists
422        let autoinc_pages = self
423            .tables
424            .values()
425            .filter(|page| page.autoincrement_registry_page.is_some())
426            .count() as MSize;
427
428        16 + (self.tables.len() as MSize * (4 * 4 + 8 + 1)) + (autoinc_pages * 4)
429    }
430}
431
432#[cfg(test)]
433mod tests {
434
435    use candid::CandidType;
436    use serde::{Deserialize, Serialize};
437    use wasm_dbms_api::prelude::{
438        ColumnDef, DbmsResult, IndexDef, InsertRecord, Int32, NoForeignFetcher, TableColumns,
439        TableRecord, UpdateRecord,
440    };
441
442    use super::*;
443    use crate::{
444        HeapMemoryProvider, MemoryAccess, MemoryManager, RecordAddress, UNCLAIMED_PAGES_CAPACITY,
445        UnclaimedPages,
446    };
447
448    fn make_mm() -> MemoryManager<HeapMemoryProvider> {
449        MemoryManager::init(HeapMemoryProvider::default())
450    }
451
452    #[test]
453    fn test_should_encode_and_decode_schema_registry() {
454        let mut mm = make_mm();
455
456        // load
457        let mut registry =
458            SchemaRegistry::load(&mut mm).expect("failed to load init schema registry");
459
460        // register table
461        let registry_page = registry
462            .register_table::<User>(&mut mm)
463            .expect("failed to register table");
464
465        // get table registry page
466        let fetched_page = registry
467            .table_registry_page::<User>()
468            .expect("failed to get table registry page");
469        assert_eq!(registry_page, fetched_page);
470
471        // encode
472        let encoded = registry.encode();
473        // decode
474        let decoded = SchemaRegistry::decode(encoded).expect("failed to decode");
475        assert_eq!(registry, decoded);
476
477        // try to actually add another
478        let another_registry_page = registry
479            .register_table::<AnotherTable>(&mut mm)
480            .expect("failed to register another table");
481        let another_fetched_page = registry
482            .table_registry_page::<AnotherTable>()
483            .expect("failed to get another table registry page");
484        assert_eq!(another_registry_page, another_fetched_page);
485
486        // re-init
487        let reloaded = SchemaRegistry::load(&mut mm).expect("failed to reload schema registry");
488        assert_eq!(registry, reloaded);
489        // should have two
490        assert_eq!(reloaded.tables.len(), 2);
491        assert_eq!(
492            reloaded
493                .table_registry_page::<User>()
494                .expect("failed to get first table registry page after reload"),
495            registry_page
496        );
497        assert_eq!(
498            reloaded
499                .table_registry_page::<AnotherTable>()
500                .expect("failed to get second table registry page after reload"),
501            another_registry_page
502        );
503    }
504
505    #[test]
506    fn test_register_table_writes_snapshot_to_ledger() {
507        let mut mm = make_mm();
508        let mut registry = SchemaRegistry::default();
509
510        let pages = registry
511            .register_table::<User>(&mut mm)
512            .expect("failed to register table");
513
514        let ledger = SchemaSnapshotLedger::load(pages.schema_snapshot_page, &mut mm)
515            .expect("failed to load snapshot ledger after register_table");
516
517        assert_eq!(ledger.get(), &User::schema_snapshot());
518        assert_eq!(ledger.get().name, "users");
519    }
520
521    #[test]
522    fn test_register_table_returns_name_collision_when_hash_slot_belongs_to_another_name() {
523        let mut mm = make_mm();
524        let mut registry = SchemaRegistry::default();
525
526        // register `User` so its snapshot lives on disk
527        let pages = registry
528            .register_table::<User>(&mut mm)
529            .expect("failed to register user");
530
531        // simulate a hash collision by rewriting the persisted snapshot to carry a different name
532        let mut tampered = User::schema_snapshot();
533        tampered.name = "imposter".to_string();
534        mm.write_at(pages.schema_snapshot_page, 0, &tampered)
535            .expect("failed to overwrite snapshot");
536
537        let result = registry.register_table::<User>(&mut mm);
538        match result {
539            Err(MemoryError::NameCollision {
540                candidate,
541                existing,
542            }) => {
543                assert_eq!(candidate, "users");
544                assert_eq!(existing, "imposter");
545            }
546            other => panic!("expected NameCollision, got {other:?}"),
547        }
548    }
549
550    #[test]
551    fn test_should_not_register_same_table_twice() {
552        let mut mm = make_mm();
553        let mut registry = SchemaRegistry::default();
554
555        let first_page = registry
556            .register_table::<User>(&mut mm)
557            .expect("failed to register table first time");
558        let second_page = registry
559            .register_table::<User>(&mut mm)
560            .expect("failed to register table second time");
561
562        assert_eq!(first_page, second_page);
563        assert_eq!(registry.tables.len(), 1);
564    }
565
566    #[test]
567    fn test_should_init_index_ledger() {
568        let mut mm = make_mm();
569        let mut registry = SchemaRegistry::default();
570
571        let pages = registry
572            .register_table::<User>(&mut mm)
573            .expect("failed to register table");
574
575        // check that index ledger is initialized with the correct indexes
576        let mut index_ledger = IndexLedger::load(pages.index_registry_page, &mut mm)
577            .expect("failed to load index ledger");
578
579        // insert an index for id
580        index_ledger
581            .insert(
582                &["id"],
583                Int32::from(1i32),
584                RecordAddress { page: 1, offset: 0 },
585                &mut mm,
586            )
587            .expect("failed to insert index");
588        // search the index
589        let result = index_ledger
590            .search(&["id"], &Int32::from(1i32), &mut mm)
591            .expect("failed to search index")
592            .get(0)
593            .copied()
594            .expect("no index at 0");
595        assert_eq!(result, RecordAddress { page: 1, offset: 0 });
596    }
597
598    #[derive(Clone, CandidType)]
599    struct AnotherTable;
600
601    impl Encode for AnotherTable {
602        const SIZE: DataSize = DataSize::Dynamic;
603
604        const ALIGNMENT: PageOffset = DEFAULT_ALIGNMENT;
605
606        fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
607            std::borrow::Cow::Owned(vec![])
608        }
609
610        fn decode(_data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
611        where
612            Self: Sized,
613        {
614            Ok(AnotherTable)
615        }
616
617        fn size(&self) -> MSize {
618            0
619        }
620    }
621
622    #[derive(Clone, CandidType, Deserialize)]
623    struct AnotherTableRecord;
624
625    impl TableRecord for AnotherTableRecord {
626        type Schema = AnotherTable;
627
628        fn from_values(_values: TableColumns) -> Self {
629            AnotherTableRecord
630        }
631
632        fn to_values(&self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
633            vec![]
634        }
635    }
636
637    #[derive(Clone, CandidType, Serialize)]
638    struct AnotherTableInsert;
639
640    impl InsertRecord for AnotherTableInsert {
641        type Record = AnotherTableRecord;
642        type Schema = AnotherTable;
643
644        fn from_values(_values: &[(ColumnDef, wasm_dbms_api::prelude::Value)]) -> DbmsResult<Self> {
645            Ok(AnotherTableInsert)
646        }
647
648        fn into_values(self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
649            vec![]
650        }
651
652        fn into_record(self) -> Self::Schema {
653            AnotherTable
654        }
655    }
656
657    #[derive(Clone, CandidType, Serialize)]
658    struct AnotherTableUpdate;
659
660    impl UpdateRecord for AnotherTableUpdate {
661        type Record = AnotherTableRecord;
662        type Schema = AnotherTable;
663
664        fn from_values(
665            _values: &[(ColumnDef, wasm_dbms_api::prelude::Value)],
666            _where_clause: Option<wasm_dbms_api::prelude::Filter>,
667        ) -> wasm_dbms_api::prelude::DbmsResult<Self> {
668            Ok(AnotherTableUpdate)
669        }
670
671        fn update_values(&self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
672            vec![]
673        }
674
675        fn where_clause(&self) -> Option<wasm_dbms_api::prelude::Filter> {
676            None
677        }
678    }
679
680    impl TableSchema for AnotherTable {
681        type Record = AnotherTableRecord;
682        type Insert = AnotherTableInsert;
683        type Update = AnotherTableUpdate;
684        type ForeignFetcher = NoForeignFetcher;
685
686        fn table_name() -> &'static str {
687            "another_table"
688        }
689
690        fn columns() -> &'static [wasm_dbms_api::prelude::ColumnDef] {
691            &[]
692        }
693
694        fn primary_key() -> &'static str {
695            ""
696        }
697
698        fn indexes() -> &'static [wasm_dbms_api::prelude::IndexDef] {
699            &[]
700        }
701
702        fn to_values(self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
703            vec![]
704        }
705
706        fn sanitizer(
707            _column_name: &'static str,
708        ) -> Option<Box<dyn wasm_dbms_api::prelude::Sanitize>> {
709            None
710        }
711
712        fn validator(
713            _column_name: &'static str,
714        ) -> Option<Box<dyn wasm_dbms_api::prelude::Validate>> {
715            None
716        }
717    }
718
719    // -- User mock for tests --
720
721    #[derive(Clone, CandidType)]
722    struct User;
723
724    impl Encode for User {
725        const SIZE: DataSize = DataSize::Dynamic;
726        const ALIGNMENT: PageOffset = DEFAULT_ALIGNMENT;
727
728        fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
729            std::borrow::Cow::Owned(vec![])
730        }
731
732        fn decode(_data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
733        where
734            Self: Sized,
735        {
736            Ok(User)
737        }
738
739        fn size(&self) -> MSize {
740            0
741        }
742    }
743
744    #[derive(Clone, CandidType, Deserialize)]
745    struct UserRecord;
746
747    impl TableRecord for UserRecord {
748        type Schema = User;
749
750        fn from_values(_values: TableColumns) -> Self {
751            UserRecord
752        }
753
754        fn to_values(&self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
755            vec![]
756        }
757    }
758
759    #[derive(Clone, CandidType, Serialize)]
760    struct UserInsert;
761
762    impl InsertRecord for UserInsert {
763        type Record = UserRecord;
764        type Schema = User;
765
766        fn from_values(_values: &[(ColumnDef, wasm_dbms_api::prelude::Value)]) -> DbmsResult<Self> {
767            Ok(UserInsert)
768        }
769
770        fn into_values(self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
771            vec![]
772        }
773
774        fn into_record(self) -> Self::Schema {
775            User
776        }
777    }
778
779    #[derive(Clone, CandidType, Serialize)]
780    struct UserUpdate;
781
782    impl UpdateRecord for UserUpdate {
783        type Record = UserRecord;
784        type Schema = User;
785
786        fn from_values(
787            _values: &[(ColumnDef, wasm_dbms_api::prelude::Value)],
788            _where_clause: Option<wasm_dbms_api::prelude::Filter>,
789        ) -> wasm_dbms_api::prelude::DbmsResult<Self> {
790            Ok(UserUpdate)
791        }
792
793        fn update_values(&self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
794            vec![]
795        }
796
797        fn where_clause(&self) -> Option<wasm_dbms_api::prelude::Filter> {
798            None
799        }
800    }
801
802    impl TableSchema for User {
803        type Record = UserRecord;
804        type Insert = UserInsert;
805        type Update = UserUpdate;
806        type ForeignFetcher = NoForeignFetcher;
807
808        fn table_name() -> &'static str {
809            "users"
810        }
811
812        fn columns() -> &'static [wasm_dbms_api::prelude::ColumnDef] {
813            &[]
814        }
815
816        fn primary_key() -> &'static str {
817            "id"
818        }
819
820        fn indexes() -> &'static [wasm_dbms_api::prelude::IndexDef] {
821            &[IndexDef(&["id"])]
822        }
823
824        fn to_values(self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
825            vec![]
826        }
827
828        fn sanitizer(
829            _column_name: &'static str,
830        ) -> Option<Box<dyn wasm_dbms_api::prelude::Sanitize>> {
831            None
832        }
833
834        fn validator(
835            _column_name: &'static str,
836        ) -> Option<Box<dyn wasm_dbms_api::prelude::Validate>> {
837            None
838        }
839    }
840
841    #[test]
842    fn test_table_registry_page_returns_none_for_unregistered_table() {
843        let registry = SchemaRegistry::default();
844        assert!(registry.table_registry_page::<User>().is_none());
845    }
846
847    #[test]
848    fn test_empty_registry_encode_decode() {
849        let registry = SchemaRegistry::default();
850        let encoded = registry.encode();
851        let decoded = SchemaRegistry::decode(encoded).expect("failed to decode empty registry");
852        assert_eq!(registry, decoded);
853        assert_eq!(decoded.tables.len(), 0);
854    }
855
856    #[test]
857    fn test_load_fresh_memory_returns_empty_registry() {
858        let mut mm = make_mm();
859        let registry = SchemaRegistry::load(&mut mm).expect("failed to load from fresh memory");
860        assert_eq!(registry.tables.len(), 0);
861    }
862
863    #[test]
864    fn test_save_and_reload() {
865        let mut mm = make_mm();
866        let mut registry = SchemaRegistry::default();
867        registry
868            .register_table::<User>(&mut mm)
869            .expect("failed to register");
870        // Modify in-memory, then explicitly save
871        registry
872            .register_table::<AnotherTable>(&mut mm)
873            .expect("failed to register another");
874        registry.save(&mut mm).expect("failed to save");
875
876        let reloaded = SchemaRegistry::load(&mut mm).expect("failed to reload");
877        assert_eq!(reloaded.tables.len(), 2);
878        assert_eq!(registry, reloaded);
879    }
880
881    #[test]
882    fn test_schema_registry_size() {
883        let mut mm = make_mm();
884        let mut registry = SchemaRegistry::default();
885        // Empty size: 8 bytes for schema_hash + 8 bytes for table count.
886        assert_eq!(registry.size(), 16);
887        registry
888            .register_table::<User>(&mut mm)
889            .expect("failed to register");
890        // One entry without autoincrement:
891        // 16 + (8 + 4 + 4 + 4 + 4 + 1) = 41
892        // (1 byte for autoincrement flag, no page bytes since User has no autoincrement column)
893        assert_eq!(registry.size(), 41);
894    }
895
896    #[test]
897    fn test_should_allocate_autoincrement_page_when_column_has_autoincrement() {
898        let mut mm = make_mm();
899        let mut registry = SchemaRegistry::default();
900
901        let pages = registry
902            .register_table::<AutoincrementTable>(&mut mm)
903            .expect("failed to register autoincrement table");
904
905        assert!(
906            pages.autoincrement_registry_page.is_some(),
907            "autoincrement registry page should be allocated for tables with autoincrement columns"
908        );
909    }
910
911    #[test]
912    fn test_should_not_allocate_autoincrement_page_when_no_autoincrement_column() {
913        let mut mm = make_mm();
914        let mut registry = SchemaRegistry::default();
915
916        let pages = registry
917            .register_table::<User>(&mut mm)
918            .expect("failed to register user table");
919
920        assert!(
921            pages.autoincrement_registry_page.is_none(),
922            "autoincrement registry page should not be allocated for tables without autoincrement columns"
923        );
924    }
925
926    #[test]
927    fn test_schema_registry_size_with_autoincrement() {
928        let mut mm = make_mm();
929        let mut registry = SchemaRegistry::default();
930
931        registry
932            .register_table::<AutoincrementTable>(&mut mm)
933            .expect("failed to register");
934        // One entry with autoincrement:
935        // 16 + (8 + 4 + 4 + 4 + 4 + 1 + 4) = 45
936        // (1 byte for autoincrement flag + 4 bytes for the autoincrement page)
937        assert_eq!(registry.size(), 45);
938    }
939
940    #[test]
941    fn test_should_encode_and_decode_registry_with_autoincrement() {
942        let mut mm = make_mm();
943        let mut registry = SchemaRegistry::default();
944
945        registry
946            .register_table::<AutoincrementTable>(&mut mm)
947            .expect("failed to register");
948
949        let encoded = registry.encode();
950        let decoded = SchemaRegistry::decode(encoded).expect("failed to decode");
951        assert_eq!(registry, decoded);
952
953        let page = decoded
954            .table_registry_page::<AutoincrementTable>()
955            .expect("missing autoincrement table");
956        assert!(page.autoincrement_registry_page.is_some());
957    }
958
959    // -- AutoincrementTable mock for tests --
960
961    #[derive(Clone, CandidType)]
962    struct AutoincrementTable;
963
964    impl Encode for AutoincrementTable {
965        const SIZE: DataSize = DataSize::Dynamic;
966        const ALIGNMENT: PageOffset = DEFAULT_ALIGNMENT;
967
968        fn encode(&'_ self) -> std::borrow::Cow<'_, [u8]> {
969            std::borrow::Cow::Owned(vec![])
970        }
971
972        fn decode(_data: std::borrow::Cow<[u8]>) -> MemoryResult<Self>
973        where
974            Self: Sized,
975        {
976            Ok(AutoincrementTable)
977        }
978
979        fn size(&self) -> MSize {
980            0
981        }
982    }
983
984    #[derive(Clone, CandidType, Deserialize)]
985    struct AutoincrementTableRecord;
986
987    impl TableRecord for AutoincrementTableRecord {
988        type Schema = AutoincrementTable;
989
990        fn from_values(_values: TableColumns) -> Self {
991            AutoincrementTableRecord
992        }
993
994        fn to_values(&self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
995            vec![]
996        }
997    }
998
999    #[derive(Clone, CandidType, Serialize)]
1000    struct AutoincrementTableInsert;
1001
1002    impl InsertRecord for AutoincrementTableInsert {
1003        type Record = AutoincrementTableRecord;
1004        type Schema = AutoincrementTable;
1005
1006        fn from_values(_values: &[(ColumnDef, wasm_dbms_api::prelude::Value)]) -> DbmsResult<Self> {
1007            Ok(AutoincrementTableInsert)
1008        }
1009
1010        fn into_values(self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
1011            vec![]
1012        }
1013
1014        fn into_record(self) -> Self::Schema {
1015            AutoincrementTable
1016        }
1017    }
1018
1019    #[derive(Clone, CandidType, Serialize)]
1020    struct AutoincrementTableUpdate;
1021
1022    impl UpdateRecord for AutoincrementTableUpdate {
1023        type Record = AutoincrementTableRecord;
1024        type Schema = AutoincrementTable;
1025
1026        fn from_values(
1027            _values: &[(ColumnDef, wasm_dbms_api::prelude::Value)],
1028            _where_clause: Option<wasm_dbms_api::prelude::Filter>,
1029        ) -> wasm_dbms_api::prelude::DbmsResult<Self> {
1030            Ok(AutoincrementTableUpdate)
1031        }
1032
1033        fn update_values(&self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
1034            vec![]
1035        }
1036
1037        fn where_clause(&self) -> Option<wasm_dbms_api::prelude::Filter> {
1038            None
1039        }
1040    }
1041
1042    impl TableSchema for AutoincrementTable {
1043        type Record = AutoincrementTableRecord;
1044        type Insert = AutoincrementTableInsert;
1045        type Update = AutoincrementTableUpdate;
1046        type ForeignFetcher = NoForeignFetcher;
1047
1048        fn table_name() -> &'static str {
1049            "autoincrement_table"
1050        }
1051
1052        fn columns() -> &'static [ColumnDef] {
1053            use wasm_dbms_api::prelude::DataTypeKind;
1054
1055            &[ColumnDef {
1056                name: "id",
1057                data_type: DataTypeKind::Uint32,
1058                auto_increment: true,
1059                nullable: false,
1060                primary_key: true,
1061                unique: true,
1062                foreign_key: None,
1063                default: None,
1064                renamed_from: &[],
1065            }]
1066        }
1067
1068        fn primary_key() -> &'static str {
1069            "id"
1070        }
1071
1072        fn indexes() -> &'static [IndexDef] {
1073            &[IndexDef(&["id"])]
1074        }
1075
1076        fn to_values(self) -> Vec<(ColumnDef, wasm_dbms_api::prelude::Value)> {
1077            vec![]
1078        }
1079
1080        fn sanitizer(
1081            _column_name: &'static str,
1082        ) -> Option<Box<dyn wasm_dbms_api::prelude::Sanitize>> {
1083            None
1084        }
1085
1086        fn validator(
1087            _column_name: &'static str,
1088        ) -> Option<Box<dyn wasm_dbms_api::prelude::Validate>> {
1089            None
1090        }
1091    }
1092
1093    // -- Migration-engine entry points -------------------------------------
1094
1095    use wasm_dbms_api::prelude::{ColumnSnapshot, DataTypeSnapshot, TableSchemaSnapshot};
1096
1097    fn dummy_snapshot(name: &str) -> TableSchemaSnapshot {
1098        TableSchemaSnapshot {
1099            version: TableSchemaSnapshot::latest_version(),
1100            name: name.to_string(),
1101            primary_key: "id".to_string(),
1102            alignment: 8,
1103            columns: vec![ColumnSnapshot {
1104                name: "id".to_string(),
1105                data_type: DataTypeSnapshot::Uint32,
1106                nullable: false,
1107                auto_increment: false,
1108                unique: true,
1109                primary_key: true,
1110                foreign_key: None,
1111                default: None,
1112            }],
1113            indexes: vec![],
1114        }
1115    }
1116
1117    #[test]
1118    fn test_table_registry_page_by_name_returns_pages_for_registered_table() {
1119        let mut mm = make_mm();
1120        let mut registry = SchemaRegistry::default();
1121        let pages = registry
1122            .register_table::<User>(&mut mm)
1123            .expect("failed to register user");
1124
1125        let by_name = registry
1126            .table_registry_page_by_name("users")
1127            .expect("missing pages by name");
1128        assert_eq!(by_name, pages);
1129    }
1130
1131    #[test]
1132    fn test_table_registry_page_by_name_returns_none_for_unknown_table() {
1133        let registry = SchemaRegistry::default();
1134        assert!(registry.table_registry_page_by_name("missing").is_none());
1135    }
1136
1137    #[test]
1138    fn test_stored_snapshots_returns_empty_for_unregistered_registry() {
1139        let mut mm = make_mm();
1140        let registry = SchemaRegistry::default();
1141        let snapshots = registry
1142            .stored_snapshots(&mut mm)
1143            .expect("failed to read snapshots");
1144        assert!(snapshots.is_empty());
1145    }
1146
1147    #[test]
1148    fn test_stored_snapshots_returns_one_entry_per_registered_table() {
1149        let mut mm = make_mm();
1150        let mut registry = SchemaRegistry::default();
1151        registry
1152            .register_table::<User>(&mut mm)
1153            .expect("failed to register user");
1154        registry
1155            .register_table::<AnotherTable>(&mut mm)
1156            .expect("failed to register another");
1157
1158        let snapshots = registry
1159            .stored_snapshots(&mut mm)
1160            .expect("failed to load snapshots");
1161        assert_eq!(snapshots.len(), 2);
1162        let names: Vec<&str> = snapshots.iter().map(|snap| snap.name.as_str()).collect();
1163        assert!(names.contains(&"users"));
1164        assert!(names.contains(&"another_table"));
1165    }
1166
1167    #[test]
1168    fn test_register_table_from_snapshot_allocates_pages_and_persists_snapshot() {
1169        let mut mm = make_mm();
1170        let mut registry = SchemaRegistry::default();
1171        let snapshot = dummy_snapshot("fresh");
1172
1173        let pages = registry
1174            .register_table_from_snapshot(&snapshot, &mut mm)
1175            .expect("failed to register from snapshot");
1176
1177        let loaded = SchemaSnapshotLedger::load(pages.schema_snapshot_page, &mut mm).expect("load");
1178        assert_eq!(loaded.get(), &snapshot);
1179        assert!(registry.table_registry_page_by_name("fresh").is_some());
1180    }
1181
1182    #[test]
1183    fn test_register_table_from_snapshot_is_idempotent_for_same_name() {
1184        let mut mm = make_mm();
1185        let mut registry = SchemaRegistry::default();
1186        let snapshot = dummy_snapshot("fresh");
1187
1188        let first = registry
1189            .register_table_from_snapshot(&snapshot, &mut mm)
1190            .expect("first");
1191        let second = registry
1192            .register_table_from_snapshot(&snapshot, &mut mm)
1193            .expect("second");
1194        assert_eq!(first, second);
1195    }
1196
1197    #[test]
1198    fn test_register_table_from_snapshot_detects_name_collision() {
1199        let mut mm = make_mm();
1200        let mut registry = SchemaRegistry::default();
1201        let snapshot = dummy_snapshot("users");
1202
1203        let pages = registry
1204            .register_table_from_snapshot(&snapshot, &mut mm)
1205            .expect("first");
1206
1207        // Tamper persisted snapshot to simulate a colliding fingerprint with a
1208        // different name.
1209        let mut tampered = snapshot.clone();
1210        tampered.name = "imposter".to_string();
1211        mm.write_at(pages.schema_snapshot_page, 0, &tampered)
1212            .expect("overwrite");
1213
1214        let result = registry.register_table_from_snapshot(&snapshot, &mut mm);
1215        assert!(matches!(
1216            result,
1217            Err(MemoryError::NameCollision {
1218                ref candidate,
1219                ref existing,
1220            }) if candidate == "users" && existing == "imposter"
1221        ));
1222    }
1223
1224    #[test]
1225    fn test_decode_rejects_truncated_schema_registry() {
1226        use std::borrow::Cow;
1227
1228        use wasm_dbms_api::prelude::DecodeError;
1229
1230        // No header at all.
1231        assert!(matches!(
1232            SchemaRegistry::decode(Cow::Borrowed(&[][..])),
1233            Err(MemoryError::DecodeError(DecodeError::TooShort))
1234        ));
1235
1236        // Header declares one entry, but only 10 entry bytes follow.
1237        let mut buf = vec![0u8; 8];
1238        buf.extend_from_slice(&1u64.to_le_bytes());
1239        buf.extend_from_slice(&[0u8; 10]);
1240        assert!(matches!(
1241            SchemaRegistry::decode(Cow::Owned(buf)),
1242            Err(MemoryError::DecodeError(DecodeError::TooShort))
1243        ));
1244    }
1245
1246    #[test]
1247    fn test_decode_rejects_entry_truncated_mid_field() {
1248        use std::borrow::Cow;
1249
1250        use wasm_dbms_api::prelude::DecodeError;
1251
1252        // One entry that passes the count check (25 bytes) but flags an
1253        // autoincrement page that is missing.
1254        let mut buf = vec![0u8; 8];
1255        buf.extend_from_slice(&1u64.to_le_bytes());
1256        buf.extend_from_slice(&[0u8; 24]);
1257        buf.push(1);
1258        assert!(matches!(
1259            SchemaRegistry::decode(Cow::Owned(buf)),
1260            Err(MemoryError::DecodeError(DecodeError::TooShort))
1261        ));
1262    }
1263
1264    #[test]
1265    fn test_register_table_from_snapshot_rejects_non_integer_autoincrement_without_side_effects() {
1266        let mut mm = make_mm();
1267        let mut registry = SchemaRegistry::default();
1268        let mut snapshot = dummy_snapshot("texts");
1269        snapshot.columns[0].data_type = DataTypeSnapshot::Text;
1270        snapshot.columns[0].auto_increment = true;
1271        let pages_before = mm.pages_count();
1272
1273        let result = registry.register_table_from_snapshot(&snapshot, &mut mm);
1274
1275        assert!(matches!(result, Err(MemoryError::ConstraintViolation(_))));
1276        assert!(registry.table_registry_page_by_name("texts").is_none());
1277        assert_eq!(mm.pages_count(), pages_before);
1278    }
1279
1280    #[test]
1281    fn test_register_table_from_snapshot_rejects_metadata_above_255_without_side_effects() {
1282        let mut long_name = dummy_snapshot(&"t".repeat(256));
1283        long_name.columns[0].name = "id".to_string();
1284
1285        let mut long_index = dummy_snapshot("indexed");
1286        long_index.indexes = vec![wasm_dbms_api::prelude::IndexSnapshot {
1287            columns: vec!["id".to_string(); 256],
1288            unique: false,
1289        }];
1290
1291        for snapshot in [long_name, long_index] {
1292            let mut mm = make_mm();
1293            let mut registry = SchemaRegistry::default();
1294            let pages_before = mm.pages_count();
1295
1296            let result = registry.register_table_from_snapshot(&snapshot, &mut mm);
1297
1298            assert!(
1299                matches!(result, Err(MemoryError::ConstraintViolation(_))),
1300                "expected ConstraintViolation, got {result:?}"
1301            );
1302            assert!(
1303                registry
1304                    .table_registry_page_by_name(&snapshot.name)
1305                    .is_none()
1306            );
1307            assert_eq!(mm.pages_count(), pages_before);
1308        }
1309    }
1310
1311    #[test]
1312    fn test_register_table_from_snapshot_round_trips_metadata_at_255() {
1313        let mut snapshot = dummy_snapshot(&"t".repeat(255));
1314        snapshot.primary_key = "p".repeat(255);
1315        snapshot.columns[0].name = "p".repeat(255);
1316        snapshot.indexes = vec![
1317            wasm_dbms_api::prelude::IndexSnapshot {
1318                columns: vec!["p".repeat(255)],
1319                unique: true,
1320            },
1321            wasm_dbms_api::prelude::IndexSnapshot {
1322                columns: (0..255).map(|i| format!("c{i}")).collect(),
1323                unique: false,
1324            },
1325        ];
1326        let mut mm = make_mm();
1327        let mut registry = SchemaRegistry::default();
1328
1329        let pages = registry
1330            .register_table_from_snapshot(&snapshot, &mut mm)
1331            .expect("255-byte metadata fits the snapshot format");
1332
1333        let loaded = SchemaSnapshotLedger::load(pages.schema_snapshot_page, &mut mm).expect("load");
1334        assert_eq!(loaded.get(), &snapshot);
1335    }
1336
1337    #[test]
1338    fn test_decode_rejects_entry_count_larger_than_payload() {
1339        use std::borrow::Cow;
1340
1341        use wasm_dbms_api::prelude::DecodeError;
1342
1343        let mut buf = vec![0u8; 8];
1344        buf.extend_from_slice(&u64::MAX.to_le_bytes());
1345        assert!(matches!(
1346            SchemaRegistry::decode(Cow::Owned(buf)),
1347            Err(MemoryError::DecodeError(DecodeError::TooShort))
1348        ));
1349    }
1350
1351    #[test]
1352    fn test_register_table_from_snapshot_initializes_autoincrement_counters() {
1353        use wasm_dbms_api::prelude::{Uint32, Value};
1354
1355        let mut mm = make_mm();
1356        let mut registry = SchemaRegistry::default();
1357        let mut snapshot = dummy_snapshot("counters");
1358        snapshot.columns[0].auto_increment = true;
1359
1360        let pages = registry
1361            .register_table_from_snapshot(&snapshot, &mut mm)
1362            .expect("register from snapshot");
1363        let mut table = TableRegistry::load(pages, &mut mm).expect("load table");
1364
1365        assert_eq!(
1366            table
1367                .next_autoincrement("id", &mut mm)
1368                .expect("first value"),
1369            Some(Value::Uint32(Uint32(1)))
1370        );
1371    }
1372
1373    #[test]
1374    fn test_unregister_table_removes_entry_and_returns_previous_pages() {
1375        let mut mm = make_mm();
1376        let mut registry = SchemaRegistry::default();
1377        let pages = registry
1378            .register_table::<User>(&mut mm)
1379            .expect("failed to register");
1380
1381        let removed = registry
1382            .unregister_table("users", &mut mm)
1383            .expect("unregister");
1384        assert_eq!(removed, Some(pages));
1385        assert!(registry.table_registry_page_by_name("users").is_none());
1386    }
1387
1388    #[test]
1389    fn test_unregister_table_returns_none_for_unknown_table() {
1390        let mut mm = make_mm();
1391        let mut registry = SchemaRegistry::default();
1392        let removed = registry
1393            .unregister_table("missing", &mut mm)
1394            .expect("unregister");
1395        assert!(removed.is_none());
1396    }
1397
1398    #[test]
1399    fn test_unregister_table_releases_pages_for_reuse() {
1400        // Registering a table allocates 4 metadata pages (no autoincrement on
1401        // `User`) plus an index B-tree root. Unregistering must hand each
1402        // back to the unclaimed-pages ledger so subsequent claims pop them
1403        // before bumping the high-water mark.
1404        let mut mm = make_mm();
1405        let mut registry = SchemaRegistry::default();
1406
1407        let pages = registry.register_table::<User>(&mut mm).expect("register");
1408
1409        let last_page_before_drop = mm.last_page().expect("at least one page");
1410
1411        registry
1412            .unregister_table("users", &mut mm)
1413            .expect("unregister")
1414            .expect("expected returned pages");
1415
1416        // Re-claim every previously-owned page; none of these should bump
1417        // the high-water mark.
1418        let mut reclaimed = Vec::new();
1419        for _ in 0..5 {
1420            let page = mm.claim_page().expect("claim");
1421            assert!(
1422                page <= last_page_before_drop,
1423                "expected reclaimed page <= {last_page_before_drop}, got {page}"
1424            );
1425            reclaimed.push(page);
1426        }
1427        // Pages handed out must include every TableRegistryPage page.
1428        assert!(reclaimed.contains(&pages.schema_snapshot_page));
1429        assert!(reclaimed.contains(&pages.pages_list_page));
1430        assert!(reclaimed.contains(&pages.free_segments_page));
1431        assert!(reclaimed.contains(&pages.index_registry_page));
1432    }
1433
1434    #[test]
1435    fn test_unregister_table_rejects_release_when_unclaimed_ledger_is_full() {
1436        let mut mm = make_mm();
1437        let mut registry = SchemaRegistry::default();
1438        let pages = registry.register_table::<User>(&mut mm).expect("register");
1439
1440        let mut ledger = UnclaimedPages::new();
1441        for page in 0..(UNCLAIMED_PAGES_CAPACITY - 1) {
1442            ledger.push(page).expect("fill ledger");
1443        }
1444        mm.write_at(mm.unclaimed_pages_page(), 0, &ledger)
1445            .expect("persist ledger");
1446
1447        let err = registry
1448            .unregister_table("users", &mut mm)
1449            .expect_err("drop must fail before partially releasing pages");
1450        assert!(matches!(err, MemoryError::UnclaimedPagesFull { .. }));
1451
1452        assert!(
1453            registry.table_registry_page_by_name("users").is_some(),
1454            "registry entry must remain when release is rejected"
1455        );
1456
1457        let snapshot = SchemaSnapshotLedger::load(pages.schema_snapshot_page, &mut mm)
1458            .expect("load snapshot ledger")
1459            .get()
1460            .clone();
1461        assert_eq!(snapshot.name, "users");
1462    }
1463
1464    #[test]
1465    fn test_register_table_from_snapshot_rejects_256_autoincrement_columns_without_side_effects() {
1466        let mut mm = make_mm();
1467        let mut registry = SchemaRegistry::default();
1468        let mut snapshot = dummy_snapshot("counters");
1469        let template = snapshot.columns[0].clone();
1470        snapshot.columns = (0..256)
1471            .map(|i| ColumnSnapshot {
1472                name: format!("c{i}"),
1473                auto_increment: true,
1474                primary_key: i == 0,
1475                ..template.clone()
1476            })
1477            .collect();
1478        snapshot.primary_key = "c0".to_string();
1479        let pages_before = mm.pages_count();
1480
1481        let result = registry.register_table_from_snapshot(&snapshot, &mut mm);
1482
1483        assert!(matches!(result, Err(MemoryError::ConstraintViolation(_))));
1484        assert!(registry.table_registry_page_by_name("counters").is_none());
1485        assert_eq!(mm.pages_count(), pages_before);
1486    }
1487
1488    #[test]
1489    fn test_register_table_from_snapshot_rejects_long_index_column_name_without_side_effects() {
1490        use wasm_dbms_api::prelude::IndexSnapshot;
1491
1492        let mut mm = make_mm();
1493        let mut registry = SchemaRegistry::default();
1494        let mut snapshot = dummy_snapshot("indexed");
1495        snapshot.indexes = vec![IndexSnapshot {
1496            columns: vec!["x".repeat(256)],
1497            unique: false,
1498        }];
1499        let pages_before = mm.pages_count();
1500
1501        let result = registry.register_table_from_snapshot(&snapshot, &mut mm);
1502
1503        assert!(matches!(result, Err(MemoryError::ConstraintViolation(_))));
1504        assert!(registry.table_registry_page_by_name("indexed").is_none());
1505        assert_eq!(mm.pages_count(), pages_before);
1506    }
1507}
wasm_dbms_memory/schema_registry.rs

wasm_dbms_memory/
schema_registry.rs
