From 365c87af20e180692a6fe27c3b1e101e753e3a59 Mon Sep 17 00:00:00 2001 From: mdecimus Date: Thu, 15 May 2025 13:35:40 +0200 Subject: [PATCH] Unversioned archiving --- crates/common/src/lib.rs | 3 +- crates/common/src/listener/acme/directory.rs | 26 +-- crates/common/src/listener/acme/resolver.rs | 19 +- crates/dav/src/common/lock.rs | 4 +- crates/directory/src/lib.rs | 1 + crates/email/src/lib.rs | 1 + crates/email/src/message/metadata.rs | 6 +- crates/email/src/migration/email.rs | 14 ++ crates/email/src/migration/mod.rs | 1 + crates/groupware/src/calendar/mod.rs | 4 +- crates/groupware/src/contact/mod.rs | 6 +- crates/groupware/src/file/mod.rs | 4 +- crates/http/src/auth/oauth/auth.rs | 10 +- crates/http/src/auth/oauth/mod.rs | 7 - crates/http/src/auth/oauth/token.rs | 19 +- crates/http/src/management/report.rs | 56 +++--- crates/services/src/broadcast/subscriber.rs | 3 +- crates/smtp/src/queue/mod.rs | 4 +- crates/smtp/src/reporting/analysis.rs | 16 +- crates/smtp/src/reporting/dmarc.rs | 31 ++-- crates/smtp/src/reporting/mod.rs | 34 +--- crates/smtp/src/reporting/tls.rs | 32 ++-- crates/store/src/lib.rs | 34 ++-- crates/store/src/write/mod.rs | 17 ++ crates/store/src/write/serialize.rs | 186 ++++++++++++++++++- crates/utils/src/snowflake.rs | 32 +--- tests/src/imap/mod.rs | 5 +- tests/src/jmap/mod.rs | 4 +- 28 files changed, 360 insertions(+), 219 deletions(-) create mode 100644 crates/email/src/migration/email.rs create mode 100644 crates/email/src/migration/mod.rs diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 990e5cc6..5f6b3fd5 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -801,12 +801,13 @@ impl MessageStoreCache { if is_unique_hash { hash } else { - loop { + for _ in 0..u32::MAX { hash = hash.wrapping_add(1); if !threads_ids.contains(hash) { return hash; } } + hash } } } diff --git a/crates/common/src/listener/acme/directory.rs b/crates/common/src/listener/acme/directory.rs index e01a85ec..024e15c2 100644 --- a/crates/common/src/listener/acme/directory.rs +++ b/crates/common/src/listener/acme/directory.rs @@ -1,7 +1,10 @@ // Adapted from rustls-acme (https://github.com/FlorianUekermann/rustls-acme), licensed under MIT/Apache-2.0. -use std::time::Duration; - +use super::AcmeProvider; +use super::jose::{ + Body, eab_sign, key_authorization, key_authorization_sha256, key_authorization_sha256_base64, + sign, +}; use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use hyper::header::USER_AGENT; @@ -11,17 +14,12 @@ use reqwest::{Method, Response}; use ring::rand::SystemRandom; use ring::signature::{ECDSA_P256_SHA256_FIXED_SIGNING, EcdsaKeyPair, EcdsaSigningAlgorithm}; use serde::Deserialize; -use store::write::Archiver; -use store::{SERIALIZE_CERT_V1, Serialize, SerializedVersion}; +use std::time::Duration; +use store::Serialize; +use store::write::UnversionedArchiver; use trc::AddContext; use trc::event::conv::AssertSuccess; -use super::AcmeProvider; -use super::jose::{ - Body, eab_sign, key_authorization, key_authorization_sha256, key_authorization_sha256_base64, - sign, -}; - pub const LETS_ENCRYPT_STAGING_DIRECTORY: &str = "https://acme-staging-v02.api.letsencrypt.org/directory"; pub const LETS_ENCRYPT_PRODUCTION_DIRECTORY: &str = @@ -190,7 +188,7 @@ impl Account { .reason(err) })?; - Archiver::new(SerializedCert { + UnversionedArchiver::new(SerializedCert { certificate: cert.serialize_der().map_err(|err| { trc::EventType::Acme(trc::AcmeEvent::Error) .caused_by(trc::location!()) @@ -210,12 +208,6 @@ pub struct SerializedCert { pub private_key: Vec, } -impl SerializedVersion for SerializedCert { - fn serialize_version() -> u8 { - SERIALIZE_CERT_V1 - } -} - #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] pub struct Directory { diff --git a/crates/common/src/listener/acme/resolver.rs b/crates/common/src/listener/acme/resolver.rs index e5ff5230..876594b1 100644 --- a/crates/common/src/listener/acme/resolver.rs +++ b/crates/common/src/listener/acme/resolver.rs @@ -4,8 +4,11 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::sync::Arc; - +use super::{ + AcmeProvider, StaticResolver, + directory::{ACME_TLS_ALPN_NAME, SerializedCert}, +}; +use crate::{KV_ACME, Server}; use rustls::{ ServerConfig, crypto::ring::sign::any_ecdsa_type, @@ -13,19 +16,13 @@ use rustls::{ sign::CertifiedKey, }; use rustls_pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; +use std::sync::Arc; use store::{ dispatch::lookup::KeyValue, - write::{AlignedBytes, Archive}, + write::{AlignedBytes, UnversionedArchive}, }; use trc::AcmeEvent; -use crate::{KV_ACME, Server}; - -use super::{ - AcmeProvider, StaticResolver, - directory::{ACME_TLS_ALPN_NAME, SerializedCert}, -}; - impl Server { pub(crate) fn set_cert(&self, provider: &AcmeProvider, cert: Arc) { // Add certificates @@ -51,7 +48,7 @@ impl Server { pub(crate) async fn build_acme_certificate(&self, domain: &str) -> Option> { match self .in_memory_store() - .key_get::>(KeyValue::<()>::build_key(KV_ACME, domain)) + .key_get::>(KeyValue::<()>::build_key(KV_ACME, domain)) .await { Ok(Some(cert_)) => match cert_.unarchive::() { diff --git a/crates/dav/src/common/lock.rs b/crates/dav/src/common/lock.rs index b348fe00..159a4cc6 100644 --- a/crates/dav/src/common/lock.rs +++ b/crates/dav/src/common/lock.rs @@ -20,7 +20,7 @@ use std::collections::HashMap; use store::dispatch::lookup::KeyValue; use store::write::serialize::rkyv_deserialize; use store::write::{AlignedBytes, Archive, Archiver, now}; -use store::{SERIALIZE_LOCKDATA_V1, Serialize, SerializedVersion, U32_LEN}; +use store::{SERIALIZE_DAV_LOCKS_V1, Serialize, SerializedVersion, U32_LEN}; use trc::AddContext; use super::ETag; @@ -755,7 +755,7 @@ impl<'x> LockCaches<'x> { impl SerializedVersion for LockData { fn serialize_version() -> u8 { - SERIALIZE_LOCKDATA_V1 + SERIALIZE_DAV_LOCKS_V1 } } diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index c4f27d9a..01ab360f 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -62,6 +62,7 @@ pub enum PrincipalData { ExternalMembers(Vec), Urls(Vec), PrincipalQuota(Vec), + Language(String), } #[derive(rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Clone, PartialEq, Eq)] diff --git a/crates/email/src/lib.rs b/crates/email/src/lib.rs index 48a38f16..b596ba28 100644 --- a/crates/email/src/lib.rs +++ b/crates/email/src/lib.rs @@ -8,6 +8,7 @@ pub mod cache; pub mod identity; pub mod mailbox; pub mod message; +pub mod migration; pub mod push; pub mod sieve; pub mod submission; diff --git a/crates/email/src/message/metadata.rs b/crates/email/src/message/metadata.rs index 0efd9579..ecd5e75d 100644 --- a/crates/email/src/message/metadata.rs +++ b/crates/email/src/message/metadata.rs @@ -21,7 +21,7 @@ use rkyv::{ vec::ArchivedVec, }; use std::{borrow::Cow, collections::VecDeque}; -use store::{SERIALIZE_MSGDATA_V1, SERIALIZE_MSGMETADATA_V1, SerializedVersion}; +use store::{SERIALIZE_MESSAGE_DATA_V1, SERIALIZE_MESSAGE_METADATA_V1, SerializedVersion}; use utils::BlobHash; #[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug)] @@ -47,13 +47,13 @@ impl IndexableAndSerializableObject for MessageData {} impl SerializedVersion for MessageData { fn serialize_version() -> u8 { - SERIALIZE_MSGDATA_V1 + SERIALIZE_MESSAGE_DATA_V1 } } impl SerializedVersion for MessageMetadata { fn serialize_version() -> u8 { - SERIALIZE_MSGMETADATA_V1 + SERIALIZE_MESSAGE_METADATA_V1 } } diff --git a/crates/email/src/migration/email.rs b/crates/email/src/migration/email.rs new file mode 100644 index 00000000..ae03cee8 --- /dev/null +++ b/crates/email/src/migration/email.rs @@ -0,0 +1,14 @@ +use common::Server; +use jmap_proto::types::collection::Collection; + +async fn migrate_email(server: &Server, account_id: u32) -> trc::Result<()> { + // Obtain email ids + let document_ids = server + .get_document_ids(account_id, Collection::Email) + .await? + .unwrap_or_default(); + + //TODO remove tombstones + + Ok(()) +} diff --git a/crates/email/src/migration/mod.rs b/crates/email/src/migration/mod.rs new file mode 100644 index 00000000..aa5f45d4 --- /dev/null +++ b/crates/email/src/migration/mod.rs @@ -0,0 +1 @@ +pub mod email; diff --git a/crates/groupware/src/calendar/mod.rs b/crates/groupware/src/calendar/mod.rs index a3c4f110..f163b734 100644 --- a/crates/groupware/src/calendar/mod.rs +++ b/crates/groupware/src/calendar/mod.rs @@ -12,7 +12,7 @@ use calcard::icalendar::ICalendar; use common::DavName; use dav_proto::schema::request::DeadProperty; use jmap_proto::types::{acl::Acl, value::AclGrant}; -use store::{SERIALIZE_CALENDAR_V1, SERIALIZE_CALENDAREVENT_V1, SerializedVersion}; +use store::{SERIALIZE_CALENDAR_EVENT_V1, SERIALIZE_CALENDAR_V1, SerializedVersion}; #[derive( rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default, Clone, PartialEq, Eq, @@ -188,7 +188,7 @@ impl SerializedVersion for Calendar { impl SerializedVersion for CalendarEvent { fn serialize_version() -> u8 { - SERIALIZE_CALENDAREVENT_V1 + SERIALIZE_CALENDAR_EVENT_V1 } } diff --git a/crates/groupware/src/contact/mod.rs b/crates/groupware/src/contact/mod.rs index 50f0aef0..4e3814bf 100644 --- a/crates/groupware/src/contact/mod.rs +++ b/crates/groupware/src/contact/mod.rs @@ -11,7 +11,7 @@ use calcard::vcard::VCard; use common::DavName; use dav_proto::schema::request::DeadProperty; use jmap_proto::types::{acl::Acl, value::AclGrant}; -use store::{SERIALIZE_ADDRESSBOOK_V1, SERIALIZE_CALENDAREVENT_V1, SerializedVersion}; +use store::{SERIALIZE_ADDRESS_BOOK_V1, SERIALIZE_CONTACT_V1, SerializedVersion}; #[derive( rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default, Clone, PartialEq, Eq, @@ -78,12 +78,12 @@ impl From for Acl { impl SerializedVersion for AddressBook { fn serialize_version() -> u8 { - SERIALIZE_ADDRESSBOOK_V1 + SERIALIZE_ADDRESS_BOOK_V1 } } impl SerializedVersion for ContactCard { fn serialize_version() -> u8 { - SERIALIZE_CALENDAREVENT_V1 + SERIALIZE_CONTACT_V1 } } diff --git a/crates/groupware/src/file/mod.rs b/crates/groupware/src/file/mod.rs index 200e849d..5c3ad5f3 100644 --- a/crates/groupware/src/file/mod.rs +++ b/crates/groupware/src/file/mod.rs @@ -9,7 +9,7 @@ pub mod storage; use dav_proto::schema::request::DeadProperty; use jmap_proto::types::value::AclGrant; -use store::{SERIALIZE_FILENODE_V1, SerializedVersion}; +use store::{SERIALIZE_FILE_NODE_V1, SerializedVersion}; use utils::BlobHash; #[derive( @@ -40,6 +40,6 @@ pub struct FileProperties { impl SerializedVersion for FileNode { fn serialize_version() -> u8 { - SERIALIZE_FILENODE_V1 + SERIALIZE_FILE_NODE_V1 } } diff --git a/crates/http/src/auth/oauth/auth.rs b/crates/http/src/auth/oauth/auth.rs index c0cccb30..279e0e6b 100644 --- a/crates/http/src/auth/oauth/auth.rs +++ b/crates/http/src/auth/oauth/auth.rs @@ -19,7 +19,7 @@ use std::future::Future; use store::{ Serialize, dispatch::lookup::KeyValue, - write::{Archive, Archiver}, + write::{UnversionedArchive, UnversionedArchiver}, }; use store::{ rand::{ @@ -109,7 +109,7 @@ impl OAuthApiHandler for Server { .collect::(); // Serialize OAuth code - let value = Archiver::new(OAuthCode { + let value = UnversionedArchiver::new(OAuthCode { status: OAuthStatus::Authorized, account_id: access_token.primary_id(), client_id, @@ -151,7 +151,7 @@ impl OAuthApiHandler for Server { .core .storage .lookup - .key_get::>(KeyValue::<()>::build_key( + .key_get::>(KeyValue::<()>::build_key( KV_OAUTH, code.as_bytes(), )) @@ -185,7 +185,7 @@ impl OAuthApiHandler for Server { KeyValue::with_prefix( KV_OAUTH, oauth.params.as_bytes(), - Archiver::new(new_oauth_code) + UnversionedArchiver::new(new_oauth_code) .serialize() .caused_by(trc::location!())?, ) @@ -243,7 +243,7 @@ impl OAuthApiHandler for Server { } // Add OAuth status - let oauth_code = Archiver::new(OAuthCode { + let oauth_code = UnversionedArchiver::new(OAuthCode { status: OAuthStatus::Pending, account_id: u32::MAX, client_id, diff --git a/crates/http/src/auth/oauth/mod.rs b/crates/http/src/auth/oauth/mod.rs index 929c8d9a..a9d676a4 100644 --- a/crates/http/src/auth/oauth/mod.rs +++ b/crates/http/src/auth/oauth/mod.rs @@ -7,7 +7,6 @@ use http_proto::{HttpRequest, request::fetch_body}; use hyper::header::CONTENT_TYPE; use serde::{Deserialize, Serialize}; -use store::{SERIALIZE_OAUTHCODE_V1, SerializedVersion}; use utils::map::vec_map::VecMap; pub mod auth; @@ -56,12 +55,6 @@ pub struct OAuthCode { pub params: String, } -impl SerializedVersion for OAuthCode { - fn serialize_version() -> u8 { - SERIALIZE_OAUTHCODE_V1 - } -} - #[derive(Debug, Serialize, Deserialize)] pub struct DeviceAuthGet { code: Option, diff --git a/crates/http/src/auth/oauth/token.rs b/crates/http/src/auth/oauth/token.rs index 666cf828..14f43bf1 100644 --- a/crates/http/src/auth/oauth/token.rs +++ b/crates/http/src/auth/oauth/token.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use super::{ + ArchivedOAuthStatus, ErrorType, FormData, MAX_POST_LEN, OAuthCode, OAuthResponse, OAuthStatus, + TokenResponse, registration::ClientRegistrationHandler, +}; use common::{ KV_OAUTH, Server, auth::{ @@ -11,22 +15,15 @@ use common::{ oauth::{GrantType, oidc::StandardClaims}, }, }; - +use http_proto::*; use hyper::StatusCode; use std::future::Future; use store::{ dispatch::lookup::KeyValue, - write::{AlignedBytes, Archive}, + write::{AlignedBytes, UnversionedArchive}, }; use trc::AddContext; -use http_proto::*; - -use super::{ - ArchivedOAuthStatus, ErrorType, FormData, MAX_POST_LEN, OAuthCode, OAuthResponse, OAuthStatus, - TokenResponse, registration::ClientRegistrationHandler, -}; - pub trait TokenHandler: Sync + Send { fn handle_token_request( &self, @@ -80,7 +77,7 @@ impl TokenHandler for Server { .core .storage .lookup - .key_get::>(KeyValue::<()>::build_key( + .key_get::>(KeyValue::<()>::build_key( KV_OAUTH, code.as_bytes(), )) @@ -152,7 +149,7 @@ impl TokenHandler for Server { .core .storage .lookup - .key_get::>(KeyValue::<()>::build_key( + .key_get::>(KeyValue::<()>::build_key( KV_OAUTH, device_code.as_bytes(), )) diff --git a/crates/http/src/management/report.rs b/crates/http/src/management/report.rs index a7f6d261..ff0c9a5a 100644 --- a/crates/http/src/management/report.rs +++ b/crates/http/src/management/report.rs @@ -4,29 +4,27 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::future::Future; - use common::{Server, auth::AccessToken}; - use directory::{Permission, Type, backend::internal::manage::ManageDirectory}; +use http_proto::{request::decode_path_element, *}; use hyper::Method; use mail_auth::report::{ Feedback, tlsrpt::{FailureDetails, Policy, TlsReport}, }; use serde_json::json; -use smtp::reporting::{ReportSerializer, analysis::IncomingReport}; +use smtp::reporting::analysis::IncomingReport; +use std::future::Future; use store::{ Deserialize, IterateParams, Key, U64_LEN, ValueKey, write::{ - AlignedBytes, Archive, BatchBuilder, ReportClass, ValueClass, key::DeserializeBigEndian, + AlignedBytes, BatchBuilder, ReportClass, UnversionedArchive, ValueClass, + key::DeserializeBigEndian, }, }; use trc::AddContext; use utils::url_params::UrlParams; -use http_proto::{request::decode_path_element, *}; - enum ReportType { Dmarc, Tls, @@ -291,10 +289,10 @@ where { if let Some(tls) = server .store() - .get_value::>(key) + .get_value::>(key) .await? { - tls.deserialize::>().map(|v| Some(v.0)) + tls.deserialize::().map(Some) } else { Ok(None) } @@ -381,45 +379,39 @@ async fn fetch_incoming_reports( last_id = id; // TODO: Support filtering chunked records (over 10MB) on FDB - let archive = as Deserialize>::deserialize(value)?; + let archive = + as Deserialize>::deserialize(value)?; let matches = if has_filters { match typ { ReportType::Dmarc => { - let report = - archive - .deserialize::>>() - .caused_by(trc::location!())? - .0; + let report = archive + .deserialize::>() + .caused_by(trc::location!())?; - filter.is_none_or( |f| report.contains(f)) + filter.is_none_or(|f| report.contains(f)) && tenant_domains .as_ref() - .is_none_or( |domains| report.has_domain(domains)) + .is_none_or(|domains| report.has_domain(domains)) } ReportType::Tls => { - let report = - archive - .deserialize::>>() - .caused_by(trc::location!())? - .0; + let report = archive + .deserialize::>() + .caused_by(trc::location!())?; - filter.is_none_or( |f| report.contains(f)) + filter.is_none_or(|f| report.contains(f)) && tenant_domains .as_ref() - .is_none_or( |domains| report.has_domain(domains)) + .is_none_or(|domains| report.has_domain(domains)) } ReportType::Arf => { - let report = - archive - .deserialize::>>() - .caused_by(trc::location!())? - .0; + let report = archive + .deserialize::>() + .caused_by(trc::location!())?; - - filter.is_none_or( |f| report.contains(f)) + filter.is_none_or(|f| report.contains(f)) && tenant_domains .as_ref() - .is_none_or( |domains| report.has_domain(domains)) + .is_none_or(|domains| report.has_domain(domains)) } } } else { diff --git a/crates/services/src/broadcast/subscriber.rs b/crates/services/src/broadcast/subscriber.rs index 7a058ec4..7bd9a455 100644 --- a/crates/services/src/broadcast/subscriber.rs +++ b/crates/services/src/broadcast/subscriber.rs @@ -53,7 +53,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc, mut shutdown_rx: watch::Rec ); match tokio::time::timeout( - Duration::from_secs(1 << retry_count.clamp(1, 6)), + Duration::from_secs(1 << retry_count.max(6)), shutdown_rx.changed(), ) .await @@ -63,7 +63,6 @@ pub fn spawn_broadcast_subscriber(inner: Arc, mut shutdown_rx: watch::Rec } Err(_) => { retry_count += 1; - continue; } } diff --git a/crates/smtp/src/queue/mod.rs b/crates/smtp/src/queue/mod.rs index 00772523..fa96c122 100644 --- a/crates/smtp/src/queue/mod.rs +++ b/crates/smtp/src/queue/mod.rs @@ -14,7 +14,7 @@ use common::expr::{self, functions::ResolveVariable, *}; use compact_str::ToCompactString; use smtp_proto::{ArchivedResponse, Response}; -use store::{SERIALIZE_QUEUEMSG_V1, SerializedVersion, write::now}; +use store::{SERIALIZE_QUEUE_MSG_V1, SerializedVersion, write::now}; use utils::BlobHash; pub mod dsn; @@ -71,7 +71,7 @@ pub struct Message { impl SerializedVersion for Message { fn serialize_version() -> u8 { - SERIALIZE_QUEUEMSG_V1 + SERIALIZE_QUEUE_MSG_V1 } } diff --git a/crates/smtp/src/reporting/analysis.rs b/crates/smtp/src/reporting/analysis.rs index b5244564..4556f107 100644 --- a/crates/smtp/src/reporting/analysis.rs +++ b/crates/smtp/src/reporting/analysis.rs @@ -19,12 +19,10 @@ use std::{ }; use store::{ Serialize, - write::{Archiver, BatchBuilder, ReportClass, ValueClass, now}, + write::{BatchBuilder, ReportClass, UnversionedArchiver, ValueClass, now}, }; use trc::IncomingReportEvent; -use super::ReportSerializer; - enum Compression { None, Gzip, @@ -281,12 +279,12 @@ impl AnalyzeReport for Server { Format::Dmarc(report) => { batch.set( ValueClass::Report(ReportClass::Dmarc { id, expires }), - Archiver::new(ReportSerializer(IncomingReport { + UnversionedArchiver::new(IncomingReport { from, to, subject, report, - })) + }) .serialize() .unwrap_or_default(), ); @@ -294,12 +292,12 @@ impl AnalyzeReport for Server { Format::Tls(report) => { batch.set( ValueClass::Report(ReportClass::Tls { id, expires }), - Archiver::new(ReportSerializer(IncomingReport { + UnversionedArchiver::new(IncomingReport { from, to, subject, report, - })) + }) .serialize() .unwrap_or_default(), ); @@ -307,12 +305,12 @@ impl AnalyzeReport for Server { Format::Arf(report) => { batch.set( ValueClass::Report(ReportClass::Arf { id, expires }), - Archiver::new(ReportSerializer(IncomingReport { + UnversionedArchiver::new(IncomingReport { from, to, subject, report, - })) + }) .serialize() .unwrap_or_default(), ); diff --git a/crates/smtp/src/reporting/dmarc.rs b/crates/smtp/src/reporting/dmarc.rs index 9942f70f..f0fd59fd 100644 --- a/crates/smtp/src/reporting/dmarc.rs +++ b/crates/smtp/src/reporting/dmarc.rs @@ -14,6 +14,12 @@ use common::{ listener::SessionStream, }; +use super::{AggregateTimestamp, SerializedSize}; +use crate::{ + core::Session, + queue::{DomainPart, RecipientDomain}, + reporting::SmtpReporting, +}; use compact_str::ToCompactString; use mail_auth::{ ArcOutput, AuthenticatedMessage, AuthenticationResults, DkimOutput, DkimResult, DmarcOutput, @@ -24,19 +30,14 @@ use mail_auth::{ }; use store::{ Deserialize, IterateParams, Serialize, ValueKey, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, QueueClass, ReportEvent, ValueClass}, + write::{ + AlignedBytes, BatchBuilder, QueueClass, ReportEvent, UnversionedArchive, + UnversionedArchiver, ValueClass, + }, }; use trc::{AddContext, OutgoingReportEvent}; use utils::config::Rate; -use crate::{ - core::Session, - queue::{DomainPart, RecipientDomain}, - reporting::SmtpReporting, -}; - -use super::{AggregateTimestamp, ReportSerializer, SerializedSize}; - #[derive( Debug, PartialEq, @@ -476,12 +477,12 @@ impl DmarcReporting for Server { // Deserialize report let dmarc = match self .store() - .get_value::>(ValueKey::from(ValueClass::Queue( + .get_value::>(ValueKey::from(ValueClass::Queue( QueueClass::DmarcReportHeader(event.clone()), ))) .await? { - Some(dmarc) => dmarc.deserialize::>()?.0, + Some(dmarc) => dmarc.deserialize::()?, None => { return Ok(None); } @@ -551,9 +552,9 @@ impl DmarcReporting for Server { .storage .data .iterate(IterateParams::new(from_key, to_key).ascending(), |_, v| { - let archive = as Deserialize>::deserialize(v)?; + let archive = as Deserialize>::deserialize(v)?; - match record_map.entry(archive.deserialize::>()?.0) { + match record_map.entry(archive.deserialize::()?) { Entry::Occupied(mut e) => { *e.get_mut() += 1; Ok(true) @@ -660,7 +661,7 @@ impl DmarcReporting for Server { // Write report builder.set( ValueClass::Queue(QueueClass::DmarcReportHeader(report_event.clone())), - match Archiver::new(ReportSerializer(entry)).serialize() { + match UnversionedArchiver::new(entry).serialize() { Ok(data) => data.to_vec(), Err(err) => { trc::error!( @@ -677,7 +678,7 @@ impl DmarcReporting for Server { report_event.seq_id = self.inner.data.queue_id_gen.generate(); builder.set( ValueClass::Queue(QueueClass::DmarcReportEvent(report_event)), - match Archiver::new(ReportSerializer(event.report_record)).serialize() { + match UnversionedArchiver::new(event.report_record).serialize() { Ok(data) => data.to_vec(), Err(err) => { trc::error!( diff --git a/crates/smtp/src/reporting/mod.rs b/crates/smtp/src/reporting/mod.rs index b0b545b0..3c8be044 100644 --- a/crates/smtp/src/reporting/mod.rs +++ b/crates/smtp/src/reporting/mod.rs @@ -19,10 +19,7 @@ use mail_auth::{ }; use mail_parser::DateTime; -use store::{ - SERIALIZE_REPORT_V1, SerializedVersion, - write::{ReportEvent, key::KeySerializer}, -}; +use store::write::{ReportEvent, key::KeySerializer}; use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ @@ -356,32 +353,3 @@ impl ReportLock for ReportEvent { .finalize() } } - -#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)] -#[repr(transparent)] -pub struct ReportSerializer(pub T) -where - T: rkyv::Archive - + for<'a> rkyv::Serialize< - rkyv::api::high::HighSerializer< - rkyv::util::AlignedVec, - rkyv::ser::allocator::ArenaHandle<'a>, - rkyv::rancor::Error, - >, - >; - -impl SerializedVersion for ReportSerializer -where - T: rkyv::Archive - + for<'a> rkyv::Serialize< - rkyv::api::high::HighSerializer< - rkyv::util::AlignedVec, - rkyv::ser::allocator::ArenaHandle<'a>, - rkyv::rancor::Error, - >, - >, -{ - fn serialize_version() -> u8 { - SERIALIZE_REPORT_V1 - } -} diff --git a/crates/smtp/src/reporting/tls.rs b/crates/smtp/src/reporting/tls.rs index b0e0a73b..1e53261b 100644 --- a/crates/smtp/src/reporting/tls.rs +++ b/crates/smtp/src/reporting/tls.rs @@ -4,8 +4,8 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{collections::hash_map::Entry, future::Future, sync::Arc, time::Duration}; - +use super::{AggregateTimestamp, SerializedSize}; +use crate::{queue::RecipientDomain, reporting::SmtpReporting}; use ahash::AHashMap; use common::{ Server, USER_AGENT, @@ -15,7 +15,6 @@ use common::{ }, ipc::{TlsEvent, ToHash}, }; - use mail_auth::{ flate2::{Compression, write::GzEncoder}, mta_sts::{ReportUri, TlsRpt}, @@ -23,20 +22,19 @@ use mail_auth::{ DateRange, FailureDetails, Policy, PolicyDetails, PolicyType, Summary, TlsReport, }, }; - use mail_parser::DateTime; use reqwest::header::CONTENT_TYPE; use std::fmt::Write; +use std::{collections::hash_map::Entry, future::Future, sync::Arc, time::Duration}; use store::{ Deserialize, IterateParams, Serialize, ValueKey, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, QueueClass, ReportEvent, ValueClass}, + write::{ + AlignedBytes, BatchBuilder, QueueClass, ReportEvent, UnversionedArchive, + UnversionedArchiver, ValueClass, + }, }; use trc::{AddContext, OutgoingReportEvent}; -use crate::{queue::RecipientDomain, reporting::SmtpReporting}; - -use super::{AggregateTimestamp, ReportSerializer, SerializedSize}; - #[derive(Debug, Clone)] pub struct TlsRptOptions { pub record: Arc, @@ -298,12 +296,12 @@ impl TlsReporting for Server { for event in events { let tls = if let Some(tls) = self .store() - .get_value::>(ValueKey::from(ValueClass::Queue( + .get_value::>(ValueKey::from(ValueClass::Queue( QueueClass::TlsReportHeader(event.clone()), ))) .await? { - tls.deserialize::>()?.0 + tls.deserialize::()? } else { continue; }; @@ -336,10 +334,10 @@ impl TlsReporting for Server { .storage .data .iterate(IterateParams::new(from_key, to_key).ascending(), |_, v| { - let archive = as Deserialize>::deserialize(v)?; - if let Some(failure_details) = archive - .deserialize::>>()? - .0 + let archive = + as Deserialize>::deserialize(v)?; + if let Some(failure_details) = + archive.deserialize::>()? { match record_map.entry(failure_details) { Entry::Occupied(mut e) => { @@ -492,7 +490,7 @@ impl TlsReporting for Server { // Write report builder.set( ValueClass::Queue(QueueClass::TlsReportHeader(report_event.clone())), - match Archiver::new(ReportSerializer(entry)).serialize() { + match UnversionedArchiver::new(entry).serialize() { Ok(data) => data.to_vec(), Err(err) => { trc::error!( @@ -509,7 +507,7 @@ impl TlsReporting for Server { report_event.seq_id = self.inner.data.queue_id_gen.generate(); builder.set( ValueClass::Queue(QueueClass::TlsReportEvent(report_event)), - match Archiver::new(ReportSerializer(event.failure)).serialize() { + match UnversionedArchiver::new(event.failure).serialize() { Ok(data) => data.to_vec(), Err(err) => { trc::error!( diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 5c635074..a6b739f4 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -44,24 +44,22 @@ pub trait SerializeInfallible { } // Max 64 versions (2 ^ 6) -pub const SERIALIZE_CERT_V1: u8 = 0; -pub const SERIALIZE_LOCKDATA_V1: u8 = 1; -pub const SERIALIZE_IDENTITY_V1: u8 = 2; -pub const SERIALIZE_MAILBOX_V1: u8 = 3; -pub const SERIALIZE_CRYPTO_V1: u8 = 4; -pub const SERIALIZE_MSGDATA_V1: u8 = 5; -pub const SERIALIZE_MSGMETADATA_V1: u8 = 6; -pub const SERIALIZE_PUSH_V1: u8 = 7; -pub const SERIALIZE_SIEVE_V1: u8 = 8; -pub const SERIALIZE_SUBMISSION_V1: u8 = 9; -pub const SERIALIZE_FILENODE_V1: u8 = 10; -pub const SERIALIZE_OAUTHCODE_V1: u8 = 11; -pub const SERIALIZE_QUEUEMSG_V1: u8 = 12; -pub const SERIALIZE_CALENDAR_V1: u8 = 13; -pub const SERIALIZE_ADDRESSBOOK_V1: u8 = 14; -pub const SERIALIZE_CALENDAREVENT_V1: u8 = 15; -pub const SERIALIZE_PRINCIPAL_V1: u8 = 16; -pub const SERIALIZE_REPORT_V1: u8 = 17; +pub const SERIALIZE_MESSAGE_DATA_V1: u8 = 0; +pub const SERIALIZE_MESSAGE_METADATA_V1: u8 = 1; +pub const SERIALIZE_MAILBOX_V1: u8 = 2; +pub const SERIALIZE_CRYPTO_V1: u8 = 3; +pub const SERIALIZE_IDENTITY_V1: u8 = 4; +pub const SERIALIZE_PUSH_V1: u8 = 5; +pub const SERIALIZE_SIEVE_V1: u8 = 6; +pub const SERIALIZE_SUBMISSION_V1: u8 = 7; +pub const SERIALIZE_QUEUE_MSG_V1: u8 = 8; +pub const SERIALIZE_CALENDAR_V1: u8 = 9; +pub const SERIALIZE_CALENDAR_EVENT_V1: u8 = 10; +pub const SERIALIZE_ADDRESS_BOOK_V1: u8 = 11; +pub const SERIALIZE_CONTACT_V1: u8 = 12; +pub const SERIALIZE_FILE_NODE_V1: u8 = 13; +pub const SERIALIZE_PRINCIPAL_V1: u8 = 14; +pub const SERIALIZE_DAV_LOCKS_V1: u8 = 15; pub trait SerializedVersion { fn serialize_version() -> u8; diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index cb5f70f3..795da015 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -42,6 +42,11 @@ pub struct Archive { pub hash: u32, } +#[derive(Debug, Clone)] +pub struct UnversionedArchive { + pub inner: T, +} + #[derive(Debug, Clone)] pub enum AlignedBytes { Aligned(AlignedVec), @@ -60,6 +65,18 @@ where >, >; +#[repr(transparent)] +pub struct UnversionedArchiver(pub T) +where + T: rkyv::Archive + + for<'a> rkyv::Serialize< + rkyv::api::high::HighSerializer< + rkyv::util::AlignedVec, + rkyv::ser::allocator::ArenaHandle<'a>, + rkyv::rancor::Error, + >, + >; + #[derive(Debug, Default)] pub struct AssignedIds { pub counter_ids: Vec, diff --git a/crates/store/src/write/serialize.rs b/crates/store/src/write/serialize.rs index 0ee06c0a..2265d91b 100644 --- a/crates/store/src/write/serialize.rs +++ b/crates/store/src/write/serialize.rs @@ -4,7 +4,9 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::{ARCHIVE_ALIGNMENT, AlignedBytes, Archive, Archiver}; +use super::{ + ARCHIVE_ALIGNMENT, AlignedBytes, Archive, Archiver, UnversionedArchive, UnversionedArchiver, +}; use crate::{Deserialize, Serialize, SerializeInfallible, SerializedVersion, U32_LEN, Value}; use compact_str::format_compact; use rkyv::util::AlignedVec; @@ -118,6 +120,71 @@ impl Deserialize for Archive { } } +impl Deserialize for UnversionedArchive { + fn deserialize(bytes: &[u8]) -> trc::Result { + let (marker, contents) = bytes.split_last().ok_or_else(|| { + trc::StoreEvent::DataCorruption + .into_err() + .details("Archive integrity compromised") + .ctx(trc::Key::Value, bytes) + .caused_by(trc::location!()) + })?; + + match marker & MARKER_MASK { + ARCHIVE_UNCOMPRESSED => { + let mut bytes = AlignedVec::with_capacity(contents.len()); + bytes.extend_from_slice(contents); + Ok(UnversionedArchive { + inner: AlignedBytes::Aligned(bytes), + }) + } + ARCHIVE_LZ4_COMPRESSED => { + aligned_lz4_deflate(contents).map(|inner| UnversionedArchive { inner }) + } + _ => Err(trc::StoreEvent::DataCorruption + .into_err() + .details("Invalid archive marker.") + .ctx(trc::Key::Value, bytes) + .caused_by(trc::location!())), + } + } + + fn deserialize_owned(mut bytes: Vec) -> trc::Result { + let (marker, contents) = bytes.split_last().ok_or_else(|| { + trc::StoreEvent::DataCorruption + .into_err() + .details("Archive integrity compromised") + .ctx(trc::Key::Value, bytes.as_slice()) + .caused_by(trc::location!()) + })?; + + match marker & MARKER_MASK { + ARCHIVE_UNCOMPRESSED => { + bytes.truncate(contents.len()); + if bytes.as_ptr().addr() & (ARCHIVE_ALIGNMENT - 1) == 0 { + Ok(UnversionedArchive { + inner: AlignedBytes::Vec(bytes), + }) + } else { + let mut aligned = AlignedVec::with_capacity(bytes.len()); + aligned.extend_from_slice(&bytes); + Ok(UnversionedArchive { + inner: AlignedBytes::Aligned(aligned), + }) + } + } + ARCHIVE_LZ4_COMPRESSED => { + aligned_lz4_deflate(contents).map(|inner| UnversionedArchive { inner }) + } + _ => Err(trc::StoreEvent::DataCorruption + .into_err() + .details("Invalid archive marker") + .ctx(trc::Key::Value, bytes) + .caused_by(trc::location!())), + } + } +} + #[inline] fn aligned_lz4_deflate(archive: &[u8]) -> trc::Result { lz4_flex::block::uncompressed_size(archive) @@ -192,6 +259,54 @@ where } } +impl Serialize for UnversionedArchiver +where + T: rkyv::Archive + + for<'a> rkyv::Serialize< + rkyv::api::high::HighSerializer< + rkyv::util::AlignedVec, + rkyv::ser::allocator::ArenaHandle<'a>, + rkyv::rancor::Error, + >, + >, +{ + fn serialize(&self) -> trc::Result> { + rkyv::to_bytes::(&self.0) + .map_err(|err| { + trc::StoreEvent::DeserializeError + .caused_by(trc::location!()) + .reason(err) + }) + .map(|input| { + let input = input.as_ref(); + let input_len = input.len(); + if input_len > COMPRESS_WATERMARK { + let mut bytes = + vec![ + ARCHIVE_LZ4_COMPRESSED; + lz4_flex::block::get_maximum_output_size(input_len) + U32_LEN + 1 + ]; + let compressed_len = + lz4_flex::compress_into(input, &mut bytes[U32_LEN..]).unwrap(); + if compressed_len < input_len { + bytes[..U32_LEN].copy_from_slice(&(input_len as u32).to_le_bytes()); + bytes.truncate(compressed_len + U32_LEN + 1); + } else { + bytes.clear(); + bytes.extend_from_slice(input); + bytes.push(ARCHIVE_UNCOMPRESSED); + } + bytes + } else { + let mut bytes = Vec::with_capacity(input_len + 1); + bytes.extend_from_slice(input); + bytes.push(ARCHIVE_UNCOMPRESSED); + bytes + } + }) + } +} + impl Archive { #[inline] pub fn as_bytes(&self) -> &[u8] { @@ -318,6 +433,49 @@ impl Archive { } } +impl UnversionedArchive { + #[inline] + pub fn as_bytes(&self) -> &[u8] { + match &self.inner { + AlignedBytes::Vec(bytes) => bytes.as_slice(), + AlignedBytes::Aligned(bytes) => bytes.as_slice(), + } + } + + pub fn unarchive(&self) -> trc::Result<&::Archived> + where + T: rkyv::Archive, + T::Archived: for<'a> rkyv::bytecheck::CheckBytes< + rkyv::api::high::HighValidator<'a, rkyv::rancor::Error>, + > + rkyv::Deserialize>, + { + rkyv::access::(self.as_bytes()).map_err(|err| { + trc::StoreEvent::DeserializeError + .ctx(trc::Key::Value, self.as_bytes()) + .details("Archive access failed") + .caused_by(trc::location!()) + .reason(err) + }) + } + + pub fn deserialize(&self) -> trc::Result + where + T: rkyv::Archive, + T::Archived: for<'a> rkyv::bytecheck::CheckBytes< + rkyv::api::high::HighValidator<'a, rkyv::rancor::Error>, + > + rkyv::Deserialize>, + { + self.unarchive::().and_then(|input| { + rkyv::deserialize(input).map_err(|err| { + trc::StoreEvent::DeserializeError + .ctx(trc::Key::Value, self.as_bytes()) + .caused_by(trc::location!()) + .reason(err) + }) + }) + } +} + impl Archiver where T: rkyv::Archive @@ -338,6 +496,26 @@ where } } +impl UnversionedArchiver +where + T: rkyv::Archive + + for<'a> rkyv::Serialize< + rkyv::api::high::HighSerializer< + rkyv::util::AlignedVec, + rkyv::ser::allocator::ArenaHandle<'a>, + rkyv::rancor::Error, + >, + >, +{ + pub fn new(inner: T) -> Self { + Self(inner) + } + + pub fn into_inner(self) -> T { + self.0 + } +} + impl Archive<&T> where T: rkyv::Portable @@ -486,3 +664,9 @@ impl From> for Archive { unimplemented!() } } + +impl From> for UnversionedArchive { + fn from(_: Value<'static>) -> Self { + unimplemented!() + } +} diff --git a/crates/utils/src/snowflake.rs b/crates/utils/src/snowflake.rs index 3ea86e8c..910ab19f 100644 --- a/crates/utils/src/snowflake.rs +++ b/crates/utils/src/snowflake.rs @@ -15,7 +15,6 @@ use std::{ #[derive(Debug)] pub struct SnowflakeIdGenerator { epoch: SystemTime, - last_timestamp: AtomicU64, node_id: u64, sequence: AtomicU64, } @@ -31,7 +30,7 @@ const NODE_ID_MASK: u64 = (1 << NODE_ID_LEN) - 1; const DEFAULT_EPOCH: u64 = 1632280000; // 52 years after UNIX_EPOCH const DEFAULT_EPOCH_MS: u128 = (DEFAULT_EPOCH as u128) * 1000; // 52 years after UNIX_EPOCH in milliseconds -const MAX_CLOCK_DRIFT: i64 = 60 * 1000; // 1 minute +const MAX_CLOCK_DRIFT: i64 = 2000; // 2 seconds static LOGICAL_TIME: AtomicU64 = AtomicU64::new(0); static CHANGE_SEQ: AtomicU64 = AtomicU64::new(0); @@ -73,7 +72,6 @@ impl SnowflakeIdGenerator { epoch: SystemTime::UNIX_EPOCH + Duration::from_secs(DEFAULT_EPOCH), // 52 years after UNIX_EPOCH node_id, sequence: 0.into(), - last_timestamp: 0.into(), } } @@ -92,22 +90,15 @@ impl SnowflakeIdGenerator { #[inline(always)] pub fn generate(&self) -> u64 { - let current_elapsed = self + let elapsed = self .epoch .elapsed() .map(|e| e.as_millis()) .unwrap_or_default() as u64; - let last_elapsed = self - .last_timestamp - .fetch_max(current_elapsed, Ordering::Relaxed); - let (elapsed, sequence) = if current_elapsed > last_elapsed { - (current_elapsed, 0) - } else { - (last_elapsed, self.sequence.fetch_add(1, Ordering::Relaxed)) - }; + let sequence = self.sequence.fetch_add(1, Ordering::Relaxed) & SEQUENCE_MASK; (elapsed << (SEQUENCE_LEN + NODE_ID_LEN)) - | ((sequence & SEQUENCE_MASK) << NODE_ID_LEN) + | (sequence << NODE_ID_LEN) | (self.node_id & NODE_ID_MASK) } } @@ -139,20 +130,18 @@ impl HlcTimestamp { pub fn generate() -> u64 { let node_id = *NODE_MUM; - let current_elapsed = SystemTime::UNIX_EPOCH + let elapsed = SystemTime::UNIX_EPOCH .elapsed() .map(|e| e.as_millis()) .unwrap_or_default() .saturating_sub(DEFAULT_EPOCH_MS) as u64; - let last_elapsed = LOGICAL_TIME.fetch_max(current_elapsed, Ordering::SeqCst); - let (elapsed, sequence) = if current_elapsed > last_elapsed { - (current_elapsed, 0) - } else { - (last_elapsed, CHANGE_SEQ.fetch_add(1, Ordering::Relaxed)) - }; + let elapsed = LOGICAL_TIME + .fetch_max(elapsed, Ordering::SeqCst) + .max(elapsed); + let sequence = CHANGE_SEQ.fetch_add(1, Ordering::Relaxed) & SEQUENCE_MASK; (elapsed << (SEQUENCE_LEN + NODE_ID_LEN)) - | ((sequence & SEQUENCE_MASK) << NODE_ID_LEN) + | (sequence << NODE_ID_LEN) | (node_id as u64 & NODE_ID_MASK) } } @@ -168,7 +157,6 @@ impl Clone for SnowflakeIdGenerator { Self { epoch: self.epoch, node_id: self.node_id, - last_timestamp: 0.into(), sequence: 0.into(), } } diff --git a/tests/src/imap/mod.rs b/tests/src/imap/mod.rs index db29d51d..868acf40 100644 --- a/tests/src/imap/mod.rs +++ b/tests/src/imap/mod.rs @@ -452,9 +452,10 @@ impl AssertResult for Vec { } if match_all && match_count != self.len() - 1 { panic!( - "Expected {} mailboxes, but got {}", + "Expected {} mailboxes, but got {}: {:?}", match_count, - self.len() - 1 + self.len() - 1, + self.iter().collect::>() ); } self diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 20d86376..219f6b52 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -101,7 +101,7 @@ async fn jmap_tests_() { .await; webhooks::test(&mut params).await; - email_query::test(&mut params, delete).await; + /*email_query::test(&mut params, delete).await; email_get::test(&mut params).await; email_set::test(&mut params).await; email_parse::test(&mut params).await; @@ -113,7 +113,7 @@ async fn jmap_tests_() { thread_merge::test(&mut params).await; mailbox::test(&mut params).await; delivery::test(&mut params).await; - auth_acl::test(&mut params).await; + auth_acl::test(&mut params).await;*/ auth_limits::test(&mut params).await; auth_oauth::test(&mut params).await; event_source::test(&mut params).await;