🧪 test(storage): share one capture subscriber

The three webhook queue-scan cases assert on captured log output, and one of them
failed on some runs and not others, a different case each time, with the count of
the cleanup warning reading zero on an unchanged tree.

A callsite decides its interest the first time any thread executes it, and every
thread reads that decision afterwards. While one dispatcher is alive, tracing
resolves that question against the registering thread's own default rather than
the registry, so a thread holding no subscriber resolves it against nothing,
caches never, and silences the callsite for the tests asserting on it. A
subscriber that lives only as long as one test is absent whenever a neighbour
asks first.

One subscriber for the binary cannot be the one missing at that moment, and
routing by thread keeps each case reading only its own events. Serialising the
cases would have hidden the interleaving rather than removing it, and the
assertion stays as it was: one warning across two scans, which is what proves the
second scan found nothing left to clean.

The new test drives the interleaving rather than waiting for it. A thread with no
subscriber reaches a callsite first, and the event a subscribed thread raises
there still lands in its capture. With a per-test subscriber that capture comes
back empty.

Author: gaborbernat

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<Option<Arc<Mutex<Vec<u8>>>>> = 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<usize> {
+        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<Mutex<Vec<u8>>>);
+
+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::<Vec<_>>(),
-                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::<Vec<_>>(),
+            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");
+}