diff --git a/crates/common/src/cache/principals.rs b/crates/common/src/cache/principals.rs index ed9bee0b..377e1f08 100644 --- a/crates/common/src/cache/principals.rs +++ b/crates/common/src/cache/principals.rs @@ -27,15 +27,13 @@ use registry::{ Role, SubAddressing, Tenant, }, }, - types::{ - id::ObjectId, - index::{IndexKey, IndexValue}, - }, + types::{EnumImpl, id::ObjectId}, }; use std::{borrow::Cow, sync::Arc}; use store::{ + U64_LEN, ValueKey, registry::{RegistryQuery, bootstrap::Bootstrap}, - write::{RegistryClass, now}, + write::{RegistryClass, ValueClass, key::KeySerializer, now}, }; use trc::AddContext; use types::id::Id; @@ -161,19 +159,20 @@ impl Server { .get(&EmailAddressRef::new(local_part, domain_id)) .is_none() { - let key = IndexKey::Global { - property: Property::Email, - value_1: IndexValue::Text(local_part.into()), - value_2: IndexValue::U64(domain_id.into()), - }; + let key = ValueKey::from(ValueClass::Registry(RegistryClass::PrimaryKey { + object_id: None, + index_id: Property::Email.to_id(), + key: KeySerializer::new(local_part.len() + U64_LEN) + .write(local_part.as_bytes()) + .write(domain_id as u64) + .finalize(), + })); + if let Some(object) = self - .registry() - .validate_primary_key( - RegistryClass::from_index_key(&key, 0, 0), - RegistryClass::from_index_key(&key, u16::MAX, u64::MAX), - None, - ) - .await? + .store() + .get_value::(key) + .await + .caused_by(trc::location!())? { let item_id = object.id().document_id(); let result = match object.object() { diff --git a/crates/common/src/config/telemetry.rs b/crates/common/src/config/telemetry.rs index 90223feb..37e5735b 100644 --- a/crates/common/src/config/telemetry.rs +++ b/crates/common/src/config/telemetry.rs @@ -508,7 +508,7 @@ impl Tracers { .copied() .unwrap_or(event_type.level()); if Level::Info.is_contained(event_level) { - global_interests.set(event_type); + global_interests.set(event_type.to_id() as usize); } } diff --git a/crates/common/src/ipc.rs b/crates/common/src/ipc.rs index 20b42a98..b6892d0b 100644 --- a/crates/common/src/ipc.rs +++ b/crates/common/src/ipc.rs @@ -154,6 +154,7 @@ pub struct DmarcEvent { pub report_record: Record, pub dmarc_record: Arc, pub interval: AggregateFrequency, + pub span_id: u64, } #[derive(Debug)] @@ -163,6 +164,7 @@ pub struct TlsEvent { pub failure: Option, pub tls_record: Arc, pub interval: AggregateFrequency, + pub span_id: u64, } #[derive(Debug, Hash, PartialEq, Eq)] diff --git a/crates/common/src/manager/backup.rs b/crates/common/src/manager/backup.rs index e1fd40ef..c2da32f5 100644 --- a/crates/common/src/manager/backup.rs +++ b/crates/common/src/manager/backup.rs @@ -316,7 +316,7 @@ impl Family { Family::Registry => &[ SUBSPACE_REGISTRY, SUBSPACE_REGISTRY_IDX, - SUBSPACE_REGISTRY_IDX_GLOBAL, + SUBSPACE_REGISTRY_PK, SUBSPACE_DIRECTORY, ], Family::Changelog => &[SUBSPACE_LOGS], diff --git a/crates/common/src/manager/console.rs b/crates/common/src/manager/console.rs index 08765c37..470f6974 100644 --- a/crates/common/src/manager/console.rs +++ b/crates/common/src/manager/console.rs @@ -9,10 +9,7 @@ use base64::engine::general_purpose; use std::env; use std::io::{self, Write}; use store::write::{AnyClass, AnyKey, BatchBuilder, ValueClass}; -use store::{ - Deserialize, IterateParams, SUBSPACE_INDEXES, SUBSPACE_REGISTRY_IDX, - SUBSPACE_REGISTRY_IDX_GLOBAL, Store, -}; +use store::{Deserialize, IterateParams, SUBSPACE_INDEXES, SUBSPACE_REGISTRY_IDX, Store}; const HELP: &str = concat!( "Stalwart Server v", @@ -76,12 +73,7 @@ pub async fn store_console(store: Store) { }, ) .set_values( - ![ - SUBSPACE_INDEXES, - SUBSPACE_REGISTRY_IDX, - SUBSPACE_REGISTRY_IDX_GLOBAL, - ] - .contains(&from_subspace), + ![SUBSPACE_INDEXES, SUBSPACE_REGISTRY_IDX].contains(&from_subspace), ), |key, value| { print!("{}", char::from(from_subspace)); diff --git a/crates/common/src/manager/restore.rs b/crates/common/src/manager/restore.rs index 5338e2c3..5e19df60 100644 --- a/crates/common/src/manager/restore.rs +++ b/crates/common/src/manager/restore.rs @@ -15,7 +15,7 @@ use std::{ }; use store::{ BlobStore, SUBSPACE_BLOBS, SUBSPACE_COUNTER, SUBSPACE_INDEXES, SUBSPACE_QUOTA, - SUBSPACE_REGISTRY_IDX, SUBSPACE_REGISTRY_IDX_GLOBAL, Store, U32_LEN, + SUBSPACE_REGISTRY_IDX, Store, U32_LEN, write::{AnyClass, BatchBuilder, ValueClass, key::DeserializeBigEndian}, }; use types::{collection::Collection, field::Field}; @@ -85,7 +85,7 @@ async fn restore_file(store: Store, blob_store: BlobStore, path: &Path) { } } } - SUBSPACE_INDEXES | SUBSPACE_REGISTRY_IDX | SUBSPACE_REGISTRY_IDX_GLOBAL => { + SUBSPACE_INDEXES | SUBSPACE_REGISTRY_IDX => { while let Some((key, _)) = reader.next() { let account_id = key .as_slice() diff --git a/crates/common/src/telemetry/metrics/store.rs b/crates/common/src/telemetry/metrics/store.rs index b172fc67..61710e87 100644 --- a/crates/common/src/telemetry/metrics/store.rs +++ b/crates/common/src/telemetry/metrics/store.rs @@ -8,40 +8,32 @@ * */ -use crate::Core; use ahash::AHashMap; use parking_lot::Mutex; -use serde::{Deserialize, Serialize}; +use registry::{ + pickle::Pickle, + schema::structs::{Metric, MetricCount, MetricSum}, +}; use std::{future::Future, sync::Arc, time::Duration}; use store::{ - IterateParams, Store, U32_LEN, U64_LEN, ValueKey, - write::{ - BatchBuilder, TelemetryClass, ValueClass, - key::{DeserializeBigEndian, KeySerializer}, - now, - }, + Store, ValueKey, + write::{BatchBuilder, TelemetryClass, ValueClass}, }; use trc::*; -use utils::codec::leb128::Leb128Reader; +use utils::snowflake::SnowflakeIdGenerator; pub trait MetricsStore: Sync + Send { fn write_metrics( &self, - core: Arc, - timestamp: u64, + timestamp: Option, history: SharedMetricHistory, ) -> impl Future> + Send; - fn query_metrics( - &self, - from_timestamp: u64, - to_timestamp: u64, - ) -> impl Future>>> + Send; fn purge_metrics(&self, period: Duration) -> impl Future> + Send; } #[derive(Default)] pub struct MetricsHistory { - events: AHashMap, + events: AHashMap, histograms: AHashMap, } @@ -51,81 +43,71 @@ struct HistogramHistory { count: u64, } -#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] -#[serde(tag = "type")] -#[serde(rename_all = "camelCase")] -pub enum Metric { - Counter { - id: CI, - timestamp: T, - value: u64, - }, - Gauge { - id: MI, - timestamp: T, - value: u64, - }, - Histogram { - id: MI, - timestamp: T, - count: u64, - sum: u64, - }, -} - pub type SharedMetricHistory = Arc>; -const TYPE_COUNTER: u64 = 0x00; -const TYPE_HISTOGRAM: u64 = 0x01; -const TYPE_GAUGE: u64 = 0x02; - impl MetricsStore for Store { async fn write_metrics( &self, - core: Arc, - timestamp: u64, + _timestamp: Option, history_: SharedMetricHistory, ) -> trc::Result<()> { let mut batch = BatchBuilder::new(); { - let node_id = core.network.node_id; let mut history = history_.lock(); for event in [ - EventType::Smtp(SmtpEvent::ConnectionStart), - EventType::Imap(ImapEvent::ConnectionStart), - EventType::Pop3(Pop3Event::ConnectionStart), - EventType::ManageSieve(ManageSieveEvent::ConnectionStart), - EventType::Http(HttpEvent::ConnectionStart), - EventType::Delivery(DeliveryEvent::AttemptStart), - EventType::Queue(QueueEvent::QueueMessage), - EventType::Queue(QueueEvent::QueueMessageAuthenticated), - EventType::Queue(QueueEvent::QueueDsn), - EventType::Queue(QueueEvent::QueueReport), - EventType::MessageIngest(MessageIngestEvent::Ham), - EventType::MessageIngest(MessageIngestEvent::Spam), - EventType::Auth(AuthEvent::Failed), - EventType::Security(SecurityEvent::AuthenticationBan), - EventType::Security(SecurityEvent::ScanBan), - EventType::Security(SecurityEvent::AbuseBan), - EventType::Security(SecurityEvent::LoiterBan), - EventType::Security(SecurityEvent::IpBlocked), - EventType::IncomingReport(IncomingReportEvent::DmarcReport), - EventType::IncomingReport(IncomingReportEvent::DmarcReportWithWarnings), - EventType::IncomingReport(IncomingReportEvent::TlsReport), - EventType::IncomingReport(IncomingReportEvent::TlsReportWithWarnings), + MetricType::SmtpConnectionStart, + MetricType::ImapConnectionStart, + MetricType::Pop3ConnectionStart, + MetricType::ManageSieveConnectionStart, + MetricType::HttpConnectionStart, + MetricType::DeliveryAttemptStart, + MetricType::QueueQueueMessage, + MetricType::QueueQueueMessageAuthenticated, + MetricType::QueueQueueDsn, + MetricType::QueueQueueReport, + MetricType::MessageIngestHam, + MetricType::MessageIngestSpam, + MetricType::AuthFailed, + MetricType::SecurityAuthenticationBan, + MetricType::SecurityScanBan, + MetricType::SecurityAbuseBan, + MetricType::SecurityLoiterBan, + MetricType::SecurityIpBlocked, + MetricType::IncomingReportDmarcReport, + MetricType::IncomingReportDmarcReportWithWarnings, + MetricType::IncomingReportTlsReport, + MetricType::IncomingReportTlsReportWithWarnings, ] { - let reading = Collector::read_metric_counter(event.to_id() as usize); + let reading = Collector::read_metric_counter(event.event_id()); if reading > 0 { let history = history.events.entry(event).or_insert(0); let diff = reading - *history; + + #[cfg(not(feature = "test_mode"))] + let metric_id = SnowflakeIdGenerator::from_sequence_id(event.to_id() as u64) + .unwrap_or_default(); + + #[cfg(feature = "test_mode")] + let metric_id = _timestamp + .map(|timestamp| { + SnowflakeIdGenerator::from_timestamp_and_sequence_id( + timestamp, + event.to_id() as u64, + ) + }) + .unwrap_or_else(|| { + SnowflakeIdGenerator::from_sequence_id(event.to_id() as u64) + }) + .unwrap_or_default(); + if diff > 0 { batch.set( - ValueClass::Telemetry(TelemetryClass::Metric { - timestamp, - metric_id: (event.to_id() << 2) as u64 | TYPE_COUNTER, - node_id, - }), - KeySerializer::new(U32_LEN).write_leb128(diff).finalize(), + ValueClass::Telemetry(TelemetryClass::Metric(metric_id)), + Metric::Counter(MetricCount { + count: diff as u64, + metric: event, + }) + .to_pickled_vec(), ); } *history = reading; @@ -133,48 +115,77 @@ impl MetricsStore for Store { } for gauge in Collector::collect_gauges(true) { - let gauge_id = gauge.id(); - if matches!(gauge_id, MetricType::QueueCount | MetricType::ServerMemory) { + let metric = gauge.id(); + if matches!(metric, MetricType::QueueCount | MetricType::ServerMemory) { let value = gauge.get(); if value > 0 { + #[cfg(not(feature = "test_mode"))] + let metric_id = + SnowflakeIdGenerator::from_sequence_id(metric.to_id() as u64) + .unwrap_or_default(); + + #[cfg(feature = "test_mode")] + let metric_id = _timestamp + .map(|timestamp| { + SnowflakeIdGenerator::from_timestamp_and_sequence_id( + timestamp, + metric.to_id() as u64, + ) + }) + .unwrap_or_else(|| { + SnowflakeIdGenerator::from_sequence_id(metric.to_id() as u64) + }) + .unwrap_or_default(); + batch.set( - ValueClass::Telemetry(TelemetryClass::Metric { - timestamp, - metric_id: (gauge_id.to_id() << 2) as u64 | TYPE_GAUGE, - node_id, - }), - KeySerializer::new(U32_LEN).write_leb128(value).finalize(), + ValueClass::Telemetry(TelemetryClass::Metric(metric_id)), + Metric::Gauge(MetricCount { + count: value, + metric, + }) + .to_pickled_vec(), ); } } } for histogram in Collector::collect_histograms(true) { - let histogram_id = histogram.id(); + let metric = histogram.id(); if matches!( - histogram_id, + metric, MetricType::MessageIngestTime | MetricType::MessageIngestIndexTime | MetricType::DeliveryTotalTime | MetricType::DeliveryAttemptTime | MetricType::DnsLookupTime ) { - let history = history.histograms.entry(histogram_id).or_default(); + let history = history.histograms.entry(metric).or_default(); let sum = histogram.sum(); let count = histogram.count(); let diff_sum = sum - history.sum; let diff_count = count - history.count; if diff_sum > 0 || diff_count > 0 { + #[cfg(not(feature = "test_mode"))] + let metric_id = + SnowflakeIdGenerator::from_sequence_id(metric.to_id() as u64) + .unwrap_or_default(); + + #[cfg(feature = "test_mode")] + let metric_id = _timestamp + .map(|timestamp| { + SnowflakeIdGenerator::from_timestamp_and_sequence_id( + timestamp, + metric.to_id() as u64, + ) + }) + .unwrap_or_else(|| { + SnowflakeIdGenerator::from_sequence_id(metric.to_id() as u64) + }) + .unwrap_or_default(); + batch.set( - ValueClass::Telemetry(TelemetryClass::Metric { - timestamp, - metric_id: (histogram_id.to_id() << 2) as u64 | TYPE_HISTOGRAM, - node_id, - }), - KeySerializer::new(U32_LEN) - .write_leb128(diff_count) - .write_leb128(diff_sum) - .finalize(), + ValueClass::Telemetry(TelemetryClass::Metric(metric_id)), + Metric::Histogram(MetricSum { count, metric, sum }).to_pickled_vec(), ); } history.sum = sum; @@ -192,105 +203,16 @@ impl MetricsStore for Store { Ok(()) } - async fn query_metrics( - &self, - from_timestamp: u64, - to_timestamp: u64, - ) -> trc::Result>> { - let mut metrics = Vec::new(); - self.iterate( - IterateParams::new( - ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric { - timestamp: from_timestamp, - metric_id: 0, - node_id: 0, - })), - ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric { - timestamp: to_timestamp, - metric_id: 0, - node_id: 0, - })), - ), - |key, value| { - let timestamp = key.deserialize_be_u64(0).caused_by(trc::location!())?; - let (metric_type, _) = key - .get(U64_LEN..) - .and_then(|bytes| bytes.read_leb128::()) - .ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?; - match metric_type & 0x03 { - TYPE_COUNTER => { - let id = - MetricType::from_id((metric_type >> 2) as u16).ok_or_else(|| { - trc::Error::corrupted_key(key, None, trc::location!()) - })?; - let (value, _) = value.read_leb128::().ok_or_else(|| { - trc::Error::corrupted_key(key, value.into(), trc::location!()) - })?; - metrics.push(Metric::Counter { - id, - timestamp, - value, - }); - } - TYPE_HISTOGRAM => { - let id = - MetricType::from_id((metric_type >> 2) as u16).ok_or_else(|| { - trc::Error::corrupted_key(key, None, trc::location!()) - })?; - let (count, bytes_read) = value.read_leb128::().ok_or_else(|| { - trc::Error::corrupted_key(key, value.into(), trc::location!()) - })?; - let (sum, _) = value - .get(bytes_read..) - .and_then(|bytes| bytes.read_leb128::()) - .ok_or_else(|| { - trc::Error::corrupted_key(key, value.into(), trc::location!()) - })?; - metrics.push(Metric::Histogram { - id, - timestamp, - count, - sum, - }); - } - TYPE_GAUGE => { - let id = - MetricType::from_id((metric_type >> 2) as u16).ok_or_else(|| { - trc::Error::corrupted_key(key, None, trc::location!()) - })?; - let (value, _) = value.read_leb128::().ok_or_else(|| { - trc::Error::corrupted_key(key, value.into(), trc::location!()) - })?; - metrics.push(Metric::Gauge { - id, - timestamp, - value, - }); - } - _ => return Err(trc::Error::corrupted_key(key, None, trc::location!())), - } - - Ok(true) - }, - ) - .await - .caused_by(trc::location!())?; - - Ok(metrics) - } - async fn purge_metrics(&self, period: Duration) -> trc::Result<()> { + let until_span_id = SnowflakeIdGenerator::from_duration(period).ok_or_else(|| { + trc::StoreEvent::UnexpectedError + .caused_by(trc::location!()) + .ctx(trc::Key::Reason, "Failed to generate reference metric id.") + })?; + self.delete_range( - ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric { - timestamp: 0, - metric_id: 0, - node_id: 0, - })), - ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric { - timestamp: now() - period.as_secs(), - metric_id: 0, - node_id: 0, - })), + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric(0))), + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric(until_span_id))), ) .await .caused_by(trc::location!()) diff --git a/crates/common/src/telemetry/tracers/store.rs b/crates/common/src/telemetry/tracers/store.rs index 9118ebe7..9779e19e 100644 --- a/crates/common/src/telemetry/tracers/store.rs +++ b/crates/common/src/telemetry/tracers/store.rs @@ -10,18 +10,25 @@ use crate::config::telemetry::StoreTracer; use ahash::{AHashMap, AHashSet}; -use registry::schema::structs::{Task, TaskIndexTrace, TaskStatus}; +use registry::{ + pickle::Pickle, + schema::structs::{ + Task, TaskIndexTrace, TaskStatus, Trace, TraceEvent, TraceKeyValue, TraceValue, + TraceValueBoolean, TraceValueDuration, TraceValueEvent, TraceValueFloat, TraceValueInteger, + TraceValueIpAddr, TraceValueList, TraceValueString, TraceValueUTCDateTime, + TraceValueUnsignedInt, + }, + types::{datetime::UTCDateTime, ipaddr::IpAddr}, +}; use std::{collections::HashSet, future::Future, time::Duration}; use store::{ - Deserialize, SearchStore, Store, ValueKey, + SearchStore, Store, ValueKey, search::{IndexDocument, SearchField, SearchFilter, SearchQuery, TracingSearchField}, write::{BatchBuilder, SearchIndex, TelemetryClass, ValueClass}, }; use trc::{ AddContext, AuthEvent, Event, EventDetails, EventType, Key, MessageIngestEvent, - OutgoingReportEvent, QueueEvent, Value, - ipc::subscriber::SubscriberBuilder, - serializers::binary::{deserialize_events, serialize_events}, + OutgoingReportEvent, QueueEvent, Value, ipc::subscriber::SubscriberBuilder, }; use utils::snowflake::SnowflakeIdGenerator; @@ -53,14 +60,15 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac // Serialize events batch .set( - ValueClass::Telemetry(TelemetryClass::Span { span_id }), - serialize_events( + ValueClass::Telemetry(TelemetryClass::Span(span_id)), + map_events( [span.as_ref()] .into_iter() .chain(events.iter().map(|event| event.as_ref())) .chain([event.as_ref()].into_iter()), events.len() + 2, - ), + ) + .to_pickled_vec(), ) .schedule_task(Task::IndexTrace(TaskIndexTrace { status: TaskStatus::now(), @@ -80,15 +88,72 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac }); } +fn map_events<'x>( + span_events: impl IntoIterator>, + num_events: usize, +) -> Trace { + let mut events = Vec::with_capacity(num_events); + + for event in span_events { + let mut key_values = Vec::with_capacity(event.keys.len()); + for (key, value) in &event.keys { + key_values.push(TraceKeyValue { + key: *key, + value: map_value(value), + }); + } + + events.push(TraceEvent { + event: event.inner.typ, + timestamp: UTCDateTime::from_timestamp(event.inner.timestamp as i64), + key_values, + }); + } + + Trace { events } +} + +fn map_value(value: &Value) -> TraceValue { + match value { + Value::String(value) => TraceValue::String(TraceValueString { + value: value.to_string(), + }), + Value::UInt(value) => TraceValue::UnsignedInt(TraceValueUnsignedInt { value: *value }), + Value::Int(value) => TraceValue::Integer(TraceValueInteger { value: *value }), + Value::Float(value) => TraceValue::Float(TraceValueFloat { value: *value }), + Value::Timestamp(value) => TraceValue::UTCDateTime(TraceValueUTCDateTime { + value: UTCDateTime::from_timestamp(*value as i64), + }), + Value::Duration(value) => TraceValue::Duration(TraceValueDuration { value: *value }), + Value::Bytes(items) => TraceValue::String(TraceValueString { + value: String::from_utf8_lossy(items).to_string(), + }), + Value::Bool(value) => TraceValue::Boolean(TraceValueBoolean { value: *value }), + Value::Ipv4(ipv4_addr) => TraceValue::IpAddr(TraceValueIpAddr { + value: IpAddr((*ipv4_addr).into()), + }), + Value::Ipv6(ipv6_addr) => TraceValue::IpAddr(TraceValueIpAddr { + value: IpAddr((*ipv6_addr).into()), + }), + Value::Event(event) => TraceValue::Event(TraceValueEvent { + value: event + .keys() + .iter() + .map(|(k, v)| TraceKeyValue { + key: *k, + value: map_value(v), + }) + .collect(), + event: event.event_type(), + }), + Value::Array(values) => TraceValue::List(TraceValueList { + value: values.iter().map(map_value).collect::>(), + }), + Value::None => TraceValue::Null, + } +} + pub trait TracingStore: Sync + Send { - fn get_span( - &self, - span_id: u64, - ) -> impl Future>>> + Send; - fn get_raw_span( - &self, - span_id: u64, - ) -> impl Future>>> + Send; fn purge_spans( &self, period: Duration, @@ -97,24 +162,6 @@ pub trait TracingStore: Sync + Send { } impl TracingStore for Store { - async fn get_span(&self, span_id: u64) -> trc::Result>> { - self.get_value::(ValueKey::from(ValueClass::Telemetry( - TelemetryClass::Span { span_id }, - ))) - .await - .caused_by(trc::location!()) - .map(|span| span.map(|span| span.0).unwrap_or_default()) - } - - async fn get_raw_span(&self, span_id: u64) -> trc::Result>> { - self.get_value::(ValueKey::from(ValueClass::Telemetry( - TelemetryClass::Span { span_id }, - ))) - .await - .caused_by(trc::location!()) - .map(|span| span.map(|span| span.0)) - } - async fn purge_spans( &self, period: Duration, @@ -127,10 +174,8 @@ impl TracingStore for Store { })?; self.delete_range( - ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span { span_id: 0 })), - ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span { - span_id: until_span_id, - })), + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span(0))), + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span(until_span_id))), ) .await .caused_by(trc::location!())?; @@ -218,76 +263,57 @@ impl StoreTracer { } } -struct RawSpan(Vec); -struct Span(Vec>); - -impl Deserialize for Span { - fn deserialize(bytes: &[u8]) -> trc::Result { - deserialize_events(bytes).map(Self) - } -} - -impl Deserialize for RawSpan { - fn deserialize(bytes: &[u8]) -> trc::Result { - Ok(Self(bytes.to_vec())) - } -} - pub fn build_span_document( span_id: u64, - events: Vec>, + trace: Trace, index_fields: &AHashSet, ) -> IndexDocument { let mut document = IndexDocument::new(SearchIndex::Tracing).with_id(span_id); let mut keywords = HashSet::new(); - for (idx, event) in events.into_iter().enumerate() { + for (idx, event) in trace.events.into_iter().enumerate() { if idx == 0 && (index_fields.is_empty() || index_fields.contains(&TracingSearchField::EventType.into())) { - document.index_unsigned(TracingSearchField::EventType, event.inner.typ.to_id()); + document.index_unsigned(TracingSearchField::EventType, event.event.to_id()); } - for (key, value) in event.keys { + for TraceKeyValue { key, value } in event.key_values { match (key, value) { - (Key::QueueId, Value::UInt(queue_id)) => { + (Key::QueueId, TraceValue::UnsignedInt(TraceValueUnsignedInt { value })) => { if index_fields.is_empty() || index_fields.contains(&TracingSearchField::QueueId.into()) { - document.index_unsigned(TracingSearchField::QueueId, queue_id); + document.index_unsigned(TracingSearchField::QueueId, value); } } - (Key::From | Key::To | Key::Domain | Key::Hostname, Value::String(address)) => { + ( + Key::From | Key::To | Key::Domain | Key::Hostname, + TraceValue::String(TraceValueString { value }), + ) => { if index_fields.is_empty() || index_fields.contains(&TracingSearchField::Keywords.into()) { - keywords.insert(address.to_string()); + keywords.insert(value); } } - (Key::To, Value::Array(value)) => { + (Key::To, TraceValue::List(TraceValueList { value })) => { if index_fields.is_empty() || index_fields.contains(&TracingSearchField::Keywords.into()) { for value in value { - if let Value::String(address) = value { - keywords.insert(address.to_string()); + if let TraceValue::String(TraceValueString { value }) = value { + keywords.insert(value); } } } } - (Key::RemoteIp, Value::Ipv4(ip)) => { + (Key::RemoteIp, TraceValue::IpAddr(TraceValueIpAddr { value })) => { if index_fields.is_empty() || index_fields.contains(&TracingSearchField::Keywords.into()) { - keywords.insert(ip.to_string()); - } - } - (Key::RemoteIp, Value::Ipv6(ip)) => { - if index_fields.is_empty() - || index_fields.contains(&TracingSearchField::Keywords.into()) - { - keywords.insert(ip.to_string()); + keywords.insert(value.to_string()); } } diff --git a/crates/http/src/management/telemetry.rs b/crates/http/src/management/telemetry.rs index d30a1c1e..54715fd2 100644 --- a/crates/http/src/management/telemetry.rs +++ b/crates/http/src/management/telemetry.rs @@ -11,10 +11,6 @@ use common::{ Server, auth::{AccessToken, oauth::GrantType}, - telemetry::{ - metrics::store::{Metric, MetricsStore}, - tracers::store::TracingStore, - }, }; use http_body_util::{StreamBody, combinators::BoxBody}; use http_proto::*; @@ -25,22 +21,18 @@ use hyper::{ use mail_parser::DateTime; use registry::schema::enums::Permission; use serde_json::json; +use std::future::Future; use std::{ fmt::Write, time::{Duration, Instant}, }; -use std::{future::Future, str::FromStr}; -use store::{ - ahash::{AHashMap, AHashSet}, - search::{SearchComparator, SearchField, SearchFilter, SearchQuery, TracingSearchField}, - write::{SearchIndex, now}, -}; +use store::ahash::{AHashMap, AHashSet}; use trc::{ - Collector, DeliveryEvent, EventType, Key, MetricType, QueueEvent, Value, + Collector, EventType, Key, MetricType, Value, ipc::{bitset::Bitset, subscriber::SubscriberBuilder}, serializers::json::JsonEventSerializer, }; -use utils::{snowflake::SnowflakeIdGenerator, url_params::UrlParams}; +use utils::url_params::UrlParams; pub trait TelemetryApi: Sync + Send { fn handle_telemetry_api_request( @@ -60,165 +52,13 @@ impl TelemetryApi for Server { ) -> trc::Result { let params = UrlParams::new(req.uri().query()); let account_id = access_token.account_id(); + let todo = "use same format as in JMAP API"; match ( path.get(1).copied().unwrap_or_default(), path.get(2).copied(), req.method(), ) { - ("traces", None, &Method::GET) => { - // Validate the access token - access_token.enforce_permission(Permission::TracingList)?; - - let page: usize = params.parse("page").unwrap_or(0); - let limit: usize = params.parse("limit").unwrap_or(0); - let mut tracing_query = Vec::new(); - tracing_query.push(SearchFilter::And); - if let Some(typ) = params.parse::("type") { - tracing_query.push(SearchFilter::eq( - TracingSearchField::EventType, - typ.to_id() as u64, - )); - } - if let Some(queue_id) = params.parse::("queue_id") { - tracing_query.push(SearchFilter::eq(TracingSearchField::QueueId, queue_id)); - } - if let Some(query) = params.get("filter") { - let mut buf = String::with_capacity(query.len()); - let mut in_quote = false; - for ch in query.chars() { - if ch.is_ascii_whitespace() { - if in_quote { - buf.push(' '); - } else if !buf.is_empty() { - tracing_query.push(SearchFilter::has_keyword( - TracingSearchField::Keywords, - buf, - )); - buf = String::new(); - } - } else if ch == '"' { - buf.push(ch); - if in_quote { - if !buf.is_empty() { - tracing_query.push(SearchFilter::has_keyword( - TracingSearchField::Keywords, - buf, - )); - buf = String::new(); - } - in_quote = false; - } else { - in_quote = true; - } - } else { - buf.push(ch); - } - } - if !buf.is_empty() { - tracing_query - .push(SearchFilter::has_keyword(TracingSearchField::Keywords, buf)); - } - } - let values = params.get("values").is_some(); - if let Some(before) = params - .parse::("before") - .map(|t| t.into_inner()) - .and_then(SnowflakeIdGenerator::from_timestamp) - { - tracing_query.push(SearchFilter::lt(SearchField::Id, before)); - } - if let Some(after) = params - .parse::("after") - .map(|t| t.into_inner()) - .and_then(SnowflakeIdGenerator::from_timestamp) - { - tracing_query.push(SearchFilter::gt(SearchField::Id, after)); - } - if !tracing_query.iter().any(|f| { - matches!( - f, - SearchFilter::Operator { - field: SearchField::Tracing( - TracingSearchField::Keywords | TracingSearchField::QueueId - ) | SearchField::Id, - .. - } - ) - }) { - tracing_query.push(SearchFilter::gt( - SearchField::Id, - SnowflakeIdGenerator::from_timestamp(now() - 86400).unwrap_or_default(), - )); - } - - tracing_query.push(SearchFilter::End); - - let store = self.tracing_store(); - - if !store.is_active() { - return Err(trc::ManageEvent::NotSupported - .ctx(trc::Key::Details, "No tracing store has been configured")); - } - - let span_ids = self - .search_store() - .query_global( - SearchQuery::new(SearchIndex::Tracing) - .with_filters(tracing_query) - .with_comparator(SearchComparator::Field { - field: SearchField::Id, - ascending: false, - }), - ) - .await?; - - let (total, span_ids) = if limit > 0 { - let offset = page.saturating_sub(1) * limit; - ( - span_ids.len(), - span_ids.into_iter().skip(offset).take(limit).collect(), - ) - } else { - (span_ids.len(), span_ids) - }; - - if values && !span_ids.is_empty() { - let mut values = Vec::with_capacity(span_ids.len()); - - for span_id in span_ids { - for event in store.get_span(span_id).await? { - if matches!( - event.inner.typ, - EventType::Delivery(DeliveryEvent::AttemptStart) - | EventType::Queue( - QueueEvent::QueueMessage - | QueueEvent::QueueMessageAuthenticated - ) - ) { - values.push(event); - break; - } - } - } - - Ok(JsonResponse::new(json!({ - "data": { - "items": JsonEventSerializer::new(values).with_spans(), - "total": total, - }, - })) - .into_http_response()) - } else { - Ok(JsonResponse::new(json!({ - "data": { - "items": span_ids, - "total": total, - }, - })) - .into_http_response()) - } - } ("traces", Some("live"), &Method::GET) => { // Validate the access token access_token.enforce_permission(Permission::TracingLive)?; @@ -346,45 +186,6 @@ impl TelemetryApi for Server { }, )))) } - ("trace", id, &Method::GET) => { - // Validate the access token - access_token.enforce_permission(Permission::TracingGet)?; - - let store = self.tracing_store(); - if !store.is_active() { - return Err(trc::ManageEvent::NotSupported - .ctx(trc::Key::Details, "No tracing store has been configured")); - } - - let mut events = Vec::new(); - for span_id in id - .or_else(|| params.get("id")) - .unwrap_or_default() - .split(',') - { - if let Ok(span_id) = span_id.parse::() { - events.push( - JsonEventSerializer::new(store.get_span(span_id).await?) - .with_description() - .with_explanation(), - ); - } else { - events.push(JsonEventSerializer::new(Vec::new())); - } - } - - if events.len() == 1 && id.is_some() { - Ok(JsonResponse::new(json!({ - "data": events.into_iter().next().unwrap(), - })) - .into_http_response()) - } else { - Ok(JsonResponse::new(json!({ - "data": events, - })) - .into_http_response()) - } - } ("live", Some("tracing-token"), &Method::GET) => { // Validate the access token access_token.enforce_permission(Permission::TracingLive)?; @@ -405,74 +206,6 @@ impl TelemetryApi for Server { })) .into_http_response()) } - ("metrics", None, &Method::GET) => { - let todo = "move to registry"; - // Validate the access token - access_token.enforce_permission(Permission::MetricsList)?; - - let before = params - .parse::("before") - .map(|t| t.into_inner()) - .unwrap_or(u64::MAX); - let after = params - .parse::("after") - .map(|t| t.into_inner()) - .unwrap_or(0); - - if !self.metrics_store().is_active() { - return Err(trc::ManageEvent::Error - .ctx(trc::Key::Details, "No metrics store has been defined") - .ctx( - trc::Key::Reason, - concat!( - "You need to configure a metrics ", - "store in order to use this feature." - ), - )); - } - - let results = self.metrics_store().query_metrics(after, before).await?; - let mut metrics = Vec::with_capacity(results.len()); - - for metric in results { - metrics.push(match metric { - Metric::Counter { - id, - timestamp, - value, - } => Metric::Counter { - id: id.as_str().to_string(), - timestamp: DateTime::from_timestamp(timestamp as i64).to_rfc3339(), - value, - }, - Metric::Histogram { - id, - timestamp, - count, - sum, - } => Metric::Histogram { - id: id.as_str(), - timestamp: DateTime::from_timestamp(timestamp as i64).to_rfc3339(), - count, - sum, - }, - Metric::Gauge { - id, - timestamp, - value, - } => Metric::Gauge { - id: id.as_str(), - timestamp: DateTime::from_timestamp(timestamp as i64).to_rfc3339(), - value, - }, - }); - } - - Ok(JsonResponse::new(json!({ - "data": metrics, - })) - .into_http_response()) - } ("metrics", Some("live"), &Method::GET) => { // Validate the access token access_token.enforce_permission(Permission::MetricsLive)?; @@ -581,23 +314,3 @@ impl TelemetryApi for Server { } } } - -pub(super) struct Timestamp(u64); - -impl FromStr for Timestamp { - type Err = (); - - fn from_str(s: &str) -> Result { - if let Some(dt) = DateTime::parse_rfc3339(s) { - Ok(Timestamp(dt.to_timestamp() as u64)) - } else { - Err(()) - } - } -} - -impl Timestamp { - pub fn into_inner(self) -> u64 { - self.0 - } -} diff --git a/crates/jmap/src/registry/get.rs b/crates/jmap/src/registry/get.rs index 550c6a0d..1acb64fe 100644 --- a/crates/jmap/src/registry/get.rs +++ b/crates/jmap/src/registry/get.rs @@ -274,6 +274,8 @@ impl RegistryGet for Server { ObjectType::Metric => {} ObjectType::Trace => {} ObjectType::SpamTrainingSample => {} + ObjectType::DmarcInternalReport => todo!(), + ObjectType::TlsInternalReport => todo!(), } Ok(response) diff --git a/crates/jmap/src/registry/mapping/mod.rs b/crates/jmap/src/registry/mapping/mod.rs new file mode 100644 index 00000000..73bfc732 --- /dev/null +++ b/crates/jmap/src/registry/mapping/mod.rs @@ -0,0 +1,7 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +pub mod queued_message; diff --git a/crates/jmap/src/registry/mapping/queued_message.rs b/crates/jmap/src/registry/mapping/queued_message.rs new file mode 100644 index 00000000..3a672468 --- /dev/null +++ b/crates/jmap/src/registry/mapping/queued_message.rs @@ -0,0 +1,166 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use common::{Server, config::smtp::queue::ArchivedQueueExpiry}; +use registry::{ + schema::{ + enums::{DeliveryErrorType, MessageFlag, RecipientFlag}, + structs::{ + DeliveryError, QueueExpiry, QueueExpiryAttempts, QueueExpiryTtl, QueuedMessage, + QueuedRecipient, RecipientStatus, ServerResponse, + }, + }, + types::{datetime::UTCDateTime, ipaddr::IpAddr}, +}; +use smtp::queue::{spool::SmtpSpool, *}; +use types::{blob::BlobId, blob_hash::BlobHash}; + +pub(crate) async fn queued_message_fetch( + server: &Server, + queue_id: u64, +) -> trc::Result> { + let Some(message_archive) = server.read_message_archive(queue_id).await? else { + return Ok(None); + }; + let message_in = message_archive.unarchive::()?; + let mut message_out = QueuedMessage { + blob_id: BlobId::new(BlobHash::from(&message_in.blob_hash), Default::default()), + created_at: UTCDateTime::from_timestamp(message_in.created.to_native() as i64), + env_id: message_in.env_id.as_ref().map(|v| v.to_string()), + flags: Vec::with_capacity(1), + priority: message_in.priority.to_native() as i64, + received_from_ip: IpAddr(message_in.received_from_ip.as_ipaddr()), + received_via_port: message_in.received_via_port.to_native() as u64, + recipients: Vec::with_capacity(message_in.recipients.len()), + return_path: message_in.return_path.to_string(), + size: message_in.size.to_native(), + }; + + // Parse flags + let flags = message_in.flags.to_native(); + for (bit, flag) in [ + (FROM_AUTHENTICATED, MessageFlag::Authenticated), + (FROM_UNAUTHENTICATED, MessageFlag::Unauthenticated), + ( + FROM_UNAUTHENTICATED_DMARC, + MessageFlag::UnauthenticatedDmarc, + ), + (FROM_DSN, MessageFlag::Dsn), + (FROM_REPORT, MessageFlag::Report), + (FROM_AUTOGENERATED, MessageFlag::Autogenerated), + ] { + if flags & bit != 0 { + message_out.flags.push(flag); + } + } + + // Parse recipients + for rcpt_in in message_in.recipients.iter() { + let mut rcpt_out = QueuedRecipient { + address: rcpt_in.address.to_string(), + expires: match &rcpt_in.expires { + ArchivedQueueExpiry::Ttl(ttl) => QueueExpiry::Ttl(QueueExpiryTtl { + expires_at: UTCDateTime::from_timestamp(ttl.to_native() as i64), + }), + ArchivedQueueExpiry::Attempts(attempts) => { + QueueExpiry::Attempts(QueueExpiryAttempts { + expires_attempts: attempts.to_native() as u64, + }) + } + }, + flags: vec![], + notify_count: rcpt_in.notify.inner.to_native() as u64, + notify_due: UTCDateTime::from_timestamp(rcpt_in.notify.due.to_native() as i64), + orcpt: rcpt_in.orcpt.as_ref().map(|v| v.to_string()), + queue_name: rcpt_in.queue.as_str().to_string(), + retry_count: rcpt_in.retry.inner.to_native() as u64, + retry_due: UTCDateTime::from_timestamp(rcpt_in.retry.due.to_native() as i64), + status: match &rcpt_in.status { + ArchivedStatus::Scheduled => RecipientStatus::Scheduled, + ArchivedStatus::Completed(status) => RecipientStatus::Completed(ServerResponse { + response_code: (status.response.code.to_native() as u64).into(), + response_enhanced: build_enhanced_code(&status.response.esc).into(), + response_hostname: status.hostname.to_string().into(), + response_message: status.response.message.to_string().into(), + }), + ArchivedStatus::TemporaryFailure(status) => { + RecipientStatus::TemporaryFailure(map_error_details(status)) + } + ArchivedStatus::PermanentFailure(status) => { + RecipientStatus::PermanentFailure(map_error_details(status)) + } + }, + }; + + // Parse recipient flags + let rcpt_flags = rcpt_in.flags.to_native(); + for (bit, flag) in [ + (RCPT_DSN_SENT, RecipientFlag::DsnSent), + (RCPT_SPAM_PAYLOAD, RecipientFlag::SpamPayload), + ] { + if rcpt_flags & bit != 0 { + rcpt_out.flags.push(flag); + } + } + + message_out.recipients.push(rcpt_out); + } + + Ok(Some(message_out)) +} + +fn map_error_details(err_in: &ArchivedErrorDetails) -> DeliveryError { + let mut err_out = DeliveryError { + response_hostname: err_in.entity.to_string().into(), + ..Default::default() + }; + + match &err_in.details { + ArchivedError::DnsError(e) => { + err_out.error_type = DeliveryErrorType::DnsError; + err_out.error_message = e.to_string().into(); + } + ArchivedError::UnexpectedResponse(e) => { + err_out.error_type = DeliveryErrorType::UnexpectedResponse; + err_out.error_command = e.command.to_string().into(); + err_out.response_code = (e.response.code.to_native() as u64).into(); + err_out.response_enhanced = build_enhanced_code(&e.response.esc).into(); + err_out.response_message = e.response.message.to_string().into(); + } + ArchivedError::ConnectionError(e) => { + err_out.error_type = DeliveryErrorType::ConnectionError; + err_out.error_message = e.to_string().into(); + } + ArchivedError::TlsError(e) => { + err_out.error_type = DeliveryErrorType::TlsError; + err_out.error_message = e.to_string().into(); + } + ArchivedError::DaneError(e) => { + err_out.error_type = DeliveryErrorType::DaneError; + err_out.error_message = e.to_string().into(); + } + ArchivedError::MtaStsError(e) => { + err_out.error_type = DeliveryErrorType::MtaStsError; + err_out.error_message = e.to_string().into(); + } + ArchivedError::RateLimited => { + err_out.error_type = DeliveryErrorType::RateLimited; + } + ArchivedError::ConcurrencyLimited => { + err_out.error_type = DeliveryErrorType::ConcurrencyLimited; + } + ArchivedError::Io(e) => { + err_out.error_type = DeliveryErrorType::Io; + err_out.error_message = e.to_string().into(); + } + } + + err_out +} + +fn build_enhanced_code(esc: &[u8; 3]) -> String { + format!("{}.{}.{}", esc[0], esc[1], esc[2]) +} diff --git a/crates/jmap/src/registry/mod.rs b/crates/jmap/src/registry/mod.rs index c036acc9..97130314 100644 --- a/crates/jmap/src/registry/mod.rs +++ b/crates/jmap/src/registry/mod.rs @@ -5,5 +5,6 @@ */ pub mod get; +pub mod mapping; pub mod query; pub mod set; diff --git a/crates/jmap/src/registry/set.rs b/crates/jmap/src/registry/set.rs index 80a6b5f5..6224e0fe 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -119,27 +119,19 @@ impl RegistrySet for Server { ObjectType::TracingStore => {} ObjectType::WebDav => {} ObjectType::WebHook => {} - - // Tenant filtered ObjectType::Account => {} ObjectType::DsnReportSettings => {} ObjectType::MailingList => {} ObjectType::OAuthClient => {} ObjectType::Role => {} ObjectType::Tenant => {} - - // Account filtered ObjectType::MaskedEmail => {} ObjectType::PublicKey => {} - - // Special ObjectType::DkimSignature => {} ObjectType::Domain => {} ObjectType::Log => {} ObjectType::QueuedMessage => {} ObjectType::Task => {} - - // Move to registry? ObjectType::ArfExternalReport => {} ObjectType::DmarcExternalReport => {} ObjectType::TlsExternalReport => {} @@ -147,6 +139,8 @@ impl RegistrySet for Server { ObjectType::Metric => {} ObjectType::Trace => {} ObjectType::SpamTrainingSample => {} + ObjectType::DmarcInternalReport => todo!(), + ObjectType::TlsInternalReport => todo!(), } let todo = "read only properties"; diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index b44c08f1..0ed1ed50 100644 --- a/crates/main/Cargo.toml +++ b/crates/main/Cargo.toml @@ -41,7 +41,7 @@ jemallocator = "0.5.0" [features] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "azure", "nats", "enterprise"] -default = ["rocks", "enterprise", "sqlite"] +default = ["rocks", "enterprise"] sqlite = ["store/sqlite", "directory/sqlite"] foundationdb = ["store/foundation", "common/foundation"] postgres = ["store/postgres", "directory/postgres"] diff --git a/crates/registry/src/jmap/patch.rs b/crates/registry/src/jmap/patch.rs index fb78d577..2583a6b2 100644 --- a/crates/registry/src/jmap/patch.rs +++ b/crates/registry/src/jmap/patch.rs @@ -208,6 +208,24 @@ impl RegistryJsonPatch for f64 { } } +impl RegistryJsonPatch for trc::Key { + fn patch( + &mut self, + pointer: JsonPointerPatch<'_>, + value: super::JmapValue<'_>, + ) -> Result<(), PatchError> { + if let Some(new_value) = value.as_str().and_then(|v| trc::Key::try_parse(v.as_ref())) { + *self = new_value; + pointer.assert_eof() + } else { + Err(PatchError::new( + pointer, + format!("Invalid value {:?} for enum type {:?}.", value, self), + )) + } + } +} + impl RegistryJsonEnumPatch for T { fn patch( &mut self, diff --git a/crates/registry/src/jmap/ser.rs b/crates/registry/src/jmap/ser.rs index a5e81a9d..8fc69e81 100644 --- a/crates/registry/src/jmap/ser.rs +++ b/crates/registry/src/jmap/ser.rs @@ -100,3 +100,9 @@ impl IntoValue for Vec { JmapValue::Array(array) } } + +impl IntoValue for trc::Key { + fn into_value(self) -> JmapValue<'static> { + JmapValue::Str(self.name().into()) + } +} diff --git a/crates/registry/src/pickle.rs b/crates/registry/src/pickle.rs index d079db55..43a1751b 100644 --- a/crates/registry/src/pickle.rs +++ b/crates/registry/src/pickle.rs @@ -250,3 +250,13 @@ where Some(map) } } + +impl Pickle for trc::Key { + fn pickle(&self, out: &mut Vec) { + self.code().pickle(out); + } + + fn unpickle(stream: &mut PickledStream<'_>) -> Option { + u64::unpickle(stream).and_then(Self::from_code) + } +} diff --git a/crates/registry/src/types/index.rs b/crates/registry/src/types/index.rs index 47a66f48..cc2c8813 100644 --- a/crates/registry/src/types/index.rs +++ b/crates/registry/src/types/index.rs @@ -16,17 +16,14 @@ use types::id::Id; pub enum IndexKey<'x> { Unique { property: Property, - value: IndexValue<'x>, + value_1: IndexValue<'x>, + value_2: IndexValue<'x>, + global: bool, }, Search { property: Property, value: IndexValue<'x>, }, - Global { - property: Property, - value_1: IndexValue<'x>, - value_2: IndexValue<'x>, - }, ForeignKey { object_id: ObjectId, type_filter: IndexValue<'x>, @@ -52,17 +49,10 @@ pub enum IndexValue<'x> { #[derive(Debug, Default)] pub struct IndexBuilder<'x> { - pub object: Option, pub keys: AHashSet>, } impl<'x> IndexBuilder<'x> { - pub fn object(&mut self, object: ObjectType) { - if self.object.is_none() { - self.object = Some(object); - } - } - pub fn typ(&mut self, typ: u16) { self.keys.insert(IndexKey::Search { property: Property::Type, @@ -73,7 +63,9 @@ impl<'x> IndexBuilder<'x> { pub fn unique(&mut self, property: Property, value: impl Into>) { self.keys.insert(IndexKey::Unique { property, - value: value.into(), + value_1: value.into(), + value_2: IndexValue::None, + global: false, }); } @@ -106,24 +98,26 @@ impl<'x> IndexBuilder<'x> { } } - pub fn global(&mut self, property: Property, value: impl Into>) { - self.keys.insert(IndexKey::Global { + pub fn unique_global(&mut self, property: Property, value: impl Into>) { + self.keys.insert(IndexKey::Unique { property, value_1: value.into(), value_2: IndexValue::None, + global: true, }); } - pub fn composite( + pub fn unique_global_composite( &mut self, property: Property, value: impl Into>, composite: impl Into>, ) { - self.keys.insert(IndexKey::Global { + self.keys.insert(IndexKey::Unique { property, value_1: value.into(), value_2: composite.into(), + global: true, }); } diff --git a/crates/registry/src/utils/task.rs b/crates/registry/src/utils/task.rs index 56bb83d2..5d675539 100644 --- a/crates/registry/src/utils/task.rs +++ b/crates/registry/src/utils/task.rs @@ -16,6 +16,8 @@ impl Task { Task::CalendarAlarmNotification(task) => task.status = status, Task::CalendarItipMessage(task) => task.status = status, Task::MergeThreads(task) => task.status = status, + Task::DmarcReport(task) => task.status = status, + Task::TlsReport(task) => task.status = status, } } @@ -28,6 +30,8 @@ impl Task { Task::CalendarAlarmNotification(task) => &task.status, Task::CalendarItipMessage(task) => &task.status, Task::MergeThreads(task) => &task.status, + Task::DmarcReport(task) => &task.status, + Task::TlsReport(task) => &task.status, } } diff --git a/crates/services/src/housekeeper/mod.rs b/crates/services/src/housekeeper/mod.rs index 16b69c72..f05ee5e8 100644 --- a/crates/services/src/housekeeper/mod.rs +++ b/crates/services/src/housekeeper/mod.rs @@ -628,11 +628,9 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver { - // SPDX-SnippetBegin - // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC - // SPDX-License-Identifier: LicenseRef-SEL + let mut spans = Vec::new(); + self.store() + .iterate( + IterateParams::new( + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span(0))), + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span(u64::MAX))), + ) + .no_values(), + |key, _| { + spans.push(key.deserialize_be_u64(0)?); + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; let mut batch = BatchBuilder::new(); for span_id in spans { @@ -542,24 +554,17 @@ async fn build_tracing_span_document( server: &Server, span_id: u64, ) -> trc::Result> { - use common::telemetry::tracers::store::{TracingStore, build_span_document}; + use common::telemetry::tracers::store::build_span_document; + use registry::schema::structs::Trace; - let Some(index_fields) = server.core.email.index_fields.get(&SearchIndex::Tracing) else { - return Ok(None); - }; - let Some(store) = server - .core - .enterprise - .as_ref() - .and_then(|e| e.trace_store.as_ref()) - else { - return Ok(None); - }; - - let span = server.tracing_store().get_span(span_id).await?; - - if !span.is_empty() { - Ok(Some(build_span_document(span_id, span, index_fields))) + if let Some(index_fields) = server.core.email.index_fields.get(&SearchIndex::Tracing) { + server + .tracing_store() + .get_value::(ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span( + span_id, + )))) + .await + .map(|trace| trace.map(|trace| build_span_document(span_id, trace, index_fields))) } else { Ok(None) } diff --git a/crates/services/src/task_manager/mod.rs b/crates/services/src/task_manager/mod.rs index 98fad65b..dcb5f763 100644 --- a/crates/services/src/task_manager/mod.rs +++ b/crates/services/src/task_manager/mod.rs @@ -8,6 +8,7 @@ use crate::task_manager::imip::SendImipTask; use crate::task_manager::index::SearchIndexTask; use crate::task_manager::lock::TaskLockManager; use crate::task_manager::merge_threads::MergeThreadsTask; +use crate::task_manager::report::SubmitReportTask; use alarm::SendAlarmTask; use common::config::server::ServerProtocol; use common::network::limiter::ConcurrencyLimiter; @@ -43,6 +44,7 @@ pub mod imip; pub mod index; pub mod lock; pub mod merge_threads; +pub mod report; const QUEUE_REFRESH_INTERVAL: u64 = 60 * 5; // 5 minutes const DEFAULT_LOCK_EXPIRY: u64 = 60 * 5; // 5 minutes @@ -207,6 +209,16 @@ pub fn spawn_task_manager(inner: Arc) { server.send_imip(task, server_instance.clone()).await } Task::MergeThreads(task) => server.merge_threads(task).await, + Task::DmarcReport(task) => { + server + .submit_report(report::ReportId::Dmarc(task.report_id.id())) + .await + } + Task::TlsReport(task) => { + server + .submit_report(report::ReportId::Tls(task.report_id.id())) + .await + } Task::IndexDocument(_) | Task::UnindexDocument(_) | Task::IndexTrace(_) => unreachable!(), @@ -385,6 +397,7 @@ impl TaskQueueManager for Server { TaskType::MergeThreads => roles .merge_threads .is_enabled_for_integer(task_job.id as u32), + TaskType::DmarcReport | TaskType::TlsReport => true, }; if enabled { @@ -569,6 +582,8 @@ impl TaskInfo for Task { Task::CalendarAlarmNotification(_) => "CalendarAlarmNotification", Task::CalendarItipMessage(_) => "CalendarItipMessage", Task::MergeThreads(_) => "MergeThreads", + Task::DmarcReport(_) => "DmarcReport", + Task::TlsReport(_) => "TlsReport", } } } diff --git a/crates/services/src/task_manager/report.rs b/crates/services/src/task_manager/report.rs new file mode 100644 index 00000000..99638039 --- /dev/null +++ b/crates/services/src/task_manager/report.rs @@ -0,0 +1,44 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::task_manager::TaskResult; +use common::Server; +use smtp::reporting::{dmarc::DmarcReporting, tls::TlsReporting}; + +pub enum ReportId { + Dmarc(u64), + Tls(u64), +} + +pub(crate) trait SubmitReportTask: Sync + Send { + fn submit_report(&self, report_id: ReportId) -> impl Future + Send; +} + +impl SubmitReportTask for Server { + async fn submit_report(&self, report_id: ReportId) -> TaskResult { + match submit_report(self, report_id).await { + Ok(result) => result, + Err(err) => { + let result = TaskResult::temporary(err.to_string()); + trc::error!(err.details("Failed to submit report")); + result + } + } + } +} + +async fn submit_report(server: &Server, report_id: ReportId) -> trc::Result { + match report_id { + ReportId::Dmarc(item_id) => server + .send_dmarc_aggregate_report(item_id) + .await + .map(|_| TaskResult::Success), + ReportId::Tls(item_id) => server + .send_tls_aggregate_report(item_id) + .await + .map(|_| TaskResult::Success), + } +} diff --git a/crates/smtp/src/outbound/delivery.rs b/crates/smtp/src/outbound/delivery.rs index e6c8e833..aa79c4ab 100644 --- a/crates/smtp/src/outbound/delivery.rs +++ b/crates/smtp/src/outbound/delivery.rs @@ -411,6 +411,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -426,6 +427,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -584,6 +586,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -715,6 +718,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -756,6 +760,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -812,6 +817,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -1054,6 +1060,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -1071,6 +1078,7 @@ impl QueuedMessage { failure: None, tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -1123,6 +1131,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } @@ -1171,6 +1180,7 @@ impl QueuedMessage { .into(), tls_record: tls_report.record.clone(), interval: tls_report.interval, + span_id: message.span_id, }) .await; } diff --git a/crates/smtp/src/reporting/dmarc.rs b/crates/smtp/src/reporting/dmarc.rs index 397410ea..67850f56 100644 --- a/crates/smtp/src/reporting/dmarc.rs +++ b/crates/smtp/src/reporting/dmarc.rs @@ -4,9 +4,8 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::{AggregateTimestamp, SerializedSize}; +use super::AggregateTimestamp; use crate::{core::Session, queue::RecipientDomain, reporting::SmtpReporting}; -use ahash::AHashMap; use common::{ Server, config::smtp::report::AggregateFrequency, @@ -18,34 +17,30 @@ use mail_auth::{ ArcOutput, AuthenticatedMessage, AuthenticationResults, DkimOutput, DkimResult, DmarcOutput, SpfResult, common::verify::VerifySignature, - dmarc::{self, URI}, - report::{AuthFailureType, IdentityAlignment, PolicyPublished, Record, Report, SPFDomainScope}, + dmarc::{self}, + report::{AuthFailureType, IdentityAlignment, PolicyPublished, Record, SPFDomainScope}, }; -use registry::schema::structs::Rate; -use std::{collections::hash_map::Entry, future::Future}; +use registry::{ + pickle::Pickle, + schema::{ + enums::FailureReportingOption, + prelude::{ObjectType, Property}, + structs::{ + DmarcInternalReport, DmarcReport, DmarcReportRecord, Rate, Task, TaskDmarcReport, + TaskStatus, + }, + }, + types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, +}; +use std::future::Future; use store::{ - Deserialize, IterateParams, Serialize, ValueKey, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, QueueClass, ValueClass}, + SerializeInfallible, U64_LEN, ValueKey, + registry::ObjectIdVersioned, + write::{BatchBuilder, RegistryClass, ValueClass, assert::AssertValue, key::KeySerializer}, }; use trc::{AddContext, OutgoingReportEvent}; use utils::DomainPart; -#[derive( - Debug, - PartialEq, - Eq, - rkyv::Serialize, - rkyv::Deserialize, - rkyv::Archive, - serde::Serialize, - serde::Deserialize, -)] -pub struct DmarcFormat { - pub rua: Vec, - pub policy: PolicyPublished, - pub records: Vec, -} - impl Session { #[allow(clippy::too_many_arguments)] pub async fn send_dmarc_report( @@ -304,108 +299,106 @@ impl Session { report_record, dmarc_record, interval, + span_id: self.data.session_id, }) .await; } } pub trait DmarcReporting: Sync + Send { - fn send_dmarc_aggregate_report(&self, event: ReportEvent) -> impl Future + Send; - fn generate_dmarc_aggregate_report( + fn send_dmarc_aggregate_report( &self, - event: &ReportEvent, - rua: &mut Vec, - serialized_size: Option<&mut serde_json::Serializer>, - span_id: u64, - ) -> impl Future>> + Send; + report_id: u64, + ) -> impl Future> + Send; fn schedule_dmarc(&self, event: Box) -> impl Future + Send; } impl DmarcReporting for Server { - async fn send_dmarc_aggregate_report(&self, event: ReportEvent) { + async fn send_dmarc_aggregate_report(&self, item_id: u64) -> trc::Result<()> { + let object_id = ObjectType::DmarcInternalReport.to_id(); + let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id }); + + let Some(report) = self + .store() + .get_value::(ValueKey::from(key.clone())) + .await + .caused_by(trc::location!())? + else { + return Ok(()); + }; + + // Delete report + let mut batch = BatchBuilder::new(); + batch.clear(key).clear(RegistryClass::PrimaryKey { + object_id: object_id.into(), + index_id: Property::Domain.to_id(), + key: KeySerializer::new(report.domain.len() + U64_LEN) + .write(&report.domain) + .write(report.policy_identifier) + .finalize(), + }); + self.core + .storage + .data + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + let span_id = self.inner.data.span_id_gen.generate(); + let event_from = report.report.date_range_begin.timestamp() as u64; + let event_to = report.report.date_range_end.timestamp() as u64; trc::event!( OutgoingReport(OutgoingReportEvent::DmarcAggregateReport), SpanId = span_id, - ReportId = event.seq_id, - Domain = event.domain.clone(), - RangeFrom = trc::Value::Timestamp(event.seq_id), - RangeTo = trc::Value::Timestamp(event.due), + ReportId = event_from, + Domain = report.domain.clone(), + RangeFrom = trc::Value::Timestamp(event_from), + RangeTo = trc::Value::Timestamp(event_to), ); - // Generate report - let mut serialized_size = serde_json::Serializer::new(SerializedSize::new( - self.eval_if( - &self.core.smtp.report.dmarc_aggregate.max_size, - &RecipientDomain::new(event.domain.as_str()), - span_id, - ) - .await - .unwrap_or(25 * 1024 * 1024), - )); - let mut rua = Vec::new(); - let report = match self - .generate_dmarc_aggregate_report(&event, &mut rua, Some(&mut serialized_size), span_id) - .await - { - Ok(Some(report)) => report, - Ok(None) => { - trc::event!( - OutgoingReport(OutgoingReportEvent::NotFound), - SpanId = span_id, - CausedBy = trc::location!() - ); - - return; - } - Err(err) => { - trc::error!(err.span_id(span_id).details("Failed to read DMARC report")); - return; - } - }; - // Verify external reporting addresses let rua = match self .core .smtp .resolvers .dns - .verify_dmarc_report_address(&event.domain, &rua, Some(&self.inner.cache.dns_txt)) + .verify_dmarc_report_address( + &report.domain, + &report.rua, + Some(&self.inner.cache.dns_txt), + ) .await { Some(rcpts) => { if !rcpts.is_empty() { rcpts - .into_iter() - .map(|u| u.uri().to_string()) - .collect::>() } else { trc::event!( OutgoingReport(OutgoingReportEvent::UnauthorizedReportingAddress), SpanId = span_id, - Url = rua - .iter() - .map(|u| trc::Value::String(u.uri().to_compact_string())) + Url = report + .rua + .into_iter() + .map(|u| trc::Value::String(u.into())) .collect::>(), ); - self.delete_dmarc_report(event).await; - return; + return Ok(()); } } None => { trc::event!( OutgoingReport(OutgoingReportEvent::ReportingAddressValidationError), SpanId = span_id, - Url = rua - .iter() - .map(|u| trc::Value::String(u.uri().to_compact_string())) + Url = report + .rua + .into_iter() + .map(|u| trc::Value::String(u.into())) .collect::>(), ); - self.delete_dmarc_report(event).await; - return; + return Ok(()); } }; @@ -414,17 +407,17 @@ impl DmarcReporting for Server { let from_addr = self .eval_if( &config.address, - &RecipientDomain::new(event.domain.as_str()), + &RecipientDomain::new(report.domain.as_str()), span_id, ) .await .unwrap_or_else(|| "MAILER-DAEMON@localhost".to_compact_string()); let mut message = Vec::with_capacity(2048); - let _ = report.write_rfc5322( + let _ = mail_auth::report::Report::from(report.report).write_rfc5322( &self .eval_if( &self.core.smtp.report.submitter, - &RecipientDomain::new(event.domain.as_str()), + &RecipientDomain::new(report.domain.as_str()), span_id, ) .await @@ -432,7 +425,7 @@ impl DmarcReporting for Server { ( self.eval_if( &config.name, - &RecipientDomain::new(event.domain.as_str()), + &RecipientDomain::new(report.domain.as_str()), span_id, ) .await @@ -451,200 +444,227 @@ impl DmarcReporting for Server { message, &config.sign, false, - event.seq_id, + span_id, ) .await; - self.delete_dmarc_report(event).await; - } - - async fn generate_dmarc_aggregate_report( - &self, - event: &ReportEvent, - rua: &mut Vec, - mut serialized_size: Option<&mut serde_json::Serializer>, - span_id: u64, - ) -> trc::Result> { - // Deserialize report - let dmarc = match self - .store() - .get_value::>(ValueKey::from(ValueClass::Queue( - QueueClass::DmarcReportHeader(event.clone()), - ))) - .await? - { - Some(dmarc) => dmarc.deserialize::()?, - None => { - return Ok(None); - } - }; - let _ = std::mem::replace(rua, dmarc.rua); - - // Create report - let config = &self.core.smtp.report.dmarc_aggregate; - let mut report = Report::new() - .with_policy_published(dmarc.policy) - .with_date_range_begin(event.seq_id) - .with_date_range_end(event.due) - .with_report_id(format!("{}_{}", event.policy_hash, event.seq_id)) - .with_email( - self.eval_if( - &config.address, - &RecipientDomain::new(event.domain.as_str()), - span_id, - ) - .await - .unwrap_or_else(|| "MAILER-DAEMON@localhost".to_compact_string()), - ); - if let Some(org_name) = self - .eval_if::( - &config.org_name, - &RecipientDomain::new(event.domain.as_str()), - span_id, - ) - .await - { - report = report.with_org_name(org_name); - } - if let Some(contact_info) = self - .eval_if::( - &config.contact_info, - &RecipientDomain::new(event.domain.as_str()), - span_id, - ) - .await - { - report = report.with_extra_contact_info(contact_info); - } - - if let Some(serialized_size) = serialized_size.as_deref_mut() { - let _ = serde::Serialize::serialize(&report, serialized_size); - } - - // Group duplicates - let from_key = ValueKey::from(ValueClass::Queue(QueueClass::DmarcReportEvent( - ReportEvent { - due: event.due, - policy_hash: event.policy_hash, - seq_id: 0, - domain: event.domain.clone(), - }, - ))); - let to_key = ValueKey::from(ValueClass::Queue(QueueClass::DmarcReportEvent( - ReportEvent { - due: event.due, - policy_hash: event.policy_hash, - seq_id: u64::MAX, - domain: event.domain.clone(), - }, - ))); - let mut record_map = AHashMap::with_capacity(dmarc.records.len()); - self.core - .storage - .data - .iterate(IterateParams::new(from_key, to_key).ascending(), |_, v| { - let archive = as Deserialize>::deserialize(v)?; - - match record_map.entry(archive.deserialize::()?) { - Entry::Occupied(mut e) => { - *e.get_mut() += 1; - Ok(true) - } - Entry::Vacant(e) => { - if serialized_size - .as_deref_mut() - .is_none_or(|serialized_size| { - serde::Serialize::serialize(e.key(), serialized_size).is_ok() - }) - { - e.insert(1u32); - Ok(true) - } else { - Ok(false) - } - } - } - }) - .await - .caused_by(trc::location!())?; - - for (record, count) in record_map { - report = report.with_record(record.with_count(count)); - } - - Ok(Some(report)) + Ok(()) } async fn schedule_dmarc(&self, event: Box) { - let created = event.interval.to_timestamp(); - let deliver_at = created + event.interval.as_secs(); - let mut report_event = ReportEvent { - due: deliver_at, - policy_hash: event.dmarc_record.to_hash(), - seq_id: created, - domain: event.domain, - }; + let object_id = ObjectType::DmarcInternalReport.to_id(); + let policy_hash = event.dmarc_record.to_hash(); + let pk = ValueClass::Registry(RegistryClass::PrimaryKey { + object_id: object_id.into(), + index_id: Property::Domain.to_id(), + key: KeySerializer::new(event.domain.len() + U64_LEN) + .write(&event.domain) + .write(policy_hash) + .finalize(), + }); + let mut rety_count = 0; - // Write policy if missing - let mut builder = BatchBuilder::new(); - if self - .core - .storage - .data - .get_value::<()>(ValueKey::from(ValueClass::Queue( - QueueClass::DmarcReportHeader(report_event.clone()), - ))) - .await - .unwrap_or_default() - .is_none() - { - // Serialize report - let entry = DmarcFormat { - rua: event.dmarc_record.rua().to_vec(), - policy: PolicyPublished::from_record( - report_event.domain.to_string(), - &event.dmarc_record, - ), - records: vec![], - }; + loop { + // Find the report by domain name + let mut batch = BatchBuilder::new(); + let report = match self + .store() + .get_value::(ValueKey::from(pk.clone())) + .await + { + Ok(Some(object_id_v)) => { + match self + .store() + .get_value::(ValueKey::from(ValueClass::Registry( + RegistryClass::Item { + object_id, + item_id: object_id_v.object_id.id().id(), + }, + ))) + .await + { + Ok(Some(report)) => Some((object_id_v, report)), + Ok(None) => { + trc::event!( + OutgoingReport(OutgoingReportEvent::NotFound), + Id = object_id_v.object_id.id().id(), + CausedBy = trc::location!(), + Details = "Failed to find DMARC report for domain" + ); - // Write report - builder.set( - ValueClass::Queue(QueueClass::DmarcReportHeader(report_event.clone())), - match Archiver::new(entry).serialize() { - Ok(data) => data.to_vec(), - Err(err) => { - trc::error!( - err.caused_by(trc::location!()) - .details("Failed to serialize DMARC report") - ); - return; + return; + } + Err(err) => { + trc::error!( + err.caused_by(trc::location!()) + .details("Failed to query registry for DMARC report") + ); + return; + } } - }, - ); - } - - // Write entry - report_event.seq_id = self.inner.data.queue_id_gen.generate(); - builder.set( - ValueClass::Queue(QueueClass::DmarcReportEvent(report_event)), - match Archiver::new(event.report_record).serialize() { - Ok(data) => data.to_vec(), + } + Ok(None) => None, Err(err) => { trc::error!( err.caused_by(trc::location!()) - .details("Failed to serialize DMARC report") + .details("Failed to query registry for DMARC report") ); return; } - }, - ); + }; - if let Err(err) = self.core.storage.data.write(builder.build_all()).await { - trc::error!( - err.caused_by(trc::location!()) - .details("Failed to write DMARC report") + // Create report if missing + let config = &self.core.smtp.report.dmarc_aggregate; + let (item_id, mut report) = if let Some((mut object_id_v, report)) = report { + batch.assert_value(pk.clone(), AssertValue::U32(object_id_v.version)); + object_id_v.version += 1; + batch.set(pk.clone(), object_id_v.serialize()); + + (object_id_v.object_id.id().id(), report) + } else { + let item_id = self.inner.data.queue_id_gen.generate(); + let deliver_at = (event.interval.to_timestamp() + event.interval.as_secs()) as i64; + + batch + .assert_value(pk.clone(), ()) + .set( + pk.clone(), + ObjectIdVersioned { + object_id: ObjectId::new( + ObjectType::DmarcInternalReport, + item_id.into(), + ), + version: 0, + } + .serialize(), + ) + .schedule_task_with_id( + item_id, + Task::DmarcReport(TaskDmarcReport { + report_id: item_id.into(), + status: TaskStatus::at(deliver_at), + }), + ); + + let created_at = UTCDateTime::now(); + let deliver_at = UTCDateTime::from_timestamp(deliver_at); + let policy = + PolicyPublished::from_record(event.domain.clone(), &event.dmarc_record); + ( + item_id, + DmarcInternalReport { + created_at, + deliver_at, + domain: event.domain.clone(), + report: DmarcReport { + report_id: format!("{}_{policy_hash}", created_at.timestamp()), + date_range_begin: created_at, + date_range_end: deliver_at, + email: self + .eval_if( + &config.address, + &RecipientDomain::new(event.domain.as_str()), + event.span_id, + ) + .await + .unwrap_or_else(|| "MAILER-DAEMON@localhost".to_string()), + extra_contact_info: self + .eval_if::( + &config.contact_info, + &RecipientDomain::new(event.domain.as_str()), + event.span_id, + ) + .await, + org_name: self + .eval_if::( + &config.org_name, + &RecipientDomain::new(event.domain.as_str()), + event.span_id, + ) + .await + .unwrap_or_default(), + policy_adkim: policy.adkim.into(), + policy_aspf: policy.aspf.into(), + policy_disposition: policy.p.into(), + policy_domain: policy.domain, + policy_failure_reporting_options: match event.dmarc_record.fo { + dmarc::Report::All => vec![FailureReportingOption::All], + dmarc::Report::Any => vec![FailureReportingOption::Any], + dmarc::Report::Dkim => vec![FailureReportingOption::DkimFailure], + dmarc::Report::Spf => vec![FailureReportingOption::SpfFailure], + dmarc::Report::DkimSpf => vec![ + FailureReportingOption::DkimFailure, + FailureReportingOption::SpfFailure, + ], + }, + policy_subdomain_disposition: policy.sp.into(), + policy_testing_mode: policy.testing, + policy_version: None, + version: 1.0, + ..Default::default() + }, + policy_identifier: policy_hash, + rua: event + .dmarc_record + .rua() + .iter() + .map(|u| u.uri.clone()) + .collect(), + }, + ) + }; + + // Add record + let mut record = DmarcReportRecord::from(event.report_record.clone()); + if let Some(idx) = report.report.records.iter().position(|d| d == &record) { + report.report.records[idx].count += 1; + } else { + record.count = 1; + report.report.records.push(record); + } + + // Write entry + let report_bytes = report.to_pickled_vec(); + let max_report_size = self + .eval_if( + &config.max_size, + &RecipientDomain::new(&event.domain), + event.span_id, + ) + .await + .unwrap_or(5 * 1024 * 1024); + if max_report_size != 0 && report_bytes.len() > max_report_size { + trc::event!( + OutgoingReport(OutgoingReportEvent::MaxSizeExceeded), + SpanId = event.span_id, + Domain = event.domain.clone(), + Details = report_bytes.len(), + Limit = max_report_size, + ); + return; + } + + batch.set( + ValueClass::Registry(RegistryClass::Item { object_id, item_id }), + report_bytes, ); + + match self.core.storage.data.write(batch.build_all()).await { + Ok(_) => { + break; + } + Err(err) => { + if err.is_assertion_failure() && rety_count < 3 { + rety_count += 1; + continue; + } + trc::error!( + err.caused_by(trc::location!()) + .details("Failed to write DMARC report") + ); + break; + } + } } } } diff --git a/crates/smtp/src/reporting/tls.rs b/crates/smtp/src/reporting/tls.rs index b88b47d1..c03d0dc6 100644 --- a/crates/smtp/src/reporting/tls.rs +++ b/crates/smtp/src/reporting/tls.rs @@ -4,9 +4,8 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::{AggregateTimestamp, SerializedSize}; +use super::AggregateTimestamp; use crate::{queue::RecipientDomain, reporting::SmtpReporting}; -use ahash::{AHashMap, AHashSet}; use common::{ Server, USER_AGENT, config::smtp::{ @@ -18,27 +17,27 @@ use common::{ use mail_auth::{ flate2::{Compression, write::GzEncoder}, mta_sts::{ReportUri, TlsRpt}, - report::tlsrpt::{ - DateRange, FailureDetails, Policy, PolicyDetails, PolicyType, Summary, TlsReport, - }, + report::tlsrpt::{FailureDetails, PolicyDetails}, }; -use mail_parser::DateTime; use registry::{ pickle::Pickle, schema::{ enums::TlsPolicyType, prelude::{ObjectType, Property}, - structs::{Task, TlsFailureDetails, TlsInternalReport, TlsReport, TlsReportPolicy}, + structs::{ + Task, TaskStatus, TaskTlsReport, TlsFailureDetails, TlsInternalReport, TlsReport, + TlsReportPolicy, + }, }, - types::{EnumImpl, datetime::UTCDateTime}, + types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, }; use reqwest::header::CONTENT_TYPE; use std::fmt::Write; -use std::{collections::hash_map::Entry, future::Future, sync::Arc, time::Duration}; +use std::{future::Future, sync::Arc, time::Duration}; use store::{ - Deserialize, IterateParams, ValueKey, - registry::RegistryQuery, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, QueueClass, RegistryClass, ValueClass}, + SerializeInfallible, ValueKey, + registry::ObjectIdVersioned, + write::{BatchBuilder, RegistryClass, ValueClass, assert::AssertValue}, }; use trc::{AddContext, OutgoingReportEvent}; @@ -61,25 +60,43 @@ pub static TLS_HTTP_REPORT: parking_lot::Mutex> = parking_lot::Mutex::ne pub trait TlsReporting: Sync + Send { fn send_tls_aggregate_report( &self, - events: Vec, - ) -> impl Future + Send; - fn generate_tls_aggregate_report( - &self, - events: &[ReportEvent], - rua: &mut Vec, - serialized_size: Option<&mut serde_json::Serializer>, - span_id: u64, - ) -> impl Future>> + Send; + report_id: u64, + ) -> impl Future> + Send; + fn schedule_tls(&self, event: Box) -> impl Future + Send; } impl TlsReporting for Server { - async fn send_tls_aggregate_report(&self, events: Vec) { - let (domain_name, event_from, event_to) = events - .first() - .map(|e| (e.domain.as_str(), e.seq_id, e.due)) - .unwrap(); + async fn send_tls_aggregate_report(&self, item_id: u64) -> trc::Result<()> { + let object_id = ObjectType::TlsInternalReport.to_id(); + let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id }); + let Some(report) = self + .store() + .get_value::(ValueKey::from(key.clone())) + .await + .caused_by(trc::location!())? + else { + return Ok(()); + }; + + // Delete report + let mut batch = BatchBuilder::new(); + batch.clear(key).clear(RegistryClass::PrimaryKey { + object_id: object_id.into(), + index_id: Property::Domain.to_id(), + key: report.domain.as_bytes().to_vec(), + }); + self.core + .storage + .data + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + + let domain_name = report.domain.as_str(); + let event_from = report.report.date_range_start.timestamp() as u64; + let event_to = report.report.date_range_end.timestamp() as u64; let span_id = self.inner.data.span_id_gen.generate(); trc::event!( @@ -92,43 +109,8 @@ impl TlsReporting for Server { ); // Generate report - let mut rua = Vec::new(); - let mut serialized_size = serde_json::Serializer::new(SerializedSize::new( - self.eval_if( - &self.core.smtp.report.tls.max_size, - &RecipientDomain::new(domain_name), - span_id, - ) - .await - .unwrap_or(25 * 1024 * 1024), - )); - let report = match self - .generate_tls_aggregate_report(&events, &mut rua, Some(&mut serialized_size), span_id) - .await - { - Ok(Some(report)) => report, - Ok(None) => { - // This should not happen - trc::event!( - OutgoingReport(OutgoingReportEvent::NotFound), - SpanId = span_id, - CausedBy = trc::location!() - ); - self.delete_tls_report(events).await; - return; - } - Err(err) => { - trc::error!( - err.span_id(span_id) - .caused_by(trc::location!()) - .details("Failed to read TLS report") - ); - return; - } - }; - - // Compress and serialize report - let json = report.to_json(); + let exported_report = mail_auth::report::tlsrpt::TlsReport::from(report.report); + let json = exported_report.to_json(); let mut e = GzEncoder::new(Vec::with_capacity(json.len()), Compression::default()); let json = match std::io::Write::write_all(&mut e, json.as_bytes()).and_then(|_| e.finish()) { @@ -141,83 +123,73 @@ impl TlsReporting for Server { Details = "Failed to compress report" ); - self.delete_tls_report(events).await; - return; + return Ok(()); } }; // Try delivering report over HTTP - let mut rcpts = Vec::with_capacity(rua.len()); - for uri in &rua { - match uri { - ReportUri::Http(uri) => { - if let Ok(client) = reqwest::Client::builder() - .user_agent(USER_AGENT) - .timeout(Duration::from_secs(2 * 60)) - .build() - { - #[cfg(feature = "test_mode")] - if uri == "https://127.0.0.1/tls" { - TLS_HTTP_REPORT.lock().extend_from_slice(&json); - self.delete_tls_report(events).await; - return; - } + for uri in &report.http_rua { + if let Ok(client) = reqwest::Client::builder() + .user_agent(USER_AGENT) + .timeout(Duration::from_secs(2 * 60)) + .build() + { + #[cfg(feature = "test_mode")] + if uri == "https://127.0.0.1/tls" { + TLS_HTTP_REPORT.lock().extend_from_slice(&json); - match client - .post(uri) - .header(CONTENT_TYPE, "application/tlsrpt+gzip") - .body(json.to_vec()) - .send() - .await - { - Ok(response) => { - if response.status().is_success() { - trc::event!( - OutgoingReport(OutgoingReportEvent::HttpSubmission), - SpanId = span_id, - Url = uri.to_string(), - Code = response.status().as_u16(), - ); + return Ok(()); + } - self.delete_tls_report(events).await; - return; - } else { - trc::event!( - OutgoingReport(OutgoingReportEvent::SubmissionError), - SpanId = span_id, - Url = uri.to_string(), - Code = response.status().as_u16(), - Details = "Invalid HTTP response" - ); - } - } - Err(err) => { - trc::event!( - OutgoingReport(OutgoingReportEvent::SubmissionError), - SpanId = span_id, - Url = uri.to_string(), - Reason = err.to_string(), - Details = "HTTP submission error" - ); - } + match client + .post(uri) + .header(CONTENT_TYPE, "application/tlsrpt+gzip") + .body(json.to_vec()) + .send() + .await + { + Ok(response) => { + if response.status().is_success() { + trc::event!( + OutgoingReport(OutgoingReportEvent::HttpSubmission), + SpanId = span_id, + Url = uri.to_string(), + Code = response.status().as_u16(), + ); + + return Ok(()); + } else { + trc::event!( + OutgoingReport(OutgoingReportEvent::SubmissionError), + SpanId = span_id, + Url = uri.to_string(), + Code = response.status().as_u16(), + Details = "Invalid HTTP response" + ); } } - } - ReportUri::Mail(mailto) => { - rcpts.push(mailto.as_str()); + Err(err) => { + trc::event!( + OutgoingReport(OutgoingReportEvent::SubmissionError), + SpanId = span_id, + Url = uri.to_string(), + Reason = err.to_string(), + Details = "HTTP submission error" + ); + } } } } // Deliver report over SMTP - if !rcpts.is_empty() { + if !report.mail_rua.is_empty() { let config = &self.core.smtp.report.tls; let from_addr = self .eval_if(&config.address, &RecipientDomain::new(domain_name), span_id) .await .unwrap_or_else(|| "MAILER-DAEMON@localhost".to_string()); let mut message = Vec::with_capacity(2048); - let _ = report.write_rfc5322_from_bytes( + let _ = exported_report.write_rfc5322_from_bytes( domain_name, &self .eval_if( @@ -234,7 +206,7 @@ impl TlsReporting for Server { .as_str(), from_addr.as_str(), ), - rcpts.iter().copied(), + report.mail_rua.iter().map(|v| v.as_str()), &json, &mut message, ); @@ -242,7 +214,7 @@ impl TlsReporting for Server { // Send report self.send_report( &from_addr, - rcpts.iter(), + report.mail_rua.iter().map(|v| v.as_str()), message, &config.sign, false, @@ -255,350 +227,273 @@ impl TlsReporting for Server { SpanId = span_id, ); } - self.delete_tls_report(events).await; - } - async fn generate_tls_aggregate_report( - &self, - events: &[ReportEvent], - rua: &mut Vec, - mut serialized_size: Option<&mut serde_json::Serializer>, - span_id: u64, - ) -> trc::Result> { - let (domain_name, event_from, event_to, policy) = events - .first() - .map(|e| (e.domain.as_str(), e.seq_id, e.due, e.policy_hash)) - .unwrap(); - let config = &self.core.smtp.report.tls; - let mut report = TlsReport { - organization_name: self - .eval_if::( - &config.org_name, - &RecipientDomain::new(domain_name), - span_id, - ) - .await - .clone(), - date_range: DateRange { - start_datetime: DateTime::from_timestamp(event_from as i64), - end_datetime: DateTime::from_timestamp(event_to as i64), - }, - contact_info: self - .eval_if::( - &config.contact_info, - &RecipientDomain::new(domain_name), - span_id, - ) - .await - .clone(), - report_id: format!("{}_{}", event_from, policy), - policies: Vec::with_capacity(events.len()), - }; - - if let Some(serialized_size) = serialized_size.as_deref_mut() { - let _ = serde::Serialize::serialize(&report, serialized_size); - } - - for event in events { - let tls = if let Some(tls) = self - .store() - .get_value::>(ValueKey::from(ValueClass::Queue( - QueueClass::TlsReportHeader(event.clone()), - ))) - .await? - { - tls.deserialize::()? - } else { - continue; - }; - - if let Some(serialized_size) = serialized_size.as_deref_mut() - && serde::Serialize::serialize(&tls, serialized_size).is_err() - { - continue; - } - - // Group duplicates - let mut total_success = 0; - let mut total_failure = 0; - let from_key = - ValueKey::from(ValueClass::Queue(QueueClass::TlsReportEvent(ReportEvent { - due: event.due, - policy_hash: event.policy_hash, - seq_id: 0, - domain: event.domain.clone(), - }))); - let to_key = - ValueKey::from(ValueClass::Queue(QueueClass::TlsReportEvent(ReportEvent { - due: event.due, - policy_hash: event.policy_hash, - seq_id: u64::MAX, - domain: event.domain.clone(), - }))); - let mut record_map = AHashMap::new(); - self.core - .storage - .data - .iterate(IterateParams::new(from_key, to_key).ascending(), |_, v| { - let archive = as Deserialize>::deserialize(v)?; - if let Some(failure_details) = - archive.deserialize::>()? - { - match record_map.entry(failure_details) { - Entry::Occupied(mut e) => { - total_failure += 1; - *e.get_mut() += 1; - Ok(true) - } - Entry::Vacant(e) => { - if serialized_size - .as_deref_mut() - .is_none_or(|serialized_size| { - serde::Serialize::serialize(e.key(), serialized_size) - .is_ok() - }) - { - total_failure += 1; - e.insert(1u32); - Ok(true) - } else { - Ok(false) - } - } - } - } else { - total_success += 1; - Ok(true) - } - }) - .await - .caused_by(trc::location!())?; - - // Add policy - report.policies.push(Policy { - policy: tls.policy, - summary: Summary { - total_success, - total_failure, - }, - failure_details: record_map - .into_iter() - .map(|(mut r, count)| { - r.failed_session_count = count; - r - }) - .collect(), - }); - - // Add report URIs - for entry in tls.rua { - if !rua.contains(&entry) { - rua.push(entry); - } - } - } - - Ok(if !report.policies.is_empty() { - Some(report) - } else { - None - }) + Ok(()) } async fn schedule_tls(&self, event: Box) { - // Find the report by domain name - let mut batch = BatchBuilder::new(); let object_id = ObjectType::TlsInternalReport.to_id(); - let item_id; - let report = match self - .registry() - .query::>( - RegistryQuery::new(ObjectType::TlsInternalReport) - .equal(Property::Domain, event.domain.clone()), - ) - .await - .map(|ids| ids.into_iter().next()) - { - Ok(Some(item_id_)) => { - match self - .store() - .get_value::(ValueKey::from(ValueClass::Registry( - RegistryClass::Item { - object_id, - item_id: item_id_, - }, - ))) - .await - { - Ok(Some(report)) => { - item_id = item_id_; - Some(report) - } - Ok(None) => { - batch.clear(ValueClass::Registry(RegistryClass::Index { - index_id: Property::Domain.to_id(), - object_id, - item_id: item_id_, - key: event.domain.as_bytes().to_vec(), - })); - item_id = self.inner.data.queue_id_gen.generate(); - None - } - Err(err) => { - trc::error!( - err.caused_by(trc::location!()) - .details("Failed to query registry for TLS report") - ); - return; + let pk = ValueClass::Registry(RegistryClass::PrimaryKey { + object_id: object_id.into(), + index_id: Property::Domain.to_id(), + key: event.domain.as_bytes().to_vec(), + }); + let mut rety_count = 0; + let policy_hash = event.policy.to_hash(); + + loop { + // Find the report by domain name + let mut batch = BatchBuilder::new(); + let report = match self + .store() + .get_value::(ValueKey::from(pk.clone())) + .await + { + Ok(Some(object_id_v)) => { + match self + .store() + .get_value::(ValueKey::from(ValueClass::Registry( + RegistryClass::Item { + object_id, + item_id: object_id_v.object_id.id().id(), + }, + ))) + .await + { + Ok(Some(report)) => Some((object_id_v, report)), + Ok(None) => { + trc::event!( + OutgoingReport(OutgoingReportEvent::NotFound), + Id = object_id_v.object_id.id().id(), + CausedBy = trc::location!(), + Details = "Failed to find TLS report for domain" + ); + + return; + } + Err(err) => { + trc::error!( + err.caused_by(trc::location!()) + .details("Failed to query registry for TLS report") + ); + return; + } } } + Ok(None) => None, + Err(err) => { + trc::error!( + err.caused_by(trc::location!()) + .details("Failed to query registry for TLS report") + ); + return; + } + }; + + // Create report if missing + let config = &self.core.smtp.report.tls; + let (item_id, mut report) = if let Some((mut object_id_v, report)) = report { + batch.assert_value(pk.clone(), AssertValue::U32(object_id_v.version)); + object_id_v.version += 1; + batch.set(pk.clone(), object_id_v.serialize()); + + (object_id_v.object_id.id().id(), report) + } else { + let item_id = self.inner.data.queue_id_gen.generate(); + let deliver_at = (event.interval.to_timestamp() + event.interval.as_secs()) as i64; + + batch + .assert_value(pk.clone(), ()) + .set( + pk.clone(), + ObjectIdVersioned { + object_id: ObjectId::new(ObjectType::TlsInternalReport, item_id.into()), + version: 0, + } + .serialize(), + ) + .schedule_task_with_id( + item_id, + Task::TlsReport(TaskTlsReport { + report_id: item_id.into(), + status: TaskStatus::at(deliver_at), + }), + ); + + let created_at = UTCDateTime::now(); + let deliver_at = UTCDateTime::from_timestamp(deliver_at); + ( + item_id, + TlsInternalReport { + created_at, + deliver_at, + domain: event.domain.clone(), + report: TlsReport { + report_id: format!("{}_{policy_hash}", created_at.timestamp()), + organization_name: self + .eval_if::( + &config.org_name, + &RecipientDomain::new(&event.domain), + event.span_id, + ) + .await + .clone(), + contact_info: self + .eval_if::( + &config.contact_info, + &RecipientDomain::new(&event.domain), + event.span_id, + ) + .await + .clone(), + date_range_end: deliver_at, + date_range_start: created_at, + policies: vec![], + }, + ..Default::default() + }, + ) + }; + + let policy = if let Some(policy) = report + .policy_identifiers + .iter() + .position(|id| *id == policy_hash) + .and_then(|idx| report.report.policies.get_mut(idx)) + { + policy + } else { + // Create policy + let mut policy = TlsReportPolicy { + policy_type: TlsPolicyType::NoPolicyFound, + policy_domain: report.domain.clone(), + ..Default::default() + }; + + match &event.policy { + common::ipc::PolicyType::Tlsa(tlsa) => { + policy.policy_type = TlsPolicyType::Tlsa; + if let Some(tlsa) = tlsa { + for entry in &tlsa.entries { + policy.policy_strings.push(format!( + "{} {} {} {}", + if entry.is_end_entity { 3 } else { 2 }, + i32::from(entry.is_spki), + if entry.is_sha256 { 1 } else { 2 }, + entry.data.iter().fold( + String::with_capacity(64), + |mut s, b| { + write!(s, "{b:02X}").ok(); + s + } + ) + )); + } + } + } + common::ipc::PolicyType::Sts(sts) => { + policy.policy_type = TlsPolicyType::Sts; + if let Some(sts) = sts { + policy.policy_strings.push("version: STSv1".to_string()); + policy.policy_strings.push(format!( + "mode: {}", + match sts.mode { + Mode::Enforce => "enforce", + Mode::Testing => "testing", + Mode::None => "none", + } + )); + policy + .policy_strings + .push(format!("max_age: {}", sts.max_age)); + for mx in &sts.mx { + let mx = match mx { + MxPattern::Equals(mx) => mx.to_string(), + MxPattern::StartsWith(mx) => format!("*.{mx}"), + }; + policy.policy_strings.push(format!("mx: {mx}")); + policy.mx_hosts.push(mx); + } + } + } + _ => (), + } + + for rua in &event.tls_record.rua { + match rua { + ReportUri::Mail(mail) => { + if !report.mail_rua.contains(mail) { + report.mail_rua.push(mail.clone()); + } + } + ReportUri::Http(uri) => { + if !report.http_rua.contains(uri) { + report.http_rua.push(uri.clone()); + } + } + } + } + + report.policy_identifiers.push(policy_hash); + report.report.policies.push(policy); + report.report.policies.last_mut().unwrap() + }; + + // Add failure details + if let Some(failure) = event.failure.clone().map(TlsFailureDetails::from) { + if let Some(idx) = policy.failure_details.iter().position(|d| d == &failure) { + policy.failure_details[idx].failed_session_count += 1; + } else { + policy.failure_details.push(failure); + } + + policy.total_failed_sessions += 1; + } else { + policy.total_successful_sessions += 1; } - Ok(None) => { - item_id = self.inner.data.queue_id_gen.generate(); - None - } - Err(err) => { - trc::error!( - err.caused_by(trc::location!()) - .details("Failed to query registry for TLS report") + + // Write entry + let report_bytes = report.to_pickled_vec(); + let max_report_size = self + .eval_if( + &config.max_size, + &RecipientDomain::new(&event.domain), + event.span_id, + ) + .await + .unwrap_or(5 * 1024 * 1024); + if max_report_size != 0 && report_bytes.len() > max_report_size { + trc::event!( + OutgoingReport(OutgoingReportEvent::MaxSizeExceeded), + SpanId = event.span_id, + Domain = event.domain.clone(), + Details = report_bytes.len(), + Limit = max_report_size, ); return; } - }; - // Generate policy if missing - let mut report = if let Some(report) = report { - report - } else { batch.set( - ValueClass::Registry(RegistryClass::Index { - index_id: Property::Domain.to_id(), - object_id, - item_id, - key: event.domain.as_bytes().to_vec(), - }), - vec![], + ValueClass::Registry(RegistryClass::Item { object_id, item_id }), + report_bytes, ); - let todo = "schedule task"; - TlsInternalReport { - created_at: UTCDateTime::now(), - deliver_at: UTCDateTime::from_timestamp( - (event.interval.to_timestamp() + event.interval.as_secs()) as i64, - ), - domain: event.domain, - ..Default::default() - } - }; - let policy_hash = event.policy.to_hash(); - let policy = if let Some(policy) = report - .policy_identifiers - .iter() - .position(|id| *id == policy_hash) - .and_then(|idx| report.report.policies.get_mut(idx)) - { - policy - } else { - // Serialize report - let mut policy = TlsReportPolicy { - policy_type: TlsPolicyType::NoPolicyFound, - policy_domain: report.domain.clone(), - ..Default::default() - }; - - match event.policy { - common::ipc::PolicyType::Tlsa(tlsa) => { - policy.policy_type = TlsPolicyType::Tlsa; - if let Some(tlsa) = tlsa { - for entry in &tlsa.entries { - policy.policy_strings.push(format!( - "{} {} {} {}", - if entry.is_end_entity { 3 } else { 2 }, - i32::from(entry.is_spki), - if entry.is_sha256 { 1 } else { 2 }, - entry - .data - .iter() - .fold(String::with_capacity(64), |mut s, b| { - write!(s, "{b:02X}").ok(); - s - }) - )); - } - } + match self.core.storage.data.write(batch.build_all()).await { + Ok(_) => { + break; } - common::ipc::PolicyType::Sts(sts) => { - policy.policy_type = TlsPolicyType::Sts; - if let Some(sts) = sts { - policy.policy_strings.push("version: STSv1".to_string()); - policy.policy_strings.push(format!( - "mode: {}", - match sts.mode { - Mode::Enforce => "enforce", - Mode::Testing => "testing", - Mode::None => "none", - } - )); - policy - .policy_strings - .push(format!("max_age: {}", sts.max_age)); - for mx in &sts.mx { - let mx = match mx { - MxPattern::Equals(mx) => mx.to_string(), - MxPattern::StartsWith(mx) => format!("*.{mx}"), - }; - policy.policy_strings.push(format!("mx: {mx}")); - policy.mx_hosts.push(mx); - } - } - } - _ => (), - } - - for rua in &event.tls_record.rua { - match rua { - ReportUri::Mail(mail) => { - if !report.mail_rua.contains(mail) { - report.mail_rua.push(mail.clone()); - } - } - ReportUri::Http(uri) => { - if !report.http_rua.contains(uri) { - report.http_rua.push(uri.clone()); - } + Err(err) => { + if err.is_assertion_failure() && rety_count < 3 { + rety_count += 1; + continue; } + trc::error!( + err.caused_by(trc::location!()) + .details("Failed to write TLS report") + ); + break; } } - - report.policy_identifiers.push(policy_hash); - report.report.policies.push(policy); - report.report.policies.last_mut().unwrap() - }; - - // Add failure details - if let Some(failure) = event.failure.map(TlsFailureDetails::from) { - if let Some(idx) = policy.failure_details.iter().position(|d| d == &failure) { - policy.failure_details[idx].failed_session_count += 1; - } else { - policy.failure_details.push(failure); - } - - policy.total_failed_sessions += 1; - } else { - policy.total_successful_sessions += 1; - } - - // Write entry - batch.set( - ValueClass::Registry(RegistryClass::Item { object_id, item_id }), - report.to_pickled_vec(), - ); - - if let Err(err) = self.core.storage.data.write(batch.build_all()).await { - trc::error!( - err.caused_by(trc::location!()) - .details("Failed to write TLS report") - ); } } } diff --git a/crates/store/src/backend/mysql/main.rs b/crates/store/src/backend/mysql/main.rs index f67ad280..8b14ce8a 100644 --- a/crates/store/src/backend/mysql/main.rs +++ b/crates/store/src/backend/mysql/main.rs @@ -93,6 +93,7 @@ impl MysqlStore { SUBSPACE_IN_MEMORY_VALUE, SUBSPACE_PROPERTY, SUBSPACE_REGISTRY, + SUBSPACE_REGISTRY_PK, SUBSPACE_DIRECTORY, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUEUE_EVENT, @@ -125,11 +126,7 @@ impl MysqlStore { .await .map_err(into_error)?; - for table in [ - SUBSPACE_INDEXES, - SUBSPACE_REGISTRY_IDX, - SUBSPACE_REGISTRY_IDX_GLOBAL, - ] { + for table in [SUBSPACE_INDEXES, SUBSPACE_REGISTRY_IDX] { let table = char::from(table); conn.query_drop(format!( "CREATE TABLE IF NOT EXISTS {table} ( diff --git a/crates/store/src/backend/postgres/main.rs b/crates/store/src/backend/postgres/main.rs index 8109a676..aa045b9c 100644 --- a/crates/store/src/backend/postgres/main.rs +++ b/crates/store/src/backend/postgres/main.rs @@ -99,6 +99,7 @@ impl PostgresStore { SUBSPACE_IN_MEMORY_VALUE, SUBSPACE_PROPERTY, SUBSPACE_REGISTRY, + SUBSPACE_REGISTRY_PK, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUEUE_EVENT, SUBSPACE_REPORT_OUT, @@ -123,11 +124,7 @@ impl PostgresStore { .map_err(into_error)?; } - for table in [ - SUBSPACE_INDEXES, - SUBSPACE_REGISTRY_IDX, - SUBSPACE_REGISTRY_IDX_GLOBAL, - ] { + for table in [SUBSPACE_INDEXES, SUBSPACE_REGISTRY_IDX] { let table = char::from(table); conn.execute( &format!( diff --git a/crates/store/src/backend/rocksdb/main.rs b/crates/store/src/backend/rocksdb/main.rs index fe7df4e2..813ab744 100644 --- a/crates/store/src/backend/rocksdb/main.rs +++ b/crates/store/src/backend/rocksdb/main.rs @@ -62,7 +62,7 @@ impl RocksDbStore { SUBSPACE_SEARCH_INDEX, SUBSPACE_SPAM_SAMPLES, SUBSPACE_REGISTRY_IDX, - SUBSPACE_REGISTRY_IDX_GLOBAL, + SUBSPACE_REGISTRY_PK, SUBSPACE_DIRECTORY, LEGACY_SUBSPACE_BITMAP_TEXT, LEGACY_SUBSPACE_FTS_INDEX, diff --git a/crates/store/src/backend/sqlite/main.rs b/crates/store/src/backend/sqlite/main.rs index 38b9eeb1..4631e770 100644 --- a/crates/store/src/backend/sqlite/main.rs +++ b/crates/store/src/backend/sqlite/main.rs @@ -70,6 +70,7 @@ impl SqliteStore { SUBSPACE_IN_MEMORY_VALUE, SUBSPACE_PROPERTY, SUBSPACE_REGISTRY, + SUBSPACE_REGISTRY_PK, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUEUE_EVENT, SUBSPACE_REPORT_OUT, @@ -94,11 +95,7 @@ impl SqliteStore { .map_err(into_error)?; } - for table in [ - SUBSPACE_INDEXES, - SUBSPACE_REGISTRY_IDX, - SUBSPACE_REGISTRY_IDX_GLOBAL, - ] { + for table in [SUBSPACE_INDEXES, SUBSPACE_REGISTRY_IDX] { let table = char::from(table); conn.execute( &format!( diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 27715a0a..09e3553f 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -117,7 +117,7 @@ pub const SUBSPACE_IN_MEMORY_COUNTER: u8 = b'y'; pub const SUBSPACE_PROPERTY: u8 = b'p'; pub const SUBSPACE_REGISTRY: u8 = b's'; pub const SUBSPACE_REGISTRY_IDX: u8 = b'b'; -pub const SUBSPACE_REGISTRY_IDX_GLOBAL: u8 = b'c'; +pub const SUBSPACE_REGISTRY_PK: u8 = b'c'; pub const SUBSPACE_DIRECTORY: u8 = b'd'; pub const SUBSPACE_QUEUE_MESSAGE: u8 = b'e'; pub const SUBSPACE_QUEUE_EVENT: u8 = b'q'; diff --git a/crates/store/src/registry/mod.rs b/crates/store/src/registry/mod.rs index e72fcea3..a2e2ebe1 100644 --- a/crates/store/src/registry/mod.rs +++ b/crates/store/src/registry/mod.rs @@ -11,14 +11,16 @@ pub mod query; pub mod write; use crate::{ - Deserialize, SerializeInfallible, U16_LEN, U64_LEN, + Deserialize, SerializeInfallible, U16_LEN, U32_LEN, U64_LEN, write::key::{DeserializeBigEndian, KeySerializer}, }; use registry::{ pickle::{Pickle, PickledStream}, schema::{ prelude::{Object, ObjectInner, ObjectType, Property}, - structs::{DeletedItem, DmarcInternalReport, SpamTrainingSample, Task, TlsInternalReport}, + structs::{ + DeletedItem, DmarcInternalReport, SpamTrainingSample, Task, TlsInternalReport, Trace, + }, }, types::{EnumImpl, ObjectImpl, id::ObjectId}, }; @@ -41,6 +43,12 @@ pub struct RegistryFilter { pub value: RegistryFilterValue, } +#[derive(Debug, PartialEq, Clone, Copy, Eq, Hash)] +pub struct ObjectIdVersioned { + pub object_id: ObjectId, + pub version: u32, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RegistryFilterOp { Equal, @@ -161,3 +169,33 @@ impl Deserialize for ObjectId { )) } } + +impl SerializeInfallible for ObjectIdVersioned { + fn serialize(&self) -> Vec { + KeySerializer::new(U16_LEN + U64_LEN + U32_LEN) + .write(self.object_id.object().to_id()) + .write(self.object_id.id().id()) + .write(self.version) + .finalize() + } +} + +impl Deserialize for ObjectIdVersioned { + fn deserialize(bytes: &[u8]) -> trc::Result { + let object_id = ObjectId::deserialize(bytes)?; + let version = bytes.deserialize_be_u32(U16_LEN + U64_LEN)?; + Ok(Self { object_id, version }) + } +} + +impl Deserialize for Trace { + fn deserialize(bytes: &[u8]) -> trc::Result { + let mut stream = PickledStream::new(bytes); + Trace::unpickle(&mut stream).ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) + } +} diff --git a/crates/store/src/registry/write.rs b/crates/store/src/registry/write.rs index ec487047..b91bbbd3 100644 --- a/crates/store/src/registry/write.rs +++ b/crates/store/src/registry/write.rs @@ -5,10 +5,9 @@ */ use crate::{ - IterateParams, RegistryStore, SUBSPACE_REGISTRY_IDX_GLOBAL, SerializeInfallible, U16_LEN, - U64_LEN, ValueKey, + IterateParams, RegistryStore, SerializeInfallible, U16_LEN, U64_LEN, ValueKey, write::{ - AnyClass, BatchBuilder, RegistryClass, ValueClass, + BatchBuilder, RegistryClass, ValueClass, assert::AssertValue, key::{DeserializeBigEndian, KeySerializer}, }, @@ -28,7 +27,6 @@ use registry::{ use std::{borrow::Cow, fmt::Display}; use trc::AddContext; use types::id::Id; -use utils::codec::leb128::Leb128Reader; pub enum RegistryWriteResult { Success(Id), @@ -268,35 +266,32 @@ impl RegistryStore { }); } } - IndexKey::Search { .. } => {} - IndexKey::Unique { property, .. } => { - let from_key = RegistryClass::from_index_key(key, object_id, 0); - let to_key = RegistryClass::from_index_key(key, object_id, u64::MAX); + IndexKey::Unique { + property, + value_1, + value_2, + global, + } => { + let key = ValueKey::from(ValueClass::Registry(RegistryClass::PrimaryKey { + object_id: (!*global).then_some(object_id), + index_id: property.to_id(), + key: serialize_composite_key(value_1, value_2), + })); if let Some(existing_id) = self - .validate_primary_key(from_key, to_key, Some(object_type)) - .await? - && existing_id.id().id() != item_id + .0 + .store + .get_value::(key) + .await + .caused_by(trc::location!())? + && existing_id != ObjectId::new(object_type, Id::new(item_id)) { return Ok(RegistryWriteResult::PrimaryKeyConflict { - existing_id, - property: *property, - }); - } - } - IndexKey::Global { property, .. } => { - let from_key = RegistryClass::from_index_key(key, 0, 0); - let to_key = RegistryClass::from_index_key(key, u16::MAX, u64::MAX); - - if let Some(existing_id) = - self.validate_primary_key(from_key, to_key, None).await? - && existing_id.id().id() != item_id - { - return Ok(RegistryWriteResult::PrimaryKeyConflict { - existing_id, property: *property, + existing_id, }); } } + IndexKey::Search { .. } => {} } } @@ -326,12 +321,13 @@ impl RegistryStore { vec![], ); } - batch.registry_index(object_id, item_id, set_index.keys.iter(), true); - batch.registry_index(object_id, item_id, clear_index.keys.iter(), false); - batch.set( - ValueClass::Registry(RegistryClass::Item { object_id, item_id }), - out, - ); + batch + .registry_index(object_id, item_id, set_index.keys.iter(), true) + .registry_index(object_id, item_id, clear_index.keys.iter(), false) + .set( + ValueClass::Registry(RegistryClass::Item { object_id, item_id }), + out, + ); Ok(RegistryWriteResult::Success(Id::new(item_id))) } @@ -378,50 +374,38 @@ impl RegistryStore { // Validate relationships let mut linked = Vec::new(); - let key = KeySerializer::new(U64_LEN + U16_LEN + 1) - .write(0u8) - .write(object_type_id) - .write(item_id) - .finalize(); - let prefix_len = key.len(); - let from_key = ValueKey::from(ValueClass::Any(AnyClass { - subspace: SUBSPACE_REGISTRY_IDX_GLOBAL, - key, + let from_key = ValueKey::from(ValueClass::Registry(RegistryClass::Reference { + to_object_id: object_type_id, + to_item_id: item_id, + from_object_id: 0, + from_item_id: 0, })); - let key = KeySerializer::new((U64_LEN * 2) + U16_LEN + 1) - .write(0u8) - .write(object_type_id) - .write(item_id) - .write(u64::MAX) - .finalize(); - let to_key = ValueKey::from(ValueClass::Any(AnyClass { - subspace: SUBSPACE_REGISTRY_IDX_GLOBAL, - key, + let to_key = ValueKey::from(ValueClass::Registry(RegistryClass::Reference { + to_object_id: object_type_id, + to_item_id: item_id, + from_object_id: u16::MAX, + from_item_id: u64::MAX, })); + self.0 .store .iterate( IterateParams::new(from_key, to_key).no_values().ascending(), |key, _| { - let object = ObjectType::from_id(key.deserialize_be_u16(prefix_len)?) - .ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Key, key) - })?; - let id = key - .get(prefix_len + U16_LEN..) - .and_then(|key| key.read_leb128::()) - .map(|r| r.0) - .ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .details(object.as_str()) - .ctx(trc::Key::Key, key) - })?; - linked.push(ObjectId::new(object, Id::new(id))); + if key.len() == (U16_LEN * 2) + (U64_LEN * 2) { + let object = + ObjectType::from_id(key.deserialize_be_u16(U64_LEN + U16_LEN)?) + .ok_or_else(|| { + trc::EventType::Registry( + trc::RegistryEvent::DeserializationError, + ) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Key, key) + })?; + let id = key.deserialize_be_u64(U64_LEN + U16_LEN + U16_LEN)?; + linked.push(ObjectId::new(object, Id::new(id))); + } Ok(true) }, @@ -463,86 +447,6 @@ impl RegistryStore { .map(|_| RegistryWriteResult::Success(Id::from(item_id))) .caused_by(trc::location!()) } - - pub async fn validate_primary_key( - &self, - from_key: RegistryClass, - to_key: RegistryClass, - object: Option, - ) -> trc::Result> { - let from_key = ValueKey::from(from_key); - let to_key = ValueKey::from(to_key); - let key_len = from_key.class.serialized_size() - 1; - - let mut result = None; - self.0 - .store - .iterate( - IterateParams::new(from_key, to_key).no_values().ascending(), - |key, _| { - if key.len() == key_len { - let item_id = key.deserialize_be_u64(key.len() - U64_LEN)?; - let object = if let Some(object) = object { - object - } else { - let object_id = - key.deserialize_be_u16(key.len() - U64_LEN - U16_LEN)?; - ObjectType::from_id(object_id).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Key, key) - })? - }; - - result = Some(ObjectId::new(object, Id::new(item_id))); - } - - Ok(false) - }, - ) - .await - .caused_by(trc::location!()) - .map(|_| result) - } -} - -impl RegistryClass { - pub fn from_index_key(key: &IndexKey<'_>, object_id: u16, item_id: u64) -> Self { - match key { - IndexKey::Unique { property, value } => RegistryClass::Index { - index_id: property.to_id(), - object_id, - item_id, - key: value.serialize(), - }, - IndexKey::Search { property, value } => RegistryClass::Index { - index_id: property.to_id(), - object_id, - item_id, - key: value.serialize(), - }, - IndexKey::Global { - property, - value_1, - value_2, - } => RegistryClass::IndexGlobal { - index_id: property.to_id(), - object_id, - item_id, - key: serialize_composite_key(value_1, value_2), - }, - IndexKey::ForeignKey { - object_id: to_object_id, - .. - } => RegistryClass::Reference { - to_object_id: to_object_id.object().to_id(), - to_item_id: to_object_id.id().id(), - from_item_id: item_id, - from_object_id: object_id, - }, - } - } } impl BatchBuilder { @@ -554,15 +458,52 @@ impl BatchBuilder { is_set: bool, ) -> &mut Self { for key in index_keys { - if is_set { - self.set( - ValueClass::Registry(RegistryClass::from_index_key(key, object_id, item_id)), + let (key, value) = match key { + IndexKey::Search { property, value } => ( + RegistryClass::Index { + index_id: property.to_id(), + object_id, + item_id, + key: value.serialize(), + }, vec![], - ); + ), + IndexKey::Unique { + property, + value_1, + value_2, + global, + } => ( + RegistryClass::PrimaryKey { + object_id: (!*global).then_some(object_id), + index_id: property.to_id(), + key: serialize_composite_key(value_1, value_2), + }, + KeySerializer::new(U16_LEN + U64_LEN) + .write(object_id) + .write(item_id) + .finalize(), + ), + IndexKey::ForeignKey { + object_id: to_object_id, + .. + } => ( + RegistryClass::Reference { + to_object_id: to_object_id.object().to_id(), + to_item_id: to_object_id.id().id(), + from_item_id: item_id, + from_object_id: object_id, + }, + vec![], + ), + }; + if is_set { + if !value.is_empty() { + self.assert_value(ValueClass::Registry(key.clone()), ()); + } + self.set(ValueClass::Registry(key), value); } else { - self.clear(ValueClass::Registry(RegistryClass::from_index_key( - key, object_id, item_id, - ))); + self.clear(ValueClass::Registry(key)); } } self diff --git a/crates/store/src/write/batch.rs b/crates/store/src/write/batch.rs index a989e390..62ceace9 100644 --- a/crates/store/src/write/batch.rs +++ b/crates/store/src/write/batch.rs @@ -483,6 +483,18 @@ impl BatchBuilder { class.serialize(), ) } + + pub fn schedule_task_with_id(&mut self, id: u64, task: Task) -> &mut Self { + let due = task.due_timestamp(); + let class = task.object_type().to_id(); + let task = task.to_pickled_vec(); + + self.set(ValueClass::TaskQueue(TaskQueueClass::Task { id }), task) + .set( + ValueClass::TaskQueue(TaskQueueClass::Due { id, due }), + class.serialize(), + ) + } } pub struct CommitPointIterator { diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 840aca6b..e054b539 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -12,7 +12,7 @@ use crate::{ SUBSPACE_DELETED_ITEMS, SUBSPACE_DIRECTORY, SUBSPACE_IN_MEMORY_COUNTER, SUBSPACE_IN_MEMORY_VALUE, SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_PROPERTY, SUBSPACE_QUEUE_EVENT, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUOTA, SUBSPACE_REGISTRY, - SUBSPACE_REGISTRY_IDX, SUBSPACE_REGISTRY_IDX_GLOBAL, SUBSPACE_REPORT_IN, SUBSPACE_REPORT_OUT, + SUBSPACE_REGISTRY_IDX, SUBSPACE_REGISTRY_PK, SUBSPACE_REPORT_IN, SUBSPACE_REPORT_OUT, SUBSPACE_SEARCH_INDEX, SUBSPACE_SPAM_SAMPLES, SUBSPACE_TASK_QUEUE, SUBSPACE_TELEMETRY_METRIC, SUBSPACE_TELEMETRY_SPAN, U16_LEN, U32_LEN, U64_LEN, ValueKey, WITH_SUBSPACE, write::{ @@ -323,22 +323,18 @@ impl ValueClass { from_object_id, from_item_id, } => serializer - .write(0u8) .write(*to_object_id) .write(*to_item_id) .write(*from_object_id) - .write_leb128(*from_item_id), - RegistryClass::IndexGlobal { - index_id, + .write(*from_item_id), + RegistryClass::PrimaryKey { object_id, - item_id, + index_id, key, } => serializer - .write(1u8) + .write((*object_id).unwrap_or(u16::MAX)) .write(*index_id) - .write(key.as_slice()) - .write(*object_id) - .write(*item_id), + .write(key.as_slice()), RegistryClass::IdCounter { object_id } => serializer.write(*object_id), }, ValueClass::Queue(queue) => match queue { @@ -351,15 +347,8 @@ impl ValueClass { QueueClass::QuotaSize(key) => serializer.write(1u8).write(key.as_slice()), }, ValueClass::Telemetry(telemetry) => match telemetry { - TelemetryClass::Span { span_id } => serializer.write(*span_id), - TelemetryClass::Metric { - timestamp, - metric_id, - node_id, - } => serializer - .write(*timestamp) - .write_leb128(*metric_id) - .write_leb128(*node_id), + TelemetryClass::Span(span_id) => serializer.write(*span_id), + TelemetryClass::Metric(metric_id) => serializer.write(*metric_id), }, ValueClass::DocumentId => serializer.write(account_id).write(collection), ValueClass::ChangeId => serializer.write(account_id), @@ -506,9 +495,9 @@ impl ValueClass { ValueClass::InMemory(InMemoryClass::Counter(v) | InMemoryClass::Key(v)) => v.len(), ValueClass::Registry(registry) => match registry { RegistryClass::Item { .. } => U16_LEN + U64_LEN + 1, - RegistryClass::Reference { .. } => ((U16_LEN + U64_LEN) * 2) + 2, + RegistryClass::Reference { .. } => ((U16_LEN + U64_LEN) * 2) + 1, RegistryClass::Index { key, .. } => (U16_LEN * 2) + U64_LEN + key.len() + 1, - RegistryClass::IndexGlobal { key, .. } => (U16_LEN * 2) + U64_LEN + key.len() + 2, + RegistryClass::PrimaryKey { key, .. } => (U16_LEN * 2) + key.len() + 1, RegistryClass::Id { .. } => U16_LEN + U64_LEN + 1, RegistryClass::IdCounter { .. } => U16_LEN + 1, }, @@ -533,8 +522,7 @@ impl ValueClass { QueueClass::QuotaCount(v) | QueueClass::QuotaSize(v) => v.len(), }, ValueClass::Telemetry(telemetry) => match telemetry { - TelemetryClass::Span { .. } => U64_LEN + 1, - TelemetryClass::Metric { .. } => U64_LEN * 2 + 1, + TelemetryClass::Span(_) | TelemetryClass::Metric(_) => U64_LEN + 1, }, ValueClass::DocumentId | ValueClass::Quota | ValueClass::TenantQuota(_) => U32_LEN + 1, ValueClass::ChangeId => U32_LEN, @@ -581,8 +569,8 @@ impl ValueClass { _ => SUBSPACE_REGISTRY, }, RegistryClass::Id { .. } | RegistryClass::Index { .. } => SUBSPACE_REGISTRY_IDX, - RegistryClass::Reference { .. } | RegistryClass::IndexGlobal { .. } => { - SUBSPACE_REGISTRY_IDX_GLOBAL + RegistryClass::Reference { .. } | RegistryClass::PrimaryKey { .. } => { + SUBSPACE_REGISTRY_PK } RegistryClass::IdCounter { .. } => SUBSPACE_COUNTER, }, diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index 9d140059..23f4d3ad 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -265,10 +265,9 @@ pub enum RegistryClass { item_id: u64, key: Vec, }, - IndexGlobal { + PrimaryKey { + object_id: Option, index_id: u16, - object_id: u16, - item_id: u64, key: Vec, }, Id { @@ -290,14 +289,8 @@ pub enum QueueClass { #[derive(Debug, PartialEq, Clone, Eq, Hash)] pub enum TelemetryClass { - Span { - span_id: u64, - }, - Metric { - timestamp: u64, - metric_id: u64, - node_id: u64, - }, + Span(u64), + Metric(u64), } #[derive(Debug, PartialEq, Clone, Eq, Hash)] diff --git a/crates/trc/src/event/enums.rs b/crates/trc/src/event/enums.rs index 31fa8e9a..2c118302 100644 --- a/crates/trc/src/event/enums.rs +++ b/crates/trc/src/event/enums.rs @@ -7,6 +7,7 @@ // This file is auto-generated. Do not edit directly. pub const TOTAL_EVENT_COUNT: usize = 596; +pub const TOTAL_METRIC_COUNT: usize = 338; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum EventType { @@ -532,6 +533,7 @@ pub enum OutgoingReportEvent { SubmissionError = 343, NoRecipientsFound = 338, Locked = 337, + MaxSizeExceeded = 59, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -607,7 +609,6 @@ pub enum RegistryEvent { BuildWarning = 55, NotSupported = 64, ValidationError = 63, - Reserved03 = 59, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -981,6 +982,7 @@ pub enum MetricType { EvalStoreNotFound = 113, HttpRequestTime = 12, HttpActiveConnections = 17, + HttpConnectionStart = 337, HttpError = 114, HttpRequestBody = 115, HttpResponseBody = 116, diff --git a/crates/trc/src/event/enums_impl.rs b/crates/trc/src/event/enums_impl.rs index 6e844483..711ed5e5 100644 --- a/crates/trc/src/event/enums_impl.rs +++ b/crates/trc/src/event/enums_impl.rs @@ -349,6 +349,7 @@ impl EventType { b"outgoing-report.submission-error" => EventType::OutgoingReport(OutgoingReportEvent::SubmissionError), b"outgoing-report.no-recipients-found" => EventType::OutgoingReport(OutgoingReportEvent::NoRecipientsFound), b"outgoing-report.locked" => EventType::OutgoingReport(OutgoingReportEvent::Locked), + b"outgoing-report.max-size-exceeded" => EventType::OutgoingReport(OutgoingReportEvent::MaxSizeExceeded), b"pop3.connection-start" => EventType::Pop3(Pop3Event::ConnectionStart), b"pop3.connection-end" => EventType::Pop3(Pop3Event::ConnectionEnd), b"pop3.delete" => EventType::Pop3(Pop3Event::Delete), @@ -399,7 +400,6 @@ impl EventType { b"registry.build-warning" => EventType::Registry(RegistryEvent::BuildWarning), b"registry.not-supported" => EventType::Registry(RegistryEvent::NotSupported), b"registry.validation-error" => EventType::Registry(RegistryEvent::ValidationError), - b"registry.reserved03" => EventType::Registry(RegistryEvent::Reserved03), b"resource.not-found" => EventType::Resource(ResourceEvent::NotFound), b"resource.bad-parameters" => EventType::Resource(ResourceEvent::BadParameters), b"resource.error" => EventType::Resource(ResourceEvent::Error), @@ -1045,6 +1045,9 @@ impl EventType { "outgoing-report.no-recipients-found" } EventType::OutgoingReport(OutgoingReportEvent::Locked) => "outgoing-report.locked", + EventType::OutgoingReport(OutgoingReportEvent::MaxSizeExceeded) => { + "outgoing-report.max-size-exceeded" + } EventType::Pop3(Pop3Event::ConnectionStart) => "pop3.connection-start", EventType::Pop3(Pop3Event::ConnectionEnd) => "pop3.connection-end", EventType::Pop3(Pop3Event::Delete) => "pop3.delete", @@ -1105,7 +1108,6 @@ impl EventType { EventType::Registry(RegistryEvent::BuildWarning) => "registry.build-warning", EventType::Registry(RegistryEvent::NotSupported) => "registry.not-supported", EventType::Registry(RegistryEvent::ValidationError) => "registry.validation-error", - EventType::Registry(RegistryEvent::Reserved03) => "registry.reserved03", EventType::Resource(ResourceEvent::NotFound) => "resource.not-found", EventType::Resource(ResourceEvent::BadParameters) => "resource.bad-parameters", EventType::Resource(ResourceEvent::Error) => "resource.error", @@ -1670,6 +1672,7 @@ impl EventType { EventType::OutgoingReport(OutgoingReportEvent::SubmissionError) => 343, EventType::OutgoingReport(OutgoingReportEvent::NoRecipientsFound) => 338, EventType::OutgoingReport(OutgoingReportEvent::Locked) => 337, + EventType::OutgoingReport(OutgoingReportEvent::MaxSizeExceeded) => 59, EventType::Pop3(Pop3Event::ConnectionStart) => 348, EventType::Pop3(Pop3Event::ConnectionEnd) => 347, EventType::Pop3(Pop3Event::Delete) => 349, @@ -1720,7 +1723,6 @@ impl EventType { EventType::Registry(RegistryEvent::BuildWarning) => 55, EventType::Registry(RegistryEvent::NotSupported) => 64, EventType::Registry(RegistryEvent::ValidationError) => 63, - EventType::Registry(RegistryEvent::Reserved03) => 59, EventType::Resource(ResourceEvent::NotFound) => 389, EventType::Resource(ResourceEvent::BadParameters) => 386, EventType::Resource(ResourceEvent::Error) => 388, @@ -2307,6 +2309,9 @@ impl EventType { OutgoingReportEvent::NoRecipientsFound, )), 337 => Some(EventType::OutgoingReport(OutgoingReportEvent::Locked)), + 59 => Some(EventType::OutgoingReport( + OutgoingReportEvent::MaxSizeExceeded, + )), 348 => Some(EventType::Pop3(Pop3Event::ConnectionStart)), 347 => Some(EventType::Pop3(Pop3Event::ConnectionEnd)), 349 => Some(EventType::Pop3(Pop3Event::Delete)), @@ -2357,7 +2362,6 @@ impl EventType { 55 => Some(EventType::Registry(RegistryEvent::BuildWarning)), 64 => Some(EventType::Registry(RegistryEvent::NotSupported)), 63 => Some(EventType::Registry(RegistryEvent::ValidationError)), - 59 => Some(EventType::Registry(RegistryEvent::Reserved03)), 389 => Some(EventType::Resource(ResourceEvent::NotFound)), 386 => Some(EventType::Resource(ResourceEvent::BadParameters)), 388 => Some(EventType::Resource(ResourceEvent::Error)), @@ -2755,7 +2759,6 @@ impl EventType { EventType::Queue(QueueEvent::RateLimitExceeded) => Level::Info, EventType::Queue(QueueEvent::ConcurrencyLimitExceeded) => Level::Info, EventType::Queue(QueueEvent::QuotaExceeded) => Level::Info, - EventType::Registry(RegistryEvent::Reserved03) => Level::Info, EventType::Resource(ResourceEvent::DownloadExternal) => Level::Info, EventType::Resource(ResourceEvent::WebadminUnpacked) => Level::Info, EventType::Security(SecurityEvent::AuthenticationBan) => Level::Info, @@ -3349,6 +3352,9 @@ impl EventType { EventType::OutgoingReport(OutgoingReportEvent::Locked) => { "Report is locked by another process" } + EventType::OutgoingReport(OutgoingReportEvent::MaxSizeExceeded) => { + "Report size exceeds maximum" + } EventType::Pop3(Pop3Event::ConnectionStart) => "POP3 connection started", EventType::Pop3(Pop3Event::ConnectionEnd) => "POP3 connection ended", EventType::Pop3(Pop3Event::Delete) => "POP3 DELETE command", @@ -3411,7 +3417,6 @@ impl EventType { "Operation not supported by local registry" } EventType::Registry(RegistryEvent::ValidationError) => "Object validation error", - EventType::Registry(RegistryEvent::Reserved03) => "Importing external configuration", EventType::Resource(ResourceEvent::NotFound) => "Resource not found", EventType::Resource(ResourceEvent::BadParameters) => "Bad resource parameters", EventType::Resource(ResourceEvent::Error) => "Resource error", @@ -4260,6 +4265,9 @@ impl EventType { EventType::OutgoingReport(OutgoingReportEvent::Locked) => { "The report is locked by another process" } + EventType::OutgoingReport(OutgoingReportEvent::MaxSizeExceeded) => { + "The report size exceeds the maximum allowed size" + } EventType::Pop3(Pop3Event::ConnectionStart) => "POP3 connection started", EventType::Pop3(Pop3Event::ConnectionEnd) => "POP3 connection ended", EventType::Pop3(Pop3Event::Delete) => "Client deleted a message", @@ -4348,9 +4356,6 @@ impl EventType { EventType::Registry(RegistryEvent::ValidationError) => { "An error occurred while validating a registry object" } - EventType::Registry(RegistryEvent::Reserved03) => { - "An external configuration is being imported" - } EventType::Resource(ResourceEvent::NotFound) => "The resource was not found", EventType::Resource(ResourceEvent::BadParameters) => "The resource parameters are bad", EventType::Resource(ResourceEvent::Error) => "An error occurred with the resource", @@ -5353,6 +5358,7 @@ impl EventType { EventType::OutgoingReport(OutgoingReportEvent::SubmissionError), EventType::OutgoingReport(OutgoingReportEvent::NoRecipientsFound), EventType::OutgoingReport(OutgoingReportEvent::Locked), + EventType::OutgoingReport(OutgoingReportEvent::MaxSizeExceeded), EventType::Pop3(Pop3Event::ConnectionStart), EventType::Pop3(Pop3Event::ConnectionEnd), EventType::Pop3(Pop3Event::Delete), @@ -5403,7 +5409,6 @@ impl EventType { EventType::Registry(RegistryEvent::BuildWarning), EventType::Registry(RegistryEvent::NotSupported), EventType::Registry(RegistryEvent::ValidationError), - EventType::Registry(RegistryEvent::Reserved03), EventType::Resource(ResourceEvent::NotFound), EventType::Resource(ResourceEvent::BadParameters), EventType::Resource(ResourceEvent::Error), @@ -5716,6 +5721,7 @@ impl MetricType { b"eval.store-not-found" => MetricType::EvalStoreNotFound, b"http.request-time" => MetricType::HttpRequestTime, b"http.active-connections" => MetricType::HttpActiveConnections, + b"http.connection-start" => MetricType::HttpConnectionStart, b"http.error" => MetricType::HttpError, b"http.request-body" => MetricType::HttpRequestBody, b"http.response-body" => MetricType::HttpResponseBody, @@ -6059,6 +6065,7 @@ impl MetricType { MetricType::EvalStoreNotFound => "eval.store-not-found", MetricType::HttpRequestTime => "http.request-time", MetricType::HttpActiveConnections => "http.active-connections", + MetricType::HttpConnectionStart => "http.connection-start", MetricType::HttpError => "http.error", MetricType::HttpRequestBody => "http.request-body", MetricType::HttpResponseBody => "http.response-body", @@ -6413,6 +6420,7 @@ impl MetricType { MetricType::EvalStoreNotFound => 113, MetricType::HttpRequestTime => 12, MetricType::HttpActiveConnections => 17, + MetricType::HttpConnectionStart => 337, MetricType::HttpError => 114, MetricType::HttpRequestBody => 115, MetricType::HttpResponseBody => 116, @@ -6755,6 +6763,7 @@ impl MetricType { 113 => Some(MetricType::EvalStoreNotFound), 12 => Some(MetricType::HttpRequestTime), 17 => Some(MetricType::HttpActiveConnections), + 337 => Some(MetricType::HttpConnectionStart), 114 => Some(MetricType::HttpError), 115 => Some(MetricType::HttpRequestBody), 116 => Some(MetricType::HttpResponseBody), @@ -7091,6 +7100,7 @@ impl MetricType { MetricType::EvalError => 138, MetricType::EvalDirectoryNotFound => 137, MetricType::EvalStoreNotFound => 140, + MetricType::HttpConnectionStart => 153, MetricType::HttpError => 154, MetricType::HttpRequestBody => 155, MetricType::HttpResponseBody => 157, @@ -7416,6 +7426,7 @@ impl MetricType { MetricType::EvalStoreNotFound => "Store not found while evaluating expression", MetricType::HttpRequestTime => "HTTP request duration", MetricType::HttpActiveConnections => "Active HTTP connections", + MetricType::HttpConnectionStart => "HTTP connection started", MetricType::HttpError => "HTTP error occurred", MetricType::HttpRequestBody => "HTTP request body", MetricType::HttpResponseBody => "HTTP response body", @@ -7771,6 +7782,7 @@ impl MetricType { | MetricType::EvalError | MetricType::EvalDirectoryNotFound | MetricType::EvalStoreNotFound + | MetricType::HttpConnectionStart | MetricType::HttpError | MetricType::HttpRequestBody | MetricType::HttpResponseBody @@ -8110,6 +8122,7 @@ impl MetricType { MetricType::EvalStoreNotFound, MetricType::HttpRequestTime, MetricType::HttpActiveConnections, + MetricType::HttpConnectionStart, MetricType::HttpError, MetricType::HttpRequestBody, MetricType::HttpResponseBody, diff --git a/crates/trc/src/event/mod.rs b/crates/trc/src/event/mod.rs index 7d2e292e..030e3014 100644 --- a/crates/trc/src/event/mod.rs +++ b/crates/trc/src/event/mod.rs @@ -703,3 +703,148 @@ impl AsRef> for Event { self } } + +impl Key { + pub fn code(&self) -> u64 { + match self { + Key::AccountName => 0, + Key::AccountId => 1, + Key::BlobId => 2, + Key::CausedBy => 3, + Key::ChangeId => 4, + Key::Code => 5, + Key::Collection => 6, + Key::Contents => 7, + Key::Details => 8, + Key::DkimFail => 9, + Key::DkimNone => 10, + Key::DkimPass => 11, + Key::DmarcNone => 12, + Key::DmarcPass => 13, + Key::DmarcQuarantine => 14, + Key::DmarcReject => 15, + Key::DocumentId => 16, + Key::Domain => 17, + Key::Due => 18, + Key::Elapsed => 19, + Key::Expires => 20, + Key::From => 21, + Key::Hostname => 22, + Key::Id => 23, + Key::Key => 24, + Key::Limit => 25, + Key::ListenerId => 26, + Key::LocalIp => 27, + Key::LocalPort => 28, + Key::MailboxName => 29, + Key::MailboxId => 30, + Key::MessageId => 31, + Key::NextDsn => 32, + Key::NextRetry => 33, + Key::Path => 34, + Key::Policy => 35, + Key::QueueId => 36, + Key::RangeFrom => 37, + Key::RangeTo => 38, + Key::Reason => 39, + Key::RemoteIp => 40, + Key::RemotePort => 41, + Key::ReportId => 42, + Key::Result => 43, + Key::Size => 44, + Key::Source => 45, + Key::SpanId => 46, + Key::SpfFail => 47, + Key::SpfNone => 48, + Key::SpfPass => 49, + Key::Strict => 50, + Key::Tls => 51, + Key::To => 52, + Key::Total => 53, + Key::TotalFailures => 54, + Key::TotalSuccesses => 55, + Key::Type => 56, + Key::Uid => 57, + Key::UidNext => 58, + Key::UidValidity => 59, + Key::Url => 60, + Key::ValidFrom => 61, + Key::ValidTo => 62, + Key::Value => 63, + Key::Version => 64, + Key::QueueName => 65, + } + } + + pub fn from_code(code: u64) -> Option { + match code { + 0 => Some(Key::AccountName), + 1 => Some(Key::AccountId), + 2 => Some(Key::BlobId), + 3 => Some(Key::CausedBy), + 4 => Some(Key::ChangeId), + 5 => Some(Key::Code), + 6 => Some(Key::Collection), + 7 => Some(Key::Contents), + 8 => Some(Key::Details), + 9 => Some(Key::DkimFail), + 10 => Some(Key::DkimNone), + 11 => Some(Key::DkimPass), + 12 => Some(Key::DmarcNone), + 13 => Some(Key::DmarcPass), + 14 => Some(Key::DmarcQuarantine), + 15 => Some(Key::DmarcReject), + 16 => Some(Key::DocumentId), + 17 => Some(Key::Domain), + 18 => Some(Key::Due), + 19 => Some(Key::Elapsed), + 20 => Some(Key::Expires), + 21 => Some(Key::From), + 22 => Some(Key::Hostname), + 23 => Some(Key::Id), + 24 => Some(Key::Key), + 25 => Some(Key::Limit), + 26 => Some(Key::ListenerId), + 27 => Some(Key::LocalIp), + 28 => Some(Key::LocalPort), + 29 => Some(Key::MailboxName), + 30 => Some(Key::MailboxId), + 31 => Some(Key::MessageId), + 32 => Some(Key::NextDsn), + 33 => Some(Key::NextRetry), + 34 => Some(Key::Path), + 35 => Some(Key::Policy), + 36 => Some(Key::QueueId), + 37 => Some(Key::RangeFrom), + 38 => Some(Key::RangeTo), + 39 => Some(Key::Reason), + 40 => Some(Key::RemoteIp), + 41 => Some(Key::RemotePort), + 42 => Some(Key::ReportId), + 43 => Some(Key::Result), + 44 => Some(Key::Size), + 45 => Some(Key::Source), + 46 => Some(Key::SpanId), + 47 => Some(Key::SpfFail), + 48 => Some(Key::SpfNone), + 49 => Some(Key::SpfPass), + 50 => Some(Key::Strict), + 51 => Some(Key::Tls), + 52 => Some(Key::To), + 53 => Some(Key::Total), + 54 => Some(Key::TotalFailures), + 55 => Some(Key::TotalSuccesses), + 56 => Some(Key::Type), + 57 => Some(Key::Uid), + 58 => Some(Key::UidNext), + 59 => Some(Key::UidValidity), + 60 => Some(Key::Url), + 61 => Some(Key::ValidFrom), + 62 => Some(Key::ValidTo), + 63 => Some(Key::Value), + 64 => Some(Key::Version), + 65 => Some(Key::QueueName), + _ => None, + } + } +} diff --git a/crates/trc/src/serializers/binary.rs b/crates/trc/src/serializers/binary.rs index 7a25bc50..bf469ac7 100644 --- a/crates/trc/src/serializers/binary.rs +++ b/crates/trc/src/serializers/binary.rs @@ -303,148 +303,3 @@ fn leb128_read<'x>(iter: &mut impl Iterator) -> Option { None } - -impl Key { - fn code(&self) -> u64 { - match self { - Key::AccountName => 0, - Key::AccountId => 1, - Key::BlobId => 2, - Key::CausedBy => 3, - Key::ChangeId => 4, - Key::Code => 5, - Key::Collection => 6, - Key::Contents => 7, - Key::Details => 8, - Key::DkimFail => 9, - Key::DkimNone => 10, - Key::DkimPass => 11, - Key::DmarcNone => 12, - Key::DmarcPass => 13, - Key::DmarcQuarantine => 14, - Key::DmarcReject => 15, - Key::DocumentId => 16, - Key::Domain => 17, - Key::Due => 18, - Key::Elapsed => 19, - Key::Expires => 20, - Key::From => 21, - Key::Hostname => 22, - Key::Id => 23, - Key::Key => 24, - Key::Limit => 25, - Key::ListenerId => 26, - Key::LocalIp => 27, - Key::LocalPort => 28, - Key::MailboxName => 29, - Key::MailboxId => 30, - Key::MessageId => 31, - Key::NextDsn => 32, - Key::NextRetry => 33, - Key::Path => 34, - Key::Policy => 35, - Key::QueueId => 36, - Key::RangeFrom => 37, - Key::RangeTo => 38, - Key::Reason => 39, - Key::RemoteIp => 40, - Key::RemotePort => 41, - Key::ReportId => 42, - Key::Result => 43, - Key::Size => 44, - Key::Source => 45, - Key::SpanId => 46, - Key::SpfFail => 47, - Key::SpfNone => 48, - Key::SpfPass => 49, - Key::Strict => 50, - Key::Tls => 51, - Key::To => 52, - Key::Total => 53, - Key::TotalFailures => 54, - Key::TotalSuccesses => 55, - Key::Type => 56, - Key::Uid => 57, - Key::UidNext => 58, - Key::UidValidity => 59, - Key::Url => 60, - Key::ValidFrom => 61, - Key::ValidTo => 62, - Key::Value => 63, - Key::Version => 64, - Key::QueueName => 65, - } - } - - fn from_code(code: u64) -> Option { - match code { - 0 => Some(Key::AccountName), - 1 => Some(Key::AccountId), - 2 => Some(Key::BlobId), - 3 => Some(Key::CausedBy), - 4 => Some(Key::ChangeId), - 5 => Some(Key::Code), - 6 => Some(Key::Collection), - 7 => Some(Key::Contents), - 8 => Some(Key::Details), - 9 => Some(Key::DkimFail), - 10 => Some(Key::DkimNone), - 11 => Some(Key::DkimPass), - 12 => Some(Key::DmarcNone), - 13 => Some(Key::DmarcPass), - 14 => Some(Key::DmarcQuarantine), - 15 => Some(Key::DmarcReject), - 16 => Some(Key::DocumentId), - 17 => Some(Key::Domain), - 18 => Some(Key::Due), - 19 => Some(Key::Elapsed), - 20 => Some(Key::Expires), - 21 => Some(Key::From), - 22 => Some(Key::Hostname), - 23 => Some(Key::Id), - 24 => Some(Key::Key), - 25 => Some(Key::Limit), - 26 => Some(Key::ListenerId), - 27 => Some(Key::LocalIp), - 28 => Some(Key::LocalPort), - 29 => Some(Key::MailboxName), - 30 => Some(Key::MailboxId), - 31 => Some(Key::MessageId), - 32 => Some(Key::NextDsn), - 33 => Some(Key::NextRetry), - 34 => Some(Key::Path), - 35 => Some(Key::Policy), - 36 => Some(Key::QueueId), - 37 => Some(Key::RangeFrom), - 38 => Some(Key::RangeTo), - 39 => Some(Key::Reason), - 40 => Some(Key::RemoteIp), - 41 => Some(Key::RemotePort), - 42 => Some(Key::ReportId), - 43 => Some(Key::Result), - 44 => Some(Key::Size), - 45 => Some(Key::Source), - 46 => Some(Key::SpanId), - 47 => Some(Key::SpfFail), - 48 => Some(Key::SpfNone), - 49 => Some(Key::SpfPass), - 50 => Some(Key::Strict), - 51 => Some(Key::Tls), - 52 => Some(Key::To), - 53 => Some(Key::Total), - 54 => Some(Key::TotalFailures), - 55 => Some(Key::TotalSuccesses), - 56 => Some(Key::Type), - 57 => Some(Key::Uid), - 58 => Some(Key::UidNext), - 59 => Some(Key::UidValidity), - 60 => Some(Key::Url), - 61 => Some(Key::ValidFrom), - 62 => Some(Key::ValidTo), - 63 => Some(Key::Value), - 64 => Some(Key::Version), - 65 => Some(Key::QueueName), - _ => None, - } - } -} diff --git a/crates/trc/src/serializers/json.rs b/crates/trc/src/serializers/json.rs index 8a0035de..58f5ab3d 100644 --- a/crates/trc/src/serializers/json.rs +++ b/crates/trc/src/serializers/json.rs @@ -281,3 +281,22 @@ impl<'de> serde::Deserialize<'de> for MetricType { Self::parse(s).ok_or_else(|| serde::de::Error::unknown_variant(s, &[])) } } + +impl serde::Serialize for Key { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + serializer.serialize_str(self.name()) + } +} + +impl<'de> serde::Deserialize<'de> for Key { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let s = <&str>::deserialize(deserializer)?; + Self::try_parse(s).ok_or_else(|| serde::de::Error::unknown_variant(s, &[])) + } +} diff --git a/crates/utils/src/snowflake.rs b/crates/utils/src/snowflake.rs index a9b4f8fb..d2996f3d 100644 --- a/crates/utils/src/snowflake.rs +++ b/crates/utils/src/snowflake.rs @@ -74,6 +74,10 @@ impl SnowflakeIdGenerator { .and_then(|diff| Self::from_duration(Duration::from_secs(diff))) } + pub fn from_timestamp_and_sequence_id(timestamp: u64, sequence: u64) -> Option { + Self::from_timestamp(timestamp).map(|id| id | (sequence << NODE_ID_LEN) | node_id()) + } + pub fn from_sequence_id(sequence: u64) -> Option { let sequence = sequence & SEQUENCE_MASK; diff --git a/tests/src/store/cleanup.rs b/tests/src/store/cleanup.rs index 67f28b56..ed91572d 100644 --- a/tests/src/store/cleanup.rs +++ b/tests/src/store/cleanup.rs @@ -38,7 +38,7 @@ pub async fn store_destroy(store: &Store) { SUBSPACE_TELEMETRY_METRIC, SUBSPACE_SEARCH_INDEX, SUBSPACE_REGISTRY_IDX, - SUBSPACE_REGISTRY_IDX_GLOBAL, + SUBSPACE_REGISTRY_PK, SUBSPACE_DIRECTORY, ] { if subspace == SUBSPACE_SEARCH_INDEX && store.is_pg_or_mysql() { @@ -251,7 +251,7 @@ pub async fn store_assert_is_empty(store: &Store, blob_store: BlobStore, include (SUBSPACE_TELEMETRY_METRIC, true), (SUBSPACE_SEARCH_INDEX, true), (SUBSPACE_REGISTRY_IDX, false), - (SUBSPACE_REGISTRY_IDX_GLOBAL, false), + (SUBSPACE_REGISTRY_PK, true), (SUBSPACE_DIRECTORY, true), ] { if (subspace == SUBSPACE_SEARCH_INDEX && store.is_pg_or_mysql())