From 5907af2484ff31251babf9a6474ca708107e03f6 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Tue, 13 Oct 2026 12:32:09 +0100 Subject: [PATCH 1/2] fix(tests): stop log-capture tests missing lines when run in parallel tracing caches, process-wide and per log line, whether any subscriber wants it, the first time a thread reaches the line. A test capturing nothing could cache a line as off, so a test capturing logs on another thread missed it. A global subscriber now keeps the answer open. --- .../src/network/service/liveness_checker.rs | 25 +----------- bin/dipper-service/src/test_support.rs | 38 +++++++++++++++++++ 2 files changed, 39 insertions(+), 24 deletions(-) diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 35810b18..2e8fcba0 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/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) -> String { - struct Buffer(Arc>>); - impl std::io::Write for Buffer { - fn write(&mut self, bytes: &[u8]) -> std::io::Result { - 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(); diff --git a/bin/dipper-service/src/test_support.rs b/bin/dipper-service/src/test_support.rs index d3c4112e..012393d7 100644 --- a/bin/dipper-service/src/test_support.rs +++ b/bin/dipper-service/src/test_support.rs @@ -178,7 +178,45 @@ pub async fn logged(run: impl std::future::Future) -> 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. `tracing` caches that answer per log line for the +/// whole process the first time any thread reaches the line, and its default +/// 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 ONCE: std::sync::Once = std::sync::Once::new(); + ONCE.call_once(|| { + tracing::subscriber::set_global_default(Askable) + .expect("the tests install no other global tracing subscriber"); + }); +} + +/// Logs nothing itself; it only keeps every log line's cached answer at "ask +/// the subscriber in place on this thread". +struct Askable; + +impl tracing::Subscriber for Askable { + fn register_callsite( + &self, + _: &'static tracing::Metadata<'static>, + ) -> tracing::subscriber::Interest { + tracing::subscriber::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) {} +} From cee373e664f0fd2bcc8c1dde443964a706c26f80 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Tue, 13 Oct 2026 12:40:08 +0100 Subject: [PATCH 2/2] test(logs): add a regression test for the cross-thread capture fix A second global tracing subscriber now fails every capture with a clear message instead of poisoning a Once, the capture helper's doc says it stops at DEBUG, and the new test reaches a log line from a thread that captures nothing before capturing the same line itself. --- bin/dipper-service/src/test_support.rs | 61 ++++++++++++++++++-------- 1 file changed, 43 insertions(+), 18 deletions(-) diff --git a/bin/dipper-service/src/test_support.rs b/bin/dipper-service/src/test_support.rs index 012393d7..2c3821fc 100644 --- a/bin/dipper-service/src/test_support.rs +++ b/bin/dipper-service/src/test_support.rs @@ -6,7 +6,7 @@ //! (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::{ @@ -14,6 +14,7 @@ use dipper_producer::{ 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) -> String { struct Buffer(Arc>>); impl std::io::Write for Buffer { @@ -185,28 +186,29 @@ pub async fn logged(run: impl std::future::Future) -> String { } /// Installs, once per test binary, a global subscriber that answers "ask each -/// time" for every log line. `tracing` caches that answer per log line for the -/// whole process the first time any thread reaches the line, and its default -/// 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. +/// 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 ONCE: std::sync::Once = std::sync::Once::new(); - ONCE.call_once(|| { - tracing::subscriber::set_global_default(Askable) - .expect("the tests install no other global tracing subscriber"); - }); + static INSTALLED: OnceLock = 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`" + ); } -/// Logs nothing itself; it only keeps every log line's cached answer at "ask -/// the subscriber in place on this thread". +/// 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>, - ) -> tracing::subscriber::Interest { - tracing::subscriber::Interest::sometimes() + fn register_callsite(&self, _: &'static tracing::Metadata<'static>) -> Interest { + Interest::sometimes() } fn enabled(&self, _: &tracing::Metadata<'_>) -> bool { false @@ -220,3 +222,26 @@ impl tracing::Subscriber for Askable { 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}" + ); + } +}