fix(server): prevent partial parallel loads from becoming valid cache

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;
@@ -450,6 +480,22 @@ 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!(
+                "Task {} parallel load incomplete: wrote {} of {} bytes; refusing to mark cache valid",
+                self.task.info.task_id,
+                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();

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(())
+    })
+}

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 "$@"
-