diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index 6d0765e..b1f08af 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/crates/pdfs-core/src/cache.rs @@ -179,6 +179,12 @@ pub struct ContentCache { /// **never** emptied on open: these are the only copy of content the user /// authored, and the whole point is that they outlive the daemon. staging_dir: PathBuf, + /// Subdirectory holding scratch files rescued from an unclean shutdown — + /// writes an application had `fsync`ed but not yet closed. Populated only at + /// open, drained by the daemon into `staging_dir` once it can address the + /// nodes they belong to. Like `staging_dir` and unlike `scratch_dir`, it is + /// never emptied blindly. + recovery_dir: PathBuf, /// JSON pin registry path. pins_path: PathBuf, /// Soft cap on total blob bytes. Exceeded only transiently: a `store` @@ -225,9 +231,14 @@ impl ContentCache { std::fs::create_dir_all(&thumb_dir)?; let block_dir = content_dir.join("blocks"); std::fs::create_dir_all(&block_dir)?; - // Scratch holds disk-backed write buffers; a previous run's leftovers are - // worthless, so start clean. + // Scratch holds disk-backed write buffers. A previous run's leftovers are + // worthless *except* where the application asked for durability: a handle + // that was `fsync`ed but never closed left a sidecar behind, and POSIX + // says those bytes are on stable storage. Rescue those, discard the rest. let scratch_dir = content_dir.join("scratch"); + let recovery_dir = content_dir.join("recovery"); + std::fs::create_dir_all(&recovery_dir)?; + Self::rescue_scratch(&scratch_dir, &recovery_dir); let _ = std::fs::remove_dir_all(&scratch_dir); std::fs::create_dir_all(&scratch_dir)?; // Staging, by contrast, is deliberately preserved across runs. @@ -242,6 +253,7 @@ impl ContentCache { block_dir, scratch_dir, staging_dir, + recovery_dir, pins_path, max_bytes: AtomicU64::new(max_bytes), blob_bytes: AtomicU64::new(0), @@ -433,6 +445,36 @@ impl ContentCache { Ok(()) } + /// Adopt an existing on-disk file as `uid`'s cached content, without reading + /// it into memory. + /// + /// [`Self::store`] takes the bytes as a slice, which means a caller holding a + /// file has to load all of it first — fine for a thumbnail, an OOM for a + /// pinned video. This takes the path instead and hardlinks it into place, + /// falling back to a copy when `src` lives on another filesystem. + /// + /// The link makes the cache blob a second name for the same inode, so `src` + /// must not be rewritten in place afterwards — unlinking it (the staging + /// case, which is what this exists for) is fine and leaves the blob intact. + pub fn store_file(&self, uid: &NodeUid, mtime: i64, size: u64, src: &Path) -> Result<()> { + let blob = self.blob_path(uid); + let tmp = blob.with_extension("tmp"); + // A stale tmp from an interrupted run would fail the link with EEXIST. + let _ = std::fs::remove_file(&tmp); + if std::fs::hard_link(src, &tmp).is_err() { + std::fs::copy(src, &tmp)?; + } + let bytes = std::fs::metadata(&tmp)?.len(); + std::fs::rename(&tmp, &blob)?; + let meta = serde_json::to_vec(&Meta { mtime, size })?; + std::fs::write(self.meta_path(uid), meta)?; + self.db + .cache_touch(&Self::key(uid), KIND_BLOB, bytes, now_millis())?; + self.blob_bytes.fetch_add(bytes, Ordering::Relaxed); + self.enforce_budget(); + Ok(()) + } + /// Cache-index key for block `idx` of `uid` — the block file's name, which /// is also its `cache_entries.cache_key`. fn block_key(&self, uid: &NodeUid, idx: u64) -> String { @@ -626,6 +668,127 @@ impl ContentCache { Ok((file, path)) } + /// The sidecar path for a scratch file, holding the [`StagedWrite`] that + /// describes it. Written by `fsync` and removed when the write is released; + /// its presence at open is what marks a scratch file as durable rather than + /// as a crashed run's rubbish. + /// + /// Safe as `with_extension` only because scratch names are dot-free (see + /// [`Self::create_scratch`]), so this appends rather than replaces. + pub fn scratch_sidecar(scratch: &Path) -> PathBuf { + scratch.with_extension("json") + } + + /// Record that the write buffered in `scratch` is durable, so a crash before + /// `close(2)` does not lose it. + /// + /// Called from `fsync(2)`, whose contract is that the bytes survive a crash. + /// The scratch file itself does survive — it is a real file — but nothing + /// else knows it holds anything worth keeping, and open() clears the scratch + /// directory. The sidecar is that knowledge: uid, logical length, and which + /// ranges are authored rather than holes, which is exactly what recovery + /// needs to hand the blob to the drain. + /// + /// Written whole to a temp name and renamed, so a crash mid-write leaves the + /// previous sidecar rather than a truncated one. The caller is expected to + /// have synced `scratch` first: a sidecar promising bytes that never reached + /// the disk is worse than no sidecar. + pub fn mark_scratch_durable(&self, scratch: &Path, meta: &StagedWrite) -> Result<()> { + let side = Self::scratch_sidecar(scratch); + let tmp = side.with_extension("json.tmp"); + std::fs::write(&tmp, serde_json::to_vec(meta)?)?; + std::fs::rename(&tmp, &side)?; + // The rename is what publishes the sidecar; fsync the directory so the + // rename itself survives, not just the bytes it points at. + if let Some(dir) = side.parent() + && let Ok(d) = std::fs::File::open(dir) + { + let _ = d.sync_all(); + } + Ok(()) + } + + /// Drop a scratch file's durability sidecar, once the write it described has + /// been staged (or turned out not to be dirty at all). Best effort: a + /// leftover sidecar whose blob is gone is ignored by recovery anyway. + pub fn clear_scratch_durable(&self, scratch: &Path) { + let _ = std::fs::remove_file(Self::scratch_sidecar(scratch)); + } + + /// Move every scratch file that carries a readable sidecar into `recovery`, + /// before the caller empties the scratch directory. + /// + /// Blob first, then sidecar: a blob in recovery without its sidecar is + /// dropped on the next open (indistinguishable from rubbish), whereas a + /// sidecar whose blob never arrived would describe nothing. Neither order is + /// lossless if we crash mid-rescue, but this one fails toward "unrecovered + /// bytes on disk" rather than "op queued against missing content". + fn rescue_scratch(scratch_dir: &Path, recovery_dir: &Path) { + let Ok(entries) = std::fs::read_dir(scratch_dir) else { + return; + }; + for entry in entries.flatten() { + let blob = entry.path(); + if blob.extension().is_some_and(|e| e == "json") { + continue; + } + let side = Self::scratch_sidecar(&blob); + // Parsed, not merely present: an unreadable sidecar cannot drive a + // recovery, so the blob is rubbish like any other leftover. + if std::fs::read(&side) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()) + .is_none() + { + continue; + } + let Some(name) = blob.file_name() else { + continue; + }; + if std::fs::rename(&blob, recovery_dir.join(name)).is_err() { + continue; + } + let Some(side_name) = side.file_name() else { + continue; + }; + let _ = std::fs::rename(&side, recovery_dir.join(side_name)); + } + } + + /// Writes rescued from an unclean shutdown, as `(blob, meta)` pairs for the + /// daemon to hand to the upload queue. A blob whose sidecar did not survive + /// the rescue is deleted here — nothing can address it. + pub fn recovered_writes(&self) -> Vec<(PathBuf, StagedWrite)> { + let Ok(entries) = std::fs::read_dir(&self.recovery_dir) else { + return Vec::new(); + }; + let mut out = Vec::new(); + for entry in entries.flatten() { + let blob = entry.path(); + if blob.extension().is_some_and(|e| e == "json") { + continue; + } + match std::fs::read(Self::scratch_sidecar(&blob)) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()) + { + Some(meta) => out.push((blob, meta)), + None => { + let _ = std::fs::remove_file(&blob); + } + } + } + out + } + + /// Retire a recovered write. `stage_write` has already moved the blob out, + /// so this is the sidecar left behind — and the blob too if the caller could + /// not use it and said so by leaving it in place. + pub fn discard_recovered(&self, blob: &Path) { + let _ = std::fs::remove_file(Self::scratch_sidecar(blob)); + let _ = std::fs::remove_file(blob); + } + /// Move a released scratch file into the staging directory with a /// [`StagedWrite`] sidecar, and return where it landed. /// @@ -1072,6 +1235,47 @@ mod tests { ); } + /// The drain promotes a staged upload into the cache by path and then + /// unlinks the staging name. The cached copy has to survive that, and the + /// promote has to charge the LRU index like [`ContentCache::store`] does. + #[test] + fn store_file_adopts_blob_and_survives_source_unlink() { + let (c, d) = cache(); + let u = uid("a"); + let data = b"staged contents"; + let src = d.path().join("staged-blob"); + std::fs::write(&src, data).unwrap(); + + c.store_file(&u, 100, data.len() as u64, &src).unwrap(); + std::fs::remove_file(&src).unwrap(); + + assert!(c.is_cached(&u, 100, data.len() as u64)); + assert_eq!( + c.read_range(&u, 100, data.len() as u64, 7, 8).unwrap(), + b"contents" + ); + assert_eq!(c.usage(), data.len() as u64); + } + + /// A cross-filesystem staging dir cannot be hardlinked from; the copy + /// blob and its metadata must both be replaced wholesale — a stale `.meta` + /// left beside new content would validate a read of the wrong bytes. The + /// link goes to a temp name and is renamed over the blob for exactly this, + /// so re-storing over an existing entry is the case worth pinning. + #[test] + fn store_file_replaces_an_existing_blob() { + let (c, d) = cache(); + let u = uid("a"); + c.store(&u, 100, 3, b"old").unwrap(); + + let src = d.path().join("new-blob"); + std::fs::write(&src, b"newer").unwrap(); + c.store_file(&u, 200, 5, &src).unwrap(); + + assert!(!c.is_cached(&u, 100, 3), "stale meta must not validate"); + assert_eq!(c.read_range(&u, 200, 5, 0, 5).unwrap(), b"newer"); + } + #[test] fn reconcile_rebuilds_index_from_disk() { let dir = TempDir::new(); @@ -1210,13 +1414,12 @@ mod tests { c.store_block(&uid("a"), 1, 100, i, b"aaaa").unwrap(); std::thread::sleep(std::time::Duration::from_millis(2)); } - let indexed: u64 = c - .db - .cache_entries_by_kind("block") - .unwrap() - .iter() - .map(|(_, s)| s) - .sum(); + let indexed: u64 = + c.db.cache_entries_by_kind("block") + .unwrap() + .iter() + .map(|(_, s)| s) + .sum(); assert_eq!( c.block_bytes.load(Ordering::Relaxed), indexed, @@ -1351,7 +1554,10 @@ mod tests { if let Some(got) = c.read_thumbnail(&u, 2, 100) { assert_eq!(got, two, "type 2 served type 1's bytes"); } - assert_eq!(store_errors, 0, "a store lost its staging file to the other type"); + assert_eq!( + store_errors, 0, + "a store lost its staging file to the other type" + ); } #[test] @@ -1577,6 +1783,89 @@ mod tests { ); } + /// A2: `fsync(2)` promises the bytes survive a crash, but queueing happens + /// at `release`. A scratch file marked durable must therefore outlive the + /// open that clears the scratch directory — and an unmarked one must not, + /// since that is a crashed run's rubbish. + #[test] + fn fsynced_scratch_survives_reopen_and_unmarked_scratch_does_not() { + use std::io::Write as _; + let dir = TempDir::new(); + let db = Arc::new(Db::open_in_memory().unwrap()); + let content = dir.path().join("content"); + let pins = dir.path().join("pins.json"); + let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap(); + + let (mut synced, synced_path) = c.create_scratch().unwrap(); + synced.write_all(b"fsynced bytes").unwrap(); + synced.sync_all().unwrap(); + let meta = StagedWrite { + uid: uid("a").to_string(), + len: 13, + base_size: 0, + base_mtime: 100, + authored: vec![(0, 13)], + complete: true, + based_on: None, + }; + c.mark_scratch_durable(&synced_path, &meta).unwrap(); + + // A second handle that was written but never fsynced: no promise was + // made about it, and it is indistinguishable from a torn buffer. + let (mut loose, loose_path) = c.create_scratch().unwrap(); + loose.write_all(b"never fsynced").unwrap(); + + // Crash: no release, no staging, just a restart. + drop(c); + let c2 = ContentCache::open(content, pins, 0, db).unwrap(); + + assert!(!synced_path.exists(), "scratch dir is still cleared"); + assert!(!loose_path.exists()); + + let recovered = c2.recovered_writes(); + assert_eq!(recovered.len(), 1, "only the fsynced write is recoverable"); + let (blob, got) = &recovered[0]; + assert_eq!(std::fs::read(blob).unwrap(), b"fsynced bytes"); + assert_eq!(got.uid, uid("a").to_string()); + assert_eq!(got.authored, vec![(0, 13)]); + + // Staging it is what a restart does with it; the sidecar goes after. + let staged = c2.stage_write(got, blob).unwrap(); + c2.discard_recovered(blob); + assert_eq!(std::fs::read(&staged).unwrap(), b"fsynced bytes"); + assert!(c2.recovered_writes().is_empty(), "recovery does not repeat"); + } + + /// Releasing a write hands the bytes to staging, so the sidecar an earlier + /// `fsync` left must not offer recovery a second, stale copy of it. + #[test] + fn cleared_durability_marker_stops_recovery() { + use std::io::Write as _; + let dir = TempDir::new(); + let db = Arc::new(Db::open_in_memory().unwrap()); + let content = dir.path().join("content"); + let pins = dir.path().join("pins.json"); + let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap(); + + let (mut f, path) = c.create_scratch().unwrap(); + f.write_all(b"released").unwrap(); + let meta = StagedWrite { + uid: uid("a").to_string(), + len: 8, + base_size: 0, + base_mtime: 100, + authored: vec![(0, 8)], + complete: true, + based_on: None, + }; + c.mark_scratch_durable(&path, &meta).unwrap(); + c.clear_scratch_durable(&path); + + drop(c); + let c2 = ContentCache::open(content, pins, 0, db).unwrap(); + assert!(c2.recovered_writes().is_empty()); + } + #[test] fn scratch_file_is_writable_and_isolated() { use std::io::{Read as _, Seek as _, SeekFrom, Write as _}; diff --git a/crates/pdfs-core/src/config.rs b/crates/pdfs-core/src/config.rs index b393985..bafe216 100644 --- a/crates/pdfs-core/src/config.rs +++ b/crates/pdfs-core/src/config.rs @@ -1,6 +1,6 @@ //! Static client identity and per-user filesystem paths. -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use directories::ProjectDirs; use serde::{Deserialize, Serialize}; @@ -165,18 +165,104 @@ impl AppDirs { .unwrap_or_else(|| self.default_mountpoint()) } - /// Create state + cache dirs if missing. + /// Create state + cache dirs if missing, owner-only. + /// + /// The mode is set explicitly rather than left to the umask. These + /// directories hold decrypted file content, a plaintext index of every node + /// name in the Drive, and the control socket — and `create_dir_all` under a + /// typical umask makes them `0755`. Until this was fixed, the only thing + /// keeping another local user out was `~/.cache` and `~/.local/state` + /// happening to be `0700`, which is a convention of the user's system and + /// not something this client established (bugs.md B6). + /// + /// Applied on every start, not just at creation: a directory that already + /// exists with a permissive mode — restored from a backup that flattened + /// modes, or created by an older build — is tightened here. pub fn ensure(&self) -> Result<()> { - std::fs::create_dir_all(self.state_dir())?; - std::fs::create_dir_all(self.cache_dir())?; - let _ = std::fs::create_dir_all(self.dirs.config_dir()); + for dir in [self.state_dir(), self.cache_dir()] { + std::fs::create_dir_all(&dir)?; + restrict_dir(&dir); + } + let config_dir = self.dirs.config_dir().to_path_buf(); + if std::fs::create_dir_all(&config_dir).is_ok() { + restrict_dir(&config_dir); + } Ok(()) } } +/// Make `dir` owner-only (`0700`). Best effort: a mode we cannot set (an +/// exotic filesystem, a directory we do not own) is not worth refusing to +/// start over, and the caller has no better answer than continuing. +fn restrict_dir(dir: &Path) { + use std::os::unix::fs::PermissionsExt; + if let Err(e) = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)) { + tracing::warn!(path = %dir.display(), error = %e, "could not restrict directory to 0700"); + } +} + +/// Make a just-bound Unix socket owner-only (`0600`). +/// +/// Connecting to the control socket drives the daemon with its authenticated +/// session — enumerate, read, upload, trash, share — without presenting any +/// credential, since the daemon already holds one. It is an authority boundary, +/// not merely private data, and `UnixListener::bind` applies the umask like any +/// other file (bugs.md B6). +/// +/// There is a window between `bind` and this call during which the socket +/// carries the umask's mode. It is closed in practice by the containing +/// directory: [`Dirs::ensure`] has already made the state directory `0700`, so +/// nothing else can reach the socket to exploit the window. +pub fn restrict_socket(path: &Path) -> Result<()> { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?; + Ok(()) +} + #[cfg(test)] mod tests { use super::*; + use std::os::unix::fs::PermissionsExt; + + fn mode_of(p: &Path) -> u32 { + std::fs::metadata(p).unwrap().permissions().mode() & 0o777 + } + + /// 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. + #[test] + fn restrict_dir_makes_a_directory_owner_only() { + let dir = std::env::temp_dir().join(format!("pdfs-cfg-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + // Start deliberately world-readable, as `create_dir_all` under a typical + // umask would leave it. + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)).unwrap(); + + restrict_dir(&dir); + + assert_eq!(mode_of(&dir), 0o700, "group and other must have no access"); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Connecting to the control socket confers the daemon's authenticated + /// session with no credential, so it is an authority boundary rather than + /// merely private data. + #[test] + fn restrict_socket_makes_a_socket_owner_only() { + let dir = std::env::temp_dir().join(format!("pdfs-sock-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let sock = dir.join("control.sock"); + let listener = std::os::unix::net::UnixListener::bind(&sock).unwrap(); + + restrict_socket(&sock).unwrap(); + + assert_eq!(mode_of(&sock), 0o600, "no other local user may connect"); + drop(listener); + let _ = std::fs::remove_dir_all(&dir); + } #[test] fn test_app_config_serialization() { diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index 5f08d32..5d1b616 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -76,11 +76,18 @@ pub enum Request { PhotoThumbs { uids: Vec }, /// Download a photo's full content into the cache; replies with its path. OpenPhoto { uid: String }, - /// Upload a photo with the given name, media type, and content bytes. + /// Upload the photo at `source_path` under the given name and media type. + /// + /// A path, not the bytes: this protocol is line-delimited JSON, and + /// `serde_json` writes a `Vec` as an array of decimal integers — a ~5-6x + /// inflation that both peers then hold in memory at once, which turned a + /// large photo or a video into an OOM of the GUI, the daemon, or both. The + /// daemon shares a filesystem with every front-end it serves (the socket is + /// a Unix socket), so it opens and streams the file itself. UploadPhoto { name: String, media_type: String, - bytes: Vec, + source_path: String, capture_time: Option, }, /// Download a Drive file's full content into the cache; replies with the @@ -106,13 +113,10 @@ pub enum Request { /// Create a new folder named `name` under the mountpoint-relative `parent`. /// Replies with [`Response::Ok`]. CreateFolder { parent: String, name: String }, - /// Upload a file named `name` with content `bytes` into the - /// mountpoint-relative `parent` folder. Replies with [`Response::Ok`]. - UploadFile { - parent: String, - name: String, - bytes: Vec, - }, + // There is deliberately no bytes-carrying single-file upload: it was + // `UploadFile { parent, name, bytes }`, and it OOMed on anything large for + // the reason described on `UploadPhoto`. `UploadPaths` covers the same case + // by path, including the one-file batch. /// Bulk-upload local files and/or directory trees into the mountpoint-relative /// `parent` folder. `sources` are absolute paths on the daemon's own /// filesystem (the daemon is local): each file is uploaded, each directory is @@ -1036,10 +1040,11 @@ mod tests { parent: "a".into(), name: "new".into(), }, - Request::UploadFile { - parent: "a".into(), - name: "f.bin".into(), - bytes: vec![0, 1, 2, 255], + Request::UploadPhoto { + name: "p.jpg".into(), + media_type: "image/jpeg".into(), + source_path: "/home/u/p.jpg".into(), + capture_time: Some(1_700_000_000), }, Request::UploadPaths { parent: "a".into(), diff --git a/crates/pdfs-core/src/db/cache.rs b/crates/pdfs-core/src/db/cache.rs index 2e7cbb0..7d20296 100644 --- a/crates/pdfs-core/src/db/cache.rs +++ b/crates/pdfs-core/src/db/cache.rs @@ -90,7 +90,11 @@ impl Db { /// pass over a large cache does not read every row to drop a few, and so the /// `(kind, last_accessed)` index can answer it as a range scan with an early /// stop. - pub fn cache_eviction_candidates(&self, kind: &str, limit: usize) -> Result> { + pub fn cache_eviction_candidates( + &self, + kind: &str, + limit: usize, + ) -> Result> { let conn = self.conn.lock(); let mut stmt = conn.prepare_cached( "SELECT cache_key, size_bytes FROM cache_entries diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index fd76c5a..49f7f52 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -373,7 +373,8 @@ fn children_if_listed_gated_on_flag() { fn a_deleted_child_leaves_a_listed_parent_serving_a_stale_listing() { let db = Db::open_in_memory().unwrap(); db.upsert_node(&folder("root", None, "My Files")).unwrap(); - db.upsert_node(&folder("dst", Some("root"), "dest")).unwrap(); + db.upsert_node(&folder("dst", Some("root"), "dest")) + .unwrap(); db.upsert_node(&file("f1", "root", "a.txt", 1)).unwrap(); db.set_listed(&uid("dst"), true).unwrap(); @@ -1302,7 +1303,10 @@ fn next_due_op_respects_backoff() { let id = db.enqueue_op(&bulk_op(1, &root)).unwrap().0; db.record_op_failure(id, "boom", 5_000).unwrap(); - assert!(db.next_due_op(4_999).unwrap().is_none(), "still backing off"); + assert!( + db.next_due_op(4_999).unwrap().is_none(), + "still backing off" + ); assert!(db.next_due_op(5_000).unwrap().is_some(), "due now"); } @@ -1362,7 +1366,9 @@ fn cache_total_bytes_agrees_with_summing_the_rows() { // query. The claim that matters — that the *common* path does not touch the // database at all — is pinned by `budget_check_is_free_when_under_budget`, // which compares against itself and so is load-independent. - println!("B4: {rounds} budget checks over {N} entries — scan {scan:?}, aggregate {aggregate:?}"); + println!( + "B4: {rounds} budget checks over {N} entries — scan {scan:?}, aggregate {aggregate:?}" + ); } /// Victims still come out least-recently-accessed first, now in bounded batches diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index a1906e3..df1c809 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -165,43 +165,54 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U Ok(CtlRequest::UploadPhoto { name, media_type, - bytes, + source_path, capture_time, }) => { - let photos = core.photos(); - let metadata = proton_drive_rs::PhotoUploadMetadata { - capture_time, - ..Default::default() - }; - let guard = core.transfers.begin( - name.clone(), - "", - TransferDirection::Upload, - bytes.len() as u64, - ); - let reader = CountingReader::new(std::io::Cursor::new(&bytes), &guard); - match core.rt.block_on(photos.upload_photo_from( - &name, - &media_type, - reader, - bytes.len() as i64, - Vec::new(), - metadata, - false, - )) { - Ok(uid) => { - // The photo we just uploaded belongs at the head of the - // timeline, and the GUI reloads the gallery the moment this - // reply lands — so refresh now rather than leaving it to a - // background pass that would land just after that reload. - if let Err(e) = core.rt.block_on(core.refresh_timeline()) { - warn!(error = %e, "timeline refresh after upload failed"); - } - CtlResponse::Ok { - message: format!("uploaded photo with uid {uid}"), + // Streamed off disk rather than carried over the socket: see the note + // on `Request::UploadPhoto`. The length comes from the same handle the + // bytes do, so a file changing size under us cannot make the two + // disagree. + match std::fs::File::open(&source_path).and_then(|f| f.metadata().map(|m| (f, m.len()))) + { + Err(e) => CtlResponse::error(CoreError::invalid(format!( + "cannot read {source_path}: {e}" + ))), + Ok((file, len)) => { + let photos = core.photos(); + let metadata = proton_drive_rs::PhotoUploadMetadata { + capture_time, + ..Default::default() + }; + let guard = + core.transfers + .begin(name.clone(), "", TransferDirection::Upload, len); + let reader = CountingReader::new(std::io::BufReader::new(file), &guard); + match core.rt.block_on(photos.upload_photo_from( + &name, + &media_type, + reader, + len as i64, + Vec::new(), + metadata, + false, + )) { + Ok(uid) => { + // The photo we just uploaded belongs at the head of the + // timeline, and the GUI reloads the gallery the moment this + // reply lands — so refresh now rather than leaving it to a + // background pass that would land just after that reload. + if let Err(e) = core.rt.block_on(core.refresh_timeline()) { + warn!(error = %e, "timeline refresh after upload failed"); + } + CtlResponse::Ok { + message: format!("uploaded photo with uid {uid}"), + } + } + Err(e) => { + CtlResponse::error(CoreError::from_api(&e, "upload photo failed")) + } } } - Err(e) => CtlResponse::error(CoreError::from_api(&e, "upload photo failed")), } } Ok(CtlRequest::OpenFile { path }) => match rel_to_mount(mountpoint, &path) { @@ -294,25 +305,6 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U }, Err(e) => CtlResponse::error(e), }, - Ok(CtlRequest::UploadFile { - parent, - name, - bytes, - }) => match rel_to_mount(mountpoint, &parent) { - Ok(parent_rel) => match core.upload(&parent_rel, &name, &bytes) { - Ok(name) => { - core.log_activity(ActivityKind::Upload, &name, "", true); - CtlResponse::Ok { - message: format!("uploaded {name}"), - } - } - Err(e) => { - core.log_activity(ActivityKind::Upload, &name, &e, false); - CtlResponse::error(e) - } - }, - Err(e) => CtlResponse::error(e), - }, Ok(CtlRequest::UploadPaths { parent, sources }) => { match rel_to_mount(mountpoint, &parent) { // Ack immediately and upload on a background thread: a directory tree diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs index fc40995..33f42dc 100644 --- a/crates/pdfs-fuse/src/devices.rs +++ b/crates/pdfs-fuse/src/devices.rs @@ -555,24 +555,44 @@ impl Core { .next() .ok_or_else(|| SwitchBlocked::Failed("remote folder not found".to_string()))?; - // Reclaim the disk: empty the local dir (keep it as the mountpoint). - evict_dir_contents(&local).map_err(|e| { - SwitchBlocked::Failed(format!("evict {}: {e}", local.display())) - })?; - - let session = self - .spawn_ondemand_mount(&local, root) - .map_err(|e| SwitchBlocked::Failed(e.to_string()))?; - self.mounts.lock().insert(id, session); - // Persist the mode only now that the mount is actually up. Writing it - // first would strand the folder on any failure below: the engine skips - // non-mirror folders, so an `ondemand` row with no mount is a folder - // that is neither mirrored nor browsable, and nothing retries it. - // Failing before this point leaves it `mirror`, and the next pass - // re-downloads whatever eviction removed. + // Persist the mode *before* touching the local tree, and never roll it + // back on failure. + // + // The eviction below deletes every local file in the folder. While the + // row still says `mirror`, that empty directory is a valid input to the + // sync engine, which reads it as "the user deleted all of this" and + // trashes the whole folder on Drive. So the window between the evict and + // the mode write is a window in which any failure — a busy mountpoint, a + // missing fuse module, a dead DB — destroys the folder. Writing the mode + // first closes it: the engine skips non-mirror folders, so the worst a + // failure can now leave behind is an `ondemand` row with no mount, which + // is inert. Inert and marked `error` is recoverable; trashed is not. + // + // The same reasoning forbids reverting to `mirror` when a step below + // fails, since eviction may already have deleted part of the tree. + // Recovery is the user's explicit switch back to `mirror`, whose arm + // clears the baseline and re-downloads. self.db .sync_folder_set_mode(id, "ondemand") .map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?; + + // Reclaim the disk: empty the local dir (keep it as the mountpoint). + if let Err(e) = evict_dir_contents(&local) { + let _ = self.db.sync_folder_set_state(id, "error", now_secs()); + return Err(SwitchBlocked::Failed(format!( + "evict {}: {e}", + local.display() + ))); + } + + let session = match self.spawn_ondemand_mount(&local, root) { + Ok(session) => session, + Err(e) => { + let _ = self.db.sync_folder_set_state(id, "error", now_secs()); + return Err(SwitchBlocked::Failed(e.to_string())); + } + }; + self.mounts.lock().insert(id, session); let _ = self.sync_tx.send(sync::SyncMsg::Rewatch); self.db.sync_folder_set_state(id, "idle", now_secs()).ok(); info!(id, path = %local.display(), "mounted sync folder on-demand"); diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 4d644b7..339040b 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -731,10 +731,11 @@ impl Core { // The staged blob now matches the sealed revision, so a pinned file keeps // it as its cached content rather than re-downloading what we just sent. - if self.cache.is_pinned(&uid) - && let Ok(bytes) = std::fs::read(&blob) - { - let _ = self.cache.store(&uid, now_secs(), meta.len, &bytes); + // Adopted by link, not by value: the blob is the whole file, and a pinned + // video read into a `Vec` here is an OOM. `discard_staged` below drops the + // staging name only — the cache keeps the inode. + if self.cache.is_pinned(&uid) { + let _ = self.cache.store_file(&uid, now_secs(), meta.len, &blob); } self.cache.discard_staged(&blob); self.evict_reader(&uid); diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index cd53887..e7d4e48 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -81,8 +81,8 @@ mod transfers; mod workers; use state::{Entry, Intervals, PendingRevision, State, WriteHandle}; use tracing::{debug, error, info, warn}; -use transfers::{CountingReader, CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry}; -use workers::{FUSE_WORKERS, Workers}; +use transfers::{CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry}; +use workers::{FUSE_WORKERS, Lane, Workers}; /// Attribute/entry cache lifetime handed back to the kernel. Long because the /// Phase 2 event poller actively invalidates changed inodes; without a remote @@ -637,6 +637,67 @@ impl Core { } } + /// Queue the writes that an unclean shutdown caught between `fsync(2)` and + /// `close(2)`. + /// + /// `fsync` promises the bytes survive a crash, but the queueing that makes a + /// write outlive the daemon happens at `release`. A crash in between used to + /// lose the data outright, because the scratch directory is cleared at open. + /// Now `fsync` leaves a sidecar, `ContentCache::open` moves those blobs to + /// `recovery/`, and this walks them into the same staging + queued-op path a + /// normal release takes. + /// + /// Runs after [`hydrate_pending`](Self::hydrate_pending), which is what makes + /// the incomplete-blob check in [`enqueue_staged_write`](Self::enqueue_staged_write) + /// meaningful: a recovered partial write whose earlier write is still queued + /// must not gap-fill from a remote revision that no longer describes the + /// file. With `pending` already loaded, that case is detected and the bytes + /// are parked rather than mis-filled. + /// + /// Failure is per-write and never fatal: a node that no longer exists, or an + /// op that cannot be queued, leaves its blob in `recovery/` for the next run + /// (and for a human) instead of taking the mount down. + fn recover_fsynced_writes(&self) { + let recovered = self.cache.recovered_writes(); + if recovered.is_empty() { + return; + } + let mut queued = 0usize; + for (blob, meta) in recovered { + let Some(uid) = parse_node_uid(&meta.uid) else { + error!(uid = %meta.uid, "recovered write has an unparseable uid; keeping bytes"); + continue; + }; + // The inode is only used to stamp the in-memory tree with the new + // size, and nothing is interned this early — `hydrate` reads the + // size back off `pending` when the node is first looked up, so 0 + // (no such inode) is correct rather than merely tolerable. + match self.enqueue_staged_write(&uid, 0, &blob, meta) { + Ok(()) => { + self.cache.discard_recovered(&blob); + queued += 1; + } + Err(e) => { + error!(%uid, blob = %blob.display(), error = ?e, + "cannot queue a recovered write; bytes kept for the next run"); + // Some failures (a partial write parked by + // `stage_orphaned_write`) still consume the blob. Its sidecar + // would then describe nothing, so retire it — the bytes are + // in staging, which is where a human looks for them. + if !blob.exists() { + self.cache.discard_recovered(&blob); + } + } + } + } + if queued > 0 { + info!( + count = queued, + "recovered fsynced writes from an unclean shutdown" + ); + } + } + /// Discard an op that can never be performed, because the bytes it was to /// upload are gone from staging (something outside the daemon deleted them). /// @@ -1392,6 +1453,11 @@ impl Core { /// The scratch file is *moved* into staging, never copied: it is the only /// copy of what the user wrote. fn queue_revision(&self, h: &WriteHandle) -> Result<(), Errno> { + // The handle is being retired either way, so any durability sidecar an + // `fsync` left has done its job: from here the bytes are tracked as a + // staged write and a queued op, and a sidecar outliving them would offer + // recovery a second, stale copy of the same write. + self.cache.clear_scratch_durable(&h.path); if !h.dirty { let _ = std::fs::remove_file(&h.path); return Ok(()); @@ -2423,46 +2489,6 @@ impl Core { Ok(name.to_string()) } - /// Upload a file named `name` with content `bytes` into the - /// mountpoint-relative `parent_rel` folder. Interns the new node directly. - fn upload(&self, parent_rel: &Path, name: &str, bytes: &[u8]) -> CoreResult { - if name.is_empty() || name.contains('/') { - return Err(CoreError::invalid(format!("invalid name: {name:?}"))); - } - let (pino, parent_uid) = self.resolve(parent_rel)?; - self.ensure_children(pino) - .map_err(|e| self.errno_error(e, "enumerate"))?; - let guard = self - .transfers - .begin(name, "", TransferDirection::Upload, bytes.len() as u64); - let reader = CountingReader::new(std::io::Cursor::new(bytes), &guard); - let new_uid = self - .rt - .block_on(self.client.upload_file_from( - &parent_uid, - name, - media_type_for(name), - reader, - bytes.len() as i64, - Vec::new(), - None, - false, - )) - .map_err(|e| CoreError::from_api(&e, "upload"))?; - drop(guard); - let node = self - .fetch_node(&new_uid) - .map_err(|e| self.errno_error(e, "fetch node"))?; - let mut st = self.state.lock(); - let ino = st.intern(pino, node); - if let Some(kids) = st.children.get_mut(&pino) - && !kids.contains(&ino) - { - kids.push(ino); - } - Ok(name.to_string()) - } - /// Bulk-upload local files and directory trees under `sources` into the /// mountpoint-relative `parent_rel` folder. Directories are recreated (or /// merged into an existing same-named folder) and walked; the resulting flat @@ -3328,7 +3354,7 @@ impl Filesystem for ProtonFs { let fs = self.clone(); self.core .workers - .run(move || fs.serve_lookup(parent, &name, reply)); + .run(Lane::Meta, move || fs.serve_lookup(parent, &name, reply)); } fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option, reply: ReplyAttr) { @@ -3360,7 +3386,7 @@ impl Filesystem for ProtonFs { let fs = self.clone(); self.core .workers - .run(move || fs.serve_readdir(ino, offset, reply)); + .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply)); } fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { @@ -3503,7 +3529,7 @@ impl Filesystem for ProtonFs { // even read off the FUSE device. It only reads `state`, so moving it // races with nothing. FUSE does not require replies in request order. let core = self.core.clone(); - self.core.workers.run(move || { + self.core.workers.run(Lane::Transfer, move || { match core.read_range(&uid, mtime, fsize, offset, size as u64, cache_blocks) { Ok(bytes) => reply.data(&bytes), Err(e) => reply.error(e), @@ -3793,23 +3819,67 @@ impl Filesystem for ProtonFs { _datasync: bool, reply: ReplyEmpty, ) { - let file = self - .core - .state - .lock() - .handles - .get(&fh.0) - .map(|h| h.file.clone()); - match file { - Some(f) => match f.sync_all() { - Ok(()) => reply.ok(), - Err(e) => { - error!(fh = fh.0, error = %e, "fsync of scratch file failed"); - reply.error(Errno::EIO); - } - }, - None => reply.error(Errno::EBADF), + // Everything the durability sidecar needs, copied out so no lock is held + // across the sync. + let handle = self.core.state.lock().handles.get(&fh.0).map(|h| { + ( + h.file.clone(), + h.path.clone(), + h.uid.clone(), + h.dirty, + h.written.clone(), + h.len, + h.base_size, + h.base_mtime, + ) + }); + let Some((f, path, uid, dirty, written, len, base_size, base_mtime)) = handle else { + reply.error(Errno::EBADF); + return; + }; + if let Err(e) = f.sync_all() { + error!(fh = fh.0, error = %e, "fsync of scratch file failed"); + reply.error(Errno::EIO); + return; } + // A clean handle has nothing on it that the remote does not already + // hold, so there is nothing for a crash to lose. + if !dirty { + reply.ok(); + return; + } + // The bytes are on stable storage, but only the scratch file knows that, + // and the scratch directory is cleared at open. Record what the blob is + // so a restart can hand it to the upload queue instead of deleting it. + // + // Deliberately *not* the full release path: staging the blob here would + // move the file out from under a handle the application still has open, + // and queueing an upload per `fsync` would push a revision for every + // barrier in a write loop. This only makes the existing file findable. + let authored: Vec<(u64, u64)> = written + .segments(0, len) + .into_iter() + .filter(|&(_, _, authored)| authored) + .map(|(s, e, _)| (s, e)) + .collect(); + let meta = StagedWrite { + uid: uid.to_string(), + len, + base_size, + base_mtime, + complete: authored == [(0, len)], + authored, + based_on: self.core.remote_baseline(&uid, base_mtime, base_size), + }; + // A failed sidecar means the write is not durable, and `fsync` promising + // otherwise is the defect this exists to fix — so it is an error, not a + // warning. + if let Err(e) = self.core.cache.mark_scratch_durable(&path, &meta) { + error!(%uid, error = %e, "recording fsync durability failed"); + reply.error(Errno::EIO); + return; + } + reply.ok(); } fn release( @@ -4037,17 +4107,10 @@ impl Filesystem for ProtonFs { reply.error(Errno::EIO); return; } - // Forget the node so it re-interns under its new parent, and drop the - // destination listing so it re-enumerates on next access. - // - // The drop has to go through `invalidate_listing`, not `children.remove`: - // `forget` deletes the node's DB row, so a destination still marked - // `listed` would be rebuilt from the DB by `ensure_children` — without the - // node we just moved into it. The file would be gone from the source and - // absent from the destination, with the rename having reported success. - let mut st = self.core.state.lock(); - st.forget(&uid); - st.invalidate_listing(newparent); + self.core + .state + .lock() + .relocate(ino, parent, newparent, &new_parent_uid, &newname); reply.ok(); } @@ -4526,6 +4589,10 @@ pub fn mount( // Writes queued by a previous run (or left behind by a crash) are still owed // an upload, and reads must be served from their staged blobs until they land. core.hydrate_pending(); + // Then the writes that were fsynced but never closed, which the cache moved + // aside at open. After `hydrate_pending`, so a recovered partial write can + // see an already-queued write to the same node. + core.recover_fsynced_writes(); { let core = core.clone(); std::thread::Builder::new() @@ -4584,6 +4651,15 @@ pub fn mount( // stale socket file from a previous run would block the bind, so clear it. let _ = std::fs::remove_file(control_socket); let listener = UnixListener::bind(control_socket)?; + // Owner-only before anything can connect: a peer on this socket commands the + // daemon's authenticated session without a credential of its own (B6). + if let Err(e) = pdfs_core::config::restrict_socket(control_socket) { + error!(error = %e, "cannot restrict control socket permissions; refusing to serve"); + let _ = std::fs::remove_file(control_socket); + return Err(std::io::Error::other(format!( + "control socket permissions: {e}" + ))); + } { let core = core.clone(); let username = username.clone(); diff --git a/crates/pdfs-fuse/src/state.rs b/crates/pdfs-fuse/src/state.rs index 9b26761..50f7c9e 100644 --- a/crates/pdfs-fuse/src/state.rs +++ b/crates/pdfs-fuse/src/state.rs @@ -269,6 +269,43 @@ impl State { } } + /// Settle the local state after a node has been moved (and possibly renamed) + /// on the remote: rewrite it in place, and drop **both** directories' + /// listings so each re-enumerates. + /// + /// Both, because each is stale for its own reason. The destination has + /// gained a child it does not know about; the source has lost one. The + /// source is the subtler half, and was audit A5: pruning the moved node from + /// the source's in-memory children looks sufficient while that entry stays + /// resident, but the source's DB row is left `listed = 1`. Once it is + /// evicted from the hot cache, or the daemon restarts, `ensure_children` + /// rebuilds the listing from the DB and declares it complete, so anything + /// else that changed remotely under the source since is never seen. + /// + /// What this must *not* do is `forget` the node. Forgetting drops its + /// `by_uid` mapping, so the re-enumeration hands it a fresh inode — while + /// the kernel has already carried the renamed dentry over to the *old* one. + /// Every lookup through that dentry then resolves to an inode `entries` no + /// longer holds and fails `ENOENT`, so the renamed directory reads as + /// missing (`ls` on it errors while `ls` of its parent lists it) until the + /// entry TTL expires. Keeping the inode keeps the kernel's dentry valid. + /// + /// A pure rename (`from == to`) invalidates that one directory once. + pub(crate) fn relocate( + &mut self, + ino: u64, + from_parent: u64, + to_parent: u64, + to_parent_uid: &NodeUid, + name: &str, + ) { + self.rename_in_place(ino, to_parent, to_parent_uid, name); + self.invalidate_listing(to_parent); + if from_parent != to_parent { + self.invalidate_listing(from_parent); + } + } + /// Update a file entry's recorded plaintext size so `getattr` reflects an /// in-progress write before the new revision is sealed. pub(crate) fn set_size(&mut self, ino: u64, size: u64) { @@ -305,3 +342,154 @@ impl State { } } } + +#[cfg(test)] +mod tests { + use super::*; + use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId}; + + fn uid(link: &str) -> NodeUid { + NodeUid::new(VolumeId::from("vol"), LinkId::from(link)) + } + + fn node(link: &str, parent: &str, name: &str, is_dir: bool) -> Node { + Node { + uid: uid(link), + parent_uid: Some(uid(parent)), + kind: if is_dir { + NodeKind::Folder + } else { + NodeKind::File { + media_type: "text/plain".into(), + total_size_on_storage: 0, + active_revision_state: None, + claimed_size: Some(0), + claimed_modification_time: None, + } + }, + name: name.into(), + creation_time: 100, + modification_time: 100, + trashed: false, + is_shared: false, + is_shared_publicly: false, + signature_email: None, + verification: Default::default(), + } + } + + /// A unique temp directory removed on drop; avoids a dev-dependency. + struct TempDir(PathBuf); + impl TempDir { + fn new() -> Self { + use std::sync::atomic::{AtomicU64, Ordering}; + static N: AtomicU64 = AtomicU64::new(0); + let p = std::env::temp_dir().join(format!( + "pdfs-state-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir_all(&p).unwrap(); + TempDir(p) + } + } + impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + /// `Db::open_in_memory` is `#[cfg(test)]` inside `pdfs-core`, so it does not + /// exist for this crate's tests — a temp file is the equivalent here. The + /// directory outlives the state because the state holds the open database. + fn state() -> (State, TempDir) { + let dir = TempDir::new(); + let db = Db::open(&dir.0.join("cache.db")).unwrap(); + let st = State { + entries: HashMap::new(), + by_uid: HashMap::new(), + children: HashMap::new(), + next_ino: 1, + handles: HashMap::new(), + next_fh: 1, + db: Arc::new(db), + }; + (st, dir) + } + + /// Two folders, each holding one file, both listed — the state a move + /// starts from. Returns `(state, src_ino, dst_ino)`. + fn two_folders() -> (State, TempDir, u64, u64) { + let (mut st, dir) = state(); + let src = st.intern(0, node("src", "root", "src", true)); + let dst = st.intern(0, node("dst", "root", "dst", true)); + let f = st.intern(src, node("f", "src", "f.txt", false)); + st.children.insert(src, vec![f]); + st.children.insert(dst, vec![]); + st.db.set_listed(&uid("src"), true).unwrap(); + st.db.set_listed(&uid("dst"), true).unwrap(); + (st, dir, src, dst) + } + + /// Audit A5. Moving a file out of a folder leaves that folder's DB row + /// claiming a complete listing, so once the hot cache drops it, remote + /// changes under it stop being seen. `relocate` has to clear both sides. + #[test] + fn relocate_invalidates_the_source_as_well_as_the_destination() { + let (mut st, _dir, src, dst) = two_folders(); + let f = st.by_uid[&uid("f")]; + st.relocate(f, src, dst, &uid("dst"), "f.txt"); + + assert!( + st.db.children_if_listed(&uid("dst")).unwrap().is_none(), + "destination re-enumerates: it gained a child whose row was just deleted" + ); + assert!( + st.db.children_if_listed(&uid("src")).unwrap().is_none(), + "source re-enumerates: its listing predates the move, and a stale \ + `listed` flag would hide every later remote change under it" + ); + assert!(!st.children.contains_key(&src)); + assert!(!st.children.contains_key(&dst)); + assert_eq!( + st.by_uid.get(&uid("f")), + Some(&f), + "the inode survives the move: the kernel has already pointed the \ + renamed dentry at it, so re-interning under a fresh one would make \ + every lookup through that dentry ENOENT" + ); + assert_eq!(st.entries[&f].parent, dst); + } + + /// A pure rename never leaves the directory, so there is one listing to + /// drop, not two — and dropping it must not depend on the parents differing. + #[test] + fn relocate_within_one_directory_still_invalidates_it() { + let (mut st, _dir, src, _dst) = two_folders(); + let f = st.by_uid[&uid("f")]; + st.relocate(f, src, src, &uid("src"), "renamed.txt"); + assert!(st.db.children_if_listed(&uid("src")).unwrap().is_none()); + assert!(!st.children.contains_key(&src)); + assert_eq!(st.by_uid.get(&uid("f")), Some(&f), "the inode is stable"); + assert_eq!(st.entries[&f].node.name, "renamed.txt"); + } + + /// The failure mode the fix exists for, stated directly: forgetting the node + /// prunes it from the resident listing but leaves the DB claiming the source + /// is fully enumerated. This pins *why* `relocate` cannot just call `forget`. + #[test] + fn forget_alone_leaves_the_source_claiming_a_complete_listing() { + let (mut st, _dir, _src, _dst) = two_folders(); + st.forget(&uid("f")); + let listed = st.db.children_if_listed(&uid("src")).unwrap(); + assert!( + listed.is_some(), + "forget does not clear the flag — which is exactly why relocate must" + ); + assert!( + listed.unwrap().is_empty(), + "and the listing it would serve is the moved file's absence, \ + with no way to notice anything else changed" + ); + } +} diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index 24e709b..6ee20ce 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -17,8 +17,8 @@ use std::collections::{HashMap, HashSet}; use std::io::Write; use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::sync::mpsc::Receiver; -use std::time::{Duration, SystemTime}; +use std::sync::mpsc::{Receiver, RecvTimeoutError}; +use std::time::{Duration, Instant, SystemTime}; use notify::{RecursiveMode, Watcher}; use proton_drive_rs::proton_sdk::ids::NodeUid; @@ -30,8 +30,20 @@ use crate::{Core, now_secs, parse_uid}; use pdfs_core::control::{ActivityKind, TransferDirection}; use pdfs_core::db::{StoredSyncEntry, StoredSyncFolder}; -/// Coalesce a burst of filesystem events into a single reconcile per folder. +/// How long the watcher must go quiet before a burst counts as finished. +/// +/// Measured from the *last* event, not the first: applications do not write a +/// file once. An editor saves by writing a temp file and renaming it over the +/// target; an export or a database dump writes continuously for as long as it +/// takes. Settling a fixed interval after the first event walks the tree while +/// the file is still growing, and uploads a torn snapshot as a real revision — +/// corrected on the next pass, but only after spending the encrypt-and-upload +/// on a file that was never in that state. const DEBOUNCE: Duration = Duration::from_secs(2); +/// Ceiling on the settle wait, so a directory under sustained change still +/// syncs. Without it, copying a large tree in — events arriving forever — +/// would postpone the reconcile for as long as the copy ran. +const MAX_SETTLE: Duration = Duration::from_secs(30); /// How often to re-walk remotes so changes made on other clients are pulled in. const POLL_INTERVAL: Duration = Duration::from_secs(120); /// How many uploads/downloads/folder-creations a reconcile runs at once. The @@ -111,11 +123,7 @@ fn engine_loop(core: Core, rx: Receiver) { let mut all = false; let mut do_rewatch = false; classify(msg, &mut ids, &mut all, &mut do_rewatch); - // Debounce: let a burst settle, then drain everything queued behind it. - std::thread::sleep(DEBOUNCE); - while let Ok(m) = rx.try_recv() { - classify(m, &mut ids, &mut all, &mut do_rewatch); - } + settle(&rx, &mut ids, &mut all, &mut do_rewatch); if do_rewatch { rewatch(&core, &mut watcher, &watched); @@ -151,10 +159,7 @@ fn poll_only_loop(core: &Core, rx: Receiver) { let mut all = false; let mut do_rewatch = false; classify(msg, &mut ids, &mut all, &mut do_rewatch); - std::thread::sleep(DEBOUNCE); - while let Ok(m) = rx.try_recv() { - classify(m, &mut ids, &mut all, &mut do_rewatch); - } + settle(&rx, &mut ids, &mut all, &mut do_rewatch); if all { reconcile_all(core); } else { @@ -167,6 +172,46 @@ fn poll_only_loop(core: &Core, rx: Receiver) { } } +/// Absorb the rest of an event burst, returning once the watcher has been quiet +/// for `quiet` or `cap` has elapsed since the first event — whichever comes +/// first. The caller has already classified the event that opened the burst. +/// +/// Split out with its timings as parameters so the settling behaviour can be +/// tested in milliseconds rather than in the tens of seconds the real constants +/// describe. +fn settle_with( + rx: &Receiver, + ids: &mut HashSet, + all: &mut bool, + do_rewatch: &mut bool, + quiet: Duration, + cap: Duration, +) { + let deadline = Instant::now() + cap; + loop { + let now = Instant::now(); + if now >= deadline { + break; + } + // Never wait past the cap, however quiet the watcher goes. + let wait = quiet.min(deadline - now); + match rx.recv_timeout(wait) { + // Another event: the burst is still going, so the quiet window + // starts again from here. + Ok(m) => classify(m, ids, all, do_rewatch), + // Quiet for a full window: whatever was being written has stopped. + Err(RecvTimeoutError::Timeout) => break, + // Every sender is gone; the caller's loop is about to end anyway. + Err(RecvTimeoutError::Disconnected) => break, + } + } +} + +/// [`settle_with`] at the real timings. +fn settle(rx: &Receiver, ids: &mut HashSet, all: &mut bool, do_rewatch: &mut bool) { + settle_with(rx, ids, all, do_rewatch, DEBOUNCE, MAX_SETTLE); +} + fn classify(msg: SyncMsg, ids: &mut HashSet, all: &mut bool, rewatch: &mut bool) { match msg { SyncMsg::Reconcile(id) => { @@ -386,6 +431,37 @@ impl From for PassAbort { } } +/// Refuse to run a pass whose local side has vanished in its entirety. +/// +/// A path in the baseline with no local copy is classified as "the user deleted +/// this", and the pass answers by trashing it on Drive. That is right for one +/// file and right for a handful. When it is true of *every* path the baseline +/// knows, the likely cause is not a user: it is an unmounted external disk, a +/// mountpoint whose FUSE mount died, a home directory that was not yet decrypted +/// when the daemon started, or a bug in this client — and the answer to all of +/// those is to trash the folder's entire contents on Drive. +/// +/// So this pass abort exists to cap the blast radius rather than to catch a +/// specific defect. Losing a pass to it costs a retry; the failure it prevents +/// costs the folder. A user who really did empty the folder can express that by +/// removing it from sync and re-adding it — the baseline goes with it. +/// +/// One surviving path is enough to clear the guard, since the classification is +/// only ever wrong about all of them at once. +fn guard_local_wipe( + baseline: &HashMap, + local: &HashMap, +) -> Result<(), String> { + if baseline.len() >= 2 && baseline.keys().all(|rel| !local.contains_key(rel)) { + return Err(format!( + "every one of the {} synced paths is missing locally; refusing to trash \ + them on Drive. Check that the folder is mounted and readable.", + baseline.len() + )); + } + Ok(()) +} + /// What a [`Pending`] op did, so the reconcile can update shared state on the /// engine thread (never inside a task). enum Applied { @@ -592,6 +668,7 @@ impl Core { let mut local: HashMap = HashMap::new(); self.walk_local(folder_id, local_root, local_root, &mut local)?; + guard_local_wipe(&baseline, &local)?; // The remote folder uids come from the baseline instead of a walk: every // directory this device has synced recorded its uid there when it was @@ -846,6 +923,7 @@ impl Core { let mut local: HashMap = HashMap::new(); self.walk_local(folder_id, local_root, local_root, &mut local)?; + guard_local_wipe(&baseline, &local)?; let mut remote: HashMap = HashMap::new(); let mut remote_dirs: HashMap = HashMap::new(); @@ -1741,6 +1819,113 @@ fn conflict_path(path: &Path, stamp: i64) -> PathBuf { mod tests { use super::*; + /// Feed `n` `Reconcile` events `gap` apart on another thread, starting + /// immediately. Returns the sender's join handle so the test can keep it + /// alive (dropping the sender early would look like a disconnect). + fn drip(n: usize, gap: Duration) -> (Receiver, std::thread::JoinHandle<()>) { + let (tx, rx) = std::sync::mpsc::channel(); + let h = std::thread::spawn(move || { + for i in 0..n { + if tx.send(SyncMsg::Reconcile(i as i64)).is_err() { + return; + } + std::thread::sleep(gap); + } + }); + (rx, h) + } + + fn collect(rx: &Receiver, quiet: Duration, cap: Duration) -> (HashSet, Duration) { + let mut ids = HashSet::new(); + let (mut all, mut rewatch) = (false, false); + // The caller of `settle` has always classified the opening event first. + let first = rx.recv().unwrap(); + classify(first, &mut ids, &mut all, &mut rewatch); + let started = Instant::now(); + settle_with(rx, &mut ids, &mut all, &mut rewatch, quiet, cap); + (ids, started.elapsed()) + } + + /// The bug this replaced a fixed sleep for: a save that keeps writing past + /// the debounce used to be walked mid-write, uploading a torn snapshot as a + /// real revision. The window has to restart on every event, so settling + /// waits for the writer to actually stop. + #[test] + fn settling_waits_for_the_last_event_not_the_first() { + // Eight events 20ms apart — a burst lasting ~160ms, far longer than the + // 60ms quiet window a fixed sleep would have used. + let (rx, h) = drip(8, Duration::from_millis(20)); + let (ids, elapsed) = collect(&rx, Duration::from_millis(60), Duration::from_secs(30)); + h.join().unwrap(); + + assert_eq!(ids.len(), 8, "every event in the burst is absorbed"); + assert!( + elapsed >= Duration::from_millis(120), + "settling returned after {elapsed:?}: it stopped waiting while events \ + were still arriving, which is the torn-revision bug" + ); + } + + /// A burst that ends is not waited on any longer than the quiet window. + #[test] + fn settling_returns_once_the_burst_stops() { + let (rx, h) = drip(3, Duration::from_millis(5)); + let (ids, elapsed) = collect(&rx, Duration::from_millis(50), Duration::from_secs(30)); + h.join().unwrap(); + + assert_eq!(ids.len(), 3); + assert!( + elapsed < Duration::from_secs(5), + "a finished burst must settle on the quiet window, not the cap" + ); + } + + /// Sustained change — copying a large tree in — must not postpone the + /// reconcile for as long as the copy runs. The cap ends the wait. + #[test] + fn settling_is_capped_under_continuous_change() { + // Events every 10ms for far longer than the 150ms cap allows. + let (rx, h) = drip(200, Duration::from_millis(10)); + let (_ids, elapsed) = collect(&rx, Duration::from_millis(100), Duration::from_millis(150)); + assert!( + elapsed < Duration::from_secs(2), + "settled after {elapsed:?}; the cap should have ended the wait near 150ms" + ); + drop(rx); + h.join().unwrap(); + } + + /// The guard that keeps a vanished local tree (dead mount, unplugged disk) + /// from being read as "the user deleted everything" and trashing the folder + /// on Drive. + #[test] + fn guard_local_wipe_blocks_only_a_total_disappearance() { + let base: HashMap = ["a.txt", "b.txt", "d/c.txt"] + .iter() + .map(|r| ((*r).to_string(), ())) + .collect(); + let none: HashMap = HashMap::new(); + let one: HashMap = [("b.txt".to_string(), ())].into_iter().collect(); + + assert!( + guard_local_wipe(&base, &none).is_err(), + "wholesale loss aborts" + ); + assert!( + guard_local_wipe(&base, &one).is_ok(), + "one survivor means the tree is readable; the rest are real deletions" + ); + // A one-entry baseline is a genuine single delete, not a wipe to detect. + let single: HashMap = [("a.txt".to_string(), ())].into_iter().collect(); + assert!(guard_local_wipe(&single, &none).is_ok()); + // Nothing synced yet: a first pass over an empty folder is not a wipe. + assert!(guard_local_wipe(&none, &none).is_ok()); + // Local paths that are not in the baseline are new uploads, and do not + // count as survivors. + let fresh: HashMap = [("new.txt".to_string(), ())].into_iter().collect(); + assert!(guard_local_wipe(&base, &fresh).is_err()); + } + #[test] fn parent_and_base_split_relative_paths() { assert_eq!(parent_rel("a"), ""); diff --git a/crates/pdfs-fuse/src/workers.rs b/crates/pdfs-fuse/src/workers.rs index bf9f68d..5b93717 100644 --- a/crates/pdfs-fuse/src/workers.rs +++ b/crates/pdfs-fuse/src/workers.rs @@ -3,10 +3,17 @@ //! fuser drives `Filesystem` from a single dispatch loop, so a cold read that //! goes to the wire would block every cheap metadata call behind it. Those //! handlers hand their work here instead and reply from the worker. +//! +//! The pool is split into two lanes because moving the work off the dispatch +//! loop is not by itself enough: with one shared queue, eight concurrent block +//! downloads occupy every worker, and a `lookup` that needs one network round +//! trip waits behind megabytes of transfer (audit A6). Some threads are +//! therefore reserved for metadata and never accept a transfer. +use std::collections::VecDeque; use std::sync::Arc; -use parking_lot::Mutex; +use parking_lot::{Condvar, Mutex}; use tracing::warn; /// How many FUSE handlers may run off the dispatch loop at once. @@ -19,75 +26,272 @@ use tracing::warn; /// /// Bounded on purpose: one worker can hold a 4 MiB block in flight, and an /// unbounded pool would let read-ahead on a big file spawn threads without -/// limit. Sized just above the SDK's in-flight block cap so the semaphore -/// there, not thread count, is what bounds download memory. -pub(crate) const FUSE_WORKERS: usize = 8; +/// limit. Sized so the SDK's in-flight block semaphore, not thread count, is +/// what bounds download memory. +/// +/// Counted as `META_WORKERS` reserved threads *plus* the eight that transfers +/// had before the lanes were split. Taking the reservation out of the original +/// eight instead would have cut concurrent reads to five — a throughput +/// regression smuggled in with a latency fix. Threads are the cheap resource +/// here; block buffers are the expensive one, and those are capped in the SDK. +pub(crate) const FUSE_WORKERS: usize = 11; + +/// How many of [`FUSE_WORKERS`] serve metadata *only*. +/// +/// This is the whole guarantee: these threads never accept a [`Lane::Transfer`] +/// job, so no number of concurrent downloads can leave a `lookup` or `readdir` +/// without a thread to run on. Three is enough because metadata jobs are short +/// — one round trip, no block fetch — so they queue behind each other briefly +/// rather than for the length of a transfer. +/// +/// The remaining workers are general: they prefer transfers and fall back to +/// metadata when there is no transfer waiting. That direction is safe (a +/// general thread picking up a cheap job frees itself again quickly) while the +/// reverse — metadata threads helping with transfers — would reintroduce +/// exactly the blocking this split exists to prevent. +const META_WORKERS: usize = 3; + +type Job = Box; + +/// Which lane a job belongs in. +#[derive(Clone, Copy, PartialEq, Eq)] +pub(crate) enum Lane { + /// Short, network-bound metadata work: `lookup`, `readdir`. Guaranteed a + /// thread by the reserved workers. + Meta, + /// Bulk data movement: block reads. May occupy every general worker. + Transfer, +} + +#[derive(Default)] +struct Queues { + meta: VecDeque, + transfer: VecDeque, + /// Set when the pool is dropped, to wake and retire every worker. + closed: bool, +} + +/// The queues and the two wait sets over them. +/// +/// Two condvars, not one, because the workers are not interchangeable: a +/// meta-only thread cannot take a transfer. With a single wait set, +/// `notify_one` for a transfer could wake a meta-only thread, which would find +/// nothing it is allowed to run and go back to sleep — leaving the transfer +/// unclaimed while a general worker sat idle. Waking *every* thread instead +/// would work but costs a thundering herd per job. So each lane notifies only +/// threads that can actually serve it. +struct Pool { + queues: Mutex, + /// Waited on by the reserved metadata workers. + meta_cv: Condvar, + /// Waited on by the general workers. + general_cv: Condvar, +} /// Bounded thread pool behind the network-touching FUSE handlers. /// /// Shared by every session forked off one [`Core`] (the main mount plus each /// on-demand sync folder), so the bound is per daemon rather than per mount. pub(crate) struct Workers { - pub(crate) tx: std::sync::mpsc::Sender>, + inner: Arc, } impl Workers { pub(crate) fn new(n: usize) -> std::io::Result { - let (tx, rx) = std::sync::mpsc::channel::>(); - let rx = Arc::new(Mutex::new(rx)); + let inner = Arc::new(Pool { + queues: Mutex::new(Queues::default()), + meta_cv: Condvar::new(), + general_cv: Condvar::new(), + }); + // Never leave the pool without a general worker, however small `n` is. + let meta_workers = META_WORKERS.min(n.saturating_sub(1)); for i in 0..n { - let rx = rx.clone(); + let inner = inner.clone(); + let meta_only = i < meta_workers; std::thread::Builder::new() - .name(format!("pdfs-fuse-{i}")) + .name(format!( + "pdfs-fuse-{}{i}", + if meta_only { "meta-" } else { "" } + )) .spawn(move || { + let cv = if meta_only { + &inner.meta_cv + } else { + &inner.general_cv + }; loop { - // Held only to take the next job, never across it. - let job = rx.lock().recv(); - match job { - // A panicking handler must not cost the pool a - // worker for the rest of the run. The dropped - // `Reply` answers EIO on its own, so the caller of - // the failed op is told; the next job is unaffected. - // - // This only holds while no shared state a job - // touches sits behind a *poisoning* lock. A - // `std::sync::Mutex` held at the point of the panic - // comes back poisoned, and the worker we just - // rescued dies on its next acquisition — as does - // every other thread. The rescue would then convert - // one recoverable EIO into a permanently broken - // daemon: strictly worse than not catching at all. - // - // This crate's state is behind `parking_lot`, which - // does not poison. **The requirement crosses the - // crate boundary**, though: a job runs deep into - // `proton-sdk`, whose entity cache is a `std` - // mutex. It recovers the guard rather than - // unwrapping (see `InMemoryCacheRepository::state`); - // its session and HTTP state are - // `tokio::sync::Mutex`, which has no poisoning at - // all. Anything new reached from a job owes the same - // check — an SDK-side `.lock().unwrap()` silently - // invalidates this comment. - Ok(job) => { - let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)); + let mut q = inner.queues.lock(); + let job = loop { + // A general worker takes transfers first: metadata + // already has threads of its own, so draining the + // bulk queue is the useful thing for it to do. + let picked = if meta_only { + q.meta.pop_front() + } else { + q.transfer.pop_front().or_else(|| q.meta.pop_front()) + }; + if let Some(job) = picked { + break Some(job); } - // Every sender is gone: the daemon is shutting down. - Err(_) => break, - } + if q.closed { + break None; + } + cv.wait(&mut q); + }; + drop(q); + let Some(job) = job else { break }; + // A panicking handler must not cost the pool a worker + // for the rest of the run. The dropped `Reply` answers + // EIO on its own, so the caller of the failed op is + // told; the next job is unaffected. + // + // This only holds while no shared state a job touches + // sits behind a *poisoning* lock. A `std::sync::Mutex` + // held at the point of the panic comes back poisoned, + // and the worker we just rescued dies on its next + // acquisition — as does every other thread. The rescue + // would then convert one recoverable EIO into a + // permanently broken daemon: strictly worse than not + // catching at all. + // + // This crate's state is behind `parking_lot`, which + // does not poison. **The requirement crosses the crate + // boundary**, though: a job runs deep into + // `proton-sdk`, whose entity cache is a `std` mutex. It + // recovers the guard rather than unwrapping (see + // `InMemoryCacheRepository::state`); its session and + // HTTP state are `tokio::sync::Mutex`, which has no + // poisoning at all. Anything new reached from a job owes + // the same check — an SDK-side `.lock().unwrap()` + // silently invalidates this comment. + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)); } })?; } - Ok(Self { tx }) + Ok(Self { inner }) } - /// Run `job` on a worker. Falls back to running it inline — the pre-pool - /// behaviour — if no worker is left to take it, so a shut-down pool degrades - /// to a slow mount rather than a mount that answers every read with EIO. - pub(crate) fn run(&self, job: impl FnOnce() + Send + 'static) { - if let Err(e) = self.tx.send(Box::new(job)) { + /// Queue `job` in `lane`. + /// + /// Neither queue is bounded, and deliberately so: the only thing that could + /// apply backpressure here is the fuser dispatch loop, and blocking *that* + /// is the stall this module exists to avoid. Queue depth is cheap anyway — + /// a pending job holds a `Reply` and a few fields, while the 4 MiB block + /// buffer is allocated inside the job once it runs, and the SDK's in-flight + /// semaphore is what bounds how many of those exist at once. + pub(crate) fn run(&self, lane: Lane, job: impl FnOnce() + Send + 'static) { + let mut q = self.inner.queues.lock(); + if q.closed { + // Pre-pool behaviour: a shut-down pool degrades to a slow mount + // rather than a mount that answers every read with EIO. + drop(q); warn!("fuse worker pool is gone; serving inline"); - (e.0)(); + job(); + return; + } + match lane { + Lane::Meta => q.meta.push_back(Box::new(job)), + Lane::Transfer => q.transfer.push_back(Box::new(job)), + } + drop(q); + match lane { + // Either class can serve metadata, and only one of them needs to: + // whichever wakes first takes it, and the other finds the queue + // empty and waits again. + Lane::Meta => { + self.inner.meta_cv.notify_one(); + self.inner.general_cv.notify_one(); + } + // Only general workers may take a transfer, so waking a reserved + // one would be the lost wakeup this split is careful to avoid. + Lane::Transfer => { + self.inner.general_cv.notify_one(); + } } } } + +impl Drop for Workers { + fn drop(&mut self) { + self.inner.queues.lock().closed = true; + self.inner.meta_cv.notify_all(); + self.inner.general_cv.notify_all(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + use std::time::Duration; + + /// Audit A6. Transfers filling every general worker must not delay a + /// metadata job: that is what the reserved threads are for. + #[test] + fn saturated_transfers_do_not_block_metadata() { + let pool = Workers::new(FUSE_WORKERS).unwrap(); + let general = FUSE_WORKERS - META_WORKERS; + + // Occupy every general worker with a transfer that will not finish + // until we say so, and wait until they are all actually running. + let (release_tx, release_rx) = mpsc::channel::<()>(); + let release_rx = Arc::new(Mutex::new(release_rx)); + let (started_tx, started_rx) = mpsc::channel(); + for _ in 0..general { + let release_rx = release_rx.clone(); + let started_tx = started_tx.clone(); + pool.run(Lane::Transfer, move || { + started_tx.send(()).unwrap(); + let _ = release_rx.lock().recv(); + }); + } + for _ in 0..general { + started_rx + .recv_timeout(Duration::from_secs(5)) + .expect("every general worker picks up a transfer"); + } + + // Queue more transfers than there are threads, so the lane is backed up + // in the way that used to starve metadata. + for _ in 0..16 { + let release_rx = release_rx.clone(); + pool.run(Lane::Transfer, move || { + let _ = release_rx.lock().recv(); + }); + } + + let (meta_tx, meta_rx) = mpsc::channel(); + pool.run(Lane::Meta, move || meta_tx.send(()).unwrap()); + meta_rx + .recv_timeout(Duration::from_secs(5)) + .expect("a metadata job runs while every transfer thread is stuck"); + + for _ in 0..(general + 16) { + let _ = release_tx.send(()); + } + } + + /// The reverse direction is allowed and matters for throughput: with no + /// transfers in flight, a general worker serves metadata rather than idling. + #[test] + fn general_workers_fall_back_to_metadata() { + // One general worker and no reserved ones, so anything that completes + // must have been served by the general worker. + let pool = Workers::new(1).unwrap(); + let (tx, rx) = mpsc::channel(); + pool.run(Lane::Meta, move || tx.send(()).unwrap()); + rx.recv_timeout(Duration::from_secs(5)) + .expect("a general worker takes metadata when no transfer is waiting"); + } + + /// A panicking job costs its caller an EIO, never the worker. + #[test] + fn a_panicking_job_does_not_retire_its_worker() { + let pool = Workers::new(1).unwrap(); + pool.run(Lane::Meta, || panic!("boom")); + let (tx, rx) = mpsc::channel(); + pool.run(Lane::Meta, move || tx.send(()).unwrap()); + rx.recv_timeout(Duration::from_secs(5)) + .expect("the pool still serves work after a panic"); + } +} diff --git a/crates/pdfs-gui/src/app/pages/photos.rs b/crates/pdfs-gui/src/app/pages/photos.rs index b347969..616936a 100644 --- a/crates/pdfs-gui/src/app/pages/photos.rs +++ b/crates/pdfs-gui/src/app/pages/photos.rs @@ -615,13 +615,6 @@ pub(crate) fn wire_gallery(ui: &Rc, list: >k4::ListView, scroll: >k4::Sc .and_then(|n| n.to_str()) .unwrap_or("photo.jpg") .to_string(); - let bytes = match std::fs::read(&path) { - Ok(b) => b, - Err(e) => { - tracing::error!("Failed to read photo: {e}"); - return; - } - }; let ext = path .extension() .and_then(|e| e.to_str()) @@ -643,10 +636,13 @@ pub(crate) fn wire_gallery(ui: &Rc, list: >k4::ListView, scroll: >k4::Sc ui.busy_begin(); let rx = spawn_request( ui.dirs.control_socket(), + // The daemon opens the file itself: it runs on this machine, + // and a photo's bytes through line-delimited JSON is an OOM + // of both processes (see `Request::UploadPhoto`). Request::UploadPhoto { name, media_type: media_type.to_string(), - bytes, + source_path: path.display().to_string(), capture_time, }, ); diff --git a/crates/pdfs-gui/src/main.rs b/crates/pdfs-gui/src/main.rs index cf02048..0bc441f 100644 --- a/crates/pdfs-gui/src/main.rs +++ b/crates/pdfs-gui/src/main.rs @@ -270,6 +270,12 @@ fn main() -> Result<()> { let _ = std::fs::remove_file(&tray_sock); let _lock_socket = std::os::unix::net::UnixListener::bind(&tray_sock) .context("failed to bind tray single-instance socket")?; + // Owner-only like the control socket (B6). This one only guards single + // instancing, so a failure is logged rather than fatal — but there is no + // reason to leave it reachable by other local users either. + if let Err(e) = pdfs_core::config::restrict_socket(&tray_sock) { + tracing::warn!(error = %e, "could not restrict tray socket permissions"); + } let socket = dirs.control_socket(); let default_mountpoint = dirs.default_mountpoint();