Repository: cberner/redb
Pull request #116 (2022-01-01 00:36:11)

Better crash handling and minor cleanups


Patch:
diff --git a/src/db.rs b/src/db.rs
index 00c3c01..9f87303 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -311,28 +311,9 @@ impl<'a> Drop for ReadOnlyDatabaseTransaction<'a> {
 
 #[cfg(test)]
 mod test {
-    use crate::{Database, ReadOnlyTable, ReadableTable, Table};
+    use crate::{Database, Table};
     use tempfile::NamedTempFile;
 
-    #[test]
-    fn non_page_size_multiple() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-
-        let db_size = 1024 * 1024 + 1;
-        let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-        let txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
-
-        let key = vec![0; 1024];
-        let value = vec![0; 1];
-        table.insert(&key, &value).unwrap();
-        txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(table.len().unwrap(), 1);
-    }
-
     #[test]
     fn transaction_id_persistence() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
diff --git a/src/error.rs b/src/error.rs
index bf887af..676f984 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -7,6 +7,7 @@ pub enum Error {
     TableTypeMismatch(String),
     DoesNotExist(String),
     LeakedWriteTransaction(String),
+    OutOfSpace,
     Io(io::Error),
 }
 
diff --git a/src/multimap_table.rs b/src/multimap_table.rs
index 85652d4..f9c7fec 100644
--- a/src/multimap_table.rs
+++ b/src/multimap_table.rs
@@ -529,151 +529,3 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<'s, K,
             .map(|x| x == 0)
     }
 }
-
-#[cfg(test)]
-mod test {
-    use crate::multimap_table::ReadableMultimapTable;
-    use crate::{Database, MultimapTable, ReadOnlyMultimapTable};
-    use tempfile::NamedTempFile;
-
-    fn get_vec(table: &ReadOnlyMultimapTable<[u8], [u8]>, key: &[u8]) -> Vec<Vec<u8>> {
-        let mut result = vec![];
-        let mut iter = table.get(key).unwrap();
-        loop {
-            let item = iter.next();
-            if let Some(item_value) = item {
-                result.push(item_value.to_vec());
-            } else {
-                return result;
-            }
-        }
-    }
-
-    #[test]
-    fn len() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
-
-        table.insert(b"hello", b"world").unwrap();
-        table.insert(b"hello", b"world2").unwrap();
-        table.insert(b"hi", b"world").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
-        assert_eq!(table.len().unwrap(), 3);
-    }
-
-    #[test]
-    fn is_empty() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-
-        let write_txn = db.begin_write().unwrap();
-        let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
-        assert!(!table.is_empty().unwrap());
-    }
-
-    #[test]
-    fn insert() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        table.insert(b"hello", b"world2").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
-        assert_eq!(
-            vec![b"world".to_vec(), b"world2".to_vec()],
-            get_vec(&table, b"hello")
-        );
-        assert_eq!(table.len().unwrap(), 2);
-    }
-
-    #[test]
-    fn range_query() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
-        for i in 0..5u8 {
-            let value = vec![i];
-            table.insert(b"0", &value).unwrap();
-        }
-        for i in 5..10u8 {
-            let value = vec![i];
-            table.insert(b"1", &value).unwrap();
-        }
-        for i in 10..15u8 {
-            let value = vec![i];
-            table.insert(b"2", &value).unwrap();
-        }
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
-        let start = b"0".as_ref();
-        let end = b"1".as_ref();
-        let mut iter = table.get_range(start..=end).unwrap();
-        for i in 0..10u8 {
-            let (key, value) = iter.next().unwrap();
-            if i < 5 {
-                assert_eq!(b"0", key);
-            } else {
-                assert_eq!(b"1", key);
-            }
-            assert_eq!(&[i], value);
-        }
-        assert!(iter.next().is_none());
-    }
-
-    #[test]
-    fn delete() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        table.insert(b"hello", b"world2").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
-        assert_eq!(
-            vec![b"world".to_vec(), b"world2".to_vec()],
-            get_vec(&table, b"hello")
-        );
-        assert_eq!(table.len().unwrap(), 2);
-
-        let write_txn = db.begin_write().unwrap();
-        let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
-        table.remove(b"hello", b"world2").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
-        assert_eq!(vec![b"world".to_vec()], get_vec(&table, b"hello"));
-        assert_eq!(table.len().unwrap(), 1);
-
-        let write_txn = db.begin_write().unwrap();
-        let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
-        table.remove_all(b"hello").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
-        assert!(table.is_empty().unwrap());
-        let empty: Vec<Vec<u8>> = vec![];
-        assert_eq!(empty, get_vec(&table, b"hello"));
-    }
-}
diff --git a/src/table.rs b/src/table.rs
index 7650b3c..7537660 100644
--- a/src/table.rs
+++ b/src/table.rs
@@ -244,111 +244,10 @@ mod test {
     use crate::types::{
         AsBytesWithLifetime, RedbKey, RedbValue, RefAsBytesLifetime, RefLifetime, WithLifetime,
     };
-    use crate::{Database, Error, ReadOnlyTable, ReadableTable, Table};
+    use crate::{Database, ReadOnlyTable, ReadableTable, Table};
     use std::cmp::Ordering;
     use tempfile::NamedTempFile;
 
-    #[test]
-    fn len() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        table.insert(b"hello2", b"world2").unwrap();
-        table.insert(b"hi", b"world").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(table.len().unwrap(), 3);
-    }
-
-    #[test]
-    fn multiple_tables() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"1").unwrap();
-        let mut table2: Table<[u8], [u8]> = write_txn.open_table(b"2").unwrap();
-
-        table.insert(b"hello", b"world").unwrap();
-        table2.insert(b"hello", b"world2").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"1").unwrap();
-        let table2: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"2").unwrap();
-        assert_eq!(table.len().unwrap(), 1);
-        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
-        assert_eq!(table2.len().unwrap(), 1);
-        assert_eq!(b"world2", table2.get(b"hello").unwrap().unwrap().as_ref());
-    }
-
-    #[test]
-    fn is_empty() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert!(!table.is_empty().unwrap());
-    }
-
-    #[test]
-    fn abort() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"aborted").unwrap();
-        assert_eq!(b"aborted", table.get(b"hello").unwrap().unwrap().as_ref());
-        write_txn.abort().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: Result<ReadOnlyTable<[u8], [u8]>, Error> = read_txn.open_table(b"x");
-        assert!(table.is_err());
-
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
-        assert_eq!(table.len().unwrap(), 1);
-    }
-
-    #[test]
-    fn insert_overwrite() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
-
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"replaced").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(b"replaced", table.get(b"hello").unwrap().unwrap().as_ref());
-    }
-
     #[test]
     fn u64_type() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
@@ -363,23 +262,6 @@ mod test {
         assert_eq!(1, table.get(&0).unwrap().unwrap().to_value());
     }
 
-    #[test]
-    fn insert_reserve() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        let value = b"world";
-        let mut reserved = table.insert_reserve(b"hello", value.len()).unwrap();
-        reserved.as_mut().copy_from_slice(value);
-        drop(reserved);
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(value, table.get(b"hello").unwrap().unwrap().as_ref());
-    }
-
     #[test]
     fn custom_ordering() {
         struct ReverseKey(Vec<u8>);
@@ -431,81 +313,4 @@ mod test {
         }
         assert!(iter.next().is_none());
     }
-
-    #[test]
-    fn delete() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        table.insert(b"hello2", b"world").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
-        assert_eq!(table.len().unwrap(), 2);
-
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.remove(b"hello").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert!(table.get(b"hello").unwrap().is_none());
-        assert_eq!(table.len().unwrap(), 1);
-    }
-
-    #[test]
-    fn no_dirty_reads() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: Result<ReadOnlyTable<[u8], [u8]>, Error> = read_txn.open_table(b"x");
-        assert!(table.is_err());
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
-    }
-
-    #[test]
-    fn read_isolation() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        table.insert(b"hello", b"world").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn = db.begin_read().unwrap();
-        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
-        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
-
-        let write_txn = db.begin_write().unwrap();
-        let mut write_table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
-        write_table.remove(b"hello").unwrap();
-        write_table.insert(b"hello2", b"world2").unwrap();
-        write_table.insert(b"hello3", b"world3").unwrap();
-        write_txn.commit().unwrap();
-
-        let read_txn2 = db.begin_read().unwrap();
-        let table2: ReadOnlyTable<[u8], [u8]> = read_txn2.open_table(b"x").unwrap();
-        assert!(table2.get(b"hello").unwrap().is_none());
-        assert_eq!(b"world2", table2.get(b"hello2").unwrap().unwrap().as_ref());
-        assert_eq!(b"world3", table2.get(b"hello3").unwrap().unwrap().as_ref());
-        assert_eq!(table2.len().unwrap(), 2);
-
-        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
-        assert!(table.get(b"hello2").unwrap().is_none());
-        assert!(table.get(b"hello3").unwrap().is_none());
-        assert_eq!(table.len().unwrap(), 1);
-    }
 }
