perf(sync): 流式发现并重新绑定所有允许文件

frontend/src-tauri/src/payloads.rs

@@ -165,6 +165,21 @@ pub(crate) fn validate_record(file: &mut fs::File, path: &str, size: u64) -> Res
     Ok(())
 }
 pub(crate) fn hash_file(path: &Path) -> Result<String> {
+    hash_file_info(path).map(|(digest, _)| digest)
+}
+pub(crate) fn sync_file_info(source: &Path, path: &str) -> Result<(String, u64)> {
+    if crate::records::is_record(path) {
+        let mut bytes = Vec::new();
+        fs::File::open(source)?
+            .take(1024 * 1024 + 1)
+            .read_to_end(&mut bytes)?;
+        crate::records::validate(path, &bytes)?;
+        Ok((hash(&bytes), bytes.len() as u64))
+    } else {
+        hash_file_info(source)
+    }
+}
+fn hash_file_info(path: &Path) -> Result<(String, u64)> {
     let mut file = fs::File::open(path)?;
     let mut buffer = vec![0u8; VERIFY_BUFFER_BYTES];
     let mut hasher = Sha256::new();
@@ -180,7 +195,7 @@ pub(crate) fn hash_file(path: &Path) -> Result<String> {
         }
         hasher.update(&buffer[..count]);
     }
