Compare commits

..

1 Commits

Author SHA1 Message Date
overtrue d39211da40 test(io-schedule): drop 24 no-op serial markers from the IO scheduler unit tests
`serial_test`'s `#[serial]` does not serialize anything under nextest (the
authoritative runner executes every test in its own process), and it only has
an effect under the `cargo test` fallback. `storage::concurrency::io_schedule`
is the one remaining `#[serial]` hotspot in the `rustfs` crate whose tests
touch no process environment and no process-global mutable state:

  * `IoPriorityQueue` is instance-local (`Arc<Mutex<..>>` fields built by
    `IoPriorityQueue::new`); no static registry.
  * `IoStrategy::from_context_with_config` is a pure function of the
    caller-supplied `IoSchedulingContext` + `IoSchedulerConfig`. The tests
    pass `concurrent_requests` explicitly, so they never read the global
    `ACTIVE_GET_REQUESTS` counter.
  * `IoSchedulerConfig::default()` / `IoPriorityQueueConfig::default()` read
    `rustfs_config::DEFAULT_*` constants only. `IoPriorityQueueConfig::from_env`
    exists but no test calls it.
  * `IoPriority::from_size` and `IoLoadLevel::from_wait_duration` resolve their
    thresholds from compile-time constants.

The single exception, `test_multi_factor_strategy_compatibility_path`, calls
`IoStrategy::from_wait_duration`, which does load `ACTIVE_GET_REQUESTS` into
the returned `concurrent_requests` field — but that field is not asserted, and
the `buffer_size` it does assert derives only from the load level.

The rest of `storage::concurrency` keeps its markers: `manager.rs` and
`multi_factor_scheduler_integration_test.rs` mutate and read the process-global
`ACTIVE_GET_REQUESTS` / `ACTIVE_PUT_REQUESTS` counters (`track_request()`,
`PutObjectGuard`, `reset_active_put_requests()`), so their `#[serial]` group is
load-bearing for the fallback runner.

