diff --git a/lib/presentation/pages/preview/cloudreve_file_app_page.dart b/lib/presentation/pages/preview/cloudreve_file_app_page.dart index 10275a0..de79b76 100644 --- a/lib/presentation/pages/preview/cloudreve_file_app_page.dart +++ b/lib/presentation/pages/preview/cloudreve_file_app_page.dart @@ -325,8 +325,12 @@ class _CloudreveFileAppPageState extends State { initialSettings: desktop.InAppWebViewSettings( javaScriptEnabled: true, domStorageEnabled: true, + isInspectable: true, + cacheMode: desktop.CacheMode.LOAD_DEFAULT, + supportMultipleWindows: true, transparentBackground: true, - supportZoom: false, + supportZoom: true, + useHybridComposition: true, ), onWebViewCreated: (controller) async { _desktopController = controller; diff --git a/lib/presentation/pages/share/share_link_page.dart b/lib/presentation/pages/share/share_link_page.dart index d5a2c96..ccb8491 100644 --- a/lib/presentation/pages/share/share_link_page.dart +++ b/lib/presentation/pages/share/share_link_page.dart @@ -1442,8 +1442,12 @@ class _OfficialDownloadDesktopWebViewState domStorageEnabled: true, supportZoom: false, transparentBackground: true, + isInspectable: true, + cacheMode: desktop.CacheMode.LOAD_DEFAULT, + supportMultipleWindows: false, useShouldOverrideUrlLoading: true, useOnDownloadStart: true, + useHybridComposition: true, ), onWebViewCreated: (controller) { _controller = controller; @@ -1457,6 +1461,18 @@ class _OfficialDownloadDesktopWebViewState ); }, onLoadStop: (_, url) => _installHook(), + shouldInterceptRequest: (_, request) async { + final url = request.url.toString(); + if (widget.onNavigationUrl(url)) { + return null; + } + return null; + }, + onLoadStart: (_, url) { + if (url != null) { + widget.onNavigationUrl(url.toString()); + } + }, shouldOverrideUrlLoading: (_, action) { final url = action.request.url?.toString(); if (url != null && widget.onNavigationUrl(url)) { diff --git a/lib/presentation/widgets/announcement_dialog.dart b/lib/presentation/widgets/announcement_dialog.dart index 6c53e10..e95405e 100644 --- a/lib/presentation/widgets/announcement_dialog.dart +++ b/lib/presentation/widgets/announcement_dialog.dart @@ -201,6 +201,9 @@ $body onLoadStop: (_, url) { if (mounted) setState(() => _loading = false); }, + onReceivedError: (_, request, error) { + if (mounted) setState(() => _loading = false); + }, ); } diff --git a/native/sync-core/src/api_client.rs b/native/sync-core/src/api_client.rs index 93012ee..4edf227 100644 --- a/native/sync-core/src/api_client.rs +++ b/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 = 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 { + 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 { 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 = 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::() + ))); + } + + // 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> { diff --git a/native/sync-core/src/errors.rs b/native/sync-core/src/errors.rs index ed7ecf4..0e872aa 100644 --- a/native/sync-core/src/errors.rs +++ b/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 }, @@ -56,7 +69,39 @@ impl From for SyncError { impl From 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}") + }) + } } } diff --git a/native/sync-core/src/lib.rs b/native/sync-core/src/lib.rs index fc451f1..6f1bd43 100644 --- a/native/sync-core/src/lib.rs +++ b/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; diff --git a/native/sync-core/src/models.rs b/native/sync-core/src/models.rs index 02bd295..9aa50d2 100644 --- a/native/sync-core/src/models.rs +++ b/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, + /// 存储策略类型: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 分页响应 ===== diff --git a/native/sync-core/src/server_error_code.rs b/native/sync-core/src/server_error_code.rs new file mode 100644 index 0000000..c4ad0f6 --- /dev/null +++ b/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)), + } +} diff --git a/native/sync-core/src/sync_engine/album.rs b/native/sync-core/src/sync_engine/album.rs index 554623f..2229496 100644 --- a/native/sync-core/src/sync_engine/album.rs +++ b/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(); diff --git a/native/sync-core/src/sync_engine/remote_events.rs b/native/sync-core/src/sync_engine/remote_events.rs index 80f51c5..bd3a06a 100644 --- a/native/sync-core/src/sync_engine/remote_events.rs +++ b/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() { diff --git a/native/sync-core/src/uploader.rs b/native/sync-core/src/uploader.rs index 4f2e9e9..c57f63c 100644 --- a/native/sync-core/src/uploader.rs +++ b/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(()) } diff --git a/native/sync-core/src/worker/pool.rs b/native/sync-core/src/worker/pool.rs index a5f7d76..767fd5d 100644 --- a/native/sync-core/src/worker/pool.rs +++ b/native/sync-core/src/worker/pool.rs @@ -26,6 +26,8 @@ pub struct WorkerPool { file_locks: Arc, ensured_dirs: Arc>, event_sink: Arc, + /// 抑制路径:上传失败清理远端碎片时,防止 SSE 删除事件误删本地文件 + suppress_paths: Arc>, shutdown_token: std::sync::Mutex, #[cfg(feature = "windows-cfapi")] platform_adapter: std::sync::Mutex>>, @@ -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, diff --git a/native/sync-core/src/worker/worker_impl.rs b/native/sync-core/src/worker/worker_impl.rs index 8f415b8..a2f78a6 100644 --- a/native/sync-core/src/worker/worker_impl.rs +++ b/native/sync-core/src/worker/worker_impl.rs @@ -37,6 +37,8 @@ pub struct Worker { ensured_dirs: Arc>, conflict_resolver: ConflictResolver, event_sink: Arc, + /// 抑制路径:上传失败清理远端碎片时,防止 SSE 删除事件误删本地文件 + suppress_paths: Arc>, shutdown_token: CancellationToken, #[cfg(feature = "windows-cfapi")] platform_adapter: Option>, @@ -55,6 +57,7 @@ impl Worker { ensured_dirs: Arc>, conflict_resolver: ConflictResolver, event_sink: Arc, + suppress_paths: Arc>, shutdown_token: CancellationToken, #[cfg(feature = "windows-cfapi")] platform_adapter: Option>, ) -> 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)