diff --git a/curvine-server/src/worker/block/block_meta.rs b/curvine-server/src/worker/block/block_meta.rs index 2d7636f80..445905741 100644 --- a/curvine-server/src/worker/block/block_meta.rs +++ b/curvine-server/src/worker/block/block_meta.rs @@ -174,6 +174,19 @@ impl BlockMeta { pub fn physical_bytes(&self) -> i64 { self.actual_len } + + /// Capacity the target worker must reserve when replicating this block. + /// + /// File-backed blocks can grow when reopened, so their committed logical + /// length is sufficient. SPDK blocks use fixed extents and must preserve + /// the source allocation so a partial final block can still be appended up + /// to its original block boundary. + pub fn replication_capacity(&self) -> i64 { + match self.storage_type { + StorageType::SpdkDisk => self.len.max(self.actual_len), + _ => self.len, + } + } } impl fmt::Display for BlockMeta { diff --git a/curvine-server/src/worker/replication/worker_replication_manager.rs b/curvine-server/src/worker/replication/worker_replication_manager.rs index 6db0d2100..e69cd896f 100644 --- a/curvine-server/src/worker/replication/worker_replication_manager.rs +++ b/curvine-server/src/worker/replication/worker_replication_manager.rs @@ -128,18 +128,21 @@ impl WorkerReplicationManager { job.with_storage_type(block_meta.storage_type()); let extend_block = ExtendedBlock::new(block_meta.id, 0, block_meta.storage_type(), FileType::File); + let target_capacity = block_meta.replication_capacity(); info!( - "Replicating block_id: {} from {} to {}", + "Replicating block_id: {} from {} to {} (copy_bytes={}, target_capacity={})", job.block_id, self.block_store.worker_id()?, - job.target_worker_addr.worker_id + job.target_worker_addr.worker_id, + block_meta.len, + target_capacity ); let mut writer = BlockWriterRemote::new( &self.fs_client_context, extend_block, job.target_worker_addr.clone(), 0, - self.fs_client_context.block_size(), + target_capacity, ) .await?; let mut reader = self.block_store.open_reader(&block_meta, 0)?; diff --git a/curvine-tests/tests/replication_test.rs b/curvine-tests/tests/replication_test.rs index 2244907a5..aabb27b69 100644 --- a/curvine-tests/tests/replication_test.rs +++ b/curvine-tests/tests/replication_test.rs @@ -22,7 +22,9 @@ use log::info; use orpc::common::Utils; use orpc::runtime::RpcRuntime; use orpc::{CommonError, CommonResult}; +use std::collections::HashMap; use std::sync::Arc; +use std::time::{Duration, Instant}; use tempfile::TempDir; /// Create a test configuration for replication testing @@ -129,12 +131,7 @@ fn test_block_replication_e2e() -> CommonResult<()> { let target_block_locations = final_locations.get(&block_id); if let Some(locations) = target_block_locations { info!("Target block locations: {}", locations.len()); - if locations.len() - > initial_locations - .get(&block_id) - .map(|l| l.len()) - .unwrap_or(0) - { + if locations.len() > replica_count(&initial_locations, block_id) { info!( "✓ Block replication succeeded - block {} now has {} replicas", block_id, @@ -271,6 +268,100 @@ fn test_replication_with_simulated_worker_failure() -> CommonResult<()> { Ok(()) } +/// Replication must size the target from source block metadata rather than the +/// worker client's unrelated default block size. +#[test] +fn test_replication_honors_source_block_capacity() -> CommonResult<()> { + const CLIENT_DEFAULT_BLOCK_SIZE: i64 = 32 * 1024; + const FILE_BLOCK_SIZE: i64 = 64 * 1024; + const FILE_SIZE: usize = 96 * 1024; + const REPLICATION_TIMEOUT: Duration = Duration::from_secs(8); + const REPLICATION_POLL_INTERVAL: Duration = Duration::from_millis(100); + + let testing = Testing::builder() + .default() + .masters(2) + .workers(3) + .mutate_conf(|conf| { + conf.client.block_size_str = format!("{CLIENT_DEFAULT_BLOCK_SIZE}B"); + conf.master.min_block_size = CLIENT_DEFAULT_BLOCK_SIZE; + conf.master.min_replication = 1; + conf.master.max_replication = 3; + conf.master.block_replication_enabled = true; + }) + .build()?; + let cluster = testing.start_cluster()?; + let mut conf = testing.get_active_cluster_conf()?; + conf.client.block_size_str = format!("{FILE_BLOCK_SIZE}B"); + conf.client.init()?; + let rt = Arc::new(conf.client_rpc_conf().create_runtime()); + let fs = testing.get_fs(Some(rt.clone()), Some(conf))?; + + let path = Path::from_str("/replication_large_block.dat")?; + let test_data = generate_test_data(FILE_SIZE); + let file_blocks = rt.block_on(async { write_test_file(&fs, &path, &test_data).await })?; + let oversized_block = file_blocks + .block_locs + .iter() + .find(|block| block.block.len > CLIENT_DEFAULT_BLOCK_SIZE) + .expect("missing block larger than the worker client default"); + let block_id = oversized_block.block.id; + let original_locations = oversized_block.locs.clone(); + + let initial_locations = rt.block_on(async { get_block_locations(&fs, &path).await })?; + let initial_replica_count = replica_count(&initial_locations, block_id); + + cluster + .get_active_master_replication_manager() + .report_under_replicated_blocks(1, vec![block_id])?; + + let deadline = Instant::now() + REPLICATION_TIMEOUT; + loop { + let current_locations = rt.block_on(async { get_block_locations(&fs, &path).await })?; + let current_replica_count = replica_count(¤t_locations, block_id); + if current_replica_count > initial_replica_count { + break; + } + if Instant::now() >= deadline { + return Err(CommonError::from(format!( + "oversized block {} did not gain a replica before timeout (before={}, after={})", + block_id, initial_replica_count, current_replica_count + ))); + } + std::thread::sleep(REPLICATION_POLL_INTERVAL); + } + + let fs_dir = cluster.get_active_master_fs().fs_dir(); + { + let mut fs_dir = fs_dir.write(); + let reports = original_locations + .iter() + .map(|location| { + ( + false, + block_id, + BlockLocation { + worker_id: location.worker_id, + storage_type: Default::default(), + }, + ) + }) + .collect(); + fs_dir.block_report(reports)?; + } + let latest_locations = rt.block_on(async { get_block_locations(&fs, &path).await })?; + assert_eq!(1, replica_count(&latest_locations, block_id)); + + let read_data = rt.block_on(async { read_test_file(&fs, &path).await })?; + if read_data != test_data { + return Err(CommonError::from( + "Data integrity check failed after oversized-block replication", + )); + } + + Ok(()) +} + async fn write_test_file( fs: &CurvineFileSystem, path: &Path, @@ -308,10 +399,10 @@ async fn read_test_file(fs: &CurvineFileSystem, path: &Path) -> CommonResult CommonResult>> { +) -> CommonResult>> { let block_locations = fs.get_block_locations(path).await?; - let mut location_map = std::collections::HashMap::new(); + let mut location_map = HashMap::new(); for block_location in block_locations.block_locs.iter() { location_map.insert(block_location.block.id, block_location.locs.clone()); } @@ -319,6 +410,13 @@ async fn get_block_locations( Ok(location_map) } +fn replica_count(locations: &HashMap>, block_id: i64) -> usize { + locations + .get(&block_id) + .map(|locations| locations.len()) + .unwrap_or(0) +} + fn generate_test_data(size: usize) -> Vec { let mut data = Vec::with_capacity(size); let pattern = b"REPLICATION_TEST_DATA_";