mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 12:26:37 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 overtrue
|
60fb7dc372 | ||
|
cxymds
|
8679570c2a |
@@ -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(());
|
||||
|
||||
@@ -297,16 +297,10 @@ impl ECStore {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::bucket::metadata_sys;
|
||||
use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
|
||||
use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
|
||||
use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
|
||||
use crate::runtime::instance::InstanceContext;
|
||||
use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
|
||||
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
|
||||
use crate::disk::{DiskOption, format::FormatV3, new_disk};
|
||||
use crate::layout::endpoints::{Endpoints, PoolEndpoints};
|
||||
use crate::store::init_format::{load_format_erasure, save_format_file};
|
||||
use crate::store::init_local_disks_with_instance_ctx;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
|
||||
let format = FormatV3::new(1, 1);
|
||||
@@ -353,51 +347,6 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) {
|
||||
let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
|
||||
let mut pool_endpoints = Vec::new();
|
||||
for pool_index in 0..2 {
|
||||
let mut endpoints = Vec::new();
|
||||
for disk_index in 0..4 {
|
||||
let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
|
||||
tokio::fs::create_dir_all(&disk_path)
|
||||
.await
|
||||
.expect("multi-pool heal test disk should be created");
|
||||
let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8"))
|
||||
.expect("test endpoint should parse");
|
||||
endpoint.set_pool_index(pool_index);
|
||||
endpoint.set_set_index(0);
|
||||
endpoint.set_disk_index(disk_index);
|
||||
endpoints.push(endpoint);
|
||||
}
|
||||
pool_endpoints.push(PoolEndpoints {
|
||||
legacy: false,
|
||||
set_count: 1,
|
||||
drives_per_set: 4,
|
||||
endpoints: Endpoints::from(endpoints),
|
||||
cmd_line: format!("heal-owner-pool-{pool_index}"),
|
||||
platform: "test".to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
let endpoint_pools = EndpointServerPools::from(pool_endpoints);
|
||||
let instance_ctx = Arc::new(InstanceContext::new());
|
||||
init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
|
||||
.await
|
||||
.expect("multi-pool local disks should initialize");
|
||||
let shutdown = CancellationToken::new();
|
||||
let store = ECStore::new_with_instance_ctx(
|
||||
"127.0.0.1:0".parse().expect("test address should parse"),
|
||||
endpoint_pools,
|
||||
shutdown.clone(),
|
||||
instance_ctx,
|
||||
)
|
||||
.await
|
||||
.expect("multi-pool test store should initialize");
|
||||
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
(temp_dir, store, shutdown)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn heal_object_pool_scope_selects_only_requested_pool() {
|
||||
let store = minimal_heal_store().await;
|
||||
@@ -557,229 +506,6 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn unscoped_heal_object_suspended_owner_semantics() {
|
||||
let (_temp_dir, store, shutdown) = multi_pool_heal_store().await;
|
||||
let bucket = format!("heal-owner-{}", Uuid::new_v4().simple());
|
||||
let active_object = "active-owner";
|
||||
let suspended_only_object = "suspended-only";
|
||||
let duplicate_object = "duplicate-owner";
|
||||
let marker_object = "marker-owner";
|
||||
let quorum_object = "quorum-owner";
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("bucket should be created in all pools");
|
||||
|
||||
let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec());
|
||||
store.pools[0]
|
||||
.put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("active owner object should be written");
|
||||
let active_disks = store.pools[0].disk_set[0].disks.read().await.clone();
|
||||
let missing_active_disk = active_disks[0].clone().expect("active disk should be online");
|
||||
missing_active_disk
|
||||
.delete(
|
||||
&bucket,
|
||||
active_object,
|
||||
DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("active owner shard should be removed for repair");
|
||||
assert!(
|
||||
missing_active_disk.read_xl(&bucket, active_object, false).await.is_err(),
|
||||
"the active owner fixture must start with one missing metadata copy"
|
||||
);
|
||||
|
||||
let mut suspended_reader = PutObjReader::from_vec(b"suspended owner".to_vec());
|
||||
store.pools[1]
|
||||
.put_object(&bucket, suspended_only_object, &mut suspended_reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("suspended owner object should be written");
|
||||
for (pool_index, mod_time) in [1_i64, 2_i64].into_iter().enumerate() {
|
||||
let mut duplicate_reader = PutObjReader::from_vec(format!("duplicate-pool-{pool_index}").into_bytes());
|
||||
store.pools[pool_index]
|
||||
.put_object(
|
||||
&bucket,
|
||||
duplicate_object,
|
||||
&mut duplicate_reader,
|
||||
&ObjectOptions {
|
||||
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("duplicate owner object should be written");
|
||||
}
|
||||
let duplicate_missing_disk = store.pools[0].disk_set[0].disks.read().await[0]
|
||||
.clone()
|
||||
.expect("duplicate active owner disk should be online");
|
||||
duplicate_missing_disk
|
||||
.delete(
|
||||
&bucket,
|
||||
duplicate_object,
|
||||
DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("duplicate active owner shard should be removed for repair");
|
||||
let history_version = Uuid::new_v4();
|
||||
let mut history_reader = PutObjReader::from_vec(b"marker history".to_vec());
|
||||
store.pools[0]
|
||||
.put_object(
|
||||
&bucket,
|
||||
marker_object,
|
||||
&mut history_reader,
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
version_id: Some(history_version.to_string()),
|
||||
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("versioned marker history should be written");
|
||||
store.pools[0]
|
||||
.delete_object(
|
||||
&bucket,
|
||||
marker_object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(2)),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("delete marker should be written");
|
||||
let mut quorum_reader = PutObjReader::from_vec(b"quorum boundary".to_vec());
|
||||
store.pools[0]
|
||||
.put_object(&bucket, quorum_object, &mut quorum_reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("quorum boundary object should be written");
|
||||
{
|
||||
let mut pool_meta = store.pool_meta.write().await;
|
||||
let mut next = PoolMeta::new(&store.pools, &pool_meta);
|
||||
next.pools[1].decommission = Some(PoolDecommissionInfo {
|
||||
start_time: Some(OffsetDateTime::UNIX_EPOCH),
|
||||
..Default::default()
|
||||
});
|
||||
*pool_meta = next;
|
||||
}
|
||||
|
||||
let (_, duplicate_owner) = store
|
||||
.get_latest_object_info_with_idx(&bucket, duplicate_object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("duplicate owner should resolve");
|
||||
assert_eq!(duplicate_owner, 1, "latest duplicate must win when all pools are eligible");
|
||||
let (_, active_duplicate_owner) = store
|
||||
.get_latest_object_info_with_idx(
|
||||
&bucket,
|
||||
duplicate_object,
|
||||
&ObjectOptions {
|
||||
skip_decommissioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("active duplicate owner should resolve");
|
||||
assert_eq!(
|
||||
active_duplicate_owner, 0,
|
||||
"suspended duplicate must be excluded from active owner selection"
|
||||
);
|
||||
let (duplicate_result, duplicate_err) = store
|
||||
.handle_heal_object(&bucket, duplicate_object, "", &HealOpts::default())
|
||||
.await
|
||||
.expect("duplicate owner heal should complete through the production path");
|
||||
assert_eq!(duplicate_result.object, duplicate_object);
|
||||
assert!(duplicate_err.is_none(), "active duplicate should be repaired: {duplicate_err:?}");
|
||||
assert!(
|
||||
duplicate_missing_disk.read_xl(&bucket, duplicate_object, false).await.is_ok(),
|
||||
"production heal must repair the active duplicate owner rather than the suspended owner"
|
||||
);
|
||||
let (marker_info, marker_owner) = store
|
||||
.get_latest_object_info_with_idx(
|
||||
&bucket,
|
||||
marker_object,
|
||||
&ObjectOptions {
|
||||
skip_decommissioned: true,
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("latest delete marker should resolve");
|
||||
assert_eq!(marker_owner, 0);
|
||||
assert!(marker_info.delete_marker, "latest version must preserve delete-marker semantics");
|
||||
|
||||
let (active_result, active_err) = store
|
||||
.handle_heal_object(&bucket, active_object, "", &HealOpts::default())
|
||||
.await
|
||||
.expect("unscoped active-owner heal should complete");
|
||||
assert_eq!(active_result.object, active_object);
|
||||
assert!(active_err.is_none(), "active owner must be selected even with a suspended pool");
|
||||
assert!(
|
||||
missing_active_disk.read_xl(&bucket, active_object, false).await.is_ok(),
|
||||
"active owner heal must write the missing disk metadata: result={active_result:?}, err={active_err:?}"
|
||||
);
|
||||
assert!(
|
||||
store.pools[1]
|
||||
.get_object_info(&bucket, active_object, &ObjectOptions::default())
|
||||
.await
|
||||
.is_err(),
|
||||
"the suspended pool must not be written for an active-owner object"
|
||||
);
|
||||
|
||||
let (suspended_result, suspended_err) = store
|
||||
.handle_heal_object(&bucket, suspended_only_object, "", &HealOpts::default())
|
||||
.await
|
||||
.expect("unscoped suspended-only heal should return a terminal result");
|
||||
assert!(suspended_result.object.is_empty());
|
||||
assert!(matches!(suspended_err, Some(Error::FileNotFound)));
|
||||
assert!(
|
||||
store.pools[1]
|
||||
.get_object_info(&bucket, suspended_only_object, &ObjectOptions::default())
|
||||
.await
|
||||
.is_ok(),
|
||||
"suspended-only data must remain untouched when unscoped heal reports absent"
|
||||
);
|
||||
|
||||
let (_, explicit_err) = store
|
||||
.handle_heal_object(
|
||||
&bucket,
|
||||
suspended_only_object,
|
||||
"",
|
||||
&HealOpts {
|
||||
pool: Some(1),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("explicit suspended-owner heal should return a mapped error");
|
||||
assert!(matches!(explicit_err, Some(Error::SlowDown)));
|
||||
|
||||
let original_quorum_disks = store.pools[0].disk_set[0].disks.read().await.clone();
|
||||
let surviving_quorum_disk = original_quorum_disks[3].clone();
|
||||
*store.pools[0].disk_set[0].disks.write().await = vec![None, None, None, surviving_quorum_disk];
|
||||
let (_, quorum_err) = store
|
||||
.handle_heal_object(&bucket, quorum_object, "", &HealOpts::default())
|
||||
.await
|
||||
.expect("quorum boundary heal should return a mapped result");
|
||||
*store.pools[0].disk_set[0].disks.write().await = original_quorum_disks;
|
||||
assert!(
|
||||
matches!(quorum_err, Some(Error::ErasureReadQuorum)),
|
||||
"quorum-boundary heal must preserve quorum error, got {quorum_err:?}"
|
||||
);
|
||||
shutdown.cancel();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn handle_heal_format_continues_after_a_pool_error() {
|
||||
let canonical_format = FormatV3::new(1, 3);
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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() {
|
||||
|
||||
Reference in New Issue
Block a user