diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c512edf..6a60dc9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ If you are upgrading from v0.16.x, replace the binary (or run `docker pull`). If - Uppercase `MAILTO` calendar addresses become invalid SMTP recipients. - Scheduling invitations on a shared, non-owned calendar fail with `MAIL FROM unauthorized`. - HTTP: Disable `allowedEndpoints` expression in recovery mode. +- Telemetry: Tasks are serialized to the wrong store when using separate stores for telemetry and data. ## [0.16.13] - 2026-07-12 diff --git a/crates/common/src/config/telemetry.rs b/crates/common/src/config/telemetry.rs index f0b272d4..81625d39 100644 --- a/crates/common/src/config/telemetry.rs +++ b/crates/common/src/config/telemetry.rs @@ -98,6 +98,7 @@ pub struct WebhookTracer { #[cfg(feature = "enterprise")] pub struct StoreTracer { pub store: store::Store, + pub data: Option, } // SPDX-SnippetEnd @@ -441,6 +442,8 @@ impl Tracers { interests: Default::default(), lossy: false, typ: TelemetrySubscriberType::StoreTracer(StoreTracer { + data: (!storage.tracing.is_same(&storage.data)) + .then(|| storage.data.clone()), store: storage.tracing.clone(), }), }; diff --git a/crates/common/src/telemetry/tracers/store.rs b/crates/common/src/telemetry/tracers/store.rs index 9d4a9653..05350b33 100644 --- a/crates/common/src/telemetry/tracers/store.rs +++ b/crates/common/src/telemetry/tracers/store.rs @@ -36,7 +36,9 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac tokio::spawn(async move { let mut active_spans = AHashMap::new(); let store = settings.store; + let data_store = settings.data; let mut batch = BatchBuilder::new(); + let mut task_batch = BatchBuilder::new(); while let Some(events) = rx.recv().await { for event in events { @@ -55,29 +57,44 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac .any(|(k, v)| matches!((k, v), (Key::QueueId, Value::UInt(_)))) { // Serialize events - batch - .set( - ValueClass::Telemetry(TelemetryClass::Span(span_id)), - Trace::from_events( - [span.as_ref()] - .into_iter() - .chain(events.iter().map(|event| event.as_ref())) - .chain([event.as_ref()]), - events.len() + 2, - ) - .to_pickled_vec(), + batch.set( + ValueClass::Telemetry(TelemetryClass::Span(span_id)), + Trace::from_events( + [span.as_ref()] + .into_iter() + .chain(events.iter().map(|event| event.as_ref())) + .chain([event.as_ref()]), + events.len() + 2, ) - .schedule_task(Task::IndexTrace(TaskIndexTrace { - status: TaskStatus::now(), - trace_id: span_id.into(), - })); + .to_pickled_vec(), + ); + + let index_task = Task::IndexTrace(TaskIndexTrace { + status: TaskStatus::now(), + trace_id: span_id.into(), + }); + if data_store.is_some() { + task_batch.schedule_task(index_task); + } else { + batch.schedule_task(index_task); + } } } } if !batch.is_empty() { - if let Err(err) = store.write(batch.build_all()).await { - trc::error!(err.caused_by(trc::location!())); + match store.write(batch.build_all()).await { + Ok(_) => { + if let Some(data_store) = &data_store { + if let Err(err) = data_store.write(task_batch.build_all()).await { + trc::error!(err.caused_by(trc::location!())); + } + task_batch = BatchBuilder::new(); + } + } + Err(err) => { + trc::error!(err.caused_by(trc::location!())); + } } batch = BatchBuilder::new(); } diff --git a/crates/services/src/task_manager/index.rs b/crates/services/src/task_manager/index.rs index 66c60cf0..d654cd57 100644 --- a/crates/services/src/task_manager/index.rs +++ b/crates/services/src/task_manager/index.rs @@ -293,7 +293,7 @@ impl SearchIndexTask for Server { pub(crate) async fn reindex_telemetry(server: &Server) -> trc::Result<()> { let mut spans = Vec::new(); server - .store() + .tracing_store() .iterate( IterateParams::new( ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span(0))), diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index dcbfa6fd..de2673c2 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -652,6 +652,26 @@ impl Store { !matches!(self, Self::None) } + pub fn is_same(&self, other: &Store) -> bool { + match (self, other) { + #[cfg(feature = "sqlite")] + (Store::SQLite(a), Store::SQLite(b)) => Arc::ptr_eq(a, b), + #[cfg(feature = "foundation")] + (Store::FoundationDb(a), Store::FoundationDb(b)) => Arc::ptr_eq(a, b), + #[cfg(feature = "postgres")] + (Store::PostgreSQL(a), Store::PostgreSQL(b)) => Arc::ptr_eq(a, b), + #[cfg(feature = "mysql")] + (Store::MySQL(a), Store::MySQL(b)) => Arc::ptr_eq(a, b), + #[cfg(feature = "rocks")] + (Store::RocksDb(a), Store::RocksDb(b)) => Arc::ptr_eq(a, b), + (Store::Ephemeral(a), Store::Ephemeral(b)) => Arc::ptr_eq(a, b), + #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] + (Store::SQLReadReplica(a), Store::SQLReadReplica(b)) => Arc::ptr_eq(a, b), + (Store::None, Store::None) => true, + _ => false, + } + } + #[inline(always)] pub fn is_sql(&self) -> bool { match self {