From b4e6c1b0816dd0968a88731f6e79cdbadcca7fae Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Tue, 25 Aug 2026 21:19:49 +0800 Subject: [PATCH] fix(connect): use persisted inventory for offline collectors (#6560) --- rustfs/src/connect/inventory.rs | 24 ++ rustfs/src/connect/offline/collectors.rs | 123 +++------- rustfs/src/connect/offline/mod.rs | 2 +- rustfs/src/connect/runtime.rs | 44 +++- rustfs/tests/connect_offline_collectors.rs | 266 +++++++++++++-------- 5 files changed, 270 insertions(+), 189 deletions(-) diff --git a/rustfs/src/connect/inventory.rs b/rustfs/src/connect/inventory.rs index 6238a581a..094af0ca1 100644 --- a/rustfs/src/connect/inventory.rs +++ b/rustfs/src/connect/inventory.rs @@ -232,6 +232,30 @@ impl InventorySnapshot { Ok(hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower)) } + pub(crate) fn rustfs_version(&self) -> &str { + &self.rustfs_version + } + + pub(crate) fn node_count(&self) -> u16 { + self.node_count + } + + pub(crate) fn drive_count(&self) -> u32 { + self.drive_count + } + + pub(crate) fn capacity_total_bytes(&self) -> u64 { + self.capacity_total_bytes + } + + pub(crate) fn capacity_used_bytes(&self) -> u64 { + self.capacity_used_bytes + } + + pub(crate) fn coarse_flags(&self) -> &[InventoryFlag] { + &self.coarse_flags + } + fn validate(&self) -> Result<(), InventoryError> { if !valid_version(&self.rustfs_version) { return Err(InventoryError::RustfsVersion); diff --git a/rustfs/src/connect/offline/collectors.rs b/rustfs/src/connect/offline/collectors.rs index 336c455ac..4cfdce16a 100644 --- a/rustfs/src/connect/offline/collectors.rs +++ b/rustfs/src/connect/offline/collectors.rs @@ -15,24 +15,23 @@ //! Fixed Q07 L0/L1 collectors for an operator-triggered offline diagnostic. use std::collections::BTreeSet; +use std::path::Path; use std::sync::{Arc, LazyLock}; use std::time::Duration; -use rustfs_madmin::{ITEM_OFFLINE, StorageInfo}; use serde::Serialize; use serde_json::{Value, json}; use sysinfo::{Disks, Networks, RefreshKind, System}; use thiserror::Error; use tokio::sync::Semaphore; use tokio_util::sync::CancellationToken; -use url::Url; +use super::super::inventory::{InventoryError, InventorySnapshot, InventoryStateStore}; use super::manifest_entry::ManifestEntry; use super::redaction::RedactionError; const COLLECT_TIMEOUT: Duration = Duration::from_secs(2); const MAX_ENTRY_BYTES: usize = 16 * 1024; -const MAX_DRIVES: usize = 4_096; static SYSTEM_SCAN_PERMIT: LazyLock> = LazyLock::new(|| Arc::new(Semaphore::new(1))); #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] @@ -119,19 +118,14 @@ impl OfflineCollector { self.field_id().split_once('.').expect("collector field ids are frozen").1 } - fn value(self, storage: &StorageSnapshot, system: &SystemSnapshot) -> Value { + fn value(self, inventory: &InventorySnapshot, system: &SystemSnapshot) -> Value { match self { - Self::RustfsVersion => json!(env!("CARGO_PKG_VERSION")), - Self::NodeCount => json!(storage.node_count), - Self::DriveCount => json!(storage.drive_count), - Self::CapacityUsedBytes => json!(storage.capacity_used_bytes), - Self::CapacityTotalBytes => json!(storage.capacity_total_bytes), - Self::CoarseHealthFlags => json!({ - "degraded": storage.degraded, - "healing": storage.healing, - "offlineDrives": storage.offline_drives, - "scanning": storage.scanning, - }), + Self::RustfsVersion => json!(inventory.rustfs_version()), + Self::NodeCount => json!(inventory.node_count()), + Self::DriveCount => json!(inventory.drive_count()), + Self::CapacityUsedBytes => json!(inventory.capacity_used_bytes()), + Self::CapacityTotalBytes => json!(inventory.capacity_total_bytes()), + Self::CoarseHealthFlags => json!(inventory.coarse_flags()), Self::OsSummary => json!(system.os_summary), Self::KernelSummary => json!(system.kernel_summary), Self::CpuSummary => json!({ "architecture": system.architecture, "cores": system.cores }), @@ -156,68 +150,22 @@ pub enum CollectorError { TimedOut, #[error("offline diagnostic collector task failed")] TaskFailed, - #[error("offline diagnostic storage topology exceeds its 4096 drive budget")] - StorageTopologyTooLarge, - #[error("offline diagnostic storage topology contains an invalid endpoint")] - InvalidStorageEndpoint, #[error("offline diagnostic field {field_id} exceeds its {limit} byte entry budget")] EntryTooLarge { field_id: &'static str, limit: usize }, #[error("offline diagnostic entry is not representable as JSON")] NotRepresentable, #[error(transparent)] + Inventory(#[from] InventoryError), + #[error(transparent)] Redaction(#[from] RedactionError), } -#[derive(Debug)] -struct StorageSnapshot { - node_count: usize, - drive_count: usize, - capacity_used_bytes: u64, - capacity_total_bytes: u64, - offline_drives: usize, - degraded: bool, - healing: bool, - scanning: bool, -} - -impl TryFrom<&StorageInfo> for StorageSnapshot { - type Error = CollectorError; - - fn try_from(info: &StorageInfo) -> Result { - if info.disks.len() > MAX_DRIVES { - return Err(CollectorError::StorageTopologyTooLarge); - } - let node_count = info - .disks - .iter() - .map(|disk| { - let endpoint = Url::parse(&disk.endpoint).map_err(|_| CollectorError::InvalidStorageEndpoint)?; - let host = endpoint.host_str().ok_or(CollectorError::InvalidStorageEndpoint)?; - Ok((host.to_owned(), endpoint.port_or_known_default())) - }) - .collect::, CollectorError>>()? - .len(); - let offline_drives = info.disks.iter().filter(|disk| disk.state == ITEM_OFFLINE).count(); - Ok(Self { - node_count, - drive_count: info.disks.len(), - capacity_used_bytes: info - .disks - .iter() - .fold(0_u64, |total, disk| total.saturating_add(disk.used_space)), - capacity_total_bytes: info - .disks - .iter() - .fold(0_u64, |total, disk| total.saturating_add(disk.total_space)), - offline_drives, - degraded: info - .disks - .iter() - .any(|disk| !matches!(disk.state.as_str(), "ok" | "unformatted" | rustfs_madmin::ITEM_ONLINE)), - healing: info.disks.iter().any(|disk| disk.healing), - scanning: info.disks.iter().any(|disk| disk.scanning), - }) - } +/// The bounded entries plus the capture time of their persisted L0 source. +#[derive(Debug, PartialEq)] +pub struct OfflineDiagnostics { + pub entries: Vec, + pub inventory_captured_at: String, + pub inventory_age: Duration, } #[derive(Debug)] @@ -293,23 +241,33 @@ async fn collect_system_snapshot(cancel: &CancellationToken) -> Result Result, CollectorError> { +) -> Result { if cancel.is_cancelled() { return Err(CollectorError::Cancelled); } - let storage = StorageSnapshot::try_from(storage_info)?; + let store = InventoryStateStore::from_state_root(state_root)?; + let _lock = store.try_runtime_lock()?; + let persisted = store.read_latest(chrono::Utc::now())?; let system = collect_system_snapshot(cancel).await?; let mut entries = Vec::with_capacity(COLLECTORS.len()); for collector in COLLECTORS { - entries.push(ManifestEntry::from_value(collector, collector.value(&storage, &system), cancel)?); + entries.push(ManifestEntry::from_value( + collector, + collector.value(&persisted.snapshot, &system), + cancel, + )?); } - Ok(entries) + Ok(OfflineDiagnostics { + entries, + inventory_captured_at: persisted.captured_at, + inventory_age: persisted.age, + }) } #[cfg(test)] @@ -346,8 +304,6 @@ mod test_support { mod tests { use std::sync::atomic::Ordering; - use rustfs_madmin::StorageInfo; - use super::*; async fn wait_for_active(expected: usize) { @@ -366,16 +322,13 @@ mod tests { test_support::DELAY_MILLIS.store((COLLECT_TIMEOUT + Duration::from_millis(200)).as_millis() as u64, Ordering::SeqCst); let first_cancel = CancellationToken::new(); - assert!(matches!( - collect_offline_diagnostics(&StorageInfo::default(), &first_cancel).await, - Err(CollectorError::TimedOut) - )); + assert!(matches!(collect_system_snapshot(&first_cancel).await, Err(CollectorError::TimedOut))); assert_eq!(test_support::ACTIVE.load(Ordering::SeqCst), 1, "timed-out blocking scan remains active"); let second_cancel = CancellationToken::new(); let second = tokio::spawn({ let second_cancel = second_cancel.clone(); - async move { collect_offline_diagnostics(&StorageInfo::default(), &second_cancel).await } + async move { collect_system_snapshot(&second_cancel).await } }); tokio::time::sleep(Duration::from_millis(50)).await; assert_eq!( @@ -392,7 +345,7 @@ mod tests { let third_cancel = CancellationToken::new(); let third = tokio::spawn({ let third_cancel = third_cancel.clone(); - async move { collect_offline_diagnostics(&StorageInfo::default(), &third_cancel).await } + async move { collect_system_snapshot(&third_cancel).await } }); wait_for_active(1).await; third_cancel.cancel(); @@ -401,7 +354,7 @@ mod tests { let fourth_cancel = CancellationToken::new(); let fourth = tokio::spawn({ let fourth_cancel = fourth_cancel.clone(); - async move { collect_offline_diagnostics(&StorageInfo::default(), &fourth_cancel).await } + async move { collect_system_snapshot(&fourth_cancel).await } }); tokio::time::sleep(Duration::from_millis(50)).await; assert_eq!( diff --git a/rustfs/src/connect/offline/mod.rs b/rustfs/src/connect/offline/mod.rs index 10b1cd6fa..37a7f3b84 100644 --- a/rustfs/src/connect/offline/mod.rs +++ b/rustfs/src/connect/offline/mod.rs @@ -34,7 +34,7 @@ pub mod key_store; pub mod manifest_entry; pub mod redaction; -pub use collectors::{CollectorError, OfflineCollector, collect_offline_diagnostics}; +pub use collectors::{CollectorError, OfflineCollector, OfflineDiagnostics, collect_offline_diagnostics}; pub use enrollment::{EnrollmentError, OfflineEnrollment, VerifiedChallenge}; pub use key_store::OfflineKeyStore; pub use manifest_entry::ManifestEntry; diff --git a/rustfs/src/connect/runtime.rs b/rustfs/src/connect/runtime.rs index 7f2b60625..cb8bf0467 100644 --- a/rustfs/src/connect/runtime.rs +++ b/rustfs/src/connect/runtime.rs @@ -13,6 +13,7 @@ // limitations under the License. use std::future::Future; +use std::sync::Arc; use std::time::Duration; use chrono::Utc; @@ -57,6 +58,7 @@ pub struct InventoryRuntime { shutdown: CancellationToken, status: watch::Receiver, task: Option>, + _lock: Arc, } impl InventoryRuntime { @@ -190,7 +192,7 @@ where let retry_schedule = config.schedule; let state_root = config.state_root().ok_or(InventoryError::StatePath)?; let store = InventoryStateStore::from_state_root(state_root)?; - let lock = store.try_runtime_lock()?; + let lock = Arc::new(store.try_runtime_lock()?); let sender = if config.transport_enabled() { Some(InventorySender::new(config)?) } else { @@ -199,8 +201,9 @@ where let shutdown = parent_shutdown.child_token(); let task_shutdown = shutdown.clone(); let (status_tx, status_rx) = watch::channel(InventoryStatus::Starting); + let task_lock = lock.clone(); let task = tokio::spawn(async move { - let _lock = lock; + let _lock = task_lock; let mut backoff = retry_schedule.initial_backoff; loop { if task_shutdown.is_cancelled() { @@ -331,6 +334,7 @@ where shutdown, status: status_rx, task: Some(task), + _lock: lock, })) } @@ -474,6 +478,7 @@ mod tests { task_inventory_shutdown.cancelled().await; let _ = inventory_stopped.send(()); }); + let inventory_lock = Arc::new(tempfile::tempfile().expect("inventory runtime lock")); let heartbeat = HeartbeatRuntime { shutdown: heartbeat_shutdown, status: heartbeat_status, @@ -483,6 +488,7 @@ mod tests { shutdown: inventory_shutdown, status: inventory_status, task: Some(inventory_task), + _lock: inventory_lock, }; let shutdown = tokio::spawn(shutdown_connect_runtimes(Some(heartbeat), Some(inventory))); @@ -496,4 +502,38 @@ mod tests { .expect("runtime shutdown") .expect("shutdown task"); } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn dropping_inventory_handle_keeps_the_lock_until_its_task_exits() { + use std::os::unix::fs::PermissionsExt as _; + + let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory"); + let state = temp.path().join("state"); + std::fs::create_dir(&state).expect("state root"); + std::fs::set_permissions(&state, std::fs::Permissions::from_mode(0o700)).expect("private state root"); + let store = InventoryStateStore::from_state_root(&state).expect("inventory store"); + let lock = Arc::new(store.try_runtime_lock().expect("runtime lock")); + let task_lock = lock.clone(); + let (release, released) = tokio::sync::oneshot::channel(); + let (finished, task_finished) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(async move { + let _ = released.await; + drop(task_lock); + let _ = finished.send(()); + }); + let (_, status) = watch::channel(InventoryStatus::Starting); + let runtime = InventoryRuntime { + shutdown: CancellationToken::new(), + status, + task: Some(task), + _lock: lock, + }; + + drop(runtime); + assert!(matches!(store.try_runtime_lock(), Err(InventoryError::AlreadyRunning))); + release.send(()).expect("release inventory task"); + task_finished.await.expect("inventory task finished"); + store.try_runtime_lock().expect("lock after task exit"); + } } diff --git a/rustfs/tests/connect_offline_collectors.rs b/rustfs/tests/connect_offline_collectors.rs index bf706baf1..ecb651cdb 100644 --- a/rustfs/tests/connect_offline_collectors.rs +++ b/rustfs/tests/connect_offline_collectors.rs @@ -16,11 +16,27 @@ use std::fs; use std::path::PathBuf; +#[cfg(target_os = "linux")] +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; +#[cfg(target_os = "linux")] +use std::time::Duration; -use rustfs::connect::offline::{CollectorError, RedactionSource, collect_offline_diagnostics, redact_json}; -use rustfs_madmin::{Disk, ITEM_OFFLINE, StorageInfo}; +#[cfg(target_os = "linux")] +use rustfs::connect::offline::{CollectorError, collect_offline_diagnostics}; +use rustfs::connect::offline::{RedactionSource, redact_json}; +#[cfg(target_os = "linux")] +use rustfs::connect::{ + CredentialStore, HeartbeatConfig, IdentityStore, InventoryError, InventoryFlag, InventorySchedule, InventorySnapshot, + InventoryStatus, spawn_inventory_runtime, +}; use serde_json::{Map, Value, json}; use sha2::{Digest as _, Sha256}; +#[cfg(target_os = "linux")] +use tokio::sync::watch; +#[cfg(target_os = "linux")] use tokio_util::sync::CancellationToken; fn fixture_dir() -> PathBuf { @@ -164,127 +180,175 @@ fn connect_offline_collectors_reject_oversize_raw_input_before_parsing() { ); } +#[cfg(target_os = "linux")] +async fn wait_for_inventory(status: &mut watch::Receiver) { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + if matches!(status.borrow_and_update().clone(), InventoryStatus::Unchanged { .. }) { + return; + } + status.changed().await.expect("inventory status channel"); + } + }) + .await + .expect("inventory persistence timeout"); +} + +#[cfg(target_os = "linux")] +async fn wait_for_inventory_failure(status: &mut watch::Receiver) { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + if matches!(status.borrow_and_update().clone(), InventoryStatus::Failed { .. }) { + return; + } + status.changed().await.expect("inventory status channel"); + } + }) + .await + .expect("inventory failure timeout"); +} + +#[cfg(target_os = "linux")] +fn private_directory(path: &std::path::Path) { + use std::os::unix::fs::PermissionsExt as _; + + fs::set_permissions(path, fs::Permissions::from_mode(0o700)).expect("private directory permissions"); +} + +#[cfg(target_os = "linux")] #[tokio::test] -async fn connect_offline_collectors_emit_only_fixed_redacted_entries_and_honor_cancellation() { - let storage = StorageInfo { - disks: vec![ - Disk { - endpoint: "https://node-a.private.example:9000/data-a".to_owned(), - drive_path: "/secret/customer/path-a".to_owned(), - uuid: "private-drive-a".to_owned(), - state: "ok".to_owned(), - total_space: 1_000, - used_space: 400, - ..Disk::default() - }, - Disk { - endpoint: "https://node-a.private.example:9000/data-b".to_owned(), - drive_path: "/secret/customer/path-b".to_owned(), - uuid: "private-drive-b".to_owned(), - state: "unformatted".to_owned(), - total_space: 2_000, - used_space: 500, - ..Disk::default() - }, - Disk { - endpoint: "https://node-b.private.example:9000/data-c".to_owned(), - drive_path: "/secret/customer/path-c".to_owned(), - uuid: "private-drive-c".to_owned(), - state: ITEM_OFFLINE.to_owned(), - total_space: 3_000, - used_space: 600, - healing: true, - ..Disk::default() - }, - ], - ..StorageInfo::default() - }; - let cancel = CancellationToken::new(); - let entries = collect_offline_diagnostics(&storage, &cancel) +async fn connect_offline_collectors_read_only_the_stopped_runtime_inventory() { + let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory"); + let state = temp.path().join("state"); + fs::create_dir(&state).expect("state root"); + private_directory(&state); + let config = HeartbeatConfig::new( + "", + Vec::new(), + IdentityStore::new(state.join("identity")), + CredentialStore::new(state.join("credential")), + state.join("heartbeat/state.json"), + ); + let shutdown = CancellationToken::new(); + let runtime = spawn_inventory_runtime( + Some(config), + InventorySchedule { + cadence: Duration::from_secs(60), + jitter: Duration::ZERO, + }, + &shutdown, + || { + std::future::ready(InventorySnapshot::new( + "1.4.2", + None, + 2, + 3, + 6_000, + 1_500, + [InventoryFlag::ClusterDegraded, InventoryFlag::DriveOffline], + )) + }, + ) + .expect("state-only inventory") + .expect("configured inventory"); + let mut status = runtime.status(); + wait_for_inventory(&mut status).await; + + assert!(matches!( + collect_offline_diagnostics(&state, &CancellationToken::new()).await, + Err(CollectorError::Inventory(InventoryError::AlreadyRunning)) + )); + runtime.shutdown().await; + + let diagnostics = collect_offline_diagnostics(&state, &CancellationToken::new()) .await .expect("collect fixed offline entries"); - assert_eq!(entries.len(), 12); - let encoded = serde_json::to_string(&entries).expect("manifest entries serialize"); - for forbidden in [ - "node-a.private.example", - "node-b.private.example", - "/secret/customer/path-a", - "/secret/customer/path-b", - "/secret/customer/path-c", - "private-drive-a", - "private-drive-b", - "private-drive-c", - ] { - assert!(!encoded.contains(forbidden), "private storage metadata must not leave the collector"); - } - assert!(entries.iter().all(|entry| entry.field_id.starts_with("offline."))); - assert!(entries.iter().all(|entry| entry.canonical_json.len() <= 16 * 1024)); + assert_eq!(diagnostics.entries.len(), 12); + assert!(diagnostics.inventory_captured_at.ends_with('Z')); + assert!(diagnostics.inventory_age < Duration::from_secs(10)); + assert!(diagnostics.entries.iter().all(|entry| entry.field_id.starts_with("offline."))); + assert!( + diagnostics + .entries + .iter() + .all(|entry| entry.canonical_json.len() <= 16 * 1024) + ); let canonical = |field_id| { - entries + diagnostics + .entries .iter() .find(|entry| entry.field_id == field_id) .unwrap_or_else(|| panic!("missing {field_id}")) .canonical_json .as_str() }; + assert_eq!(canonical("offline.rustfsVersion"), r#"{"rustfsVersion":"1.4.2"}"#); assert_eq!(canonical("offline.nodeCount"), r#"{"nodeCount":2}"#); assert_eq!(canonical("offline.driveCount"), r#"{"driveCount":3}"#); assert_eq!(canonical("offline.capacityUsedBytes"), r#"{"capacityUsedBytes":1500}"#); assert_eq!(canonical("offline.capacityTotalBytes"), r#"{"capacityTotalBytes":6000}"#); assert_eq!( canonical("offline.coarseHealthFlags"), - r#"{"coarseHealthFlags":{"degraded":true,"healing":true,"offlineDrives":1,"scanning":false}}"# + r#"{"coarseHealthFlags":["cluster.degraded","drive.offline"]}"# ); - let healthy = StorageInfo { - disks: ["ok", "unformatted", "online"] - .into_iter() - .enumerate() - .map(|(index, state)| Disk { - endpoint: format!("https://healthy.example:9000/data-{index}"), - state: state.to_owned(), - ..Disk::default() - }) - .collect(), - ..StorageInfo::default() - }; - let healthy_entries = collect_offline_diagnostics(&healthy, &CancellationToken::new()) - .await - .expect("collect healthy storage summary"); - assert_eq!( - healthy_entries - .iter() - .find(|entry| entry.field_id == "offline.coarseHealthFlags") - .expect("healthy coarse health entry") - .canonical_json, - r#"{"coarseHealthFlags":{"degraded":false,"healing":false,"offlineDrives":0,"scanning":false}}"# - ); - - cancel.cancel(); + let cancelled = CancellationToken::new(); + cancelled.cancel(); assert!(matches!( - collect_offline_diagnostics(&storage, &cancel).await, + collect_offline_diagnostics(&state, &cancelled).await, Err(CollectorError::Cancelled) )); - let oversized = StorageInfo { - disks: vec![Disk::default(); 4_097], - ..StorageInfo::default() - }; - let active = CancellationToken::new(); + fs::write(state.join("inventory/latest.json"), b"{}").expect("corrupt latest inventory"); assert!(matches!( - collect_offline_diagnostics(&oversized, &active).await, - Err(CollectorError::StorageTopologyTooLarge) - )); - - let invalid_endpoint = StorageInfo { - disks: vec![Disk { - endpoint: "not-an-endpoint".to_owned(), - ..Disk::default() - }], - ..StorageInfo::default() - }; - assert!(matches!( - collect_offline_diagnostics(&invalid_endpoint, &active).await, - Err(CollectorError::InvalidStorageEndpoint) + collect_offline_diagnostics(&state, &CancellationToken::new()).await, + Err(CollectorError::Inventory(InventoryError::EnvelopeInvalid)) )); } + +#[cfg(target_os = "linux")] +#[tokio::test] +async fn connect_offline_collectors_reject_a_live_runtime_after_its_task_fails() { + let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory"); + let state = temp.path().join("state"); + fs::create_dir(&state).expect("state root"); + private_directory(&state); + let config = HeartbeatConfig::new( + "", + Vec::new(), + IdentityStore::new(state.join("identity")), + CredentialStore::new(state.join("credential")), + state.join("heartbeat/state.json"), + ); + let attempts = Arc::new(AtomicUsize::new(0)); + let sample_attempts = attempts.clone(); + let runtime = spawn_inventory_runtime( + Some(config), + InventorySchedule { + cadence: Duration::from_millis(1), + jitter: Duration::ZERO, + }, + &CancellationToken::new(), + move || { + let attempt = sample_attempts.fetch_add(1, Ordering::SeqCst); + std::future::ready(if attempt == 0 { + InventorySnapshot::new("1.4.2", None, 1, 1, 100, 50, []) + } else { + Err(InventoryError::Capacity) + }) + }, + ) + .expect("state-only inventory") + .expect("configured inventory"); + let mut status = runtime.status(); + wait_for_inventory_failure(&mut status).await; + + assert!(matches!( + collect_offline_diagnostics(&state, &CancellationToken::new()).await, + Err(CollectorError::Inventory(InventoryError::AlreadyRunning)) + )); + runtime.shutdown().await; + collect_offline_diagnostics(&state, &CancellationToken::new()) + .await + .expect("collector after runtime shutdown"); +}