filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

CardDAV skeleton + Message ingestion performance improvements

Browse Source
This commit is contained in:
mdecimus
2025-03-28 17:25:30 +01:00
parent a5e6f77b26
commit 372f2bec70
43 changed files with 1007 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/email/src/message/copy.rs
View File

@@ -108,7 +108,7 @@ impl EmailCopy for Server {
| HeaderName::ResentMessageId => {
header.value.visit_text(|id| {
if !id.is_empty() && id.len() < MAX_ID_LENGTH {
references.push(id);
references.push(id.as_bytes());
}
});
}
@@ -128,7 +128,7 @@ impl EmailCopy for Server {
// Obtain threadId
let thread_id = self
.find_or_merge_thread(account_id, subject, &references)
.find_or_merge_thread(account_id, subject, references, None)
.await
.caused_by(trc::location!())?;

23
crates/email/src/message/index.rs
View File

@@ -116,13 +116,9 @@ impl MessageMetadata {
// Add ids to inverted index
if id.len() < MAX_ID_LENGTH {
if set {
batch
.index(Property::MessageId, id.serialize())
.index(Property::References, id.serialize());
batch.index(Property::References, encode_message_id(id));
} else {
batch
.unindex(Property::MessageId, id.serialize())
.unindex(Property::References, id.serialize());
batch.unindex(Property::References, encode_message_id(id));
}
}
});
@@ -229,6 +225,13 @@ impl MessageMetadata {
}
}
fn encode_message_id(message_id: &str) -> Vec<u8> {
let mut msg_id = Vec::with_capacity(message_id.len() + 1);
msg_id.extend_from_slice(message_id.as_bytes());
msg_id.push(0);
msg_id
}
impl ArchivedMessageMetadata {
#[inline(always)]
pub fn root_part(&self) -> &ArchivedMessageMetadataPart {
@@ -307,13 +310,9 @@ impl ArchivedMessageMetadata {
// Add ids to inverted index
if id.len() < MAX_ID_LENGTH {
if set {
batch
.index(Property::MessageId, id.serialize())
.index(Property::References, id.serialize());
batch.index(Property::References, encode_message_id(id));
} else {
batch
.unindex(Property::MessageId, id.serialize())
.unindex(Property::References, id.serialize());
batch.unindex(Property::References, encode_message_id(id));
}
}
});

110
crates/email/src/message/ingest.rs
View File

@@ -6,7 +6,6 @@
use std::{
borrow::Cow,
collections::BTreeSet,
fmt::Write,
time::{Duration, Instant},
};
@@ -37,7 +36,6 @@ use std::future::Future;
use store::{
BlobClass, IndexKey, IndexKeyPrefix, IterateParams, U32_LEN,
ahash::AHashMap,
query::Filter,
roaring::RoaringBitmap,
write::{
AlignedBytes, Archive, AssignedIds, BatchBuilder, MaybeDynamicValue, SerializeWithId,
@@ -106,7 +104,8 @@ pub trait EmailIngest: Sync + Send {
&self,
account_id: u32,
thread_name: &str,
references: &[&str],
references: Vec<&[u8]>,
skip_duplicate: Option<(&[u8], u32)>,
) -> impl Future<Output = trc::Result<u32>> + Send;
fn assign_imap_uid(
&self,
@@ -258,7 +257,7 @@ impl EmailIngest for Server {
}
// Obtain message references and thread name
let mut message_id = String::new();
let mut message_id = None;
let thread_id = {
let mut references = Vec::with_capacity(5);
let mut subject = "";
@@ -266,10 +265,11 @@ impl EmailIngest for Server {
match &header.name {
HeaderName::MessageId => header.value.visit_text(|id| {
if !id.is_empty() && id.len() < MAX_ID_LENGTH {
if message_id.is_empty() {
message_id = id.to_string();
// Used by find_or_merge_thread to skip duplicates
if params.source.is_smtp() && message_id.is_none() {
message_id = references.len().into();
}
references.push(id);
references.push(id.as_bytes());
}
}),
HeaderName::InReplyTo
@@ -277,7 +277,7 @@ impl EmailIngest for Server {
| HeaderName::ResentMessageId => {
header.value.visit_text(|id| {
if !id.is_empty() && id.len() < MAX_ID_LENGTH {
references.push(id);
references.push(id.as_bytes());
}
});
}
@@ -295,29 +295,19 @@ impl EmailIngest for Server {
}
}
// Check for duplicates
if params.source.is_smtp()
&& !message_id.is_empty()
&& !self
.core
.storage
.data
.filter(
account_id,
Collection::Email,
vec![
Filter::eq(Property::MessageId, message_id.as_str().serialize()),
Filter::is_in_bitmap(
Property::MailboxIds,
params.mailbox_ids.first().copied().unwrap_or(INBOX_ID),
),
],
)
.await
.caused_by(trc::location!())?
.results
.is_empty()
{
let skip_duplicate = message_id.map(|idx| {
(
references[idx],
params.mailbox_ids.first().copied().unwrap_or(INBOX_ID),
)
});
let thread_id = self
.find_or_merge_thread(account_id, subject, references, skip_duplicate)
.await?;
if thread_id != u32::MAX {
thread_id
} else {
// Duplicate message
trc::event!(
MessageIngest(MessageIngestEvent::Duplicate),
SpanId = params.session_id,
@@ -333,9 +323,6 @@ impl EmailIngest for Server {
size: 0,
});
}
self.find_or_merge_thread(account_id, subject, &references)
.await?
};
// Add additional headers to message
@@ -586,7 +573,8 @@ impl EmailIngest for Server {
&self,
account_id: u32,
thread_name: &str,
references: &[&str],
mut references: Vec<&[u8]>,
skip_duplicate: Option<(&[u8], u32)>,
) -> trc::Result<u32> {
if references.is_empty() {
return self.create_thread_id(account_id).await;
@@ -599,10 +587,9 @@ impl EmailIngest for Server {
"!"
}
.serialize();
let references = references
.iter()
.map(|r| r.as_bytes())
.collect::<BTreeSet<_>>();
// Sort references ascending
references.sort_unstable();
loop {
// Find messages with a matching subject
@@ -642,12 +629,15 @@ impl EmailIngest for Server {
)
.await
.caused_by(trc::location!())?;
// No matching subjects were found, skip early
if subj_results.is_empty() {
return self.create_thread_id(account_id).await;
}
// Find messages with matching references
let mut results = RoaringBitmap::new();
let mut found_message_id = Vec::new();
self.store()
.iterate(
IterateParams::new(
@@ -670,13 +660,27 @@ impl EmailIngest for Server {
.ascending(),
|key, _| {
let id_pos = key.len() - U32_LEN;
let value = key.get(IndexKeyPrefix::len()..id_pos).ok_or_else(|| {
trc::Error::corrupted_key(key, None, trc::location!())
})?;
let mut value =
key.get(IndexKeyPrefix::len()..id_pos).ok_or_else(|| {
trc::Error::corrupted_key(key, None, trc::location!())
})?;
let document_id = key.deserialize_be_u32(id_pos)?;
if subj_results.contains(document_id) && references.contains(value) {
if let Some(message_id) = value.strip_suffix(&[0]) {
value = message_id;
if skip_duplicate.is_some_and(|(message_id, _)| message_id == value) {
found_message_id.push(document_id);
}
}
if subj_results.contains(document_id)
&& references.binary_search(&value).is_ok()
{
results.insert(document_id);
if subj_results.len() == results.len() {
return Ok(false);
}
}
Ok(true)
@@ -684,10 +688,30 @@ impl EmailIngest for Server {
)
.await
.caused_by(trc::location!())?;
// No matching messages
if results.is_empty() {
return self.create_thread_id(account_id).await;
}
// Skip duplicate messages
if !found_message_id.is_empty() {
if let Some(ids) = self
.get_tag(
account_id,
Collection::Email,
Property::MailboxIds,
skip_duplicate.unwrap().1,
)
.await
.caused_by(trc::location!())?
{
if found_message_id.iter().any(|id| ids.contains(*id)) {
return Ok(u32::MAX);
}
}
}
// Find the most common threadId
let mut thread_counts = AHashMap::<u32, u32>::with_capacity(16);
let mut thread_id = u32::MAX;