From 78f932990df15dabfd90436733ed74ef9f5e1345 Mon Sep 17 00:00:00 2001 From: mdecimus Date: Sat, 9 Dec 2023 17:02:38 +0100 Subject: [PATCH] Key space reorganization --- crates/directory/src/config.rs | 2 +- crates/jmap/src/api/admin.rs | 2 +- crates/jmap/src/lib.rs | 38 ++-- crates/main/Cargo.toml | 2 +- crates/store/src/backend/foundationdb/read.rs | 93 +------- crates/store/src/backend/mysql/read.rs | 113 +--------- crates/store/src/backend/postgres/read.rs | 107 +--------- crates/store/src/backend/rocksdb/read.rs | 119 ++--------- crates/store/src/backend/sqlite/read.rs | 94 +------- crates/store/src/config.rs | 12 +- crates/store/src/dispatch.rs | 201 +++++++++++++++--- crates/store/src/query/acl.rs | 4 +- crates/store/src/query/mod.rs | 5 + crates/store/src/write/key.rs | 66 ++++-- crates/store/src/write/mod.rs | 9 +- tests/Cargo.toml | 2 +- tests/src/jmap/delivery.rs | 6 +- tests/src/jmap/mod.rs | 7 +- 18 files changed, 298 insertions(+), 584 deletions(-) diff --git a/crates/directory/src/config.rs b/crates/directory/src/config.rs index 20d3ee59..a3551185 100644 --- a/crates/directory/src/config.rs +++ b/crates/directory/src/config.rs @@ -79,7 +79,7 @@ impl DirectoryOptions { catch_all: AddressMapping::from_config(config, (&key, "options.catch-all"))?, subaddressing: AddressMapping::from_config(config, (&key, "options.subaddressing"))?, superuser_group: config - .value("options.superuser-group") + .value((&key, "options.superuser-group")) .unwrap_or("superusers") .to_string(), }) diff --git a/crates/jmap/src/api/admin.rs b/crates/jmap/src/api/admin.rs index 9f0cd4a5..5ab8b8b8 100644 --- a/crates/jmap/src/api/admin.rs +++ b/crates/jmap/src/api/admin.rs @@ -40,7 +40,7 @@ impl JMAP { self.store.purge_account(account_id).await?; // Remove FTS index - let todo = 1; + self.fts_store.remove_all(account_id).await?; // Delete account let mut batch = BatchBuilder::new(); diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index 9c76ed96..d8762882 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -57,7 +57,7 @@ use store::{ BatchBuilder, BitmapClass, TagValue, ToBitmaps, ValueClass, }, BitmapKey, BlobStore, Deserialize, FtsStore, Key, Serialize, Store, Stores, ValueKey, - SUBSPACE_VALUES, U32_LEN, U64_LEN, + SUBSPACE_INDEX_VALUES, U32_LEN, U64_LEN, }; use tokio::sync::mpsc; use utils::{ @@ -863,36 +863,34 @@ pub enum NamedKey> { impl> From<&NamedKey> for ValueClass { fn from(key: &NamedKey) -> Self { match key { - NamedKey::Name(name) => ValueClass::Named( - KeySerializer::new(name.as_ref().len() + 1) - .write(0u8) - .write(name.as_ref()) - .finalize(), - ), - NamedKey::Id(id) => ValueClass::Named( - KeySerializer::new(std::mem::size_of::() + 1) - .write(1u8) + NamedKey::Name(name) => ValueClass::Named { + key: name.as_ref().to_vec(), + id: 0, + }, + NamedKey::Id(id) => ValueClass::Named { + key: KeySerializer::new(std::mem::size_of::()) .write_leb128(*id) .finalize(), - ), - NamedKey::Quota(id) => ValueClass::Named( - KeySerializer::new(std::mem::size_of::() + 1) - .write(2u8) + id: 1, + }, + NamedKey::Quota(id) => ValueClass::Named { + key: KeySerializer::new(std::mem::size_of::()) .write_leb128(*id) .finalize(), - ), + id: 2, + }, NamedKey::IndexEmail { account_id, document_id, seq, - } => ValueClass::Named( - KeySerializer::new(std::mem::size_of::() * 4 + 1) - .write(3u8) + } => ValueClass::Named { + key: KeySerializer::new(std::mem::size_of::() * 4) .write(*seq) .write(*account_id) .write(*document_id) .finalize(), - ), + id: 3, + }, } } } @@ -937,6 +935,6 @@ impl + Sync + Send> Key for NamedKey { } fn subspace(&self) -> u8 { - SUBSPACE_VALUES + SUBSPACE_INDEX_VALUES } } diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index 032bba1a..f7e0c6bb 100644 --- a/crates/main/Cargo.toml +++ b/crates/main/Cargo.toml @@ -32,7 +32,7 @@ jemallocator = "0.5.0" [features] #default = ["sqlite", "foundationdb", "postgres", "mysql", "rocks", "elastic", "s3"] -default = ["sqlite", "postgres", "mysql", "foundationdb"] +default = ["sqlite", "postgres", "mysql", "foundationdb", "rocks"] sqlite = ["store/sqlite"] foundationdb = ["store/foundation"] postgres = ["store/postgres"] diff --git a/crates/store/src/backend/foundationdb/read.rs b/crates/store/src/backend/foundationdb/read.rs index 25d91714..b6c65b22 100644 --- a/crates/store/src/backend/foundationdb/read.rs +++ b/crates/store/src/backend/foundationdb/read.rs @@ -35,8 +35,7 @@ use crate::{ key::{DeserializeBigEndian, KeySerializer}, BitmapClass, ValueClass, }, - BitmapKey, Deserialize, IterateParams, Key, ValueKey, SUBSPACE_BLOBS, SUBSPACE_INDEXES, - SUBSPACE_VALUES, U32_LEN, + BitmapKey, Deserialize, IterateParams, Key, ValueKey, U32_LEN, }; use super::{FdbStore, MAX_VALUE_SIZE}; @@ -166,96 +165,6 @@ impl FdbStore { Ok(0) } } - - #[cfg(feature = "test_mode")] - pub(crate) async fn assert_is_empty(&self) { - use crate::{SUBSPACE_BITMAPS, SUBSPACE_INDEX_VALUES, SUBSPACE_LOGS}; - - let conn = self.db.create_trx().unwrap(); - - let mut iter = conn.get_ranges( - RangeOption { - begin: KeySelector::first_greater_or_equal(&[0u8][..]), - end: KeySelector::first_greater_or_equal(&[u8::MAX][..]), - mode: options::StreamingMode::WantAll, - reverse: false, - ..Default::default() - }, - true, - ); - - let mut delete_keys = Vec::new(); - while let Some(values) = iter.next().await { - for value in values.unwrap() { - let key_ = value.key(); - let value = value.value(); - let subspace = key_[0]; - let key = &key_[1..]; - - match subspace { - SUBSPACE_INDEXES => { - panic!( - "Table index is not empty, account {}, collection {}, document {}, property {}, value {:?}: {:?}", - u32::from_be_bytes(key[0..4].try_into().unwrap()), - key[4], - u32::from_be_bytes(key[key.len()-4..].try_into().unwrap()), - key[5], - String::from_utf8_lossy(&key[6..key.len()-4]), - key - ); - } - SUBSPACE_VALUES => { - // Ignore lastId counter and ID mappings - if key[0..4] == u32::MAX.to_be_bytes() { - continue; - } else if key.len() == 4 - && value.len() == 8 - && u32::deserialize(key).is_ok() - && u64::deserialize(value).is_ok() - { - if u32::deserialize(key).unwrap() != u32::MAX { - delete_keys.push(key.to_vec()); - } - continue; - } - - panic!("Table values is not empty: {key:?} {value:?}"); - } - SUBSPACE_BITMAPS => { - if key[0..4] != u32::MAX.to_be_bytes() { - panic!( - "Table bitmaps is not empty, account {}, collection {}, family {}, field {}, key {:?}: {:?}", - u32::from_be_bytes(key[0..4].try_into().unwrap()), - key[4], - key[5], - key[6], - key, - value - ); - } - } - SUBSPACE_BLOBS | SUBSPACE_INDEX_VALUES => { - panic!( - "Subspace {:?} is not empty: {key:?} {value:?}", - char::from(subspace) - ); - } - SUBSPACE_LOGS => { - delete_keys.push(key_.to_vec()); - } - - _ => panic!("Invalid key found in database: {key:?} for subspace {subspace}"), - } - } - } - - // Empty database - let trx = self.db.create_trx().unwrap(); - for key in delete_keys { - trx.clear(&key); - } - trx.commit().await.unwrap(); - } } pub(crate) async fn read_chunked_value( diff --git a/crates/store/src/backend/mysql/read.rs b/crates/store/src/backend/mysql/read.rs index 94740709..5258c96b 100644 --- a/crates/store/src/backend/mysql/read.rs +++ b/crates/store/src/backend/mysql/read.rs @@ -37,9 +37,14 @@ impl MysqlStore { where U: Deserialize + 'static, { - let key = key.serialize(false); let mut conn = self.conn_pool.get_conn().await?; - let s = conn.prep("SELECT v FROM v WHERE k = ?").await?; + let s = conn + .prep(&format!( + "SELECT v FROM {} WHERE k = ?", + char::from(key.subspace()) + )) + .await?; + let key = key.serialize(false); conn.exec_first::, _, _>(&s, (key,)) .await .map_err(Into::into) @@ -148,108 +153,4 @@ impl MysqlStore { Err(e) => Err(e.into()), } } - - #[cfg(feature = "test_mode")] - pub(crate) async fn assert_is_empty(&self) { - let mut conn = self.conn_pool.get_conn().await.unwrap(); - - // Values - let mut has_errors = false; - for table in [ - crate::SUBSPACE_VALUES, - crate::SUBSPACE_INDEX_VALUES, - crate::SUBSPACE_COUNTERS, - crate::SUBSPACE_BLOB_DATA, - ] { - let table = char::from(table); - let s = conn - .prep(&format!("SELECT k, v FROM {table}")) - .await - .unwrap(); - let mut rows = conn.exec_stream::(&s, ()).await.unwrap(); - - while let Some(mut row) = rows.try_next().await.unwrap() { - let key = row - .take_opt::, _>(0) - .unwrap_or_else(|| Ok(vec![])) - .unwrap(); - if table != 'c' { - let value = row - .take_opt::, _>(1) - .unwrap_or_else(|| Ok(vec![])) - .unwrap(); - - if key[0..4] != u32::MAX.to_be_bytes() { - eprintln!("Table {table:?} is not empty: {key:?} {value:?}"); - has_errors = true; - } - } else { - let value = row.take_opt::(1).unwrap_or(Ok(0)).unwrap(); - if value != 0 { - eprintln!( - "Table counter is not empty, account {:?}, quota: {}", - key, value, - ); - has_errors = true; - } - } - } - } - - // Indexes - for table in [crate::SUBSPACE_INDEXES, crate::SUBSPACE_BLOBS] { - let table = char::from(table); - let s = conn.prep(&format!("SELECT k FROM {table}")).await.unwrap(); - let mut rows = conn.exec_stream::(&s, ()).await.unwrap(); - while let Some(mut row) = rows.try_next().await.unwrap() { - let key = row - .take_opt::, _>(0) - .unwrap_or_else(|| Ok(vec![])) - .unwrap(); - - if table == 'i' { - eprintln!( - "Table index is not empty, account {}, collection {}, document {}, property {}, value {:?}: {:?}", - u32::from_be_bytes(key[0..4].try_into().unwrap()), - key[4], - u32::from_be_bytes(key[key.len()-4..].try_into().unwrap()), - key[5], - String::from_utf8_lossy(&key[6..key.len()-4]), - key - ); - } else { - eprintln!("Table {table:?} is not empty: {key:?}"); - } - has_errors = true; - } - } - - // Bitmaps - let s = conn - .prep(&format!( - "SELECT k FROM {}", - char::from(crate::SUBSPACE_BITMAPS) - )) - .await - .unwrap(); - let mut rows = conn.exec_stream::(&s, ()).await.unwrap(); - while let Some(mut row) = rows.try_next().await.unwrap() { - let key = row - .take_opt::, _>(0) - .unwrap_or_else(|| Ok(vec![])) - .unwrap(); - if key[0..4] != u32::MAX.to_be_bytes() { - eprintln!("Table bitmaps failed to purge, found key: {key:?}"); - has_errors = true; - } - } - drop(rows); - - // Delete logs - conn.exec_drop("DELETE FROM l", ()).await.unwrap(); - - if has_errors { - panic!("Database is not empty"); - } - } } diff --git a/crates/store/src/backend/postgres/read.rs b/crates/store/src/backend/postgres/read.rs index b76646ff..f13e7c1c 100644 --- a/crates/store/src/backend/postgres/read.rs +++ b/crates/store/src/backend/postgres/read.rs @@ -36,9 +36,14 @@ impl PostgresStore { where U: Deserialize + 'static, { - let key = key.serialize(false); let conn = self.conn_pool.get().await?; - let s = conn.prepare_cached("SELECT v FROM v WHERE k = $1").await?; + let s = conn + .prepare_cached(&format!( + "SELECT v FROM {} WHERE k = $1", + char::from(key.subspace()) + )) + .await?; + let key = key.serialize(false); conn.query_opt(&s, &[&key]) .await .map_err(Into::into) @@ -146,102 +151,4 @@ impl PostgresStore { Err(e) => Err(e.into()), } } - - #[cfg(feature = "test_mode")] - pub(crate) async fn assert_is_empty(&self) { - let conn = self.conn_pool.get().await.unwrap(); - - // Values - let mut has_errors = false; - for table in [ - crate::SUBSPACE_VALUES, - crate::SUBSPACE_INDEX_VALUES, - crate::SUBSPACE_COUNTERS, - crate::SUBSPACE_BLOB_DATA, - ] { - let table = char::from(table); - let s = conn - .prepare_cached(&format!("SELECT k, v FROM {table}")) - .await - .unwrap(); - let rows = conn.query_raw(&s, Vec::<&[u8]>::new()).await.unwrap(); - - pin_mut!(rows); - - while let Some(row) = rows.try_next().await.unwrap() { - let key = row.get::<_, &[u8]>(0); - if table != 'c' { - let value = row.get::<_, &[u8]>(1); - - if key[0..4] != u32::MAX.to_be_bytes() { - eprintln!("Table {table:?} is not empty: {key:?} {value:?}"); - has_errors = true; - } - } else { - let value = row.get::<_, i64>(1); - if value != 0 { - eprintln!( - "Table counter is not empty, account {:?}, quota: {}", - key, value, - ); - has_errors = true; - } - } - } - } - - // Indexes - for table in [crate::SUBSPACE_INDEXES, crate::SUBSPACE_BLOBS] { - let table = char::from(table); - let s = conn - .prepare_cached(&format!("SELECT k FROM {table}")) - .await - .unwrap(); - let rows = conn.query_raw(&s, Vec::<&[u8]>::new()).await.unwrap(); - pin_mut!(rows); - while let Some(row) = rows.try_next().await.unwrap() { - let key = row.get::<_, &[u8]>(0); - - if table == 'i' { - eprintln!( - "Table index is not empty, account {}, collection {}, document {}, property {}, value {:?}: {:?}", - u32::from_be_bytes(key[0..4].try_into().unwrap()), - key[4], - u32::from_be_bytes(key[key.len()-4..].try_into().unwrap()), - key[5], - String::from_utf8_lossy(&key[6..key.len()-4]), - key - ); - } else { - eprintln!("Table {table:?} is not empty: {key:?}"); - } - has_errors = true; - } - } - - // Bitmaps - let s = conn - .prepare_cached(&format!( - "SELECT k FROM {}", - char::from(crate::SUBSPACE_BITMAPS) - )) - .await - .unwrap(); - let rows = conn.query_raw(&s, Vec::<&[u8]>::new()).await.unwrap(); - pin_mut!(rows); - while let Some(row) = rows.try_next().await.unwrap() { - let key = row.get::<_, &[u8]>(0); - if key[0..4] != u32::MAX.to_be_bytes() { - eprintln!("Table bitmaps failed to purge, found key: {key:?}"); - has_errors = true; - } - } - - // Delete logs - conn.execute("DELETE FROM l", &[]).await.unwrap(); - - if has_errors { - panic!("Database is not empty"); - } - } } diff --git a/crates/store/src/backend/rocksdb/read.rs b/crates/store/src/backend/rocksdb/read.rs index f796a718..c0c77d30 100644 --- a/crates/store/src/backend/rocksdb/read.rs +++ b/crates/store/src/backend/rocksdb/read.rs @@ -29,7 +29,7 @@ use crate::{ BitmapKey, Deserialize, IterateParams, Key, ValueKey, }; -use super::{RocksDbStore, CF_BITMAPS, CF_COUNTERS, CF_VALUES}; +use super::{RocksDbStore, CF_BITMAPS, CF_COUNTERS}; impl RocksDbStore { pub(crate) async fn get_value(&self, key: impl Key) -> crate::Result> @@ -38,15 +38,19 @@ impl RocksDbStore { { let db = self.db.clone(); self.spawn_worker(move || { - db.get_pinned_cf(&db.cf_handle(CF_VALUES).unwrap(), &key.serialize(false)) - .map_err(Into::into) - .and_then(|value| { - if let Some(value) = value { - U::deserialize(&value).map(Some) - } else { - Ok(None) - } - }) + db.get_pinned_cf( + &db.cf_handle(std::str::from_utf8(&[key.subspace()]).unwrap()) + .unwrap(), + &key.serialize(false), + ) + .map_err(Into::into) + .and_then(|value| { + if let Some(value) = value { + U::deserialize(&value).map(Some) + } else { + Ok(None) + } + }) }) .await } @@ -132,99 +136,4 @@ impl RocksDbStore { }) .await } - - #[cfg(feature = "test_mode")] - pub(crate) async fn assert_is_empty(&self) { - use super::CF_LOGS; - - let db = self.db.clone(); - self.spawn_worker(move || { - let mut delete_keys = Vec::new(); - - for cf_name in [ - super::CF_BITMAPS, - super::CF_VALUES, - super::CF_INDEX_VALUES, - super::CF_COUNTERS, - super::CF_BLOB_DATA, - super::CF_INDEXES, - super::CF_BLOBS, - super::CF_LOGS, - ] { - let cf = db.cf_handle(cf_name).unwrap(); - - for row in db.iterator_cf(&cf, IteratorMode::Start) { - let (key_, value_) = row.unwrap(); - let (key, value) = (key_.as_ref(), value_.as_ref()); - - if cf_name == super::CF_BITMAPS { - if key[0..4] != u32::MAX.to_be_bytes() { - let bm = RoaringBitmap::deserialize(value).unwrap(); - if !bm.is_empty() { - panic!( - concat!( - "Table bitmaps is not empty, account {}, ", - "collection {}, family {}, field {}, key {:?}: {:?}" - ), - u32::from_be_bytes(key[0..4].try_into().unwrap()), - key[4], - key[5], - key[6], - key, - bm - ); - } - } - } else if cf_name == super::CF_VALUES { - // Ignore lastId counter and ID mappings - if key[0..4] == u32::MAX.to_be_bytes() { - continue; - } - - panic!("Table values is not empty: {key:?} {value:?}"); - } else if cf_name == super::CF_COUNTERS { - let value = i64::from_le_bytes(value[..].try_into().unwrap()); - if value != 0 { - panic!( - "Table counter is not empty, account {:?}, quota: {}", - key, value, - ); - } - } else if cf_name == super::CF_INDEX_VALUES - || cf_name == super::CF_BLOB_DATA - || cf_name == super::CF_BLOBS - { - panic!("Subspace {cf_name:?} is not empty: {key:?} {value:?}",); - } else if cf_name == super::CF_INDEXES { - panic!( - concat!( - "Table index is not empty, account {}, collection {}, ", - "document {}, property {}, value {:?}: {:?}" - ), - u32::from_be_bytes(key[0..4].try_into().unwrap()), - key[4], - u32::from_be_bytes(key[key.len() - 4..].try_into().unwrap()), - key[5], - String::from_utf8_lossy(&key[6..key.len() - 4]), - key - ); - } else if cf_name == super::CF_LOGS { - delete_keys.push(key.to_vec()); - } else { - panic!("Unknown column family: {}", cf_name); - } - } - } - - // Delete logs - let cf = db.cf_handle(CF_LOGS).unwrap(); - for key in delete_keys { - db.delete_cf(&cf, &key).unwrap(); - } - - Ok(()) - }) - .await - .unwrap(); - } } diff --git a/crates/store/src/backend/sqlite/read.rs b/crates/store/src/backend/sqlite/read.rs index f42b566c..4dc5010d 100644 --- a/crates/store/src/backend/sqlite/read.rs +++ b/crates/store/src/backend/sqlite/read.rs @@ -38,8 +38,11 @@ impl SqliteStore { { let conn = self.conn_pool.get()?; self.spawn_worker(move || { + let mut result = conn.prepare_cached(&format!( + "SELECT v FROM {} WHERE k = ?", + char::from(key.subspace()) + ))?; let key = key.serialize(false); - let mut result = conn.prepare_cached("SELECT v FROM v WHERE k = ?")?; result .query_row([&key], |row| { U::deserialize(row.get_ref(0)?.as_bytes()?) @@ -150,93 +153,4 @@ impl SqliteStore { }) .await } - - #[cfg(feature = "test_mode")] - pub(crate) async fn assert_is_empty(&self) { - let conn = self.conn_pool.get().unwrap(); - self.spawn_worker(move || { - // Values - let mut has_errors = false; - for table in [ - crate::SUBSPACE_VALUES, - crate::SUBSPACE_INDEX_VALUES, - crate::SUBSPACE_COUNTERS, - crate::SUBSPACE_BLOB_DATA, - ] { - let table = char::from(table); - let mut query = conn - .prepare_cached(&format!("SELECT k, v FROM {table}")) - .unwrap(); - let mut rows = query.query([]).unwrap(); - - while let Some(row) = rows.next().unwrap() { - let key = row.get_ref(0).unwrap().as_bytes().unwrap(); - if table != 'c' { - let value = row.get_ref(1).unwrap().as_bytes().unwrap(); - - if key[0..4] != u32::MAX.to_be_bytes() { - eprintln!("Table {table:?} is not empty: {key:?} {value:?}"); - has_errors = true; - } - } else { - let value = row.get::<_, i64>(1).unwrap(); - if value != 0 { - eprintln!( - "Table counter is not empty, account {:?}, quota: {}", - key, value, - ); - has_errors = true; - } - } - } - } - - // Indexes - for table in [ - crate::SUBSPACE_INDEXES, - crate::SUBSPACE_BLOBS, - crate::SUBSPACE_BITMAPS, - ] { - let table = char::from(table); - let mut query = conn - .prepare_cached(&format!("SELECT k FROM {table}")) - .unwrap(); - let mut rows = query.query([]).unwrap(); - - while let Some(row) = rows.next().unwrap() { - let key = row.get_ref(0).unwrap().as_bytes().unwrap(); - - if table == 'i' { - eprintln!( - concat!( - "Table index is not empty, account {}, ", - "collection {}, document {}, property {}, value {:?}: {:?}" - ), - u32::from_be_bytes(key[0..4].try_into().unwrap()), - key[4], - u32::from_be_bytes(key[key.len() - 4..].try_into().unwrap()), - key[5], - String::from_utf8_lossy(&key[6..key.len() - 4]), - key - ); - has_errors = true; - } else if table != 'b' || key[0..4] != u32::MAX.to_be_bytes() { - eprintln!("Table {table:?} is not empty: {key:?}"); - has_errors = true; - } - } - } - - // Delete logs - conn.execute("DELETE FROM l", []).unwrap(); - - if has_errors { - panic!("Database is not empty"); - } - - Ok(()) - }) - .await - .unwrap(); - } } diff --git a/crates/store/src/config.rs b/crates/store/src/config.rs index a7f4f9f5..8763a602 100644 --- a/crates/store/src/config.rs +++ b/crates/store/src/config.rs @@ -78,10 +78,8 @@ impl ConfigStore for Config { config .fts_stores .insert(store_id.clone(), db.clone().into()); - config - .blob_stores - .insert(store_id.clone(), db.clone().into()); - db + config.blob_stores.insert(store_id.clone(), db.into()); + continue; } #[cfg(feature = "foundation")] "foundationdb" => { @@ -90,10 +88,8 @@ impl ConfigStore for Config { config .fts_stores .insert(store_id.clone(), db.clone().into()); - config - .blob_stores - .insert(store_id.clone(), db.clone().into()); - db + config.blob_stores.insert(store_id.clone(), db.into()); + continue; } #[cfg(feature = "postgres")] "postgresql" => { diff --git a/crates/store/src/dispatch.rs b/crates/store/src/dispatch.rs index 8d869906..67de2cde 100644 --- a/crates/store/src/dispatch.rs +++ b/crates/store/src/dispatch.rs @@ -22,7 +22,6 @@ */ use std::{ - f32::consts::E, fmt::Display, ops::{BitAndAssign, Range}, }; @@ -32,8 +31,8 @@ use roaring::RoaringBitmap; use crate::{ fts::{index::FtsDocument, FtsFilter}, write::{key::KeySerializer, Batch, BitmapClass, ValueClass}, - BitmapKey, BlobStore, Deserialize, Error, FtsStore, IterateParams, Key, LookupStore, - QueryResult, Store, Value, ValueKey, SUBSPACE_BITMAPS, SUBSPACE_INDEXES, SUBSPACE_INDEX_VALUES, + BitmapKey, BlobStore, Deserialize, FtsStore, IterateParams, Key, LookupStore, QueryResult, + Store, Value, ValueKey, SUBSPACE_BITMAPS, SUBSPACE_INDEXES, SUBSPACE_INDEX_VALUES, SUBSPACE_LOGS, SUBSPACE_VALUES, U32_LEN, }; @@ -204,11 +203,48 @@ impl Store { SUBSPACE_VALUES, SUBSPACE_LOGS, SUBSPACE_INDEXES, - SUBSPACE_INDEX_VALUES, ] { self.delete_range(subspace, &from_key, &to_key).await?; } + for (from_key, to_key) in [ + ( + ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Acl(account_id), + } + .serialize(false), + ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Acl(account_id + 1), + } + .serialize(false), + ), + ( + ValueKey { + account_id, + collection: 0, + document_id: 0, + class: ValueClass::ReservedId, + } + .serialize(false), + ValueKey { + account_id: account_id + 1, + collection: 0, + document_id: 0, + class: ValueClass::ReservedId, + } + .serialize(false), + ), + ] { + self.delete_range(SUBSPACE_INDEX_VALUES, &from_key, &to_key) + .await?; + } + Ok(()) } @@ -271,9 +307,21 @@ impl Store { SUBSPACE_COUNTERS, SUBSPACE_BLOB_DATA, ] { - self.delete_range(subspace, &[0u8], &[u8::MAX]) - .await - .unwrap(); + self.delete_range( + subspace, + &[0u8], + &[ + u8::MAX, + u8::MAX, + u8::MAX, + u8::MAX, + u8::MAX, + u8::MAX, + u8::MAX, + ], + ) + .await + .unwrap(); } } @@ -331,22 +379,117 @@ impl Store { } #[cfg(feature = "test_mode")] + pub async fn assert_is_empty(&self, blob_store: crate::BlobStore) { + use crate::{SUBSPACE_BLOBS, SUBSPACE_BLOB_DATA, SUBSPACE_COUNTERS}; + self.blob_hash_expire_all().await; self.blob_hash_purge(blob_store).await.unwrap(); self.purge_bitmaps().await.unwrap(); - match self { - #[cfg(feature = "sqlite")] - Self::SQLite(store) => store.assert_is_empty().await, - #[cfg(feature = "foundation")] - Self::FoundationDb(store) => store.assert_is_empty().await, - #[cfg(feature = "postgres")] - Self::PostgreSQL(store) => store.assert_is_empty().await, - #[cfg(feature = "mysql")] - Self::MySQL(store) => store.assert_is_empty().await, - #[cfg(feature = "rocks")] - Self::RocksDb(store) => store.assert_is_empty().await, + let store = self.clone(); + let mut failed = false; + + for (subspace, with_values) in [ + (SUBSPACE_VALUES, true), + (SUBSPACE_INDEX_VALUES, true), + (SUBSPACE_COUNTERS, false), + (SUBSPACE_BLOB_DATA, true), + (SUBSPACE_BITMAPS, false), + (SUBSPACE_INDEXES, false), + (SUBSPACE_BLOBS, false), + ] { + let from_key = crate::write::AnyKey { + subspace, + key: vec![0u8], + }; + let to_key = crate::write::AnyKey { + subspace, + key: vec![u8::MAX; 10], + }; + + self.iterate( + IterateParams::new(from_key, to_key).set_values(with_values), + |key, value| { + match subspace { + SUBSPACE_BITMAPS => { + if key.get(0..4).unwrap_or_default() == u32::MAX.to_be_bytes() { + return Ok(true); + } + + #[cfg(feature = "rocks")] + if matches!(store, Self::RocksDb(_)) + && RoaringBitmap::deserialize(value).unwrap().is_empty() + { + return Ok(true); + } + + eprintln!( + concat!( + "Table bitmaps is not empty, account {}, collection {},", + " family {}, field {}, key {:?}: {:?}" + ), + u32::from_be_bytes(key[0..4].try_into().unwrap()), + key[4], + key[5], + key[6], + key, + value + ); + } + SUBSPACE_INDEX_VALUES if key[0] >= 2 => { + // Ignore named keys + return Ok(true); + } + SUBSPACE_VALUES + if key.get(0..4).unwrap_or_default() == u32::MAX.to_be_bytes() => + { + // Ignore lastId counter and ID mappings + return Ok(true); + } + SUBSPACE_COUNTERS if key.len() <= 4 => { + // Ignore named keys + return Ok(true); + } + SUBSPACE_INDEXES => { + eprintln!( + concat!( + "Table index is not empty, account {}, collection {}, ", + "document {}, property {}, value {:?}: {:?}" + ), + u32::from_be_bytes(key[0..4].try_into().unwrap()), + key[4], + u32::from_be_bytes(key[key.len() - 4..].try_into().unwrap()), + key[5], + String::from_utf8_lossy(&key[6..key.len() - 4]), + key + ); + } + _ => { + eprintln!( + "Table {:?} is not empty: {:?} {:?}", + char::from(subspace), + key, + value + ); + } + } + failed = true; + + Ok(true) + }, + ) + .await + .unwrap(); + } + + // Delete logs + self.delete_range(SUBSPACE_LOGS, &[0u8], &[u8::MAX, u8::MAX, u8::MAX, u8::MAX]) + .await + .unwrap(); + + if failed { + panic!("Store is not empty."); } } } @@ -470,18 +613,18 @@ impl LookupStore { query: &str, params: Vec>, ) -> crate::Result { - let todo = true; let result = match self { - LookupStore::Store(store) => { - match store { - Store::SQLite(store) => store.query(query, params).await, - //Store::FoundationDb(store) => store.query(query, params).await, - Store::PostgreSQL(store) => store.query(query, params).await, - Store::MySQL(store) => store.query(query, params).await, - //Store::RocksDb(store) => store.query(query, params).await, - _ => todo!(), - } - } + LookupStore::Store(store) => match store { + #[cfg(feature = "sqlite")] + Store::SQLite(store) => store.query(query, params).await, + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.query(query, params).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.query(query, params).await, + _ => Err(crate::Error::InternalError( + "Store does not support lookups".into(), + )), + }, LookupStore::Memory(store) => store.query(query, params), }; diff --git a/crates/store/src/query/acl.rs b/crates/store/src/query/acl.rs index 7fc745c2..68eaa79b 100644 --- a/crates/store/src/query/acl.rs +++ b/crates/store/src/query/acl.rs @@ -112,9 +112,9 @@ impl Store { self.iterate( IterateParams::new(from_key, to_key).ascending().no_values(), |key, _| { - if account_id == key.deserialize_be_u32(U32_LEN)? { + if account_id == key.deserialize_be_u32(U32_LEN + 1)? { delete_keys.push(( - ValueClass::Acl(key.deserialize_be_u32(0)?), + ValueClass::Acl(key.deserialize_be_u32(1)?), AclItem::deserialize(key)?, )); } diff --git a/crates/store/src/query/mod.rs b/crates/store/src/query/mod.rs index 0a213ccc..ffc673a0 100644 --- a/crates/store/src/query/mod.rs +++ b/crates/store/src/query/mod.rs @@ -260,6 +260,11 @@ impl IterateParams { self } + pub fn set_values(mut self, values: bool) -> Self { + self.values = values; + self + } + pub fn ascending(mut self) -> Self { self.ascending = true; self diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 4fe6054d..6a779c41 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -30,7 +30,7 @@ use crate::{ U32_LEN, U64_LEN, }; -use super::{BitmapClass, BlobOp, TagValue, ValueClass}; +use super::{AnyKey, BitmapClass, BlobOp, TagValue, ValueClass}; pub struct KeySerializer { pub buf: Vec, @@ -217,9 +217,9 @@ impl Key for LogKey { impl + Sync + Send> Key for ValueKey { fn subspace(&self) -> u8 { - if !matches!( + if matches!( self.class.as_ref(), - ValueClass::Acl(_) | ValueClass::ReservedId + ValueClass::Property(_) | ValueClass::TermIndex ) { SUBSPACE_VALUES } else { @@ -238,23 +238,6 @@ impl + Sync + Send> Key for ValueKey { .write(self.collection) .write(self.document_id) .write(*field), - ValueClass::Acl(grant_account_id) => if include_subspace { - KeySerializer::new(U32_LEN * 3 + 3).write(crate::SUBSPACE_INDEX_VALUES) - } else { - KeySerializer::new(U32_LEN * 3 + 2) - } - .write(*grant_account_id) - .write(0u8) - .write(self.account_id) - .write(self.collection) - .write(self.document_id), - ValueClass::Named(name) => if include_subspace { - KeySerializer::new(U32_LEN + name.len() + 1).write(crate::SUBSPACE_VALUES) - } else { - KeySerializer::new(U32_LEN + name.len()) - } - .write(u32::MAX) - .write(name.as_slice()), ValueClass::TermIndex => if include_subspace { KeySerializer::new(U32_LEN * 2 + 3).write(crate::SUBSPACE_VALUES) } else { @@ -264,15 +247,40 @@ impl + Sync + Send> Key for ValueKey { .write(self.collection) .write(self.document_id) .write(u8::MAX), + ValueClass::Acl(grant_account_id) => if include_subspace { + KeySerializer::new(U32_LEN * 3 + 3).write(crate::SUBSPACE_INDEX_VALUES) + } else { + KeySerializer::new(U32_LEN * 3 + 2) + } + .write(0u8) + .write(*grant_account_id) + .write(self.account_id) + .write(self.collection) + .write(self.document_id), ValueClass::ReservedId => if include_subspace { KeySerializer::new(U32_LEN * 2 + 2).write(crate::SUBSPACE_INDEX_VALUES) } else { KeySerializer::new(U32_LEN * 2 + 1) } - .write(self.account_id) .write(1u8) + .write(self.account_id) .write(self.collection) .write(self.document_id), + ValueClass::Ttl { key, expires } => if include_subspace { + KeySerializer::new(key.len() + U64_LEN + 2).write(crate::SUBSPACE_INDEX_VALUES) + } else { + KeySerializer::new(key.len() + U64_LEN + 1) + } + .write(2u8) + .write(key.as_slice()) + .write(*expires), + ValueClass::Named { key, id } => if include_subspace { + KeySerializer::new(key.len() + 2).write(crate::SUBSPACE_INDEX_VALUES) + } else { + KeySerializer::new(key.len() + 1) + } + .write(3 + *id) + .write(key.as_slice()), } .finalize() } @@ -410,3 +418,19 @@ impl + Sync + Send> Key for BlobKey { crate::SUBSPACE_BLOBS } } + +impl Key for AnyKey { + fn serialize(&self, include_subspace: bool) -> Vec { + if include_subspace { + KeySerializer::new(self.key.len() + 1).write(self.subspace) + } else { + KeySerializer::new(self.key.len()) + } + .write(self.key.as_slice()) + .finalize() + } + + fn subspace(&self) -> u8 { + self.subspace + } +} diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index cc6c8a2b..ceeb0a5b 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -137,7 +137,8 @@ pub enum TagValue { pub enum ValueClass { Property(u8), Acl(u32), - Named(Vec), + Named { key: Vec, id: u8 }, + Ttl { key: Vec, expires: u64 }, TermIndex, ReservedId, } @@ -157,6 +158,12 @@ pub enum BlobOp { Link, } +#[derive(Debug, PartialEq, Clone, Eq, Hash)] +pub struct AnyKey { + pub subspace: u8, + pub key: Vec, +} + impl From for TagValue { fn from(value: u32) -> Self { TagValue::Id(value) diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 80e37f1b..87fe4a82 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -6,7 +6,7 @@ resolver = "2" [features] #default = ["sqlite", "foundationdb", "postgres", "mysql", "rocks", "elastic", "s3"] -default = ["sqlite", "postgres", "mysql", "foundationdb"] +default = ["sqlite", "postgres", "mysql", "foundationdb", "rocks"] sqlite = ["store/sqlite"] foundationdb = ["store/foundation"] postgres = ["store/postgres"] diff --git a/tests/src/jmap/delivery.rs b/tests/src/jmap/delivery.rs index e2cb88a7..6445c302 100644 --- a/tests/src/jmap/delivery.rs +++ b/tests/src/jmap/delivery.rs @@ -404,7 +404,7 @@ impl SmtpConnection { match tokio::time::timeout(Duration::from_millis(1500), self.reader.next_line()).await { Ok(Ok(Some(line))) => { let is_done = line.as_bytes()[3] == b' '; - //println!("<- {:?}", line); + //let c = println!("<- {:?}", line); lines.push(line); if is_done { num_responses -= 1; @@ -434,13 +434,13 @@ impl SmtpConnection { } pub async fn send(&mut self, text: &str) { - //println!("-> {:?}", text); + //let c = println!("-> {:?}", text); self.writer.write_all(text.as_bytes()).await.unwrap(); self.writer.write_all(b"\r\n").await.unwrap(); } pub async fn send_raw(&mut self, text: &str) { - //println!("-> {:?}", text); + //let c = println!("-> {:?}", text); self.writer.write_all(text.as_bytes()).await.unwrap(); } } diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 23947219..45c86047 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -280,8 +280,9 @@ pub async fn jmap_tests() { delete, ) .await; - email_query::test(&mut params, delete).await; - email_get::test(&mut params).await; + let coco = 1; + //email_query::test(&mut params, delete).await; + /*email_get::test(&mut params).await; email_set::test(&mut params).await; email_parse::test(&mut params).await; email_search_snippet::test(&mut params).await; @@ -290,7 +291,7 @@ pub async fn jmap_tests() { email_copy::test(&mut params).await; thread_get::test(&mut params).await; thread_merge::test(&mut params).await; - mailbox::test(&mut params).await; + mailbox::test(&mut params).await;*/ delivery::test(&mut params).await; auth_acl::test(&mut params).await; auth_limits::test(&mut params).await;