-    Ok(format!("{:x}", hasher.finalize()))
+    Ok((format!("{:x}", hasher.finalize()), total))
 }
 pub(crate) fn verify(path: &Path, digest: &str, size: u64) -> Result<()> {
     open_verified(path, digest, size).map(drop)

frontend/src-tauri/src/sync_discovery.rs

@@ -1,7 +1,7 @@
 //! Reconcile externally edited files against committed snapshots, never against UI read caches.
-use crate::workspace::{hash, HostError, Result, Workspace};
+use crate::workspace::{HostError, Result, Workspace};
 use rusqlite::{params, OptionalExtension};
-use std::{collections::HashSet, fs, io::Read, path::Path};
+use std::{collections::HashSet, fs, path::Path};
 use uuid::Uuid;
 /// File transport only. Logical records receive their own versioned whitelist separately.
 pub fn allowed(path: &str) -> bool {
@@ -93,20 +93,7 @@ impl Workspace {
         let mut changes = 0;
         for path in paths {
             let target = self.resolve(&path)?;
-            if fs::metadata(&target)?.len() > 104857600 {
+            let (digest, _) = crate::payloads::sync_file_info(&target, &path)?;
-                return Err(HostError::new("FILE_TOO_LARGE"));
-            }
-            let mut bytes = Vec::new();
-            fs::File::open(&target)?
-                .take(104857601)
-                .read_to_end(&mut bytes)?;
-            if bytes.len() > 104857600 {
-                return Err(HostError::new("FILE_TOO_LARGE"));
-            }
-            if crate::records::is_record(&path) {
-                crate::records::validate(&path, &bytes)?;
-            }
-            let digest = hash(&bytes);
             let previous = self.entry(&path)?;
             let observed: Option<(String, String, bool)> = if let Some(entry) = &previous {
                 self.db
@@ -129,9 +116,8 @@ impl Workspace {
                 continue;
             }
             // Confirm the snapshot without writing back over an external editor.
-            crate::payloads::verify(&target, &digest, bytes.len() as u64)?;
             let operation = Uuid::new_v4().to_string();
-            self.store_payload(&operation, &bytes)?;
+            self.store_payload_file(&operation, &target, &digest)?;
             let file_id = previous.map_or_else(|| Uuid::new_v4().to_string(), |e| e.file_id);
             let tx = self.db.transaction()?;
             tx.execute("INSERT INTO files VALUES (?1,?2,?3,1,0) ON CONFLICT(path) DO UPDATE SET hash=excluded.hash,revision=files.revision+1,deleted=0",params![file_id,path,digest])?;

frontend/src-tauri/src/sync_initial.rs

@@ -6,7 +6,9 @@ use crate::{
 };
 use rusqlite::{params, OptionalExtension};
 use serde::{Deserialize, Serialize};
-use std::{collections::BTreeMap, fs};
+use std::collections::BTreeMap;
+#[cfg(test)]
+use std::fs;
 use uuid::Uuid;
 #[derive(Clone, Serialize, Deserialize)]
 pub struct Snapshot {
@@ -36,17 +38,11 @@ impl Workspace {
             .into_iter()
             .map(|path| {
                 let source = self.resolve(&path)?;
-                if fs::metadata(&source)?.len() > 104857600 {
+                let (hash, size) = crate::payloads::sync_file_info(&source, &path)?;
-                    return Err(HostError::new("FILE_TOO_LARGE"));
-                }
-                let bytes = fs::read(source)?;
-                if crate::records::is_record(&path) {
-                    crate::records::validate(&path, &bytes)?;
-                }
                 Ok(Local {
                     path,
-                    hash: hash(&bytes),
+                    hash,
-                    size: bytes.len(),
+                    size: size as usize,
                 })
             })
             .collect()
@@ -131,11 +127,14 @@ impl Workspace {
         let mut prepared = Vec::new();
         for item in local {
             let operation = Uuid::new_v4().to_string();
-            let bytes = fs::read(self.resolve(&item.path)?)?;
+            self.store_payload_file(&operation, &self.resolve(&item.path)?, &item.hash)
-            if hash(&bytes) != item.hash {
+                .map_err(|error| {
-                return Err(HostError::new("SYNC_PREVIEW_CHANGED"));
+                    if error.code == "REVISION_CONFLICT" {
+                        HostError::new("SYNC_PREVIEW_CHANGED")
+                    } else {
+                        error
                     }
-            self.store_payload(&operation, &bytes)?;
+                })?;
             let old = self.entry(&item.path)?;
             let remote = snapshot
                 .items
@@ -271,6 +270,98 @@ mod tests {
         }
     }
     #[test]
+    fn hundred_mib_discovery_preview_and_rebinding_preserve_all_current_files() {
+        use std::io::{Seek, SeekFrom, Write};
+        for initial in [false, true] {
+            let root = tempfile::tempdir().unwrap();
+            fs::create_dir(root.path().join("attachments")).unwrap();
+            let source = root.path().join("large.md");
+            let mut file = fs::File::create(&source).unwrap();
+            let block = vec![b'a'; 64 * 1024];
+            for _ in 0..1600 {
+                file.write_all(&block).unwrap();
+            }
+            file.sync_all().unwrap();
+            drop(file);
+            fs::copy(&source, root.path().join("attachments/large.bin")).unwrap();
+            let mut ws = Workspace::open(root.path()).unwrap();
+            let snapshot = Snapshot {
+                boundary: 0,
+                items: Vec::new(),
+            };
+            let binding = if initial {
+                let preview = ws
+                    .sync_preview("https://sync.example", "remote", "account", &snapshot)
+                    .unwrap();
+                let mut file = fs::OpenOptions::new().write(true).open(&source).unwrap();
+                file.seek(SeekFrom::End(-1)).unwrap();
+                file.write_all(b"b").unwrap();
+                file.sync_all().unwrap();
+                drop(file);
+                assert_eq!(
+                    ws.sync_bind_initial(
+                        "https://sync.example",
+                        "remote",
+                        "account",
+                        &snapshot,
+                        &preview.fingerprint
+                    )
+                    .err()
+                    .unwrap()
+                    .code,
+                    "SYNC_PREVIEW_CHANGED"
+                );
+                let preview = ws
+                    .sync_preview("https://sync.example", "remote", "account", &snapshot)
+                    .unwrap();
+                ws.sync_bind_initial(
+                    "https://sync.example",
+                    "remote",
+                    "account",
+                    &snapshot,
+                    &preview.fingerprint,
+                )
+                .unwrap()
+            } else {
+                ws.sync_bind_empty("https://sync.example", "remote", "account")
+                    .unwrap()
+            };
+            assert_eq!(ws.sync_discover(&binding.id).unwrap(), 0);
+            ws.sync_capture(&binding.id).unwrap();
+            assert_eq!(ws.pending_count().unwrap(), 2);
+            let identity = ws.entry("large.md").unwrap().unwrap().file_id;
+            let mut file = fs::OpenOptions::new().write(true).open(&source).unwrap();
+            file.seek(SeekFrom::End(-1)).unwrap();
+            file.write_all(b"c").unwrap();
+            file.sync_all().unwrap();
+            drop(file);
+            ws.scan().unwrap();
+            assert_eq!(ws.sync_discover(&binding.id).unwrap(), 1);
+            assert_eq!(ws.entry("large.md").unwrap().unwrap().file_id, identity);
+            drop(ws);
+            ws = Workspace::open(root.path()).unwrap();
+            assert_eq!(ws.sync_discover(&binding.id).unwrap(), 0);
+            assert_eq!(ws.pending_count().unwrap(), 3);
+            ws.sync_unbind(&binding.id).unwrap();
+            let rebound = ws
+                .sync_bind_empty("https://sync.example", "another-vault", "another-account")
+                .unwrap();
+            assert_eq!(ws.pending_count().unwrap(), 2);
+            assert_eq!(
+                ws.db
+                    .query_row(
+                        "SELECT COUNT(*) FROM sync_jobs WHERE binding=?1",
+                        [&rebound.id],
+                        |r| r.get::<_, i64>(0)
+                    )
+                    .unwrap(),
+                2
+            );
+            assert_eq!(ws.sync_discover(&rebound.id).unwrap(), 0);
+            assert_eq!(ws.entry("large.md").unwrap().unwrap().file_id, identity);
+        }
+    }
+    #[test]
     fn initial_snapshot_cursor_waits_for_all_files_and_recovers_twenty_rounds() {
         for _ in 0..20 {
             for committed in [false, true] {

frontend/src-tauri/src/sync_state.rs

@@ -71,7 +71,7 @@ impl Workspace {
                 .query_row("SELECT EXISTS(SELECT 1 FROM sync_bindings)", [], |r| {
                     r.get(0)
                 })?;
-        let entries = self.scan()?;
+        let paths = self.sync_paths()?;
         let id = Uuid::new_v4().to_string();
         // Rebinding explicitly starts from the current snapshot, never an old account's queue.
         if had_binding {
@@ -80,15 +80,30 @@ impl Workspace {
                 [],
             )?;
         }
-        for entry in entries.into_iter().filter(|e| !e.is_folder && !e.deleted) {
+        for path in paths {
-            let queued: bool = self.db.query_row(
+            let entry = self.entry(&path)?;
+            let queued = if let Some(entry) = &entry {
+                self.db.query_row(
                     "SELECT EXISTS(SELECT 1 FROM outbox WHERE file_id=?1 AND state='pending')",
                     [&entry.file_id],
-                |r| r.get(0),
+                    |r| r.get::<_, bool>(0),
-            )?;
+                )?
+            } else {
+                false
+            };
             if !queued {
-                let content = fs::read(self.resolve(&entry.path)?)?;
+                let source = self.resolve(&path)?;
-                self.write(&entry.path, &entry.hash, &content, "local")?;
+                let (digest, size) = crate::payloads::sync_file_info(&source, &path)?;
+                let operation = Uuid::new_v4().to_string();
+                self.store_payload_file(&operation, &source, &digest)?;
+                self.write_spooled_with_identity(
+                    &path,
+                    &digest,
+                    (&digest, size),
+                    "local",
+                    &operation,
+                    None,
+                )?;
             }
         }
         self.db.execute(

frontend/src-tauri/src/workspace.rs

@@ -298,8 +298,7 @@ impl Workspace {
                 });
                 continue;
             }
-            let content = fs::read(self.resolve(&path)?)?;
+            let digest = crate::payloads::hash_file(&self.resolve(&path)?)?;
-            let digest = hash(&content);
             let previous = self.entry(&path)?;
             if previous
                 .as_ref()