use std::ops::Range;
use std::pin::Pin;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt};
use tracing::{debug, info};
use xet_client::cas_client::Client;
use xet_client::cas_types::{FileChunkHashesResponse, FileRange, HexMerkleHash};
use xet_core_structures::merklehash::{ChunkHashList, MerkleHash, MerkleHashSubtree};
use xet_core_structures::metadata_shard::file_structs::{
FileDataSequenceEntry, FileDataSequenceHeader, FileVerificationEntry, MDBFileInfo,
};
use xet_runtime::core::XetContext;
use super::XetFileInfo;
use super::configurations::TranslatorConfig;
use super::file_cleaner::Sha256Policy;
use super::file_upload_session::FileUploadSession;
use crate::error::{DataError, Result};
use crate::file_reconstruction::FileReconstructor;
pub struct DirtyInput {
pub original_range: Range<u64>,
pub reader: Pin<Box<dyn AsyncRead + Send>>,
pub new_length: u64,
}
const STREAM_BLOCK_SIZE: usize = 4 * 1024 * 1024;
struct UploadedWindow {
start: u64,
end: u64,
chunks: ChunkHashList,
mdb: MDBFileInfo,
}
pub async fn upload_ranges(
config: Arc<TranslatorConfig>,
cas_client: Arc<dyn Client>,
original_hash: MerkleHash,
original_size: u64,
mut dirty_inputs: Vec<DirtyInput>,
) -> Result<XetFileInfo> {
validate_dirty_ranges(&dirty_inputs, original_size)?;
let total_size = compute_total_size(original_size, &dirty_inputs)?;
if dirty_inputs.is_empty() {
debug_assert_eq!(total_size, original_size);
return Ok(XetFileInfo::new(original_hash.hex(), original_size));
}
if original_size == 0 {
return upload_fresh_file(config, dirty_inputs, total_size).await;
}
let recon_result = cas_client.get_file_reconstruction_info(&original_hash).await?;
let original_mdb = recon_result
.map(|(mdb, _)| mdb)
.ok_or_else(|| DataError::ParameterError(format!("file {} not found in CAS", original_hash.hex())))?;
if original_mdb.file_size() != original_size {
return Err(DataError::ParameterError(format!(
"caller said original_size={original_size} but reconstruction info reports {}",
original_mdb.file_size()
)));
}
let mut seg_byte_starts: Vec<u64> = Vec::with_capacity(original_mdb.segments.len() + 1);
seg_byte_starts.push(0);
let mut acc = 0u64;
for s in &original_mdb.segments {
acc += s.unpacked_segment_bytes as u64;
seg_byte_starts.push(acc);
}
let n_segs = original_mdb.segments.len();
let mut snapped: Vec<(u64, u64)> = Vec::with_capacity(dirty_inputs.len());
for input in &dirty_inputs {
let r = &input.original_range;
let (s, e) = if r.start == r.end {
if r.start == original_size {
(seg_byte_starts[n_segs - 1], seg_byte_starts[n_segs])
} else {
(snap_to_segment_start(&seg_byte_starts, r.start), snap_to_segment_end(&seg_byte_starts, r.start + 1))
}
} else {
(snap_to_segment_start(&seg_byte_starts, r.start), snap_to_segment_end(&seg_byte_starts, r.end))
};
snapped.push((s, e));
}
snapped.sort_by_key(|&(s, _)| s);
let mut coalesced: Vec<(u64, u64)> = Vec::with_capacity(snapped.len());
for r in snapped {
if let Some(last) = coalesced.last_mut()
&& r.0 <= last.1
{
last.1 = last.1.max(r.1);
continue;
}
coalesced.push(r);
}
let server_query: Vec<FileRange> = coalesced.iter().map(|&(s, e)| FileRange::new(s, e)).collect();
if server_query.is_empty() {
return Err(DataError::InternalError("internal: non-empty dirty_inputs produced no server query".into()));
}
let response: FileChunkHashesResponse = cas_client.get_file_chunk_hashes(&original_hash, server_query).await?;
if response.windows.is_empty() {
return Err(DataError::InternalError("server returned no windows".into()));
}
if response.hash_ranges.len() != response.windows.len() + 1 {
return Err(DataError::InternalError(format!(
"server returned {} hash_ranges, expected {} (n_windows + 1)",
response.hash_ranges.len(),
response.windows.len() + 1
)));
}
let gap_verification = response.gap_verification;
let ctx = config.ctx.clone();
let session = FileUploadSession::new(config.clone()).await?;
let mut input_idx = 0usize;
let mut uploaded: Vec<UploadedWindow> = Vec::with_capacity(response.windows.len());
let mut buf = vec![0u8; STREAM_BLOCK_SIZE];
for window in response.windows.iter() {
let w_start = window.dirty_byte_range[0];
let w_end = window.dirty_byte_range[1];
let edits_end = dirty_inputs[input_idx..]
.iter()
.take_while(|d| {
let r = &d.original_range;
if r.start == r.end {
r.start < w_end || (r.start == w_end && w_end == original_size)
} else {
r.end <= w_end
}
})
.count()
+ input_idx;
let window_edits = &mut dirty_inputs[input_idx..edits_end];
let (removed, added): (u64, u64) = window_edits
.iter()
.map(|d| (d.original_range.end - d.original_range.start, d.new_length))
.fold((0, 0), |(rm, ad), (r, a)| (rm + r, ad + a));
let middle_size = (w_end - w_start) + added - removed;
let (_id, mut cleaner) = session.start_clean(None, Some(middle_size), Sha256Policy::Skip)?;
let mut cursor = w_start;
for input in window_edits.iter_mut() {
let edit_start = input.original_range.start;
let edit_end = input.original_range.end;
debug_assert!(edit_start >= w_start && edit_end <= w_end, "edit straddles window (validation bug)");
if cursor < edit_start {
stream_cas_range(&ctx, &cas_client, original_hash, cursor, edit_start, &mut cleaner).await?;
}
let mut remaining = input.new_length as usize;
while remaining > 0 {
let to_read = buf.len().min(remaining);
input.reader.read_exact(&mut buf[..to_read]).await.map_err(|err| {
DataError::InternalError(format!(
"failed to read dirty input [{}, {}): {err}",
input.original_range.start, input.original_range.end
))
})?;
cleaner.add_data(&buf[..to_read]).await?;
remaining -= to_read;
}
cursor = edit_end;
}
input_idx = edits_end;
if cursor < w_end {
stream_cas_range(&ctx, &cas_client, original_hash, cursor, w_end, &mut cleaner).await?;
}
let (_info, chunks, mdb, _metrics) = cleaner.finish_with_chunks_detached().await?;
uploaded.push(UploadedWindow {
start: w_start,
end: w_end,
chunks,
mdb,
});
}
if input_idx != dirty_inputs.len() {
return Err(DataError::InternalError(format!(
"{} dirty edits not assigned to any window (input_idx={input_idx}, total={})",
dirty_inputs.len() - input_idx,
dirty_inputs.len()
)));
}
let mut hash_ranges = response.hash_ranges;
let trailing_gap = hash_ranges.pop().flatten();
let first_window_at_start = matches!(hash_ranges.first(), Some(None));
let last_window_at_end = trailing_gap.is_none();
let last_idx = uploaded.len() - 1;
let mut merge_seq: Vec<MerkleHashSubtree> = Vec::with_capacity(2 * uploaded.len() + 1);
for (i, (w, gap)) in uploaded.iter().zip(hash_ranges).enumerate() {
if let Some(g) = gap {
merge_seq.push(g);
}
let at_start = i == 0 && first_window_at_start;
let at_end = i == last_idx && last_window_at_end;
merge_seq.push(MerkleHashSubtree::from_chunks(at_start, &w.chunks, at_end));
}
if let Some(g) = trailing_gap {
merge_seq.push(g);
}
let merged = MerkleHashSubtree::merge(&merge_seq)
.map_err(|err| DataError::InternalError(format!("MerkleHashSubtree::merge failed: {err}")))?;
let aggregated_hash = merged.final_hash().ok_or_else(|| {
DataError::InternalError("merged subtree is not fully closed; cannot derive final hash".into())
})?;
let combined_hash = if total_size == 0 {
MerkleHash::default()
} else {
aggregated_hash.hmac(MerkleHash::default())
};
let composed_mdb =
compose_mdb(&original_mdb, &seg_byte_starts, &uploaded, gap_verification, combined_hash, original_size)?;
debug!(
"upload_ranges: composed hash={}, {} segments, {} windows",
combined_hash.hex(),
composed_mdb.segments.len(),
uploaded.len()
);
session.register_composed_file(composed_mdb).await?;
session.finalize().await?;
let total_dirty: u64 = dirty_inputs.iter().map(|d| d.new_length).sum();
info!(
"upload_ranges: hash={} size={} (original={}, {} windows, {} dirty bytes)",
combined_hash.hex(),
total_size,
original_size,
uploaded.len(),
total_dirty
);
Ok(XetFileInfo::new(combined_hash.hex(), total_size))
}
fn compose_mdb(
original_mdb: &MDBFileInfo,
seg_byte_starts: &[u64],
uploaded: &[UploadedWindow],
gap_verification: Vec<HexMerkleHash>,
combined_hash: MerkleHash,
original_size: u64,
) -> Result<MDBFileInfo> {
let mut all_segments: Vec<FileDataSequenceEntry> = Vec::new();
let mut all_verification: Vec<FileVerificationEntry> = Vec::new();
let mut seg_idx = 0usize;
let n_segs = original_mdb.segments.len();
let mut gap_idx = 0usize;
for w in uploaded {
while seg_idx < n_segs && seg_byte_starts[seg_idx] < w.start {
if seg_byte_starts[seg_idx + 1] > w.start {
return Err(DataError::InternalError(format!(
"server returned a window starting at {} that straddles segment {} \
({}..{}); composition requires segment-aligned windows",
w.start,
seg_idx,
seg_byte_starts[seg_idx],
seg_byte_starts[seg_idx + 1]
)));
}
all_segments.push(original_mdb.segments[seg_idx].clone());
let entry = gap_verification.get(gap_idx).ok_or_else(|| {
DataError::InternalError(format!(
"ran out of gap_verification entries at stable segment {seg_idx}; \
server response is inconsistent with the segment layout"
))
})?;
all_verification.push(FileVerificationEntry::new(entry.into()));
gap_idx += 1;
seg_idx += 1;
}
debug_assert!(w.end <= original_size, "window end {} exceeds original_size {}", w.end, original_size);
while seg_idx < n_segs && seg_byte_starts[seg_idx] < w.end {
seg_idx += 1;
}
if w.mdb.verification.len() != w.mdb.segments.len() {
return Err(DataError::InternalError(format!(
"window MDB has {} segments but {} verification entries",
w.mdb.segments.len(),
w.mdb.verification.len()
)));
}
all_segments.extend_from_slice(&w.mdb.segments);
all_verification.extend_from_slice(&w.mdb.verification);
}
while seg_idx < n_segs {
all_segments.push(original_mdb.segments[seg_idx].clone());
let entry = gap_verification.get(gap_idx).ok_or_else(|| {
DataError::InternalError(format!(
"ran out of gap_verification entries at stable segment {seg_idx}; \
server response is inconsistent with the segment layout"
))
})?;
all_verification.push(FileVerificationEntry::new(entry.into()));
gap_idx += 1;
seg_idx += 1;
}
if gap_idx < gap_verification.len() {
return Err(DataError::InternalError(format!(
"server returned {} gap_verification entries but only {} stable segments were emitted",
gap_verification.len(),
gap_idx
)));
}
debug_assert_eq!(all_segments.len(), all_verification.len());
Ok(MDBFileInfo {
metadata: FileDataSequenceHeader::new(combined_hash, all_segments.len(), true, false),
segments: all_segments,
verification: all_verification,
metadata_ext: None,
})
}
fn validate_dirty_ranges(dirty_inputs: &[DirtyInput], original_size: u64) -> Result<()> {
let mut prev_end = 0u64;
for (i, input) in dirty_inputs.iter().enumerate() {
let r = &input.original_range;
if r.start > r.end {
return Err(DataError::ParameterError(format!(
"dirty_inputs[{i}].original_range is reversed: {}..{}",
r.start, r.end
)));
}
if r.end > original_size {
return Err(DataError::ParameterError(format!(
"dirty_inputs[{i}].original_range end ({}) exceeds original_size ({original_size})",
r.end
)));
}
if i > 0 && r.start < prev_end {
return Err(DataError::ParameterError(format!(
"dirty_inputs[{i}].original_range overlaps the previous edit (starts at {} < {prev_end})",
r.start
)));
}
prev_end = r.end;
}
Ok(())
}
fn compute_total_size(original_size: u64, dirty_inputs: &[DirtyInput]) -> Result<u64> {
let (removed, added) = dirty_inputs
.iter()
.fold((0u64, 0u64), |(r, a), d| (r + (d.original_range.end - d.original_range.start), a + d.new_length));
original_size
.checked_add(added)
.and_then(|s| s.checked_sub(removed))
.ok_or_else(|| {
DataError::ParameterError(format!(
"total size overflows: original_size={original_size}, added={added}, removed={removed}"
))
})
}
async fn upload_fresh_file(
config: Arc<TranslatorConfig>,
mut dirty_inputs: Vec<DirtyInput>,
total_size: u64,
) -> Result<XetFileInfo> {
let session = FileUploadSession::new(config).await?;
let (_id, mut cleaner) = session.start_clean(None, Some(total_size), Sha256Policy::Skip)?;
for input in &mut dirty_inputs {
let mut remaining = input.new_length as usize;
let mut buf = vec![0u8; STREAM_BLOCK_SIZE.min(remaining.max(1))];
while remaining > 0 {
let to_read = buf.len().min(remaining);
input.reader.read_exact(&mut buf[..to_read]).await.map_err(|err| {
DataError::InternalError(format!("failed to read dirty input at {}: {err}", input.original_range.start))
})?;
cleaner.add_data(&buf[..to_read]).await?;
remaining -= to_read;
}
}
let (info, _metrics) = cleaner.finish().await?;
session.finalize().await?;
Ok(info)
}
async fn stream_cas_range(
ctx: &XetContext,
cas_client: &Arc<dyn Client>,
file_hash: MerkleHash,
start: u64,
end: u64,
cleaner: &mut super::SingleFileCleaner,
) -> Result<()> {
let reconstructor = FileReconstructor::new(ctx, cas_client, file_hash).with_byte_range(FileRange::new(start, end));
let mut stream = reconstructor.reconstruct_to_stream();
while let Some(chunk) = stream.next().await? {
cleaner.add_data(&chunk).await?;
}
Ok(())
}
fn snap_to_segment_start(seg_byte_starts: &[u64], byte: u64) -> u64 {
let idx = seg_byte_starts.partition_point(|&s| s <= byte);
seg_byte_starts[idx.saturating_sub(1)]
}
fn snap_to_segment_end(seg_byte_starts: &[u64], byte: u64) -> u64 {
let idx = seg_byte_starts.partition_point(|&s| s < byte);
seg_byte_starts[idx]
}
#[cfg(all(test, feature = "simulation"))]
mod tests {
use std::io::Cursor;
use std::ops::Range;
use std::path::Path;
use std::sync::Arc;
use tempfile::TempDir;
use xet_client::cas_client::{Client, LocalTestServerBuilder};
use xet_core_structures::merklehash::MerkleHash;
use super::*;
use crate::processing::configurations::TranslatorConfig;
use crate::processing::file_cleaner::Sha256Policy;
use crate::processing::file_download_session::FileDownloadSession;
use crate::processing::file_upload_session::FileUploadSession;
fn test_config(endpoint: impl AsRef<str>, base_dir: impl AsRef<Path>) -> Arc<TranslatorConfig> {
let ctx = XetContext::default().unwrap();
Arc::new(TranslatorConfig::test_server_config(&ctx, endpoint, base_dir).unwrap())
}
async fn fetch_segment_sizes(cas_client: &Arc<dyn Client>, hash: &MerkleHash) -> Vec<u64> {
let (mdb, _) = cas_client.get_file_reconstruction_info(hash).await.unwrap().unwrap();
mdb.segments.iter().map(|s| s.unpacked_segment_bytes as u64).collect()
}
fn make_dirty_inputs(ranges: &[(u64, u64)], data: &[u8]) -> Vec<DirtyInput> {
ranges
.iter()
.map(|&(start, end)| {
let slice = data[start as usize..end as usize].to_vec();
DirtyInput {
original_range: start..end,
new_length: end - start,
reader: Box::pin(Cursor::new(slice)),
}
})
.collect()
}
fn make_dummy_inputs(ranges: &[(u64, u64)]) -> Vec<DirtyInput> {
ranges
.iter()
.map(|&(start, end)| DirtyInput {
original_range: start..end,
new_length: end - start,
reader: Box::pin(Cursor::new(Vec::new())),
})
.collect()
}
fn make_legacy_inputs(specs: &[(u64, u64)], data: &[u8], original_size: u64, total_size: u64) -> Vec<DirtyInput> {
let mut out: Vec<DirtyInput> = Vec::new();
for &(start, end) in specs {
let bytes = data[start as usize..end as usize].to_vec();
if end <= original_size {
out.push(DirtyInput {
original_range: start..end,
new_length: bytes.len() as u64,
reader: Box::pin(Cursor::new(bytes)),
});
} else if start >= original_size {
out.push(DirtyInput {
original_range: original_size..original_size,
new_length: bytes.len() as u64,
reader: Box::pin(Cursor::new(bytes)),
});
} else {
let split = (original_size - start) as usize;
let (head, tail) = bytes.split_at(split);
out.push(DirtyInput {
original_range: start..original_size,
new_length: head.len() as u64,
reader: Box::pin(Cursor::new(head.to_vec())),
});
out.push(DirtyInput {
original_range: original_size..original_size,
new_length: tail.len() as u64,
reader: Box::pin(Cursor::new(tail.to_vec())),
});
}
}
if total_size < original_size {
out.push(DirtyInput {
original_range: total_size..original_size,
new_length: 0,
reader: Box::pin(Cursor::new(Vec::new())),
});
}
out
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_upload_ranges_mid_file_edit() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let endpoint = server.http_endpoint().to_string();
let config = test_config(&endpoint, base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(42, 256 * 1024);
let original_hash = {
let upload_session = FileUploadSession::new(config.clone()).await.unwrap();
let (_id, mut cleaner) = upload_session
.start_clean(Some("original".into()), Some(original_data.len() as u64), Sha256Policy::Skip)
.unwrap();
cleaner.add_data(&original_data).await.unwrap();
let (xfi, _metrics) = cleaner.finish().await.unwrap();
upload_session.finalize().await.unwrap();
MerkleHash::from_hex(xfi.hash()).unwrap()
};
let original_size = original_data.len() as u64;
let mut modified_data = original_data.clone();
let dirty_start = 100_000usize;
let dirty_end = 101_000usize;
modified_data[dirty_start..dirty_end].fill(0xBB);
let total_size = modified_data.len() as u64;
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[(dirty_start as u64, dirty_end as u64)], &modified_data, original_size, total_size),
)
.await
.unwrap();
assert_eq!(result.file_size, Some(total_size));
let composed_hash = MerkleHash::from_hex(result.hash()).unwrap();
let session = FileDownloadSession::new(config.clone(), None).await.unwrap();
let file_info = crate::processing::XetFileInfo::new(composed_hash.hex(), total_size);
let out_path = base_dir.path().join("output");
session.download_file(&file_info, &out_path).await.unwrap();
let downloaded = std::fs::read(&out_path).unwrap();
assert_eq!(downloaded.len(), modified_data.len());
assert_eq!(downloaded, modified_data);
let clean_hash = upload_file(&config, &modified_data).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_upload_ranges_truncation() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(43, 256 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let truncated_size = 100_000u64;
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[], &[], original_size, truncated_size),
)
.await
.unwrap();
assert_eq!(result.file_size(), Some(truncated_size));
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), truncated_size).await;
assert_eq!(downloaded.len(), truncated_size as usize);
assert_eq!(downloaded, &original_data[..truncated_size as usize]);
let clean_hash = upload_file(&config, &original_data[..truncated_size as usize]).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_upload_ranges_append() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(44, 100 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let mut full_data = original_data.clone();
full_data.extend(random_data(99, 50 * 1024));
let total_size = full_data.len() as u64;
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[(original_size, total_size)], &full_data, original_size, total_size),
)
.await
.unwrap();
assert_eq!(result.file_size(), Some(total_size));
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), total_size).await;
assert_eq!(downloaded, full_data);
let clean_hash = upload_file(&config, &full_data).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_upload_ranges_at_file_start() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(45, 256 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let mut modified_data = original_data.clone();
modified_data[..4096].fill(0xBB);
let total_size = modified_data.len() as u64;
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[(0, 4096)], &modified_data, original_size, total_size),
)
.await
.unwrap();
assert_eq!(result.file_size(), Some(total_size));
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), total_size).await;
assert_eq!(downloaded.len(), modified_data.len());
assert_eq!(downloaded, modified_data);
let clean_hash = upload_file(&config, &modified_data).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_upload_ranges_multiple_regions() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(46, 256 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let mut modified_data = original_data.clone();
modified_data[10_000..12_000].fill(0xBB); modified_data[200_000..202_000].fill(0xCC); let total_size = modified_data.len() as u64;
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[(10_000, 12_000), (200_000, 202_000)], &modified_data, original_size, total_size),
)
.await
.unwrap();
assert_eq!(result.file_size(), Some(total_size));
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), total_size).await;
assert_eq!(downloaded.len(), modified_data.len());
assert_eq!(downloaded, modified_data);
let clean_hash = upload_file(&config, &modified_data).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_append_with_gap_before_dirty_range() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(50, 100 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let gap = 500u64;
let write_data = random_data(101, 4096);
let total_size = original_size + gap + write_data.len() as u64;
let mut full_data = original_data.clone();
full_data.extend(vec![0x00u8; gap as usize]);
full_data.extend(&write_data);
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[(original_size, total_size)], &full_data, original_size, total_size),
)
.await
.unwrap();
assert_eq!(result.file_size(), Some(total_size));
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), total_size).await;
assert_eq!(downloaded.len(), full_data.len(), "size mismatch");
assert_eq!(&downloaded[..], &full_data[..], "content mismatch: gap bytes were lost");
let clean_hash = upload_file(&config, &full_data).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_append_sparse_staging_file() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = vec![0xDDu8; 100 * 1024];
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let append_data = vec![0xEEu8; 50 * 1024];
let total_size = original_size + append_data.len() as u64;
let mut sparse_staging = vec![0u8; total_size as usize];
sparse_staging[original_size as usize..].copy_from_slice(&append_data);
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[(original_size, total_size)], &sparse_staging, original_size, total_size),
)
.await
.unwrap();
let mut expected = original_data.clone();
expected.extend(&append_data);
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), total_size).await;
assert_eq!(downloaded.len(), expected.len(), "size mismatch");
assert_eq!(&downloaded[..], &expected[..], "content mismatch: CAS data replaced by zeros from sparse file");
let clean_hash = upload_file(&config, &expected).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_data_integrity_scenarios() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
{
let original = vec![0xAAu8; 256 * 1024];
let mut expected = original[..100_000].to_vec();
expected[90_000..100_000].fill(0xBB);
assert_range_edit(&config, &cas_client, &original, &expected, &[(90_000, 100_000)], 100_000).await;
}
{
let original = vec![0xAAu8; 128 * 1024];
let expected = vec![0xBBu8; 128 * 1024];
let size = original.len() as u64;
assert_range_edit(&config, &cas_client, &original, &expected, &[(0, size)], size).await;
}
{
let original = vec![0xAAu8; 256 * 1024];
let mut expected = original.clone();
expected[50_000..51_000].fill(0xBB);
expected[51_000..52_000].fill(0xCC);
expected[52_000..53_000].fill(0xDD);
let size = original.len() as u64;
assert_range_edit(
&config,
&cas_client,
&original,
&expected,
&[(50_000, 51_000), (51_000, 52_000), (52_000, 53_000)],
size,
)
.await;
}
{
let original = vec![0xAAu8; 100 * 1024];
let mut expected = original.clone();
expected.extend(vec![0xEEu8; 50 * 1024]);
let total = expected.len() as u64;
assert_range_edit(&config, &cas_client, &original, &expected, &[], total).await;
}
{
let original: Vec<u8> = (0..256 * 1024)
.map(|i: usize| {
let x = i.wrapping_mul(2654435761);
(x >> 16) as u8
})
.collect();
let original_hash = upload_file(&config, &original).await;
let seg_sizes = fetch_segment_sizes(&cas_client, &original_hash).await;
if seg_sizes.len() >= 3 {
let boundary: u64 = seg_sizes[0] + seg_sizes[1];
let dirty_end = boundary + seg_sizes[2];
let mut expected = original.clone();
expected[boundary as usize..dirty_end as usize].fill(0xFF);
let size = original.len() as u64;
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
size,
make_legacy_inputs(&[(boundary, dirty_end)], &expected, size, size),
)
.await
.unwrap();
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), size).await;
assert_eq!(downloaded, expected, "chunk-boundary edit mismatch");
let clean_hash = upload_file(&config, &expected).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_noop_returns_original_hash() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let data = random_data(70, 256 * 1024);
let hash = upload_file(&config, &data).await;
let size = data.len() as u64;
let result = upload_ranges(config, cas_client, hash, size, make_legacy_inputs(&[], &[], size, size))
.await
.unwrap();
assert_eq!(result.hash(), hash.hex());
assert_eq!(result.file_size(), Some(size));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_rejects_dirty_range_past_total_size() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let data = random_data(71, 256 * 1024);
let hash = upload_file(&config, &data).await;
let size = data.len() as u64;
let err = upload_ranges(config, cas_client, hash, size, make_dummy_inputs(&[(100, size + 1)])).await;
assert!(err.is_err(), "dirty range past total_size should be rejected");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_rejects_overlapping_dirty_ranges() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let data = random_data(60, 256 * 1024);
let hash = upload_file(&config, &data).await;
let size = data.len() as u64;
let err = upload_ranges(config, cas_client, hash, size, make_dummy_inputs(&[(100, 300), (200, 400)])).await;
assert!(err.is_err(), "overlapping ranges should be rejected");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_empty_original_validates_ranges() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_hash = upload_file(&config, &[]).await;
let inputs = vec![
DirtyInput {
original_range: 0..10,
new_length: 10,
reader: Box::pin(Cursor::new(vec![0xAA; 10])),
},
DirtyInput {
original_range: 5..15,
new_length: 10,
reader: Box::pin(Cursor::new(vec![0xBB; 10])),
},
];
let err = upload_ranges(config, cas_client, original_hash, 0, inputs).await;
assert!(err.is_err(), "ranges with end > original_size must be rejected for empty originals too");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_rejects_unsorted_dirty_ranges() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let data = random_data(62, 256 * 1024);
let hash = upload_file(&config, &data).await;
let size = data.len() as u64;
let err = upload_ranges(config, cas_client, hash, size, make_dummy_inputs(&[(300, 400), (100, 200)])).await;
assert!(err.is_err(), "unsorted ranges should be rejected");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_single_input_spanning_many_chunks() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(99, 256 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let mut modified = original_data.clone();
let dirty_start = 10_000u64;
let dirty_end = 200_000u64;
modified[dirty_start as usize..dirty_end as usize].fill(0xFF);
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[(dirty_start, dirty_end)], &modified, original_size, original_size),
)
.await
.unwrap();
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), original_size).await;
assert_eq!(downloaded, modified, "large spanning input produced wrong content");
let clean_hash = upload_file(&config, &modified).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_upload_ranges_small_file_mid_edit() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = b"AAAA_HEADER_AAAA|";
let original_hash = upload_file(&config, original_data).await;
let original_size = original_data.len() as u64;
let dirty_data = b"SPARSE";
let dirty_inputs = vec![DirtyInput {
original_range: 5..11,
new_length: dirty_data.len() as u64,
reader: Box::pin(Cursor::new(dirty_data.to_vec())),
}];
let result = upload_ranges(config.clone(), cas_client.clone(), original_hash, original_size, dirty_inputs)
.await
.unwrap();
assert_eq!(result.file_size(), Some(original_size));
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), original_size).await;
assert_eq!(downloaded.len(), original_size as usize, "reconstructed size mismatch");
assert_eq!(&downloaded[..5], b"AAAA_", "prefix from CAS");
assert_eq!(&downloaded[5..11], b"SPARSE", "dirty range");
assert_eq!(&downloaded[11..], b"_AAAA|", "suffix from CAS");
let expected = b"AAAA_SPARSE_AAAA|";
let clean_hash = upload_file(&config, expected).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_upload_ranges_truncation_empty_staging() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(77, 256 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let truncated_size = 100_000u64;
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[], &[], original_size, truncated_size),
)
.await
.unwrap();
assert_eq!(result.file_size(), Some(truncated_size));
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), truncated_size).await;
assert_eq!(downloaded.len(), truncated_size as usize);
assert_eq!(
&downloaded[..],
&original_data[..truncated_size as usize],
"truncated content should match original CAS data, not staging zeros"
);
let clean_hash = upload_file(&config, &original_data[..truncated_size as usize]).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_upload_ranges_truncation_with_overlapping_dirty() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(88, 256 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let truncated_size = 100_000u64;
let dirty_start = 90_000u64;
let dirty_end = 95_000u64;
let mut expected = original_data[..truncated_size as usize].to_vec();
expected[dirty_start as usize..dirty_end as usize].fill(0xBB);
let mut staging = vec![0u8; truncated_size as usize];
staging[dirty_start as usize..dirty_end as usize].fill(0xBB);
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[(dirty_start, dirty_end)], &staging, original_size, truncated_size),
)
.await
.unwrap();
assert_eq!(result.file_size(), Some(truncated_size));
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), truncated_size).await;
assert_eq!(downloaded.len(), expected.len());
assert_eq!(&downloaded[..], &expected[..], "dirty bytes should come from staging, boundary bytes from CAS");
let clean_hash = upload_file(&config, &expected).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
fn random_data(seed: u64, len: usize) -> Vec<u8> {
(0..len)
.map(|i| {
let x = (i as u64).wrapping_add(seed).wrapping_mul(2654435761);
(x >> 16) as u8
})
.collect()
}
#[derive(Clone, Debug)]
struct DeterministicRng {
state: u64,
}
impl DeterministicRng {
fn new(seed: u64) -> Self {
Self { state: seed }
}
fn next_u64(&mut self) -> u64 {
self.state = self.state.wrapping_mul(6364136223846793005).wrapping_add(1);
self.state
}
fn gen_range(&mut self, start: usize, end: usize) -> usize {
if end <= start {
return start;
}
start + (self.next_u64() as usize % (end - start))
}
fn gen_bytes(&mut self, len: usize) -> Vec<u8> {
(0..len).map(|_| (self.next_u64() >> 56) as u8).collect()
}
}
#[derive(Clone, Debug)]
struct PlannedEdit {
original_range: Range<usize>,
replacement: Vec<u8>,
}
fn build_random_non_overlapping_edits(
rng: &mut DeterministicRng,
original_len: usize,
max_edits: usize,
) -> Vec<PlannedEdit> {
if original_len == 0 {
let replacement_len = 1 + rng.gen_range(0, 8 * 1024);
return vec![PlannedEdit {
original_range: 0..0,
replacement: rng.gen_bytes(replacement_len),
}];
}
let target_edits = 1 + rng.gen_range(0, max_edits.max(1));
let mut edits: Vec<PlannedEdit> = Vec::with_capacity(target_edits);
let mut cursor = 0usize;
while edits.len() < target_edits && cursor <= original_len {
let remaining = original_len - cursor;
let max_gap = remaining.min(64 * 1024);
let start = cursor + rng.gen_range(0, max_gap + 1);
let (end, replacement_len) = if start == original_len {
(start, 1 + rng.gen_range(0, 32 * 1024))
} else {
let op = rng.gen_range(0, 5);
let max_span = (original_len - start).clamp(1, 64 * 1024);
let span = 1 + rng.gen_range(0, max_span);
let end = start + span;
match op {
0 => (start, 1 + rng.gen_range(0, 32 * 1024)),
1 => (end, span),
2 => (end, span + 1 + rng.gen_range(0, 16 * 1024)),
3 => (end, rng.gen_range(0, span + 1)),
_ => (end, 0),
}
};
edits.push(PlannedEdit {
original_range: start..end,
replacement: rng.gen_bytes(replacement_len),
});
cursor = if end > start { end } else { start.saturating_add(1) };
}
if edits.is_empty() {
let replacement_len = 1 + rng.gen_range(0, 32 * 1024);
edits.push(PlannedEdit {
original_range: original_len..original_len,
replacement: rng.gen_bytes(replacement_len),
});
}
for w in edits.windows(2) {
assert!(w[0].original_range.end <= w[1].original_range.start);
}
edits
}
fn apply_planned_edits(original: &[u8], edits: &[PlannedEdit]) -> Vec<u8> {
let removed: usize = edits.iter().map(|e| e.original_range.end - e.original_range.start).sum();
let added: usize = edits.iter().map(|e| e.replacement.len()).sum();
let mut out: Vec<u8> = Vec::with_capacity(original.len() + added.saturating_sub(removed));
let mut cursor = 0usize;
for edit in edits {
assert!(edit.original_range.start >= cursor);
out.extend_from_slice(&original[cursor..edit.original_range.start]);
out.extend_from_slice(&edit.replacement);
cursor = edit.original_range.end;
}
out.extend_from_slice(&original[cursor..]);
out
}
fn edits_to_dirty_inputs(edits: &[PlannedEdit]) -> Vec<DirtyInput> {
edits
.iter()
.map(|e| DirtyInput {
original_range: e.original_range.start as u64..e.original_range.end as u64,
new_length: e.replacement.len() as u64,
reader: Box::pin(Cursor::new(e.replacement.clone())),
})
.collect()
}
fn summarize_edits(edits: &[PlannedEdit]) -> String {
edits
.iter()
.map(|e| format!("[{}..{}, new_len={}]", e.original_range.start, e.original_range.end, e.replacement.len()))
.collect::<Vec<_>>()
.join(", ")
}
async fn upload_file(config: &Arc<TranslatorConfig>, data: &[u8]) -> MerkleHash {
let session = FileUploadSession::new(config.clone()).await.unwrap();
let (_id, mut cleaner) = session
.start_clean(Some("test".into()), Some(data.len() as u64), Sha256Policy::Skip)
.unwrap();
cleaner.add_data(data).await.unwrap();
let (xfi, _metrics) = cleaner.finish().await.unwrap();
session.finalize().await.unwrap();
MerkleHash::from_hex(xfi.hash()).unwrap()
}
async fn download_file(config: &Arc<TranslatorConfig>, hash: MerkleHash, size: u64) -> Vec<u8> {
let session = FileDownloadSession::new(config.clone(), None).await.unwrap();
let xfi = crate::processing::XetFileInfo::new(hash.hex(), size);
let dir = TempDir::new().unwrap();
let out = dir.path().join("out");
session.download_file(&xfi, &out).await.unwrap();
std::fs::read(&out).unwrap()
}
fn empty_reader() -> Pin<Box<dyn AsyncRead + Send>> {
Box::pin(Cursor::new(Vec::<u8>::new()))
}
async fn assert_edits(
config: &Arc<TranslatorConfig>,
cas_client: &Arc<dyn Client>,
original: &[u8],
inputs: Vec<DirtyInput>,
expected: &[u8],
) {
let original_hash = upload_file(config, original).await;
let original_size = original.len() as u64;
let result = upload_ranges(config.clone(), cas_client.clone(), original_hash, original_size, inputs)
.await
.unwrap();
assert_eq!(result.file_size(), Some(expected.len() as u64), "file size mismatch");
let downloaded =
download_file(config, MerkleHash::from_hex(result.hash()).unwrap(), expected.len() as u64).await;
assert_eq!(downloaded, expected, "content mismatch");
let clean = upload_file(config, expected).await;
assert_eq!(result.hash(), clean.hex(), "hash diverges from clean upload");
}
async fn assert_range_edit(
config: &Arc<TranslatorConfig>,
cas_client: &Arc<dyn Client>,
original_data: &[u8],
expected: &[u8],
dirty_ranges: &[(u64, u64)],
total_size: u64,
) {
let original_hash = upload_file(config, original_data).await;
let original_size = original_data.len() as u64;
let mut inputs = make_dirty_inputs(dirty_ranges, expected);
if total_size > original_size {
let append_start = original_size;
let already_covered = dirty_ranges.iter().any(|&(s, e)| s <= append_start && e >= total_size);
if !already_covered {
inputs.push(DirtyInput {
original_range: original_size..original_size,
new_length: total_size - original_size,
reader: Box::pin(Cursor::new(expected[append_start as usize..total_size as usize].to_vec())),
});
inputs.sort_by_key(|d| d.original_range.start);
}
}
if total_size < original_size {
inputs.push(DirtyInput {
original_range: total_size..original_size,
new_length: 0,
reader: empty_reader(),
});
inputs.sort_by_key(|d| d.original_range.start);
}
let result = upload_ranges(config.clone(), cas_client.clone(), original_hash, original_size, inputs)
.await
.unwrap();
assert_eq!(result.file_size(), Some(total_size), "file size mismatch");
let downloaded = download_file(config, MerkleHash::from_hex(result.hash()).unwrap(), total_size).await;
assert_eq!(downloaded.len(), expected.len(), "downloaded length mismatch");
assert_eq!(&downloaded[..], expected, "content mismatch");
let clean_hash = upload_file(config, expected).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_mid_edit_plus_append() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(7, 256 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let dirty_start = 50_000usize;
let dirty_end = 51_000usize;
let append_extra: Vec<u8> = (0..16 * 1024).map(|i| (i % 251) as u8).collect();
let mut expected = original_data.clone();
expected[dirty_start..dirty_end].fill(0xAA);
expected.extend_from_slice(&append_extra);
let total_size = expected.len() as u64;
let inputs = vec![
DirtyInput {
original_range: dirty_start as u64..dirty_end as u64,
new_length: (dirty_end - dirty_start) as u64,
reader: Box::pin(Cursor::new(expected[dirty_start..dirty_end].to_vec())),
},
DirtyInput {
original_range: original_size..original_size,
new_length: append_extra.len() as u64,
reader: Box::pin(Cursor::new(append_extra)),
},
];
let result = upload_ranges(config.clone(), cas_client.clone(), original_hash, original_size, inputs)
.await
.unwrap();
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), total_size).await;
assert_eq!(downloaded, expected, "content mismatch (mid-edit + append regression)");
let clean_hash = upload_file(&config, &expected).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_empty_original_append() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data: &[u8] = &[];
let original_hash = upload_file(&config, original_data).await;
let new_data: Vec<u8> = (0..32 * 1024).map(|i| (i % 251) as u8).collect();
let total_size = new_data.len() as u64;
let inputs = vec![DirtyInput {
original_range: 0..0,
new_length: total_size,
reader: Box::pin(Cursor::new(new_data.clone())),
}];
let result = upload_ranges(config.clone(), cas_client.clone(), original_hash, 0, inputs)
.await
.unwrap();
let downloaded = download_file(&config, MerkleHash::from_hex(result.hash()).unwrap(), total_size).await;
assert_eq!(downloaded, new_data);
let clean_hash = upload_file(&config, &new_data).await;
assert_eq!(result.hash(), clean_hash.hex(), "hash mismatch with clean upload");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_truncate_to_empty_matches_clean_empty() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original_data = random_data(11, 64 * 1024);
let original_hash = upload_file(&config, &original_data).await;
let original_size = original_data.len() as u64;
let result = upload_ranges(
config.clone(),
cas_client.clone(),
original_hash,
original_size,
make_legacy_inputs(&[], &[], original_size, 0),
)
.await
.unwrap();
let clean_empty = upload_file(&config, &[]).await;
assert_eq!(result.hash(), clean_empty.hex(), "truncate-to-empty must match clean empty upload hash");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_resize_edits_abc() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
assert_edits(
&config,
&cas_client,
b"abc",
vec![DirtyInput {
original_range: 0..1,
new_length: 3,
reader: Box::pin(Cursor::new(b"foo".to_vec())),
}],
b"foobc",
)
.await;
assert_edits(
&config,
&cas_client,
b"abc",
vec![DirtyInput {
original_range: 0..0,
new_length: 3,
reader: Box::pin(Cursor::new(b"foo".to_vec())),
}],
b"fooabc",
)
.await;
assert_edits(
&config,
&cas_client,
b"abc",
vec![DirtyInput {
original_range: 0..1,
new_length: 0,
reader: empty_reader(),
}],
b"bc",
)
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_resize_large_replace_grows_file() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original = random_data(101, 256 * 1024);
let drop_start = 100_000usize;
let drop_end = 104_000usize;
let new_bytes: Vec<u8> = (0..32 * 1024).map(|i| (i % 251) as u8).collect();
let mut expected = Vec::with_capacity(original.len() - (drop_end - drop_start) + new_bytes.len());
expected.extend_from_slice(&original[..drop_start]);
expected.extend_from_slice(&new_bytes);
expected.extend_from_slice(&original[drop_end..]);
assert_edits(
&config,
&cas_client,
&original,
vec![DirtyInput {
original_range: drop_start as u64..drop_end as u64,
new_length: new_bytes.len() as u64,
reader: Box::pin(Cursor::new(new_bytes)),
}],
&expected,
)
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_resize_large_replace_shrinks_file() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original = random_data(102, 256 * 1024);
let drop_start = 80_000usize;
let drop_end = 160_000usize;
let new_bytes: Vec<u8> = (0..4 * 1024).map(|i| (0xCC ^ i) as u8).collect();
let mut expected = Vec::with_capacity(original.len() - (drop_end - drop_start) + new_bytes.len());
expected.extend_from_slice(&original[..drop_start]);
expected.extend_from_slice(&new_bytes);
expected.extend_from_slice(&original[drop_end..]);
assert_edits(
&config,
&cas_client,
&original,
vec![DirtyInput {
original_range: drop_start as u64..drop_end as u64,
new_length: new_bytes.len() as u64,
reader: Box::pin(Cursor::new(new_bytes)),
}],
&expected,
)
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_resize_mid_file_insert() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original = random_data(103, 100 * 1024);
let at = 40_000usize;
let new_bytes: Vec<u8> = (0..8u32 * 1024).map(|i| (i.wrapping_mul(13) % 251) as u8).collect();
let mut expected = Vec::with_capacity(original.len() + new_bytes.len());
expected.extend_from_slice(&original[..at]);
expected.extend_from_slice(&new_bytes);
expected.extend_from_slice(&original[at..]);
assert_edits(
&config,
&cas_client,
&original,
vec![DirtyInput {
original_range: at as u64..at as u64,
new_length: new_bytes.len() as u64,
reader: Box::pin(Cursor::new(new_bytes)),
}],
&expected,
)
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_resize_mid_file_delete() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original = random_data(104, 256 * 1024);
let drop_start = 80_000usize;
let drop_end = 144_000usize;
let mut expected = Vec::with_capacity(original.len() - (drop_end - drop_start));
expected.extend_from_slice(&original[..drop_start]);
expected.extend_from_slice(&original[drop_end..]);
assert_edits(
&config,
&cas_client,
&original,
vec![DirtyInput {
original_range: drop_start as u64..drop_end as u64,
new_length: 0,
reader: empty_reader(),
}],
&expected,
)
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_resize_multi_edit_mix() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original = random_data(105, 384 * 1024);
let (a_start, a_end) = (10 * 1024usize, 20 * 1024usize);
let a_new: Vec<u8> = vec![0xAA; 2 * 1024];
let b_at = 150 * 1024usize;
let b_new: Vec<u8> = vec![0xBB; 4 * 1024];
let (c_start, c_end) = (300 * 1024usize, 320 * 1024usize);
let mut expected = Vec::with_capacity(original.len() + b_new.len());
expected.extend_from_slice(&original[..a_start]);
expected.extend_from_slice(&a_new);
expected.extend_from_slice(&original[a_end..b_at]);
expected.extend_from_slice(&b_new);
expected.extend_from_slice(&original[b_at..c_start]);
expected.extend_from_slice(&original[c_end..]);
assert_edits(
&config,
&cas_client,
&original,
vec![
DirtyInput {
original_range: a_start as u64..a_end as u64,
new_length: a_new.len() as u64,
reader: Box::pin(Cursor::new(a_new)),
},
DirtyInput {
original_range: b_at as u64..b_at as u64,
new_length: b_new.len() as u64,
reader: Box::pin(Cursor::new(b_new)),
},
DirtyInput {
original_range: c_start as u64..c_end as u64,
new_length: 0,
reader: empty_reader(),
},
],
&expected,
)
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_resize_insert_at_segment_boundary() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original = random_data(106, 200 * 1024);
let original_hash = upload_file(&config, &original).await;
let original_size = original.len() as u64;
let seg_sizes = fetch_segment_sizes(&cas_client, &original_hash).await;
let Some(boundary) = seg_sizes
.iter()
.scan(0u64, |acc, s| {
*acc += s;
Some(*acc)
})
.find(|&b| b > 0 && b < original_size)
else {
return;
};
let new_bytes: Vec<u8> = vec![0x42; 4 * 1024];
let mut expected = Vec::with_capacity(original.len() + new_bytes.len());
expected.extend_from_slice(&original[..boundary as usize]);
expected.extend_from_slice(&new_bytes);
expected.extend_from_slice(&original[boundary as usize..]);
assert_edits(
&config,
&cas_client,
&original,
vec![DirtyInput {
original_range: boundary..boundary,
new_length: new_bytes.len() as u64,
reader: Box::pin(Cursor::new(new_bytes)),
}],
&expected,
)
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "stress test"]
async fn test_stress_random_resize_sequences() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
for seed in 0..6u64 {
let mut rng = DeterministicRng::new(0x9E37_79B9_7F4A_7C15 ^ seed.wrapping_mul(0xD1B5_4A32_D192_ED03));
let mut expected = random_data(10_000 + seed, 1_048_576 + (seed as usize * 91_117 % 262_144));
let mut original_hash = upload_file(&config, &expected).await;
let mut original_size = expected.len() as u64;
for round in 0..25usize {
let edits = build_random_non_overlapping_edits(&mut rng, expected.len(), 8);
let expected_next = apply_planned_edits(&expected, &edits);
let inputs = edits_to_dirty_inputs(&edits);
let result = upload_ranges(config.clone(), cas_client.clone(), original_hash, original_size, inputs)
.await
.unwrap();
let result_hash = MerkleHash::from_hex(result.hash()).unwrap();
assert_eq!(
result.file_size(),
Some(expected_next.len() as u64),
"seed={seed}, round={round}: size mismatch"
);
let clean_hash = upload_file(&config, &expected_next).await;
assert_eq!(result.hash(), clean_hash.hex(), "seed={seed}, round={round}: hash mismatch");
let downloaded = download_file(&config, result_hash, expected_next.len() as u64).await;
assert_eq!(downloaded, expected_next, "seed={seed}, round={round}: content mismatch");
expected = expected_next;
original_hash = result_hash;
original_size = expected.len() as u64;
}
}
}
#[cfg(not(feature = "smoke-test"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_regression_hash_matches_clean_upload_seed1_round17() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let seed = 1u64;
let mut rng = DeterministicRng::new(0x9E37_79B9_7F4A_7C15 ^ seed.wrapping_mul(0xD1B5_4A32_D192_ED03));
let mut expected = random_data(10_000 + seed, 1_048_576 + (seed as usize * 91_117 % 262_144));
let mut original_hash = upload_file(&config, &expected).await;
let mut original_size = expected.len() as u64;
for round in 0..=17usize {
let edits = build_random_non_overlapping_edits(&mut rng, expected.len(), 8);
let edits_summary = summarize_edits(&edits);
let expected_next = apply_planned_edits(&expected, &edits);
let inputs = edits_to_dirty_inputs(&edits);
let result = upload_ranges(config.clone(), cas_client.clone(), original_hash, original_size, inputs)
.await
.unwrap();
let result_hash = MerkleHash::from_hex(result.hash()).unwrap();
let clean_hash = upload_file(&config, &expected_next).await;
assert_eq!(
result.hash(),
clean_hash.hex(),
"seed={seed}, round={round}: hash mismatch; original_size={original_size}, expected_size={}, edits={edits_summary}",
expected_next.len()
);
let downloaded = download_file(&config, result_hash, expected_next.len() as u64).await;
assert_eq!(downloaded, expected_next, "seed={seed}, round={round}: content mismatch");
expected = expected_next;
original_hash = result_hash;
original_size = expected.len() as u64;
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "stress test"]
async fn test_stress_many_sparse_windows_single_call() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let original = random_data(13_337, 16 * 1024 * 1024);
let mut rng = DeterministicRng::new(0xA5A5_5A5A_0123_4567);
let mut edits: Vec<PlannedEdit> = Vec::new();
let stride = original.len() / 200;
let mut cursor = stride / 2;
while edits.len() < 128 && cursor < original.len() {
let start = cursor;
let max_span = (original.len() - start).clamp(1, 1536);
let span = 128 + rng.gen_range(0, max_span);
let end = (start + span).min(original.len());
let replacement_len = match rng.gen_range(0, 4) {
0 => end - start,
1 => (end - start) + 64 + rng.gen_range(0, 512),
2 => rng.gen_range(0, end - start + 1),
_ => 0,
};
edits.push(PlannedEdit {
original_range: start..end,
replacement: rng.gen_bytes(replacement_len),
});
cursor = cursor.saturating_add(stride.max(1));
}
let expected = apply_planned_edits(&original, &edits);
let inputs = edits_to_dirty_inputs(&edits);
assert_edits(&config, &cas_client, &original, inputs, &expected).await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "stress test"]
async fn test_stress_parallel_random_resize_sequences() {
let server = LocalTestServerBuilder::new().start().await;
let base_dir = TempDir::new().unwrap();
let config = test_config(server.http_endpoint(), base_dir.path());
let cas_client: Arc<dyn Client> = Arc::new(server);
let mut handles = Vec::new();
for worker in 0..8u64 {
let config = config.clone();
let cas_client = cas_client.clone();
handles.push(tokio::spawn(async move {
let mut rng = DeterministicRng::new(0xC0FF_EE00_1234_5678 ^ worker.wrapping_mul(0x94D0_49BB_1331_11EB));
let mut expected = random_data(20_000 + worker, 786_432 + worker as usize * 17_321);
let mut original_hash = upload_file(&config, &expected).await;
let mut original_size = expected.len() as u64;
for round in 0..18usize {
let edits = build_random_non_overlapping_edits(&mut rng, expected.len(), 6);
let expected_next = apply_planned_edits(&expected, &edits);
let inputs = edits_to_dirty_inputs(&edits);
let result =
upload_ranges(config.clone(), cas_client.clone(), original_hash, original_size, inputs)
.await
.unwrap();
let result_hash = MerkleHash::from_hex(result.hash()).unwrap();
assert_eq!(
result.file_size(),
Some(expected_next.len() as u64),
"worker={worker}, round={round}: size mismatch"
);
let clean_hash = upload_file(&config, &expected_next).await;
assert_eq!(result.hash(), clean_hash.hex(), "worker={worker}, round={round}: hash mismatch");
let downloaded = download_file(&config, result_hash, expected_next.len() as u64).await;
assert_eq!(downloaded, expected_next, "worker={worker}, round={round}: content mismatch");
expected = expected_next;
original_hash = result_hash;
original_size = expected.len() as u64;
}
}));
}
for handle in handles {
handle.await.unwrap();
}
}
}