fix: stop log-capture tests missing lines when run in parallel

bin/dipper-service/src/network/service/liveness_checker.rs

@@ -1227,6 +1227,7 @@ mod tests {
             IndexingAgreementTermsMetadata, IndexingRequest, IndexingRequestRegistry,
             PendingCancellationRegistry, Result as RegistryResult, StubAgreementRegistry,
         },
+        test_support::logged,
         worker::service::{JobPriority, WorkerQueue},
     };
 
@@ -1921,30 +1922,6 @@ mod tests {
         .await;
     }
 
-    /// What dipper logs, at every level, while `run` runs.
-    async fn logged(run: impl std::future::Future<Output = ()>) -> String {
-        struct Buffer(Arc<Mutex<Vec<u8>>>);
-        impl std::io::Write for Buffer {
-            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
-                self.0.lock().unwrap().extend_from_slice(bytes);
-                Ok(bytes.len())
-            }
-            fn flush(&mut self) -> std::io::Result<()> {
-                Ok(())
-            }
-        }
-        let lines = Arc::new(Mutex::new(Vec::new()));
-        let writer = Arc::clone(&lines);
-        let subscriber = tracing_subscriber::fmt()
-            .with_ansi(false)
-            .with_max_level(tracing::Level::DEBUG)
-            .with_writer(move || Buffer(Arc::clone(&writer)))
-            .finish();
-        let _logging = tracing::subscriber::set_default(subscriber);
-        run.await;
-        String::from_utf8(lines.lock().unwrap().clone()).unwrap()
-    }
-
     #[tokio::test]
     async fn ends_a_stale_agreement_through_cancelling_and_reassesses() {
         let agreement = stale_agreement();

bin/dipper-service/src/test_support.rs

@@ -6,14 +6,15 @@
 //! (and with which payloads) without a Kafka backend, and [`logged`], which captures what
 //! dipper logs.
 
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, OnceLock};
 
 use async_trait::async_trait;
 use dipper_producer::{
     events::{EmitError, SubgraphIndexingAgreementEventsProducer},
     proto,
 };
 use thegraph_core::{DeploymentId, alloy::primitives::ChainId};
+use tracing::subscriber::Interest;
 
 /// One captured lifecycle event: the routing metadata plus the typed payload.
 #[derive(Clone, Debug, PartialEq)]
@@ -159,7 +160,7 @@ impl SubgraphIndexingAgreementEventsProducer for CapturingEventsProducer {
     }
 }
 
-/// What dipper logs, at every level, while `run` runs.
+/// What dipper logs, at DEBUG and above, while `run` runs.
 pub async fn logged(run: impl std::future::Future<Output = ()>) -> String {
     struct Buffer(Arc<Mutex<Vec<u8>>>);
     impl std::io::Write for Buffer {
@@ -178,7 +179,69 @@ pub async fn logged(run: impl std::future::Future<Output = ()>) -> String {
         .with_max_level(tracing::Level::DEBUG)
         .with_writer(move || Buffer(Arc::clone(&writer)))
         .finish();
+    keep_every_log_line_askable();
     let _logging = tracing::subscriber::set_default(subscriber);
     run.await;
     String::from_utf8(lines.lock().unwrap().clone()).unwrap()
 }
+
+/// Installs, once per test binary, a global subscriber that answers "ask each
+/// time" for every log line and then declines them all. `tracing` caches that
+/// answer per log line for the whole process the first time any thread reaches
+/// the line, and with no global subscriber the answer is "never": a test that
+/// captures nothing can reach a line first and cache it as off, so a test
+/// capturing logs on another thread never sees it.
+fn keep_every_log_line_askable() {
+    static INSTALLED: OnceLock<bool> = OnceLock::new();
+    let installed =
+        *INSTALLED.get_or_init(|| tracing::subscriber::set_global_default(Askable).is_ok());
+    assert!(
+        installed,
+        "the tests install no other global tracing subscriber; capture logs through `logged`"
+    );
+}
+
+/// Answers "ask each time" for every log line and then declines them all, so it
+/// never sees a span or event; `new_span` is unreachable because `enabled` is
+/// always false.
+struct Askable;
+
+impl tracing::Subscriber for Askable {
+    fn register_callsite(&self, _: &'static tracing::Metadata<'static>) -> Interest {
+        Interest::sometimes()
+    }
+    fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
+        false
+    }
+    fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
+        tracing::span::Id::from_u64(1)
+    }
+    fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
+    fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
+    fn event(&self, _: &tracing::Event<'_>) {}
+    fn enter(&self, _: &tracing::span::Id) {}
+    fn exit(&self, _: &tracing::span::Id) {}
+}
+
+#[cfg(test)]
+mod tests {
+    use super::logged;
+
+    fn reach_a_shared_line() {
+        tracing::info!("A line another thread reached first");
+    }
+
+    #[tokio::test]
+    async fn captures_a_line_that_a_thread_capturing_nothing_reached_first() {
+        let logs = logged(async {
+            std::thread::spawn(reach_a_shared_line).join().unwrap();
+            reach_a_shared_line();
+        })
+        .await;
+
+        assert!(
+            logs.contains("A line another thread reached first"),
+            "{logs}"
+        );
+    }
+}