diff --git a/crates/peryx-storage/tests/unit/tests/meta/webhook_tests.rs b/crates/peryx-storage/tests/unit/tests/meta/webhook_tests.rs index a2e857b8c..2b77c81fd 100644 --- a/crates/peryx-storage/tests/unit/tests/meta/webhook_tests.rs +++ b/crates/peryx-storage/tests/unit/tests/meta/webhook_tests.rs @@ -1,12 +1,82 @@ +use std::cell::RefCell; use std::collections::HashSet; -use std::io::{Read as _, Seek as _}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex, OnceLock}; use rstest::rstest; use super::store; use crate::meta::{MetaStore, NewWebhookDelivery, WebhookDeliveryAttempt, WebhookDeliveryStatus, WebhookEventIntent}; +thread_local! { + /// Where this thread's events go while it holds a [`Captured`], and nowhere otherwise. + static CAPTURE: RefCell>>>> = const { RefCell::new(None) }; +} + +/// A writer that hands each event to the capture belonging to the thread that raised it, so one +/// subscriber serves every test at once. +struct ThreadCapture; + +impl std::io::Write for ThreadCapture { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + CAPTURE.with_borrow(|sink| { + if let Some(sink) = sink { + sink.lock().unwrap().extend_from_slice(buf); + } + }); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl tracing_subscriber::fmt::MakeWriter<'_> for ThreadCapture { + type Writer = Self; + + fn make_writer(&self) -> Self { + Self + } +} + +/// The events this thread raises, until it drops. +/// +/// The subscriber is installed once for the whole binary rather than per test, because a callsite +/// decides its interest the first time any thread executes it and every thread reads that decision +/// afterwards. A per-test subscriber leaves that decision to whichever thread arrives first: one +/// holding no subscriber at all resolves the callsite against nothing, caches `never`, and silences +/// the callsite for the tests that are asserting on it. A subscriber that outlives every test cannot +/// be the one missing when that question is asked. +struct Captured(Arc>>); + +impl Captured { + fn install() -> Self { + static SUBSCRIBER: OnceLock<()> = OnceLock::new(); + SUBSCRIBER.get_or_init(|| { + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_writer(ThreadCapture) + .finish(); + tracing::subscriber::set_global_default(subscriber) + .expect("the test binary installs one global subscriber"); + }); + let sink = Arc::new(Mutex::new(Vec::new())); + CAPTURE.with_borrow_mut(|slot| *slot = Some(Arc::clone(&sink))); + Self(sink) + } + + fn output(&self) -> String { + String::from_utf8(self.0.lock().unwrap().clone()).expect("the fmt subscriber writes utf-8") + } +} + +impl Drop for Captured { + fn drop(&mut self) { + CAPTURE.with_borrow_mut(|slot| *slot = None); + } +} + fn none() -> HashSet<(String, String)> { HashSet::new() } @@ -346,31 +416,22 @@ fn test_webhook_queue_scan_skips_and_cleans_damaged_rows(#[case] damage: QueueDa let store = MetaStore::open_existing(&path).unwrap(); assert_eq!(store.next_webhook_delivery_at().unwrap(), Some(20)); - let mut capture = tempfile::tempfile().unwrap(); - let subscriber = tracing_subscriber::fmt() - .with_ansi(false) - .without_time() - .with_writer(Mutex::new(capture.try_clone().unwrap())) - .finish(); - tracing::subscriber::with_default(subscriber, || { - for _ in 0..2 { - assert_eq!( - store - .list_due_webhook_deliveries(100, 10, &none()) - .unwrap() - .into_iter() - .map(|record| record.id) - .collect::>(), - std::slice::from_ref(&healthy) - ); - } - }); + let captured = Captured::install(); + for _ in 0..2 { + assert_eq!( + store + .list_due_webhook_deliveries(100, 10, &none()) + .unwrap() + .into_iter() + .map(|record| record.id) + .collect::>(), + std::slice::from_ref(&healthy) + ); + } if matches!(damage, QueueDamage::InvalidJson) { assert_eq!(store.get_webhook_delivery(damaged.as_deref().unwrap()).unwrap(), None); } - capture.rewind().unwrap(); - let mut output = String::new(); - capture.read_to_string(&mut output).unwrap(); + let output = captured.output(); assert_eq!(output.matches("discarding damaged webhook queue rows").count(), 1); assert!(output.contains(&format!("{count}=1")), "{output}"); assert!(!output.contains("unbounded-corrupt-identifier")); @@ -457,3 +518,26 @@ fn test_list_due_separates_same_target_name_across_indexes() { let ids: Vec<&str> = due.iter().map(|record| record.id.as_str()).collect(); assert_eq!(ids, [first.as_str(), second.as_str()]); } + +/// The interleaving that silenced this file's captures: a thread holding no subscriber is the first to +/// execute a callsite, which is when its interest is decided for every thread that follows. A capture +/// that only exists for the length of one test is absent at that moment; the binary's subscriber is +/// not, so the event still reaches the thread asserting on it. +#[test] +fn test_a_capture_survives_a_neighbour_reaching_the_callsite_first() { + let captured = Captured::install(); + std::thread::spawn(unsubscribed_first_event).join().unwrap(); + unsubscribed_first_event(); + + assert!( + captured.output().contains("callsite registered by a bare thread"), + "a thread with no subscriber decided this callsite for everyone: {:?}", + captured.output() + ); +} + +/// A callsite of its own, so the test asserts on a first execution rather than on one some earlier test +/// already resolved. +fn unsubscribed_first_event() { + tracing::warn!(target: "peryx::webhook", "callsite registered by a bare thread"); +}