diff --git a/Cargo.lock b/Cargo.lock index 9e155cf..0d9fe27 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3529,6 +3529,7 @@ dependencies = [ "sha1 0.11.0", "tokio", "tracing", + "tracing-subscriber", "zip", ] diff --git a/crates/pdfs-fuse/Cargo.toml b/crates/pdfs-fuse/Cargo.toml index bf923ba..7a44d36 100644 --- a/crates/pdfs-fuse/Cargo.toml +++ b/crates/pdfs-fuse/Cargo.toml @@ -25,3 +25,6 @@ base64.workspace = true image.workspace = true notify = "8" parking_lot.workspace = true + +[dev-dependencies] +tracing-subscriber.workspace = true diff --git a/crates/pdfs-fuse/src/drive.rs b/crates/pdfs-fuse/src/drive.rs index 6eaff84..46e64c0 100644 --- a/crates/pdfs-fuse/src/drive.rs +++ b/crates/pdfs-fuse/src/drive.rs @@ -22,6 +22,7 @@ use std::sync::Arc; use async_trait::async_trait; use futures::StreamExt as _; use futures::stream::BoxStream; +use proton_drive_rs::proton_sdk::account::Quota; use proton_drive_rs::proton_sdk::error::Result; use proton_drive_rs::proton_sdk::ids::{DriveEventId, NodeUid}; use proton_drive_rs::{ @@ -149,6 +150,9 @@ pub(crate) trait DriveApi: Send + Sync { fn delete_nodes_streaming<'a>(&'a self, uids: &[NodeUid]) -> OutcomeStream<'a>; async fn enumerate_trash_node_uids(&self) -> Result>; + + /// The account's storage, across every Proton product. + async fn quota(&self) -> Result; } /// The generic signatures of the client, over the object-safe ones above. @@ -455,4 +459,8 @@ impl DriveApi for ProtonDriveClient { async fn enumerate_trash_node_uids(&self) -> Result> { ProtonDriveClient::enumerate_trash_node_uids(self).await } + + async fn quota(&self) -> Result { + ProtonDriveClient::quota(self).await + } } diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 1de7b7f..596c294 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -207,11 +207,11 @@ const ROOT_UID_KEY: &str = "root_uid"; /// means an empty tree, and mounting that would show the user an empty Drive /// rather than an honest error. fn fetch_or_recall_root( - client: &ProtonDriveClient, + drive: &dyn DriveApi, rt: &tokio::runtime::Handle, db: &Db, ) -> std::io::Result<(Node, bool)> { - let err = match rt.block_on(client.get_my_files_folder()) { + let err = match rt.block_on(drive.get_my_files_folder()) { Ok(root) => { if let Err(e) = db.set_state_str(ROOT_UID_KEY, &root.uid.to_string()) { warn!(error = %e, "persist root uid failed"); @@ -339,6 +339,45 @@ pub fn mount( control_socket: &Path, db: Arc, options: MountOptions, +) -> std::io::Result { + let drive = Arc::new(client.clone()); + mount_with( + client, + drive, + rt, + mountpoint, + cache, + control_socket, + db, + options, + Host::Daemon, + ) +} + +/// What a mount runs in. +pub(crate) enum Host { + /// The daemon: stopped by SIGTERM or SIGINT, and indexing the user's home + /// for the launcher prompt. + Daemon, + /// One of several simulated daemons in a test process (`sim`): stopped by + /// a message on this channel, and with no home to index. + #[cfg(test)] + Simulation(std::sync::mpsc::Receiver<()>), +} + +/// [`mount`], reaching Drive through `drive` and hosted by `host`; `client` +/// serves only what [`DriveApi`] does not cover. +#[allow(clippy::too_many_arguments)] +pub(crate) fn mount_with( + client: ProtonDriveClient, + drive: Arc, + rt: tokio::runtime::Handle, + mountpoint: &Path, + cache: ContentCache, + control_socket: &Path, + db: Arc, + options: MountOptions, + host: Host, ) -> std::io::Result { let MountOptions { username, @@ -349,7 +388,7 @@ pub fn mount( // Start the uptime clock here rather than at the first `pdfs diagnostics`, // so the age it reports is the daemon's own. super::diagnostics::start_clock(); - let (root, online) = fetch_or_recall_root(&client, &rt, &db)?; + let (root, online) = fetch_or_recall_root(drive.as_ref(), &rt, &db)?; let scope = root.tree_event_scope_id(); db.mount_upsert_my_files(&mountpoint.to_string_lossy(), &root.uid.to_string(), None) .map_err(|error| std::io::Error::other(format!("project My Files mount: {error}")))?; @@ -364,8 +403,8 @@ pub fn mount( let paused_until = pause::load_paused_until(&db); let core = Core { - drive: Arc::new(client.clone()), - client: client.clone(), + drive, + client, rt: rt.clone(), maintenance: Arc::new(Mutex::new(Default::default())), primary_root_uid: root.uid.clone(), @@ -502,7 +541,7 @@ pub fn mount( // Keep the launcher prompt's "This computer" index fresh. Its own thread: // the walk is I/O-heavy and must never sit in front of a FUSE callback. - { + if matches!(host, Host::Daemon) { let db = core.db.clone(); let indexing = core.indexing.clone(); let transfers = core.transfers.clone(); @@ -645,30 +684,11 @@ pub fn mount( // are delivered onto the async runtime; bridge them onto a sync channel so // the loop below can react without blocking a worker thread. A bounded // channel of 1 is enough — we only need to know that *a* stop arrived. - let (sig_tx, sig_rx) = std::sync::mpsc::sync_channel::<()>(1); - rt.spawn(async move { - let mut sigterm = - match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { - Ok(s) => s, - Err(e) => { - warn!(error = %e, "install SIGTERM handler failed"); - return; - } - }; - let mut sigint = - match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) { - Ok(s) => s, - Err(e) => { - warn!(error = %e, "install SIGINT handler failed"); - return; - } - }; - tokio::select! { - _ = sigterm.recv() => info!("received SIGTERM"), - _ = sigint.recv() => info!("received SIGINT"), - } - let _ = sig_tx.try_send(()); - }); + let sig_rx = match host { + Host::Daemon => signal_stop(&rt), + #[cfg(test)] + Host::Simulation(rx) => rx, + }; // Wait for whichever happens first: a stop signal (→ we unmount ourselves // via the lazy MNT_DETACH path, clean even mid-download), or the kernel @@ -725,6 +745,35 @@ pub fn mount( Ok(outcome) } +/// A channel that receives once SIGTERM or SIGINT arrives. +fn signal_stop(rt: &tokio::runtime::Handle) -> std::sync::mpsc::Receiver<()> { + let (sig_tx, sig_rx) = std::sync::mpsc::sync_channel::<()>(1); + rt.spawn(async move { + let mut sigterm = + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { + Ok(s) => s, + Err(e) => { + warn!(error = %e, "install SIGTERM handler failed"); + return; + } + }; + let mut sigint = + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) { + Ok(s) => s, + Err(e) => { + warn!(error = %e, "install SIGINT handler failed"); + return; + } + }; + tokio::select! { + _ = sigterm.recv() => info!("received SIGTERM"), + _ = sigint.recv() => info!("received SIGINT"), + } + let _ = sig_tx.try_send(()); + }); + sig_rx +} + /// How long teardown waits for the FUSE worker pool before proceeding without /// it. Long enough for an in-flight block to finish, short enough that /// `systemctl --user stop` still returns inside `TimeoutStopSec`. diff --git a/crates/pdfs-fuse/src/sharing.rs b/crates/pdfs-fuse/src/sharing.rs index c22ff0b..7a16eb2 100644 --- a/crates/pdfs-fuse/src/sharing.rs +++ b/crates/pdfs-fuse/src/sharing.rs @@ -610,7 +610,7 @@ impl Core { /// Proton products (not Drive-only). A remote round-trip; nothing here is /// cached. pub(crate) fn account_quota(&self) -> CoreResult<(i64, i64)> { - let q = self.block_on_bounded(self.client.quota()).map_err(|e| { + let q = self.block_on_bounded(self.drive.quota()).map_err(|e| { self.lost_link(&e, "account quota"); CoreError::from_api(&e, "account quota") })?; diff --git a/crates/pdfs-fuse/src/sim/daemon.rs b/crates/pdfs-fuse/src/sim/daemon.rs new file mode 100644 index 0000000..0f84204 --- /dev/null +++ b/crates/pdfs-fuse/src/sim/daemon.rs @@ -0,0 +1,239 @@ +//! One simulated daemon: a real FUSE mount on its own state directory, whose +//! Drive is a [`FakeClient`]. +//! +//! Everything the daemon does runs as in production (drain workers, sync +//! engine, event feed, online probe, control socket) except what the host +//! changes (`mount::Host::Simulation`): a channel stops it instead of a signal, +//! and it does not index the developer's home. + +use std::io; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::mpsc::SyncSender; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +use pdfs_core::cache::ContentCache; +use pdfs_core::config::AppConfig; +use pdfs_core::db::Db; +use proton_drive_rs::ProtonDriveClient; +use proton_drive_rs::proton_sdk::config::ProtonClientConfiguration; +use proton_drive_rs::proton_sdk::session::{PasswordMode, ProtonApiSession, ResumeParameters}; + +use super::fake_drive::FakeClient; +use crate::mount::{Host, MountOptions, MountOutcome, mount_with}; + +/// How long a mount may take to come up. +const MOUNT_DEADLINE: Duration = Duration::from_secs(20); + +pub(crate) struct Daemon { + pub(crate) client: FakeClient, + pub(crate) mountpoint: PathBuf, + stop: SyncSender<()>, + mount: Option>>, + rt: Option, +} + +impl Daemon { + /// Mount a daemon on the state under `dir`, creating it on the first start. + /// A later start on the same `dir` is a restart: it finds the queue, the + /// cache and the database the last run left. + pub(crate) fn start(dir: &Path, client: FakeClient) -> io::Result { + log_to_test_output(); + let mountpoint = dir.join("mnt"); + std::fs::create_dir_all(&mountpoint)?; + let db = Arc::new(Db::open(&dir.join("pdfs.db")).map_err(io::Error::other)?); + let cache = ContentCache::open(dir.join("cache"), dir.join("pins.json"), 0, db.clone()) + .map_err(io::Error::other)?; + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build()?; + let (stop, stopped) = std::sync::mpsc::sync_channel(1); + let mount = { + let drive = Arc::new(client.clone()); + let handle = rt.handle().clone(); + let mountpoint = mountpoint.clone(); + let socket = dir.join("control.sock"); + std::thread::Builder::new() + .name("sim-mount".into()) + .spawn(move || { + mount_with( + offline_client(), + drive, + handle, + &mountpoint, + cache, + &socket, + db, + MountOptions { + username: "sim".into(), + sweep_mode: AppConfig::default().resolved_conflict_sweep(), + upload_limit: 0, + download_limit: 0, + }, + Host::Simulation(stopped), + ) + })? + }; + let deadline = Instant::now() + MOUNT_DEADLINE; + while !is_mounted(&mountpoint) { + if mount.is_finished() { + return Err(match mount.join() { + Ok(Err(error)) => error, + _ => io::Error::other("mount ended before it came up"), + }); + } + if Instant::now() > deadline { + return Err(io::Error::other("mount did not come up")); + } + std::thread::sleep(Duration::from_millis(20)); + } + Ok(Self { + client, + mountpoint, + stop, + mount: Some(mount), + rt: Some(rt), + }) + } + + /// Stop the daemon the way `systemctl --user stop` does. + pub(crate) fn stop(mut self) -> io::Result { + self.shut_down() + } + + fn shut_down(&mut self) -> io::Result { + let Some(mount) = self.mount.take() else { + return Ok(MountOutcome::Shutdown); + }; + let _ = self.stop.try_send(()); + let outcome = mount + .join() + .unwrap_or_else(|_| Err(io::Error::other("mount thread panicked"))); + if let Some(rt) = self.rt.take() { + rt.shutdown_timeout(Duration::from_secs(5)); + } + outcome + } +} + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.shut_down(); + } +} + +/// Whether a file system is mounted at `path`: it sits on another device than +/// its parent. +fn is_mounted(path: &Path) -> bool { + use std::os::unix::fs::MetadataExt; + let device = |path: &Path| std::fs::metadata(path).map(|meta| meta.dev()).ok(); + let parent = path.parent().and_then(device); + device(path).is_some_and(|dev| Some(dev) != parent) +} + +/// The daemons' logs, filtered by `RUST_LOG` and shown with `--nocapture`. +fn log_to_test_output() { + let _ = tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_test_writer() + .try_init(); +} + +/// A client for what `DriveApi` does not cover (sharing, Photos, devices), +/// pointed at a closed local port so none of it leaves the machine. +fn offline_client() -> ProtonDriveClient { + let config = ProtonClientConfiguration::new("external-drive-pdfs-sim@0.0.0-dev") + .with_base_url("http://127.0.0.1:9"); + let session = ProtonApiSession::resume( + config, + ResumeParameters { + session_id: "sim".to_owned().into(), + username: "sim".into(), + user_id: "sim".to_owned().into(), + access_token: String::new(), + refresh_token: String::new(), + scopes: Vec::new(), + is_waiting_for_second_factor_code: false, + password_mode: PasswordMode::Single, + }, + ) + .expect("an offline session needs no network"); + ProtonDriveClient::new(&session, Vec::new()) +} + +/// A fresh directory for one simulation, under the system temp dir. +pub(crate) fn scratch(label: &str) -> PathBuf { + use std::sync::atomic::{AtomicU64, Ordering}; + static N: AtomicU64 = AtomicU64::new(0); + let dir = std::env::temp_dir().join(format!( + "pdfs-sim-{label}-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create a simulation directory"); + dir +} + +/// Wait up to `limit` for `done`, polling. +pub(crate) fn wait_until(limit: Duration, mut done: impl FnMut() -> bool) -> bool { + let deadline = Instant::now() + limit; + loop { + if done() { + return true; + } + if Instant::now() > deadline { + return false; + } + std::thread::sleep(Duration::from_millis(50)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sim::fake_drive::{Entry, FakeDrive, Faults}; + + #[test] + #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"] + fn a_daemon_on_the_fake_drive_syncs_both_ways() { + let drive = FakeDrive::new(); + let dir = scratch("smoke"); + let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap(); + + std::fs::write(daemon.mountpoint.join("hello.txt"), b"hi").unwrap(); + assert!( + wait_until(Duration::from_secs(30), || { + drive.tree().get("hello.txt") == Some(&Entry::File(Arc::new(b"hi".to_vec()))) + }), + "a local write landed on Drive: {:?}", + drive.tree() + ); + + drive.device().write("phone.txt", b"from the phone"); + let remote = daemon.mountpoint.join("phone.txt"); + assert!( + wait_until(Duration::from_secs(30), || { + std::fs::read(&remote).ok().as_deref() == Some(b"from the phone".as_slice()) + }), + "a remote write reached the mount" + ); + + daemon.client.set_online(false); + std::fs::write(daemon.mountpoint.join("offline.txt"), b"queued").unwrap(); + std::thread::sleep(Duration::from_secs(2)); + assert!(!drive.tree().contains_key("offline.txt")); + daemon.client.set_online(true); + assert!( + wait_until(Duration::from_secs(60), || drive + .tree() + .contains_key("offline.txt")), + "a write made offline landed once the link came back" + ); + + assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown))); + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/crates/pdfs-fuse/src/sim/fake_drive.rs b/crates/pdfs-fuse/src/sim/fake_drive.rs index 7e9b352..2c98cfb 100644 --- a/crates/pdfs-fuse/src/sim/fake_drive.rs +++ b/crates/pdfs-fuse/src/sim/fake_drive.rs @@ -23,6 +23,8 @@ //! then loses its reply (B107, B121's half state). //! - **Latency** per request, from a range ([`Faults::lan`], [`Faults::wifi`]). //! - **Non-uniform blocks** ([`FakeDrive::set_block_pattern`]; B84, B85, B87). +//! - **A full account** ([`FakeDrive::set_quota`]) refusing uploads with +//! `InsufficientQuota`. use std::collections::{BTreeMap, HashMap}; use std::io::{Read, Write}; @@ -34,6 +36,7 @@ use async_trait::async_trait; use futures::FutureExt as _; use futures::StreamExt as _; use parking_lot::Mutex; +use proton_drive_rs::proton_sdk::account::Quota; use proton_drive_rs::proton_sdk::api::ResponseCode; use proton_drive_rs::proton_sdk::error::{ProtonApiError, ProtonError, Result}; use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId}; @@ -91,6 +94,8 @@ struct Server { events: Vec, /// Plaintext block sizes of new content, cycled; the last one repeats. block_pattern: Vec, + /// The account's storage limit in bytes. + max_space: i64, } impl Server { @@ -186,6 +191,27 @@ impl Server { }); } + /// Bytes stored: every revision of every node not deleted, trashed or not, + /// as Drive counts them. + fn used_space(&self) -> i64 { + self.nodes + .values() + .filter_map(|versions| versions.last()?.1.as_ref()?.file.as_ref()) + .flat_map(|file| &file.revisions) + .map(|(_, content)| content.len() as i64) + .sum() + } + + fn make_room(&self, bytes: usize) -> Result<()> { + if self.used_space() + bytes as i64 > self.max_space { + return Err(api( + ResponseCode::InsufficientQuota, + "storage quota exceeded", + )); + } + Ok(()) + } + fn new_file( &mut self, parent: &NodeUid, @@ -203,6 +229,7 @@ impl Server { "a file or folder with that name already exists", )); } + self.make_room(content.len())?; let uid = self.mint_uid(); let revision = self.mint_revision(); let content = Arc::new(content); @@ -252,6 +279,7 @@ impl FakeDrive { next_revision: 0, events: Vec::new(), block_pattern: vec![pdfs_core::cache::BLOCK_SIZE], + max_space: 1 << 40, }; let now = now_secs(); server.nodes.insert( @@ -292,6 +320,12 @@ impl FakeDrive { self.server.lock().block_pattern = pattern; } + /// The account's storage limit; a limit below what is stored refuses every + /// upload until something is deleted. + pub(crate) fn set_quota(&self, max_space: i64) { + self.server.lock().max_space = max_space; + } + pub(crate) fn root(&self) -> NodeUid { self.server.lock().root.clone() } @@ -954,6 +988,7 @@ impl DriveApi for FakeClient { if remote.file.is_none() { return Err(api(ResponseCode::InvalidRequirements, "not a file")); } + server.make_room(content.len())?; let revision = server.mint_revision(); let blocks = server.blocks_of(content.len() as u64); let file = remote.file.as_mut().expect("checked above"); @@ -1081,6 +1116,16 @@ impl DriveApi for FakeClient { outcome_stream(self.batch(uids, delete_one)) } + async fn quota(&self) -> Result { + self.request(false, |server| { + Ok(Quota { + max_space: server.max_space, + used_space: server.used_space(), + }) + }) + .await + } + async fn enumerate_trash_node_uids(&self) -> Result> { self.request(false, |server| { Ok(server @@ -1447,6 +1492,30 @@ mod tests { }); } + #[test] + fn an_exhausted_quota_refuses_uploads_until_space_is_freed() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client(1, Faults::none()); + let root = drive.root(); + let big = client.upload_file(&root, "big", "", &[0; 8]).await.unwrap(); + drive.set_quota(10); + let full = client + .upload_file(&root, "more", "", &[0; 4]) + .await + .unwrap_err(); + assert_eq!(code(&full), Some(ResponseCode::InsufficientQuota)); + assert_eq!(client.quota().await.unwrap().used_space, 8); + let mut deleted = client.delete_nodes_streaming(std::slice::from_ref(&big)); + assert!(deleted.next().await.unwrap().unwrap().1.is_ok()); + drop(deleted); + client + .upload_file(&root, "more", "", &[0; 4]) + .await + .unwrap(); + }); + } + #[test] fn content_is_split_by_the_block_pattern() { rt().block_on(async { diff --git a/crates/pdfs-fuse/src/sim/mod.rs b/crates/pdfs-fuse/src/sim/mod.rs index 89e635b..3224e25 100644 --- a/crates/pdfs-fuse/src/sim/mod.rs +++ b/crates/pdfs-fuse/src/sim/mod.rs @@ -1,5 +1,6 @@ //! The simulation tests of `docs/MILESTONE-3.0.0.md` §8: a fake Drive with //! the faults the real one has, and the tools to drive daemons against it. +pub(crate) mod daemon; pub(crate) mod fake_drive; pub(crate) mod rng;