Compare commits

..

16 Commits

Author SHA1 Message Date
overtrue c2ce237d5a fix(ecstore): restore decommission test imports 2026-08-22 19:19:08 +08:00
overtrue 241271c256 fix(ecstore): satisfy decommission clippy checks 2026-08-22 18:55:44 +08:00
overtrue 3be1511214 style(ecstore): format decommission test imports 2026-08-22 18:44:46 +08:00
overtrue bc1caf0dc7 fix(ecstore): drop stale decommission test import 2026-08-22 18:44:46 +08:00
overtrue ce4a72869d style(ecstore): format decommission supervision tests 2026-08-22 18:44:14 +08:00
overtrue 255943fa43 test(ecstore): cover terminal save retry fencing 2026-08-22 18:43:56 +08:00
overtrue 5583e8373c fix(ecstore): supervise decommission worker exits 2026-08-22 18:43:56 +08:00
overtrue e1c657652f fix(ecstore): fence decommission canceler cleanup 2026-08-22 18:43:47 +08:00
Zhengchao An 1a3be70d98 fix(ecstore): preserve remote delete error types (#6371) 2026-08-22 17:07:02 +08:00
cxymds 8679570c2a fix(heal): retain displaced task status (#6370) 2026-08-22 08:23:40 +00:00
Zhengchao An 7143697a5f fix(rpc): bound internode concurrency under multipart load (#6368) 2026-08-22 06:42:10 +00:00
cxymds a34310a58f fix(ecstore): fail closed on unresolved decommission entries (#6367)
* fix(ecstore): fail closed on unresolved decommission entries

* perf(ecstore): avoid successful listing name clone
2026-08-22 14:12:28 +08:00
cxymds 2e60029079 perf(ecstore): throttle decommission checkpoints (#6356) 2026-08-22 13:49:23 +08:00
Zhengchao An 2c3e68ad89 ci: let feature validation jobs finish (#6364) 2026-08-22 04:09:21 +00:00
houseme 2f0918f60b feat(disk): fsync dedicated blocking pool (default-off) (#6366) 2026-08-22 11:24:24 +08:00
Zhengchao An 5b951de2b7 test(ci): bound s3-tests failure logs (#6361) 2026-08-22 02:58:01 +00:00
17 changed files with 2016 additions and 417 deletions
+3 -3
View File
@@ -400,7 +400,7 @@ jobs:
if: github.event_name != 'pull_request' || github.event.action != 'closed'
needs: [ quick-checks ]
runs-on: sm-standard-4
timeout-minutes: 45
timeout-minutes: 90
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
@@ -440,7 +440,7 @@ jobs:
if: github.event_name != 'pull_request' || github.event.action != 'closed'
needs: [ quick-checks ]
runs-on: sm-standard-4
timeout-minutes: 60
timeout-minutes: 90
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
@@ -470,7 +470,7 @@ jobs:
if: github.event_name != 'pull_request' || github.event.action != 'closed'
needs: [ quick-checks ]
runs-on: sm-standard-4
timeout-minutes: 60
timeout-minutes: 90
strategy:
# On a PR, one failing protocol leg is enough to know the PR is not ready,
# so stop the sibling leg instead of paying another ~40 minutes for it.
+7
View File
@@ -57,6 +57,13 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024;
pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
/// Dedicated blocking thread pool for fsync/fdatasync operations.
/// When > 1, fsync operations are isolated from the main blocking pool to
/// prevent device-bound fsync from starving read operations (pread/stat/open).
/// Default 0 means auto (no isolation, use main runtime).
pub const ENV_FSYNC_BLOCKING_THREADS: &str = "RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS";
pub const DEFAULT_FSYNC_BLOCKING_THREADS: usize = 0;
// Dial9 Tokio Telemetry Default values
pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";
+84 -17
View File
@@ -50,10 +50,10 @@ use rustfs_protos::evict_failed_connection;
use rustfs_protos::proto_gen::node_service::RenamePartRequest;
use rustfs_protos::proto_gen::node_service::{
BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest,
DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest,
MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest,
ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
DeleteVersionRequest, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest,
ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest,
ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
WriteMetadataRequest, node_service_client::NodeServiceClient,
};
@@ -112,6 +112,28 @@ const EVENT_REMOTE_DISK_RPC: &str = "remote_disk_rpc";
const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);
fn decode_delete_versions_errors(response: DeleteVersionsResponse, expected_len: usize) -> Vec<Option<Error>> {
if !response.item_errors.is_empty() {
if response.item_errors.len() != expected_len {
return vec![Some(Error::other("malformed delete_versions item errors")); expected_len];
}
return response
.item_errors
.into_iter()
.map(|error| (error.code != 0).then(|| error.into()))
.collect();
}
if response.errors.len() != expected_len {
return vec![Some(Error::other("malformed delete_versions errors")); expected_len];
}
response
.errors
.into_iter()
.map(|error| (!error.is_empty()).then(|| Error::other(error)))
.collect()
}
fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result<SnapshotLeaseToken> {
if !response.success {
return Err(response.error.unwrap_or_default().into());
@@ -2406,8 +2428,6 @@ impl DiskAPI for RemoteDisk {
return errors;
}
// TODO(backlog): replace string errors with typed `StorageError` variants
let result = self
.execute_with_timeout(
|| async {
@@ -2439,17 +2459,7 @@ impl DiskAPI for RemoteDisk {
}
return errors;
}
response
.errors
.iter()
.map(|error| {
if error.is_empty() {
None
} else {
Some(Error::other(error.to_string()))
}
})
.collect()
decode_delete_versions_errors(response, versions.len())
}
#[tracing::instrument(level = "trace", skip_all)]
@@ -3760,6 +3770,63 @@ mod tests {
static INIT: Once = Once::new();
#[test]
fn delete_versions_response_preserves_typed_item_errors() {
let errors = decode_delete_versions_errors(
DeleteVersionsResponse {
success: true,
errors: vec!["file not found".to_string(), String::new()],
error: None,
item_errors: vec![
rustfs_protos::proto_gen::node_service::Error {
code: DiskError::FileNotFound.to_u32(),
error_info: "file not found".to_string(),
},
rustfs_protos::proto_gen::node_service::Error::default(),
],
},
2,
);
assert!(matches!(errors.as_slice(), [Some(DiskError::FileNotFound), None]));
}
#[test]
fn delete_versions_response_accepts_legacy_string_errors() {
let errors = decode_delete_versions_errors(
DeleteVersionsResponse {
success: true,
errors: vec!["legacy error".to_string(), String::new()],
error: None,
item_errors: Vec::new(),
},
2,
);
assert_eq!(errors.len(), 2);
assert_eq!(errors[0].as_ref().map(ToString::to_string).as_deref(), Some("io error legacy error"));
assert!(errors[1].is_none());
}
#[test]
fn delete_versions_response_rejects_misaligned_item_errors() {
let errors = decode_delete_versions_errors(
DeleteVersionsResponse {
success: true,
errors: vec!["file not found".to_string()],
error: None,
item_errors: vec![rustfs_protos::proto_gen::node_service::Error {
code: DiskError::FileNotFound.to_u32(),
error_info: "file not found".to_string(),
}],
},
2,
);
assert_eq!(errors.len(), 2);
assert!(errors.iter().all(Option::is_some));
}
#[test]
fn disk_mutation_digest_marks_rolling_compatibility() {
let mut request = Request::new(());
File diff suppressed because it is too large Load Diff
+42 -4
View File
@@ -315,7 +315,7 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
#[cfg(unix)]
{
let dir = dir.as_ref().to_path_buf();
tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
fsync_spawn_blocking(move || fsync_dir_std(dir)).await?
}
#[cfg(not(unix))]
@@ -683,7 +683,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
#[cfg(test)]
let dir = group.dir.clone();
let dir_file = group.dir_file.clone();
tokio::task::spawn_blocking(move || {
fsync_spawn_blocking(move || {
#[cfg(test)]
{
if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
@@ -1080,6 +1080,44 @@ const TEST_GLOBAL_FILE_SYNCS: usize = 64;
static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
/// configured with >1 threads, isolates device-bound fsync from the main
/// blocking pool so reads (pread/stat/open) are not starved. `None` means
/// fall back to the main runtime (zero behavior change).
static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(|| {
let threads =
rustfs_utils::get_env_usize(rustfs_config::ENV_FSYNC_BLOCKING_THREADS, rustfs_config::DEFAULT_FSYNC_BLOCKING_THREADS);
if threads <= 1 {
return None;
}
let mut builder = tokio::runtime::Builder::new_multi_thread();
builder
.worker_threads(num_cpus::get().min(8))
.max_blocking_threads(threads)
.thread_name("rustfs-fsync")
.thread_stack_size(512 * 1024)
.enable_all();
match builder.build() {
Ok(rt) => {
tracing::info!(threads, "fsync dedicated blocking pool enabled");
Some(rt)
}
Err(err) => {
tracing::warn!(%err, "failed to build fsync runtime, falling back to main pool");
None
}
}
});
/// Spawn a blocking task on the fsync-dedicated runtime if configured,
/// otherwise fall back to the main tokio blocking pool.
fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
match FSYNC_RUNTIME.as_ref() {
Some(rt) => rt.spawn_blocking(f),
None => tokio::task::spawn_blocking(f),
}
}
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
type NamespaceMutationLock = AsyncMutex<()>;
@@ -1217,7 +1255,7 @@ where
F: FnOnce() -> io::Result<T> + Send + 'static,
{
let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
let result = tokio::task::spawn_blocking(move || {
let result = fsync_spawn_blocking(move || {
let _disk_permit = disk_permit;
work()
})
@@ -2146,7 +2184,7 @@ async fn run_blocking_namespace_file_sync_operation_with_global<T: Send + 'stati
wait_started,
);
let disk_permit = admission.disk_permit.clone();
let result = tokio::task::spawn_blocking(move || {
let result = fsync_spawn_blocking(move || {
let _lease = lease;
let _disk_permit = disk_permit;
operation()
+2 -2
View File
@@ -33,7 +33,7 @@ use crate::bucket::utils::check_put_object_part_args;
use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
use crate::cluster::rpc::{RemoteClient, S3PeerSys};
use crate::config::storageclass;
use crate::core::pools::PoolMeta;
use crate::core::pools::{DecommissionCanceler, PoolMeta};
use crate::disk::endpoint::{Endpoint, EndpointType};
use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions};
use crate::error::{Error, Result};
@@ -176,7 +176,7 @@ pub struct ECStore {
// pub local_disks: Vec<DiskStore>,
pub pool_meta: RwLock<PoolMeta>,
pub rebalance_meta: RwLock<Option<RebalanceMeta>>,
pub decommission_cancelers: RwLock<Vec<Option<CancellationToken>>>,
pub decommission_cancelers: RwLock<Vec<Option<DecommissionCanceler>>>,
/// Serializes rebalance/decommission start transitions.
///
/// Lock order: acquire `start_gate` before `pool_meta`, `rebalance_meta`,
+54
View File
@@ -1640,6 +1640,60 @@ mod tests {
assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
}
#[tokio::test]
async fn test_process_query_request_reports_displaced_terminal_detail() {
let heal_manager = Arc::new(HealManager::new(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 1,
..HealConfig::default()
}),
));
let mut displaced = HealRequest::new(
HealType::Bucket {
bucket: "displaced-channel".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
displaced.id = "displaced-channel-task".to_string();
let displaced_id = displaced.id.clone();
heal_manager
.submit_heal_request(displaced)
.await
.expect("initial channel task should queue");
heal_manager
.submit_heal_request(HealRequest::new(
HealType::Bucket {
bucket: "successor-channel".to_string(),
},
HealOptions::default(),
HealPriority::High,
))
.await
.expect("successor channel task should displace the initial task");
let processor = HealChannelProcessor::new(heal_manager);
let (tx, rx) = oneshot::channel();
processor
.process_query_request("displaced-channel".to_string(), displaced_id, None, tx)
.await
.expect("displaced query should process");
let response = rx
.await
.expect("query response should be returned")
.expect("displaced query should remain successful");
let payload: serde_json::Value = serde_json::from_slice(response.data.as_deref().expect("status payload should exist"))
.expect("status payload should be json");
assert_eq!(payload["summary"], "stopped");
assert!(
response
.error
.as_deref()
.is_some_and(|detail| detail.contains("reason=displaced"))
);
}
#[tokio::test]
async fn test_process_query_request_reports_running_for_queued_task() {
let heal_manager = create_test_heal_manager();
+105 -10
View File
@@ -40,6 +40,7 @@ use tracing::{debug, error, info, warn};
use super::{DiskError, Endpoint, HealDiskExt as _, local_disk_map_read};
const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
const DISPLACED_HEAL_REASON: &str = "reason=displaced; retry_hint=submit_again";
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner";
const LOG_SUBSYSTEM_MANAGER: &str = "manager";
@@ -120,26 +121,30 @@ struct MrfRepairNoticeTarget {
version_id: Option<[u8; 16]>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[derive(Debug, Clone)]
struct HealAdmissionDecision {
result: HealAdmissionResult,
displaced_task_id: Option<String>,
displaced_request: Option<HealRequest>,
}
impl HealAdmissionDecision {
const fn new(result: HealAdmissionResult) -> Self {
Self {
result,
displaced_task_id: None,
displaced_request: None,
}
}
fn accepted_with_displacement(displaced_task_id: String) -> Self {
fn accepted_with_displacement(displaced_request: HealRequest) -> Self {
Self {
result: HealAdmissionResult::Accepted,
displaced_task_id: Some(displaced_task_id),
displaced_request: Some(displaced_request),
}
}
fn displaced_task_id(&self) -> Option<&str> {
self.displaced_request.as_ref().map(|request| request.id.as_str())
}
}
fn lock_mrf_repair_notice_targets(
@@ -151,6 +156,55 @@ fn lock_mrf_repair_notice_targets(
}
}
fn lock_displaced_terminals(
registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
) -> StdMutexGuard<'_, HashMap<String, Arc<CompletedHealStatus>>> {
match registry.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
fn record_displaced_terminal(
registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
request: &HealRequest,
) -> Arc<CompletedHealStatus> {
let terminal = Arc::new(CompletedHealStatus {
heal_type: request.heal_type.clone(),
status: HealTaskStatus::Failed {
error: format!("heal task displaced by a higher-priority request ({DISPLACED_HEAL_REASON})"),
},
result_items_truncated: false,
completed_at: SystemTime::now(),
seqed_items: Vec::new(),
next_seq: 0,
min_seq: 0,
});
let mut terminals = lock_displaced_terminals(registry);
prune_completed_heal_statuses(&mut terminals);
terminals.insert(request.id.clone(), Arc::clone(&terminal));
terminal
}
async fn remove_displaced_task_aliases(
aliases: &Arc<Mutex<HashMap<String, HealTaskAlias>>>,
terminals: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
task_id: &str,
terminal: &Arc<CompletedHealStatus>,
) {
let mut aliases = aliases.lock().await;
let alias_ids = aliases
.iter()
.filter_map(|(alias_id, alias)| (alias.task_id == task_id).then_some(alias_id.clone()))
.collect::<Vec<_>>();
let mut displaced_terminals = lock_displaced_terminals(terminals);
prune_completed_heal_statuses(&mut displaced_terminals);
for alias_id in alias_ids {
displaced_terminals.insert(alias_id, Arc::clone(terminal));
}
aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
}
async fn remove_task_aliases_for_task(registry: &Arc<Mutex<HashMap<String, HealTaskAlias>>>, task_id: &str) {
registry
.lock()
@@ -618,6 +672,14 @@ pub struct HealManager {
/// are shared so the lookup helper can hand a completed entry to a
/// caller without cloning the retained result window.
completed_heals: Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
/// Terminals for requests removed by priority displacement. An Accepted
/// task ID remains queryable for the same process lifetime and the normal
/// ten-minute status TTL; clients should treat `reason=displaced` as a
/// terminal result and submit a fresh request. This sidecar is synchronous
/// so admission can publish the terminal while the queue transition is
/// still under its lock, without awaiting another tokio lock. Queue state
/// is process-local, so this guarantee does not extend across restart.
displaced_terminals: Arc<StdMutex<HashMap<String, Arc<CompletedHealStatus>>>>,
/// Client tokens merged into an existing task id.
task_aliases: Arc<Mutex<HashMap<String, HealTaskAlias>>>,
/// Heal tasks waiting for a retry backoff to expire.
@@ -659,6 +721,7 @@ struct HealQueueContext<'a> {
heal_queue: &'a Arc<Mutex<PriorityHealQueue>>,
active_heals: &'a Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
completed_heals: &'a Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
displaced_terminals: &'a Arc<StdMutex<HashMap<String, Arc<CompletedHealStatus>>>>,
task_aliases: &'a Arc<Mutex<HashMap<String, HealTaskAlias>>>,
retrying_heals: &'a Arc<Mutex<HashMap<String, RetryingHeal>>>,
mrf_repair_notice_targets: &'a Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>,
@@ -874,7 +937,7 @@ impl HealManager {
result = "accepted_by_displacement",
"Heal queue request accepted by displacement"
});
return HealAdmissionDecision::accepted_with_displacement(displaced.id);
return HealAdmissionDecision::accepted_with_displacement(displaced);
}
demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", {
@@ -1105,6 +1168,7 @@ impl HealManager {
active_heals: Arc::new(Mutex::new(HashMap::new())),
heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())),
completed_heals: Arc::new(Mutex::new(HashMap::new())),
displaced_terminals: Arc::new(StdMutex::new(HashMap::new())),
task_aliases: Arc::new(Mutex::new(HashMap::new())),
retrying_heals: Arc::new(Mutex::new(HashMap::new())),
mrf_repair_notice_targets: Arc::new(StdMutex::new(HashMap::new())),
@@ -1209,6 +1273,10 @@ impl HealManager {
active_heals.clear();
publish_active_heal_count(&active_heals);
self.completed_heals.lock().await.clear();
// Do not let the synchronous guard live across the following async lock.
{
lock_displaced_terminals(&self.displaced_terminals).clear();
}
self.task_aliases.lock().await.clear();
self.retrying_heals.lock().await.clear();
lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).clear();
@@ -1459,7 +1527,11 @@ impl HealManager {
task_id = queued_id.to_owned();
}
let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
let displaced_task_id = admission_decision.displaced_task_id;
let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
let displaced_terminal = admission_decision
.displaced_request
.as_ref()
.map(|request| record_displaced_terminal(&self.displaced_terminals, request));
if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
&& let Some(target) = mrf_notice_target
{
@@ -1473,8 +1545,12 @@ impl HealManager {
drop(queue);
drop(active_heals);
if let Some(displaced_task_id) = displaced_task_id {
self.remove_aliases_for_task(&displaced_task_id).await;
if let (Some(displaced_task_id), Some(displaced_terminal)) = (displaced_task_id, displaced_terminal) {
// The queue has already removed the displaced request, so the
// synchronous terminal sidecar was published before aliases and
// MRF ownership are cleaned up.
remove_displaced_task_aliases(&self.task_aliases, &self.displaced_terminals, &displaced_task_id, &displaced_terminal)
.await;
}
if should_notify {
@@ -1549,6 +1625,15 @@ impl HealManager {
}
}
if terminal_completed.is_none() {
let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
prune_completed_heal_statuses(&mut displaced_terminals);
terminal_completed = displaced_terminals
.get(canonical_task_id)
.filter(|terminal| matches_path(&terminal.heal_type))
.cloned();
}
match terminal_completed {
Some(completed) => TaskStateLookup::Completed(completed),
None => TaskStateLookup::NotFound,
@@ -1669,9 +1754,19 @@ impl HealManager {
let mut completed_heals = self.completed_heals.lock().await;
prune_completed_heal_statuses(&mut completed_heals);
completed_heals
if completed_heals
.values()
.any(|completed| heal_type_matches_path(&completed.heal_type, heal_path))
{
return true;
}
drop(completed_heals);
let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
prune_completed_heal_statuses(&mut displaced_terminals);
displaced_terminals
.values()
.any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path))
}
/// Get task progress
+15 -2
View File
@@ -21,6 +21,7 @@ impl HealManager {
let heal_queue = self.heal_queue.clone();
let active_heals = self.active_heals.clone();
let task_aliases = self.task_aliases.clone();
let displaced_terminals = self.displaced_terminals.clone();
let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
let storage = self.storage.clone();
let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
@@ -481,6 +482,10 @@ impl HealManager {
let admission = admission_decision.result;
let should_notify =
matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
let displaced_terminal = admission_decision
.displaced_request
.as_ref()
.map(|request| record_displaced_terminal(&displaced_terminals, request));
if matches!(admission, HealAdmissionResult::Accepted)
&& let Some(anchor) = recovery_anchor
{
@@ -491,8 +496,16 @@ impl HealManager {
}
drop(queue);
drop(config);
if let Some(displaced_task_id) = admission_decision.displaced_task_id {
remove_task_aliases_for_task(&task_aliases, &displaced_task_id).await;
if let (Some(displaced_task_id), Some(displaced_terminal)) =
(admission_decision.displaced_task_id().map(ToOwned::to_owned), displaced_terminal)
{
remove_displaced_task_aliases(
&task_aliases,
&displaced_terminals,
&displaced_task_id,
&displaced_terminal,
)
.await;
lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id);
}
if matches!(admission, HealAdmissionResult::Accepted) {
+25 -3
View File
@@ -21,6 +21,7 @@ impl HealManager {
let heal_queue = self.heal_queue.clone();
let active_heals = self.active_heals.clone();
let completed_heals = self.completed_heals.clone();
let displaced_terminals = self.displaced_terminals.clone();
let task_aliases = self.task_aliases.clone();
let retrying_heals = self.retrying_heals.clone();
let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
@@ -53,6 +54,7 @@ impl HealManager {
heal_queue: &heal_queue,
active_heals: &active_heals,
completed_heals: &completed_heals,
displaced_terminals: &displaced_terminals,
task_aliases: &task_aliases,
retrying_heals: &retrying_heals,
mrf_repair_notice_targets: &mrf_repair_notice_targets,
@@ -71,6 +73,7 @@ impl HealManager {
heal_queue: &heal_queue,
active_heals: &active_heals,
completed_heals: &completed_heals,
displaced_terminals: &displaced_terminals,
task_aliases: &task_aliases,
retrying_heals: &retrying_heals,
mrf_repair_notice_targets: &mrf_repair_notice_targets,
@@ -98,6 +101,7 @@ impl HealManager {
heal_queue,
active_heals,
completed_heals,
displaced_terminals,
task_aliases,
retrying_heals,
mrf_repair_notice_targets,
@@ -183,6 +187,7 @@ impl HealManager {
let active_heals_clone = active_heals.clone();
let heal_queue_clone = heal_queue.clone();
let completed_heals_clone = completed_heals.clone();
let displaced_terminals_clone = displaced_terminals.clone();
let task_aliases_clone = task_aliases.clone();
let retrying_heals_clone = retrying_heals.clone();
let mrf_repair_notice_targets_clone = mrf_repair_notice_targets.clone();
@@ -363,6 +368,7 @@ impl HealManager {
let retry_heal_queue = heal_queue_clone.clone();
let retrying_heals_for_spawn = retrying_heals_clone.clone();
let retry_task_aliases = task_aliases_clone.clone();
let retry_displaced_terminals = displaced_terminals_clone.clone();
let retry_mrf_repair_notice_targets = mrf_repair_notice_targets_clone.clone();
let retry_completed_heals = completed_heals_clone.clone();
let retry_notify = notify_clone.clone();
@@ -430,6 +436,14 @@ impl HealManager {
let admission = admission_decision.result;
let should_notify = matches!(admission, HealAdmissionResult::Accepted)
&& retry_config.event_driven_scheduler_enable;
// Publish the terminal synchronously while the
// queue transition is protected. The subsequent
// queue -> retrying handoff retains the lock order
// used by operations_snapshot.
let displaced_terminal = admission_decision
.displaced_request
.as_ref()
.map(|request| record_displaced_terminal(&retry_displaced_terminals, request));
match admission {
HealAdmissionResult::Accepted => {
// Transfer ownership while holding queue -> retrying,
@@ -437,10 +451,18 @@ impl HealManager {
#[cfg(test)]
pause_retry_ownership_transition(&retry_request_id, true).await;
retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
let displaced_task_id = admission_decision.displaced_task_id;
let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
drop(queue);
if let Some(displaced_task_id) = displaced_task_id {
remove_task_aliases_for_task(&retry_task_aliases, &displaced_task_id).await;
if let (Some(displaced_task_id), Some(displaced_terminal)) =
(displaced_task_id, displaced_terminal)
{
remove_displaced_task_aliases(
&retry_task_aliases,
&retry_displaced_terminals,
&displaced_task_id,
&displaced_terminal,
)
.await;
remove_mrf_repair_notice_targets(
&retry_mrf_repair_notice_targets,
&displaced_task_id,
+262 -1
View File
@@ -84,6 +84,7 @@ async fn process_manager_queue_once(manager: &HealManager) {
heal_queue: &manager.heal_queue,
active_heals: &manager.active_heals,
completed_heals: &manager.completed_heals,
displaced_terminals: &manager.displaced_terminals,
task_aliases: &manager.task_aliases,
retrying_heals: &manager.retrying_heals,
mrf_repair_notice_targets: &manager.mrf_repair_notice_targets,
@@ -2778,7 +2779,10 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
HealAdmissionResult::Accepted
);
assert_eq!(manager.get_queue_length().await, 1);
assert!(matches!(manager.get_task_status(&low_id).await, Err(Error::TaskNotFound { .. })));
assert!(matches!(
manager.get_task_status(&low_id).await,
Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
));
assert_eq!(
manager
.get_task_status(&high_id)
@@ -2788,6 +2792,263 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
);
}
#[tokio::test]
async fn displaced_task_remains_queryable() {
let manager = HealManager::new(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 1,
..HealConfig::default()
}),
);
let mut displaced = HealRequest::new(
HealType::Bucket {
bucket: "displaced-bucket".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
displaced.id = "displaced-task".to_string();
let displaced_id = displaced.id.clone();
manager
.submit_heal_request(displaced)
.await
.expect("displaced request should queue");
let successor = HealRequest::new(
HealType::Bucket {
bucket: "successor-bucket".to_string(),
},
HealOptions::default(),
HealPriority::High,
);
manager
.submit_heal_request(successor)
.await
.expect("successor should displace low work");
let report = manager
.get_task_report(&displaced_id)
.await
.expect("displaced report should remain queryable");
assert!(matches!(report.status, HealTaskStatus::Failed { ref error } if error.contains("reason=displaced")));
}
#[tokio::test]
async fn displaced_archive_failure_keeps_queryable_terminal() {
let manager = HealManager::new(Arc::new(MockStorage), None);
let mut request = HealRequest::new(
HealType::Bucket {
bucket: "archive-failure".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
request.id = "archive-failure-task".to_string();
let request_id = request.id.clone();
// The synchronous sidecar is the authoritative fallback when the normal
// completed-task archive has no entry (the failure window that must not
// turn an Accepted ID into NotFound).
record_displaced_terminal(&manager.displaced_terminals, &request);
assert!(manager.completed_heals.lock().await.is_empty());
assert!(matches!(
manager.get_task_status(&request_id).await,
Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
));
}
#[tokio::test]
async fn scheduler_retry_displacement_keeps_evicted_task_queryable() {
let manager = Arc::new(HealManager::new(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 1,
event_driven_scheduler_enable: false,
..HealConfig::default()
}),
));
let mut retry_request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
retry_request.priority = HealPriority::High;
let retry_id = retry_request.id.clone();
manager
.submit_heal_request(retry_request)
.await
.expect("retry request should queue");
// Process exactly one queue cycle so the retry task is spawned without a
// background scheduler consuming the filler request before the retry wakes.
process_manager_queue_once(&manager).await;
tokio::time::timeout(Duration::from_secs(1), async {
loop {
if manager.retrying_heals.lock().await.contains_key(&retry_id) {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("retry request should enter backoff");
let filler = HealRequest::new(
HealType::Bucket {
bucket: "retry-displaced-filler".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
let filler_id = filler.id.clone();
manager
.submit_heal_request(filler)
.await
.expect("filler request should occupy the queue");
tokio::time::timeout(Duration::from_secs(5), async {
loop {
if matches!(
manager.get_task_status(&filler_id).await,
Ok(HealTaskStatus::Failed { ref error }) if error.contains("reason=displaced")
) {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("retry admission should displace the filler request");
assert_eq!(manager.get_queue_length().await, 1);
assert_eq!(
manager.get_task_status(&retry_id).await.expect("retry should be queued"),
HealTaskStatus::Pending
);
}
#[tokio::test]
async fn concurrent_displacers_produce_one_terminal_generation() {
let manager = Arc::new(HealManager::new(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 1,
..HealConfig::default()
}),
));
let mut displaced = HealRequest::new(
HealType::Bucket {
bucket: "concurrent-displaced".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
displaced.id = "concurrent-displaced-task".to_string();
let displaced_id = displaced.id.clone();
manager
.submit_heal_request(displaced)
.await
.expect("initial request should queue");
let first = HealRequest::new(
HealType::Bucket {
bucket: "concurrent-successor-a".to_string(),
},
HealOptions::default(),
HealPriority::High,
);
let second = HealRequest::new(
HealType::Bucket {
bucket: "concurrent-successor-b".to_string(),
},
HealOptions::default(),
HealPriority::High,
);
let (first_result, second_result) = tokio::join!(manager.submit_heal_request(first), manager.submit_heal_request(second));
let accepted = [&first_result, &second_result]
.into_iter()
.filter(|result| matches!(result, Ok(HealAdmissionResult::Accepted)))
.count();
assert_eq!(accepted, 1, "exactly one concurrent displacer should win the full queue");
assert!(
first_result.is_ok() && second_result.is_ok(),
"the losing request should receive a typed Full result"
);
let terminals = lock_displaced_terminals(&manager.displaced_terminals);
assert_eq!(terminals.len(), 1);
assert!(terminals.contains_key(&displaced_id));
}
#[tokio::test]
async fn successor_chain_is_bounded_and_authorized() {
let manager = HealManager::new(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 1,
..HealConfig::default()
}),
);
let mut original = HealRequest::new(
HealType::Bucket {
bucket: "authorized-original".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
original.id = "authorized-original-task".to_string();
let original_id = original.id.clone();
manager.submit_heal_request(original).await.expect("original should queue");
let mut duplicate = HealRequest::new(
HealType::Bucket {
bucket: "authorized-original".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
duplicate.id = "authorized-duplicate-task".to_string();
let duplicate_id = duplicate.id.clone();
manager
.submit_heal_request(duplicate)
.await
.expect("same-target duplicate should merge");
let successor = HealRequest::new(
HealType::Bucket {
bucket: "authorized-successor".to_string(),
},
HealOptions::default(),
HealPriority::High,
);
let successor_id = successor.id.clone();
manager.submit_heal_request(successor).await.expect("successor should queue");
assert!(manager.task_aliases.lock().await.is_empty());
assert!(matches!(manager.get_task_status(&original_id).await, Ok(HealTaskStatus::Failed { .. })));
assert!(matches!(manager.get_task_status(&duplicate_id).await, Ok(HealTaskStatus::Failed { .. })));
assert_eq!(
manager
.get_task_status(&successor_id)
.await
.expect("successor should remain queued"),
HealTaskStatus::Pending
);
}
#[tokio::test]
async fn displaced_terminal_expires_after_bounded_ttl() {
let manager = HealManager::new(Arc::new(MockStorage), None);
let mut request = HealRequest::new(
HealType::Bucket {
bucket: "expires".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
request.id = "expires-task".to_string();
let request_id = request.id.clone();
record_displaced_terminal(&manager.displaced_terminals, &request);
{
let mut terminals = lock_displaced_terminals(&manager.displaced_terminals);
let entry =
Arc::get_mut(terminals.get_mut(&request_id).expect("terminal should be retained")).expect("test owns terminal entry");
entry.completed_at = SystemTime::now() - KEEP_HEAL_TASK_STATUS_DURATION - Duration::from_secs(1);
}
assert!(matches!(manager.get_task_status(&request_id).await, Err(Error::TaskNotFound { .. })));
}
#[tokio::test]
async fn test_displacing_registered_mrf_task_drops_notice_ownership() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -722,6 +722,10 @@ pub struct DeleteVersionsResponse {
pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
#[prost(message, optional, tag = "3")]
pub error: ::core::option::Option<Error>,
/// Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
/// when present and fall back to strings for peers that predate this field. Code zero means success.
#[prost(message, repeated, tag = "4")]
pub item_errors: ::prost::alloc::vec::Vec<Error>,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct ReadMultipleRequest {
+4
View File
@@ -2106,6 +2106,9 @@ pub enum ChannelClass {
Bulk,
}
// Keep multiplexed unary RPCs below h2's per-connection small-frame budget.
const INTERNODE_RPC_CONCURRENCY_LIMIT: usize = 64;
/// Whether control/bulk channel isolation is enabled (env-gated, default off for safe rollout).
fn channel_isolation_enabled() -> bool {
rustfs_utils::get_env_bool(
@@ -2188,6 +2191,7 @@ async fn build_channel(dial_addr: &str, cache_key: &str) -> Result<Channel, Box<
let mut connector = Endpoint::from_shared(dial_addr.to_string())?
// Fast connection timeout for dead peer detection
.connect_timeout(connect_timeout)
.concurrency_limit(INTERNODE_RPC_CONCURRENCY_LIMIT)
// TCP-level keepalive - OS will probe connection
.tcp_keepalive(Some(tcp_keepalive))
// Disable Nagle so latency-sensitive control-plane RPCs (locks/health) are not batched
+3
View File
@@ -493,6 +493,9 @@ message DeleteVersionsResponse {
bool success = 1;
repeated string errors = 2;
optional Error error = 3;
// Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
// when present and fall back to strings for peers that predate this field. Code zero means success.
repeated Error item_errors = 4;
}
message ReadMultipleRequest {
+43 -11
View File
@@ -146,6 +146,29 @@ fn encode_file_info_msgpack(value: &FileInfo) -> std::result::Result<Vec<u8>, Di
encode_msgpack_with_capacity(value, "FileInfo", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT)
}
fn encode_delete_versions_errors(disk_errors: Vec<Option<DiskError>>) -> (Vec<String>, Vec<Error>) {
let mut errors = Vec::with_capacity(disk_errors.len());
let mut item_errors = Vec::with_capacity(disk_errors.len());
for error in disk_errors {
match error {
Some(error) => {
let code = match &error {
DiskError::Io(source) if source.kind() == std::io::ErrorKind::NotFound => DiskError::FileNotFound.to_u32(),
_ => error.to_u32(),
};
let error_info = error.to_string();
errors.push(error_info.clone());
item_errors.push(Error { code, error_info });
}
None => {
errors.push(String::new());
item_errors.push(Error::default());
}
}
}
(errors, item_errors)
}
fn encode_msgpack_named<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map();
value
@@ -552,6 +575,7 @@ impl NodeService {
success: false,
errors: Vec::new(),
error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
item_errors: Vec::new(),
}));
}
};
@@ -563,30 +587,26 @@ impl NodeService {
success: false,
errors: Vec::new(),
error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
item_errors: Vec::new(),
}));
}
};
let errors = disk
.delete_versions(&request.volume, versions, opts)
.await
.into_iter()
.map(|error| match error {
Some(e) => e.to_string(),
None => "".to_string(),
})
.collect();
let (errors, item_errors) =
encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await);
Ok(Response::new(DeleteVersionsResponse {
success: true,
errors,
error: None,
item_errors,
}))
} else {
Ok(Response::new(DeleteVersionsResponse {
success: false,
errors: Vec::new(),
error: Some(DiskError::other("cannot find disk".to_string()).into()),
item_errors: Vec::new(),
}))
}
}
@@ -1612,8 +1632,8 @@ impl NodeService {
mod tests {
use super::{
compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info,
encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named,
encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
encode_batch_read_version_response_payloads, encode_delete_versions_errors, encode_file_info_msgpack, encode_msgpack,
encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
};
use crate::storage::rpc::node_service::make_server;
use crate::storage::storage_api::ReadMultipleResp;
@@ -1632,6 +1652,18 @@ mod tests {
count: u32,
}
#[test]
fn delete_versions_response_dual_writes_typed_item_errors() {
let raw_not_found = super::DiskError::Io(std::io::Error::from(std::io::ErrorKind::NotFound));
let (errors, item_errors) = encode_delete_versions_errors(vec![Some(raw_not_found), None]);
assert!(errors[0].starts_with("io error "));
assert!(errors[1].is_empty());
assert_eq!(item_errors[0].code, super::DiskError::FileNotFound.to_u32());
assert_eq!(item_errors[0].error_info, errors[0]);
assert_eq!(item_errors[1].code, 0);
}
#[tokio::test]
#[serial]
async fn handle_read_version_records_attribution_for_missing_disk() {
+26 -1
View File
@@ -206,6 +206,13 @@ def check_runner_selection(root: Path) -> list[str]:
return errors
def check_s3_tests_runner(root: Path) -> list[str]:
runner = (root / "scripts/s3-tests/run.sh").read_text()
if "--showlocals" in runner:
return ["scripts/s3-tests/run.sh: pytest failure diagnostics must not dump local values"]
return []
def profile_selection(root: Path, profile: str) -> str:
if not re.fullmatch(r"e2e-[a-z0-9-]+", profile):
raise ValueError(f"invalid e2e profile name: {profile}")
@@ -272,6 +279,7 @@ def validate(root: Path) -> list[str]:
errors.extend(check_e2e_modules(root))
errors.extend(check_fuzz_targets(root))
errors.extend(check_runner_selection(root))
errors.extend(check_s3_tests_runner(root))
errors.extend(check_profile_definitions(root))
return errors
@@ -341,6 +349,23 @@ class SelfTests(unittest.TestCase):
)
self.assertEqual(len(check_fuzz_targets(root)), 1)
def test_s3_runner_rejects_unbounded_failure_locals(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
runner = root / "scripts/s3-tests/run.sh"
runner.parent.mkdir(parents=True)
runner.write_text("tox -- -vv -ra --tb=long\n")
self.assertEqual(check_s3_tests_runner(root), [])
runner.write_text("tox -- -vv -ra --showlocals --tb=long\n")
self.assertEqual(len(check_s3_tests_runner(root)), 1)
with (
mock.patch(__name__ + ".check_e2e_modules", return_value=[]),
mock.patch(__name__ + ".check_fuzz_targets", return_value=[]),
mock.patch(__name__ + ".check_runner_selection", return_value=[]),
mock.patch(__name__ + ".check_profile_definitions", return_value=[]),
):
self.assertEqual(len(validate(root)), 1)
def test_profile_listing_enforces_selection(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
@@ -411,7 +436,7 @@ def main() -> int:
for error in errors:
print(f"ERROR: {error}", file=sys.stderr)
return 1
print("OK: e2e modules, runner selection, fuzz matrices, and profile guards are wired")
print("OK: e2e modules, runner selection, fuzz matrices, profiles, and bounded diagnostics are wired")
return 0
+2 -1
View File
@@ -1028,10 +1028,11 @@ else
fi
# Run tests from s3tests/functional
# Failure locals can contain multi-MiB request bodies; keep tracebacks without expanding local values.
set +e
S3TEST_CONF="${CONF_OUTPUT_PATH}" \
tox -- \
-vv -ra --showlocals --tb=long \
-vv -ra --tb=long \
--maxfail="${MAXFAIL}" \
--timeout="${TEST_TIMEOUT}" \
--junitxml="${ARTIFACTS_DIR}/junit.xml" \