diff --git a/src/tree_store/base_types.rs b/src/tree_store/base_types.rs
index bb072da..7d705db 100644
--- a/src/tree_store/base_types.rs
+++ b/src/tree_store/base_types.rs
@@ -7,12 +7,10 @@ pub(in crate) struct NodeHandle {
 }
 
 impl NodeHandle {
-    pub(crate) fn new(page_number: PageNumber, messages: u32) -> Self {
-        // TODO: change messages to a u8
-        assert!(messages <= u8::MAX as u32);
+    pub(crate) fn new(page_number: PageNumber, messages: u8) -> Self {
         Self {
             page_number,
-            messages: messages as u8,
+            messages,
         }
     }
 
@@ -23,8 +21,8 @@ impl NodeHandle {
     // Messages are btree node defined, but must form a strict ordering of subsets of page memory.
     // That is, if m1 & m2 are counts of messages and m1 < m2, then the bytes in a page p,
     // used in the representation of m1 is a strict subset of the bytes used for m2
-    pub(crate) fn get_valid_messages(&self) -> u32 {
-        self.messages as u32
+    pub(crate) fn get_valid_messages(&self) -> u8 {
+        self.messages
     }
 
     #[inline(always)]
@@ -41,7 +39,7 @@ impl NodeHandle {
     pub(crate) fn from_be_bytes(bytes: [u8; Self::serialized_size()]) -> Self {
         let page_number = PageNumber::from_be_bytes(bytes);
         let temp = u64::from_be_bytes(bytes[..8].try_into().unwrap());
-        let messages = (temp >> 56) as u32;
+        let messages = (temp >> 56) as u8;
 
         Self::new(page_number, messages)
     }
diff --git a/src/tree_store/btree_utils.rs b/src/tree_store/btree_utils.rs
index 8510e3b..5add45d 100644
--- a/src/tree_store/btree_utils.rs
+++ b/src/tree_store/btree_utils.rs
@@ -3,6 +3,7 @@ use crate::tree_store::btree_utils::RangeIterState::{Internal, LeafLeft, LeafRig
 use crate::tree_store::page_store::{Page, PageImpl, PageMut, PageNumber, TransactionalMemory};
 use crate::tree_store::NodeHandle;
 use crate::types::{RedbKey, RedbValue};
+use crate::Error;
 use std::cmp::{max, Ordering};
 use std::convert::TryInto;
 use std::marker::PhantomData;
@@ -52,7 +53,7 @@ pub(crate) enum RangeIterState<'a> {
     },
     Internal {
         page: PageImpl<'a>,
-        valid_message_bytes: u32,
+        valid_messages: u8,
         child: usize,
         parent: Option<Box<RangeIterState<'a>>>,
         reversed: bool,
@@ -60,6 +61,14 @@ pub(crate) enum RangeIterState<'a> {
 }
 
 impl<'a> RangeIterState<'a> {
+    fn page_number(&self) -> PageNumber {
+        match self {
+            RangeIterState::LeafLeft { page, .. } => page.get_page_number(),
+            RangeIterState::LeafRight { page, .. } => page.get_page_number(),
+            RangeIterState::Internal { page, .. } => page.get_page_number(),
+        }
+    }
+
     fn forward_next(self, manager: &'a TransactionalMemory) -> Option<RangeIterState> {
         match self {
             RangeIterState::LeafLeft { page, parent, .. } => Some(LeafRight {
@@ -70,19 +79,19 @@ impl<'a> RangeIterState<'a> {
             RangeIterState::LeafRight { parent, .. } => parent.map(|x| *x),
             RangeIterState::Internal {
                 page,
-                valid_message_bytes,
+                valid_messages,
                 child,
                 mut parent,
                 ..
             } => {
-                let accessor = InternalAccessor::new(&page, valid_message_bytes);
+                let accessor = InternalAccessor::new(&page, valid_messages);
                 let child_page = accessor.child_page(child).unwrap();
-                let child_message_bytes = child_page.get_valid_messages();
+                let child_messages = child_page.get_valid_messages();
                 let child_page = manager.get_page(child_page.get_page_number());
                 if child < BTREE_ORDER - 1 && accessor.child_page(child + 1).is_some() {
                     parent = Some(Box::new(Internal {
                         page,
-                        valid_message_bytes,
+                        valid_messages,
                         child: child + 1,
                         parent,
                         reversed: false,
@@ -96,7 +105,7 @@ impl<'a> RangeIterState<'a> {
                     }),
                     INTERNAL => Some(Internal {
                         page: child_page,
-                        valid_message_bytes: child_message_bytes,
+                        valid_messages: child_messages,
                         child: 0,
                         parent,
                         reversed: false,
@@ -117,20 +126,20 @@ impl<'a> RangeIterState<'a> {
             }),
             RangeIterState::Internal {
                 page,
-                valid_message_bytes,
+                valid_messages,
                 child,
                 mut parent,
                 ..
             } => {
-                let child_page = InternalAccessor::new(&page, valid_message_bytes)
+                let child_page = InternalAccessor::new(&page, valid_messages)
                     .child_page(child)
                     .unwrap();
-                let child_message_bytes = child_page.get_valid_messages();
+                let child_messages = child_page.get_valid_messages();
                 let child_page = manager.get_page(child_page.get_page_number());
                 if child > 0 {
                     parent = Some(Box::new(Internal {
                         page,
-                        valid_message_bytes,
+                        valid_messages,
                         child: child - 1,
                         parent,
                         reversed: true,
@@ -143,7 +152,7 @@ impl<'a> RangeIterState<'a> {
                         reversed: true,
                     }),
                     INTERNAL => {
-                        let accessor = InternalAccessor::new(&child_page, child_message_bytes);
+                        let accessor = InternalAccessor::new(&child_page, child_messages);
                         let mut index = 0;
                         for i in (0..BTREE_ORDER).rev() {
                             if accessor.child_page(i).is_some() {
@@ -154,7 +163,7 @@ impl<'a> RangeIterState<'a> {
                         assert!(index > 0);
                         Some(Internal {
                             page: child_page,
-                            valid_message_bytes: child_message_bytes,
+                            valid_messages: child_messages,
                             child: index,
                             parent,
                             reversed: true,
@@ -201,6 +210,67 @@ impl<'a> RangeIterState<'a> {
     }
 }
 
+pub(crate) struct AllPageNumbersBtreeIter<'a> {
+    next: Option<RangeIterState<'a>>,
+    manager: &'a TransactionalMemory,
+}
+
+impl<'a> AllPageNumbersBtreeIter<'a> {
+    pub(crate) fn new(start: RangeIterState<'a>, manager: &'a TransactionalMemory) -> Self {
+        Self {
+            next: Some(start),
+            manager,
+        }
+    }
+}
+
+impl<'a> Iterator for AllPageNumbersBtreeIter<'a> {
+    type Item = PageNumber;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let state = self.next.take()?;
+        let value = state.page_number();
+        self.next = state.forward_next(self.manager);
+
+        Some(value)
+    }
+}
+
+pub(in crate) fn page_numbers_iter_start_state<'a>(
+    page: PageImpl<'a>,
+    valid_messages: u8,
+    mut parent: Option<Box<RangeIterState<'a>>>,
+    manager: &'a TransactionalMemory,
+) -> Option<RangeIterState<'a>> {
+    let node_mem = page.memory();
+    match node_mem[0] {
+        LEAF => Some(RangeIterState::LeafLeft {
+            page,
+            parent,
+            reversed: false,
+        }),
+        INTERNAL => {
+            let accessor = InternalAccessor::new(&page, valid_messages);
+            let child_page_number = accessor.child_page(0).unwrap();
+            let child_page = manager.get_page(child_page_number.get_page_number());
+            parent = Some(Box::new(Internal {
+                page,
+                valid_messages,
+                child: 0,
+                parent,
+                reversed: false,
+            }));
+            page_numbers_iter_start_state(
+                child_page,
+                child_page_number.get_valid_messages(),
+                parent,
+                manager,
+            )
+        }
+        _ => unreachable!(),
+    }
+}
+
 pub struct BtreeRangeIter<
     'a,
     T: RangeBounds<KR>,
@@ -544,16 +614,16 @@ impl<'a: 'b, 'b> LeafBuilder<'a, 'b> {
 // Provides a simple zero-copy way to access an index page
 struct InternalAccessor<'a: 'b, 'b, T: Page + 'a> {
     page: &'b T,
-    valid_messages: u32,
+    valid_messages: u8,
     _page_lifetime: PhantomData<&'a ()>,
 }
 
 impl<'a: 'b, 'b, T: Page + 'a> InternalAccessor<'a, 'b, T> {
-    fn new(page: &'b T, message_bytes: u32) -> Self {
+    fn new(page: &'b T, messages: u8) -> Self {
         debug_assert_eq!(page.memory()[0], INTERNAL);
         InternalAccessor {
             page,
-            valid_messages: message_bytes,
+            valid_messages: messages,
             _page_lifetime: Default::default(),
         }
     }
@@ -825,19 +895,19 @@ impl<'a: 'b, 'b> InternalMutator<'a, 'b> {
         Self { page }
     }
 
-    fn can_write_delta_message(&self, message_byte_offset: u32) -> bool {
-        message_byte_offset < MESSAGE_BUFFER as u32
+    fn can_write_delta_message(&self, message_offset: u8) -> bool {
+        message_offset < MESSAGE_BUFFER as u8
     }
 
     // Returns the new valid message offset
     fn write_delta_message(
         &mut self,
-        existing_messages: u32,
+        existing_messages: u8,
         child_index: u8,
         handle: NodeHandle,
-    ) -> u32 {
+    ) -> u8 {
         assert!(child_index < BTREE_ORDER as u8);
-        assert!(existing_messages < MESSAGE_BUFFER as u32);
+        assert!(existing_messages < MESSAGE_BUFFER as u8);
         let offset = 1
             + NodeHandle::serialized_size() * BTREE_ORDER
             + 8 * (BTREE_ORDER - 1) * 3
@@ -863,14 +933,14 @@ impl<'a: 'b, 'b> InternalMutator<'a, 'b> {
 
 pub(in crate) fn tree_height<'a>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     manager: &'a TransactionalMemory,
 ) -> usize {
     let node_mem = page.memory();
     match node_mem[0] {
         LEAF => 1,
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             let mut max_child_height = 0;
             for i in 0..BTREE_ORDER {
                 if let Some(child) = accessor.child_page(i) {
@@ -889,7 +959,7 @@ pub(in crate) fn tree_height<'a>(
     }
 }
 
-pub(in crate) fn print_node(page: &impl Page, valid_message_bytes: u32) {
+pub(in crate) fn print_node(page: &impl Page, valid_messages: u8) {
     let node_mem = page.memory();
     match node_mem[0] {
         LEAF => {
@@ -910,11 +980,11 @@ pub(in crate) fn print_node(page: &impl Page, valid_message_bytes: u32) {
             eprint!("]");
         }
         INTERNAL => {
-            let accessor = InternalAccessor::new(page, valid_message_bytes);
+            let accessor = InternalAccessor::new(page, valid_messages);
             eprint!(
                 "Internal[ (page={:?}/{}), child_0={:?}",
                 page.get_page_number(),
-                valid_message_bytes,
+                valid_messages,
                 accessor.child_page(0).unwrap().get_page_number()
             );
             for i in 0..(BTREE_ORDER - 1) {
@@ -934,9 +1004,9 @@ pub(in crate) fn print_node(page: &impl Page, valid_message_bytes: u32) {
 
 pub(in crate) fn node_children<'a>(
     page: &PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     manager: &'a TransactionalMemory,
-) -> Vec<(PageImpl<'a>, u32)> {
+) -> Vec<(PageImpl<'a>, u8)> {
     let node_mem = page.memory();
     match node_mem[0] {
         LEAF => {
@@ -944,7 +1014,7 @@ pub(in crate) fn node_children<'a>(
         }
         INTERNAL => {
             let mut children = vec![];
-            let accessor = InternalAccessor::new(page, valid_message_bytes);
+            let accessor = InternalAccessor::new(page, valid_messages);
             for i in 0..BTREE_ORDER {
                 if let Some(child) = accessor.child_page(i) {
                     children.push((
@@ -961,15 +1031,15 @@ pub(in crate) fn node_children<'a>(
 
 pub(in crate) fn print_tree<'a>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     manager: &'a TransactionalMemory,
 ) {
-    let mut pages = vec![(page, valid_message_bytes)];
+    let mut pages = vec![(page, valid_messages)];
     while !pages.is_empty() {
         let mut next_children = vec![];
-        for (page, message_bytes) in pages.drain(..) {
-            next_children.extend(node_children(&page, message_bytes, manager));
-            print_node(&page, message_bytes);
+        for (page, messages) in pages.drain(..) {
+            next_children.extend(node_children(&page, messages, manager));
+            print_node(&page, messages);
             eprint!("  ");
         }
         eprintln!();
@@ -981,14 +1051,14 @@ pub(in crate) fn print_tree<'a>(
 // Returns the new root, and a list of freed pages
 pub(in crate) fn tree_delete<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     table: u64,
     key: &[u8],
     manager: &'a TransactionalMemory,
-) -> (Option<NodeHandle>, Vec<PageNumber>) {
+) -> Result<(Option<NodeHandle>, Vec<PageNumber>), Error> {
     let mut freed = vec![];
     let result =
-        match tree_delete_helper::<K>(page, valid_message_bytes, table, key, &mut freed, manager) {
+        match tree_delete_helper::<K>(page, valid_messages, table, key, &mut freed, manager)? {
             DeletionResult::Subtree(page) => Some(page),
             DeletionResult::PartialLeaf(entries) => {
                 assert!(entries.is_empty());
@@ -999,7 +1069,7 @@ pub(in crate) fn tree_delete<'a, K: RedbKey + ?Sized>(
                 Some(pages[0])
             }
         };
-    (result, freed)
+    Ok((result, freed))
 }
 
 #[derive(Debug)]
@@ -1018,18 +1088,18 @@ fn split_leaf(
     partial: &[(u64, Vec<u8>, Vec<u8>)],
     manager: &TransactionalMemory,
     freed: &mut Vec<PageNumber>,
-) -> Option<(NodeHandle, NodeHandle)> {
+) -> Result<Option<(NodeHandle, NodeHandle)>, Error> {
     assert!(partial.is_empty());
     let page = manager.get_page(leaf.get_page_number());
     let accessor = LeafAccessor::new(&page);
     if let Some(greater) = accessor.greater() {
         let lesser = accessor.lesser();
-        let page1 = make_single_leaf(lesser.table_id(), lesser.key(), lesser.value(), manager);
-        let page2 = make_single_leaf(greater.table_id(), greater.key(), greater.value(), manager);
+        let page1 = make_single_leaf(lesser.table_id(), lesser.key(), lesser.value(), manager)?;
+        let page2 = make_single_leaf(greater.table_id(), greater.key(), greater.value(), manager)?;
         freed.push(page.get_page_number());
-        Some((page1, page2))
+        Ok(Some((page1, page2)))
     } else {
-        None
+        Ok(None)
     }
 }
 
@@ -1052,7 +1122,7 @@ fn split_index(
     partial: &[NodeHandle],
     manager: &TransactionalMemory,
     freed: &mut Vec<PageNumber>,
-) -> Option<(NodeHandle, NodeHandle)> {
+) -> Result<Option<(NodeHandle, NodeHandle)>, Error> {
     let page = manager.get_page(index.get_page_number());
     let accessor = InternalAccessor::new(&page, index.get_valid_messages());
 
@@ -1073,7 +1143,7 @@ fn split_index(
     // The total_key_length calculation below does it correctly
     let has_space = accessor.total_key_length() + required_key_space < MAX_KEY_SPACE_PER_PAGE;
     if has_space && has_enough_slots {
-        return None;
+        return Ok(None);
     }
 
     let mut pages = vec![];
@@ -1127,15 +1197,18 @@ fn split_index(
         index
     };
 
-    let page1 = make_index_many_pages(&pages[0..division], manager);
-    let page2 = make_index_many_pages(&pages[division..], manager);
+    let page1 = make_index_many_pages(&pages[0..division], manager)?;
+    let page2 = make_index_many_pages(&pages[division..], manager)?;
     freed.push(page.get_page_number());
 
-    Some((page1, page2))
+    Ok(Some((page1, page2)))
 }
 
 // Pages must be in sorted order
-fn make_index_many_pages(children: &[NodeHandle], manager: &TransactionalMemory) -> NodeHandle {
+fn make_index_many_pages(
+    children: &[NodeHandle],
+    manager: &TransactionalMemory,
+) -> Result<NodeHandle, Error> {
     let mut tables_and_keys = vec![];
     let mut key_size = 0;
     for i in 1..children.len() {
@@ -1147,14 +1220,14 @@ fn make_index_many_pages(children: &[NodeHandle], manager: &TransactionalMemory)
         key_size += entry.1.len();
         tables_and_keys.push(entry);
     }
-    let mut page = manager.allocate(InternalBuilder::required_bytes(key_size));
+    let mut page = manager.allocate(InternalBuilder::required_bytes(key_size))?;
     let mut builder = InternalBuilder::new(&mut page);
     builder.write_first_page(children[0]);
     for i in 1..children.len() {
         let (table, key) = &tables_and_keys[i - 1];
         builder.write_nth_key(*table, key, children[i], i - 1);
     }
-    NodeHandle::new(page.get_page_number(), 0)
+    Ok(NodeHandle::new(page.get_page_number(), 0))
 }
 
 fn merge_index(
@@ -1162,7 +1235,7 @@ fn merge_index(
     partial: &[NodeHandle],
     manager: &TransactionalMemory,
     freed: &mut Vec<PageNumber>,
-) -> NodeHandle {
+) -> Result<NodeHandle, Error> {
     let page = manager.get_page(index.get_page_number());
     let accessor = InternalAccessor::new(&page, index.get_valid_messages());
     assert!(accessor.child_page(BTREE_ORDER - partial.len()).is_none());
@@ -1193,15 +1266,16 @@ fn repair_children(
     children: Vec<DeletionResult>,
     manager: &TransactionalMemory,
     freed: &mut Vec<PageNumber>,
-) -> Vec<NodeHandle> {
+) -> Result<Vec<NodeHandle>, Error> {
     if children.iter().all(|x| matches!(x, Subtree(_))) {
-        children
+        let handles: Vec<NodeHandle> = children
             .iter()
             .map(|x| match x {
                 Subtree(page_number) => *page_number,
                 _ => unreachable!(),
             })
-            .collect()
+            .collect();
+        Ok(handles)
     } else if children.iter().any(|x| matches!(x, PartialLeaf(_))) {
         let mut result = vec![];
         let mut repaired = false;
@@ -1214,7 +1288,7 @@ fn repair_children(
                 }
                 let offset = if i > 0 { i - 1 } else { i + 1 };
                 if let Some(PartialLeaf(partials)) = children.get(offset) {
-                    if let Some((page1, page2)) = split_leaf(*handle, partials, manager, freed) {
+                    if let Some((page1, page2)) = split_leaf(*handle, partials, manager, freed)? {
                         result.push(page1);
                         result.push(page2);
                     } else {
@@ -1227,7 +1301,7 @@ fn repair_children(
                 }
             }
         }
-        result
+        Ok(result)
     } else if children.iter().any(|x| matches!(x, PartialInternal(_))) {
         let mut result = vec![];
         let mut repaired = false;
@@ -1241,12 +1315,12 @@ fn repair_children(
                 let offset = if i > 0 { i - 1 } else { i + 1 };
                 if let Some(PartialInternal(partials)) = children.get(offset) {
                     if let Some((page1, page2)) =
-                        split_index(*page_number, partials, manager, freed)
+                        split_index(*page_number, partials, manager, freed)?
                     {
                         result.push(page1);
                         result.push(page2);
                     } else {
-                        result.push(merge_index(*page_number, partials, manager, freed));
+                        result.push(merge_index(*page_number, partials, manager, freed)?);
                     }
                     repaired = true;
                 } else {
@@ -1255,7 +1329,7 @@ fn repair_children(
                 }
             }
         }
-        result
+        Ok(result)
     } else {
         unreachable!()
     }
@@ -1263,7 +1337,7 @@ fn repair_children(
 
 fn max_table_key(
     page: PageImpl,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     manager: &TransactionalMemory,
 ) -> (u64, Vec<u8>) {
     let node_mem = page.memory();
@@ -1280,7 +1354,7 @@ fn max_table_key(
             }
         }
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             for i in (0..BTREE_ORDER).rev() {
                 if let Some(child) = accessor.child_page(i) {
                     return max_table_key(
@@ -1301,12 +1375,12 @@ fn max_table_key(
 #[allow(clippy::needless_return)]
 fn tree_delete_helper<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     table: u64,
     key: &[u8],
     freed: &mut Vec<PageNumber>,
     manager: &'a TransactionalMemory,
-) -> DeletionResult {
+) -> Result<DeletionResult, Error> {
     let node_mem = page.memory();
     match node_mem[0] {
         LEAF => {
@@ -1317,7 +1391,10 @@ fn tree_delete_helper<'a, K: RedbKey + ?Sized>(
                     && greater.compare::<K>(table, key).is_ne()
                 {
                     // Not found
-                    return Subtree(NodeHandle::new(page.get_page_number(), valid_message_bytes));
+                    return Ok(Subtree(NodeHandle::new(
+                        page.get_page_number(),
+                        valid_messages,
+                    )));
                 }
                 let new_leaf = if accessor.lesser().compare::<K>(table, key).is_eq() {
                     (greater.table_id(), greater.key(), greater.value())
@@ -1330,22 +1407,25 @@ fn tree_delete_helper<'a, K: RedbKey + ?Sized>(
                 };
 
                 freed.push(page.get_page_number());
-                Subtree(make_single_leaf(
+                Ok(Subtree(make_single_leaf(
                     new_leaf.0, new_leaf.1, new_leaf.2, manager,
-                ))
+                )?))
             } else {
                 if accessor.lesser().compare::<K>(table, key).is_eq() {
                     // Deleted the entire left
                     freed.push(page.get_page_number());
-                    PartialLeaf(vec![])
+                    Ok(PartialLeaf(vec![]))
                 } else {
                     // Not found
-                    Subtree(NodeHandle::new(page.get_page_number(), valid_message_bytes))
+                    Ok(Subtree(NodeHandle::new(
+                        page.get_page_number(),
+                        valid_messages,
+                    )))
                 }
             }
         }
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             let original_page_number = page.get_page_number();
             let mut children = vec![];
             let mut found = false;
@@ -1363,14 +1443,14 @@ fn tree_delete_helper<'a, K: RedbKey + ?Sized>(
                             key,
                             freed,
                             manager,
-                        );
+                        )?;
                         // The key must not have been found, since the subtree didn't change
                         if let Subtree(page_number) = result {
                             if page_number == child_page {
-                                return Subtree(NodeHandle::new(
+                                return Ok(Subtree(NodeHandle::new(
                                     original_page_number,
-                                    valid_message_bytes,
-                                ));
+                                    valid_messages,
+                                )));
                             }
                         }
                         children.push(result);
@@ -1394,12 +1474,15 @@ fn tree_delete_helper<'a, K: RedbKey + ?Sized>(
                     key,
                     freed,
                     manager,
-                );
+                )?;
                 found = true;
                 // The key must not have been found, since the subtree didn't change
                 if let Subtree(page_number) = result {
                     if page_number == last_page {
-                        return Subtree(NodeHandle::new(original_page_number, valid_message_bytes));
+                        return Ok(Subtree(NodeHandle::new(
+                            original_page_number,
+                            valid_messages,
+                        )));
                     }
                 }
                 children.push(result);
@@ -1407,12 +1490,12 @@ fn tree_delete_helper<'a, K: RedbKey + ?Sized>(
             assert!(found);
             assert!(children.len() > 1);
             freed.push(original_page_number);
-            let children = repair_children(children, manager, freed);
+            let children = repair_children(children, manager, freed)?;
             if children.len() == 1 {
-                return PartialInternal(children);
+                return Ok(PartialInternal(children));
             }
 
-            Subtree(make_index_many_pages(&children, manager))
+            Ok(Subtree(make_index_many_pages(&children, manager)?))
         }
         _ => unreachable!(),
     }
@@ -1423,8 +1506,8 @@ pub(in crate) fn make_mut_single_leaf<'a>(
     key: &[u8],
     value: &[u8],
     manager: &'a TransactionalMemory,
-) -> (NodeHandle, AccessGuardMut<'a>) {
-    let mut page = manager.allocate(LeafBuilder::required_bytes(&[key, value]));
+) -> Result<(NodeHandle, AccessGuardMut<'a>), Error> {
+    let mut page = manager.allocate(LeafBuilder::required_bytes(&[key, value]))?;
     let mut builder = LeafBuilder::new(&mut page);
     builder.write_lesser(table, key, value);
     builder.write_greater(None);
@@ -1435,7 +1518,7 @@ pub(in crate) fn make_mut_single_leaf<'a>(
     let page_num = page.get_page_number();
     let guard = AccessGuardMut::new(page, offset, value.len());
 
-    (NodeHandle::new(page_num, 0), guard)
+    Ok((NodeHandle::new(page_num, 0), guard))
 }
 
 pub(in crate) fn make_mut_double_leaf_right<'a, K: RedbKey + ?Sized>(
@@ -1446,9 +1529,9 @@ pub(in crate) fn make_mut_double_leaf_right<'a, K: RedbKey + ?Sized>(
     key2: &[u8],
     value2: &[u8],
     manager: &'a TransactionalMemory,
-) -> (NodeHandle, AccessGuardMut<'a>) {
+) -> Result<(NodeHandle, AccessGuardMut<'a>), Error> {
     debug_assert!(cmp_keys::<K>(table1, key1, table2, key2).is_lt());
-    let mut page = manager.allocate(LeafBuilder::required_bytes(&[key1, value1, key2, value2]));
+    let mut page = manager.allocate(LeafBuilder::required_bytes(&[key1, value1, key2, value2]))?;
     let mut builder = LeafBuilder::new(&mut page);
     builder.write_lesser(table1, key1, value1);
     builder.write_greater(Some((table2, key2, value2)));
@@ -1459,7 +1542,7 @@ pub(in crate) fn make_mut_double_leaf_right<'a, K: RedbKey + ?Sized>(
     let page_num = page.get_page_number();
     let guard = AccessGuardMut::new(page, offset, value2.len());
 
-    (NodeHandle::new(page_num, 0), guard)
+    Ok((NodeHandle::new(page_num, 0), guard))
 }
 
 pub(in crate) fn make_mut_double_leaf_left<'a, K: RedbKey + ?Sized>(
@@ -1470,9 +1553,9 @@ pub(in crate) fn make_mut_double_leaf_left<'a, K: RedbKey + ?Sized>(
     key2: &[u8],
     value2: &[u8],
     manager: &'a TransactionalMemory,
-) -> (NodeHandle, AccessGuardMut<'a>) {
+) -> Result<(NodeHandle, AccessGuardMut<'a>), Error> {
     debug_assert!(cmp_keys::<K>(table1, key1, table2, key2).is_lt());
-    let mut page = manager.allocate(LeafBuilder::required_bytes(&[key1, value1, key2, value2]));
+    let mut page = manager.allocate(LeafBuilder::required_bytes(&[key1, value1, key2, value2]))?;
     let mut builder = LeafBuilder::new(&mut page);
     builder.write_lesser(table1, key1, value1);
     builder.write_greater(Some((table2, key2, value2)));
@@ -1483,7 +1566,7 @@ pub(in crate) fn make_mut_double_leaf_left<'a, K: RedbKey + ?Sized>(
     let page_num = page.get_page_number();
     let guard = AccessGuardMut::new(page, offset, value1.len());
 
-    (NodeHandle::new(page_num, 0), guard)
+    Ok((NodeHandle::new(page_num, 0), guard))
 }
 
 pub(in crate) fn make_single_leaf<'a>(
@@ -1491,12 +1574,12 @@ pub(in crate) fn make_single_leaf<'a>(
     key: &[u8],
     value: &[u8],
     manager: &'a TransactionalMemory,
-) -> NodeHandle {
-    let mut page = manager.allocate(LeafBuilder::required_bytes(&[key, value]));
+) -> Result<NodeHandle, Error> {
+    let mut page = manager.allocate(LeafBuilder::required_bytes(&[key, value]))?;
     let mut builder = LeafBuilder::new(&mut page);
     builder.write_lesser(table, key, value);
     builder.write_greater(None);
-    NodeHandle::new(page.get_page_number(), 0)
+    Ok(NodeHandle::new(page.get_page_number(), 0))
 }
 
 pub(in crate) fn make_index(
@@ -1505,40 +1588,33 @@ pub(in crate) fn make_index(
     lte_page: NodeHandle,
     gt_page: NodeHandle,
     manager: &TransactionalMemory,
-) -> NodeHandle {
-    let mut page = manager.allocate(InternalBuilder::required_bytes(key.len()));
+) -> Result<NodeHandle, Error> {
+    let mut page = manager.allocate(InternalBuilder::required_bytes(key.len()))?;
     let mut builder = InternalBuilder::new(&mut page);
     builder.write_first_page(lte_page);
     builder.write_nth_key(table, key, gt_page, 0);
-    NodeHandle::new(page.get_page_number(), 0)
+    Ok(NodeHandle::new(page.get_page_number(), 0))
 }
 
 // Returns the page number of the sub-tree into which the key was inserted,
 // and the guard which can be used to access the value, and a list of freed pages
 pub(in crate) fn tree_insert<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     table: u64,
     key: &[u8],
     value: &[u8],
     manager: &'a TransactionalMemory,
-) -> (NodeHandle, AccessGuardMut<'a>, Vec<PageNumber>) {
+) -> Result<(NodeHandle, AccessGuardMut<'a>, Vec<PageNumber>), Error> {
     let mut freed = vec![];
-    let (page1, more, guard) = tree_insert_helper::<K>(
-        page,
-        valid_message_bytes,
-        table,
-        key,
-        value,
-        &mut freed,
-        manager,
-    );
+    let (page1, more, guard) =
+        tree_insert_helper::<K>(page, valid_messages, table, key, value, &mut freed, manager)?;
 
     if let Some((table, key, page2)) = more {
-        let index_page = make_index(table, &key, page1, page2, manager);
-        (index_page, guard, freed)
+        let index_page = make_index(table, &key, page1, page2, manager)?;
+        Ok((index_page, guard, freed))
     } else {
-        (page1, guard, freed)
+        Ok((page1, guard, freed))
     }
 }
 
@@ -1589,19 +1665,22 @@ fn copy_to_builder_and_patch<'a>(
 #[allow(clippy::type_complexity)]
 fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     table: u64,
     key: &[u8],
     value: &[u8],
     freed: &mut Vec<PageNumber>,
     manager: &'a TransactionalMemory,
-) -> (
-    NodeHandle,
-    Option<(u64, Vec<u8>, NodeHandle)>,
-    AccessGuardMut<'a>,
-) {
+) -> Result<
+    (
+        NodeHandle,
+        Option<(u64, Vec<u8>, NodeHandle)>,
+        AccessGuardMut<'a>,
+    ),
+    Error,
+> {
     let node_mem = page.memory();
-    match node_mem[0] {
+    Ok(match node_mem[0] {
         LEAF => {
             let accessor = LeafAccessor::new(&page);
             if let Some(entry) = accessor.greater() {
@@ -1610,10 +1689,10 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                         // New entry goes in a new page to the right, so leave this page untouched
                         let left_page = page.get_page_number();
 
-                        let (right_page, guard) = make_mut_single_leaf(table, key, value, manager);
+                        let (right_page, guard) = make_mut_single_leaf(table, key, value, manager)?;
 
                         (
-                            NodeHandle::new(left_page, valid_message_bytes),
+                            NodeHandle::new(left_page, valid_messages),
                             Some((entry.table_id(), entry.key().to_vec(), right_page)),
                             guard,
                         )
@@ -1627,9 +1706,9 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                             key,
                             value,
                             manager,
-                        );
+                        )?;
 
-                        if !manager.free_if_uncommitted(page.get_page_number()) {
+                        if !manager.free_if_uncommitted(page.get_page_number())? {
                             freed.push(page.get_page_number());
                         }
 
@@ -1648,11 +1727,11 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                             Ordering::Less => {
                                 let (left, guard) = make_mut_double_leaf_right::<K>(
                                     left_table, left_key, left_value, table, key, value, manager,
-                                );
+                                )?;
                                 let right =
-                                    make_single_leaf(right_table, right_key, right_value, manager);
+                                    make_single_leaf(right_table, right_key, right_value, manager)?;
 
-                                if !manager.free_if_uncommitted(page.get_page_number()) {
+                                if !manager.free_if_uncommitted(page.get_page_number())? {
                                     freed.push(page.get_page_number());
                                 }
 
@@ -1667,9 +1746,9 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                                     right_key,
                                     right_value,
                                     manager,
-                                );
+                                )?;
 
-                                if !manager.free_if_uncommitted(page.get_page_number()) {
+                                if !manager.free_if_uncommitted(page.get_page_number())? {
                                     freed.push(page.get_page_number());
                                 }
 
@@ -1678,11 +1757,11 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                             Ordering::Greater => {
                                 let (left, guard) = make_mut_double_leaf_left::<K>(
                                     table, key, value, left_table, left_key, left_value, manager,
-                                );
+                                )?;
                                 let right =
-                                    make_single_leaf(right_table, right_key, right_value, manager);
+                                    make_single_leaf(right_table, right_key, right_value, manager)?;
 
-                                if !manager.free_if_uncommitted(page.get_page_number()) {
+                                if !manager.free_if_uncommitted(page.get_page_number())? {
                                     freed.push(page.get_page_number());
                                 }
 
@@ -1706,8 +1785,8 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                         key,
                         value,
                         manager,
-                    ),
-                    Ordering::Equal => make_mut_single_leaf(table, key, value, manager),
+                    )?,
+                    Ordering::Equal => make_mut_single_leaf(table, key, value, manager)?,
                     Ordering::Greater => make_mut_double_leaf_left::<K>(
                         table,
                         key,
@@ -1716,10 +1795,10 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                         accessor.lesser().key(),
                         accessor.lesser().value(),
                         manager,
-                    ),
+                    )?,
                 };
 
-                if !manager.free_if_uncommitted(page.get_page_number()) {
+                if !manager.free_if_uncommitted(page.get_page_number())? {
                     freed.push(page.get_page_number());
                 }
 
@@ -1727,7 +1806,7 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
             }
         }
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             // Delta message can only be used if the keys did not change
             let (child_index, child_page) = accessor.child_for_key::<K>(table, key);
             let (page1, more, guard) = tree_insert_helper::<K>(
@@ -1738,7 +1817,7 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                 value,
                 freed,
                 manager,
-            );
+            )?;
 
             if let Some((index_table2, index_key2, page2)) = more {
                 let new_children_count = 1 + accessor.count_children();
@@ -1748,7 +1827,7 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                 {
                     // Rewrite page since we're splitting a child
                     let mut new_page =
-                        manager.allocate(accessor.total_key_length() + index_key2.len());
+                        manager.allocate(accessor.total_key_length() + index_key2.len())?;
                     let mut builder = InternalBuilder::new(&mut new_page);
 
                     copy_to_builder_and_patch(
@@ -1761,7 +1840,7 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                         Some((index_table2, &index_key2, page2)),
                     );
                     // Free the original page, since we've replaced it
-                    if !manager.free_if_uncommitted(page.get_page_number()) {
+                    if !manager.free_if_uncommitted(page.get_page_number())? {
                         freed.push(page.get_page_number());
                     }
                     (NodeHandle::new(new_page.get_page_number(), 0), None, guard)
@@ -1814,7 +1893,8 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                         .iter()
                         .map(|(_, k)| k.len())
                         .sum();
-                    let mut new_page = manager.allocate(InternalBuilder::required_bytes(key_size));
+                    let mut new_page =
+                        manager.allocate(InternalBuilder::required_bytes(key_size))?;
                     let mut builder = InternalBuilder::new(&mut new_page);
 
                     builder.write_first_page(children[0]);
@@ -1829,7 +1909,8 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                         .iter()
                         .map(|(_, k)| k.len())
                         .sum();
-                    let mut new_page2 = manager.allocate(InternalBuilder::required_bytes(key_size));
+                    let mut new_page2 =
+                        manager.allocate(InternalBuilder::required_bytes(key_size))?;
                     let mut builder2 = InternalBuilder::new(&mut new_page2);
                     builder2.write_first_page(children[division + 1]);
                     for i in (division + 1)..index_table_keys.len() {
@@ -1838,7 +1919,7 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                     }
 
                     // Free the original page, since we've replaced it
-                    if !manager.free_if_uncommitted(page.get_page_number()) {
+                    if !manager.free_if_uncommitted(page.get_page_number())? {
                         freed.push(page.get_page_number());
                     }
                     (
@@ -1857,21 +1938,21 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                 if page1 == child_page {
                     // NO-OP. One of our descendants is uncommitted, so there was no change
                     (
-                        NodeHandle::new(mutpage.get_page_number(), valid_message_bytes),
+                        NodeHandle::new(mutpage.get_page_number(), valid_messages),
                         None,
                         guard,
                     )
                 } else if manager.uncommitted(page.get_page_number()) {
-                    assert_eq!(valid_message_bytes, 0);
+                    assert_eq!(valid_messages, 0);
                     mutator.write_child_page(child_index, page1);
                     (
-                        NodeHandle::new(mutpage.get_page_number(), valid_message_bytes),
+                        NodeHandle::new(mutpage.get_page_number(), valid_messages),
                         None,
                         guard,
                     )
-                } else if mutator.can_write_delta_message(valid_message_bytes) {
+                } else if mutator.can_write_delta_message(valid_messages) {
                     let new_messages =
-                        mutator.write_delta_message(valid_message_bytes, child_index as u8, page1);
+                        mutator.write_delta_message(valid_messages, child_index as u8, page1);
                     (
                         NodeHandle::new(mutpage.get_page_number(), new_messages),
                         None,
@@ -1880,7 +1961,7 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                 } else {
                     // Page is full of delta messages, so rewrite it
                     let mut new_page = manager
-                        .allocate(InternalBuilder::required_bytes(accessor.total_key_length()));
+                        .allocate(InternalBuilder::required_bytes(accessor.total_key_length()))?;
                     let mut builder = InternalBuilder::new(&mut new_page);
                     copy_to_builder_and_patch(
                         &accessor,
@@ -1893,7 +1974,7 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
                     );
 
                     // Free the original page, since we've replaced it
-                    if !manager.free_if_uncommitted(page.get_page_number()) {
+                    if !manager.free_if_uncommitted(page.get_page_number())? {
                         freed.push(page.get_page_number());
                     }
                     (NodeHandle::new(new_page.get_page_number(), 0), None, guard)
@@ -1901,13 +1982,13 @@ fn tree_insert_helper<'a, K: RedbKey + ?Sized>(
             }
         }
         _ => unreachable!(),
-    }
+    })
 }
 
 // Returns the (offset, len) of the value for the queried key, if present
 pub(in crate) fn lookup_in_raw<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     table: u64,
     query: &[u8],
     manager: &'a TransactionalMemory,
@@ -1944,7 +2025,7 @@ pub(in crate) fn lookup_in_raw<'a, K: RedbKey + ?Sized>(
             }
         }
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             let (_, child_page) = accessor.child_for_key::<K>(table, query);
             return lookup_in_raw::<K>(
                 manager.get_page(child_page.get_page_number()),
@@ -1960,7 +2041,7 @@ pub(in crate) fn lookup_in_raw<'a, K: RedbKey + ?Sized>(
 
 pub(in crate) fn find_iter_unbounded_start<'a>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     mut parent: Option<Box<RangeIterState<'a>>>,
     table: u64,
     manager: &'a TransactionalMemory,
@@ -1973,13 +2054,13 @@ pub(in crate) fn find_iter_unbounded_start<'a>(
             reversed: false,
         }),
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             let (child_index, child_page_number) = accessor.first_child_for_table(table);
             let child_page = manager.get_page(child_page_number.get_page_number());
             if child_index < BTREE_ORDER - 1 && accessor.child_page(child_index + 1).is_some() {
                 parent = Some(Box::new(Internal {
                     page,
-                    valid_message_bytes,
+                    valid_messages,
                     child: child_index + 1,
                     parent,
                     reversed: false,
@@ -1999,7 +2080,7 @@ pub(in crate) fn find_iter_unbounded_start<'a>(
 
 pub(in crate) fn find_iter_unbounded_reversed<'a>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     mut parent: Option<Box<RangeIterState<'a>>>,
     table: u64,
     manager: &'a TransactionalMemory,
@@ -2012,13 +2093,13 @@ pub(in crate) fn find_iter_unbounded_reversed<'a>(
             reversed: false,
         }),
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             let (child_index, child_page_number) = accessor.last_child_for_table(table);
             let child_page = manager.get_page(child_page_number.get_page_number());
             if child_index < BTREE_ORDER - 1 && accessor.child_page(child_index + 1).is_some() {
                 parent = Some(Box::new(Internal {
                     page,
-                    valid_message_bytes,
+                    valid_messages,
                     child: child_index + 1,
                     parent,
                     reversed: false,
@@ -2038,7 +2119,7 @@ pub(in crate) fn find_iter_unbounded_reversed<'a>(
 
 pub(in crate) fn find_iter_start<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     mut parent: Option<Box<RangeIterState<'a>>>,
     table: u64,
     query: &[u8],
@@ -2052,13 +2133,13 @@ pub(in crate) fn find_iter_start<'a, K: RedbKey + ?Sized>(
             reversed: false,
         }),
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             let (child_index, child_page_number) = accessor.child_for_key::<K>(table, query);
             let child_page = manager.get_page(child_page_number.get_page_number());
             if child_index < BTREE_ORDER - 1 && accessor.child_page(child_index + 1).is_some() {
                 parent = Some(Box::new(Internal {
                     page,
-                    valid_message_bytes,
+                    valid_messages,
                     child: child_index + 1,
                     parent,
                     reversed: false,
@@ -2079,7 +2160,7 @@ pub(in crate) fn find_iter_start<'a, K: RedbKey + ?Sized>(
 
 pub(in crate) fn find_iter_start_reversed<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
-    valid_message_bytes: u32,
+    valid_messages: u8,
     mut parent: Option<Box<RangeIterState<'a>>>,
     table: u64,
     query: &[u8],
@@ -2093,13 +2174,13 @@ pub(in crate) fn find_iter_start_reversed<'a, K: RedbKey + ?Sized>(
             reversed: true,
         }),
         INTERNAL => {
-            let accessor = InternalAccessor::new(&page, valid_message_bytes);
+            let accessor = InternalAccessor::new(&page, valid_messages);
             let (child_index, child_page_number) = accessor.child_for_key::<K>(table, query);
             let child_page = manager.get_page(child_page_number.get_page_number());
             if child_index > 0 && accessor.child_page(child_index - 1).is_some() {
                 parent = Some(Box::new(Internal {
                     page,
-                    valid_message_bytes,
+                    valid_messages,
                     child: child_index - 1,
                     parent,
                     reversed: true,
diff --git a/src/tree_store/page_store/page_allocator.rs b/src/tree_store/page_store/page_allocator.rs
index 4944dd6..49dc56d 100644
--- a/src/tree_store/page_store/page_allocator.rs
+++ b/src/tree_store/page_store/page_allocator.rs
@@ -1,3 +1,4 @@
+use crate::Error;
 use std::convert::TryInto;
 use std::mem::size_of;
 
@@ -227,7 +228,7 @@ impl PageAllocator {
     }
 
     /// data must have been initialized by Self::init_new()
-    pub(crate) fn alloc(&self, data: &mut [u8]) -> Option<u64> {
+    pub(crate) fn alloc(&self, data: &mut [u8]) -> Result<u64, Error> {
         if let Some(mut entry) = self.get_level(data, 0).first_unset(0, 64) {
             let mut height = 0;
 
@@ -242,9 +243,9 @@ impl PageAllocator {
 
             assert!(entry < self.get_num_pages() as usize);
             self.record_alloc(data, entry as u64);
-            Some(entry as u64)
+            Ok(entry as u64)
         } else {
-            None
+            Err(Error::OutOfSpace)
         }
     }
 
@@ -269,6 +270,7 @@ impl PageAllocator {
 #[cfg(test)]
 mod test {
     use crate::tree_store::page_store::page_allocator::PageAllocator;
+    use crate::Error;
     use rand::prelude::IteratorRandom;
     use rand::rngs::StdRng;
     use rand::{Rng, SeedableRng};
@@ -283,7 +285,10 @@ mod test {
         for i in 0..num_pages {
             assert_eq!(i as u64, allocator.alloc(&mut data).unwrap());
         }
-        assert!(allocator.alloc(&mut data).is_none());
+        assert!(matches!(
+            allocator.alloc(&mut data).unwrap_err(),
+            Error::OutOfSpace
+        ));
     }
 
     #[test]
@@ -292,7 +297,10 @@ mod test {
         let allocator = PageAllocator::init_new(&mut data, 2);
         allocator.record_alloc(&mut data, 0);
         assert_eq!(1, allocator.alloc(&mut data).unwrap());
-        assert!(allocator.alloc(&mut data).is_none());
+        assert!(matches!(
+            allocator.alloc(&mut data).unwrap_err(),
+            Error::OutOfSpace
+        ));
     }
 
     #[test]
@@ -300,7 +308,10 @@ mod test {
         let mut data = vec![0; PageAllocator::required_space(1)];
         let allocator = PageAllocator::init_new(&mut data, 1);
         assert_eq!(0, allocator.alloc(&mut data).unwrap());
-        assert!(allocator.alloc(&mut data).is_none());
+        assert!(matches!(
+            allocator.alloc(&mut data).unwrap_err(),
+            Error::OutOfSpace
+        ));
         allocator.free(&mut data, 0);
         assert_eq!(0, allocator.alloc(&mut data).unwrap());
     }
@@ -317,7 +328,10 @@ mod test {
         allocator.free(&mut data, 15);
         assert_eq!(5, allocator.alloc(&mut data).unwrap());
         assert_eq!(15, allocator.alloc(&mut data).unwrap());
-        assert!(allocator.alloc(&mut data).is_none());
+        assert!(matches!(
+            allocator.alloc(&mut data).unwrap_err(),
+            Error::OutOfSpace
+        ));
     }
 
     #[test]
@@ -326,7 +340,7 @@ mod test {
         let mut data = vec![0; PageAllocator::required_space(num_pages)];
         let allocator = PageAllocator::init_new(&mut data, num_pages);
         // Allocate everything
-        while allocator.alloc(&mut data).is_some() {}
+        while allocator.alloc(&mut data).is_ok() {}
         // The last u64 must be used, since the leaf layer is compact
         let l = data.len();
         assert_ne!(0, u64::from_be_bytes(data[(l - 8)..].try_into().unwrap()));
@@ -346,7 +360,7 @@ mod test {
 
         for _ in 0..(num_pages * 2) {
             if rng.gen_bool(0.75) {
-                if let Some(page) = allocator.alloc(&mut data) {
+                if let Ok(page) = allocator.alloc(&mut data) {
                     allocated.insert(page);
                 } else {
                     assert_eq!(allocated.len(), num_pages);
@@ -362,7 +376,10 @@ mod test {
         for _ in allocated.len()..num_pages {
             allocator.alloc(&mut data).unwrap();
         }
-        assert!(allocator.alloc(&mut data).is_none());
+        assert!(matches!(
+            allocator.alloc(&mut data).unwrap_err(),
+            Error::OutOfSpace
+        ));
 
         for i in 0..num_pages {
             allocator.free(&mut data, i as u64);
@@ -371,6 +388,9 @@ mod test {
         for _ in 0..num_pages {
             allocator.alloc(&mut data).unwrap();
         }
-        assert!(allocator.alloc(&mut data).is_none());
+        assert!(matches!(
+            allocator.alloc(&mut data).unwrap_err(),
+            Error::OutOfSpace
+        ));
     }
 }
diff --git a/src/tree_store/page_store/page_manager.rs b/src/tree_store/page_store/page_manager.rs
index ac6a0ea..25aa84f 100644
--- a/src/tree_store/page_store/page_manager.rs
+++ b/src/tree_store/page_store/page_manager.rs
@@ -25,8 +25,8 @@ const DB_METAPAGE_SIZE: usize = TRANSACTION_1_OFFSET + TRANSACTION_SIZE;
 
 // Structure of each metapage
 const ROOT_PAGE_OFFSET: usize = 0;
-const ROOT_PAGE_MESSAGE_BYTES_OFFSET: usize = ROOT_PAGE_OFFSET + size_of::<u64>();
-const TRANSACTION_ID_OFFSET: usize = ROOT_PAGE_MESSAGE_BYTES_OFFSET + size_of::<u32>();
+const ROOT_PAGE_MESSAGES_OFFSET: usize = ROOT_PAGE_OFFSET + size_of::<u64>();
+const TRANSACTION_ID_OFFSET: usize = ROOT_PAGE_MESSAGES_OFFSET + size_of::<u8>();
 // Memory pointed to by this ptr is logically part of the metapage
 const ALLOCATOR_STATE_PTR_OFFSET: usize = TRANSACTION_ID_OFFSET + size_of::<u128>();
 const ALLOCATOR_STATE_LEN_OFFSET: usize = ALLOCATOR_STATE_PTR_OFFSET + size_of::<u64>();
@@ -137,22 +137,17 @@ impl<'a> TransactionAccessor<'a> {
         TransactionAccessor { mem, _guard: guard }
     }
 
-    fn get_root_page(&self) -> Option<(PageNumber, u32)> {
+    fn get_root_page(&self) -> Option<(PageNumber, u8)> {
         let num = PageNumber::from_be_bytes(
             self.mem[ROOT_PAGE_OFFSET..(ROOT_PAGE_OFFSET + 8)]
                 .try_into()
                 .unwrap(),
         );
-        let message_bytes = u32::from_be_bytes(
-            self.mem[ROOT_PAGE_MESSAGE_BYTES_OFFSET
-                ..(ROOT_PAGE_MESSAGE_BYTES_OFFSET + size_of::<u32>())]
-                .try_into()
-                .unwrap(),
-        );
+        let messages = self.mem[ROOT_PAGE_MESSAGES_OFFSET];
         if num.page_index == 0 {
             None
         } else {
-            Some((num, message_bytes))
+            Some((num, messages))
         }
     }
 
@@ -204,11 +199,10 @@ impl<'a> TransactionMutator<'a> {
         TransactionMutator { mem, _guard: guard }
     }
 
-    fn set_root_page(&mut self, page_number: PageNumber, valid_message_bytes: u32) {
+    fn set_root_page(&mut self, page_number: PageNumber, valid_messages: u8) {
         self.mem[ROOT_PAGE_OFFSET..(ROOT_PAGE_OFFSET + 8)]
             .copy_from_slice(&page_number.to_be_bytes());
-        self.mem[ROOT_PAGE_MESSAGE_BYTES_OFFSET..(ROOT_PAGE_MESSAGE_BYTES_OFFSET + 4)]
-            .copy_from_slice(&valid_message_bytes.to_be_bytes());
+        self.mem[ROOT_PAGE_MESSAGES_OFFSET] = valid_messages;
     }
 
     fn set_last_committed_transaction_id(&mut self, transaction_id: u128) {
@@ -324,7 +318,8 @@ pub(crate) struct TransactionalMemory {
     allocated_since_commit: RefCell<HashSet<PageNumber>>,
     freed_since_commit: RefCell<Vec<PageNumber>>,
     // Metapage guard lock should be held when using this to modify the page allocator state
-    page_allocator: PageAllocator,
+    // May be None, if the allocator state was corrupted when the file was opened
+    page_allocator: Option<PageAllocator>,
     mmap: MmapMut,
     // We use unsafe to access the metapage (page 0), and so guard it with this mutex
     // It would be nice if this was a RefCell<&[u8]> on the metapage. However, that would be
@@ -358,14 +353,14 @@ impl TransactionalMemory {
         requested_page_size: Option<usize>,
     ) -> Result<Self, Error> {
         let mutex = Mutex::new(MetapageGuard {});
-        let usable_pages = Self::calculate_usable_pages(mmap.len());
-        let page_allocator = PageAllocator::new(usable_pages);
         if mmap[0..MAGICNUMBER.len()] != MAGICNUMBER {
             let page_size = requested_page_size.unwrap_or_else(get_page_size);
             // Ensure that the database metadata fits into the first page
             assert!(page_size >= DB_METAPAGE_SIZE);
             assert!(page_size.is_power_of_two());
 
+            let usable_pages = Self::calculate_usable_pages(mmap.len());
+
             // Explicitly zero the memory
             mmap[0..DB_METAPAGE_SIZE].copy_from_slice(&[0; DB_METAPAGE_SIZE]);
             for i in &mut mmap[(usable_pages * page_size)..] {
@@ -440,13 +435,19 @@ impl TransactionalMemory {
         );
 
         let accessor = TransactionAccessor::new(get_primary(&mmap), mutex.lock().unwrap());
-        // TODO: repair instead of crashing
-        assert!(!accessor.get_allocator_dirty());
+        let mut allocator_dirty = accessor.get_allocator_dirty();
         drop(accessor);
         let accessor = TransactionAccessor::new(get_secondary(&mut mmap), mutex.lock().unwrap());
-        assert!(!accessor.get_allocator_dirty());
+        allocator_dirty |= accessor.get_allocator_dirty();
         drop(accessor);
 
+        let page_allocator = if allocator_dirty {
+            None
+        } else {
+            let usable_pages = Self::calculate_usable_pages(mmap.len());
+            Some(PageAllocator::new(usable_pages))
+        };
+
         Ok(TransactionalMemory {
             allocated_since_commit: RefCell::new(HashSet::new()),
             freed_since_commit: RefCell::new(vec![]),
@@ -459,6 +460,76 @@ impl TransactionalMemory {
         })
     }
 
+    pub(crate) fn needs_repair(&self) -> bool {
+        let (mmap, guard) = self.acquire_mutable_metapage();
+        let accessor = TransactionAccessor::new(get_primary(mmap), guard);
+        let mut allocator_dirty = accessor.get_allocator_dirty();
+        drop(accessor);
+
+        let (mmap, guard) = self.acquire_mutable_metapage();
+        let accessor = TransactionAccessor::new(get_secondary(mmap), guard);
+        allocator_dirty |= accessor.get_allocator_dirty();
+
+        allocator_dirty
+    }
+
+    // Returns true if the repair is complete. If false, this method must be called again
+    pub(crate) fn repair_allocator(
+        &self,
+        allocated_pages: impl Iterator<Item = PageNumber>,
+    ) -> Result<bool, Error> {
+        // TODO: clean up all the duplicated code in this function
+        let (mmap, guard) = self.acquire_mutable_metapage();
+        let mut mutator = TransactionMutator::new(get_secondary(mmap), guard);
+        if mutator.get_allocator_dirty() {
+            let (mem, guard) = self.acquire_mutable_page_allocator(mutator)?;
+
+            let usable_pages = Self::calculate_usable_pages(self.mmap.len());
+            let allocator = PageAllocator::init_new(mem, usable_pages);
+            for page in allocated_pages {
+                allocator.record_alloc(mem, page.page_index);
+            }
+            self.mmap.flush()?;
+
+            drop(guard);
+            let (mmap, guard) = self.acquire_mutable_metapage();
+            let mut mutator = TransactionMutator::new(get_secondary(mmap), guard);
+            mutator.set_allocator_dirty(false);
+            self.mmap.flush()?;
+
+            Ok(false)
+        } else {
+            // Repair the primary instead
+            drop(mutator);
+            let (mmap, guard) = self.acquire_mutable_metapage();
+            mutator = TransactionMutator::new(get_primary_mut(mmap), guard);
+
+            let (mem, guard) = self.acquire_mutable_page_allocator(mutator)?;
+
+            let usable_pages = Self::calculate_usable_pages(self.mmap.len());
+            let allocator = PageAllocator::init_new(mem, usable_pages);
+            for page in allocated_pages {
+                allocator.record_alloc(mem, page.page_index);
+            }
+            self.mmap.flush()?;
+
+            drop(guard);
+            let (mmap, guard) = self.acquire_mutable_metapage();
+            mutator = TransactionMutator::new(get_primary_mut(mmap), guard);
+            mutator.set_allocator_dirty(false);
+            self.mmap.flush()?;
+
+            Ok(true)
+        }
+    }
+
+    pub(crate) fn finalize_repair_allocator(&mut self) {
+        assert!(!self.needs_repair());
+        let usable_pages = Self::calculate_usable_pages(self.mmap.len());
+        let allocator = PageAllocator::new(usable_pages);
+        self.page_allocator = Some(allocator);
+    }
+
     fn acquire_mutable_metapage(&self) -> (&mut [u8], MutexGuard<MetapageGuard>) {
         let guard = self.metapage_guard.lock().unwrap();
         let ptr = &self.mmap as *const MmapMut as *mut MmapMut;
@@ -494,6 +565,7 @@ impl TransactionalMemory {
         // no more writes can happen to the pages it allocated. Thus it is safe to make them visible
         // to future read transactions
         assert!(self.open_dirty_pages.borrow().is_empty());
+        assert!(self.page_allocator.is_some());
 
         let (mmap, guard) = self.acquire_mutable_metapage();
         let mut mutator = TransactionMutator::new(get_secondary(mmap), guard);
@@ -523,11 +595,16 @@ impl TransactionalMemory {
         for page_number in self.allocated_since_commit.borrow_mut().drain() {
             assert_eq!(page_number.page_order, 0);
             self.page_allocator
+                .as_ref()
+                .unwrap()
                 .record_alloc(mem, page_number.page_index);
         }
         for page_number in self.freed_since_commit.borrow_mut().drain(..) {
             assert_eq!(page_number.page_order, 0);
-            self.page_allocator.free(mem, page_number.page_index);
+            self.page_allocator
+                .as_ref()
+                .unwrap()
+                .free(mem, page_number.page_index);
         }
         drop(guard); // Ensure the guard lives past all the writes to the page allocator state
         self.read_from_secondary.store(false, Ordering::SeqCst);
@@ -541,6 +618,7 @@ impl TransactionalMemory {
         // no more writes can happen to the pages it allocated. Thus it is safe to make them visible
         // to future read transactions
         assert!(self.open_dirty_pages.borrow().is_empty());
+        assert!(self.page_allocator.is_some());
 
         let (mmap, guard) = self.acquire_mutable_metapage();
         let mut mutator = TransactionMutator::new(get_secondary(mmap), guard);
@@ -562,6 +640,8 @@ impl TransactionalMemory {
         for page_number in self.allocated_since_commit.borrow_mut().drain() {
             assert_eq!(page_number.page_order, 0);
             self.page_allocator
+                .as_ref()
+                .unwrap()
                 .record_alloc(mem, page_number.page_index);
         }
         assert!(self.freed_since_commit.borrow().is_empty());
@@ -578,11 +658,16 @@ impl TransactionalMemory {
         let (mem, guard) = self.acquire_mutable_page_allocator(mutator)?;
         for page_number in self.allocated_since_commit.borrow_mut().drain() {
             assert_eq!(page_number.page_order, 0);
-            self.page_allocator.free(mem, page_number.page_index);
+            self.page_allocator
+                .as_ref()
+                .unwrap()
+                .free(mem, page_number.page_index);
         }
         for page_number in self.freed_since_commit.borrow_mut().drain(..) {
             assert_eq!(page_number.page_order, 0);
             self.page_allocator
+                .as_ref()
+                .unwrap()
                 .record_alloc(mem, page_number.page_index);
         }
         // Drop guard only after page_allocator calls are completed
@@ -625,7 +710,7 @@ impl TransactionalMemory {
         }
     }
 
-    pub(crate) fn get_primary_root_page(&self) -> Option<(PageNumber, u32)> {
+    pub(crate) fn get_primary_root_page(&self) -> Option<(PageNumber, u8)> {
         if self.read_from_secondary.load(Ordering::SeqCst) {
             TransactionAccessor::new(
                 get_secondary_const(&self.mmap),
@@ -653,36 +738,43 @@ impl TransactionalMemory {
 
     // TODO: valid_message_bytes kind of breaks the separation of concerns for the PageManager.
     // It's only used by the delta message protocol of the b-tree
-    pub(crate) fn set_secondary_root_page(&self, root_page: PageNumber, valid_message_bytes: u32) {
+    pub(crate) fn set_secondary_root_page(&self, root_page: PageNumber, valid_messages: u8) {
         let (mmap, guard) = self.acquire_mutable_metapage();
         let mut mutator = TransactionMutator::new(get_secondary(mmap), guard);
-        mutator.set_root_page(root_page, valid_message_bytes);
+        mutator.set_root_page(root_page, valid_messages);
     }
 
-    pub(crate) fn free(&self, page: PageNumber) {
+    pub(crate) fn free(&self, page: PageNumber) -> Result<(), Error> {
         let (mmap, guard) = self.acquire_mutable_metapage();
         let mutator = TransactionMutator::new(get_secondary(mmap), guard);
-        // TODO: should propagate this error
-        let (mem, guard) = self.acquire_mutable_page_allocator(mutator).unwrap();
+        let (mem, guard) = self.acquire_mutable_page_allocator(mutator)?;
         assert_eq!(page.page_order, 0);
-        self.page_allocator.free(mem, page.page_index);
+        self.page_allocator
+            .as_ref()
+            .unwrap()
+            .free(mem, page.page_index);
         drop(guard);
         self.freed_since_commit.borrow_mut().push(page);
+
+        Ok(())
     }
 
     // Frees the page if it was allocated since the last commit. Returns true, if the page was freed
-    pub(crate) fn free_if_uncommitted(&self, page: PageNumber) -> bool {
+    pub(crate) fn free_if_uncommitted(&self, page: PageNumber) -> Result<bool, Error> {
         if self.allocated_since_commit.borrow_mut().remove(&page) {
             let (mmap, guard) = self.acquire_mutable_metapage();
             let mutator = TransactionMutator::new(get_secondary(mmap), guard);
-            // TODO: should propagate this error
-            let (mem, guard) = self.acquire_mutable_page_allocator(mutator).unwrap();
+            let (mem, guard) = self.acquire_mutable_page_allocator(mutator)?;
             assert_eq!(page.page_order, 0);
-            self.page_allocator.free(mem, page.page_index);
+            self.page_allocator
+                .as_ref()
+                .unwrap()
+                .free(mem, page.page_index);
             drop(guard);
-            true
+
+            Ok(true)
         } else {
-            false
+            Ok(false)
         }
     }
 
@@ -691,15 +783,13 @@ impl TransactionalMemory {
         self.allocated_since_commit.borrow().contains(&page)
     }
 
-    pub(crate) fn allocate(&self, allocation_size: usize) -> PageMut {
+    pub(crate) fn allocate(&self, allocation_size: usize) -> Result<PageMut, Error> {
         assert!(allocation_size <= self.page_size);
 
         let (mmap, guard) = self.acquire_mutable_metapage();
         let mutator = TransactionMutator::new(get_secondary(mmap), guard);
-        // TODO: should propagate this error
-        let (mem, guard) = self.acquire_mutable_page_allocator(mutator).unwrap();
-        // TODO: handle out-of-memory and return an error
-        let page_number = PageNumber::new(self.page_allocator.alloc(mem).unwrap(), 0);
+        let (mem, guard) = self.acquire_mutable_page_allocator(mutator)?;
+        let page_number = PageNumber::new(self.page_allocator.as_ref().unwrap().alloc(mem)?, 0);
         // Drop guard only after page_allocator.alloc() is completed
         drop(guard);
 
@@ -717,11 +807,11 @@ impl TransactionalMemory {
         // Zero the memory
         mem.copy_from_slice(&vec![0u8; page_number.page_size_bytes(self.page_size)]);
 
-        PageMut {
+        Ok(PageMut {
             mem,
             page_number,
             open_pages: &self.open_dirty_pages,
-        }
+        })
     }
 
     pub(crate) fn count_free_pages(&self) -> usize {
@@ -730,7 +820,7 @@ impl TransactionalMemory {
         // and avoid dirtying the allocator state
         let mutator = TransactionMutator::new(get_secondary(mmap), guard);
         let (mem, guard) = self.acquire_mutable_page_allocator(mutator).unwrap();
-        let count = self.page_allocator.count_free_pages(mem);
+        let count = self.page_allocator.as_ref().unwrap().count_free_pages(mem);
         // Drop guard only after page_allocator.count_free() is completed
         drop(guard);
 
@@ -746,7 +836,7 @@ impl Drop for TransactionalMemory {
             self.commit(non_durable_transaction_id)
                 .expect("Failure while finalizing non-durable commit. Database may be corrupted");
         }
-        if self.mmap.flush().is_ok() {
+        if self.mmap.flush().is_ok() && self.page_allocator.is_some() {
             let (metamem, guard) = self.acquire_mutable_metapage();
             let mut mutator = TransactionMutator::new(get_secondary(metamem), guard);
             mutator.set_allocator_dirty(false);
@@ -754,3 +844,55 @@ impl Drop for TransactionalMemory {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use crate::tree_store::page_store::page_manager::{
+        get_primary_mut, get_secondary, MetapageGuard, TransactionMutator,
+    };
+    use crate::tree_store::page_store::TransactionalMemory;
+    use crate::{Database, Table};
+    use memmap2::MmapMut;
+    use std::fs::OpenOptions;
+    use std::sync::Mutex;
+    use tempfile::NamedTempFile;
+
+    #[test]
+    fn repair_allocator() {
+        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
+        write_txn.commit().unwrap();
+        drop(db);
+
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(tmpfile.path())
+            .unwrap();
+
+        let mut mmap = unsafe { MmapMut::map_mut(&file) }.unwrap();
+
+        let mutex = Mutex::new(MetapageGuard {});
+        let mut mutator = TransactionMutator::new(get_secondary(&mut mmap), mutex.lock().unwrap());
+        mutator.set_allocator_dirty(true);
+        drop(mutator);
+        mmap.flush().unwrap();
+
+        let mut mutator =
+            TransactionMutator::new(get_primary_mut(&mut mmap), mutex.lock().unwrap());
+        mutator.set_allocator_dirty(true);
+        drop(mutator);
+        mmap.flush().unwrap();
+
+        assert!(TransactionalMemory::new(mmap, None).unwrap().needs_repair());
+
+        let db2 = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+        let write_txn = db2.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello2", b"world2").unwrap();
+        write_txn.commit().unwrap();
+    }
+}
diff --git a/src/tree_store/storage.rs b/src/tree_store/storage.rs
index 14f993d..aeae2cb 100644
--- a/src/tree_store/storage.rs
+++ b/src/tree_store/storage.rs
@@ -1,8 +1,9 @@
 use crate::tree_store::base_types::NodeHandle;
 use crate::tree_store::btree_utils::{
     find_iter_start, find_iter_start_reversed, find_iter_unbounded_reversed,
-    find_iter_unbounded_start, lookup_in_raw, make_mut_single_leaf, print_tree, tree_delete,
-    tree_height, tree_insert, AccessGuardMut, BtreeEntry, BtreeRangeIter,
+    find_iter_unbounded_start, lookup_in_raw, make_mut_single_leaf, page_numbers_iter_start_state,
+    print_tree, tree_delete, tree_height, tree_insert, AccessGuardMut, AllPageNumbersBtreeIter,
+    BtreeEntry, BtreeRangeIter,
 };
 use crate::tree_store::page_store::{Page, PageImpl, PageNumber, TransactionalMemory};
 use crate::types::{RedbKey, RedbValue, WithLifetime};
@@ -121,7 +122,18 @@ pub(in crate) struct Storage {
 
 impl Storage {
     pub(in crate) fn new(mmap: MmapMut, page_size: Option<usize>) -> Result<Storage, Error> {
-        let mem = TransactionalMemory::new(mmap, page_size)?;
+        let mut mem = TransactionalMemory::new(mmap, page_size)?;
+        while mem.needs_repair() {
+            let (root, messages) = mem
+                .get_primary_root_page()
+                .expect("Tried to repair an empty database");
+            let root_page = mem.get_page(root);
+            let start = page_numbers_iter_start_state(root_page, messages, None, &mem);
+            let iter = AllPageNumbersBtreeIter::new(start.unwrap(), &mem);
+            mem.repair_allocator(iter)?;
+        }
+        mem.finalize_repair_allocator();
+
         let next_transaction_id = mem.get_last_committed_transaction_id() + 1;
 
         Ok(Storage {
@@ -288,13 +300,13 @@ impl Storage {
                 key,
                 value,
                 &self.mem,
-            );
+            )?;
             self.pending_freed_pages
                 .borrow_mut()
                 .extend_from_slice(&freed);
             Ok(new_root)
         } else {
-            let (new_root, _) = make_mut_single_leaf(table_id, key, value, &self.mem);
+            let (new_root, _) = make_mut_single_leaf(table_id, key, value, &self.mem)?;
             Ok(new_root)
         }
     }
@@ -319,13 +331,13 @@ impl Storage {
                 key,
                 &value,
                 &self.mem,
-            );
+            )?;
             self.pending_freed_pages
                 .borrow_mut()
                 .extend_from_slice(&freed);
             (new_root, guard)
         } else {
-            make_mut_single_leaf(table_id, key, &value, &self.mem)
+            make_mut_single_leaf(table_id, key, &value, &self.mem)?
         };
         Ok((new_root, guard))
     }
@@ -343,7 +355,7 @@ impl Storage {
     pub(in crate) fn get_root_page_number(&self) -> Option<NodeHandle> {
         self.mem
             .get_primary_root_page()
-            .map(|(page, message_bytes)| NodeHandle::new(page, message_bytes))
+            .map(|(page, messages)| NodeHandle::new(page, messages))
     }
 
     pub(in crate) fn commit(
@@ -364,7 +376,7 @@ impl Storage {
             new_root = self.store_freed_pages(transaction_id, new_root)?;
         } else {
             for page in self.pending_freed_pages.borrow_mut().drain(..) {
-                self.mem.free(page);
+                self.mem.free(page)?;
             }
         }
 
@@ -427,7 +439,7 @@ impl Storage {
             // 1..=length because the array is length prefixed
             for i in 1..=length {
                 let page = PageNumber::from_be_bytes(value[i * 8..(i + 1) * 8].try_into().unwrap());
-                self.mem.free(page);
+                self.mem.free(page)?;
             }
         }
         drop(iter);
@@ -658,7 +670,7 @@ impl Storage {
                 table_id,
                 key,
                 &self.mem,
-            );
+            )?;
             self.pending_freed_pages
                 .borrow_mut()
                 .extend_from_slice(&freed);


Test patch:
diff --git a/tests/basic_tests.rs b/tests/basic_tests.rs
new file mode 100644
index 0000000..bf713ca
--- /dev/null
+++ b/tests/basic_tests.rs
@@ -0,0 +1,197 @@
+use redb::{Database, Error, ReadOnlyTable, ReadableTable, Table};
+use tempfile::NamedTempFile;
+
+#[test]
+fn len() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    table.insert(b"hello2", b"world2").unwrap();
+    table.insert(b"hi", b"world").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(table.len().unwrap(), 3);
+}
+
+#[test]
+fn multiple_tables() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"1").unwrap();
+    let mut table2: Table<[u8], [u8]> = write_txn.open_table(b"2").unwrap();
+
+    table.insert(b"hello", b"world").unwrap();
+    table2.insert(b"hello", b"world2").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"1").unwrap();
+    let table2: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"2").unwrap();
+    assert_eq!(table.len().unwrap(), 1);
+    assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+    assert_eq!(table2.len().unwrap(), 1);
+    assert_eq!(b"world2", table2.get(b"hello").unwrap().unwrap().as_ref());
+}
+
+#[test]
+fn is_empty() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert!(!table.is_empty().unwrap());
+}
+
+#[test]
+fn abort() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"aborted").unwrap();
+    assert_eq!(b"aborted", table.get(b"hello").unwrap().unwrap().as_ref());
+    write_txn.abort().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: Result<ReadOnlyTable<[u8], [u8]>, Error> = read_txn.open_table(b"x");
+    assert!(table.is_err());
+
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+    assert_eq!(table.len().unwrap(), 1);
+}
+
+#[test]
+fn insert_overwrite() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"replaced").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(b"replaced", table.get(b"hello").unwrap().unwrap().as_ref());
+}
+
+#[test]
+fn insert_reserve() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    let value = b"world";
+    let mut reserved = table.insert_reserve(b"hello", value.len()).unwrap();
+    reserved.as_mut().copy_from_slice(value);
+    drop(reserved);
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(value, table.get(b"hello").unwrap().unwrap().as_ref());
+}
+
+#[test]
+fn delete() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    table.insert(b"hello2", b"world").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+    assert_eq!(table.len().unwrap(), 2);
+
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.remove(b"hello").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert!(table.get(b"hello").unwrap().is_none());
+    assert_eq!(table.len().unwrap(), 1);
+}
+
+#[test]
+fn no_dirty_reads() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: Result<ReadOnlyTable<[u8], [u8]>, Error> = read_txn.open_table(b"x");
+    assert!(table.is_err());
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+}
+
+#[test]
+fn read_isolation() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+
+    let write_txn = db.begin_write().unwrap();
+    let mut write_table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    write_table.remove(b"hello").unwrap();
+    write_table.insert(b"hello2", b"world2").unwrap();
+    write_table.insert(b"hello3", b"world3").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn2 = db.begin_read().unwrap();
+    let table2: ReadOnlyTable<[u8], [u8]> = read_txn2.open_table(b"x").unwrap();
+    assert!(table2.get(b"hello").unwrap().is_none());
+    assert_eq!(b"world2", table2.get(b"hello2").unwrap().unwrap().as_ref());
+    assert_eq!(b"world3", table2.get(b"hello3").unwrap().unwrap().as_ref());
+    assert_eq!(table2.len().unwrap(), 2);
+
+    assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+    assert!(table.get(b"hello2").unwrap().is_none());
+    assert!(table.get(b"hello3").unwrap().is_none());
+    assert_eq!(table.len().unwrap(), 1);
+}
diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs
index 2cb169c..9c34270 100644
--- a/tests/integration_tests.rs
+++ b/tests/integration_tests.rs
@@ -377,3 +377,22 @@ fn leaked_write() {
         panic!();
     }
 }
+
+#[test]
+fn non_page_size_multiple() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+
+    let db_size = 1024 * 1024 + 1;
+    let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
+
+    let key = vec![0; 1024];
+    let value = vec![0; 1];
+    table.insert(&key, &value).unwrap();
+    txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(table.len().unwrap(), 1);
+}
diff --git a/tests/multimap_tests.rs b/tests/multimap_tests.rs
new file mode 100644
index 0000000..6bde502
--- /dev/null
+++ b/tests/multimap_tests.rs
@@ -0,0 +1,143 @@
+use redb::{Database, MultimapTable, ReadOnlyMultimapTable, ReadableMultimapTable};
+use tempfile::NamedTempFile;
+
+fn get_vec(table: &ReadOnlyMultimapTable<[u8], [u8]>, key: &[u8]) -> Vec<Vec<u8>> {
+    let mut result = vec![];
+    let mut iter = table.get(key).unwrap();
+    loop {
+        let item = iter.next();
+        if let Some(item_value) = item {
+            result.push(item_value.to_vec());
+        } else {
+            return result;
+        }
+    }
+}
+
+#[test]
+fn len() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+
+    table.insert(b"hello", b"world").unwrap();
+    table.insert(b"hello", b"world2").unwrap();
+    table.insert(b"hi", b"world").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+    assert_eq!(table.len().unwrap(), 3);
+}
+
+#[test]
+fn is_empty() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+
+    let write_txn = db.begin_write().unwrap();
+    let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+    assert!(!table.is_empty().unwrap());
+}
+
+#[test]
+fn insert() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    table.insert(b"hello", b"world2").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+    assert_eq!(
+        vec![b"world".to_vec(), b"world2".to_vec()],
+        get_vec(&table, b"hello")
+    );
+    assert_eq!(table.len().unwrap(), 2);
+}
+
+#[test]
+fn range_query() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+    for i in 0..5u8 {
+        let value = vec![i];
+        table.insert(b"0", &value).unwrap();
+    }
+    for i in 5..10u8 {
+        let value = vec![i];
+        table.insert(b"1", &value).unwrap();
+    }
+    for i in 10..15u8 {
+        let value = vec![i];
+        table.insert(b"2", &value).unwrap();
+    }
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+    let start = b"0".as_ref();
+    let end = b"1".as_ref();
+    let mut iter = table.get_range(start..=end).unwrap();
+    for i in 0..10u8 {
+        let (key, value) = iter.next().unwrap();
+        if i < 5 {
+            assert_eq!(b"0", key);
+        } else {
+            assert_eq!(b"1", key);
+        }
+        assert_eq!(&[i], value);
+    }
+    assert!(iter.next().is_none());
+}
+
+#[test]
+fn delete() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+    let write_txn = db.begin_write().unwrap();
+    let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+    table.insert(b"hello", b"world").unwrap();
+    table.insert(b"hello", b"world2").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+    assert_eq!(
+        vec![b"world".to_vec(), b"world2".to_vec()],
+        get_vec(&table, b"hello")
+    );
+    assert_eq!(table.len().unwrap(), 2);
+
+    let write_txn = db.begin_write().unwrap();
+    let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+    table.remove(b"hello", b"world2").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+    assert_eq!(vec![b"world".to_vec()], get_vec(&table, b"hello"));
+    assert_eq!(table.len().unwrap(), 1);
+
+    let write_txn = db.begin_write().unwrap();
+    let mut table: MultimapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+    table.remove_all(b"hello").unwrap();
+    write_txn.commit().unwrap();
+
+    let read_txn = db.begin_read().unwrap();
+    let table: ReadOnlyMultimapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+    assert!(table.is_empty().unwrap());
+    let empty: Vec<Vec<u8>> = vec![];
+    assert_eq!(empty, get_vec(&table, b"hello"));
+}