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

Database schema optimization - part 11

Browse Source
This commit is contained in:
mdecimus
2025-11-12 19:09:06 +01:00
parent 3e181ae467
commit 8bbcb999d1
40 changed files with 541 additions and 493 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
crates/services/src/task_manager/imip.rs
View File

@@ -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<ServerInstance>,
) -> impl Future<Output = bool> + Send;
}
@@ -57,7 +57,7 @@ impl SendImipTask for Server {
&self,
account_id: u32,
document_id: u32,
due: u64,
due: TaskEpoch,
server_instance: Arc<ServerInstance>,
) -> 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<ServerInstance>,
) -> trc::Result<bool> {
// Obtain access token

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

@@ -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::<MessageMetadata>()
.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 <hello@stalw.art>
// 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!(

21
crates/services/src/task_manager/lock.rs
View File

@@ -93,7 +93,7 @@ impl TaskLock for Task<IndexAction> {
fn lock_key(&self) -> Vec<u8> {
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<CalendarAlarm> {
fn lock_key(&self) -> Vec<u8> {
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<ImipAction> {
fn lock_key(&self) -> Vec<u8> {
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<ImipAction> {
fn value_classes(&self) -> impl Iterator<Item = ValueClass> {
[
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<MergeThreadIds<AHashSet<u32>>> {
fn lock_key(&self) -> Vec<u8> {
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<TaskAction> {
}
}
pub(crate) fn deserialize(key: &[u8], value: &[u8]) -> trc::Result<Self> {
pub fn deserialize(key: &[u8], value: &[u8]) -> trc::Result<Self> {
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) {

20
crates/services/src/task_manager/mod.rs
View File

@@ -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<T> {
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)
}
},