Part of rustfs/backlog#1846 (T1, batch 3).
2026-08-19 10:36:12 +08:00
6 changed files with 32 additions and 146 deletions
+4 -23
View File
@@ -853,32 +853,13 @@ mod tests {
#[test]
fn test_error_conversions() {
// A plain io::Error carries no typed payload to recover, so it lands in
// `Io` rather than being guessed at from its kind — `NotFound` here must
// not silently become `FileNotFound`, which quorum aggregation counts as
// a different error (rustfs/backlog#1836).
// Test From implementations
let io_error = std::io::Error::new(std::io::ErrorKind::NotFound, "test");
let disk_error: DiskError = io_error.into();
match &disk_error {
DiskError::Io(inner) => assert_eq!(inner.kind(), std::io::ErrorKind::NotFound),
other => panic!("a plain io::Error must stay typed as Io, got {other:?}"),
}
let _disk_error: DiskError = io_error.into();
// A typed DiskError boxed through io::Error round-trips back to itself
// instead of degrading to `Io`.
let boxed: std::io::Error = std::io::Error::other(DiskError::VolumeNotFound);
assert_eq!(DiskError::from(boxed), DiskError::VolumeNotFound);
// serde_json errors have no dedicated variant and fold into `other`,
// keeping the original message.
let json_str = r#"{"invalid": json}"#;
let json_str = r#"{"invalid": json}"#; // Invalid JSON
let json_error = serde_json::from_str::<serde_json::Value>(json_str).unwrap_err();
let json_message = json_error.to_string();
let disk_error: DiskError = json_error.into();
assert!(
disk_error.to_string().contains(&json_message),
"the json error message must survive the conversion: {disk_error}"
);
let _disk_error: DiskError = json_error.into();
}
#[test]
+20 -35
View File
@@ -436,45 +436,30 @@ mod tests {
assert_eq!(unknown_profile.sequential_boost_multiplier, 1.0);
}
// What platform probing returns depends on the machine, so these pin the two
// rules that do not: the override wins over probing, and probing that is
// switched off reports Unknown rather than guessing (rustfs/backlog#1836).
#[cfg(target_os = "linux")]
#[test]
fn storage_media_override_wins_over_platform_detection() {
for (override_value, expected) in [
("nvme", StorageMedia::Nvme),
("ssd", StorageMedia::Ssd),
("hdd", StorageMedia::Hdd),
] {
assert_eq!(detect_storage_media(true, override_value), expected);
assert_eq!(
detect_storage_media(false, override_value),
expected,
"an override must be honoured even with detection disabled"
);
fn test_linux_storage_detection_exists() {
// This test just verifies the detection function exists and doesn't panic
// The actual result depends on the system it's running on
let result = detect_storage_media(true, "");
// We should get some result (not panic)
match result {
StorageMedia::Nvme | StorageMedia::Ssd | StorageMedia::Hdd | StorageMedia::Unknown => {
// All valid results
}
}
}
#[cfg(target_os = "macos")]
#[test]
fn disabled_detection_reports_unknown_instead_of_guessing() {
assert_eq!(detect_storage_media(false, ""), StorageMedia::Unknown);
assert_eq!(
detect_storage_media(false, "not-a-medium"),
StorageMedia::Unknown,
"an unparseable override falls through to the disabled path"
);
}
#[test]
fn enabled_detection_returns_a_medium_for_this_platform() {
// Whatever this machine reports, it must be one of the known variants and
// it must be stable across calls — a probe that flapped would make the
// scheduler's profile depend on when it asked.
let first = detect_storage_media(true, "");
assert!(matches!(
first,
StorageMedia::Nvme | StorageMedia::Ssd | StorageMedia::Hdd | StorageMedia::Unknown
));
assert_eq!(detect_storage_media(true, ""), first);
fn test_macos_storage_detection_exists() {
// This test just verifies the detection function exists and doesn't panic
let result = detect_storage_media(true, "");
// We should get some result (not panic)
match result {
StorageMedia::Nvme | StorageMedia::Ssd | StorageMedia::Hdd | StorageMedia::Unknown => {
// All valid results
}
}
}
}
+1 -11
View File
@@ -527,19 +527,9 @@ mod tests {
}
#[tokio::test]
async fn stopping_replay_workers_is_a_no_op_when_there_are_none() {
async fn runtime_facade_stops_empty_replay_workers() {
let (facade, _, _) = build_facade();
facade.stop_replay_workers().await;
// The stop path takes the worker list and hands it to the adapter, so an
// empty facade must come back with the list still empty and dispatch
// released rather than left paused (rustfs/backlog#1836).
assert!(facade.replay_workers.read().await.is_empty());
// Calling it twice must stay harmless: shutdown paths do exactly that.
facade.stop_replay_workers().await;
assert!(facade.replay_workers.read().await.is_empty());
}
#[tokio::test]
+3 -10
View File
@@ -873,16 +873,9 @@ mod tests {
/// now return a finite, non-panicking mask.
#[test]
fn test_mask_never_recurses_for_any_variant() {
// Terminating is the point — a regression here overflows the stack rather
// than failing an assertion — but the masks are collected and checked so
// the loop cannot be optimised into nothing and so a variant that starts
// returning an empty mask is caught too (rustfs/backlog#1836).
let masks: Vec<u64> = ALL_EVENT_NAMES.iter().map(|ev| ev.mask()).collect();
assert_eq!(masks.len(), ALL_EVENT_NAMES.len());
for (ev, mask) in ALL_EVENT_NAMES.iter().zip(&masks) {
assert_ne!(*mask, 0, "{ev:?} must carry at least one bit");
assert_eq!(ev.mask(), *mask, "{ev:?} must return the same mask every call");
for ev in ALL_EVENT_NAMES {
// Must terminate (no infinite recursion / stack overflow).
let _ = ev.mask();
}
}
@@ -1766,11 +1766,9 @@ mod tests {
};
use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
use rustfs_io_metrics::bandwidth::{BandwidthSnapshot, BandwidthTier};
use serial_test::serial;
use std::time::Duration;
#[tokio::test]
#[serial]
async fn test_io_priority_queue_basic() {
let config = IoPriorityQueueConfig::default();
let queue = IoPriorityQueue::new(config);
@@ -1789,7 +1787,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_priority_queue_dequeue_order() {
let config = IoPriorityQueueConfig::default();
let queue = IoPriorityQueue::new(config);
@@ -1817,7 +1814,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_priority_queue_status() {
let config = IoPriorityQueueConfig::default();
let queue = IoPriorityQueue::new(config);
@@ -1835,7 +1831,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_priority_queue_starvation_prevention() {
let config = IoPriorityQueueConfig {
starvation_threshold_secs: 1,
@@ -1859,7 +1854,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_priority_from_size() {
// High priority: < 1MB
assert_eq!(IoPriority::from_size(100 * 1024), IoPriority::High);
@@ -1875,7 +1869,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_load_level_from_wait_duration() {
use std::time::Duration;
@@ -1893,7 +1886,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_scheduler_config_default() {
let config = IoSchedulerConfig::default();
@@ -1907,7 +1899,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_scheduler_config_to_core_config() {
let config = IoSchedulerConfig::default();
let core = config.to_core_config();
@@ -1923,7 +1914,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_priority_queue_config_to_core_config() {
let config = IoPriorityQueueConfig::default();
let core = config.to_core_config();
@@ -1935,7 +1925,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_io_priority_queue_config_from_scheduler_config() {
let scheduler_config = IoSchedulerConfig {
queue_high_capacity: 128,
@@ -1958,7 +1947,6 @@ mod tests {
// ============================================
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_nvme_sequential_low_load() {
// NVMe + Sequential + Low load = maximum buffer size
let context = IoSchedulingContext {
@@ -1985,7 +1973,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_hdd_random_high_load() {
// HDD + Random + High load = conservative buffer size
let context = IoSchedulingContext {
@@ -2012,7 +1999,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_ssd_mixed_medium_load() {
// SSD + Mixed + Medium load = moderate buffer
let context = IoSchedulingContext {
@@ -2040,7 +2026,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_critical_load_disables_features() {
// Any media + Critical load = minimal features
let context = IoSchedulingContext {
@@ -2065,7 +2050,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_buffer_cap_enforcement() {
// Test that storage media caps are enforced
let context = IoSchedulingContext {
@@ -2090,7 +2074,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_applies_sequential_hint_when_pattern_unknown() {
let context = IoSchedulingContext {
file_size: 2 * 1024 * 1024 * 1024, // 2GiB
@@ -2115,7 +2098,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_bandwidth_low_reduces_buffer() {
// Low bandwidth should reduce buffer
let context = IoSchedulingContext {
@@ -2139,7 +2121,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_high_concurrency_reduction() {
// High concurrency should reduce buffer
let context = IoSchedulingContext {
@@ -2162,7 +2143,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_sequential_boost() {
// Sequential reads should get boost
let sequential_context = IoSchedulingContext {
@@ -2204,7 +2184,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_unknown_media_conservative() {
// Unknown media should be conservative
let context = IoSchedulingContext {
@@ -2230,7 +2209,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_priority_classification() {
// Test priority classification based on file size
let small_context = IoSchedulingContext {
@@ -2277,7 +2255,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_readahead_decision_matrix() {
// Test readahead enable/disable logic
let configs = vec![
@@ -2363,7 +2340,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_buffer_multiplier_stages() {
// Test that all multiplier stages are applied
let context = IoSchedulingContext {
@@ -2398,7 +2374,6 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn test_multi_factor_strategy_compatibility_path() {
// Test that compatibility path (from_wait_duration) still works
let wait_duration = Duration::from_millis(50);
+4 -42
View File
@@ -49,47 +49,19 @@ import sys
from pathlib import Path
VERIFY_SIGNALS = re.compile(
r"assert[a-z0-9_]*!|debug_assert|panic!\(|\.expect\(|\.unwrap\(|"
r"assert!|assert_eq!|assert_ne!|debug_assert|panic!\(|\.expect\(|\.unwrap\(|"
r"unreachable!|matches!\(|insta::|proptest!|\.await\?|\)\?|\?;|should_panic"
)
DELEGATION = re.compile(
r"\b(?:assert|verify|check|expect|ensure|run)_[a-z0-9_]*(?:::<[^>]*>)?\s*\(|"
r"\b[a-z0-9_]+_(?:case|cases|harness|roundtrip|round_trip)(?:::<[^>]*>)?\s*\("
r"\b(?:assert|verify|check|expect|ensure|run)_[a-z0-9_]*\s*\(|"
r"\b[a-z0-9_]+_(?:case|cases|harness|roundtrip|round_trip)\s*\("
)
# A body whose whole content is one call delegates by construction, whatever the
# callee is named: `run(DurabilityMode::Strict).await` and
# `aborting_encode_drops_blocked_producer(EncodePipeline::Vec).await` both hand
# every assertion to a shared harness.
SINGLE_CALL_BODY = re.compile(
r"\A\s*[a-zA-Z_][a-zA-Z0-9_:]*(?:::<[^>]*>)?\s*\([^;]*\)\s*(?:\.await\s*)?;?\s*\Z",
re.S,
)
# A nested `fn` that is only bound and discarded is a signature guard: the type
# system is the assertion, exactly like the `fn _name()` form below.
SIGNATURE_GUARD = re.compile(r"\bfn\s+[a-zA-Z0-9_]+\s*(?:<[^>]*>)?\s*\([^;]*\)[^;]*\{", re.S)
DISCARDED_BINDING = re.compile(r"\blet\s+_\s*=\s*[a-zA-Z_][a-zA-Z0-9_]*\s*;")
# `let _ = Type::<T>::method;` — a path item referenced but never called can only
# be a signature guard; the call form (`let _ = x.foo();`) is excluded by the
# absence of parens before the semicolon.
DISCARDED_PATH_ITEM = re.compile(r"\blet\s+_\s*=\s*[a-zA-Z_][a-zA-Z0-9_]*(?:::(?:<[^>]*>|[a-zA-Z_][a-zA-Z0-9_]*))+\s*;")
COMPILE_TIME_CHECK = re.compile(r"\bfn\s+_[a-zA-Z0-9_]*\s*(?:<[^>]*>)?\s*\(")
TEST_ATTR = re.compile(r"#\[(?:tokio::)?test[\](]")
TEST_CASE_ATTR = re.compile(r"#\[test_case")
FN_LINE = re.compile(r"^\s*(?:pub\s+)?(?:async\s+)?fn\s+([a-zA-Z0-9_]+)")
def extract_body(text: str) -> str:
"""Return what is between the outermost braces of a scanned function."""
start = text.find("{")
end = text.rfind("}")
if start == -1 or end <= start:
return text
return text[start + 1 : end]
def scan_file(path: Path):
try:
lines = path.read_text(encoding="utf-8").split("\n")
@@ -133,17 +105,7 @@ def scan_file(path: Path):
break
k += 1
text = "\n".join(body)
# The attribute block carries verification too: `#[should_panic(expected
# = "...")]` makes the panic message the assertion.
attr_text = "\n".join(attrs)
inner = extract_body(text)
delegates = (
DELEGATION.search(text)
or SINGLE_CALL_BODY.match(inner)
or (SIGNATURE_GUARD.search(inner) and DISCARDED_BINDING.search(inner))
or DISCARDED_PATH_ITEM.search(inner)
)
if not VERIFY_SIGNALS.search(text) and not VERIFY_SIGNALS.search(attr_text) and not delegates and not COMPILE_TIME_CHECK.search(text):
if not VERIFY_SIGNALS.search(text) and not DELEGATION.search(text) and not COMPILE_TIME_CHECK.search(text):
print(f"{path}:{j + 1}: {name}")
i = k + 1