Split integration-test from cooldb example

Cargo.toml

@@ -1,5 +1,6 @@
 [workspace]
 members = [
     "tokio-bin-process",
-    "cooldb"
+    "cooldb",
+    "integration-test"
 ]

cooldb/src/main.rs

@@ -1,9 +1,6 @@
 use clap::Parser;
-use std::time::Duration;
-use tokio::{
-    signal::unix::{signal, SignalKind},
-    sync::watch,
-};
+use tokio::signal::unix::{signal, SignalKind};
+use tokio::sync::watch;
 use tracing_appender::non_blocking::WorkerGuard;
 
 mod tracing_panic_handler;
@@ -14,21 +11,11 @@ pub enum LogFormat {
     Json,
 }
 
-#[derive(clap::ValueEnum, Clone, Copy)]
-pub enum Mode {
-    Standard,
-    ErrorAtRuntime,
-    ErrorAtStartup,
-    StdErrSpam,
-}
-
 #[derive(Parser, Clone)]
 #[clap()]
 pub struct ConfigOpts {
     #[arg(long, value_enum, default_value = "human")]
     pub log_format: LogFormat,
-    #[arg(long, value_enum)]
-    pub mode: Mode,
 }
 
 #[tokio::main]
@@ -56,7 +43,7 @@ async fn main() {
         trigger_shutdown_tx.send(true).unwrap();
     });
 
