v0.1.2-pre.005

crates/ksp-logging-lib/unit_tests/runtime.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-logging-lib/unit_tests/runtime.rs
-// version: 4
+// version: 5
 
 #[test]
 fn level_mapping_covers_all_ksp_levels() {
@@ -88,3 +88,107 @@ fn dropped_line_snapshots_add_saturating_by_sink() {
     assert_eq!(combined.file(), 11);
     assert_eq!(combined.total(), usize::MAX);
 }
+
+#[test]
+fn takeover_filter_prefers_more_specific_ksp_prefixes_and_supports_off() {
+    let settings = crate::LoggingSettings::new(crate::LogFilterLevel::Info, crate::SpanEvents::Off, std::option::Option::None, std::option::Option::None)
+        .with_target_filter(crate::TargetFilter::new("ksp-store-", crate::LogFilterLevel::Debug))
+        .with_target_filter(crate::TargetFilter::new("ksp-store-lib", crate::LogFilterLevel::Trace))
+        .with_target_filter(crate::TargetFilter::new("ksp-wallet-lib", crate::LogFilterLevel::Off));
+    let filter = super::build_target_filter(&settings);
+    assert!(filter.would_enable("ksp-store-other", &tracing::Level::DEBUG));
+    assert!(!filter.would_enable("ksp-store-other", &tracing::Level::TRACE));
+    assert!(filter.would_enable("ksp-store-lib", &tracing::Level::TRACE));
+    assert!(!filter.would_enable("ksp-wallet-lib", &tracing::Level::ERROR));
+}
+
+struct BlockingWriter {
+    first_write: bool,
+    started: std::sync::mpsc::SyncSender<()>,
+    release: std::sync::Arc<(std::sync::Mutex<bool>, std::sync::Condvar)>,
+}
+
+impl BlockingWriter {
+    fn new(started: std::sync::mpsc::SyncSender<()>, release: std::sync::Arc<(std::sync::Mutex<bool>, std::sync::Condvar)>) -> Self {
+        return Self { first_write: true, started, release };
+    }
+}
+
+impl std::io::Write for BlockingWriter {
+    fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
+        if self.first_write {
+            self.first_write = false;
+            if self.started.send(()).is_err() {
+                return std::result::Result::Err(std::io::Error::other("unable to notify saturation test that the writer is blocked"));
+            }
+            let (lock, condition) = self.release.as_ref();
+            let lock_result = lock.lock();
+            let mut released = match lock_result {
+                std::result::Result::Ok(released) => released,
+                std::result::Result::Err(_) => {
+                    return std::result::Result::Err(std::io::Error::other("saturation test release lock is poisoned"));
+                },
+            };
+            while !*released {
+                let wait_result = condition.wait(released);
+                released = match wait_result {
+                    std::result::Result::Ok(released) => released,
+                    std::result::Result::Err(_) => {
+                        return std::result::Result::Err(std::io::Error::other("saturation test release wait is poisoned"));
+                    },
+                };
+            }
+        }
+        return std::result::Result::Ok(buffer.len());
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        return std::result::Result::Ok(());
+    }
+}
+
+fn release_blocked_writer(release: &std::sync::Arc<(std::sync::Mutex<bool>, std::sync::Condvar)>) {
+    let (lock, condition) = release.as_ref();
+    let lock_result = lock.lock();
+    let mut released = match lock_result {
+        std::result::Result::Ok(released) => released,
+        std::result::Result::Err(error) => error.into_inner(),
+    };
+    *released = true;
+    condition.notify_all();
+}
+
+#[test]
+fn lossy_non_blocking_builder_drops_lines_instead_of_blocking_a_stalled_producer() {
+    let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1);
+    let release = std::sync::Arc::new((std::sync::Mutex::new(false), std::sync::Condvar::new()));
+    let writer = BlockingWriter::new(started_sender, std::sync::Arc::clone(&release));
+    let (mut non_blocking, worker_guard) = super::non_blocking_builder("ksp-logging-saturation-test").buffered_lines_limit(1).finish(writer);
+    let error_counter = non_blocking.error_counter();
+    let first_write = std::io::Write::write_all(&mut non_blocking, b"block worker\n");
+    assert!(first_write.is_ok());
+    let writer_started = started_receiver.recv_timeout(std::time::Duration::from_secs(2));
+    assert!(writer_started.is_ok());
+    let mut producer = non_blocking.clone();
+    let (finished_sender, finished_receiver) = std::sync::mpsc::sync_channel(1);
+    let producer_thread = std::thread::spawn(move || {
+        let mut succeeded = true;
+        for _ in 0..1_024 {
+            let write_result = std::io::Write::write_all(&mut producer, b"queued line\n");
+            if write_result.is_err() {
+                succeeded = false;
+                break;
+            }
+        }
+        let _send_result = finished_sender.send(succeeded);
+        return;
+    });
+    let producer_finished = finished_receiver.recv_timeout(std::time::Duration::from_secs(2));
+    release_blocked_writer(&release);
+    let join_result = producer_thread.join();
+    assert!(join_result.is_ok());
+    assert_eq!(producer_finished, std::result::Result::Ok(true));
+    assert!(error_counter.dropped_lines() > 0);
+    drop(non_blocking);
+    drop(worker_guard);
+}