gongyun/app
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 7 Activity

主要合入内容：

Browse Source 
- 文件管理器分页加载、排序偏好、桌面表头排序、移动端排序菜单、Ctrl/Cmd+F搜索、Esc 关闭搜索、鼠标右滑返回上级。
- 移动端同步状态页、同步统计卡片、同步任务/累计统计事件、同步页布局更新。
- Android 相册同步默认路径改为 DCIM/Camera，下载路径改用 external_path。
- Rust sync-core 更新：Android 日志依赖、AlbumUpload/AlbumDownload、MirrorWcf 统计修复、Linux FUSE镜像挂载与读写同步支持。

- 依赖更新：external_path、fl_chart、pdfrx 升级，并修了 pdfrx 新版本的deprecated API。
- 新增文件包括：sort_options.dart、sync_page_android.dart、sync_stats_card.dart、platform/fuse.rs、sync_engine/fuse.rs。
This commit is contained in:
公云存储
2026-06-04 07:11:43 +08:00
parent 5ee6ba1e28
commit b02daf1448
56 changed files with 7589 additions and 1200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
native/sync-core/src/sync_engine/album.rs
+65 -10
View File
@@ -7,29 +7,54 @@ use super::SyncEngine;
impl SyncEngine {
pub async fn sync_album(&self, album_paths: Vec<String>, remote_dcim_uri: &str) -> Result<()> {
let synced = self.db.get_album_sync_records().await?;
let new_photos: Vec<_> = album_paths.iter().filter(|p| !synced.contains_key(*p)).collect();
let new_photos: Vec<_> = album_paths
.iter()
.filter(|p| !synced.contains_key(*p))
.collect();
let total = new_photos.len();
if total == 0 { return Ok(()); }
if total == 0 {
return Ok(());
}
for (i, photo_path) in new_photos.iter().enumerate() {
let local_path = Path::new(photo_path);
let file_name = local_path.file_name()
let file_name = local_path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_else(|| format!("photo_{}", i));
match tokio::fs::metadata(photo_path).await {
Ok(metadata) => {
let file_size = metadata.len();
match self.api.create_upload_session(remote_dcim_uri, file_size, false, None, None, None).await {
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, "album").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();
if let Err(e) = self.db.add_album_sync_record(photo_path, &remote_uri, &hash).await {
let hash = crate::utils::quick_hash(local_path, file_size)
.await
.unwrap_or_default();
if let Err(e) = self
.db
.add_album_sync_record(photo_path, &remote_uri, &hash)
.await
{
tracing::warn!("记录同步状态失败: {}", e);
}
tracing::info!("照片上传完成 ({}/{}): {}", i + 1, total, file_name);
tracing::info!(
"照片上传完成 ({}/{}): {}",
i + 1,
total,
file_name
);
}
Err(e) => tracing::error!("上传照片失败 {}: {}", file_name, e),
}
@@ -47,17 +72,47 @@ impl SyncEngine {
let files = self.api.list_files_page(base_uri, 0, 200, None).await?;
let dcim_exists = files.files.iter().any(|f| f.name == "DCIM" && f.is_dir);
let pictures_exists = files.files.iter().any(|f| f.name == "Pictures" && f.is_dir);
let dcim_uri = if dcim_exists {
Some(format!("{}/DCIM", base_uri))
} else {
None
};
// 检查 DCIM/Camera 子目录
let camera_exists = if let Some(ref dcim_uri) = dcim_uri {
let dcim_files = self.api.list_files_page(dcim_uri, 0, 200, None).await?;
dcim_files
.files
.iter()
.any(|f| f.name == "Camera" && f.is_dir)
} else {
false
};
Ok(CloudAlbumCheckResult {
dcim_exists,
pictures_exists,
dcim_uri: if dcim_exists { Some(format!("{}/DCIM", base_uri)) } else { None },
pictures_uri: if pictures_exists { Some(format!("{}/Pictures", base_uri)) } else { None },
dcim_uri,
pictures_uri: if pictures_exists {
Some(format!("{}/Pictures", base_uri))
} else {
None
},
camera_exists,
camera_uri: if camera_exists {
Some(format!("{}/DCIM/Camera", base_uri))
} else {
None
},
})
}
pub async fn create_album_dirs(&self, base_uri: &str) -> Result<()> {
self.api.create_directory(base_uri, "DCIM").await?;
self.api.create_directory(base_uri, "Pictures").await?;
// 创建 DCIM/Camera 子目录
let dcim_uri = format!("{}/DCIM", base_uri);
self.api.create_directory(&dcim_uri, "Camera").await?;
Ok(())
}
}
native/sync-core/src/sync_engine/continuous_sync.rs
+103 -55
View File
@@ -7,27 +7,37 @@ use super::SyncEngine;
impl SyncEngine {
/// 持续同步：双事件源驱动 (SSE + 本地文件监听)，按 sync_mode 选择事件源
pub async fn run_continuous(&self) -> Result<()> {
let event_handler = EventHandler::new(
self.api.clone(),
self.api.client_id().to_string(),
);
let event_handler = EventHandler::new(self.api.clone(), self.api.client_id().to_string());
let (local_root, remote_root, sync_mode) = {
let config = self.config.read().await;
(config.local_root.clone(), config.remote_root.clone(), config.sync_mode.clone())
(
config.local_root.clone(),
config.remote_root.clone(),
config.sync_mode.clone(),
)
};
// 仅 DownloadOnly、Full 和 MirrorWcf 订阅 SSE
let mut remote_rx = if matches!(sync_mode, SyncMode::DownloadOnly | SyncMode::Full | SyncMode::MirrorWcf) {
// 仅 DownloadOnly、Full、MirrorWcf、AlbumDownload 订阅 SSE
let mut remote_rx = if matches!(
sync_mode,
SyncMode::DownloadOnly | SyncMode::Full | SyncMode::MirrorWcf | SyncMode::AlbumDownload
) {
Some(event_handler.subscribe_sse(&remote_root).await?)
} else {
tracing::info!("仅上传模式: 不订阅 SSE 远程事件");
None
};
// 仅 UploadOnly、Full 和 MirrorWcf 启动本地文件监听
let mut local_rx = if matches!(sync_mode, SyncMode::UploadOnly | SyncMode::Full | SyncMode::MirrorWcf) {
Some(spawn_local_watcher(&local_root, self.shutdown_token.lock().unwrap().clone()))
// 仅 UploadOnly、Full、MirrorWcf、AlbumUpload 启动本地文件监听
let mut local_rx = if matches!(
sync_mode,
SyncMode::UploadOnly | SyncMode::Full | SyncMode::MirrorWcf | SyncMode::AlbumUpload
) {
Some(spawn_local_watcher(
&local_root,
self.shutdown_token.lock().unwrap().clone(),
))
} else {
tracing::info!("仅下载模式: 不启动本地文件监听");
None
@@ -46,9 +56,21 @@ impl SyncEngine {
#[cfg(not(feature = "windows-cfapi"))]
let _wcf_fetch_rx: Option<()> = None;
let mut debounce = crate::event_handler::EventDebouncer::new(
std::time::Duration::from_millis(500),
);
// MirrorFUSE: 取走 FUSE 请求接收端
#[cfg(feature = "linux-fuse")]
let mut fuse_request_rx = if matches!(sync_mode, SyncMode::MirrorWcf) {
self.fuse_request_rx
.lock()
.ok()
.and_then(|mut rx| rx.take())
} else {
None
};
#[cfg(not(feature = "linux-fuse"))]
let _fuse_request_rx: Option<()> = None;
let mut debounce =
crate::event_handler::EventDebouncer::new(std::time::Duration::from_millis(500));
let shutdown_token = self.shutdown_token.lock().unwrap().clone();
loop {
@@ -109,6 +131,29 @@ impl SyncEngine {
}
}
// FUSE 请求（仅 MirrorFUSE）
request = async {
#[cfg(feature = "linux-fuse")]
{
match &mut fuse_request_rx {
Some(rx) => rx.recv().await,
None => std::future::pending().await,
}
}
#[cfg(not(feature = "linux-fuse"))]
{
let _: Option<()> = std::future::pending().await;
None::<()>
}
} => {
if let Some(req) = request {
#[cfg(feature = "linux-fuse")]
self.handle_fuse_request(req, &local_root).await;
#[cfg(not(feature = "linux-fuse"))]
let _: () = req;
}
}
// 定期心跳
_ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {
tracing::trace!("持续同步心跳");
@@ -130,9 +175,9 @@ fn spawn_local_watcher(
let watch_root = watch_root.to_path_buf();
std::thread::spawn(move || {
use notify_debouncer_full::notify::{RecursiveMode, EventKind};
use notify_debouncer_full::notify::event::{ModifyKind, RenameMode};
use notify_debouncer_full::new_debouncer;
use notify_debouncer_full::notify::event::{ModifyKind, RenameMode};
use notify_debouncer_full::notify::{EventKind, RecursiveMode};
let tx = local_tx.clone();
let shutdown = shutdown_token.clone();
@@ -140,51 +185,54 @@ fn spawn_local_watcher(
let mut debouncer = match new_debouncer(
std::time::Duration::from_millis(500),
None,
move |result: notify_debouncer_full::DebounceEventResult| {
match result {
Ok(events) => {
for event in events {
if shutdown.is_cancelled() { return; }
let kind = event.kind;
let paths = &event.paths;
move |result: notify_debouncer_full::DebounceEventResult| match result {
Ok(events) => {
for event in events {
if shutdown.is_cancelled() {
return;
}
let kind = event.kind;
let paths = &event.paths;
let filtered: Vec<_> = paths.iter()
.filter(|p| !p.extension().map(|e| e == "sync_tmp").unwrap_or(false))
.cloned()
.collect();
if filtered.is_empty() { continue; }
let filtered: Vec<_> = paths
.iter()
.filter(|p| !p.extension().map(|e| e == "sync_tmp").unwrap_or(false))
.cloned()
.collect();
if filtered.is_empty() {
continue;
}
match kind {
EventKind::Create(_) => {
let _ = tx.blocking_send(LocalFileEvent::Created(filtered));
}
EventKind::Modify(ModifyKind::Name(RenameMode::From)) => {}
EventKind::Modify(ModifyKind::Name(RenameMode::To)) => {}
EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => {
if filtered.len() == 2 {
let _ = tx.blocking_send(LocalFileEvent::Renamed {
old_paths: vec![filtered[0].clone()],
new_paths: vec![filtered[1].clone()],
});
}
}
EventKind::Modify(ModifyKind::Name(RenameMode::Other)) => {
let _ = tx.blocking_send(LocalFileEvent::Modified(filtered));
}
EventKind::Modify(_) => {
let _ = tx.blocking_send(LocalFileEvent::Modified(filtered));
}
EventKind::Remove(_) => {
let _ = tx.blocking_send(LocalFileEvent::Deleted(filtered));
}
_ => {}
match kind {
EventKind::Create(_) => {
let _ = tx.blocking_send(LocalFileEvent::Created(filtered));
}
EventKind::Modify(ModifyKind::Name(RenameMode::From)) => {}
EventKind::Modify(ModifyKind::Name(RenameMode::To)) => {}
EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => {
if filtered.len() == 2 {
let _ = tx.blocking_send(LocalFileEvent::Renamed {
old_paths: vec![filtered[0].clone()],
new_paths: vec![filtered[1].clone()],
});
}
}
EventKind::Modify(ModifyKind::Name(RenameMode::Other)) => {
let _ = tx.blocking_send(LocalFileEvent::Modified(filtered));
}
EventKind::Modify(_) => {
let _ = tx.blocking_send(LocalFileEvent::Modified(filtered));
}
EventKind::Remove(_) => {
let _ = tx.blocking_send(LocalFileEvent::Deleted(filtered));
}
_ => {}
}
}
Err(errors) => {
for e in errors {
tracing::warn!("文件监听去抖错误: {}", e);
}
}
Err(errors) => {
for e in errors {
tracing::warn!("文件监听去抖错误: {}", e);
}
}
},
native/sync-core/src/sync_engine/fuse.rs
+657
View File
@@ -0,0 +1,657 @@
//! FUSE (Linux) 相关的 SyncEngine 方法
//! 仅在 linux-fuse feature 启用时编译
use crate::models::*;
use super::SyncEngine;
#[cfg(feature = "linux-fuse")]
impl SyncEngine {
/// 处理 FUSE 请求（统一分发）
pub(crate) async fn handle_fuse_request(
&self,
request: crate::platform::fuse::FuseRequest,
local_root: &std::path::Path,
) {
use crate::platform::fuse::FuseRequest;
match request {
FuseRequest::Read {
inode,
remote_uri,
offset,
length,
reply_tx,
} => {
self.handle_fuse_read(inode, &remote_uri, offset, length, reply_tx, local_root)
.await;
}
FuseRequest::Upload {
inode,
parent_ino,
name,
relative_path,
data,
tmp_path,
mtime_ms,
overwrite,
reply_tx,
} => {
self.handle_fuse_upload(
inode,
parent_ino,
&name,
&relative_path,
data,
tmp_path.as_deref(),
mtime_ms,
overwrite,
reply_tx,
)
.await;
}
FuseRequest::Mkdir {
inode,
parent_ino,
name,
relative_path,
reply_tx,
} => {
self.handle_fuse_mkdir(inode, parent_ino, &name, &relative_path, reply_tx)
.await;
}
FuseRequest::Unlink {
inode,
name,
is_dir,
remote_uri,
relative_path,
reply_tx,
} => {
self.handle_fuse_unlink(
inode,
&name,
is_dir,
&remote_uri,
&relative_path,
reply_tx,
)
.await;
}
FuseRequest::Rename {
inode,
old_name,
old_relative_path,
old_remote_uri,
new_parent_ino,
new_name,
new_relative_path,
reply_tx,
} => {
self.handle_fuse_rename(
inode,
&old_name,
&old_relative_path,
&old_remote_uri,
new_parent_ino,
&new_name,
&new_relative_path,
reply_tx,
)
.await;
}
}
}
/// MirrorFUSE: 处理 FUSE read 水合请求（按需下载）
pub(crate) async fn handle_fuse_read(
&self,
inode: u64,
remote_uri: &str,
offset: i64,
length: i64,
reply_tx: tokio::sync::oneshot::Sender<Result<Vec<u8>, String>>,
_local_root: &std::path::Path,
) {
tracing::debug!(
"FUSE 水合请求: ino={}, uri={}, offset={}, length={}",
inode,
remote_uri,
offset,
length
);
let root_id = match &self.sync_root_id {
Some(id) => id.clone(),
None => {
let _ = reply_tx.send(Err("sync_root_id 为空".to_string()));
return;
}
};
let remote_uri_owned = remote_uri.to_string();
let now = std::time::Instant::now();
self.hydration_cache
.retain(|_, (_, ts)| now.duration_since(*ts).as_secs() < 300);
let data = if let Some(cached) = self.hydration_cache.get(&remote_uri_owned) {
tracing::debug!("FUSE 水合缓存命中: {}", remote_uri_owned);
cached.0.clone()
} else {
tracing::info!("FUSE 水合下载: {}", remote_uri_owned);
let config = self.snapshot_worker_config().await;
let download_result = async {
let urls = self.api.get_download_url(&[&remote_uri_owned]).await;
let urls = match urls {
Ok(u) => u,
Err(crate::errors::SyncError::Auth(_)) => {
tracing::info!("FUSE 水合: token 过期，尝试刷新后重试");
self.api.refresh_access_token().await?;
self.api.get_download_url(&[&remote_uri_owned]).await?
}
Err(e) => return Err(e),
};
let download_url = urls.into_iter().next().ok_or_else(|| {
crate::errors::SyncError::Network("获取下载 URL 返回空列表".into())
})?;
let data = crate::downloader::download_to_buffer(
&self.api,
&download_url,
config.bandwidth_limit,
)
.await?;
Ok::<Vec<u8>, crate::errors::SyncError>(data)
}
.await;
match download_result {
Ok(data) => {
self.hydration_cache.insert(
remote_uri_owned.clone(),
(data.clone(), std::time::Instant::now()),
);
data
}
Err(e) => {
tracing::error!("FUSE 水合下载失败: {}: {}", remote_uri_owned, e);
let _ = reply_tx.send(Err(format!("下载失败: {}", e)));
return;
}
}
};
if let Ok(Some(mapping)) = self
.db
.find_mapping_by_remote_uri(&root_id, &remote_uri_owned)
.await
{
let _ = self
.db
.upsert_file_mapping(&FileMapping {
id: mapping.id,
sync_root_id: mapping.sync_root_id,
local_path: mapping.local_path.clone(),
remote_uri: mapping.remote_uri.clone(),
remote_file_id: mapping.remote_file_id.clone(),
local_hash: None,
remote_hash: mapping.remote_hash.clone(),
local_mtime: mapping.local_mtime,
remote_mtime: mapping.remote_mtime,
local_size: None,
remote_size: Some(data.len() as u64),
sync_status: SyncFileStatus::Synced,
is_placeholder: false,
})
.await;
}
let _ = reply_tx.send(Ok(data));
}
/// FUSE 上传：将写入的文件上传到云端
#[allow(clippy::too_many_arguments)]
async fn handle_fuse_upload(
&self,
inode: u64,
_parent_ino: u64,
name: &str,
relative_path: &str,
data: Vec<u8>,
tmp_path: Option<&str>,
_mtime_ms: i64,
overwrite: bool,
reply_tx: tokio::sync::oneshot::Sender<Result<crate::platform::fuse::UploadResult, String>>,
) {
let root_id = match &self.sync_root_id {
Some(id) => id.clone(),
None => {
let _ = reply_tx.send(Err("sync_root_id 为空".to_string()));
return;
}
};
let config = self.snapshot_worker_config().await;
let remote_root = &config.remote_root;
let file_uri = format!("{}/{}", remote_root, relative_path);
tracing::info!(
"FUSE 上传: {} ({}bytes, overwrite={})",
relative_path,
data.len(),
overwrite
);
// 确保远程父目录存在
if let Some(parent) = std::path::PathBuf::from(relative_path).parent() {
let parent_str = parent.to_string_lossy().to_string();
if !parent_str.is_empty() {
if let Err(e) = crate::uploader::ensure_remote_dirs(
"fuse",
remote_root,
&parent_str,
&self.api,
&self.ensured_dirs,
)
.await
{
tracing::warn!("FUSE 上传: 确保远程父目录失败 {}: {}", parent_str, e);
}
}
}
// 读取文件数据
let file_data = if !data.is_empty() {
data
} else if let Some(tmp) = tmp_path {
match tokio::fs::read(tmp).await {
Ok(d) => d,
Err(e) => {
let _ = reply_tx.send(Err(format!("读取临时文件失败: {}", e)));
return;
}
}
} else {
Vec::new()
};
let file_size = file_data.len() as u64;
// 创建上传会话
let session = match crate::uploader::retry_upload_session(
"fuse", &file_uri, file_size, 3, overwrite, None, None, None, &self.api,
)
.await
{
Ok(s) => s,
Err(e) => {
let _ = reply_tx.send(Err(format!("创建上传会话失败: {}", e)));
return;
}
};
let chunk_size = session.chunk_size as usize;
// 逐块上传
let mut offset = 0usize;
let mut index: u32 = 0;
while offset < file_data.len() {
let end = (offset + chunk_size).min(file_data.len());
let chunk = &file_data[offset..end];
let mut chunk_retries = 0u32;
loop {
match self
.api
.upload_chunk(&session, index, chunk, file_size, "fuse")
.await
{
Ok(_) => break,
Err(e) if chunk_retries < 3 => {
chunk_retries += 1;
tracing::warn!("FUSE 上传重试 ({}/{}): {}", chunk_retries, 3, e);
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
Err(e) => {
let _ = reply_tx.send(Err(format!("上传分片失败: {}", e)));
return;
}
}
}
offset = end;
index += 1;
}
// 远程存储策略回调
if session.is_remote_storage() {
if let Err(e) = self.api.callback_upload_complete(&session, "fuse").await {
tracing::warn!("FUSE 上传完成回调失败: {}", e);
}
}
// 获取远程文件信息
let (remote_file_id, remote_hash) = match self.api.get_file_info(&file_uri).await {
Ok(info) => (info.file_id.clone(), info.hash.clone()),
Err(e) => {
tracing::warn!("FUSE 上传后获取文件信息失败: {}", e);
(None, None)
}
};
// 更新 DB mapping
let _ = self
.db
.upsert_file_mapping(&FileMapping {
id: 0,
sync_root_id: root_id,
local_path: std::path::PathBuf::from(relative_path),
remote_uri: file_uri.clone(),
remote_file_id,
local_hash: None,
remote_hash: remote_hash.clone(),
local_mtime: Some(_mtime_ms),
remote_mtime: Some(_mtime_ms),
local_size: Some(file_size),
remote_size: Some(file_size),
sync_status: SyncFileStatus::Synced,
is_placeholder: false,
})
.await;
// 抑制 SSE 回弹
self.suppress_paths
.insert(relative_path.to_string(), std::time::Instant::now());
tracing::info!(
"FUSE 上传完成: {} → {} ({}bytes)",
name,
file_uri,
file_size
);
let _ = reply_tx.send(Ok(crate::platform::fuse::UploadResult {
remote_uri: file_uri,
remote_hash,
size: file_size,
}));
let _ = inode; // used for logging
}
/// FUSE 创建远程目录
async fn handle_fuse_mkdir(
&self,
_inode: u64,
_parent_ino: u64,
name: &str,
relative_path: &str,
reply_tx: tokio::sync::oneshot::Sender<Result<(), String>>,
) {
let root_id = match &self.sync_root_id {
Some(id) => id.clone(),
None => {
let _ = reply_tx.send(Err("sync_root_id 为空".to_string()));
return;
}
};
let config = self.snapshot_worker_config().await;
let parent_rel = std::path::PathBuf::from(relative_path)
.parent()
.map(|p| crate::utils::normalize_path(&p.to_string_lossy()))
.unwrap_or_default();
let parent_uri = if parent_rel.is_empty() {
config.remote_root.clone()
} else {
format!("{}/{}", config.remote_root, parent_rel)
};
tracing::info!("FUSE 创建目录: {} (parent_uri={})", name, parent_uri);
match self.api.create_directory(&parent_uri, name).await {
Ok(remote_entry) => {
let remote_uri = remote_entry.uri.clone();
// 更新 InodeStore 中的 remote_uri（在 await 之前释放锁）
{
let adapter = match self.fuse_adapter.lock() {
Ok(guard) => guard,
Err(e) => {
let _ = reply_tx.send(Err(format!("锁失败: {}", e)));
return;
}
};
if let Some(ref fuse) = *adapter {
fuse.inode_store().update_remote_uri(_inode, &remote_uri);
}
}
// 更新 DB mapping
let _ = self
.db
.upsert_file_mapping(&FileMapping {
id: 0,
sync_root_id: root_id,
local_path: std::path::PathBuf::from(relative_path),
remote_uri: remote_uri.clone(),
remote_file_id: remote_entry.file_id.clone(),
local_hash: None,
remote_hash: remote_entry.hash.clone(),
local_mtime: None,
remote_mtime: Some(remote_entry.mtime_ms),
local_size: None,
remote_size: Some(0),
sync_status: SyncFileStatus::Synced,
is_placeholder: false,
})
.await;
// 抑制 SSE 回弹
self.suppress_paths
.insert(relative_path.to_string(), std::time::Instant::now());
tracing::info!("FUSE 目录创建成功: {} → {}", name, remote_uri);
let _ = reply_tx.send(Ok(()));
}
Err(e) => {
tracing::error!("FUSE 创建目录失败: {}: {}", name, e);
let _ = reply_tx.send(Err(format!("创建目录失败: {}", e)));
}
}
}
/// FUSE 删除远程文件/目录
async fn handle_fuse_unlink(
&self,
_inode: u64,
name: &str,
_is_dir: bool,
remote_uri: &str,
relative_path: &str,
reply_tx: tokio::sync::oneshot::Sender<Result<(), String>>,
) {
let root_id = match &self.sync_root_id {
Some(id) => id.clone(),
None => {
let _ = reply_tx.send(Err("sync_root_id 为空".to_string()));
return;
}
};
tracing::info!("FUSE 删除: {} ({})", name, remote_uri);
match self.api.delete_files(&[remote_uri]).await {
Ok(()) => {
// 删除 DB mapping
let _ = self
.db
.delete_mapping_by_remote_uri(&root_id, remote_uri)
.await;
// 抑制 SSE 回弹
self.suppress_paths
.insert(relative_path.to_string(), std::time::Instant::now());
tracing::info!("FUSE 删除成功: {}", name);
let _ = reply_tx.send(Ok(()));
}
Err(e) => {
tracing::error!("FUSE 删除失败: {}: {}", name, e);
let _ = reply_tx.send(Err(format!("删除失败: {}", e)));
}
}
}
/// FUSE 重命名/移动
#[allow(clippy::too_many_arguments)]
async fn handle_fuse_rename(
&self,
_inode: u64,
old_name: &str,
old_relative_path: &str,
old_remote_uri: &str,
new_parent_ino: u64,
new_name: &str,
new_relative_path: &str,
reply_tx: tokio::sync::oneshot::Sender<Result<(), String>>,
) {
let root_id = match &self.sync_root_id {
Some(id) => id.clone(),
None => {
let _ = reply_tx.send(Err("sync_root_id 为空".to_string()));
return;
}
};
let config = self.snapshot_worker_config().await;
let _ = (new_parent_ino, new_name);
// 判断是同目录重命名还是跨目录移动
let old_parent_rel = std::path::PathBuf::from(old_relative_path)
.parent()
.map(|p| crate::utils::normalize_path(&p.to_string_lossy()))
.unwrap_or_default();
let new_parent_rel = std::path::PathBuf::from(new_relative_path)
.parent()
.map(|p| crate::utils::normalize_path(&p.to_string_lossy()))
.unwrap_or_default();
let result = if old_parent_rel == new_parent_rel {
// 同目录重命名
tracing::info!("FUSE 重命名: {} → {}", old_name, new_name);
self.api.rename_file(old_remote_uri, new_name).await
} else {
// 跨目录移动
let dst_uri = if new_parent_rel.is_empty() {
config.remote_root.clone()
} else {
format!("{}/{}", config.remote_root, new_parent_rel)
};
tracing::info!("FUSE 移动: {} → {}", old_remote_uri, dst_uri);
self.api
.move_files(&[old_remote_uri], &dst_uri, false)
.await
};
match result {
Ok(()) => {
let new_remote_uri = format!("{}/{}", config.remote_root, new_relative_path);
// 更新 InodeStore 中的 remote_uri（在 await 之前释放锁）
{
let adapter = match self.fuse_adapter.lock() {
Ok(guard) => guard,
Err(e) => {
let _ = reply_tx.send(Err(format!("锁失败: {}", e)));
return;
}
};
if let Some(ref fuse) = *adapter {
fuse.inode_store()
.update_remote_uri(_inode, &new_remote_uri);
}
}
// 更新 DB mapping
let _ = self
.db
.update_mapping_remote_uri(&root_id, old_remote_uri, &new_remote_uri)
.await;
// 抑制 SSE 回弹
self.suppress_paths
.insert(old_relative_path.to_string(), std::time::Instant::now());
self.suppress_paths
.insert(new_relative_path.to_string(), std::time::Instant::now());
tracing::info!(
"FUSE 重命名/移动成功: {} → {}",
old_relative_path,
new_relative_path
);
let _ = reply_tx.send(Ok(()));
}
Err(e) => {
tracing::error!("FUSE 重命名/移动失败: {}: {}", old_name, e);
let _ = reply_tx.send(Err(format!("重命名/移动失败: {}", e)));
}
}
}
/// MirrorFUSE: 为远程文件注册 FUSE inode（持续同步时远程新建/修改文件调用）
pub(crate) async fn _create_placeholder_for_remote(
&self,
relative: &str,
remote: &RemoteFileEntry,
_local_root: &std::path::Path,
_root_id: &str,
) {
let adapter = match self.fuse_adapter.lock() {
Ok(guard) => guard,
Err(e) => {
tracing::error!("FUSE adapter lock 失败: {}", e);
return;
}
};
if let Some(ref fuse) = *adapter {
let parent_rel = std::path::PathBuf::from(relative)
.parent()
.map(|p| crate::utils::normalize_path(&p.to_string_lossy()))
.unwrap_or_default();
let name = std::path::PathBuf::from(relative)
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
fuse.create_placeholder_for_remote(
&parent_rel,
&name,
relative,
remote.is_dir,
remote.size,
&remote.uri,
remote.hash.as_deref(),
remote.mtime_ms,
);
}
}
/// FUSE 清理（卸载挂载点）
pub(crate) fn cleanup_fuse(&self) {
let adapter_opt = match self.fuse_adapter.lock() {
Ok(mut guard) => guard.take(),
Err(e) => {
tracing::error!("FUSE adapter lock 失败: {}", e);
return;
}
};
if let Some(adapter) = adapter_opt {
if let Err(e) = adapter.unmount() {
tracing::warn!("FUSE 卸载失败: {}", e);
}
tracing::info!("FUSE 适配器已清理");
}
}
}
native/sync-core/src/sync_engine/initial_sync.rs
+87 -6
View File
@@ -23,7 +23,11 @@ impl SyncEngine {
let (local_root, remote_root, sync_mode) = {
let config = self.config.read().await;
(config.local_root.clone(), config.remote_root.clone(), config.sync_mode.clone())
(
config.local_root.clone(),
config.remote_root.clone(),
config.sync_mode.clone(),
)
};
tracing::info!("开始初始同步, 模式={:?}", sync_mode);
@@ -47,7 +51,13 @@ impl SyncEngine {
}
let db_mappings = self.load_all_mappings().await?;
let plan = crate::diff::compute_diff(&local_files, &remote_files, &db_mappings, &remote_root, &sync_mode);
let plan = crate::diff::compute_diff(
&local_files,
&remote_files,
&db_mappings,
&remote_root,
&sync_mode,
);
tracing::info!(
"差异计算完成: 上传={}, 下载={}, 删本地={}, 删远程={}, 冲突={}",
plan.uploads.len(),
@@ -83,7 +93,8 @@ impl SyncEngine {
self.db.clone(),
self.api.clone(),
config.clone(),
).map_err(|e| crate::errors::SyncError::Internal(e.to_string()))?;
)
.map_err(|e| crate::errors::SyncError::Internal(e.to_string()))?;
let fetch_rx = adapter.take_fetch_receiver();
*self.wcf_fetch_rx.lock().unwrap() = fetch_rx;
let adapter_arc = std::sync::Arc::new(adapter);
@@ -96,9 +107,79 @@ impl SyncEngine {
}
}
let result = self.worker_pool.submit(
plan, worker_config, WorkerTrigger::InitialSync, conflict_resolver,
).await;
// MirrorFUSE 模式：初始化 FUSE 平台适配器（直接挂载到 local_root）
#[cfg(feature = "linux-fuse")]
if matches!(sync_mode, SyncMode::MirrorWcf) {
let already_initialized = self
.fuse_adapter
.lock()
.map(|g| g.is_some())
.unwrap_or(false);
if !already_initialized {
let config = self.config.read().await;
let mount_path = config.local_root.clone();
let adapter = crate::platform::fuse::FusePlatformAdapter::new(
&mount_path,
self.db.clone(),
self.api.clone(),
config.clone(),
)
.map_err(|e| crate::errors::SyncError::Internal(e.to_string()))?;
// 注册所有远程文件到 FUSE inode 表
for remote in &remote_files {
let relative = crate::diff::remote_relative_path(
&remote_root,
&remote.path,
&remote.name,
remote.is_dir,
);
let parent_rel = std::path::PathBuf::from(&relative)
.parent()
.map(|p| crate::utils::normalize_path(&p.to_string_lossy()))
.unwrap_or_default();
let name = std::path::PathBuf::from(&relative)
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
adapter.create_placeholder_for_remote(
&parent_rel,
&name,
&relative,
remote.is_dir,
remote.size,
&remote.uri,
remote.hash.as_deref(),
remote.mtime_ms,
);
}
let request_rx = adapter.take_request_receiver();
if let Ok(mut rx) = self.fuse_request_rx.lock() {
*rx = request_rx;
}
if let Ok(mut adapter_guard) = self.fuse_adapter.lock() {
*adapter_guard = Some(std::sync::Arc::new(adapter));
}
tracing::info!(
"MirrorFUSE: FUSE 平台适配器已初始化, 挂载点={}, inode 数={}",
mount_path.display(),
remote_files.len()
);
} else {
tracing::info!("MirrorFUSE: FUSE 平台适配器已存在，跳过重复初始化");
}
}
let result = self
.worker_pool
.submit(
plan,
worker_config,
WorkerTrigger::InitialSync,
conflict_resolver,
)
.await;
match result {
Ok(summary) => {
native/sync-core/src/sync_engine/local_events.rs
+272 -93
View File
@@ -24,28 +24,60 @@ impl SyncEngine {
// 清理过期的 suppress 记录（超过 30 秒）
let now = std::time::Instant::now();
self.suppress_paths.retain(|_, ts| now.duration_since(*ts).as_secs() < 30);
self.suppress_paths
.retain(|_, ts| now.duration_since(*ts).as_secs() < 30);
// === 第一步：提取 Renamed/Moved 事件，查 DB 构建操作 ===
let mut rename_remote: Vec<RenameAction> = Vec::new();
let mut move_remote: Vec<MoveAction> = Vec::new();
let mut handled_old_rels: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut handled_new_rels: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut handled_old_rels: std::collections::HashSet<String> =
std::collections::HashSet::new();
let mut handled_new_rels: std::collections::HashSet<String> =
std::collections::HashSet::new();
for event in &all_events {
match event {
LocalFileEvent::Renamed { old_paths, new_paths } => {
LocalFileEvent::Renamed {
old_paths,
new_paths,
} => {
for (old_path, new_path) in old_paths.iter().zip(new_paths.iter()) {
if let Some((old_rel, new_rel)) = rel_pair(local_root, old_path, new_path) {
if self.suppress_paths.contains_key(&old_rel) || self.suppress_paths.contains_key(&new_rel) {
tracing::trace!("本地重命名被抑制(远程操作导致): {} -> {}", old_rel, new_rel);
if self.suppress_paths.contains_key(&old_rel)
|| self.suppress_paths.contains_key(&new_rel)
{
tracing::trace!(
"本地重命名被抑制(远程操作导致): {} -> {}",
old_rel,
new_rel
);
continue;
}
let new_name = new_path.file_name()
let new_name = new_path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
if let Ok(Some(mapping)) = self.db.get_file_mapping(&root_id, &old_rel).await {
// Android 回收站：.trashed- 前缀视为本地删除，不重命名远程
if new_name.starts_with(".trashed-") {
if let Ok(Some(_)) =
self.db.get_file_mapping(&root_id, &old_rel).await
{
tracing::info!(
"检测到本地回收站重命名，视为删除: {} -> {}",
old_rel,
new_rel
);
let _ = self.db.delete_file_mapping(&root_id, &old_rel).await;
handled_old_rels.insert(old_rel);
handled_new_rels.insert(new_rel);
}
continue;
}
if let Ok(Some(mapping)) =
self.db.get_file_mapping(&root_id, &old_rel).await
{
tracing::info!("检测到本地重命名: {} -> {}", old_rel, new_rel);
rename_remote.push(RenameAction {
old_relative_path: old_rel.clone(),
@@ -56,27 +88,67 @@ impl SyncEngine {
handled_old_rels.insert(old_rel);
handled_new_rels.insert(new_rel);
} else {
tracing::info!("本地重命名但旧路径无DB映射，按新建处理: {} -> {}", old_rel, new_rel);
tracing::info!(
"本地重命名但旧路径无DB映射，按新建处理: {} -> {}",
old_rel,
new_rel
);
}
}
}
}
LocalFileEvent::Moved { old_paths, new_paths } => {
LocalFileEvent::Moved {
old_paths,
new_paths,
} => {
for (old_path, new_path) in old_paths.iter().zip(new_paths.iter()) {
if let Some((old_rel, new_rel)) = rel_pair(local_root, old_path, new_path) {
if self.suppress_paths.contains_key(&old_rel) || self.suppress_paths.contains_key(&new_rel) {
tracing::trace!("本地移动被抑制(远程操作导致): {} -> {}", old_rel, new_rel);
if self.suppress_paths.contains_key(&old_rel)
|| self.suppress_paths.contains_key(&new_rel)
{
tracing::trace!(
"本地移动被抑制(远程操作导致): {} -> {}",
old_rel,
new_rel
);
continue;
}
if let Ok(Some(mapping)) = self.db.get_file_mapping(&root_id, &old_rel).await {
let new_name = new_path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
// Android 回收站：移动到 .trashed- 前缀文件，视为本地删除
if new_name.starts_with(".trashed-") {
if let Ok(Some(_)) =
self.db.get_file_mapping(&root_id, &old_rel).await
{
tracing::info!(
"检测到本地回收站移动，视为删除: {} -> {}",
old_rel,
new_rel
);
let _ = self.db.delete_file_mapping(&root_id, &old_rel).await;
handled_old_rels.insert(old_rel);
handled_new_rels.insert(new_rel);
}
continue;
}
if let Ok(Some(mapping)) =
self.db.get_file_mapping(&root_id, &old_rel).await
{
let remote_root = { self.config.read().await.remote_root.clone() };
let new_rel_path = std::path::PathBuf::from(&new_rel);
let dst_dir_rel = new_rel_path.parent()
let dst_dir_rel = new_rel_path
.parent()
.map(|p| crate::utils::normalize_path(&p.to_string_lossy()))
.unwrap_or_default();
let dst_remote_dir_uri = format!("{}/{}",
let dst_remote_dir_uri = format!(
"{}/{}",
remote_root.trim_end_matches('/'),
dst_dir_rel.trim_start_matches('/'));
dst_dir_rel.trim_start_matches('/')
);
tracing::info!("检测到本地移动: {} -> {}", old_rel, new_rel);
move_remote.push(MoveAction {
@@ -88,7 +160,11 @@ impl SyncEngine {
handled_old_rels.insert(old_rel);
handled_new_rels.insert(new_rel);
} else {
tracing::info!("本地移动但旧路径无DB映射，按新建处理: {} -> {}", old_rel, new_rel);
tracing::info!(
"本地移动但旧路径无DB映射，按新建处理: {} -> {}",
old_rel,
new_rel
);
}
}
}
@@ -98,8 +174,10 @@ impl SyncEngine {
}
// === 第二步：按事件类型分类路径，跳过已识别为 rename/move 的路径 ===
let mut create_paths: std::collections::BTreeMap<String, std::path::PathBuf> = std::collections::BTreeMap::new();
let mut delete_paths: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
let mut create_paths: std::collections::BTreeMap<String, std::path::PathBuf> =
std::collections::BTreeMap::new();
let mut delete_paths: std::collections::BTreeSet<String> =
std::collections::BTreeSet::new();
for event in &all_events {
for path in event.paths() {
@@ -107,21 +185,31 @@ impl SyncEngine {
continue;
}
let file_name = path.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default();
if crate::fs_scanner::SKIP_NAMES.iter().any(|s| file_name == *s)
let file_name = path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
if crate::fs_scanner::SKIP_NAMES
.iter()
.any(|s| file_name == *s)
|| file_name.starts_with(".sync_")
|| crate::utils::is_conflict_file(&file_name) {
|| file_name.starts_with(".trashed-")
|| crate::utils::is_conflict_file(&file_name)
{
continue;
}
let relative = path.strip_prefix(local_root)
let relative = path
.strip_prefix(local_root)
.unwrap_or(path)
.to_string_lossy()
.to_string();
let relative = crate::utils::normalize_path(&relative);
if handled_old_rels.contains(&relative) || handled_new_rels.contains(&relative)
|| self.suppress_paths.contains_key(&relative) {
if handled_old_rels.contains(&relative)
|| handled_new_rels.contains(&relative)
|| self.suppress_paths.contains_key(&relative)
{
continue;
}
@@ -145,33 +233,54 @@ impl SyncEngine {
// === hash 匹配回退：检测 delete+create 为 rename 的情况 ===
// MirrorWcf: 跳过 hash 匹配回退，因为读取占位符文件会被 CFApi 拦截导致 426 超时
if !delete_paths.is_empty() && !create_paths.is_empty() && !matches!(sync_mode, SyncMode::MirrorWcf) {
let mut matched_deletes: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut matched_creates: std::collections::HashSet<String> = std::collections::HashSet::new();
if !delete_paths.is_empty()
&& !create_paths.is_empty()
&& !matches!(sync_mode, SyncMode::MirrorWcf)
{
let mut matched_deletes: std::collections::HashSet<String> =
std::collections::HashSet::new();
let mut matched_creates: std::collections::HashSet<String> =
std::collections::HashSet::new();
for (new_rel, new_path) in &create_paths {
if let Ok(metadata) = tokio::fs::metadata(new_path).await {
if metadata.is_dir() || metadata.len() == 0 { continue; }
let new_hash = crate::utils::quick_hash(new_path, metadata.len()).await.unwrap_or_default();
if new_hash.is_empty() { continue; }
if metadata.is_dir() || metadata.len() == 0 {
continue;
}
let new_hash = crate::utils::quick_hash(new_path, metadata.len())
.await
.unwrap_or_default();
if new_hash.is_empty() {
continue;
}
for del_rel in &delete_paths {
if matched_deletes.contains(del_rel.as_str()) { continue; }
if let Ok(Some(mapping)) = self.db.get_file_mapping(&root_id, del_rel).await {
if matched_deletes.contains(del_rel.as_str()) {
continue;
}
if let Ok(Some(mapping)) = self.db.get_file_mapping(&root_id, del_rel).await
{
if mapping.local_hash.as_deref() == Some(&new_hash) {
let new_name = new_path.file_name()
let new_name = new_path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
let old_dir = std::path::PathBuf::from(del_rel).parent()
let old_dir = std::path::PathBuf::from(del_rel)
.parent()
.map(|p| crate::utils::normalize_path(&p.to_string_lossy()))
.unwrap_or_default();
let new_dir = std::path::PathBuf::from(new_rel.as_str()).parent()
let new_dir = std::path::PathBuf::from(new_rel.as_str())
.parent()
.map(|p| crate::utils::normalize_path(&p.to_string_lossy()))
.unwrap_or_default();
if old_dir == new_dir {
tracing::info!("hash匹配检测到重命名: {} -> {}", del_rel, new_rel);
tracing::info!(
"hash匹配检测到重命名: {} -> {}",
del_rel,
new_rel
);
rename_remote.push(RenameAction {
old_relative_path: del_rel.clone(),
new_relative_path: new_rel.clone(),
@@ -179,11 +288,18 @@ impl SyncEngine {
new_name,
});
} else {
let remote_root = { self.config.read().await.remote_root.clone() };
let dst_remote_dir_uri = format!("{}/{}",
let remote_root =
{ self.config.read().await.remote_root.clone() };
let dst_remote_dir_uri = format!(
"{}/{}",
remote_root.trim_end_matches('/'),
new_dir.trim_start_matches('/'));
tracing::info!("hash匹配检测到移动: {} -> {}", del_rel, new_rel);
new_dir.trim_start_matches('/')
);
tracing::info!(
"hash匹配检测到移动: {} -> {}",
del_rel,
new_rel
);
move_remote.push(MoveAction {
old_relative_path: del_rel.clone(),
new_relative_path: new_rel.clone(),
@@ -212,9 +328,14 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
}
// === 提交移动任务 ===
@@ -225,9 +346,14 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
}
// === 提交上传任务 (Create/Modify) ===
@@ -265,12 +391,21 @@ impl SyncEngine {
let quick_hash = if matches!(sync_mode, SyncMode::MirrorWcf) {
String::new()
} else {
crate::utils::quick_hash(path, size).await.unwrap_or_default()
crate::utils::quick_hash(path, size)
.await
.unwrap_or_default()
};
let db_mapping = self.db.get_file_mapping(&root_id, relative).await.ok().flatten();
let db_mapping = self
.db
.get_file_mapping(&root_id, relative)
.await
.ok()
.flatten();
if let Some(ref mapping) = db_mapping {
if !quick_hash.is_empty() && mapping.local_hash.as_deref() == Some(&quick_hash) {
if !quick_hash.is_empty()
&& mapping.local_hash.as_deref() == Some(&quick_hash)
{
continue;
}
if mapping.is_placeholder {
@@ -278,7 +413,8 @@ impl SyncEngine {
}
}
let mtime_ms = metadata.modified()
let mtime_ms = metadata
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_millis() as i64)
@@ -303,28 +439,33 @@ impl SyncEngine {
let scan_dirs = find_top_level_dirs(&dir_paths);
let all_handled: std::collections::HashSet<&String> = handled_old_rels.iter()
let all_handled: std::collections::HashSet<&String> = handled_old_rels
.iter()
.chain(handled_new_rels.iter())
.collect();
let filtered_scan_dirs: Vec<String> = scan_dirs.into_iter().filter(|dir| {
if all_handled.iter().any(|rel| {
rel.starts_with(dir.as_str())
&& rel.as_bytes().get(dir.len()) == Some(&b'/')
}) {
return false;
}
for entry in self.suppress_paths.iter() {
let rel = entry.key();
if rel.starts_with(dir.as_str())
&& rel.as_bytes().get(dir.len()) == Some(&b'/') {
let filtered_scan_dirs: Vec<String> = scan_dirs
.into_iter()
.filter(|dir| {
if all_handled.iter().any(|rel| {
rel.starts_with(dir.as_str())
&& rel.as_bytes().get(dir.len()) == Some(&b'/')
}) {
return false;
}
if dir.as_str() == rel.as_str() {
return false;
for entry in self.suppress_paths.iter() {
let rel = entry.key();
if rel.starts_with(dir.as_str())
&& rel.as_bytes().get(dir.len()) == Some(&b'/')
{
return false;
}
if dir.as_str() == rel.as_str() {
return false;
}
}
}
true
}).collect();
true
})
.collect();
if !filtered_scan_dirs.is_empty() {
uploads.retain(|action| {
@@ -338,8 +479,10 @@ impl SyncEngine {
if !uploads.is_empty() || !filtered_scan_dirs.is_empty() {
tracing::info!(
"本地事件收集完成: 上传={}, 目录扫描={:?}, 跳过(未稳定)={}, 跳过(重复上传)={}",
uploads.len(), filtered_scan_dirs,
skipped_unstable, skipped_uploading,
uploads.len(),
filtered_scan_dirs,
skipped_unstable,
skipped_uploading,
);
let plan = SyncPlan {
uploads,
@@ -348,9 +491,14 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
}
}
@@ -359,7 +507,9 @@ impl SyncEngine {
// MirrorWcf 模式：仅在 wcf_delete_mode == SyncRemote 时删除远程，否则仅删除本地（保留远程以便重新水合）
let wcf_should_delete_remote = matches!(sync_mode, SyncMode::MirrorWcf)
&& matches!(wcf_delete_mode, WcfDeleteMode::SyncRemote);
if !delete_paths.is_empty() && (matches!(sync_mode, SyncMode::Full) || wcf_should_delete_remote) {
if !delete_paths.is_empty()
&& (matches!(sync_mode, SyncMode::Full) || wcf_should_delete_remote)
{
let mut delete_remote: Vec<SyncAction> = Vec::new();
for relative in &delete_paths {
tracing::info!("检测到本地文件删除: {}", relative);
@@ -392,9 +542,14 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
}
}
@@ -405,20 +560,47 @@ impl SyncEngine {
tracing::debug!("仅上传模式: 本地删除仅清理映射，不删除远程: {}", relative);
}
}
// MirrorWcf DeleteLocalOnly 模式下：本地删除仅清理 DB 映射，保留远程可重新水合
let wcf_delete_local_only = matches!(sync_mode, SyncMode::MirrorWcf)
&& !matches!(wcf_delete_mode, WcfDeleteMode::SyncRemote);
if !delete_paths.is_empty() && wcf_delete_local_only {
for relative in &delete_paths {
if let Ok(Some(_)) = self.db.get_file_mapping(&root_id, relative).await {
let _ = self.db.delete_file_mapping(&root_id, relative).await;
tracing::info!("MirrorWcf(仅删除本地): 清理映射，保留远程: {}", relative);
}
}
}
}
}
/// 从两个绝对路径生成相对于 local_root 的相对路径对
fn rel_pair(local_root: &std::path::Path, old_path: &std::path::Path, new_path: &std::path::Path) -> Option<(String, String)> {
let old_rel = old_path.strip_prefix(local_root).ok()?
.to_string_lossy().to_string();
let new_rel = new_path.strip_prefix(local_root).ok()?
.to_string_lossy().to_string();
Some((crate::utils::normalize_path(&old_rel), crate::utils::normalize_path(&new_rel)))
fn rel_pair(
local_root: &std::path::Path,
old_path: &std::path::Path,
new_path: &std::path::Path,
) -> Option<(String, String)> {
let old_rel = old_path
.strip_prefix(local_root)
.ok()?
.to_string_lossy()
.to_string();
let new_rel = new_path
.strip_prefix(local_root)
.ok()?
.to_string_lossy()
.to_string();
Some((
crate::utils::normalize_path(&old_rel),
crate::utils::normalize_path(&new_rel),
))
}
fn find_top_level_dirs(dirs: &[String]) -> Vec<String> {
if dirs.is_empty() { return Vec::new(); }
if dirs.is_empty() {
return Vec::new();
}
let mut sorted: Vec<&String> = dirs.iter().collect();
sorted.sort();
@@ -426,8 +608,7 @@ fn find_top_level_dirs(dirs: &[String]) -> Vec<String> {
let mut top_level = Vec::new();
for dir in &sorted {
let dominated = top_level.iter().any(|parent: &String| {
dir.starts_with(parent.as_str())
&& dir.as_bytes().get(parent.len()) == Some(&b'/')
dir.starts_with(parent.as_str()) && dir.as_bytes().get(parent.len()) == Some(&b'/')
});
if !dominated {
top_level.retain(|existing: &String| {
@@ -449,13 +630,11 @@ async fn is_file_stable(path: &Path) -> bool {
};
let path_display = path.display().to_string();
let can_open = tokio::task::spawn_blocking(move || {
match std::fs::File::open(&path_display) {
Ok(_) => true,
Err(e) => {
tracing::trace!("文件稳定性检测打开失败: {}", e);
false
}
let can_open = tokio::task::spawn_blocking(move || match std::fs::File::open(&path_display) {
Ok(_) => true,
Err(e) => {
tracing::trace!("文件稳定性检测打开失败: {}", e);
false
}
})
.await
native/sync-core/src/sync_engine/mod.rs
+131 -58
View File
@@ -1,13 +1,15 @@
mod initial_sync;
mod album;
mod continuous_sync;
#[cfg(feature = "linux-fuse")]
mod fuse;
mod initial_sync;
mod local_events;
mod remote_events;
mod album;
#[cfg(feature = "windows-cfapi")]
mod wcf;
// 非 WCF feature 下的 stub 方法，供 remote_events.rs 编译通过
#[cfg(not(feature = "windows-cfapi"))]
// 非 WCF/FUSE feature 下的 stub 方法，供 remote_events.rs 编译通过
#[cfg(not(any(feature = "windows-cfapi", feature = "linux-fuse")))]
impl SyncEngine {
async fn _create_placeholder_for_remote(
&self,
@@ -16,7 +18,7 @@ impl SyncEngine {
_local_root: &std::path::Path,
_root_id: &str,
) {
// MirrorWcf 模式在非 Windows 平台不可用，此方法不应被调用
// MirrorWcf 模式在非 Windows/Linux-FUSE 平台不可用，此方法不应被调用
}
}
@@ -31,9 +33,9 @@ use crate::worker::WorkerPool;
use dashmap::DashMap;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
#[cfg(feature = "windows-cfapi")]
use tokio::sync::mpsc;
use tokio::sync::RwLock;
use tokio_util::sync::CancellationToken;
pub struct SyncEngine {
@@ -66,15 +68,35 @@ pub struct SyncEngine {
/// 缓存的本地同步根路径（WCF 清理时同步读取，避免 await）
#[cfg(feature = "windows-cfapi")]
cached_local_root: std::sync::Mutex<std::path::PathBuf>,
/// FUSE 平台适配器（仅 MirrorWcf + linux-fuse 模式下初始化）
#[cfg(feature = "linux-fuse")]
fuse_adapter: std::sync::Mutex<Option<Arc<crate::platform::fuse::FusePlatformAdapter>>>,
/// FUSE 请求接收端（在适配器初始化时提取）
#[cfg(feature = "linux-fuse")]
fuse_request_rx:
std::sync::Mutex<Option<tokio::sync::mpsc::Receiver<crate::platform::fuse::FuseRequest>>>,
/// FUSE 水合缓存：uri → 已下载的完整文件数据
#[cfg(feature = "linux-fuse")]
hydration_cache: Arc<DashMap<String, (Vec<u8>, std::time::Instant)>>,
}
impl SyncEngine {
pub async fn new(config: SyncConfig) -> Result<Self> {
let db_path = config.data_dir.join("sync_core").join("datas").join(".sync_db.sqlite3");
let db_path = config
.data_dir
.join("sync_core")
.join("datas")
.join(".sync_db.sqlite3");
let db_path_clone = db_path.clone();
let db = Arc::new(tokio::task::spawn_blocking(move || SyncDb::open(&db_path_clone)).await??);
let db =
Arc::new(tokio::task::spawn_blocking(move || SyncDb::open(&db_path_clone)).await??);
let api = Arc::new(ApiClient::new(&config.base_url, &config.access_token, &config.refresh_token, &config.client_id));
let api = Arc::new(ApiClient::new(
&config.base_url,
&config.access_token,
&config.refresh_token,
&config.client_id,
));
let conflict = ConflictResolver::new(config.conflict_strategy.clone());
@@ -127,6 +149,12 @@ impl SyncEngine {
hydration_cache: Arc::new(DashMap::new()),
#[cfg(feature = "windows-cfapi")]
cached_local_root: std::sync::Mutex::new(std::path::PathBuf::new()),
#[cfg(feature = "linux-fuse")]
fuse_adapter: std::sync::Mutex::new(None),
#[cfg(feature = "linux-fuse")]
fuse_request_rx: std::sync::Mutex::new(None),
#[cfg(feature = "linux-fuse")]
hydration_cache: Arc::new(DashMap::new()),
})
}
@@ -184,8 +212,8 @@ impl SyncEngine {
}
/// 重置同步：停止任务 → 清空 DB → 清空本地目录 → 回到初始状态
pub async fn reset_sync(&self) -> Result<()> {
tracing::info!("开始重置同步...");
pub async fn reset_sync(&self, delete_local_files: bool) -> Result<()> {
tracing::info!("开始重置同步... delete_local_files={}", delete_local_files);
// 1. 停止同步
self.stop().await?;
@@ -196,6 +224,12 @@ impl SyncEngine {
self.cleanup_wcf();
}
// 2b. 清理 FUSE
#[cfg(feature = "linux-fuse")]
{
self.cleanup_fuse();
}
// 3. 终止所有活跃 Worker
// 2. 终止所有活跃 Worker
@@ -206,19 +240,27 @@ impl SyncEngine {
tracing::info!("同步数据库已清空");
// 4. 清空本地同步目录（保留目录本身，只删内容）
let local_root = self.config.read().await.local_root.clone();
if local_root.exists() {
let entries = std::fs::read_dir(&local_root)
.map_err(|_| crate::errors::SyncError::DiskFull { needed: 0, available: 0 })?;
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
let _ = std::fs::remove_dir_all(&path);
} else {
let _ = std::fs::remove_file(&path);
if delete_local_files {
let local_root = self.config.read().await.local_root.clone();
if local_root.exists() {
let entries = std::fs::read_dir(&local_root).map_err(|_| {
crate::errors::SyncError::DiskFull {
needed: 0,
available: 0,
}
})?;
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
let _ = std::fs::remove_dir_all(&path);
} else {
let _ = std::fs::remove_file(&path);
}
}
tracing::info!("本地同步目录已清空: {}", local_root.display());
}
tracing::info!("本地同步目录已清空: {}", local_root.display());
} else {
tracing::info!("跳过清空本地同步目录");
}
// 5. 清空内存缓存
@@ -232,14 +274,30 @@ impl SyncEngine {
Ok(())
}
pub fn status(&self) -> SyncStatusSnapshot {
let state = self.state.try_read().map(|g| g.clone()).unwrap_or(SyncState::Idle);
pub async fn status(&self) -> SyncStatusSnapshot {
let state = self
.state
.try_read()
.map(|g| g.clone())
.unwrap_or(SyncState::Idle);
let (synced_files, total_files) = match &state {
SyncState::InitialSync { progress } => {
let done = progress.uploaded + progress.downloaded;
(done, progress.total_to_sync)
}
SyncState::Continuous | SyncState::Paused => {
// 持续同步/暂停：从活跃任务聚合进度
let mut synced: u64 = 0;
let mut total: u64 = 0;
if let Ok(tasks) = self.db.get_active_sync_tasks().await {
for t in &tasks {
synced += t.completed_count as u64;
total += t.total_count as u64;
}
}
(synced, total)
}
_ => (0, 0),
};
@@ -288,7 +346,11 @@ impl SyncEngine {
}
tracing::info!(
"同步配置已更新: 模式={}, 冲突策略={}, WCF删除={}, 并发={}, 带宽限制={:?}",
new_mode, new_conflict, new_wcf_delete, new_max_concurrent, new_bandwidth
new_mode,
new_conflict,
new_wcf_delete,
new_max_concurrent,
new_bandwidth
);
Ok(())
}
@@ -297,7 +359,10 @@ impl SyncEngine {
self.api.update_token(token).await;
}
pub async fn register_event_sink(&self, sink: crate::frb_generated::StreamSink<crate::api::ffi_types::SyncEventFfi>) {
pub async fn register_event_sink(
&self,
sink: crate::frb_generated::StreamSink<crate::api::ffi_types::SyncEventFfi>,
) {
self.event_sink.register(sink).await;
}
@@ -317,6 +382,10 @@ impl SyncEngine {
self.db.query_task_items(filter).await
}
pub async fn get_cum_stats(&self) -> Result<SyncCumStats> {
self.db.get_cum_stats().await
}
pub async fn hydrate_file(&self, local_path: &str) -> Result<()> {
#[cfg(feature = "windows-cfapi")]
{
@@ -340,42 +409,46 @@ impl SyncEngine {
};
let pool = self.db.read_pool();
let result = tokio::task::spawn_blocking(move || -> Result<HashMap<String, FileMapping>> {
let conn = pool.get()?;
let mut stmt = conn.prepare(
"SELECT id, sync_root_id, local_path, remote_uri, remote_file_id,
let result =
tokio::task::spawn_blocking(move || -> Result<HashMap<String, FileMapping>> {
let conn = pool.get()?;
let mut stmt = conn.prepare(
"SELECT id, sync_root_id, local_path, remote_uri, remote_file_id,
local_hash, remote_hash, local_mtime, remote_mtime,
local_size, remote_size, sync_status, is_placeholder
FROM file_mapping WHERE sync_root_id = ?1"
)?;
FROM file_mapping WHERE sync_root_id = ?1",
)?;
let mappings: HashMap<String, FileMapping> = stmt.query_map(
rusqlite::params![root_id],
|row| {
let local_path: String = row.get(2)?;
Ok((
crate::utils::normalize_path(&local_path),
FileMapping {
id: row.get(0)?,
sync_root_id: row.get(1)?,
local_path: std::path::PathBuf::from(local_path),
remote_uri: row.get(3)?,
remote_file_id: row.get(4)?,
local_hash: row.get(5)?,
remote_hash: row.get(6)?,
local_mtime: row.get(7)?,
remote_mtime: row.get(8)?,
local_size: row.get(9)?,
remote_size: row.get(10)?,
sync_status: crate::diff::parse_sync_status_from_str(&row.get::<_, String>(11)?),
is_placeholder: row.get::<_, i32>(12)? != 0,
},
))
},
)?.filter_map(|r| r.ok()).collect();
let mappings: HashMap<String, FileMapping> = stmt
.query_map(rusqlite::params![root_id], |row| {
let local_path: String = row.get(2)?;
Ok((
crate::utils::normalize_path(&local_path),
FileMapping {
id: row.get(0)?,
sync_root_id: row.get(1)?,
local_path: std::path::PathBuf::from(local_path),
remote_uri: row.get(3)?,
remote_file_id: row.get(4)?,
local_hash: row.get(5)?,
remote_hash: row.get(6)?,
local_mtime: row.get(7)?,
remote_mtime: row.get(8)?,
local_size: row.get(9)?,
remote_size: row.get(10)?,
sync_status: crate::diff::parse_sync_status_from_str(
&row.get::<_, String>(11)?,
),
is_placeholder: row.get::<_, i32>(12)? != 0,
},
))
})?
.filter_map(|r| r.ok())
.collect();
Ok(mappings)
}).await??;
Ok(mappings)
})
.await??;
Ok(result)
}
native/sync-core/src/sync_engine/remote_events.rs
+165 -45
View File
@@ -11,6 +11,7 @@ impl SyncEngine {
remote_root: &str,
) {
// UploadOnly: 忽略所有远程事件
// AlbumDownload: 忽略删除事件（不删本地照片）
let sync_mode = {
let config = self.config.read().await;
config.sync_mode.clone()
@@ -30,12 +31,21 @@ impl SyncEngine {
&remote.name,
remote.is_dir,
);
tracing::info!("[远程事件] {}/{:?}: {}", event_type_name(&event), remote.file_id, relative);
tracing::info!(
"[远程事件] {}/{:?}: {}",
event_type_name(&event),
remote.file_id,
relative
);
let remote_entry = if remote.size == 0 && !remote.is_dir {
match self.api.get_file_info(&remote.uri).await {
Ok(info) => {
tracing::debug!("[远程事件] 获取文件详情成功: {} ({}bytes)", relative, info.size);
tracing::debug!(
"[远程事件] 获取文件详情成功: {} ({}bytes)",
relative,
info.size
);
info
}
Err(e) => {
@@ -58,8 +68,19 @@ impl SyncEngine {
if is_mirror_wcf {
self._create_placeholder_for_remote(
&relative, &remote_entry, local_root, &root_id,
).await;
&relative,
&remote_entry,
local_root,
&root_id,
)
.await;
self._record_wcf_stats(
&relative,
TaskActionType::CreatePlaceholder,
remote_entry.size,
None,
)
.await;
} else {
let plan = SyncPlan {
downloads: vec![SyncAction {
@@ -72,24 +93,32 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
}
}
RemoteFileEvent::Deleted { uri, name } => {
// 清理过期抑制条目
let now = std::time::Instant::now();
self.suppress_paths.retain(|_, ts| now.duration_since(*ts).as_secs() < 30);
self.suppress_paths
.retain(|_, ts| now.duration_since(*ts).as_secs() < 30);
let relative = crate::diff::remote_relative_path(
remote_root,
uri,
name,
false,
);
let relative = crate::diff::remote_relative_path(remote_root, uri, name, false);
tracing::info!("[远程事件] 删除: {}", relative);
// AlbumDownload: 远程删除不删除本地文件，仅清理 DB 映射
if matches!(sync_mode, SyncMode::AlbumDownload) {
tracing::info!("[远程事件] AlbumDownload 模式，跳过本地删除: {}", relative);
let _ = self.db.delete_file_mapping(&root_id, &relative).await;
return;
}
// 被抑制的路径：上传失败清理远端碎片等场景，不应删除本地文件
if self.suppress_paths.contains_key(&relative) {
tracing::info!("[远程事件] 删除已抑制，跳过本地删除: {}", relative);
@@ -98,7 +127,8 @@ impl SyncEngine {
}
let local_path = local_root.join(&relative);
if local_path.exists() {
let existed = local_path.exists();
if existed {
if local_path.is_dir() {
let _ = tokio::fs::remove_dir_all(&local_path).await;
} else {
@@ -107,15 +137,27 @@ impl SyncEngine {
tracing::info!("[远程事件] 已删除本地文件: {}", relative);
}
let _ = self.db.delete_file_mapping(&root_id, &relative).await;
self.suppress_paths.insert(relative.clone(), std::time::Instant::now());
self.suppress_paths
.insert(relative.clone(), std::time::Instant::now());
// MirrorFUSE: 远程删除 → 从 FUSE inode 缓存移除
#[cfg(feature = "linux-fuse")]
if is_mirror_wcf {
let adapter = self.fuse_adapter.lock().unwrap();
if let Some(ref fuse) = *adapter {
fuse.remove_inode(&relative);
}
}
// MirrorWcf: 远程删除 → 删本地，记录统计
if is_mirror_wcf && existed {
self._record_wcf_stats(&relative, TaskActionType::DeleteLocal, 0, None)
.await;
}
}
RemoteFileEvent::Renamed { old_uri, new_entry } => {
let old_relative = crate::diff::remote_relative_path(
remote_root,
old_uri,
&new_entry.name,
false,
);
let old_relative =
crate::diff::remote_relative_path(remote_root, old_uri, &new_entry.name, false);
let new_relative = crate::diff::remote_relative_path(
remote_root,
&new_entry.path,
@@ -141,14 +183,30 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
} else if is_mirror_wcf {
let remote_entry = self.get_remote_entry_or_fallback(new_entry).await;
self._create_placeholder_for_remote(
&new_relative, &remote_entry, local_root, &root_id,
).await;
&new_relative,
&remote_entry,
local_root,
&root_id,
)
.await;
self._record_wcf_stats(
&format!("{} -> {}", old_relative, new_relative),
TaskActionType::Rename,
0,
None,
)
.await;
} else {
let remote_entry = self.get_remote_entry_or_fallback(new_entry).await;
let plan = SyncPlan {
@@ -162,18 +220,19 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
}
}
RemoteFileEvent::Moved { old_uri, new_entry } => {
let old_relative = crate::diff::remote_relative_path(
remote_root,
old_uri,
&new_entry.name,
false,
);
let old_relative =
crate::diff::remote_relative_path(remote_root, old_uri, &new_entry.name, false);
let new_relative = crate::diff::remote_relative_path(
remote_root,
&new_entry.path,
@@ -199,14 +258,30 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
} else if is_mirror_wcf {
let remote_entry = self.get_remote_entry_or_fallback(new_entry).await;
self._create_placeholder_for_remote(
&new_relative, &remote_entry, local_root, &root_id,
).await;
&new_relative,
&remote_entry,
local_root,
&root_id,
)
.await;
self._record_wcf_stats(
&format!("{} -> {}", old_relative, new_relative),
TaskActionType::Move,
0,
None,
)
.await;
} else {
let remote_entry = self.get_remote_entry_or_fallback(new_entry).await;
let plan = SyncPlan {
@@ -220,16 +295,61 @@ impl SyncEngine {
};
let worker_config = self.snapshot_worker_config().await;
let conflict_resolver = self.conflict.read().await.clone();
self.worker_pool.submit_background(
plan, worker_config, WorkerTrigger::Continuous, conflict_resolver,
).await;
self.worker_pool
.submit_background(
plan,
worker_config,
WorkerTrigger::Continuous,
conflict_resolver,
)
.await;
}
}
}
}
/// MirrorWcf 专用：记录绕过 WorkerPool 的操作统计
async fn _record_wcf_stats(
&self,
relative_path: &str,
action_type: TaskActionType,
file_size: u64,
error_message: Option<String>,
) {
let now = chrono::Utc::now().to_rfc3339();
let status = if error_message.is_none() {
TaskItemStatus::Completed
} else {
TaskItemStatus::Failed
};
let task_id = format!("wcf_{}", uuid::Uuid::new_v4());
if let Err(e) = self
.db
.record_standalone_task_item(
&WorkerTrigger::WcfEvent,
&SyncTaskItem {
id: 0,
task_id,
relative_path: relative_path.to_string(),
action_type,
status,
file_size,
error_message,
created_at: now.clone(),
updated_at: now,
},
)
.await
{
tracing::warn!("WCF 统计记录失败: {}", e);
}
}
/// 获取远程文件详情，失败则使用 SSE 数据回退
pub(crate) async fn get_remote_entry_or_fallback(&self, entry: &RemoteFileEntry) -> RemoteFileEntry {
pub(crate) async fn get_remote_entry_or_fallback(
&self,
entry: &RemoteFileEntry,
) -> RemoteFileEntry {
if entry.size == 0 && !entry.is_dir {
match self.api.get_file_info(&entry.uri).await {
Ok(info) => info,
native/sync-core/src/sync_engine/wcf.rs
+168 -70
View File
@@ -18,7 +18,8 @@ impl SyncEngine {
Err(e) => {
tracing::error!("WCF 水合: FileIdentity 反序列化失败: {}", e);
let _ = crate::platform::wcf::WcfPlatformAdapter::reject_fetch_data(
request.connection_key, request.transfer_key,
request.connection_key,
request.transfer_key,
);
return;
}
@@ -31,19 +32,26 @@ impl SyncEngine {
if remote_uri.is_empty() {
tracing::error!("WCF 水合: FileIdentity 中 uri 为空");
let _ = crate::platform::wcf::WcfPlatformAdapter::reject_fetch_data(
request.connection_key, request.transfer_key,
request.connection_key,
request.transfer_key,
);
return;
}
tracing::debug!("WCF 水合请求: uri={}, size={}, offset={}, length={}",
remote_uri, remote_size, request.required_offset, request.required_length);
tracing::debug!(
"WCF 水合请求: uri={}, size={}, offset={}, length={}",
remote_uri,
remote_size,
request.required_offset,
request.required_length
);
let root_id = match &self.sync_root_id {
Some(id) => id.clone(),
None => {
let _ = crate::platform::wcf::WcfPlatformAdapter::reject_fetch_data(
request.connection_key, request.transfer_key,
request.connection_key,
request.transfer_key,
);
return;
}
@@ -51,12 +59,13 @@ impl SyncEngine {
// 清理过期缓存（超过 5 分钟）
let now = std::time::Instant::now();
self.hydration_cache.retain(|_, (_, ts)| now.duration_since(*ts).as_secs() < 300);
self.hydration_cache
.retain(|_, (_, ts)| now.duration_since(*ts).as_secs() < 300);
// 尝试从缓存获取已下载的数据
let data = if let Some(cached) = self.hydration_cache.get(&remote_uri) {
// 尝试从缓存获取已下载的数据；cache miss 时下载并标记 is_new_download
let (data, is_new_download) = if let Some(cached) = self.hydration_cache.get(&remote_uri) {
tracing::debug!("WCF 水合缓存命中: {}", remote_uri);
cached.0.clone()
(cached.0.clone(), false)
} else {
tracing::info!("WCF 水合下载: {} ({}bytes)", remote_uri, remote_size);
let config = self.snapshot_worker_config().await;
@@ -72,27 +81,57 @@ impl SyncEngine {
}
Err(e) => return Err(e),
};
let download_url = urls.into_iter().next()
.ok_or_else(|| crate::errors::SyncError::Network("获取下载 URL 返回空列表".into()))?;
let download_url = urls.into_iter().next().ok_or_else(|| {
crate::errors::SyncError::Network("获取下载 URL 返回空列表".into())
})?;
let data = crate::downloader::download_to_buffer(
&self.api,
&download_url,
config.bandwidth_limit,
).await?;
)
.await?;
Ok::<Vec<u8>, crate::errors::SyncError>(data)
}.await;
}
.await;
match download_result {
Ok(data) => {
self.hydration_cache.insert(remote_uri.clone(), (data.clone(), std::time::Instant::now()));
data
self.hydration_cache.insert(
remote_uri.clone(),
(data.clone(), std::time::Instant::now()),
);
(data, true)
}
Err(e) => {
tracing::error!("WCF 水合下载失败: {}: {}", remote_uri, e);
// 下载失败时记录一次统计
let now_str = chrono::Utc::now().to_rfc3339();
let task_id = format!("hydration_{}", uuid::Uuid::new_v4());
if let Err(db_err) = self
.db
.record_standalone_task_item(
&crate::models::WorkerTrigger::Hydration,
&crate::models::SyncTaskItem {
id: 0,
task_id,
relative_path: remote_uri.clone(),
action_type: crate::models::TaskActionType::Hydration,
status: crate::models::TaskItemStatus::Failed,
file_size: remote_size,
error_message: Some(e.to_string()),
created_at: now_str.clone(),
updated_at: now_str,
},
)
.await
{
tracing::warn!("WCF 水合失败统计记录失败: {}", db_err);
}
let _ = crate::platform::wcf::WcfPlatformAdapter::reject_fetch_data(
request.connection_key, request.transfer_key,
request.connection_key,
request.transfer_key,
);
return;
}
@@ -120,25 +159,72 @@ impl SyncEngine {
offset as i64,
) {
Ok(_) => {
tracing::debug!("WCF 水合数据推送: {} offset={} len={}", remote_uri, offset, transfer_data.len());
tracing::debug!(
"WCF 水合数据推送: {} offset={} len={}",
remote_uri,
offset,
transfer_data.len()
);
if let Ok(Some(mapping)) = self.db.find_mapping_by_remote_uri(&root_id, &remote_uri).await {
self.suppress_paths.insert(mapping.local_path.to_string_lossy().into_owned(), std::time::Instant::now());
let _ = self.db.upsert_file_mapping(&FileMapping {
id: mapping.id,
sync_root_id: mapping.sync_root_id,
local_path: mapping.local_path.clone(),
remote_uri: mapping.remote_uri.clone(),
remote_file_id: mapping.remote_file_id.clone(),
local_hash: None,
remote_hash: if remote_hash.is_empty() { mapping.remote_hash.clone() } else { Some(remote_hash.clone()) },
local_mtime: mapping.local_mtime,
remote_mtime: mapping.remote_mtime,
local_size: mapping.local_size,
remote_size: Some(remote_size),
sync_status: SyncFileStatus::Synced,
is_placeholder: false,
}).await;
// 仅在首次下载（cache miss）时更新映射和记录统计，避免同一文件多次 range 请求重复计数
if is_new_download {
if let Ok(Some(mapping)) = self
.db
.find_mapping_by_remote_uri(&root_id, &remote_uri)
.await
{
self.suppress_paths.insert(
mapping.local_path.to_string_lossy().into_owned(),
std::time::Instant::now(),
);
let _ = self
.db
.upsert_file_mapping(&FileMapping {
id: mapping.id,
sync_root_id: mapping.sync_root_id,
local_path: mapping.local_path.clone(),
remote_uri: mapping.remote_uri.clone(),
remote_file_id: mapping.remote_file_id.clone(),
local_hash: None,
remote_hash: if remote_hash.is_empty() {
mapping.remote_hash.clone()
} else {
Some(remote_hash.clone())
},
local_mtime: mapping.local_mtime,
remote_mtime: mapping.remote_mtime,
local_size: mapping.local_size,
remote_size: Some(remote_size),
sync_status: SyncFileStatus::Synced,
is_placeholder: false,
})
.await;
// 记录水合操作到统计
let now_str = chrono::Utc::now().to_rfc3339();
let local_path_str = mapping.local_path.to_string_lossy().to_string();
let task_id = format!("hydration_{}", uuid::Uuid::new_v4());
if let Err(e) = self
.db
.record_standalone_task_item(
&crate::models::WorkerTrigger::Hydration,
&crate::models::SyncTaskItem {
id: 0,
task_id,
relative_path: local_path_str,
action_type: crate::models::TaskActionType::Hydration,
status: crate::models::TaskItemStatus::Completed,
file_size: remote_size,
error_message: None,
created_at: now_str.clone(),
updated_at: now_str,
},
)
.await
{
tracing::warn!("WCF 水合统计记录失败: {}", e);
}
}
}
}
Err(e) => {
@@ -169,7 +255,11 @@ impl SyncEngine {
if let Some(adapter) = self.platform_adapter.lock().unwrap().as_ref() {
match adapter.create_placeholder_for_remote(
local_path.parent().unwrap_or(local_root),
local_path.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default().as_str(),
local_path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default()
.as_str(),
remote.size,
&remote.uri,
remote.hash.as_deref(),
@@ -181,21 +271,24 @@ impl SyncEngine {
}
}
let _ = self.db.upsert_file_mapping(&FileMapping {
id: 0,
sync_root_id: root_id.to_string(),
local_path: std::path::PathBuf::from(relative),
remote_uri: remote.uri.clone(),
remote_file_id: remote.file_id.clone(),
local_hash: None,
remote_hash: remote.hash.clone(),
local_mtime: None,
remote_mtime: Some(remote.mtime_ms),
local_size: None,
remote_size: Some(remote.size),
sync_status: SyncFileStatus::Placeholder,
is_placeholder: true,
}).await;
let _ = self
.db
.upsert_file_mapping(&FileMapping {
id: 0,
sync_root_id: root_id.to_string(),
local_path: std::path::PathBuf::from(relative),
remote_uri: remote.uri.clone(),
remote_file_id: remote.file_id.clone(),
local_hash: None,
remote_hash: remote.hash.clone(),
local_mtime: None,
remote_mtime: Some(remote.mtime_ms),
local_size: None,
remote_size: Some(remote.size),
sync_status: SyncFileStatus::Placeholder,
is_placeholder: true,
})
.await;
}
}
@@ -241,31 +334,36 @@ impl SyncEngine {
fn list_placeholders_sync(&self, sync_root_id: &str) -> anyhow::Result<Vec<FileMapping>> {
let pool = self.db.read_pool();
let conn = pool.get().map_err(|e| anyhow::anyhow!("{}", e))?;
let mut stmt = conn.prepare(
"SELECT id, sync_root_id, local_path, remote_uri, remote_file_id,
let mut stmt = conn
.prepare(
"SELECT id, sync_root_id, local_path, remote_uri, remote_file_id,
local_hash, remote_hash, local_mtime, remote_mtime,
local_size, remote_size, sync_status, is_placeholder
FROM file_mapping WHERE sync_root_id = ?1 AND is_placeholder = 1",
).map_err(|e| anyhow::anyhow!("{}", e))?;
)
.map_err(|e| anyhow::anyhow!("{}", e))?;
let mappings = stmt.query_map(rusqlite::params![sync_root_id], |row| {
Ok(FileMapping {
id: row.get(0)?,
sync_root_id: row.get(1)?,
local_path: std::path::PathBuf::from(row.get::<_, String>(2)?),
remote_uri: row.get(3)?,
remote_file_id: row.get(4)?,
local_hash: row.get(5)?,
remote_hash: row.get(6)?,
local_mtime: row.get(7)?,
remote_mtime: row.get(8)?,
local_size: row.get(9)?,
remote_size: row.get(10)?,
sync_status: crate::sync_db::parse_sync_status(&row.get::<_, String>(11)?),
is_placeholder: row.get::<_, i32>(12)? != 0,
let mappings = stmt
.query_map(rusqlite::params![sync_root_id], |row| {
Ok(FileMapping {
id: row.get(0)?,
sync_root_id: row.get(1)?,
local_path: std::path::PathBuf::from(row.get::<_, String>(2)?),
remote_uri: row.get(3)?,
remote_file_id: row.get(4)?,
local_hash: row.get(5)?,
remote_hash: row.get(6)?,
local_mtime: row.get(7)?,
remote_mtime: row.get(8)?,
local_size: row.get(9)?,
remote_size: row.get(10)?,
sync_status: crate::sync_db::parse_sync_status(&row.get::<_, String>(11)?),
is_placeholder: row.get::<_, i32>(12)? != 0,
})
})
}).map_err(|e| anyhow::anyhow!("{}", e))?
.filter_map(|m| m.ok()).collect();
.map_err(|e| anyhow::anyhow!("{}", e))?
.filter_map(|m| m.ok())
.collect();
Ok(mappings)
}