From c6b9994b5089687cadd70e78e03d166a93105a14 Mon Sep 17 00:00:00 2001 From: Iisyourdad Date: Sun, 29 Mar 2026 03:06:02 -0500 Subject: [PATCH] Preserve versions for multipart completion --- src/api/multipart.rs | 23 +- src/storage/filesystem.rs | 354 +++++++---- tests/integration.rs | 1195 ++++++++++++++++++++++++++++++------- 3 files changed, 1235 insertions(+), 337 deletions(-) diff --git a/src/api/multipart.rs b/src/api/multipart.rs index b0a7843..99fd269 100644 --- a/src/api/multipart.rs +++ b/src/api/multipart.rs @@ -120,6 +120,9 @@ pub async fn complete_multipart_upload( let mut builder = Response::builder() .status(StatusCode::OK) .header("content-type", "application/xml"); + if let Some(vid) = &result.version_id { + builder = builder.header("x-amz-version-id", vid.as_str()); + } if let (Some(algo), Some(val)) = (&result.checksum_algorithm, &result.checksum_value) { builder = builder.header(algo.header_name(), val.as_str()); } @@ -233,7 +236,9 @@ pub(super) fn map_storage_err(err: StorageError) -> S3Error { match err { StorageError::ChecksumMismatch(_) => S3Error::bad_checksum("x-amz-checksum"), StorageError::UploadNotFound(upload_id) => S3Error::no_such_upload(&upload_id), - StorageError::InvalidKey(msg) if msg.contains("part too small") => S3Error::entity_too_small(), + StorageError::InvalidKey(msg) if msg.contains("part too small") => { + S3Error::entity_too_small() + } StorageError::InvalidKey(msg) if msg.contains("part") || msg.contains("etag") @@ -272,11 +277,21 @@ fn parse_complete_parts(xml: &str) -> Result, S3Error> { }, Ok(quick_xml::events::Event::Text(e)) => { if in_part_number { - let value = e.unescape().map_err(|_| S3Error::malformed_xml())?.into_owned(); - part_number = Some(value.parse::().map_err(|_| S3Error::invalid_part("invalid part number"))?); + let value = e + .unescape() + .map_err(|_| S3Error::malformed_xml())? + .into_owned(); + part_number = Some( + value + .parse::() + .map_err(|_| S3Error::invalid_part("invalid part number"))?, + ); in_part_number = false; } else if in_etag { - let value = e.unescape().map_err(|_| S3Error::malformed_xml())?.into_owned(); + let value = e + .unescape() + .map_err(|_| S3Error::malformed_xml())? + .into_owned(); let normalized = if value.starts_with('"') && value.ends_with('"') { value } else { diff --git a/src/storage/filesystem.rs b/src/storage/filesystem.rs index 7438d25..91ea847 100644 --- a/src/storage/filesystem.rs +++ b/src/storage/filesystem.rs @@ -1,5 +1,8 @@ -use super::{BucketMeta, ByteStream, ChecksumAlgorithm, ChunkInfo, ChunkKind, ChunkManifest, DeleteResult, MultipartUploadMeta, ObjectMeta, PartMeta, PutResult, StorageError}; use super::chunk_reader::VerifiedChunkReader; +use super::{ + BucketMeta, ByteStream, ChecksumAlgorithm, ChunkInfo, ChunkKind, ChunkManifest, DeleteResult, + MultipartUploadMeta, ObjectMeta, PartMeta, PutResult, StorageError, +}; use base64::Engine; use md5::{Digest, Md5}; use rand::RngExt; @@ -46,7 +49,6 @@ impl ChecksumHasher { Self::Sha256(h) => b64.encode(Digest::finalize(h)), } } - } pub struct FilesystemStorage { @@ -62,7 +64,9 @@ fn validate_key(key: &str) -> Result<(), StorageError> { return Err(StorageError::InvalidKey("Key must not be empty".into())); } if key.len() > 1024 { - return Err(StorageError::InvalidKey("Key must not exceed 1024 bytes".into())); + return Err(StorageError::InvalidKey( + "Key must not exceed 1024 bytes".into(), + )); } let path = Path::new(key); for component in path.components() { @@ -94,10 +98,20 @@ fn validate_upload_id(upload_id: &str) -> Result<(), StorageError> { } impl FilesystemStorage { - pub async fn new(data_dir: &str, erasure_coding: bool, chunk_size: u64, parity_shards: u32) -> Result { + pub async fn new( + data_dir: &str, + erasure_coding: bool, + chunk_size: u64, + parity_shards: u32, + ) -> Result { let buckets_dir = Path::new(data_dir).join("buckets"); fs::create_dir_all(&buckets_dir).await?; - Ok(Self { buckets_dir, erasure_coding, chunk_size, parity_shards }) + Ok(Self { + buckets_dir, + erasure_coding, + chunk_size, + parity_shards, + }) } // --- Bucket operations --- @@ -149,7 +163,9 @@ impl FilesystemStorage { // is effectively deleted (head_bucket checks .bucket.json). let meta = BucketMeta { name: name.to_string(), - created_at: chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(), + created_at: chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(), region: String::new(), versioning: false, cors_rules: None, @@ -157,7 +173,8 @@ impl FilesystemStorage { let _ = fs::write( bucket_dir.join(".bucket.json"), serde_json::to_string_pretty(&meta).unwrap_or_default(), - ).await; + ) + .await; Err(StorageError::BucketNotEmpty) } Err(e) => Err(e.into()), @@ -207,9 +224,7 @@ impl FilesystemStorage { } fn ec_dir(&self, bucket: &str, key: &str) -> PathBuf { - self.buckets_dir - .join(bucket) - .join(format!("{}.ec", key)) + self.buckets_dir.join(bucket).join(format!("{}.ec", key)) } fn chunk_path(&self, bucket: &str, key: &str, index: u32) -> PathBuf { @@ -274,7 +289,15 @@ impl FilesystemStorage { } if self.erasure_coding { - return self.put_object_chunked(bucket, key, content_type, body, checksum.as_ref().map(|(a, _)| *a)).await; + return self + .put_object_chunked( + bucket, + key, + content_type, + body, + checksum.as_ref().map(|(a, _)| *a), + ) + .await; } let obj_path = self.object_path(bucket, key); @@ -285,7 +308,9 @@ impl FilesystemStorage { let file = fs::File::create(&obj_path).await?; let mut writer = BufWriter::with_capacity(IO_BUFFER_SIZE, file); let mut hasher = Md5::new(); - let mut checksum_hasher = checksum.as_ref().map(|(algo, _)| ChecksumHasher::new(*algo)); + let mut checksum_hasher = checksum + .as_ref() + .map(|(algo, _)| ChecksumHasher::new(*algo)); let mut size: u64 = 0; let mut buf = vec![0u8; IO_BUFFER_SIZE]; @@ -312,7 +337,8 @@ impl FilesystemStorage { if let Some(expected_val) = expected { if computed != expected_val { return Err(StorageError::ChecksumMismatch(format!( - "expected {}, got {}", expected_val, computed + "expected {}, got {}", + expected_val, computed ))); } } @@ -321,7 +347,9 @@ impl FilesystemStorage { (None, None) }; - let now = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(); + let now = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); let versioned = self.is_versioned(bucket).await.unwrap_or(false); let version_id = if versioned { @@ -393,7 +421,9 @@ impl FilesystemStorage { if n == 0 { // Flush remaining chunk_buf if !chunk_buf.is_empty() { - let ci = self.write_chunk(bucket, key, chunk_index, &chunk_buf).await?; + let ci = self + .write_chunk(bucket, key, chunk_index, &chunk_buf) + .await?; chunks.push(ci); } break; @@ -408,7 +438,9 @@ impl FilesystemStorage { while chunk_buf.len() >= self.chunk_size as usize { let chunk_data: Vec = chunk_buf.drain(..self.chunk_size as usize).collect(); - let ci = self.write_chunk(bucket, key, chunk_index, &chunk_data).await?; + let ci = self + .write_chunk(bucket, key, chunk_index, &chunk_data) + .await?; chunks.push(ci); chunk_index += 1; } @@ -435,8 +467,16 @@ impl FilesystemStorage { chunk_size: self.chunk_size, chunk_count: data_chunk_count, chunks, - parity_shards: if has_parity { Some(self.parity_shards) } else { None }, - shard_size: if has_parity { Some(self.chunk_size) } else { None }, + parity_shards: if has_parity { + Some(self.parity_shards) + } else { + None + }, + shard_size: if has_parity { + Some(self.chunk_size) + } else { + None + }, }; let manifest_json = serde_json::to_string_pretty(&manifest)?; fs::write(self.manifest_path(bucket, key), manifest_json).await?; @@ -445,7 +485,9 @@ impl FilesystemStorage { let etag_quoted = format!("\"{}\"", etag); let checksum_value = checksum_hasher.map(|h| h.finalize_base64()); - let now = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(); + let now = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); let versioned = self.is_versioned(bucket).await.unwrap_or(false); let version_id = if versioned { @@ -454,7 +496,11 @@ impl FilesystemStorage { None }; - let storage_format = if has_parity { "chunked-v2" } else { "chunked-v1" }; + let storage_format = if has_parity { + "chunked-v2" + } else { + "chunked-v1" + }; let meta = ObjectMeta { key: key.to_string(), size: total_size, @@ -525,7 +571,9 @@ impl FilesystemStorage { if k + m > 255 { return Err(StorageError::InvalidKey(format!( "too many shards: {} data + {} parity = {} > 255 (GF(2^8) limit). Increase --chunk-size", - k, m, k + m + k, + m, + k + m ))); } @@ -546,12 +594,10 @@ impl FilesystemStorage { } // Encode parity - let rs = ReedSolomon::new(k, m).map_err(|e| { - StorageError::InvalidKey(format!("Reed-Solomon init error: {e}")) - })?; - rs.encode(&mut all_shards).map_err(|e| { - StorageError::InvalidKey(format!("Reed-Solomon encode error: {e}")) - })?; + let rs = ReedSolomon::new(k, m) + .map_err(|e| StorageError::InvalidKey(format!("Reed-Solomon init error: {e}")))?; + rs.encode(&mut all_shards) + .map_err(|e| StorageError::InvalidKey(format!("Reed-Solomon encode error: {e}")))?; // Write parity chunks to disk let mut parity_infos = Vec::with_capacity(m); @@ -596,7 +642,8 @@ impl FilesystemStorage { let mut buf = vec![0u8; IO_BUFFER_SIZE]; for part in selected { - let mut part_file = fs::File::open(self.part_path(bucket, upload_id, part.part_number)).await?; + let mut part_file = + fs::File::open(self.part_path(bucket, upload_id, part.part_number)).await?; loop { let n = part_file.read(&mut buf).await?; if n == 0 { @@ -607,7 +654,9 @@ impl FilesystemStorage { while chunk_buf.len() >= self.chunk_size as usize { let chunk_data: Vec = chunk_buf.drain(..self.chunk_size as usize).collect(); - let ci = self.write_chunk(bucket, key, chunk_index, &chunk_data).await?; + let ci = self + .write_chunk(bucket, key, chunk_index, &chunk_data) + .await?; chunks.push(ci); chunk_index += 1; } @@ -620,7 +669,9 @@ impl FilesystemStorage { // Flush remaining if !chunk_buf.is_empty() { - let ci = self.write_chunk(bucket, key, chunk_index, &chunk_buf).await?; + let ci = self + .write_chunk(bucket, key, chunk_index, &chunk_buf) + .await?; chunks.push(ci); } @@ -644,45 +695,75 @@ impl FilesystemStorage { chunk_size: self.chunk_size, chunk_count: data_chunk_count, chunks, - parity_shards: if has_parity { Some(self.parity_shards) } else { None }, - shard_size: if has_parity { Some(self.chunk_size) } else { None }, + parity_shards: if has_parity { + Some(self.parity_shards) + } else { + None + }, + shard_size: if has_parity { + Some(self.chunk_size) + } else { + None + }, }; - fs::write(self.manifest_path(bucket, key), serde_json::to_string_pretty(&manifest)?).await?; + fs::write( + self.manifest_path(bucket, key), + serde_json::to_string_pretty(&manifest)?, + ) + .await?; - let etag = format!("\"{}-{}\"", hex::encode(etag_hasher.finalize()), selected.len()); + let etag = format!( + "\"{}-{}\"", + hex::encode(etag_hasher.finalize()), + selected.len() + ); // Compute composite checksum if algorithm was specified - let (checksum_algorithm, checksum_value) = if let Some(algo) = upload_meta.checksum_algorithm { - let b64 = base64::engine::general_purpose::STANDARD; - let mut raw_checksums = Vec::new(); - for part in selected { - if let Some(ref val) = part.checksum_value { - if let Ok(raw) = b64.decode(val) { - raw_checksums.extend_from_slice(&raw); + let (checksum_algorithm, checksum_value) = + if let Some(algo) = upload_meta.checksum_algorithm { + let b64 = base64::engine::general_purpose::STANDARD; + let mut raw_checksums = Vec::new(); + for part in selected { + if let Some(ref val) = part.checksum_value { + if let Ok(raw) = b64.decode(val) { + raw_checksums.extend_from_slice(&raw); + } } } - } - if !raw_checksums.is_empty() { - let mut composite_hasher = ChecksumHasher::new(algo); - composite_hasher.update(&raw_checksums); - let composite = format!("{}-{}", composite_hasher.finalize_base64(), selected.len()); - (Some(algo), Some(composite)) + if !raw_checksums.is_empty() { + let mut composite_hasher = ChecksumHasher::new(algo); + composite_hasher.update(&raw_checksums); + let composite = + format!("{}-{}", composite_hasher.finalize_base64(), selected.len()); + (Some(algo), Some(composite)) + } else { + (Some(algo), None) + } } else { - (Some(algo), None) - } - } else { - (None, None) - }; + (None, None) + }; let part_sizes: Vec = selected.iter().map(|p| p.size).collect(); - let storage_format = if has_parity { "chunked-v2" } else { "chunked-v1" }; + let versioned = self.is_versioned(bucket).await.unwrap_or(false); + let version_id = if versioned { + Some(Self::generate_version_id()) + } else { + None + }; + let storage_format = if has_parity { + "chunked-v2" + } else { + "chunked-v1" + }; let object_meta = ObjectMeta { key: key.to_string(), size: total_size, etag: etag.clone(), content_type: upload_meta.content_type.clone(), - last_modified: chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(), - version_id: None, + last_modified: chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(), + version_id: version_id.clone(), is_delete_marker: false, storage_format: Some(storage_format.to_string()), checksum_algorithm, @@ -696,22 +777,22 @@ impl FilesystemStorage { fs::create_dir_all(parent).await?; } fs::write(&meta_path, serde_json::to_string_pretty(&object_meta)?).await?; + if versioned { + self.write_version_chunked(bucket, key, &object_meta) + .await?; + } let _ = fs::remove_dir_all(self.upload_dir(bucket, upload_id)).await; Ok(PutResult { size: total_size, etag, - version_id: None, + version_id, checksum_algorithm, checksum_value, }) } - async fn put_folder_marker( - &self, - bucket: &str, - key: &str, - ) -> Result { + async fn put_folder_marker(&self, bucket: &str, key: &str) -> Result { let folder_dir = self .buckets_dir .join(bucket) @@ -813,7 +894,9 @@ impl FilesystemStorage { StorageError::Io(e) } })?; - file.seek(std::io::SeekFrom::Start(offset)).await.map_err(StorageError::Io)?; + file.seek(std::io::SeekFrom::Start(offset)) + .await + .map_err(StorageError::Io)?; let mut data = vec![0u8; length as usize]; file.read_exact(&mut data).await.map_err(StorageError::Io)?; return Ok((Box::pin(std::io::Cursor::new(data)), meta)); @@ -825,17 +908,15 @@ impl FilesystemStorage { StorageError::Io(e) } })?; - file.seek(std::io::SeekFrom::Start(offset)).await.map_err(StorageError::Io)?; + file.seek(std::io::SeekFrom::Start(offset)) + .await + .map_err(StorageError::Io)?; let limited = file.take(length); let reader = BufReader::with_capacity(IO_BUFFER_SIZE, limited); Ok((Box::pin(reader), meta)) } - pub async fn head_object( - &self, - bucket: &str, - key: &str, - ) -> Result { + pub async fn head_object(&self, bucket: &str, key: &str) -> Result { validate_key(key)?; self.read_object_meta(bucket, key).await } @@ -864,11 +945,7 @@ impl FilesystemStorage { Ok(()) } - pub async fn delete_object_tagging( - &self, - bucket: &str, - key: &str, - ) -> Result<(), StorageError> { + pub async fn delete_object_tagging(&self, bucket: &str, key: &str) -> Result<(), StorageError> { validate_key(key)?; let mut meta = self.read_object_meta(bucket, key).await?; meta.tags = None; @@ -947,7 +1024,9 @@ impl FilesystemStorage { bucket: bucket.to_string(), key: key.to_string(), content_type: content_type.to_string(), - initiated: chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(), + initiated: chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(), checksum_algorithm, }; @@ -966,7 +1045,9 @@ impl FilesystemStorage { ) -> Result { validate_upload_id(upload_id)?; if part_number == 0 || part_number > 10_000 { - return Err(StorageError::InvalidKey("part number must be 1..=10000".into())); + return Err(StorageError::InvalidKey( + "part number must be 1..=10000".into(), + )); } let upload_dir = self.upload_dir(bucket, upload_id); if !fs::try_exists(&upload_dir).await? { @@ -977,7 +1058,9 @@ impl FilesystemStorage { let file = fs::File::create(&part_path).await?; let mut writer = BufWriter::with_capacity(IO_BUFFER_SIZE, file); let mut hasher = Md5::new(); - let mut checksum_hasher = checksum.as_ref().map(|(algo, _)| ChecksumHasher::new(*algo)); + let mut checksum_hasher = checksum + .as_ref() + .map(|(algo, _)| ChecksumHasher::new(*algo)); let mut size: u64 = 0; let mut buf = vec![0u8; IO_BUFFER_SIZE]; @@ -1002,7 +1085,8 @@ impl FilesystemStorage { if computed != expected_val { let _ = fs::remove_file(&part_path).await; return Err(StorageError::ChecksumMismatch(format!( - "expected {}, got {}", expected_val, computed + "expected {}, got {}", + expected_val, computed ))); } } @@ -1065,7 +1149,9 @@ impl FilesystemStorage { } if self.erasure_coding { - return self.complete_multipart_chunked(bucket, upload_id, &upload_meta, &selected).await; + return self + .complete_multipart_chunked(bucket, upload_id, &upload_meta, &selected) + .await; } let obj_path = self.object_path(bucket, &upload_meta.key); @@ -1079,7 +1165,8 @@ impl FilesystemStorage { let mut buf = vec![0u8; IO_BUFFER_SIZE]; for part in &selected { - let mut part_file = fs::File::open(self.part_path(bucket, upload_id, part.part_number)).await?; + let mut part_file = + fs::File::open(self.part_path(bucket, upload_id, part.part_number)).await?; loop { let n = part_file.read(&mut buf).await?; if n == 0 { @@ -1095,39 +1182,53 @@ impl FilesystemStorage { } writer.flush().await?; - let etag = format!("\"{}-{}\"", hex::encode(etag_hasher.finalize()), selected.len()); + let etag = format!( + "\"{}-{}\"", + hex::encode(etag_hasher.finalize()), + selected.len() + ); // Compute composite checksum if algorithm was specified - let (checksum_algorithm, checksum_value) = if let Some(algo) = upload_meta.checksum_algorithm { - let b64 = base64::engine::general_purpose::STANDARD; - let mut raw_checksums = Vec::new(); - for part in &selected { - if let Some(ref val) = part.checksum_value { - if let Ok(raw) = b64.decode(val) { - raw_checksums.extend_from_slice(&raw); + let (checksum_algorithm, checksum_value) = + if let Some(algo) = upload_meta.checksum_algorithm { + let b64 = base64::engine::general_purpose::STANDARD; + let mut raw_checksums = Vec::new(); + for part in &selected { + if let Some(ref val) = part.checksum_value { + if let Ok(raw) = b64.decode(val) { + raw_checksums.extend_from_slice(&raw); + } } } - } - if !raw_checksums.is_empty() { - let mut composite_hasher = ChecksumHasher::new(algo); - composite_hasher.update(&raw_checksums); - let composite = format!("{}-{}", composite_hasher.finalize_base64(), selected.len()); - (Some(algo), Some(composite)) + if !raw_checksums.is_empty() { + let mut composite_hasher = ChecksumHasher::new(algo); + composite_hasher.update(&raw_checksums); + let composite = + format!("{}-{}", composite_hasher.finalize_base64(), selected.len()); + (Some(algo), Some(composite)) + } else { + (Some(algo), None) + } } else { - (Some(algo), None) - } - } else { - (None, None) - }; + (None, None) + }; let part_sizes: Vec = selected.iter().map(|p| p.size).collect(); + let versioned = self.is_versioned(bucket).await.unwrap_or(false); + let version_id = if versioned { + Some(Self::generate_version_id()) + } else { + None + }; let object_meta = ObjectMeta { key: upload_meta.key.clone(), size: total_size, etag: etag.clone(), content_type: upload_meta.content_type, - last_modified: chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(), - version_id: None, + last_modified: chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(), + version_id: version_id.clone(), is_delete_marker: false, storage_format: None, checksum_algorithm, @@ -1140,12 +1241,16 @@ impl FilesystemStorage { fs::create_dir_all(parent).await?; } fs::write(meta_path, serde_json::to_string_pretty(&object_meta)?).await?; + if versioned { + self.write_version(bucket, &upload_meta.key, &object_meta, &obj_path) + .await?; + } let _ = fs::remove_dir_all(self.upload_dir(bucket, upload_id)).await; Ok(PutResult { size: total_size, etag, - version_id: None, + version_id, checksum_algorithm, checksum_value, }) @@ -1246,11 +1351,7 @@ impl FilesystemStorage { }) } - async fn read_object_meta( - &self, - bucket: &str, - key: &str, - ) -> Result { + async fn read_object_meta(&self, bucket: &str, key: &str) -> Result { let meta_path = self.meta_path(bucket, key); let data = fs::read_to_string(&meta_path).await.map_err(|e| { if e.kind() == std::io::ErrorKind::NotFound { @@ -1297,10 +1398,10 @@ impl FilesystemStorage { // Strip the .ec suffix to get the key let key = rel_str.strip_suffix(".ec").unwrap_or(&rel_str).to_string(); if key.starts_with(prefix) { - if let Ok(meta) = self.read_object_meta( - base.file_name().unwrap().to_str().unwrap(), - &key, - ).await { + if let Ok(meta) = self + .read_object_meta(base.file_name().unwrap().to_str().unwrap(), &key) + .await + { results.push(meta); } } @@ -1328,10 +1429,10 @@ impl FilesystemStorage { if let Ok(rel) = path.strip_prefix(base) { let key = rel.to_string_lossy().to_string(); if key.starts_with(prefix) { - if let Ok(meta) = self.read_object_meta( - base.file_name().unwrap().to_str().unwrap(), - &key, - ).await { + if let Ok(meta) = self + .read_object_meta(base.file_name().unwrap().to_str().unwrap(), &key) + .await + { results.push(meta); } } @@ -1419,11 +1520,7 @@ impl FilesystemStorage { Ok(meta.versioning) } - pub async fn set_versioning( - &self, - bucket: &str, - enabled: bool, - ) -> Result<(), StorageError> { + pub async fn set_versioning(&self, bucket: &str, enabled: bool) -> Result<(), StorageError> { let meta_path = self.buckets_dir.join(bucket).join(".bucket.json"); let data = fs::read_to_string(&meta_path).await.map_err(|e| { if e.kind() == std::io::ErrorKind::NotFound { @@ -1620,11 +1717,7 @@ impl FilesystemStorage { } /// Scan versions for a key and update the top-level files to reflect the latest non-delete-marker. - async fn update_current_version( - &self, - bucket: &str, - key: &str, - ) -> Result<(), StorageError> { + async fn update_current_version(&self, bucket: &str, key: &str) -> Result<(), StorageError> { let ver_dir = self.versions_dir(bucket, key); if !fs::try_exists(&ver_dir).await.unwrap_or(false) { return Ok(()); @@ -1711,7 +1804,9 @@ impl FilesystemStorage { } // Check for chunked version - let ver_ec_dir = self.versions_dir(bucket, key).join(format!("{}.ec", version_id)); + let ver_ec_dir = self + .versions_dir(bucket, key) + .join(format!("{}.ec", version_id)); if ver_ec_dir.is_dir() { let manifest_path = ver_ec_dir.join("manifest.json"); let manifest_data = fs::read_to_string(&manifest_path).await.map_err(|e| { @@ -1734,7 +1829,10 @@ impl FilesystemStorage { StorageError::Io(e) } })?; - Ok((Box::pin(BufReader::with_capacity(IO_BUFFER_SIZE, file)), meta)) + Ok(( + Box::pin(BufReader::with_capacity(IO_BUFFER_SIZE, file)), + meta, + )) } pub async fn head_object_version( @@ -1780,7 +1878,9 @@ impl FilesystemStorage { let _ = fs::remove_file(&ver_meta_path).await; let ver_data_path = self.version_data_path(bucket, key, version_id); let _ = fs::remove_file(&ver_data_path).await; - let ver_ec_dir = self.versions_dir(bucket, key).join(format!("{}.ec", version_id)); + let ver_ec_dir = self + .versions_dir(bucket, key) + .join(format!("{}.ec", version_id)); let _ = fs::remove_dir_all(&ver_ec_dir).await; // Clean up empty versions dir diff --git a/tests/integration.rs b/tests/integration.rs index 98e9453..f5086f5 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -19,7 +19,9 @@ async fn start_server() -> (String, TempDir) { let tmp = TempDir::new().unwrap(); let data_dir = tmp.path().to_str().unwrap().to_string(); - let storage = FilesystemStorage::new(&data_dir, false, 10 * 1024 * 1024, 0).await.unwrap(); + let storage = FilesystemStorage::new(&data_dir, false, 10 * 1024 * 1024, 0) + .await + .unwrap(); let config = Config { port: 0, @@ -45,19 +47,19 @@ async fn start_server() -> (String, TempDir) { let base_url = format!("http://{}", addr); tokio::spawn(async move { - axum::serve(listener, app.into_make_service_with_connect_info::()).await.unwrap(); + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); }); (base_url, tmp) } /// Sign a request with AWS Signature V4. -fn sign_request( - method: &str, - url: &str, - headers: &mut Vec<(String, String)>, - body: &[u8], -) { +fn sign_request(method: &str, url: &str, headers: &mut Vec<(String, String)>, body: &[u8]) { let parsed = reqwest::Url::parse(url).unwrap(); let host = parsed.host_str().unwrap(); let port = parsed.port().unwrap(); @@ -101,7 +103,11 @@ fn sign_request( }) .collect(); pairs.sort(); - pairs.iter().map(|(k, v)| format!("{}={}", k, v)).collect::>().join("&") + pairs + .iter() + .map(|(k, v)| format!("{}={}", k, v)) + .collect::>() + .join("&") }; let canonical_request = format!( @@ -151,12 +157,7 @@ fn client() -> reqwest::Client { } /// Sign a request using comma-only separators (no spaces), like mc does. -fn sign_request_compact( - method: &str, - url: &str, - headers: &mut Vec<(String, String)>, - body: &[u8], -) { +fn sign_request_compact(method: &str, url: &str, headers: &mut Vec<(String, String)>, body: &[u8]) { // Reuse the same signing logic but produce compact auth header let parsed = reqwest::Url::parse(url).unwrap(); let host = parsed.host_str().unwrap(); @@ -199,7 +200,11 @@ fn sign_request_compact( }) .collect(); pairs.sort(); - pairs.iter().map(|(k, v)| format!("{}={}", k, v)).collect::>().join("&") + pairs + .iter() + .map(|(k, v)| format!("{}={}", k, v)) + .collect::>() + .join("&") }; let canonical_request = format!( @@ -242,11 +247,7 @@ fn sign_request_compact( } /// Build a signed request and send it. -async fn s3_request( - method: &str, - url: &str, - body: Vec, -) -> reqwest::Response { +async fn s3_request(method: &str, url: &str, body: Vec) -> reqwest::Response { let mut headers = Vec::new(); sign_request(method, url, &mut headers, &body); @@ -336,11 +337,7 @@ async fn s3_request_with_headers( } /// Build a signed request with compact auth header (no spaces after commas). -async fn s3_request_compact( - method: &str, - url: &str, - body: Vec, -) -> reqwest::Response { +async fn s3_request_compact(method: &str, url: &str, body: Vec) -> reqwest::Response { let mut headers = Vec::new(); sign_request_compact(method, url, &mut headers, &body); @@ -366,10 +363,7 @@ async fn s3_request_compact( } /// Build a PUT request with STREAMING-AWS4-HMAC-SHA256-PAYLOAD (AWS chunked encoding). -async fn s3_put_chunked( - url: &str, - data: &[u8], -) -> reqwest::Response { +async fn s3_put_chunked(url: &str, data: &[u8]) -> reqwest::Response { let parsed = reqwest::Url::parse(url).unwrap(); let host = parsed.host_str().unwrap(); let port = parsed.port().unwrap(); @@ -388,7 +382,10 @@ async fn s3_put_chunked( ("host".to_string(), host_header.clone()), ("x-amz-content-sha256".to_string(), payload_hash.to_string()), ("x-amz-date".to_string(), amz_date.clone()), - ("x-amz-decoded-content-length".to_string(), data.len().to_string()), + ( + "x-amz-decoded-content-length".to_string(), + data.len().to_string(), + ), ]; sign_headers.sort_by(|a, b| a.0.cmp(&b.0)); @@ -446,9 +443,7 @@ async fn s3_put_chunked( ); chunked_body.extend_from_slice(data); chunked_body.extend_from_slice(b"\r\n"); - chunked_body.extend_from_slice( - format!("0;chunk-signature={}\r\n", chunk_sig).as_bytes(), - ); + chunked_body.extend_from_slice(format!("0;chunk-signature={}\r\n", chunk_sig).as_bytes()); client() .put(url) @@ -472,6 +467,15 @@ fn extract_xml_tag(body: &str, tag: &str) -> Option { Some(body[from..to].to_string()) } +async fn enable_bucket_versioning(base_url: &str, bucket: &str) -> reqwest::Response { + let xml = br#"Enabled"#; + s3_request( + "PUT", + &format!("{}/{}?versioning=", base_url, bucket), + xml.to_vec(), + ) + .await +} // ---- Tests ---- @@ -652,12 +656,7 @@ async fn test_list_objects() { ) .await; - let resp = s3_request( - "GET", - &format!("{}/mybucket?list-type=2", base_url), - vec![], - ) - .await; + let resp = s3_request("GET", &format!("{}/mybucket?list-type=2", base_url), vec![]).await; assert_eq!(resp.status(), 200); let body = resp.text().await.unwrap(); assert!(body.contains("a.txt")); @@ -697,21 +696,55 @@ async fn test_last_modified_http_date_format() { // HEAD should return RFC 7231 Last-Modified let resp = s3_request("HEAD", &format!("{}/mybucket/file.txt", base_url), vec![]).await; assert_eq!(resp.status(), 200); - let last_modified = resp.headers().get("last-modified").unwrap().to_str().unwrap(); + let last_modified = resp + .headers() + .get("last-modified") + .unwrap() + .to_str() + .unwrap(); // Should match pattern like "Mon, 17 Feb 2026 22:17:45 GMT" - assert!(last_modified.ends_with(" GMT"), "Last-Modified should end with GMT: {}", last_modified); - assert!(last_modified.contains(", "), "Last-Modified should contain comma-space: {}", last_modified); + assert!( + last_modified.ends_with(" GMT"), + "Last-Modified should end with GMT: {}", + last_modified + ); + assert!( + last_modified.contains(", "), + "Last-Modified should contain comma-space: {}", + last_modified + ); // Must NOT be ISO 8601 (no "T" between date and time digits) - assert!(!last_modified.contains("T0"), "Last-Modified must not be ISO 8601: {}", last_modified); - assert!(!last_modified.contains("T1"), "Last-Modified must not be ISO 8601: {}", last_modified); - assert!(!last_modified.contains("T2"), "Last-Modified must not be ISO 8601: {}", last_modified); + assert!( + !last_modified.contains("T0"), + "Last-Modified must not be ISO 8601: {}", + last_modified + ); + assert!( + !last_modified.contains("T1"), + "Last-Modified must not be ISO 8601: {}", + last_modified + ); + assert!( + !last_modified.contains("T2"), + "Last-Modified must not be ISO 8601: {}", + last_modified + ); // GET should also return RFC 7231 Last-Modified let resp = s3_request("GET", &format!("{}/mybucket/file.txt", base_url), vec![]).await; - let last_modified = resp.headers().get("last-modified").unwrap().to_str().unwrap(); + let last_modified = resp + .headers() + .get("last-modified") + .unwrap() + .to_str() + .unwrap(); assert!(last_modified.ends_with(" GMT")); // Verify it parses as HTTP date (day-of-week, DD Mon YYYY HH:MM:SS GMT) - assert!(last_modified.len() > 25, "Last-Modified should be full HTTP date: {}", last_modified); + assert!( + last_modified.len() > 25, + "Last-Modified should be full HTTP date: {}", + last_modified + ); } #[tokio::test] @@ -723,11 +756,7 @@ async fn test_put_object_aws_chunked_encoding() { s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; let data = b"hello chunked world"; - let resp = s3_put_chunked( - &format!("{}/mybucket/chunked.txt", base_url), - data, - ) - .await; + let resp = s3_put_chunked(&format!("{}/mybucket/chunked.txt", base_url), data).await; assert_eq!(resp.status(), 200); assert!(resp.headers().contains_key("etag")); @@ -735,7 +764,11 @@ async fn test_put_object_aws_chunked_encoding() { let resp = s3_request("GET", &format!("{}/mybucket/chunked.txt", base_url), vec![]).await; assert_eq!(resp.status(), 200); let body = resp.bytes().await.unwrap(); - assert_eq!(body.as_ref(), data, "Chunked upload content should be decoded"); + assert_eq!( + body.as_ref(), + data, + "Chunked upload content should be decoded" + ); } #[tokio::test] @@ -753,7 +786,11 @@ async fn test_put_object_response_headers() { .await; assert_eq!(resp.status(), 200); let etag = resp.headers().get("etag").unwrap().to_str().unwrap(); - assert!(etag.starts_with('"') && etag.ends_with('"'), "ETag should be quoted: {}", etag); + assert!( + etag.starts_with('"') && etag.ends_with('"'), + "ETag should be quoted: {}", + etag + ); // HEAD should return Content-Type, Content-Length, ETag, Last-Modified let resp = s3_request("HEAD", &format!("{}/mybucket/file.txt", base_url), vec![]).await; @@ -777,9 +814,24 @@ async fn test_delete_objects_batch() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - s3_request("PUT", &format!("{}/mybucket/a.txt", base_url), b"aaa".to_vec()).await; - s3_request("PUT", &format!("{}/mybucket/b.txt", base_url), b"bbb".to_vec()).await; - s3_request("PUT", &format!("{}/mybucket/c.txt", base_url), b"ccc".to_vec()).await; + s3_request( + "PUT", + &format!("{}/mybucket/a.txt", base_url), + b"aaa".to_vec(), + ) + .await; + s3_request( + "PUT", + &format!("{}/mybucket/b.txt", base_url), + b"bbb".to_vec(), + ) + .await; + s3_request( + "PUT", + &format!("{}/mybucket/c.txt", base_url), + b"ccc".to_vec(), + ) + .await; // Batch delete a.txt and b.txt let delete_xml = r#" @@ -796,7 +848,10 @@ async fn test_delete_objects_batch() { .await; assert_eq!(resp.status(), 200); let body = resp.text().await.unwrap(); - assert!(body.contains(""), "Response should contain Deleted elements"); + assert!( + body.contains(""), + "Response should contain Deleted elements" + ); assert!(body.contains("a.txt")); assert!(body.contains("b.txt")); @@ -825,7 +880,12 @@ async fn test_trailing_slash_bucket_routes() { assert_eq!(resp.status(), 200); // GET (list) with trailing slash - let resp = s3_request("GET", &format!("{}/mybucket/?list-type=2", base_url), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/mybucket/?list-type=2", base_url), + vec![], + ) + .await; assert_eq!(resp.status(), 200); // DELETE with trailing slash @@ -861,7 +921,10 @@ async fn test_chunked_upload_interrupted_then_retry() { ("host".to_string(), host_header.clone()), ("x-amz-content-sha256".to_string(), payload_hash.to_string()), ("x-amz-date".to_string(), amz_date.clone()), - ("x-amz-decoded-content-length".to_string(), "1000".to_string()), + ( + "x-amz-decoded-content-length".to_string(), + "1000".to_string(), + ), ]; sign_headers.sort_by(|a, b| a.0.cmp(&b.0)); @@ -879,7 +942,8 @@ async fn test_chunked_upload_interrupted_then_retry() { let scope = format!("{}/{}/s3/aws4_request", date_stamp, REGION); let string_to_sign = format!( "AWS4-HMAC-SHA256\n{}\n{}\n{}", - amz_date, scope, + amz_date, + scope, hex::encode(Sha256::digest(canonical_request.as_bytes())) ); let key = format!("AWS4{}", SECRET_KEY); @@ -929,7 +993,11 @@ async fn test_chunked_upload_interrupted_then_retry() { // Now do a proper chunked upload to the same key — this MUST succeed let good_data = b"hello after interrupted upload"; let resp = s3_put_chunked(&url, good_data).await; - assert_eq!(resp.status(), 200, "Retry upload after interrupted should succeed"); + assert_eq!( + resp.status(), + 200, + "Retry upload after interrupted should succeed" + ); // Verify content is from the successful retry, not the partial upload let resp = s3_request("GET", &url, vec![]).await; @@ -970,7 +1038,10 @@ async fn test_chunked_upload_multi_chunk() { ("host".to_string(), host_header.clone()), ("x-amz-content-sha256".to_string(), payload_hash.to_string()), ("x-amz-date".to_string(), amz_date.clone()), - ("x-amz-decoded-content-length".to_string(), total_len.to_string()), + ( + "x-amz-decoded-content-length".to_string(), + total_len.to_string(), + ), ]; sign_headers.sort_by(|a, b| a.0.cmp(&b.0)); @@ -988,7 +1059,8 @@ async fn test_chunked_upload_multi_chunk() { let scope = format!("{}/{}/s3/aws4_request", date_stamp, REGION); let string_to_sign = format!( "AWS4-HMAC-SHA256\n{}\n{}\n{}", - amz_date, scope, + amz_date, + scope, hex::encode(Sha256::digest(canonical_request.as_bytes())) ); let key = format!("AWS4{}", SECRET_KEY); @@ -1024,9 +1096,7 @@ async fn test_chunked_upload_multi_chunk() { chunked_body.extend_from_slice(b"\r\n"); } // Terminating chunk - chunked_body.extend_from_slice( - format!("0;chunk-signature={}\r\n", chunk_sig).as_bytes(), - ); + chunked_body.extend_from_slice(format!("0;chunk-signature={}\r\n", chunk_sig).as_bytes()); let resp = client() .put(&url) @@ -1048,7 +1118,11 @@ async fn test_chunked_upload_multi_chunk() { assert_eq!(resp.status(), 200); let body = resp.bytes().await.unwrap(); let expected = b"first chunk data second chunk data third chunk data"; - assert_eq!(body.as_ref(), expected, "Multi-chunk content should be concatenated"); + assert_eq!( + body.as_ref(), + expected, + "Multi-chunk content should be concatenated" + ); // Verify content-length matches let resp = s3_request("HEAD", &url, vec![]).await; @@ -1063,7 +1137,12 @@ async fn test_multipart_create_upload() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let resp = s3_request("POST", &format!("{}/mybucket/large.bin?uploads=", base_url), vec![]).await; + let resp = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; assert_eq!(resp.status(), 200); let body = resp.text().await.unwrap(); let upload_id = extract_xml_tag(&body, "UploadId").unwrap(); @@ -1074,12 +1153,20 @@ async fn test_multipart_create_upload() { async fn test_multipart_upload_part() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/large.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); let resp = s3_request( "PUT", - &format!("{}/mybucket/large.bin?partNumber=1&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/large.bin?partNumber=1&uploadId={}", + base_url, upload_id + ), b"part-one".to_vec(), ) .await; @@ -1092,25 +1179,48 @@ async fn test_multipart_upload_part() { async fn test_multipart_complete() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/large.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); let p1 = vec![b'a'; 5 * 1024 * 1024]; let p2 = b"tail".to_vec(); let r1 = s3_request( "PUT", - &format!("{}/mybucket/large.bin?partNumber=1&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/large.bin?partNumber=1&uploadId={}", + base_url, upload_id + ), p1.clone(), ) .await; - let e1 = r1.headers().get("etag").unwrap().to_str().unwrap().to_string(); + let e1 = r1 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); let r2 = s3_request( "PUT", - &format!("{}/mybucket/large.bin?partNumber=2&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/large.bin?partNumber=2&uploadId={}", + base_url, upload_id + ), p2.clone(), ) .await; - let e2 = r2.headers().get("etag").unwrap().to_str().unwrap().to_string(); + let e2 = r2 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); let complete_xml = format!( "1{}2{}", @@ -1132,27 +1242,172 @@ async fn test_multipart_complete() { assert_eq!(body.as_ref(), expected.as_slice()); } +#[tokio::test] +async fn test_multipart_complete_in_versioned_bucket_preserves_prior_versions() { + let (base_url, _tmp) = start_server().await; + assert_eq!( + s3_request("PUT", &format!("{}/mybucket", base_url), vec![]) + .await + .status(), + 200 + ); + assert_eq!( + enable_bucket_versioning(&base_url, "mybucket") + .await + .status(), + 200 + ); + + let create_v1 = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; + let upload_id_v1 = extract_xml_tag(&create_v1.text().await.unwrap(), "UploadId").unwrap(); + let part_v1 = vec![b'a'; 5 * 1024 * 1024]; + let upload_part_v1 = s3_request( + "PUT", + &format!( + "{}/mybucket/large.bin?partNumber=1&uploadId={}", + base_url, upload_id_v1 + ), + part_v1.clone(), + ) + .await; + let etag_v1 = upload_part_v1 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + let complete_v1 = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploadId={}", base_url, upload_id_v1), + format!( + "1{}", + etag_v1 + ) + .into_bytes(), + ) + .await; + assert_eq!(complete_v1.status(), 200); + let version_id_v1 = complete_v1 + .headers() + .get("x-amz-version-id") + .expect("multipart completion should return version id") + .to_str() + .unwrap() + .to_string(); + + let create_v2 = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; + let upload_id_v2 = extract_xml_tag(&create_v2.text().await.unwrap(), "UploadId").unwrap(); + let part_v2 = vec![b'b'; 5 * 1024 * 1024]; + let upload_part_v2 = s3_request( + "PUT", + &format!( + "{}/mybucket/large.bin?partNumber=1&uploadId={}", + base_url, upload_id_v2 + ), + part_v2.clone(), + ) + .await; + let etag_v2 = upload_part_v2 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + let complete_v2 = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploadId={}", base_url, upload_id_v2), + format!( + "1{}", + etag_v2 + ) + .into_bytes(), + ) + .await; + assert_eq!(complete_v2.status(), 200); + let version_id_v2 = complete_v2 + .headers() + .get("x-amz-version-id") + .expect("multipart overwrite should return version id") + .to_str() + .unwrap() + .to_string(); + assert_ne!(version_id_v1, version_id_v2); + + let current = s3_request("GET", &format!("{}/mybucket/large.bin", base_url), vec![]).await; + assert_eq!(current.status(), 200); + assert_eq!(current.bytes().await.unwrap().as_ref(), part_v2.as_slice()); + + let prior = s3_request( + "GET", + &format!( + "{}/mybucket/large.bin?versionId={}", + base_url, version_id_v1 + ), + vec![], + ) + .await; + assert_eq!(prior.status(), 200); + assert_eq!(prior.bytes().await.unwrap().as_ref(), part_v1.as_slice()); +} + #[tokio::test] async fn test_multipart_get_part_number() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/parts.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/parts.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); let p1 = vec![b'A'; 5 * 1024 * 1024]; let p2 = vec![b'B'; 3 * 1024 * 1024]; let r1 = s3_request( "PUT", - &format!("{}/mybucket/parts.bin?partNumber=1&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/parts.bin?partNumber=1&uploadId={}", + base_url, upload_id + ), p1.clone(), - ).await; - let e1 = r1.headers().get("etag").unwrap().to_str().unwrap().to_string(); + ) + .await; + let e1 = r1 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); let r2 = s3_request( "PUT", - &format!("{}/mybucket/parts.bin?partNumber=2&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/parts.bin?partNumber=2&uploadId={}", + base_url, upload_id + ), p2.clone(), - ).await; - let e2 = r2.headers().get("etag").unwrap().to_str().unwrap().to_string(); + ) + .await; + let e2 = r2 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); let complete_xml = format!( "1{}2{}", @@ -1162,26 +1417,34 @@ async fn test_multipart_get_part_number() { "POST", &format!("{}/mybucket/parts.bin?uploadId={}", base_url, upload_id), complete_xml.into_bytes(), - ).await; + ) + .await; assert_eq!(complete.status(), 200); // GET partNumber=1 should return only part 1 data - let get_p1 = s3_request("GET", &format!("{}/mybucket/parts.bin?partNumber=1", base_url), vec![]).await; + let get_p1 = s3_request( + "GET", + &format!("{}/mybucket/parts.bin?partNumber=1", base_url), + vec![], + ) + .await; assert_eq!(get_p1.status(), 206); assert_eq!( get_p1.headers().get("content-length").unwrap(), &(5 * 1024 * 1024).to_string() ); - assert_eq!( - get_p1.headers().get("x-amz-mp-parts-count").unwrap(), - "2" - ); + assert_eq!(get_p1.headers().get("x-amz-mp-parts-count").unwrap(), "2"); let body1 = get_p1.bytes().await.unwrap(); assert_eq!(body1.len(), 5 * 1024 * 1024); assert!(body1.iter().all(|&b| b == b'A')); // GET partNumber=2 should return only part 2 data - let get_p2 = s3_request("GET", &format!("{}/mybucket/parts.bin?partNumber=2", base_url), vec![]).await; + let get_p2 = s3_request( + "GET", + &format!("{}/mybucket/parts.bin?partNumber=2", base_url), + vec![], + ) + .await; assert_eq!(get_p2.status(), 206); assert_eq!( get_p2.headers().get("content-length").unwrap(), @@ -1192,39 +1455,64 @@ async fn test_multipart_get_part_number() { assert!(body2.iter().all(|&b| b == b'B')); // HEAD partNumber=1 should return part-specific headers - let head_p1 = s3_request("HEAD", &format!("{}/mybucket/parts.bin?partNumber=1", base_url), vec![]).await; + let head_p1 = s3_request( + "HEAD", + &format!("{}/mybucket/parts.bin?partNumber=1", base_url), + vec![], + ) + .await; assert_eq!(head_p1.status(), 206); assert_eq!( head_p1.headers().get("content-length").unwrap(), &(5 * 1024 * 1024).to_string() ); - assert_eq!( - head_p1.headers().get("x-amz-mp-parts-count").unwrap(), - "2" - ); + assert_eq!(head_p1.headers().get("x-amz-mp-parts-count").unwrap(), "2"); } #[tokio::test] async fn test_multipart_complete_part_too_small() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/large.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); let r1 = s3_request( "PUT", - &format!("{}/mybucket/large.bin?partNumber=1&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/large.bin?partNumber=1&uploadId={}", + base_url, upload_id + ), b"tiny".to_vec(), ) .await; - let e1 = r1.headers().get("etag").unwrap().to_str().unwrap().to_string(); + let e1 = r1 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); let r2 = s3_request( "PUT", - &format!("{}/mybucket/large.bin?partNumber=2&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/large.bin?partNumber=2&uploadId={}", + base_url, upload_id + ), b"tail".to_vec(), ) .await; - let e2 = r2.headers().get("etag").unwrap().to_str().unwrap().to_string(); + let e2 = r2 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); let complete_xml = format!( "1{}2{}", @@ -1245,7 +1533,12 @@ async fn test_multipart_complete_part_too_small() { async fn test_multipart_abort() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/large.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); let abort = s3_request( @@ -1261,12 +1554,20 @@ async fn test_multipart_abort() { async fn test_multipart_list_parts() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/large.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); s3_request( "PUT", - &format!("{}/mybucket/large.bin?partNumber=1&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/large.bin?partNumber=1&uploadId={}", + base_url, upload_id + ), b"part-one".to_vec(), ) .await; @@ -1286,7 +1587,12 @@ async fn test_multipart_list_parts() { async fn test_multipart_list_uploads() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/large.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/large.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); let list = s3_request("GET", &format!("{}/mybucket?uploads=", base_url), vec![]).await; @@ -1316,7 +1622,12 @@ async fn test_multipart_no_such_upload() { async fn test_multipart_excluded_from_list_objects() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/in-progress.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/in-progress.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); s3_request( "PUT", @@ -1338,25 +1649,48 @@ async fn test_multipart_excluded_from_list_objects() { async fn test_multipart_etag_format() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - let create = s3_request("POST", &format!("{}/mybucket/etag.bin?uploads=", base_url), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/mybucket/etag.bin?uploads=", base_url), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); let p1 = vec![b'a'; 5 * 1024 * 1024]; let p2 = b"tail".to_vec(); let r1 = s3_request( "PUT", - &format!("{}/mybucket/etag.bin?partNumber=1&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/etag.bin?partNumber=1&uploadId={}", + base_url, upload_id + ), p1, ) .await; - let e1 = r1.headers().get("etag").unwrap().to_str().unwrap().to_string(); + let e1 = r1 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); let r2 = s3_request( "PUT", - &format!("{}/mybucket/etag.bin?partNumber=2&uploadId={}", base_url, upload_id), + &format!( + "{}/mybucket/etag.bin?partNumber=2&uploadId={}", + base_url, upload_id + ), p2, ) .await; - let e2 = r2.headers().get("etag").unwrap().to_str().unwrap().to_string(); + let e2 = r2 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); let complete_xml = format!( "1{}2{}", e1, e2 @@ -1379,7 +1713,12 @@ async fn test_copy_object_basic() { s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; // Upload source object - s3_request("PUT", &format!("{}/mybucket/src.txt", base_url), b"copy me".to_vec()).await; + s3_request( + "PUT", + &format!("{}/mybucket/src.txt", base_url), + b"copy me".to_vec(), + ) + .await; // Copy to new key in same bucket let resp = s3_request_with_headers( @@ -1408,7 +1747,12 @@ async fn test_copy_object_cross_bucket() { s3_request("PUT", &format!("{}/src-bucket", base_url), vec![]).await; s3_request("PUT", &format!("{}/dst-bucket", base_url), vec![]).await; - s3_request("PUT", &format!("{}/src-bucket/file.txt", base_url), b"cross bucket".to_vec()).await; + s3_request( + "PUT", + &format!("{}/src-bucket/file.txt", base_url), + b"cross bucket".to_vec(), + ) + .await; let resp = s3_request_with_headers( "PUT", @@ -1450,7 +1794,11 @@ async fn test_copy_object_metadata_copy() { // HEAD destination — content-type should be preserved let resp = s3_request("HEAD", &format!("{}/mybucket/dst.txt", base_url), vec![]).await; assert_eq!( - resp.headers().get("content-type").unwrap().to_str().unwrap(), + resp.headers() + .get("content-type") + .unwrap() + .to_str() + .unwrap(), "text/plain" ); } @@ -1483,7 +1831,11 @@ async fn test_copy_object_metadata_replace() { let resp = s3_request("HEAD", &format!("{}/mybucket/dst.txt", base_url), vec![]).await; assert_eq!( - resp.headers().get("content-type").unwrap().to_str().unwrap(), + resp.headers() + .get("content-type") + .unwrap() + .to_str() + .unwrap(), "application/json" ); } @@ -1509,7 +1861,12 @@ async fn test_copy_object_source_not_found() { async fn test_copy_object_no_leading_slash() { let (base_url, _tmp) = start_server().await; s3_request("PUT", &format!("{}/mybucket", base_url), vec![]).await; - s3_request("PUT", &format!("{}/mybucket/src.txt", base_url), b"no slash".to_vec()).await; + s3_request( + "PUT", + &format!("{}/mybucket/src.txt", base_url), + b"no slash".to_vec(), + ) + .await; // Copy source without leading slash let resp = s3_request_with_headers( @@ -1538,7 +1895,10 @@ fn presign_url(base_url: &str, method: &str, path: &str, expires_secs: u64) -> S let credential = format!("{}/{}/{}/s3/aws4_request", ACCESS_KEY, date_stamp, REGION); let mut qs_params = vec![ - ("X-Amz-Algorithm".to_string(), "AWS4-HMAC-SHA256".to_string()), + ( + "X-Amz-Algorithm".to_string(), + "AWS4-HMAC-SHA256".to_string(), + ), ("X-Amz-Credential".to_string(), credential.clone()), ("X-Amz-Date".to_string(), amz_date.clone()), ("X-Amz-Expires".to_string(), expires_secs.to_string()), @@ -1561,7 +1921,8 @@ fn presign_url(base_url: &str, method: &str, path: &str, expires_secs: u64) -> S let scope = format!("{}/{}/s3/aws4_request", date_stamp, REGION); let string_to_sign = format!( "AWS4-HMAC-SHA256\n{}\n{}\n{}", - amz_date, scope, + amz_date, + scope, hex::encode(Sha256::digest(canonical_request.as_bytes())) ); @@ -1582,7 +1943,10 @@ fn presign_url(base_url: &str, method: &str, path: &str, expires_secs: u64) -> S mac.update(string_to_sign.as_bytes()); let signature = hex::encode(mac.finalize().into_bytes()); - format!("{}{}?{}&X-Amz-Signature={}", base_url, path, canonical_qs, signature) + format!( + "{}{}?{}&X-Amz-Signature={}", + base_url, path, canonical_qs, signature + ) } fn percent_encode_s3(input: &str) -> String { @@ -1620,7 +1984,12 @@ async fn test_presigned_put_object() { let presigned = presign_url(&base_url, "PUT", "/presign-put-bucket/uploaded.txt", 300); let body = b"uploaded via presigned PUT"; - let resp = client().put(&presigned).body(body.to_vec()).send().await.unwrap(); + let resp = client() + .put(&presigned) + .body(body.to_vec()) + .send() + .await + .unwrap(); assert_eq!(resp.status(), 200); let url = format!("{}/presign-put-bucket/uploaded.txt", base_url); @@ -1642,7 +2011,14 @@ async fn test_presigned_head_object() { let presigned = presign_url(&base_url, "HEAD", "/presign-head-bucket/test.txt", 300); let resp = client().head(&presigned).send().await.unwrap(); assert_eq!(resp.status(), 200); - assert_eq!(resp.headers().get("content-length").unwrap().to_str().unwrap(), "9"); + assert_eq!( + resp.headers() + .get("content-length") + .unwrap() + .to_str() + .unwrap(), + "9" + ); } #[tokio::test] @@ -1655,7 +2031,8 @@ async fn test_presigned_expired_url() { s3_request("PUT", &url, b"data".to_vec()).await; // Manually craft a presigned URL with a timestamp from 2 hours ago - let parsed = reqwest::Url::parse(&format!("{}/presign-expire-bucket/test.txt", base_url)).unwrap(); + let parsed = + reqwest::Url::parse(&format!("{}/presign-expire-bucket/test.txt", base_url)).unwrap(); let host = parsed.host_str().unwrap(); let port = parsed.port().unwrap(); let host_header = format!("{}:{}", host, port); @@ -1666,7 +2043,10 @@ async fn test_presigned_expired_url() { let credential = format!("{}/{}/{}/s3/aws4_request", ACCESS_KEY, date_stamp, REGION); let mut qs_params = vec![ - ("X-Amz-Algorithm".to_string(), "AWS4-HMAC-SHA256".to_string()), + ( + "X-Amz-Algorithm".to_string(), + "AWS4-HMAC-SHA256".to_string(), + ), ("X-Amz-Credential".to_string(), credential.clone()), ("X-Amz-Date".to_string(), amz_date.clone()), ("X-Amz-Expires".to_string(), "60".to_string()), @@ -1686,7 +2066,8 @@ async fn test_presigned_expired_url() { let scope = format!("{}/{}/s3/aws4_request", date_stamp, REGION); let string_to_sign = format!( "AWS4-HMAC-SHA256\n{}\n{}\n{}", - amz_date, scope, + amz_date, + scope, hex::encode(Sha256::digest(canonical_request.as_bytes())) ); @@ -1772,7 +2153,12 @@ async fn test_console_presign_simple_key() { // Create bucket and upload object via S3 API s3_request("PUT", &format!("{}/cpresign-bucket", base_url), vec![]).await; let body = b"console presign test"; - s3_request("PUT", &format!("{}/cpresign-bucket/test.txt", base_url), body.to_vec()).await; + s3_request( + "PUT", + &format!("{}/cpresign-bucket/test.txt", base_url), + body.to_vec(), + ) + .await; // Login to console API let session = console_login(&base_url).await; @@ -1789,11 +2175,18 @@ async fn test_console_presign_simple_key() { .unwrap(); assert_eq!(resp.status(), 200); let json: serde_json::Value = resp.json().await.unwrap(); - let presigned_url = json["url"].as_str().expect("response should have url field"); + let presigned_url = json["url"] + .as_str() + .expect("response should have url field"); // Fetch the presigned URL without any auth — should succeed let resp = client().get(presigned_url).send().await.unwrap(); - assert_eq!(resp.status(), 200, "presigned URL should return 200, got {}", resp.status()); + assert_eq!( + resp.status(), + 200, + "presigned URL should return 200, got {}", + resp.status() + ); assert_eq!(resp.bytes().await.unwrap().as_ref(), body); } @@ -1825,10 +2218,17 @@ async fn test_console_presign_key_with_spaces() { .unwrap(); assert_eq!(resp.status(), 200); let json: serde_json::Value = resp.json().await.unwrap(); - let presigned_url = json["url"].as_str().expect("response should have url field"); + let presigned_url = json["url"] + .as_str() + .expect("response should have url field"); let resp = client().get(presigned_url).send().await.unwrap(); - assert_eq!(resp.status(), 200, "presigned URL for key with spaces should return 200, got {}", resp.status()); + assert_eq!( + resp.status(), + 200, + "presigned URL for key with spaces should return 200, got {}", + resp.status() + ); assert_eq!(resp.bytes().await.unwrap().as_ref(), body); } @@ -1858,10 +2258,17 @@ async fn test_console_presign_nested_key() { .unwrap(); assert_eq!(resp.status(), 200); let json: serde_json::Value = resp.json().await.unwrap(); - let presigned_url = json["url"].as_str().expect("response should have url field"); + let presigned_url = json["url"] + .as_str() + .expect("response should have url field"); let resp = client().get(presigned_url).send().await.unwrap(); - assert_eq!(resp.status(), 200, "presigned URL for nested key should return 200, got {}", resp.status()); + assert_eq!( + resp.status(), + 200, + "presigned URL for nested key should return 200, got {}", + resp.status() + ); assert_eq!(resp.bytes().await.unwrap().as_ref(), body); } @@ -2166,17 +2573,27 @@ async fn test_folder_marker_with_children() { .await; let body = resp.text().await.unwrap(); assert!(body.contains("docs/"), "body: {}", body); - assert!(!body.contains("readme.txt"), "readme.txt should not appear at root"); + assert!( + !body.contains("readme.txt"), + "readme.txt should not appear at root" + ); // List inside docs/ — should see readme.txt let resp = s3_request( "GET", - &format!("{}/mybucket?list-type=2&prefix=docs%2F&delimiter=%2F", base_url), + &format!( + "{}/mybucket?list-type=2&prefix=docs%2F&delimiter=%2F", + base_url + ), vec![], ) .await; let body = resp.text().await.unwrap(); - assert!(body.contains("docs/readme.txt"), "body: {}", body); + assert!( + body.contains("docs/readme.txt"), + "body: {}", + body + ); // Delete folder marker — the child object should still exist s3_request("DELETE", &format!("{}/mybucket/docs/", base_url), vec![]).await; @@ -2196,7 +2613,12 @@ async fn test_delete_folder_marker() { // Create and then delete folder marker s3_request("PUT", &format!("{}/mybucket/empty-dir/", base_url), vec![]).await; - s3_request("DELETE", &format!("{}/mybucket/empty-dir/", base_url), vec![]).await; + s3_request( + "DELETE", + &format!("{}/mybucket/empty-dir/", base_url), + vec![], + ) + .await; // HeadObject should now return 404 let resp = s3_request("HEAD", &format!("{}/mybucket/empty-dir/", base_url), vec![]).await; @@ -2211,7 +2633,9 @@ async fn start_server_ec() -> (String, TempDir) { let data_dir = tmp.path().to_str().unwrap().to_string(); // Use 1KB chunk size for easy multi-chunk testing - let storage = FilesystemStorage::new(&data_dir, true, 1024, 0).await.unwrap(); + let storage = FilesystemStorage::new(&data_dir, true, 1024, 0) + .await + .unwrap(); let config = Config { port: 0, @@ -2237,7 +2661,12 @@ async fn start_server_ec() -> (String, TempDir) { let base_url = format!("http://{}", addr); tokio::spawn(async move { - axum::serve(listener, app.into_make_service_with_connect_info::()).await.unwrap(); + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); }); (base_url, tmp) @@ -2259,13 +2688,138 @@ async fn test_ec_put_and_get_object() { .await; // GET should return identical data - let resp = s3_request("GET", &format!("{}/testbucket/bigfile.bin", base_url), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/testbucket/bigfile.bin", base_url), + vec![], + ) + .await; assert_eq!(resp.status(), 200); let body = resp.bytes().await.unwrap(); assert_eq!(body.len(), 3 * 1024); assert_eq!(&body[..], &data[..]); } +#[tokio::test] +async fn test_ec_multipart_complete_in_versioned_bucket_preserves_prior_versions() { + let (base_url, _tmp) = start_server_ec().await; + assert_eq!( + s3_request("PUT", &format!("{}/mybucket", base_url), vec![]) + .await + .status(), + 200 + ); + assert_eq!( + enable_bucket_versioning(&base_url, "mybucket") + .await + .status(), + 200 + ); + + let create_v1 = s3_request( + "POST", + &format!("{}/mybucket/chunked.bin?uploads=", base_url), + vec![], + ) + .await; + let upload_id_v1 = extract_xml_tag(&create_v1.text().await.unwrap(), "UploadId").unwrap(); + let part_v1 = vec![b'a'; 2048]; + let upload_part_v1 = s3_request( + "PUT", + &format!( + "{}/mybucket/chunked.bin?partNumber=1&uploadId={}", + base_url, upload_id_v1 + ), + part_v1.clone(), + ) + .await; + let etag_v1 = upload_part_v1 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + let complete_v1 = s3_request( + "POST", + &format!("{}/mybucket/chunked.bin?uploadId={}", base_url, upload_id_v1), + format!( + "1{}", + etag_v1 + ) + .into_bytes(), + ) + .await; + assert_eq!(complete_v1.status(), 200); + let version_id_v1 = complete_v1 + .headers() + .get("x-amz-version-id") + .expect("erasure-coded multipart completion should return version id") + .to_str() + .unwrap() + .to_string(); + + let create_v2 = s3_request( + "POST", + &format!("{}/mybucket/chunked.bin?uploads=", base_url), + vec![], + ) + .await; + let upload_id_v2 = extract_xml_tag(&create_v2.text().await.unwrap(), "UploadId").unwrap(); + let part_v2 = vec![b'b'; 3072]; + let upload_part_v2 = s3_request( + "PUT", + &format!( + "{}/mybucket/chunked.bin?partNumber=1&uploadId={}", + base_url, upload_id_v2 + ), + part_v2.clone(), + ) + .await; + let etag_v2 = upload_part_v2 + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + let complete_v2 = s3_request( + "POST", + &format!("{}/mybucket/chunked.bin?uploadId={}", base_url, upload_id_v2), + format!( + "1{}", + etag_v2 + ) + .into_bytes(), + ) + .await; + assert_eq!(complete_v2.status(), 200); + let version_id_v2 = complete_v2 + .headers() + .get("x-amz-version-id") + .expect("erasure-coded multipart overwrite should return version id") + .to_str() + .unwrap() + .to_string(); + assert_ne!(version_id_v1, version_id_v2); + + let current = s3_request("GET", &format!("{}/mybucket/chunked.bin", base_url), vec![]).await; + assert_eq!(current.status(), 200); + assert_eq!(current.bytes().await.unwrap().as_ref(), part_v2.as_slice()); + + let prior = s3_request( + "GET", + &format!( + "{}/mybucket/chunked.bin?versionId={}", + base_url, version_id_v1 + ), + vec![], + ) + .await; + assert_eq!(prior.status(), 200); + assert_eq!(prior.bytes().await.unwrap().as_ref(), part_v1.as_slice()); +} + #[tokio::test] async fn test_ec_small_object() { let (base_url, _tmp) = start_server_ec().await; @@ -2333,10 +2887,20 @@ async fn test_ec_delete_object() { let ec_dir = tmp.path().join("buckets/testbucket/todelete.txt.ec"); assert!(ec_dir.exists(), "EC dir should exist after PUT"); - s3_request("DELETE", &format!("{}/testbucket/todelete.txt", base_url), vec![]).await; + s3_request( + "DELETE", + &format!("{}/testbucket/todelete.txt", base_url), + vec![], + ) + .await; assert!(!ec_dir.exists(), "EC dir should be removed after DELETE"); - let resp = s3_request("GET", &format!("{}/testbucket/todelete.txt", base_url), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/testbucket/todelete.txt", base_url), + vec![], + ) + .await; assert_eq!(resp.status(), 404); } @@ -2366,9 +2930,24 @@ async fn test_ec_etag_matches_flat_file() { ) .await; - let etag_flat = resp_flat.headers().get("etag").unwrap().to_str().unwrap().to_string(); - let etag_ec = resp_ec.headers().get("etag").unwrap().to_str().unwrap().to_string(); - assert_eq!(etag_flat, etag_ec, "ETags should match between flat and EC storage"); + let etag_flat = resp_flat + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + let etag_ec = resp_ec + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + assert_eq!( + etag_flat, etag_ec, + "ETags should match between flat and EC storage" + ); } #[tokio::test] @@ -2397,8 +2976,10 @@ async fn test_ec_bitrot_detection() { // If we get a response, reading the body should fail or status should be 500 if resp.status() == 200 { let body_result = resp.bytes().await; - assert!(body_result.is_err() || body_result.unwrap() != vec![0xAA; 2048], - "Should not return original uncorrupted data"); + assert!( + body_result.is_err() || body_result.unwrap() != vec![0xAA; 2048], + "Should not return original uncorrupted data" + ); } } Err(_) => { @@ -2437,7 +3018,11 @@ async fn test_ec_list_objects() { assert!(body.contains("file1.txt"), "body: {}", body); assert!(body.contains("file2.txt"), "body: {}", body); // .ec directories should NOT appear as objects - assert!(!body.contains(".ec"), "body should not contain .ec: {}", body); + assert!( + !body.contains(".ec"), + "body should not contain .ec: {}", + body + ); } // --- Checksum tests --- @@ -2464,23 +3049,45 @@ async fn test_put_object_with_crc32_checksum() { .await; assert_eq!(resp.status(), 200); assert_eq!( - resp.headers().get("x-amz-checksum-crc32").unwrap().to_str().unwrap(), + resp.headers() + .get("x-amz-checksum-crc32") + .unwrap() + .to_str() + .unwrap(), crc_b64 ); // GET should return the checksum header - let resp = s3_request("GET", &format!("{}/checksum-bucket/test.txt", base_url), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/checksum-bucket/test.txt", base_url), + vec![], + ) + .await; assert_eq!(resp.status(), 200); assert_eq!( - resp.headers().get("x-amz-checksum-crc32").unwrap().to_str().unwrap(), + resp.headers() + .get("x-amz-checksum-crc32") + .unwrap() + .to_str() + .unwrap(), crc_b64 ); // HEAD should also return it - let resp = s3_request("HEAD", &format!("{}/checksum-bucket/test.txt", base_url), vec![]).await; + let resp = s3_request( + "HEAD", + &format!("{}/checksum-bucket/test.txt", base_url), + vec![], + ) + .await; assert_eq!(resp.status(), 200); assert_eq!( - resp.headers().get("x-amz-checksum-crc32").unwrap().to_str().unwrap(), + resp.headers() + .get("x-amz-checksum-crc32") + .unwrap() + .to_str() + .unwrap(), crc_b64 ); } @@ -2500,7 +3107,11 @@ async fn test_put_object_with_wrong_checksum() { .await; assert_eq!(resp.status(), 400); let body = resp.text().await.unwrap(); - assert!(body.contains("BadDigest"), "expected BadDigest error: {}", body); + assert!( + body.contains("BadDigest"), + "expected BadDigest error: {}", + body + ); } #[tokio::test] @@ -2521,7 +3132,12 @@ async fn test_put_object_with_algorithm_only() { assert_eq!(resp.status(), 200); // Verify a CRC32C header was returned - let checksum = resp.headers().get("x-amz-checksum-crc32c").unwrap().to_str().unwrap(); + let checksum = resp + .headers() + .get("x-amz-checksum-crc32c") + .unwrap() + .to_str() + .unwrap(); assert!(!checksum.is_empty()); // Verify it's the correct value @@ -2571,7 +3187,11 @@ async fn test_put_object_with_sha256_checksum() { .await; assert_eq!(resp.status(), 200); assert_eq!( - resp.headers().get("x-amz-checksum-sha256").unwrap().to_str().unwrap(), + resp.headers() + .get("x-amz-checksum-sha256") + .unwrap() + .to_str() + .unwrap(), hash_b64 ); } @@ -2584,7 +3204,9 @@ async fn start_server_parity(parity_shards: u32) -> (String, TempDir) { let data_dir = tmp.path().to_str().unwrap().to_string(); // 100-byte chunks for easy multi-chunk testing - let storage = FilesystemStorage::new(&data_dir, true, 100, parity_shards).await.unwrap(); + let storage = FilesystemStorage::new(&data_dir, true, 100, parity_shards) + .await + .unwrap(); let config = Config { port: 0, @@ -2610,7 +3232,12 @@ async fn start_server_parity(parity_shards: u32) -> (String, TempDir) { let base_url = format!("http://{}", addr); tokio::spawn(async move { - axum::serve(listener, app.into_make_service_with_connect_info::()).await.unwrap(); + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); }); (base_url, tmp) @@ -2636,9 +3263,9 @@ async fn test_parity_write_creates_parity_chunks() { assert_eq!(entries.len(), 7, "expected 4 data + 2 parity + 1 manifest"); // Verify manifest - let manifest: serde_json::Value = serde_json::from_str( - &std::fs::read_to_string(ec_dir.join("manifest.json")).unwrap() - ).unwrap(); + let manifest: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string(ec_dir.join("manifest.json")).unwrap()) + .unwrap(); assert_eq!(manifest["version"], 2); assert_eq!(manifest["chunk_count"], 4); assert_eq!(manifest["parity_shards"], 2); @@ -2662,7 +3289,12 @@ async fn test_parity_read_healthy() { s3_request("PUT", &format!("{}/parity-test", base_url), vec![]).await; let data = vec![0xCDu8; 350]; - s3_request("PUT", &format!("{}/parity-test/file.bin", base_url), data.clone()).await; + s3_request( + "PUT", + &format!("{}/parity-test/file.bin", base_url), + data.clone(), + ) + .await; let resp = s3_request("GET", &format!("{}/parity-test/file.bin", base_url), vec![]).await; assert_eq!(resp.status(), 200); @@ -2677,7 +3309,12 @@ async fn test_parity_recovery_corrupted_chunk() { s3_request("PUT", &format!("{}/parity-test", base_url), vec![]).await; let data = vec![0xEFu8; 350]; - s3_request("PUT", &format!("{}/parity-test/file.bin", base_url), data.clone()).await; + s3_request( + "PUT", + &format!("{}/parity-test/file.bin", base_url), + data.clone(), + ) + .await; // Corrupt data chunk 1 (overwrite with zeros) let chunk_path = tmp.path().join("buckets/parity-test/file.bin.ec/000001"); @@ -2697,7 +3334,12 @@ async fn test_parity_recovery_missing_chunk() { s3_request("PUT", &format!("{}/parity-test", base_url), vec![]).await; let data = vec![0x42u8; 350]; - s3_request("PUT", &format!("{}/parity-test/file.bin", base_url), data.clone()).await; + s3_request( + "PUT", + &format!("{}/parity-test/file.bin", base_url), + data.clone(), + ) + .await; // Delete data chunk 0 let chunk_path = tmp.path().join("buckets/parity-test/file.bin.ec/000000"); @@ -2720,14 +3362,17 @@ async fn test_parity_too_many_failures() { // Delete 3 chunks (more than m=2 parity can handle) for i in 0..3 { - let chunk_path = tmp.path().join(format!("buckets/parity-test/file.bin.ec/{:06}", i)); + let chunk_path = tmp + .path() + .join(format!("buckets/parity-test/file.bin.ec/{:06}", i)); std::fs::remove_file(&chunk_path).unwrap(); } // The server will return an error or drop the connection when RS recovery fails. // Since the object is streamed, the error may manifest as a connection reset // rather than a clean HTTP error status. - let result = s3_request_result("GET", &format!("{}/parity-test/file.bin", base_url), vec![]).await; + let result = + s3_request_result("GET", &format!("{}/parity-test/file.bin", base_url), vec![]).await; match result { Err(_) => {} // Connection error — expected Ok(resp) => { @@ -2755,7 +3400,12 @@ async fn test_parity_range_read_degraded() { data.extend(std::iter::repeat(i + 1).take(chunk_len)); } assert_eq!(data.len(), 350); - s3_request("PUT", &format!("{}/parity-test/file.bin", base_url), data.clone()).await; + s3_request( + "PUT", + &format!("{}/parity-test/file.bin", base_url), + data.clone(), + ) + .await; // Corrupt chunk 1 let chunk_path = tmp.path().join("buckets/parity-test/file.bin.ec/000001"); @@ -2782,7 +3432,12 @@ async fn test_parity_backward_compat_v1_manifest() { s3_request("PUT", &format!("{}/compat-test", base_url), vec![]).await; let data = vec![0xAAu8; 2048]; - s3_request("PUT", &format!("{}/compat-test/file.bin", base_url), data.clone()).await; + s3_request( + "PUT", + &format!("{}/compat-test/file.bin", base_url), + data.clone(), + ) + .await; let resp = s3_request("GET", &format!("{}/compat-test/file.bin", base_url), vec![]).await; assert_eq!(resp.status(), 200); @@ -2797,16 +3452,26 @@ async fn test_parity_empty_object() { s3_request("PUT", &format!("{}/parity-test", base_url), vec![]).await; // Empty object — should skip parity - s3_request("PUT", &format!("{}/parity-test/empty.bin", base_url), vec![]).await; + s3_request( + "PUT", + &format!("{}/parity-test/empty.bin", base_url), + vec![], + ) + .await; let ec_dir = tmp.path().join("buckets/parity-test/empty.bin.ec"); - let manifest: serde_json::Value = serde_json::from_str( - &std::fs::read_to_string(ec_dir.join("manifest.json")).unwrap() - ).unwrap(); + let manifest: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string(ec_dir.join("manifest.json")).unwrap()) + .unwrap(); assert_eq!(manifest["version"], 1); // no parity for empty assert!(manifest.get("parity_shards").is_none() || manifest["parity_shards"].is_null()); - let resp = s3_request("GET", &format!("{}/parity-test/empty.bin", base_url), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/parity-test/empty.bin", base_url), + vec![], + ) + .await; assert_eq!(resp.status(), 200); assert_eq!(resp.bytes().await.unwrap().len(), 0); } @@ -2817,17 +3482,28 @@ async fn test_parity_empty_object() { async fn test_put_and_get_object_tagging() { let (base, _tmp) = start_server().await; s3_request("PUT", &format!("{}/tag-bucket", base), vec![]).await; - s3_request("PUT", &format!("{}/tag-bucket/obj.txt", base), b"hello".to_vec()).await; + s3_request( + "PUT", + &format!("{}/tag-bucket/obj.txt", base), + b"hello".to_vec(), + ) + .await; let tagging_xml = r#"envprodteamplatform"#; let resp = s3_request( "PUT", &format!("{}/tag-bucket/obj.txt?tagging", base), tagging_xml.as_bytes().to_vec(), - ).await; + ) + .await; assert_eq!(resp.status(), 200); - let resp = s3_request("GET", &format!("{}/tag-bucket/obj.txt?tagging", base), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/tag-bucket/obj.txt?tagging", base), + vec![], + ) + .await; assert_eq!(resp.status(), 200); let body = resp.text().await.unwrap(); assert!(body.contains("env")); @@ -2840,9 +3516,19 @@ async fn test_put_and_get_object_tagging() { async fn test_get_object_tagging_no_tags() { let (base, _tmp) = start_server().await; s3_request("PUT", &format!("{}/notag-bucket", base), vec![]).await; - s3_request("PUT", &format!("{}/notag-bucket/obj.txt", base), b"hello".to_vec()).await; + s3_request( + "PUT", + &format!("{}/notag-bucket/obj.txt", base), + b"hello".to_vec(), + ) + .await; - let resp = s3_request("GET", &format!("{}/notag-bucket/obj.txt?tagging", base), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/notag-bucket/obj.txt?tagging", base), + vec![], + ) + .await; assert_eq!(resp.status(), 200); let body = resp.text().await.unwrap(); assert!(body.contains("") || body.contains("envprod"#; - s3_request("PUT", &format!("{}/deltag-bucket/obj.txt?tagging", base), tagging_xml.as_bytes().to_vec()).await; + let tagging_xml = + r#"envprod"#; + s3_request( + "PUT", + &format!("{}/deltag-bucket/obj.txt?tagging", base), + tagging_xml.as_bytes().to_vec(), + ) + .await; - let resp = s3_request("DELETE", &format!("{}/deltag-bucket/obj.txt?tagging", base), vec![]).await; + let resp = s3_request( + "DELETE", + &format!("{}/deltag-bucket/obj.txt?tagging", base), + vec![], + ) + .await; assert_eq!(resp.status(), 204); - let resp = s3_request("GET", &format!("{}/deltag-bucket/obj.txt?tagging", base), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/deltag-bucket/obj.txt?tagging", base), + vec![], + ) + .await; assert_eq!(resp.status(), 200); let body = resp.text().await.unwrap(); assert!(!body.contains("")); @@ -2872,7 +3579,12 @@ async fn test_get_object_tagging_no_such_key() { let (base, _tmp) = start_server().await; s3_request("PUT", &format!("{}/nsk-bucket", base), vec![]).await; - let resp = s3_request("GET", &format!("{}/nsk-bucket/nonexistent.txt?tagging", base), vec![]).await; + let resp = s3_request( + "GET", + &format!("{}/nsk-bucket/nonexistent.txt?tagging", base), + vec![], + ) + .await; assert_eq!(resp.status(), 404); let body = resp.text().await.unwrap(); assert!(body.contains("NoSuchKey")); @@ -2882,7 +3594,12 @@ async fn test_get_object_tagging_no_such_key() { async fn test_put_object_tagging_too_many_tags() { let (base, _tmp) = start_server().await; s3_request("PUT", &format!("{}/manytagbucket", base), vec![]).await; - s3_request("PUT", &format!("{}/manytagbucket/obj.txt", base), b"data".to_vec()).await; + s3_request( + "PUT", + &format!("{}/manytagbucket/obj.txt", base), + b"data".to_vec(), + ) + .await; let tags: String = (1..=11) .map(|i| format!("key{}val{}", i, i)) @@ -2892,7 +3609,8 @@ async fn test_put_object_tagging_too_many_tags() { "PUT", &format!("{}/manytagbucket/obj.txt?tagging", base), tagging_xml.into_bytes(), - ).await; + ) + .await; assert_eq!(resp.status(), 400); let body = resp.text().await.unwrap(); assert!(body.contains("InvalidArgument")); @@ -2902,15 +3620,24 @@ async fn test_put_object_tagging_too_many_tags() { async fn test_put_object_tagging_key_too_long() { let (base, _tmp) = start_server().await; s3_request("PUT", &format!("{}/longtag-bucket", base), vec![]).await; - s3_request("PUT", &format!("{}/longtag-bucket/obj.txt", base), b"data".to_vec()).await; + s3_request( + "PUT", + &format!("{}/longtag-bucket/obj.txt", base), + b"data".to_vec(), + ) + .await; let long_key = "k".repeat(129); - let tagging_xml = format!("{}v", long_key); + let tagging_xml = format!( + "{}v", + long_key + ); let resp = s3_request( "PUT", &format!("{}/longtag-bucket/obj.txt?tagging", base), tagging_xml.into_bytes(), - ).await; + ) + .await; assert_eq!(resp.status(), 400); let body = resp.text().await.unwrap(); assert!(body.contains("InvalidArgument")); @@ -2920,15 +3647,24 @@ async fn test_put_object_tagging_key_too_long() { async fn test_put_object_tagging_value_too_long() { let (base, _tmp) = start_server().await; s3_request("PUT", &format!("{}/longval-bucket", base), vec![]).await; - s3_request("PUT", &format!("{}/longval-bucket/obj.txt", base), b"data".to_vec()).await; + s3_request( + "PUT", + &format!("{}/longval-bucket/obj.txt", base), + b"data".to_vec(), + ) + .await; let long_val = "v".repeat(257); - let tagging_xml = format!("k{}", long_val); + let tagging_xml = format!( + "k{}", + long_val + ); let resp = s3_request( "PUT", &format!("{}/longval-bucket/obj.txt?tagging", base), tagging_xml.into_bytes(), - ).await; + ) + .await; assert_eq!(resp.status(), 400); let body = resp.text().await.unwrap(); assert!(body.contains("InvalidArgument")); @@ -2942,25 +3678,46 @@ async fn test_upload_part_copy_full() { // Create source bucket and object s3_request("PUT", &format!("{}/src-upc", base), vec![]).await; let src_data: Vec = (0u8..255).cycle().take(5 * 1024 * 1024).collect(); // 5 MiB - s3_request("PUT", &format!("{}/src-upc/source.bin", base), src_data.clone()).await; + s3_request( + "PUT", + &format!("{}/src-upc/source.bin", base), + src_data.clone(), + ) + .await; // Create destination bucket and start multipart upload s3_request("PUT", &format!("{}/dst-upc", base), vec![]).await; - let create = s3_request("POST", &format!("{}/dst-upc/dest.bin?uploads=", base), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/dst-upc/dest.bin?uploads=", base), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); // UploadPartCopy: copy full source as part 1 let resp = s3_request_with_headers( "PUT", - &format!("{}/dst-upc/dest.bin?partNumber=1&uploadId={}", base, upload_id), + &format!( + "{}/dst-upc/dest.bin?partNumber=1&uploadId={}", + base, upload_id + ), vec![], vec![("x-amz-copy-source", "/src-upc/source.bin")], - ).await; + ) + .await; assert_eq!(resp.status(), 200, "upload_part_copy should return 200"); let body = resp.text().await.unwrap(); - assert!(body.contains(""), "response should be CopyPartResult XML, got: {}", body); + assert!( + body.contains(""), + "response should be CopyPartResult XML, got: {}", + body + ); let etag = extract_xml_tag(&body, "ETag").unwrap(); - assert!(etag.starts_with('"') && etag.ends_with('"'), "ETag should be quoted"); + assert!( + etag.starts_with('"') && etag.ends_with('"'), + "ETag should be quoted" + ); // Complete the multipart upload let complete_xml = format!( @@ -2971,7 +3728,8 @@ async fn test_upload_part_copy_full() { "POST", &format!("{}/dst-upc/dest.bin?uploadId={}", base, upload_id), complete_xml.into_bytes(), - ).await; + ) + .await; assert_eq!(complete.status(), 200); // Verify content matches source @@ -2992,23 +3750,40 @@ async fn test_upload_part_copy_range() { let part2: Vec = vec![b'B'; 1024]; let mut src_data = part1.clone(); src_data.extend_from_slice(&part2); - s3_request("PUT", &format!("{}/src-upcr/source.bin", base), src_data.clone()).await; + s3_request( + "PUT", + &format!("{}/src-upcr/source.bin", base), + src_data.clone(), + ) + .await; // Create destination and start multipart upload s3_request("PUT", &format!("{}/dst-upcr", base), vec![]).await; - let create = s3_request("POST", &format!("{}/dst-upcr/dest.bin?uploads=", base), vec![]).await; + let create = s3_request( + "POST", + &format!("{}/dst-upcr/dest.bin?uploads=", base), + vec![], + ) + .await; let upload_id = extract_xml_tag(&create.text().await.unwrap(), "UploadId").unwrap(); // Part 1: bytes 0 to (5MiB - 1) let r1 = s3_request_with_headers( "PUT", - &format!("{}/dst-upcr/dest.bin?partNumber=1&uploadId={}", base, upload_id), + &format!( + "{}/dst-upcr/dest.bin?partNumber=1&uploadId={}", + base, upload_id + ), vec![], vec![ ("x-amz-copy-source", "/src-upcr/source.bin"), - ("x-amz-copy-source-range", &format!("bytes=0-{}", 5 * 1024 * 1024 - 1)), + ( + "x-amz-copy-source-range", + &format!("bytes=0-{}", 5 * 1024 * 1024 - 1), + ), ], - ).await; + ) + .await; assert_eq!(r1.status(), 200); let body1 = r1.text().await.unwrap(); assert!(body1.contains("")); @@ -3017,13 +3792,20 @@ async fn test_upload_part_copy_range() { // Part 2: remaining bytes let r2 = s3_request_with_headers( "PUT", - &format!("{}/dst-upcr/dest.bin?partNumber=2&uploadId={}", base, upload_id), + &format!( + "{}/dst-upcr/dest.bin?partNumber=2&uploadId={}", + base, upload_id + ), vec![], vec![ ("x-amz-copy-source", "/src-upcr/source.bin"), - ("x-amz-copy-source-range", &format!("bytes={}-{}", 5 * 1024 * 1024, src_data.len() - 1)), + ( + "x-amz-copy-source-range", + &format!("bytes={}-{}", 5 * 1024 * 1024, src_data.len() - 1), + ), ], - ).await; + ) + .await; assert_eq!(r2.status(), 200); let body2 = r2.text().await.unwrap(); assert!(body2.contains("")); @@ -3041,7 +3823,8 @@ async fn test_upload_part_copy_range() { "POST", &format!("{}/dst-upcr/dest.bin?uploadId={}", base, upload_id), complete_xml.into_bytes(), - ).await; + ) + .await; assert_eq!(complete.status(), 200); // Verify reconstructed content matches original source