diff --git a/upstairs/src/volume.rs b/upstairs/src/volume.rs index 74ced4821..e9224a62e 100644 --- a/upstairs/src/volume.rs +++ b/upstairs/src/volume.rs @@ -455,6 +455,55 @@ impl VolumeInner { sv_vec } + fn checked_lba_end(start: u64, length: u64) -> Result { + let Some(end) = start.checked_add(length) else { + crucible_bail!(OffsetInvalid); + }; + + Ok(end) + } + + fn lba_ranges_fully_cover( + start: u64, + end: u64, + ranges: impl Iterator>, + ) -> bool { + if start >= end { + return false; + } + + let mut next = start; + + for range in ranges { + if range.start != next + || range.end <= range.start + || range.end > end + { + return false; + } + + next = range.end; + } + + next == end + } + + fn check_lba_range_is_fully_covered( + start: u64, + end: u64, + affected_sub_volumes: &[(Range, &SubVolume)], + ) -> Result<(), CrucibleError> { + if !Self::lba_ranges_fully_cover( + start, + end, + affected_sub_volumes.iter().map(|(range, _)| range.clone()), + ) { + crucible_bail!(OffsetInvalid); + } + + Ok(()) + } + #[allow(clippy::if_same_then_else)] pub fn read_only_parent_for_lba_range( &self, @@ -508,14 +557,17 @@ impl VolumeInner { self.check_data_size(data.len()).await?; - let affected_sub_volumes = self.sub_volumes_for_lba_range( - offset.0, - data.len() as u64 / self.block_size, - ); + let block_count = data.len() as u64 / self.block_size; + let end = Self::checked_lba_end(offset.0, block_count)?; - if affected_sub_volumes.is_empty() { - crucible_bail!(OffsetInvalid); - } + let affected_sub_volumes = + self.sub_volumes_for_lba_range(offset.0, block_count); + + Self::check_lba_range_is_fully_covered( + offset.0, + end, + &affected_sub_volumes, + )?; // TODO parallel dispatch! for (coverage, sub_volume) in affected_sub_volumes { @@ -960,14 +1012,17 @@ impl BlockIO for VolumeInner { let bs = self.check_data_size(data.len()).await? as usize; - let affected_sub_volumes = self.sub_volumes_for_lba_range( - offset.0, - data.len() as u64 / self.block_size, - ); + let block_count = data.len() as u64 / self.block_size; + let end = Self::checked_lba_end(offset.0, block_count)?; - if affected_sub_volumes.is_empty() { - crucible_bail!(OffsetInvalid); - } + let affected_sub_volumes = + self.sub_volumes_for_lba_range(offset.0, block_count); + + Self::check_lba_range_is_fully_covered( + offset.0, + end, + &affected_sub_volumes, + )?; // TODO parallel dispatch! let mut data_index = 0; @@ -1018,7 +1073,10 @@ impl BlockIO for VolumeInner { * self.block_size as usize; } - assert_eq!(data.len(), data_index); + debug_assert_eq!(data.len(), data_index); + if data.len() != data_index { + crucible_bail!(OffsetInvalid); + } cdt::volume__read__done!(|| (cc, self.uuid)); Ok(()) @@ -1189,60 +1247,15 @@ impl SubVolume { ) -> Option> { assert!(length >= 1); - let end = start + length - 1; - - // No coverage: - // - // lba_range: |-------------| - // argument range: |-------| - // argument range: |--------| - // - - if end < self.lba_range.start { - return None; - } - - if start >= self.lba_range.end { - return None; - } - - // Total coverage: - // - // lba_range: |-------------| - // argument range: |-------| - // argument range: |--------| + let end = start.checked_add(length)?; - if self.lba_range.contains(&start) && self.lba_range.contains(&end) { - return Some(start..(start + length)); - } + let overlap_start = std::cmp::max(start, self.lba_range.start); + let overlap_end = std::cmp::min(end, self.lba_range.end); - // Partial coverage: - - if self.lba_range.contains(&start) { - assert!(!self.lba_range.contains(&end)); - - // lba_range: |-------------| - // argument range: |--------| - // coverage: ^^^ - - Some(start..self.lba_range.end) - } else if self.lba_range.contains(&end) { - assert!(!self.lba_range.contains(&start)); - - // lba_range: |-------------| - // argument range: |-------| - // coverage: ^^^^^^ - Some(self.lba_range.start..(end + 1)) - } else if start < self.lba_range.start && end > self.lba_range.end { - // lba_range: |-------------| - // argument range: |--------------------| - // coverage: ^^^^^^^^^^^^^^^ - Some(self.lba_range.clone()) + if overlap_start < overlap_end { + Some(overlap_start..overlap_end) } else { - panic!( - "should never get here! {:?} {} {}", - self.lba_range, start, length - ); + None } } } @@ -6097,3 +6110,194 @@ mod test { Volume::construct(vcr, None, csl()).await.unwrap_err(); } } + +#[cfg(test)] +mod volume_partial_coverage_tests { + use crate::volume::SubVolume; + use crate::{ + BlockIO, BlockIndex, Buffer, BytesMut, CrucibleError, InMemoryBlockIO, + Volume, VolumeBuilder, + }; + use proptest::prelude::*; + use slog::Logger; + use std::sync::Arc; + use uuid::Uuid; + + async fn two_subvolume_volume( + block_size: usize, + blocks_per_subvolume: usize, + ) -> Volume { + let log = Logger::root(slog::Discard, slog::o!()); + let mut builder = VolumeBuilder::new(block_size as u64, log); + + for _ in 0..2 { + let block_io: Arc = + Arc::new(InMemoryBlockIO::new( + Uuid::new_v4(), + block_size as u64, + block_size * blocks_per_subvolume, + )); + + builder.add_subvolume(block_io).await.unwrap(); + } + + builder.into() + } + + fn subvolume_for_test(lba_range: std::ops::Range) -> SubVolume { + let block_size = 512; + let block_count = lba_range.end - lba_range.start; + let block_io: Arc = + Arc::new(InMemoryBlockIO::new( + Uuid::new_v4(), + block_size, + block_size as usize * block_count as usize, + )); + + SubVolume { + lba_range, + block_io, + } + } + + #[test] + fn subvolume_lba_range_coverage_handles_request_covering_entire_subvolume() + { + let block_size = 512; + let block_io: Arc = + Arc::new(InMemoryBlockIO::new( + Uuid::new_v4(), + block_size as u64, + 2 * block_size, + )); + + let sub_volume = SubVolume { + lba_range: 2..4, + block_io, + }; + + /* + * request: [1, 5) + * subvolume: [2, 4) + * + * The request overlaps the entire subvolume. The helper should return the + * subvolume's coverage range, not panic. + */ + assert_eq!(sub_volume.lba_range_coverage(1, 4), Some(2..4)); + } + + proptest! { + #[test] + fn subvolume_lba_range_coverage_matches_block_intersection( + sub_start in 0u64..32, + sub_len in 1u64..32, + req_start in 0u64..64, + req_len in 1u64..64, + ) { + let sub_end = sub_start + sub_len; + let req_end = req_start + req_len; + let sub_volume = subvolume_for_test(sub_start..sub_end); + + let universe_end = std::cmp::max(sub_end, req_end); + + let covered: Vec = (0..universe_end) + .filter(|&block| { + sub_start <= block + && block < sub_end + && req_start <= block + && block < req_end + }) + .collect(); + + let expected = covered.first().map(|first| { + let last = *covered.last().unwrap(); + *first..(last + 1) + }); + + prop_assert_eq!( + sub_volume.lba_range_coverage(req_start, req_len), + expected, + ); + } + } + + #[test] + fn subvolume_lba_range_coverage_returns_none_when_request_end_overflows() { + let sub_volume = subvolume_for_test(2..4); + + assert_eq!(sub_volume.lba_range_coverage(u64::MAX, 1), None,); + } + + #[tokio::test] + async fn write_rejects_partially_covered_range_before_dispatch() { + let block_size = 512; + let blocks_per_subvolume = 2; + let volume = + two_subvolume_volume(block_size, blocks_per_subvolume).await; + + let last_valid_block = + BlockIndex((2 * blocks_per_subvolume - 1) as u64); + + volume + .write( + last_valid_block, + BytesMut::from(&vec![0x11; block_size][..]), + ) + .await + .unwrap(); + + let result = volume + .write( + last_valid_block, + BytesMut::from(&vec![0x22; 2 * block_size][..]), + ) + .await; + + assert!(matches!(result, Err(CrucibleError::OffsetInvalid))); + + let mut after = Buffer::new(1, block_size); + volume.read(last_valid_block, &mut after).await.unwrap(); + + assert_eq!(&after[..], &vec![0x11; block_size][..]); + } + + #[tokio::test] + async fn write_unwritten_rejects_partially_covered_range_before_dispatch() { + let block_size = 512; + let blocks_per_subvolume = 2; + let volume = + two_subvolume_volume(block_size, blocks_per_subvolume).await; + + let last_valid_block = + BlockIndex((2 * blocks_per_subvolume - 1) as u64); + + let result = volume + .write_unwritten( + last_valid_block, + BytesMut::from(&vec![0x33; 2 * block_size][..]), + ) + .await; + + assert!(matches!(result, Err(CrucibleError::OffsetInvalid))); + } + + #[tokio::test] + async fn read_rejects_partially_covered_range() { + let block_size = 512; + let blocks_per_subvolume = 2; + let volume = + two_subvolume_volume(block_size, blocks_per_subvolume).await; + + /* + * subvolumes: [0, 2) [2, 4) + * requested read: [1, 5) + * + * This intersects the existing subvolumes but is not fully covered by them. + */ + let mut data = Buffer::new(4, block_size); + let result = volume.read(BlockIndex(1), &mut data).await; + + assert!(matches!(result, Err(CrucibleError::OffsetInvalid))); + assert_eq!(&data[..], &vec![0; 4 * block_size][..]); + } +}