test(fuse): run real FUSE daemons against the FakeDrive

mount() becomes a thin wrapper over mount_with(), which takes the
DriveApi to reach Drive through and a Host. The daemon's Host is
unchanged: stopped by SIGTERM or SIGINT, installed at the same point as
before, and indexing the user's home. A simulated daemon is stopped over
a channel and skips the home index, so several can share one test
process without walking the developer's home.

The account quota moves onto DriveApi. It was the only call outside the
trait that marks the link lost, so a simulated daemon would have gone
offline on it; the FakeDrive now also models a full account
(InsufficientQuota).

sim::daemon starts one daemon on its own state directory with a real
FUSE mount, and a smoke test (ignored; needs /dev/fuse) checks a local
write lands, a remote write arrives, and a write made offline lands once
the link is back.

Cargo.lock

@@ -3529,6 +3529,7 @@ dependencies = [
  "sha1 0.11.0",
  "tokio",
  "tracing",
+ "tracing-subscriber",
  "zip",
 ]
 

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

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<Vec<NodeUid>>;
+
+    /// The account's storage, across every Proton product.
+    async fn quota(&self) -> Result<Quota>;
 }
 
 /// 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<Vec<NodeUid>> {
         ProtonDriveClient::enumerate_trash_node_uids(self).await
     }
+
+    async fn quota(&self) -> Result<Quota> {
+        ProtonDriveClient::quota(self).await
+    }
 }

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<Db>,
     options: MountOptions,
+) -> std::io::Result<MountOutcome> {
+    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<dyn DriveApi>,
+    rt: tokio::runtime::Handle,
+    mountpoint: &Path,
+    cache: ContentCache,
+    control_socket: &Path,
+    db: Arc<Db>,
+    options: MountOptions,
+    host: Host,
 ) -> std::io::Result<MountOutcome> {
     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`.

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")
         })?;

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<JoinHandle<io::Result<MountOutcome>>>,
+    rt: Option<tokio::runtime::Runtime>,
+}
+
+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<Self> {
+        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<MountOutcome> {
+        self.shut_down()
+    }
+
+    fn shut_down(&mut self) -> io::Result<MountOutcome> {
+        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);
+    }
+}

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<Logged>,
     /// Plaintext block sizes of new content, cycled; the last one repeats.
     block_pattern: Vec<u64>,
+    /// 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<Quota> {
+        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<Vec<NodeUid>> {
         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 {

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;