Skip to content

upstairs: validate subvolume coverage before IO dispatch #1956

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
SG-devel wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

upstairs: validate subvolume coverage before IO dispatch #1956

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d468a8c
upstairs: add tests for partially covered volume IO
SG-devel Jul 5, 2026
c2ec938
upstairs: reject partially covered volume IO
SG-devel Jul 5, 2026
83d9565
upstairs: add tests for `SubVolume::lba_range_coverage()`
SG-devel Jul 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
336 changes: 270 additions & 66 deletions upstairs/src/volume.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -455,6 +455,55 @@ impl VolumeInner {
sv_vec
}

fn checked_lba_end(start: u64, length: u64) -> Result<u64, CrucibleError> {
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<Item = Range<u64>>,
) -> 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<u64>, &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,
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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(())
Expand Down Expand Up @@ -1189,60 +1247,15 @@ impl SubVolume {
) -> Option<Range<u64>> {
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
}
}
}
Expand Down Expand Up @@ -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<dyn BlockIO + Send + Sync> =
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<u64>) -> SubVolume {
let block_size = 512;
let block_count = lba_range.end - lba_range.start;
let block_io: Arc<dyn BlockIO + Send + Sync> =
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<dyn BlockIO + Send + Sync> =
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<u64> = (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][..]);
}
}