test(ecstore): fix two cluster::rpc flakes from process-global test state (#5438)

peer_rest_recovery_probe_logs_keep_request_id_span_context failed ~10% of
`cargo test -p rustfs-ecstore --lib -- cluster::rpc::` runs with
left: "request-span", right: "recovery-monitor". The recovery-monitor
info_span! was evaluating to Span::none(), so the probe's log line landed
under the caller's span.

tracing caches each 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 that derives the
interest from the registering thread's own subscriber, and registration is
once-only (CAS). Under libtest a sibling test reaches recovery_monitor_span
via mark_offline_and_spawn_recovery from a thread with no subscriber, so
the interest is derived from NoSubscriber and cached as Interest::never()
for the whole process.

Add pin_callsite_interest_for_test(): registering a second, inert
dispatcher rebuilds every registered callsite's interest against the live
dispatcher set (repairing a poisoned value) and keeps tracing-core off the
single-dispatcher fast path (preventing new ones). This also covers the
production marked_suspect / recovery_monitor_started event callsites that
remote_disk_network_error_starts_recovery_monitor_with_request_context
asserts on.

rename_data_response_accepts_legacy_json_without_decode_error is a
separate root cause: it snapshots the process-global internode metrics and
asserts the decode-error counter did not move, which siblings that record
decode errors (or reset the counters) invalidate. Put the 11 tests that
observe those counters in one #[serial(internode_metrics)] group.

Both races are impossible under nextest, which runs each test in its own
process, so neither test belongs in the ecstore-serial-flaky test-group
(that serializes across process boundaries) nor in the ci-profile
quarantine (they never redden CI).

Verified: cluster::rpc:: subset 0/30 failures under libtest (was 3/30);
target test paired with its poisoner 0/30 (was 4/20); 5/5 clean under
nextest at 179/179.

Cargo.lock

@@ -9212,6 +9212,7 @@ dependencies = [
  "tonic",
  "tower",
  "tracing",
+ "tracing-core",
  "tracing-opentelemetry",
  "tracing-subscriber",
  "url",

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" }

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"

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::*;

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::{

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");

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 {