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

Registry testing - part 11

Browse Source
This commit is contained in:
Maurus Decimus
2026-03-21 19:51:25 +01:00
parent 9b118bceef
commit 21c541e505
22 changed files with 519 additions and 464 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
crates/services/src/task_manager/index.rs
View File

@@ -585,10 +585,8 @@ async fn delete_email_metadata(
use store::{
SerializeInfallible,
write::{BlobLink, BlobOp, RegistryClass, now},
xxhash_rust,
};
use types::blob::BlobId;
use utils::snowflake::SnowflakeIdGenerator;
let root_part = metadata.root_part();
let from: Option<String> = root_part.headers.iter().find_map(|h| {
@@ -632,10 +630,7 @@ async fn delete_email_metadata(
})
.to_pickled_vec();
let object_id = ObjectType::ArchivedItem.to_id();
let item_id = SnowflakeIdGenerator::from_sequence_id(
xxhash_rust::xxh3::xxh3_64(item.as_slice()),
)
.unwrap_or_default();
let item_id = server.inner.data.registry_id_gen.generate();
batch
.set(

142
crates/services/src/task_manager/merge_threads.rs
View File

@@ -6,15 +6,18 @@
use crate::task_manager::TaskResult;
use common::{Server, storage::index::ObjectIndexBuilder};
use email::message::{ingest::ThreadMerge, metadata::MessageData};
use email::message::{
ingest::{ThreadMerge, has_message_id},
metadata::MessageData,
};
use registry::schema::structs::TaskMergeThreads;
use std::{str::FromStr, time::Duration};
use store::{
IndexKeyPrefix, IterateParams, U32_LEN, ValueKey,
ahash::{AHashMap, AHashSet},
IterateParams, Key, U32_LEN, ValueKey,
ahash::AHashMap,
rand::Rng,
write::{
AlignedBytes, Archive, BatchBuilder, IndexPropertyClass, ValueClass,
AlignedBytes, Archive, BatchBuilder, IndexPropertyClass, MergeResult, Params, ValueClass,
key::DeserializeBigEndian,
},
};
@@ -51,55 +54,70 @@ async fn merge_threads(
server: &Server,
task_merge_threads: &TaskMergeThreads,
) -> trc::Result<TaskResult> {
let Ok(thread_hash) = CheekyHash::from_str(&task_merge_threads.thread_hash) else {
let Ok(thread_hash) = CheekyHash::from_str(&task_merge_threads.thread_name) else {
return Ok(TaskResult::permanent("Invalid thread hash"));
};
let account_id = task_merge_threads.account_id.document_id();
let key_len = IndexKeyPrefix::len() + thread_hash.len() + U32_LEN;
let document_id_pos = key_len - U32_LEN;
let merge_thread_ids = task_merge_threads
.thread_ids
let Ok(mut message_ids) = task_merge_threads
.message_ids
.iter()
.map(|id| id.document_id())
.collect::<AHashSet<_>>();
let mut thread_merge = ThreadMerge::new();
let mut thread_index = AHashMap::new();
.map(|id| CheekyHash::from_str(id))
.collect::<Result<Vec<_>, _>>()
else {
return Ok(TaskResult::permanent("Invalid message ids"));
};
message_ids.sort_unstable();
let account_id = task_merge_threads.account_id.document_id();
let mut try_count = 0;
let from_key = ValueKey {
account_id,
collection: Collection::Email.into(),
document_id: 0,
class: ValueClass::IndexProperty(IndexPropertyClass::Hash {
property: EmailField::Threading.into(),
hash: thread_hash,
}),
};
let to_key = ValueKey {
account_id,
collection: Collection::Email.into(),
document_id: u32::MAX,
class: ValueClass::IndexProperty(IndexPropertyClass::Hash {
property: EmailField::Threading.into(),
hash: thread_hash,
}),
};
let mut prefix = from_key.serialize(0);
let key_len = prefix.len();
let document_id_pos = key_len - U32_LEN;
prefix.truncate(document_id_pos);
'retry: loop {
// Merge threads
let mut thread_merge = ThreadMerge::new();
let mut same_subject_messages: AHashMap<u32, Vec<u32>> = AHashMap::new();
// Find thread ids
server
.store()
.iterate(
IterateParams::new(
ValueKey {
account_id,
collection: Collection::Email.into(),
document_id: 0,
class: ValueClass::IndexProperty(IndexPropertyClass::Hash {
property: EmailField::Threading.into(),
hash: thread_hash,
}),
},
ValueKey {
account_id,
collection: Collection::Email.into(),
document_id: u32::MAX,
class: ValueClass::IndexProperty(IndexPropertyClass::Hash {
property: EmailField::Threading.into(),
hash: thread_hash,
}),
},
)
.ascending(),
IterateParams::new(from_key.clone(), to_key.clone()).ascending(),
|key, value| {
if key.len() == key_len {
if key.len() == key_len && key.starts_with(&prefix) {
// Find matching references
let references = value.get(U32_LEN..).unwrap_or_default();
let thread_id = value.deserialize_be_u32(0)?;
if merge_thread_ids.contains(&thread_id) {
let document_id = key.deserialize_be_u32(document_id_pos)?;
let document_id = key.deserialize_be_u32(document_id_pos)?;
if has_message_id(&message_ids, references) {
thread_merge.add(thread_id, document_id);
thread_index.insert(document_id, value.to_vec());
} else {
// Keep track of messages with the same subject for potential future merges
same_subject_messages
.entry(thread_id)
.or_default()
.push(document_id);
}
}
@@ -113,6 +131,17 @@ async fn merge_threads(
// Another process merged the threads already?
return Ok(TaskResult::Success);
}
// Add other messages with the same subject to the merge if they share a
// thread id with a message that has a matching message id
for thread_id in thread_merge.thread_ids().copied().collect::<Vec<_>>() {
if let Some(document_ids) = same_subject_messages.get(&thread_id) {
for &document_id in document_ids {
thread_merge.add(thread_id, document_id);
}
}
}
let thread_id = thread_merge.merge_thread_id();
// Delete all but the most common threadId
@@ -168,14 +197,41 @@ async fn merge_threads(
.caused_by(trc::location!())?;
// Update thread index property
let mut thread_index = thread_index.remove(&document_id).unwrap();
thread_index[0..U32_LEN].copy_from_slice(&thread_id.to_be_bytes());
batch.set(
batch.merge_fnc(
ValueClass::IndexProperty(IndexPropertyClass::Hash {
property: EmailField::Threading.into(),
hash: thread_hash,
}),
thread_index,
Params::with_capacity(3)
.with_u64(thread_id as u64)
.with_u64(group_thread_id as u64),
|params, _, bytes| {
let new_thread_id = params.u64(0) as u32;
let old_thread_id = params.u64(1) as u32;
let mut thread_index = bytes
.filter(|v| v.len() > U32_LEN)
.ok_or_else(|| {
trc::StoreEvent::AssertValueFailed
.into_err()
.details("Message no longer exists.")
.caused_by(trc::location!())
})?
.to_vec();
if thread_index.as_slice().deserialize_be_u32(0)? != old_thread_id {
return Err(
trc::StoreEvent::AssertValueFailed
.into_err()
.details("Thread id mismatch, likely due to concurrent modification.")
.caused_by(trc::location!())
);
}
thread_index[0..U32_LEN].copy_from_slice(&new_thread_id.to_be_bytes());
Ok(MergeResult::Update(thread_index))
},
);
}
}