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.