diff --git a/crates/nunciod/src/calendar_sync.rs b/crates/nunciod/src/calendar_sync.rs index 3b9c52f..6b698a7 100644 --- a/crates/nunciod/src/calendar_sync.rs +++ b/crates/nunciod/src/calendar_sync.rs @@ -43,14 +43,33 @@ pub async fn sync_with_backend( start_window: i64, end_window: i64, ) -> Result { + tracing::debug!( + calendar_id = %calendar_id, + start_window, + end_window, + "calendar sync: fetching events from backend" + ); let events = backend .fetch_events(calendar_id, start_window, end_window) - .await?; + .await + .inspect_err(|e| { + tracing::warn!( + calendar_id = %calendar_id, + "calendar sync: backend fetch failed: {e}" + ); + })?; + let fetched = events.len(); let mut synced = 0usize; for event in events { db.save_calendar_event(&event).await?; synced += 1; } + tracing::info!( + calendar_id = %calendar_id, + fetched, + synced, + "calendar sync: persisted fetched events" + ); Ok(synced) } @@ -72,6 +91,11 @@ pub async fn sync_caldav_account( start_window: i64, end_window: i64, ) -> Result { + tracing::info!( + account_id = %account.id, + calendar_id = %calendar_id, + "calendar sync: dispatching CalDAV sync for account" + ); let client = CalDavClient::new(CalDavAccountConfig { account_id: account.id.clone(), caldav_url: account.dav_collection_url().unwrap_or_default().to_string(), @@ -170,4 +194,50 @@ mod tests { assert!(matches!(err, CalendarSyncError::Backend(_))); assert!(err.to_string().contains("calendar backend error")); } + + /// A calendar sync must emit a followable INFO event carrying the calendar + /// id and the persisted count, so an operator can trace what a sync did. + #[test] + fn sync_with_backend_logs_calendar_id_and_synced_count() { + use crate::test_tracing::with_recorder; + use tracing::Level; + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + + let (recorder, ()) = with_recorder(|| { + runtime.block_on(async { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let mock = MockCalendarBackend::new(); + mock.add_event(mock_event( + "evt-1", + "cal-work", + 1_700_000_000, + 1_700_003_600, + )); + sync_with_backend(&db, &mock, "cal-work", 0, i64::MAX) + .await + .expect("sync succeeds"); + }); + }); + + let events = recorder.events(); + let persisted = events + .iter() + .find(|e| e.message() == "calendar sync: persisted fetched events") + .expect("a domain event for the persisted sync must be logged"); + assert_eq!(persisted.level, Level::INFO); + assert_eq!( + persisted.fields.get("calendar_id").map(String::as_str), + Some("cal-work") + ); + assert_eq!( + persisted.fields.get("synced").map(String::as_str), + Some("1") + ); + } } diff --git a/crates/nunciod/src/contacts_sync.rs b/crates/nunciod/src/contacts_sync.rs index b5cb883..640f8f7 100644 --- a/crates/nunciod/src/contacts_sync.rs +++ b/crates/nunciod/src/contacts_sync.rs @@ -43,12 +43,28 @@ pub async fn sync_with_backend( backend: &dyn ContactsBackend, account_id: &str, ) -> Result { - let contacts = backend.fetch_contacts(account_id).await?; + tracing::info!( + account_id = %account_id, + "contacts sync: dispatching sync for account" + ); + let contacts = backend.fetch_contacts(account_id).await.inspect_err(|e| { + tracing::warn!( + account_id = %account_id, + "contacts sync: backend fetch failed: {e}" + ); + })?; + let fetched = contacts.len(); let mut synced = 0usize; for contact in contacts { db.save_contact(&contact).await?; synced += 1; } + tracing::info!( + account_id = %account_id, + fetched, + synced, + "contacts sync: persisted fetched contacts" + ); Ok(synced) } @@ -120,4 +136,45 @@ mod tests { assert!(matches!(err, ContactsSyncError::Backend(_))); assert!(err.to_string().contains("contacts backend error")); } + + /// A contacts sync must emit a followable INFO event carrying the account + /// id and the persisted count. + #[test] + fn sync_with_backend_logs_account_id_and_synced_count() { + use crate::test_tracing::with_recorder; + use tracing::Level; + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + + let (recorder, ()) = with_recorder(|| { + runtime.block_on(async { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let mock = MockContactsBackend::new(); + mock.add_contact(mock_contact("ct-1", "acct-contacts-1", "Alice")); + sync_with_backend(&db, &mock, "acct-contacts-1") + .await + .expect("sync succeeds"); + }); + }); + + let events = recorder.events(); + let persisted = events + .iter() + .find(|e| e.message() == "contacts sync: persisted fetched contacts") + .expect("a domain event for the persisted sync must be logged"); + assert_eq!(persisted.level, Level::INFO); + assert_eq!( + persisted.fields.get("account_id").map(String::as_str), + Some("acct-contacts-1") + ); + assert_eq!( + persisted.fields.get("synced").map(String::as_str), + Some("1") + ); + } } diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 6562824..00659cf 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -68,6 +68,7 @@ use tokio_stream::{Stream, StreamExt}; use tonic::service::interceptor::InterceptedService; use tonic::transport::Server; use tonic::{Request, Response, Status}; +use tracing::Instrument; // Resource bounds applied to the tonic `Server` builder in // `serve_on_listener_with_overrides_and_shutdown`. These exist so a @@ -298,6 +299,17 @@ fn map_tls_mode_from_proto(mode: TlsModeProto) -> Result &'static str { + match transport { + nuncio_core::Transport::ImapSmtp(_) => "imap_smtp", + nuncio_core::Transport::Jmap(_) => "jmap", + nuncio_core::Transport::Dav(_) => "dav", + } +} + /// Maps a `nuncio_core::AccountConfig` onto its wire-format /// `nuncio.v1.AccountConfig` representation for `ListAccounts` responses. /// The password credential is NEVER part of `AccountConfig` (it lives only @@ -662,6 +674,12 @@ impl Accounts for AccountsGrpcService { return Err(Status::internal(message)); } + tracing::info!( + account_id = %config.id, + transport = transport_kind(&config.transport), + "Accounts: account added" + ); + Ok(Response::new(AddAccountResponse { config: Some(map_account_config_to_proto(config)), })) @@ -671,7 +689,7 @@ impl Accounts for AccountsGrpcService { &self, _request: Request, ) -> Result, Status> { - let accounts = self + let accounts: Vec = self .db .list_accounts() .await @@ -680,6 +698,8 @@ impl Accounts for AccountsGrpcService { .map(map_account_config_to_proto) .collect(); + tracing::debug!(count = accounts.len(), "Accounts: listed accounts"); + Ok(Response::new(ListAccountsResponse { accounts })) } @@ -712,6 +732,7 @@ impl Accounts for AccountsGrpcService { ))); } + let password_rotated = matches!(&req.password, Some(p) if !p.is_empty()); match req.password { // A password rotation writes the new credential first, then // persists. Unlike `add_account` (where no prior secret exists, @@ -735,6 +756,13 @@ impl Accounts for AccountsGrpcService { } } + tracing::info!( + account_id = %config.id, + transport = transport_kind(&config.transport), + password_rotated, + "Accounts: account updated" + ); + Ok(Response::new(UpdateAccountResponse {})) } @@ -777,6 +805,8 @@ impl Accounts for AccountsGrpcService { ); } + tracing::info!(account_id = %req.account_id, "Accounts: account removed"); + Ok(Response::new(RemoveAccountResponse {})) } @@ -809,6 +839,13 @@ impl Accounts for AccountsGrpcService { let report = self.connection_tester.probe(&config, &password).await; + tracing::info!( + account_id = %config.id, + imap_ok = report.imap.ok, + smtp_ok = report.smtp.ok, + "Accounts: connection probe finished" + ); + Ok(Response::new(TestAccountConnectionResponse { imap_ok: report.imap.ok, smtp_ok: report.smtp.ok, @@ -1137,6 +1174,12 @@ impl Calendar for CalendarGrpcService { } }; + tracing::info!( + calendar_id = %req.calendar_id, + synced, + "Calendar: sync finished" + ); + Ok(Response::new(CalendarSyncResponse { synced_count: synced as u64, })) @@ -1306,6 +1349,12 @@ impl Contacts for ContactsGrpcService { } }; + tracing::info!( + account_id = %req.account_id, + synced, + "Contacts: sync finished" + ); + Ok(Response::new(ContactsSyncResponse { synced_count: synced as u64, })) @@ -1421,6 +1470,12 @@ impl Contacts for ContactsGrpcService { .await .map_err(|e| Status::internal(format!("failed to save contact: {e}")))?; + tracing::info!( + contact_id = %contact.id, + account_id = ?contact.account_id, + "Contacts: contact created" + ); + Ok(Response::new(CreateContactResponse { contact: Some(map_contact_to_proto(contact)), })) @@ -1601,6 +1656,12 @@ impl Mail for MailGrpcService { ) })?; + tracing::info!( + message_id = %req.message_id, + read = req.read, + "Mail: mark-read applied" + ); + self.event_bus.process_command(CoreCommand::MarkRead { message_id: req.message_id, read: req.read, @@ -1724,6 +1785,11 @@ impl Mail for MailGrpcService { async fn sync(&self, request: Request) -> Result, Status> { let account_id = request.into_inner().account_id; + match &account_id { + Some(id) => tracing::info!(account_id = %id, "Mail: dispatching sync for account"), + None => tracing::info!("Mail: dispatching sync for all configured accounts"), + } + let synced = if let Some(backend) = &self.overrides.mail_backend { crate::sync::sync_with_backend( &self.db, @@ -1743,6 +1809,8 @@ impl Mail for MailGrpcService { sync_all_configured(&self.sync_dispatcher, &self.db, &self.event_bus).await }; + tracing::info!(synced, "Mail: sync finished"); + Ok(Response::new(SyncResponse { synced_count: synced as u64, })) @@ -1906,6 +1974,8 @@ impl Filters for FiltersGrpcService { self.reload_engine_from_store().await; + tracing::info!(rule_id = %rule.id, "Filters: rule created"); + Ok(Response::new(CreateRuleResponse { rule: Some(map_filter_rule_to_proto(rule)), })) @@ -1957,6 +2027,8 @@ impl Filters for FiltersGrpcService { self.reload_engine_from_store().await; + tracing::info!(rule_id = %req.rule_id, "Filters: rule deleted"); + Ok(Response::new(DeleteRuleResponse {})) } @@ -2082,6 +2154,8 @@ impl Filters for FiltersGrpcService { self.reload_engine_from_store().await; + tracing::info!(rule_id = %rule.id, "Filters: rule updated"); + Ok(Response::new(UpdateRuleResponse { rule: Some(map_filter_rule_to_proto(rule)), })) @@ -2117,6 +2191,12 @@ impl Filters for FiltersGrpcService { } }; + tracing::info!( + format = ?format, + rule_count = rules.len(), + "Filters: rules exported" + ); + Ok(Response::new(ExportRulesResponse { content })) } @@ -2156,6 +2236,12 @@ impl Filters for FiltersGrpcService { self.reload_engine_from_store().await; + tracing::info!( + imported_count, + error_count = errors.len(), + "Filters: rules imported" + ); + Ok(Response::new(ImportRulesResponse { imported_count, errors, @@ -2214,66 +2300,80 @@ impl Filters for FiltersGrpcService { req.chunk_size as usize }; + tracing::info!(chunk_size, "Filters: triage rescan started"); + let (tx, rx) = tokio::sync::mpsc::channel::>(16); let db = self.db.clone(); let engine = self.filter_engine.clone(); + // Carry the per-RPC span into the detached worker so its completion + // event inherits the same request_id as the RPC that started the scan. + let triage_span = tracing::Span::current(); - tokio::spawn(async move { - let mut last_id = String::new(); - let mut scanned: u64 = 0; - let mut matched: u64 = 0; - let mut applied: u64 = 0; + tokio::spawn( + async move { + let mut last_id = String::new(); + let mut scanned: u64 = 0; + let mut matched: u64 = 0; + let mut applied: u64 = 0; - loop { - let batch = match db.get_message_chunk(&last_id, chunk_size).await { - Ok(batch) => batch, - Err(e) => { - let _ = tx - .send(Err(Status::internal(format!( - "failed to read message chunk during triage: {e}" - )))) - .await; + loop { + let batch = match db.get_message_chunk(&last_id, chunk_size).await { + Ok(batch) => batch, + Err(e) => { + let _ = tx + .send(Err(Status::internal(format!( + "failed to read message chunk during triage: {e}" + )))) + .await; + return; + } + }; + let is_last_page = batch.len() < chunk_size; + + for email in &batch { + let actions_for_email = + crate::sync::apply_filter_actions(&db, &engine, email).await; + scanned += 1; + applied += actions_for_email as u64; + if actions_for_email > 0 { + matched += 1; + } + } + + if let Some(last) = batch.last() { + last_id = last.id.clone(); + } + + let done = is_last_page; + if tx + .send(Ok(TriageProgress { + scanned_count: scanned, + matched_count: matched, + actions_applied_count: applied, + last_message_id: last_id.clone(), + done, + })) + .await + .is_err() + { + // Receiver dropped: the caller disconnected, so there is + // no one left to report progress to. return; } - }; - let is_last_page = batch.len() < chunk_size; - for email in &batch { - let actions_for_email = - crate::sync::apply_filter_actions(&db, &engine, email).await; - scanned += 1; - applied += actions_for_email as u64; - if actions_for_email > 0 { - matched += 1; + if is_last_page { + tracing::info!( + scanned, + matched, + applied, + "Filters: triage rescan finished" + ); + return; } } - - if let Some(last) = batch.last() { - last_id = last.id.clone(); - } - - let done = is_last_page; - if tx - .send(Ok(TriageProgress { - scanned_count: scanned, - matched_count: matched, - actions_applied_count: applied, - last_message_id: last_id.clone(), - done, - })) - .await - .is_err() - { - // Receiver dropped: the caller disconnected, so there is - // no one left to report progress to. - return; - } - - if is_last_page { - return; - } } - }); + .instrument(triage_span), + ); Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) } @@ -2323,6 +2423,13 @@ impl Export for ExportGrpcService { } let format = map_export_format_from_proto(req.format())?; + let scope_kind = match &req.scope { + Some(export_request::Scope::AccountId(_)) => "account", + Some(export_request::Scope::FolderId(_)) => "folder", + None => "all", + }; + tracing::info!(format = ?format, scope = scope_kind, "Export: mailbox export started"); + let messages = match req.scope { Some(export_request::Scope::AccountId(account_id)) => { self.db @@ -2345,6 +2452,12 @@ impl Export for ExportGrpcService { .await .map_err(|e| Status::internal(format!("export failed: {e}")))?; + tracing::info!( + message_count = summary.message_count, + bytes_written = summary.bytes_written, + "Export: mailbox export finished" + ); + Ok(Response::new(ExportResponse { output_path: summary.output_path, message_count: summary.message_count as u64, @@ -2427,6 +2540,12 @@ impl Audit for AuditGrpcService { .await .map_err(|e| Status::internal(format!("failed to verify audit chain: {e}")))?; + tracing::info!( + valid = report.valid, + record_count = report.record_count, + "Audit: chain verification finished" + ); + Ok(Response::new(VerifyChainResponse { valid: report.valid, record_count: report.record_count as u64, @@ -2929,6 +3048,73 @@ mod tests { request } + /// A state-changing RPC must emit its own domain event, and (because every + /// handler event inherits the OBS-2 per-RPC span) that event must carry the + /// request-scoped `request_id`. Here the `Filters::create_rule` handler is + /// invoked inside a stand-in request span to prove the domain event both + /// fires and inherits the correlation id -- without duplicating the layer's + /// bare entry/exit lines. + #[test] + fn create_rule_handler_logs_domain_event_under_request_scope() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + + let (recorder, ()) = with_recorder(|| { + runtime.block_on(async { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let db = Arc::new(db); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rules")); + let service = FiltersGrpcService { db, filter_engine }; + + // Stand in for the RpcTraceLayer's per-RPC span so the handler's + // domain event inherits a request_id, exactly as in production. + let span = tracing::info_span!("grpc.request", request_id = "test-req-42"); + async { + service + .create_rule(Request::new(CreateRuleRequest { + name: "Domain Event Rule".to_string(), + priority: 1, + nsql: SAMPLE_RULE_NSQL.to_string(), + })) + .await + .expect("rule creation succeeds"); + } + .instrument(span) + .await; + }); + }); + + let events = recorder.events(); + let created = events + .iter() + .find(|e| e.message() == "Filters: rule created") + .expect("the create_rule handler must emit its domain event"); + assert_eq!(created.level, Level::INFO); + assert!( + created.fields.contains_key("rule_id"), + "the domain event must carry the rule id" + ); + + // The domain event was emitted while the handler ran inside the stand-in + // request span, so an operator can correlate it back to that request via + // the span's `request_id` (the same mechanism the RpcTraceLayer provides + // in production -- see `rpc_trace`'s own test). + let request_span = recorder + .spans() + .into_iter() + .find(|s| s.name == "grpc.request") + .expect("the request span must be recorded"); + assert_eq!( + request_span.fields.get("request_id").map(String::as_str), + Some("test-req-42"), + "the request span must carry the request-scoped correlation id" + ); + } + #[test] fn auth_accept_logs_debug_without_leaking_the_token() { let (recorder, result) = with_recorder(|| { diff --git a/crates/nunciod/src/lifecycle.rs b/crates/nunciod/src/lifecycle.rs index 107fd09..ce108bf 100644 --- a/crates/nunciod/src/lifecycle.rs +++ b/crates/nunciod/src/lifecycle.rs @@ -88,7 +88,12 @@ impl ShutdownController { /// Initiates shutdown. Idempotent: calling it more than once (e.g. a /// second Ctrl+C while already draining) is a no-op beyond the first. pub fn trigger(&self) { + // Drives the event bus to `ShuttingDown` and publishes the + // `CoreEvent::ShuttingDown` event to every subscriber (e.g. clients on + // `System/Subscribe`), so a graceful shutdown is observable over the + // API and not just in local logs. self.event_bus.process_command(CoreCommand::Shutdown); + tracing::info!("shutdown: ShuttingDown event published to subscribers"); let _ = self.tx.send(true); } } @@ -166,6 +171,29 @@ mod tests { assert!(late_clone.is_requested()); } + /// Triggering shutdown must both publish the `ShuttingDown` event (flipping + /// bus state) and emit a traceable lifecycle log line, so a clean shutdown + /// is observable in the logs and not only over the API. + #[test] + fn trigger_emits_a_traceable_shutting_down_lifecycle_log() { + use crate::test_tracing::with_recorder; + + let (recorder, ()) = with_recorder(|| { + let event_bus = Arc::new(EventBus::new()); + let (controller, _signal) = ShutdownController::new(event_bus.clone()); + controller.trigger(); + assert_eq!(event_bus.current_state().status, EngineStatus::ShuttingDown); + }); + + assert!( + recorder + .events() + .iter() + .any(|e| e.message() == "shutdown: ShuttingDown event published to subscribers"), + "trigger must emit the ShuttingDown lifecycle log" + ); + } + #[tokio::test] async fn trigger_is_idempotent() { let event_bus = Arc::new(EventBus::new()); diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index 3b9a892..134a805 100644 --- a/crates/nunciod/src/main.rs +++ b/crates/nunciod/src/main.rs @@ -308,10 +308,17 @@ async fn main() -> Result<(), Box> { } }; + // From here down is the graceful-shutdown drain: the gRPC server has + // stopped accepting new connections, so log each remaining teardown step + // to make a clean shutdown fully traceable end to end. + tracing::info!( + "shutdown: gRPC server stopped accepting connections; draining background workers" + ); + // Bound how long the background workers get to notice the shutdown // signal and exit their loops before the database is closed out from // under them. - let _ = tokio::time::timeout(SHUTDOWN_GRACE_PERIOD, async { + let workers_drained = tokio::time::timeout(SHUTDOWN_GRACE_PERIOD, async { if let Some(task) = sync_command_task { let _ = task.await; } @@ -321,12 +328,22 @@ async fn main() -> Result<(), Box> { } }) .await; + match workers_drained { + Ok(()) => tracing::info!("shutdown: background workers (sync, outbox, update) drained"), + Err(_) => tracing::warn!( + "shutdown: background workers did not all exit within the {:?} grace period; \ + proceeding to close the database", + SHUTDOWN_GRACE_PERIOD + ), + } signal_task.abort(); // Force the WAL checkpoint before the process exits so no committed // page is left for SQLite's own deferred "last connection closes" // teardown to race (see `DatabaseEngine::close`'s doc comment). + tracing::info!("shutdown: closing database and checkpointing the WAL"); db.close().await; + tracing::info!("shutdown: complete"); grpc_result.map_err(|e| format!("nunciod gRPC server failed: {e}"))?; diff --git a/crates/nunciod/src/outbox.rs b/crates/nunciod/src/outbox.rs index 813a659..7de867e 100644 --- a/crates/nunciod/src/outbox.rs +++ b/crates/nunciod/src/outbox.rs @@ -155,6 +155,16 @@ pub async fn execute_pending_mutations( let mut summary = ExecutionSummary::default(); for item in pending { + // Log the executor's decision-level attempt before issuing the real + // remote op: the mutation identity, its operation, and the surrogate + // message id it targets (never the message body/subject/recipients). + tracing::info!( + mutation_id = %item.id, + op = %item.mutation_type, + message_id = %item.message_id, + retry_count = item.retry_count, + "outbox: attempting mutation" + ); let outcome = tokio::select! { biased; _ = shutdown.wait() => { @@ -174,7 +184,14 @@ pub async fn execute_pending_mutations( }; let next_retry = item.retry_count + 1; let (status, retry, tally): (&str, i32, fn(&mut ExecutionSummary)) = match disposition { - Disposition::Completed => ("completed", item.retry_count, |s| s.completed += 1), + Disposition::Completed => { + tracing::info!( + mutation_id = %item.id, + op = %item.mutation_type, + "outbox: mutation completed" + ); + ("completed", item.retry_count, |s| s.completed += 1) + } Disposition::Permanent(reason) => { tracing::warn!( mutation_id = %item.id, @@ -298,6 +315,14 @@ async fn dispatch_mailbox_mutation( kind, }; + tracing::debug!( + op = %mutation.mutation_type, + remote_id = %spec.remote_id, + folder_id = %spec.folder_id, + uid_validity = %spec.uid_validity, + "outbox: issuing mailbox mutation against remote backend" + ); + match backend.apply_mutation(&spec).await { Ok(()) => Disposition::Completed, Err(e) => Disposition::Retry(format!("remote mutation failed: {e}")), @@ -505,3 +530,157 @@ impl RemoteExecutionEnv for ProductionExecutionEnv { .await } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::lifecycle::{ShutdownController, ShutdownSignal}; + use crate::test_tracing::with_recorder; + use nuncio_core::model::Email; + use nuncio_core::EventBus; + use nuncio_filter::MutationPayload; + use nuncio_mail::MockMailBackend; + use tracing::Level; + + /// Minimal test [`RemoteExecutionEnv`] whose mailbox mutations always + /// genuinely succeed against a recording [`MockMailBackend`]; the forward + /// and webhook paths are unused by these tests. + struct SucceedingEnv { + backend: MockMailBackend, + } + + #[async_trait] + impl RemoteExecutionEnv for SucceedingEnv { + async fn mail_backend( + &self, + _account_id: &str, + ) -> Result, OutboxExecuteError> { + Ok(Box::new(self.backend.clone())) + } + + async fn message_sender( + &self, + account_id: &str, + ) -> Result, OutboxExecuteError> { + // The FLAG-only tests never take the forward path; surface an error + // rather than a panic if that assumption ever changes. + Err(OutboxExecuteError::NotAMailAccount(account_id.to_string())) + } + + async fn dispatch_webhook( + &self, + _url: &str, + _rule_id: &str, + _message_id: &str, + _subject: &str, + _sender: &str, + ) -> Result { + // The FLAG-only tests never take the webhook path. + Err(WebhookError::SigningError( + "webhook path unused in test".into(), + )) + } + } + + fn no_shutdown() -> (ShutdownController, ShutdownSignal) { + ShutdownController::new(Arc::new(EventBus::new())) + } + + fn sample_email() -> Email { + Email { + id: "msg-outbox-log".to_string(), + account_id: "acct-outbox-log".to_string(), + folder_id: "INBOX".to_string(), + remote_id: "77".to_string(), + uid_validity: "9".to_string(), + subject: "Confidential outbox subject".to_string(), + sender: "alice@example.com".to_string(), + recipient: "owner@nuncio.mx".to_string(), + received_at: 1_700_000_000, + read: false, + body_plain: Some("secret outbox body".to_string()), + body_html: None, + attachments: Vec::new(), + } + } + + fn flag_mutation() -> PendingRemoteMutation { + PendingRemoteMutation { + id: "mut-1".to_string(), + rule_id: "rule-1".to_string(), + message_id: "msg-outbox-log".to_string(), + mutation_type: "FLAG".to_string(), + payload: serde_json::to_string(&MutationPayload { + action_type: "FLAG".to_string(), + target: None, + }) + .expect("serialize payload"), + status: "pending".to_string(), + retry_count: 0, + created_at: 1_700_000_000, + } + } + + /// An outbox drain must log the executor's decision-level attempt BEFORE + /// issuing the remote op and the Completed result AFTER, carrying the + /// mutation id and operation -- and never the message body or subject. + #[test] + fn drain_logs_attempt_and_completion_without_leaking_message_content() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + + let (recorder, ()) = with_recorder(|| { + runtime.block_on(async { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + db.save_folder_sync_state("acct-outbox-log", "INBOX", "9:100") + .await + .expect("save folder checkpoint"); + db.save_email(&sample_email()).await.expect("save email"); + db.save_pending_mutation(&flag_mutation()) + .await + .expect("save pending mutation"); + + let env = SucceedingEnv { + backend: MockMailBackend::new(), + }; + let (_ctl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 10, &mut shutdown).await; + assert_eq!(summary.completed, 1, "the mutation must complete"); + }); + }); + + let events = recorder.events(); + let attempt = events + .iter() + .find(|e| e.message() == "outbox: attempting mutation") + .expect("an attempt event must be logged before issuing the op"); + assert_eq!(attempt.level, Level::INFO); + assert_eq!(attempt.fields.get("op").map(String::as_str), Some("FLAG")); + assert_eq!( + attempt.fields.get("mutation_id").map(String::as_str), + Some("mut-1") + ); + + let completed = events + .iter() + .find(|e| e.message() == "outbox: mutation completed") + .expect("a completion event must be logged after a genuine success"); + assert_eq!(completed.level, Level::INFO); + assert_eq!(completed.fields.get("op").map(String::as_str), Some("FLAG")); + + for value in recorder.all_field_values() { + assert!( + !value.contains("Confidential outbox subject"), + "subject leaked into telemetry: {value}" + ); + assert!( + !value.contains("secret outbox body"), + "body leaked into telemetry: {value}" + ); + } + } +} diff --git a/crates/nunciod/src/send.rs b/crates/nunciod/src/send.rs index eebcba4..de07d15 100644 --- a/crates/nunciod/src/send.rs +++ b/crates/nunciod/src/send.rs @@ -183,6 +183,15 @@ pub async fn send_message_for_account( request: ComposeRequest, ) -> Result { let config = resolve_sending_account(db, request.account_id.as_deref()).await?; + let account_id = config.id.clone(); + let attachment_count = request.attachments.len(); + tracing::info!( + account_id = %account_id, + transport = "smtp", + has_cc = request.cc.is_some(), + attachment_count, + "mail send: dispatching outbound message over real SMTP transport" + ); let password = secrets.get_secret(&config.keyring_secret_key)?; let sender = build_smtp_sender(&config, &password)?; @@ -196,8 +205,24 @@ pub async fn send_message_for_account( attachments: request.attachments, }; - send_with_sender(&sender, &outbound).await?; - Ok(generate_message_id()) + match send_with_sender(&sender, &outbound).await { + Ok(()) => { + let message_id = generate_message_id(); + tracing::info!( + account_id = %account_id, + message_id = %message_id, + "mail send: outbound message accepted by SMTP transport" + ); + Ok(message_id) + } + Err(e) => { + tracing::warn!( + account_id = %account_id, + "mail send: SMTP transport rejected the outbound message: {e}" + ); + Err(e) + } + } } /// Test/E2E entry point for the `SendMessage` RPC when a @@ -215,6 +240,14 @@ pub async fn send_message_with_injected_sender( request: ComposeRequest, ) -> Result { let config = resolve_sending_account(db, request.account_id.as_deref()).await?; + let account_id = config.id.clone(); + tracing::info!( + account_id = %account_id, + transport = "injected", + has_cc = request.cc.is_some(), + attachment_count = request.attachments.len(), + "mail send: dispatching outbound message via injected sender" + ); let outbound = OutboundMessage { from: config.email_address, @@ -226,8 +259,24 @@ pub async fn send_message_with_injected_sender( attachments: request.attachments, }; - send_with_sender(sender, &outbound).await?; - Ok(generate_message_id()) + match send_with_sender(sender, &outbound).await { + Ok(()) => { + let message_id = generate_message_id(); + tracing::info!( + account_id = %account_id, + message_id = %message_id, + "mail send: outbound message accepted by injected sender" + ); + Ok(message_id) + } + Err(e) => { + tracing::warn!( + account_id = %account_id, + "mail send: injected sender rejected the outbound message: {e}" + ); + Err(e) + } + } } #[cfg(test)] @@ -534,4 +583,77 @@ mod tests { assert!(matches!(err, SendError::AccountNotFound(id) if id == "acct-does-not-exist")); assert!(mock.sent_messages().is_empty()); } + + /// A send must be followable in the log: a dispatch event before the send + /// and an acceptance event after, both scoped to the account id -- and the + /// recipient address, subject, and body must never appear in any logged + /// field. + #[test] + fn send_logs_dispatch_and_acceptance_without_leaking_recipient_or_subject() { + use crate::test_tracing::with_recorder; + use tracing::Level; + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + + let (recorder, ()) = with_recorder(|| { + runtime.block_on(async { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let config = sample_account("acct-log"); + db.save_account(&config).await.expect("save account"); + + let mock = MockMessageSender::new(); + send_message_with_injected_sender( + &db, + &mock, + ComposeRequest { + to: "secret-recipient@example.com".to_string(), + cc: None, + subject: "Confidential Subject Line".to_string(), + body_text: Some("secret body".to_string()), + body_html: None, + attachments: Vec::new(), + account_id: None, + }, + ) + .await + .expect("send succeeds"); + }); + }); + + let events = recorder.events(); + let dispatch = events + .iter() + .find(|e| e.message() == "mail send: dispatching outbound message via injected sender") + .expect("dispatch event must be logged"); + assert_eq!(dispatch.level, Level::INFO); + assert_eq!( + dispatch.fields.get("account_id").map(String::as_str), + Some("acct-log") + ); + assert!(events + .iter() + .any(|e| e.message() == "mail send: outbound message accepted by injected sender")); + + // Secret-safety: no recipient, subject, or body text may appear in any + // recorded telemetry field. + for value in recorder.all_field_values() { + assert!( + !value.contains("secret-recipient@example.com"), + "recipient leaked into telemetry: {value}" + ); + assert!( + !value.contains("Confidential Subject Line"), + "subject leaked into telemetry: {value}" + ); + assert!( + !value.contains("secret body"), + "body leaked into telemetry: {value}" + ); + } + } }