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

MTA queue enhancements (closes #1246 closes #1035 closes #457)

Browse Source
This commit is contained in:
mdecimus
2025-07-11 17:51:53 +02:00
parent d04a4331c3
commit e28769c5ce
68 changed files with 1518 additions and 598 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/migration/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "migration"
version = "0.12.5"
version = "0.13.0"
edition = "2024"
resolver = "2"

110
crates/migration/src/lib.rs
View File

@@ -4,25 +4,25 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::calendar::migrate_calendar_events;
use crate::{
calendar::migrate_calendar_events,
queue::{migrate_queue_v011, migrate_queue_v012},
tasks::migrate_tasks_v011,
};
use changelog::reset_changelog;
use common::{DATABASE_SCHEMA_VERSION, KV_LOCK_HOUSEKEEPER, Server};
use jmap_proto::types::{collection::Collection, property::Property};
use principal::{migrate_principal, migrate_principals};
use queue::migrate_queue;
use report::migrate_reports;
use std::time::Duration;
use store::{
Deserialize, IterateParams, SUBSPACE_PROPERTY, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_REPORT_IN,
SUBSPACE_REPORT_OUT, SUBSPACE_TASK_QUEUE, SerializeInfallible, U32_LEN, Value, ValueKey,
SUBSPACE_REPORT_OUT, SerializeInfallible, U32_LEN, Value, ValueKey,
dispatch::{DocumentSet, lookup::KeyValue},
rand::{self, seq::SliceRandom},
write::{
AnyClass, AnyKey, BatchBuilder, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now,
},
write::{AnyClass, AnyKey, BatchBuilder, ValueClass, key::DeserializeBigEndian},
};
use trc::AddContext;
use utils::BlobHash;
pub mod calendar;
pub mod changelog;
@@ -37,6 +37,7 @@ pub mod queue;
pub mod report;
pub mod sieve;
pub mod submission;
pub mod tasks;
pub mod threads;
const LOCK_WAIT_TIME_ACCOUNT: u64 = 3 * 60;
@@ -45,7 +46,14 @@ const LOCK_RETRY_TIME: Duration = Duration::from_secs(30);
pub async fn try_migrate(server: &Server) -> trc::Result<()> {
if std::env::var("FORCE_MIGRATE_QUEUE").is_ok() {
migrate_queue(server).await.caused_by(trc::location!())?;
migrate_queue_v012(server)
.await
.caused_by(trc::location!())?;
return Ok(());
} else if std::env::var("FORCE_MIGRATE_QUEUE_V011").is_ok() {
migrate_queue_v011(server)
.await
.caused_by(trc::location!())?;
return Ok(());
} else if let Some(account_id) = std::env::var("FORCE_MIGRATE_ACCOUNT")
.ok()
@@ -57,9 +65,6 @@ pub async fn try_migrate(server: &Server) -> trc::Result<()> {
return Ok(());
}
let todo =
"migrate queue + new LegacyRecipient with domain_idx u32 / size u64 + migrate error enum";
match server
.store()
.get_value::<u32>(AnyKey {
@@ -73,7 +78,14 @@ pub async fn try_migrate(server: &Server) -> trc::Result<()> {
return Ok(());
}
Some(1) => {
migrate_v0_12_0(server).await.caused_by(trc::location!())?;
migrate_v0_12(server, true)
.await
.caused_by(trc::location!())?;
}
Some(2) => {
migrate_v0_12(server, false)
.await
.caused_by(trc::location!())?;
}
Some(version) => {
panic!(
@@ -105,7 +117,7 @@ pub async fn try_migrate(server: &Server) -> trc::Result<()> {
Ok(())
}
async fn migrate_v0_12_0(server: &Server) -> trc::Result<()> {
async fn migrate_v0_12(server: &Server, migrate_tasks: bool) -> trc::Result<()> {
let force_lock = std::env::var("FORCE_LOCK").is_ok();
let in_memory = server.in_memory_store();
@@ -120,76 +132,14 @@ async fn migrate_v0_12_0(server: &Server) -> trc::Result<()> {
.await
.caused_by(trc::location!())?
{
let from_key = ValueKey::<ValueClass> {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail {
due: 0,
hash: BlobHash::default(),
}),
};
let to_key = ValueKey::<ValueClass> {
account_id: u32::MAX,
collection: u8::MAX,
document_id: u32::MAX,
class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail {
due: u64::MAX,
hash: BlobHash::default(),
}),
};
let now = now();
let mut migrate_tasks = Vec::new();
server
.core
.storage
.data
.iterate(
IterateParams::new(from_key, to_key).ascending(),
|key, value| {
let due = key.deserialize_be_u64(0)?;
if due > now {
migrate_tasks.push((key.to_vec(), value.to_vec()));
}
Ok(true)
},
)
migrate_queue_v012(server)
.await
.caused_by(trc::location!())?;
if !migrate_tasks.is_empty() {
let num_migrated = migrate_tasks.len();
let mut batch = BatchBuilder::new();
for (key, value) in migrate_tasks {
let mut new_key = key.clone();
new_key[0..8].copy_from_slice(&now.to_be_bytes());
batch
.clear(ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key,
}))
.set(
ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key: new_key,
}),
value,
);
}
server
.store()
.write(batch.build_all())
if migrate_tasks {
migrate_tasks_v011(server)
.await
.caused_by(trc::location!())?;
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {num_migrated} tasks")
);
}
in_memory
@@ -237,7 +187,9 @@ async fn migrate_v0_11(server: &Server) -> trc::Result<()> {
.caused_by(trc::location!())?
.is_none()
{
migrate_queue(server).await.caused_by(trc::location!())?;
migrate_queue_v011(server)
.await
.caused_by(trc::location!())?;
migrate_reports(server).await.caused_by(trc::location!())?;
reset_changelog(server).await.caused_by(trc::location!())?;
principal_ids = migrate_principals(server)

214
crates/migration/src/queue.rs
View File

@@ -25,7 +25,7 @@ use store::{
use trc::AddContext;
use utils::BlobHash;
pub(crate) async fn migrate_queue(server: &Server) -> trc::Result<()> {
pub(crate) async fn migrate_queue_v011(server: &Server) -> trc::Result<()> {
let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: 0,
@@ -75,7 +75,7 @@ pub(crate) async fn migrate_queue(server: &Server) -> trc::Result<()> {
for queue_id in queue_ids {
match server
.store()
.get_value::<LegacyBincode<LegacyMessage>>(ValueKey::from(ValueClass::Queue(
.get_value::<LegacyBincode<MessageV011>>(ValueKey::from(ValueClass::Queue(
QueueClass::Message(queue_id),
)))
.await
@@ -123,8 +123,123 @@ pub(crate) async fn migrate_queue(server: &Server) -> trc::Result<()> {
Ok(())
}
impl From<LegacyMessage> for Message {
fn from(message: LegacyMessage) -> Self {
pub(crate) async fn migrate_queue_v012(server: &Server) -> trc::Result<()> {
let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: 0,
queue_id: 0,
queue_name: [0; 8],
},
)));
let to_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(
store::write::QueueEvent {
due: u64::MAX,
queue_id: u64::MAX,
queue_name: [u8::MAX; 8],
},
)));
let mut queue_ids = AHashSet::new();
server
.store()
.iterate(
IterateParams::new(from_key, to_key).ascending().no_values(),
|key, _| {
queue_ids.insert(key.deserialize_be_u64(U64_LEN)?);
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
let from_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(0)));
let to_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(u64::MAX)));
server
.store()
.iterate(
IterateParams::new(from_key, to_key).ascending().no_values(),
|key, _| {
queue_ids.insert(key.deserialize_be_u64(0)?);
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
let mut count = 0;
for queue_id in queue_ids {
match server
.store()
.get_value::<Archive<AlignedBytes>>(ValueKey::from(ValueClass::Queue(
QueueClass::Message(queue_id),
)))
.await
.and_then(|archive| {
if let Some(archive) = archive {
archive.deserialize::<MessageV012>().map(Some)
} else {
Ok(None)
}
}) {
Ok(Some(archive)) => {
let mut batch = BatchBuilder::new();
batch.set(
ValueClass::Queue(QueueClass::Message(queue_id)),
Archiver::new(Message::from(archive))
.serialize()
.caused_by(trc::location!())?,
);
count += 1;
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
Ok(None) => (),
Err(err) => {
if server
.store()
.get_value::<Archive<AlignedBytes>>(ValueKey::from(ValueClass::Queue(
QueueClass::Message(queue_id),
)))
.await
.and_then(|archive| {
if let Some(archive) = archive {
archive.deserialize::<Message>().map(Some)
} else {
Ok(None)
}
})
.is_err()
{
return Err(err
.ctx(trc::Key::QueueId, queue_id)
.caused_by(trc::location!()));
}
}
}
}
if count > 0 {
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {count} queued messages",)
);
}
Ok(())
}
impl<SIZE, IDX> From<LegacyMessage<SIZE, IDX>> for Message
where
SIZE: AsU64,
IDX: AsU64,
{
fn from(message: LegacyMessage<SIZE, IDX>) -> Self {
let domains = message.domains;
Message {
created: message.created,
@@ -136,7 +251,7 @@ impl From<LegacyMessage> for Message {
.recipients
.into_iter()
.map(|r| {
let domain = &domains[r.domain_idx];
let domain = &domains[r.domain_idx.as_u64() as usize];
Recipient {
address: r.address,
address_lcase: r.address_lcase,
@@ -170,7 +285,7 @@ impl From<LegacyMessage> for Message {
flags: message.flags,
env_id: message.env_id,
priority: message.priority,
size: message.size as u64,
size: message.size.as_u64(),
quota_keys: message.quota_keys,
received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
received_via_port: 0,
@@ -178,6 +293,25 @@ impl From<LegacyMessage> for Message {
}
}
trait AsU64 {
fn as_u64(&self) -> u64;
}
impl AsU64 for usize {
fn as_u64(&self) -> u64 {
*self as u64
}
}
impl AsU64 for u32 {
fn as_u64(&self) -> u64 {
*self as u64
}
}
impl AsU64 for u64 {
fn as_u64(&self) -> u64 {
*self
}
}
fn migrate_legacy_error(domain: &str, err: &LegacyError) -> ErrorDetails {
match err {
LegacyError::DnsError(err) => ErrorDetails {
@@ -232,8 +366,20 @@ fn migrate_host_response(response: HostResponse<LegacyErrorDetails>) -> ErrorDet
}
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
pub struct LegacyMessage {
pub type MessageV011 = LegacyMessage<usize, usize>;
pub type MessageV012 = LegacyMessage<u64, u32>;
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
)]
pub struct LegacyMessage<SIZE, IDX> {
pub queue_id: QueueId,
pub created: u64,
pub blob_hash: BlobHash,
@@ -241,23 +387,32 @@ pub struct LegacyMessage {
pub return_path: String,
pub return_path_lcase: String,
pub return_path_domain: String,
pub recipients: Vec<LegacyRecipient>,
pub recipients: Vec<LegacyRecipient<IDX>>,
pub domains: Vec<LegacyDomain>,
pub flags: u64,
pub env_id: Option<String>,
pub priority: i16,
pub size: usize,
pub size: SIZE,
pub quota_keys: Vec<QuotaKey>,
#[serde(skip)]
pub span_id: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
pub struct LegacyRecipient {
pub domain_idx: usize,
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
)]
pub struct LegacyRecipient<IDX> {
pub domain_idx: IDX,
pub address: String,
pub address_lcase: String,
pub status: Status<HostResponse<String>, HostResponse<LegacyErrorDetails>>,
@@ -265,7 +420,16 @@ pub struct LegacyRecipient {
pub orcpt: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
)]
pub struct LegacyDomain {
pub domain: String,
pub retry: Schedule<u32>,
@@ -274,7 +438,16 @@ pub struct LegacyDomain {
pub status: Status<(), LegacyError>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
)]
pub enum LegacyError {
DnsError(String),
UnexpectedResponse(HostResponse<LegacyErrorDetails>),
@@ -287,7 +460,16 @@ pub enum LegacyError {
Io(String),
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
)]
pub struct LegacyErrorDetails {
pub entity: String,
pub details: String,

89
crates/migration/src/tasks.rs Normal file
View File

@@ -0,0 +1,89 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::Server;
use store::{
IterateParams, SUBSPACE_TASK_QUEUE, ValueKey,
write::{AnyClass, BatchBuilder, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now},
};
use trc::AddContext;
use utils::BlobHash;
pub(crate) async fn migrate_tasks_v011(server: &Server) -> trc::Result<()> {
let from_key = ValueKey::<ValueClass> {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail {
due: 0,
hash: BlobHash::default(),
}),
};
let to_key = ValueKey::<ValueClass> {
account_id: u32::MAX,
collection: u8::MAX,
document_id: u32::MAX,
class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail {
due: u64::MAX,
hash: BlobHash::default(),
}),
};
let now = now();
let mut migrate_tasks = Vec::new();
server
.core
.storage
.data
.iterate(
IterateParams::new(from_key, to_key).ascending(),
|key, value| {
let due = key.deserialize_be_u64(0)?;
if due > now {
migrate_tasks.push((key.to_vec(), value.to_vec()));
}
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
if !migrate_tasks.is_empty() {
let num_migrated = migrate_tasks.len();
let mut batch = BatchBuilder::new();
for (key, value) in migrate_tasks {
let mut new_key = key.clone();
new_key[0..8].copy_from_slice(&now.to_be_bytes());
batch
.clear(ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key,
}))
.set(
ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key: new_key,
}),
value,
);
}
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {num_migrated} tasks")
);
}
Ok(())
}