From 566a2a0ab8f06bde4eb8086942be2b630086f7f2 Mon Sep 17 00:00:00 2001 From: mdecimus Date: Sat, 16 Dec 2023 21:24:34 +0100 Subject: [PATCH] Key space optimization --- .../directory/src/backend/internal/lookup.rs | 18 +- .../directory/src/backend/internal/manage.rs | 56 ++-- crates/jmap-proto/src/types/blob.rs | 11 +- crates/jmap/src/blob/copy.rs | 17 +- crates/jmap/src/blob/download.rs | 6 +- crates/jmap/src/blob/get.rs | 3 +- crates/jmap/src/blob/upload.rs | 35 ++- crates/jmap/src/email/index.rs | 26 +- crates/jmap/src/lib.rs | 4 +- crates/jmap/src/sieve/get.rs | 13 +- crates/jmap/src/sieve/set.rs | 34 ++- crates/jmap/src/vacation/set.rs | 17 +- crates/main/Cargo.toml | 2 +- crates/managesieve/src/op/putscript.rs | 24 +- crates/store/src/backend/foundationdb/blob.rs | 12 +- .../store/src/backend/foundationdb/write.rs | 18 +- crates/store/src/backend/mysql/main.rs | 17 +- crates/store/src/backend/mysql/write.rs | 19 +- crates/store/src/backend/postgres/main.rs | 8 +- crates/store/src/backend/postgres/write.rs | 22 +- crates/store/src/backend/rocksdb/blob.rs | 8 +- crates/store/src/backend/rocksdb/main.rs | 8 +- crates/store/src/backend/rocksdb/mod.rs | 5 +- crates/store/src/backend/rocksdb/write.rs | 21 +- crates/store/src/backend/sqlite/main.rs | 8 +- crates/store/src/backend/sqlite/write.rs | 20 +- crates/store/src/dispatch/store.rs | 51 ++-- crates/store/src/lib.rs | 20 +- crates/store/src/write/batch.rs | 16 +- crates/store/src/write/blob.rs | 251 ++++++++---------- crates/store/src/write/key.rs | 99 ++++--- crates/store/src/write/mod.rs | 29 +- tests/Cargo.toml | 2 +- tests/src/jmap/mod.rs | 2 +- tests/src/store/blob.rs | 186 ++++++++----- 35 files changed, 528 insertions(+), 560 deletions(-) diff --git a/crates/directory/src/backend/internal/lookup.rs b/crates/directory/src/backend/internal/lookup.rs index f353cba7..d173b793 100644 --- a/crates/directory/src/backend/internal/lookup.rs +++ b/crates/directory/src/backend/internal/lookup.rs @@ -23,7 +23,7 @@ use mail_send::Credentials; use store::{ - write::{DirectoryValue, ValueClass}, + write::{DirectoryClass, ValueClass}, IterateParams, Store, ValueKey, }; @@ -39,7 +39,7 @@ impl Directory for Store { QueryBy::Id(account_id) => { return self .get_value::>(ValueKey::from(ValueClass::Directory( - DirectoryValue::Principal(account_id), + DirectoryClass::Principal(account_id), ))) .await .map_err(Into::into); @@ -58,7 +58,7 @@ impl Directory for Store { if let Some(account_id) = self.get_account_id(username).await? { match ( self.get_value::>(ValueKey::from(ValueClass::Directory( - DirectoryValue::Principal(account_id), + DirectoryClass::Principal(account_id), ))) .await?, secret, @@ -76,7 +76,7 @@ impl Directory for Store { async fn email_to_ids(&self, email: &str) -> crate::Result> { self.get_value::>(ValueKey::from(ValueClass::Directory( - DirectoryValue::EmailToId(email.as_bytes().to_vec()), + DirectoryClass::EmailToId(email.as_bytes().to_vec()), ))) .await .map(|ids| ids.unwrap_or_default()) @@ -85,7 +85,7 @@ impl Directory for Store { async fn is_local_domain(&self, domain: &str) -> crate::Result { self.get_value::<()>(ValueKey::from(ValueClass::Directory( - DirectoryValue::Domain(domain.as_bytes().to_vec()), + DirectoryClass::Domain(domain.as_bytes().to_vec()), ))) .await .map(|ids| ids.is_some()) @@ -94,7 +94,7 @@ impl Directory for Store { async fn rcpt(&self, address: &str) -> crate::Result { self.get_value::<()>(ValueKey::from(ValueClass::Directory( - DirectoryValue::EmailToId(address.as_bytes().to_vec()), + DirectoryClass::EmailToId(address.as_bytes().to_vec()), ))) .await .map(|ids| ids.is_some()) @@ -107,8 +107,8 @@ impl Directory for Store { if address.len() > 3 { self.iterate( IterateParams::new( - ValueKey::from(ValueClass::Directory(DirectoryValue::EmailToId(vec![0u8]))), - ValueKey::from(ValueClass::Directory(DirectoryValue::EmailToId( + ValueKey::from(ValueClass::Directory(DirectoryClass::EmailToId(vec![0u8]))), + ValueKey::from(ValueClass::Directory(DirectoryClass::EmailToId( vec![u8::MAX; 10], ))), ) @@ -133,7 +133,7 @@ impl Directory for Store { for account_id in self.email_to_ids(address).await? { if let Some(email) = self .get_value::>(ValueKey::from(ValueClass::Directory( - DirectoryValue::Principal(account_id), + DirectoryClass::Principal(account_id), ))) .await? .and_then(|p| p.emails.into_iter().next()) diff --git a/crates/directory/src/backend/internal/manage.rs b/crates/directory/src/backend/internal/manage.rs index 91597b97..18e1e69b 100644 --- a/crates/directory/src/backend/internal/manage.rs +++ b/crates/directory/src/backend/internal/manage.rs @@ -23,7 +23,7 @@ use jmap_proto::types::collection::Collection; use store::{ - write::{assert::HashedValue, BatchBuilder, DirectoryValue, ValueClass}, + write::{assert::HashedValue, BatchBuilder, DirectoryClass, ValueClass}, IterateParams, Serialize, Store, ValueKey, }; @@ -62,7 +62,7 @@ pub trait ManageDirectory { impl ManageDirectory for Store { async fn get_account_name(&self, account_id: u32) -> crate::Result> { self.get_value::>(ValueKey::from(ValueClass::Directory( - DirectoryValue::Principal(account_id), + DirectoryClass::Principal(account_id), ))) .await .map_err(Into::into) @@ -84,7 +84,7 @@ impl ManageDirectory for Store { async fn get_account_id(&self, name: &str) -> crate::Result> { self.get_value::(ValueKey::from(ValueClass::Directory( - DirectoryValue::NameToId(name.as_bytes().to_vec()), + DirectoryClass::NameToId(name.as_bytes().to_vec()), ))) .await .map_err(Into::into) @@ -107,7 +107,7 @@ impl ManageDirectory for Store { // Write account ID let name_key = - ValueClass::Directory(DirectoryValue::NameToId(name.as_bytes().to_vec())); + ValueClass::Directory(DirectoryClass::NameToId(name.as_bytes().to_vec())); let mut batch = BatchBuilder::new(); batch .with_account_id(u32::MAX) @@ -116,7 +116,7 @@ impl ManageDirectory for Store { .assert_value(name_key.clone(), ()) .set(name_key, account_id.serialize()) .set( - ValueClass::Directory(DirectoryValue::Principal(account_id)), + ValueClass::Directory(DirectoryClass::Principal(account_id)), Principal { id: account_id, typ: Type::Individual, @@ -190,17 +190,17 @@ impl ManageDirectory for Store { let mut batch = BatchBuilder::new(); batch .assert_value( - ValueClass::Directory(DirectoryValue::NameToId( + ValueClass::Directory(DirectoryClass::NameToId( principal.name.clone().into_bytes(), )), (), ) .set( - ValueClass::Directory(DirectoryValue::Principal(account_id)), + ValueClass::Directory(DirectoryClass::Principal(account_id)), (&principal).serialize(), ) .set( - ValueClass::Directory(DirectoryValue::NameToId(principal.name.into_bytes())), + ValueClass::Directory(DirectoryClass::NameToId(principal.name.into_bytes())), account_id.serialize(), ); @@ -213,7 +213,7 @@ impl ManageDirectory for Store { for email in principal.emails { batch.set( - ValueClass::Directory(DirectoryValue::EmailToId(email.into_bytes())), + ValueClass::Directory(DirectoryClass::EmailToId(email.into_bytes())), (&ids).serialize(), ); } @@ -234,7 +234,7 @@ impl ManageDirectory for Store { let principal = self .get_value::>(ValueKey::from(ValueClass::Directory( - DirectoryValue::Principal(account_id), + DirectoryClass::Principal(account_id), ))) .await? .ok_or_else(|| { @@ -254,12 +254,12 @@ impl ManageDirectory for Store { let mut batch = BatchBuilder::new(); batch .with_account_id(account_id) - .clear(DirectoryValue::NameToId(principal.name.into_bytes())) - .clear(DirectoryValue::Principal(account_id)) - .clear(DirectoryValue::UsedQuota(account_id)); + .clear(DirectoryClass::NameToId(principal.name.into_bytes())) + .clear(DirectoryClass::Principal(account_id)) + .clear(DirectoryClass::UsedQuota(account_id)); for email in principal.emails { - batch.clear(DirectoryValue::EmailToId(email.into_bytes())); + batch.clear(DirectoryClass::EmailToId(email.into_bytes())); } self.write(batch.build()).await?; @@ -283,7 +283,7 @@ impl ManageDirectory for Store { // Fetch principal let mut principal = self .get_value::>>(ValueKey::from(ValueClass::Directory( - DirectoryValue::Principal(account_id), + DirectoryClass::Principal(account_id), ))) .await? .ok_or_else(|| { @@ -295,7 +295,7 @@ impl ManageDirectory for Store { let is_list = matches!(principal.inner.typ, Type::List); let mut has_list_changes = false; batch.assert_value( - ValueClass::Directory(DirectoryValue::Principal(account_id)), + ValueClass::Directory(DirectoryClass::Principal(account_id)), &principal, ); for change in changes { @@ -310,14 +310,14 @@ impl ManageDirectory for Store { )); } - batch.clear(ValueClass::Directory(DirectoryValue::NameToId( + batch.clear(ValueClass::Directory(DirectoryClass::NameToId( principal.inner.name.as_bytes().to_vec(), ))); principal.inner.name = new_name.clone(); batch.set( - ValueClass::Directory(DirectoryValue::NameToId(new_name.into_bytes())), + ValueClass::Directory(DirectoryClass::NameToId(new_name.into_bytes())), account_id.serialize(), ); } @@ -374,7 +374,7 @@ impl ManageDirectory for Store { } if !is_list { batch.set( - ValueClass::Directory(DirectoryValue::EmailToId( + ValueClass::Directory(DirectoryClass::EmailToId( email.as_bytes().to_vec(), )), vec![account_id].serialize(), @@ -385,7 +385,7 @@ impl ManageDirectory for Store { if !is_list { for email in &principal.inner.emails { if !emails.contains(email) { - batch.clear(ValueClass::Directory(DirectoryValue::EmailToId( + batch.clear(ValueClass::Directory(DirectoryClass::EmailToId( email.as_bytes().to_vec(), ))); } @@ -446,7 +446,7 @@ impl ManageDirectory for Store { } if !is_list { batch.set( - ValueClass::Directory(DirectoryValue::EmailToId( + ValueClass::Directory(DirectoryClass::EmailToId( email.as_bytes().to_vec(), )), vec![account_id].serialize(), @@ -482,7 +482,7 @@ impl ManageDirectory for Store { let email = email.to_lowercase(); if let Some(pos) = principal.inner.emails.iter().position(|v| *v == email) { if !is_list { - batch.clear(ValueClass::Directory(DirectoryValue::EmailToId( + batch.clear(ValueClass::Directory(DirectoryClass::EmailToId( email.as_bytes().to_vec(), ))); } @@ -498,14 +498,14 @@ impl ManageDirectory for Store { if has_list_changes { for email in &principal.inner.emails { batch.set( - ValueClass::Directory(DirectoryValue::EmailToId(email.as_bytes().to_vec())), + ValueClass::Directory(DirectoryClass::EmailToId(email.as_bytes().to_vec())), (&principal.inner.member_of).serialize(), ); } } batch.set( - ValueClass::Directory(DirectoryValue::Principal(account_id)), + ValueClass::Directory(DirectoryClass::Principal(account_id)), principal.inner.serialize(), ); @@ -522,7 +522,7 @@ impl ManageDirectory for Store { } let mut batch = BatchBuilder::new(); batch.set( - ValueClass::Directory(DirectoryValue::Domain(domain.to_lowercase().into_bytes())), + ValueClass::Directory(DirectoryClass::Domain(domain.to_lowercase().into_bytes())), vec![], ); self.write(batch.build()).await.map_err(Into::into) @@ -535,7 +535,7 @@ impl ManageDirectory for Store { ))); } let mut batch = BatchBuilder::new(); - batch.clear(ValueClass::Directory(DirectoryValue::Domain( + batch.clear(ValueClass::Directory(DirectoryClass::Domain( domain.to_lowercase().into_bytes(), ))); self.write(batch.build()).await.map_err(Into::into) @@ -597,10 +597,10 @@ impl ManageDirectory for Store { start_from: Option<&str>, limit: usize, ) -> crate::Result> { - let from_key = ValueKey::from(ValueClass::Directory(DirectoryValue::NameToId( + let from_key = ValueKey::from(ValueClass::Directory(DirectoryClass::NameToId( start_from.unwrap_or("").as_bytes().to_vec(), ))); - let to_key = ValueKey::from(ValueClass::Directory(DirectoryValue::NameToId(vec![ + let to_key = ValueKey::from(ValueClass::Directory(DirectoryClass::NameToId(vec![ u8::MAX; 10 ]))); diff --git a/crates/jmap-proto/src/types/blob.rs b/crates/jmap-proto/src/types/blob.rs index c34344c0..3f9a2f7b 100644 --- a/crates/jmap-proto/src/types/blob.rs +++ b/crates/jmap-proto/src/types/blob.rs @@ -123,7 +123,10 @@ impl BlobId { document_id: it.next_leb128()?, } } else { - BlobClass::Reserved { account_id } + BlobClass::Reserved { + account_id, + expires: it.next_leb128()?, + } }, section: if encoding != 0 { BlobSection { @@ -160,8 +163,12 @@ impl BlobId { let _ = writer.write(self.hash.as_ref()); match &self.class { - BlobClass::Reserved { account_id } => { + BlobClass::Reserved { + account_id, + expires, + } => { let _ = writer.write_leb128(*account_id); + let _ = writer.write_leb128(*expires); } BlobClass::Linked { account_id, diff --git a/crates/jmap/src/blob/copy.rs b/crates/jmap/src/blob/copy.rs index 66d93b8a..dc15974f 100644 --- a/crates/jmap/src/blob/copy.rs +++ b/crates/jmap/src/blob/copy.rs @@ -32,7 +32,7 @@ use jmap_proto::{ use store::{ write::{now, BatchBuilder, BlobOp}, - BlobClass, + BlobClass, Serialize, }; use utils::map::vec_map::VecMap; @@ -55,18 +55,21 @@ impl JMAP { for blob_id in request.blob_ids { if self.has_access_blob(&blob_id, access_token).await? { let mut batch = BatchBuilder::new(); - batch.with_account_id(account_id).blob( - blob_id.hash.clone(), + let until = now() + self.config.upload_tmp_ttl; + batch.with_account_id(account_id).set( BlobOp::Reserve { - until: now() + self.config.upload_tmp_ttl, - size: 0, + until, + hash: blob_id.hash.clone(), }, - 0, + 0u32.serialize(), ); self.write_batch(batch).await?; let dest_blob_id = BlobId { hash: blob_id.hash.clone(), - class: BlobClass::Reserved { account_id }, + class: BlobClass::Reserved { + account_id, + expires: until, + }, section: blob_id.section.clone(), }; diff --git a/crates/jmap/src/blob/download.rs b/crates/jmap/src/blob/download.rs index f27d2132..14ae820d 100644 --- a/crates/jmap/src/blob/download.rs +++ b/crates/jmap/src/blob/download.rs @@ -48,7 +48,7 @@ impl JMAP { ) -> Result>, MethodError> { if !self .store - .blob_hash_can_read(&blob_id.hash, &blob_id.class) + .blob_has_access(&blob_id.hash, &blob_id.class) .await .map_err(|err| { tracing::error!(event = "error", @@ -149,7 +149,7 @@ impl JMAP { ) -> Result { Ok(self .store - .blob_hash_can_read(&blob_id.hash, &blob_id.class) + .blob_has_access(&blob_id.hash, &blob_id.class) .await .map_err(|err| { tracing::error!(event = "error", @@ -184,7 +184,7 @@ impl JMAP { .await?) } } - BlobClass::Reserved { account_id } => access_token.is_member(*account_id), + BlobClass::Reserved { account_id, .. } => access_token.is_member(*account_id), }) } } diff --git a/crates/jmap/src/blob/get.rs b/crates/jmap/src/blob/get.rs index 06b72ae0..2ec97e0c 100644 --- a/crates/jmap/src/blob/get.rs +++ b/crates/jmap/src/blob/get.rs @@ -268,7 +268,8 @@ impl JMAP { } } } - BlobClass::Reserved { account_id } if *account_id == req_account_id => (), + BlobClass::Reserved { account_id, .. } if *account_id == req_account_id => { + } _ => { response.not_found.push(MaybeUnparsable::Value(id)); continue; diff --git a/crates/jmap/src/blob/upload.rs b/crates/jmap/src/blob/upload.rs index 4f55954a..086e5d9f 100644 --- a/crates/jmap/src/blob/upload.rs +++ b/crates/jmap/src/blob/upload.rs @@ -33,7 +33,7 @@ use jmap_proto::{ }; use store::{ write::{now, BatchBuilder, BlobOp}, - BlobClass, BlobHash, + BlobClass, BlobHash, Serialize, }; use crate::{auth::AccessToken, JMAP}; @@ -155,18 +155,14 @@ impl JMAP { } // Enforce quota - let used = self - .store - .blob_hash_quota(account_id) - .await - .map_err(|err| { - tracing::error!(event = "error", + let used = self.store.blob_quota(account_id).await.map_err(|err| { + tracing::error!(event = "error", context = "blob_store", account_id = account_id, error = ?err, "Failed to obtain blob quota"); - MethodError::ServerPartialFail - })?; + MethodError::ServerPartialFail + })?; if ((self.config.upload_tmp_quota_size > 0 && used.bytes + data.len() > self.config.upload_tmp_quota_size) @@ -219,7 +215,7 @@ impl JMAP { // Enforce quota let used = self .store - .blob_hash_quota(account_id.document_id()) + .blob_quota(account_id.document_id()) .await .map_err(|err| { tracing::error!(event = "error", @@ -271,18 +267,18 @@ impl JMAP { // First reserve the hash let hash = BlobHash::from(data); let mut batch = BatchBuilder::new(); + let until = now() + self.config.upload_tmp_ttl; - batch.with_account_id(account_id).blob( - hash.clone(), + batch.with_account_id(account_id).set( BlobOp::Reserve { - until: now() + self.config.upload_tmp_ttl, - size: if set_quota { data.len() } else { 0 }, + hash: hash.clone(), + until, }, - 0, + (if set_quota { data.len() as u32 } else { 0u32 }).serialize(), ); self.write_batch(batch).await?; - if !self.store.blob_hash_exists(&hash).await.map_err(|err| { + if !self.store.blob_exists(&hash).await.map_err(|err| { tracing::error!( event = "error", context = "put_blob", @@ -305,13 +301,16 @@ impl JMAP { // Commit blob let mut batch = BatchBuilder::new(); - batch.blob(hash.clone(), BlobOp::Commit, 0); + batch.set(BlobOp::Commit { hash: hash.clone() }, Vec::new()); self.write_batch(batch).await?; } Ok(BlobId { hash, - class: BlobClass::Reserved { account_id }, + class: BlobClass::Reserved { + account_id, + expires: until, + }, section: None, }) } diff --git a/crates/jmap/src/email/index.rs b/crates/jmap/src/email/index.rs index c43e7acf..7b2e1735 100644 --- a/crates/jmap/src/email/index.rs +++ b/crates/jmap/src/email/index.rs @@ -35,7 +35,7 @@ use store::{ backend::MAX_TOKEN_LENGTH, fts::{index::FtsDocument, Field}, write::{ - BatchBuilder, BlobOp, DirectoryValue, IntoOperations, F_BITMAP, F_CLEAR, F_INDEX, F_VALUE, + BatchBuilder, BlobOp, DirectoryClass, IntoOperations, F_BITMAP, F_CLEAR, F_INDEX, F_VALUE, }, BlobHash, }; @@ -92,7 +92,7 @@ impl IndexMessage for BatchBuilder { let account_id = self.last_account_id().unwrap(); self.value(Property::Size, message.raw_message.len() as u32, F_INDEX) .add( - DirectoryValue::UsedQuota(account_id), + DirectoryClass::UsedQuota(account_id), message.raw_message.len() as i64, ); @@ -152,7 +152,12 @@ impl IndexMessage for BatchBuilder { } // Link blob - self.blob(blob_hash.clone(), BlobOp::Link, 0); + self.set( + BlobOp::Link { + hash: blob_hash.clone(), + }, + Vec::new(), + ); // Store message metadata self.value( @@ -415,7 +420,7 @@ impl<'x> IntoOperations for EmailIndexBuilder<'x> { batch .value(Property::Size, metadata.size as u32, F_INDEX | options) .add( - DirectoryValue::UsedQuota(account_id), + DirectoryClass::UsedQuota(account_id), if self.set { metadata.size as i64 } else { @@ -435,7 +440,18 @@ impl<'x> IntoOperations for EmailIndexBuilder<'x> { batch.index_headers(&metadata.contents.parts[0].headers, options); // Link blob - batch.blob(metadata.blob_hash.clone(), BlobOp::Link, options); + if self.set { + batch.set( + BlobOp::Link { + hash: metadata.blob_hash.clone(), + }, + Vec::new(), + ); + } else { + batch.clear(BlobOp::Link { + hash: metadata.blob_hash.clone(), + }); + } } } diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index ebfd006d..e09fca66 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -52,7 +52,7 @@ use store::{ parking_lot::Mutex, query::{sort::Pagination, Comparator, Filter, ResultSet, SortedResultSet}, roaring::RoaringBitmap, - write::{BatchBuilder, BitmapClass, DirectoryValue, TagValue, ToBitmaps, ValueClass}, + write::{BatchBuilder, BitmapClass, DirectoryClass, TagValue, ToBitmaps, ValueClass}, BitmapKey, BlobStore, Deserialize, FtsStore, Serialize, Store, Stores, ValueKey, }; use tokio::sync::mpsc; @@ -617,7 +617,7 @@ impl JMAP { pub async fn get_used_quota(&self, account_id: u32) -> Result { self.store - .get_counter(DirectoryValue::UsedQuota(account_id)) + .get_counter(DirectoryClass::UsedQuota(account_id)) .await .map_err(|err| { tracing::error!( diff --git a/crates/jmap/src/sieve/get.rs b/crates/jmap/src/sieve/get.rs index 3b16acde..c6cc7466 100644 --- a/crates/jmap/src/sieve/get.rs +++ b/crates/jmap/src/sieve/get.rs @@ -32,7 +32,7 @@ use jmap_proto::{ use sieve::Sieve; use store::{ query::Filter, - write::{assert::HashedValue, BatchBuilder, BlobOp, F_CLEAR}, + write::{assert::HashedValue, BatchBuilder, BlobOp}, Deserialize, Serialize, }; @@ -276,8 +276,15 @@ impl JMAP { .update_document(document_id) .assert_value(Property::Value, &script_object) .set(Property::Value, (&new_script_object).serialize()) - .blob(blob_id.hash.clone(), BlobOp::Link, F_CLEAR) - .blob(new_blob_id.hash, BlobOp::Link, 0); + .clear(BlobOp::Link { + hash: blob_id.hash.clone(), + }) + .set( + BlobOp::Link { + hash: new_blob_id.hash, + }, + Vec::new(), + ); self.write_batch(batch).await?; Ok((sieve.inner, new_script_object)) diff --git a/crates/jmap/src/sieve/set.rs b/crates/jmap/src/sieve/set.rs index f27818ea..a76ee122 100644 --- a/crates/jmap/src/sieve/set.rs +++ b/crates/jmap/src/sieve/set.rs @@ -47,7 +47,7 @@ use store::{ query::Filter, rand::{distributions::Alphanumeric, thread_rng, Rng}, write::{ - assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, BlobOp, DirectoryValue, F_CLEAR, + assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, BlobOp, DirectoryClass, F_CLEAR, F_VALUE, }, BlobClass, @@ -122,8 +122,13 @@ impl JMAP { .with_account_id(account_id) .with_collection(Collection::SieveScript) .create_document(document_id) - .add(DirectoryValue::UsedQuota(account_id), script_size as i64) - .blob(blob_id.hash.clone(), BlobOp::Link, 0) + .add(DirectoryClass::UsedQuota(account_id), script_size as i64) + .set( + BlobOp::Link { + hash: blob_id.hash.clone(), + }, + Vec::new(), + ) .custom(builder); sieve_ids.insert(document_id); self.write_batch(batch).await?; @@ -218,15 +223,20 @@ impl JMAP { std::cmp::Ordering::Equal => 0, }; if update_quota != 0 { - batch.add(DirectoryValue::UsedQuota(account_id), update_quota); + batch.add(DirectoryClass::UsedQuota(account_id), update_quota); } // Update blobId - batch.blob(prev_blob_id.hash, BlobOp::Link, F_CLEAR).blob( - blob_id.hash.clone(), - BlobOp::Link, - 0, - ); + batch + .clear(BlobOp::Link { + hash: prev_blob_id.hash, + }) + .set( + BlobOp::Link { + hash: blob_id.hash.clone(), + }, + Vec::new(), + ); blob_id.into() } else { @@ -394,9 +404,11 @@ impl JMAP { .with_collection(Collection::SieveScript) .delete_document(document_id) .value(Property::EmailIds, (), F_VALUE | F_CLEAR) - .blob(blob_id.hash.clone(), BlobOp::Link, F_CLEAR) + .clear(BlobOp::Link { + hash: blob_id.hash.clone(), + }) .add( - DirectoryValue::UsedQuota(account_id), + DirectoryClass::UsedQuota(account_id), -(blob_id.section.as_ref().unwrap().size as i64), ) .custom(ObjectIndexBuilder::new(SCHEMA).with_current(obj)); diff --git a/crates/jmap/src/vacation/set.rs b/crates/jmap/src/vacation/set.rs index 0ece164a..5ec1c8c4 100644 --- a/crates/jmap/src/vacation/set.rs +++ b/crates/jmap/src/vacation/set.rs @@ -43,7 +43,7 @@ use mail_builder::MessageBuilder; use mail_parser::decoders::html::html_to_text; use store::{ write::{ - assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, BlobOp, DirectoryValue, F_CLEAR, + assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, BlobOp, DirectoryClass, F_CLEAR, F_VALUE, }, BlobClass, @@ -262,7 +262,12 @@ impl JMAP { }; // Link blob - batch.blob(blob_id.hash.clone(), BlobOp::Link, 0); + batch.set( + BlobOp::Link { + hash: blob_id.hash.clone(), + }, + Vec::new(), + ); let script_size = blob_id.section.as_ref().unwrap().size as i64; @@ -279,7 +284,9 @@ impl JMAP { })?; // Unlink previous blob - batch.blob(current_blob_id.hash.clone(), BlobOp::Link, F_CLEAR); + batch.clear(BlobOp::Link { + hash: current_blob_id.hash.clone(), + }); // Update quota let current_script_size = current_blob_id.section.as_ref().unwrap().size as i64; @@ -289,10 +296,10 @@ impl JMAP { std::cmp::Ordering::Equal => 0, }; if quota != 0 { - batch.add(DirectoryValue::UsedQuota(account_id), quota); + batch.add(DirectoryClass::UsedQuota(account_id), quota); } } else { - batch.add(DirectoryValue::UsedQuota(account_id), script_size); + batch.add(DirectoryClass::UsedQuota(account_id), script_size); } }; diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index 95f17b52..465f5ef4 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", "redis"] -default = ["sqlite", "postgres", "mysql", "redis"] +default = ["sqlite", "postgres", "mysql", "redis", "rocks", "foundationdb"] sqlite = ["store/sqlite"] foundationdb = ["store/foundation"] postgres = ["store/postgres"] diff --git a/crates/managesieve/src/op/putscript.rs b/crates/managesieve/src/op/putscript.rs index 9ecac091..94aa6e8c 100644 --- a/crates/managesieve/src/op/putscript.rs +++ b/crates/managesieve/src/op/putscript.rs @@ -30,7 +30,7 @@ use jmap_proto::{ use sieve::compiler::ErrorType; use store::{ query::Filter, - write::{assert::HashedValue, BatchBuilder, BlobOp, DirectoryValue, F_CLEAR}, + write::{assert::HashedValue, BatchBuilder, BlobOp, DirectoryClass}, BlobClass, }; use tokio::io::{AsyncRead, AsyncWrite}; @@ -128,8 +128,15 @@ impl Session { .with_account_id(account_id) .with_collection(Collection::SieveScript) .update_document(document_id) - .blob(prev_blob_id.hash.clone(), BlobOp::Link, F_CLEAR) - .blob(blob_id.hash.clone(), BlobOp::Link, 0); + .clear(BlobOp::Link { + hash: prev_blob_id.hash.clone(), + }) + .set( + BlobOp::Link { + hash: blob_id.hash.clone(), + }, + Vec::new(), + ); // Update quota let prev_script_size = prev_blob_id.section.as_ref().unwrap().size as i64; @@ -139,7 +146,7 @@ impl Session { std::cmp::Ordering::Equal => 0, }; if update_quota != 0 { - batch.add(DirectoryValue::UsedQuota(account_id), update_quota); + batch.add(DirectoryClass::UsedQuota(account_id), update_quota); } batch.custom( @@ -180,8 +187,13 @@ impl Session { .with_account_id(account_id) .with_collection(Collection::SieveScript) .create_document(document_id) - .add(DirectoryValue::UsedQuota(account_id), script_size) - .blob(blob_id.hash.clone(), BlobOp::Link, 0) + .add(DirectoryClass::UsedQuota(account_id), script_size) + .set( + BlobOp::Link { + hash: blob_id.hash.clone(), + }, + Vec::new(), + ) .custom( ObjectIndexBuilder::new(SCHEMA).with_changes( Object::with_capacity(3) diff --git a/crates/store/src/backend/foundationdb/blob.rs b/crates/store/src/backend/foundationdb/blob.rs index 9c27d156..390575a7 100644 --- a/crates/store/src/backend/foundationdb/blob.rs +++ b/crates/store/src/backend/foundationdb/blob.rs @@ -26,7 +26,7 @@ use std::ops::Range; use foundationdb::{options::StreamingMode, FdbError, KeySelector, RangeOption}; use futures::StreamExt; -use crate::{write::key::KeySerializer, Error, BLOB_HASH_LEN, SUBSPACE_BLOB_DATA}; +use crate::{write::key::KeySerializer, Error, BLOB_HASH_LEN, SUBSPACE_BLOBS}; use super::{FdbStore, MAX_VALUE_SIZE}; @@ -41,12 +41,12 @@ impl FdbStore { let block_end = (range.end as usize / MAX_VALUE_SIZE) + 1; let begin = KeySerializer::new(key.len() + 3) - .write(SUBSPACE_BLOB_DATA) + .write(SUBSPACE_BLOBS) .write(key) .write(block_start as u16) .finalize(); let end = KeySerializer::new(key.len() + 3) - .write(SUBSPACE_BLOB_DATA) + .write(SUBSPACE_BLOBS) .write(key) .write(block_end as u16) .finalize(); @@ -129,7 +129,7 @@ impl FdbStore { for (chunk_pos, chunk_bytes) in data.chunks(MAX_VALUE_SIZE).enumerate() { trx.set( &KeySerializer::new(key.len() + 3) - .write(SUBSPACE_BLOB_DATA) + .write(SUBSPACE_BLOBS) .write(key) .write(chunk_pos as u16) .finalize(), @@ -158,12 +158,12 @@ impl FdbStore { let trx = self.db.create_trx()?; trx.clear_range( &KeySerializer::new(key.len() + 3) - .write(SUBSPACE_BLOB_DATA) + .write(SUBSPACE_BLOBS) .write(key) .write(0u16) .finalize(), &KeySerializer::new(key.len() + 3) - .write(SUBSPACE_BLOB_DATA) + .write(SUBSPACE_BLOBS) .write(key) .write(u16::MAX) .finalize(), diff --git a/crates/store/src/backend/foundationdb/write.rs b/crates/store/src/backend/foundationdb/write.rs index 8fd41f40..1de4d4d7 100644 --- a/crates/store/src/backend/foundationdb/write.rs +++ b/crates/store/src/backend/foundationdb/write.rs @@ -40,7 +40,7 @@ use crate::{ key::KeySerializer, Batch, BitmapClass, Operation, ValueClass, ValueOp, MAX_COMMIT_ATTEMPTS, MAX_COMMIT_TIME, }, - BitmapKey, BlobKey, IndexKey, Key, LogKey, ValueKey, SUBSPACE_BITMAPS, SUBSPACE_VALUES, + BitmapKey, IndexKey, Key, LogKey, ValueKey, SUBSPACE_BITMAPS, SUBSPACE_VALUES, }; use super::{ @@ -235,22 +235,6 @@ impl FdbStore { .push(BitmapOp::new(document_id, *set)); } } - Operation::Blob { hash, op, set } => { - let key = BlobKey { - account_id, - collection, - document_id, - hash, - op: *op, - } - .serialize(true); - - if *set { - trx.set(&key, &[]); - } else { - trx.clear(&key); - } - } Operation::Log { collection, change_id, diff --git a/crates/store/src/backend/mysql/main.rs b/crates/store/src/backend/mysql/main.rs index e12ca6f6..f1001afc 100644 --- a/crates/store/src/backend/mysql/main.rs +++ b/crates/store/src/backend/mysql/main.rs @@ -25,8 +25,8 @@ use mysql_async::{prelude::Queryable, OptsBuilder, Pool, PoolConstraints, PoolOp use utils::config::utils::AsKey; use crate::{ - SUBSPACE_BITMAPS, SUBSPACE_BLOBS, SUBSPACE_BLOB_DATA, SUBSPACE_COUNTERS, SUBSPACE_INDEXES, - SUBSPACE_LOGS, SUBSPACE_VALUES, + SUBSPACE_BITMAPS, SUBSPACE_BLOBS, SUBSPACE_COUNTERS, SUBSPACE_INDEXES, SUBSPACE_LOGS, + SUBSPACE_VALUES, }; use super::MysqlStore; @@ -98,7 +98,7 @@ impl MysqlStore { v LONGBLOB NOT NULL, PRIMARY KEY (k(255)) ) ENGINE=InnoDB", - char::from(SUBSPACE_BLOB_DATA), + char::from(SUBSPACE_BLOBS), )) .await?; @@ -113,17 +113,6 @@ impl MysqlStore { .await?; } - for table in [SUBSPACE_BLOBS] { - let table = char::from(table); - conn.query_drop(&format!( - "CREATE TABLE IF NOT EXISTS {table} ( - k TINYBLOB, - PRIMARY KEY (k(255)) - ) ENGINE=InnoDB" - )) - .await?; - } - conn.query_drop(&format!( "CREATE TABLE IF NOT EXISTS {} ( k TINYBLOB, diff --git a/crates/store/src/backend/mysql/write.rs b/crates/store/src/backend/mysql/write.rs index f8bf8a58..461e3178 100644 --- a/crates/store/src/backend/mysql/write.rs +++ b/crates/store/src/backend/mysql/write.rs @@ -31,7 +31,7 @@ use crate::{ write::{ Batch, BitmapClass, Operation, ValueClass, ValueOp, MAX_COMMIT_ATTEMPTS, MAX_COMMIT_TIME, }, - BitmapKey, BlobKey, IndexKey, Key, LogKey, ValueKey, + BitmapKey, IndexKey, Key, LogKey, ValueKey, }; use super::MysqlStore; @@ -219,23 +219,6 @@ impl MysqlStore { }; trx.exec_drop(&s, (key,)).await?; } - Operation::Blob { hash, op, set } => { - let key = BlobKey { - account_id, - collection, - document_id, - hash, - op: *op, - } - .serialize(false); - - let s = if *set { - trx.prep("INSERT IGNORE INTO o (k) VALUES (?)").await? - } else { - trx.prep("DELETE FROM o WHERE k = ?").await? - }; - trx.exec_drop(&s, (key,)).await?; - } Operation::Log { collection, change_id, diff --git a/crates/store/src/backend/postgres/main.rs b/crates/store/src/backend/postgres/main.rs index 91abcba9..3757f4c4 100644 --- a/crates/store/src/backend/postgres/main.rs +++ b/crates/store/src/backend/postgres/main.rs @@ -22,8 +22,8 @@ */ use crate::{ - backend::postgres::tls::MakeRustlsConnect, SUBSPACE_BITMAPS, SUBSPACE_BLOBS, - SUBSPACE_BLOB_DATA, SUBSPACE_COUNTERS, SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_VALUES, + backend::postgres::tls::MakeRustlsConnect, SUBSPACE_BITMAPS, SUBSPACE_BLOBS, SUBSPACE_COUNTERS, + SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_VALUES, }; use super::PostgresStore; @@ -74,7 +74,7 @@ impl PostgresStore { pub(super) async fn create_tables(&self) -> crate::Result<()> { let conn = self.conn_pool.get().await?; - for table in [SUBSPACE_VALUES, SUBSPACE_LOGS, SUBSPACE_BLOB_DATA] { + for table in [SUBSPACE_VALUES, SUBSPACE_LOGS, SUBSPACE_BLOBS] { let table = char::from(table); conn.execute( &format!( @@ -88,7 +88,7 @@ impl PostgresStore { .await?; } - for table in [SUBSPACE_INDEXES, SUBSPACE_BITMAPS, SUBSPACE_BLOBS] { + for table in [SUBSPACE_INDEXES, SUBSPACE_BITMAPS] { let table = char::from(table); conn.execute( &format!( diff --git a/crates/store/src/backend/postgres/write.rs b/crates/store/src/backend/postgres/write.rs index 6948797f..e7eb08ad 100644 --- a/crates/store/src/backend/postgres/write.rs +++ b/crates/store/src/backend/postgres/write.rs @@ -32,7 +32,7 @@ use crate::{ write::{ Batch, BitmapClass, Operation, ValueClass, ValueOp, MAX_COMMIT_ATTEMPTS, MAX_COMMIT_TIME, }, - BitmapKey, BlobKey, IndexKey, Key, LogKey, ValueKey, + BitmapKey, IndexKey, Key, LogKey, ValueKey, }; use super::PostgresStore; @@ -229,26 +229,6 @@ impl PostgresStore { }; trx.execute(&s, &[&key]).await?; } - Operation::Blob { hash, op, set } => { - let key = BlobKey { - account_id, - collection, - document_id, - hash, - op: *op, - } - .serialize(false); - - let s = if *set { - trx.prepare_cached( - "INSERT INTO o (k) VALUES ($1) ON CONFLICT (k) DO NOTHING", - ) - .await? - } else { - trx.prepare_cached("DELETE FROM o WHERE k = $1").await? - }; - trx.execute(&s, &[&key]).await?; - } Operation::Log { collection, change_id, diff --git a/crates/store/src/backend/rocksdb/blob.rs b/crates/store/src/backend/rocksdb/blob.rs index 5022d2ba..d8f34fdf 100644 --- a/crates/store/src/backend/rocksdb/blob.rs +++ b/crates/store/src/backend/rocksdb/blob.rs @@ -23,7 +23,7 @@ use std::ops::Range; -use super::{RocksDbStore, CF_BLOB_DATA}; +use super::{RocksDbStore, CF_BLOBS}; impl RocksDbStore { pub(crate) async fn get_blob( @@ -33,7 +33,7 @@ impl RocksDbStore { ) -> crate::Result>> { let db = self.db.clone(); self.spawn_worker(move || { - db.get_pinned_cf(&db.cf_handle(CF_BLOB_DATA).unwrap(), key) + db.get_pinned_cf(&db.cf_handle(CF_BLOBS).unwrap(), key) .map(|obj| { obj.map(|bytes| { if range.start == 0 && range.end == u32::MAX { @@ -57,7 +57,7 @@ impl RocksDbStore { pub(crate) async fn put_blob(&self, key: &[u8], data: &[u8]) -> crate::Result<()> { let db = self.db.clone(); self.spawn_worker(move || { - db.put_cf(&db.cf_handle(CF_BLOB_DATA).unwrap(), key, data) + db.put_cf(&db.cf_handle(CF_BLOBS).unwrap(), key, data) .map_err(|e| crate::Error::InternalError(format!("Failed to insert blob: {}", e))) }) .await @@ -66,7 +66,7 @@ impl RocksDbStore { pub(crate) async fn delete_blob(&self, key: &[u8]) -> crate::Result { let db = self.db.clone(); self.spawn_worker(move || { - db.delete_cf(&db.cf_handle(CF_BLOB_DATA).unwrap(), key) + db.delete_cf(&db.cf_handle(CF_BLOBS).unwrap(), key) .map_err(|e| crate::Error::InternalError(format!("Failed to delete blob: {}", e))) .map(|_| true) }) diff --git a/crates/store/src/backend/rocksdb/main.rs b/crates/store/src/backend/rocksdb/main.rs index 75836006..fb6eb001 100644 --- a/crates/store/src/backend/rocksdb/main.rs +++ b/crates/store/src/backend/rocksdb/main.rs @@ -37,9 +37,7 @@ use utils::{ use crate::{Deserialize, Error}; -use super::{ - RocksDbStore, CF_BITMAPS, CF_BLOBS, CF_BLOB_DATA, CF_COUNTERS, CF_INDEXES, CF_LOGS, CF_VALUES, -}; +use super::{RocksDbStore, CF_BITMAPS, CF_BLOBS, CF_COUNTERS, CF_INDEXES, CF_LOGS, CF_VALUES}; impl RocksDbStore { pub async fn open(config: &Config, prefix: impl AsKey) -> crate::Result { @@ -76,10 +74,10 @@ impl RocksDbStore { let mut cf_opts = Options::default(); cf_opts.set_enable_blob_files(true); cf_opts.set_min_blob_size(config.property_or_static((&prefix, "min-blob-size"), "16834")?); - cfs.push(ColumnFamilyDescriptor::new(CF_BLOB_DATA, cf_opts)); + cfs.push(ColumnFamilyDescriptor::new(CF_BLOBS, cf_opts)); // Other cfs - for cf in [CF_BLOBS, CF_INDEXES, CF_LOGS, CF_VALUES] { + for cf in [CF_INDEXES, CF_LOGS, CF_VALUES] { let cf_opts = Options::default(); cfs.push(ColumnFamilyDescriptor::new(cf, cf_opts)); } diff --git a/crates/store/src/backend/rocksdb/mod.rs b/crates/store/src/backend/rocksdb/mod.rs index 9dd3cc5e..d6cc5082 100644 --- a/crates/store/src/backend/rocksdb/mod.rs +++ b/crates/store/src/backend/rocksdb/mod.rs @@ -26,8 +26,8 @@ use std::sync::Arc; use rocksdb::{MultiThreaded, OptimisticTransactionDB}; use crate::{ - SUBSPACE_BITMAPS, SUBSPACE_BLOBS, SUBSPACE_BLOB_DATA, SUBSPACE_COUNTERS, SUBSPACE_INDEXES, - SUBSPACE_LOGS, SUBSPACE_VALUES, + SUBSPACE_BITMAPS, SUBSPACE_BLOBS, SUBSPACE_COUNTERS, SUBSPACE_INDEXES, SUBSPACE_LOGS, + SUBSPACE_VALUES, }; pub mod bitmap; @@ -41,7 +41,6 @@ static CF_VALUES: &str = unsafe { std::str::from_utf8_unchecked(&[SUBSPACE_VALUE static CF_LOGS: &str = unsafe { std::str::from_utf8_unchecked(&[SUBSPACE_LOGS]) }; static CF_INDEXES: &str = unsafe { std::str::from_utf8_unchecked(&[SUBSPACE_INDEXES]) }; static CF_BLOBS: &str = unsafe { std::str::from_utf8_unchecked(&[SUBSPACE_BLOBS]) }; -static CF_BLOB_DATA: &str = unsafe { std::str::from_utf8_unchecked(&[SUBSPACE_BLOB_DATA]) }; static CF_COUNTERS: &str = unsafe { std::str::from_utf8_unchecked(&[SUBSPACE_COUNTERS]) }; impl From for crate::Error { diff --git a/crates/store/src/backend/rocksdb/write.rs b/crates/store/src/backend/rocksdb/write.rs index fd003bf8..a58ec225 100644 --- a/crates/store/src/backend/rocksdb/write.rs +++ b/crates/store/src/backend/rocksdb/write.rs @@ -31,11 +31,11 @@ use rocksdb::{Direction, ErrorKind, IteratorMode}; use super::{ bitmap::{clear_bit, set_bit}, - RocksDbStore, CF_BITMAPS, CF_BLOBS, CF_COUNTERS, CF_INDEXES, CF_LOGS, CF_VALUES, + RocksDbStore, CF_BITMAPS, CF_COUNTERS, CF_INDEXES, CF_LOGS, CF_VALUES, }; use crate::{ write::{Batch, Operation, ValueOp, MAX_COMMIT_ATTEMPTS, MAX_COMMIT_TIME}, - BitmapKey, BlobKey, IndexKey, Key, LogKey, ValueKey, SUBSPACE_VALUES, + BitmapKey, IndexKey, Key, LogKey, ValueKey, }; impl RocksDbStore { @@ -50,7 +50,6 @@ impl RocksDbStore { let cf_values = db.cf_handle(CF_VALUES).unwrap(); let cf_indexes = db.cf_handle(CF_INDEXES).unwrap(); let cf_logs = db.cf_handle(CF_LOGS).unwrap(); - let cf_blobs = db.cf_handle(CF_BLOBS).unwrap(); let cf_counters = db.cf_handle(CF_COUNTERS).unwrap(); loop { @@ -140,22 +139,6 @@ impl RocksDbStore { wb.merge_cf(&cf_bitmaps, key, value); } - Operation::Blob { hash, op, set } => { - let key = BlobKey { - account_id, - collection, - document_id, - hash, - op: *op, - } - .serialize(false); - - if *set { - wb.put_cf(&cf_blobs, &key, []); - } else { - wb.delete_cf(&cf_blobs, &key); - } - } Operation::Log { collection, change_id, diff --git a/crates/store/src/backend/sqlite/main.rs b/crates/store/src/backend/sqlite/main.rs index 1a48cd4d..5424479c 100644 --- a/crates/store/src/backend/sqlite/main.rs +++ b/crates/store/src/backend/sqlite/main.rs @@ -29,8 +29,8 @@ use utils::{ }; use crate::{ - SUBSPACE_BITMAPS, SUBSPACE_BLOBS, SUBSPACE_BLOB_DATA, SUBSPACE_COUNTERS, SUBSPACE_INDEXES, - SUBSPACE_LOGS, SUBSPACE_VALUES, + SUBSPACE_BITMAPS, SUBSPACE_BLOBS, SUBSPACE_COUNTERS, SUBSPACE_INDEXES, SUBSPACE_LOGS, + SUBSPACE_VALUES, }; use super::{pool::SqliteConnectionManager, SqliteStore}; @@ -75,7 +75,7 @@ impl SqliteStore { pub(super) fn create_tables(&self) -> crate::Result<()> { let conn = self.conn_pool.get()?; - for table in [SUBSPACE_VALUES, SUBSPACE_LOGS, SUBSPACE_BLOB_DATA] { + for table in [SUBSPACE_VALUES, SUBSPACE_LOGS, SUBSPACE_BLOBS] { let table = char::from(table); conn.execute( &format!( @@ -88,7 +88,7 @@ impl SqliteStore { )?; } - for table in [SUBSPACE_INDEXES, SUBSPACE_BLOBS, SUBSPACE_BITMAPS] { + for table in [SUBSPACE_INDEXES, SUBSPACE_BITMAPS] { let table = char::from(table); conn.execute( &format!( diff --git a/crates/store/src/backend/sqlite/write.rs b/crates/store/src/backend/sqlite/write.rs index e4904726..e4acf8de 100644 --- a/crates/store/src/backend/sqlite/write.rs +++ b/crates/store/src/backend/sqlite/write.rs @@ -25,7 +25,7 @@ use rusqlite::{params, OptionalExtension, TransactionBehavior}; use crate::{ write::{Batch, Operation, ValueOp}, - BitmapKey, BlobKey, IndexKey, Key, LogKey, ValueKey, + BitmapKey, IndexKey, Key, LogKey, ValueKey, }; use super::SqliteStore; @@ -135,24 +135,6 @@ impl SqliteStore { .execute(params![&key])?; }; } - Operation::Blob { hash, op, set } => { - let key = BlobKey { - account_id, - collection, - document_id, - hash, - op: *op, - } - .serialize(false); - - if *set { - trx.prepare_cached("INSERT OR IGNORE INTO o (k) VALUES (?)")? - .execute([&key])?; - } else { - trx.prepare_cached("DELETE FROM o WHERE k = ?")? - .execute([&key])?; - } - } Operation::Log { collection, change_id, diff --git a/crates/store/src/dispatch/store.rs b/crates/store/src/dispatch/store.rs index 02801437..7487b339 100644 --- a/crates/store/src/dispatch/store.rs +++ b/crates/store/src/dispatch/store.rs @@ -272,16 +272,15 @@ impl Store { #[cfg(feature = "test_mode")] pub async fn destroy(&self) { - use crate::{SUBSPACE_BLOBS, SUBSPACE_BLOB_DATA, SUBSPACE_COUNTERS}; + use crate::{SUBSPACE_BLOBS, SUBSPACE_COUNTERS}; for subspace in [ SUBSPACE_VALUES, SUBSPACE_LOGS, SUBSPACE_BITMAPS, SUBSPACE_INDEXES, - SUBSPACE_BLOBS, SUBSPACE_COUNTERS, - SUBSPACE_BLOB_DATA, + SUBSPACE_BLOBS, ] { self.delete_range( AnyKey { @@ -309,24 +308,28 @@ impl Store { #[cfg(feature = "test_mode")] pub async fn blob_hash_expire_all(&self) { use crate::{ - write::{key::DeserializeBigEndian, BatchBuilder, BlobOp, F_CLEAR}, - BlobHash, BlobKey, BLOB_HASH_LEN, U64_LEN, + write::{key::DeserializeBigEndian, BatchBuilder, BlobOp, Operation, ValueOp}, + BlobHash, BLOB_HASH_LEN, U64_LEN, }; // Delete all temporary hashes - let from_key = BlobKey { + let from_key = ValueKey { account_id: 0, collection: 0, document_id: 0, - op: BlobOp::Reserve { until: 0, size: 0 }, - hash: BlobHash::default(), + class: ValueClass::Blob(BlobOp::Reserve { + hash: BlobHash::default(), + until: 0, + }), }; - let to_key = BlobKey { + let to_key = ValueKey { account_id: u32::MAX, collection: 0, document_id: 0, - op: BlobOp::Reserve { until: 0, size: 0 }, - hash: BlobHash::default(), + class: ValueClass::Blob(BlobOp::Reserve { + hash: BlobHash::default(), + until: 0, + }), }; let mut batch = BatchBuilder::new(); let mut last_account_id = u32::MAX; @@ -339,17 +342,16 @@ impl Store { batch.with_account_id(account_id); } - batch.blob( - BlobHash::try_from_hash_slice( - key.get(1 + U32_LEN..1 + U32_LEN + BLOB_HASH_LEN).unwrap(), - ) - .unwrap(), - BlobOp::Reserve { - until: key.deserialize_be_u64(key.len() - (U64_LEN + U32_LEN))?, - size: key.deserialize_be_u32(key.len() - U32_LEN)? as usize, - }, - F_CLEAR, - ); + batch.ops.push(Operation::Value { + class: ValueClass::Blob(BlobOp::Reserve { + hash: BlobHash::try_from_hash_slice( + key.get(1 + U32_LEN..1 + U32_LEN + BLOB_HASH_LEN).unwrap(), + ) + .unwrap(), + until: key.deserialize_be_u64(key.len() - U64_LEN)?, + }), + op: ValueOp::Clear, + }); Ok(true) }, @@ -363,7 +365,7 @@ impl Store { #[allow(unused_variables)] pub async fn assert_is_empty(&self, blob_store: crate::BlobStore) { - use crate::{SUBSPACE_BLOBS, SUBSPACE_BLOB_DATA, SUBSPACE_COUNTERS}; + use crate::{SUBSPACE_BLOBS, SUBSPACE_COUNTERS}; self.blob_hash_expire_all().await; self.purge_blobs(blob_store).await.unwrap(); @@ -375,10 +377,9 @@ impl Store { for (subspace, with_values) in [ (SUBSPACE_VALUES, true), (SUBSPACE_COUNTERS, false), - (SUBSPACE_BLOB_DATA, true), + (SUBSPACE_BLOBS, true), (SUBSPACE_BITMAPS, false), (SUBSPACE_INDEXES, false), - (SUBSPACE_BLOBS, false), ] { let from_key = crate::write::AnyKey { subspace, diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 9e747c6a..0cc31dbd 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -37,7 +37,7 @@ pub use blake3; pub use parking_lot; pub use rand; pub use roaring; -use write::{BitmapClass, BlobOp, ValueClass}; +use write::{BitmapClass, ValueClass}; #[cfg(feature = "s3")] use backend::s3::S3Store; @@ -108,15 +108,6 @@ pub struct ValueKey> { pub class: T, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct BlobKey> { - pub account_id: u32, - pub collection: u8, - pub document_id: u32, - pub hash: T, - pub op: BlobOp, -} - #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct LogKey { pub account_id: u32, @@ -137,6 +128,7 @@ pub type Result = std::result::Result; pub enum BlobClass { Reserved { account_id: u32, + expires: u64, }, Linked { account_id: u32, @@ -147,7 +139,10 @@ pub enum BlobClass { impl Default for BlobClass { fn default() -> Self { - BlobClass::Reserved { account_id: 0 } + BlobClass::Reserved { + account_id: 0, + expires: 0, + } } } @@ -178,8 +173,7 @@ pub const SUBSPACE_BITMAPS: u8 = b'b'; pub const SUBSPACE_VALUES: u8 = b'v'; pub const SUBSPACE_LOGS: u8 = b'l'; pub const SUBSPACE_INDEXES: u8 = b'i'; -pub const SUBSPACE_BLOBS: u8 = b'o'; -pub const SUBSPACE_BLOB_DATA: u8 = b't'; +pub const SUBSPACE_BLOBS: u8 = b't'; pub const SUBSPACE_COUNTERS: u8 = b'c'; pub struct IterateParams { diff --git a/crates/store/src/write/batch.rs b/crates/store/src/write/batch.rs index cfa78e43..720e1ab6 100644 --- a/crates/store/src/write/batch.rs +++ b/crates/store/src/write/batch.rs @@ -21,12 +21,9 @@ * for more details. */ -use crate::BlobHash; - use super::{ - assert::ToAssertValue, Batch, BatchBuilder, BitmapClass, BlobOp, HasFlag, IntoOperations, - Operation, Serialize, TagValue, ToBitmaps, ValueClass, ValueOp, F_BITMAP, F_CLEAR, F_INDEX, - F_VALUE, + assert::ToAssertValue, Batch, BatchBuilder, BitmapClass, HasFlag, IntoOperations, Operation, + Serialize, TagValue, ToBitmaps, ValueClass, ValueOp, F_BITMAP, F_CLEAR, F_INDEX, F_VALUE, }; impl BatchBuilder { @@ -145,15 +142,6 @@ impl BatchBuilder { self } - pub fn blob(&mut self, hash: BlobHash, op: BlobOp, options: u32) -> &mut Self { - self.ops.push(Operation::Blob { - hash, - op, - set: !options.has_flag(F_CLEAR), - }); - self - } - pub fn add(&mut self, class: impl Into, value: i64) -> &mut Self { self.ops.push(Operation::Value { class: class.into(), diff --git a/crates/store/src/write/blob.rs b/crates/store/src/write/blob.rs index b5c67c80..9e385728 100644 --- a/crates/store/src/write/blob.rs +++ b/crates/store/src/write/blob.rs @@ -24,11 +24,11 @@ use ahash::AHashSet; use crate::{ - write::{BatchBuilder, F_CLEAR}, - BlobClass, BlobHash, BlobKey, BlobStore, IterateParams, Store, BLOB_HASH_LEN, U32_LEN, U64_LEN, + write::BatchBuilder, BlobClass, BlobHash, BlobStore, Deserialize, IterateParams, Store, + ValueKey, BLOB_HASH_LEN, U32_LEN, U64_LEN, }; -use super::{key::DeserializeBigEndian, now, BlobOp}; +use super::{key::DeserializeBigEndian, now, BlobOp, Operation, ValueClass, ValueOp}; #[derive(Debug, PartialEq, Eq)] pub struct BlobQuota { @@ -37,69 +37,53 @@ pub struct BlobQuota { } impl Store { - pub async fn blob_hash_exists( + pub async fn blob_exists( &self, hash: impl AsRef + Sync + Send, ) -> crate::Result { - let from_key = BlobKey { - account_id: u32::MAX, + self.get_value::<()>(ValueKey { + account_id: 0, collection: 0, document_id: 0, - op: BlobOp::Link, - hash: hash.as_ref().clone(), - }; - let to_key = BlobKey { - account_id: u32::MAX, - collection: 1, - document_id: 0, - op: BlobOp::Link, - hash: hash.as_ref().clone(), - }; - - let mut exists = false; - - self.iterate( - IterateParams::new(from_key, to_key) - .ascending() - .no_values() - .only_first(), - |_, _| { - exists = true; - Ok(false) - }, - ) - .await?; - - Ok(exists) + class: ValueClass::Blob(BlobOp::Commit { + hash: hash.as_ref().clone(), + }), + }) + .await + .map(|v| v.is_some()) } - pub async fn blob_hash_quota(&self, account_id: u32) -> crate::Result { - let from_key = BlobKey { + pub async fn blob_quota(&self, account_id: u32) -> crate::Result { + let from_key = ValueKey { account_id, collection: 0, document_id: 0, - op: BlobOp::Reserve { until: 0, size: 0 }, - hash: BlobHash::default(), + class: ValueClass::Blob(BlobOp::Reserve { + hash: BlobHash::default(), + until: 0, + }), }; - let to_key = BlobKey { + let to_key = ValueKey { account_id: account_id + 1, collection: 0, document_id: 0, - op: BlobOp::Reserve { until: 0, size: 0 }, - hash: BlobHash::default(), + class: ValueClass::Blob(BlobOp::Reserve { + hash: BlobHash::default(), + until: 0, + }), }; let now = now(); let mut quota = BlobQuota { bytes: 0, count: 0 }; self.iterate( - IterateParams::new(from_key, to_key).ascending().no_values(), - |key, _| { - let until = key.deserialize_be_u64(key.len() - (U64_LEN + U32_LEN))?; + IterateParams::new(from_key, to_key).ascending(), + |key, value| { + let until = key.deserialize_be_u64(key.len() - U64_LEN)?; if until > now { - let bytes = key.deserialize_be_u32(key.len() - U32_LEN)? as usize; + let bytes = u32::deserialize(value)?; if bytes > 0 { - quota.bytes += bytes; + quota.bytes += bytes as usize; quota.count += 1; } } @@ -111,88 +95,61 @@ impl Store { Ok(quota) } - pub async fn blob_hash_can_read( + pub async fn blob_has_access( &self, hash: impl AsRef + Sync + Send, class: impl AsRef + Sync + Send, ) -> crate::Result { - let (from_key, to_key) = match class.as_ref() { - BlobClass::Reserved { account_id } => ( - BlobKey { - account_id: *account_id, - collection: 0, - document_id: 0, - op: BlobOp::Reserve { - until: now(), - size: 0, - }, + let key = match class.as_ref() { + BlobClass::Reserved { + account_id, + expires, + } if *expires > now() => ValueKey { + account_id: *account_id, + collection: 0, + document_id: 0, + class: ValueClass::Blob(BlobOp::Reserve { hash: hash.as_ref().clone(), - }, - BlobKey { - account_id: *account_id, - collection: 0, - document_id: 0, - op: BlobOp::Reserve { - until: u64::MAX, - size: u32::MAX as usize, - }, - hash: hash.as_ref().clone(), - }, - ), + until: *expires, + }), + }, BlobClass::Linked { account_id, collection, document_id, - } => ( - BlobKey { - account_id: *account_id, - collection: *collection, - document_id: *document_id, - op: BlobOp::Link, + } => ValueKey { + account_id: *account_id, + collection: *collection, + document_id: *document_id, + class: ValueClass::Blob(BlobOp::Link { hash: hash.as_ref().clone(), - }, - BlobKey { - account_id: *account_id, - collection: *collection, - document_id: *document_id + 1, - op: BlobOp::Link, - hash: hash.as_ref().clone(), - }, - ), + }), + }, + _ => return Ok(false), }; - let mut has_access = false; - - self.iterate( - IterateParams::new(from_key, to_key) - .ascending() - .no_values() - .only_first(), - |_, _| { - has_access = true; - Ok(false) - }, - ) - .await?; - - Ok(has_access) + self.get_value::<()>(key).await.map(|v| v.is_some()) } pub async fn purge_blobs(&self, blob_store: BlobStore) -> crate::Result<()> { // Remove expired temporary blobs - let from_key = BlobKey { + let from_key = ValueKey { account_id: 0, collection: 0, document_id: 0, - op: BlobOp::Reserve { until: 0, size: 0 }, - hash: BlobHash::default(), + class: ValueClass::Blob(BlobOp::Reserve { + until: 0, + hash: BlobHash::default(), + }), }; - let to_key = BlobKey { + let to_key = ValueKey { account_id: u32::MAX, collection: 0, document_id: 0, - op: BlobOp::Reserve { until: 0, size: 0 }, - hash: BlobHash::default(), + class: ValueClass::Blob(BlobOp::Reserve { + until: 0, + hash: BlobHash::default(), + }), }; let mut delete_keys = Vec::new(); let mut active_hashes = AHashSet::new(); @@ -209,16 +166,13 @@ impl Store { })?, ) .unwrap(); - let until = key.deserialize_be_u64(key.len() - (U64_LEN + U32_LEN))?; - if until < now { - let account_id = key.deserialize_be_u32(1)?; - let size = key.deserialize_be_u32(key.len() - U32_LEN)? as usize; - delete_keys.push(BlobKey { - account_id, + let until = key.deserialize_be_u64(key.len() - U64_LEN)?; + if until <= now { + delete_keys.push(ValueKey { + account_id: key.deserialize_be_u32(1)?, collection: 0, document_id: 0, - hash, - op: BlobOp::Reserve { until, size }, + class: ValueClass::Blob(BlobOp::Reserve { until, hash }), }); } else { active_hashes.insert(hash); @@ -229,19 +183,21 @@ impl Store { .await?; // Validate linked blobs - let from_key = BlobKey { + let from_key = ValueKey { account_id: 0, collection: 0, document_id: 0, - op: BlobOp::Link, - hash: BlobHash::default(), + class: ValueClass::Blob(BlobOp::Link { + hash: BlobHash::default(), + }), }; - let to_key = BlobKey { + let to_key = ValueKey { account_id: u32::MAX, collection: u8::MAX, document_id: u32::MAX, - op: BlobOp::Link, - hash: BlobHash::new_max(), + class: ValueClass::Blob(BlobOp::Link { + hash: BlobHash::new_max(), + }), }; let mut last_hash = BlobHash::default(); self.iterate( @@ -263,12 +219,11 @@ impl Store { } } else if last_hash != hash && !active_hashes.contains(&hash) { // Unlinked or expired blob, delete. - delete_keys.push(BlobKey { + delete_keys.push(ValueKey { account_id: 0, collection: 0, document_id: 0, - hash, - op: BlobOp::Commit, + class: ValueClass::Blob(BlobOp::Commit { hash }), }); } @@ -279,8 +234,8 @@ impl Store { // Delete expired or unlinked blobs for key in &delete_keys { - if matches!(key.op, BlobOp::Commit) { - blob_store.delete_blob(key.hash.as_ref()).await?; + if let ValueClass::Blob(BlobOp::Commit { hash }) = &key.class { + blob_store.delete_blob(hash.as_ref()).await?; } } @@ -293,11 +248,16 @@ impl Store { self.write(batch.build()).await?; batch = BatchBuilder::new(); } - if matches!(key.op, BlobOp::Reserve { .. }) && key.account_id != last_account_id { + if matches!(key.class, ValueClass::Blob(BlobOp::Reserve { .. })) + && key.account_id != last_account_id + { batch.with_account_id(key.account_id); last_account_id = key.account_id; } - batch.blob(key.hash, key.op, F_CLEAR); + batch.ops.push(Operation::Value { + class: key.class, + op: ValueOp::Clear, + }) } if !batch.is_empty() { self.write(batch.build()).await?; @@ -308,19 +268,21 @@ impl Store { pub async fn blob_hash_unlink_account(&self, account_id: u32) -> crate::Result<()> { // Validate linked blobs - let from_key = BlobKey { + let from_key = ValueKey { account_id: 0, collection: 0, document_id: 0, - op: BlobOp::Link, - hash: BlobHash::default(), + class: ValueClass::Blob(BlobOp::Link { + hash: BlobHash::default(), + }), }; - let to_key = BlobKey { + let to_key = ValueKey { account_id: u32::MAX, collection: u8::MAX, document_id: u32::MAX, - op: BlobOp::Link, - hash: BlobHash::new_max(), + class: ValueClass::Blob(BlobOp::Link { + hash: BlobHash::new_max(), + }), }; let mut delete_keys = Vec::new(); self.iterate( @@ -331,19 +293,20 @@ impl Store { if document_id != u32::MAX && key.deserialize_be_u32(1 + BLOB_HASH_LEN)? == account_id { - delete_keys.push(BlobKey { + delete_keys.push(ValueKey { account_id, collection: key[1 + BLOB_HASH_LEN + U32_LEN], document_id, - hash: BlobHash::try_from_hash_slice( - key.get(1..1 + BLOB_HASH_LEN).ok_or_else(|| { - crate::Error::InternalError(format!( - "Invalid key {key:?} in blob hash tables" - )) - })?, - ) - .unwrap(), - op: BlobOp::Link, + class: ValueClass::Blob(BlobOp::Link { + hash: BlobHash::try_from_hash_slice( + key.get(1..1 + BLOB_HASH_LEN).ok_or_else(|| { + crate::Error::InternalError(format!( + "Invalid key {key:?} in blob hash tables" + )) + })?, + ) + .unwrap(), + }), }); } @@ -367,9 +330,11 @@ impl Store { batch.with_collection(key.collection); last_collection = key.collection; } - batch - .update_document(key.document_id) - .blob(key.hash, key.op, F_CLEAR); + batch.update_document(key.document_id); + batch.ops.push(Operation::Value { + class: key.class, + op: ValueOp::Clear, + }); } if !batch.is_empty() { self.write(batch.build()).await?; diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 7a53e3cd..5027294d 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -25,11 +25,11 @@ use std::convert::TryInto; use utils::codec::leb128::Leb128_; use crate::{ - BitmapKey, BlobHash, BlobKey, IndexKey, IndexKeyPrefix, Key, LogKey, ValueKey, BLOB_HASH_LEN, - SUBSPACE_BITMAPS, SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_VALUES, U32_LEN, U64_LEN, + BitmapKey, IndexKey, IndexKeyPrefix, Key, LogKey, ValueKey, BLOB_HASH_LEN, SUBSPACE_BITMAPS, + SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_VALUES, U32_LEN, U64_LEN, }; -use super::{AnyKey, BitmapClass, BlobOp, DirectoryValue, TagValue, ValueClass}; +use super::{AnyKey, BitmapClass, BlobOp, DirectoryClass, TagValue, ValueClass}; pub struct KeySerializer { pub buf: Vec, @@ -255,12 +255,31 @@ impl + Sync + Send> Key for ValueKey { .write(*seq) .write(self.account_id) .write(self.document_id), + ValueClass::Blob(op) => match op { + BlobOp::Reserve { hash, until } => serializer + .write(6u8) + .write(self.account_id) + .write::<&[u8]>(hash.as_ref()) + .write(*until), + BlobOp::Commit { hash } => serializer + .write(7u8) + .write::<&[u8]>(hash.as_ref()) + .write(u32::MAX) + .write(0u8) + .write(u32::MAX), + BlobOp::Link { hash } => serializer + .write(7u8) + .write::<&[u8]>(hash.as_ref()) + .write(self.account_id) + .write(self.collection) + .write(self.document_id), + }, ValueClass::Directory(directory) => match directory { - DirectoryValue::NameToId(name) => serializer.write(6u8).write(name.as_slice()), - DirectoryValue::EmailToId(email) => serializer.write(7u8).write(email.as_slice()), - DirectoryValue::Principal(uid) => serializer.write(8u8).write_leb128(*uid), - DirectoryValue::Domain(name) => serializer.write(9u8).write(name.as_slice()), - DirectoryValue::UsedQuota(uid) => serializer.write(10u8).write_leb128(*uid), + DirectoryClass::NameToId(name) => serializer.write(8u8).write(name.as_slice()), + DirectoryClass::EmailToId(email) => serializer.write(9u8).write(email.as_slice()), + DirectoryClass::Principal(uid) => serializer.write(10u8).write_leb128(*uid), + DirectoryClass::Domain(name) => serializer.write(11u8).write(name.as_slice()), + DirectoryClass::UsedQuota(uid) => serializer.write(12u8).write_leb128(*uid), }, } .finalize() @@ -362,44 +381,6 @@ impl + Sync + Send> Key for BitmapKey { } } -impl + Sync + Send> Key for BlobKey { - fn serialize(&self, include_subspace: bool) -> Vec { - let ks = { - if include_subspace { - KeySerializer::new(BLOB_HASH_LEN + (U64_LEN * 3) + 1).write(crate::SUBSPACE_BLOBS) - } else { - KeySerializer::new(BLOB_HASH_LEN + (U64_LEN * 3)) - } - }; - - match self.op { - BlobOp::Reserve { until, size } => ks - .write(1u8) - .write(self.account_id) - .write::<&[u8]>(self.hash.as_ref().as_ref()) - .write(until) - .write(size as u32), - BlobOp::Commit => ks - .write(0u8) - .write::<&[u8]>(self.hash.as_ref().as_ref()) - .write(u32::MAX) - .write(0u8) - .write(u32::MAX), - BlobOp::Link => ks - .write(0u8) - .write::<&[u8]>(self.hash.as_ref().as_ref()) - .write(self.account_id) - .write(self.collection) - .write(self.document_id), - } - .finalize() - } - - fn subspace(&self) -> u8 { - crate::SUBSPACE_BLOBS - } -} - impl + Sync + Send> Key for AnyKey { fn serialize(&self, include_subspace: bool) -> Vec { let key = self.key.as_ref(); @@ -426,10 +407,14 @@ impl ValueClass { ValueClass::Acl(_) => U32_LEN * 3 + 2, ValueClass::Key(v) => v.len(), ValueClass::Directory(d) => match d { - DirectoryValue::NameToId(v) - | DirectoryValue::EmailToId(v) - | DirectoryValue::Domain(v) => v.len(), - DirectoryValue::Principal(_) | DirectoryValue::UsedQuota(_) => U32_LEN, + DirectoryClass::NameToId(v) + | DirectoryClass::EmailToId(v) + | DirectoryClass::Domain(v) => v.len(), + DirectoryClass::Principal(_) | DirectoryClass::UsedQuota(_) => U32_LEN, + }, + ValueClass::Blob(op) => match op { + BlobOp::Reserve { .. } => BLOB_HASH_LEN + U64_LEN + U32_LEN + 1, + BlobOp::Commit { .. } | BlobOp::Link { .. } => BLOB_HASH_LEN + U32_LEN * 2 + 2, }, ValueClass::IndexEmail { .. } => U64_LEN * 2, } @@ -447,8 +432,8 @@ impl From for ValueKey { } } -impl From for ValueKey { - fn from(value: DirectoryValue) -> Self { +impl From for ValueKey { + fn from(value: DirectoryClass) -> Self { ValueKey { account_id: 0, collection: 0, @@ -458,8 +443,14 @@ impl From for ValueKey { } } -impl From for ValueClass { - fn from(value: DirectoryValue) -> Self { +impl From for ValueClass { + fn from(value: DirectoryClass) -> Self { ValueClass::Directory(value) } } + +impl From for ValueClass { + fn from(value: BlobOp) -> Self { + ValueClass::Blob(value) + } +} diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index 2f93bf5e..08f5c54c 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -102,11 +102,6 @@ pub enum Operation { class: BitmapClass, set: bool, }, - Blob { - hash: BlobHash, - op: BlobOp, - set: bool, - }, Log { change_id: u64, collection: u8, @@ -141,12 +136,13 @@ pub enum ValueClass { Key(Vec), TermIndex, ReservedId, - Directory(DirectoryValue), + Directory(DirectoryClass), + Blob(BlobOp), IndexEmail(u64), } #[derive(Debug, PartialEq, Clone, Eq, Hash)] -pub enum DirectoryValue { +pub enum DirectoryClass { NameToId(Vec), EmailToId(Vec), Domain(Vec), @@ -162,11 +158,11 @@ pub enum ValueOp { Clear, } -#[derive(Debug, PartialEq, Clone, Copy, Eq, Hash)] +#[derive(Debug, PartialEq, Clone, Eq, Hash)] pub enum BlobOp { - Reserve { until: u64, size: usize }, - Commit, - Link, + Reserve { hash: BlobHash, until: u64 }, + Commit { hash: BlobHash }, + Link { hash: BlobHash }, } #[derive(Debug, PartialEq, Clone, Eq, Hash)] @@ -270,7 +266,7 @@ impl Deserialize for u64 { impl Deserialize for u32 { fn deserialize(bytes: &[u8]) -> crate::Result { Ok(u32::from_be_bytes(bytes.try_into().map_err(|_| { - crate::Error::InternalError("Failed to deserialize u64".to_string()) + crate::Error::InternalError("Failed to deserialize u32".to_string()) })?)) } } @@ -595,9 +591,16 @@ impl From for Vec { impl BlobClass { pub fn account_id(&self) -> u32 { match self { - BlobClass::Reserved { account_id } | BlobClass::Linked { account_id, .. } => { + BlobClass::Reserved { account_id, .. } | BlobClass::Linked { account_id, .. } => { *account_id } } } + + pub fn is_valid(&self) -> bool { + match self { + BlobClass::Reserved { expires, .. } => *expires > now(), + BlobClass::Linked { .. } => true, + } + } } diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 95173913..08432d7a 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -6,7 +6,7 @@ resolver = "2" [features] #default = ["sqlite", "foundationdb", "postgres", "mysql", "rocks", "elastic", "s3", "redis"] -default = ["sqlite", "postgres", "mysql", "redis"] +default = ["sqlite", "postgres", "mysql", "redis", "rocks", "foundationdb"] sqlite = ["store/sqlite"] foundationdb = ["store/foundation"] postgres = ["store/postgres"] diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 7790cb96..47bcba43 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -282,7 +282,7 @@ pub async fn jmap_tests() { .await; let coco = 1; //email_query::test(&mut params, delete).await; - //email_get::test(&mut params).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; diff --git a/tests/src/store/blob.rs b/tests/src/store/blob.rs index 7bd80cfe..20455b77 100644 --- a/tests/src/store/blob.rs +++ b/tests/src/store/blob.rs @@ -21,10 +21,11 @@ * for more details. */ +use ahash::AHashMap; use store::{ config::ConfigStore, - write::{blob::BlobQuota, now, BatchBuilder, BlobOp, F_CLEAR}, - BlobClass, BlobHash, BlobStore, + write::{blob::BlobQuota, now, BatchBuilder, BlobOp}, + BlobClass, BlobHash, BlobStore, Serialize, }; use utils::config::Config; @@ -53,20 +54,20 @@ pub async fn blob_tests() { // Blob hash exists let hash = BlobHash::from(b"abc".as_slice()); - assert!(!store.blob_hash_exists(&hash).await.unwrap()); + assert!(!store.blob_exists(&hash).await.unwrap()); - // Reserve blob but mark it as expired + // Reserve blob + let until = now() + 1; store .write( BatchBuilder::new() .with_account_id(0) - .blob( - hash.clone(), + .set( BlobOp::Reserve { - until: now() - 10, - size: 1024, + until, + hash: hash.clone(), }, - 0, + 1024u32.serialize(), ) .build_batch(), ) @@ -74,7 +75,7 @@ pub async fn blob_tests() { .unwrap(); // Uncommitted blob, should not exist - assert!(!store.blob_hash_exists(&hash).await.unwrap()); + assert!(!store.blob_exists(&hash).await.unwrap()); // Write blob to store blob_store.put_blob(hash.as_ref(), b"abc").await.unwrap(); @@ -83,30 +84,48 @@ pub async fn blob_tests() { store .write( BatchBuilder::new() - .blob(hash.clone(), BlobOp::Commit, 0) + .set(BlobOp::Commit { hash: hash.clone() }, Vec::new()) .build_batch(), ) .await .unwrap(); // Blob hash should now exist - assert!(store.blob_hash_exists(&hash).await.unwrap()); + assert!(store.blob_exists(&hash).await.unwrap()); + assert!(blob_store + .get_blob(hash.as_ref(), 0..u32::MAX) + .await + .unwrap() + .is_some()); // AccountId 0 should be able to read blob assert!(store - .blob_hash_can_read(&hash, BlobClass::Reserved { account_id: 0 }) + .blob_has_access( + &hash, + BlobClass::Reserved { + account_id: 0, + expires: until + } + ) .await .unwrap()); // AccountId 1 should not be able to read blob assert!(!store - .blob_hash_can_read(&hash, BlobClass::Reserved { account_id: 1 }) + .blob_has_access( + &hash, + BlobClass::Reserved { + account_id: 1, + expires: until + } + ) .await .unwrap()); // Blob already expired, quota should be 0 + tokio::time::sleep(std::time::Duration::from_secs(1)).await; assert_eq!( - store.blob_hash_quota(0).await.unwrap(), + store.blob_quota(0).await.unwrap(), BlobQuota { bytes: 0, count: 0 } ); @@ -114,11 +133,17 @@ pub async fn blob_tests() { store.purge_blobs(blob_store.clone()).await.unwrap(); // Blob hash should no longer exist - assert!(!store.blob_hash_exists(&hash).await.unwrap()); + assert!(!store.blob_exists(&hash).await.unwrap()); // AccountId 0 should not be able to read blob assert!(!store - .blob_hash_can_read(&hash, BlobClass::Reserved { account_id: 0 }) + .blob_has_access( + &hash, + BlobClass::Reserved { + account_id: 0, + expires: until + } + ) .await .unwrap()); @@ -130,44 +155,39 @@ pub async fn blob_tests() { .is_none()); // Upload one linked blob to accountId 1, two linked blobs to accountId 0, and three unlinked (reserved) blobs to accountId 2 - for (document_id, (blob, blob_op)) in [ - (b"123", BlobOp::Link), - (b"456", BlobOp::Link), - (b"789", BlobOp::Link), - ( - b"abc", - BlobOp::Reserve { - until: now() - 10, - size: 5000, - }, - ), - ( - b"efg", - BlobOp::Reserve { - until: now() + 10, - size: 1000, - }, - ), - ( - b"hij", - BlobOp::Reserve { - until: now() + 10, - size: 2000, - }, - ), + let expiry_times = AHashMap::from_iter([ + (b"abc", now() - 10), + (b"efg", now() + 10), + (b"hij", now() + 10), + ]); + for (document_id, (blob, blob_value)) in [ + (b"123", vec![]), + (b"456", vec![]), + (b"789", vec![]), + (b"abc", 5000u32.serialize()), + (b"efg", 1000u32.serialize()), + (b"hij", 2000u32.serialize()), ] .into_iter() .enumerate() { let hash = BlobHash::from(blob.as_slice()); + let blob_op = if let Some(until) = expiry_times.get(blob) { + BlobOp::Reserve { + until: *until, + hash: hash.clone(), + } + } else { + BlobOp::Link { hash: hash.clone() } + }; store .write( BatchBuilder::new() .with_account_id(if document_id > 0 { 0 } else { 1 }) .with_collection(0) .update_document(document_id as u32) - .blob(hash.clone(), blob_op, 0) - .blob(hash.clone(), BlobOp::Commit, 0) + .set(blob_op, blob_value) + .set(BlobOp::Commit { hash: hash.clone() }, vec![]) .build_batch(), ) .await @@ -180,21 +200,27 @@ pub async fn blob_tests() { // One of the reserved blobs expired and should not count towards quota assert_eq!( - store.blob_hash_quota(0).await.unwrap(), + store.blob_quota(0).await.unwrap(), BlobQuota { bytes: 3000, count: 2 } ); assert_eq!( - store.blob_hash_quota(1).await.unwrap(), + store.blob_quota(1).await.unwrap(), BlobQuota { bytes: 0, count: 0 } ); // Purge expired blobs and make sure nothing else is deleted store.purge_blobs(blob_store.clone()).await.unwrap(); for (pos, (blob, blob_class)) in [ - (b"abc", BlobClass::Reserved { account_id: 0 }), + ( + b"abc", + BlobClass::Reserved { + account_id: 0, + expires: expiry_times[&b"abc"], + }, + ), ( b"123", BlobClass::Linked { @@ -219,16 +245,28 @@ pub async fn blob_tests() { document_id: 2, }, ), - (b"efg", BlobClass::Reserved { account_id: 0 }), - (b"hij", BlobClass::Reserved { account_id: 0 }), + ( + b"efg", + BlobClass::Reserved { + account_id: 0, + expires: expiry_times[&b"efg"], + }, + ), + ( + b"hij", + BlobClass::Reserved { + account_id: 0, + expires: expiry_times[&b"hij"], + }, + ), ] .into_iter() .enumerate() { let ct = pos == 0; let hash = BlobHash::from(blob.as_slice()); - assert!(store.blob_hash_can_read(&hash, blob_class).await.unwrap() ^ ct); - assert!(store.blob_hash_exists(&hash).await.unwrap() ^ ct); + assert!(store.blob_has_access(&hash, blob_class).await.unwrap() ^ ct); + assert!(store.blob_exists(&hash).await.unwrap() ^ ct); assert!( blob_store .get_blob(hash.as_ref(), 0..u32::MAX) @@ -241,7 +279,7 @@ pub async fn blob_tests() { // AccountId 0 should not have access to accountId 1's blobs assert!(!store - .blob_hash_can_read( + .blob_has_access( BlobHash::from(b"123".as_slice()), BlobClass::Linked { account_id: 0, @@ -259,7 +297,9 @@ pub async fn blob_tests() { .with_account_id(0) .with_collection(0) .update_document(2) - .blob(BlobHash::from(b"789".as_slice()), BlobOp::Link, F_CLEAR) + .clear(BlobOp::Link { + hash: BlobHash::from(b"789".as_slice()), + }) .build_batch(), ) .await @@ -292,16 +332,28 @@ pub async fn blob_tests() { document_id: 1, }, ), - (b"efg", BlobClass::Reserved { account_id: 0 }), - (b"hij", BlobClass::Reserved { account_id: 0 }), + ( + b"efg", + BlobClass::Reserved { + account_id: 0, + expires: expiry_times[&b"efg"], + }, + ), + ( + b"hij", + BlobClass::Reserved { + account_id: 0, + expires: expiry_times[&b"hij"], + }, + ), ] .into_iter() .enumerate() { let ct = pos == 0; let hash = BlobHash::from(blob.as_slice()); - assert!(store.blob_hash_can_read(&hash, blob_class).await.unwrap() ^ ct); - assert!(store.blob_hash_exists(&hash).await.unwrap() ^ ct); + assert!(store.blob_has_access(&hash, blob_class).await.unwrap() ^ ct); + assert!(store.blob_exists(&hash).await.unwrap() ^ ct); assert!( blob_store .get_blob(hash.as_ref(), 0..u32::MAX) @@ -334,16 +386,28 @@ pub async fn blob_tests() { document_id: 1, }, ), - (b"efg", BlobClass::Reserved { account_id: 0 }), - (b"hij", BlobClass::Reserved { account_id: 0 }), + ( + b"efg", + BlobClass::Reserved { + account_id: 0, + expires: expiry_times[&b"efg"], + }, + ), + ( + b"hij", + BlobClass::Reserved { + account_id: 0, + expires: expiry_times[&b"hij"], + }, + ), ] .into_iter() .enumerate() { let ct = pos == 0; let hash = BlobHash::from(blob.as_slice()); - assert!(store.blob_hash_can_read(&hash, blob_class).await.unwrap() ^ ct); - assert!(store.blob_hash_exists(&hash).await.unwrap() ^ ct); + assert!(store.blob_has_access(&hash, blob_class).await.unwrap() ^ ct); + assert!(store.blob_exists(&hash).await.unwrap() ^ ct); assert!( blob_store .get_blob(hash.as_ref(), 0..u32::MAX)