diff --git a/Cargo.lock b/Cargo.lock index bc15956fd..a4a358ab2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9212,6 +9212,7 @@ dependencies = [ "tonic", "tower", "tracing", + "tracing-core", "tracing-opentelemetry", "tracing-subscriber", "url", diff --git a/Cargo.toml b/Cargo.toml index f4de6dbf5..49e91d08a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -303,6 +303,7 @@ test-case = "3.3.1" thiserror = "2.0.19" tracing = { version = "0.1.44" } tracing-appender = "0.2.5" +tracing-core = "0.1.36" tracing-error = "0.2.1" tracing-opentelemetry = { version = "0.33" } tracing-subscriber = { version = "0.3.23" } diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index c35289f72..6d3962e1e 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -158,6 +158,9 @@ tokio = { workspace = true, features = ["rt-multi-thread", "macros", "test-util" criterion = { workspace = true, features = ["html_reports"] } temp-env = { workspace = true, features = ["async_closure"] } tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] } +# Only for `pin_callsite_interest_for_test`, which registers a `NoSubscriber` +# dispatcher to keep tracing's process-global callsite-interest cache honest. +tracing-core = { workspace = true } serial_test = { workspace = true } opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] } proptest = "1" diff --git a/crates/ecstore/src/cluster/rpc/background_monitor.rs b/crates/ecstore/src/cluster/rpc/background_monitor.rs index 897b6895e..946409943 100644 --- a/crates/ecstore/src/cluster/rpc/background_monitor.rs +++ b/crates/ecstore/src/cluster/rpc/background_monitor.rs @@ -84,6 +84,44 @@ where ) } +/// Stop a sibling test thread from poisoning tracing's process-global callsite +/// interest cache while this test asserts on span or event context. +/// +/// `tracing` caches every callsite's `Interest` in **process-global** state, and +/// the first thread to reach a callsite fixes that value. While at most one +/// dispatcher is registered, tracing-core takes a fast path +/// (`Dispatchers::rebuilder` -> `Rebuilder::JustOne`) that derives a +/// newly-registered callsite's interest from whichever subscriber is current +/// *on the registering thread*. +/// +/// In a multi-threaded `cargo test` binary a sibling test therefore routinely +/// reaches a production callsite first, from a thread with no subscriber +/// installed: the interest is derived from that thread's `NoSubscriber` and +/// cached as `Interest::never()` for the whole process. From then on the +/// callsite is dead for *every* later caller, including a test that installed +/// its own subscriber — an `info_span!` silently evaluates to `Span::none()` +/// (so the monitor's log line lands under the caller's span instead of +/// `recovery-monitor`), and a `warn!`/`info!` event never fires at all. +/// +/// Registering a second, inert dispatcher closes both halves of that race: +/// +/// * constructing it rebuilds every *already-registered* callsite's interest +/// against the live dispatcher set — which includes the caller's subscriber — +/// repairing whatever a sibling may already have poisoned; and +/// * holding it alive keeps tracing-core off the single-dispatcher fast path, so +/// a callsite registered *later* by any thread is resolved against that live +/// set instead of the registering thread's `NoSubscriber`. +/// +/// Call this **after** installing the test's subscriber, and hold the returned +/// guard for the rest of the test. Only the `cargo test` fallback needs it: +/// nextest runs each test in its own process, where there is no sibling thread +/// to lose the race to (see `docs/testing/README.md`). +#[cfg(test)] +#[must_use = "callsite interest is only pinned while the returned guard is held"] +pub(crate) fn pin_callsite_interest_for_test() -> tracing::Dispatch { + tracing::Dispatch::new(tracing_core::subscriber::NoSubscriber::default()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index 637fb5849..74d5bfbe8 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -23,6 +23,8 @@ pub(crate) mod remote_disk; pub(crate) mod remote_locker; pub(crate) mod runtime_sources; +#[cfg(test)] +pub(crate) use background_monitor::pin_callsite_interest_for_test; pub use background_monitor::shutdown_background_monitors; pub(crate) use background_monitor::spawn_background_monitor; pub use client::{ diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 203e6064d..eea06629d 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -2741,6 +2741,11 @@ mod tests { .with_span_list(true), ); let _guard = tracing::subscriber::set_default(subscriber); + // The `recovery-monitor` callsite is shared with the production + // `mark_offline_and_spawn_recovery` path that sibling tests exercise from + // subscriber-less threads; without this the span can be cached as + // `Interest::never()` and silently degrade to `Span::none()`. + let _callsite_pin = crate::cluster::rpc::pin_callsite_interest_for_test(); let client = test_peer_client(); let span = tracing::info_span!("request-span", request_id = "req-peer-rest"); diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index ffeaa2791..5151275d2 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -3005,6 +3005,7 @@ mod tests { use crate::cluster::rpc::internode_data_transport::{InternodeDataTransportCapabilities, TcpHttpInternodeDataTransport}; use crate::runtime::sources as runtime_sources; use serde_json::Value; + use serial_test::serial; use std::io::{self as std_io, Write}; use std::pin::Pin; use std::sync::{Arc, Mutex, Mutex as StdMutex, Once}; @@ -3018,6 +3019,20 @@ mod tests { static INIT: Once = Once::new(); + // `#[serial(internode_metrics)]` marks every test that observes + // `global_internode_metrics()`. Those counters are a process-wide singleton: + // some of these tests snapshot a counter, run one decode, and assert on the + // delta, while others deliberately record decode errors or call + // `reset_internode_metrics_for_test()`. Run concurrently in one process they + // corrupt each other's deltas — a sibling's error bumps the "no decode error" + // assertion off zero, and a sibling's reset can drive an `after > before` + // assertion backwards. + // + // The marker only takes effect under the `cargo test` fallback; nextest + // already isolates each test in its own process, so every test there gets its + // own copy of the counters (see `docs/testing/README.md`). Any new test that + // reads or mutates the global internode metrics belongs in this group. + #[test] fn snapshot_lease_response_requires_current_protocol_and_valid_token() { let token = SnapshotLeaseToken::new(); @@ -3306,6 +3321,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn read_multiple_response_decode_prefers_msgpack_payloads() { let endpoint = sample_remote_endpoint(); let msgpack_resp = sample_read_multiple_resp("msgpack", b"binary"); @@ -3328,6 +3344,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn read_multiple_response_decode_falls_back_to_json_payloads() { let endpoint = sample_remote_endpoint(); let json_resp = sample_read_multiple_resp("json", b"fallback"); @@ -3349,6 +3366,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn rename_data_response_accepts_legacy_json_without_decode_error() { crate::cluster::rpc::runtime_sources::reset_internode_metrics_for_test(); let response = RenameDataResp { @@ -3500,6 +3518,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn read_multiple_response_decode_reports_corrupt_msgpack_item() { let endpoint = sample_remote_endpoint(); let response = ReadMultipleResponse { @@ -3525,6 +3544,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn read_multiple_response_decode_reports_corrupt_json_item() { let endpoint = sample_remote_endpoint(); let response = ReadMultipleResponse { @@ -3565,6 +3585,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn batch_read_version_response_decode_prefers_msgpack_payloads() { let endpoint = sample_remote_endpoint(); let msgpack_resp = sample_batch_read_version_resp(7, "msgpack-object", true); @@ -3585,6 +3606,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn batch_read_version_response_rejects_invalid_success_metadata() { let endpoint = sample_remote_endpoint(); let mut response_item = sample_batch_read_version_resp(0, "invalid-object", true); @@ -3604,6 +3626,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn batch_read_version_response_decode_reports_corrupt_msgpack_item() { let endpoint = sample_remote_endpoint(); let response = BatchReadVersionResponse { @@ -3630,6 +3653,7 @@ mod tests { } #[test] + #[serial(internode_metrics)] fn batch_read_version_response_decode_reports_corrupt_json_item() { let endpoint = sample_remote_endpoint(); let response = BatchReadVersionResponse { @@ -4419,6 +4443,7 @@ mod tests { } #[tokio::test] + #[serial(internode_metrics)] async fn test_remote_disk_create_file_retries_once_on_retryable_open_write_error() { let transport = RetryingOpenWriteInternodeDataTransport::with_steps(vec![ OpenWriteTestStep::Error(DiskError::from(rustfs_rio::new_test_internode_http_io_error( @@ -4457,6 +4482,7 @@ mod tests { } #[tokio::test] + #[serial(internode_metrics)] async fn test_remote_disk_read_file_stream_retries_once_on_retryable_open_read_error() { // A transient reset-by-peer on a shard read during the read-after-write window must be // absorbed by one re-dial rather than eroding read quorum (issue #2761). @@ -5306,6 +5332,11 @@ mod tests { .with_span_list(true), ); let _guard = tracing::subscriber::set_default(subscriber); + // The `recovery-monitor` span and the monitor's own log events are + // production callsites that sibling tests exercise from subscriber-less + // threads; without this they can be cached as `Interest::never()` and go + // silently missing here. + let _callsite_pin = crate::cluster::rpc::pin_callsite_interest_for_test(); let endpoint = Endpoint { url: url::Url::parse("http://127.0.0.1:59996/data").expect("endpoint URL should parse"), @@ -5360,6 +5391,11 @@ mod tests { .with_span_list(true), ); let _guard = tracing::subscriber::set_default(subscriber); + // The `recovery-monitor` span and the monitor's own log events are + // production callsites that sibling tests exercise from subscriber-less + // threads; without this they can be cached as `Interest::never()` and go + // silently missing here. + let _callsite_pin = crate::cluster::rpc::pin_callsite_interest_for_test(); let addr = "http://127.0.0.1:59997".to_string(); let endpoint = Endpoint {