//! The control socket: the daemon`s IPC surface. //! //! The CLI and the GUI never touch the mount or the database directly — they //! open this Unix socket, write one JSON [`CtlRequest`] line, and read one JSON //! [`CtlResponse`] line back. That keeps the daemon the single writer of both //! the FUSE state and `cache.db`. //! //! Connections are served concurrently, one thread each, because requests differ //! wildly in cost: an `OpenPhoto` downloads a whole photo, and a serial loop //! would stall the GUI`s 2s status poll behind it. use std::io::{BufRead, BufReader, Write as _}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Component, Path, PathBuf}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Duration, Instant}; use pdfs_core::config::AppDirs; use pdfs_core::control::{ ActivityKind, PhotoMonth, RefreshScope, Request as CtlRequest, Response as CtlResponse, TransferDirection, TrashFailure, }; use pdfs_core::{CoreError, CoreResult}; use proton_drive_rs::proton_sdk::ids::NodeUid; use tracing::{debug, info, warn}; use super::transfers::CountingReader; use super::{Core, count_noun, human_bytes, human_duration, parse_uid}; const MAX_CONTROL_REQUEST_BYTES: u64 = 1024 * 1024; const CONTROL_READ_TIMEOUT: Duration = Duration::from_secs(10); /// Answer a [`CtlRequest::StopSharing`]: log it, and turn a partial removal /// into an error that names what is still shared. fn stop_sharing_reply( core: &Core, target: &str, result: CoreResult, ) -> CtlResponse { match result.map(|outcome| outcome.into_result()) { Ok(Ok(message)) => { core.log_activity(ActivityKind::Unshare, target, "stopped sharing", true); CtlResponse::Ok { message } } Ok(Err(left)) => { core.log_activity(ActivityKind::Unshare, target, &left, false); CtlResponse::error(CoreError::remote(left)) } Err(e) => CtlResponse::error(e), } } /// A client that connects, sends a request and then never reads its reply used /// to park a handler thread forever in `write_all`; 64 of those exhaust /// [`MAX_CONTROL_HANDLERS`] permanently. Generous enough for the largest /// responses (photo bytes) over a local socket, finite so a wedged peer can't /// hold a permit. const CONTROL_WRITE_TIMEOUT: Duration = Duration::from_secs(30); const MAX_CONTROL_HANDLERS: usize = 64; /// Ceiling on any client-supplied row `limit`. Every in-tree caller asks for far /// less (200 for activity, 60 for a photo page), but the value goes straight /// into `SQL LIMIT ?` under the daemon's single connection — an unclamped one /// materialises the whole account, and then serialises it into one JSON line. const MAX_ROW_LIMIT: usize = 1000; /// Clamp a client-supplied row limit to [`MAX_ROW_LIMIT`]. fn clamp_limit(limit: usize) -> usize { limit.min(MAX_ROW_LIMIT) } /// Answer an album add or remove: parse the uids, run `change`, and report /// per photo. fn album_change( uid: &str, photos: &[String], change: impl FnOnce(&NodeUid, &[NodeUid]) -> CoreResult, ) -> CtlResponse { let Some(album) = parse_uid(uid) else { return CtlResponse::error(CoreError::invalid(format!("bad album uid: {uid}"))); }; let photos = match Core::parse_uids(photos) { Ok(photos) => photos, Err(e) => return CtlResponse::error(e), }; match change(&album, &photos) { Ok((changed, failed)) => CtlResponse::AlbumChanged { changed, failed: failed .into_iter() .map(|(uid, message)| TrashFailure { uid, message }) .collect(), }, Err(e) => CtlResponse::error(e), } } static ACTIVE_CONTROL_HANDLERS: AtomicUsize = AtomicUsize::new(0); struct ControlHandlerPermit; impl ControlHandlerPermit { fn acquire() -> Option { ACTIVE_CONTROL_HANDLERS .fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| { (active < MAX_CONTROL_HANDLERS).then_some(active + 1) }) .ok() .map(|_| Self) } } impl Drop for ControlHandlerPermit { fn drop(&mut self) { ACTIVE_CONTROL_HANDLERS.fetch_sub(1, Ordering::AcqRel); } } /// Handler threads in use, for the diagnostics. At the limit the daemon refuses /// connections, which a front-end cannot tell apart from a dead daemon. pub(crate) fn active_handlers() -> usize { ACTIVE_CONTROL_HANDLERS.load(Ordering::Acquire) } pub(crate) fn handler_limit() -> usize { MAX_CONTROL_HANDLERS } fn read_request_line(reader: &mut impl BufRead) -> std::io::Result> { let mut bytes = Vec::new(); loop { let available = reader.fill_buf()?; if available.is_empty() { break; } let take = available .iter() .position(|byte| *byte == b'\n') .map_or(available.len(), |index| index + 1); if bytes.len().saturating_add(take) > MAX_CONTROL_REQUEST_BYTES as usize { return Err(std::io::Error::new( std::io::ErrorKind::InvalidData, "control request exceeds size limit", )); } bytes.extend_from_slice(&available[..take]); reader.consume(take); if bytes.ends_with(b"\n") { break; } } if bytes.is_empty() { return Ok(None); } if bytes.len() as u64 > MAX_CONTROL_REQUEST_BYTES || !bytes.ends_with(b"\n") { return Err(std::io::Error::new( std::io::ErrorKind::InvalidData, "control request exceeds limit or is not newline terminated", )); } String::from_utf8(bytes) .map(Some) .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e)) } /// Turn a CLI-supplied path into the mount that owns it and the path relative /// to *that* mount's root. /// /// [`rel_to_mount`] is the older, primary-only reading, and it is what every /// mutating request still uses. It rejects anything outside the primary /// mountpoint, which leaves a secondary on-demand mount — a synced folder /// switched out of mirror mode — with no way to be named at all: `pdfs ls` and /// `pdfs refresh` both answered "is not under the mountpoint", so a folder /// serving a stale listing had no manual escape hatch (`docs/BUGS.md` B86). /// /// A relative path keeps the old meaning, relative to the primary mount, since /// there is nothing in it to choose a mount by. fn route_to_mount(core: &Core, mountpoint: &Path, path: &str) -> CoreResult<(Core, PathBuf)> { let p = Path::new(path); if !p.is_absolute() { return Ok((core.clone(), p.to_path_buf())); } if let Some(routed) = core.rooted_at(p) { return Ok(routed); } // Fall back to the primary reading so the error text stays the familiar one // when the path is simply outside everything this daemon has mounted. rel_to_mount(mountpoint, path).map(|rel| (core.clone(), rel)) } /// Turn a CLI-supplied path into a mountpoint-relative path. An absolute path /// must live under `mountpoint`; a relative path is taken as already relative to /// the mount root. fn rel_to_mount(mountpoint: &Path, path: &str) -> CoreResult { let p = Path::new(path); let relative = if p.is_absolute() { p.strip_prefix(mountpoint) .map(Path::to_path_buf) .map_err(|_| CoreError::invalid(format!("{path} is not under the mountpoint"))) } else { Ok(p.to_path_buf()) }?; if relative .components() .all(|component| matches!(component, Component::Normal(_) | Component::CurDir)) { Ok(relative) } else { Err(CoreError::invalid(format!("{path} escapes the mountpoint"))) } } /// Handle one control-socket connection: read a single JSON request line, /// dispatch it against `core`, and write back a JSON response line. /// Refresh scopes being served right now, one entry per running refresh. /// /// A `Vec` because there are three scopes and at most `MAX_CONTROL_HANDLERS` /// handlers, so a set would buy nothing. static REFRESHING: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); /// What makes two refreshes the same work: the scope, and for a directory the /// folder, since refreshing one folder says nothing about another. A full /// photos refresh is not the same work as an ordinary one — it forgets what the /// daemon knows first — so it keys apart. fn scope_key(scope: &RefreshScope) -> String { match scope { RefreshScope::Dir { path } => format!("dir:{path}"), RefreshScope::Trash => "trash".to_string(), RefreshScope::Photos { full: true } => "photos:full".to_string(), RefreshScope::Photos { full: false } => "photos".to_string(), } } /// Holds one refresh scope for as long as it is being served. `acquire` /// returns `None` when that scope is already in flight. struct RefreshGuard(String); impl RefreshGuard { fn acquire(scope: &RefreshScope) -> Option { let key = scope_key(scope); let mut running = REFRESHING.lock().unwrap_or_else(|e| e.into_inner()); if running.contains(&key) { return None; } running.push(key.clone()); Some(Self(key)) } } impl Drop for RefreshGuard { fn drop(&mut self) { let mut running = REFRESHING.lock().unwrap_or_else(|e| e.into_inner()); if let Some(at) = running.iter().position(|key| *key == self.0) { running.swap_remove(at); } } } fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: UnixStream) { if let Err(e) = stream.set_read_timeout(Some(CONTROL_READ_TIMEOUT)) { warn!(error = %e, "control: could not set request timeout"); return; } if let Err(e) = stream.set_write_timeout(Some(CONTROL_WRITE_TIMEOUT)) { warn!(error = %e, "control: could not set response timeout"); return; } let mut reader = BufReader::new(match stream.try_clone() { Ok(s) => s, Err(e) => { warn!(error = %e, "control: clone stream failed"); return; } }); let line = match read_request_line(&mut reader) { Ok(Some(line)) if !line.trim().is_empty() => line, Ok(_) => return, Err(e) => { warn!(error = %e, "control: rejected request"); return; } }; // Registered before the request is even parsed, so a handler that wedges // inside the daemon shows up by name in `pdfs diagnostics`. let _inflight = super::diagnostics::InflightGuard::new(super::diagnostics::request_kind(&line)); let response = match serde_json::from_str::(line.trim()) { Ok(CtlRequest::Status) => { let pins = core.cache.list_pins(); let queued = core.db.pending_op_counts().unwrap_or_default(); let staging = core.cache.staging_usage(); CtlResponse::Status { parked_uploads: queued.parked.max(0) as u64, failing_ops: queued.failing.max(0) as u64, failing_error: queued.last_error.clone(), staged_bytes: staging.bytes, staged_oldest_secs: staging.oldest_secs, paused: core.sync_pause().is_some(), paused_until: core.sync_pause().flatten(), username: username.to_string(), mountpoint: mountpoint.display().to_string(), pinned: pins.len(), used: core.cache.usage(), budget: core.cache.budget(), pins, online: core.online.load(Ordering::Relaxed), pending_uploads: queued.uploads.max(0) as u64, pending_changes: queued.changes.max(0) as u64, } } Ok(CtlRequest::Pin { path }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.pin(&rel) { Ok(name) => { core.touch_profile(); CtlResponse::Ok { message: format!("pinned {name}"), } } Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::Unpin { path }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.unpin(&rel) { Ok(name) => { core.touch_profile(); CtlResponse::Ok { message: format!("unpinned {name}"), } } Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListPins) => CtlResponse::Pins { pins: core.list_pins(), }, Ok(CtlRequest::ListDir { path }) => match route_to_mount(core, mountpoint, &path) { Ok((core, rel)) => match core.list_dir(&rel) { Ok(entries) => CtlResponse::Entries { entries }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::Refresh { scope }) => match RefreshGuard::acquire(&scope) { // The same scope is already being refreshed by another handler. // Running it again cannot make the answer newer, and the duplicate // would spend its whole run contending for the state and database // locks the first one holds — which is how five stacked photo // refreshes sat in flight for twenty-three minutes at mount time. None => { debug!(scope = %scope_key(&scope), "a refresh of this scope is already running"); CtlResponse::Ok { message: "refreshing".to_string(), } } Some(_refreshing) => { let result = match &scope { RefreshScope::Dir { path } => route_to_mount(core, mountpoint, path) .and_then(|(core, rel)| core.refresh_dir(&rel)), RefreshScope::Trash => { core.invalidate_trash(); Ok(()) } RefreshScope::Photos { full } if !full && core.timeline_refreshing() => { // A refresh is already running and will publish exactly the // timeline this request asks for. Doing the work again // cannot make it newer, and the invalidation plus the // album read contend with the running refresh for the // database lock — which is how a handful of stacked photo // refreshes at mount time left five control requests // in flight for twenty-three minutes. debug!("a photos refresh is already running; joining it"); Ok(()) } RefreshScope::Photos { full } => { core.invalidate_photos(); // A full refresh is "forget what you know", not a second code // path: the ordinary refresh then finds every photo stale. if let Some(Err(e)) = full.then(|| core.db.photos_mark_all_unresolved()) { warn!(error = %e, "marking the whole timeline unresolved failed"); } // Whatever changed the timeline can just as easily have // changed an album; one scope covers the whole photos view. core.invalidate_albums(); // The refresh is started here, not awaited. A library of // twelve thousand photos is re-read in chunks of // `TIMELINE_ENRICH_CHUNK` nodes, which takes minutes — // longer than the client's read bound, so awaiting it // returned `Resource temporarily unavailable` to a caller // whose refresh was in fact running fine (B92). The front // end follows it with `PhotosRefreshStatus`. core.spawn_timeline_refresh(); Ok(()) } }; match result { Ok(()) => CtlResponse::Ok { message: match scope { RefreshScope::Photos { .. } => "refreshing".to_string(), _ => "refreshed".to_string(), }, }, Err(e) => CtlResponse::error(e), } } }, Ok(CtlRequest::PhotosTimeline { offset, limit, kind, range, favorites, }) => match core.photos_timeline(offset, clamp_limit(limit), kind, range, favorites) { Ok(Some(items)) => CtlResponse::Photos { available: true, items, counts: core.db.photos_counts().ok(), }, Ok(None) => CtlResponse::Photos { available: false, items: Vec::new(), counts: None, }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::PhotoGroup { uid }) => match core.photo_group(&uid) { Ok(items) => CtlResponse::Photos { available: true, items, counts: None, }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::SetPhotoFavorite { uid, favorite }) => match parse_uid(&uid) { Some(node_uid) => match core.set_photo_favorite(&node_uid, favorite) { Ok(()) => CtlResponse::Ok { message: if favorite { "added to favorites".to_string() } else { "removed from favorites".to_string() }, }, Err(e) => CtlResponse::error(e), }, None => CtlResponse::error(CoreError::invalid(format!("invalid uid: {uid}"))), }, Ok(CtlRequest::PhotoAlbums) => match core.albums() { Ok(Some(items)) => CtlResponse::Albums { available: true, items, }, Ok(None) => CtlResponse::Albums { available: false, items: Vec::new(), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::AlbumPhotos { uid, offset, limit }) => match parse_uid(&uid) { // An album page is a timeline page as far as the front-end is // concerned, so it comes back in the same reply — with `counts` // unset, which describe the whole timeline rather than this album. Some(album) => match core.album_photos(&album, offset, clamp_limit(limit)) { Ok(items) => CtlResponse::Photos { available: true, items, counts: None, }, Err(e) => CtlResponse::error(e), }, None => CtlResponse::error(CoreError::invalid(format!("bad album uid: {uid}"))), }, Ok(CtlRequest::CreateAlbum { name }) => match core.create_album(&name) { Ok(uid) => CtlResponse::AlbumCreated { uid }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RenameAlbum { uid, name }) => match parse_uid(&uid) { Some(album) => match core.rename_album(&album, &name) { Ok(()) => CtlResponse::Ok { message: format!("renamed album to {name}"), }, Err(e) => CtlResponse::error(e), }, None => CtlResponse::error(CoreError::invalid(format!("bad album uid: {uid}"))), }, Ok(CtlRequest::DeleteAlbum { uid, delete_photos }) => match parse_uid(&uid) { Some(album) => match core.delete_album(&album, delete_photos) { Ok(()) => CtlResponse::Ok { message: "album deleted".to_string(), }, Err(e) => CtlResponse::error(e), }, None => CtlResponse::error(CoreError::invalid(format!("bad album uid: {uid}"))), }, Ok(CtlRequest::AddToAlbum { uid, photos }) => { album_change(&uid, &photos, |album, photos| { core.add_to_album(album, photos) }) } Ok(CtlRequest::RemoveFromAlbum { uid, photos }) => { album_change(&uid, &photos, |album, photos| { core.remove_from_album(album, photos) }) } Ok(CtlRequest::PhotoMonths { kind }) => match core.db.photos_months(kind) { Ok(months) => CtlResponse::PhotoMonths { months: months .into_iter() .map(|(year, month, count)| PhotoMonth { year, month, count }) .collect(), }, Err(e) => CtlResponse::error(e.into()), }, Ok(CtlRequest::PhotoThumbs { uids }) => { let parsed: Vec = uids.iter().filter_map(|u| parse_uid(u)).collect(); CtlResponse::Thumbs { items: core.photo_thumbs(&parsed), } } Ok(CtlRequest::TrashNodes { uids }) => match Core::parse_uids(&uids) { Ok(parsed) => match core.trash_photos(&parsed) { Ok((trashed, failed)) => CtlResponse::Trashed { trashed, failed: failed .into_iter() .map(|(uid, message)| TrashFailure { uid, message }) .collect(), }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::FileThumbs { items, generation }) => { match core.file_thumbs(&items, generation) { Some(items) => CtlResponse::Thumbs { items }, None => CtlResponse::FileThumbsStale, } } Ok(CtlRequest::ReserveFileThumbGeneration) => CtlResponse::FileThumbGeneration { generation: core.reserve_file_thumb_generation(), }, Ok(CtlRequest::CancelFileThumbs { generation }) => { core.cancel_file_thumbs(generation); CtlResponse::Ok { message: "thumbnail requests cancelled".to_string(), } } Ok(CtlRequest::StartThumbnailBuild { path }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.start_thumbnail_build(rel) { Ok(status) => CtlResponse::ThumbnailBuild { status }, Err(error) => CtlResponse::error(error), }, Err(error) => CtlResponse::error(error), }, Ok(CtlRequest::CancelThumbnailBuild) => CtlResponse::ThumbnailBuild { status: core.cancel_thumbnail_build(), }, Ok(CtlRequest::ThumbnailBuildStatus) => CtlResponse::ThumbnailBuild { status: core.thumbnail_build_status(), }, Ok(CtlRequest::PhotosRefreshStatus) => CtlResponse::PhotosRefresh { running: core.timeline_refreshing(), }, Ok(CtlRequest::OpenPhoto { uid }) => match parse_uid(&uid) { Some(u) => match core.open_photo(&u) { Ok(p) => CtlResponse::FilePath { path: p.display().to_string(), }, Err(e) => CtlResponse::error(e), }, None => CtlResponse::error(CoreError::invalid(format!("bad uid: {uid}"))), }, // The scan alone walks every archive's central directory, so even the // dry run goes to a background thread and is read back through // `ImportStatus` rather than answered inline. Ok(CtlRequest::ImportTakeout { archives, dry_run }) => { if archives.is_empty() { CtlResponse::error(CoreError::invalid("no Takeout archives given")) } else { let paths: Vec = archives.into_iter().map(PathBuf::from).collect(); let missing: Vec = paths .iter() .filter(|path| !path.is_file()) .map(|path| path.display().to_string()) .collect(); if !missing.is_empty() { CtlResponse::error(CoreError::not_found(format!( "cannot read: {}", missing.join(", ") ))) } else { let count = paths.len(); match core.spawn_takeout_import(paths, dry_run) { Ok(()) => CtlResponse::Ok { message: format!( "{} {count} archive(s)", if dry_run { "scanning" } else { "importing" } ), }, Err(e) => CtlResponse::error(e), } } } } Ok(CtlRequest::ImportStatus) => { let (running, summary) = crate::takeout::import_status(); CtlResponse::ImportStatus { running, summary } } Ok(CtlRequest::CancelImport) => { if crate::takeout::cancel_import() { CtlResponse::Ok { message: "stopping the import".to_string(), } } else { CtlResponse::error(CoreError::invalid("no import is running")) } } // The scan reads the whole persisted timeline and every candidate is a // download plus an upload, so this goes to a background thread and is // read back through `RedateStatus`. Ok(CtlRequest::RedatePhotos { dry_run, limit, range, }) => match core.spawn_redate(dry_run, limit, range) { Ok(()) => CtlResponse::Ok { message: if dry_run { "checking photo dates".to_string() } else { "fixing photo dates".to_string() }, }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RedateStatus) => { let (running, summary) = crate::redate::redate_status(); CtlResponse::RedateStatus { running, summary } } Ok(CtlRequest::CancelRedate) => { if crate::redate::cancel_redate() { CtlResponse::Ok { message: "stopping the re-date".to_string(), } } else { CtlResponse::error(CoreError::invalid("no photo re-date is running")) } } Ok(CtlRequest::UploadPhoto { name, media_type, source_path, capture_time, }) => { // 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")) } } } } } // A uid addresses the node wherever it lives; the path only works for // the primary mount's tree. Prefer the uid when the client sent one. Ok(CtlRequest::OpenFile { path, uid }) => { let opened = match uid.as_deref() { Some(uid) => core.open_file_uid(uid), None => rel_to_mount(mountpoint, &path).and_then(|rel| core.open_file(&rel)), }; match opened { Ok(p) => CtlResponse::FilePath { path: p.display().to_string(), }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::Search { query, limit, scope, }) => match core.search(&query, clamp_limit(limit), scope.as_deref()) { Ok(hits) => CtlResponse::SearchResults { hits }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::SearchLocal { query, limit }) => { match core.search_local(&query, clamp_limit(limit)) { Ok(hits) => CtlResponse::LocalResults { hits, indexing: core.indexing.load(Ordering::Relaxed), }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::SearchV2 { query, limit, filters, }) => match core.search_v2(&query, clamp_limit(limit), &filters) { Ok((drive_hits, local_hits)) => CtlResponse::SearchResultsV2 { drive_hits, local_hits, local_indexing: core.indexing.load(Ordering::Relaxed), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::Rename { path, new_name }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.rename(&rel, &new_name) { Ok(name) => { core.log_activity(ActivityKind::Rename, &name, format!("was {path}"), true); CtlResponse::Ok { message: format!("renamed to {name}"), } } Err(e) => { core.log_activity(ActivityKind::Rename, &path, &e, false); CtlResponse::error(e) } }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::Move { path, new_parent }) => { match ( rel_to_mount(mountpoint, &path), rel_to_mount(mountpoint, &new_parent), ) { (Ok(rel), Ok(parent_rel)) => match core.move_to(&rel, &parent_rel) { Ok(name) => { let dest = if new_parent.is_empty() { "My files".to_string() } else { new_parent.clone() }; core.log_activity(ActivityKind::Move, &name, format!("to {dest}"), true); CtlResponse::Ok { message: format!("moved {name}"), } } Err(e) => { core.log_activity(ActivityKind::Move, &path, &e, false); CtlResponse::error(e) } }, (Err(e), _) | (_, Err(e)) => CtlResponse::error(e), } } Ok(CtlRequest::Delete { path }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.delete(&rel) { Ok(name) => { core.log_activity(ActivityKind::Trash, &name, "", true); CtlResponse::Ok { message: format!("trashed {name}"), } } Err(e) => { core.log_activity(ActivityKind::Trash, &path, &e, false); CtlResponse::error(e) } }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::CreateFolder { parent, name }) => match rel_to_mount(mountpoint, &parent) { Ok(parent_rel) => match core.create_folder(&parent_rel, &name) { Ok(name) => { core.log_activity(ActivityKind::CreateFolder, &name, "", true); CtlResponse::Ok { message: format!("created folder {name}"), } } Err(e) => { core.log_activity(ActivityKind::CreateFolder, &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 // far outlasts the socket read timeout. Progress and completion are // observed through GetQueueStatus; the activity log gets the summary // when the whole batch finishes. Ok(parent_rel) => { let core = core.clone(); let paths: Vec = sources.into_iter().map(PathBuf::from).collect(); let n = paths.len(); std::thread::spawn(move || { let started = Instant::now(); match core.upload_paths(&parent_rel, &paths) { Ok(stats) => { // e.g. "Uploaded 700 files to Photos" — name the destination // so the log reads like a sentence rather than a bare count. let dest = parent_rel .file_name() .and_then(|n| n.to_str()) .unwrap_or("your Drive"); let target = format!( "{} to {dest}", count_noun(stats.uploaded, "file", "files") ); // Size · folders · elapsed, with a trailing failure count // when some files didn't make it. let mut parts = vec![human_bytes(stats.bytes)]; if stats.folders > 0 { parts.push(count_noun(stats.folders, "folder", "folders")); } parts.push(human_duration(started.elapsed())); if stats.failed > 0 { parts.push(format!("{} failed", stats.failed)); } core.log_activity( ActivityKind::Upload, target, parts.join(" · "), stats.failed == 0, ); } Err(e) => { warn!(error = %e, "bulk upload failed"); core.log_activity(ActivityKind::Upload, "bulk upload", &e, false); } } }); CtlResponse::Ok { message: format!("uploading {n} item(s)"), } } Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::Diagnostics) => CtlResponse::Diagnostics { diagnostics: core.diagnostics(), }, Ok(CtlRequest::ListTrash) => match core.list_trash() { Ok(entries) => CtlResponse::Entries { entries }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::Restore { uids }) => match core.restore(&uids) { Ok(n) => { core.log_activity(ActivityKind::Restore, format!("{n} item(s)"), "", true); CtlResponse::Ok { message: format!("restored {n} item(s)"), } } Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::DeleteForever { uids }) => match core.delete_forever(&uids) { Ok(n) => { core.log_activity( ActivityKind::DeleteForever, format!("{n} item(s)"), "", true, ); CtlResponse::Ok { message: format!("permanently deleted {n} item(s)"), } } Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::EmptyTrash) => match core.empty_trash() { Ok(n) => { let items = if n == 1 { "1 item".to_string() } else { format!("{n} items") }; core.log_activity(ActivityKind::EmptyTrash, items.clone(), "", true); CtlResponse::Ok { message: format!("emptied trash ({items})"), } } Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::PurgeCache) => { let freed = core.cache.clear_unpinned(); CtlResponse::Ok { message: format!( "purged {:.1} MiB of unpinned cache", freed as f64 / 1_048_576.0 ), } } Ok(CtlRequest::CacheInspect { deep }) => match core.db.stats() { Ok(stats) => { // The deep check reads every page — tens of seconds on a real // database, all of it holding the connection every FUSE handler // needs — so it runs on a thread and this request returns // whatever the last completed check found. `integrity_checked` // tells the caller which answer it is holding, since "no // problems found" and "never looked" are both an empty list; // `integrity_running` says a fresh answer is on its way. let (integrity_problems, integrity_checked, integrity_running) = if deep { core.start_or_read_integrity_check() } else { (Vec::new(), false, false) }; CtlResponse::CacheReport { integrity_running, schema_version: stats.schema_version, db_bytes: stats.total_bytes(), db_reclaimable_bytes: stats.reclaimable_bytes(), tables: stats.tables, cache_used: core.cache.usage(), cache_budget: core.cache.budget(), integrity_problems, integrity_checked, } } Err(e) => CtlResponse::error(CoreError::internal(format!("read db stats: {e:?}"))), }, // Ack and vacuum on a thread. `VACUUM` rewrites the whole database // under the connection mutex — minutes on a large one — and the // request's own socket timeout is 10 s, so running it inline both froze // the mount and lost its own answer. Progress shows up as a job in // `GetQueueStatus`; the outcome goes to the log. Ok(CtlRequest::CacheVacuum) => match core.start_vacuum() { Ok(()) => CtlResponse::Ok { message: "compacting the cache database in the background".into(), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::GetQueueStatus) => CtlResponse::Transfers { items: core.transfers.snapshot(), jobs: core.jobs_snapshot(), }, Ok(CtlRequest::SetCacheBudget { bytes }) => { core.cache.set_budget(bytes); // Persist so the next mount keeps the new cap. Best-effort: a config // write failure is reported but the live cap is already applied. match AppDirs::new().map(|dirs| { let mut cfg = dirs.load_config(); cfg.cache_budget = Some(bytes); dirs.save_config(&cfg) }) { Ok(Ok(())) => { core.touch_profile(); CtlResponse::Ok { message: format!("cache budget set to {bytes} bytes"), } } Ok(Err(e)) => CtlResponse::error(CoreError::internal(format!( "budget applied but config write failed: {e}" ))), Err(e) => CtlResponse::error(CoreError::internal(format!( "budget applied but config unavailable: {e}" ))), } } Ok(CtlRequest::SetBandwidthLimits { upload, download }) => { core.transfers.set_limits(upload, download); // Persisted like the cache budget: best-effort, the live caps are // already applied. match AppDirs::new().map(|dirs| { let mut cfg = dirs.load_config(); cfg.upload_limit = Some(upload); cfg.download_limit = Some(download); dirs.save_config(&cfg) }) { Ok(Ok(())) => CtlResponse::Ok { message: format!( "bandwidth limits set to {upload} B/s up, {download} B/s down" ), }, Ok(Err(e)) => CtlResponse::error(CoreError::internal(format!( "limits applied but config write failed: {e}" ))), Err(e) => CtlResponse::error(CoreError::internal(format!( "limits applied but config unavailable: {e}" ))), } } Ok(CtlRequest::ListDevices) => match core.list_devices() { Ok(items) => CtlResponse::Devices { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListLocations) => match core.list_locations() { Ok(items) => CtlResponse::Locations { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RenameDevice { uid, name }) => match core.rename_device(&uid, &name) { Ok(()) => CtlResponse::Ok { message: format!("renamed device to {name}"), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::DeleteDevice { uid }) => match core.delete_device(&uid) { Ok(()) => CtlResponse::Ok { message: "device deleted".to_string(), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::AdoptDevice { uid }) => match core.adopt_device(uid.as_deref()) { Ok(message) => CtlResponse::Ok { message }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::AddSyncFolder { local_path }) => { // Registering the device and uploading a folder tree far outlasts the // socket read timeout, so ack immediately and work on a background // thread. The folder appears in ListSyncFolders once the row lands; // completion (and any failures) go to the activity log. let core = core.clone(); let path = PathBuf::from(&local_path); std::thread::spawn(move || { let started = Instant::now(); match core.add_sync_folder(&path) { Ok(folder) => { core.touch_profile(); let name = Path::new(&folder.local_path) .file_name() .and_then(|n| n.to_str()) .unwrap_or(&folder.local_path) .to_string(); core.log_activity( ActivityKind::Upload, format!("synced {name}"), human_duration(started.elapsed()), folder.state != "error", ); } Err(e) => { warn!(error = %e, "add sync folder failed"); core.log_activity(ActivityKind::Upload, "add sync folder", &e, false); } } }); CtlResponse::Ok { message: format!("adding {local_path}"), } } Ok(CtlRequest::ListSyncFolders) => match core.list_sync_folders() { Ok(items) => CtlResponse::SyncFolders { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RemoveSyncFolder { id, delete_remote }) => { match core.remove_sync_folder(id, delete_remote) { Ok(()) => { core.touch_profile(); CtlResponse::Ok { message: "removed synced folder".to_string(), } } Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::SetSyncFolderMode { id, mode }) => { match core.request_sync_folder_mode(id, &mode) { Ok(message) => { core.touch_profile(); core.log_activity(ActivityKind::Upload, &message, "", true); CtlResponse::Ok { message } } Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::SetSyncFolderPaused { id, paused }) => { match core.set_sync_folder_paused(id, paused) { Ok(()) => CtlResponse::Ok { message: match paused { true => format!("folder {id} paused"), false => format!("folder {id} resumed"), }, }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::SyncNow { id }) => { core.sync_now(id); CtlResponse::Ok { message: match id { Some(id) => format!("reconciling folder {id}"), None => "reconciling all folders".to_string(), }, } } Ok(CtlRequest::SetSyncPaused { paused, until }) => { match core.set_sync_paused(paused, until) { Ok(()) => CtlResponse::Ok { message: match (paused, until) { (false, _) => "sync resumed".to_string(), (true, None) => "sync paused until resumed".to_string(), (true, Some(_)) => "sync paused".to_string(), }, }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::ListConflicts) => match core.list_conflicts() { Ok(items) => CtlResponse::Conflicts { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ResolveConflict { path, keep }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.resolve_conflict(&rel, &keep) { Ok(message) => CtlResponse::Ok { message }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListPendingOps) => match core.pending_op_infos() { Ok(items) => CtlResponse::PendingOps { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RetryPendingOp { id }) => match core.retry_pending_ops(id) { Ok(0) if id.is_some() => CtlResponse::Ok { message: "nothing to retry: the operation is already due or waiting for a rename" .to_string(), }, Ok(moved) => CtlResponse::Ok { message: format!("retrying {moved} queued operation(s) now"), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListRestorableFolders) => match core.list_restorable_folders(None) { Ok(items) => CtlResponse::RestorableFolders { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListDeviceRestorableFolders { device }) => { match core.list_restorable_folders(Some(&device)) { Ok(items) => CtlResponse::RestorableFolders { items }, Err(e) => CtlResponse::error(e), } } Ok( request @ (CtlRequest::RestoreSyncFolders { .. } | CtlRequest::RestoreDeviceFolders { .. }), ) => { let (device, items) = match request { CtlRequest::RestoreDeviceFolders { device, items } => (Some(device), items), CtlRequest::RestoreSyncFolders { items } => (None, items), _ => unreachable!("matched above"), }; // Like AddSyncFolder: each restored folder downloads a whole tree, // which outlasts the socket timeout. Ack, then work. let core = core.clone(); std::thread::spawn(move || { match core.restore_sync_folders(&items, device.as_deref()) { Ok(message) => { core.touch_profile(); core.log_activity(ActivityKind::Download, &message, "", true); } Err(e) => { warn!(error = %e, "restore failed"); core.log_activity(ActivityKind::Download, "restore folders", &e, false); } } }); CtlResponse::Ok { message: "restoring folders".to_string(), } } Ok(CtlRequest::ShareNode { path, emails, role, message, }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.share_node(&rel, &emails, &role, message.as_deref()) { Ok((proton, external)) => { core.log_activity( ActivityKind::Share, &path, format!("{} recipient(s) as {role}", proton + external), true, ); CtlResponse::Ok { message: format!("invited {proton} Proton and {external} external user(s)"), } } Err(e) => { core.log_activity(ActivityKind::Share, &path, &e, false); CtlResponse::error(e) } }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ShareNodeByUid { uid, emails, role, message, }) => match core.share_node_by_uid(&uid, &emails, &role, message.as_deref()) { Ok((proton, external)) => { core.log_activity( ActivityKind::Share, &uid, format!("{} recipient(s) as {role}", proton + external), true, ); CtlResponse::Ok { message: format!("invited {proton} Proton and {external} external user(s)"), } } Err(e) => { core.log_activity(ActivityKind::Share, &uid, &e, false); CtlResponse::error(e) } }, Ok(CtlRequest::ListShare { path }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.list_share(&rel) { Ok((entries, link)) => CtlResponse::Share { entries, link }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListShareByUid { uid }) => match core.list_share_by_uid(&uid) { Ok((entries, link)) => CtlResponse::Share { entries, link }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::UpdateShareRole { path, id, kind, role, }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.update_share_role(&rel, &id, kind, &role) { Ok(()) => CtlResponse::Ok { message: format!("role updated to {role}"), }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::UpdateShareRoleByUid { uid, id, kind, role, }) => match core.update_share_role_by_uid(&uid, &id, kind, &role) { Ok(()) => CtlResponse::Ok { message: format!("role updated to {role}"), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RemoveShareEntry { path, id, kind }) => { match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.remove_share_entry(&rel, &id, kind) { Ok(()) => { core.log_activity(ActivityKind::Unshare, &path, "access removed", true); CtlResponse::Ok { message: "removed".to_string(), } } Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::RemoveShareEntryByUid { uid, id, kind }) => { match core.remove_share_entry_by_uid(&uid, &id, kind) { Ok(()) => { core.log_activity(ActivityKind::Unshare, &uid, "access removed", true); CtlResponse::Ok { message: "removed".to_string(), } } Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::CreatePublicLink { path, role, password, expires, }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.create_public_link(&rel, &role, password.as_deref(), expires) { Ok(link) => { core.log_activity(ActivityKind::PublicLink, &path, "link created", true); CtlResponse::PublicLink { link } } Err(e) => { core.log_activity(ActivityKind::PublicLink, &path, &e, false); CtlResponse::error(e) } }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::CreatePublicLinkByUid { uid, role, password, expires, }) => match core.create_public_link_by_uid(&uid, &role, password.as_deref(), expires) { Ok(link) => { core.log_activity(ActivityKind::PublicLink, &uid, "link created", true); CtlResponse::PublicLink { link } } Err(e) => { core.log_activity(ActivityKind::PublicLink, &uid, &e, false); CtlResponse::error(e) } }, Ok(CtlRequest::RemovePublicLink { path, id }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.remove_public_link(&rel, &id) { Ok(()) => { core.log_activity(ActivityKind::Unshare, &path, "link removed", true); CtlResponse::Ok { message: "public link removed".to_string(), } } Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RemovePublicLinkByUid { uid, id }) => { match core.remove_public_link_by_uid(&uid, &id) { Ok(()) => { core.log_activity(ActivityKind::Unshare, &uid, "link removed", true); CtlResponse::Ok { message: "public link removed".to_string(), } } Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::StopSharing { path }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => stop_sharing_reply(core, &path, core.stop_sharing(&rel)), Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::StopSharingByUid { uid }) => { stop_sharing_reply(core, &uid, core.stop_sharing_by_uid(&uid)) } Ok(CtlRequest::ListRevisions { path }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.list_revisions(&rel) { Ok(items) => CtlResponse::Revisions { items }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListRevisionsByUid { uid }) => match core.list_revisions_by_uid(&uid) { Ok(items) => CtlResponse::Revisions { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RestoreRevision { path, revision_id }) => { match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.restore_revision(&rel, &revision_id) { Ok(()) => CtlResponse::Ok { message: format!("restoring version {revision_id} of {path}"), }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::RestoreRevisionByUid { uid, revision_id }) => { match core.restore_revision_by_uid(&uid, &revision_id) { Ok(()) => CtlResponse::Ok { message: format!("restoring version {revision_id}"), }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::DeleteRevision { path, revision_id }) => { match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.delete_revision(&rel, &revision_id) { Ok(()) => CtlResponse::Ok { message: format!("deleted version {revision_id} of {path}"), }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::DeleteRevisionByUid { uid, revision_id }) => { match core.delete_revision_by_uid(&uid, &revision_id) { Ok(()) => CtlResponse::Ok { message: format!("deleted version {revision_id}"), }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::SaveRevisionAs { path, revision_id, dest, }) => match rel_to_mount(mountpoint, &path) { Ok(rel) => match core.save_revision_as(&rel, &revision_id, Path::new(&dest)) { Ok(written) => CtlResponse::Ok { message: format!("wrote {} to {dest}", human_bytes(written)), }, Err(e) => CtlResponse::error(e), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::SaveRevisionAsByUid { uid, revision_id, dest, }) => match core.save_revision_as_by_uid(&uid, &revision_id, Path::new(&dest)) { Ok(written) => CtlResponse::Ok { message: format!("wrote {} to {dest}", human_bytes(written)), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListSharedWithMe) => match core.list_shared_with_me() { Ok(entries) => CtlResponse::Entries { entries }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::AccountQuota) => match core.account_quota() { Ok((max_space, used_space)) => CtlResponse::AccountQuota { max_space, used_space, }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListSharedFolder { uid }) => match core.list_shared_folder(&uid) { Ok(entries) => CtlResponse::Entries { entries }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::OpenSharedFile { uid }) => match core.open_shared_file(&uid) { Ok(p) => CtlResponse::FilePath { path: p.display().to_string(), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::LeaveShared { uid }) => match core.leave_shared(&uid) { Ok(()) => { core.log_activity(ActivityKind::Unshare, "shared item", "left", true); CtlResponse::Ok { message: "left shared item".to_string(), } } Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListInvitations) => match core.list_invitations() { Ok(items) => CtlResponse::Invitations { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::AcceptInvitation { id }) => match core.accept_invitation(&id) { Ok(()) => CtlResponse::Ok { message: "invitation accepted".to_string(), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::RejectInvitation { id }) => match core.reject_invitation(&id) { Ok(()) => CtlResponse::Ok { message: "invitation rejected".to_string(), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListBookmarks) => match core.list_bookmarks() { Ok(items) => CtlResponse::Bookmarks { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::CreateBookmark { url, password }) => { match core.create_bookmark(&url, password.as_deref()) { Ok(()) => CtlResponse::Ok { message: "bookmark saved".to_string(), }, Err(e) => CtlResponse::error(e), } } Ok(CtlRequest::DeleteBookmark { token }) => match core.delete_bookmark(&token) { Ok(()) => CtlResponse::Ok { message: "bookmark removed".to_string(), }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListSharedByMe) => match core.list_shared_by_me() { Ok(items) => CtlResponse::SharedByMe { items }, Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListActivity { limit }) => CtlResponse::Activity { items: core.list_activity(clamp_limit(limit)), }, // The request did not parse: the caller sent something malformed. Err(e) => CtlResponse::error(CoreError::invalid(format!("bad request: {e}"))), }; let mut out = match serde_json::to_vec(&response) { Ok(v) => v, Err(e) => { warn!(error = %e, "control: serialize response failed"); return; } }; out.push(b'\n'); let mut stream = stream; let _ = stream.write_all(&out); } /// Listen on the control socket, serving one command per connection, each on its /// own thread. Runs on its own thread; returns only if the listener itself fails. /// /// Concurrent rather than serial because requests differ wildly in cost: an /// `OpenPhoto` downloads a whole photo, and while it ran the accept loop used to /// stall every other caller behind it — the GUI's 2s status poll, and the /// thumbnail batches the gallery needs to paint. [`Core`] is a bundle of handles /// (`Arc`/`Clone`), so each connection gets its own copy of it. pub(crate) fn run_control_socket( core: Core, username: String, mountpoint: PathBuf, listener: UnixListener, ) { info!("control socket listening"); for conn in listener.incoming() { // Checked on every accept, including the deliberate poke teardown sends // to wake this thread out of a blocking accept (bugs.md B44). if core.shutdown.is_stopping() { info!("control socket closing"); return; } match conn { Ok(stream) => { let Some(permit) = ControlHandlerPermit::acquire() else { warn!( limit = MAX_CONTROL_HANDLERS, "control: rejecting connection at handler limit" ); continue; }; let core = core.clone(); let username = username.clone(); let mountpoint = mountpoint.clone(); if let Err(e) = std::thread::Builder::new() .name("pdfs-control".into()) .spawn(move || { let _permit = permit; handle_control_conn(&core, &username, &mountpoint, stream); }) { warn!(error = %e, "control: spawn handler failed"); } } Err(e) => { warn!(error = %e, "control: accept failed"); } } } } #[cfg(test)] mod request_limit_tests { use super::*; use std::io::Cursor; #[test] fn accepts_one_bounded_newline_terminated_request() { let mut input = Cursor::new(b"{\"Status\":null}\n".to_vec()); assert_eq!( read_request_line(&mut input).unwrap().as_deref(), Some("{\"Status\":null}\n") ); } #[test] fn rejects_a_request_without_a_terminating_newline() { let mut input = Cursor::new(b"{\"Status\":null}".to_vec()); assert_eq!( read_request_line(&mut input).unwrap_err().kind(), std::io::ErrorKind::InvalidData ); } #[test] fn rejects_a_request_larger_than_the_limit() { let mut bytes = vec![b'x'; MAX_CONTROL_REQUEST_BYTES as usize + 1]; bytes.push(b'\n'); let mut input = Cursor::new(bytes); assert_eq!( read_request_line(&mut input).unwrap_err().kind(), std::io::ErrorKind::InvalidData ); } #[test] fn accepts_paths_inside_the_mountpoint() { let mountpoint = Path::new("/mnt/proton"); assert_eq!( rel_to_mount(mountpoint, "Photos/2026").unwrap(), Path::new("Photos/2026") ); assert_eq!( rel_to_mount(mountpoint, "/mnt/proton/Photos/2026").unwrap(), Path::new("Photos/2026") ); assert_eq!( rel_to_mount(mountpoint, "/mnt/proton").unwrap(), Path::new("") ); } #[test] fn rejects_paths_that_escape_the_mountpoint() { let mountpoint = Path::new("/mnt/proton"); assert!(rel_to_mount(mountpoint, "../escape").is_err()); assert!(rel_to_mount(mountpoint, "Photos/../../escape").is_err()); assert!(rel_to_mount(mountpoint, "/mnt/elsewhere").is_err()); } } #[cfg(test)] mod tests { use super::*; /// Two refreshes of the same scope are one piece of work; two refreshes of /// different scopes are not. #[test] fn a_scope_refreshes_once_at_a_time() { let photos = RefreshScope::Photos { full: false }; let held = RefreshGuard::acquire(&photos).expect("first refresh runs"); assert!(RefreshGuard::acquire(&photos).is_none()); // A full refresh forgets what the daemon knows, so it is other work. assert!(RefreshGuard::acquire(&RefreshScope::Photos { full: true }).is_some()); assert!( RefreshGuard::acquire(&RefreshScope::Dir { path: "a".to_string() }) .is_some() ); drop(held); assert!(RefreshGuard::acquire(&photos).is_some()); } }