From 016cd7b58158cee187b61b5e140ddec5021c1bfa Mon Sep 17 00:00:00 2001 From: Alessandro Passaro Date: Thu, 5 Oct 2023 07:47:26 +0000 Subject: [PATCH 1/3] Use new ChecksummedBlock in DataCache Introduce a new `ChecksummedBlock` type which represents a bytes buffer and its matching checksum. It is a simpler version of `ChecksummedBytes` in that the checksum is always matching the exposed bytes buffer, rather than potentially a containing larger buffer. The new type is better suited to be used in the `DataCache` because it can be more efficiently serialized/deserialized preserving its checksum. This change also introduces a new `checksums` module, containing both `ChecksummedBytes` and `ChecksummedBlock`, in addition to other checksum functions and types. Signed-off-by: Alessandro Passaro --- mountpoint-s3/src/checksums.rs | 39 ++++ mountpoint-s3/src/checksums/block.rs | 211 ++++++++++++++++++ .../bytes.rs} | 28 ++- mountpoint-s3/src/data_cache.rs | 6 +- .../src/data_cache/in_memory_data_cache.rs | 24 +- mountpoint-s3/src/lib.rs | 1 + mountpoint-s3/src/prefetch.rs | 3 +- mountpoint-s3/src/prefetch/feed.rs | 3 +- mountpoint-s3/src/prefetch/part.rs | 2 +- mountpoint-s3/src/prefetch/part_queue.rs | 2 +- mountpoint-s3/src/upload.rs | 21 +- 11 files changed, 289 insertions(+), 51 deletions(-) create mode 100644 mountpoint-s3/src/checksums.rs create mode 100644 mountpoint-s3/src/checksums/block.rs rename mountpoint-s3/src/{prefetch/checksummed_bytes.rs => checksums/bytes.rs} (94%) diff --git a/mountpoint-s3/src/checksums.rs b/mountpoint-s3/src/checksums.rs new file mode 100644 index 000000000..31364f5f5 --- /dev/null +++ b/mountpoint-s3/src/checksums.rs @@ -0,0 +1,39 @@ +mod block; +mod bytes; + +use mountpoint_s3_crt::checksums::crc32c::Crc32c; + +use thiserror::Error; + +pub use block::ChecksummedBlock; +pub use bytes::ChecksummedBytes; + +/// Calculates the combined checksum for `AB` where `prefix_crc` is the checksum for `A`, +/// `suffix_crc` is the checksum for `B`, and `suffic_len` is the length of `B`. +pub fn combine_checksums(prefix_crc: Crc32c, suffix_crc: Crc32c, suffix_len: usize) -> Crc32c { + let combined = ::crc32c::crc32c_combine(prefix_crc.value(), suffix_crc.value(), suffix_len); + Crc32c::new(combined) +} + +#[derive(Debug, Error)] +pub enum IntegrityError { + #[error("Checksum mismatch. expected: {0:?}, actual: {1:?}")] + ChecksumMismatch(Crc32c, Crc32c), +} + +#[cfg(test)] +mod tests { + use super::*; + use mountpoint_s3_crt::checksums::crc32c; + + #[test] + fn test_combine_checksums() { + let buf: &[u8] = b"123456789"; + let (buf1, buf2) = buf.split_at(4); + let crc = crc32c::checksum(buf); + let crc1 = crc32c::checksum(buf1); + let crc2 = crc32c::checksum(buf2); + let combined = combine_checksums(crc1, crc2, buf2.len()); + assert_eq!(combined, crc); + } +} diff --git a/mountpoint-s3/src/checksums/block.rs b/mountpoint-s3/src/checksums/block.rs new file mode 100644 index 000000000..d9a79e808 --- /dev/null +++ b/mountpoint-s3/src/checksums/block.rs @@ -0,0 +1,211 @@ +use bytes::{Bytes, BytesMut}; +use mountpoint_s3_crt::checksums::crc32c::{self, Crc32c}; + +use crate::checksums::{bytes::ChecksummedBytes, combine_checksums, IntegrityError}; + +/// A `ChecksummedBlock` is a bytes buffer that carries its checksum. +/// The implementation guarantees that its integrity will be validated when data is accessed. +#[derive(Debug, Clone)] +pub struct ChecksummedBlock { + bytes: Bytes, + checksum: Crc32c, +} + +impl ChecksummedBlock { + pub fn new(bytes: Bytes, checksum: Crc32c) -> Self { + Self { bytes, checksum } + } + + /// Create `ChecksummedBlock` from `Bytes`, calculating its checksum. + pub fn from_bytes(bytes: Bytes) -> Self { + let checksum = crc32c::checksum(&bytes); + Self::new(bytes, checksum) + } + + /// Convert the `ChecksummedBlock` into `Bytes`, data integrity will be validated before converting. + /// + /// Return `IntegrityError` on data corruption. + pub fn into_bytes(self) -> Result { + self.validate()?; + + Ok(self.bytes) + } + + /// Convert into a `ChecksummedBytes`. + pub fn into_checksummed_bytes(self) -> ChecksummedBytes { + ChecksummedBytes::new(self.bytes, self.checksum) + } + + /// Returns the number of bytes contained in this `ChecksummedBlock`. + pub fn len(&self) -> usize { + self.bytes.len() + } + + /// Returns true if the `ChecksummedBlock` has a length of 0. + pub fn is_empty(&self) -> bool { + self.bytes.is_empty() + } + + /// Append the given bytes to current `ChecksummedBlock`. + pub fn extend(&mut self, extend: ChecksummedBlock) { + if self.is_empty() { + *self = extend; + return; + } + if extend.is_empty() { + return; + } + + let total_len = self.bytes.len() + extend.len(); + let mut bytes_mut = BytesMut::with_capacity(total_len); + bytes_mut.extend_from_slice(&self.bytes); + bytes_mut.extend_from_slice(&extend.bytes); + let new_bytes = bytes_mut.freeze(); + let new_checksum = combine_checksums(self.checksum, extend.checksum, extend.len()); + *self = ChecksummedBlock { + bytes: new_bytes, + checksum: new_checksum, + }; + } + + /// Validate data integrity in this `ChecksummedBlock`. + /// + /// Return `IntegrityError` on data corruption. + pub fn validate(&self) -> Result<(), IntegrityError> { + let checksum = crc32c::checksum(&self.bytes); + if self.checksum != checksum { + return Err(IntegrityError::ChecksumMismatch(self.checksum, checksum)); + } + Ok(()) + } +} + +impl Default for ChecksummedBlock { + fn default() -> Self { + let bytes = Bytes::new(); + let checksum = Crc32c::new(0); + Self { bytes, checksum } + } +} + +impl From for ChecksummedBytes { + fn from(value: ChecksummedBlock) -> Self { + value.into_checksummed_bytes() + } +} + +// Implement equality for tests only. We implement equality, and will panic if the data is corrupted. +#[cfg(test)] +impl PartialEq for ChecksummedBlock { + fn eq(&self, other: &Self) -> bool { + if self.bytes != other.bytes { + return false; + } + + if self.checksum == other.checksum { + return true; + } + + self.validate().expect("should be valid"); + other.validate().expect("should be valid"); + + true + } +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + use mountpoint_s3_crt::checksums::crc32c; + + use super::*; + + #[test] + fn test_into_bytes() { + let bytes = Bytes::from_static(b"some bytes"); + let expected = bytes.clone(); + let checksum = crc32c::checksum(&bytes); + let checksummed_block = ChecksummedBlock::new(bytes, checksum); + + let actual = checksummed_block.into_bytes().unwrap(); + assert_eq!(expected, actual); + } + + #[test] + fn test_into_bytes_integrity_error() { + let bytes = Bytes::from_static(b"some bytes"); + let checksum = crc32c::checksum(&bytes); + let mut checksummed_block = ChecksummedBlock::new(bytes, checksum); + checksummed_block.bytes = Bytes::from_static(b"new bytes"); + + let actual = checksummed_block.into_bytes(); + assert!(matches!(actual, Err(IntegrityError::ChecksumMismatch(_, _)))); + } + + #[test] + fn test_into_checksummed_bytes() { + let bytes = Bytes::from_static(b"some bytes"); + let checksum = crc32c::checksum(&bytes); + let checksummed_block = ChecksummedBlock::new(bytes, checksum); + let checksummed_bytes = checksummed_block.clone().into_checksummed_bytes(); + + assert_eq!( + checksummed_block.into_bytes().unwrap(), + checksummed_bytes.into_bytes().unwrap() + ); + } + + #[test] + fn test_extend() { + let bytes = Bytes::from_static(b"some bytes"); + let expected = Bytes::from_static(b"some bytes extended"); + let checksum = crc32c::checksum(&bytes); + let mut checksummed_block = ChecksummedBlock::new(bytes, checksum); + + let extend = Bytes::from_static(b" extended"); + let extend_checksum = crc32c::checksum(&extend); + let extend = ChecksummedBlock::new(extend, extend_checksum); + checksummed_block.extend(extend); + let actual = checksummed_block.bytes; + assert_eq!(expected, actual); + } + + #[test] + fn test_extend_self_corrupted() { + let bytes = Bytes::from_static(b"some bytes"); + let checksum = crc32c::checksum(&bytes); + let mut checksummed_block = ChecksummedBlock::new(bytes, checksum); + + let currupted_bytes = Bytes::from_static(b"corrupted data"); + checksummed_block.bytes = currupted_bytes.clone(); + + let extend = Bytes::from_static(b" extended"); + let extend_checksum = crc32c::checksum(&extend); + let extend = ChecksummedBlock::new(extend, extend_checksum); + checksummed_block.extend(extend); + assert!(matches!( + checksummed_block.validate(), + Err(IntegrityError::ChecksumMismatch(_, _)) + )); + } + + #[test] + fn test_extend_other_corrupted() { + let bytes = Bytes::from_static(b"some bytes"); + let checksum = crc32c::checksum(&bytes); + let mut checksummed_block = ChecksummedBlock::new(bytes, checksum); + + let extend = Bytes::from_static(b" extended"); + let extend_checksum = crc32c::checksum(&extend); + let mut extend = ChecksummedBlock::new(extend, extend_checksum); + + let currupted_bytes = Bytes::from_static(b"corrupted data"); + extend.bytes = currupted_bytes.clone(); + + checksummed_block.extend(extend); + assert!(matches!( + checksummed_block.validate(), + Err(IntegrityError::ChecksumMismatch(_, _)) + )); + } +} diff --git a/mountpoint-s3/src/prefetch/checksummed_bytes.rs b/mountpoint-s3/src/checksums/bytes.rs similarity index 94% rename from mountpoint-s3/src/prefetch/checksummed_bytes.rs rename to mountpoint-s3/src/checksums/bytes.rs index 4f9592326..b12988722 100644 --- a/mountpoint-s3/src/prefetch/checksummed_bytes.rs +++ b/mountpoint-s3/src/checksums/bytes.rs @@ -1,6 +1,9 @@ +use std::ops::RangeBounds; + use bytes::{Bytes, BytesMut}; use mountpoint_s3_crt::checksums::crc32c::{self, Crc32c}; -use thiserror::Error; + +use crate::checksums::IntegrityError; /// A `ChecksummedBytes` is a bytes buffer that carries its checksum. /// The implementation guarantees that its integrity will be validated when data transformation occurs. @@ -56,6 +59,18 @@ impl ChecksummedBytes { } } + /// Returns a slice of self for the provided range. + /// + /// This operation just increases the reference count and sets a few indices, + /// so there will be no validation and the checksum will not be recomputed. + pub fn slice(&self, range: impl RangeBounds) -> Self { + Self { + orig_bytes: self.orig_bytes.clone(), + curr_slice: self.curr_slice.slice(range), + checksum: self.checksum, + } + } + /// Append the given checksummed bytes to current `ChecksummedBytes`, ensure that data integrity will /// be validated. /// @@ -118,13 +133,6 @@ impl Default for ChecksummedBytes { } } } - -#[derive(Debug, Error)] -pub enum IntegrityError { - #[error("Checksum mismatch. expected: {0:?}, actual: {1:?}")] - ChecksumMismatch(Crc32c, Crc32c), -} - // Implement equality for tests only. We implement equality, and will panic if the data is corrupted. #[cfg(test)] impl PartialEq for ChecksummedBytes { @@ -149,9 +157,7 @@ mod tests { use bytes::Bytes; use mountpoint_s3_crt::checksums::crc32c; - use crate::prefetch::checksummed_bytes::IntegrityError; - - use super::ChecksummedBytes; + use super::*; #[test] fn test_into_bytes() { diff --git a/mountpoint-s3/src/data_cache.rs b/mountpoint-s3/src/data_cache.rs index 8ea85a00a..d19bfc4a0 100644 --- a/mountpoint-s3/src/data_cache.rs +++ b/mountpoint-s3/src/data_cache.rs @@ -10,7 +10,7 @@ use std::ops::Range; use thiserror::Error; -pub use crate::prefetch::checksummed_bytes::ChecksummedBytes; +pub use crate::checksums::ChecksummedBlock; /// Indexes blocks within a given object. pub type BlockIndex = u64; @@ -36,10 +36,10 @@ pub trait DataCache { /// Get block of data from the cache for the given [Key] and [BlockIndex], if available. /// /// Operation may fail due to errors, or return [None] if the block was not available in the cache. - fn get_block(&self, cache_key: &Key, block_idx: BlockIndex) -> DataCacheResult>; + fn get_block(&self, cache_key: &Key, block_idx: BlockIndex) -> DataCacheResult>; /// Put block of data to the cache for the given [Key] and [BlockIndex]. - fn put_block(&self, cache_key: Key, block_idx: BlockIndex, bytes: ChecksummedBytes) -> DataCacheResult<()>; + fn put_block(&self, cache_key: Key, block_idx: BlockIndex, bytes: ChecksummedBlock) -> DataCacheResult<()>; /// Returns the block size for the data cache. fn block_size(&self) -> u64; diff --git a/mountpoint-s3/src/data_cache/in_memory_data_cache.rs b/mountpoint-s3/src/data_cache/in_memory_data_cache.rs index 6843b44e2..bb6a492f1 100644 --- a/mountpoint-s3/src/data_cache/in_memory_data_cache.rs +++ b/mountpoint-s3/src/data_cache/in_memory_data_cache.rs @@ -5,12 +5,12 @@ use std::default::Default; use std::hash::Hash; use std::ops::Range; -use super::{BlockIndex, ChecksummedBytes, DataCache, DataCacheResult}; +use super::{BlockIndex, ChecksummedBlock, DataCache, DataCacheResult}; use crate::sync::RwLock; /// Simple in-memory (RAM) implementation of [DataCache]. Recommended for use in testing only. pub struct InMemoryDataCache { - data: RwLock>>, + data: RwLock>>, block_size: u64, } @@ -25,13 +25,13 @@ impl InMemoryDataCache { } impl DataCache for InMemoryDataCache { - fn get_block(&self, cache_key: &Key, block_idx: BlockIndex) -> DataCacheResult> { + fn get_block(&self, cache_key: &Key, block_idx: BlockIndex) -> DataCacheResult> { let data = self.data.read().unwrap(); let block_data = data.get(cache_key).and_then(|blocks| blocks.get(&block_idx)).cloned(); Ok(block_data) } - fn put_block(&self, cache_key: Key, block_idx: BlockIndex, bytes: ChecksummedBytes) -> DataCacheResult<()> { + fn put_block(&self, cache_key: Key, block_idx: BlockIndex, bytes: ChecksummedBlock) -> DataCacheResult<()> { let mut data = self.data.write().unwrap(); let blocks = data.entry(cache_key).or_default(); blocks.insert(block_idx, bytes); @@ -59,20 +59,18 @@ mod tests { use bytes::Bytes; - use mountpoint_s3_crt::checksums::crc32c; - type TestCacheKey = String; #[test] fn test_put_get() { let data_1 = Bytes::from_static(b"Hello world"); - let data_1 = ChecksummedBytes::new(data_1.clone(), crc32c::checksum(&data_1)); + let data_1 = ChecksummedBlock::from_bytes(data_1.clone()); let data_2 = Bytes::from_static(b"Foo bar"); - let data_2 = ChecksummedBytes::new(data_2.clone(), crc32c::checksum(&data_2)); + let data_2 = ChecksummedBlock::from_bytes(data_2.clone()); let data_3 = Bytes::from_static(b"Baz"); - let data_3 = ChecksummedBytes::new(data_3.clone(), crc32c::checksum(&data_3)); + let data_3 = ChecksummedBlock::from_bytes(data_3.clone()); - let mut cache = InMemoryDataCache::new(8 * 1024 * 1024); + let cache = InMemoryDataCache::new(8 * 1024 * 1024); let cache_key_1: TestCacheKey = String::from("a"); let cache_key_2: TestCacheKey = String::from("b"); @@ -136,11 +134,11 @@ mod tests { #[test] fn test_cached_indices() { let data_1 = Bytes::from_static(b"Hello world"); - let data_1 = ChecksummedBytes::new(data_1.clone(), crc32c::checksum(&data_1)); + let data_1 = ChecksummedBlock::from_bytes(data_1.clone()); let data_2 = Bytes::from_static(b"Foo bar"); - let data_2 = ChecksummedBytes::new(data_2.clone(), crc32c::checksum(&data_2)); + let data_2 = ChecksummedBlock::from_bytes(data_2.clone()); - let mut cache = InMemoryDataCache::new(8 * 1024 * 1024); + let cache = InMemoryDataCache::new(8 * 1024 * 1024); let cache_key_1: TestCacheKey = String::from("a"); let cache_key_2: TestCacheKey = String::from("b"); diff --git a/mountpoint-s3/src/lib.rs b/mountpoint-s3/src/lib.rs index d4550f489..a17cb14b3 100644 --- a/mountpoint-s3/src/lib.rs +++ b/mountpoint-s3/src/lib.rs @@ -1,3 +1,4 @@ +mod checksums; mod data_cache; pub mod fs; pub mod fuse; diff --git a/mountpoint-s3/src/prefetch.rs b/mountpoint-s3/src/prefetch.rs index b4717bc0b..550aa229b 100644 --- a/mountpoint-s3/src/prefetch.rs +++ b/mountpoint-s3/src/prefetch.rs @@ -7,7 +7,6 @@ //! we increase the size of the GetObject requests up to some maximum. If the reader ever makes a //! non-sequential read, we abandon the prefetching and start again with the minimum request size. -pub mod checksummed_bytes; mod feed; mod part; mod part_queue; @@ -26,7 +25,7 @@ use mountpoint_s3_client::ObjectClient; use thiserror::Error; use tracing::{debug_span, error, trace, Instrument}; -use crate::prefetch::checksummed_bytes::{ChecksummedBytes, IntegrityError}; +use crate::checksums::{ChecksummedBytes, IntegrityError}; use crate::prefetch::feed::{ClientPartFeed, ObjectPartFeed}; use crate::prefetch::part::Part; use crate::prefetch::part_queue::{unbounded_part_queue, PartQueue}; diff --git a/mountpoint-s3/src/prefetch/feed.rs b/mountpoint-s3/src/prefetch/feed.rs index 0a7cf2819..b6067cafe 100644 --- a/mountpoint-s3/src/prefetch/feed.rs +++ b/mountpoint-s3/src/prefetch/feed.rs @@ -11,7 +11,8 @@ use mountpoint_s3_client::{ use mountpoint_s3_crt::checksums::crc32c; use tracing::{error, trace}; -use crate::prefetch::{checksummed_bytes::ChecksummedBytes, part::Part, part_queue::PartQueueProducer}; +use crate::checksums::ChecksummedBytes; +use crate::prefetch::{part::Part, part_queue::PartQueueProducer}; /// A generic interface to retrieve data from objects in a S3-like store. #[async_trait] diff --git a/mountpoint-s3/src/prefetch/part.rs b/mountpoint-s3/src/prefetch/part.rs index 8a8af6b25..c3a7b34c0 100644 --- a/mountpoint-s3/src/prefetch/part.rs +++ b/mountpoint-s3/src/prefetch/part.rs @@ -1,6 +1,6 @@ use thiserror::Error; -use super::checksummed_bytes::ChecksummedBytes; +use crate::checksums::ChecksummedBytes; /// A self-identifying part of an S3 object. Users can only retrieve the bytes from this part if /// they can prove they have the correct offset and key. diff --git a/mountpoint-s3/src/prefetch/part_queue.rs b/mountpoint-s3/src/prefetch/part_queue.rs index 42355eff5..07a353bab 100644 --- a/mountpoint-s3/src/prefetch/part_queue.rs +++ b/mountpoint-s3/src/prefetch/part_queue.rs @@ -99,7 +99,7 @@ impl PartQueueProducer { #[cfg(test)] mod tests { - use crate::prefetch::checksummed_bytes::ChecksummedBytes; + use crate::checksums::ChecksummedBytes; use super::*; diff --git a/mountpoint-s3/src/upload.rs b/mountpoint-s3/src/upload.rs index 10cb14b10..c12a1638f 100644 --- a/mountpoint-s3/src/upload.rs +++ b/mountpoint-s3/src/upload.rs @@ -9,6 +9,8 @@ use mountpoint_s3_crt::checksums::crc32c::{Crc32c, Hasher}; use thiserror::Error; use tracing::error; +use crate::checksums::combine_checksums; + type PutRequestError = ObjectClientError::ClientError>; const MAX_S3_MULTIPART_UPLOAD_PARTS: usize = 10000; @@ -180,13 +182,6 @@ fn verify_checksums(review: UploadReview, expected_size: u64, expected_checksum: true } -/// Calculates the combined checksum for `AB` where `prefix_crc` is the checksum for `A`, -/// `suffix_crc` is the checksum for `B`, and `suffic_len` is the length of `B`. -fn combine_checksums(prefix_crc: Crc32c, suffix_crc: Crc32c, suffix_len: usize) -> Crc32c { - let combined = ::crc32c::crc32c_combine(prefix_crc.value(), suffix_crc.value(), suffix_len); - Crc32c::new(combined) -} - #[cfg(test)] mod tests { use std::collections::HashMap; @@ -196,7 +191,6 @@ mod tests { failure_client::countdown_failure_client, mock_client::{MockClient, MockClientConfig, MockClientError}, }; - use mountpoint_s3_crt::checksums::crc32c; use test_case::test_case; #[tokio::test] @@ -341,15 +335,4 @@ mod tests { assert!(!client.contains_key(key)); assert!(!client.is_upload_in_progress(key)); } - - #[test] - fn test_combine_checksums() { - let buf: &[u8] = b"123456789"; - let (buf1, buf2) = buf.split_at(4); - let crc = crc32c::checksum(buf); - let crc1 = crc32c::checksum(buf1); - let crc2 = crc32c::checksum(buf2); - let combined = combine_checksums(crc1, crc2, buf2.len()); - assert_eq!(combined, crc); - } } From 4b18b783b218d9a9ec3d5f095a9cd39dfe706732 Mon Sep 17 00:00:00 2001 From: Alessandro Passaro Date: Tue, 24 Oct 2023 14:20:32 +0100 Subject: [PATCH 2/3] Simplify ChecksummedBlock tests Signed-off-by: Alessandro Passaro --- mountpoint-s3/src/checksums.rs | 2 +- mountpoint-s3/src/checksums/block.rs | 34 ++++++++++------------------ 2 files changed, 13 insertions(+), 23 deletions(-) diff --git a/mountpoint-s3/src/checksums.rs b/mountpoint-s3/src/checksums.rs index 31364f5f5..ef577b292 100644 --- a/mountpoint-s3/src/checksums.rs +++ b/mountpoint-s3/src/checksums.rs @@ -9,7 +9,7 @@ pub use block::ChecksummedBlock; pub use bytes::ChecksummedBytes; /// Calculates the combined checksum for `AB` where `prefix_crc` is the checksum for `A`, -/// `suffix_crc` is the checksum for `B`, and `suffic_len` is the length of `B`. +/// `suffix_crc` is the checksum for `B`, and `suffix_len` is the length of `B`. pub fn combine_checksums(prefix_crc: Crc32c, suffix_crc: Crc32c, suffix_len: usize) -> Crc32c { let combined = ::crc32c::crc32c_combine(prefix_crc.value(), suffix_crc.value(), suffix_len); Crc32c::new(combined) diff --git a/mountpoint-s3/src/checksums/block.rs b/mountpoint-s3/src/checksums/block.rs index d9a79e808..cf216121d 100644 --- a/mountpoint-s3/src/checksums/block.rs +++ b/mountpoint-s3/src/checksums/block.rs @@ -158,31 +158,25 @@ mod tests { #[test] fn test_extend() { let bytes = Bytes::from_static(b"some bytes"); + let extend = Bytes::from_static(b" extended"); let expected = Bytes::from_static(b"some bytes extended"); - let checksum = crc32c::checksum(&bytes); - let mut checksummed_block = ChecksummedBlock::new(bytes, checksum); - let extend = Bytes::from_static(b" extended"); - let extend_checksum = crc32c::checksum(&extend); - let extend = ChecksummedBlock::new(extend, extend_checksum); - checksummed_block.extend(extend); + let mut checksummed_block = ChecksummedBlock::from_bytes(bytes); + let extend_block = ChecksummedBlock::from_bytes(extend); + checksummed_block.extend(extend_block); let actual = checksummed_block.bytes; assert_eq!(expected, actual); } #[test] fn test_extend_self_corrupted() { - let bytes = Bytes::from_static(b"some bytes"); - let checksum = crc32c::checksum(&bytes); - let mut checksummed_block = ChecksummedBlock::new(bytes, checksum); - + let checksum = crc32c::checksum(b"some bytes"); let currupted_bytes = Bytes::from_static(b"corrupted data"); - checksummed_block.bytes = currupted_bytes.clone(); + let mut checksummed_block = ChecksummedBlock::new(currupted_bytes, checksum); let extend = Bytes::from_static(b" extended"); - let extend_checksum = crc32c::checksum(&extend); - let extend = ChecksummedBlock::new(extend, extend_checksum); - checksummed_block.extend(extend); + let extend_block = ChecksummedBlock::from_bytes(extend); + checksummed_block.extend(extend_block); assert!(matches!( checksummed_block.validate(), Err(IntegrityError::ChecksumMismatch(_, _)) @@ -192,17 +186,13 @@ mod tests { #[test] fn test_extend_other_corrupted() { let bytes = Bytes::from_static(b"some bytes"); - let checksum = crc32c::checksum(&bytes); - let mut checksummed_block = ChecksummedBlock::new(bytes, checksum); - - let extend = Bytes::from_static(b" extended"); - let extend_checksum = crc32c::checksum(&extend); - let mut extend = ChecksummedBlock::new(extend, extend_checksum); + let mut checksummed_block = ChecksummedBlock::from_bytes(bytes); let currupted_bytes = Bytes::from_static(b"corrupted data"); - extend.bytes = currupted_bytes.clone(); + let extend_checksum = crc32c::checksum(b" extended"); + let extend_block = ChecksummedBlock::new(currupted_bytes, extend_checksum); - checksummed_block.extend(extend); + checksummed_block.extend(extend_block); assert!(matches!( checksummed_block.validate(), Err(IntegrityError::ChecksumMismatch(_, _)) From 162aa165fa7cfecfc866a3ba83990df9dd382ef5 Mon Sep 17 00:00:00 2001 From: Alessandro Passaro Date: Tue, 24 Oct 2023 14:42:10 +0100 Subject: [PATCH 3/3] Implement From/TryFrom for ChecksummedBlock/Bytes Signed-off-by: Alessandro Passaro --- mountpoint-s3/src/checksums/block.rs | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/mountpoint-s3/src/checksums/block.rs b/mountpoint-s3/src/checksums/block.rs index cf216121d..88e11f236 100644 --- a/mountpoint-s3/src/checksums/block.rs +++ b/mountpoint-s3/src/checksums/block.rs @@ -94,6 +94,20 @@ impl From for ChecksummedBytes { } } +impl From for ChecksummedBlock { + fn from(value: Bytes) -> Self { + Self::from_bytes(value) + } +} + +impl TryFrom for Bytes { + type Error = IntegrityError; + + fn try_from(value: ChecksummedBlock) -> Result { + value.into_bytes() + } +} + // Implement equality for tests only. We implement equality, and will panic if the data is corrupted. #[cfg(test)] impl PartialEq for ChecksummedBlock { @@ -162,8 +176,7 @@ mod tests { let expected = Bytes::from_static(b"some bytes extended"); let mut checksummed_block = ChecksummedBlock::from_bytes(bytes); - let extend_block = ChecksummedBlock::from_bytes(extend); - checksummed_block.extend(extend_block); + checksummed_block.extend(extend.into()); let actual = checksummed_block.bytes; assert_eq!(expected, actual); } @@ -175,8 +188,7 @@ mod tests { let mut checksummed_block = ChecksummedBlock::new(currupted_bytes, checksum); let extend = Bytes::from_static(b" extended"); - let extend_block = ChecksummedBlock::from_bytes(extend); - checksummed_block.extend(extend_block); + checksummed_block.extend(extend.into()); assert!(matches!( checksummed_block.validate(), Err(IntegrityError::ChecksumMismatch(_, _))