KofTwentyTwo/nuncio
2
0 Fork 0
Code Issues 73 Pull Requests 12 Actions Packages Projects Releases Wiki Activity

fix(test): install one global capture subscriber; delete schema migrations #411

Merged
KofTwentyTwo merged 4 commits from fix/test-flakes-tracing-race into dev 2026-08-14 14:46:38 +00:00
Conversation 0 Commits 4 Files Changed 13
+648 -788
13 changed files with 648 additions and 788 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/nuncio-cal/src/test_tracing.rs
+100 -33
View File
@@ -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<Recorder>);
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<Option<Arc<Recorder>>> = const { RefCell::new(None) };
}
/// Run `f` against the recorder installed for this thread, if any.
fn with_active<R>(f: impl FnOnce(&Recorder) -> R) -> Option<R> {
ACTIVE_RECORDER.with(|slot| slot.borrow().as_ref().map(|rec| f(rec)))
}
struct CaptureLayer;
impl<S> tracing_subscriber::Layer<S> 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<T>(f: impl FnOnce() -> T) -> (Arc<Recorder>, 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)
}
crates/nuncio-contacts/src/test_tracing.rs
+100 -33
View File
@@ -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<Recorder>);
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<Option<Arc<Recorder>>> = const { RefCell::new(None) };
}
/// Run `f` against the recorder installed for this thread, if any.
fn with_active<R>(f: impl FnOnce(&Recorder) -> R) -> Option<R> {
ACTIVE_RECORDER.with(|slot| slot.borrow().as_ref().map(|rec| f(rec)))
}
struct CaptureLayer;
impl<S> tracing_subscriber::Layer<S> 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<T>(f: impl FnOnce() -> T) -> (Arc<Recorder>, 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)
}
crates/nuncio-filter/src/engine.rs
+2 -51
View File
@@ -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<Mutex<Vec<String>>>,
}
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<Mutex<Vec<String>>> = 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")
crates/nuncio-filter/src/lib.rs
+2
View File
@@ -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;
crates/nuncio-filter/src/test_tracing.rs
+146
View File
@@ -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<Option<Arc<Mutex<Vec<String>>>>> = 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<Mutex<Vec<String>>>,
_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<String> {
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,
}
}
crates/nuncio-filter/src/webhook.rs
+4 -59
View File
@@ -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<Mutex<Vec<String>>>,
}
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<Mutex<Vec<String>>> =
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<Mutex<Vec<String>>> =
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()
crates/nuncio-store/src/db.rs
+10 -510
View File
@@ -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<String> = sqlx::query("PRAGMA table_info(messages)")
.fetch_all(&self.pool)
.await
.map_err(DatabaseError::Query)?
.iter()
.map(|row| row.get::<String, _>("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<String> = sqlx::query("PRAGMA table_info(accounts)")
.fetch_all(&self.pool)
.await
.map_err(DatabaseError::Query)?
.iter()
.map(|row| row.get::<String, _>("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<String> = sqlx::query("PRAGMA table_info(accounts)")
.fetch_all(&self.pool)
.await
.map_err(DatabaseError::Query)?
.iter()
.map(|row| row.get::<String, _>("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<String> = sqlx::query("PRAGMA table_info(accounts)")
.fetch_all(&self.pool)
.await
.map_err(DatabaseError::Query)?
.iter()
.map(|row| row.get::<String, _>("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<Mutex<Vec<u8>>>);
impl std::io::Write for CapturedLogs {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
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(),
crates/nuncio-store/src/lib.rs
+2
View File
@@ -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};
crates/nuncio-store/src/test_tracing.rs
+143
View File
@@ -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<Option<Arc<Mutex<Vec<u8>>>>> = 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<usize> {
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<Mutex<Vec<u8>>>,
_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,
}
}
crates/nuncio-store/src/vault.rs
+2 -36
View File
@@ -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<Mutex<Vec<u8>>>);
impl std::io::Write for CapturedLogs {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
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}"
crates/nunciod/src/outbox.rs
+19 -2
View File
@@ -107,6 +107,22 @@ pub trait RemoteExecutionEnv: Send + Sync {
subject: &str,
sender: &str,
) -> Result<u16, WebhookError>;
/// 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;
crates/nunciod/src/test_tracing.rs
+100 -33
View File
@@ -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<Recorder>);
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<Option<Arc<Recorder>>> = const { RefCell::new(None) };
}
/// Run `f` against the recorder installed for this thread, if any.
fn with_active<R>(f: impl FnOnce(&Recorder) -> R) -> Option<R> {
ACTIVE_RECORDER.with(|slot| slot.borrow().as_ref().map(|rec| f(rec)))
}
struct CaptureLayer;
impl<S> tracing_subscriber::Layer<S> 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<T>(f: impl FnOnce() -> T) -> (Arc<Recorder>, 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)
}
crates/nunciod/tests/outbox_executor_e2e_test.rs
+18 -31
View File
@@ -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<Box<dyn MailBackend>, 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