From 9629234609869ef3d04cce5baa5112c8d0120771 Mon Sep 17 00:00:00 2001 From: GongXun Date: Thu, 23 Jul 2026 21:22:02 +0800 Subject: [PATCH 1/2] fix(server): prevent partial parallel loads from becoming valid cache --- .../src/worker/task/load_task_runner.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/curvine-server/src/worker/task/load_task_runner.rs b/curvine-server/src/worker/task/load_task_runner.rs index c38528e09..b8bfcd535 100644 --- a/curvine-server/src/worker/task/load_task_runner.rs +++ b/curvine-server/src/worker/task/load_task_runner.rs @@ -450,6 +450,21 @@ impl LoadTaskRunner { } } + // Cancellation may happen after the loop's pre-await check while the + // final stream is still running. That stream returns Ok(0), so recheck + // here before stamping the pre-resized target as a valid cache entry. + if self.task.is_cancel() { + info!("task {} was cancelled", self.task.info.task_id); + return Ok(()); + } + if written != src_len { + return err_box!( + "parallel load incomplete: wrote {} of {} bytes; refusing to mark cache valid", + written, + src_len + ); + } + // ufs -> cv: stamp the source mtime onto the cached file (cache validity). let ufs_mtime = self.get_ufs()?.get_status(source_path).await?.mtime; let attr_opts = SetAttrOptsBuilder::new().ufs_mtime(ufs_mtime).build(); From 96cc408f48e1c1793426cc457c924ce8dc0832c6 Mon Sep 17 00:00:00 2001 From: GongXun Date: Sat, 25 Jul 2026 14:23:15 +0800 Subject: [PATCH 2/2] fix(server): add parallel load cancellation regression --- .../src/worker/task/load_task_runner.rs | 33 +++- .../tests/load_task_runner_fault_test.rs | 160 ++++++++++++++++++ .../regression/daily_regression_test.sh | 25 ++- 3 files changed, 211 insertions(+), 7 deletions(-) create mode 100644 curvine-server/tests/load_task_runner_fault_test.rs diff --git a/curvine-server/src/worker/task/load_task_runner.rs b/curvine-server/src/worker/task/load_task_runner.rs index b8bfcd535..a7ab95058 100644 --- a/curvine-server/src/worker/task/load_task_runner.rs +++ b/curvine-server/src/worker/task/load_task_runner.rs @@ -80,6 +80,15 @@ impl LoadTaskRunner { warn!("report task {}", e) } } + + crate::fault_point! { + async, + name: "worker.load_task.after_run", + description: "After a worker load task runner has fully exited", + context: { + "task_id" => self.task.info.task_id.clone(), + }, + } } async fn run0(&self) -> FsResult<()> { @@ -359,6 +368,18 @@ impl LoadTaskRunner { let mut writer = fs.open_for_write(&dst, false).await?; writer.seek(off).await?; + crate::fault_point! { + async, + name: "worker.load_task.parallel.before_segment_copy", + description: "After a parallel load segment opens its streams and before it copies data", + context: { + "task_id" => task.info.task_id.clone(), + "stream_index" => i as i64, + "segment_offset" => off, + "segment_len" => len, + }, + } + let mut remaining = len; while remaining > 0 { // Honor cancellation and the shared deadline INSIDE the @@ -426,6 +447,15 @@ impl LoadTaskRunner { Self::abort_remaining(&handles, idx); return Ok(()); } + crate::fault_point! { + async, + name: "worker.load_task.parallel.before_join_await", + description: "After the parent cancellation check and before awaiting a parallel load stream", + context: { + "task_id" => self.task.info.task_id.clone(), + "stream_index" => idx as i64, + }, + } match (&mut handles[idx]).await { Ok(Ok(n)) => { written += n; @@ -459,7 +489,8 @@ impl LoadTaskRunner { } if written != src_len { return err_box!( - "parallel load incomplete: wrote {} of {} bytes; refusing to mark cache valid", + "Task {} parallel load incomplete: wrote {} of {} bytes; refusing to mark cache valid", + self.task.info.task_id, written, src_len ); diff --git a/curvine-server/tests/load_task_runner_fault_test.rs b/curvine-server/tests/load_task_runner_fault_test.rs new file mode 100644 index 000000000..f13c9dea1 --- /dev/null +++ b/curvine-server/tests/load_task_runner_fault_test.rs @@ -0,0 +1,160 @@ +// Copyright 2025 OPPO. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#![cfg(feature = "fault-injection")] + +use curvine_client::rpc::JobMasterClient; +use curvine_common::conf::ClusterConf; +use curvine_common::fs::Path; +use curvine_common::state::{JobTaskState, LoadJobCommand, MountOptions}; +use curvine_fault::{FaultHttpController, FaultRuleBuilder, FaultRuntime, FaultTestSession}; +use curvine_server::test::MiniCluster; +use orpc::common::Utils; +use orpc::runtime::RpcRuntime; +use orpc::CommonResult; +use std::sync::Arc; +use std::time::Duration; + +const TOKEN_ENV: &str = "CURVINE_LOAD_TASK_FAULT_TEST_TOKEN"; +const BLOCK_SIZE: i64 = 1024 * 1024; +const SOURCE_LEN: usize = 4 * 1024 * 1024; +const FINAL_STREAM: i64 = 1; + +/// Reproduces cancellation after the parent's final pre-await check: +/// +/// - stream 0 completes normally; +/// - stream 1 is paused before copying; +/// - the parent is paused after checking cancellation and before awaiting stream 1; +/// - cancellation is delivered, so stream 1 later returns the soft `Ok(0)` exit. +/// +/// The target has already been resized to the source length, so stamping its +/// UFS mtime would turn this partial file into a false cache hit. +#[test] +fn canceled_final_parallel_stream_does_not_mark_cache_valid() -> CommonResult<()> { + std::env::set_var(TOKEN_ENV, "load-task-fault-test-secret"); + FaultRuntime::process().clear()?; + + let test_id = Utils::rand_str(8); + let base_dir = Utils::test_sub_dir(format!("load-task-fault-{test_id}")); + let ufs_dir = format!("{base_dir}/ufs"); + std::fs::create_dir_all(&ufs_dir)?; + let source_file = format!("{ufs_dir}/source.bin"); + std::fs::write(&source_file, vec![0x5a; SOURCE_LEN])?; + + let mut conf = ClusterConf::default(); + conf.master.meta_dir = format!("{base_dir}/meta"); + conf.journal.journal_dir = format!("{base_dir}/journal"); + conf.worker.data_dir = vec![format!("[MEM:64MB]{base_dir}/worker")]; + conf.journal.raft_tick_interval_ms = 100; + conf.client.short_circuit = false; + conf.fault_injection.enabled = true; + conf.fault_injection.auth_token_env = TOKEN_ENV.to_string(); + + let cluster = Arc::new(MiniCluster::with_num(&conf, 1, 1)); + cluster.start_cluster(); + + let worker_conf = &cluster.worker_conf[0]; + let fault_base = format!( + "http://{}:{}", + worker_conf.worker.hostname, worker_conf.worker.web_port + ); + let controller = Arc::new(FaultHttpController::new( + fault_base, + std::env::var(TOKEN_ENV)?, + )?); + let mut faults = FaultTestSession::new(); + faults.add_target("worker", controller); + + cluster.clone_client_rt().block_on(async move { + faults.preflight().await?; + + let final_segment_delay = + FaultRuleBuilder::named("worker.load_task.parallel.before_segment_copy") + .matches("stream_index", FINAL_STREAM)? + .times(1)? + .delay(8_000)?; + faults + .configure("worker", "delay-final-segment", final_segment_delay) + .await?; + + let final_join_delay = + FaultRuleBuilder::named("worker.load_task.parallel.before_join_await") + .matches("stream_index", FINAL_STREAM)? + .times(1)? + .delay(2_000)?; + faults + .configure("worker", "delay-final-join", final_join_delay) + .await?; + + let runner_finished = FaultRuleBuilder::named("worker.load_task.after_run") + .times(1)? + .record()?; + faults + .configure("worker", "runner-finished", runner_finished) + .await?; + + let fs = cluster.new_fs(); + let mount_path = Path::from_str("/cache")?; + let ufs_root = Path::from_str(format!("file://{ufs_dir}"))?; + let mount_opts = MountOptions::builder() + .add_property("load_task.parallel_streams", "2") + .add_property("load_task.min_bytes_per_stream", "1") + .block_size(BLOCK_SIZE) + .replicas(1) + .auto_cache(false) + .build(); + fs.mount(&ufs_root, &mount_path, mount_opts).await?; + + let job_client = JobMasterClient::new(fs.fs_client()); + let source_path = format!("file://{source_file}"); + let load = job_client + .submit_load_job( + LoadJobCommand::builder(&source_path) + .block_size(BLOCK_SIZE) + .replicas(1) + .build(), + ) + .await?; + + faults + .wait_for_executions("worker", "delay-final-segment", 1, Duration::from_secs(20)) + .await?; + faults + .wait_for_executions("worker", "delay-final-join", 1, Duration::from_secs(20)) + .await?; + + job_client.cancel_job(&load.job_id).await?; + + faults + .wait_for_executions("worker", "runner-finished", 1, Duration::from_secs(20)) + .await?; + + let status = job_client.get_job_status(&load.job_id).await?; + assert_eq!(status.state, JobTaskState::Canceled); + + let target_path = Path::from_str(&load.target_path)?; + let target_status = fs.get_status(&target_path).await?; + assert_eq!( + target_status.storage_policy.ufs_mtime, 0, + "a canceled parallel load must not stamp UFS cache validity" + ); + assert!( + !target_status.cv_valid(None), + "a canceled parallel load must not become a valid cache entry" + ); + + faults.cleanup().await?; + CommonResult::Ok(()) + }) +} diff --git a/curvine-tests/regression/daily_regression_test.sh b/curvine-tests/regression/daily_regression_test.sh index a42f0f3e7..85baf514b 100755 --- a/curvine-tests/regression/daily_regression_test.sh +++ b/curvine-tests/regression/daily_regression_test.sh @@ -104,11 +104,15 @@ run_specific_test_and_summary() { local safe_tf="${test_file//::/_}" local safe_tc="${test_case//::/_}" local log_file="$TEST_DIR/logs/$package/${safe_tf}_${safe_tc}.log" + local cargo_feature_args=() + if [ "$package" = "curvine-server" ]; then + cargo_feature_args=(--features fault-injection) + fi cd "$PROJECT_ROOT" if [ "$test_file" = "lib" ]; then - cargo test --package "$package" --lib -- "$test_case" --exact --nocapture > "$log_file" 2>&1 + cargo test --package "$package" "${cargo_feature_args[@]}" --lib -- "$test_case" --exact --nocapture > "$log_file" 2>&1 else - cargo test --package "$package" --test "$test_file" -- "$test_case" --exact --nocapture > "$log_file" 2>&1 + cargo test --package "$package" "${cargo_feature_args[@]}" --test "$test_file" -- "$test_case" --exact --nocapture > "$log_file" 2>&1 fi local exit_code=$? set -e @@ -150,7 +154,17 @@ run_tests_nextest() { local nextest_log="$TEST_DIR/nextest_run.log" local nextest_profile_dir="default" # Include skipped count in final summary (default is fail-only, so "N skipped" is hidden) - local nextest_args=(--config-file "$SCRIPT_DIR/nextest.toml" --no-fail-fast --workspace --final-status-level=skip) + # Include Curvine Server fault-injection tests in the regular Rust test + # discovery. Runtime fault endpoints remain disabled unless an individual + # test explicitly enables them in its ClusterConf. + local nextest_feature_args=(--features curvine-server/fault-injection) + local nextest_args=( + --config-file "$SCRIPT_DIR/nextest.toml" + --no-fail-fast + --workspace + --final-status-level=skip + "${nextest_feature_args[@]}" + ) if [ -n "${NEXTEST_PROFILE}" ]; then nextest_profile_dir="${NEXTEST_PROFILE}" nextest_args+=(--profile "$NEXTEST_PROFILE") @@ -184,8 +198,8 @@ run_tests_nextest() { rm -f "$skipped_list_file" if [ "$nextest_profile_dir" != "default" ]; then comm -23 \ - <(cd "$PROJECT_ROOT" && cargo nextest list --config-file "$SCRIPT_DIR/nextest.toml" --profile default 2>/dev/null | sort) \ - <(cd "$PROJECT_ROOT" && cargo nextest list --config-file "$SCRIPT_DIR/nextest.toml" --profile "$nextest_profile_dir" 2>/dev/null | sort) \ + <(cd "$PROJECT_ROOT" && cargo nextest list --config-file "$SCRIPT_DIR/nextest.toml" --workspace --profile default "${nextest_feature_args[@]}" 2>/dev/null | sort) \ + <(cd "$PROJECT_ROOT" && cargo nextest list --config-file "$SCRIPT_DIR/nextest.toml" --workspace --profile "$nextest_profile_dir" "${nextest_feature_args[@]}" 2>/dev/null | sort) \ > "$skipped_list_file" skipped_count=$(wc -l < "$skipped_list_file") [ "$skipped_count" -lt 0 ] && skipped_count=0 @@ -316,4 +330,3 @@ main() { } main "$@" -