Upstream source code synchronization

native/sync-core/src/api_client.rs

@@ -1,5 +1,6 @@
 use crate::errors::{Result, SyncError};
 use crate::models::*;
+use crate::server_error_code::api_code_to_error;
 use reqwest::Client;
 use serde::Deserialize;
 use std::sync::Arc;
@@ -154,12 +155,11 @@ impl ApiClient {
             return Err(SyncError::Network(format!("HTTP {}", resp.status())));
         }
         let api_resp: ApiResponse<serde_json::Value> = resp.json().await?;
-        if api_resp.code == 401 {
-            return Err(SyncError::Auth("Login required".into()));
-        }
-        if api_resp.code == 40004 {
-            return Err(SyncError::ObjectExisted);
+        if api_resp.code == 0 {
+            return Ok(api_resp.data.unwrap_or_default());
         }
+
+        // 40073 锁冲突需要特殊处理 data
         if api_resp.code == 40073 {
             let items = api_resp.data
                 .and_then(|d| d.as_array().cloned())
@@ -177,12 +177,11 @@ impl ApiClient {
                 .collect();
             return Err(SyncError::LockConflict { tokens: items });
         }
-        if api_resp.code != 0 {
-            return Err(SyncError::Network(
-                api_resp.msg.unwrap_or_else(|| format!("错误码: {}", api_resp.code)),
-            ));
-        }
-        Ok(api_resp.data.unwrap_or_default())
+
+        let msg = api_resp.msg
+            .filter(|m| !m.is_empty())
+            .unwrap_or_default();
+        Err(api_code_to_error(api_resp.code, &msg))
     }
 
     /// 发送带认证的请求，自动处理 401（刷新 token 后重试一次）
@@ -196,9 +195,22 @@ impl ApiClient {
 
         // 第一次尝试
         let token = self.token().await;
-        let resp = request_builder(token)
+        let resp = match request_builder(token)
             .header("X-Cr-Client-Id", &client_id)
-            .send().await?;
+            .send()
+            .await
+        {
+            Ok(r) => r,
+            Err(e) => {
+                tracing::warn!(
+                    "请求发送失败: kind={:?}, url={:?}, error={}",
+                    e.is_connect(),
+                    e.url().map(|u| u.as_str()),
+                    e
+                );
+                return Err(e.into());
+            }
+        };
         let result = self.parse_response(resp).await;
 
         if let Err(SyncError::Auth(_)) = result {
@@ -206,9 +218,22 @@ impl ApiClient {
             self.refresh_access_token().await?;
             // 用新 token 重试
             let new_token = self.token().await;
-            let resp = request_builder(new_token)
+            let resp = match request_builder(new_token)
                 .header("X-Cr-Client-Id", &client_id)
-                .send().await?;
+                .send()
+                .await
+            {
+                Ok(r) => r,
+                Err(e) => {
+                    tracing::warn!(
+                        "重试请求发送失败: kind={:?}, url={:?}, error={}",
+                        e.is_connect(),
+                        e.url().map(|u| u.as_str()),
+                        e
+                    );
+                    return Err(e.into());
+                }
+            };
             return self.parse_response(resp).await;
         }
 
@@ -261,7 +286,22 @@ impl ApiClient {
             }
 
             for dir_uri in dirs_to_recurse {
-                self.list_all_files_recursive(&dir_uri, result).await?;
+                let mut retry = 0u32;
+                loop {
+                    match self.list_all_files_recursive(&dir_uri, result).await {
+                        Ok(()) => break,
+                        Err(e) => {
+                            retry += 1;
+                            if retry > 3 {
+                                tracing::error!("递归列出目录失败，跳过: {}: {}", dir_uri, e);
+                                break;
+                            }
+                            let delay = crate::utils::retry_delay_ms(retry, 2000, 30000);
+                            tracing::warn!("递归列出目录失败 (重试 {}/3): {}: {}, {}ms后重试", retry, dir_uri, e, delay);
+                            tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
+                        }
+                    }
+                }
             }
 
             Ok(())
@@ -274,6 +314,33 @@ impl ApiClient {
         page: u32,
         page_size: u32,
         next_page_token: Option<&str>,
+    ) -> Result<ListFilesResponse> {
+        let max_retries = 3u32;
+        let mut attempt = 0u32;
+        loop {
+            attempt += 1;
+            match self.list_files_page_inner(uri, page, page_size, next_page_token).await {
+                Ok(resp) => return Ok(resp),
+                Err(SyncError::Auth(_)) => return Err(SyncError::Auth("Token 过期".into())),
+                Err(e) if attempt <= max_retries => {
+                    let delay = crate::utils::retry_delay_ms(attempt, 2000, 30000);
+                    tracing::warn!(
+                        "列出文件失败 (重试 {}/{}): uri={}, error={}, {}ms后重试",
+                        attempt, max_retries, uri, e, delay,
+                    );
+                    tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
+                }
+                Err(e) => return Err(e),
+            }
+        }
+    }
+
+    async fn list_files_page_inner(
+        &self,
+        uri: &str,
+        page: u32,
+        page_size: u32,
+        next_page_token: Option<&str>,
     ) -> Result<ListFilesResponse> {
         let data = self.send_with_auth_retry(|token| {
             let mut req = self.client
@@ -389,14 +456,56 @@ impl ApiClient {
         let chunk_size = data.get("chunk_size")
             .and_then(|c| c.as_u64())
             .unwrap_or(10 * 1024 * 1024);
+        let upload_urls: Vec<String> = data.get("upload_urls")
+            .and_then(|u| u.as_array())
+            .map(|arr| arr.iter().filter_map(|v| v.as_str().map(String::from)).collect())
+            .unwrap_or_default();
+        let storage_policy_type = data.get("storage_policy")
+            .and_then(|sp| sp.get("type"))
+            .and_then(|t| t.as_str())
+            .unwrap_or("local")
+            .to_string();
+        let callback_secret = data.get("callback_secret")
+            .and_then(|s| s.as_str())
+            .unwrap_or("")
+            .to_string();
+
+        let file_name = uri.rsplit('/').next().unwrap_or(uri).to_string();
+
+        tracing::info!(
+            "[{}] 创建上传会话: policy={}, urls={}, chunk_size={}",
+            file_name, storage_policy_type, upload_urls.len(), chunk_size,
+        );
 
         Ok(UploadSession {
             session_id,
             chunk_size,
+            upload_urls,
+            storage_policy_type,
+            callback_secret,
+            file_name,
         })
     }
 
     pub async fn upload_chunk(
+        &self,
+        session: &UploadSession,
+        index: u32,
+        data: &[u8],
+        file_size: u64,
+        task_id: &str,
+    ) -> Result<()> {
+        if let Some(url) = session.chunk_upload_url(index as usize) {
+            // 远程存储策略：直接上传到外部 URL（OneDrive/S3/OSS 等）
+            self.upload_chunk_to_remote(url, data, index, file_size, session, task_id).await
+        } else {
+            // 本地存储策略：上传到 Cloudreve 服务端
+            self.upload_chunk_local(&session.session_id, index, data).await
+        }
+    }
+
+    /// 本地存储：上传分片到 /file/upload/{session_id}/{index}
+    async fn upload_chunk_local(
         &self,
         session_id: &str,
         index: u32,
@@ -417,6 +526,88 @@ impl ApiClient {
         Ok(())
     }
 
+    /// 远程存储：上传分片到外部 URL（OneDrive/S3 等），无需 Cloudreve token
+    /// 必须带 Content-Range 头：bytes {start}-{end}/{total}
+    async fn upload_chunk_to_remote(
+        &self,
+        url: &str,
+        data: &[u8],
+        index: u32,
+        file_size: u64,
+        session: &UploadSession,
+        task_id: &str,
+    ) -> Result<()> {
+        let chunk_size = session.chunk_size;
+        let file_name = &session.file_name;
+        let start = index as u64 * chunk_size;
+        let end = start + data.len() as u64 - 1;
+        let content_range = format!("bytes {}-{}/{}", start, end, file_size);
+        let content_len = data.len().to_string();
+
+        tracing::debug!("[{}][{}] 远程存储上传分片 {}: Content-Range={}", task_id, file_name, index, content_range);
+
+        let resp = self.client
+            .put(url)
+            .header("Content-Length", &content_len)
+            .header("Content-Range", &content_range)
+            .body(data.to_vec())
+            .timeout(std::time::Duration::from_secs(300))
+            .send()
+            .await
+            .map_err(|e| {
+                tracing::warn!("[{}][{}] 远程存储上传失败: error={}", task_id, file_name, e);
+                SyncError::from(e)
+            })?;
+
+        if !resp.status().is_success() {
+            let status = resp.status();
+            let body = resp.text().await.unwrap_or_default();
+            return Err(SyncError::UploadFailed(format!(
+                "远程存储返回 HTTP {}: {}",
+                status,
+                body.chars().take(200).collect::<String>()
+            )));
+        }
+
+        // 202 = 分片已接收，上传未完成，继续下一个分片
+        // 200/201 = 上传完成，文件已创建
+        let status = resp.status();
+        if status.as_u16() == 202 {
+            tracing::debug!("[{}][{}] 远程存储分片 {} 已接收(202)，继续上传", task_id, file_name, index);
+        } else if status.as_u16() == 200 || status.as_u16() == 201 {
+            tracing::info!("[{}][{}] 远程存储上传完成({}), 分片 {}", task_id, file_name, status, index);
+        }
+
+        Ok(())
+    }
+
+    /// 远程存储上传完成后回调 Cloudreve 服务端
+    /// POST /callback/{storage_policy_type}/{session_id}/{callback_secret}
+    pub async fn callback_upload_complete(&self, session: &UploadSession, task_id: &str) -> Result<()> {
+        if session.callback_secret.is_empty() {
+            tracing::warn!("[{}][{}] 上传回调跳过: callback_secret 为空", task_id, session.file_name);
+            return Ok(());
+        }
+
+        let url = format!(
+            "{}/callback/{}/{}/{}",
+            self.base_url,
+            session.storage_policy_type,
+            session.session_id,
+            session.callback_secret,
+        );
+        tracing::info!("[{}][{}] 上传完成回调: policy={}, session={}", task_id, session.file_name, session.storage_policy_type, session.session_id);
+
+        self.send_with_auth_retry(|token| {
+            self.client
+                .post(&url)
+                .bearer_auth(&token)
+        }).await?;
+
+        tracing::info!("[{}][{}] 上传完成回调成功: policy={}, session={}", task_id, session.file_name, session.storage_policy_type, session.session_id);
+        Ok(())
+    }
+
     // ===== 下载 =====
 
     pub async fn get_download_url(&self, uris: &[&str]) -> Result<Vec<String>> {

native/sync-core/src/errors.rs

@@ -1,3 +1,4 @@
+use std::error::Error as StdError;
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -14,6 +15,18 @@ pub enum SyncError {
     #[error("远程文件已存在")]
     ObjectExisted,
 
+    #[error("存储策略不允许: {0}")]
+    StoragePolicyDenied(String),
+
+    #[error("上传失败: {0}")]
+    UploadFailed(String),
+
+    #[error("文件未找到: {0}")]
+    FileNotFound(String),
+
+    #[error("权限不足: {0}")]
+    PermissionDenied(String),
+
     #[error("文件锁定冲突")]
     LockConflict { tokens: Vec<LockConflictItem> },
 
@@ -56,7 +69,39 @@ impl From<r2d2::Error> for SyncError {
 
 impl From<reqwest::Error> for SyncError {
     fn from(e: reqwest::Error) -> Self {
-        SyncError::Network(e.to_string())
+        let mut detail = String::new();
+        if e.is_connect() {
+            detail.push_str("连接失败");
+        } else if e.is_timeout() {
+            detail.push_str("请求超时");
+        } else if e.is_request() {
+            detail.push_str("请求构建失败");
+        } else if e.is_body() {
+            detail.push_str("请求体错误");
+        } else if e.is_decode() {
+            detail.push_str("响应解码失败");
+        } else if e.is_redirect() {
+            detail.push_str("重定向过多");
+        }
+        let url = e.url().map(|u| u.to_string()).unwrap_or_default();
+        let source = StdError::source(&e)
+            .map(|s| format!(": {s}"))
+            .unwrap_or_default();
+        let msg = e.to_string();
+        // 如果 detail 为空，用 reqwest 原始消息
+        if detail.is_empty() {
+            SyncError::Network(if url.is_empty() {
+                format!("{msg}{source}")
+            } else {
+                format!("{msg} [{url}]{source}")
+            })
+        } else {
+            SyncError::Network(if url.is_empty() {
+                format!("{detail}: {msg}{source}")
+            } else {
+                format!("{detail}: {msg} [{url}]{source}")
+            })
+        }
     }
 }
 

native/sync-core/src/lib.rs

@@ -5,6 +5,7 @@ pub mod errors;
 pub mod utils;
 pub mod sync_db;
 pub mod api_client;
+pub mod server_error_code;
 pub mod fs_scanner;
 pub mod conflict_resolver;
 pub mod transfer;

native/sync-core/src/models.rs

@@ -283,6 +283,35 @@ pub enum PlatformCallbackEvent {
 pub struct UploadSession {
     pub session_id: String,
     pub chunk_size: u64,
+    /// 远程存储的上传 URL 列表（每个分片一个 URL）
+    /// 仅当 storage_policy_type != "local" 时有值
+    pub upload_urls: Vec<String>,
+    /// 存储策略类型：local / onedrive / s3 / oss / cos / etc.
+    pub storage_policy_type: String,
+    /// 回调密钥（远程存储上传完成后需要回调通知服务端）
+    pub callback_secret: String,
+    /// 上传文件名（仅用于日志标识，并发时区分不同文件）
+    pub file_name: String,
+}
+
+impl UploadSession {
+    /// 是否使用远程存储直接上传（非本地策略）
+    pub fn is_remote_storage(&self) -> bool {
+        self.storage_policy_type != "local" && !self.upload_urls.is_empty()
+    }
+
+    /// 获取第 index 个分片的上传 URL
+    /// 远程存储：返回 upload_urls[index]（不足则用最后一个）
+    /// 本地存储：返回 None，由调用方拼接 /file/upload/{session_id}/{index}
+    pub fn chunk_upload_url(&self, index: usize) -> Option<&str> {
+        if !self.is_remote_storage() {
+            return None;
+        }
+        if self.upload_urls.is_empty() {
+            return None;
+        }
+        Some(self.upload_urls[index.min(self.upload_urls.len() - 1)].as_str())
+    }
 }
 
 // ===== API 分页响应 =====

native/sync-core/src/server_error_code.rs

@@ -0,0 +1,92 @@
+use crate::errors::SyncError;
+
+/// Cloudreve 服务端错误码 → 中文描述映射
+pub fn server_code_desc(code: i32) -> Option<&'static str> {
+    Some(match code {
+        // HTTP 语义码
+        203 => "部分操作未成功",
+        401 => "未登录",
+        403 => "无权限访问",
+        404 => "资源不存在",
+        409 => "资源冲突",
+        // 4xxxx 业务码
+        40001 => "参数错误",
+        40002 => "上传失败",
+        40003 => "文件夹创建失败",
+        40004 => "对象已存在",
+        40005 => "签名已过期",
+        40006 => "当前存储策略不允许",
+        40007 => "用户组不允许此操作",
+        40008 => "需要管理员权限",
+        40009 => "主节点未注册",
+        40010 => "需要绑定手机",
+        40011 => "上传会话已过期",
+        40012 => "无效的分片序号",
+        40013 => "无效的 Content-Length",
+        40014 => "批量源大小超限",
+        40016 => "父目录不存在",
+        40017 => "用户被封禁",
+        40018 => "用户未激活",
+        40019 => "功能未启用",
+        40020 => "凭据无效",
+        40021 => "用户不存在",
+        40022 => "两步验证码错误",
+        40023 => "登录会话不存在",
+        40026 => "验证码错误",
+        40027 => "验证码需要刷新",
+        40035 => "存储策略不存在",
+        40044 => "文件未找到",
+        40045 => "列出文件失败",
+        40049 => "文件过大",
+        40050 => "文件类型不允许",
+        40051 => "用户容量不足",
+        40052 => "非法对象名",
+        40053 => "根目录受保护",
+        40054 => "同名文件正在上传",
+        40055 => "元数据不匹配",
+        40057 => "可用存储策略已变更",
+        40071 => "签名无效",
+        40073 => "文件锁定冲突",
+        40074 => "URI 数量过多",
+        40075 => "锁令牌已过期",
+        40077 => "实体不存在",
+        40078 => "文件在回收站中",
+        40079 => "文件数量已达上限",
+        40081 => "批量操作未完全完成",
+        40082 => "仅所有者可操作",
+        // 5xxxx 服务端内部错误
+        50001 => "数据库操作失败",
+        50002 => "加密失败",
+        50004 => "IO 操作失败",
+        50006 => "缓存操作失败",
+        50007 => "回调失败",
+        50010 => "节点离线",
+        _ => return None,
+    })
+}
+
+/// 将服务端业务错误码转为 SyncError
+pub fn api_code_to_error(code: i32, msg: &str) -> SyncError {
+    let desc = server_code_desc(code);
+    let detail = if msg.is_empty() {
+        desc.unwrap_or("未知错误").to_string()
+    } else if let Some(d) = desc {
+        if msg != d {
+            format!("{}: {}", d, msg)
+        } else {
+            msg.to_string()
+        }
+    } else {
+        msg.to_string()
+    };
+
+    match code {
+        401 => SyncError::Auth(detail),
+        40004 => SyncError::ObjectExisted,
+        40006 | 40035 | 40057 => SyncError::StoragePolicyDenied(detail),
+        40002 | 40011 | 40012 | 40013 | 40054 => SyncError::UploadFailed(detail),
+        40044 | 40077 => SyncError::FileNotFound(detail),
+        40017 | 40018 | 40007 | 40008 => SyncError::PermissionDenied(detail),
+        _ => SyncError::Network(format!("[{}] {}", code, detail)),
+    }
+}

native/sync-core/src/sync_engine/album.rs

@@ -22,7 +22,7 @@ impl SyncEngine {
                     let file_size = metadata.len();
                     match self.api.create_upload_session(remote_dcim_uri, file_size, false, None, None, None).await {
                         Ok(session) => {
-                            match crate::uploader::upload_file_chunked(&self.api, local_path, &session).await {
+                            match crate::uploader::upload_file_chunked(&self.api, local_path, &session, "album").await {
                                 Ok(_) => {
                                     let remote_uri = format!("{}/{}", remote_dcim_uri, file_name);
                                     let hash = crate::utils::quick_hash(local_path, file_size).await.unwrap_or_default();

native/sync-core/src/sync_engine/remote_events.rs

@@ -78,6 +78,10 @@ impl SyncEngine {
                 }
             }
             RemoteFileEvent::Deleted { uri, name } => {
+                // 清理过期抑制条目
+                let now = std::time::Instant::now();
+                self.suppress_paths.retain(|_, ts| now.duration_since(*ts).as_secs() < 30);
+
                 let relative = crate::diff::remote_relative_path(
                     remote_root,
                     uri,
@@ -86,6 +90,13 @@ impl SyncEngine {
                 );
                 tracing::info!("[远程事件] 删除: {}", relative);
 
+                // 被抑制的路径：上传失败清理远端碎片等场景，不应删除本地文件
+                if self.suppress_paths.contains_key(&relative) {
+                    tracing::info!("[远程事件] 删除已抑制，跳过本地删除: {}", relative);
+                    self.suppress_paths.remove(&relative);
+                    return;
+                }
+
                 let local_path = local_root.join(&relative);
                 if local_path.exists() {
                     if local_path.is_dir() {

native/sync-core/src/uploader.rs

@@ -126,9 +126,15 @@ pub async fn upload_file(
         let chunk = &buf[..filled];
         let mut chunk_retries = 0u32;
         loop {
-            match api.upload_chunk(&session.session_id, index, chunk).await {
+            match api.upload_chunk(&session, index, chunk, local.size, task_id).await {
                 Ok(_) => break,
                 Err(SyncError::Auth(_)) => return Err(SyncError::Auth("Token 过期".into())),
+                // 业务错误，重试无意义，直接失败
+                Err(e @ SyncError::StoragePolicyDenied(_))
+                | Err(e @ SyncError::UploadFailed(_))
+                | Err(e @ SyncError::FileNotFound(_))
+                | Err(e @ SyncError::PermissionDenied(_))
+                | Err(e @ SyncError::ObjectExisted) => return Err(e),
                 Err(e) if chunk_retries < max_retries => {
                     chunk_retries += 1;
                     let delay = crate::utils::retry_delay_ms(chunk_retries, 1000, 30000);
@@ -141,6 +147,13 @@ pub async fn upload_file(
         index += 1;
     }
 
+    // 远程存储策略：上传完成后回调 Cloudreve 服务端
+    if session.is_remote_storage() {
+        if let Err(e) = api.callback_upload_complete(&session, task_id).await {
+            tracing::warn!("[{}][{}] 上传完成回调失败: {}", task_id, session.file_name, e);
+        }
+    }
+
     // 上传完成后获取远程文件信息
     let remote_uri = file_uri.clone();
     let (remote_file_id, remote_hash) = match api.get_file_info(&remote_uri).await {
@@ -206,6 +219,11 @@ pub async fn retry_upload_session(
                 }
                 return Err(SyncError::LockConflict { tokens });
             }
+            // 业务错误，重试无意义，直接失败
+            Err(e @ SyncError::StoragePolicyDenied(_))
+            | Err(e @ SyncError::UploadFailed(_))
+            | Err(e @ SyncError::FileNotFound(_))
+            | Err(e @ SyncError::PermissionDenied(_)) => return Err(e),
             Err(e) if attempt <= max_retries => {
                 let delay = crate::utils::retry_delay_ms(attempt, 1000, 30000);
                 tracing::warn!("[{}] 创建上传会话失败，{}ms后重试 ({}): {}", task_id, delay, attempt, e);
@@ -273,9 +291,11 @@ pub async fn upload_file_chunked(
     api: &ApiClient,
     local_path: &Path,
     session: &UploadSession,
+    task_id: &str,
 ) -> Result<()> {
     let chunk_size = session.chunk_size as usize;
     let file = tokio::fs::File::open(local_path).await?;
+    let file_size = file.metadata().await.map(|m| m.len()).unwrap_or(0);
     let mut reader = tokio::io::BufReader::new(file);
     let mut buf = vec![0u8; chunk_size];
     let mut index: u32 = 0;
@@ -291,9 +311,16 @@ pub async fn upload_file_chunked(
         }
         if filled == 0 { break; }
 
-        api.upload_chunk(&session.session_id, index, &buf[..filled]).await?;
+        api.upload_chunk(session, index, &buf[..filled], file_size, task_id).await?;
         index += 1;
     }
 
+    // 远程存储策略：上传完成后回调 Cloudreve 服务端
+    if session.is_remote_storage() {
+        if let Err(e) = api.callback_upload_complete(session, task_id).await {
+            tracing::warn!("[{}][{}] 上传完成回调失败: {}", task_id, session.file_name, e);
+        }
+    }
+
     Ok(())
 }

native/sync-core/src/worker/pool.rs

@@ -26,6 +26,8 @@ pub struct WorkerPool {
     file_locks: Arc<FileLockRegistry>,
     ensured_dirs: Arc<DashMap<String, ()>>,
     event_sink: Arc<crate::event_sink::EventSink>,
+    /// 抑制路径：上传失败清理远端碎片时，防止 SSE 删除事件误删本地文件
+    suppress_paths: Arc<DashMap<String, std::time::Instant>>,
     shutdown_token: std::sync::Mutex<CancellationToken>,
     #[cfg(feature = "windows-cfapi")]
     platform_adapter: std::sync::Mutex<Option<Arc<dyn PlaceholderCreator>>>,
@@ -67,6 +69,7 @@ impl WorkerPool {
             file_locks,
             ensured_dirs,
             event_sink,
+            suppress_paths: Arc::new(DashMap::new()),
             shutdown_token: std::sync::Mutex::new(shutdown_token),
             #[cfg(feature = "windows-cfapi")]
             platform_adapter: std::sync::Mutex::new(None),
@@ -133,6 +136,7 @@ impl WorkerPool {
             self.ensured_dirs.clone(),
             conflict_resolver,
             self.event_sink.clone(),
+            self.suppress_paths.clone(),
             self.shutdown_token.lock().unwrap().clone(),
             #[cfg(feature = "windows-cfapi")]
             self.platform_adapter.lock().unwrap().clone(),
@@ -232,6 +236,7 @@ impl WorkerPool {
         let file_locks = self.file_locks.clone();
         let ensured_dirs = self.ensured_dirs.clone();
         let event_sink = self.event_sink.clone();
+        let suppress_paths = self.suppress_paths.clone();
         let shutdown_token = self.shutdown_token.lock().unwrap().clone();
         let active_workers = self.active_workers.clone();
         let active_upload_paths = self.active_upload_paths.clone();
@@ -255,6 +260,7 @@ impl WorkerPool {
                 ensured_dirs,
                 conflict_resolver,
                 event_sink,
+                suppress_paths,
                 shutdown_token,
                 #[cfg(feature = "windows-cfapi")]
                 platform_adapter,

native/sync-core/src/worker/worker_impl.rs

@@ -37,6 +37,8 @@ pub struct Worker {
     ensured_dirs: Arc<DashMap<String, ()>>,
     conflict_resolver: ConflictResolver,
     event_sink: Arc<crate::event_sink::EventSink>,
+    /// 抑制路径：上传失败清理远端碎片时，防止 SSE 删除事件误删本地文件
+    suppress_paths: Arc<DashMap<String, std::time::Instant>>,
     shutdown_token: CancellationToken,
     #[cfg(feature = "windows-cfapi")]
     platform_adapter: Option<Arc<dyn PlaceholderCreator>>,
@@ -55,6 +57,7 @@ impl Worker {
         ensured_dirs: Arc<DashMap<String, ()>>,
         conflict_resolver: ConflictResolver,
         event_sink: Arc<crate::event_sink::EventSink>,
+        suppress_paths: Arc<DashMap<String, std::time::Instant>>,
         shutdown_token: CancellationToken,
         #[cfg(feature = "windows-cfapi")] platform_adapter: Option<Arc<dyn PlaceholderCreator>>,
     ) -> Self {
@@ -69,6 +72,7 @@ impl Worker {
             ensured_dirs,
             conflict_resolver,
             event_sink,
+            suppress_paths,
             shutdown_token,
             #[cfg(feature = "windows-cfapi")]
             platform_adapter,
@@ -783,6 +787,8 @@ impl Worker {
                             Some(&e.to_string()),
                         )
                         .await;
+                    // 清理远端碎片：强制解锁 + 删除
+                    self.cleanup_failed_upload(&rel_path).await;
                 }
                 Err(e) => {
                     tracing::error!("[{}] 上传任务异常: {}: {}", tid, rel_path, e);
@@ -798,6 +804,8 @@ impl Worker {
                             Some(&e.to_string()),
                         )
                         .await;
+                    // 清理远端碎片：强制解锁 + 删除
+                    self.cleanup_failed_upload(&rel_path).await;
                 }
             }
         }
@@ -1208,7 +1216,22 @@ impl Worker {
         }
     }
 
-    /// 7. 删除远程文件（Full 始终删除；MirrorWcf 仅在 WcfDeleteMode::SyncRemote 时删除）
+    // 上传失败后清理远端碎片：删除远端可能已创建的部分文件
+    // delete_files 内部遇到锁冲突(40073)会自动解锁后重试
+    // 删除前先插入 suppress_paths，防止 SSE 删除事件误删本地文件
+    async fn cleanup_failed_upload(&self, relative_path: &str) {
+        let remote_uri = format!("{}/{}", self.config.remote_root, relative_path);
+        let tid = &self.task_id;
+
+        // 先标记抑制，30s 内 SSE 删除事件不会删本地文件
+        self.suppress_paths.insert(relative_path.to_string(), std::time::Instant::now());
+
+        match self.api.delete_files(&[&remote_uri]).await {
+            Ok(_) => tracing::info!("[{}] 上传失败清理-已删除远端碎片: {}", tid, remote_uri),
+            Err(e) => tracing::debug!("[{}] 上传失败清理-删除失败（可能不存在）: {}", tid, e),
+        }
+    }
+
     async fn step_delete_remote(&self, summary: &mut SyncSummary) {
         let should_delete = matches!(self.config.sync_mode, SyncMode::Full)
             || (matches!(self.config.sync_mode, SyncMode::MirrorWcf)