Compare commits

..

3 Commits

Author SHA1 Message Date
houseme 022f1ecce5 fix(ecstore): roll back failed CAS directory fsync
Restore the previous control-file bytes, or remove a newly created file, when the Unix compare-and-update path reaches the rename but then fails to fsync the parent directory. This keeps failed metadata CAS publications from advancing recovery anchors.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
2026-09-07 11:42:41 +08:00
Zhengchao An 672087ec0d ci(docker): move latest on every prerelease until first stable tag (#7341)
Since rc.1 the latest tag has been frozen at 1.0.0-beta.12 because the
docker workflow only allowed alpha/beta prereleases to update latest
(#2732 dropped the rc case). Before 1.0.0 GA, latest is expected to
track the newest test build.

- Prereleases (alpha/beta/rc) now update latest as long as no stable
  vX.Y.Z tag exists on origin, so the rule retires itself at GA.
- Channel tags (alpha/beta/rc) are now always added for prereleases;
  the previous if/elif skipped the channel tag whenever latest was
  created, which is why no :beta tag was ever published.
2026-09-07 11:25:40 +08:00
houseme 9dbeae1b45 chore(deps): refresh scanner batch baseline (#7336)
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-09-07 11:04:34 +08:00
7 changed files with 292 additions and 164 deletions
+26 -8
View File
@@ -121,6 +121,22 @@ jobs:
create_latest=false
source_ref="$GITHUB_SHA"
# Pre-GA policy: until the first stable (vX.Y.Z) tag exists, every
# prerelease (alpha/beta/rc) also moves `latest`, so users pulling
# `latest` get the newest test build. Once a stable tag is published
# this returns false and `latest` follows stable releases only.
prerelease_moves_latest() {
local stable_tags
stable_tags=$(git ls-remote --tags --refs origin 2>/dev/null \
| awk '{print $2}' \
| grep -E '^refs/tags/v?[0-9]+\.[0-9]+\.[0-9]+$' || true)
if [[ -z "$stable_tags" ]]; then
return 0
fi
echo "️ Stable release tag(s) already exist; prereleases no longer update latest"
return 1
}
if [[ "${{ github.event_name }}" == "workflow_run" ]]; then
# Triggered by build workflow completion
echo "🔗 Triggered by build workflow completion"
@@ -184,8 +200,8 @@ jobs:
if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]] || [[ "$version" == *"rc"* ]]; then
build_type="prerelease"
is_prerelease=true
# Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
# Pre-GA policy: prereleases update latest until the first stable tag exists.
if prerelease_moves_latest; then
create_latest=true
echo "🧪 Building Docker image for prerelease: $version (creating latest tag)"
else
@@ -243,8 +259,8 @@ jobs:
v*alpha*|v*beta*|v*rc*|*alpha*|*beta*|*rc*)
build_type="prerelease"
is_prerelease=true
# Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
# Pre-GA policy: prereleases update latest until the first stable tag exists.
if prerelease_moves_latest; then
create_latest=true
echo "🧪 Building with prerelease version: $input_version (creating latest tag)"
else
@@ -394,11 +410,13 @@ jobs:
TAG_BASE="${VERSION}${VARIANT_SUFFIX}"
TAGS="${{ env.REGISTRY_DOCKERHUB }}:$TAG_BASE,${{ env.REGISTRY_GHCR }}:$TAG_BASE,${{ env.REGISTRY_QUAY }}:$TAG_BASE"
# Add channel tags for prereleases and latest for stable
# Add latest when requested (stable releases, and prereleases before GA)
if [[ "$CREATE_LATEST" == "true" ]]; then
# Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
TAGS="$TAGS,${{ env.REGISTRY_DOCKERHUB }}:latest${VARIANT_SUFFIX},${{ env.REGISTRY_GHCR }}:latest${VARIANT_SUFFIX},${{ env.REGISTRY_QUAY }}:latest${VARIANT_SUFFIX}"
elif [[ "$BUILD_TYPE" == "prerelease" ]]; then
fi
# Always add the channel tag for prereleases, independent of latest
if [[ "$BUILD_TYPE" == "prerelease" ]]; then
# Prerelease channel tags (alpha, beta, rc)
if [[ "$VERSION" == *"alpha"* ]]; then
CHANNEL="alpha"
@@ -555,7 +573,7 @@ jobs:
"prerelease")
echo "🧪 Prerelease Docker image has been built with ${VERSION} tags"
echo "⚠️ This is a prerelease image - use with caution"
# Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
# Prereleases move latest until the first stable tag exists (pre-GA policy).
if [[ "$CREATE_LATEST" == "true" ]]; then
echo "🏷️ Latest tag has been created for prerelease: $VERSION"
else
Generated
+5 -5
View File
@@ -6139,9 +6139,9 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
[[package]]
name = "libflate"
version = "2.3.1"
version = "2.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4da9b700e758e57152a1fd1c52cbdc5727c1aa6d8743dc1acda917398f1d76c"
checksum = "561a8da1a50e1428d3c51321dafeca849df992a5bb67720c386131234caba82e"
dependencies = [
"adler32",
"crc32fast",
@@ -10943,9 +10943,9 @@ dependencies = [
[[package]]
name = "rustfs-uring"
version = "0.2.1"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0486e62d0efe25db95c00aeacb2da84368adcba299216cda99fcb11328061c84"
checksum = "b29bc57b4bd62a73f4fae408b536adf578332e50e464797d09dc2382c7cb68c2"
dependencies = [
"io-uring",
"libc",
@@ -12406,7 +12406,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
dependencies = [
"fastrand",
"getrandom 0.3.4",
"getrandom 0.4.3",
"once_cell",
"rustix",
"windows-sys 0.59.0",
+1 -1
View File
@@ -226,7 +226,7 @@ metrics = { workspace = true }
# crates.io. The guard scripts/check_no_tokio_io_uring.sh allows an explicit
# io-uring integration; only the tokio "io-uring" runtime feature is banned.
[target.'cfg(target_os = "linux")'.dependencies]
rustfs-uring = "0.2.1"
rustfs-uring = "0.2.2"
[target.'cfg(windows)'.dependencies]
winapi-util.workspace = true
-11
View File
@@ -195,13 +195,6 @@ fn resolve_drive_timeout_profile_from_env() -> DriveTimeoutProfile {
DriveTimeoutProfile::parse(rustfs_config::DEFAULT_DRIVE_TIMEOUT_PROFILE).unwrap_or(DriveTimeoutProfile::Default)
}
#[cfg(test)]
tokio::task_local! {
/// Artificial `disk_info` latency for tests that pin how the admin storage
/// walk composes per-drive probe time.
pub(crate) static DISK_INFO_PROBE_DELAY_FOR_TEST: Duration;
}
fn get_drive_timeout_profile() -> DriveTimeoutProfile {
#[cfg(test)]
{
@@ -2043,10 +2036,6 @@ impl DiskAPI for LocalDiskWrapper {
.track_disk_health_with_op_and_timeout_action(
"disk_info",
|| async {
#[cfg(test)]
if let Ok(delay) = DISK_INFO_PROBE_DELAY_FOR_TEST.try_with(|delay| *delay) {
tokio::time::sleep(delay).await;
}
let result = self.disk.disk_info(opts).await?;
if let Some(current_disk_id) = *self.disk_id.read().await
+141 -4
View File
@@ -8708,6 +8708,34 @@ impl DiskAPI for LocalDisk {
});
}
let rollback_after_fsync_failure = |parent: &Path, current: Option<&[u8]>| -> std::io::Result<()> {
match current {
Some(previous) => {
let rollback_temporary =
parent.join(format!(".{}.{}.rollback.tmp", path.replace('/', "_"), Uuid::new_v4()));
let rollback_result = (|| -> std::io::Result<()> {
let mut staged = std::fs::OpenOptions::new()
.create_new(true)
.write(true)
.open(&rollback_temporary)?;
staged.write_all(previous)?;
staged.sync_all()?;
std::fs::rename(&rollback_temporary, &file_path)
})();
if let Err(err) = rollback_result {
let _ = std::fs::remove_file(&rollback_temporary);
return Err(err);
}
}
None => match std::fs::remove_file(&file_path) {
Ok(()) => {}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(err),
},
}
os::fsync_dir_std(parent)
};
match replacement {
Some(replacement) => {
let parent = file_path
@@ -8727,14 +8755,19 @@ impl DiskAPI for LocalDisk {
let _ = std::fs::remove_file(&temporary);
return Err(err);
}
if sync_metadata {
os::fsync_dir_std(parent)?;
if sync_metadata && let Err(err) = os::fsync_dir_std(parent) {
rollback_after_fsync_failure(parent, current.as_deref())?;
return Err(err);
}
}
None => {
std::fs::remove_file(&file_path)?;
if sync_metadata && let Some(parent) = file_path.parent() {
os::fsync_dir_std(parent)?;
if sync_metadata
&& let Some(parent) = file_path.parent()
&& let Err(err) = os::fsync_dir_std(parent)
{
rollback_after_fsync_failure(parent, current.as_deref())?;
return Err(err);
}
}
}
@@ -22166,6 +22199,110 @@ mod test {
));
}
#[cfg(unix)]
#[tokio::test]
async fn conditional_file_update_dir_fsync_failure_restores_previous_bytes() {
use tempfile::tempdir;
let dir = tempdir().expect("temp dir should be created");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
let previous = Bytes::from_static(b"previous-owner");
let successor = Bytes::from_static(b"successor-owner");
assert_eq!(
disk.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, None, Some(previous.clone()))
.await
.expect("previous owner should commit"),
ConditionalFileUpdate::Updated
);
let marker_path = disk
.get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
.expect("marker path should resolve");
let parent = marker_path.parent().expect("marker path should have a parent");
os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
let err = disk
.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, Some(previous.clone()), Some(successor))
.await
.expect_err("directory fsync failure must fail the CAS update");
assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
assert_eq!(
disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
.await
.expect("previous bytes should remain readable after rollback"),
previous
);
}
#[cfg(unix)]
#[tokio::test]
async fn conditional_file_update_dir_fsync_failure_removes_new_file_without_anchor() {
use tempfile::tempdir;
let dir = tempdir().expect("temp dir should be created");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
ensure_test_volume(&disk, RUSTFS_META_BUCKET).await;
let marker_path = disk
.get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
.expect("marker path should resolve");
let parent = marker_path.parent().expect("marker path should have a parent");
os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
let err = disk
.compare_and_update_file(
RUSTFS_META_BUCKET,
HEALING_MARKER_PATH,
None,
Some(Bytes::from_static(b"successor-owner")),
)
.await
.expect_err("directory fsync failure must fail the CAS create");
assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
assert!(
matches!(disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH).await, Err(DiskError::FileNotFound)),
"uncommitted successor bytes must be removed when no previous anchor exists"
);
}
#[cfg(unix)]
#[tokio::test]
async fn conditional_file_delete_dir_fsync_failure_restores_previous_bytes() {
use tempfile::tempdir;
let dir = tempdir().expect("temp dir should be created");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
let previous = Bytes::from_static(b"previous-owner");
assert_eq!(
disk.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, None, Some(previous.clone()))
.await
.expect("previous owner should commit"),
ConditionalFileUpdate::Updated
);
let marker_path = disk
.get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
.expect("marker path should resolve");
let parent = marker_path.parent().expect("marker path should have a parent");
os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
let err = disk
.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, Some(previous.clone()), None)
.await
.expect_err("directory fsync failure must fail the CAS delete");
assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
assert_eq!(
disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
.await
.expect("previous bytes should be restored after failed delete"),
previous
);
}
#[cfg(unix)]
#[tokio::test]
async fn conditional_file_update_returns_would_block_when_marker_lock_is_contended() {
+20
View File
@@ -92,6 +92,9 @@ pub(crate) mod fsync_dir_recorder {
static LIMITED: Mutex<Vec<PathBuf>> = Mutex::new(Vec::new());
static GROUPED: Mutex<Vec<(PathBuf, usize)>> = Mutex::new(Vec::new());
#[cfg(unix)]
static FAILURES: std::sync::LazyLock<Mutex<HashMap<PathBuf, io::ErrorKind>>> =
std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
#[cfg(unix)]
static BEFORE_LIMITED: std::sync::LazyLock<Mutex<HashMap<PathBuf, Hook>>> =
std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
static BEFORE_GROUP_BATCH: std::sync::LazyLock<Mutex<HashMap<PathBuf, Hook>>> =
@@ -151,6 +154,19 @@ pub(crate) mod fsync_dir_recorder {
contains_path(&RECORDED.lock().expect("fsync dir recorder poisoned"), dir)
}
#[cfg(unix)]
pub(crate) fn set_failure(dir: &Path, kind: io::ErrorKind) {
FAILURES
.lock()
.expect("fsync dir failure hook poisoned")
.insert(dir.to_path_buf(), kind);
}
#[cfg(unix)]
pub(crate) fn take_failure(dir: &Path) -> Option<io::ErrorKind> {
remove_path_keyed(&FAILURES, dir, "fsync dir failure hook poisoned")
}
#[cfg(unix)]
pub(crate) fn record_limited(dir: &Path) {
record_path(&LIMITED, dir, "limited fsync dir recorder");
@@ -426,6 +442,10 @@ pub fn fsync_dir_std(dir: impl AsRef<Path>) -> io::Result<()> {
fsync_dir_recorder::record(dir.as_ref());
#[cfg(unix)]
{
#[cfg(test)]
if let Some(kind) = fsync_dir_recorder::take_failure(dir.as_ref()) {
return Err(io::Error::from(kind));
}
std::fs::File::open(dir.as_ref())?.sync_all()?;
}
#[cfg(not(unix))]
+99 -135
View File
@@ -6463,114 +6463,113 @@ pub fn should_heal_object_on_disk(
(false, false, None)
}
/// Probe every drive of the set at once. Each live probe is bounded by the
/// drive `disk_info` timeout, and the admin peer probe budget only covers one
/// such timeout; a sequential walk over several stalled drives after a power
/// cut would exceed it and make healthy peers render as unknown (#6488).
async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<rustfs_madmin::Disk> {
join_all(disks.iter().zip(eps).map(|(disk, ep)| disk_admin_info(disk.as_ref(), ep))).await
}
let mut ret = Vec::new();
async fn disk_admin_info(disk: Option<&DiskStore>, ep: &Endpoint) -> rustfs_madmin::Disk {
let Some(disk) = disk else {
return rustfs_madmin::Disk {
endpoint: ep.to_string(),
drive_path: ep.get_file_path(),
local: ep.is_local,
pool_index: ep.pool_idx,
set_index: ep.set_idx,
disk_index: ep.disk_idx,
runtime_state: None,
offline_duration_seconds: None,
state: DiskError::DiskNotFound.to_string(),
capacity_observation_source: Some("missing".to_owned()),
capacity_observation_age_seconds: Some(0),
..Default::default()
};
};
for (i, pool) in disks.iter().enumerate() {
if let Some(disk) = pool {
let runtime_state = disk.runtime_state();
let offline_duration_seconds = disk.offline_duration_secs();
let capacity_snapshot = disk.last_capacity_snapshot();
let cached_disk_id = disk.cached_disk_id().await;
if runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect {
match disk
.disk_info(&DiskInfoOptions {
metrics: true,
..Default::default()
})
.await
{
Ok(res) => {
disk.record_capacity_probe(res.total, res.used, res.free);
ret.push(rustfs_madmin::Disk {
endpoint: eps[i].to_string(),
local: eps[i].is_local,
pool_index: eps[i].pool_idx,
set_index: eps[i].set_idx,
disk_index: eps[i].disk_idx,
state: "ok".to_owned(),
let runtime_state = disk.runtime_state();
let offline_duration_seconds = disk.offline_duration_secs();
let capacity_snapshot = disk.last_capacity_snapshot();
let cached_disk_id = disk.cached_disk_id().await;
if !(runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect) {
let mut disk_info = build_runtime_snapshot_disk(ep, runtime_state, offline_duration_seconds, capacity_snapshot);
disk_info.metrics = disk.metrics_snapshot();
disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string());
return disk_info;
}
root_disk: res.root_disk,
drive_path: res.mount_path.clone(),
healing: res.healing,
scanning: res.scanning,
runtime_state: Some(runtime_state.as_str().to_string()),
offline_duration_seconds,
capacity_observation_source: Some("live_probe".to_owned()),
capacity_observation_age_seconds: Some(0),
match disk
.disk_info(&DiskInfoOptions {
metrics: true,
..Default::default()
})
.await
{
Ok(res) => {
disk.record_capacity_probe(res.total, res.used, res.free);
rustfs_madmin::Disk {
endpoint: ep.to_string(),
local: ep.is_local,
pool_index: ep.pool_idx,
set_index: ep.set_idx,
disk_index: ep.disk_idx,
state: "ok".to_owned(),
root_disk: res.root_disk,
drive_path: res.mount_path.clone(),
healing: res.healing,
scanning: res.scanning,
runtime_state: Some(runtime_state.as_str().to_string()),
offline_duration_seconds,
capacity_observation_source: Some("live_probe".to_owned()),
capacity_observation_age_seconds: Some(0),
uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
major: res.major as u32,
minor: res.minor as u32,
model: None,
total_space: res.total,
used_space: res.used,
available_space: res.free,
physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
utilization: utilization_percent(res.total, res.used),
used_inodes: res.used_inodes,
free_inodes: res.free_inodes,
metrics: Some(res.metrics),
..Default::default()
}
}
Err(err) => {
let mut disk_info = rustfs_madmin::Disk {
state: err.to_string(),
endpoint: ep.to_string(),
drive_path: ep.get_file_path(),
local: ep.is_local,
pool_index: ep.pool_idx,
set_index: ep.set_idx,
disk_index: ep.disk_idx,
runtime_state: Some(runtime_state.as_str().to_string()),
offline_duration_seconds,
metrics: disk.metrics_snapshot(),
uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()),
..Default::default()
};
if let Some((total, used, free, _)) = capacity_snapshot {
disk_info.total_space = total;
disk_info.used_space = used;
disk_info.available_space = free;
disk_info.utilization = utilization_percent(total, used);
disk_info.capacity_observation_source = Some("snapshot".to_owned());
disk_info.capacity_observation_age_seconds =
capacity_snapshot.map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
major: res.major as u32,
minor: res.minor as u32,
model: None,
total_space: res.total,
used_space: res.used,
available_space: res.free,
physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
utilization: utilization_percent(res.total, res.used),
used_inodes: res.used_inodes,
free_inodes: res.free_inodes,
metrics: Some(res.metrics),
..Default::default()
});
}
Err(err) => {
let mut disk_info = rustfs_madmin::Disk {
state: err.to_string(),
endpoint: eps[i].to_string(),
drive_path: eps[i].get_file_path(),
local: eps[i].is_local,
pool_index: eps[i].pool_idx,
set_index: eps[i].set_idx,
disk_index: eps[i].disk_idx,
runtime_state: Some(runtime_state.as_str().to_string()),
offline_duration_seconds,
metrics: disk.metrics_snapshot(),
uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()),
..Default::default()
};
if let Some((total, used, free, _)) = capacity_snapshot {
disk_info.total_space = total;
disk_info.used_space = used;
disk_info.available_space = free;
disk_info.utilization = utilization_percent(total, used);
disk_info.capacity_observation_source = Some("snapshot".to_owned());
disk_info.capacity_observation_age_seconds = capacity_snapshot
.map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
} else {
disk_info.capacity_observation_source = Some("missing".to_owned());
disk_info.capacity_observation_age_seconds = Some(0);
}
ret.push(disk_info);
}
}
} else {
disk_info.capacity_observation_source = Some("missing".to_owned());
disk_info.capacity_observation_age_seconds = Some(0);
let mut disk_info =
build_runtime_snapshot_disk(&eps[i], runtime_state, offline_duration_seconds, capacity_snapshot);
disk_info.metrics = disk.metrics_snapshot();
disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string());
ret.push(disk_info);
}
disk_info
} else {
ret.push(rustfs_madmin::Disk {
endpoint: eps[i].to_string(),
drive_path: eps[i].get_file_path(),
local: eps[i].is_local,
pool_index: eps[i].pool_idx,
set_index: eps[i].set_idx,
disk_index: eps[i].disk_idx,
runtime_state: None,
offline_duration_seconds: None,
state: DiskError::DiskNotFound.to_string(),
capacity_observation_source: Some("missing".to_owned()),
capacity_observation_age_seconds: Some(0),
..Default::default()
})
}
}
ret
}
fn build_runtime_snapshot_disk(
@@ -10692,41 +10691,6 @@ mod tests {
);
}
#[tokio::test(start_paused = true)]
async fn test_get_disks_info_probes_drives_concurrently() {
use crate::disk::disk_store::DISK_INFO_PROBE_DELAY_FOR_TEST;
let format = FormatV3::new(1, 4);
let mut temp_dirs = Vec::new();
let mut endpoints = Vec::new();
let mut disks = Vec::new();
for disk_idx in 0..4 {
let (dir, endpoint, disk) = make_formatted_local_disk_for_info_test(disk_idx, &format).await;
temp_dirs.push(dir);
endpoints.push(endpoint);
disks.push(Some(disk));
}
let probe_delay = std::time::Duration::from_secs(2);
let started = tokio::time::Instant::now();
let info = DISK_INFO_PROBE_DELAY_FOR_TEST
.scope(probe_delay, get_disks_info(&disks, &endpoints))
.await;
let elapsed = started.elapsed();
assert_eq!(info.len(), 4);
assert!(info.iter().all(|disk| disk.state == "ok"), "every drive should still report a live probe");
assert_eq!(
info.iter().map(|disk| disk.disk_index).collect::<Vec<_>>(),
endpoints.iter().map(|ep| ep.disk_idx).collect::<Vec<_>>(),
"concurrent probes must keep endpoint order"
);
assert!(
elapsed < probe_delay * 2,
"four stalled drives must cost one probe delay, not four; took {elapsed:?}"
);
}
#[tokio::test]
async fn test_get_disks_info_preserves_remote_cached_disk_id_when_offline() {
let (endpoint, disk) = make_remote_disk_for_info_test(0).await;