-    db_logic(trigger_shutdown_rx, opts.mode).await;
+    db_logic(trigger_shutdown_rx).await;
 }
 
 pub fn init_tracing(format: LogFormat) -> WorkerGuard {
@@ -82,29 +69,10 @@ pub fn init_tracing(format: LogFormat) -> WorkerGuard {
     guard
 }
 
-async fn db_logic(mut trigger_shutdown_rx: watch::Receiver<bool>, mode: Mode) {
-    if let Mode::ErrorAtStartup = mode {
-        tracing::error!("An error occurs during startup");
-    }
-
+async fn db_logic(mut trigger_shutdown_rx: watch::Receiver<bool>) {
     tracing::info!("accepting inbound connections");
 
-    let start = std::time::Instant::now();
-    match mode {
-        Mode::Standard | Mode::ErrorAtStartup => tracing::info!("some functionality occurs"),
-        Mode::ErrorAtRuntime => tracing::error!("some error occurs"),
-        Mode::StdErrSpam => {
-            tracing::info!("some functionality occurs");
-            loop {
-                eprintln!("some library is spitting out nonsense you dont care about");
-                tokio::task::yield_now().await;
-                if start.elapsed() > Duration::from_secs(5) {
-                    break;
-                }
-            }
-            tracing::info!("other functionality occurs");
-        }
-    }
+    tracing::info!("some functionality occurs");
 
     trigger_shutdown_rx.changed().await.unwrap();
 }

cooldb/tests/test.rs

@@ -8,7 +8,7 @@ use tokio_bin_process::BinProcess;
 #[tokio::test(flavor = "multi_thread")]
 async fn test_cooldb() {
     // Setup cooldb
-    let mut cooldb = cooldb("standard").await;
+    let mut cooldb = cooldb().await;
 
     // Assert that some functionality occured.
     // Use a timeout to prevent the test hanging if no events occur.
@@ -24,65 +24,9 @@ async fn test_cooldb() {
     cooldb.shutdown_and_then_consume_events(&[]).await;
 }
 
-// Generally tokio-bin-process only cares about the contents of stdout.
-// However if the application is spamming stderr then we need to ensure
-// we are reading from stderr otherwise the application will deadlock.
-#[tokio::test(flavor = "multi_thread")]
-async fn test_cooldb_stderr_spam() {
-    // Setup cooldb
-    let mut cooldb = cooldb("std-err-spam").await;
-
-    // Assert that some functionality occured.
-    // Use a timeout to prevent the test hanging if no events occur.
-    timeout(Duration::from_secs(5), cooldb.consume_events(1, &[]))
-        .await
-        .unwrap()
-        .assert_contains(
-            &EventMatcher::new()
-                .with_level(Level::Info)
-                .with_message("some functionality occurs"),
-        );
-
-    timeout(Duration::from_secs(10), cooldb.consume_events(1, &[]))
-        .await
-        .unwrap()
-        .assert_contains(
-            &EventMatcher::new()
-                .with_level(Level::Info)
-                .with_message("other functionality occurs"),
-        );
-
-    timeout(
-        Duration::from_secs(10),
-        cooldb.shutdown_and_then_consume_events(&[]),
-    )
-    .await
-    .unwrap();
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[should_panic(expected = r#"some error occurs
-Any ERROR or WARN events that occur in integration tests must be explicitly allowed by adding an appropriate EventMatcher to the method call."#)]
-async fn test_cooldb_error_at_runtime() {
-    let cooldb = cooldb("error-at-runtime").await;
-    cooldb.shutdown_and_then_consume_events(&[]).await;
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[should_panic(expected = r#"An error occurs during startup
-Any ERROR or WARN events that occur in integration tests must be explicitly allowed by adding an appropriate EventMatcher to the method call."#)]
-async fn test_cooldb_error_at_startup() {
-    let cooldb = cooldb("error-at-startup").await;
-    cooldb.shutdown_and_then_consume_events(&[]).await;
-}
-
-async fn cooldb(mode: &str) -> BinProcess {
-    let mut cooldb = BinProcess::start_binary(
-        bin_path!("cooldb"),
-        "cooldb",
-        &["--log-format", "json", "--mode", mode],
-    )
-    .await;
+async fn cooldb() -> BinProcess {
+    let mut cooldb =
+        BinProcess::start_binary(bin_path!("cooldb"), "cooldb", &["--log-format", "json"]).await;
 
     timeout(
         Duration::from_secs(30),

integration-test/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "integration-test"
+version = "0.1.0"
+edition = "2021"
+license = "Apache-2.0"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+clap = { version = "4.2.1", features = ["derive"] }
+tokio = "1.27.0"
+tracing = "0.1.15"
+tracing-subscriber = { version = "0.3.1", features = ["env-filter", "json"] }
+tracing-appender = "0.2.2"
+backtrace = "0.3.67"
+backtrace-ext = "0.2"
+
+[dev-dependencies]
+tokio-bin-process = { path = "../tokio-bin-process" }

integration-test/src/main.rs

@@ -0,0 +1,110 @@
+use clap::Parser;
+use std::time::Duration;
+use tokio::{
+    signal::unix::{signal, SignalKind},
+    sync::watch,
+};
+use tracing_appender::non_blocking::WorkerGuard;
+
+mod tracing_panic_handler;
+
+#[derive(clap::ValueEnum, Clone, Copy)]
+pub enum LogFormat {
+    Human,
+    Json,
+}
+
+#[derive(clap::ValueEnum, Clone, Copy)]
+pub enum Mode {
+    Standard,
+    ErrorAtRuntime,
+    ErrorAtStartup,
+    StdErrSpam,
+}
+
+#[derive(Parser, Clone)]
+#[clap()]
+pub struct ConfigOpts {
+    #[arg(long, value_enum, default_value = "human")]
+    pub log_format: LogFormat,
+    #[arg(long, value_enum)]
+    pub mode: Mode,
+}
+
+#[tokio::main]
+async fn main() {
+    let opts = ConfigOpts::parse();
+    let _guard = init_tracing(opts.log_format);
+
+    tracing::info!("Initializing!");
+
+    // We need to block on this part to ensure that we immediately register these signals.
+    // Otherwise if we included signal creation in the below spawned task we would be at the mercy of whenever tokio decides to start running the task.
+    let mut interrupt = signal(SignalKind::interrupt()).unwrap();
+    let mut terminate = signal(SignalKind::terminate()).unwrap();
+    let (trigger_shutdown_tx, trigger_shutdown_rx) = watch::channel(false);
+    tokio::spawn(async move {
+        tokio::select! {
+            _ = interrupt.recv() => {
+                tracing::info!("received SIGINT");
+            },
+            _ = terminate.recv() => {
+                tracing::info!("received SIGTERM");
+            },
+        };
+
+        trigger_shutdown_tx.send(true).unwrap();
+    });
+
+    db_logic(trigger_shutdown_rx, opts.mode).await;
+}
+
+pub fn init_tracing(format: LogFormat) -> WorkerGuard {
+    let (non_blocking, guard) = tracing_appender::non_blocking(std::io::stdout());
+
+    let builder = tracing_subscriber::fmt().with_writer(non_blocking);
+
+    match format {
+        LogFormat::Json => builder.json().init(),
+        LogFormat::Human => builder.init(),
+    }
+
+    // When in json mode we need to process panics as events instead of printing directly to stdout.
+    // This is so that:
+    // * We dont include invalid json in stdout
+    // * panics can be received by whatever is processing the json events
+    //
+    // We dont do this for LogFormat::Human because the default panic messages are more readable for humans
+    if let LogFormat::Json = format {
+        crate::tracing_panic_handler::setup();
+    }
+
+    guard
+}
+
+async fn db_logic(mut trigger_shutdown_rx: watch::Receiver<bool>, mode: Mode) {
+    if let Mode::ErrorAtStartup = mode {
+        tracing::error!("An error occurs during startup");
+    }
+
+    tracing::info!("accepting inbound connections");
+
+    let start = std::time::Instant::now();
+    match mode {
+        Mode::Standard | Mode::ErrorAtStartup => tracing::info!("some functionality occurs"),
+        Mode::ErrorAtRuntime => tracing::error!("some error occurs"),
+        Mode::StdErrSpam => {
+            tracing::info!("some functionality occurs");
+            loop {
+                eprintln!("some library is spitting out nonsense you dont care about");
+                tokio::task::yield_now().await;
+                if start.elapsed() > Duration::from_secs(5) {
+                    break;
+                }
+            }
+            tracing::info!("other functionality occurs");
+        }
+    }
+
+    trigger_shutdown_rx.changed().await.unwrap();
+}

integration-test/src/tracing_panic_handler.rs

@@ -0,0 +1,57 @@
+use backtrace::{Backtrace, BacktraceFmt, BytesOrWideString, PrintFmt};
+use std::fmt;
+
+pub fn setup() {
+    std::panic::set_hook(Box::new(|panic| {
+        let backtrace = BacktraceFormatter(Backtrace::new());
+        // If the panic has a source location, record it as structured fields.
+        if let Some(location) = panic.location() {
+            tracing::error!(
+                message = %panic,
+                panic.file = location.file(),
+                panic.line = location.line(),
+                panic.column = location.column(),
+                panic.backtrace = format!("{backtrace}"),
+            );
+        } else {
+            tracing::error!(
+                message = %panic,
+                panic.backtrace = format!("{backtrace}"),
+            );
+        }
+    }));
+}
+
+/// The `std::backtrace::Backtrace` formatting is really noisy because it includes all the pre-main and panic handling frames.
+/// Internal panics have logic to remove that but that is missing from `std::backtrace::Backtrace`.
+/// <https://github.com/rust-lang/rust/issues/105413>
+///
+/// As a workaround we use the backtrace crate and manually perform the required formatting
+struct BacktraceFormatter(Backtrace);
+
+// based on https://github.com/rust-lang/backtrace-rs/blob/5be2e8ba9cf6e391c5fa45219fc091b4075eb6be/src/capture.rs#L371
+impl fmt::Display for BacktraceFormatter {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // When printing paths we try to strip the cwd if it exists, otherwise
+        // we just print the path as-is. Note that we also only do this for the
+        // short format, because if it's full we presumably want to print
+        // everything.
+        let cwd = std::env::current_dir();
+        let mut print_path = move |fmt: &mut fmt::Formatter<'_>, path: BytesOrWideString<'_>| {
+            let path = path.into_path_buf();
+            if let Ok(cwd) = &cwd {
+                if let Ok(suffix) = path.strip_prefix(cwd) {
+                    return fmt::Display::fmt(&suffix.display(), fmt);
+                }
+            }
+            fmt::Display::fmt(&path.display(), fmt)
+        };
+
+        let mut f = BacktraceFmt::new(f, PrintFmt::Short, &mut print_path);
+        f.add_context()?;
+        for (frame, _) in backtrace_ext::short_frames_strict(&self.0) {
+            f.frame().backtrace_frame(frame)?;
+        }
+        f.finish()
+    }
+}