From 5218e68554be6a586cac6b6101f9803e9f68f76f Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Sun, 27 Sep 2026 14:38:23 +0300 Subject: [PATCH] volume server: collect heartbeats under the store read lock, off the runtime (#11480) * volume server: collect heartbeats under the store read lock, off the runtime Every pulse tick, options change and volume-state notification took store.write() for a whole heartbeat pass, directly on the async heartbeat task. The pass fstats every volume's .dat twice and hashes its report, so on a server with many volumes it held the store exclusively for the whole scan: reads and writes stalled, and with the writer-preferring RwLock a pending pass parked every new reader too. The pass only needs to mutate the store for a few rare actions: removing expired EC volumes, deleting expired volumes past their removal delay, and setting no-write on IO-quarantined volumes. It now runs under store.read(), records those as (disk, volume id) actions, and applies them afterwards under a short store.write() that is only taken when there is something to do. Each action re-checks its target under the write lock, so a volume written to, replaced or removed in between is left alone. Expired EC volumes are still removed before the volume pass, as before, because the EC shard count feeds the disk-space-low max volume count. Every pass runs on the blocking pool via spawn_blocking. The heartbeat message is unchanged for the same store state. Co-Authored-By: Claude Opus 5.5 (1M context) * volume server: take has_no_ec_shards with the heartbeat's volume list The heartbeat pass took has_no_ec_shards from the EC phase's read lock, then built the volume list under a second one. An EC shard mounted in between went out as "no EC shards" beside a volume list taken after the mount, and the master clears a server's EC registrations on that flag. has_no_ec_shards is now computed under the same read lock as the volume list, with the EC phase's filter: not expired, not quarantined, at least one shard. Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) --- seaweed-volume/src/server/heartbeat.rs | 435 ++++++++++++++++++++++--- seaweed-volume/src/storage/store.rs | 93 ++++-- 2 files changed, 455 insertions(+), 73 deletions(-) diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 64e9dc5d9..30a978304 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -24,8 +24,10 @@ use crate::storage::volume_report_hash::report_hash; const DUPLICATE_UUID_RETRY_MESSAGE: &str = "duplicate UUIDs detected, retrying connection"; const MAX_DUPLICATE_UUID_RETRIES: u32 = 3; +const MAX_TTL_VOLUME_REMOVAL_DELAY: u32 = 10; /// Configuration for the heartbeat client. +#[derive(Clone)] pub struct HeartbeatConfig { pub ip: String, pub port: u16, @@ -396,7 +398,8 @@ async fn do_heartbeat( state.store.read().unwrap().volume_report.reset(); // Keep track of what we sent, to generate delta updates - let (initial_hb, initial_volumes) = collect_heartbeat_with_snapshot(config, state); + let (initial_hb, initial_volumes) = + off_runtime(config, state, collect_heartbeat_with_snapshot).await?; let mut last_volumes: HashMap = volume_identities(&initial_volumes); let mut last_ec_shards = { let store = state.store.read().unwrap(); @@ -462,7 +465,8 @@ async fn do_heartbeat( }; if changed { let (adjusted_hb, adjusted_volumes) = - collect_heartbeat_with_snapshot(config, state); + off_runtime(config, state, collect_heartbeat_with_snapshot) + .await?; last_volumes = volume_identities(&adjusted_volumes); last_ec_shards = { let store = state.store.read().unwrap(); @@ -496,7 +500,8 @@ async fn do_heartbeat( let s = state.store.read().unwrap(); s.maybe_adjust_volume_max(); } - let (current_hb, current_volumes) = collect_heartbeat_with_snapshot(config, state); + let (current_hb, current_volumes) = + off_runtime(config, state, collect_heartbeat_with_snapshot).await?; last_volumes = volume_identities(¤t_volumes); last_ec_shards = { let store = state.store.read().unwrap(); @@ -525,7 +530,7 @@ async fn do_heartbeat( info!("Heartbeat stopping"); return Ok(None); } - let held_volumes = collect_volume_snapshot(config, state); + let held_volumes = off_runtime(config, state, collect_volume_snapshot).await?; let current_volumes = volume_identities(&held_volumes); let current_ec_shards = { let store = state.store.read().unwrap(); @@ -781,6 +786,16 @@ fn volume_identities( .collect() } +/// Runs a store pass on the blocking pool: it stats every volume's files. +async fn off_runtime( + config: &HeartbeatConfig, + state: &Arc, + pass: fn(&HeartbeatConfig, &Arc) -> T, +) -> Result { + let (config, state) = (config.clone(), state.clone()); + tokio::task::spawn_blocking(move || pass(&config, &state)).await +} + /// Collect volume information into a Heartbeat message. fn collect_heartbeat_with_snapshot( config: &HeartbeatConfig, @@ -789,15 +804,32 @@ fn collect_heartbeat_with_snapshot( master_pb::Heartbeat, Vec, ) { - let mut store = state.store.write().unwrap(); - let (ec_shards, deleted_ec_shards) = store.delete_expired_ec_volumes(); - build_heartbeat_with_ec_status( - config, - &mut store, - deleted_ec_shards, - ec_shards.is_empty(), - true, - ) + let (_, expired_ec) = state.store.read().unwrap().find_expired_ec_volumes(); + let mut deleted_ec_shards = Vec::new(); + if !expired_ec.is_empty() { + deleted_ec_shards = state + .store + .write() + .unwrap() + .remove_expired_ec_volumes(expired_ec) + .0; + } + #[cfg(test)] + { + let store_id = state.store.read().unwrap().id.clone(); + read_phase_hook::park(read_phase_hook::Point::BeforeVolumePass, &store_id); + } + let (heartbeat, volumes, actions) = { + let store = state.store.read().unwrap(); + #[cfg(test)] + read_phase_hook::park(read_phase_hook::Point::VolumePass, &store.id); + // Taken with the volume list: a shard mounted since the EC phase must + // not go out as "no EC shards", which clears it on the master. + let has_no_ec_shards = !has_reportable_ec_shards(&store); + build_heartbeat_with_ec_status(config, &store, deleted_ec_shards, has_no_ec_shards, true) + }; + apply_volume_actions(state, actions); + (heartbeat, volumes) } /// Lists the volumes the server holds without touching reporting state or @@ -807,8 +839,100 @@ fn collect_volume_snapshot( config: &HeartbeatConfig, state: &Arc, ) -> Vec { - let mut store = state.store.write().unwrap(); - build_heartbeat_with_ec_status(config, &mut store, Vec::new(), true, false).1 + let (_, volumes, actions) = build_heartbeat_with_ec_status( + config, + &state.store.read().unwrap(), + Vec::new(), + true, + false, + ); + apply_volume_actions(state, actions); + volumes +} + +/// Store changes a heartbeat pass decides on under the read lock, applied +/// under a short write lock afterwards. Entries are (disk index, volume id). +#[derive(Default)] +struct VolumeActions { + delete_expired: Vec<(usize, VolumeId)>, + quarantine: Vec<(usize, VolumeId)>, +} + +fn apply_volume_actions(state: &VolumeServerState, actions: VolumeActions) { + if actions.delete_expired.is_empty() && actions.quarantine.is_empty() { + return; + } + apply_volume_actions_to(&mut state.store.write().unwrap(), actions); +} + +/// Each target is re-checked: it may have been written to, replaced or removed +/// since the read pass chose it. +fn apply_volume_actions_to(store: &mut Store, actions: VolumeActions) { + let volume_size_limit = store.volume_size_limit.load(Ordering::Relaxed); + for (disk_id, vid) in actions.delete_expired { + let Some(loc) = store.locations.get_mut(disk_id) else { + continue; + }; + let still_expired = loc.find_volume(vid).is_some_and(|vol| { + !vol.should_quarantine() + && vol.is_expired(vol.dat_file_size().unwrap_or(0), volume_size_limit) + && vol.is_expired_long_enough(MAX_TTL_VOLUME_REMOVAL_DELAY) + }); + if still_expired { + let _ = loc.delete_volume(vid, false, false, false); + } + } + for (disk_id, vid) in actions.quarantine { + if let Some(vol) = store + .locations + .get_mut(disk_id) + .and_then(|loc| loc.find_volume_mut(vid)) + && vol.should_quarantine() + { + vol.set_no_write_or_delete(true); + } + } +} + +#[cfg(test)] +mod read_phase_hook { + use std::sync::Mutex; + use std::sync::mpsc::Receiver; + use tokio::sync::oneshot::Sender; + + #[derive(Clone, Copy, PartialEq)] + pub(super) enum Point { + /// Between the EC phase and the volume pass, holding no lock. + BeforeVolumePass, + /// Inside the volume pass, holding the store read lock. + VolumePass, + } + + type Park = (Point, String, Sender<()>, Receiver<()>); + static ARMED: Mutex> = Mutex::new(Vec::new()); + + /// Parks the next pass over the store with this id at `point`, announcing + /// itself on `entered` and waiting until `release` is dropped. + pub(super) fn arm(point: Point, store_id: &str, entered: Sender<()>, release: Receiver<()>) { + ARMED + .lock() + .unwrap() + .push((point, store_id.to_string(), entered, release)); + } + + pub(super) fn park(point: Point, store_id: &str) { + let armed = { + let mut armed = ARMED.lock().unwrap(); + armed + .iter() + .position(|(p, id, _, _)| *p == point && !id.is_empty() && id == store_id) + .map(|i| armed.swap_remove(i)) + }; + if let Some((_, _, entered, release)) = armed { + let _ = entered.send(()); + let _ = release.recv(); + } + } } /// The heartbeat alone, without the volume snapshot the send loop pairs it @@ -851,27 +975,42 @@ fn collect_location_metadata( (location_uuids, disk_tags) } +/// Whether a heartbeat would report any EC shard: Go's non-empty +/// `ecVolumeMessages` from `deleteExpiredEcVolumes`. +fn has_reportable_ec_shards(store: &Store) -> bool { + store.locations.iter().any(|loc| { + loc.ec_volumes().any(|(_, ec_vol)| { + !ec_vol.is_time_to_destroy() + && !ec_vol.should_quarantine() + && ec_vol.shards.iter().any(Option::is_some) + }) + }) +} + #[cfg(test)] fn build_heartbeat(config: &HeartbeatConfig, store: &mut Store) -> master_pb::Heartbeat { let has_no_ec_shards = collect_live_ec_shards(store, false).is_empty(); - build_heartbeat_with_ec_status(config, store, Vec::new(), has_no_ec_shards, true).0 + let (heartbeat, _, actions) = + build_heartbeat_with_ec_status(config, store, Vec::new(), has_no_ec_shards, true); + apply_volume_actions_to(store, actions); + heartbeat } -/// Returns the heartbeat to send and, separately, every volume held. The -/// caller derives mount and unmount deltas by diffing successive snapshots, so -/// it must not be handed the partial list a heartbeat may carry. +/// Returns the heartbeat to send, every volume held, and the store changes +/// the pass decided on. The caller derives mount and unmount deltas by diffing +/// successive snapshots, so it must not be handed the partial list a heartbeat +/// may carry. fn build_heartbeat_with_ec_status( config: &HeartbeatConfig, - store: &mut Store, + store: &Store, deleted_ec_shards: Vec, has_no_ec_shards: bool, commit_report: bool, ) -> ( master_pb::Heartbeat, Vec, + VolumeActions, ) { - const MAX_TTL_VOLUME_REMOVAL_DELAY: u32 = 10; - #[derive(Default)] struct ReadOnlyCounts { is_read_only: u32, @@ -901,8 +1040,9 @@ fn build_heartbeat_with_ec_status( // Per-disk effective max for DiskTag, captured alongside the per-type sum. let mut disk_max_by_id = vec![0i32; store.locations.len()]; + let mut actions = VolumeActions::default(); - for (disk_id, loc) in store.locations.iter_mut().enumerate() { + for (disk_id, loc) in store.locations.iter().enumerate() { let disk_type_str = loc.disk_type.to_string(); let mut effective_max_count = loc.max_volume_count.load(Ordering::Relaxed); if loc.is_disk_space_low.load(Ordering::Relaxed) { @@ -926,8 +1066,6 @@ fn build_heartbeat_with_ec_status( *disk_free_bytes.entry(disk_type_str).or_insert(0) += loc.disk_free_bytes.load(Ordering::Relaxed); - let mut delete_vids = Vec::new(); - let mut quarantine_vids: Vec = Vec::new(); for (_, vol) in loc.iter_volumes() { let cur_max = vol.max_file_key(); if cur_max > max_file_key { @@ -947,7 +1085,7 @@ fn build_heartbeat_with_ec_status( ); } quarantined_volumes += 1; - quarantine_vids.push(vol.id); + actions.quarantine.push((disk_id, vol.id)); continue; } else if !vol.is_expired(volume_size, volume_size_limit) { // Detect phantom volumes: the .dat was unlinked from disk but is still @@ -1015,7 +1153,7 @@ fn build_heartbeat_with_ec_status( } volumes.push(volume_message); } else if vol.is_expired_long_enough(MAX_TTL_VOLUME_REMOVAL_DELAY) { - delete_vids.push(vol.id); + actions.delete_expired.push((disk_id, vol.id)); should_delete_volume = true; } @@ -1046,16 +1184,6 @@ fn build_heartbeat_with_ec_status( } } } - - for vid in delete_vids { - let _ = loc.delete_volume(vid, false, false, false); - } - - for vid in quarantine_vids { - if let Some(vol) = loc.find_volume_mut(vid) { - vol.set_no_write_or_delete(true); - } - } } // Update disk size and read-only gauges @@ -1152,7 +1280,7 @@ fn build_heartbeat_with_ec_status( disk_tags, ..Default::default() }; - (heartbeat, volumes) + (heartbeat, volumes, actions) } fn collect_live_ec_shards( @@ -1585,7 +1713,7 @@ mod tests { // What the notify path does: collect a snapshot, send a message of its own. let snapshot = - build_heartbeat_with_ec_status(&test_config(), &mut store, Vec::new(), true, false).1; + build_heartbeat_with_ec_status(&test_config(), &store, Vec::new(), true, false).1; assert_eq!(snapshot.len(), 3); let heartbeat = build_heartbeat(&test_config(), &mut store); @@ -2012,6 +2140,235 @@ mod tests { assert!(volume.is_no_write_or_delete()); } + // The pass only reads the store, so it must not shut out the readers that + // serve traffic while it stats every volume. + #[tokio::test] + async fn test_heartbeat_collection_leaves_the_store_readable() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2); + store.id = "heartbeat-read-phase-park".to_string(); + let state = test_state_with_store(store); + + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + read_phase_hook::arm( + read_phase_hook::Point::VolumePass, + "heartbeat-read-phase-park", + entered_tx, + release_rx, + ); + let collection = { + let state = state.clone(); + tokio::spawn(async move { + off_runtime(&test_config(), &state, collect_heartbeat_with_snapshot).await + }) + }; + tokio::time::timeout(Duration::from_secs(10), entered_rx) + .await + .expect("the pass never reached its read phase") + .unwrap(); + + let readable = state.store.try_read().is_ok(); + drop(release_tx); + let (heartbeat, volumes) = collection.await.unwrap().unwrap(); + + assert!(readable, "a parked heartbeat pass shut out store readers"); + assert_eq!(heartbeat.volumes.len(), 2); + assert_eq!(volumes.len(), 2); + } + + // What the read pass decided on can go stale before the write lock is + // taken: each action must re-check its target, not act on whatever now + // holds the id. + #[test] + fn test_volume_actions_skip_volumes_changed_since_the_read_pass() { + let temp_dir = tempfile::tempdir().unwrap(); + let dir = temp_dir.path().to_str().unwrap(); + + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location( + dir, + dir, + 8, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + store.volume_size_limit.store(1, Ordering::Relaxed); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + for id in [61, 62, 63] { + store + .add_volume( + VolumeId(id), + DiskType::HardDrive, + &VolumeSpec { + collection: "stale_action_case", + ttl: Some(crate::storage::needle::ttl::TTL::read("20m").unwrap()), + ..Default::default() + }, + ) + .unwrap(); + let (_, volume) = store.find_volume_mut(VolumeId(id)).unwrap(); + volume.set_last_io_error_for_test(None); + volume.set_last_modified_ts_for_test(now.saturating_sub(60 * 60)); + std::fs::OpenOptions::new() + .write(true) + .open(volume.dat_path()) + .unwrap() + .set_len((crate::storage::super_block::SUPER_BLOCK_SIZE + 1) as u64) + .unwrap(); + } + store + .add_volume( + VolumeId(64), + DiskType::HardDrive, + &VolumeSpec { + collection: "stale_action_case", + ..Default::default() + }, + ) + .unwrap(); + let (_, volume) = store.find_volume_mut(VolumeId(64)).unwrap(); + volume.set_last_io_error_for_test(Some("input/output error")); + + let (heartbeat, _, actions) = + build_heartbeat_with_ec_status(&test_config(), &store, Vec::new(), true, true); + assert!(heartbeat.volumes.is_empty()); + assert_eq!(actions.delete_expired.len(), 3); + assert_eq!(actions.quarantine, vec![(0, VolumeId(64))]); + + // Between the phases: 61 is deleted by someone else, 62 and 64 are + // replaced by fresh copies under the same ids; 63 is left alone. + store + .delete_volume(VolumeId(61), false, false, false) + .unwrap(); + for id in [62, 64] { + store + .delete_volume(VolumeId(id), false, false, false) + .unwrap(); + store + .add_volume( + VolumeId(id), + DiskType::HardDrive, + &VolumeSpec { + collection: "stale_action_case", + ..Default::default() + }, + ) + .unwrap(); + } + + apply_volume_actions_to(&mut store, actions); + + assert!(!store.has_volume(VolumeId(61))); + assert!(store.has_volume(VolumeId(62)), "a fresh copy was deleted"); + assert!( + !store.has_volume(VolumeId(63)), + "the expired volume survived" + ); + let (_, fresh) = store.find_volume(VolumeId(64)).unwrap(); + assert!( + !fresh.is_no_write_or_delete(), + "a fresh copy was quarantined" + ); + } + + // A shard mounted after the EC phase is held when the volume list is + // taken; reporting "no EC shards" alongside it would clear it on the master. + #[tokio::test] + async fn test_ec_shard_mounted_after_the_ec_phase_is_not_reported_absent() { + let temp_dir = tempfile::tempdir().unwrap(); + let dir = temp_dir.path().to_str().unwrap(); + let mut store = reporting_store(dir, 1); + store.id = "heartbeat-ec-mount-between-passes".to_string(); + let state = test_state_with_store(store); + std::fs::write(format!("{}/ec_mount_race_73.ec00", dir), b"shard").unwrap(); + std::fs::write(format!("{}/ec_mount_race_73.ecx", dir), [0u8; 16]).unwrap(); + + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + read_phase_hook::arm( + read_phase_hook::Point::BeforeVolumePass, + "heartbeat-ec-mount-between-passes", + entered_tx, + release_rx, + ); + let collection = { + let state = state.clone(); + tokio::spawn(async move { + off_runtime(&test_config(), &state, collect_heartbeat_with_snapshot).await + }) + }; + tokio::time::timeout(Duration::from_secs(10), entered_rx) + .await + .expect("the pass never finished its EC phase") + .unwrap(); + + state.store.write().unwrap().locations[0] + .mount_ec_shards(VolumeId(73), "ec_mount_race", &[0], "") + .unwrap(); + drop(release_tx); + let (heartbeat, _) = collection.await.unwrap().unwrap(); + + assert!( + !heartbeat.has_no_ec_shards, + "a mounted EC shard was reported absent" + ); + } + + #[test] + fn test_expired_ec_volume_gone_before_removal_is_not_reported_deleted() { + let temp_dir = tempfile::tempdir().unwrap(); + let dir = temp_dir.path().to_str().unwrap(); + + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location( + dir, + dir, + 8, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + for id in [71, 72] { + std::fs::write(format!("{}/stale_ec_case_{}.ec00", dir, id), b"expired").unwrap(); + std::fs::write(format!("{}/stale_ec_case_{}.ecx", dir, id), [0u8; 16]).unwrap(); + store.locations[0] + .mount_ec_shards(VolumeId(id), "stale_ec_case", &[0], "") + .unwrap(); + store + .find_ec_volume_mut(VolumeId(id)) + .unwrap() + .expire_at_sec = 1; + } + + let (ec_shards, mut expired) = store.find_expired_ec_volumes(); + expired.sort(); + assert!(ec_shards.is_empty()); + assert_eq!(expired, vec![(0, VolumeId(71)), (0, VolumeId(72))]); + + // Between the phases: 71 is destroyed by someone else, 72 is remounted + // without an expiry. + store.remove_ec_volume(VolumeId(71)).unwrap().destroy(); + store.remove_ec_volume(VolumeId(72)).unwrap(); + store.locations[0] + .mount_ec_shards(VolumeId(72), "stale_ec_case", &[0], "") + .unwrap(); + let (deleted, still_held) = store.remove_expired_ec_volumes(expired); + + assert!(deleted.is_empty()); + assert_eq!(still_held.len(), 1); + assert_eq!(still_held[0].id, 72); + assert!(store.has_ec_volume(VolumeId(72))); + } + #[test] fn test_build_heartbeat_includes_remote_storage_name_and_key() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 5a7553048..c53e55fed 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -1206,43 +1206,28 @@ impl Store { Vec, Vec, ) { - let mut ec_shards = Vec::new(); - let mut deleted = Vec::new(); + let (mut ec_shards, expired) = self.find_expired_ec_volumes(); + let (deleted, still_held) = self.remove_expired_ec_volumes(expired); + ec_shards.extend(still_held); + (ec_shards, deleted) + } - for (disk_id, loc) in self.locations.iter_mut().enumerate() { - let mut expired_vids = Vec::new(); - let mut io_quarantined_vids = Vec::new(); + /// The read half of `delete_expired_ec_volumes`: the shards to report, and + /// the expired EC volumes, by disk index, for `remove_expired_ec_volumes`. + pub fn find_expired_ec_volumes( + &self, + ) -> ( + Vec, + Vec<(usize, VolumeId)>, + ) { + let mut ec_shards = Vec::new(); + let mut expired = Vec::new(); + + for (disk_id, loc) in self.locations.iter().enumerate() { for (vid, ec_vol) in loc.ec_volumes() { if ec_vol.is_time_to_destroy() { - expired_vids.push(*vid); + expired.push((disk_id, *vid)); } else if ec_vol.should_quarantine() { - io_quarantined_vids.push(*vid); - } else { - ec_shards - .extend(ec_vol.to_volume_ec_shard_information_messages(disk_id as u32)); - } - } - - for vid in expired_vids { - let messages = loc - .find_ec_volume(vid) - .map(|ec_vol| ec_vol.to_volume_ec_shard_information_messages(disk_id as u32)) - .unwrap_or_default(); - if let Some(mut ec_vol) = loc.remove_ec_volume(vid) { - for _ in 0..ec_vol.shard_count() { - crate::metrics::VOLUME_GAUGE - .with_label_values(&[&ec_vol.collection, "ec_shards"]) - .dec(); - } - ec_vol.destroy(); - deleted.extend(messages); - } else { - ec_shards.extend(messages); - } - } - - for vid in io_quarantined_vids { - if let Some(ec_vol) = loc.find_ec_volume(vid) { let (_, io_count, quarantined) = ec_vol.get_io_error_state(); if !quarantined { ec_vol.mark_io_quarantined(); @@ -1252,11 +1237,51 @@ impl Store { "ec volume quarantined after consecutive IO errors" ); } + } else { + ec_shards + .extend(ec_vol.to_volume_ec_shard_information_messages(disk_id as u32)); } } } - (ec_shards, deleted) + (ec_shards, expired) + } + + /// The write half of `delete_expired_ec_volumes`: destroys each volume that + /// is still there and still expired, returning the shards deleted and the + /// shards of any that no longer qualify. + pub fn remove_expired_ec_volumes( + &mut self, + expired: Vec<(usize, VolumeId)>, + ) -> ( + Vec, + Vec, + ) { + let mut deleted = Vec::new(); + let mut still_held = Vec::new(); + for (disk_id, vid) in expired { + let Some(loc) = self.locations.get_mut(disk_id) else { + continue; + }; + let Some(ec_vol) = loc.find_ec_volume(vid) else { + continue; + }; + let messages = ec_vol.to_volume_ec_shard_information_messages(disk_id as u32); + if !ec_vol.is_time_to_destroy() { + still_held.extend(messages); + continue; + } + if let Some(mut ec_vol) = loc.remove_ec_volume(vid) { + for _ in 0..ec_vol.shard_count() { + crate::metrics::VOLUME_GAUGE + .with_label_values(&[&ec_vol.collection, "ec_shards"]) + .dec(); + } + ec_vol.destroy(); + deleted.extend(messages); + } + } + (deleted, still_held) } /// Remove an EC volume from whichever location has it.