From d53ba1be21c12a74ce9dc35d20116fcad1088651 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sat, 3 Oct 2026 04:51:35 +0200 Subject: [PATCH] test(fuse): add an in-memory FakeDrive with Drive's known faults The first half of the simulation harness in docs/MILESTONE-3.0.0.md section 8.1. FakeDrive is one shared server; each simulated daemon reaches it through a FakeClient that implements DriveApi and carries its own link state, latency, faults and an SDK-style entity cache. The faults are the ones the account runs met: listings that lag (B118, B48), renames and moves refused for a stale name hash because the cache is only emptied by events (B111, B121), names that refuse a replace (B13, B104, B106), events delayed, duplicated, reordered and echoed (B70, B105), requests that stall, fail, or change Drive and lose their reply (B107, B121), and non-uniform block sizes (B84, B85, B87). Test-only; no behaviour change. --- crates/pdfs-fuse/src/lib.rs | 2 + crates/pdfs-fuse/src/sim/fake_drive.rs | 1508 ++++++++++++++++++++++++ crates/pdfs-fuse/src/sim/mod.rs | 5 + crates/pdfs-fuse/src/sim/rng.rs | 79 ++ 4 files changed, 1594 insertions(+) create mode 100644 crates/pdfs-fuse/src/sim/fake_drive.rs create mode 100644 crates/pdfs-fuse/src/sim/mod.rs create mode 100644 crates/pdfs-fuse/src/sim/rng.rs diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 30c2f85..18d2c0e 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -103,6 +103,8 @@ mod relocate; mod revisions; mod sharing; mod shutdown; +#[cfg(test)] +mod sim; mod state; mod supervisor; mod sweep; diff --git a/crates/pdfs-fuse/src/sim/fake_drive.rs b/crates/pdfs-fuse/src/sim/fake_drive.rs new file mode 100644 index 0000000..7e9b352 --- /dev/null +++ b/crates/pdfs-fuse/src/sim/fake_drive.rs @@ -0,0 +1,1508 @@ +//! An in-memory Drive for the simulation tests (`docs/MILESTONE-3.0.0.md` §8.1). +//! +//! [`FakeDrive`] is the server: one tree of nodes with uids, revisions and +//! names, and an event log. Each simulated daemon reaches it through its own +//! [`FakeClient`], which implements [`DriveApi`] and carries what is particular +//! to one connection: whether its link is up, how slow it is, which faults it +//! injects, and an SDK-style entity cache. +//! +//! The faults are the ones Drive is known to have: +//! +//! - **Listing lag.** A read returns a node as it was [`Faults::listing_lag`] +//! ago, so a create or rename is not listed at once (B118, B48). +//! - **Stale name hash.** A rename or move sends the name the client's entity +//! cache holds, and Drive refuses it with `InvalidRequirements` when the node +//! has been renamed or moved since. Like the real SDK, the cache is filled by +//! reads and only emptied by `invalidate_caches_for_event`, not by the +//! client's own rename (B111, B121). +//! - **No atomic replace.** A name that is taken refuses a create, upload, +//! rename, move or restore with `AlreadyExists` (B13, B104, B106). +//! - **Events** delayed, duplicated or reordered, and our own changes echoed +//! back, because every change is logged for every client (B70, B105). +//! - **A request** that stalls, fails before it reaches Drive, or succeeds and +//! 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). + +use std::collections::{BTreeMap, HashMap}; +use std::io::{Read, Write}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use async_trait::async_trait; +use futures::FutureExt as _; +use futures::StreamExt as _; +use parking_lot::Mutex; +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}; +use proton_drive_rs::{ + DriveEvent, DriveEventScopeId, Node, NodeKind, NodeMoveItem, RevisionState, Thumbnail, +}; +use sha1::{Digest as _, Sha1}; + +use super::rng::Rng; +use crate::drive::{DriveApi, NodeOutcomes, OutcomeStream, RevisionRead}; + +/// The volume every fake node lives on. +pub(crate) const VOLUME: &str = "simvol"; + +/// How long superseded versions of a node are kept for lagging reads. Longer +/// than any lag a test sets. +const HISTORY: Duration = Duration::from_secs(120); + +/// What a file holds on the fake Drive. +#[derive(Clone, Debug)] +struct FileData { + media_type: String, + revision: String, + content: Arc>, + block_sizes: Vec, + /// Every revision the file has had, oldest first, the active one last. + revisions: Vec<(String, Arc>)>, +} + +/// One node as Drive has it at one moment. +#[derive(Clone, Debug)] +struct Remote { + parent: Option, + name: String, + trashed: bool, + created: i64, + modified: i64, + file: Option, +} + +/// One entry of the event log. +struct Logged { + seq: u64, + at: Instant, + event: DriveEvent, +} + +struct Server { + root: NodeUid, + /// Every version of every node within [`HISTORY`], oldest first. `None` is + /// a permanent delete. + nodes: HashMap)>>, + next_link: u64, + next_revision: u64, + events: Vec, + /// Plaintext block sizes of new content, cycled; the last one repeats. + block_pattern: Vec, +} + +impl Server { + fn current(&self, uid: &NodeUid) -> Option<&Remote> { + self.nodes.get(uid)?.last()?.1.as_ref() + } + + /// The node as a read `lag` behind the present sees it. + fn visible(&self, uid: &NodeUid, lag: Duration, now: Instant) -> Option { + let versions = self.nodes.get(uid)?; + versions + .iter() + .rev() + .find(|(at, _)| *at + lag <= now) + .and_then(|(_, remote)| remote.clone()) + } + + fn live_child_named(&self, parent: &NodeUid, name: &str) -> Option { + self.nodes.iter().find_map(|(uid, versions)| { + let remote = versions.last()?.1.as_ref()?; + (remote.parent.as_ref() == Some(parent) && remote.name == name && !remote.trashed) + .then(|| uid.clone()) + }) + } + + fn is_folder(&self, uid: &NodeUid) -> bool { + self.current(uid) + .is_some_and(|remote| remote.file.is_none() && !remote.trashed) + } + + fn is_below(&self, uid: &NodeUid, ancestor: &NodeUid) -> bool { + let mut cursor = Some(uid.clone()); + while let Some(at) = cursor { + if &at == ancestor { + return true; + } + cursor = self.current(&at).and_then(|remote| remote.parent.clone()); + } + false + } + + fn mint_uid(&mut self) -> NodeUid { + self.next_link += 1; + uid(&format!("L{}", self.next_link)) + } + + fn mint_revision(&mut self) -> String { + self.next_revision += 1; + format!("rev{}", self.next_revision) + } + + fn blocks_of(&self, len: u64) -> Vec { + let mut sizes = Vec::new(); + let mut left = len; + let mut i = 0; + while left > 0 { + let size = self.block_pattern[i.min(self.block_pattern.len() - 1)].min(left); + sizes.push(size); + left -= size; + i += 1; + } + sizes + } + + /// Record a new version of `uid` and log the change. + fn put(&mut self, uid: &NodeUid, remote: Option) { + let now = Instant::now(); + let event_id = DriveEventId::from(format!("ev{}", self.events.len() + 1)); + let event = match &remote { + Some(remote) => DriveEvent::NodeUpdated { + id: event_id, + node_uid: uid.clone(), + parent_node_uid: remote.parent.clone(), + is_trashed: remote.trashed, + is_shared: false, + }, + None => DriveEvent::NodeDeleted { + id: event_id, + node_uid: uid.clone(), + parent_node_uid: self.current(uid).and_then(|r| r.parent.clone()), + }, + }; + let versions = self.nodes.entry(uid.clone()).or_default(); + versions.push((now, remote)); + while versions.len() > 1 && versions[1].0 + HISTORY < now { + versions.remove(0); + } + let seq = self.events.len() as u64 + 1; + self.events.push(Logged { + seq, + at: now, + event, + }); + } + + fn new_file( + &mut self, + parent: &NodeUid, + name: &str, + media_type: &str, + content: Vec, + mtime: Option, + ) -> Result { + if !self.is_folder(parent) { + return Err(api(ResponseCode::DoesNotExist, "parent folder not found")); + } + if self.live_child_named(parent, name).is_some() { + return Err(api( + ResponseCode::AlreadyExists, + "a file or folder with that name already exists", + )); + } + let uid = self.mint_uid(); + let revision = self.mint_revision(); + let content = Arc::new(content); + let now = now_secs(); + let file = FileData { + media_type: media_type.to_owned(), + revision: revision.clone(), + block_sizes: self.blocks_of(content.len() as u64), + content: content.clone(), + revisions: vec![(revision, content)], + }; + self.put( + &uid, + Some(Remote { + parent: Some(parent.clone()), + name: name.to_owned(), + trashed: false, + created: now, + modified: mtime.unwrap_or(now), + file: Some(file), + }), + ); + Ok(uid) + } +} + +/// An in-memory Drive shared by every simulated client. See the module +/// documentation. +pub(crate) struct FakeDrive { + server: Mutex, +} + +/// A node of [`FakeDrive::tree`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum Entry { + Folder, + File(Arc>), +} + +impl FakeDrive { + pub(crate) fn new() -> Arc { + let root = uid("root"); + let mut server = Server { + root: root.clone(), + nodes: HashMap::new(), + next_link: 0, + next_revision: 0, + events: Vec::new(), + block_pattern: vec![pdfs_core::cache::BLOCK_SIZE], + }; + let now = now_secs(); + server.nodes.insert( + root, + vec![( + Instant::now() - HISTORY, + Some(Remote { + parent: None, + name: "root".into(), + trashed: false, + created: now, + modified: now, + file: None, + }), + )], + ); + Arc::new(Self { + server: Mutex::new(server), + }) + } + + /// A connection to this Drive for one simulated daemon. + pub(crate) fn client(self: &Arc, seed: u64, faults: Faults) -> FakeClient { + FakeClient(Arc::new(ClientInner { + drive: self.clone(), + faults: Mutex::new(faults), + rng: Mutex::new(Rng::new(seed)), + entity_cache: Mutex::new(HashMap::new()), + online: AtomicBool::new(true), + requests: AtomicU64::new(0), + })) + } + + /// Plaintext block sizes for content uploaded from now on, cycled with the + /// last one repeating: `[1 MiB, 4 MiB]` makes a short first block. + pub(crate) fn set_block_pattern(&self, pattern: Vec) { + assert!(!pattern.is_empty() && pattern.iter().all(|&size| size > 0)); + self.server.lock().block_pattern = pattern; + } + + pub(crate) fn root(&self) -> NodeUid { + self.server.lock().root.clone() + } + + /// The live tree as Drive has it now, by path below the root: every node + /// that is neither trashed nor below a trashed folder. + pub(crate) fn tree(&self) -> BTreeMap { + let server = self.server.lock(); + let mut tree = BTreeMap::new(); + for uid in server.nodes.keys() { + if *uid == server.root { + continue; + } + let mut names = Vec::new(); + let mut cursor = Some(uid.clone()); + let mut live = true; + while let Some(at) = cursor { + if at == server.root { + break; + } + match server.current(&at) { + Some(remote) if !remote.trashed => { + names.push(remote.name.clone()); + cursor = remote.parent.clone(); + } + _ => { + live = false; + break; + } + } + } + if !live { + continue; + } + names.reverse(); + let entry = match &server.current(uid).and_then(|r| r.file.clone()) { + Some(file) => Entry::File(file.content.clone()), + None => Entry::Folder, + }; + tree.insert(names.join("/"), entry); + } + tree + } + + /// The uid at `path` below the root, as Drive has it now. + pub(crate) fn lookup(&self, path: &str) -> Option { + let server = self.server.lock(); + let mut at = server.root.clone(); + for name in path.split('/').filter(|name| !name.is_empty()) { + at = server.live_child_named(&at, name)?; + } + Some(at) + } + + /// Every change Drive has logged, oldest first. + pub(crate) fn event_count(&self) -> usize { + self.server.lock().events.len() + } + + /// Changes made by another device, straight on the server: what a second + /// client would do, without the daemon that client would need. + pub(crate) fn device(&self) -> Device<'_> { + Device(self) + } +} + +/// See [`FakeDrive::device`]. +pub(crate) struct Device<'a>(&'a FakeDrive); + +impl Device<'_> { + /// Write `content` at `path`, as a new file or a new revision of one. + pub(crate) fn write(&self, path: &str, content: &[u8]) -> NodeUid { + let (parent, name) = split_path(path); + let parent = self + .0 + .lookup(parent) + .expect("device write into a missing folder"); + let mut server = self.0.server.lock(); + match server.live_child_named(&parent, name) { + Some(uid) => { + let mut remote = server.current(&uid).cloned().expect("live child"); + let revision = server.mint_revision(); + let content = Arc::new(content.to_vec()); + let blocks = server.blocks_of(content.len() as u64); + let file = remote.file.as_mut().expect("device write over a folder"); + file.revisions.push((revision.clone(), content.clone())); + file.revision = revision; + file.content = content; + file.block_sizes = blocks; + remote.modified = now_secs(); + server.put(&uid, Some(remote)); + uid + } + None => server + .new_file( + &parent, + name, + "application/octet-stream", + content.to_vec(), + None, + ) + .expect("device create"), + } + } + + /// Move the node at `path` to the trash. + pub(crate) fn trash(&self, path: &str) { + let uid = self.0.lookup(path).expect("device trash of a missing node"); + let mut server = self.0.server.lock(); + let mut remote = server.current(&uid).cloned().expect("live node"); + remote.trashed = true; + server.put(&uid, Some(remote)); + } + + /// Rename the node at `path` in place. + pub(crate) fn rename(&self, path: &str, new_name: &str) { + let uid = self + .0 + .lookup(path) + .expect("device rename of a missing node"); + let mut server = self.0.server.lock(); + let mut remote = server.current(&uid).cloned().expect("live node"); + remote.name = new_name.to_owned(); + server.put(&uid, Some(remote)); + } +} + +/// The faults and timing of one connection. See the module documentation. +#[derive(Clone, Debug)] +pub(crate) struct Faults { + /// Each request takes a time drawn from this range, in milliseconds. + pub(crate) latency_ms: (u64, u64), + /// How far behind the present reads are. + pub(crate) listing_lag: Duration, + /// How long a change takes to reach the event feed. + pub(crate) event_delay: Duration, + /// Chance that an event is delivered twice. + pub(crate) duplicate_events: f64, + /// Chance that two neighbouring events of a page swap places. + pub(crate) reorder_events: f64, + /// Chance that a request never reaches Drive. + pub(crate) fail: f64, + /// Chance that a change Drive made loses its reply. + pub(crate) drop_reply: f64, + /// Chance that a request stalls for [`Faults::stall_for`] and then fails. + pub(crate) stall: f64, + pub(crate) stall_for: Duration, + /// Whether renames and moves send the entity cache's name, as the SDK does. + pub(crate) stale_name_cache: bool, +} + +impl Faults { + /// No latency and no faults beyond the SDK's own stale name cache. + pub(crate) fn none() -> Self { + Self { + latency_ms: (0, 0), + listing_lag: Duration::ZERO, + event_delay: Duration::ZERO, + duplicate_events: 0.0, + reorder_events: 0.0, + fail: 0.0, + drop_reply: 0.0, + stall: 0.0, + stall_for: Duration::ZERO, + stale_name_cache: true, + } + } + + /// A good link: a few milliseconds a request, listings barely behind. + pub(crate) fn lan() -> Self { + Self { + latency_ms: (2, 15), + listing_lag: Duration::from_millis(50), + event_delay: Duration::from_millis(200), + ..Self::none() + } + } + + /// The Wi-Fi of the 2026-10 account runs: slow and lossy, with listings + /// and events well behind. + pub(crate) fn wifi() -> Self { + Self { + latency_ms: (60, 400), + listing_lag: Duration::from_millis(800), + event_delay: Duration::from_secs(2), + duplicate_events: 0.05, + reorder_events: 0.1, + fail: 0.02, + drop_reply: 0.02, + stall: 0.005, + stall_for: Duration::from_secs(3), + ..Self::none() + } + } +} + +struct ClientInner { + drive: Arc, + faults: Mutex, + rng: Mutex, + /// What the SDK's entity cache holds: filled by reads, emptied only by + /// `invalidate_caches_for_event`. + entity_cache: Mutex>, + online: AtomicBool, + requests: AtomicU64, +} + +/// One daemon's connection to a [`FakeDrive`]. Cheap to clone; clones share +/// the link, the faults and the entity cache. +#[derive(Clone)] +pub(crate) struct FakeClient(Arc); + +impl FakeClient { + pub(crate) fn set_online(&self, online: bool) { + self.0.online.store(online, Ordering::SeqCst); + } + + pub(crate) fn set_faults(&self, faults: Faults) { + *self.0.faults.lock() = faults; + } + + /// Requests made through this client so far. + pub(crate) fn requests(&self) -> u64 { + self.0.requests.load(Ordering::SeqCst) + } + + /// Run one request against the server: wait out the latency, apply the + /// connection's faults, and call `apply` if the request reaches Drive. + async fn request( + &self, + change: bool, + apply: impl FnOnce(&mut Server) -> Result, + ) -> Result { + self.0.requests.fetch_add(1, Ordering::SeqCst); + let faults = self.0.faults.lock().clone(); + let (latency, stall, fail, drop_reply) = { + let mut rng = self.0.rng.lock(); + ( + Duration::from_millis(rng.between(faults.latency_ms.0, faults.latency_ms.1)), + rng.chance(faults.stall), + rng.chance(faults.fail), + change && rng.chance(faults.drop_reply), + ) + }; + if !self.0.online.load(Ordering::SeqCst) { + tokio::time::sleep(latency.min(Duration::from_millis(20))).await; + return Err(no_answer()); + } + tokio::time::sleep(latency / 2).await; + if stall { + tokio::time::sleep(faults.stall_for).await; + return Err(no_answer()); + } + if fail || !self.0.online.load(Ordering::SeqCst) { + return Err(no_answer()); + } + let outcome = apply(&mut self.0.drive.server.lock()); + tokio::time::sleep(latency - latency / 2).await; + if drop_reply || !self.0.online.load(Ordering::SeqCst) { + return Err(no_answer()); + } + outcome + } + + fn lag(&self) -> Duration { + self.0.faults.lock().listing_lag + } + + /// The `(parent, name)` a rename or move of `uid` sends as its original + /// hash: the entity cache's copy, or Drive's as a read sees it. + fn original(&self, server: &Server, uid: &NodeUid) -> Option<(Option, String)> { + if self.0.faults.lock().stale_name_cache { + if let Some(node) = self.0.entity_cache.lock().get(uid) { + return Some((node.parent_uid.clone(), node.name.clone())); + } + let remote = server.visible(uid, self.lag(), Instant::now())?; + self.0 + .entity_cache + .lock() + .insert(uid.clone(), to_node(uid, &remote, false)); + return Some((remote.parent, remote.name)); + } + server + .current(uid) + .map(|r| (r.parent.clone(), r.name.clone())) + } + + fn cache(&self, nodes: &[Node]) { + let mut cache = self.0.entity_cache.lock(); + for node in nodes { + cache.insert(node.uid.clone(), node.clone()); + } + } + + fn rename_or_move( + &self, + server: &mut Server, + uid: &NodeUid, + new_parent: Option<&NodeUid>, + new_name: Option<&str>, + ) -> Result<()> { + let original = self.original(server, uid); + let Some(mut remote) = server.current(uid).cloned() else { + return Err(api(ResponseCode::DoesNotExist, "node not found")); + }; + if original != Some((remote.parent.clone(), remote.name.clone())) { + return Err(api( + ResponseCode::InvalidRequirements, + "original hash is out of date", + )); + } + let parent = new_parent.cloned().or(remote.parent.clone()); + let name = new_name.unwrap_or(&remote.name).to_owned(); + if let Some(parent) = &parent { + if !server.is_folder(parent) { + return Err(api( + ResponseCode::DoesNotExist, + "destination folder not found", + )); + } + if server.is_below(parent, uid) { + return Err(api( + ResponseCode::InvalidRequirements, + "cannot move a folder into itself", + )); + } + if server + .live_child_named(parent, &name) + .is_some_and(|other| &other != uid) + { + return Err(api( + ResponseCode::AlreadyExists, + "a file or folder with that name already exists", + )); + } + } + remote.parent = parent; + remote.name = name; + server.put(uid, Some(remote)); + Ok(()) + } + + fn batch<'a>( + &'a self, + uids: &[NodeUid], + each: fn(&mut Server, &NodeUid) -> Result<()>, + ) -> futures::future::BoxFuture<'a, Result> { + let uids = uids.to_vec(); + async move { + self.request(true, |server| { + Ok(uids + .iter() + .map(|uid| (uid.clone(), each(server, uid))) + .collect()) + }) + .await + } + .boxed() + } +} + +fn outcome_stream( + outcomes: futures::future::BoxFuture<'_, Result>, +) -> OutcomeStream<'_> { + outcomes + .map(|outcomes| match outcomes { + Ok(outcomes) => futures::stream::iter(outcomes.into_iter().map(Ok).collect::>()), + Err(error) => futures::stream::iter(vec![Err(error)]), + }) + .flatten_stream() + .boxed() +} + +fn trash_one(server: &mut Server, uid: &NodeUid) -> Result<()> { + let Some(mut remote) = server.current(uid).cloned() else { + return Err(api(ResponseCode::DoesNotExist, "node not found")); + }; + if !remote.trashed { + remote.trashed = true; + server.put(uid, Some(remote)); + } + Ok(()) +} + +fn restore_one(server: &mut Server, uid: &NodeUid) -> Result<()> { + let Some(mut remote) = server.current(uid).cloned() else { + return Err(api(ResponseCode::DoesNotExist, "node not found")); + }; + if !remote.trashed { + return Ok(()); + } + if let Some(parent) = &remote.parent + && server.live_child_named(parent, &remote.name).is_some() + { + return Err(api( + ResponseCode::AlreadyExists, + "a file or folder with that name already exists", + )); + } + remote.trashed = false; + server.put(uid, Some(remote)); + Ok(()) +} + +fn delete_one(server: &mut Server, uid: &NodeUid) -> Result<()> { + if server.current(uid).is_none() { + return Err(api(ResponseCode::DoesNotExist, "node not found")); + } + server.put(uid, None); + Ok(()) +} + +#[async_trait] +impl DriveApi for FakeClient { + async fn get_my_files_folder(&self) -> Result { + self.request(false, |server| { + let root = server.root.clone(); + let remote = server.current(&root).expect("the root always exists"); + Ok(to_node(&root, remote, false)) + }) + .await + } + + async fn get_node(&self, uid: &NodeUid) -> Result> { + if let Some(node) = self.0.entity_cache.lock().get(uid) { + return Ok(Some(node.clone())); + } + let lag = self.lag(); + let node = self + .request(false, |server| { + Ok(server + .visible(uid, lag, Instant::now()) + .map(|remote| to_node(uid, &remote, false))) + }) + .await?; + self.cache(node.as_slice()); + Ok(node) + } + + async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result> { + let lag = self.lag(); + let nodes: Vec = self + .request(false, |server| { + let now = Instant::now(); + Ok(uids + .iter() + .filter_map(|uid| Some(to_node(uid, &server.visible(uid, lag, now)?, false))) + .collect()) + }) + .await?; + self.cache(&nodes); + Ok(nodes) + } + + async fn enumerate_nodes_light(&self, uids: &[NodeUid]) -> Result> { + let lag = self.lag(); + self.request(false, |server| { + let now = Instant::now(); + Ok(uids + .iter() + .filter_map(|uid| Some(to_node(uid, &server.visible(uid, lag, now)?, true))) + .collect()) + }) + .await + } + + async fn enumerate_folder_children_node_uids( + &self, + folder_uid: &NodeUid, + ) -> Result> { + let lag = self.lag(); + self.request(false, |server| { + let now = Instant::now(); + if server.visible(folder_uid, lag, now).is_none() { + return Err(api(ResponseCode::DoesNotExist, "folder not found")); + } + let mut children: Vec = server + .nodes + .keys() + .filter(|uid| { + server.visible(uid, lag, now).is_some_and(|remote| { + remote.parent.as_ref() == Some(folder_uid) && !remote.trashed + }) + }) + .cloned() + .collect(); + children.sort_by(|a, b| a.link_id.cmp(&b.link_id)); + Ok(children) + }) + .await + } + + async fn enumerate_events( + &self, + _scope: &DriveEventScopeId, + cursor: Option<&DriveEventId>, + ) -> Result> { + let faults = self.0.faults.lock().clone(); + let mut events = self + .request(false, |server| { + let Some(cursor) = cursor else { + let head = DriveEventId::from(format!("ev{}", server.events.len())); + return Ok(vec![DriveEvent::CursorAdvanced { id: head }]); + }; + let Some(after) = cursor + .as_str() + .strip_prefix("ev") + .and_then(|seq| seq.parse::().ok()) + else { + return Ok(vec![DriveEvent::ContinuityLost { id: cursor.clone() }]); + }; + let now = Instant::now(); + Ok(server + .events + .iter() + .filter(|logged| logged.seq > after) + .take_while(|logged| logged.at + faults.event_delay <= now) + .map(|logged| logged.event.clone()) + .collect()) + }) + .await?; + let mut rng = self.0.rng.lock(); + let mut i = 0; + while i < events.len() { + if rng.chance(faults.duplicate_events) { + events.insert(i + 1, events[i].clone()); + i += 1; + } + i += 1; + } + for i in 1..events.len() { + if rng.chance(faults.reorder_events) { + events.swap(i - 1, i); + } + } + Ok(events) + } + + async fn invalidate_caches_for_event(&self, event: &DriveEvent) -> Result<()> { + let mut cache = self.0.entity_cache.lock(); + match event { + DriveEvent::NodeUpdated { node_uid, .. } | DriveEvent::NodeDeleted { node_uid, .. } => { + cache.remove(node_uid); + } + DriveEvent::ContinuityLost { .. } | DriveEvent::ScopeAccessLost { .. } => cache.clear(), + DriveEvent::CursorAdvanced { .. } | DriveEvent::SharedWithMeUpdated { .. } => {} + } + Ok(()) + } + + async fn create_folder( + &self, + parent_uid: &NodeUid, + name: &str, + last_modification_time: Option, + ) -> Result { + self.request(true, |server| { + if !server.is_folder(parent_uid) { + return Err(api(ResponseCode::DoesNotExist, "parent folder not found")); + } + if server.live_child_named(parent_uid, name).is_some() { + return Err(api( + ResponseCode::AlreadyExists, + "a file or folder with that name already exists", + )); + } + let uid = server.mint_uid(); + let now = now_secs(); + server.put( + &uid, + Some(Remote { + parent: Some(parent_uid.clone()), + name: name.to_owned(), + trashed: false, + created: now, + modified: last_modification_time.unwrap_or(now), + file: None, + }), + ); + Ok(uid) + }) + .await + } + + async fn upload_file( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + contents: &[u8], + ) -> Result { + self.request(true, |server| { + server.new_file(parent_uid, name, media_type, contents.to_vec(), None) + }) + .await + } + + async fn upload_file_from_dyn( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + reader: &mut (dyn Read + Send), + _intended_size: i64, + _thumbnails: Vec, + last_modification_time: Option, + _aead: bool, + ) -> Result { + let content = read_all(reader)?; + self.request(true, |server| { + server.new_file( + parent_uid, + name, + media_type, + content, + last_modification_time, + ) + }) + .await + } + + async fn upload_file_replacing_draft_from_dyn( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + reader: &mut (dyn Read + Send), + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + aead: bool, + ) -> Result { + // The fake Drive keeps no drafts, so there is never one to replace. + self.upload_file_from_dyn( + parent_uid, + name, + media_type, + reader, + intended_size, + thumbnails, + last_modification_time, + aead, + ) + .await + } + + async fn upload_new_revision_from_dyn( + &self, + file_uid: &NodeUid, + reader: &mut (dyn Read + Send), + _intended_size: i64, + _thumbnails: Vec, + last_modification_time: Option, + ) -> Result<()> { + let content = Arc::new(read_all(reader)?); + self.request(true, |server| { + let Some(mut remote) = server.current(file_uid).cloned() else { + return Err(api(ResponseCode::DoesNotExist, "file not found")); + }; + if remote.file.is_none() { + return Err(api(ResponseCode::InvalidRequirements, "not a file")); + } + let revision = server.mint_revision(); + let blocks = server.blocks_of(content.len() as u64); + let file = remote.file.as_mut().expect("checked above"); + file.revisions.push((revision.clone(), content.clone())); + file.revision = revision; + file.content = content; + file.block_sizes = blocks; + remote.modified = last_modification_time.unwrap_or_else(now_secs); + server.put(file_uid, Some(remote)); + Ok(()) + }) + .await + } + + async fn open_revision(&self, uid: &NodeUid) -> Result> { + let file = self + .request(false, |server| { + server + .current(uid) + .and_then(|remote| remote.file.clone()) + .ok_or_else(|| api(ResponseCode::DoesNotExist, "file not found")) + }) + .await?; + Ok(Arc::new(FakeRevision { + client: self.clone(), + content: file.content, + block_sizes: file.block_sizes, + })) + } + + async fn download_file_to_dyn( + &self, + uid: &NodeUid, + output: &mut (dyn Write + Send), + ) -> Result<()> { + let content = self + .request(false, |server| { + server + .current(uid) + .and_then(|remote| remote.file.as_ref().map(|file| file.content.clone())) + .ok_or_else(|| api(ResponseCode::DoesNotExist, "file not found")) + }) + .await?; + output.write_all(&content).map_err(io_error) + } + + async fn download_revision_to_dyn( + &self, + file_uid: &NodeUid, + revision_id: &str, + writer: &mut (dyn Write + Send), + ) -> Result<()> { + let content = self + .request(false, |server| { + server + .current(file_uid) + .and_then(|remote| remote.file.as_ref()) + .and_then(|file| file.revisions.iter().find(|(id, _)| id == revision_id)) + .map(|(_, content)| content.clone()) + .ok_or_else(|| api(ResponseCode::DoesNotExist, "revision not found")) + }) + .await?; + writer.write_all(&content).map_err(io_error) + } + + async fn rename_node( + &self, + uid: &NodeUid, + new_name: &str, + _new_media_type: Option<&str>, + ) -> Result<()> { + self.request(true, |server| { + self.rename_or_move(server, uid, None, Some(new_name)) + }) + .await + } + + async fn move_node(&self, uid: &NodeUid, new_parent: &NodeUid) -> Result<()> { + self.request(true, |server| { + self.rename_or_move(server, uid, Some(new_parent), None) + }) + .await + } + + fn move_nodes_streaming<'a>( + &'a self, + items: Vec, + new_parent: NodeUid, + ) -> OutcomeStream<'a> { + outcome_stream( + async move { + self.request(true, |server| { + Ok(items + .iter() + .map(|item| { + let outcome = self.rename_or_move( + server, + &item.uid, + Some(&new_parent), + item.target_name.as_deref(), + ); + (item.uid.clone(), outcome) + }) + .collect()) + }) + .await + } + .boxed(), + ) + } + + async fn trash_nodes(&self, uids: &[NodeUid]) -> Result { + self.batch(uids, trash_one).await + } + + async fn restore_nodes(&self, uids: &[NodeUid]) -> Result { + self.batch(uids, restore_one).await + } + + fn restore_nodes_streaming<'a>(&'a self, uids: &[NodeUid]) -> OutcomeStream<'a> { + outcome_stream(self.batch(uids, restore_one)) + } + + fn delete_nodes_streaming<'a>(&'a self, uids: &[NodeUid]) -> OutcomeStream<'a> { + outcome_stream(self.batch(uids, delete_one)) + } + + async fn enumerate_trash_node_uids(&self) -> Result> { + self.request(false, |server| { + Ok(server + .nodes + .iter() + .filter(|(_, versions)| { + versions + .last() + .and_then(|(_, remote)| remote.as_ref()) + .is_some_and(|remote| remote.trashed) + }) + .map(|(uid, _)| uid.clone()) + .collect()) + }) + .await + } +} + +/// An open revision of a fake file. Its block reads are requests like any +/// other, so a link that drops under a read fails it. +struct FakeRevision { + client: FakeClient, + content: Arc>, + block_sizes: Vec, +} + +#[async_trait] +impl RevisionRead for FakeRevision { + fn size(&self) -> u64 { + self.content.len() as u64 + } + + fn block_sizes(&self) -> &[u64] { + &self.block_sizes + } + + async fn read_at(&self, offset: u64, length: u64) -> Result> { + let content = self.content.clone(); + self.client + .request(false, move |_| { + let start = (offset as usize).min(content.len()); + let end = (offset.saturating_add(length) as usize).min(content.len()); + Ok(content[start..end].to_vec()) + }) + .await + } +} + +/// A uid on the fake volume. +pub(crate) fn uid(link: &str) -> NodeUid { + NodeUid::new(VolumeId::from(VOLUME), LinkId::from(link)) +} + +fn to_node(uid: &NodeUid, remote: &Remote, light: bool) -> Node { + let kind = match &remote.file { + None => NodeKind::Folder, + Some(file) => { + let size = file.content.len() as i64; + NodeKind::File { + media_type: file.media_type.clone(), + // Ciphertext is larger than the plaintext: a packet header per + // block, which is what B12's provisional size was off by. + total_size_on_storage: size + 512 * file.block_sizes.len().max(1) as i64, + active_revision_state: Some(RevisionState::Active), + active_revision_id: Some(file.revision.clone()), + claimed_size: (!light).then_some(size), + claimed_modification_time: None, + content_sha1: (!light).then(|| { + Sha1::digest(file.content.as_slice()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() + }), + } + } + }; + Node { + uid: uid.clone(), + parent_uid: remote.parent.clone(), + kind, + name: remote.name.clone(), + creation_time: remote.created, + modification_time: remote.modified, + trashed: remote.trashed, + is_shared: false, + is_shared_publicly: false, + signature_email: None, + membership: None, + direct_role: None, + share_id: None, + photo: None, + album: None, + verification: Default::default(), + } +} + +fn split_path(path: &str) -> (&str, &str) { + match path.rsplit_once('/') { + Some((parent, name)) => (parent, name), + None => ("", path), + } +} + +fn read_all(reader: &mut (dyn Read + Send)) -> Result> { + let mut content = Vec::new(); + reader.read_to_end(&mut content).map_err(io_error)?; + Ok(content) +} + +fn io_error(error: std::io::Error) -> ProtonError { + ProtonError::invalid_operation(format!("local i/o: {error}")) +} + +fn api(code: ResponseCode, message: &str) -> ProtonError { + let http_status = match code { + ResponseCode::DoesNotExist => 404, + ResponseCode::AlreadyExists | ResponseCode::InvalidRequirements => 422, + _ => 400, + }; + ProtonError::Api(ProtonApiError { + code, + http_status, + message: format!("fake drive: {message}"), + details: None, + }) +} + +/// What a request that never got an answer reports; the daemon reads it as +/// the network (see `link::is_network_error`). +fn no_answer() -> ProtonError { + ProtonError::Api(ProtonApiError { + code: ResponseCode::RequestTimeout, + http_status: 408, + message: "fake drive: no answer".into(), + details: None, + }) +} + +fn now_secs() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::link::is_network_error; + + fn code(error: &ProtonError) -> Option { + match error { + ProtonError::Api(api) => Some(api.code), + _ => None, + } + } + + fn rt() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap() + } + + #[test] + fn a_tree_built_through_the_api_reads_back() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client(1, Faults::none()); + let root = client.get_my_files_folder().await.unwrap(); + let docs = client.create_folder(&root.uid, "docs", None).await.unwrap(); + let file = client + .upload_file(&docs, "a.txt", "text/plain", b"hello") + .await + .unwrap(); + let children = client + .enumerate_folder_children_node_uids(&docs) + .await + .unwrap(); + assert_eq!(children, vec![file.clone()]); + let node = client + .enumerate_nodes(std::slice::from_ref(&file)) + .await + .unwrap() + .remove(0); + assert!(matches!( + node.kind, + NodeKind::File { + claimed_size: Some(5), + .. + } + )); + let light = client + .enumerate_nodes_light(std::slice::from_ref(&file)) + .await + .unwrap() + .remove(0); + assert!(matches!( + light.kind, + NodeKind::File { + claimed_size: None, + .. + } + )); + let mut out = Vec::new(); + (&client as &dyn DriveApi) + .download_file_to(&file, &mut out) + .await + .unwrap(); + assert_eq!(out, b"hello"); + assert_eq!( + drive.tree(), + BTreeMap::from([ + ("docs".to_owned(), Entry::Folder), + ( + "docs/a.txt".to_owned(), + Entry::File(Arc::new(b"hello".to_vec())) + ), + ]) + ); + }); + } + + #[test] + fn a_taken_name_refuses_a_create_and_a_rename() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client(1, Faults::none()); + let root = drive.root(); + client.upload_file(&root, "a", "", b"1").await.unwrap(); + let b = client.upload_file(&root, "b", "", b"2").await.unwrap(); + let taken = client.upload_file(&root, "a", "", b"3").await.unwrap_err(); + assert_eq!(code(&taken), Some(ResponseCode::AlreadyExists)); + let taken = client.rename_node(&b, "a", None).await.unwrap_err(); + assert_eq!(code(&taken), Some(ResponseCode::AlreadyExists)); + }); + } + + #[test] + fn a_second_rename_sends_the_cached_name_until_an_event_evicts_it() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client(1, Faults::none()); + let root = drive.root(); + let file = client.upload_file(&root, "a", "", b"1").await.unwrap(); + client.rename_node(&file, "b", None).await.unwrap(); + let stale = client.rename_node(&file, "c", None).await.unwrap_err(); + assert_eq!(code(&stale), Some(ResponseCode::InvalidRequirements)); + let changed = DriveEvent::NodeUpdated { + id: "x".into(), + node_uid: file.clone(), + parent_node_uid: None, + is_trashed: false, + is_shared: false, + }; + client.invalidate_caches_for_event(&changed).await.unwrap(); + client.rename_node(&file, "c", None).await.unwrap(); + assert!(drive.lookup("c").is_some()); + }); + } + + #[test] + fn a_lagging_listing_does_not_show_a_fresh_create() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client( + 1, + Faults { + listing_lag: Duration::from_millis(200), + ..Faults::none() + }, + ); + let root = drive.root(); + let file = client.upload_file(&root, "a", "", b"1").await.unwrap(); + assert!( + client + .enumerate_folder_children_node_uids(&root) + .await + .unwrap() + .is_empty() + ); + assert!( + client + .enumerate_nodes(std::slice::from_ref(&file)) + .await + .unwrap() + .is_empty() + ); + tokio::time::sleep(Duration::from_millis(250)).await; + assert_eq!( + client + .enumerate_folder_children_node_uids(&root) + .await + .unwrap(), + vec![file] + ); + }); + } + + #[test] + fn every_client_sees_every_change_on_the_feed() { + rt().block_on(async { + let drive = FakeDrive::new(); + let (one, two) = ( + drive.client(1, Faults::none()), + drive.client(2, Faults::none()), + ); + let scope = DriveEventScopeId::new(VolumeId::from(VOLUME)); + let seed = two.enumerate_events(&scope, None).await.unwrap(); + let cursor = seed[0].id().clone(); + let file = one.upload_file(&drive.root(), "a", "", b"1").await.unwrap(); + one.trash_nodes(std::slice::from_ref(&file)).await.unwrap(); + let events = two.enumerate_events(&scope, Some(&cursor)).await.unwrap(); + assert!(matches!( + &events[..], + [ + DriveEvent::NodeUpdated { + is_trashed: false, + .. + }, + DriveEvent::NodeUpdated { + is_trashed: true, + .. + } + ] + )); + let own = one.enumerate_events(&scope, Some(&cursor)).await.unwrap(); + assert_eq!(own.len(), 2, "our own changes are echoed back"); + }); + } + + #[test] + fn a_lost_reply_still_changed_drive() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client( + 1, + Faults { + drop_reply: 1.0, + ..Faults::none() + }, + ); + let error = client + .create_folder(&drive.root(), "half", None) + .await + .unwrap_err(); + assert!(is_network_error(&error)); + assert!(drive.lookup("half").is_some()); + }); + } + + #[test] + fn an_offline_client_reaches_nothing() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client(1, Faults::none()); + client.set_online(false); + let error = client + .create_folder(&drive.root(), "x", None) + .await + .unwrap_err(); + assert!(is_network_error(&error)); + assert!(drive.lookup("x").is_none()); + }); + } + + #[test] + fn content_is_split_by_the_block_pattern() { + rt().block_on(async { + let drive = FakeDrive::new(); + drive.set_block_pattern(vec![3, 5]); + let client = drive.client(1, Faults::none()); + let file = client + .upload_file(&drive.root(), "f", "", b"0123456789abc") + .await + .unwrap(); + let reader = client.open_revision(&file).await.unwrap(); + assert_eq!(reader.block_sizes(), &[3, 5, 5]); + assert_eq!(reader.read_at(3, 5).await.unwrap(), b"34567"); + }); + } + + #[test] + fn a_rename_by_another_device_makes_our_cached_name_stale() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client(1, Faults::none()); + let file = client + .upload_file(&drive.root(), "a", "", b"1") + .await + .unwrap(); + client.get_node(&file).await.unwrap(); + drive.device().rename("a", "b"); + let stale = client.rename_node(&file, "c", None).await.unwrap_err(); + assert_eq!(code(&stale), Some(ResponseCode::InvalidRequirements)); + }); + } + + #[test] + fn faults_set_later_apply_to_the_next_request() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client(1, Faults::lan()); + client.get_my_files_folder().await.unwrap(); + client.set_faults(Faults { + fail: 1.0, + ..Faults::wifi() + }); + let error = client.get_my_files_folder().await.unwrap_err(); + assert!(is_network_error(&error)); + assert_eq!(client.requests(), 2); + }); + } + + #[test] + fn a_device_change_lands_on_the_server_and_the_feed() { + let drive = FakeDrive::new(); + drive.device().write("notes", b"v1"); + drive.device().write("notes", b"v2"); + assert_eq!(drive.tree()["notes"], Entry::File(Arc::new(b"v2".to_vec()))); + drive.device().trash("notes"); + assert!(drive.tree().is_empty()); + assert_eq!(drive.event_count(), 3); + } +} diff --git a/crates/pdfs-fuse/src/sim/mod.rs b/crates/pdfs-fuse/src/sim/mod.rs new file mode 100644 index 0000000..89e635b --- /dev/null +++ b/crates/pdfs-fuse/src/sim/mod.rs @@ -0,0 +1,5 @@ +//! 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 fake_drive; +pub(crate) mod rng; diff --git a/crates/pdfs-fuse/src/sim/rng.rs b/crates/pdfs-fuse/src/sim/rng.rs new file mode 100644 index 0000000..3c856b9 --- /dev/null +++ b/crates/pdfs-fuse/src/sim/rng.rs @@ -0,0 +1,79 @@ +//! A small seeded random number generator, so a failing simulation is a seed +//! that can be replayed. SplitMix64: fast, good enough for choosing operations +//! and faults, and no dependency. + +#[derive(Clone, Debug)] +pub(crate) struct Rng(u64); + +impl Rng { + pub(crate) fn new(seed: u64) -> Self { + Self(seed) + } + + pub(crate) fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + + /// A value in `0..n`; `0` when `n` is `0`. + pub(crate) fn below(&mut self, n: u64) -> u64 { + if n == 0 { 0 } else { self.next_u64() % n } + } + + /// A value in `lo..=hi`. + pub(crate) fn between(&mut self, lo: u64, hi: u64) -> u64 { + lo + self.below(hi.saturating_sub(lo) + 1) + } + + /// True with probability `p`. + pub(crate) fn chance(&mut self, p: f64) -> bool { + p > 0.0 && (self.next_u64() >> 11) as f64 / (1u64 << 53) as f64 <= p + } + + pub(crate) fn pick<'a, T>(&mut self, items: &'a [T]) -> Option<&'a T> { + items.get(self.below(items.len() as u64) as usize) + } + + /// A child generator, so one consumer's draws do not shift another's. + pub(crate) fn fork(&mut self) -> Self { + Self::new(self.next_u64()) + } +} + +#[cfg(test)] +mod tests { + use super::Rng; + + #[test] + fn a_seed_replays_the_same_sequence() { + let (mut a, mut b) = (Rng::new(7), Rng::new(7)); + for _ in 0..100 { + assert_eq!(a.next_u64(), b.next_u64()); + } + assert_ne!(Rng::new(7).next_u64(), Rng::new(8).next_u64()); + } + + #[test] + fn bounded_draws_stay_in_range() { + let mut rng = Rng::new(1); + for _ in 0..1000 { + assert!(rng.below(5) < 5); + assert!((3..=9).contains(&rng.between(3, 9))); + } + assert_eq!(rng.below(0), 0); + assert!(!rng.chance(0.0)); + assert!(rng.chance(1.0)); + assert!([1, 2, 3].contains(rng.pick(&[1, 2, 3]).unwrap())); + assert_eq!(rng.pick::(&[]), None); + } + + #[test] + fn a_fork_draws_apart_from_its_parent() { + let mut parent = Rng::new(3); + let mut child = parent.fork(); + assert_ne!(parent.next_u64(), child.next_u64()); + } +}