diff --git a/frontend/src-tauri/src/payloads.rs b/frontend/src-tauri/src/payloads.rs index c6cc029..84bf036 100644 --- a/frontend/src-tauri/src/payloads.rs +++ b/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 { + 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 { } 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) diff --git a/frontend/src-tauri/src/sync_discovery.rs b/frontend/src-tauri/src/sync_discovery.rs index 82ac11d..b4575b3 100644 --- a/frontend/src-tauri/src/sync_discovery.rs +++ b/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 { - 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 (digest, _) = crate::payloads::sync_file_info(&target, &path)?; 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])?; diff --git a/frontend/src-tauri/src/sync_initial.rs b/frontend/src-tauri/src/sync_initial.rs index 4e61eb2..6a5a85e 100644 --- a/frontend/src-tauri/src/sync_initial.rs +++ b/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 { - return Err(HostError::new("FILE_TOO_LARGE")); - } - let bytes = fs::read(source)?; - if crate::records::is_record(&path) { - crate::records::validate(&path, &bytes)?; - } + let (hash, size) = crate::payloads::sync_file_info(&source, &path)?; Ok(Local { path, - hash: hash(&bytes), - size: bytes.len(), + hash, + 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)?)?; - if hash(&bytes) != item.hash { - return Err(HostError::new("SYNC_PREVIEW_CHANGED")); - } - self.store_payload(&operation, &bytes)?; + self.store_payload_file(&operation, &self.resolve(&item.path)?, &item.hash) + .map_err(|error| { + if error.code == "REVISION_CONFLICT" { + HostError::new("SYNC_PREVIEW_CHANGED") + } else { + error + } + })?; 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] { diff --git a/frontend/src-tauri/src/sync_state.rs b/frontend/src-tauri/src/sync_state.rs index 685a050..8637564 100644 --- a/frontend/src-tauri/src/sync_state.rs +++ b/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) { - let queued: bool = self.db.query_row( - "SELECT EXISTS(SELECT 1 FROM outbox WHERE file_id=?1 AND state='pending')", - [&entry.file_id], - |r| r.get(0), - )?; + for path in paths { + 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::<_, bool>(0), + )? + } else { + false + }; if !queued { - let content = fs::read(self.resolve(&entry.path)?)?; - self.write(&entry.path, &entry.hash, &content, "local")?; + let source = self.resolve(&path)?; + 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( diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index 13971ab..16af59c 100644 --- a/frontend/src-tauri/src/workspace.rs +++ b/frontend/src-tauri/src/workspace.rs @@ -298,8 +298,7 @@ impl Workspace { }); continue; } - let content = fs::read(self.resolve(&path)?)?; - let digest = hash(&content); + let digest = crate::payloads::hash_file(&self.resolve(&path)?)?; let previous = self.entry(&path)?; if previous .as_ref()