diff --git a/crates/common/src/manager/restore.rs b/crates/common/src/manager/restore.rs index cb06cd25..55f0e913 100644 --- a/crates/common/src/manager/restore.rs +++ b/crates/common/src/manager/restore.rs @@ -14,7 +14,7 @@ use store::{ BlobStore, Key, LogKey, SUBSPACE_LOGS, SerializeInfallible, Store, U32_LEN, write::{ AnyClass, BatchBuilder, BlobOp, DirectoryClass, InMemoryClass, Operation, SearchIndex, - TaskQueueClass, ValueClass, ValueOp, key::DeserializeBigEndian, now, + TaskEpoch, TaskQueueClass, ValueClass, ValueOp, key::DeserializeBigEndian, now, }, }; use store::{ @@ -147,7 +147,7 @@ async fn restore_file(store: Store, blob_store: BlobStore, path: &Path) { if reader.version == 1 && collection == Collection::Email { batch.set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due, + due: TaskEpoch::from_inner(due), index: SearchIndex::Email, is_insert: true, }), diff --git a/crates/common/src/storage/index.rs b/crates/common/src/storage/index.rs index a5cb71ea..399dfaf2 100644 --- a/crates/common/src/storage/index.rs +++ b/crates/common/src/storage/index.rs @@ -15,7 +15,7 @@ use store::{ Serialize, SerializeInfallible, write::{ Archive, Archiver, BatchBuilder, BlobOp, DirectoryClass, IntoOperations, Params, - SearchIndex, TaskQueueClass, ValueClass, now, + SearchIndex, TaskEpoch, TaskQueueClass, ValueClass, }, }; use types::{ @@ -407,7 +407,7 @@ fn build_index( IndexValue::SearchIndex { index, .. } => { batch.set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due: now(), + due: TaskEpoch::now().with_random_sequence_id(), index, is_insert: set, }), @@ -539,7 +539,7 @@ fn merge_index( (IndexValue::SearchIndex { index, .. }, IndexValue::SearchIndex { .. }) => { batch.set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due: now(), + due: TaskEpoch::now().with_random_sequence_id(), index, is_insert: true, }), diff --git a/crates/common/src/telemetry/tracers/store.rs b/crates/common/src/telemetry/tracers/store.rs index 26ed76fd..3560c6ab 100644 --- a/crates/common/src/telemetry/tracers/store.rs +++ b/crates/common/src/telemetry/tracers/store.rs @@ -10,12 +10,11 @@ use crate::config::telemetry::StoreTracer; use ahash::{AHashMap, AHashSet}; -use nlp::language::Language; -use std::{future::Future, time::Duration}; +use std::{collections::HashSet, future::Future, time::Duration}; use store::{ Deserialize, SearchStore, Store, ValueKey, search::{IndexDocument, SearchField, SearchFilter, SearchQuery, TracingSearchField}, - write::{BatchBuilder, SearchIndex, TaskQueueClass, TelemetryClass, ValueClass, now}, + write::{BatchBuilder, SearchIndex, TaskEpoch, TaskQueueClass, TelemetryClass, ValueClass}, }; use trc::{ AddContext, AuthEvent, Event, EventDetails, EventType, Key, MessageIngestEvent, @@ -35,8 +34,6 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac let mut batch = BatchBuilder::new(); while let Some(events) = rx.recv().await { - let now = now(); - for event in events { if let Some(span) = &event.inner.span { let span_id = span.span_id().unwrap(); @@ -68,7 +65,7 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac .with_document(span_id as u32) .set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due: now, + due: TaskEpoch::now(), index: SearchIndex::Tracing, is_insert: true, }), @@ -244,16 +241,17 @@ pub fn build_span_document( index_fields: &AHashSet, ) -> IndexDocument { let mut document = IndexDocument::new(SearchIndex::Tracing).with_id(span_id); + let mut keywords = HashSet::new(); - for event in events { - for (idx, (key, value)) in event.keys.into_iter().enumerate() { - if idx == 0 - && (index_fields.is_empty() - || index_fields.contains(&TracingSearchField::EventType.into())) - { - document.index_unsigned(TracingSearchField::EventType, event.inner.typ.code()); - } + for (idx, event) in events.into_iter().enumerate() { + if idx == 0 + && (index_fields.is_empty() + || index_fields.contains(&TracingSearchField::EventType.into())) + { + document.index_unsigned(TracingSearchField::EventType, event.inner.typ.code()); + } + for (key, value) in event.keys { match (key, value) { (Key::QueueId, Value::UInt(queue_id)) => { if index_fields.is_empty() @@ -266,11 +264,7 @@ pub fn build_span_document( if index_fields.is_empty() || index_fields.contains(&TracingSearchField::Keywords.into()) { - document.index_text( - TracingSearchField::Keywords, - &address, - Language::Unknown, - ); + keywords.insert(address.to_string()); } } (Key::To, Value::Array(value)) => { @@ -279,11 +273,7 @@ pub fn build_span_document( { for value in value { if let Value::String(address) = value { - document.index_text( - TracingSearchField::Keywords, - &address, - Language::Unknown, - ); + keywords.insert(address.to_string()); } } } @@ -292,22 +282,14 @@ pub fn build_span_document( if index_fields.is_empty() || index_fields.contains(&TracingSearchField::Keywords.into()) { - document.index_text( - TracingSearchField::Keywords, - &ip.to_string(), - Language::Unknown, - ); + keywords.insert(ip.to_string()); } } (Key::RemoteIp, Value::Ipv6(ip)) => { if index_fields.is_empty() || index_fields.contains(&TracingSearchField::Keywords.into()) { - document.index_text( - TracingSearchField::Keywords, - &ip.to_string(), - Language::Unknown, - ); + keywords.insert(ip.to_string()); } } @@ -316,5 +298,17 @@ pub fn build_span_document( } } + if !keywords.is_empty() { + let mut keyword_str = String::new(); + for keyword in keywords { + if !keyword_str.is_empty() { + keyword_str.push(' '); + } + keyword_str.push_str(&keyword); + } + + document.index_keyword(TracingSearchField::Keywords, keyword_str); + } + document } diff --git a/crates/email/src/message/copy.rs b/crates/email/src/message/copy.rs index a3203f66..285747f6 100644 --- a/crates/email/src/message/copy.rs +++ b/crates/email/src/message/copy.rs @@ -18,7 +18,7 @@ use crate::{ use common::{Server, auth::ResourceToken, storage::index::ObjectIndexBuilder}; use mail_parser::{HeaderName, HeaderValue, parsers::fields::thread::thread_name}; use store::write::{ - BatchBuilder, IndexPropertyClass, SearchIndex, TaskQueueClass, ValueClass, now, + BatchBuilder, IndexPropertyClass, SearchIndex, TaskEpoch, TaskQueueClass, ValueClass, }; use trc::AddContext; use types::{ @@ -179,7 +179,6 @@ impl EmailCopy for Server { .log_container_insert(SyncCollection::Thread); document_id }; - let due = now(); batch .with_collection(Collection::Email) .with_document(document_id) @@ -201,7 +200,7 @@ impl EmailCopy for Server { .set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { index: SearchIndex::Email, - due, + due: TaskEpoch::now(), is_insert: true, }), vec![], @@ -210,7 +209,9 @@ impl EmailCopy for Server { // Merge threads if necessary if let Some(merge_threads) = MergeThreadIds::new(thread_result).serialize() { batch.set( - ValueClass::TaskQueue(TaskQueueClass::MergeThreads { due }), + ValueClass::TaskQueue(TaskQueueClass::MergeThreads { + due: TaskEpoch::now(), + }), merge_threads, ); } diff --git a/crates/email/src/message/delete.rs b/crates/email/src/message/delete.rs index f0319aa6..bdd1108f 100644 --- a/crates/email/src/message/delete.rs +++ b/crates/email/src/message/delete.rs @@ -12,7 +12,7 @@ use groupware::calendar::storage::ItipAutoExpunge; use std::future::Future; use store::rand::prelude::SliceRandom; use store::write::key::DeserializeBigEndian; -use store::write::{IndexPropertyClass, SearchIndex, TaskQueueClass, now}; +use store::write::{IndexPropertyClass, SearchIndex, TaskEpoch, TaskQueueClass, now}; use store::{IterateParams, SerializeInfallible, U32_LEN, U64_LEN, ValueKey}; use store::{ roaring::RoaringBitmap, @@ -54,7 +54,6 @@ impl EmailDeletion for Server { batch: &mut BatchBuilder, document_ids: RoaringBitmap, ) -> trc::Result { - let due = now(); let mut deleted_ids = RoaringBitmap::new(); batch .with_account_id(account_id) @@ -81,7 +80,7 @@ impl EmailDeletion for Server { .set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { index: SearchIndex::Email, - due, + due: TaskEpoch::now(), is_insert: false, }), 0u64.serialize(), @@ -359,113 +358,4 @@ impl EmailDeletion for Server { Ok(()) } - - /*async fn emails_purge_tombstoned(&self, account_id: u32) -> trc::Result<()> { - // Obtain tombstoned messages - let tombstoned_ids = self - .core - .storage - .data - .get_bitmap(BitmapKey { - account_id, - collection: Collection::Email.into(), - class: BitmapClass::Tag { - field: EmailField::MailboxIds.into(), - value: TagValue::Id(TOMBSTONE_ID), - }, - document_id: 0, - }) - .await? - .unwrap_or_default(); - - if tombstoned_ids.is_empty() { - return Ok(()); - } - - trc::event!( - Purge(trc::PurgeEvent::TombstoneCleanup), - AccountId = account_id, - Total = tombstoned_ids.len(), - ); - - // Delete full-text index - self.core - .storage - .fts - .remove(account_id, Collection::Email, &tombstoned_ids) - .await?; - - // Obtain tenant id - let tenant_id = self - .get_access_token(account_id) - .await - .caused_by(trc::location!())? - .tenant - .map(|t| t.id); - - // Delete messages - let mut batch = BatchBuilder::new(); - batch.with_account_id(account_id); - - for document_id in tombstoned_ids { - batch - .with_collection(Collection::Email) - .delete_document(document_id) - .clear(EmailField::Archive) - .untag(EmailField::MailboxIds, TagValue::Id(TOMBSTONE_ID)); - - // Remove message metadata - if let Some(metadata_) = self - .core - .storage - .data - .get_value::>(ValueKey { - account_id, - collection: Collection::Email.into(), - document_id, - class: ValueClass::Property(EmailField::Metadata.into()), - }) - .await? - { - let metadata = metadata_ - .unarchive::() - .caused_by(trc::location!())?; - - // SPDX-SnippetBegin - // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC - // SPDX-License-Identifier: LicenseRef-SEL - - // Hold blob for undeletion - #[cfg(feature = "enterprise")] - self.core.hold_undelete( - &mut batch, - Collection::Email.into(), - &BlobHash::from(&metadata.blob_hash), - u32::from(metadata.size) as usize, - ); - - // SPDX-SnippetEnd - - // Delete message - metadata - .index(&mut batch, account_id, tenant_id, false) - .caused_by(trc::location!())?; - - // Commit point - batch.commit_point(); - } else { - trc::event!( - Purge(trc::PurgeEvent::Error), - AccountId = account_id, - DocumentId = document_id, - Reason = "Failed to fetch message metadata.", - CausedBy = trc::location!(), - ); - } - } - - self.commit_batch(batch).await?; - - Ok(()) - }*/ } diff --git a/crates/email/src/message/ingest.rs b/crates/email/src/message/ingest.rs index b465fe52..e40a2f5c 100644 --- a/crates/email/src/message/ingest.rs +++ b/crates/email/src/message/ingest.rs @@ -33,7 +33,7 @@ use store::{ IndexKeyPrefix, IterateParams, U32_LEN, ValueKey, ahash::{AHashMap, AHashSet}, write::{ - BatchBuilder, IndexPropertyClass, SearchIndex, TaskQueueClass, ValueClass, + BatchBuilder, IndexPropertyClass, SearchIndex, TaskEpoch, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now, }, }; @@ -623,7 +623,6 @@ impl EmailIngest for Server { .log_container_insert(SyncCollection::Thread); document_id }; - let due = now(); batch .with_collection(Collection::Email) @@ -651,7 +650,7 @@ impl EmailIngest for Server { .set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { index: SearchIndex::Email, - due, + due: TaskEpoch::now(), is_insert: true, }), vec![], @@ -660,7 +659,9 @@ impl EmailIngest for Server { // Merge threads if necessary if let Some(merge_threads) = MergeThreadIds::new(thread_result).serialize() { batch.set( - ValueClass::TaskQueue(TaskQueueClass::MergeThreads { due }), + ValueClass::TaskQueue(TaskQueueClass::MergeThreads { + due: TaskEpoch::now(), + }), merge_threads, ); } @@ -668,7 +669,10 @@ impl EmailIngest for Server { // Request spam training if let Some(learn_spam) = train_spam { batch.set( - ValueClass::TaskQueue(TaskQueueClass::BayesTrain { due, learn_spam }), + ValueClass::TaskQueue(TaskQueueClass::BayesTrain { + due: TaskEpoch::now(), + learn_spam, + }), vec![], ); } diff --git a/crates/groupware/src/calendar/index.rs b/crates/groupware/src/calendar/index.rs index 36120b3b..9ef6e392 100644 --- a/crates/groupware/src/calendar/index.rs +++ b/crates/groupware/src/calendar/index.rs @@ -378,15 +378,21 @@ impl ArchivedCalendarEvent { .filter(|e| e.component_type.is_scheduling_object()) { for entry in component.entries.iter() { - let (is_lang, field) = match entry.name { - ArchivedICalendarProperty::Summary => (true, CalendarSearchField::Title), + let (is_lang, is_keyword, field) = match entry.name { + ArchivedICalendarProperty::Summary => (true, false, CalendarSearchField::Title), ArchivedICalendarProperty::Description => { - (true, CalendarSearchField::Description) + (true, false, CalendarSearchField::Description) } - ArchivedICalendarProperty::Location => (false, CalendarSearchField::Location), - ArchivedICalendarProperty::Organizer => (false, CalendarSearchField::Owner), - ArchivedICalendarProperty::Attendee => (false, CalendarSearchField::Attendee), - ArchivedICalendarProperty::Uid => (false, CalendarSearchField::Uid), + ArchivedICalendarProperty::Location => { + (false, false, CalendarSearchField::Location) + } + ArchivedICalendarProperty::Organizer => { + (false, false, CalendarSearchField::Owner) + } + ArchivedICalendarProperty::Attendee => { + (false, false, CalendarSearchField::Attendee) + } + ArchivedICalendarProperty::Uid => (false, true, CalendarSearchField::Uid), _ => continue, }; let field = SearchField::Calendar(field); @@ -406,7 +412,7 @@ impl ArchivedCalendarEvent { _ => None, })) { - let value = value.strip_prefix("mailto:").unwrap_or(value); + let value = value.strip_prefix("mailto:").unwrap_or(value).trim(); let lang = if is_lang { detector.detect(value, MIN_LANGUAGE_SCORE); Language::Unknown @@ -414,7 +420,11 @@ impl ArchivedCalendarEvent { Language::None }; - document.index_text(field.clone(), value, lang); + if !is_keyword { + document.index_text(field.clone(), value, lang); + } else { + document.index_keyword(field.clone(), value); + } } } } diff --git a/crates/groupware/src/calendar/storage.rs b/crates/groupware/src/calendar/storage.rs index 1cf79ef5..ef20c0d4 100644 --- a/crates/groupware/src/calendar/storage.rs +++ b/crates/groupware/src/calendar/storage.rs @@ -21,7 +21,7 @@ use store::{ IterateParams, U16_LEN, U32_LEN, U64_LEN, ValueKey, roaring::RoaringBitmap, write::{ - Archive, BatchBuilder, IndexPropertyClass, TaskQueueClass, ValueClass, + Archive, BatchBuilder, IndexPropertyClass, TaskEpoch, TaskQueueClass, ValueClass, key::{DeserializeBigEndian, KeySerializer}, now, }, @@ -520,7 +520,7 @@ impl CalendarAlarm { } => { batch.set( ValueClass::TaskQueue(TaskQueueClass::SendAlarm { - due: self.alarm_time as u64, + due: TaskEpoch::new(self.alarm_time as u64), event_id: self.event_id, alarm_id: self.alarm_id, is_email_alert: true, @@ -536,7 +536,7 @@ impl CalendarAlarm { CalendarAlarmType::Display { recurrence_id } => { batch.set( ValueClass::TaskQueue(TaskQueueClass::SendAlarm { - due: self.alarm_time as u64, + due: TaskEpoch::new(self.alarm_time as u64), event_id: self.event_id, alarm_id: self.alarm_id, is_email_alert: false, @@ -551,7 +551,7 @@ impl CalendarAlarm { pub fn delete_task(&self, batch: &mut BatchBuilder) { batch.clear(ValueClass::TaskQueue(TaskQueueClass::SendAlarm { - due: self.alarm_time as u64, + due: TaskEpoch::new(self.alarm_time as u64), event_id: self.event_id, alarm_id: self.alarm_id, is_email_alert: matches!(self.typ, CalendarAlarmType::Email { .. }), diff --git a/crates/groupware/src/contact/index.rs b/crates/groupware/src/contact/index.rs index b3963c85..809ea595 100644 --- a/crates/groupware/src/contact/index.rs +++ b/crates/groupware/src/contact/index.rs @@ -273,20 +273,20 @@ impl ArchivedContactCard { let mut detector = LanguageDetector::new(); for entry in self.card.entries.iter() { - let (is_text, field) = match entry.name { - ArchivedVCardProperty::N => (false, ContactSearchField::Name), - ArchivedVCardProperty::Nickname => (false, ContactSearchField::Nickname), - ArchivedVCardProperty::Org => (false, ContactSearchField::Organization), - ArchivedVCardProperty::Email => (false, ContactSearchField::Email), - ArchivedVCardProperty::Tel => (false, ContactSearchField::Phone), + let (is_text, is_keyword, field) = match entry.name { + ArchivedVCardProperty::N => (false, false, ContactSearchField::Name), + ArchivedVCardProperty::Nickname => (false, false, ContactSearchField::Nickname), + ArchivedVCardProperty::Org => (false, false, ContactSearchField::Organization), + ArchivedVCardProperty::Email => (false, false, ContactSearchField::Email), + ArchivedVCardProperty::Tel => (false, false, ContactSearchField::Phone), ArchivedVCardProperty::Impp | ArchivedVCardProperty::Socialprofile => { - (false, ContactSearchField::OnlineService) + (false, false, ContactSearchField::OnlineService) } - ArchivedVCardProperty::Adr => (false, ContactSearchField::Address), - ArchivedVCardProperty::Note => (true, ContactSearchField::Note), - ArchivedVCardProperty::Kind => (false, ContactSearchField::Kind), - ArchivedVCardProperty::Uid => (false, ContactSearchField::Uid), - ArchivedVCardProperty::Member => (false, ContactSearchField::Member), + ArchivedVCardProperty::Adr => (false, false, ContactSearchField::Address), + ArchivedVCardProperty::Note => (true, false, ContactSearchField::Note), + ArchivedVCardProperty::Kind => (false, true, ContactSearchField::Kind), + ArchivedVCardProperty::Uid => (false, true, ContactSearchField::Uid), + ArchivedVCardProperty::Member => (false, false, ContactSearchField::Member), _ => continue, }; let field = SearchField::Contact(field); @@ -295,28 +295,32 @@ impl ArchivedContactCard { for value in entry.values.iter() { match value { ArchivedVCardValue::Text(v) => { - let lang = if is_text { - detector.detect(v.as_str(), MIN_LANGUAGE_SCORE); - Language::Unknown - } else { - Language::None - }; + if !is_keyword { + let lang = if is_text { + detector.detect(v.as_str().trim(), MIN_LANGUAGE_SCORE); + Language::Unknown + } else { + Language::None + }; - document.index_text(field.clone(), v, lang); + document.index_text(field.clone(), v, lang); + } else { + document.index_keyword(field.clone(), v.as_str()); + } } ArchivedVCardValue::Kind(v) => { - document.index_text(field.clone(), v.as_str(), Language::None); + document.index_keyword(field.clone(), v.as_str()); } ArchivedVCardValue::Component(v) => { for item in v.iter() { - document.index_text(field.clone(), item, Language::None); + document.index_text(field.clone(), item.trim(), Language::None); } } _ => (), } } - for param in entry.params.iter() { + /*for param in entry.params.iter() { if let ArchivedVCardParameterValue::Text(value) = ¶m.value { let lang = if is_text { detector.detect(value.as_str(), MIN_LANGUAGE_SCORE); @@ -326,7 +330,7 @@ impl ArchivedContactCard { }; document.index_text(field.clone(), value, lang); } - } + }*/ } } diff --git a/crates/groupware/src/scheduling/itip.rs b/crates/groupware/src/scheduling/itip.rs index 48909596..7b1b4c03 100644 --- a/crates/groupware/src/scheduling/itip.rs +++ b/crates/groupware/src/scheduling/itip.rs @@ -16,7 +16,7 @@ use calcard::{ use common::PROD_ID; use store::{ Serialize, - write::{Archiver, BatchBuilder, TaskQueueClass, ValueClass, now}, + write::{Archiver, BatchBuilder, TaskEpoch, TaskQueueClass, ValueClass}, }; use trc::AddContext; @@ -278,7 +278,7 @@ impl ItipMessages { } pub fn queue(self, batch: &mut BatchBuilder) -> trc::Result<()> { - let due = now(); + let due = TaskEpoch::now().with_random_sequence_id(); batch.set( ValueClass::TaskQueue(TaskQueueClass::SendImip { due, diff --git a/crates/http/src/management/enterprise/telemetry.rs b/crates/http/src/management/enterprise/telemetry.rs index e9e12b1e..9347bbe3 100644 --- a/crates/http/src/management/enterprise/telemetry.rs +++ b/crates/http/src/management/enterprise/telemetry.rs @@ -78,7 +78,7 @@ impl TelemetryApi for Server { let mut tracing_query = Vec::new(); tracing_query.push(SearchFilter::And); if let Some(typ) = params.parse::("type") { - tracing_query.push(SearchFilter::eq(TracingSearchField::EventType, typ.id())); + tracing_query.push(SearchFilter::eq(TracingSearchField::EventType, typ.code())); } if let Some(queue_id) = params.parse::("queue_id") { tracing_query.push(SearchFilter::eq(TracingSearchField::QueueId, queue_id)); @@ -91,7 +91,7 @@ impl TelemetryApi for Server { if in_quote { buf.push(' '); } else if !buf.is_empty() { - tracing_query.push(SearchFilter::has_unknown_text( + tracing_query.push(SearchFilter::has_keyword( TracingSearchField::Keywords, buf, )); @@ -101,7 +101,7 @@ impl TelemetryApi for Server { buf.push(ch); if in_quote { if !buf.is_empty() { - tracing_query.push(SearchFilter::has_unknown_text( + tracing_query.push(SearchFilter::has_keyword( TracingSearchField::Keywords, buf, )); @@ -116,10 +116,8 @@ impl TelemetryApi for Server { } } if !buf.is_empty() { - tracing_query.push(SearchFilter::has_unknown_text( - TracingSearchField::Keywords, - buf, - )); + tracing_query + .push(SearchFilter::has_keyword(TracingSearchField::Keywords, buf)); } } let values = params.get("values").is_some(); diff --git a/crates/imap/src/op/copy_move.rs b/crates/imap/src/op/copy_move.rs index 37e6d458..61dc88fc 100644 --- a/crates/imap/src/op/copy_move.rs +++ b/crates/imap/src/op/copy_move.rs @@ -27,7 +27,7 @@ use imap_proto::{ use std::{sync::Arc, time::Instant}; use store::{ roaring::RoaringBitmap, - write::{AlignedBytes, Archive, BatchBuilder, TaskQueueClass, ValueClass, now}, + write::{AlignedBytes, Archive, BatchBuilder, TaskEpoch, TaskQueueClass, ValueClass}, }; use types::{ acl::Acl, @@ -317,7 +317,7 @@ impl SessionData { if dest_mailbox_id.mailbox_id == JUNK_ID { batch.set( ValueClass::TaskQueue(TaskQueueClass::BayesTrain { - due: now(), + due: TaskEpoch::now(), learn_spam: true, }), vec![], @@ -328,7 +328,7 @@ impl SessionData { { batch.set( ValueClass::TaskQueue(TaskQueueClass::BayesTrain { - due: now(), + due: TaskEpoch::now(), learn_spam: false, }), vec![], diff --git a/crates/imap/src/op/expunge.rs b/crates/imap/src/op/expunge.rs index 1978e082..3f4d0e7b 100644 --- a/crates/imap/src/op/expunge.rs +++ b/crates/imap/src/op/expunge.rs @@ -22,7 +22,7 @@ use std::{sync::Arc, time::Instant}; use store::{ SerializeInfallible, roaring::RoaringBitmap, - write::{BatchBuilder, SearchIndex, TaskQueueClass, ValueClass, now}, + write::{BatchBuilder, SearchIndex, TaskEpoch, TaskQueueClass, ValueClass}, }; use trc::AddContext; use types::{ @@ -154,6 +154,7 @@ impl SessionData { .commit_batch(batch) .await .caused_by(trc::location!())?; + self.server.notify_task_queue(); } Ok(()) @@ -169,7 +170,6 @@ impl SessionData { batch .with_account_id(account_id) .with_collection(Collection::Email); - let due = now(); self.server .archives( @@ -197,7 +197,7 @@ impl SessionData { .set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { index: SearchIndex::Email, - due, + due: TaskEpoch::now(), is_insert: false, }), 0u64.serialize(), diff --git a/crates/imap/src/op/store.rs b/crates/imap/src/op/store.rs index f5b659c1..0fba7800 100644 --- a/crates/imap/src/op/store.rs +++ b/crates/imap/src/op/store.rs @@ -25,7 +25,7 @@ use imap_proto::{ use std::{sync::Arc, time::Instant}; use store::{ query::log::{Change, Query}, - write::{BatchBuilder, TaskQueueClass, ValueClass, now}, + write::{BatchBuilder, TaskEpoch, TaskQueueClass, ValueClass}, }; use trc::AddContext; use types::{ @@ -304,7 +304,7 @@ impl SessionData { if let Some(learn_spam) = train_spam { batch.set( ValueClass::TaskQueue(TaskQueueClass::BayesTrain { - due: now(), + due: TaskEpoch::now(), learn_spam, }), vec![], diff --git a/crates/jmap-proto/src/method/query.rs b/crates/jmap-proto/src/method/query.rs index ec6d1d65..cc094478 100644 --- a/crates/jmap-proto/src/method/query.rs +++ b/crates/jmap-proto/src/method/query.rs @@ -70,7 +70,7 @@ where Close, } -#[derive(Debug, Clone, PartialEq, Eq, Default)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct Comparator where T: for<'de> DeserializeArguments<'de> + Default, @@ -358,3 +358,16 @@ where } } } + +impl Default for Comparator +where + T: for<'de> DeserializeArguments<'de> + Default, +{ + fn default() -> Self { + Self { + is_ascending: true, + collation: None, + property: T::default(), + } + } +} diff --git a/crates/jmap/src/calendar_event/query.rs b/crates/jmap/src/calendar_event/query.rs index 261e3f88..658d2c6f 100644 --- a/crates/jmap/src/calendar_event/query.rs +++ b/crates/jmap/src/calendar_event/query.rs @@ -49,7 +49,6 @@ impl CalendarEventQuery for Server { .await?; let default_tz = request.arguments.time_zone.unwrap_or(Tz::UTC); let mut filter: Option = None; - let mut did_filter_by_time = false; // Extract from/to arguments for cond in &request.filter { @@ -141,20 +140,34 @@ impl CalendarEventQuery for Server { Language::None, )); } - CalendarEventFilter::After(_) | CalendarEventFilter::Before(_) => { - if let Some(filter) = &filter - && !did_filter_by_time - { + CalendarEventFilter::After(after) => { + /* + The end of the event, or any recurrence of the event, in the time zone given + as the "timeZone" argument, must be after this date to match the condition. + */ + if let Some(after) = local_timestamp(&after, default_tz) { filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( cache.resources.iter().filter_map(|r| { - r.event_time_range().and_then(|(start, end)| { - filter - .is_in_range(false, start, end) - .then_some(r.document_id) + r.event_time_range() + .and_then(|(_, end)| (after < end).then_some(r.document_id)) + }), + ))); + } + } + CalendarEventFilter::Before(before) => { + /* + The start of the event, or any recurrence of the event, in the time zone given + as the "timeZone" argument, must be before this date to match the condition. + */ + + if let Some(before) = local_timestamp(&before, default_tz) { + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + cache.resources.iter().filter_map(|r| { + r.event_time_range().and_then(|(start, _)| { + (before > start).then_some(r.document_id) }) }), ))); - did_filter_by_time = true; } } unsupported => { @@ -225,117 +238,120 @@ impl CalendarEventQuery for Server { ) .await?; - let mut response = QueryResponseBuilder::new( - results.len(), - self.core.jmap.query_max_results, - cache.get_state(false), - &request, - ); + // Extract comparators + let comparators = request + .sort + .as_deref() + .filter(|s| !s.is_empty()) + .unwrap_or_default(); - if !results.is_empty() { - // Extract comparators - let comparators = request - .sort - .as_deref() - .filter(|s| !s.is_empty()) - .unwrap_or_default(); + if expand_recurrences && !results.is_empty() { + let Some(time_range) = filter.filter(|f| f.start != i64::MIN && f.end != i64::MAX) + else { + return Err(trc::JmapEvent::InvalidArguments.into_err().details( + "Both 'after' and 'before' filters are required when expanding recurrences", + )); + }; + let max_instances = self.core.groupware.max_ical_instances; + let mut expanded_results = Vec::with_capacity(results.len() as usize); + let has_uid_comparator = comparators + .iter() + .any(|c| matches!(c.property, CalendarEventComparator::Uid)); - if expand_recurrences { - let Some(time_range) = filter.filter(|f| f.start != i64::MIN && f.end != i64::MAX) + for document_id in results { + let Some(_calendar_event) = self + .archive(account_id, Collection::CalendarEvent, document_id) + .await? else { - return Err(trc::JmapEvent::InvalidArguments.into_err().details( - "Both 'after' and 'before' filters are required when expanding recurrences", - )); + continue; }; - let max_instances = self.core.groupware.max_ical_instances; - let mut expanded_results = Vec::with_capacity(results.len() as usize); - let has_uid_comparator = comparators - .iter() - .any(|c| matches!(c.property, CalendarEventComparator::Uid)); + let calendar_event = _calendar_event + .unarchive::() + .caused_by(trc::location!())?; - for document_id in results { - let Some(_calendar_event) = self - .archive(account_id, Collection::CalendarEvent, document_id) - .await? - else { - continue; - }; - let calendar_event = _calendar_event - .unarchive::() - .caused_by(trc::location!())?; - - // Expand recurrences - let uid = if has_uid_comparator { - Arc::new( - calendar_event - .data - .event - .uids() - .next() - .unwrap_or_default() - .to_string(), - ) + // Expand recurrences + let uid = if has_uid_comparator { + Arc::new( + calendar_event + .data + .event + .uids() + .next() + .unwrap_or_default() + .to_string(), + ) + } else { + Arc::new(String::new()) + }; + for expansion in calendar_event + .data + .expand(default_tz, time_range) + .unwrap_or_default() + { + if expanded_results.len() < max_instances { + expanded_results.push(SearchResult { + created: calendar_event.created.to_native().to_be_bytes(), + updated: calendar_event.modified.to_native().to_be_bytes(), + start: expansion.start.to_be_bytes(), + uid: uid.clone(), + document_id, + expansion_id: expansion.expansion_id.into(), + }); } else { - Arc::new(String::new()) - }; - for expansion in calendar_event - .data - .expand(default_tz, time_range) - .unwrap_or_default() - { - if expanded_results.len() < max_instances { - expanded_results.push(SearchResult { - created: calendar_event.created.to_native().to_be_bytes(), - updated: calendar_event.modified.to_native().to_be_bytes(), - start: expansion.start.to_be_bytes(), - uid: uid.clone(), - document_id, - expansion_id: expansion.expansion_id.into(), - }); + return Err(trc::JmapEvent::InvalidArguments.into_err().details( + "The number of expanded recurrences exceeds the server limit", + )); + } + } + } + + let mut response = QueryResponseBuilder::new( + expanded_results.len(), + self.core.jmap.query_max_results, + cache.get_state(false), + &request, + ); + // Sort results + if !expanded_results.is_empty() { + expanded_results.sort_by(|a, b| { + for comparator in comparators { + let ordering = if comparator.is_ascending { + a.get_property(&comparator.property) + .cmp(b.get_property(&comparator.property)) } else { - return Err(trc::JmapEvent::InvalidArguments.into_err().details( - "The number of expanded recurrences exceeds the server limit", - )); + b.get_property(&comparator.property) + .cmp(a.get_property(&comparator.property)) + }; + + if ordering != Ordering::Equal { + return ordering; } } - } + Ordering::Equal + }); - // Sort results - if !expanded_results.is_empty() { - expanded_results.sort_by(|a, b| { - for comparator in comparators { - let ordering = if comparator.is_ascending { - a.get_property(&comparator.property) - .cmp(b.get_property(&comparator.property)) - } else { - b.get_property(&comparator.property) - .cmp(a.get_property(&comparator.property)) - }; - - if ordering != Ordering::Equal { - return ordering; - } - } - Ordering::Equal - }); - - // Add results - for result in expanded_results { - if !response.add(result.expansion_id.unwrap() + 1, result.document_id) { - break; - } - } - } - } else { - for document_id in results { - if !response.add(0, document_id) { + // Add results + for result in expanded_results { + if !response.add(result.expansion_id.unwrap() + 1, result.document_id) { break; } } } + response.build() + } else { + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + cache.get_state(false), + &request, + ); + for document_id in results { + if !response.add(0, document_id) { + break; + } + } + response.build() } - - response.build() } } diff --git a/crates/jmap/src/contact/query.rs b/crates/jmap/src/contact/query.rs index 08c26e3f..e8c9d1e9 100644 --- a/crates/jmap/src/contact/query.rs +++ b/crates/jmap/src/contact/query.rs @@ -121,37 +121,31 @@ impl ContactCardQuery for Server { | ContactCardFilter::NameGiven(value) | ContactCardFilter::NameSurname(value) | ContactCardFilter::NameSurname2(value) => { - filters.push(SearchFilter::has_unknown_text( - ContactSearchField::Name, - value, - )); + filters.push(SearchFilter::has_keyword(ContactSearchField::Name, value)); } ContactCardFilter::Nickname(value) => { - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Nickname, value, )); } ContactCardFilter::Organization(value) => { - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Organization, value, )); } ContactCardFilter::Phone(value) => { - filters.push(SearchFilter::has_unknown_text( - ContactSearchField::Phone, - value, - )); + filters.push(SearchFilter::has_keyword(ContactSearchField::Phone, value)); } ContactCardFilter::OnlineService(value) => { - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::OnlineService, value, )); } ContactCardFilter::Address(value) => { - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Address, value, )); @@ -164,10 +158,7 @@ impl ContactCardQuery for Server { )); } ContactCardFilter::HasMember(value) => { - filters.push(SearchFilter::has_unknown_text( - ContactSearchField::Member, - value, - )); + filters.push(SearchFilter::has_keyword(ContactSearchField::Member, value)); } ContactCardFilter::Kind(value) => { filters.push(SearchFilter::eq(ContactSearchField::Kind, value)); @@ -175,39 +166,37 @@ impl ContactCardQuery for Server { ContactCardFilter::Uid(value) => { filters.push(SearchFilter::eq(ContactSearchField::Uid, value)) } - ContactCardFilter::Email(email) => { - filters.push(SearchFilter::has_unknown_text( - ContactSearchField::Email, - sanitize_email(&email).unwrap_or(email), - )) - } + ContactCardFilter::Email(email) => filters.push(SearchFilter::has_keyword( + ContactSearchField::Email, + sanitize_email(&email).unwrap_or(email), + )), ContactCardFilter::Text(value) => { filters.push(SearchFilter::Or); - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Name, value.clone(), )); - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Nickname, value.clone(), )); - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Organization, value.clone(), )); - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Email, value.clone(), )); - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Phone, value.clone(), )); - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::OnlineService, value.clone(), )); - filters.push(SearchFilter::has_unknown_text( + filters.push(SearchFilter::has_keyword( ContactSearchField::Address, value.clone(), )); diff --git a/crates/services/src/task_manager/imip.rs b/crates/services/src/task_manager/imip.rs index abb411a3..dce407fc 100644 --- a/crates/services/src/task_manager/imip.rs +++ b/crates/services/src/task_manager/imip.rs @@ -37,7 +37,7 @@ use store::{ ValueKey, ahash::AHashMap, rkyv::rend::{i16_le, i32_le}, - write::{AlignedBytes, Archive, TaskQueueClass, ValueClass, now}, + write::{AlignedBytes, Archive, TaskEpoch, TaskQueueClass, ValueClass, now}, }; use trc::AddContext; use utils::template::{Variable, Variables}; @@ -47,7 +47,7 @@ pub trait SendImipTask: Sync + Send { &self, account_id: u32, document_id: u32, - due: u64, + due: TaskEpoch, server_instance: Arc, ) -> impl Future + Send; } @@ -57,7 +57,7 @@ impl SendImipTask for Server { &self, account_id: u32, document_id: u32, - due: u64, + due: TaskEpoch, server_instance: Arc, ) -> bool { match send_imip(self, account_id, document_id, due, server_instance).await { @@ -79,7 +79,7 @@ async fn send_imip( server: &Server, account_id: u32, document_id: u32, - due: u64, + due: TaskEpoch, server_instance: Arc, ) -> trc::Result { // Obtain access token diff --git a/crates/services/src/task_manager/index.rs b/crates/services/src/task_manager/index.rs index 30270249..ccc5e32c 100644 --- a/crates/services/src/task_manager/index.rs +++ b/crates/services/src/task_manager/index.rs @@ -19,8 +19,8 @@ use store::{ roaring::RoaringBitmap, search::{IndexDocument, SearchField, SearchFilter, SearchQuery}, write::{ - BatchBuilder, BlobOp, SearchIndex, TaskQueueClass, ValueClass, key::DeserializeBigEndian, - now, + BatchBuilder, BlobOp, SearchIndex, TaskEpoch, TaskQueueClass, ValueClass, + key::DeserializeBigEndian, }, }; use trc::{AddContext, TaskQueueEvent}; @@ -203,7 +203,7 @@ impl SearchIndexTask for Server { .details("Failed to index documents") ); for r in results.iter_mut() { - if r.task_type == TaskType::Delete && r.status == TaskStatus::Success { + if r.task_type == TaskType::Insert && r.status == TaskStatus::Success { r.status = TaskStatus::Failed; } } @@ -301,7 +301,7 @@ impl ReindexIndexTask for Server { } accounts }; - let due = now(); + let due = TaskEpoch::now(); match index { SearchIndex::Email => { @@ -563,7 +563,7 @@ async fn delete_email_metadata( ) .await? { - Some(metadata) => { + Some(metadata_) => { let tenant_id = server .core .storage @@ -577,10 +577,25 @@ async fn delete_email_metadata( .with_account_id(account_id) .with_collection(Collection::Email) .with_document(document_id); - metadata + let metadata = metadata_ .unarchive::() - .caused_by(trc::location!())? - .unindex(batch, account_id, tenant_id); + .caused_by(trc::location!())?; + metadata.unindex(batch, account_id, tenant_id); + + // SPDX-SnippetBegin + // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + // SPDX-License-Identifier: LicenseRef-SEL + + // Hold blob for undeletion + #[cfg(feature = "enterprise")] + server.core.hold_undelete( + batch, + Collection::Email.into(), + &BlobHash::from(&metadata.blob_hash), + u32::from(metadata.size) as usize, + ); + + // SPDX-SnippetEnd } None => { trc::event!( diff --git a/crates/services/src/task_manager/lock.rs b/crates/services/src/task_manager/lock.rs index 95f737ea..befc429d 100644 --- a/crates/services/src/task_manager/lock.rs +++ b/crates/services/src/task_manager/lock.rs @@ -93,7 +93,7 @@ impl TaskLock for Task { fn lock_key(&self) -> Vec { KeySerializer::new((U32_LEN * 2) + U64_LEN + 2) .write(0u8) - .write(self.due) + .write(self.due.inner()) .write_leb128(self.account_id) .write_leb128(self.document_id) .write(self.action.index.to_u8()) @@ -162,7 +162,7 @@ impl TaskLock for Task { fn lock_key(&self) -> Vec { KeySerializer::new((U32_LEN * 2) + U64_LEN + 1) .write(2u8) - .write(self.due) + .write(self.due.inner()) .write_leb128(self.account_id) .write_leb128(self.document_id) .finalize() @@ -198,7 +198,7 @@ impl TaskLock for Task { fn lock_key(&self) -> Vec { KeySerializer::new((U32_LEN * 2) + U64_LEN + 1) .write(3u8) - .write(self.due) + .write(self.due.inner()) .write_leb128(self.account_id) .write_leb128(self.document_id) .finalize() @@ -210,17 +210,16 @@ impl TaskLock for Task { fn value_classes(&self) -> impl Iterator { [ - Some(ValueClass::TaskQueue(TaskQueueClass::SendImip { + ValueClass::TaskQueue(TaskQueueClass::SendImip { due: self.due, is_payload: false, - })), - Some(ValueClass::TaskQueue(TaskQueueClass::SendImip { + }), + ValueClass::TaskQueue(TaskQueueClass::SendImip { due: self.due, is_payload: true, - })), + }), ] .into_iter() - .flatten() } } @@ -240,7 +239,7 @@ impl TaskLock for Task>> { fn lock_key(&self) -> Vec { KeySerializer::new((U32_LEN * 2) + U64_LEN + 1) .write(4u8) - .write(self.due) + .write(self.due.inner()) .write_leb128(self.account_id) .write_leb128(self.document_id) .finalize() @@ -267,11 +266,11 @@ impl Task { } } - pub(crate) fn deserialize(key: &[u8], value: &[u8]) -> trc::Result { + pub fn deserialize(key: &[u8], value: &[u8]) -> trc::Result { let document_id = key.deserialize_be_u32(U64_LEN + U32_LEN + 1)?; Ok(Task { - due: key.deserialize_be_u64(0)?, + due: TaskEpoch::from_inner(key.deserialize_be_u64(0)?), account_id: key.deserialize_be_u32(U64_LEN)?, document_id, action: match key.get(U64_LEN + U32_LEN) { diff --git a/crates/services/src/task_manager/mod.rs b/crates/services/src/task_manager/mod.rs index e71ab14c..726bce0a 100644 --- a/crates/services/src/task_manager/mod.rs +++ b/crates/services/src/task_manager/mod.rs @@ -24,7 +24,7 @@ use std::{sync::Arc, time::Instant}; use store::ahash::AHashSet; use store::rand; use store::rand::seq::SliceRandom; -use store::write::SearchIndex; +use store::write::{SearchIndex, TaskEpoch}; use store::{ IterateParams, U16_LEN, U32_LEN, U64_LEN, ValueKey, ahash::AHashMap, @@ -49,12 +49,12 @@ pub mod merge_threads; pub struct Task { pub account_id: u32, pub document_id: u32, - pub due: u64, + pub due: TaskEpoch, pub action: T, } #[derive(Debug, Clone, Hash, PartialEq, Eq)] -pub(crate) enum TaskAction { +pub enum TaskAction { UpdateIndex(IndexAction), BayesTrain(bool), SendAlarm(CalendarAlarm), @@ -63,7 +63,7 @@ pub(crate) enum TaskAction { } #[derive(Debug, Clone, Hash, PartialEq, Eq)] -pub(crate) struct IndexAction { +pub struct IndexAction { pub index: SearchIndex, pub is_insert: bool, } @@ -370,7 +370,7 @@ impl TaskQueueManager for Server { collection: 0, document_id: 0, class: ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due: 0, + due: TaskEpoch::from_inner(0), index: SearchIndex::Email, is_insert: true, }), @@ -380,7 +380,9 @@ impl TaskQueueManager for Server { collection: u8::MAX, document_id: u32::MAX, class: ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due: now_timestamp + QUEUE_REFRESH_INTERVAL, + due: TaskEpoch::new(now_timestamp + QUEUE_REFRESH_INTERVAL) + .with_attempt(u16::MAX) + .with_sequence_id(u16::MAX), index: SearchIndex::Email, is_insert: true, }), @@ -397,7 +399,9 @@ impl TaskQueueManager for Server { IterateParams::new(from_key, to_key).ascending(), |key, value| { let task = Task::deserialize(key, value)?; - if task.due <= now_timestamp { + + let task_due = task.due.due(); + if task_due <= now_timestamp { match ipc.locked.entry(key.to_vec()) { Entry::Occupied(mut entry) => { let locked = entry.get_mut(); @@ -420,7 +424,7 @@ impl TaskQueueManager for Server { Ok(true) } else { - next_event = Some(task.due); + next_event = Some(task_due); Ok(false) } }, diff --git a/crates/store/src/backend/postgres/blob.rs b/crates/store/src/backend/postgres/blob.rs index decacfe7..66d437c0 100644 --- a/crates/store/src/backend/postgres/blob.rs +++ b/crates/store/src/backend/postgres/blob.rs @@ -6,6 +6,8 @@ use std::ops::Range; +use crate::backend::postgres::into_pool_error; + use super::{PostgresStore, into_error}; impl PostgresStore { @@ -14,7 +16,7 @@ impl PostgresStore { key: &[u8], range: Range, ) -> trc::Result>> { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn .prepare_cached("SELECT v FROM t WHERE k = $1") .await @@ -40,7 +42,7 @@ impl PostgresStore { } pub(crate) async fn put_blob(&self, key: &[u8], data: &[u8]) -> trc::Result<()> { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn .prepare_cached( "INSERT INTO t (k, v) VALUES ($1, $2) ON CONFLICT (k) DO UPDATE SET v = EXCLUDED.v", @@ -54,7 +56,7 @@ impl PostgresStore { } pub(crate) async fn delete_blob(&self, key: &[u8]) -> trc::Result { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn .prepare_cached("DELETE FROM t WHERE k = $1") .await diff --git a/crates/store/src/backend/postgres/lookup.rs b/crates/store/src/backend/postgres/lookup.rs index c8d90d44..71374798 100644 --- a/crates/store/src/backend/postgres/lookup.rs +++ b/crates/store/src/backend/postgres/lookup.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{QueryResult, QueryType}; +use crate::{QueryResult, QueryType, backend::postgres::into_pool_error}; use bytes::BytesMut; use futures::{TryStreamExt, pin_mut}; @@ -20,7 +20,7 @@ impl PostgresStore { query: &str, params_: &[crate::Value<'_>], ) -> trc::Result { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn.prepare_cached(query).await.map_err(into_error)?; let params = params_ .iter() diff --git a/crates/store/src/backend/postgres/main.rs b/crates/store/src/backend/postgres/main.rs index bdf4262b..282fc06a 100644 --- a/crates/store/src/backend/postgres/main.rs +++ b/crates/store/src/backend/postgres/main.rs @@ -6,7 +6,7 @@ use super::{PostgresStore, into_error}; use crate::{ - backend::postgres::{PsqlSearchField, tls::MakeRustlsConnect}, + backend::postgres::{PsqlSearchField, into_pool_error, tls::MakeRustlsConnect}, search::{ CalendarSearchField, ContactSearchField, EmailSearchField, SearchableField, TracingSearchField, @@ -96,7 +96,7 @@ impl PostgresStore { } pub(crate) async fn create_storage_tables(&self) -> trc::Result<()> { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; for table in [ SUBSPACE_ACL, @@ -163,7 +163,7 @@ impl PostgresStore { } pub(crate) async fn create_search_tables(&self) -> trc::Result<()> { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; create_search_tables::(&conn).await?; create_search_tables::(&conn).await?; diff --git a/crates/store/src/backend/postgres/mod.rs b/crates/store/src/backend/postgres/mod.rs index 3a831f7c..121a0ad9 100644 --- a/crates/store/src/backend/postgres/mod.rs +++ b/crates/store/src/backend/postgres/mod.rs @@ -14,7 +14,6 @@ use crate::{ use ahash::AHashSet; use deadpool_postgres::Pool; use nlp::language::Language; -use std::fmt::Display; pub mod blob; pub mod lookup; @@ -30,7 +29,23 @@ pub struct PostgresStore { } #[inline(always)] -fn into_error(err: impl Display) -> trc::Error { +fn into_error(err: tokio_postgres::error::Error) -> trc::Error { + let mut local_err = trc::StoreEvent::PostgresqlError.reason(err.to_string()); + if let Some(db_err) = err.as_db_error() { + local_err = local_err.code(db_err.code().code().to_string()); + if let Some(detail) = db_err.detail() { + local_err = local_err.details(detail.to_string()); + } + + if let Some(hint) = db_err.hint() { + local_err = local_err.caused_by(hint.to_string()); + } + } + local_err +} + +#[inline(always)] +fn into_pool_error(err: deadpool::managed::PoolError) -> trc::Error { trc::StoreEvent::PostgresqlError.reason(err) } diff --git a/crates/store/src/backend/postgres/read.rs b/crates/store/src/backend/postgres/read.rs index d137c7e7..25c0a8d1 100644 --- a/crates/store/src/backend/postgres/read.rs +++ b/crates/store/src/backend/postgres/read.rs @@ -5,7 +5,10 @@ */ use super::{PostgresStore, into_error}; -use crate::{Deserialize, IterateParams, Key, ValueKey, write::ValueClass}; +use crate::{ + Deserialize, IterateParams, Key, ValueKey, backend::postgres::into_pool_error, + write::ValueClass, +}; use futures::{TryStreamExt, pin_mut}; impl PostgresStore { @@ -13,7 +16,7 @@ impl PostgresStore { where U: Deserialize + 'static, { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn .prepare_cached(&format!( "SELECT v FROM {} WHERE k = $1", @@ -39,7 +42,7 @@ impl PostgresStore { params: IterateParams, mut cb: impl for<'x> FnMut(&'x [u8], &'x [u8]) -> trc::Result + Sync + Send, ) -> trc::Result<()> { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let table = char::from(params.begin.subspace()); let begin = params.begin.serialize(0); let end = params.end.serialize(0); @@ -100,7 +103,7 @@ impl PostgresStore { let table = char::from(key.subspace()); let key = key.serialize(0); - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn .prepare_cached(&format!("SELECT v FROM {table} WHERE k = $1")) .await diff --git a/crates/store/src/backend/postgres/search.rs b/crates/store/src/backend/postgres/search.rs index 80f55053..af272564 100644 --- a/crates/store/src/backend/postgres/search.rs +++ b/crates/store/src/backend/postgres/search.rs @@ -5,7 +5,7 @@ */ use crate::{ - backend::postgres::{PostgresStore, PsqlSearchField, into_error}, + backend::postgres::{PostgresStore, PsqlSearchField, into_error, into_pool_error}, search::{ IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchOperator, SearchQuery, SearchValue, @@ -21,7 +21,7 @@ use tokio_postgres::{ impl PostgresStore { pub async fn index(&self, documents: Vec) -> trc::Result<()> { - let mut conn = self.conn_pool.get().await.map_err(into_error)?; + let mut conn = self.conn_pool.get().await.map_err(into_pool_error)?; let trx = conn .build_transaction() .isolation_level(IsolationLevel::ReadCommitted) @@ -121,10 +121,10 @@ impl PostgresStore { if !sort.is_empty() { build_sort(&mut query, sort); } - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn.prepare_cached(&query).await.map_err(into_error)?; - let c = println!("Executing search query: {}", query); + let c = println!("Executing search query: {} and values {:?}", query, params); conn.query(&s, params.as_slice()) .await @@ -140,7 +140,7 @@ impl PostgresStore { debug_assert!(!filter.filters.is_empty()); let mut query = format!("DELETE FROM {} ", filter.index.psql_table()); let params = self.build_filter(&mut query, &filter.filters); - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn.prepare_cached(&query).await.map_err(into_error)?; conn.execute(&s, params.as_slice()) diff --git a/crates/store/src/backend/postgres/write.rs b/crates/store/src/backend/postgres/write.rs index 59612b64..88b7f23d 100644 --- a/crates/store/src/backend/postgres/write.rs +++ b/crates/store/src/backend/postgres/write.rs @@ -7,6 +7,7 @@ use super::{PostgresStore, into_error}; use crate::{ IndexKey, Key, LogKey, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER, SUBSPACE_QUOTA, + backend::postgres::into_pool_error, write::{ AssignedIds, Batch, MAX_COMMIT_ATTEMPTS, MAX_COMMIT_TIME, MergeResult, Operation, ValueClass, ValueOp, @@ -27,7 +28,7 @@ enum CommitError { impl PostgresStore { pub(crate) async fn write(&self, mut batch: Batch<'_>) -> trc::Result { - let mut conn = self.conn_pool.get().await.map_err(into_error)?; + let mut conn = self.conn_pool.get().await.map_err(into_pool_error)?; let start = Instant::now(); let mut retry_count = 0; @@ -379,7 +380,7 @@ impl PostgresStore { } pub(crate) async fn purge_store(&self) -> trc::Result<()> { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; for subspace in [SUBSPACE_QUOTA, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER] { let s = conn @@ -396,7 +397,7 @@ impl PostgresStore { } pub(crate) async fn delete_range(&self, from: impl Key, to: impl Key) -> trc::Result<()> { - let conn = self.conn_pool.get().await.map_err(into_error)?; + let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn .prepare_cached(&format!( diff --git a/crates/store/src/dispatch/search.rs b/crates/store/src/dispatch/search.rs index c5da3608..40740859 100644 --- a/crates/store/src/dispatch/search.rs +++ b/crates/store/src/dispatch/search.rs @@ -83,7 +83,6 @@ impl SearchStore { .. } => {} SearchFilter::Operator { .. } => { - let mut internal_item = None; let mut depth = 0; let mut external = Vec::with_capacity(5); @@ -106,11 +105,61 @@ impl SearchStore { depth -= 1; external.push(item); } + SearchFilter::Operator { + field: SearchField::AccountId, + .. + } => {} SearchFilter::Operator { .. } => { external.push(item); } _ => { - internal_item = Some(item); + let mut new_filters = Vec::new(); + let mut pop_count = depth; + while pop_count > 0 { + let prev_item = external.pop().unwrap(); + if matches!( + prev_item, + SearchFilter::And | SearchFilter::Or | SearchFilter::Not + ) { + pop_count -= 1; + } + new_filters.push(prev_item); + } + let is_end = matches!(item, SearchFilter::End); + new_filters.push(item); + + if !is_end { + if logical_op.is_some() { + depth += 1; + } + for item in iter { + match item { + SearchFilter::And + | SearchFilter::Or + | SearchFilter::Not => { + depth += 1; + new_filters.push(item); + } + SearchFilter::End => { + depth -= 1; + new_filters.push(item); + } + SearchFilter::Operator { + field: SearchField::AccountId, + .. + } => {} + SearchFilter::Operator { .. } if depth == 0 => { + external.push(item); + } + _ => { + new_filters.push(item); + } + } + } + } else { + new_filters.extend(iter); + } + iter = new_filters.into_iter(); break; } } @@ -120,20 +169,6 @@ impl SearchStore { external.push(SearchFilter::End); } - let mut internal_items = Vec::with_capacity(depth * 2); - if depth > 0 { - while depth > 0 { - let item = external.pop().unwrap(); - if matches!( - item, - SearchFilter::And | SearchFilter::Or | SearchFilter::Not - ) { - depth -= 1; - } - internal_items.push(item); - } - } - // Add account id if external.len() == 1 { external.push(SearchFilter::Operator { @@ -158,11 +193,6 @@ impl SearchStore { .into_iter() .collect(), )); - filters.extend(internal_items); - - if let Some(item) = internal_item { - filters.push(item); - } } _ => { match &item { diff --git a/crates/store/src/dispatch/store.rs b/crates/store/src/dispatch/store.rs index 0fbeed21..e609e8b7 100644 --- a/crates/store/src/dispatch/store.rs +++ b/crates/store/src/dispatch/store.rs @@ -503,19 +503,11 @@ impl Store { self.delete_range( AnyKey { subspace, - key: &[0u8], + key: vec![0u8], }, AnyKey { subspace, - key: &[ - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - ], + key: vec![u8::MAX; 16], }, ) .await diff --git a/crates/store/src/search/document.rs b/crates/store/src/search/document.rs index 4d7c57d8..a5ac3461 100644 --- a/crates/store/src/search/document.rs +++ b/crates/store/src/search/document.rs @@ -69,6 +69,16 @@ impl IndexDocument { .insert(field.into(), SearchValue::Uint(value.into())); } + pub fn index_keyword(&mut self, field: impl Into, value: impl Into) { + self.fields.insert( + field.into(), + SearchValue::Text { + value: value.into(), + language: Language::None, + }, + ); + } + pub fn insert_key_value( &mut self, field: impl Into, @@ -218,11 +228,6 @@ impl SearchFilter { Self::has_text(field, text, Language::English) } - #[inline(always)] - pub fn has_unknown_text(field: impl Into, text: impl Into) -> Self { - Self::has_text(field, text, Language::Unknown) - } - #[inline(always)] pub fn has_keyword(field: impl Into, text: impl Into) -> Self { Self::has_text(field, text, Language::None) diff --git a/crates/store/src/search/fields.rs b/crates/store/src/search/fields.rs index a93be113..7cf7e263 100644 --- a/crates/store/src/search/fields.rs +++ b/crates/store/src/search/fields.rs @@ -104,14 +104,7 @@ impl SearchableField for CalendarSearchField { } fn is_text(&self) -> bool { - matches!( - self, - CalendarSearchField::Title - | CalendarSearchField::Description - | CalendarSearchField::Location - | CalendarSearchField::Owner - | CalendarSearchField::Attendee - ) + !self.is_indexed() } } @@ -145,17 +138,7 @@ impl SearchableField for ContactSearchField { } fn is_text(&self) -> bool { - matches!( - self, - ContactSearchField::Name - | ContactSearchField::Nickname - | ContactSearchField::Organization - | ContactSearchField::Email - | ContactSearchField::Phone - | ContactSearchField::OnlineService - | ContactSearchField::Address - | ContactSearchField::Note - ) + !self.is_indexed() } } diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 9bc1b5a9..6ca6c623 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -284,13 +284,13 @@ impl ValueClass { is_insert, due, } => serializer - .write(*due) + .write(due.inner()) .write(account_id) .write(if *is_insert { 7u8 } else { 8u8 }) .write(document_id) .write(index.to_u8()), TaskQueueClass::BayesTrain { due, learn_spam } => serializer - .write(*due) + .write(due.inner()) .write(account_id) .write(if *learn_spam { 1u8 } else { 2u8 }) .write(document_id), @@ -300,7 +300,7 @@ impl ValueClass { alarm_id, is_email_alert, } => serializer - .write(*due) + .write(due.inner()) .write(account_id) .write(if *is_email_alert { 3u8 } else { 6u8 }) .write(document_id) @@ -309,7 +309,7 @@ impl ValueClass { TaskQueueClass::SendImip { due, is_payload } => { if !*is_payload { serializer - .write(*due) + .write(due.inner()) .write(account_id) .write(4u8) .write(document_id) @@ -319,11 +319,11 @@ impl ValueClass { .write(account_id) .write(5u8) .write(document_id) - .write(*due) + .write(due.inner()) } } TaskQueueClass::MergeThreads { due } => serializer - .write(*due) + .write(due.inner()) .write(account_id) .write(9u8) .write(document_id), diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index 8ae98c69..b326f01c 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -223,29 +223,33 @@ pub enum SearchIndexId { #[derive(Debug, PartialEq, Clone, Eq, Hash)] pub enum TaskQueueClass { UpdateIndex { - due: u64, + due: TaskEpoch, index: SearchIndex, is_insert: bool, }, BayesTrain { - due: u64, + due: TaskEpoch, learn_spam: bool, }, SendAlarm { - due: u64, + due: TaskEpoch, event_id: u16, alarm_id: u16, is_email_alert: bool, }, SendImip { - due: u64, + due: TaskEpoch, is_payload: bool, }, MergeThreads { - due: u64, + due: TaskEpoch, }, } +#[derive(Debug, PartialEq, Clone, Copy, Eq, Hash)] +#[repr(transparent)] +pub struct TaskEpoch(pub(crate) u64); + #[derive(Debug, PartialEq, Clone, Copy, Eq, Hash)] pub enum SearchIndex { Email, @@ -713,3 +717,56 @@ impl AsRef<[Param]> for Params { &self.0 } } + +impl TaskEpoch { + /* + Structure of the 64-bit epoch: + 4 bytes: seconds since custom epoch (1632280000) + 2 bytes: attempt number + 2 bytes: sequence id + */ + + const EPOCH_OFFSET: u64 = 1632280000; + + pub fn now() -> Self { + Self::new(now()) + } + + pub fn new(timestamp: u64) -> Self { + Self(timestamp.saturating_sub(Self::EPOCH_OFFSET) << 32) + } + + pub fn with_attempt(mut self, attempt: u16) -> Self { + self.0 |= (attempt as u64) << 16; + self + } + + pub fn with_sequence_id(mut self, sequence_id: u16) -> Self { + self.0 |= sequence_id as u64; + self + } + + pub fn with_random_sequence_id(self) -> Self { + self.with_sequence_id(rand::random()) + } + + pub fn due(&self) -> u64 { + (self.0 >> 32) + Self::EPOCH_OFFSET + } + + pub fn attempt(&self) -> u16 { + (self.0 >> 16) as u16 + } + + pub fn sequence_id(&self) -> u16 { + self.0 as u16 + } + + pub fn inner(&self) -> u64 { + self.0 + } + + pub fn from_inner(inner: u64) -> Self { + Self(inner) + } +} diff --git a/crates/types/src/lib.rs b/crates/types/src/lib.rs index 690a6a76..422c41aa 100644 --- a/crates/types/src/lib.rs +++ b/crates/types/src/lib.rs @@ -27,6 +27,10 @@ pub struct TimeRange { } impl TimeRange { + pub fn new(start: i64, end: i64) -> Self { + Self { start, end } + } + pub fn is_in_range(&self, match_overlap: bool, start: i64, end: i64) -> bool { if !match_overlap { // RFC4791#9.9: (start < DTEND AND end > DTSTART) diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 26418855..e53f6b95 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -5,7 +5,7 @@ edition = "2024" [features] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "nats", "azure", "foundationdb", "enterprise"] -default = ["rocks"] +default = ["postgres"] sqlite = ["store/sqlite"] foundationdb = ["store/foundation", "common/foundation"] postgres = ["store/postgres"] diff --git a/tests/src/jmap/calendar/notification.rs b/tests/src/jmap/calendar/notification.rs index 2a4ed7de..d459508d 100644 --- a/tests/src/jmap/calendar/notification.rs +++ b/tests/src/jmap/calendar/notification.rs @@ -72,7 +72,7 @@ pub async fn test(params: &mut JMAPTest) { .await; let john_event_id = response.created(0).id().to_string(); - tokio::time::sleep(std::time::Duration::from_millis(200)).await; + tokio::time::sleep(std::time::Duration::from_millis(600)).await; // Verify Jane and Bill received the share notification let mut jane_event_id = String::new(); diff --git a/tests/src/jmap/contacts/contact.rs b/tests/src/jmap/contacts/contact.rs index 943affc3..1b02b86b 100644 --- a/tests/src/jmap/contacts/contact.rs +++ b/tests/src/jmap/contacts/contact.rs @@ -5,7 +5,7 @@ */ use crate::{ - jmap::{ChangeType, IntoJmapSet, JMAPTest, JmapUtils}, + jmap::{ChangeType, IntoJmapSet, JMAPTest, JmapUtils, wait_for_index}, webdav::DummyWebDavClient, }; use ahash::AHashSet; @@ -336,6 +336,7 @@ pub async fn test(params: &mut JMAPTest) { })); // Query tests + wait_for_index(¶ms.server).await; assert_eq!( account .jmap_query( diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 23634638..4c62ab67 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -40,7 +40,10 @@ use pop3::Pop3SessionManager; use reqwest::header; use serde::{Deserialize, Serialize, de::DeserializeOwned}; use serde_json::{Value, json}; -use services::SpawnServices; +use services::{ + SpawnServices, + task_manager::{Task, TaskAction}, +}; use smtp::{SpawnQueueManager, core::SmtpSessionManager}; use std::{ fmt::{Debug, Display}, @@ -48,7 +51,10 @@ use std::{ sync::Arc, time::Duration, }; -use store::{IterateParams, SUBSPACE_TASK_QUEUE, Stores, write::AnyKey}; +use store::{ + IterateParams, SUBSPACE_TASK_QUEUE, Stores, U32_LEN, U64_LEN, + write::{AnyKey, TaskEpoch, key::DeserializeBigEndian}, +}; use tokio::sync::watch; use types::id::Id; use utils::config::Config; @@ -71,7 +77,7 @@ async fn jmap_tests() { /*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; @@ -95,7 +101,7 @@ async fn jmap_tests() { auth::limits::test(&mut params).await; auth::oauth::test(&mut params).await; auth::quota::test(&mut params).await; - auth::permissions::test(¶ms).await;*/ + auth::permissions::test(¶ms).await; contacts::addressbook::test(&mut params).await; contacts::contact::test(&mut params).await; @@ -198,7 +204,7 @@ impl Account { pub async fn wait_for_index(server: &Server) { let mut count = 0; loop { - let mut has_index_tasks = false; + let mut has_index_tasks = None; server .core .storage @@ -211,12 +217,21 @@ pub async fn wait_for_index(server: &Server) { }, AnyKey { subspace: SUBSPACE_TASK_QUEUE, - key: vec![u8::MAX, u8::MAX, u8::MAX, u8::MAX], + key: vec![u8::MAX; 16], }, ) .ascending(), - |_, _| { - has_index_tasks = true; + |key, value| { + has_index_tasks = Some( + Task::::deserialize(key, value).unwrap_or_else(|_| Task { + due: TaskEpoch::from_inner( + key.deserialize_be_u64(key.len() - U64_LEN).unwrap(), + ), + account_id: key.deserialize_be_u32(U64_LEN).unwrap(), + document_id: key.deserialize_be_u32(U64_LEN + U32_LEN + 1).unwrap(), + action: TaskAction::SendImip, + }), + ); Ok(false) }, @@ -224,10 +239,10 @@ pub async fn wait_for_index(server: &Server) { .await .unwrap(); - if has_index_tasks { + if let Some(task) = has_index_tasks { count += 1; if count % 10 == 0 { - println!("Waiting for pending index tasks..."); + println!("Waiting for pending task {:?}...", task); } tokio::time::sleep(Duration::from_millis(300)).await; } else { @@ -314,6 +329,10 @@ async fn init_jmap_tests(delete_if_exists: bool) -> JMAPTest { cache, }); + if delete_if_exists { + store.destroy().await; + } + // Parse acceptors servers.parse_tcp_acceptors(&mut config, inner.clone()); @@ -361,10 +380,6 @@ async fn init_jmap_tests(delete_if_exists: bool) -> JMAPTest { }; }); - if delete_if_exists { - store.destroy().await; - } - // Create tables let server = inner.build_server(); let mut accounts = AHashMap::new(); diff --git a/tests/src/jmap/server/enterprise.rs b/tests/src/jmap/server/enterprise.rs index 5baa68d3..49266dff 100644 --- a/tests/src/jmap/server/enterprise.rs +++ b/tests/src/jmap/server/enterprise.rs @@ -16,6 +16,7 @@ use crate::{ JMAPTest, ManagementApi, mail::delivery::{AssertResult, SmtpConnection}, server::List, + wait_for_index, }, }; use common::{ @@ -258,9 +259,7 @@ async fn tracing(params: &mut JMAPTest) { SearchFilter::Operator { field: SearchField::Tracing(TracingSearchField::EventType), op: SearchOperator::Equal, - value: SearchValue::Uint( - EventType::Smtp(SmtpEvent::ConnectionStart).id() as u64 - ) + value: SearchValue::Uint(EventType::Smtp(SmtpEvent::ConnectionStart).code()) } ])) .await @@ -285,11 +284,14 @@ async fn tracing(params: &mut JMAPTest) { ) .await; lmtp.quit().await; - tokio::time::sleep(Duration::from_millis(200)).await; + tokio::time::sleep(Duration::from_millis(300)).await; + + params.server.notify_task_queue(); + wait_for_index(¶ms.server).await; // Purge should not delete anything at this point store - .purge_spans(Duration::from_secs(1), Some(&query)) + .purge_spans(Duration::from_secs(2), Some(&query)) .await .unwrap(); @@ -303,7 +305,7 @@ async fn tracing(params: &mut JMAPTest) { SearchFilter::Operator { field: SearchField::Tracing(TracingSearchField::EventType), op: SearchOperator::Equal, - value: SearchValue::Uint(span_type.id() as u64), + value: SearchValue::Uint(span_type.code()), }, ])) .await @@ -332,7 +334,7 @@ async fn tracing(params: &mut JMAPTest) { .unwrap(); assert_eq!(spans.len(), 2, "keyword: {keyword}"); - assert!(spans[0] > spans[1], "keyword: {keyword}"); + assert!(spans[0] != spans[1], "keyword: {keyword}"); } // Purge should delete the span entries @@ -377,7 +379,7 @@ async fn metrics(params: &mut JMAPTest) { ); } -async fn undelete(_params: &mut JMAPTest) { +async fn undelete(params: &mut JMAPTest) { // Authenticate let mut imap = ImapConnection::connect(b"_x ").await; imap.authenticate("jdoe@example.com", "12345").await; @@ -430,6 +432,7 @@ async fn undelete(_params: &mut JMAPTest) { api.get::("/api/store/purge/account/jdoe@example.com") .await .unwrap(); + wait_for_index(¶ms.server).await; tokio::time::sleep(Duration::from_millis(200)).await; let deleted = api .get::>>("/api/store/undelete/jdoe@example.com")