diff --git a/rustfs/src/connect/inventory.rs b/rustfs/src/connect/inventory.rs index 40c4f6cc5..5969ce56f 100644 --- a/rustfs/src/connect/inventory.rs +++ b/rustfs/src/connect/inventory.rs @@ -550,6 +550,8 @@ pub enum InventoryError { DriveCount, #[error("the RustFS inventory capacity is outside protocol bounds")] Capacity, + #[error("the RustFS inventory snapshot is incomplete: observed {observed} of {expected} configured drives")] + SnapshotIncomplete { expected: usize, observed: usize }, #[error("the Connect inventory schedule is invalid")] Schedule, #[error("the Connect inventory sequence is exhausted")] diff --git a/rustfs/src/connect/runtime.rs b/rustfs/src/connect/runtime.rs index 0f8882bca..d4e82d1e1 100644 --- a/rustfs/src/connect/runtime.rs +++ b/rustfs/src/connect/runtime.rs @@ -199,6 +199,15 @@ where Ok(None) => { let snapshot = match cancellable(&task_shutdown, sample()).await { Some(Ok(snapshot)) => snapshot, + Some(Err(InventoryError::SnapshotIncomplete { .. })) => { + let delay = backoff; + backoff = backoff.saturating_mul(2).min(retry_schedule.max_backoff); + let _ = status_tx.send(InventoryStatus::BackingOff { delay }); + if sleep_or_cancel(&task_shutdown, delay).await { + break; + } + continue; + } Some(Err(error)) => return failed_inventory(&status_tx, error), None => break, }; diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 51ecc0b2a..f71258590 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -17,7 +17,7 @@ use crate::storage_api::startup::services::{ECStore, EndpointServerPools, Server use crate::{ config::Config, connect::{ - CoarseNodeSummary, HeartbeatConfig, HeartbeatRuntime, InventoryFlag, InventoryRuntime, InventorySchedule, + CoarseNodeSummary, HeartbeatConfig, HeartbeatRuntime, InventoryError, InventoryFlag, InventoryRuntime, InventorySchedule, InventorySnapshot, spawn_heartbeat_runtime, spawn_inventory_runtime, }, init::{init_buffer_profile_system, init_kms_system}, @@ -80,6 +80,9 @@ pub(crate) async fn init_startup_runtime_services( let optional_runtimes = init_optional_runtime_services().await?; let heartbeat_config = HeartbeatConfig::from_env().map_err(std::io::Error::other)?; let heartbeat_nodes = heartbeat_config.as_ref().map(|_| endpoint_pools.get_nodes().len()); + let inventory_drives = heartbeat_config + .as_ref() + .map(|_| endpoint_pools.as_ref().iter().map(|pool| pool.endpoints.as_ref().len()).sum()); init_buffer_profile_system(config); init_deadlock_detector_runtime(); @@ -100,7 +103,7 @@ pub(crate) async fn init_startup_runtime_services( let enable_scanner = init_background_service_runtime(store.clone()).await?; init_observability_runtime(store.clone(), ctx.clone()).await; let heartbeat = start_heartbeat_runtime(heartbeat_config.clone(), heartbeat_nodes, &ctx)?; - let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, store, &ctx)?; + let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, inventory_drives, store, &ctx)?; let heartbeat = heartbeat.map(|heartbeat| heartbeat.with_inventory(inventory)); Ok(StartupServiceRuntime { @@ -129,6 +132,7 @@ fn start_heartbeat_runtime( fn start_inventory_runtime( config: Option, node_count: Option, + expected_drive_count: Option, store: Arc, shutdown: &CancellationToken, ) -> Result> { @@ -136,21 +140,57 @@ fn start_inventory_runtime( return Ok(None); }; let node_count = node_count.unwrap_or_default(); + let expected_drive_count = expected_drive_count.unwrap_or_default(); spawn_inventory_runtime(Some(config), InventorySchedule::default(), shutdown, move || { let store = store.clone(); async move { let info = StorageAdminApi::storage_info(store.as_ref()).await; - let total = crate::app::storage_api::capacity::get_total_usable_capacity(&info.disks, &info) as u64; - let free = crate::app::storage_api::capacity::get_total_usable_capacity_free(&info.disks, &info) as u64; - let mut flags = Vec::with_capacity(3); - if info.disks.iter().any(|disk| disk.state == rustfs_madmin::ITEM_OFFLINE) { - flags.extend([InventoryFlag::ClusterDegraded, InventoryFlag::DriveOffline]); - } - if info.disks.iter().any(|disk| disk.healing) { - flags.push(InventoryFlag::ClusterHealing); - } - InventorySnapshot::current(node_count, info.disks.len(), total, free, flags) + inventory_snapshot(node_count, expected_drive_count, info) } }) .map_err(std::io::Error::other) } + +fn inventory_snapshot( + node_count: usize, + expected_drive_count: usize, + info: rustfs_madmin::StorageInfo, +) -> std::result::Result { + if info.disks.len() != expected_drive_count { + return Err(InventoryError::SnapshotIncomplete { + expected: expected_drive_count, + observed: info.disks.len(), + }); + } + let total = crate::app::storage_api::capacity::get_total_usable_capacity(&info.disks, &info) as u64; + let free = crate::app::storage_api::capacity::get_total_usable_capacity_free(&info.disks, &info) as u64; + let mut flags = Vec::with_capacity(3); + if info.disks.iter().any(|disk| disk.state == rustfs_madmin::ITEM_OFFLINE) { + flags.extend([InventoryFlag::ClusterDegraded, InventoryFlag::DriveOffline]); + } + if info.disks.iter().any(|disk| disk.healing) { + flags.push(InventoryFlag::ClusterHealing); + } + InventorySnapshot::current(node_count, info.disks.len(), total, free, flags) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn inventory_rejects_a_partial_startup_storage_snapshot() { + let info = rustfs_madmin::StorageInfo { + disks: vec![rustfs_madmin::Disk::default()], + ..Default::default() + }; + + assert!(matches!( + inventory_snapshot(2, 2, info), + Err(InventoryError::SnapshotIncomplete { + expected: 2, + observed: 1 + }) + )); + } +} diff --git a/rustfs/tests/connect_inventory.rs b/rustfs/tests/connect_inventory.rs index 494a2f005..6c193e997 100644 --- a/rustfs/tests/connect_inventory.rs +++ b/rustfs/tests/connect_inventory.rs @@ -500,6 +500,39 @@ async fn connect_inventory_disconnect_retries_without_resampling() { runtime.shutdown().await; } +#[tokio::test] +async fn connect_inventory_retries_an_incomplete_sample_before_delivery() { + let pki = TestPki::new(); + let content_hash = snapshot().content_hash().expect("content hash"); + let server = server(&pki, vec![Reply::ok(&content_hash)]).await; + let temp = tempfile::tempdir().expect("tempdir"); + let shutdown = CancellationToken::new(); + let samples = Arc::new(AtomicUsize::new(0)); + let sampled = samples.clone(); + let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &server)), schedule(), &shutdown, move || { + let attempt = sampled.fetch_add(1, Ordering::Relaxed); + std::future::ready(if attempt == 0 { + Err(rustfs::connect::InventoryError::SnapshotIncomplete { + expected: 96, + observed: 12, + }) + } else { + Ok(snapshot()) + }) + }) + .expect("start inventory") + .expect("configured inventory"); + let mut status = runtime.status(); + + assert!(matches!( + wait_for(&mut status, |status| matches!(status, InventoryStatus::Online { .. })).await, + InventoryStatus::Online { content_hash: accepted, .. } if accepted == content_hash + )); + assert_eq!(samples.load(Ordering::Relaxed), 2); + assert_eq!(server.seen.lock().expect("seen lock").len(), 1); + runtime.shutdown().await; +} + #[tokio::test] async fn connect_inventory_revoked_device_stops_without_retrying() { let pki = TestPki::new();