diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c5b54f7..be30d78 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -45,7 +45,7 @@ jobs: run: cargo fmt --all -- --check - name: Lints - run: cargo clippy --workspace --all-targets -- -D warnings + run: cargo clippy --workspace --all-targets --locked -- -D warnings - name: Unit tests run: cargo test --workspace --locked @@ -66,8 +66,9 @@ jobs: # The seeded simulation runs (docs/TESTING.md): real FUSE # daemons against an in-memory Drive, so they need /dev/fuse but no account. + # They run on pull requests too: most data-loss fixes are only exercised + # here, and a change that breaks one should not reach main first. simulation: - if: github.event_name != 'pull_request' runs-on: ubuntu-24.04 timeout-minutes: 30 diff --git a/Cargo.lock b/Cargo.lock index c571294..0bac2c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3474,7 +3474,7 @@ checksum = "2ee67f1008b1ba2321834326597b8e186293b049a023cdef258527550b9935b4" [[package]] name = "pdfs-cli" -version = "3.4.1" +version = "3.5.0" dependencies = [ "anyhow", "clap", @@ -3490,13 +3490,14 @@ dependencies = [ [[package]] name = "pdfs-core" -version = "3.4.1" +version = "3.5.0" dependencies = [ "async-trait", "base64 0.23.1", "dbus-secret-service-keyring-store", "directories", "ignore", + "kamadak-exif", "keyring-core", "libc", "parking_lot", @@ -3516,7 +3517,7 @@ dependencies = [ [[package]] name = "pdfs-fuse" -version = "3.4.1" +version = "3.5.0" dependencies = [ "async-trait", "base64 0.23.1", @@ -3538,7 +3539,7 @@ dependencies = [ [[package]] name = "pdfs-gui" -version = "3.4.1" +version = "3.5.0" dependencies = [ "anyhow", "async-channel", diff --git a/Cargo.toml b/Cargo.toml index ce6da98..82ae7b6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"] [workspace.package] -version = "3.4.1" +version = "3.5.0" edition = "2024" license = "MIT" authors = ["Nils Pukropp "] @@ -61,3 +61,15 @@ async-trait = "0.1" futures = "0.3" fuser = "0.18" libc = "0.2" + +[profile.release] +# No package ships a separate debug package (the PKGBUILD sets `!debug`, the +# RPM spec sets `debug_package` to nil), and the unstripped binaries weighed +# 43 MB (`pdfs`), 24 MB (`pdfs-app`) and 12 MB (`pdfs-tray`). +strip = "symbols" +# Off on purpose: LTO has broken GTK links for this workspace, which is why +# the packages turn it off as well. +lto = false +# `Workers` and the thumbnail build catch panics to keep the daemon alive, so +# unwinding must stay on. +panic = "unwind" diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index 3b60cd6..d358c42 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -630,6 +630,16 @@ enum DeviceCmd { #[arg(long, conflicts_with = "uid")] clear: bool, }, + /// Sync the same folders as another computer: this machine continues that + /// computer's device alongside it, and its folders are offered for syncing + /// here. Both computers keep them in step. + Join { + /// Device uid (from `devices list`). + uid: String, + /// Skip the confirmation and accept every proposed local path. + #[arg(long)] + yes: bool, + }, /// Move another computer's folders into this computer's backup and sync /// them here. The other computer stays registered, without those folders. Migrate { @@ -831,6 +841,20 @@ fn main() -> Result<()> { let cli = Cli::parse(); JSON_OUTPUT.store(cli.json, std::sync::atomic::Ordering::Relaxed); + // The commands that hold the password, the key passphrases or the session + // tokens in memory: keep them out of core dumps and away from same-user + // ptrace. Only the daemon did, though `mount` holds the same keys for as + // long as it runs. + if matches!( + cli.command, + Command::Login { .. } + | Command::Unlock { .. } + | Command::Logout + | Command::Mount { .. } + | Command::Daemon { .. } + ) { + pdfs_core::kernelkey::forbid_dumps(); + } match cli.command { Command::Login { username, remember } => cmd_login(username, remember), Command::Unlock { remember } => cmd_unlock(remember), @@ -978,6 +1002,26 @@ fn cmd_devices(action: DeviceCmd) -> Result<()> { let uid = if clear { None } else { uid }; ok_or_bail(control_request(CtlRequest::AdoptDevice { uid })?)? } + DeviceCmd::Join { uid, yes } => { + if !yes { + println!( + "This computer syncs into that device alongside it. Its folders are kept\n\ + in step on both computers: an edit or a deletion on one reaches the other.\n\ + New synced folders here go under that device too. Folders already synced\n\ + here stay where they are." + ); + confirm("Sync with it?")?; + } + let response = control_request(CtlRequest::AdoptDevice { uid: Some(uid) })?; + // A script gets the one reply and runs `sync restore` itself. + if emit_json(&response)? { + return Ok(()); + } + ok_or_bail(response)?; + // The device is this machine's now, so the plain restore offers its + // folders. + cmd_sync_restore(yes, None)? + } DeviceCmd::Migrate { uid, yes } => { if !yes { println!( @@ -1977,8 +2021,14 @@ fn key_state_name() -> &'static str { } fn cmd_logout() -> Result<()> { - auth::logout()?; - println!("Stored session cleared."); + if auth::logout()? { + println!("Signed out: the session is ended on Proton and cleared here."); + } else { + println!( + "Stored session cleared. Proton could not be reached to end the session; \ + it stays valid until it expires or is signed out in your account's security settings." + ); + } Ok(()) } @@ -2221,9 +2271,6 @@ where /// an external unmount triggers a remount; errors back off and retry. This is /// the entry point for the systemd user service. fn cmd_daemon(mountpoint: Option) -> Result<()> { - // The daemon holds the account's key passphrases in memory; keep them out - // of core dumps and away from same-user ptrace. - pdfs_core::kernelkey::forbid_dumps(); loop { // Wait until a session is stored. The GUI enables this service on login, // but the service may also start at boot before the user has logged in. @@ -2592,7 +2639,10 @@ fn cmd_import_google_photos(archives: Vec, dry_run: bool, wait: bool) - } => continue, CtlResponse::ImportStatus { summary, .. } => { print_import_summary(summary.as_ref()); - return Ok(()); + return finished_run( + "the import", + summary.map(|s| (s.failed, s.cancelled, s.error)), + ); } CtlResponse::Error { message, kind } => bail!("{}", cli_error(kind, &message)), other => bail!("unexpected response: {other:?}"), @@ -2635,7 +2685,10 @@ fn cmd_redate_photos( CtlResponse::RedateStatus { running: true, .. } => continue, CtlResponse::RedateStatus { summary, .. } => { print_redate_summary(summary.as_ref()); - return Ok(()); + return finished_run( + "the re-date", + summary.map(|s| (s.failed, s.cancelled, s.error)), + ); } CtlResponse::Error { message, kind } => bail!("{}", cli_error(kind, &message)), other => bail!("unexpected response: {other:?}"), @@ -2643,6 +2696,22 @@ fn cmd_redate_photos( } } +/// How a run `--wait` followed ended, as the exit status: an error when it +/// failed, was cancelled, or left photos failed, so a script can tell. +/// `outcome` is `(failed, cancelled, error)`; `None` when the daemon has no +/// result, which happens when it restarted during the run. +fn finished_run(what: &str, outcome: Option<(usize, bool, Option)>) -> Result<()> { + match outcome { + None => bail!("{what} ended without a result; the service may have restarted"), + Some((_, _, Some(error))) => bail!("{what} failed: {error}"), + Some((_, true, None)) => bail!("{what} was cancelled"), + Some((failed, false, None)) if failed > 0 => { + bail!("{what} finished, but {failed} photo(s) failed") + } + Some(_) => Ok(()), + } +} + /// A `YYYY-MM-DD` day as epoch seconds at UTC midnight. Reuses the Takeout /// scanner's date reader, which already accepts exactly this spelling. fn parse_day(day: &str) -> Result { @@ -2678,6 +2747,10 @@ fn print_redate_summary(summary: Option<&pdfs_core::control::RedateSummary>) { println!("No re-date has run yet."); return; }; + if let Some(error) = &summary.error { + println!("Failed : {error}"); + return; + } println!("Photos examined : {}", summary.examined); println!("Mis-dated : {}", summary.candidates); if summary.dry_run { @@ -2737,6 +2810,10 @@ fn print_import_summary(summary: Option<&pdfs_core::control::ImportSummary>) { println!("No import has run yet."); return; }; + if let Some(error) = &summary.error { + println!("Failed : {error}"); + return; + } println!("Photos in export : {}", summary.found); println!("Uploaded : {}", summary.uploaded); println!("Already on Proton: {}", summary.duplicates); @@ -3751,6 +3828,28 @@ mod login_wait_tests { assert_eq!(login_poll_delay(u32::MAX), Duration::from_secs(15)); } + /// The file-manager script passes each selected path after `--`, so a + /// file named like an option is still the path. + #[test] + fn a_path_after_a_double_dash_is_a_path_even_when_it_looks_like_a_flag() { + let cli = Cli::try_parse_from(["pdfs", "pin", "--", "--json"]).unwrap(); + assert!(!cli.json); + match cli.command { + Command::Pin { path } => assert_eq!(path, PathBuf::from("--json")), + _ => panic!("not a pin"), + } + } + + #[test] + fn a_waited_for_run_that_did_not_fully_succeed_exits_with_an_error() { + assert!(finished_run("the import", Some((0, false, None))).is_ok()); + assert!(finished_run("the import", Some((3, false, None))).is_err()); + assert!(finished_run("the import", Some((0, true, None))).is_err()); + let failed = finished_run("the import", Some((0, false, Some("not a zip".into())))); + assert!(failed.unwrap_err().to_string().contains("not a zip")); + assert!(finished_run("the import", None).is_err()); + } + #[test] fn waiting_for_a_session_stops_hammering_the_keyring() { let mut checks_left = 20; diff --git a/crates/pdfs-core/Cargo.toml b/crates/pdfs-core/Cargo.toml index 0804486..82acd1a 100644 --- a/crates/pdfs-core/Cargo.toml +++ b/crates/pdfs-core/Cargo.toml @@ -30,3 +30,4 @@ async-trait.workspace = true libc.workspace = true zeroize.workspace = true base64.workspace = true +exif = { version = "0.6", package = "kamadak-exif" } diff --git a/crates/pdfs-core/src/auth.rs b/crates/pdfs-core/src/auth.rs index 796db3c..5acb3a0 100644 --- a/crates/pdfs-core/src/auth.rs +++ b/crates/pdfs-core/src/auth.rs @@ -516,8 +516,59 @@ pub fn lock(forget: bool) -> Result<()> { Ok(()) } -/// Forget the persisted session (best-effort; absent entry is not an error). -pub fn logout() -> Result<()> { +/// How long [`logout`] waits for Proton to end the session before it signs +/// out locally without. +const END_SESSION_TIMEOUT: Duration = Duration::from_secs(10); + +/// End the stored session on Proton's side (`DELETE auth/v4`). Returns false +/// when there is none or Proton could not be reached in time. +/// +/// Runs on a thread of its own, so it works the same whether the caller is +/// inside a tokio runtime or not. +fn end_server_session() -> bool { + let Ok(stored) = load() else { + return false; + }; + let ended = std::thread::scope(|scope| { + scope + .spawn(|| -> std::result::Result<(), String> { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| e.to_string())?; + rt.block_on(async { + let session = resume_stored(&stored).map_err(|e| e.to_string())?; + match tokio::time::timeout(END_SESSION_TIMEOUT, session.end()).await { + Ok(result) => result.map_err(|e| e.to_string()), + Err(_) => Err(format!( + "no answer within {}s", + END_SESSION_TIMEOUT.as_secs() + )), + } + }) + }) + .join() + .unwrap_or_else(|_| Err("the sign-out thread panicked".into())) + }); + match ended { + Ok(()) => true, + Err(error) => { + tracing::warn!(%error, "ending the session on Proton failed; signing out locally"); + false + } + } +} + +/// End the session on Proton, then forget it here. Returns whether Proton +/// ended it: offline, the session stays valid on Proton's side until it +/// expires or is signed out from the account's security settings. +/// +/// Before this, signing out only deleted the keyring entry. The tokens stayed +/// valid on Proton, so a copy of them taken from this machine still worked. +/// +/// The local half is best effort; an absent entry is not an error. +pub fn logout() -> Result { + let ended = end_server_session(); // The persisted entity cache describes the account being logged out of. It // is encrypted under a key derived from that account's passphrases, but // leaving it behind would also mean the next account inherits a store it can @@ -539,7 +590,7 @@ pub fn logout() -> Result<()> { } let _ = kernelkey::UNSAVED_TOKENS.clear(); match keyring_entry()?.delete_credential() { - Ok(()) | Err(keyring_core::Error::NoEntry) => Ok(()), + Ok(()) | Err(keyring_core::Error::NoEntry) => Ok(ended), Err(e) => Err(e.into()), } } diff --git a/crates/pdfs-core/src/config.rs b/crates/pdfs-core/src/config.rs index 74b7342..4206655 100644 --- a/crates/pdfs-core/src/config.rs +++ b/crates/pdfs-core/src/config.rs @@ -163,6 +163,10 @@ pub struct AppConfig { /// Defaulted for configs predating the field. #[serde(default)] pub files_view: FilesView, + /// The desktop app's main window size, remembered between runs. Defaulted + /// for configs predating the field. + #[serde(default)] + pub window: WindowState, /// The user hid the tray icon from its own menu. The tray exits at start /// while this is set, and the app does not spawn it; Preferences turns it /// back on. Defaulted for configs predating the field. @@ -211,6 +215,44 @@ impl Default for FilesView { } } +/// The desktop app's main window, as it was when it last closed. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] +#[serde(default)] +pub struct WindowState { + /// The unmaximised size, in logical pixels. + pub width: i32, + pub height: i32, + pub maximized: bool, +} + +impl WindowState { + /// Below this the saved size is taken for damage and the default is used. + pub const MIN_WIDTH: i32 = 480; + pub const MIN_HEIGHT: i32 = 400; + + /// This state, with a size too small to use replaced by the default. + pub fn usable(self) -> Self { + if self.width < Self::MIN_WIDTH || self.height < Self::MIN_HEIGHT { + Self { + maximized: self.maximized, + ..Self::default() + } + } else { + self + } + } +} + +impl Default for WindowState { + fn default() -> Self { + Self { + width: 980, + height: 680, + maximized: false, + } + } +} + /// The key a My Files listing is ordered by. #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)] #[serde(rename_all = "lowercase")] @@ -310,6 +352,22 @@ impl AppConfig { } } +/// Fail when the config file at `path` exists but does not parse, so that +/// [`AppDirs::save_config`] never writes over settings it could not read. +fn refuse_to_replace_unreadable(path: &Path) -> Result<()> { + match std::fs::read_to_string(path) { + Ok(content) => match serde_json::from_str::(&content) { + Ok(_) => Ok(()), + Err(e) => Err(Error::Other(format!( + "{} cannot be read ({e}); fix it or move it aside before changing settings", + path.display() + ))), + }, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(e.into()), + } +} + /// Resolved XDG locations for state, cache, and the default mountpoint. pub struct AppDirs { dirs: ProjectDirs, @@ -341,6 +399,7 @@ impl AppDirs { // Preserve the original bytes for diagnosis/recovery. In // particular, never replace a truncated config with defaults // and silently lose mount, device, or ignore settings. + // `save_config` refuses to write over it for the same reason. tracing::error!(path = %path.display(), error = %e, "config is unreadable; using in-memory defaults without overwriting it"); return AppConfig::default(); } @@ -348,9 +407,8 @@ impl AppDirs { } let default_config = AppConfig::default(); - if let Ok(content) = serde_json::to_string_pretty(&default_config) { - let _ = std::fs::create_dir_all(self.dirs.config_dir()); - let _ = std::fs::write(&path, content); + if let Err(e) = self.save_config(&default_config) { + tracing::warn!(path = %path.display(), error = %e, "could not write the default config"); } default_config } @@ -358,8 +416,15 @@ impl AppDirs { /// Persist `config` to [`config_path`](Self::config_path), creating the /// config dir if missing. The Settings page calls this after editing the /// cache budget or mountpoint; the daemon re-reads the file on its next mount. + /// + /// Refuses while the file on disk does not parse. Every caller loads, + /// changes one field and saves, and [`load_config`](Self::load_config) + /// gives defaults for a file it cannot read, so the first change after + /// the file was damaged wrote defaults over the mountpoint, the device and + /// the ignore rules the user had (`docs/BUGS.md` B223). pub fn save_config(&self, config: &AppConfig) -> Result<()> { let path = self.config_path(); + refuse_to_replace_unreadable(&path)?; let parent = path .parent() .ok_or_else(|| Error::Other("configuration path has no parent".into()))?; @@ -383,6 +448,46 @@ impl AppDirs { Ok(()) } + /// This machine's id (e.g. ~/.config/proton-drive-linux/machine-id). See + /// [`machine_id`](Self::machine_id). + pub fn machine_id_path(&self) -> PathBuf { + self.dirs.config_dir().join("machine-id") + } + + /// A random id for this machine, made on first use and kept in + /// [`machine_id_path`](Self::machine_id_path). + /// + /// It names this machine's profile in a device it may share with other + /// computers. Random rather than `/etc/machine-id`, so nothing that + /// identifies the host goes to Drive. In a file of its own rather than + /// `config.json`, because the app saves the config from a copy it loaded + /// earlier, and that would drop an id the daemon had just added. + pub fn machine_id(&self) -> Result { + let path = self.machine_id_path(); + match std::fs::read_to_string(&path) { + Ok(id) if is_machine_id(id.trim()) => return Ok(id.trim().to_string()), + Ok(_) => { + tracing::warn!(path = %path.display(), "machine id is unusable; making a new one"); + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(e.into()), + } + let mut bytes = [0u8; 16]; + use std::io::Read as _; + std::fs::File::open("/dev/urandom")?.read_exact(&mut bytes)?; + let id: String = bytes.iter().map(|b| format!("{b:02x}")).collect(); + let parent = path + .parent() + .ok_or_else(|| Error::Other("machine id path has no parent".into()))?; + std::fs::create_dir_all(parent)?; + let tmp = path.with_extension(format!("tmp-{}", std::process::id())); + std::fs::write(&tmp, format!("{id}\n"))?; + std::fs::File::open(&tmp)?.sync_all()?; + std::fs::rename(&tmp, &path)?; + std::fs::File::open(parent)?.sync_all()?; + Ok(id) + } + /// Persistent state: inode map / cache index DB lives here. pub fn state_dir(&self) -> PathBuf { // `state_dir` is Linux-only in `directories`; fall back to data dir. @@ -471,6 +576,12 @@ impl AppDirs { } } +/// Whether `id` is what [`AppDirs::machine_id`] writes: 32 lowercase hex +/// digits. +fn is_machine_id(id: &str) -> bool { + id.len() == 32 && id.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) +} + /// Make `dir` owner-only (`0700`). These directories contain decrypted bytes, /// plaintext names, and an authenticated control socket, so failure is fatal. fn restrict_dir(dir: &Path) -> Result<()> { @@ -526,6 +637,28 @@ mod tests { std::fs::metadata(p).unwrap().permissions().mode() & 0o777 } + /// A damaged config must not be replaced by defaults the next time a + /// setting changes (`docs/BUGS.md` B223). + #[test] + fn an_unreadable_config_is_never_written_over() { + let dir = std::env::temp_dir().join(format!( + "pdfs-cfg-guard-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("config.json"); + + assert!(refuse_to_replace_unreadable(&path).is_ok(), "no file yet"); + std::fs::write(&path, serde_json::to_vec(&AppConfig::default()).unwrap()).unwrap(); + assert!(refuse_to_replace_unreadable(&path).is_ok()); + std::fs::write(&path, br#"{"mountpoint": "/home/me/Pro"#).unwrap(); + assert!(refuse_to_replace_unreadable(&path).is_err(), "truncated"); + + let _ = std::fs::remove_dir_all(&dir); + } + /// `docs/BUGS.md` B6. A permissive mode here is not a cosmetic problem: these /// directories hold decrypted content, a plaintext index of every node name, /// and the control socket that commands the daemon's session. @@ -611,6 +744,11 @@ mod tests { descending: true, folders_first: false, }, + window: WindowState { + width: 1200, + height: 800, + maximized: true, + }, tray_hidden: true, language: Some("de".to_string()), online_map: true, @@ -630,6 +768,39 @@ mod tests { assert_eq!(decoded.upload_limit, Some(500_000)); assert_eq!(decoded.download_limit, None); assert_eq!(decoded.files_view, config.files_view); + assert_eq!(decoded.window, config.window); + } + + #[test] + fn a_config_without_a_window_opens_at_the_default_size() { + let decoded: AppConfig = + serde_json::from_str(r#"{"app_version":"a","user_agent":"b"}"#).unwrap(); + assert_eq!(decoded.window, WindowState::default()); + let partial: WindowState = serde_json::from_str(r#"{"maximized":true}"#).unwrap(); + assert_eq!((partial.width, partial.height), (980, 680)); + assert!(partial.maximized); + } + + #[test] + fn a_window_saved_too_small_opens_at_the_default_size() { + let tiny = WindowState { + width: 0, + height: 500, + maximized: true, + }; + assert_eq!( + tiny.usable(), + WindowState { + maximized: true, + ..WindowState::default() + } + ); + let fine = WindowState { + width: 640, + height: 480, + maximized: false, + }; + assert_eq!(fine.usable(), fine); } #[test] @@ -760,6 +931,15 @@ mod tests { assert_eq!(decoded.conflict_sweep, Some(SweepMode::Enforce)); } + #[test] + fn a_machine_id_is_32_lowercase_hex_digits() { + assert!(is_machine_id("0123456789abcdef0123456789abcdef")); + assert!(!is_machine_id("0123456789ABCDEF0123456789ABCDEF")); + assert!(!is_machine_id("0123456789abcdef")); + assert!(!is_machine_id("../../../../etc/passwd/0123456789")); + assert!(!is_machine_id("")); + } + #[test] fn a_config_predating_ignore_patterns_still_loads() { // Every installed config on disk looks like this; the field must not be diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index 1a01fa9..202240d 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -1591,6 +1591,10 @@ pub struct RedateSummary { /// name implies)` — what a dry run is read for. #[serde(default)] pub samples: Vec<(String, i64, i64)>, + /// Why the run as a whole failed, when it did. The counts are then + /// empty: the run never got to the photos. + #[serde(default)] + pub error: Option, } /// How a Google Photos Takeout import is going, or how the last one ended @@ -1620,6 +1624,10 @@ pub struct ImportSummary { pub bytes: u64, /// True when the run stopped early because it was cancelled. pub cancelled: bool, + /// Why the run as a whole failed, when it did, such as an archive that + /// is not a zip. The counts are then empty. + #[serde(default)] + pub error: Option, } /// One album in a [`Response::Albums`] listing. diff --git a/crates/pdfs-core/src/db/local.rs b/crates/pdfs-core/src/db/local.rs index f5f4ae2..f15eb22 100644 --- a/crates/pdfs-core/src/db/local.rs +++ b/crates/pdfs-core/src/db/local.rs @@ -111,9 +111,80 @@ impl Db { )?; let count: i64 = tx.query_row("SELECT COUNT(*) FROM local_files", [], |r| r.get(0))?; tx.commit()?; + drop(conn); + self.compact_search_indexes_if_due(finished_at)?; Ok(count) } + /// Close a scan that did not see everything: it stopped at the entry cap, + /// or a batch failed to write. Nothing is pruned, since a path it missed is + /// not a path that is gone; the next complete scan prunes. Returns the + /// number of indexed entries. + /// + /// Pruning after such a scan dropped every entry it had not reached, and + /// the next scan put them back. That churn is most of what grew + /// `local_fts` to 16 times its size (`docs/BUGS.md` B217). + pub fn local_finish_partial_scan(&self, finished_at: i64) -> Result { + let conn = self.conn.lock(); + conn.execute( + "INSERT INTO sync_state (key, value) VALUES ('local_indexed_at', ?1) + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + [finished_at.to_string()], + )?; + let count: i64 = conn.query_row("SELECT COUNT(*) FROM local_files", [], |r| r.get(0))?; + Ok(count) + } + + /// Compact both full-text indexes, at most once a day. + /// + /// FTS5 keeps a deleted entry's data until the segments holding it are + /// merged, and with the few large segments these indexes settle into that + /// is close to never. Every file the local scan sees disappear or appear + /// leaves some behind, and every node row the tree replaces does too: on + /// one machine `local_fts` held 317 MiB for 40,000 paths, against 20 MiB + /// rebuilt, and a search took four times as long (`docs/BUGS.md` B217). + /// + /// `local_fts` is rebuilt from `local_files`, which also drops anything + /// left over, and `nodes_fts`, which has no content table to rebuild from, + /// is merged into one segment. Each runs in its own transaction on the + /// write connection, about half a second on an index that size; once a + /// day keeps the leftovers small without paying that at every scan. + pub fn compact_search_indexes_if_due(&self, now: i64) -> Result<()> { + const EVERY_SECS: i64 = 24 * 60 * 60; + let last: i64 = self + .read() + .query_row( + "SELECT value FROM sync_state WHERE key = 'fts_compacted_at'", + [], + |row| row.get::<_, String>(0), + ) + .optional()? + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + if now - last < EVERY_SECS { + return Ok(()); + } + let started = std::time::Instant::now(); + { + let conn = self.conn.lock(); + conn.execute("INSERT INTO local_fts(local_fts) VALUES('rebuild')", [])?; + } + { + let conn = self.conn.lock(); + conn.execute("INSERT INTO nodes_fts(nodes_fts) VALUES('optimize')", [])?; + conn.execute( + "INSERT INTO sync_state (key, value) VALUES ('fts_compacted_at', ?1) + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + [now.to_string()], + )?; + } + tracing::info!( + elapsed_ms = started.elapsed().as_millis() as u64, + "compacted the search indexes" + ); + Ok(()) + } + /// When the last local scan completed (epoch seconds), or `None` if the index /// has never been built. The daemon uses this to decide whether a fresh mount /// needs an immediate rescan or can serve the existing index. diff --git a/crates/pdfs-core/src/db/maintenance.rs b/crates/pdfs-core/src/db/maintenance.rs index 09e2005..8f0f9dc 100644 --- a/crates/pdfs-core/src/db/maintenance.rs +++ b/crates/pdfs-core/src/db/maintenance.rs @@ -91,8 +91,13 @@ impl Db { /// Returns an empty vector when the database is sound; otherwise each entry /// is one problem SQLite found. Read-only and safe against a live daemon, /// though it reads every page, so it is not free on a large file. + /// + /// Runs on a read connection: it reads every page, which took 34 s on a + /// 567 MiB file, and on the write connection every write in the daemon + /// waited that long behind it (`docs/BUGS.md` B218). WAL lets it read a + /// consistent snapshot while the writer goes on. pub fn integrity_check(&self) -> Result> { - let conn = self.conn.lock(); + let conn = self.read(); let mut stmt = conn.prepare("PRAGMA integrity_check")?; let rows = stmt.query_map([], |r| r.get::<_, String>(0))?; let mut problems = Vec::new(); @@ -107,9 +112,11 @@ impl Db { Ok(problems) } - /// Size and row-count snapshot for `pdfs cache inspect`. + /// Size and row-count snapshot for `pdfs cache inspect`. On a read + /// connection, for the reason [`integrity_check`](Self::integrity_check) + /// gives: counting the rows of every table reads most of the file. pub fn stats(&self) -> Result { - let conn = self.conn.lock(); + let conn = self.read(); let page_count: i64 = conn.query_row("PRAGMA page_count", [], |r| r.get(0))?; let free_pages: i64 = conn.query_row("PRAGMA freelist_count", [], |r| r.get(0))?; let page_size: i64 = conn.query_row("PRAGMA page_size", [], |r| r.get(0))?; @@ -218,6 +225,40 @@ mod tests { assert!(db.integrity_check().unwrap().is_empty()); } + /// `pdfs diagnose` and `pdfs cache inspect` read the whole file. Behind the + /// write lock they froze every write in the daemon for as long as that + /// took (`docs/BUGS.md` B218). + #[test] + fn the_integrity_check_and_stats_do_not_wait_for_the_writer() { + let path = std::env::temp_dir().join(format!( + "pdfs-check-{}-{}.db", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let db = std::sync::Arc::new(Db::open(&path).unwrap()); + let writer = db.conn.lock(); + let (tx, rx) = std::sync::mpsc::channel(); + let reader = std::sync::Arc::clone(&db); + std::thread::spawn(move || { + let checked = reader.integrity_check().map(|p| p.is_empty()); + let counted = reader.stats().map(|s| s.page_count > 0); + let _ = tx.send((checked.ok(), counted.ok())); + }); + let answer = rx.recv_timeout(std::time::Duration::from_secs(10)); + drop(writer); + assert_eq!(answer.ok(), Some((Some(true), Some(true)))); + drop(db); + for suffix in ["", "-wal", "-shm"] { + let mut file = path.as_os_str().to_owned(); + file.push(suffix); + let _ = std::fs::remove_file(std::path::PathBuf::from(file)); + } + let _ = std::fs::remove_file(path.with_extension("db.lock")); + } + #[test] fn stats_report_the_schema_and_the_known_tables() { let db = Db::open_in_memory().unwrap(); diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs index 5a4d5b2..08b946b 100644 --- a/crates/pdfs-core/src/db/migrations.rs +++ b/crates/pdfs-core/src/db/migrations.rs @@ -11,7 +11,7 @@ use super::Db; use crate::Result; /// Current schema version. Bump on every forward migration added below. -pub(super) const SCHEMA_VERSION: i64 = 42; +pub(super) const SCHEMA_VERSION: i64 = 44; impl Db { pub(super) fn migrate(&self) -> Result<()> { @@ -440,6 +440,33 @@ impl Db { tx.execute_batch(MIGRATION_V42)?; } } + if current < 43 { + let has_ops: bool = tx.query_row( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'pending_op'", + [], + |row| row.get::<_, i64>(0), + )? > 0; + if has_ops { + tx.execute_batch(MIGRATION_V43)?; + } + } + if current < 44 { + // Same guards as V26-V42. + let has_column: bool = tx.query_row( + "SELECT COUNT(*) FROM pragma_table_info('sync_entry') \ + WHERE name = 'remote_revision_id'", + [], + |row| row.get::<_, i64>(0), + )? > 0; + let has_entries: bool = tx.query_row( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'sync_entry'", + [], + |row| row.get::<_, i64>(0), + )? > 0; + if has_entries && !has_column { + tx.execute_batch(MIGRATION_V44)?; + } + } tx.execute( "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -1330,3 +1357,28 @@ const MIGRATION_V42: &str = " ALTER TABLE pending_op ADD COLUMN sent_sha1 TEXT; ALTER TABLE pending_op ADD COLUMN sent_to TEXT; "; + +/// Schema v43: queued ops by kind and name. +/// +/// Before each create, rename and trash the drain asks whether a queued trash +/// or rename holds the name it wants, and a released name wakes the ops +/// waiting for it. Each of those scanned the whole queue, so a large offline +/// batch made draining it quadratic. The kind leads, so a lookup of every +/// queued rename uses the index too. +const MIGRATION_V43: &str = " +CREATE INDEX IF NOT EXISTS pending_op_kind_name ON pending_op(kind, name); +"; + +/// Schema v44: a synced file's baseline remembers the revision it was synced +/// at. +/// +/// A mirrored file edited here is uploaded as a new revision of whatever is +/// current on Drive, and another computer syncing the same folder may have +/// committed one since this pass looked. The revision id is what the upload +/// checks first, so that edit becomes a conflict copy instead of being +/// replaced (B249). `NULL` means "this row predates the column", and the check +/// falls back to the modification time and size; each row gains a real value +/// the next time its path is synced. +const MIGRATION_V44: &str = " +ALTER TABLE sync_entry ADD COLUMN remote_revision_id TEXT; +"; diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs index 611e0e3..c397024 100644 --- a/crates/pdfs-core/src/db/mod.rs +++ b/crates/pdfs-core/src/db/mod.rs @@ -179,6 +179,45 @@ impl Drop for Reader<'_> { } } +/// The write connection with `synchronous=FULL`, back to NORMAL on drop. +/// +/// NORMAL in WAL mode keeps a commit across a crash of the process but not +/// across a power loss: the WAL is synced only at checkpoints. That is fine for +/// metadata the next listing re-fetches, and not for the queue. A staged blob +/// is synced before its op is queued, and the op row is what makes it +/// reachable; losing that row loses a write `close()` already reported as +/// done. A FULL commit syncs the WAL, which covers every NORMAL commit before +/// it as well. +pub(super) struct Durable<'a>(parking_lot::MutexGuard<'a, Connection>); + +impl<'a> Durable<'a> { + pub(super) fn new(conn: parking_lot::MutexGuard<'a, Connection>) -> Result { + conn.pragma_update(None, "synchronous", "FULL")?; + Ok(Self(conn)) + } +} + +impl std::ops::Deref for Durable<'_> { + type Target = Connection; + + fn deref(&self) -> &Connection { + &self.0 + } +} + +impl std::ops::DerefMut for Durable<'_> { + fn deref_mut(&mut self) -> &mut Connection { + &mut self.0 + } +} + +impl Drop for Durable<'_> { + fn drop(&mut self) { + // A failure leaves the writer at FULL: slower, never less safe. + let _ = self.0.pragma_update(None, "synchronous", "NORMAL"); + } +} + /// Open a second connection to the same file for reads only. /// /// `SQLITE_OPEN_READ_ONLY` plus `query_only` is belt and braces: the flag stops @@ -216,18 +255,34 @@ impl Db { let conn = match Self::open_configured(path) { Ok(conn) => conn, Err(e) if is_corrupt(&e) => { - // Every byte in here is derived from the account or from the - // content cache directory, so the recovery is to start over — - // loudly, and keeping the damaged file for a post-mortem rather - // than deleting evidence. + // Most of what is in here is derived from the account and is + // fetched again. The queue is not: it holds writes accepted + // here that Drive does not have yet, and the mirror baselines + // and pins are this machine's alone. So the damaged file is + // moved aside, kept as evidence, and those tables are copied + // out of it into the new one as far as SQLite can still read + // them (`docs/BUGS.md` B216). tracing::error!( db = %path.display(), error = %e, - "cache database is corrupt — moving it aside and rebuilding \ - (metadata will be re-fetched; no user data is stored here)" + "cache database is corrupt; moving it aside, rebuilding, and \ + copying the upload queue, sync baselines and pins out of it" ); - quarantine_corrupt_db(path)?; - Self::open_configured(path)? + let damaged = quarantine_corrupt_db(path)?; + let conn = Self::open_configured(path)?; + for (table, result) in salvage(&conn, &damaged) { + match result { + Ok(rows) => { + tracing::warn!(table, rows, "salvaged from the damaged database") + } + Err(error) => tracing::error!( + table, + %error, + "could not salvage from the damaged database" + ), + } + } + conn } Err(e) => return Err(e), }; @@ -254,6 +309,12 @@ impl Db { Ok(db) } + /// The write connection for a commit that accepts a user's change into + /// the queue. See [`Durable`]. + pub(super) fn durable(&self) -> Result> { + Durable::new(self.conn.lock()) + } + /// A connection for a `SELECT`-only method. /// /// Reads see the last committed state, which is the same thing they saw @@ -281,7 +342,8 @@ impl Db { fn open_configured(path: &Path) -> Result { let conn = Connection::open(path)?; // WAL: readers never block the single writer. NORMAL sync is the - // standard durability/throughput tradeoff for WAL. + // standard durability/throughput tradeoff for WAL; the queue commits + // that need more raise it to FULL for themselves (`Durable`). conn.pragma_update(None, "journal_mode", "WAL")?; conn.pragma_update(None, "synchronous", "NORMAL")?; conn.pragma_update(None, "foreign_keys", "ON")?; @@ -428,12 +490,96 @@ fn is_corrupt(e: &Error) -> bool { ) } -/// Move a damaged database (and its WAL sidecars) out of the way so the next -/// open starts from empty. +/// The tables [`salvage`] copies out of a damaged database, in order, each +/// with the rows it keeps. /// -/// Renamed rather than deleted: this is a cache, but it is also the only -/// evidence of what went wrong, and it costs one file the user can remove. -fn quarantine_corrupt_db(path: &Path) -> Result<()> { +/// The queue goes with the node rows it names by local id: a create made here +/// has no other record of its node, and a real node keeps its local id when its +/// folder is listed again. Those rows come back unlisted, so their folders are +/// fetched again rather than trusted. Claims are cleared by [`Db::open`] as +/// after any other crash. +const SALVAGE: &[(&str, &str)] = &[ + ( + "nodes", + "lid IN (SELECT lid FROM old.pending_op WHERE lid IS NOT NULL + UNION SELECT parent_lid FROM old.pending_op WHERE parent_lid IS NOT NULL) + OR uid LIKE 'local~%'", + ), + ("pending_op", "1"), + ("sync_folder", "1"), + ("sync_entry", "1"), + ("pins", "1"), + ("own_sealed_rev", "1"), +]; + +/// Copy what only this machine has out of the damaged database at `damaged` +/// into the fresh one `conn` just created. Returns how each table went. +/// +/// Each table is copied in its own transaction, so a table SQLite cannot read +/// to the end is left out whole rather than half copied, and does not stop +/// the others. Only the columns both files have are copied, which covers a +/// file that broke in the middle of an upgrade. The damaged file is opened read +/// only and left as it was. +fn salvage(conn: &Connection, damaged: &Path) -> Vec<(&'static str, Result)> { + let uri = format!( + "file:{}?mode=ro", + damaged + .to_string_lossy() + .replace('%', "%25") + .replace('?', "%3f") + .replace('#', "%23") + ); + if let Err(e) = conn.execute("ATTACH DATABASE ?1 AS old", [&uri]) { + let e = Error::Db(e); + tracing::error!(error = %e, "could not open the damaged database to salvage it"); + return Vec::new(); + } + let results = SALVAGE + .iter() + .map(|&(table, rows)| (table, salvage_table(conn, table, rows))) + .collect(); + let _ = conn.execute("UPDATE nodes SET listed = 0", []); + if let Err(e) = conn.execute("DETACH DATABASE old", []) { + tracing::warn!(error = %e, "could not detach the damaged database"); + } + results +} + +fn salvage_table(conn: &Connection, table: &str, rows: &str) -> Result { + let columns = |schema: &str| -> Result> { + let mut stmt = conn.prepare(&format!("PRAGMA {schema}.table_info({table})"))?; + let names = stmt + .query_map([], |r| r.get::<_, String>(1))? + .collect::>>()?; + Ok(names) + }; + let new = columns("main")?; + let shared: Vec = columns("old")? + .into_iter() + .filter(|c| new.contains(c)) + .map(|c| format!("\"{c}\"")) + .collect(); + if shared.is_empty() { + return Ok(0); + } + let shared = shared.join(", "); + let tx = conn.unchecked_transaction()?; + let copied = tx.execute( + &format!( + "INSERT OR IGNORE INTO main.{table} ({shared}) SELECT {shared} FROM old.{table} WHERE {rows}" + ), + [], + )?; + tx.commit()?; + Ok(copied) +} + +/// Move a damaged database (and its WAL sidecars) out of the way so the next +/// open starts from empty. Returns where the database itself went. +/// +/// Renamed rather than deleted: it is the only evidence of what went wrong, +/// and [`salvage`] reads what only this machine has back out of it. +fn quarantine_corrupt_db(path: &Path) -> Result { let stamp = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs()) @@ -457,7 +603,9 @@ fn quarantine_corrupt_db(path: &Path) -> Result<()> { Err(e) => tracing::warn!(file = %from.display(), error = %e, "could not quarantine"), } } - Ok(()) + let mut kept = path.as_os_str().to_owned(); + kept.push(format!(".corrupt-{stamp}")); + Ok(PathBuf::from(kept)) } #[cfg(test)] diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index 9edabc3..a04be71 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -279,6 +279,45 @@ fn claimable() -> String { /// drain replaces it with the uid the server assigns. pub const LOCAL_VOLUME: &str = "local"; +/// One place a queued create was sent, with when it first was. +struct SentTarget { + parent: String, + name: String, + at_ms: Option, +} + +/// The `sent_to` column: `[[parent, name, at_ms], …]`, or `[[parent, name], …]` +/// as written before the time was kept. `None` when it does not parse. +fn parse_sent_to(json: Option<&str>) -> Option> { + #[derive(Deserialize)] + #[serde(untagged)] + enum Entry { + At(String, String, i64), + Bare(String, String), + } + let Some(json) = json else { + return Some(Vec::new()); + }; + let entries: Vec = serde_json::from_str(json).ok()?; + Some( + entries + .into_iter() + .map(|entry| match entry { + Entry::At(parent, name, at) => SentTarget { + parent, + name, + at_ms: Some(at), + }, + Entry::Bare(parent, name) => SentTarget { + parent, + name, + at_ms: None, + }, + }) + .collect(), + ) +} + /// The uid a node made on this machine goes by until Drive gives it one: its /// row's local id under [`LOCAL_VOLUME`]. /// @@ -541,7 +580,7 @@ impl Db { /// land in the order they were made (`docs/BUGS.md` B141). Should it fail, /// [`Db::record_op_failure`] or [`Db::release_op_claim`] drops it. pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option)> { - let mut conn = self.conn.lock(); + let mut conn = self.durable()?; let tx = conn.transaction()?; let superseded: Option = if op_supersedes(&op.kind) { let lid = node_lid(&tx, &op.uid)?; @@ -598,7 +637,7 @@ impl Db { name: &str, created_at: i64, ) -> Result<(i64, Vec)> { - let mut conn = self.conn.lock(); + let mut conn = self.durable()?; let tx = conn.transaction()?; let blobs = drop_doomed_ops(&tx, uid)?; tx.execute( @@ -628,7 +667,7 @@ impl Db { blob_path: &str, meta_json: &str, ) -> Result> { - let conn = self.conn.lock(); + let conn = self.durable()?; let lid = node_lid(&conn, uid)?; let existing: Option<(i64, Option)> = conn .query_row( @@ -691,7 +730,7 @@ impl Db { rewrite: impl FnOnce(&str) -> Option, ) -> Result { let (local, real) = (landing.local, landing.real); - let mut conn = self.conn.lock(); + let mut conn = self.durable()?; let tx = conn.transaction()?; let row: Option = tx .query_row( @@ -808,12 +847,19 @@ impl Db { Ok(()) } - /// Record that queued create `id` is being sent to `parent` under `name`. + /// Record that queued create `id` is being sent to `parent` under `name` + /// at `at_ms`. /// /// A rename while the op waits rewrites its target, but the file a lost /// answer left stays where it was sent. The retry and the trash look - /// there too (`docs/BUGS.md` B184). - pub fn note_create_target(&self, id: i64, parent: &str, name: &str) -> Result<()> { + /// there too (`docs/BUGS.md` B184). A target sent before keeps its first + /// time. + /// + /// A record that does not parse is left as it is rather than replaced + /// with this one target: the earlier ones are what the duplicate checks + /// look for. Without them a lost answer forks a conflict copy, which + /// loses nothing. + pub fn note_create_target(&self, id: i64, parent: &str, name: &str, at_ms: i64) -> Result<()> { let mut conn = self.conn.lock(); let tx = conn.transaction()?; let sent: Option = tx @@ -822,15 +868,26 @@ impl Db { }) .optional()? .flatten(); - let mut targets: Vec<(String, String)> = sent - .and_then(|json| serde_json::from_str(&json).ok()) - .unwrap_or_default(); - let target = (parent.to_string(), name.to_string()); - if !targets.contains(&target) { - targets.push(target); + let Some(mut targets) = parse_sent_to(sent.as_deref()) else { + tracing::warn!( + id, + "a queued create's send record does not parse; keeping it" + ); + return Ok(()); + }; + if !targets.iter().any(|t| t.parent == parent && t.name == name) { + targets.push(SentTarget { + parent: parent.to_string(), + name: name.to_string(), + at_ms: Some(at_ms), + }); + let json: Vec<(&str, &str, i64)> = targets + .iter() + .map(|t| (t.parent.as_str(), t.name.as_str(), t.at_ms.unwrap_or(0))) + .collect(); tx.execute( "UPDATE pending_op SET sent_to = ?2 WHERE id = ?1", - params![id, serde_json::to_string(&targets)?], + params![id, serde_json::to_string(&json)?], )?; } tx.commit()?; @@ -840,6 +897,24 @@ impl Db { /// The parent and name queued op `id` was sent under as a create, in the /// order it was ([`Db::note_create_target`]). pub fn create_targets(&self, id: i64) -> Result> { + Ok(self + .sent_targets(id)? + .into_iter() + .map(|t| (t.parent, t.name)) + .collect()) + } + + /// When queued create `id` was first sent, if that was recorded. Rows + /// written before the time was kept have none. + pub fn create_first_sent(&self, id: i64) -> Result> { + Ok(self + .sent_targets(id)? + .into_iter() + .filter_map(|t| t.at_ms.filter(|&at| at > 0)) + .min()) + } + + fn sent_targets(&self, id: i64) -> Result> { let sent: Option = self .conn .lock() @@ -848,9 +923,7 @@ impl Db { }) .optional()? .flatten(); - Ok(sent - .and_then(|json| serde_json::from_str(&json).ok()) - .unwrap_or_default()) + Ok(parse_sent_to(sent.as_deref()).unwrap_or_default()) } /// The SHA-1 of each blob queued op `id` sent as a create @@ -898,6 +971,27 @@ impl Db { Ok(n > 0) } + /// Replace the sidecar of op `id`, but only while it still holds the blob + /// at `blob_path`. + /// + /// The drain fills an incomplete blob in place and then records it as + /// whole, so a retry after a lost answer can recognise its own bytes on the + /// server. A write that closed meanwhile has pointed the row at a newer + /// blob with its own sidecar; that one must not be overwritten. Returns + /// false when the row is gone or holds another blob. + pub fn update_op_meta_for_blob( + &self, + id: i64, + blob_path: &str, + meta_json: &str, + ) -> Result { + let n = self.conn.lock().execute( + "UPDATE pending_op SET meta_json = ?3 WHERE id = ?1 AND blob_path = ?2", + params![id, blob_path, meta_json], + )?; + Ok(n > 0) + } + /// Point a queued create at a new parent and name, for a node renamed or /// moved before it ever reached the server. /// @@ -911,7 +1005,7 @@ impl Db { /// name it no longer wants, such as one held by a queued change. A parked /// create stays parked. pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result { - let conn = self.conn.lock(); + let conn = self.durable()?; let lid = node_lid(&conn, uid)?; let parent_lid = node_lid(&conn, parent_uid)?; let n = conn.execute( @@ -1389,12 +1483,26 @@ impl Db { } /// [`retry_op_now`](Self::retry_op_now) for every op that has failed at - /// least once. Returns how many rows changed. + /// least once, for the user's "Retry all". Returns how many rows changed. pub fn retry_failed_ops_now(&self, now: i64) -> Result { + self.retry_ops_now(now, "") + } + + /// [`retry_failed_ops_now`](Self::retry_failed_ops_now) for the ops Drive + /// did not refuse, for a reconnect. A link coming back changes nothing + /// about a full quota or a missing permission, so those keep their backoff + /// instead of all being sent again at once. + pub fn retry_unrefused_ops_now(&self, now: i64) -> Result { + self.retry_ops_now(now, "AND issue IS NULL") + } + + fn retry_ops_now(&self, now: i64, filter: &str) -> Result { let conn = self.conn.lock(); let changed = conn.execute( - "UPDATE pending_op SET next_attempt_at = ?1 - WHERE attempts > 0 AND next_attempt_at > ?1 AND next_attempt_at < ?2", + &format!( + "UPDATE pending_op SET next_attempt_at = ?1 + WHERE attempts > 0 AND next_attempt_at > ?1 AND next_attempt_at < ?2 {filter}" + ), params![now, PARK_UNTIL], )?; Ok(changed) diff --git a/crates/pdfs-core/src/db/sync.rs b/crates/pdfs-core/src/db/sync.rs index af93b07..960c00a 100644 --- a/crates/pdfs-core/src/db/sync.rs +++ b/crates/pdfs-core/src/db/sync.rs @@ -45,6 +45,12 @@ pub struct StoredSyncEntry { pub local_mtime_ns: Option, pub remote_rev: Option, pub remote_hash: Option, + /// The id of the remote revision this row was synced at. A mirror upload + /// checks it before it supersedes what Drive holds, because another client + /// may have committed a revision since (`docs/BUGS.md` B249). `None` on a + /// row written before schema 44, or by a pass that did not learn it; the + /// check then falls back to `remote_rev` and `remote_hash`. + pub remote_revision_id: Option, } impl Db { @@ -188,7 +194,7 @@ impl Db { let conn = self.read(); let mut stmt = conn.prepare( "SELECT rel_path, remote_uid, local_mtime, local_size, remote_rev, remote_hash, - local_mtime_ns + local_mtime_ns, remote_revision_id FROM sync_entry WHERE folder_id = ?1", )?; let rows = stmt.query_map(params![folder_id], |r| { @@ -200,6 +206,7 @@ impl Db { remote_rev: r.get(4)?, remote_hash: r.get(5)?, local_mtime_ns: r.get(6)?, + remote_revision_id: r.get(7)?, }) })?; let mut map = HashMap::new(); @@ -238,15 +245,16 @@ impl Db { conn.execute( "INSERT INTO sync_entry (folder_id, rel_path, remote_uid, local_mtime, local_size, remote_rev, - remote_hash, local_mtime_ns) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + remote_hash, local_mtime_ns, remote_revision_id) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) ON CONFLICT(folder_id, rel_path) DO UPDATE SET - remote_uid = excluded.remote_uid, - local_mtime = excluded.local_mtime, - local_size = excluded.local_size, - remote_rev = excluded.remote_rev, - remote_hash = excluded.remote_hash, - local_mtime_ns = excluded.local_mtime_ns", + remote_uid = excluded.remote_uid, + local_mtime = excluded.local_mtime, + local_size = excluded.local_size, + remote_rev = excluded.remote_rev, + remote_hash = excluded.remote_hash, + local_mtime_ns = excluded.local_mtime_ns, + remote_revision_id = excluded.remote_revision_id", params![ folder_id, e.rel_path, @@ -256,6 +264,7 @@ impl Db { e.remote_rev, e.remote_hash, e.local_mtime_ns, + e.remote_revision_id, ], )?; Ok(()) @@ -306,7 +315,7 @@ impl Db { { let mut select = tx.prepare( "SELECT rel_path, remote_uid, local_mtime, local_size, remote_rev, remote_hash, - local_mtime_ns + local_mtime_ns, remote_revision_id FROM sync_entry WHERE folder_id = ?1 AND (rel_path = ?2 OR substr(rel_path, 1, length(?2) + 1) = ?2 || '/')", @@ -321,6 +330,7 @@ impl Db { remote_rev: r.get(4)?, remote_hash: r.get(5)?, local_mtime_ns: r.get(6)?, + remote_revision_id: r.get(7)?, }) })? .collect::>>()?; @@ -332,8 +342,8 @@ impl Db { let mut insert = tx.prepare( "INSERT OR REPLACE INTO sync_entry (folder_id, rel_path, remote_uid, local_mtime, local_size, remote_rev, - remote_hash, local_mtime_ns) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + remote_hash, local_mtime_ns, remote_revision_id) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", )?; for row in rows { let rel = format!("{to_rel}{}", &row.rel_path[from_rel.len()..]); @@ -346,6 +356,7 @@ impl Db { row.remote_rev, row.remote_hash, row.local_mtime_ns, + row.remote_revision_id, ])?; } } diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 324378e..549eb8c 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -1283,6 +1283,90 @@ fn a_corrupt_database_is_moved_aside_and_rebuilt() { let _ = std::fs::remove_file(path.with_extension("db.lock")); } +/// A rebuilt database must not lose what only this machine has: writes Drive +/// has not received, the mirror baselines and the pins (`docs/BUGS.md` B216). +#[test] +fn a_rebuild_keeps_the_queue_baselines_and_pins_of_the_damaged_file() { + let id = now_test_id(); + let damaged = + std::env::temp_dir().join(format!("pdfs-salvage-old-{}-{id}.db", std::process::id())); + let fresh = + std::env::temp_dir().join(format!("pdfs-salvage-new-{}-{id}.db", std::process::id())); + + let db = Db::open(&damaged).unwrap(); + db.upsert_node(&folder("root", None, "My Files")).unwrap(); + db.upsert_node(&file("listed", "root", "listed.txt", 1)) + .unwrap(); + db.upsert_node(&file("queued", "root", "queued.txt", 2)) + .unwrap(); + db.set_listed(&uid("root"), true).unwrap(); + db.enqueue_op(&PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("queued").to_string(), + parent_uid: None, + name: None, + blob_path: Some("/staging/queued".into()), + meta_json: None, + created_at: 1, + attempts: 3, + last_error: None, + next_attempt_at: 0, + }) + .unwrap(); + let folder_id = db + .sync_folder_add("/home/me/Docs", "vol~docs", "share") + .unwrap(); + db.sync_entry_upsert( + folder_id, + &StoredSyncEntry { + rel_path: "a.txt".into(), + remote_uid: Some("vol~a".into()), + local_mtime: 5, + local_mtime_ns: None, + local_size: 6, + remote_rev: Some("7".into()), + remote_hash: Some("8".into()), + remote_revision_id: None, + }, + ) + .unwrap(); + db.pin_add("vol~pinned", "/pinned", true).unwrap(); + drop(db); + + let db = Db::open(&fresh).unwrap(); + let results = salvage(&db.conn.lock(), &damaged); + assert!(results.iter().all(|(_, r)| r.is_ok()), "{results:?}"); + + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0].uid, uid("queued").to_string()); + assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/queued")); + // The op's node came with it; the rest of the tree is fetched again. + assert!( + db.node_by_uid(&uid("queued").to_string()) + .unwrap() + .is_some() + ); + assert!( + db.node_by_uid(&uid("listed").to_string()) + .unwrap() + .is_none() + ); + assert_eq!(db.sync_entries(folder_id).unwrap().len(), 1); + assert!(db.is_pinned("vol~pinned").unwrap()); + drop(db); + + for path in [&damaged, &fresh] { + for suffix in ["", "-wal", "-shm"] { + let mut file = path.as_os_str().to_owned(); + file.push(suffix); + let _ = std::fs::remove_file(std::path::PathBuf::from(file)); + } + let _ = std::fs::remove_file(path.with_extension("db.lock")); + } +} + fn local(path: &str, name: &str, is_dir: bool) -> LocalEntry { LocalEntry { path: path.into(), @@ -1346,6 +1430,77 @@ fn rescanning_an_unchanged_file_keeps_exactly_one_index_entry() { } } +/// A scan that stopped early or lost a batch did not see every file, so it +/// must not prune the ones it missed (`docs/BUGS.md` B217). +#[test] +fn a_partial_local_scan_prunes_nothing() { + let db = Db::open_in_memory().unwrap(); + let g = db.local_begin_scan().unwrap(); + db.local_upsert_batch( + g, + &[ + local("/home/u/a.txt", "a.txt", false), + local("/home/u/b.txt", "b.txt", false), + ], + ) + .unwrap(); + assert_eq!(db.local_finish_scan(g, 1).unwrap(), 2); + + let g = db.local_begin_scan().unwrap(); + db.local_upsert_batch(g, &[local("/home/u/a.txt", "a.txt", false)]) + .unwrap(); + assert_eq!(db.local_finish_partial_scan(2).unwrap(), 2); + assert_eq!(db.search_local("b.txt", 10).unwrap().len(), 1); + assert_eq!(db.local_indexed_at().unwrap(), Some(2)); +} + +/// Deleted entries pile up in an FTS5 index until it is compacted. The daily +/// compaction must leave both indexes whole and searchable, and must not run +/// again the same day (`docs/BUGS.md` B217). +#[test] +fn compacting_the_search_indexes_keeps_them_whole_and_runs_once_a_day() { + let db = Db::open_in_memory().unwrap(); + db.upsert_node(&folder("root", None, "My Files")).unwrap(); + db.upsert_node(&file("f1", "root", "quarterly-report.pdf", 1)) + .unwrap(); + for round in 0..20 { + let g = db.local_begin_scan().unwrap(); + let name = format!("churn-{round}.txt"); + db.local_upsert_batch( + g, + &[ + local("/home/u/keep.txt", "keep.txt", false), + local(&format!("/home/u/{name}"), &name, false), + ], + ) + .unwrap(); + // Far in the past, so no scan compacts on its own. + db.local_finish_scan(g, 0).unwrap(); + } + + let day = 24 * 60 * 60; + db.compact_search_indexes_if_due(day).unwrap(); + { + let conn = db.conn.lock(); + for table in ["local_fts", "nodes_fts"] { + conn.execute( + &format!("INSERT INTO {table}({table}) VALUES('integrity-check')"), + [], + ) + .unwrap(); + } + } + assert_eq!(db.search_local("keep", 10).unwrap().len(), 1); + assert_eq!(db.search_local("churn-19", 10).unwrap().len(), 1); + assert!(db.search_local("churn-3.txt", 10).unwrap().is_empty()); + assert_eq!(db.search("quarterly", 10).unwrap().len(), 1); + let stamp = |db: &Db| db.state_str("fts_compacted_at").unwrap(); + assert_eq!(stamp(&db).as_deref(), Some(day.to_string().as_str())); + + db.compact_search_indexes_if_due(day + 60).unwrap(); + assert_eq!(stamp(&db).as_deref(), Some(day.to_string().as_str())); +} + /// Queries below the trigram minimum still match, via the `LIKE` fallback. #[test] fn local_search_short_query_like_fallback() { @@ -2198,6 +2353,100 @@ fn refuses_a_schema_newer_than_the_running_build() { let _ = std::fs::remove_file(path); } +/// Queueing a write syncs it to disk, and leaves the writer at NORMAL after. +/// +/// The op row is what makes a staged blob reachable once `close()` has +/// reported success, so its commit must survive a power loss. Everything else +/// the writer commits is metadata the next listing re-fetches, and pays only +/// NORMAL's cost. +#[test] +fn queueing_a_write_commits_at_full_sync_and_restores_normal() { + let path = std::env::temp_dir().join(format!("pdfs-db-sync-{}.db", std::process::id())); + let _ = std::fs::remove_file(&path); + let db = Db::open(&path).unwrap(); + let level = |db: &Db| -> i64 { + db.conn + .lock() + .query_row("PRAGMA synchronous", [], |r| r.get(0)) + .unwrap() + }; + // 1 = NORMAL, 2 = FULL. + assert_eq!(level(&db), 1); + { + let durable = db.durable().unwrap(); + let during: i64 = durable + .query_row("PRAGMA synchronous", [], |r| r.get(0)) + .unwrap(); + assert_eq!(during, 2); + } + assert_eq!(level(&db), 1); + + db.enqueue_op(&PendingOp { + id: 0, + kind: OP_TRASH.to_string(), + uid: uid("a").to_string(), + parent_uid: None, + name: Some("a".to_string()), + blob_path: None, + meta_json: None, + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }) + .unwrap(); + assert_eq!(level(&db), 1); + assert_eq!(db.pending_ops().unwrap().len(), 1); + drop(db); + for suffix in ["", "-wal", "-shm"] { + let _ = std::fs::remove_file(format!("{}{suffix}", path.display())); + } +} + +/// A gap-filled blob is recorded as whole only while the op still holds it. +/// +/// A write that closes during the drain points the row at a newer blob with +/// its own sidecar. Overwriting that sidecar with the older blob's would +/// describe bytes the row no longer holds. +#[test] +fn marking_a_blob_whole_leaves_a_newer_blob_alone() { + let db = Db::open_in_memory().unwrap(); + let op = PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("a").to_string(), + parent_uid: None, + name: None, + blob_path: Some("/staging/first".to_string()), + meta_json: Some(r#"{"complete":false}"#.to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }; + db.enqueue_op(&op).unwrap(); + let id = db.pending_ops().unwrap()[0].id; + + assert!( + db.update_op_meta_for_blob(id, "/staging/first", r#"{"complete":true}"#) + .unwrap() + ); + assert_eq!( + db.pending_ops().unwrap()[0].meta_json.as_deref(), + Some(r#"{"complete":true}"#) + ); + + assert!( + !db.update_op_meta_for_blob(id, "/staging/older", r#"{"stale":true}"#) + .unwrap(), + "the row holds another blob" + ); + assert_eq!( + db.pending_ops().unwrap()[0].meta_json.as_deref(), + Some(r#"{"complete":true}"#) + ); +} + /// A queued write whose baseline is restamped must keep everything else. /// /// The restamp happens after our *own* upload seals a new revision under a @@ -3927,15 +4176,15 @@ fn what_a_create_sent_outlives_a_newer_blob_its_trash_and_a_restart() { .unwrap(); assert!(db.create_sent(id).unwrap().is_empty()); assert!(db.create_targets(id).unwrap().is_empty()); - db.note_create_target(id, &parent, "c").unwrap(); - db.note_create_target(id, &parent, "c").unwrap(); + db.note_create_target(id, &parent, "c", 10).unwrap(); + db.note_create_target(id, &parent, "c", 20).unwrap(); db.note_create_sent(id, "aaaa").unwrap(); db.note_create_sent(id, "aaaa").unwrap(); db.record_op_failure(id, "no answer", 0).unwrap(); db.attach_blob_to_create("local~sent", "/staging/second", "{}") .unwrap() .expect("the create is still queued"); - db.note_create_target(id, &parent, "d").unwrap(); + db.note_create_target(id, &parent, "d", 30).unwrap(); db.note_create_sent(id, "bbbb").unwrap(); db.record_op_failure(id, "no answer", 0).unwrap(); db.delete_ops_for_uid("local~sent").unwrap(); @@ -3958,6 +4207,73 @@ fn what_a_create_sent_outlives_a_newer_blob_its_trash_and_a_restart() { let _ = std::fs::remove_file(path); } +/// A queued create knows when it first went out, which is where the window +/// for adopting a node its lost answer left starts. Records written before +/// the time was kept still list their targets, and one that does not parse is +/// not replaced by the next target. +#[test] +fn a_create_remembers_when_it_was_first_sent() { + let db = Db::open_in_memory().unwrap(); + let parent = uid("parent").to_string(); + let (id, _) = db + .enqueue_op(&PendingOp { + id: 0, + kind: OP_CREATE.to_string(), + uid: "local~first".to_string(), + parent_uid: Some(parent.clone()), + name: Some("a.txt".to_string()), + blob_path: None, + meta_json: None, + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }) + .unwrap(); + assert_eq!(db.create_first_sent(id).unwrap(), None, "never sent"); + db.note_create_target(id, &parent, "a.txt", 500).unwrap(); + db.note_create_target(id, &parent, "a.txt", 900).unwrap(); + db.note_create_target(id, &parent, "b.txt", 700).unwrap(); + assert_eq!(db.create_first_sent(id).unwrap(), Some(500)); + + let set = |json: &str| { + db.conn + .lock() + .execute( + "UPDATE pending_op SET sent_to = ?2 WHERE id = ?1", + rusqlite::params![id, json], + ) + .unwrap(); + }; + set(&format!(r#"[["{parent}","old.txt"]]"#)); + assert_eq!( + db.create_targets(id).unwrap(), + vec![(parent.clone(), "old.txt".to_string())] + ); + assert_eq!(db.create_first_sent(id).unwrap(), None); + db.note_create_target(id, &parent, "a.txt", 900).unwrap(); + assert_eq!( + db.create_targets(id).unwrap().len(), + 2, + "the old target stays" + ); + assert_eq!(db.create_first_sent(id).unwrap(), Some(900)); + + set("not json"); + db.note_create_target(id, &parent, "c.txt", 1000).unwrap(); + let raw: String = db + .conn + .lock() + .query_row("SELECT sent_to FROM pending_op WHERE id = ?1", [id], |r| { + r.get(0) + }) + .unwrap(); + assert_eq!( + raw, "not json", + "an unreadable record is kept, not replaced" + ); +} + /// A write after a create failed clears its attempts, so the next try goes /// out at once. The create was still sent: deleting the file then left no /// trash, and the file a lost answer made stayed on Drive under the name it @@ -3981,7 +4297,7 @@ fn a_create_rewritten_after_it_failed_still_leaves_a_trash_when_deleted() { next_attempt_at: 0, }) .unwrap(); - db.note_create_target(id, &parent, "a.txt").unwrap(); + db.note_create_target(id, &parent, "a.txt", 10).unwrap(); db.note_create_sent(id, "aaaa").unwrap(); db.record_op_failure(id, "no answer", 0).unwrap(); db.attach_blob_to_create("local~sent", "/staging/second", "{}") @@ -5028,6 +5344,7 @@ fn baseline_row(rel: &str) -> StoredSyncEntry { local_size: 7, remote_rev: Some("1700000000".into()), remote_hash: Some("7".into()), + remote_revision_id: None, } } @@ -5122,6 +5439,7 @@ fn mirror_uid_lookup_covers_the_root_and_synced_descendants_only() { local_size: 2, remote_rev: Some("3".into()), remote_hash: Some("2".into()), + remote_revision_id: None, local_mtime_ns: None, }, ) @@ -5139,6 +5457,7 @@ fn mirror_uid_lookup_covers_the_root_and_synced_descendants_only() { local_size: 2, remote_rev: Some("3".into()), remote_hash: Some("2".into()), + remote_revision_id: None, local_mtime_ns: None, }, ) @@ -6316,6 +6635,71 @@ fn migration_v27_adds_the_access_deferral_column_to_a_v26_queue() { let _ = std::fs::remove_file(format!("{}.lock", path.display())); } +/// V44 adds the remote revision a baseline row was synced at. A row carried +/// over from V43 has none, and the mirror then compares the modification time +/// and size as before, so upgrading changes nothing about what is in step +/// (`docs/BUGS.md` B249). +#[test] +fn migration_v44_adds_the_synced_revision_without_disturbing_a_v43_baseline() { + let path = std::env::temp_dir().join(format!( + "pdfs-db-v43-fixture-{}-{}.db", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let row = StoredSyncEntry { + rel_path: "notes.txt".into(), + remote_uid: Some("vol~notes".into()), + local_mtime: 1_700_000_000, + local_mtime_ns: Some(1_700_000_000_123_456_789), + local_size: 42, + remote_rev: Some("1700000000".into()), + remote_hash: Some("42".into()), + remote_revision_id: Some("rev-1".into()), + }; + { + let db = Db::open(&path).unwrap(); + let folder = db + .sync_folder_add("/home/me/Mirror", "vol~root", "share") + .unwrap(); + db.sync_entry_upsert(folder, &row).unwrap(); + } + { + // Put the file back in the state a released V43 database was in. + let conn = rusqlite::Connection::open(&path).unwrap(); + conn.execute_batch( + "ALTER TABLE sync_entry DROP COLUMN remote_revision_id; + UPDATE sync_state SET value = '43' WHERE key = 'schema_version';", + ) + .unwrap(); + } + + let db = Db::open(&path).unwrap(); + let entries = db.sync_entries(1).unwrap(); + let entry = entries.get("notes.txt").expect("the baseline row survives"); + assert_eq!( + entry, + &StoredSyncEntry { + remote_revision_id: None, + ..row.clone() + } + ); + + // And the column is live, including through a move of the row. + db.sync_entry_upsert(1, &row).unwrap(); + db.sync_entries_move(1, "notes.txt", 1, "moved.txt") + .unwrap(); + let entries = db.sync_entries(1).unwrap(); + assert_eq!( + entries["moved.txt"].remote_revision_id.as_deref(), + Some("rev-1") + ); + drop(db); + let _ = std::fs::remove_file(&path); +} + /// A mirror baseline carried over from V27 keeps its rows, and each one reads /// back with no nanosecond time. That `None` is the whole point: it makes the /// comparison fall back to whole seconds for rows written before the column @@ -6346,6 +6730,7 @@ fn migration_v28_adds_sub_second_local_times_without_disturbing_a_v27_baseline() local_size: 42, remote_rev: Some("1700000000".into()), remote_hash: Some("42".into()), + remote_revision_id: None, }, ) .unwrap(); @@ -7216,6 +7601,56 @@ fn a_refused_change_names_its_node_by_local_id() { ); } +/// The name checks the drain makes before each op look the name up in an +/// index. Scanning the queue instead made a large offline batch quadratic. +#[test] +fn queue_name_checks_use_an_index() { + let db = Db::open_in_memory().unwrap(); + let conn = db.conn.lock(); + let plan = |sql: &str| -> String { + let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap(); + stmt.query_map([], |r| r.get::<_, String>(3)) + .unwrap() + .map(|row| row.unwrap()) + .collect::>() + .join("; ") + }; + for sql in [ + "SELECT 1 FROM pending_op WHERE kind = 'trash' AND name = 'a'", + "SELECT 1 FROM pending_op WHERE name = 'a' AND kind IN ('create', 'mkdir', 'rename')", + "SELECT 1 FROM pending_op WHERE kind = 'rename'", + ] { + let plan = plan(sql); + assert!(plan.contains("pending_op_kind_name"), "{sql}: {plan}"); + } +} + +/// A reconnect moves up the ops that failed for want of a link. One Drive +/// refused, for quota or access, keeps its backoff: the link coming back +/// changes nothing about it. "Retry all" still moves both. +#[test] +fn a_reconnect_leaves_refused_ops_to_their_backoff() { + let db = Db::open_in_memory().unwrap(); + let offline = queued_op(&db, "vol~a", 2, 90_000); + let refused = queued_op(&db, "vol~b", 2, 90_000); + db.set_op_issue(refused, Some("quota")).unwrap(); + + assert_eq!(db.retry_unrefused_ops_now(5_000).unwrap(), 1); + let due = |id: i64| { + db.pending_ops() + .unwrap() + .into_iter() + .find(|op| op.id == id) + .unwrap() + .next_attempt_at + }; + assert_eq!(due(offline), 5_000); + assert_eq!(due(refused), 90_000); + + assert_eq!(db.retry_failed_ops_now(5_000).unwrap(), 1); + assert_eq!(due(refused), 5_000); +} + #[test] fn retry_all_touches_only_ops_that_have_failed() { let db = Db::open_in_memory().unwrap(); diff --git a/crates/pdfs-core/src/exiftime.rs b/crates/pdfs-core/src/exiftime.rs new file mode 100644 index 0000000..a7cb391 --- /dev/null +++ b/crates/pdfs-core/src/exiftime.rs @@ -0,0 +1,133 @@ +//! When a photo was taken, as its own EXIF tags say. +//! +//! The fallback for a photo nothing else dates: a Takeout photo without a +//! sidecar or a date in its name, and a photo uploaded from the app. Without +//! it the first was filed under the moment of the import and the second under +//! the file's modification time, which a copy resets. + +use std::io::{BufRead, Seek}; +use std::path::Path; + +use crate::takeout::days_from_civil; + +/// The capture time in the EXIF of the file at `path`, in epoch seconds. +pub fn capture_time_of(path: &Path) -> Option { + let file = std::fs::File::open(path).ok()?; + capture_time(&mut std::io::BufReader::new(file)) +} + +/// The capture time in the EXIF `reader` holds: `DateTimeOriginal`, then +/// `DateTimeDigitized`, each adjusted by its `OffsetTime…` tag when the camera +/// wrote one. Without an offset the time is read as UTC, as a date in a file +/// name is (`takeout::capture_time_from_name`). +pub fn capture_time(reader: &mut (impl BufRead + Seek)) -> Option { + let exif = exif::Reader::new().read_from_container(reader).ok()?; + let ascii = |tag: exif::Tag| -> Option { + let field = exif.get_field(tag, exif::In::PRIMARY)?; + match &field.value { + exif::Value::Ascii(parts) => parts.first().map(|part| { + String::from_utf8_lossy(part) + .trim_matches('\0') + .trim() + .to_string() + }), + _ => None, + } + }; + [ + (exif::Tag::DateTimeOriginal, exif::Tag::OffsetTimeOriginal), + (exif::Tag::DateTimeDigitized, exif::Tag::OffsetTimeDigitized), + ] + .into_iter() + .find_map(|(time, offset)| { + let local = parse_exif_time(&ascii(time)?)?; + let offset = ascii(offset).and_then(|o| parse_offset(&o)).unwrap_or(0); + Some(local - offset) + }) +} + +/// `YYYY:MM:DD HH:MM:SS` as epoch seconds, read as UTC. `None` for the +/// all-zero or blank value cameras write when they do not know. +fn parse_exif_time(text: &str) -> Option { + let bytes = text.as_bytes(); + if bytes.len() < 19 || bytes[4] != b':' || bytes[7] != b':' || bytes[13] != b':' { + return None; + } + let number = |range: std::ops::Range| text.get(range)?.parse::().ok(); + let (year, month, day) = (number(0..4)?, number(5..7)?, number(8..10)?); + let (hour, minute, second) = (number(11..13)?, number(14..16)?, number(17..19)?); + if !(1800..=9999).contains(&year) + || !(1..=12).contains(&month) + || !(1..=31).contains(&day) + || hour > 23 + || minute > 59 + || second > 60 + { + return None; + } + Some(days_from_civil(year, month, day) * 86_400 + hour * 3600 + minute * 60 + second) +} + +/// `+HH:MM` or `-HH:MM` as seconds east of UTC. +fn parse_offset(text: &str) -> Option { + let sign = match text.as_bytes().first()? { + b'+' => 1, + b'-' => -1, + _ => return None, + }; + let (hours, minutes) = text.get(1..)?.split_once(':')?; + let (hours, minutes) = (hours.parse::().ok()?, minutes.parse::().ok()?); + (hours <= 14 && minutes < 60).then_some(sign * (hours * 3600 + minutes * 60)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_exif_time_reads_as_utc_and_its_offset_moves_it() { + // 2023-06-01 12:00:00 UTC. + let noon = 1_685_620_800; + assert_eq!(parse_exif_time("2023:06:01 12:00:00"), Some(noon)); + assert_eq!(parse_offset("+02:00"), Some(7200)); + assert_eq!(parse_offset("-05:30"), Some(-19_800)); + assert_eq!( + parse_exif_time("2023:06:01 14:00:00").map(|t| t - parse_offset("+02:00").unwrap()), + Some(noon) + ); + } + + #[test] + fn an_unknown_exif_time_is_no_time() { + assert_eq!(parse_exif_time("0000:00:00 00:00:00"), None); + assert_eq!(parse_exif_time(" : : : : "), None); + assert_eq!(parse_exif_time(""), None); + assert_eq!(parse_offset("02:00"), None); + } + + #[test] + fn a_photo_is_dated_by_its_original_time_and_offset() { + let ascii = |tag, text: &str| exif::Field { + tag, + ifd_num: exif::In::PRIMARY, + value: exif::Value::Ascii(vec![text.as_bytes().to_vec()]), + }; + let taken = ascii(exif::Tag::DateTimeOriginal, "2023:06:01 14:00:00"); + let offset = ascii(exif::Tag::OffsetTimeOriginal, "+02:00"); + let digitized = ascii(exif::Tag::DateTimeDigitized, "2024:01:01 00:00:00"); + let mut writer = exif::experimental::Writer::new(); + writer.push_field(&taken); + writer.push_field(&offset); + writer.push_field(&digitized); + let mut tiff = std::io::Cursor::new(Vec::new()); + writer.write(&mut tiff, false).unwrap(); + tiff.set_position(0); + assert_eq!(capture_time(&mut tiff), Some(1_685_620_800)); + } + + #[test] + fn a_file_without_exif_has_no_capture_time() { + let mut png = std::io::Cursor::new(b"\x89PNG\r\n\x1a\n".to_vec()); + assert_eq!(capture_time(&mut png), None); + } +} diff --git a/crates/pdfs-core/src/lib.rs b/crates/pdfs-core/src/lib.rs index 5338996..73a84a7 100644 --- a/crates/pdfs-core/src/lib.rs +++ b/crates/pdfs-core/src/lib.rs @@ -10,6 +10,7 @@ pub mod config; pub mod control; pub mod db; pub mod error; +pub mod exiftime; pub mod kernelkey; pub mod localindex; pub mod menu; @@ -23,6 +24,7 @@ pub mod service; pub mod shell; pub mod similar; pub mod syncignore; +pub mod syncroot; pub mod takeout; pub mod tray; diff --git a/crates/pdfs-core/src/localindex.rs b/crates/pdfs-core/src/localindex.rs index 8335ea8..c77996f 100644 --- a/crates/pdfs-core/src/localindex.rs +++ b/crates/pdfs-core/src/localindex.rs @@ -42,7 +42,7 @@ const SKIP_PATH_SUFFIXES: &[&str] = &["go/pkg/mod"]; /// Hard cap on indexed entries, so a pathological home directory cannot grow the /// database without bound. Reaching it stops the walk early. -const MAX_ENTRIES: usize = 400_000; +pub const MAX_ENTRIES: usize = 400_000; /// Rows handed to the sink at a time. Large enough that the per-transaction cost /// amortises, small enough that the writer lock is released often (FUSE diff --git a/crates/pdfs-core/src/opener.rs b/crates/pdfs-core/src/opener.rs index f1e356b..6795f86 100644 --- a/crates/pdfs-core/src/opener.rs +++ b/crates/pdfs-core/src/opener.rs @@ -189,11 +189,21 @@ fn spawn(argv: &[String], path: &Path) { .stderr(Stdio::null()) .process_group(0) .spawn(); - if let Err(e) = spawned { - tracing::error!(command = ?argv, "open failed: {e}"); + match spawned { + Ok(child) => reap(child), + Err(e) => tracing::error!(command = ?argv, "open failed: {e}"), } } +/// Wait for `child` on a thread of its own. A child nobody waits for stays a +/// zombie for as long as its parent lives, which for the app and the tray is +/// the whole session. +pub fn reap(mut child: std::process::Child) { + std::thread::spawn(move || { + let _ = child.wait(); + }); +} + /// [`open`] with the policy from the user's `config.json`, read once per /// process. For callers that open the occasional path and have no reason to /// carry a policy around; long-lived windows should hold their own copy. @@ -472,6 +482,21 @@ fn glob_match(pattern: &str, name: &str) -> bool { mod tests { use super::*; + #[test] + fn a_reaped_child_does_not_stay_a_zombie() { + let child = std::process::Command::new("true").spawn().unwrap(); + let proc_entry = std::path::PathBuf::from(format!("/proc/{}", child.id())); + reap(child); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while proc_entry.exists() && std::time::Instant::now() < deadline { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + assert!( + !proc_entry.exists(), + "the child is still in the process table" + ); + } + fn rule(patterns: &[&str], command: &[&str], terminal: bool) -> OpenRule { OpenRule { patterns: patterns.iter().map(|p| (*p).to_string()).collect(), diff --git a/crates/pdfs-core/src/profile.rs b/crates/pdfs-core/src/profile.rs index a0c439e..91ee174 100644 --- a/crates/pdfs-core/src/profile.rs +++ b/crates/pdfs-core/src/profile.rs @@ -19,9 +19,31 @@ use serde::{Deserialize, Serialize}; use crate::mounts::MountMode; -/// File name of the profile. +/// File name of the profile clients before 3.5.0 wrote, one per device. +/// +/// Read, never written: two computers sharing a device each overwrote the +/// other's. See [`profile_file_name`]. pub const PROFILE_FILE_NAME: &str = "profile.json"; +/// File name of one machine's profile: `profile-.json`. +/// +/// One per machine, because several computers may sync into one device, and +/// each describes its own arrangement. A single shared file went to whichever +/// computer saved last. +pub fn profile_file_name(machine_id: &str) -> String { + format!("profile-{machine_id}.json") +} + +/// Whether `name` is a profile: [`PROFILE_FILE_NAME`] or one machine's +/// [`profile_file_name`]. +pub fn is_profile_file_name(name: &str) -> bool { + name == PROFILE_FILE_NAME + || name + .strip_prefix("profile-") + .and_then(|rest| rest.strip_suffix(".json")) + .is_some_and(|id| !id.is_empty() && id.bytes().all(|b| b.is_ascii_alphanumeric())) +} + /// Folder, directly under the device root, that holds the profile. /// /// The API refuses `Cannot create file at the root of a device` (HTTP 422), so @@ -74,6 +96,10 @@ pub struct Profile { /// Hostname of the machine that wrote it, so a restore prompt can say /// *"restore settings from laptop-xyz?"* rather than quoting a uid. pub hostname: String, + /// The writing machine's id, as in its [`profile_file_name`]. `None` in a + /// profile written before 3.5.0. + #[serde(default)] + pub machine_id: Option, /// When it was written, epoch seconds. pub saved_at: i64, #[serde(default)] @@ -92,6 +118,25 @@ pub struct Profile { pub mountpoint: Option, } +/// The saved arrangement of the folder with uid `remote_uid`, from the +/// profiles of every computer that syncs the device. +/// +/// The profile of the machine named `hostname` wins, since it describes this +/// computer; otherwise the most recently saved one that has the folder. +pub fn saved_folder<'a>( + profiles: &'a [Profile], + remote_uid: &str, + hostname: &str, +) -> Option<&'a ProfileFolder> { + let holding = |p: &&'a Profile| p.folders.iter().any(|f| f.remote_uid == remote_uid); + let profile = profiles + .iter() + .filter(holding) + .find(|p| p.hostname == hostname) + .or_else(|| profiles.iter().filter(holding).max_by_key(|p| p.saved_at))?; + profile.folders.iter().find(|f| f.remote_uid == remote_uid) +} + impl Profile { /// Parse a profile document, refusing one written by a newer client. pub fn parse(bytes: &[u8]) -> Result { @@ -122,6 +167,7 @@ mod tests { version: PROFILE_VERSION, device_uid: "dev-1".to_string(), hostname: "laptop-xyz".to_string(), + machine_id: Some("0123456789abcdef0123456789abcdef".to_string()), saved_at: 1_700_000_000, folders: vec![ProfileFolder { remote_uid: "vol~link".to_string(), @@ -219,6 +265,52 @@ mod tests { assert!(p.ignore_patterns.is_none()); } + #[test] + fn a_profile_name_is_the_legacy_one_or_one_per_machine() { + assert!(is_profile_file_name(PROFILE_FILE_NAME)); + let name = profile_file_name("0123456789abcdef0123456789abcdef"); + assert_eq!(name, "profile-0123456789abcdef0123456789abcdef.json"); + assert!(is_profile_file_name(&name)); + assert!(!is_profile_file_name("profile-.json")); + assert!(!is_profile_file_name("profile-a/b.json")); + assert!(!is_profile_file_name("profile (1).json")); + assert!(!is_profile_file_name("notes.json")); + } + + /// A profile written before 3.5.0 has no machine id and still parses. + #[test] + fn a_profile_without_a_machine_id_parses() { + let json = br#"{"version":1,"device_uid":"d","hostname":"h","saved_at":0}"#; + assert_eq!(Profile::parse(json).unwrap().machine_id, None); + } + + /// Several computers sync one device, each with its own profile. This + /// computer's own wins, then the newest that has the folder. + #[test] + fn the_saved_folder_comes_from_this_computer_then_the_newest_profile() { + let profile = |hostname: &str, saved_at: i64, path: &str| Profile { + hostname: hostname.to_string(), + saved_at, + folders: vec![ProfileFolder { + remote_uid: "vol~docs".to_string(), + local_path: path.to_string(), + mode: MountMode::Mirror, + }], + ..sample() + }; + let profiles = vec![ + profile("desktop", 10, "/home/a/Docs"), + profile("laptop", 5, "/home/b/Docs"), + profile("old", 1, "/home/c/Docs"), + ]; + let path = + |host: &str| saved_folder(&profiles, "vol~docs", host).map(|f| f.local_path.as_str()); + assert_eq!(path("laptop"), Some("/home/b/Docs")); + assert_eq!(path("new-machine"), Some("/home/a/Docs")); + assert_eq!(saved_folder(&profiles, "vol~other", "laptop"), None); + assert_eq!(saved_folder(&[], "vol~docs", "laptop"), None); + } + #[test] fn future_version_is_refused() { let mut p = sample(); diff --git a/crates/pdfs-core/src/shell.rs b/crates/pdfs-core/src/shell.rs index 22932a7..a5323a4 100644 --- a/crates/pdfs-core/src/shell.rs +++ b/crates/pdfs-core/src/shell.rs @@ -23,6 +23,11 @@ const SCRIPTS: [(&str, &str); 2] = [ /// `NAUTILUS_SCRIPT_SELECTED_FILE_PATHS`; the positional args carry the same /// selection when the script runs on a local (non-URI) folder, so we prefer the /// environment and fall back to `"$@"` for the Nemo/Caja variants. +/// +/// The `--` keeps a file named like an option, such as `--json`, a path. A +/// name with a newline in it still arrives split in two from Nautilus, whose +/// list is newline-separated; `pdfs` then fails on both halves, which touches +/// nothing. fn script_body(subcommand: &str) -> String { format!( "#!/bin/sh\n\ @@ -41,7 +46,7 @@ fn script_body(subcommand: &str) -> String { \n\ printf '%s' \"$paths\" | while IFS= read -r path; do\n\ \x20 [ -n \"$path\" ] || continue\n\ - \x20 pdfs {subcommand} \"$path\" || exit 1\n\ + \x20 pdfs {subcommand} -- \"$path\" || exit 1\n\ done\n" ) } @@ -95,3 +100,13 @@ fn write_scripts(dir: &PathBuf) -> io::Result<()> { tracing::debug!("installed file-manager scripts in {dir:?}"); Ok(()) } + +#[cfg(test)] +mod tests { + use super::script_body; + + #[test] + fn the_script_ends_options_before_each_path() { + assert!(script_body("pin").contains(r#"pdfs pin -- "$path""#)); + } +} diff --git a/crates/pdfs-core/src/syncroot.rs b/crates/pdfs-core/src/syncroot.rs new file mode 100644 index 0000000..2fffafa --- /dev/null +++ b/crates/pdfs-core/src/syncroot.rs @@ -0,0 +1,209 @@ +//! Which local folders may become a sync root. +//! +//! A mirror uploads everything under its root and deletes what Drive deleted, +//! so a root that holds more than the user meant is dangerous. `$HOME` takes +//! the decrypted content cache, `cache.db` and `~/.ssh` up to Drive, and walks +//! into the mount, which downloads the whole Drive only to upload it again. +//! Every place a root is added or restored goes through [`problem`]. + +use std::path::{Component, Path, PathBuf}; + +use crate::config::{AppConfig, AppDirs}; + +/// The paths a sync root must stay clear of. +#[derive(Debug, Clone, Default)] +pub struct Reserved { + /// The user's home folder. A root may sit inside it, not be it or above it. + pub home: Option, + /// The Proton Drive mount. + pub mountpoint: Option, + /// This app's own configuration, state and cache folders. + pub app_dirs: Vec, +} + +impl Reserved { + /// The reserved paths of this user and `config`, resolved like a root. + pub fn current(config: &AppConfig) -> Self { + let Ok(dirs) = AppDirs::new() else { + return Self::default(); + }; + let resolved = |path: PathBuf| resolve(&path).unwrap_or(path); + let mut app_dirs = vec![dirs.state_dir(), dirs.cache_dir()]; + if let Some(config_dir) = dirs.config_path().parent() { + app_dirs.push(config_dir.to_path_buf()); + } + Self { + home: dirs.home_dir().map(resolved), + mountpoint: Some(resolved(dirs.resolved_mountpoint(config))), + app_dirs: app_dirs.into_iter().map(resolved).collect(), + } + } +} + +/// `path` as a root is compared: absolute, with symlinks resolved as far as the +/// path exists. A folder a restore is about to create does not exist yet, so +/// the part past the deepest existing folder is kept as written. +/// +/// A `..` is refused rather than resolved: before a symlink it means something +/// else than after it, and a path recorded on another machine has no business +/// holding one. +pub fn resolve(path: &Path) -> Result { + if !path.is_absolute() { + return Err(format!("{} is not an absolute path", path.display())); + } + if path.components().any(|c| c == Component::ParentDir) { + return Err(format!("{} goes up with `..`", path.display())); + } + let mut existing = path.to_path_buf(); + let mut rest = Vec::new(); + loop { + match existing.canonicalize() { + Ok(canonical) => { + return Ok(rest + .into_iter() + .rev() + .fold(canonical, |path: PathBuf, part| path.join(part))); + } + Err(_) => { + let Some(name) = existing.file_name().map(|n| n.to_os_string()) else { + return Ok(path.to_path_buf()); + }; + rest.push(name); + if !existing.pop() { + return Ok(path.to_path_buf()); + } + } + } + } +} + +/// Why `root` cannot be synced, or `None` when it can. `root` is resolved +/// ([`resolve`]); `others` are the roots already synced. +pub fn problem(root: &Path, others: &[PathBuf], reserved: &Reserved) -> Option { + let shown = root.display(); + if root.parent().is_none() { + return Some(format!("{shown} is the root of the file system")); + } + if let Some(home) = &reserved.home { + if root == home { + return Some(format!( + "{shown} is your home folder; pick a folder inside it" + )); + } + if home.starts_with(root) { + return Some(format!("{shown} holds your home folder")); + } + } + if let Some(mountpoint) = &reserved.mountpoint + && overlaps(root, mountpoint) + { + return Some(format!( + "{shown} overlaps the Proton Drive folder {}", + mountpoint.display() + )); + } + if let Some(dir) = reserved.app_dirs.iter().find(|dir| overlaps(root, dir)) { + return Some(format!( + "{shown} overlaps this app's own data in {}", + dir.display() + )); + } + if others.iter().any(|other| other == root) { + return Some(format!("{shown} is already synced")); + } + if let Some(other) = others.iter().find(|other| overlaps(root, other)) { + return Some(format!( + "{shown} overlaps the synced folder {}", + other.display() + )); + } + None +} + +/// Whether one of `a` and `b` is the other or inside it. +fn overlaps(a: &Path, b: &Path) -> bool { + a.starts_with(b) || b.starts_with(a) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn reserved() -> Reserved { + Reserved { + home: Some("/home/u".into()), + mountpoint: Some("/home/u/ProtonDrive".into()), + app_dirs: vec![ + "/home/u/.local/state/proton-drive-linux".into(), + "/home/u/.cache/proton-drive-linux".into(), + "/home/u/.config/proton-drive-linux".into(), + ], + } + } + + fn refused(root: &str, others: &[&str]) -> bool { + let others: Vec = others.iter().map(PathBuf::from).collect(); + problem(Path::new(root), &others, &reserved()).is_some() + } + + #[test] + fn a_folder_inside_home_can_be_synced() { + assert!(!refused("/home/u/Documents", &[])); + assert!(!refused("/home/u/Pictures", &["/home/u/Documents"])); + assert!(!refused("/mnt/data/photos", &[])); + } + + #[test] + fn the_file_system_root_home_and_its_parents_are_refused() { + assert!(refused("/", &[])); + assert!(refused("/home/u", &[])); + assert!(refused("/home", &[])); + } + + #[test] + fn the_mount_and_the_apps_own_data_are_refused_either_way_round() { + assert!(refused("/home/u/ProtonDrive", &[])); + assert!(refused("/home/u/ProtonDrive/Documents", &[])); + assert!(refused("/home/u/.cache", &[])); + assert!(refused( + "/home/u/.local/state/proton-drive-linux/staging", + &[] + )); + assert!(refused("/home/u/.config", &[])); + } + + #[test] + fn a_root_inside_or_around_another_root_is_refused() { + assert!(refused("/home/u/Documents", &["/home/u/Documents"])); + assert!(refused("/home/u/Documents/Work", &["/home/u/Documents"])); + assert!(refused("/home/u/stuff", &["/home/u/stuff/Documents"])); + // A shared prefix of characters is not a shared folder. + assert!(!refused("/home/u/Documents2", &["/home/u/Documents"])); + } + + #[test] + fn a_path_that_goes_up_or_is_relative_does_not_resolve() { + assert!(resolve(Path::new("/home/u/../../etc")).is_err()); + assert!(resolve(Path::new("Documents")).is_err()); + } + + #[test] + fn a_folder_still_to_be_created_resolves_under_its_existing_parent() { + let base = std::env::temp_dir().join(format!("pdfs-syncroot-{}", std::process::id())); + std::fs::create_dir_all(&base).unwrap(); + let canonical = base.canonicalize().unwrap(); + assert_eq!( + resolve(&base.join("new/inner")).unwrap(), + canonical.join("new/inner") + ); + let link = base.join("link"); + let _ = std::fs::remove_file(&link); + std::os::unix::fs::symlink(&canonical, &link).unwrap(); + assert_eq!( + resolve(&link.join("later")).unwrap(), + canonical.join("later"), + "a symlink is followed to where it points" + ); + std::fs::remove_dir_all(&base).unwrap(); + } +} diff --git a/crates/pdfs-core/src/takeout.rs b/crates/pdfs-core/src/takeout.rs index 8df5db3..df9cf40 100644 --- a/crates/pdfs-core/src/takeout.rs +++ b/crates/pdfs-core/src/takeout.rs @@ -346,6 +346,11 @@ fn assemble(entries: Vec, sidecars: &HashMap>) -> Take .insert(file.to_string(), body); } + let indexes: HashMap<&str, SidecarIndex<'_>> = per_folder_sidecars + .iter() + .map(|(folder, in_folder)| (folder.as_str(), SidecarIndex::new(in_folder))) + .collect(); + let mut scan = TakeoutScan::default(); let mut seen_albums: HashMap = HashMap::new(); @@ -364,9 +369,13 @@ fn assemble(entries: Vec, sidecars: &HashMap>) -> Take continue; }; - let sidecar = per_folder_sidecars - .get(&folder) - .and_then(|in_folder| find_sidecar(&file_name, in_folder)) + let found = indexes + .get(folder.as_str()) + .and_then(|index| find_sidecar(&file_name, index)); + // An edited copy keeps its own name: the title is its original's. + let title_is_ours = found.is_some_and(|((rule, _), _)| rule != EDITED_COPY); + let sidecar = found + .map(|(_, body)| body) .and_then(|body| serde_json::from_slice::(body).ok()) .unwrap_or_default(); @@ -388,6 +397,7 @@ fn assemble(entries: Vec, sidecars: &HashMap>) -> Take let name = sidecar .title .as_deref() + .filter(|_| title_is_ours) .map(str::trim) .filter(|title| !title.is_empty() && same_extension(title, &file_name)) .unwrap_or(&file_name) @@ -598,7 +608,7 @@ pub fn capture_time_from_name(name: &str) -> Option { /// Days between 1970-01-01 and a proleptic-Gregorian date (Howard Hinnant's /// `days_from_civil`). Avoids pulling a date crate in for one conversion. -fn days_from_civil(year: i64, month: i64, day: i64) -> i64 { +pub(crate) fn days_from_civil(year: i64, month: i64, day: i64) -> i64 { let year = year - i64::from(month <= 2); let era = if year >= 0 { year } else { year - 399 } / 400; let year_of_era = year - era * 400; @@ -627,17 +637,106 @@ fn swap_duplicate_marker(file_name: &str) -> Option { Some(format!("{}{}{}", &file_name[..marker_start], tail, marker)) } -/// Find the sidecar belonging to `file_name` among the JSON files of its folder. +/// How long Google lets a sidecar's file name get, `.json` included. A longer +/// one is cut, into its suffix first and then into the photo name. +const SIDECAR_NAME_CAP: usize = 51; + +/// The rank [`sidecar_rank`] gives an edited copy that shares its original's +/// sidecar. Its date is the original's, but its name is its own. +const EDITED_COPY: u8 = 4; + +/// The sidecars of one folder by their `.json`-stripped stem, so a photo is +/// ranked only against those that can match it. +/// +/// Every rule in [`sidecar_rank`] needs the stem to start with a prefix of the +/// photo name that the name itself determines, and a sorted map finds the +/// stems with a given prefix as one range. Ranking every photo against every +/// sidecar instead took 27 s for a folder of 10,000 photos. +struct SidecarIndex<'a> { + by_stem: BTreeMap<&'a str, Vec<(&'a str, &'a Vec)>>, +} + +impl<'a> SidecarIndex<'a> { + fn new(in_folder: &'a BTreeMap>) -> Self { + let mut by_stem: BTreeMap<&'a str, Vec<(&'a str, &'a Vec)>> = BTreeMap::new(); + for (name, body) in in_folder { + by_stem + .entry(strip_json_suffix(name)) + .or_default() + .push((name.as_str(), *body)); + } + Self { by_stem } + } + + /// Add every sidecar whose stem starts with `prefix` to `out`. + fn starting_with(&self, prefix: &str, out: &mut BTreeMap<&'a str, &'a Vec>) { + let from = ( + std::ops::Bound::Included(prefix), + std::ops::Bound::Unbounded, + ); + for (stem, sidecars) in self.by_stem.range::(from) { + if !stem.starts_with(prefix) { + break; + } + out.extend(sidecars.iter().copied()); + } + } + + /// Add every sidecar whose stem is `head`, or starts with `head` and a `.`: + /// those that have `head` as a boundary ([`suffix_boundaries`]). + fn bounded_by(&self, head: &str, out: &mut BTreeMap<&'a str, &'a Vec>) { + if let Some(sidecars) = self.by_stem.get(head) { + out.extend(sidecars.iter().copied()); + } + self.starting_with(&format!("{head}."), out); + } + + /// Every sidecar [`sidecar_rank`] could match `file_name` to, by name. A + /// superset: each one is still ranked. + fn candidates(&self, file_name: &str) -> BTreeMap<&'a str, &'a Vec> { + let mut out = BTreeMap::new(); + // Rules 0 and 2: the stem starts with the name, or with its swapped form. + self.starting_with(file_name, &mut out); + if let Some(swapped) = swap_duplicate_marker(file_name) { + self.starting_with(&swapped, &mut out); + } + // Rule 1: the stem of `A(N).B` starts with `A.B(N)`, so with `A.`. + for (paren, _) in file_name.match_indices('(') { + self.starting_with(&format!("{}.", &file_name[..paren]), &mut out); + } + // Rule 3: a boundary of the stem is a prefix of the name. + for end in file_name + .char_indices() + .map(|(index, _)| index) + .skip(1) + .chain([file_name.len()]) + { + self.bounded_by(&file_name[..end], &mut out); + } + // Rule 4: `-.` has `.` as a boundary. + if let Some((_, extension)) = file_name.rsplit_once('.') { + for (dash, _) in file_name.match_indices('-') { + self.bounded_by(&format!("{}.{extension}", &file_name[..dash]), &mut out); + } + } + out + } +} + +/// Find the sidecar belonging to `file_name` among the JSON files of its folder, +/// with the rank it matched at. /// /// Tried in order of confidence: the exact name, the `(N)`-swapped name, an -/// unknown trailing suffix, then a prefix match for the case where truncation -/// cut into the photo name itself. The best-ranked candidate in the folder wins. +/// unknown trailing suffix, a prefix match for the case where truncation cut +/// into the photo name itself, and last an edited copy's original. The +/// best-ranked candidate in the folder wins; of equal ranks, the first by +/// sidecar name. fn find_sidecar<'a>( file_name: &str, - in_folder: &BTreeMap>, -) -> Option<&'a Vec> { + index: &SidecarIndex<'a>, +) -> Option<((u8, usize), &'a Vec)> { let mut best: Option<((u8, usize), &'a Vec)> = None; - for (sidecar_name, body) in in_folder { + for (sidecar_name, body) in index.candidates(file_name) { let Some(rank) = sidecar_rank(strip_json_suffix(sidecar_name), file_name) else { continue; }; @@ -645,7 +744,7 @@ fn find_sidecar<'a>( best = Some((rank, body)); } } - best.map(|(_, body)| body) + best } /// How well a sidecar's `.json`-stripped `stem` matches `file_name`, as a sort @@ -683,19 +782,44 @@ fn sidecar_rank(stem: &str, file_name: &str) -> Option<(u8, usize)> { } // 3 — truncation ate into the photo name itself, leaving only a prefix of // it. Every suffix boundary in the stem is tried, so this holds for a - // localized suffix too; the longest surviving prefix wins, because a folder - // with `IMG_1.jpg` and `IMG_12.jpg` must not let the shorter stem claim the - // longer photo. - let mut longest: Option = None; - for head in suffix_boundaries(stem) { - if !head.is_empty() - && file_name.starts_with(head) - && longest.is_none_or(|len| head.len() > len) - { - longest = Some(head.len()); + // localized suffix too; the longest surviving prefix wins. + // + // Only for a sidecar whose name is as long as Google lets one get: a name + // that short of the cap was not cut, so a prefix of it is the name of a + // different photo. Without this, `IMG_1.jpg`'s sidecar gave its name and + // date to an `IMG_12.jpg` that had none (`docs/BUGS.md` B225). + if stem.chars().count() + ".json".len() >= SIDECAR_NAME_CAP - 4 { + let mut longest: Option = None; + for head in suffix_boundaries(stem) { + if !head.is_empty() + && file_name.starts_with(head) + && longest.is_none_or(|len| head.len() > len) + { + longest = Some(head.len()); + } + } + if let Some(len) = longest { + return Some((3, usize::MAX - len)); } } - longest.map(|len| (3, usize::MAX - len)) + // 4 — an edited copy, `IMG_1-edited.jpg` (the word is localized), which + // Google exports without a sidecar of its own next to `IMG_1.jpg`'s. + suffix_boundaries(stem) + .filter_map(|head| head.rsplit_once('.')) + .any(|(original, extension)| is_edited_copy_of(file_name, original, extension)) + .then_some((EDITED_COPY, 0)) +} + +/// Whether `file_name` is `-.`, the name Google +/// gives an edited copy of `.`. +fn is_edited_copy_of(file_name: &str, original: &str, extension: &str) -> bool { + !original.is_empty() + && file_name + .strip_prefix(original) + .and_then(|rest| rest.strip_prefix('-')) + .and_then(|rest| rest.strip_suffix(extension)) + .and_then(|rest| rest.strip_suffix('.')) + .is_some_and(|word| !word.is_empty() && word.chars().all(char::is_alphabetic)) } /// The stem plus each of its prefixes ending before a `.`, longest first — the @@ -994,21 +1118,25 @@ mod tests { #[test] fn longest_prefix_wins_when_the_photo_name_itself_was_truncated() { + // Both sidecar names are at Google's cap, so both were cut into the + // photo name; the longer surviving prefix is the better claim. + let short = "Takeout/Google Photos/Photos from 2019/PXL_20190101_120000000.PORTRAIT_ORIGINAL_E.json"; + let long = "Takeout/Google Photos/Photos from 2019/PXL_20190101_120000000.PORTRAIT_ORIGINAL_EXPO.json"; let mut sidecars = HashMap::new(); sidecar( &mut sidecars, - "Takeout/Google Photos/Photos from 2019/IMG_1.json", + short, r#"{"photoTakenTime":{"timestamp":"1"}}"#, ); sidecar( &mut sidecars, - "Takeout/Google Photos/Photos from 2019/IMG_12.json", + long, r#"{"photoTakenTime":{"timestamp":"2"}}"#, ); let scan = assemble( vec![entry( 0, - "Takeout/Google Photos/Photos from 2019/IMG_12345.jpg", + "Takeout/Google Photos/Photos from 2019/PXL_20190101_120000000.PORTRAIT_ORIGINAL_EXPOSURE.jpg", 10, )], &sidecars, @@ -1017,6 +1145,104 @@ mod tests { assert_eq!(scan.photos[0].capture_time, Some(2)); } + /// A short sidecar name was not cut, so a prefix of it names another photo: + /// `IMG_1.jpg`'s sidecar gave its title and date to `IMG_12.jpg` + /// (`docs/BUGS.md` B225). + #[test] + fn a_photo_without_a_sidecar_does_not_take_a_shorter_named_siblings() { + let mut sidecars = HashMap::new(); + sidecar( + &mut sidecars, + "Takeout/Google Photos/Photos from 2019/IMG_1.jpg.supplemental-metadata.json", + r#"{"title":"IMG_1.jpg","photoTakenTime":{"timestamp":"1"}}"#, + ); + let scan = assemble( + vec![ + entry(0, "Takeout/Google Photos/Photos from 2019/IMG_1.jpg", 10), + entry(0, "Takeout/Google Photos/Photos from 2019/IMG_12.jpg", 10), + ], + &sidecars, + ); + + assert_eq!(scan.photos[0].capture_time, Some(1)); + assert_eq!(scan.photos[1].name, "IMG_12.jpg"); + assert_eq!(scan.photos[1].capture_time, None); + } + + /// An edited copy has no sidecar of its own. It takes its original's date, + /// but keeps its own name rather than the original's title. + #[test] + fn an_edited_copy_takes_its_originals_date_and_keeps_its_name() { + let mut sidecars = HashMap::new(); + sidecar( + &mut sidecars, + "Takeout/Google Fotos/Urlaub/IMG_2.jpg.ergänzende-Metadaten.json", + r#"{"title":"Holiday.jpg","photoTakenTime":{"timestamp":"7"}}"#, + ); + let scan = assemble( + vec![ + entry(0, "Takeout/Google Fotos/Urlaub/IMG_2.jpg", 10), + entry(0, "Takeout/Google Fotos/Urlaub/IMG_2-bearbeitet.jpg", 10), + ], + &sidecars, + ); + + assert_eq!(scan.photos[0].name, "Holiday.jpg"); + assert_eq!(scan.photos[1].name, "IMG_2-bearbeitet.jpg"); + assert_eq!(scan.photos[1].capture_time, Some(7)); + } + + /// The index looks only at the sidecars that can match, and must find the + /// same one that ranking every sidecar in the folder finds. + #[test] + fn the_sidecar_index_finds_what_ranking_every_sidecar_finds() { + let long = "A_very_long_holiday_photo_name_from_the_beach_2019"; + let names = [ + "IMG_1.jpg".to_string(), + "IMG_12.jpg".to_string(), + "IMG_1(1).jpg".to_string(), + "IMG_1-edited.jpg".to_string(), + "IMG_1-bearbeitet.jpg".to_string(), + "my.photo(2).jpg".to_string(), + "x(1).jpg".to_string(), + "Ölbild.png".to_string(), + "PXL_20230101_101010.mp4".to_string(), + format!("{long}.jpg"), + format!("{long}_2.jpg"), + "nothing.heic".to_string(), + ]; + let sidecar_names = [ + "IMG_1.jpg.supplemental-metadata.json".to_string(), + "IMG_1.jpg(1).supplemental-metadata.json".to_string(), + "IMG_12.jpg.ergänzende-Metadaten.json".to_string(), + "my.photo.jpg(2).json".to_string(), + "x.jpg(1).json".to_string(), + "Ölbild.png.suppl.json".to_string(), + "PXL_20230101_101010.mp4.JSON".to_string(), + format!("{}.json", &long[..44]), + format!("{long}.jp.json"), + "IMG.json".to_string(), + "unrelated.json".to_string(), + ]; + let bodies: Vec> = (0..sidecar_names.len()) + .map(|i| i.to_string().into_bytes()) + .collect(); + let in_folder: BTreeMap> = + sidecar_names.iter().cloned().zip(bodies.iter()).collect(); + let index = SidecarIndex::new(&in_folder); + for name in &names { + let mut every: Option<((u8, usize), &Vec)> = None; + for (sidecar_name, body) in &in_folder { + if let Some(rank) = sidecar_rank(strip_json_suffix(sidecar_name), name) + && every.as_ref().is_none_or(|(current, _)| rank < *current) + { + every = Some((rank, *body)); + } + } + assert_eq!(find_sidecar(name, &index), every, "{name}"); + } + } + #[test] fn sidecar_title_replaces_a_mangled_on_disk_name() { let mut sidecars = HashMap::new(); diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs index 64afdd2..9026a4c 100644 --- a/crates/pdfs-fuse/src/background.rs +++ b/crates/pdfs-fuse/src/background.rs @@ -723,16 +723,24 @@ fn scan_local_once( // has seen and stays indeterminate. let job = transfers.begin_background_job("Indexing this computer"); job.detail("Scanning your files"); + let mut failed = false; let walked = localindex::scan(&[home], &excludes, |batch| { if let Err(e) = db.local_upsert_batch(generation, &batch) { warn!(error = %e, "local index: batch write failed"); + failed = true; } }); - // Prune what this scan did not see and rebuild the FTS index over the rest, - // even if some batches failed — a partial index still beats none. + // Prune what this scan did not see, unless it did not see everything: + // a path it never reached, or one whose batch failed to write, is not a + // path that is gone (`docs/BUGS.md` B217). job.detail("Building the search index"); - match db.local_finish_scan(generation, now_secs()) { + let finished = if failed || walked >= localindex::MAX_ENTRIES { + db.local_finish_partial_scan(now_secs()) + } else { + db.local_finish_scan(generation, now_secs()) + }; + match finished { Ok(indexed) => info!( walked, indexed, diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index e08b5a0..1e0f5ee 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -911,10 +911,22 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U ))), Ok((file, len)) => { let photos = core.photos(); + // Unless the caller knows better: the photo's own EXIF, + // then the file's modification time, which a copy resets. + let capture_time = capture_time.or_else(|| { + let path = Path::new(&source_path); + pdfs_core::exiftime::capture_time_of(path).or_else(|| { + let modified = + std::fs::metadata(path).and_then(|m| m.modified()).ok()?; + let since = modified.duration_since(std::time::UNIX_EPOCH).ok()?; + Some(since.as_secs() as i64) + }) + }); let metadata = proton_drive_rs::PhotoUploadMetadata { capture_time, ..Default::default() }; + let thumbnails = core.upload_thumbnails(Path::new(&source_path), &name); let guard = core.transfers .begin(name.clone(), "", TransferDirection::Upload, len); @@ -924,7 +936,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U &media_type, reader, len as i64, - Vec::new(), + thumbnails, metadata, false, )) { diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs index 3f660c0..72560b7 100644 --- a/crates/pdfs-fuse/src/devices.rs +++ b/crates/pdfs-fuse/src/devices.rs @@ -105,6 +105,15 @@ pub(crate) fn adopted_device_uid() -> Option { (!uid.is_empty()).then_some(uid) } +/// The paths no sync root may overlap, with the mountpoint as the config says +/// now. +pub(crate) fn sync_root_reserved() -> pdfs_core::syncroot::Reserved { + let config = pdfs_core::config::AppDirs::new() + .map(|dirs| dirs.load_config()) + .unwrap_or_default(); + pdfs_core::syncroot::Reserved::current(&config) +} + /// Pin (or, with `None`, unpin) the adopted device uid in `config.json`. pub(crate) fn set_adopted_device_uid(uid: Option<&str>) -> CoreResult<()> { let dirs = pdfs_core::config::AppDirs::new() @@ -517,18 +526,20 @@ impl Core { .map_err(|e| CoreError::internal(format!("canonicalize {}: {e}", local.display())))?; let local_str = local.to_string_lossy().to_string(); - // Reject duplicates up front for a clear error (UNIQUE would also catch it). - if self + // `/`, the home folder, the mount, this app's own data, and a folder + // inside or around another root are refused (`syncroot::problem`). A + // duplicate is one of those, refused here for a clear error before + // the UNIQUE constraint would catch it. + let others = self .db .sync_folder_list() .map_err(|e| CoreError::internal(format!("db: {e:?}")))? .iter() - .any(|f| f.local_path == local_str) - { - return Err(CoreError::invalid(format!( - "{} is already synced", - local.display() - ))); + .map(|f| PathBuf::from(&f.local_path)) + .collect::>(); + let reserved = sync_root_reserved(); + if let Some(problem) = pdfs_core::syncroot::problem(&local, &others, &reserved) { + return Err(CoreError::invalid(problem)); } let name = local diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 5cc7b35..e530787 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -389,13 +389,7 @@ impl Core { // A name a change of ours is about to free is a wait, not a // failure: the trash or rename holding it lands in moments. let held = waits_for_name(e.as_ref()); - let backoff = if held { - DRAIN_BACKOFF_MIN - } else { - DRAIN_BACKOFF_MIN - .saturating_mul(1u32 << attempts.min(6)) - .min(DRAIN_BACKOFF_MAX) - }; + let backoff = drain_backoff(attempts, held, op.id); warn!(uid = %op.uid, attempts, error = %err_str, "pending upload failed; will retry"); if let Err(e) = self.db.record_attempt_failure( op.id, @@ -1118,7 +1112,7 @@ impl Core { _ => None, }; let sent = self.create_sent_digests(op.id, current); - Ok(adoptable(is_dir, &twin, op.created_at, &sent).then_some(twin.uid)) + Ok(adoptable(is_dir, &twin, self.first_sent(op), &sent).then_some(twin.uid)) } /// Record the name the remote actually gave a node, after a conflict forced @@ -1513,8 +1507,10 @@ impl Core { .as_deref() .is_some_and(|sha| holds_bytes(&twin, sha)); let sent = self.create_sent_digests(op.id, current); - Ok(adoptable(op.kind == OP_MKDIR, &twin, op.created_at, &sent) - .then_some((twin.uid, holds_blob))) + Ok( + adoptable(op.kind == OP_MKDIR, &twin, self.first_sent(op), &sent) + .then_some((twin.uid, holds_blob)), + ) } /// The SHA-1s of the bytes queued create `id` may have made its file with: @@ -1529,6 +1525,22 @@ impl Core { sent } + /// When create `op` first went out, which is the earliest a node its lost + /// answer left can have been made. An op queued offline may wait for days + /// before that; measured from when it was queued, the adoption window let + /// it take a node another device made in the meantime. Ops sent before the + /// time was recorded fall back to when they were queued. + fn first_sent(&self, op: &PendingOp) -> i64 { + match self.db.create_first_sent(op.id) { + Ok(Some(at)) => at.max(op.created_at), + Ok(None) => op.created_at, + Err(e) => { + warn!(id = op.id, error = %e, "reading when a queued create was sent failed"); + op.created_at + } + } + } + /// Make one queued `create`/`mkdir` real under a given name, and hand back /// the API's own error so the caller can tell a name clash from a failure. /// @@ -1541,7 +1553,7 @@ impl Core { name: &str, ) -> Result> { self.db - .note_create_target(op.id, &parent.to_string(), name)?; + .note_create_target(op.id, &parent.to_string(), name, now_millis())?; match op.kind == OP_MKDIR { true => Ok(self.block_on_bounded(self.drive.create_folder( parent, @@ -1586,7 +1598,7 @@ impl Core { .begin(name, op.uid.clone(), TransferDirection::Upload, meta.len); let reader = CountingReader::new(File::open(blob)?, &guard); let uid = self.block_on_within( - link::upload_deadline(meta.len), + link::upload_deadline(meta.len, self.transfers.upload_cap()), self.drive.upload_file_from( parent, name, @@ -1983,7 +1995,7 @@ impl Core { .begin(&alt, meta.uid.clone(), TransferDirection::Upload, meta.len); let reader = CountingReader::new(File::open(blob)?, &guard); self.block_on_within( - link::upload_deadline(meta.len), + link::upload_deadline(meta.len, self.transfers.upload_cap()), self.drive.upload_file_from( &parent, &alt, @@ -2236,7 +2248,7 @@ impl Core { .clone() .ok_or("pending op has no staged blob")?; let blob = PathBuf::from(blob); - let meta: StagedWrite = serde_json::from_str(op.meta_json.as_deref().unwrap_or(""))?; + let mut meta: StagedWrite = serde_json::from_str(op.meta_json.as_deref().unwrap_or(""))?; let uid = parse_node_uid(&meta.uid).ok_or("staged write has an unparseable uid")?; if let Some(reason) = self.revision_conflict(&uid, &meta, &blob)? { @@ -2275,6 +2287,22 @@ impl Core { &written, ) .map_err(|e| self.errno_error(e, "gap-fill from base failed"))?; + // The blob is whole now. Recorded, so that a retry after a lost + // answer can match these bytes against the server's revision + // (`revision_conflict`) instead of forking a conflict copy of the + // user's own write. + meta.complete = true; + match serde_json::to_string(&meta) { + Ok(json) => { + let path = op.blob_path.as_deref().unwrap_or_default(); + if let Err(e) = self.db.update_op_meta_for_blob(op.id, path, &json) { + warn!(%uid, error = %e, + "recording a gap-filled blob as whole failed; \ + a lost answer may land it as a conflict copy"); + } + } + Err(e) => warn!(%uid, error = %e, "serializing a gap-filled sidecar failed"), + } } let name = self.node_name(&uid); @@ -2295,7 +2323,7 @@ impl Core { let reader = CountingReader::new(File::open(&blob)?, &guard).with_cancel(cancel.flag()); let started = Instant::now(); let sent = self.block_on_within( - link::upload_deadline(meta.len), + link::upload_deadline(meta.len, self.transfers.upload_cap()), self.drive .upload_new_revision_from(&uid, reader, meta.len as i64, thumbnails, None), ); @@ -2753,24 +2781,43 @@ fn waits_for_name(e: &(dyn std::error::Error + 'static)) -> bool { std::iter::successors(Some(e), |e| e.source()).any(|e| e.is::()) } +/// How long an op waits after its `attempts`-th failure: doubling from +/// [`DRAIN_BACKOFF_MIN`] up to [`DRAIN_BACKOFF_MAX`], plus up to a fifth more, +/// different for each op (`seed`), so that ops that failed together are not +/// all retried in the same second. One that waits for a name (`held`) waits +/// the minimum. +/// +/// The doubling stopped at 2^6, 128 s, so the maximum was never reached. +fn drain_backoff(attempts: i64, held: bool, seed: i64) -> Duration { + if held { + return DRAIN_BACKOFF_MIN; + } + let base = DRAIN_BACKOFF_MIN + .saturating_mul(1u32 << attempts.clamp(0, 8)) + .min(DRAIN_BACKOFF_MAX); + // A cheap, stable spread: the same op and attempt always wait the same. + let spread = (seed.wrapping_mul(0x9E37_79B9).wrapping_add(attempts) as u64 % 21) as u32; + base + base * spread / 100 +} + /// How many attempts an op waits for a rename or a create on the wire to free /// the name it wants. Two files swapping names each wait for the other, so the /// wait has to end; a conflict name ends it. const NAME_HOLD_ATTEMPTS: i64 = 4; -/// How much older than its op a remote node may be and still be the one an -/// unanswered call made. The op is queued the moment the call gives up, so the -/// node is at most one call's deadline older; the rest is clock skew between -/// this machine and Proton. +/// How much older than its op's first send a remote node may be and still be +/// the one an unanswered call made. The node is made during that call, so it +/// is at most one call's deadline older; the rest is clock skew between this +/// machine and Proton. const ADOPT_WINDOW_MS: i64 = 2 * 60 * 1000; /// Whether `twin`, found holding the name a queued create wants, is the node an -/// earlier, unanswered attempt at that create made: the same kind, created -/// around when the op was queued, and for a file still empty, as the mount's +/// earlier, unanswered attempt at that create made: the same kind, created no +/// earlier than around when the op was first sent (`first_sent_ms`), and for a file still empty, as the mount's /// create leaves it, or holding exactly bytes the op uploaded, whose SHA-1 is /// one of `sent`. -fn adoptable(is_dir: bool, twin: &Node, op_created_ms: i64, sent: &[String]) -> bool { - if twin.creation_time.saturating_mul(1000) < op_created_ms - ADOPT_WINDOW_MS { +fn adoptable(is_dir: bool, twin: &Node, first_sent_ms: i64, sent: &[String]) -> bool { + if twin.creation_time.saturating_mul(1000) < first_sent_ms - ADOPT_WINDOW_MS { return false; } match &twin.kind { @@ -3352,6 +3399,22 @@ mod tests { } } + #[test] + fn a_failing_op_backs_off_to_the_maximum_with_a_spread() { + let at = |attempts| drain_backoff(attempts, false, 7); + assert!(at(1) >= DRAIN_BACKOFF_MIN * 2); + assert!(at(2) > at(1)); + // Reached, which the doubling capped at 2^6 never did. + assert!(at(8) >= DRAIN_BACKOFF_MAX); + assert!(at(30) <= DRAIN_BACKOFF_MAX + DRAIN_BACKOFF_MAX / 5); + // Ops that failed together do not all come back together. + let waits: std::collections::HashSet<_> = + (1..=20).map(|id| drain_backoff(8, false, id)).collect(); + assert!(waits.len() > 5, "{waits:?}"); + // A wait for a name stays short however often it repeats. + assert_eq!(drain_backoff(30, true, 7), DRAIN_BACKOFF_MIN); + } + #[test] fn only_a_fresh_empty_twin_of_the_same_kind_is_adopted() { // The op was queued at t=1000 s, after a create that got no answer. diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 9dd60cb..5059202 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -261,10 +261,13 @@ impl Filesystem for ProtonFs { // in-flight (possibly unsaved) content: authored bytes come from the // scratch file, untouched bytes from the base. // - // This path stays on the dispatch loop. `write` runs there too, so - // keeping its reads there as well preserves the ordering the write path - // has always had between a write and a read of the same handle; a mostly - // local read is not worth reasoning about that for. + // Answered on the dispatch loop when everything it needs is on this + // machine, which keeps the order `write` has with reads of the same + // handle. A base range that has to be downloaded goes to a worker + // instead, with the handle as it was when the read came in: the kernel + // issued the read before any write that comes after it, so that is + // the content it asked for. On the loop, the download held every + // request of the mount (`docs/BUGS.md` B227). let handle = { let st = self.core.state(); st.active_writes.get(&ino.0).map(|h| { @@ -279,7 +282,7 @@ impl Filesystem for ProtonFs { }) }; if let Some((file, len, uid, base_mtime, base_size, written)) = handle { - match self.core.serve_open_read( + match self.core.serve_open_read_local( &file, len, &uid, @@ -289,7 +292,25 @@ impl Filesystem for ProtonFs { offset, size as u64, ) { - Ok(bytes) => reply.data(&bytes), + Ok(Some(bytes)) => reply.data(&bytes), + Ok(None) => { + let fs = self.clone(); + self.core.workers.run(Lane::Transfer, "read", move || { + match fs.core.serve_open_read( + &file, + len, + &uid, + base_mtime, + base_size, + &written, + offset, + size as u64, + ) { + Ok(bytes) => reply.data(&bytes), + Err(e) => reply.error(e), + } + }); + } Err(e) => reply.error(e), } return; @@ -556,48 +577,62 @@ impl Filesystem for ProtonFs { return; } let new_end = offset.saturating_add(length); - let keep_size = (mode & 1) != 0; // FALLOC_FL_KEEP_SIZE - - let is_punch_hole = (mode & 0x02) != 0; // FALLOC_FL_PUNCH_HOLE - - let res = { + // Only the modes whose effect on the scratch file this handle can + // record. Anything else went through to the scratch file untracked: + // a zeroed range was refilled from the base at commit, and a collapse + // or insert shifted bytes that `len` and `written` did not follow + // (`docs/BUGS.md` B224). `EOPNOTSUPP` is also what makes glibc's + // `posix_fallocate` fall back to writing zeros. + let Some(kind) = falloc_kind(mode) else { + reply.error(Errno::EOPNOTSUPP); + return; + }; + let file = { + let st = self.core.state(); + match st.handles.get(&_fh.0).and_then(|i| st.active_writes.get(i)) { + Some(aw) => aw.file.clone(), + None => { + reply.error(if st.entries.contains_key(&ino.0) { + Errno::EBADF + } else { + Errno::ENOENT + }); + return; + } + } + }; + // The syscall can take a while on a large range; the state lock is + // not held across it, as `write` does not hold it across its I/O. + let ret = { + use std::os::unix::io::AsRawFd; + unsafe { libc::fallocate(file.as_raw_fd(), mode, offset as i64, length as i64) } + }; + let res = if ret != 0 { + Err(Errno::from(std::io::Error::last_os_error())) + } else { let mut st = self.core.state(); - match st + if let Some(aw) = st .handles .get(&_fh.0) .copied() .and_then(|i| st.active_writes.get_mut(&i)) { - Some(aw) => { - use std::os::unix::io::AsRawFd; - let fd = aw.file.as_raw_fd(); - let ret = unsafe { libc::fallocate(fd, mode, offset as i64, length as i64) }; - if ret == 0 { - if is_punch_hole { - // A punched range reads as zero. Mark it authored so - // queue_revision does not refill it from the remote - // baseline and silently undo the hole at commit. - aw.written.add(offset, new_end.min(aw.len)); - aw.dirty = true; - } else if !keep_size && new_end > aw.len { - aw.written.add(aw.len, new_end); - aw.len = new_end; - aw.dirty = true; - st.set_size(ino.0, new_end); - } - Ok(()) - } else { - Err(Errno::EIO) - } + let len = aw.len; + let (authored, grown) = kind.effect(offset, new_end, len); + if let Some((from, to)) = authored { + // Zeroed or punched bytes read as zero. Marked authored so + // queue_revision does not refill them from the remote + // baseline and silently undo them at commit. + aw.written.add(from, to); + aw.dirty = true; } - None => { - if st.entries.contains_key(&ino.0) { - Err(Errno::EBADF) - } else { - Err(Errno::ENOENT) - } + if let Some(end) = grown { + aw.len = end; + aw.dirty = true; + st.set_size(ino.0, end); } } + Ok(()) }; match res { @@ -606,18 +641,47 @@ impl Filesystem for ProtonFs { } } - /// `close(2)` calls flush before release. The upload is queued in `release` - /// and performed in the background, so there is nothing - /// to push here — the written bytes are already in the scratch file. + /// `close(2)` calls flush before release, and flush is the only one of the + /// two whose answer reaches the program: the kernel drops `release`'s. So + /// the scratch file of a handle with writes on it is synced here, on a + /// worker, and a full disk or a failing one is what `close` returns. + /// + /// Before, flush answered `ok` at once and `release` synced the file, so an + /// `ENOSPC` that delayed allocation held back until then was reported to + /// nobody, and the program took the save as done (`docs/BUGS.md` B222). + /// It also means `release`, which stays on the dispatch loop for ordering, + /// finds the pages already on disk instead of writing back a whole file + /// while every other request on the mount waits. fn flush( &self, _req: &Request, _ino: INodeNo, - _fh: FileHandle, + fh: FileHandle, _lock_owner: LockOwner, reply: ReplyEmpty, ) { - reply.ok(); + let file = { + let st = self.core.state(); + st.handles + .get(&fh.0) + .and_then(|ino| st.active_writes.get(ino)) + .filter(|aw| aw.dirty) + .map(|aw| aw.file.clone()) + }; + let Some(file) = file else { + reply.ok(); + return; + }; + let fs = self.clone(); + self.core + .workers + .run(Lane::Transfer, "flush", move || match file.sync_data() { + Ok(()) => reply.ok(), + Err(e) => { + error!(fh = fh.0, error = %e, "flush of scratch file failed"); + reply.error(fs.scratch_sync_errno(&e)); + } + }); } /// Durability here means the scratch file, not the remote: a caller asking @@ -1022,6 +1086,52 @@ fn build_listing(st: &State, ino: u64) -> DirListing { /// The cookie handed back is the index of the *next* entry, so a resumed call /// continues exactly where the last one stopped — which is only true because /// the listing behind those indexes is a snapshot (`docs/BUGS.md` B43). +/// What an `fallocate` mode does to a write handle, for the modes it can +/// record (`docs/BUGS.md` B224). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FallocKind { + /// Reserve space; past the end it grows the file unless `keep_size`. + Allocate { keep_size: bool }, + /// Deallocate a range inside the file, which then reads as zero. + Punch, + /// Zero a range; past the end it grows the file unless `keep_size`. + Zero { keep_size: bool }, +} + +fn falloc_kind(mode: i32) -> Option { + const KEEP: i32 = libc::FALLOC_FL_KEEP_SIZE; + const PUNCH: i32 = libc::FALLOC_FL_PUNCH_HOLE; + const ZERO: i32 = libc::FALLOC_FL_ZERO_RANGE; + match mode { + 0 => Some(FallocKind::Allocate { keep_size: false }), + KEEP => Some(FallocKind::Allocate { keep_size: true }), + m if m == PUNCH | KEEP => Some(FallocKind::Punch), + ZERO => Some(FallocKind::Zero { keep_size: false }), + m if m == ZERO | KEEP => Some(FallocKind::Zero { keep_size: true }), + _ => None, + } +} + +impl FallocKind { + /// For `[offset, end)` on a file `len` bytes long: the range that now + /// reads as zero and must count as written, and the new length if it grew. + fn effect(self, offset: u64, end: u64, len: u64) -> (Option<(u64, u64)>, Option) { + let grow = |keep_size: bool| (!keep_size && end > len).then_some(end); + let zeroed = |to: u64| (offset < to).then_some((offset, to)); + match self { + // Space past the old end reads as zero, and nothing before it moved. + Self::Allocate { keep_size } => { + (grow(keep_size).map(|end| (len, end)), grow(keep_size)) + } + Self::Punch => (zeroed(end.min(len)), None), + Self::Zero { keep_size } => { + let to = if keep_size { end.min(len) } else { end }; + (zeroed(to), grow(keep_size)) + } + } + } +} + fn reply_listing(listing: &DirListing, offset: u64, root: u64, mut reply: ReplyDirectory) { for (i, (ino, is_dir, name)) in listing.iter().enumerate().skip(offset as usize) { let ft = if *is_dir { @@ -1435,6 +1545,20 @@ impl ProtonFs { } /// The body of [`Filesystem::fsync`], run off the dispatch loop. + /// The errno to give a program for a failed sync of a scratch file. + /// Delayed allocation means a full disk usually surfaces at the sync + /// rather than at `write(2)`, and `ENOSPC` is what a program checks for. + fn scratch_sync_errno(&self, e: &std::io::Error) -> Errno { + match e.raw_os_error() { + Some(libc::ENOSPC) => { + self.core.cache.emergency_evict(); + Errno::ENOSPC + } + Some(libc::EDQUOT) => Errno::EDQUOT, + _ => Errno::EIO, + } + } + fn serve_fsync(&self, fh: FileHandle, reply: ReplyEmpty) { // Everything the durability sidecar needs, copied out so no lock is held // across the sync. @@ -1455,22 +1579,18 @@ impl ProtonFs { ) }) }; + // A handle opened for reading has nothing of its own to make durable, + // and POSIX allows `fsync` on it; `EBADF` failed `sync FILE` and + // backup tools that sync what they read (`docs/BUGS.md` B222). let Some((f, path, uid, dirty, written, len, base_size, base_mtime, base_revision_id)) = handle else { - reply.error(Errno::EBADF); + reply.ok(); return; }; if let Err(e) = f.sync_all() { error!(fh = fh.0, error = %e, "fsync of scratch file failed"); - // Delayed allocation means a full disk usually surfaces here rather - // than at `write(2)`; `ENOSPC` is what an application checks for. - if e.raw_os_error() == Some(libc::ENOSPC) { - self.core.cache.emergency_evict(); - reply.error(Errno::ENOSPC); - } else { - reply.error(Errno::EIO); - } + reply.error(self.scratch_sync_errno(&e)); return; } // A clean handle has nothing on it that the remote does not already @@ -1717,9 +1837,7 @@ impl ProtonFs { // empty, which it is, so reads stay offline-clean across a restart too. if local { st.children.insert(ino, Vec::new()); - if let Err(e) = self.core.db.set_listed(&uid, true) { - warn!(%uid, error = %e, "db set_listed(true) failed for local folder"); - } + st.record_listed(uid); } let entry = st.entries.get(&ino).unwrap(); let attr = self.attr(ino, &entry.node, entry.access); @@ -2337,3 +2455,60 @@ impl ProtonFs { } } } + +#[cfg(test)] +mod falloc_tests { + use super::*; + + #[test] + fn only_the_modes_a_handle_can_record_are_accepted() { + let keep = libc::FALLOC_FL_KEEP_SIZE; + assert_eq!( + falloc_kind(0), + Some(FallocKind::Allocate { keep_size: false }) + ); + assert_eq!( + falloc_kind(keep), + Some(FallocKind::Allocate { keep_size: true }) + ); + assert_eq!( + falloc_kind(libc::FALLOC_FL_PUNCH_HOLE | keep), + Some(FallocKind::Punch) + ); + assert_eq!( + falloc_kind(libc::FALLOC_FL_ZERO_RANGE), + Some(FallocKind::Zero { keep_size: false }) + ); + for mode in [ + libc::FALLOC_FL_PUNCH_HOLE, + libc::FALLOC_FL_COLLAPSE_RANGE, + libc::FALLOC_FL_INSERT_RANGE, + libc::FALLOC_FL_UNSHARE_RANGE, + ] { + assert_eq!(falloc_kind(mode), None, "{mode:#x}"); + } + } + + /// A zeroed range was left unwritten, so the commit filled it back in + /// from the remote base (`docs/BUGS.md` B224). + #[test] + fn a_zeroed_or_punched_range_counts_as_written() { + let zero = FallocKind::Zero { keep_size: false }; + assert_eq!(zero.effect(10, 20, 100), (Some((10, 20)), None)); + assert_eq!(zero.effect(90, 120, 100), (Some((90, 120)), Some(120))); + let zero_keep = FallocKind::Zero { keep_size: true }; + assert_eq!(zero_keep.effect(90, 120, 100), (Some((90, 100)), None)); + assert_eq!( + FallocKind::Punch.effect(90, 120, 100), + (Some((90, 100)), None) + ); + assert_eq!(FallocKind::Punch.effect(120, 130, 100), (None, None)); + let alloc = FallocKind::Allocate { keep_size: false }; + assert_eq!(alloc.effect(0, 50, 100), (None, None)); + assert_eq!(alloc.effect(90, 120, 100), (Some((100, 120)), Some(120))); + assert_eq!( + FallocKind::Allocate { keep_size: true }.effect(90, 120, 100), + (None, None) + ); + } +} diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 30d5aa3..8d02340 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -116,6 +116,7 @@ mod upload; mod r#virtual; mod workers; use background::{forget_and_notify, run_event_sync, run_local_index, run_photos_event_sync}; +use link::BATCH_CALL_TIMEOUT; pub(crate) use mount::is_stale_mount; pub use mount::{MountOptions, MountOutcome, mount}; use mount::{ @@ -2120,12 +2121,39 @@ impl Core { } } - /// Whether `ino`'s listing is already in memory, i.e. whether - /// [`Core::ensure_children`] would return without touching the network. - /// Lets a handler decide between answering inline and handing off to a - /// worker, at the cost of one uncontended map lookup. + /// Whether [`Core::ensure_children`] would answer for `ino` without + /// touching the network. Lets a handler decide between answering inline + /// and handing off to a worker. + /// + /// A resident listing is not enough on its own. A shared folder's, or the + /// Shared with me root's, is fetched again once it is a minute old, and + /// that fetch ran on the dispatch loop, so every request on the mount + /// waited for Drive to answer (`docs/BUGS.md` B219). fn children_cached(&self, ino: u64) -> bool { - self.state().children.contains_key(&ino) + let (uid, resident) = { + let st = self.state(); + match st.entries.get(&ino) { + Some(e) => (e.uid.clone(), st.children.contains_key(&ino)), + // `list_children` answers ENOENT without the network. + None => return true, + } + }; + if !resident { + return false; + } + let freshness = if is_virtual_uid(&uid) { + Some(SHARED_WITH_ME_SYNCED_MS.to_string()) + } else if !self.is_own_or_virtual(&uid) && !is_local_uid(&uid) { + Some(Self::shared_folder_freshness_key(&uid)) + } else { + None + }; + freshness.is_none_or(|key| { + !listing_needs_refresh( + self.online.load(Ordering::Relaxed), + self.shared_listing_stale(&key), + ) + }) } /// Re-apply the optimistic size of any queued write to `nodes`. @@ -2293,8 +2321,7 @@ impl Core { let generation = self.shared_generation.load(Ordering::SeqCst); let accepted = accepted_share_provenance( - self.rt - .block_on(self.client.enumerate_shared_with_me()) + self.block_on_within(BATCH_CALL_TIMEOUT, self.client.enumerate_shared_with_me()) .map_err(|error| { error!(%error, "enumerating shared roots failed"); Errno::EIO @@ -2303,12 +2330,14 @@ impl Core { let nodes = if accepted.uids.is_empty() { Vec::new() } else { - self.rt - .block_on(self.drive.enumerate_nodes_light(&accepted.uids)) - .map_err(|error| { - error!(%error, "materializing shared roots failed"); - Errno::EIO - })? + self.block_on_within( + BATCH_CALL_TIMEOUT, + self.drive.enumerate_nodes_light(&accepted.uids), + ) + .map_err(|error| { + error!(%error, "materializing shared roots failed"); + Errno::EIO + })? }; let virtual_uid = shared_with_me_uid(); let published = prepare_shared_roots(&accepted, nodes, &virtual_uid); @@ -2687,6 +2716,7 @@ impl Core { return Ok(()); } + let preassigned = state::inos_for(&self.db, &filtered_nodes); let mut st = self.state(); // Lost the race? Another thread already populated it. if st.children.contains_key(&ino) { @@ -2708,15 +2738,14 @@ impl Core { }) .map(|n| n.uid.clone()) .collect(); - let inos = st.intern_batch(ino, filtered_nodes); + let inos = st.intern_batch_with_inos(ino, filtered_nodes, preassigned); child_inos.extend(inos); st.children.insert(ino, child_inos); let root_snapshot = primary_root.then(|| RootListingSnapshot::capture(&st, ino).unwrap()); // Record the listing as complete so a later restart (or a trimmed hot - // cache) can rebuild it from the DB without the API. - if let Err(e) = self.db.set_listed(&folder_uid, true) { - warn!(%folder_uid, error = %e, "db set_listed(true) failed"); - } + // cache) can rebuild it from the DB without the API. Written with the + // children's rows when the lock goes, not under it. + st.record_listed(folder_uid.clone()); drop(st); self.flush_access_changes(); if let Some(snapshot) = root_snapshot { @@ -2879,7 +2908,8 @@ impl Core { let (mut key, mut uids, mut slot) = (key, uids, slot); loop { for chunk in uids.chunks(SIZE_UPGRADE_CHUNK) { - let result = self.rt.block_on(self.drive.enumerate_nodes(chunk)); + let result = + self.block_on_within(BATCH_CALL_TIMEOUT, self.drive.enumerate_nodes(chunk)); self.apply_size_upgrade(key, result); slot.chunk_done(); } @@ -4378,9 +4408,15 @@ impl Core { if self.thumbnail_misses.lock().remote_contains(&key, mtime) { return Ok(None); } + // A file manager asks for one per file in a folder, on the metadata + // workers. Offline, or on a stalled link, each one held a worker for + // the SDK's whole retry budget, and every cold lookup, create and + // rename on the mount queued behind them (`docs/BUGS.md` B220). + if !self.is_online() { + return Ok(None); + } let bytes = self - .rt - .block_on(self.client.download_thumbnail(&uid, ttype)) + .block_on_bounded(self.client.download_thumbnail(&uid, ttype)) .map_err(|e| { warn!(%uid, error = %e, "download thumbnail failed"); Errno::EIO @@ -6129,11 +6165,61 @@ fn is_gone(e: &(dyn std::error::Error + 'static)) -> bool { /// (`sync.rs`, `conflict_path`), so the two halves of the product name the same /// situation the same way and a user only has to learn it once. fn conflict_name(name: &str, stamp: i64) -> String { + conflict_name_as(name, conflict_label(), stamp) +} + +/// [`conflict_name`] with the machine's label given. +fn conflict_name_as(name: &str, label: &str, stamp: i64) -> String { let path = Path::new(name); let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or(name); + let tag = conflict_tag(label, stamp, 0); match path.extension().and_then(|s| s.to_str()) { - Some(ext) => format!("{stem} (sync-conflict {stamp}).{ext}"), - None => format!("{stem} (sync-conflict {stamp})"), + Some(ext) => format!("{stem} (sync-conflict {tag}).{ext}"), + None => format!("{stem} (sync-conflict {tag})"), + } +} + +/// This machine's name as its conflict copies carry it (`docs/BUGS.md` B251). +/// +/// Two computers syncing one folder that each keep a copy of the same file in +/// the same second would otherwise make the same name on Drive, and nobody +/// could tell which computer a copy came from. Read once, from the hostname. +fn conflict_label() -> &'static str { + static LABEL: OnceLock = OnceLock::new(); + LABEL.get_or_init(|| conflict_label_of(&this_hostname())) +} + +/// `host` as one token: anything but ASCII letters, digits, `-`, `_` and `.` +/// becomes `-`, so the label holds no space for `conflict_base_name` to trip +/// on and no `/` or `)` to break the name. +fn conflict_label_of(host: &str) -> String { + let label: String = host + .chars() + .take(32) + .map(|c| { + if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.') { + c + } else { + '-' + } + }) + .collect(); + label.trim_matches('-').to_string() +} + +/// The inside of a conflict copy's marker: `