diff --git a/crates/nuncio-cal/src/test_tracing.rs b/crates/nuncio-cal/src/test_tracing.rs index 178898c..cdb5e73 100644 --- a/crates/nuncio-cal/src/test_tracing.rs +++ b/crates/nuncio-cal/src/test_tracing.rs @@ -1,15 +1,21 @@ //! Shared in-test `tracing` capture helper. //! -//! Installs a thread-local subscriber that records every span (with its -//! fields) and every event (with its level, fields, and message) into an -//! in-memory buffer, so tests can assert on what the code logged -- including -//! the negative assertion that a secret never appears in any recorded field. -//! Deterministic and fully offline: no global subscriber, no I/O. +//! Records every span (with its fields) and every event (with its level, +//! fields, and message) into an in-memory buffer, so tests can assert on what +//! the code logged -- including the negative assertion that a secret never +//! appears in any recorded field. Fully offline: no I/O. +//! +//! The subscriber is installed **once, globally**, and events are routed to a +//! per-thread recorder. That split is deliberate and load-bearing: a +//! thread-local subscriber cannot work here, because `tracing` caches each +//! callsite's `Interest` process-wide on first use. See +//! [`install_global_capture`]. #![cfg(test)] +use std::cell::RefCell; use std::collections::HashMap; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, OnceLock}; use tracing::field::{Field, Visit}; use tracing::span::Attributes; @@ -89,46 +95,107 @@ impl Visit for FieldVisitor { } } -struct CaptureLayer(Arc); +thread_local! { + /// The recorder collecting telemetry for the test running on THIS thread, + /// or `None` on any thread not currently inside [`with_recorder`]. + /// + /// Routing per thread is what keeps the single global subscriber from + /// letting concurrent tests read each other's telemetry. + static ACTIVE_RECORDER: RefCell>> = const { RefCell::new(None) }; +} + +/// Run `f` against the recorder installed for this thread, if any. +fn with_active(f: impl FnOnce(&Recorder) -> R) -> Option { + ACTIVE_RECORDER.with(|slot| slot.borrow().as_ref().map(|rec| f(rec))) +} + +struct CaptureLayer; impl tracing_subscriber::Layer for CaptureLayer where S: Subscriber + for<'a> LookupSpan<'a>, { fn on_new_span(&self, attrs: &Attributes<'_>, _id: &Id, _ctx: LayerContext<'_, S>) { - let mut visitor = FieldVisitor(HashMap::new()); - attrs.record(&mut visitor); - self.0 - .spans - .lock() - .unwrap_or_else(|e| e.into_inner()) - .push(RecordedSpan { - name: attrs.metadata().name().to_string(), - fields: visitor.0, - }); + with_active(|rec| { + let mut visitor = FieldVisitor(HashMap::new()); + attrs.record(&mut visitor); + rec.spans + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(RecordedSpan { + name: attrs.metadata().name().to_string(), + fields: visitor.0, + }); + }); } fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) { - let mut visitor = FieldVisitor(HashMap::new()); - event.record(&mut visitor); - self.0 - .events - .lock() - .unwrap_or_else(|e| e.into_inner()) - .push(RecordedEvent { - level: *event.metadata().level(), - fields: visitor.0, - }); + with_active(|rec| { + let mut visitor = FieldVisitor(HashMap::new()); + event.record(&mut visitor); + rec.events + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(RecordedEvent { + level: *event.metadata().level(), + fields: visitor.0, + }); + }); } } -/// Runs `f` with a thread-local capturing subscriber installed and returns the -/// recorder alongside `f`'s result. Because the subscriber is thread-local, -/// any async work driven inside `f` must run on the calling thread (e.g. a -/// current-thread runtime's `block_on`) for its telemetry to be captured. +/// Install the capturing subscriber as the process-wide default, exactly once. +/// +/// It MUST be global rather than thread-local, and this is the whole reason the +/// helper is shaped this way. `tracing` resolves each callsite's `Interest` +/// **once for the process** and caches it; a callsite first executed on a thread +/// with no subscriber caches `Interest::never()` and is then skipped forever -- +/// before any later thread-local subscriber is ever consulted. Under a parallel +/// test harness, whichever thread reaches a callsite first therefore decides +/// whether any other test can ever capture it, which made every log-capture +/// assertion in this crate a coin flip. +/// +/// A single global subscriber removes the race: every callsite registers against +/// a real subscriber, so interest is always enabled, and per-thread routing +/// (`ACTIVE_RECORDER`) decides what is actually recorded. +fn install_global_capture() { + static INSTALLED: OnceLock<()> = OnceLock::new(); + INSTALLED.get_or_init(|| { + let subscriber = tracing_subscriber::registry().with(CaptureLayer); + // A global default may already be set by another component of the same + // test binary; capturing still works through the thread-local routing + // above, so a lost race here is not fatal. + let _ = tracing::subscriber::set_global_default(subscriber); + // Callsites executed before this point resolved their interest against + // the no-op dispatcher and cached "never". Recompute so they can be + // captured from here on. Done once, at install -- not per test. + tracing::callsite::rebuild_interest_cache(); + }); +} + +/// Resets this thread's recorder slot on drop, so a panicking test cannot leak +/// its recorder into whatever the harness runs next on the same thread. +struct RecorderGuard; + +impl Drop for RecorderGuard { + fn drop(&mut self) { + ACTIVE_RECORDER.with(|slot| *slot.borrow_mut() = None); + } +} + +/// Runs `f` with telemetry from THIS thread captured, returning the recorder +/// alongside `f`'s result. +/// +/// Only this thread's telemetry is recorded, so any async work driven inside +/// `f` must run on the calling thread (e.g. a current-thread runtime's +/// `block_on`) to be captured -- unchanged from the previous thread-local +/// implementation. What changed is that the subscriber itself is now global; +/// see [`install_global_capture`] for why that is load-bearing. pub(crate) fn with_recorder(f: impl FnOnce() -> T) -> (Arc, T) { + install_global_capture(); let recorder = Arc::new(Recorder::default()); - let subscriber = tracing_subscriber::registry().with(CaptureLayer(recorder.clone())); - let value = tracing::subscriber::with_default(subscriber, f); + ACTIVE_RECORDER.with(|slot| *slot.borrow_mut() = Some(recorder.clone())); + let _guard = RecorderGuard; + let value = f(); (recorder, value) } diff --git a/crates/nuncio-contacts/src/test_tracing.rs b/crates/nuncio-contacts/src/test_tracing.rs index 7e7c4bf..745b352 100644 --- a/crates/nuncio-contacts/src/test_tracing.rs +++ b/crates/nuncio-contacts/src/test_tracing.rs @@ -1,15 +1,21 @@ //! Shared in-test `tracing` capture helper. //! -//! Installs a thread-local subscriber that records every span (with its -//! fields) and every event (with its level, fields, and message) into an -//! in-memory buffer, so tests can assert on what the code logged -- including -//! the negative assertion that a secret never appears in any recorded field. -//! Deterministic and fully offline: no global subscriber, no I/O. +//! Records every span (with its fields) and every event (with its level, +//! fields, and message) into an in-memory buffer, so tests can assert on what +//! the code logged -- including the negative assertion that a secret never +//! appears in any recorded field. Fully offline: no I/O. +//! +//! The subscriber is installed **once, globally**, and events are routed to a +//! per-thread recorder. That split is deliberate and load-bearing: a +//! thread-local subscriber cannot work here, because `tracing` caches each +//! callsite's `Interest` process-wide on first use. See +//! [`install_global_capture`]. #![cfg(test)] +use std::cell::RefCell; use std::collections::HashMap; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, OnceLock}; use tracing::field::{Field, Visit}; use tracing::span::Attributes; @@ -90,46 +96,107 @@ impl Visit for FieldVisitor { } } -struct CaptureLayer(Arc); +thread_local! { + /// The recorder collecting telemetry for the test running on THIS thread, + /// or `None` on any thread not currently inside [`with_recorder`]. + /// + /// Routing per thread is what keeps the single global subscriber from + /// letting concurrent tests read each other's telemetry. + static ACTIVE_RECORDER: RefCell>> = const { RefCell::new(None) }; +} + +/// Run `f` against the recorder installed for this thread, if any. +fn with_active(f: impl FnOnce(&Recorder) -> R) -> Option { + ACTIVE_RECORDER.with(|slot| slot.borrow().as_ref().map(|rec| f(rec))) +} + +struct CaptureLayer; impl tracing_subscriber::Layer for CaptureLayer where S: Subscriber + for<'a> LookupSpan<'a>, { fn on_new_span(&self, attrs: &Attributes<'_>, _id: &Id, _ctx: LayerContext<'_, S>) { - let mut visitor = FieldVisitor(HashMap::new()); - attrs.record(&mut visitor); - self.0 - .spans - .lock() - .unwrap_or_else(|e| e.into_inner()) - .push(RecordedSpan { - name: attrs.metadata().name().to_string(), - fields: visitor.0, - }); + with_active(|rec| { + let mut visitor = FieldVisitor(HashMap::new()); + attrs.record(&mut visitor); + rec.spans + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(RecordedSpan { + name: attrs.metadata().name().to_string(), + fields: visitor.0, + }); + }); } fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) { - let mut visitor = FieldVisitor(HashMap::new()); - event.record(&mut visitor); - self.0 - .events - .lock() - .unwrap_or_else(|e| e.into_inner()) - .push(RecordedEvent { - level: *event.metadata().level(), - fields: visitor.0, - }); + with_active(|rec| { + let mut visitor = FieldVisitor(HashMap::new()); + event.record(&mut visitor); + rec.events + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(RecordedEvent { + level: *event.metadata().level(), + fields: visitor.0, + }); + }); } } -/// Runs `f` with a thread-local capturing subscriber installed and returns the -/// recorder alongside `f`'s result. Because the subscriber is thread-local, -/// any async work driven inside `f` must run on the calling thread (e.g. a -/// current-thread runtime's `block_on`) for its telemetry to be captured. +/// Install the capturing subscriber as the process-wide default, exactly once. +/// +/// It MUST be global rather than thread-local, and this is the whole reason the +/// helper is shaped this way. `tracing` resolves each callsite's `Interest` +/// **once for the process** and caches it; a callsite first executed on a thread +/// with no subscriber caches `Interest::never()` and is then skipped forever -- +/// before any later thread-local subscriber is ever consulted. Under a parallel +/// test harness, whichever thread reaches a callsite first therefore decides +/// whether any other test can ever capture it, which made every log-capture +/// assertion in this crate a coin flip. +/// +/// A single global subscriber removes the race: every callsite registers against +/// a real subscriber, so interest is always enabled, and per-thread routing +/// (`ACTIVE_RECORDER`) decides what is actually recorded. +fn install_global_capture() { + static INSTALLED: OnceLock<()> = OnceLock::new(); + INSTALLED.get_or_init(|| { + let subscriber = tracing_subscriber::registry().with(CaptureLayer); + // A global default may already be set by another component of the same + // test binary; capturing still works through the thread-local routing + // above, so a lost race here is not fatal. + let _ = tracing::subscriber::set_global_default(subscriber); + // Callsites executed before this point resolved their interest against + // the no-op dispatcher and cached "never". Recompute so they can be + // captured from here on. Done once, at install -- not per test. + tracing::callsite::rebuild_interest_cache(); + }); +} + +/// Resets this thread's recorder slot on drop, so a panicking test cannot leak +/// its recorder into whatever the harness runs next on the same thread. +struct RecorderGuard; + +impl Drop for RecorderGuard { + fn drop(&mut self) { + ACTIVE_RECORDER.with(|slot| *slot.borrow_mut() = None); + } +} + +/// Runs `f` with telemetry from THIS thread captured, returning the recorder +/// alongside `f`'s result. +/// +/// Only this thread's telemetry is recorded, so any async work driven inside +/// `f` must run on the calling thread (e.g. a current-thread runtime's +/// `block_on`) to be captured -- unchanged from the previous thread-local +/// implementation. What changed is that the subscriber itself is now global; +/// see [`install_global_capture`] for why that is load-bearing. pub(crate) fn with_recorder(f: impl FnOnce() -> T) -> (Arc, T) { + install_global_capture(); let recorder = Arc::new(Recorder::default()); - let subscriber = tracing_subscriber::registry().with(CaptureLayer(recorder.clone())); - let value = tracing::subscriber::with_default(subscriber, f); + ACTIVE_RECORDER.with(|slot| *slot.borrow_mut() = Some(recorder.clone())); + let _guard = RecorderGuard; + let value = f(); (recorder, value) } diff --git a/crates/nuncio-filter/src/engine.rs b/crates/nuncio-filter/src/engine.rs index d5d1230..72605aa 100644 --- a/crates/nuncio-filter/src/engine.rs +++ b/crates/nuncio-filter/src/engine.rs @@ -434,51 +434,6 @@ impl FilterEngine { mod tests { use super::*; use nuncio_core::model::Email; - use std::sync::Mutex; - use tracing::field::{Field, Visit}; - use tracing::span; - - /// Minimal `tracing::Subscriber` that records a formatted line per event - /// so tests can assert on emitted level + fields without pulling in - /// `tracing-subscriber`'s registry machinery. - struct CapturingSubscriber { - events: Arc>>, - } - - struct LineVisitor<'a>(&'a mut String); - - impl Visit for LineVisitor<'_> { - fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { - self.0.push_str(&format!(" {}={:?}", field.name(), value)); - } - } - - impl tracing::Subscriber for CapturingSubscriber { - fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { - true - } - - fn new_span(&self, _span: &span::Attributes<'_>) -> span::Id { - span::Id::from_u64(1) - } - - fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {} - - fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {} - - fn event(&self, event: &tracing::Event<'_>) { - let mut line = format!("{}", event.metadata().level()); - let mut visitor = LineVisitor(&mut line); - event.record(&mut visitor); - if let Ok(mut events) = self.events.lock() { - events.push(line); - } - } - - fn enter(&self, _span: &span::Id) {} - - fn exit(&self, _span: &span::Id) {} - } fn test_email(account_id: &str, subject: &str, folder_id: &str) -> Email { Email { @@ -697,11 +652,7 @@ mod tests { // real blocking-pool thread while the timer races it on the async // task, so the timer reliably wins for any evaluation slower than a // few microseconds. - let events: Arc>> = Arc::new(Mutex::new(Vec::new())); - let subscriber = CapturingSubscriber { - events: events.clone(), - }; - let _guard = tracing::subscriber::set_default(subscriber); + let logs = crate::test_tracing::capture_logs(); let rule = FilterRule { id: "slow-rule".to_string(), @@ -734,7 +685,7 @@ mod tests { not fabricated as a match" ); - let captured = events.lock().unwrap(); + let captured = logs.lines(); assert!( captured.iter().any(|line| line.contains("WARN") && line.contains("timed out") diff --git a/crates/nuncio-filter/src/lib.rs b/crates/nuncio-filter/src/lib.rs index 7056052..d96fdf3 100644 --- a/crates/nuncio-filter/src/lib.rs +++ b/crates/nuncio-filter/src/lib.rs @@ -8,6 +8,8 @@ pub mod codegen; pub mod engine; pub mod outbox; pub mod parser; +#[cfg(test)] +mod test_tracing; pub mod validator; pub mod webhook; diff --git a/crates/nuncio-filter/src/test_tracing.rs b/crates/nuncio-filter/src/test_tracing.rs new file mode 100644 index 0000000..bfe547e --- /dev/null +++ b/crates/nuncio-filter/src/test_tracing.rs @@ -0,0 +1,146 @@ +//! Shared in-test `tracing` capture helper. +//! +//! Records one formatted line per event (level plus `Debug`-rendered fields, +//! including the message) into an in-memory buffer, so tests can assert that +//! a level tag and specific fields appear on the SAME logged event. This +//! crate has no dependency on `tracing-subscriber`, so the capture is a +//! minimal, hand-rolled `tracing::Subscriber` rather than a `Layer` -- that +//! keeps this helper free of a new dependency. Fully offline: no I/O. +//! +//! The subscriber is installed **once, globally**, and events are routed to a +//! per-thread line buffer. That split is deliberate and load-bearing: a +//! thread-local subscriber cannot work here, because `tracing` caches each +//! callsite's `Interest` process-wide on first use. See +//! [`install_global_capture`]. + +#![cfg(test)] + +use std::cell::RefCell; +use std::sync::{Arc, Mutex, OnceLock}; + +use tracing::field::{Field, Visit}; +use tracing::span; + +thread_local! { + /// The line buffer collecting telemetry for the test running on THIS + /// thread, or `None` on any thread not currently inside [`capture_logs`]. + /// + /// Routing per thread is what keeps the single global subscriber from + /// letting concurrent tests read each other's telemetry. + static ACTIVE_LINES: RefCell>>>> = const { RefCell::new(None) }; +} + +struct LineVisitor<'a>(&'a mut String); + +impl Visit for LineVisitor<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0.push_str(&format!(" {}={:?}", field.name(), value)); + } +} + +/// Minimal `tracing::Subscriber` that records a formatted line per event +/// (level plus fields) without pulling in `tracing-subscriber`'s registry +/// machinery. Spans are not tracked (`new_span` returns a constant id) since +/// no test in this crate asserts on span structure, only on event lines. +struct CaptureSubscriber; + +impl tracing::Subscriber for CaptureSubscriber { + fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { + true + } + + fn new_span(&self, _span: &span::Attributes<'_>) -> span::Id { + span::Id::from_u64(1) + } + + fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {} + + fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {} + + fn event(&self, event: &tracing::Event<'_>) { + let mut line = format!("{}", event.metadata().level()); + let mut visitor = LineVisitor(&mut line); + event.record(&mut visitor); + ACTIVE_LINES.with(|slot| { + if let Some(lines) = slot.borrow().as_ref() { + lines.lock().unwrap_or_else(|e| e.into_inner()).push(line); + } + }); + } + + fn enter(&self, _span: &span::Id) {} + + fn exit(&self, _span: &span::Id) {} +} + +/// Install the capturing subscriber as the process-wide default, exactly once. +/// +/// It MUST be global rather than thread-local, and this is the whole reason the +/// helper is shaped this way. `tracing` resolves each callsite's `Interest` +/// **once for the process** and caches it; a callsite first executed on a thread +/// with no subscriber caches `Interest::never()` and is then skipped forever -- +/// before any later thread-local subscriber is ever consulted. Under a parallel +/// test harness, whichever thread reaches a callsite first therefore decides +/// whether any other test can ever capture it, which made every log-capture +/// assertion in this crate a coin flip. +/// +/// A single global subscriber removes the race: every callsite registers against +/// a real subscriber, so interest is always enabled, and per-thread routing +/// (`ACTIVE_LINES`) decides what is actually recorded. +fn install_global_capture() { + static INSTALLED: OnceLock<()> = OnceLock::new(); + INSTALLED.get_or_init(|| { + // A global default may already be set by another component of the same + // test binary; capturing still works through the thread-local routing + // above, so a lost race here is not fatal. + let _ = tracing::subscriber::set_global_default(CaptureSubscriber); + // Callsites executed before this point resolved their interest against + // the no-op dispatcher and cached "never". Recompute so they can be + // captured from here on. Done once, at install -- not per test. + tracing::callsite::rebuild_interest_cache(); + }); +} + +/// Resets this thread's line buffer slot on drop, so a panicking test cannot +/// leak its buffer into whatever the harness runs next on the same thread. +struct LinesGuard; + +impl Drop for LinesGuard { + fn drop(&mut self) { + ACTIVE_LINES.with(|slot| *slot.borrow_mut() = None); + } +} + +/// A capture in progress on the current thread. Held across the code under +/// test -- including any `.await` points, as long as the test stays on a +/// single OS thread (e.g. the default current-thread `#[tokio::test]` +/// runtime) -- then queried for the lines captured so far. +pub(crate) struct CapturedLogs { + lines: Arc>>, + _guard: LinesGuard, +} + +impl CapturedLogs { + /// A snapshot of every formatted event line captured on this thread so + /// far: level plus `Debug`-rendered fields (including `message`). + pub fn lines(&self) -> Vec { + self.lines.lock().unwrap_or_else(|e| e.into_inner()).clone() + } +} + +/// Start capturing this thread's `tracing` events as formatted lines. +/// +/// Only this thread's events are captured, so any async work driven while the +/// returned [`CapturedLogs`] is alive must run on the calling thread (e.g. a +/// current-thread runtime's `block_on`) to be captured -- the subscriber +/// itself is global, but the destination buffer is per-thread; see +/// [`install_global_capture`] for why that split is load-bearing. +pub(crate) fn capture_logs() -> CapturedLogs { + install_global_capture(); + let lines = Arc::new(Mutex::new(Vec::new())); + ACTIVE_LINES.with(|slot| *slot.borrow_mut() = Some(lines.clone())); + CapturedLogs { + lines, + _guard: LinesGuard, + } +} diff --git a/crates/nuncio-filter/src/webhook.rs b/crates/nuncio-filter/src/webhook.rs index 123f82e..0a18320 100644 --- a/crates/nuncio-filter/src/webhook.rs +++ b/crates/nuncio-filter/src/webhook.rs @@ -240,54 +240,9 @@ impl WebhookDispatcher { #[cfg(test)] mod tests { use super::*; - use std::sync::Mutex; - use tracing::field::{Field, Visit}; - use tracing::span; use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; - /// Minimal `tracing::Subscriber` that records a formatted line per event - /// so tests can assert on emitted level + fields without pulling in - /// `tracing-subscriber`'s registry machinery. - struct CapturingSubscriber { - events: std::sync::Arc>>, - } - - struct LineVisitor<'a>(&'a mut String); - - impl Visit for LineVisitor<'_> { - fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { - self.0.push_str(&format!(" {}={:?}", field.name(), value)); - } - } - - impl tracing::Subscriber for CapturingSubscriber { - fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { - true - } - - fn new_span(&self, _span: &span::Attributes<'_>) -> span::Id { - span::Id::from_u64(1) - } - - fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {} - - fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {} - - fn event(&self, event: &tracing::Event<'_>) { - let mut line = format!("{}", event.metadata().level()); - let mut visitor = LineVisitor(&mut line); - event.record(&mut visitor); - if let Ok(mut events) = self.events.lock() { - events.push(line); - } - } - - fn enter(&self, _span: &span::Id) {} - - fn exit(&self, _span: &span::Id) {} - } - #[tokio::test] async fn test_pinned_client_connects_to_checked_ip_not_dns() { // `.invalid` is reserved by RFC 2606 to never resolve: any real DNS @@ -393,12 +348,7 @@ mod tests { #[tokio::test] async fn test_blocked_webhook_logs_warn_with_host_and_reason() { - let events: std::sync::Arc>> = - std::sync::Arc::new(Mutex::new(Vec::new())); - let subscriber = CapturingSubscriber { - events: events.clone(), - }; - let _guard = tracing::subscriber::set_default(subscriber); + let logs = crate::test_tracing::capture_logs(); let dispatcher = WebhookDispatcher::new("secret_key_123"); let result = dispatcher @@ -413,7 +363,7 @@ mod tests { assert!(matches!(result, Err(WebhookError::SecurityViolation(_)))); - let captured = events.lock().unwrap(); + let captured = logs.lines(); assert!( captured.iter().any(|line| line.contains("WARN") && line.contains("blocked") @@ -435,12 +385,7 @@ mod tests { .mount(&mock_server) .await; - let events: std::sync::Arc>> = - std::sync::Arc::new(Mutex::new(Vec::new())); - let subscriber = CapturingSubscriber { - events: events.clone(), - }; - let _guard = tracing::subscriber::set_default(subscriber); + let logs = crate::test_tracing::capture_logs(); let dispatcher = WebhookDispatcher::new("secret_key_123"); let opts = ValidationOptions { @@ -455,7 +400,7 @@ mod tests { .expect("dispatch to the mock server must succeed"); assert_eq!(status, 200); - let captured = events.lock().unwrap(); + let captured = logs.lines(); assert!( captured .iter() diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index c22ba55..50b6e37 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -747,239 +747,22 @@ impl DatabaseEngine { .await .map_err(DatabaseError::Query)?; - self.ensure_accounts_smtp_columns().await?; - self.ensure_accounts_tls_mode_columns().await?; - self.ensure_accounts_dav_columns().await?; - self.ensure_messages_identity_columns().await?; + // The statements above create the schema in full, so there is nothing to + // upgrade: every column this engine reads is declared in its own + // `CREATE TABLE IF NOT EXISTS`. Databases are created, never migrated. self.backfill_message_fts().await?; Ok(()) } - /// Additive, backfill-safe migration that adds the `remote_id` / - /// `uid_validity` message-identity columns to a pre-existing `messages` - /// table that predated the opaque surrogate id, and installs the - /// `UNIQUE(account_id, folder_id, uid_validity, remote_id)` index that is - /// the real correctness guard against a shared protocol id (e.g. an IMAP - /// UID reused across folders/accounts) overwriting an unrelated message. - /// - /// A fresh database already gets both columns from `CREATE TABLE IF NOT - /// EXISTS messages` above, so on a fresh database the `PRAGMA table_info` - /// check finds them present and the backfill matches no rows -- only the - /// index is created. For a pre-existing database file the columns are added - /// with an empty-string default and every legacy row is backfilled to - /// `remote_id = id` (the old primary key, which was globally unique) so the - /// UNIQUE index can be built without collisions. SQLite has no `ADD COLUMN - /// IF NOT EXISTS`, so column presence is checked explicitly first, making - /// this safe to run on every daemon startup. - /// - /// The whole step runs in a single transaction, and the backfill is - /// unconditional (idempotent `WHERE remote_id = ''`) rather than gated on - /// having just added the column. Both properties make a partial run - /// self-healing: a crash mid-migration rolls back atomically, and even a - /// committed intermediate state (column present but a legacy row still - /// carrying the empty default) is completed by the next startup's backfill - /// before the index is built -- so the daemon can never wedge on an index - /// collision it could have avoided. - async fn ensure_messages_identity_columns(&self) -> Result<(), DatabaseError> { - let existing_columns: Vec = sqlx::query("PRAGMA table_info(messages)") - .fetch_all(&self.pool) - .await - .map_err(DatabaseError::Query)? - .iter() - .map(|row| row.get::("name")) - .collect(); - - let mut tx = self.pool.begin().await.map_err(DatabaseError::Query)?; - - if !existing_columns.iter().any(|c| c == "remote_id") { - sqlx::query("ALTER TABLE messages ADD COLUMN remote_id TEXT NOT NULL DEFAULT ''") - .execute(&mut *tx) - .await - .map_err(DatabaseError::Query)?; - tracing::info!( - table = "messages", - column = "remote_id", - "applying schema migration" - ); - } - if !existing_columns.iter().any(|c| c == "uid_validity") { - sqlx::query("ALTER TABLE messages ADD COLUMN uid_validity TEXT NOT NULL DEFAULT ''") - .execute(&mut *tx) - .await - .map_err(DatabaseError::Query)?; - tracing::info!( - table = "messages", - column = "uid_validity", - "applying schema migration" - ); - } - - // Seed any legacy row still carrying the empty-string default with a - // unique, non-empty remote id (the old primary key) so the UNIQUE - // identity index is satisfiable. Unconditional so a partial prior run - // is always completed before the index is (re)built; a no-op on a fresh - // or already-migrated database. - sqlx::query("UPDATE messages SET remote_id = id WHERE remote_id = ''") - .execute(&mut *tx) - .await - .map_err(DatabaseError::Query)?; - - sqlx::query( - "CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_identity \ - ON messages (account_id, folder_id, uid_validity, remote_id)", - ) - .execute(&mut *tx) - .await - .map_err(DatabaseError::Query)?; - - tx.commit().await.map_err(DatabaseError::Query)?; - - Ok(()) - } - - /// Additive, backfill-safe migration that adds the `imap_tls_mode` / - /// `smtp_tls_mode` columns to a pre-existing `accounts` table that - /// predates persisted per-protocol TLS transport modes. - /// - /// A fresh database already gets these columns from `CREATE TABLE IF NOT - /// EXISTS accounts` above, so on a fresh database this is a no-op. For a - /// pre-existing database file the columns are added with a - /// `'implicit_tls'` default, so every existing row keeps loading - /// successfully and retains the historical behavior (implicit TLS) it was - /// created under. SQLite has no `ADD COLUMN IF NOT EXISTS`, so column - /// presence is checked explicitly via `PRAGMA table_info` first, making - /// this safe to run on every daemon startup. - async fn ensure_accounts_tls_mode_columns(&self) -> Result<(), DatabaseError> { - let existing_columns: Vec = sqlx::query("PRAGMA table_info(accounts)") - .fetch_all(&self.pool) - .await - .map_err(DatabaseError::Query)? - .iter() - .map(|row| row.get::("name")) - .collect(); - - if !existing_columns.iter().any(|c| c == "imap_tls_mode") { - sqlx::query( - "ALTER TABLE accounts ADD COLUMN imap_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls'", - ) - .execute(&self.pool) - .await - .map_err(DatabaseError::Query)?; - tracing::info!( - table = "accounts", - column = "imap_tls_mode", - "applying schema migration" - ); - } - if !existing_columns.iter().any(|c| c == "smtp_tls_mode") { - sqlx::query( - "ALTER TABLE accounts ADD COLUMN smtp_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls'", - ) - .execute(&self.pool) - .await - .map_err(DatabaseError::Query)?; - tracing::info!( - table = "accounts", - column = "smtp_tls_mode", - "applying schema migration" - ); - } - - Ok(()) - } - - /// Additive, backfill-safe migration that adds the - /// `smtp_host` / `smtp_port` columns to a pre-existing `accounts` table - /// that predates the SMTP account-endpoint feature. - /// - /// A fresh database already gets these columns from `CREATE TABLE IF NOT - /// EXISTS accounts` above, so on a fresh database this is a no-op (the - /// `PRAGMA table_info` check below finds both columns already present). - /// For a pre-existing database file, `CREATE TABLE IF NOT EXISTS` is - /// itself a no-op (the table already exists), so this is what actually - /// adds the new columns -- as `NULL`-able columns, so every existing row - /// keeps loading successfully (see [`Self::list_accounts`], which falls - /// back to `server_host`/`server_port` for any row where `smtp_host` / - /// `smtp_port` is still `NULL`). SQLite has no `ADD COLUMN IF NOT - /// EXISTS`, so column presence is checked explicitly via `PRAGMA - /// table_info` first, making this safe to run on every daemon startup. - async fn ensure_accounts_smtp_columns(&self) -> Result<(), DatabaseError> { - let existing_columns: Vec = sqlx::query("PRAGMA table_info(accounts)") - .fetch_all(&self.pool) - .await - .map_err(DatabaseError::Query)? - .iter() - .map(|row| row.get::("name")) - .collect(); - - if !existing_columns.iter().any(|c| c == "smtp_host") { - sqlx::query("ALTER TABLE accounts ADD COLUMN smtp_host TEXT") - .execute(&self.pool) - .await - .map_err(DatabaseError::Query)?; - tracing::info!( - table = "accounts", - column = "smtp_host", - "applying schema migration" - ); - } - if !existing_columns.iter().any(|c| c == "smtp_port") { - sqlx::query("ALTER TABLE accounts ADD COLUMN smtp_port INTEGER") - .execute(&self.pool) - .await - .map_err(DatabaseError::Query)?; - tracing::info!( - table = "accounts", - column = "smtp_port", - "applying schema migration" - ); - } - - Ok(()) - } - - /// Additive, backfill-safe migration that adds the `collection_url` - /// column to a pre-existing `accounts` table that predates DAV-protocol - /// (CalDAV/CardDAV) accounts. - /// - /// A fresh database already gets this column from `CREATE TABLE IF NOT - /// EXISTS accounts` above, so on a fresh database this is a no-op. For a - /// pre-existing database file the column is added as a `NULL`-able column, - /// so every existing (mail) row keeps loading successfully with an empty - /// collection URL (see [`Self::list_accounts`], which maps a `NULL` - /// `collection_url` to an empty string). SQLite has no `ADD COLUMN IF NOT - /// EXISTS`, so column presence is checked explicitly via `PRAGMA - /// table_info` first, making this safe to run on every daemon startup. - async fn ensure_accounts_dav_columns(&self) -> Result<(), DatabaseError> { - let existing_columns: Vec = sqlx::query("PRAGMA table_info(accounts)") - .fetch_all(&self.pool) - .await - .map_err(DatabaseError::Query)? - .iter() - .map(|row| row.get::("name")) - .collect(); - - if !existing_columns.iter().any(|c| c == "collection_url") { - sqlx::query("ALTER TABLE accounts ADD COLUMN collection_url TEXT") - .execute(&self.pool) - .await - .map_err(DatabaseError::Query)?; - tracing::info!( - table = "accounts", - column = "collection_url", - "applying schema migration" - ); - } - - Ok(()) - } - /// Backfill `messages_fts` for any `messages` row that does not yet have a matching FTS - /// entry -- e.g. rows written before the FTS5 index existed, or written directly by a - /// process that bypassed `save_email`'s explicit index population. Run automatically as - /// part of [`DatabaseEngine::migrate`] (idempotent: a fully-indexed database performs no - /// work). The stored `body_plain` column holds AES-256-GCM ciphertext, so each candidate + /// entry -- rows written directly by a process that bypassed `save_email`'s explicit + /// index population. That is not a legacy concern: `SqliteRecoveryEngine::salvage` + /// restores `messages` rows straight into the table, so a database that has been + /// through recovery has content the trigram index has never seen, and search would + /// silently miss it. Run automatically as part of [`DatabaseEngine::migrate`] + /// (idempotent: a fully-indexed database performs no work). The stored `body_plain` + /// column holds AES-256-GCM ciphertext, so each candidate /// row is decrypted with this engine's storage key before being written into the /// plaintext-derived trigram index -- see the confidentiality tradeoff documented above /// `messages_fts`'s `CREATE VIRTUAL TABLE` statement in [`DatabaseEngine::migrate`]. @@ -3756,230 +3539,6 @@ mod tests { ); } - /// Proves the additive `smtp_host`/`smtp_port` - /// migration is backfill-safe. Simulates a database created BEFORE this - /// feature existed (an `accounts` table with no `smtp_host`/`smtp_port` - /// columns at all, populated via a direct `INSERT` bypassing - /// `save_account`), then opens it through `DatabaseEngine::connect_file` - /// (which runs `migrate()`, including `ensure_accounts_smtp_columns`) - /// and confirms: (1) opening an old-schema database never errors, (2) - /// the pre-existing row still loads, falling back to `server_host`/ - /// `server_port` for the missing SMTP endpoint, and (3) re-running - /// `migrate()` a second time (simulating a daemon restart) is a no-op - /// that does not error or duplicate columns. - #[tokio::test] - async fn migrate_backfills_smtp_columns_for_a_pre_existing_accounts_table() { - let dir = tempfile::tempdir().expect("tempdir"); - let db_path = dir.path().join("pre_smtp_migration.db"); - let secrets = crate::vault::SecretManager::mock(); - - // Step 1: create an OLD-schema `accounts` table (no smtp_host/smtp_port - // columns) directly, bypassing `DatabaseEngine::migrate` entirely, and - // insert one pre-existing row -- exactly what a database created - // before the `smtp_host`/`smtp_port` columns existed would look like on disk. - { - let url = format!("sqlite://{}", db_path.to_string_lossy()); - let options = sqlx::sqlite::SqliteConnectOptions::from_str(&url) - .expect("valid sqlite url") - .create_if_missing(true); - let pool = sqlx::SqlitePool::connect_with(options) - .await - .expect("connect to fresh old-schema db"); - - sqlx::query( - r#" - CREATE TABLE accounts ( - id TEXT PRIMARY KEY NOT NULL, - name TEXT NOT NULL, - email_address TEXT NOT NULL, - protocol TEXT NOT NULL, - server_host TEXT NOT NULL, - server_port INTEGER NOT NULL, - use_tls INTEGER NOT NULL, - keyring_secret_key TEXT NOT NULL, - sync_interval_secs INTEGER NOT NULL - ) - "#, - ) - .execute(&pool) - .await - .expect("create old-schema accounts table"); - - sqlx::query( - r#" - INSERT INTO accounts - (id, name, email_address, protocol, server_host, server_port, use_tls, keyring_secret_key, sync_interval_secs) - VALUES ('acct-pre-migration', 'Pre-Migration Account', 'pre@nuncio.mx', '"imap-smtp"', 'imap.nuncio.mx', 993, 1, 'nuncio/acct-pre-migration', 60) - "#, - ) - .execute(&pool) - .await - .expect("insert pre-existing row"); - - pool.close().await; - } - - // Step 2: open through the real production path -- this is what - // actually runs `migrate()` / `ensure_accounts_smtp_columns`. - let engine = DatabaseEngine::connect_file(&db_path, &secrets) - .await - .expect("opening an old-schema database must never error"); - - let accounts = engine - .list_accounts() - .await - .expect("list accounts succeeds on a migrated old-schema table"); - assert_eq!(accounts.len(), 1); - let acct = &accounts[0]; - assert_eq!(acct.id, "acct-pre-migration"); - // Falls back to the IMAP/JMAP endpoint since smtp_host/smtp_port - // were NULL for this pre-existing row. - let acct_t = acct.imap_smtp().expect("imap-smtp transport"); - assert_eq!(acct_t.smtp_host, "imap.nuncio.mx"); - assert_eq!(acct_t.smtp_port, 993); - - // Step 3: re-running migrate() (e.g. a second daemon startup against - // the same file) must be a no-op, not an error. - engine - .migrate() - .await - .expect("re-running migrate on an already-migrated table must not error"); - - // A newly saved account (going through the real `save_account` path) - // now gets its own genuine smtp_host/smtp_port persisted, proving - // the migrated columns are fully writable/readable going forward. - let new_acct = nuncio_core::AccountConfig { - id: "acct-post-migration".to_string(), - name: "Post Migration Account".to_string(), - email_address: "post@nuncio.mx".to_string(), - keyring_secret_key: "nuncio/acct-post-migration".to_string(), - sync_interval_secs: 60, - transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { - imap_host: "imap.nuncio.mx".to_string(), - imap_port: 993, - imap_tls_mode: nuncio_core::TlsMode::ImplicitTls, - smtp_host: "smtp.nuncio.mx".to_string(), - smtp_port: 587, - smtp_tls_mode: nuncio_core::TlsMode::StartTls, - }), - }; - engine - .save_account(&new_acct) - .await - .expect("save account succeeds after migration"); - let accounts = engine.list_accounts().await.expect("list accounts"); - let post = accounts - .iter() - .find(|a| a.id == "acct-post-migration") - .expect("newly saved account present"); - let post_t = post.imap_smtp().expect("imap-smtp transport"); - assert_eq!(post_t.smtp_host, "smtp.nuncio.mx"); - assert_eq!(post_t.smtp_port, 587); - - engine.close().await; - } - - /// Applying an additive column migration against a pre-existing (old-schema) - /// database must be visible in telemetry -- an `INFO` log naming the table and - /// column -- rather than only observable indirectly via the resulting schema. - #[tokio::test] - async fn migrate_logs_when_an_additive_column_migration_actually_runs() { - use std::sync::{Arc, Mutex}; - use tracing_subscriber::fmt::MakeWriter; - - #[derive(Clone, Default)] - struct CapturedLogs(Arc>>); - - impl std::io::Write for CapturedLogs { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - self.0 - .lock() - .expect("log buffer lock") - .extend_from_slice(buf); - Ok(buf.len()) - } - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } - } - - impl<'a> MakeWriter<'a> for CapturedLogs { - type Writer = CapturedLogs; - fn make_writer(&'a self) -> Self::Writer { - self.clone() - } - } - - let dir = tempfile::tempdir().expect("tempdir"); - let db_path = dir.path().join("pre_dav_migration.db"); - let secrets = crate::vault::SecretManager::mock(); - - // An old-schema `accounts` table predating `collection_url`. - { - let url = format!("sqlite://{}", db_path.to_string_lossy()); - let options = sqlx::sqlite::SqliteConnectOptions::from_str(&url) - .expect("valid sqlite url") - .create_if_missing(true); - let pool = sqlx::SqlitePool::connect_with(options) - .await - .expect("connect to fresh old-schema db"); - - sqlx::query( - r#" - CREATE TABLE accounts ( - id TEXT PRIMARY KEY NOT NULL, - name TEXT NOT NULL, - email_address TEXT NOT NULL, - protocol TEXT NOT NULL, - server_host TEXT NOT NULL, - server_port INTEGER NOT NULL, - use_tls INTEGER NOT NULL, - keyring_secret_key TEXT NOT NULL, - sync_interval_secs INTEGER NOT NULL, - smtp_host TEXT, - smtp_port INTEGER, - imap_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls', - smtp_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls' - ) - "#, - ) - .execute(&pool) - .await - .expect("create old-schema accounts table"); - - pool.close().await; - } - - let logs = CapturedLogs::default(); - let subscriber = tracing_subscriber::fmt() - .with_writer(logs.clone()) - .with_ansi(false) - .finish(); - - // `set_default` returns a guard that stays active across `.await` - // points as long as this test stays on the single `#[tokio::test]` - // current-thread executor, which is the default runtime flavor and - // exactly what is used here. - let guard = tracing::subscriber::set_default(subscriber); - let engine = DatabaseEngine::connect_file(&db_path, &secrets) - .await - .expect("opening an old-schema database must never error"); - drop(guard); - - let captured = String::from_utf8(logs.0.lock().expect("log buffer lock").clone()) - .expect("captured log is valid utf8"); - assert!( - captured.contains("collection_url"), - "expected the migrated column to be named in the INFO log, got: {captured}" - ); - assert!( - captured.to_ascii_uppercase().contains("INFO"), - "expected an INFO-level log for the applied migration, got: {captured}" - ); - - engine.close().await; - } - #[test] fn database_error_display() { let err = DatabaseError::PoolCreation("failed".to_string()); @@ -4309,65 +3868,6 @@ mod tests { assert_eq!(fetched.folder_id, "INBOX"); } - /// A crash between `ADD COLUMN remote_id` and its backfill can commit an - /// intermediate state -- the column present but a legacy row still carrying - /// the empty-string default -- with the UNIQUE index not yet built. The - /// migration must complete that backfill on the next open and build the - /// index without a collision, rather than wedging the daemon. - #[tokio::test] - async fn messages_identity_migration_completes_from_a_partial_run() { - let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); - - // Reproduce the partial state: drop the identity index, then insert two - // rows in the same folder that both still hold the empty default. With - // the index gone this is permitted, and the two would collide on - // (account, folder, uid_validity='', remote_id='') if it were rebuilt now. - sqlx::query("DROP INDEX IF EXISTS idx_messages_identity") - .execute(engine.pool()) - .await - .unwrap(); - for id in ["legacy-a", "legacy-b"] { - sqlx::query( - "INSERT INTO messages \ - (id, account_id, folder_id, remote_id, uid_validity, subject, sender, recipient, received_at, read_flag) \ - VALUES (?, 'acct-1', 'INBOX', '', '', 'subj', 'a@nuncio.mx', 'b@nuncio.mx', 0, 0)", - ) - .bind(id) - .execute(engine.pool()) - .await - .unwrap(); - } - - // Re-running the migration must finish the backfill and build the index. - engine - .ensure_messages_identity_columns() - .await - .expect("a partial migration must complete on the next open"); - - // Both legacy rows survive with a unique, non-empty backfilled remote id. - assert_eq!( - engine.get_message("legacy-a").await.unwrap().remote_id, - "legacy-a" - ); - assert_eq!( - engine.get_message("legacy-b").await.unwrap().remote_id, - "legacy-b" - ); - - // The unique identity index is now present, and re-running is idempotent. - let index: Option<(String,)> = sqlx::query_as( - "SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_messages_identity'", - ) - .fetch_optional(engine.pool()) - .await - .unwrap(); - assert!(index.is_some(), "the unique identity index must be built"); - engine - .ensure_messages_identity_columns() - .await - .expect("re-running an already-complete migration is a no-op"); - } - fn export_test_email(id: &str, account_id: &str, folder_id: &str) -> nuncio_core::model::Email { nuncio_core::model::Email { id: id.to_string(), diff --git a/crates/nuncio-store/src/lib.rs b/crates/nuncio-store/src/lib.rs index 5783c30..ad82973 100644 --- a/crates/nuncio-store/src/lib.rs +++ b/crates/nuncio-store/src/lib.rs @@ -7,6 +7,8 @@ pub mod cipher; pub mod db; pub mod recovery; pub mod search; +#[cfg(test)] +mod test_tracing; pub mod vault; pub use cipher::{CipherError, PayloadCipher}; diff --git a/crates/nuncio-store/src/test_tracing.rs b/crates/nuncio-store/src/test_tracing.rs new file mode 100644 index 0000000..64175ca --- /dev/null +++ b/crates/nuncio-store/src/test_tracing.rs @@ -0,0 +1,143 @@ +//! Shared in-test `tracing` capture helper. +//! +//! Captures the process's formatted `tracing` output (the same text a real +//! log line would carry, level tag and all) into an in-memory buffer, so +//! tests can assert on substrings such as a table/column name or an `"INFO"`/ +//! `"WARN"` level tag. Fully offline: no I/O beyond the in-memory buffer. +//! +//! The subscriber is installed **once, globally**, and each thread's output is +//! routed to that thread's own buffer. That split is deliberate and +//! load-bearing: a thread-local subscriber cannot work here, because +//! `tracing` caches each callsite's `Interest` process-wide on first use. See +//! [`install_global_capture`]. + +#![cfg(test)] + +use std::cell::RefCell; +use std::io; +use std::sync::{Arc, Mutex, OnceLock}; + +use tracing_subscriber::fmt::MakeWriter; + +thread_local! { + /// The buffer collecting formatted log output for the test running on + /// THIS thread, or `None` on any thread not currently inside + /// [`capture_logs`]. + /// + /// Routing per thread is what keeps the single global subscriber from + /// letting concurrent tests read each other's telemetry. + static ACTIVE_BUFFER: RefCell>>>> = const { RefCell::new(None) }; +} + +/// A `Write`/`MakeWriter` that appends into whichever buffer is active for +/// the calling thread, discarding the write if no test on this thread is +/// currently capturing. +#[derive(Clone, Default)] +struct ThreadRoutedWriter; + +impl io::Write for ThreadRoutedWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + ACTIVE_BUFFER.with(|slot| { + if let Some(buffer) = slot.borrow().as_ref() { + buffer + .lock() + .unwrap_or_else(|e| e.into_inner()) + .extend_from_slice(buf); + } + }); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl<'a> MakeWriter<'a> for ThreadRoutedWriter { + type Writer = ThreadRoutedWriter; + fn make_writer(&'a self) -> Self::Writer { + ThreadRoutedWriter + } +} + +/// Install the capturing subscriber as the process-wide default, exactly once. +/// +/// It MUST be global rather than thread-local, and this is the whole reason the +/// helper is shaped this way. `tracing` resolves each callsite's `Interest` +/// **once for the process** and caches it; a callsite first executed on a thread +/// with no subscriber caches `Interest::never()` and is then skipped forever -- +/// before any later thread-local subscriber is ever consulted. Under a parallel +/// test harness, whichever thread reaches a callsite first therefore decides +/// whether any other test can ever capture it, which made every log-capture +/// assertion in this crate a coin flip. +/// +/// A single global subscriber removes the race: every callsite registers against +/// a real subscriber, so interest is always enabled, and per-thread routing +/// (`ACTIVE_BUFFER`) decides what is actually recorded. +fn install_global_capture() { + static INSTALLED: OnceLock<()> = OnceLock::new(); + INSTALLED.get_or_init(|| { + let subscriber = tracing_subscriber::fmt() + .with_writer(ThreadRoutedWriter) + .with_ansi(false) + .finish(); + // A global default may already be set by another component of the same + // test binary; capturing still works through the thread-local routing + // above, so a lost race here is not fatal. + let _ = tracing::subscriber::set_global_default(subscriber); + // Callsites executed before this point resolved their interest against + // the no-op dispatcher and cached "never". Recompute so they can be + // captured from here on. Done once, at install -- not per test. + tracing::callsite::rebuild_interest_cache(); + }); +} + +/// Resets this thread's buffer slot on drop, so a panicking test cannot leak +/// its buffer into whatever the harness runs next on the same thread. +struct BufferGuard; + +impl Drop for BufferGuard { + fn drop(&mut self) { + ACTIVE_BUFFER.with(|slot| *slot.borrow_mut() = None); + } +} + +/// A capture in progress on the current thread. Held across the code under +/// test -- including any `.await` points, as long as the test stays on a +/// single OS thread (e.g. the default current-thread `#[tokio::test]` +/// runtime) -- then queried for the formatted text captured so far. +pub(crate) struct CapturedLogs { + buffer: Arc>>, + _guard: BufferGuard, +} + +impl CapturedLogs { + /// A snapshot of everything captured on this thread so far, as the + /// formatted text a real log line would carry (level tag included). + pub fn text(&self) -> String { + let bytes = self + .buffer + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + String::from_utf8(bytes) + .unwrap_or_else(|e| String::from_utf8_lossy(&e.into_bytes()).into_owned()) + } +} + +/// Start capturing this thread's formatted `tracing` output. +/// +/// Only this thread's output is captured, so any async work driven while the +/// returned [`CapturedLogs`] is alive must run on the calling thread (e.g. a +/// current-thread runtime's `block_on`) to be captured -- the subscriber +/// itself is global, but the destination buffer is per-thread; see +/// [`install_global_capture`] for why that split is load-bearing. +pub(crate) fn capture_logs() -> CapturedLogs { + install_global_capture(); + let buffer = Arc::new(Mutex::new(Vec::new())); + ACTIVE_BUFFER.with(|slot| *slot.borrow_mut() = Some(buffer.clone())); + CapturedLogs { + buffer, + _guard: BufferGuard, + } +} diff --git a/crates/nuncio-store/src/vault.rs b/crates/nuncio-store/src/vault.rs index 26ec80c..a3fb2a7 100644 --- a/crates/nuncio-store/src/vault.rs +++ b/crates/nuncio-store/src/vault.rs @@ -365,47 +365,13 @@ mod tests { #[test] fn get_or_create_key_bytes_logs_minting_without_leaking_the_key() { - use std::sync::{Arc, Mutex}; - use tracing_subscriber::fmt::MakeWriter; - - #[derive(Clone, Default)] - struct CapturedLogs(Arc>>); - - impl std::io::Write for CapturedLogs { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - self.0 - .lock() - .expect("log buffer lock") - .extend_from_slice(buf); - Ok(buf.len()) - } - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } - } - - impl<'a> MakeWriter<'a> for CapturedLogs { - type Writer = CapturedLogs; - fn make_writer(&'a self) -> Self::Writer { - self.clone() - } - } - let manager = SecretManager::mock(); - let logs = CapturedLogs::default(); - let subscriber = tracing_subscriber::fmt() - .with_writer(logs.clone()) - .with_ansi(false) - .finish(); - - let guard = tracing::subscriber::set_default(subscriber); + let logs = crate::test_tracing::capture_logs(); let key = manager .get_or_create_key_bytes(WEBHOOK_SIGNING_KEY_ACCOUNT, 32) .expect("key provisioned from vault"); - drop(guard); - let captured = String::from_utf8(logs.0.lock().expect("log buffer lock").clone()) - .expect("captured log is valid utf8"); + let captured = logs.text(); assert!( captured.contains(WEBHOOK_SIGNING_KEY_ACCOUNT), "expected the key's purpose to be named in the INFO log, got: {captured}" diff --git a/crates/nunciod/src/outbox.rs b/crates/nunciod/src/outbox.rs index 7de867e..5c00417 100644 --- a/crates/nunciod/src/outbox.rs +++ b/crates/nunciod/src/outbox.rs @@ -107,6 +107,22 @@ pub trait RemoteExecutionEnv: Send + Sync { subject: &str, sender: &str, ) -> Result; + + /// How long a single mutation may run before it is abandoned as a + /// retryable timeout. Production keeps [`PER_ITEM_EXECUTION_TIMEOUT`]. + /// + /// Overridable so a test can pick a duration it can actually wait out in + /// real time. The alternative -- pausing the Tokio clock and letting + /// auto-advance jump the 30 seconds -- cannot be made reliable here, + /// because this executor performs real SQLite I/O on a blocking pool + /// within the same window. Auto-advance fires whenever the runtime looks + /// idle, and a paused clock also fires sqlx's own pool-acquire timeout + /// (30s by default -- the same duration), which turns a healthy query into + /// `PoolTimedOut` and makes this function return an all-zero summary. That + /// is precisely the shape the flaky CI failures took. + fn per_item_timeout(&self) -> Duration { + PER_ITEM_EXECUTION_TIMEOUT + } } /// The disposition of a single mutation attempt. @@ -174,12 +190,13 @@ pub async fn execute_pending_mutations( ); break; } - result = tokio::time::timeout(PER_ITEM_EXECUTION_TIMEOUT, execute_one(db, env, &item)) => result, + result = tokio::time::timeout(env.per_item_timeout(), execute_one(db, env, &item)) => result, }; let disposition = match outcome { Ok(disposition) => disposition, Err(_elapsed) => Disposition::Retry(format!( - "mutation execution exceeded the {PER_ITEM_EXECUTION_TIMEOUT:?} timeout" + "mutation execution exceeded the {:?} timeout", + env.per_item_timeout() )), }; let next_retry = item.retry_count + 1; diff --git a/crates/nunciod/src/test_tracing.rs b/crates/nunciod/src/test_tracing.rs index 9c68c9b..57eabd5 100644 --- a/crates/nunciod/src/test_tracing.rs +++ b/crates/nunciod/src/test_tracing.rs @@ -1,15 +1,21 @@ //! Shared in-test `tracing` capture helper. //! -//! Installs a thread-local subscriber that records every span (with its -//! fields) and every event (with its level, fields, and message) into an -//! in-memory buffer, so tests can assert on what the code logged -- including -//! the negative assertion that a secret never appears in any recorded field. -//! Deterministic and fully offline: no global subscriber, no I/O. +//! Records every span (with its fields) and every event (with its level, +//! fields, and message) into an in-memory buffer, so tests can assert on what +//! the code logged -- including the negative assertion that a secret never +//! appears in any recorded field. Fully offline: no I/O. +//! +//! The subscriber is installed **once, globally**, and events are routed to a +//! per-thread recorder. That split is deliberate and load-bearing: a +//! thread-local subscriber cannot work here, because `tracing` caches each +//! callsite's `Interest` process-wide on first use. See +//! [`install_global_capture`]. #![cfg(test)] +use std::cell::RefCell; use std::collections::HashMap; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, OnceLock}; use tracing::field::{Field, Visit}; use tracing::span::Attributes; @@ -88,46 +94,107 @@ impl Visit for FieldVisitor { } } -struct CaptureLayer(Arc); +thread_local! { + /// The recorder collecting telemetry for the test running on THIS thread, + /// or `None` on any thread not currently inside [`with_recorder`]. + /// + /// Routing per thread is what keeps the single global subscriber from + /// letting concurrent tests read each other's telemetry. + static ACTIVE_RECORDER: RefCell>> = const { RefCell::new(None) }; +} + +/// Run `f` against the recorder installed for this thread, if any. +fn with_active(f: impl FnOnce(&Recorder) -> R) -> Option { + ACTIVE_RECORDER.with(|slot| slot.borrow().as_ref().map(|rec| f(rec))) +} + +struct CaptureLayer; impl tracing_subscriber::Layer for CaptureLayer where S: Subscriber + for<'a> LookupSpan<'a>, { fn on_new_span(&self, attrs: &Attributes<'_>, _id: &Id, _ctx: LayerContext<'_, S>) { - let mut visitor = FieldVisitor(HashMap::new()); - attrs.record(&mut visitor); - self.0 - .spans - .lock() - .unwrap_or_else(|e| e.into_inner()) - .push(RecordedSpan { - name: attrs.metadata().name().to_string(), - fields: visitor.0, - }); + with_active(|rec| { + let mut visitor = FieldVisitor(HashMap::new()); + attrs.record(&mut visitor); + rec.spans + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(RecordedSpan { + name: attrs.metadata().name().to_string(), + fields: visitor.0, + }); + }); } fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) { - let mut visitor = FieldVisitor(HashMap::new()); - event.record(&mut visitor); - self.0 - .events - .lock() - .unwrap_or_else(|e| e.into_inner()) - .push(RecordedEvent { - level: *event.metadata().level(), - fields: visitor.0, - }); + with_active(|rec| { + let mut visitor = FieldVisitor(HashMap::new()); + event.record(&mut visitor); + rec.events + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(RecordedEvent { + level: *event.metadata().level(), + fields: visitor.0, + }); + }); } } -/// Runs `f` with a thread-local capturing subscriber installed and returns the -/// recorder alongside `f`'s result. Because the subscriber is thread-local, -/// any async work driven inside `f` must run on the calling thread (e.g. a -/// current-thread runtime's `block_on`) for its telemetry to be captured. +/// Install the capturing subscriber as the process-wide default, exactly once. +/// +/// It MUST be global rather than thread-local, and this is the whole reason the +/// helper is shaped this way. `tracing` resolves each callsite's `Interest` +/// **once for the process** and caches it; a callsite first executed on a thread +/// with no subscriber caches `Interest::never()` and is then skipped forever -- +/// before any later thread-local subscriber is ever consulted. Under a parallel +/// test harness, whichever thread reaches a callsite first therefore decides +/// whether any other test can ever capture it, which made every log-capture +/// assertion in this crate a coin flip. +/// +/// A single global subscriber removes the race: every callsite registers against +/// a real subscriber, so interest is always enabled, and per-thread routing +/// (`ACTIVE_RECORDER`) decides what is actually recorded. +fn install_global_capture() { + static INSTALLED: OnceLock<()> = OnceLock::new(); + INSTALLED.get_or_init(|| { + let subscriber = tracing_subscriber::registry().with(CaptureLayer); + // A global default may already be set by another component of the same + // test binary; capturing still works through the thread-local routing + // above, so a lost race here is not fatal. + let _ = tracing::subscriber::set_global_default(subscriber); + // Callsites executed before this point resolved their interest against + // the no-op dispatcher and cached "never". Recompute so they can be + // captured from here on. Done once, at install -- not per test. + tracing::callsite::rebuild_interest_cache(); + }); +} + +/// Resets this thread's recorder slot on drop, so a panicking test cannot leak +/// its recorder into whatever the harness runs next on the same thread. +struct RecorderGuard; + +impl Drop for RecorderGuard { + fn drop(&mut self) { + ACTIVE_RECORDER.with(|slot| *slot.borrow_mut() = None); + } +} + +/// Runs `f` with telemetry from THIS thread captured, returning the recorder +/// alongside `f`'s result. +/// +/// Only this thread's telemetry is recorded, so any async work driven inside +/// `f` must run on the calling thread (e.g. a current-thread runtime's +/// `block_on`) to be captured -- unchanged from the previous thread-local +/// implementation. What changed is that the subscriber itself is now global; +/// see [`install_global_capture`] for why that is load-bearing. pub(crate) fn with_recorder(f: impl FnOnce() -> T) -> (Arc, T) { + install_global_capture(); let recorder = Arc::new(Recorder::default()); - let subscriber = tracing_subscriber::registry().with(CaptureLayer(recorder.clone())); - let value = tracing::subscriber::with_default(subscriber, f); + ACTIVE_RECORDER.with(|slot| *slot.borrow_mut() = Some(recorder.clone())); + let _guard = RecorderGuard; + let value = f(); (recorder, value) } diff --git a/crates/nunciod/tests/outbox_executor_e2e_test.rs b/crates/nunciod/tests/outbox_executor_e2e_test.rs index 5fccdcd..6a9c42e 100644 --- a/crates/nunciod/tests/outbox_executor_e2e_test.rs +++ b/crates/nunciod/tests/outbox_executor_e2e_test.rs @@ -421,19 +421,30 @@ impl HangEnv { #[async_trait] impl RemoteExecutionEnv for HangEnv { + /// Short enough to wait out in real time, so these tests never pause the + /// Tokio clock. A paused clock auto-advances whenever the runtime looks + /// idle, and this executor does real SQLite I/O on a blocking pool in the + /// same window -- the advance also fires sqlx's 30s pool-acquire timer, + /// turning a healthy query into `PoolTimedOut` and the pass into an + /// all-zero summary. Generous enough to survive a loaded CI runner: the + /// hang it bounds is infinite, so a slow machine only makes the test + /// slower, never wrong. + fn per_item_timeout(&self) -> std::time::Duration { + std::time::Duration::from_millis(250) + } + async fn mail_backend( &self, account_id: &str, ) -> Result, nunciod::outbox::OutboxExecuteError> { let call = self.calls.fetch_add(1, Ordering::SeqCst); if call < self.hang_calls { - // A `tokio::time`-tracked sleep far longer than any per-item - // timeout under test -- NOT `std::future::pending`, which never - // registers with the time driver and so behaves unpredictably - // once `tokio::time::pause`'s auto-advance-on-idle interacts with - // real (non-timer) I/O elsewhere in the same test, like the - // SQLite pool's own connection bookkeeping. A real (virtual) - // sleep is what the timeout under test is meant to race against. + // Stands in for a remote operation that never answers. An hour is + // simply "longer than any timeout under test"; the executor + // abandons the attempt and drops this future, so nothing actually + // waits. A sleep rather than `std::future::pending` keeps the + // future registered with the time driver, which is the shape a + // real stalled network call has. tokio::time::sleep(std::time::Duration::from_secs(3600)).await; } self.inner.mail_backend(account_id).await @@ -489,23 +500,9 @@ async fn a_hung_mutation_times_out_and_is_retried_never_completed() { ) .await; - // Paused only AFTER setup, and this pass touches the database exactly - // ONCE (a single `get_message`) before the hang -- mixing - // `tokio::time::pause`'s auto-advance-on-idle with real (non-timer) - // background I/O is only safe when there is a single, already-in-flight - // real operation for the auto-advance to race past; see the sibling - // two-pass test below for why a SECOND real DB operation in the same - // paused pass is deliberately avoided. - tokio::time::pause(); let env = HangEnv::new(usize::MAX); let (_shutdown_ctrl, mut shutdown) = no_shutdown(); let summary = execute_pending_mutations(&db, &env, 10, &mut shutdown).await; - // Resume real time before any further DB access: `tokio::time::pause`'s - // auto-advance-on-idle only plays safely with the ONE real DB operation - // this pass performs (the `get_message` before the hang); a second real - // query made while still paused is what made the sibling two-pass test - // flaky before it was split across a resume point. - tokio::time::resume(); // The hung item is NEVER marked completed -- it is a retryable timeout, // exactly like any other transient remote failure. @@ -533,13 +530,6 @@ async fn a_timed_out_item_does_not_block_a_later_item_in_a_subsequent_pass() { let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); seed_two_move_mutations(&db).await; - // First pass: bound to the FIRST (created-earliest) mutation only, so - // this pass's real DB work is a single `get_message` call before the - // hang, same as the sibling single-item timeout test. Real threaded - // SQLite I/O racing `tokio::time::pause`'s auto-advance is only - // exercised once per paused pass here -- see that test's comment for - // why a second real DB call in the SAME paused pass is unreliable. - tokio::time::pause(); let env = HangEnv::new(usize::MAX); let (_shutdown_ctrl, mut shutdown) = no_shutdown(); let first_pass = execute_pending_mutations(&db, &env, 1, &mut shutdown).await; @@ -551,7 +541,6 @@ async fn a_timed_out_item_does_not_block_a_later_item_in_a_subsequent_pass() { // tick: the item that timed out in the first pass did not jam the queue // -- both it (now against a healthy backend) and the item queued behind // it are processed and genuinely complete. - tokio::time::resume(); let normal_env = MockEnv::new(); let (_shutdown_ctrl2, mut shutdown2) = no_shutdown(); let second_pass = execute_pending_mutations(&db, &normal_env, 10, &mut shutdown2).await; @@ -586,7 +575,6 @@ async fn shutdown_signal_interrupts_a_hung_item_instead_of_waiting_out_the_timeo // Paused only after setup -- see the comment on the sibling timeout test // for why pausing from the start would spuriously fail the DB pool. - tokio::time::pause(); // Hangs on every call -- the shutdown signal, not the timeout, must be // what unblocks this drain pass. @@ -598,7 +586,6 @@ async fn shutdown_signal_interrupts_a_hung_item_instead_of_waiting_out_the_timeo }); let summary = execute_pending_mutations(&db, &env, 10, &mut shutdown).await; - tokio::time::resume(); // The drain pass stopped on shutdown, well before the per-item timeout // would have elapsed -- the hung item is untouched: not completed, not