From 99a8d1a96fb972b8b220cbb4c13995bff955e2ce Mon Sep 17 00:00:00 2001 From: James Maes Date: Wed, 12 Aug 2026 11:47:34 -0500 Subject: [PATCH 1/4] fix(test): capture logs through one global subscriber, not a thread-local tracing resolves a callsite's Interest once per process and caches it. A callsite first executed on a thread with no subscriber caches never and is skipped forever, before any thread-local subscriber is consulted. Under the parallel harness, whichever thread reached a callsite first decided whether any other test could capture it, so every log-capture assertion in this crate was a coin flip -- reproducible at --test-threads=2, which is what CI runners give us. Install the capture layer once as the global default and route events to a per-thread recorder instead. Rebuild the interest cache once at install so callsites already resolved against the no-op dispatcher are recoverable. Measured on nunciod: ~90% of runs red at --test-threads=2 before, 0/12 after. --- crates/nunciod/src/test_tracing.rs | 133 ++++++++++++++++++++++------- 1 file changed, 100 insertions(+), 33 deletions(-) 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) } -- 2.54.0 From 1cd1bcc13c764432c9343530684024b18e77d418 Mon Sep 17 00:00:00 2001 From: James Maes Date: Wed, 12 Aug 2026 12:26:46 -0500 Subject: [PATCH 2/4] fix(test): capture logs through a global subscriber in 4 more crates Ports the nunciod fix (99a8d1a) to nuncio-cal, nuncio-contacts, nuncio-store, and nuncio-filter: install the tracing capture subscriber once as the process default and route events to a per-thread buffer, instead of a thread-local subscriber that a racing callsite's cached Interest::never() can permanently defeat under a parallel test harness. nuncio-cal and nuncio-contacts get the same structured Recorder shape as nunciod. nuncio-store gains a formatted-text test_tracing helper (its assertions match on rendered level tags and substrings) used by vault.rs; the two migration-log CapturedLogs sites in db.rs are left untouched because those tests are being deleted separately. nuncio-filter keeps its dependency-free hand-rolled Subscriber but makes it global and per-thread-routed, replacing the local CapturingSubscriber in engine.rs and webhook.rs. --- crates/nuncio-cal/src/test_tracing.rs | 133 ++++++++++++++----- crates/nuncio-contacts/src/test_tracing.rs | 133 ++++++++++++++----- crates/nuncio-filter/src/engine.rs | 53 +------- crates/nuncio-filter/src/lib.rs | 2 + crates/nuncio-filter/src/test_tracing.rs | 146 +++++++++++++++++++++ crates/nuncio-filter/src/webhook.rs | 63 +-------- crates/nuncio-store/src/lib.rs | 2 + crates/nuncio-store/src/test_tracing.rs | 143 ++++++++++++++++++++ crates/nuncio-store/src/vault.rs | 38 +----- 9 files changed, 501 insertions(+), 212 deletions(-) create mode 100644 crates/nuncio-filter/src/test_tracing.rs create mode 100644 crates/nuncio-store/src/test_tracing.rs 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/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}" -- 2.54.0 From 15226b62a146e52b33bd31c4440ac2fe3093f6d5 Mon Sep 17 00:00:00 2001 From: James Maes Date: Wed, 12 Aug 2026 12:32:20 -0500 Subject: [PATCH 3/4] refactor(store): create the schema, never migrate it Four ALTER TABLE helpers upgraded a pre-existing on-disk schema: smtp columns, TLS-mode columns, the DAV collection_url, and the messages identity columns. Every column they add is already declared in the CREATE TABLE IF NOT EXISTS in the same migrate(), so on any database this engine actually creates they were unconditional no-ops. Nothing is released and every store is a local development database, so there is no older schema to meet. Delete the helpers and the three tests that existed only to exercise them. backfill_message_fts stays: salvage restores rows straight into messages, bypassing save_email's index write, so a recovered database really does have content the trigram index has never seen. Its doc no longer claims the legacy justification. Removes 495 lines, and with them the migrate_logs flake outright. --- crates/nuncio-store/src/db.rs | 520 +--------------------------------- 1 file changed, 10 insertions(+), 510 deletions(-) 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(), -- 2.54.0 From bbebd576d8c444fd90e565a183eb67bb33bbdd8f Mon Sep 17 00:00:00 2001 From: James Maes Date: Wed, 12 Aug 2026 13:10:12 -0500 Subject: [PATCH 4/4] fix(test): time out outbox mutations in real time, not on a paused clock The two hung-mutation tests paused the Tokio clock so auto-advance would jump the 30s per-item timeout. That cannot be made reliable: the executor does real SQLite I/O on a blocking pool inside the same window, and auto-advance fires whenever the runtime looks idle. It also fires sqlx's own pool-acquire timer, which defaults to the same 30 seconds -- turning a healthy list_pending_mutations into PoolTimedOut, whose error arm returns an all-zero summary. That is exactly the shape CI reported: completed 0 and failed 0 both as expected, retried 0 instead of 1. Make the per-item timeout an overridable method on RemoteExecutionEnv and give the hanging test env 250ms, so the tests wait it out in real time and no test pauses the clock. The hang is infinite, so a loaded runner makes these slower, never wrong. 50 runs at --test-threads=2 under 16-way CPU load: 0 failures. --- crates/nunciod/src/outbox.rs | 21 +++++++- .../nunciod/tests/outbox_executor_e2e_test.rs | 49 +++++++------------ 2 files changed, 37 insertions(+), 33 deletions(-) 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/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 -- 2.54.0