diff --git a/crates/integration-tests/maildir-sink.lua b/crates/integration-tests/maildir-sink.lua index 0cd860df..d002e4de 100644 --- a/crates/integration-tests/maildir-sink.lua +++ b/crates/integration-tests/maildir-sink.lua @@ -25,6 +25,15 @@ kumo.on('init', function() } end) +kumo.on('smtp_server_rcpt_to', function(recipient) + if string.find(recipient.user, 'tempfail') then + kumo.reject(400, 'tempfail requested') + end + if string.find(recipient.user, 'permfail') then + kumo.reject(500, 'permfail requested') + end +end) + kumo.on('smtp_server_message_received', function(msg) msg:set_meta('queue', 'maildir') end) diff --git a/crates/integration-tests/source.lua b/crates/integration-tests/source.lua index 76fbb487..b7d128b5 100644 --- a/crates/integration-tests/source.lua +++ b/crates/integration-tests/source.lua @@ -12,7 +12,7 @@ kumo.on('init', function() kumo.configure_local_logs { log_dir = TEST_DIR .. '/logs', - max_segement_duration = '1s', + max_segment_duration = '1s', } kumo.define_spool { diff --git a/crates/integration-tests/src/kumod.rs b/crates/integration-tests/src/kumod.rs index 1087b54b..48a9f753 100644 --- a/crates/integration-tests/src/kumod.rs +++ b/crates/integration-tests/src/kumod.rs @@ -86,7 +86,6 @@ impl MailGenParams<'_> { }; let sender = self.sender.unwrap_or("sender@example.com"); let recip = self.recip.unwrap_or("recip@example.com"); - eprintln!("using {body}"); let message = mail_builder::MessageBuilder::new() .from(sender) .to(recip) @@ -138,6 +137,28 @@ impl DaemonWithMaildir { self.sink.wait_for_maildir_count(count, timeout).await } + pub async fn wait_for_source_summary(&self, mut func: F, timeout: Duration) -> bool + where + F: FnMut(&BTreeMap) -> bool, + { + tokio::select! { + _ = async { + loop { + if let Ok(summary) = self.source.dump_logs() { + let done = (func)(&summary); + if done { + return true; + } + } + + + tokio::time::sleep(Duration::from_millis(100)).await; + } + } => true, + _ = tokio::time::sleep(timeout) => false, + } + } + pub async fn stop_both(&mut self) -> anyhow::Result<()> { let (res_1, res_2) = tokio::join!(self.source.stop(), self.sink.stop()); res_1?; @@ -290,7 +311,7 @@ impl KumoDaemon { let mut counts = BTreeMap::new(); for entry in std::fs::read_dir(&dir)? { - let entry = dbg!(entry)?; + let entry = entry?; if entry.file_type()?.is_file() { let f = std::fs::File::open(entry.path())?; let data = zstd::stream::decode_all(f)?; diff --git a/crates/integration-tests/src/main.rs b/crates/integration-tests/src/main.rs index ae78b7dd..d8a9b578 100644 --- a/crates/integration-tests/src/main.rs +++ b/crates/integration-tests/src/main.rs @@ -9,10 +9,87 @@ fn main() { mod test { use super::kumod::*; use k9::assert_equal; + use kumo_log_types::RecordType::{Bounce, TransientFailure}; use mailparse::MailHeaderMap; use rfc5321::*; use std::time::Duration; + #[tokio::test] + async fn temp_fail() -> anyhow::Result<()> { + let mut daemon = DaemonWithMaildir::start().await?; + let mut client = daemon.smtp_client().await?; + + let response = MailGenParams { + recip: Some("tempfail@example.com"), + ..Default::default() + } + .send(&mut client) + .await?; + eprintln!("{response:?}"); + anyhow::ensure!(response.code == 250); + + daemon + .wait_for_source_summary( + |summary| summary.get(&TransientFailure).copied().unwrap_or(0) > 0, + Duration::from_secs(5), + ) + .await; + + daemon.stop_both().await?; + let delivery_summary = daemon.dump_logs()?; + k9::snapshot!( + delivery_summary, + " +DeliverySummary { + source_counts: { + Reception: 1, + TransientFailure: 1, + }, + sink_counts: {}, +} +" + ); + Ok(()) + } + + #[tokio::test] + async fn perm_fail() -> anyhow::Result<()> { + let mut daemon = DaemonWithMaildir::start().await?; + let mut client = daemon.smtp_client().await?; + + let response = MailGenParams { + recip: Some("permfail@example.com"), + ..Default::default() + } + .send(&mut client) + .await?; + eprintln!("{response:?}"); + anyhow::ensure!(response.code == 250); + + daemon + .wait_for_source_summary( + |summary| summary.get(&Bounce).copied().unwrap_or(0) > 0, + Duration::from_secs(5), + ) + .await; + + daemon.stop_both().await?; + let delivery_summary = daemon.dump_logs()?; + k9::snapshot!( + delivery_summary, + " +DeliverySummary { + source_counts: { + Reception: 1, + Bounce: 1, + }, + sink_counts: {}, +} +" + ); + Ok(()) + } + /// test maximum line length #[tokio::test] async fn max_line_length() -> anyhow::Result<()> { diff --git a/crates/kumod/src/logging.rs b/crates/kumod/src/logging.rs index c16de9c0..32fdeb08 100644 --- a/crates/kumod/src/logging.rs +++ b/crates/kumod/src/logging.rs @@ -124,6 +124,7 @@ impl LogFileParams { } } +#[derive(Debug)] enum LogCommand { Record(JsonLogRecord), Terminate, @@ -171,21 +172,25 @@ impl Logger { let headers = params.headers.clone(); let meta = params.meta.clone(); let (sender, receiver) = async_channel::bounded(params.back_pressure); - let thread = std::thread::spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_time() - .build() - .expect("create logger runtime"); - runtime.block_on(async move { - let mut state = LogThreadState { - params, - receiver, - template_engine, - file_map: HashMap::new(), - }; - state.logger_thread().await - }); - }); + let thread = std::thread::Builder::new() + .name("logger".to_string()) + .spawn(move || { + tracing::info!("started logger thread"); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .expect("create logger runtime"); + runtime.block_on(async move { + tracing::info!("calling state.logger_thread()"); + let mut state = LogThreadState { + params, + receiver, + template_engine, + file_map: HashMap::new(), + }; + state.logger_thread().await + }); + })?; let logger = Self { sender, @@ -446,13 +451,16 @@ struct LogThreadState { impl LogThreadState { async fn logger_thread(&mut self) { + tracing::debug!("LogFileParams: {:#?}", self.params); loop { let deadline = self.get_deadline(); + tracing::debug!("waiting until {deadline:?} for log"); let cmd = if let Some(deadline) = deadline { tokio::select! { cmd = self.receiver.recv() => cmd, _ = tokio::time::sleep_until(deadline.into()) => { + tracing::debug!("deadline reached, running expiration"); self.expire(); continue; } @@ -462,10 +470,14 @@ impl LogThreadState { }; let cmd = match cmd { Ok(cmd) => cmd, - _ => return, + other => { + tracing::debug!("logging channel closed {other:?}"); + return; + } }; match cmd { LogCommand::Terminate => { + tracing::debug!("LogCommand::Terminate received. Stopping writing logs"); break; } LogCommand::Record(record) => { @@ -480,7 +492,10 @@ impl LogThreadState { fn expire(&mut self) { let now = Instant::now(); self.file_map.retain(|_, of| match of.expires { - Some(exp) => exp > now, + Some(exp) => { + tracing::trace!("check {exp:?} vs {now:?} -> {}", exp > now); + exp > now + } None => true, }); } @@ -520,6 +535,7 @@ impl LogThreadState { } fn do_record(&mut self, mut record: JsonLogRecord) -> anyhow::Result<()> { + tracing::trace!("do_record {record:?}"); let file_key = if let Some(per_rec) = self.per_record(record.kind) { FileNameKey { log_dir: per_rec