diff --git a/Cargo.lock b/Cargo.lock index 90028b35..f7992ebe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4386,6 +4386,7 @@ dependencies = [ "mail-auth", "mail-parser", "nlp", + "num_cpus", "proc_macros", "rkyv", "serde", diff --git a/README.md b/README.md index 225f09ce..0dcfd9b1 100644 --- a/README.md +++ b/README.md @@ -157,7 +157,7 @@ Your support is crucial in helping us continue to improve the project, add new f These are some of our open-source sponsors: -User avatar: James BrumondUser avatar: MailRoute, Inc.User avatar: Starsong GmbHUser avatar: Ming Fu Design Ltd. 明孚設計有限公司User avatar: TaminoUser avatar: panascaisUser avatar: Merlijn +User avatar: James BrumondUser avatar: MailRoute, Inc.User avatar: Starsong GmbHUser avatar: Ming Fu Design Ltd. 明孚設計有限公司User avatar: TaminoUser avatar: panascaisUser avatar: Jan JonssonUser avatar: GitHub SecureOSS Fund
If you would like to support our work, please consider [becoming a sponsor](https://opencollective.com/stalwart). diff --git a/crates/email/src/message/index/search.rs b/crates/email/src/message/index/search.rs index dce77e37..81f9f983 100644 --- a/crates/email/src/message/index/search.rs +++ b/crates/email/src/message/index/search.rs @@ -217,7 +217,7 @@ impl ArchivedMessageMetadata { document.insert_key_value( EmailSearchField::Headers, - header.name.as_str().to_string(), + header.name.as_str(), value, ); } diff --git a/crates/http/src/management/enterprise/telemetry.rs b/crates/http/src/management/enterprise/telemetry.rs index 9347bbe3..98139395 100644 --- a/crates/http/src/management/enterprise/telemetry.rs +++ b/crates/http/src/management/enterprise/telemetry.rs @@ -33,7 +33,7 @@ use serde_json::json; use std::future::Future; use store::{ ahash::{AHashMap, AHashSet}, - search::{SearchField, SearchFilter, SearchQuery, TracingSearchField}, + search::{SearchComparator, SearchField, SearchFilter, SearchQuery, TracingSearchField}, write::{SearchIndex, now}, }; use trc::{ @@ -165,7 +165,12 @@ impl TelemetryApi for Server { let span_ids = self .search_store() .query_global( - SearchQuery::new(SearchIndex::Tracing).with_filters(tracing_query), + SearchQuery::new(SearchIndex::Tracing) + .with_filters(tracing_query) + .with_comparator(SearchComparator::Field { + field: SearchField::Id, + ascending: false, + }), ) .await?; diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index 34c08d37..153fdbc3 100644 --- a/crates/main/Cargo.toml +++ b/crates/main/Cargo.toml @@ -64,4 +64,5 @@ enterprise = [ "jmap/enterprise", "dav/enterprise", "groupware/enterprise", "trc/enterprise", - "services/enterprise" ] + "services/enterprise", + "migration/enterprise" ] diff --git a/crates/migration/Cargo.toml b/crates/migration/Cargo.toml index 76b11d3a..01cd9e9e 100644 --- a/crates/migration/Cargo.toml +++ b/crates/migration/Cargo.toml @@ -31,6 +31,7 @@ bincode = "1.3.3" lz4_flex = { version = "0.12", default-features = false } base64 = "0.22" futures = "0.3" +num_cpus = "1.13.1" [features] test_mode = [] diff --git a/crates/migration/src/blob.rs b/crates/migration/src/blob.rs index 3ec8df4e..a0add790 100644 --- a/crates/migration/src/blob.rs +++ b/crates/migration/src/blob.rs @@ -18,6 +18,7 @@ use types::blob_hash::{BLOB_HASH_LEN, BlobHash}; const SUBSPACE_BLOB_RESERVE: u8 = b'j'; pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> { + let mut num_blobs = 0; for byte in 0..=u8::MAX { // Validate linked blobs let mut from_hash = BlobHash::default(); @@ -78,6 +79,7 @@ pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> { .caused_by(trc::location!())?; let mut batch = BatchBuilder::new(); + num_blobs += keys.len(); for (key, op) in keys { batch .clear(ValueClass::Any(AnyClass { @@ -104,6 +106,11 @@ pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> { } } + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!("Migrated {num_blobs} blob links") + ); + enum OldType { Quota { size: u32 }, Undelete { deleted_at: u64, size: u32 }, @@ -181,6 +188,7 @@ pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> { .caused_by(trc::location!())?; let mut batch = BatchBuilder::new(); + let num_entries = entries.len(); for entry in entries { batch .clear(ValueClass::Any(AnyClass { @@ -259,6 +267,11 @@ pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> { } } + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!("Migrated {num_entries} temporary blob links") + ); + if !batch.is_empty() { server .store() diff --git a/crates/migration/src/principal_v2.rs b/crates/migration/src/principal_v2.rs index 0b833401..5f34a5fa 100644 --- a/crates/migration/src/principal_v2.rs +++ b/crates/migration/src/principal_v2.rs @@ -4,19 +4,6 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::time::Instant; - -use common::Server; -use directory::{Principal, PrincipalData, Type, backend::internal::SpecialSecrets}; -use proc_macros::EnumMethods; -use store::{ - Serialize, ValueKey, - roaring::RoaringBitmap, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, DirectoryClass, ValueClass}, -}; -use trc::AddContext; -use types::collection::Collection; - use crate::{ addressbook_v2::migrate_addressbook_v013, calendar_v2::migrate_calendar_v013, @@ -26,6 +13,17 @@ use crate::{ push_v2::migrate_push_subscriptions_v013, sieve_v2::migrate_sieve_v013, }; +use common::Server; +use directory::{Principal, PrincipalData, Type, backend::internal::SpecialSecrets}; +use proc_macros::EnumMethods; +use std::time::Instant; +use store::{ + Serialize, ValueKey, + roaring::RoaringBitmap, + write::{AlignedBytes, Archive, Archiver, BatchBuilder, DirectoryClass, ValueClass}, +}; +use trc::AddContext; +use types::collection::Collection; pub(crate) async fn migrate_principals_v0_13(server: &Server) -> trc::Result { // Obtain email ids diff --git a/crates/migration/src/queue_v2.rs b/crates/migration/src/queue_v2.rs index a68968b8..9fa4bddc 100644 --- a/crates/migration/src/queue_v2.rs +++ b/crates/migration/src/queue_v2.rs @@ -152,8 +152,6 @@ pub enum LegacyQuotaKey { } pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> { - let mut count = 0; - let mut messages = Vec::new(); server .store() @@ -176,8 +174,6 @@ pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> { } } - count += 1; - Ok(true) }, ) @@ -185,6 +181,7 @@ pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> { .caused_by(trc::location!())?; let mut batch = BatchBuilder::new(); + let count = messages.len(); for (queue_id, message) in messages { batch.set( ValueClass::Queue(QueueClass::Message(queue_id)), @@ -211,12 +208,10 @@ pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> { .caused_by(trc::location!())?; } - if count > 0 { - trc::event!( - Server(trc::ServerEvent::Startup), - Details = format!("Migrated {count} queued messages",) - ); - } + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!("Migrated {count} queued messages",) + ); Ok(()) } diff --git a/crates/migration/src/tasks_v2.rs b/crates/migration/src/tasks_v2.rs index c5a20b54..701012eb 100644 --- a/crates/migration/src/tasks_v2.rs +++ b/crates/migration/src/tasks_v2.rs @@ -7,7 +7,10 @@ use common::Server; use store::{ IterateParams, SUBSPACE_TASK_QUEUE, U32_LEN, U64_LEN, ValueKey, - write::{AnyClass, BatchBuilder, ValueClass, key::KeySerializer}, + write::{ + AnyClass, BatchBuilder, TaskEpoch, ValueClass, + key::{DeserializeBigEndian, KeySerializer}, + }, }; use trc::AddContext; @@ -31,16 +34,15 @@ pub(crate) async fn migrate_tasks_v014(server: &Server) -> trc::Result<()> { }), }; - let todo = "task epochs"; - let mut delete_tasks = Vec::new(); + let mut insert_tasks = Vec::new(); server .core .storage .data .iterate( - IterateParams::new(from_key, to_key).ascending().no_values(), - |key, _| { + IterateParams::new(from_key, to_key).ascending(), + |key, value| { match key.get(U64_LEN + U32_LEN) { Some(0..=2) => { delete_tasks.push(key.to_vec()); @@ -48,7 +50,18 @@ pub(crate) async fn migrate_tasks_v014(server: &Server) -> trc::Result<()> { None => { return Err(trc::Error::corrupted_key(key, None, trc::location!())); } - _ => {} + _ => { + let due = key.deserialize_be_u64(0)?; + let maybe_epoch = TaskEpoch::from_inner(due); + if maybe_epoch.attempt() != 0 { + delete_tasks.push(key.to_vec()); + let epoch = TaskEpoch::new(due).inner(); + let mut new_key = Vec::with_capacity(key.len()); + new_key.extend_from_slice(&epoch.to_be_bytes()); + new_key.extend_from_slice(&key[U64_LEN..]); + insert_tasks.push((new_key, value.to_vec())); + } + } }; Ok(true) }, @@ -56,26 +69,57 @@ pub(crate) async fn migrate_tasks_v014(server: &Server) -> trc::Result<()> { .await .caused_by(trc::location!())?; - if !delete_tasks.is_empty() { - let num_migrated = delete_tasks.len(); + let num_migrated = delete_tasks.len() + insert_tasks.len(); + if num_migrated != 0 { let mut batch = BatchBuilder::new(); + let mut batch_len = 0; + for (key, value) in insert_tasks { + batch_len += key.len() + value.len(); + batch.set( + ValueClass::Any(AnyClass { + subspace: SUBSPACE_TASK_QUEUE, + key, + }), + value, + ); + if batch_len > 4 * 1024 * 1024 { + server + .store() + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + batch = BatchBuilder::new(); + batch_len = 0; + } + } + for key in delete_tasks { + batch_len += key.len(); batch.clear(ValueClass::Any(AnyClass { subspace: SUBSPACE_TASK_QUEUE, key, })); + if batch_len > 4 * 1024 * 1024 { + server + .store() + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + batch = BatchBuilder::new(); + batch_len = 0; + } } server .store() .write(batch.build_all()) .await .caused_by(trc::location!())?; - - trc::event!( - Server(trc::ServerEvent::Startup), - Details = format!("Migrated {num_migrated} tasks") - ); } + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!("Migrated {num_migrated} tasks") + ); + Ok(()) } diff --git a/crates/migration/src/v014.rs b/crates/migration/src/v014.rs index 2bc60b54..00ed8145 100644 --- a/crates/migration/src/v014.rs +++ b/crates/migration/src/v014.rs @@ -35,7 +35,7 @@ pub const SUBSPACE_BITMAP_TEXT: u8 = b'v'; pub const SUBSPACE_FTS_INDEX: u8 = b'g'; pub const SUBSPACE_TELEMETRY_INDEX: u8 = b'w'; -pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> { +pub async fn migrate_v0_14(server: &Server) -> trc::Result<()> { // Migrate global data let mut tasks = Vec::new(); let _server = server.clone(); @@ -74,9 +74,10 @@ pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> { std::env::var("NUM_THREADS") .ok() .and_then(|s| s.parse::().ok()) - .unwrap_or(8), + .unwrap_or_else(|| num_cpus::get().min(2) * 2), )); let mut tasks = Vec::with_capacity(principal_ids.len()); + let num_principals = principal_ids.len(); for principal_id in principal_ids { let permit = semaphore.clone().acquire_owned().await.unwrap(); let _server = server.clone(); @@ -97,6 +98,11 @@ pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> { .details("Join Error") })??; + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!("Migrated {num_principals} accounts") + ); + // Delete old subspaces for subspace in [ SUBSPACE_BITMAP_ID, @@ -121,16 +127,34 @@ pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> { .caused_by(trc::location!())?; } + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!("Migration to v0.15 completed") + ); + Ok(()) } pub(crate) async fn migrate_principal_v0_14(server: &Server, account_id: u32) -> trc::Result<()> { - migrate_emails_v014(server, account_id).await?; - migrate_encryption_params_v014(server, account_id).await?; - migrate_indexes(server, account_id).await + let emails = migrate_emails_v014(server, account_id).await?; + let params = migrate_encryption_params_v014(server, account_id).await?; + let (num_contacts, num_calendars, num_email_submissions, num_identities) = + migrate_indexes(server, account_id).await?; + + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!( + "Migrated account {account_id}: {emails} emails, {params} encryption params, {num_contacts} contacts, {num_calendars} calendars, {num_email_submissions} submissions, and {num_identities} identities" + ) + ); + + Ok(()) } -pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Result<()> { +pub(crate) async fn migrate_indexes( + server: &Server, + account_id: u32, +) -> trc::Result<(usize, usize, usize, usize)> { /* EmailSubmissionField::UndoStatus => 41, @@ -211,6 +235,10 @@ pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Re } let mut indexes = Vec::new(); + let mut num_contacts = 0; + let mut num_calendars = 0; + let mut num_email_submissions = 0; + let mut num_identities = 0; for collection in [ Collection::ContactCard, Collection::CalendarEventNotification, @@ -218,104 +246,101 @@ pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Re Collection::Identity, ] { server - .archives( - account_id, - Collection::ContactCard, - &(), - |document_id, archive| { - match collection { - Collection::ContactCard => { - let data = archive - .unarchive_untrusted::() - .caused_by(trc::location!())?; + .archives(account_id, collection, &(), |document_id, archive| { + match collection { + Collection::ContactCard => { + let data = archive + .unarchive_untrusted::() + .caused_by(trc::location!())?; - if let Some(email) = data.emails().next() { - indexes.push(( - collection, - document_id, - Operation::Index { - field: ContactField::Email.into(), - key: email.into_bytes(), - set: true, - }, - )); - } - indexes.push(( - collection, - document_id, - Operation::Value { - class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: ContactField::CreatedToUpdated.into(), - value: data.created.to_native() as u64, - }), - op: ValueOp::Set( - (data.modified.to_native() as u64).serialize(), - ), - }, - )); - } - Collection::CalendarEventNotification => { - let data = archive - .unarchive_untrusted::() - .caused_by(trc::location!())?; - indexes.push(( - collection, - document_id, - Operation::Value { - class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: CalendarNotificationField::CreatedToId.into(), - value: data.created.to_native() as u64, - }), - op: ValueOp::Set( - data.event_id - .as_ref() - .map(|v| v.to_native()) - .unwrap_or(u32::MAX) - .serialize(), - ), - }, - )); - } - Collection::EmailSubmission => { - let data = archive - .unarchive_untrusted::() - .caused_by(trc::location!())?; - indexes.push(( - collection, - document_id, - Operation::Value { - class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailSubmissionField::Metadata.into(), - value: data.send_at.to_native(), - }), - op: ValueOp::Set( - KeySerializer::new(U32_LEN * 3 + 1) - .write(data.email_id.to_native()) - .write(data.thread_id.to_native()) - .write(data.identity_id.to_native()) - .write(data.undo_status.as_index()) - .finalize(), - ), - }, - )); - } - Collection::Identity => { + if let Some(email) = data.emails().next() { indexes.push(( collection, document_id, Operation::Index { - field: IdentityField::DocumentId.into(), - key: vec![], + field: ContactField::Email.into(), + key: email.into_bytes(), set: true, }, )); } - _ => unreachable!(), + num_contacts += 1; + indexes.push(( + collection, + document_id, + Operation::Value { + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: ContactField::CreatedToUpdated.into(), + value: data.created.to_native() as u64, + }), + op: ValueOp::Set((data.modified.to_native() as u64).serialize()), + }, + )); } + Collection::CalendarEventNotification => { + let data = archive + .unarchive_untrusted::() + .caused_by(trc::location!())?; + num_calendars += 1; + indexes.push(( + collection, + document_id, + Operation::Value { + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: CalendarNotificationField::CreatedToId.into(), + value: data.created.to_native() as u64, + }), + op: ValueOp::Set( + data.event_id + .as_ref() + .map(|v| v.to_native()) + .unwrap_or(u32::MAX) + .serialize(), + ), + }, + )); + } + Collection::EmailSubmission => { + let data = archive + .unarchive_untrusted::() + .caused_by(trc::location!())?; + num_email_submissions += 1; + indexes.push(( + collection, + document_id, + Operation::Value { + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: data.send_at.to_native(), + }), + op: ValueOp::Set( + KeySerializer::new(U32_LEN * 3 + 1) + .write(data.email_id.to_native()) + .write(data.thread_id.to_native()) + .write(data.identity_id.to_native()) + .write(data.undo_status.as_index()) + .finalize(), + ), + }, + )); + } + Collection::Identity => { + num_identities += 1; + indexes.push(( + collection, + document_id, + Operation::Index { + field: IdentityField::DocumentId.into(), + key: vec![], + set: true, + }, + )); + } + _ => unreachable!(), + } - Ok(true) - }, - ) + Ok(true) + }) .await .caused_by(trc::location!())?; } @@ -345,5 +370,10 @@ pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Re .caused_by(trc::location!())?; } - Ok(()) + Ok(( + num_contacts, + num_calendars, + num_email_submissions, + num_identities, + )) } diff --git a/crates/services/src/task_manager/index.rs b/crates/services/src/task_manager/index.rs index 96df283b..bcbd234e 100644 --- a/crates/services/src/task_manager/index.rs +++ b/crates/services/src/task_manager/index.rs @@ -17,12 +17,13 @@ use email::{cache::MessageCacheFetch, message::metadata::MessageMetadata}; use groupware::{cache::GroupwareCache, calendar::CalendarEvent, contact::ContactCard}; use std::cmp::Ordering; use store::{ - SerializeInfallible, ValueKey, + IterateParams, SerializeInfallible, ValueKey, ahash::AHashMap, roaring::RoaringBitmap, search::{IndexDocument, SearchField, SearchFilter, SearchQuery}, write::{ - AlignedBytes, Archive, BatchBuilder, SearchIndex, TaskEpoch, TaskQueueClass, ValueClass, + AlignedBytes, Archive, BatchBuilder, SearchIndex, TaskEpoch, TaskQueueClass, + TelemetryClass, ValueClass, key::DeserializeBigEndian, }, }; use trc::{AddContext, TaskQueueEvent}; @@ -384,7 +385,51 @@ impl ReindexIndexTask for Server { } } } - SearchIndex::File | SearchIndex::Tracing | SearchIndex::InMemory => (), + SearchIndex::Tracing => { + let mut spans = Vec::new(); + self.store() + .iterate( + IterateParams::new( + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span { + span_id: 0, + })), + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span { + span_id: u64::MAX, + })), + ) + .no_values(), + |key, _| { + spans.push(key.deserialize_be_u64(0)?); + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + let mut batch = BatchBuilder::new(); + for span_id in spans { + batch + .with_account_id((span_id >> 32) as u32) // TODO: This is hacky, improve + .with_document(span_id as u32) + .set( + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + due: TaskEpoch::now(), + index: SearchIndex::Tracing, + is_insert: true, + }), + vec![], + ); + if batch.len() >= 2000 { + self.core.storage.data.write(batch.build_all()).await?; + batch = BatchBuilder::new(); + } + } + + if !batch.is_empty() { + self.core.storage.data.write(batch.build_all()).await?; + } + } + SearchIndex::File | SearchIndex::InMemory => (), } // Request indexing diff --git a/crates/spam-filter/src/modules/html.rs b/crates/spam-filter/src/modules/html.rs index 88871bfd..93808c85 100644 --- a/crates/spam-filter/src/modules/html.rs +++ b/crates/spam-filter/src/modules/html.rs @@ -145,11 +145,11 @@ pub fn html_to_tokens(input: &str) -> Vec { match iter.peek() { Some(&(_, &b'/')) => { is_end_tag = true; - pos += 1; + //pos += 1; iter.next(); } Some((_, ch)) if ch.is_ascii_whitespace() => { - pos += 1; + //pos += 1; iter.next(); } _ => break, diff --git a/crates/store/src/backend/foundationdb/write.rs b/crates/store/src/backend/foundationdb/write.rs index 9ca638fc..1590e301 100644 --- a/crates/store/src/backend/foundationdb/write.rs +++ b/crates/store/src/backend/foundationdb/write.rs @@ -9,13 +9,12 @@ use super::{ read::{ChunkedValue, read_chunked_value}, }; use crate::{ - IndexKey, Key, LogKey, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER, SUBSPACE_QUOTA, - WITH_SUBSPACE, backend::deserialize_i64_le, write::{ - AssignedIds, Batch, MAX_COMMIT_ATTEMPTS, MAX_COMMIT_TIME, MergeResult, Operation, - ValueClass, ValueOp, key::KeySerializer, + AssignedIds, Batch, DirectoryClass, MAX_COMMIT_ATTEMPTS, MAX_COMMIT_TIME, MergeResult, + Operation, TaskQueueClass, TelemetryClass, ValueClass, ValueOp, key::KeySerializer, }, + *, }; use foundationdb::{ FdbError, KeySelector, RangeOption, Transaction, @@ -82,7 +81,6 @@ impl FdbStore { Operation::Value { class, op } => { let mut key = class.serialize(account_id, collection, document_id, WITH_SUBSPACE); - let do_chunk = !class.is_counter(collection); match op { ValueOp::Set(value) => { @@ -168,7 +166,32 @@ impl FdbStore { result.push_counter_id(num); } ValueOp::Clear => { - if do_chunk { + if matches!( + key[0], + SUBSPACE_DIRECTORY + | SUBSPACE_TASK_QUEUE + | SUBSPACE_IN_MEMORY_VALUE + | SUBSPACE_PROPERTY + | SUBSPACE_QUEUE_MESSAGE + | SUBSPACE_REPORT_OUT + | SUBSPACE_REPORT_IN + | SUBSPACE_TELEMETRY_SPAN + | SUBSPACE_SEARCH_INDEX + | SUBSPACE_LOGS + ) && matches!( + class, + ValueClass::Property(_) + | ValueClass::Queue(_) + | ValueClass::Report(_) + | ValueClass::Directory(DirectoryClass::Principal(_)) + | ValueClass::ShareNotification { .. } + | ValueClass::Telemetry(TelemetryClass::Metric { .. }) + | ValueClass::TaskQueue(TaskQueueClass::SendImip { + is_payload: true, + .. + }) + | ValueClass::InMemory(_) + ) { trx.clear_range( &key, &KeySerializer::new(key.len() + 1) diff --git a/crates/store/src/backend/rocksdb/main.rs b/crates/store/src/backend/rocksdb/main.rs index dba1cdca..b16d6053 100644 --- a/crates/store/src/backend/rocksdb/main.rs +++ b/crates/store/src/backend/rocksdb/main.rs @@ -4,17 +4,13 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::path::PathBuf; - +use super::{CF_BLOBS, RocksDbStore}; +use crate::*; use rocksdb::{ColumnFamilyDescriptor, MergeOperands, OptimisticTransactionDB, Options}; - +use std::path::PathBuf; use tokio::sync::oneshot; use utils::config::{Config, utils::AsKey}; -use crate::*; - -use super::{CF_BLOBS, RocksDbStore}; - impl RocksDbStore { pub async fn open(config: &mut Config, prefix: impl AsKey) -> Option { let prefix = prefix.as_key(); @@ -75,6 +71,11 @@ impl RocksDbStore { SUBSPACE_TELEMETRY_SPAN, SUBSPACE_TELEMETRY_METRIC, SUBSPACE_SEARCH_INDEX, + LEGACY_SUBSPACE_BITMAP_ID, + LEGACY_SUBSPACE_BITMAP_TAG, + LEGACY_SUBSPACE_BITMAP_TEXT, + LEGACY_SUBSPACE_FTS_INDEX, + LEGACY_SUBSPACE_TELEMETRY_INDEX, ] { let cf_opts = Options::default(); cfs.push(ColumnFamilyDescriptor::new( diff --git a/crates/store/src/search/query.rs b/crates/store/src/search/query.rs index 4b155cc4..b0dfc753 100644 --- a/crates/store/src/search/query.rs +++ b/crates/store/src/search/query.rs @@ -393,6 +393,23 @@ impl Store { } } - Ok(state.bm.unwrap_or_default().into_iter().collect::>()) + if query.comparators.iter().all(|c| { + matches!( + c, + SearchComparator::Field { + field: SearchField::Id, + ascending: false + } + ) + }) { + Ok(state + .bm + .unwrap_or_default() + .into_iter() + .rev() + .collect::>()) + } else { + Ok(state.bm.unwrap_or_default().into_iter().collect::>()) + } } } diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 2742b6fc..60cc7f44 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -78,9 +78,9 @@ async fn jmap_tests() { server::webhooks::test(&mut params).await; - /*mail::get::test(&mut params).await; + mail::get::test(&mut params).await; mail::set::test(&mut params).await; - mail::parse::test(&mut params).await;*/ + mail::parse::test(&mut params).await; mail::query::test(&mut params, delete).await; mail::search_snippet::test(&mut params).await; mail::changes::test(&mut params).await; diff --git a/tests/src/store/cleanup.rs b/tests/src/store/cleanup.rs index 497d0c05..37c12842 100644 --- a/tests/src/store/cleanup.rs +++ b/tests/src/store/cleanup.rs @@ -155,9 +155,7 @@ pub async fn store_blob_expire_all(store: &Store) { until, })); } - _ => { - eprintln!("Unknown blob link type for key {key:?}: {value:?}",); - } + _ => {} } batch.clear(ValueClass::Blob(BlobOp::Link {