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

All tests passing

Browse Source
This commit is contained in:
mdecimus
2025-12-12 15:34:38 +01:00
parent 3df5327ac8
commit e944316312
18 changed files with 331 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/migration/Cargo.toml
View File

@@ -31,6 +31,7 @@ bincode = "1.3.3"
lz4_flex = { version = "0.12", default-features = false }
base64 = "0.22"
futures = "0.3"
num_cpus = "1.13.1"
[features]
test_mode = []

13
crates/migration/src/blob.rs
View File

@@ -18,6 +18,7 @@ use types::blob_hash::{BLOB_HASH_LEN, BlobHash};
const SUBSPACE_BLOB_RESERVE: u8 = b'j';
pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> {
let mut num_blobs = 0;
for byte in 0..=u8::MAX {
// Validate linked blobs
let mut from_hash = BlobHash::default();
@@ -78,6 +79,7 @@ pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> {
.caused_by(trc::location!())?;
let mut batch = BatchBuilder::new();
num_blobs += keys.len();
for (key, op) in keys {
batch
.clear(ValueClass::Any(AnyClass {
@@ -104,6 +106,11 @@ pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> {
}
}
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {num_blobs} blob links")
);
enum OldType {
Quota { size: u32 },
Undelete { deleted_at: u64, size: u32 },
@@ -181,6 +188,7 @@ pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> {
.caused_by(trc::location!())?;
let mut batch = BatchBuilder::new();
let num_entries = entries.len();
for entry in entries {
batch
.clear(ValueClass::Any(AnyClass {
@@ -259,6 +267,11 @@ pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> {
}
}
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {num_entries} temporary blob links")
);
if !batch.is_empty() {
server
.store()

24
crates/migration/src/principal_v2.rs
View File

@@ -4,19 +4,6 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::time::Instant;
use common::Server;
use directory::{Principal, PrincipalData, Type, backend::internal::SpecialSecrets};
use proc_macros::EnumMethods;
use store::{
Serialize, ValueKey,
roaring::RoaringBitmap,
write::{AlignedBytes, Archive, Archiver, BatchBuilder, DirectoryClass, ValueClass},
};
use trc::AddContext;
use types::collection::Collection;
use crate::{
addressbook_v2::migrate_addressbook_v013,
calendar_v2::migrate_calendar_v013,
@@ -26,6 +13,17 @@ use crate::{
push_v2::migrate_push_subscriptions_v013,
sieve_v2::migrate_sieve_v013,
};
use common::Server;
use directory::{Principal, PrincipalData, Type, backend::internal::SpecialSecrets};
use proc_macros::EnumMethods;
use std::time::Instant;
use store::{
Serialize, ValueKey,
roaring::RoaringBitmap,
write::{AlignedBytes, Archive, Archiver, BatchBuilder, DirectoryClass, ValueClass},
};
use trc::AddContext;
use types::collection::Collection;
pub(crate) async fn migrate_principals_v0_13(server: &Server) -> trc::Result<RoaringBitmap> {
// Obtain email ids

15
crates/migration/src/queue_v2.rs
View File

@@ -152,8 +152,6 @@ pub enum LegacyQuotaKey {
}
pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> {
let mut count = 0;
let mut messages = Vec::new();
server
.store()
@@ -176,8 +174,6 @@ pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> {
}
}
count += 1;
Ok(true)
},
)
@@ -185,6 +181,7 @@ pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> {
.caused_by(trc::location!())?;
let mut batch = BatchBuilder::new();
let count = messages.len();
for (queue_id, message) in messages {
batch.set(
ValueClass::Queue(QueueClass::Message(queue_id)),
@@ -211,12 +208,10 @@ pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> {
.caused_by(trc::location!())?;
}
if count > 0 {
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {count} queued messages",)
);
}
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {count} queued messages",)
);
Ok(())
}

70
crates/migration/src/tasks_v2.rs
View File

@@ -7,7 +7,10 @@
use common::Server;
use store::{
IterateParams, SUBSPACE_TASK_QUEUE, U32_LEN, U64_LEN, ValueKey,
write::{AnyClass, BatchBuilder, ValueClass, key::KeySerializer},
write::{
AnyClass, BatchBuilder, TaskEpoch, ValueClass,
key::{DeserializeBigEndian, KeySerializer},
},
};
use trc::AddContext;
@@ -31,16 +34,15 @@ pub(crate) async fn migrate_tasks_v014(server: &Server) -> trc::Result<()> {
}),
};
let todo = "task epochs";
let mut delete_tasks = Vec::new();
let mut insert_tasks = Vec::new();
server
.core
.storage
.data
.iterate(
IterateParams::new(from_key, to_key).ascending().no_values(),
|key, _| {
IterateParams::new(from_key, to_key).ascending(),
|key, value| {
match key.get(U64_LEN + U32_LEN) {
Some(0..=2) => {
delete_tasks.push(key.to_vec());
@@ -48,7 +50,18 @@ pub(crate) async fn migrate_tasks_v014(server: &Server) -> trc::Result<()> {
None => {
return Err(trc::Error::corrupted_key(key, None, trc::location!()));
}
_ => {}
_ => {
let due = key.deserialize_be_u64(0)?;
let maybe_epoch = TaskEpoch::from_inner(due);
if maybe_epoch.attempt() != 0 {
delete_tasks.push(key.to_vec());
let epoch = TaskEpoch::new(due).inner();
let mut new_key = Vec::with_capacity(key.len());
new_key.extend_from_slice(&epoch.to_be_bytes());
new_key.extend_from_slice(&key[U64_LEN..]);
insert_tasks.push((new_key, value.to_vec()));
}
}
};
Ok(true)
},
@@ -56,26 +69,57 @@ pub(crate) async fn migrate_tasks_v014(server: &Server) -> trc::Result<()> {
.await
.caused_by(trc::location!())?;
if !delete_tasks.is_empty() {
let num_migrated = delete_tasks.len();
let num_migrated = delete_tasks.len() + insert_tasks.len();
if num_migrated != 0 {
let mut batch = BatchBuilder::new();
let mut batch_len = 0;
for (key, value) in insert_tasks {
batch_len += key.len() + value.len();
batch.set(
ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key,
}),
value,
);
if batch_len > 4 * 1024 * 1024 {
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
batch_len = 0;
}
}
for key in delete_tasks {
batch_len += key.len();
batch.clear(ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key,
}));
if batch_len > 4 * 1024 * 1024 {
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
batch_len = 0;
}
}
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {num_migrated} tasks")
);
}
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {num_migrated} tasks")
);
Ok(())
}

218
crates/migration/src/v014.rs
View File

@@ -35,7 +35,7 @@ pub const SUBSPACE_BITMAP_TEXT: u8 = b'v';
pub const SUBSPACE_FTS_INDEX: u8 = b'g';
pub const SUBSPACE_TELEMETRY_INDEX: u8 = b'w';
pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> {
pub async fn migrate_v0_14(server: &Server) -> trc::Result<()> {
// Migrate global data
let mut tasks = Vec::new();
let _server = server.clone();
@@ -74,9 +74,10 @@ pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> {
std::env::var("NUM_THREADS")
.ok()
.and_then(|s| s.parse::<usize>().ok())
.unwrap_or(8),
.unwrap_or_else(|| num_cpus::get().min(2) * 2),
));
let mut tasks = Vec::with_capacity(principal_ids.len());
let num_principals = principal_ids.len();
for principal_id in principal_ids {
let permit = semaphore.clone().acquire_owned().await.unwrap();
let _server = server.clone();
@@ -97,6 +98,11 @@ pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> {
.details("Join Error")
})??;
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {num_principals} accounts")
);
// Delete old subspaces
for subspace in [
SUBSPACE_BITMAP_ID,
@@ -121,16 +127,34 @@ pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> {
.caused_by(trc::location!())?;
}
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migration to v0.15 completed")
);
Ok(())
}
pub(crate) async fn migrate_principal_v0_14(server: &Server, account_id: u32) -> trc::Result<()> {
migrate_emails_v014(server, account_id).await?;
migrate_encryption_params_v014(server, account_id).await?;
migrate_indexes(server, account_id).await
let emails = migrate_emails_v014(server, account_id).await?;
let params = migrate_encryption_params_v014(server, account_id).await?;
let (num_contacts, num_calendars, num_email_submissions, num_identities) =
migrate_indexes(server, account_id).await?;
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!(
"Migrated account {account_id}: {emails} emails, {params} encryption params, {num_contacts} contacts, {num_calendars} calendars, {num_email_submissions} submissions, and {num_identities} identities"
)
);
Ok(())
}
pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Result<()> {
pub(crate) async fn migrate_indexes(
server: &Server,
account_id: u32,
) -> trc::Result<(usize, usize, usize, usize)> {
/*
EmailSubmissionField::UndoStatus => 41,
@@ -211,6 +235,10 @@ pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Re
}
let mut indexes = Vec::new();
let mut num_contacts = 0;
let mut num_calendars = 0;
let mut num_email_submissions = 0;
let mut num_identities = 0;
for collection in [
Collection::ContactCard,
Collection::CalendarEventNotification,
@@ -218,104 +246,101 @@ pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Re
Collection::Identity,
] {
server
.archives(
account_id,
Collection::ContactCard,
&(),
|document_id, archive| {
match collection {
Collection::ContactCard => {
let data = archive
.unarchive_untrusted::<ContactCard>()
.caused_by(trc::location!())?;
.archives(account_id, collection, &(), |document_id, archive| {
match collection {
Collection::ContactCard => {
let data = archive
.unarchive_untrusted::<ContactCard>()
.caused_by(trc::location!())?;
if let Some(email) = data.emails().next() {
indexes.push((
collection,
document_id,
Operation::Index {
field: ContactField::Email.into(),
key: email.into_bytes(),
set: true,
},
));
}
indexes.push((
collection,
document_id,
Operation::Value {
class: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: ContactField::CreatedToUpdated.into(),
value: data.created.to_native() as u64,
}),
op: ValueOp::Set(
(data.modified.to_native() as u64).serialize(),
),
},
));
}
Collection::CalendarEventNotification => {
let data = archive
.unarchive_untrusted::<CalendarEventNotification>()
.caused_by(trc::location!())?;
indexes.push((
collection,
document_id,
Operation::Value {
class: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: CalendarNotificationField::CreatedToId.into(),
value: data.created.to_native() as u64,
}),
op: ValueOp::Set(
data.event_id
.as_ref()
.map(|v| v.to_native())
.unwrap_or(u32::MAX)
.serialize(),
),
},
));
}
Collection::EmailSubmission => {
let data = archive
.unarchive_untrusted::<EmailSubmission>()
.caused_by(trc::location!())?;
indexes.push((
collection,
document_id,
Operation::Value {
class: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: EmailSubmissionField::Metadata.into(),
value: data.send_at.to_native(),
}),
op: ValueOp::Set(
KeySerializer::new(U32_LEN * 3 + 1)
.write(data.email_id.to_native())
.write(data.thread_id.to_native())
.write(data.identity_id.to_native())
.write(data.undo_status.as_index())
.finalize(),
),
},
));
}
Collection::Identity => {
if let Some(email) = data.emails().next() {
indexes.push((
collection,
document_id,
Operation::Index {
field: IdentityField::DocumentId.into(),
key: vec![],
field: ContactField::Email.into(),
key: email.into_bytes(),
set: true,
},
));
}
_ => unreachable!(),
num_contacts += 1;
indexes.push((
collection,
document_id,
Operation::Value {
class: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: ContactField::CreatedToUpdated.into(),
value: data.created.to_native() as u64,
}),
op: ValueOp::Set((data.modified.to_native() as u64).serialize()),
},
));
}
Collection::CalendarEventNotification => {
let data = archive
.unarchive_untrusted::<CalendarEventNotification>()
.caused_by(trc::location!())?;
num_calendars += 1;
indexes.push((
collection,
document_id,
Operation::Value {
class: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: CalendarNotificationField::CreatedToId.into(),
value: data.created.to_native() as u64,
}),
op: ValueOp::Set(
data.event_id
.as_ref()
.map(|v| v.to_native())
.unwrap_or(u32::MAX)
.serialize(),
),
},
));
}
Collection::EmailSubmission => {
let data = archive
.unarchive_untrusted::<EmailSubmission>()
.caused_by(trc::location!())?;
num_email_submissions += 1;
indexes.push((
collection,
document_id,
Operation::Value {
class: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: EmailSubmissionField::Metadata.into(),
value: data.send_at.to_native(),
}),
op: ValueOp::Set(
KeySerializer::new(U32_LEN * 3 + 1)
.write(data.email_id.to_native())
.write(data.thread_id.to_native())
.write(data.identity_id.to_native())
.write(data.undo_status.as_index())
.finalize(),
),
},
));
}
Collection::Identity => {
num_identities += 1;
indexes.push((
collection,
document_id,
Operation::Index {
field: IdentityField::DocumentId.into(),
key: vec![],
set: true,
},
));
}
_ => unreachable!(),
}
Ok(true)
},
)
Ok(true)
})
.await
.caused_by(trc::location!())?;
}
@@ -345,5 +370,10 @@ pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Re
.caused_by(trc::location!())?;
}
Ok(())
Ok((
num_contacts,
num_calendars,
num_email_submissions,
num_identities,
))
}