add temp and perm fail integration tests

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)

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 {

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<F>(&self, mut func: F, timeout: Duration) -> bool
+    where
+        F: FnMut(&BTreeMap<RecordType, usize>) -> 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)?;

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<()> {

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