diff --git a/crates/pdfs-fuse/src/sim/fake_drive.rs b/crates/pdfs-fuse/src/sim/fake_drive.rs index 3288f4b..ea853da 100644 --- a/crates/pdfs-fuse/src/sim/fake_drive.rs +++ b/crates/pdfs-fuse/src/sim/fake_drive.rs @@ -30,6 +30,10 @@ //! - **Non-uniform blocks** ([`FakeDrive::set_block_pattern`]; B84, B85, B87). //! - **A full account** ([`FakeDrive::set_quota`]) refusing uploads with //! `InsufficientQuota`. +//! +//! A test can also hold the next read of one node on its way to Drive +//! ([`FakeClient::hold_next_read`]), to make a change while the daemon waits +//! for the answer. use std::collections::{BTreeMap, HashMap}; use std::io::{Read, Write}; @@ -325,6 +329,7 @@ impl FakeDrive { entity_cache: Mutex::new(HashMap::new()), online: AtomicBool::new(true), requests: AtomicU64::new(0), + hold: Mutex::new(None), })) } @@ -559,6 +564,31 @@ struct ClientInner { entity_cache: Mutex>, online: AtomicBool, requests: AtomicU64, + /// The read [`FakeClient::hold_next_read`] is waiting for. + hold: Mutex>>, +} + +struct HoldState { + uid: NodeUid, + reached: AtomicBool, + released: AtomicBool, +} + +/// A read held on its way to Drive, from [`FakeClient::hold_next_read`]. It +/// goes on when this is dropped. +pub(crate) struct Held(Arc); + +impl Held { + /// Whether the read has come and is being held. + pub(crate) fn reached(&self) -> bool { + self.0.reached.load(Ordering::SeqCst) + } +} + +impl Drop for Held { + fn drop(&mut self) { + self.0.released.store(true, Ordering::SeqCst); + } } /// One daemon's connection to a [`FakeDrive`]. Cheap to clone; clones share @@ -575,6 +605,34 @@ impl FakeClient { *self.0.faults.lock() = faults; } + /// Hold the next read of `uid` alone, before it reaches Drive, until the + /// answer is dropped. Only one read is held at a time. + pub(crate) fn hold_next_read(&self, uid: &NodeUid) -> Held { + let state = Arc::new(HoldState { + uid: uid.clone(), + reached: AtomicBool::new(false), + released: AtomicBool::new(false), + }); + *self.0.hold.lock() = Some(state.clone()); + Held(state) + } + + /// Wait here if `uids` is the read a [`Held`] is for. + async fn wait_if_held(&self, uids: &[NodeUid]) { + let state = { + let mut hold = self.0.hold.lock(); + match hold.as_ref() { + Some(state) if uids == std::slice::from_ref(&state.uid) => hold.take(), + _ => None, + } + }; + let Some(state) = state else { return }; + state.reached.store(true, Ordering::SeqCst); + while !state.released.load(Ordering::SeqCst) { + tokio::time::sleep(Duration::from_millis(5)).await; + } + } + /// Requests made through this client so far. pub(crate) fn requests(&self) -> u64 { self.0.requests.load(Ordering::SeqCst) @@ -799,6 +857,7 @@ impl DriveApi for FakeClient { } async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result> { + self.wait_if_held(uids).await; let lag = self.lag(); let nodes: Vec = self .request(false, |server| { @@ -1459,6 +1518,31 @@ mod tests { }); } + #[test] + fn a_held_read_answers_with_what_drive_holds_when_released() { + 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(); + let held = client.hold_next_read(&file); + let read = tokio::spawn({ + let client = client.clone(); + let file = file.clone(); + async move { client.enumerate_nodes(std::slice::from_ref(&file)).await } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(held.reached()); + assert!(!read.is_finished()); + drive.device().trash("a"); + drop(held); + let nodes = read.await.unwrap().unwrap(); + assert!(nodes[0].trashed); + // Only the next read is held. + assert_eq!(client.enumerate_nodes(&[file]).await.unwrap().len(), 1); + }); + } + #[test] fn a_rename_lags_only_by_the_update_lag() { rt().block_on(async { diff --git a/crates/pdfs-fuse/src/sim/model.rs b/crates/pdfs-fuse/src/sim/model.rs index 588f5f4..ea96aae 100644 --- a/crates/pdfs-fuse/src/sim/model.rs +++ b/crates/pdfs-fuse/src/sim/model.rs @@ -224,7 +224,7 @@ impl Model { self.tree.keys().any(|other| is_below(other, path)) } - fn files(&self) -> Vec<&String> { + pub(crate) fn files(&self) -> Vec<&String> { self.tree .iter() .filter(|(_, entry)| entry.is_some()) @@ -318,7 +318,7 @@ impl Model { } /// Content for a write: short and recognisable, now and then empty. -fn data(rng: &mut Rng) -> Vec { +pub(crate) fn data(rng: &mut Rng) -> Vec { let len = if rng.chance(0.1) { 0 } else { diff --git a/crates/pdfs-fuse/src/sim/run.rs b/crates/pdfs-fuse/src/sim/run.rs index 38cae0d..f20883d 100644 --- a/crates/pdfs-fuse/src/sim/run.rs +++ b/crates/pdfs-fuse/src/sim/run.rs @@ -3,14 +3,16 @@ //! A run starts one or more daemons on a shared [`FakeDrive`], each owning a //! folder `c` that only it writes, so there is one writer per file. A seed //! draws a sequence of steps: syscalls through a client's mount, the link of a -//! client going down or up, a client restarting, a pause, or a settle (every -//! queue must drain within a budget). Every syscall is checked against the +//! client going down or up, a client restarting, a pause, a settle (every +//! queue must drain within a budget), or a file deleted while the drain holds +//! a write to it. Every syscall is checked against the //! client's [`Model`] as it happens. After the sequence every link comes up //! and the run checks: //! //! 1. **No loss** and **POSIX**: Drive's copy of each client's folder is what //! its model says. -//! 2. **No false conflicts**: no conflict copy anywhere. +//! 2. **No false conflicts**: no conflict copy anywhere, and the drain never +//! takes a client's own delete for another device's. //! 3. **Convergence**: each mount shows Drive's tree. //! 4. **Liveness**: no syscall took longer than its budget, settles drained in //! theirs, and every queue drained once the links were up. A call that @@ -42,9 +44,11 @@ use std::time::{Duration, Instant}; use std::collections::BTreeMap; +use proton_drive_rs::proton_sdk::ids::NodeUid; + use super::daemon::{Daemon, scratch, take_logged, wait_until}; use super::fake_drive::{Entry, FakeDrive, Faults}; -use super::model::{FsOp, Model, Tree, join, name_of, parent_of}; +use super::model::{FsOp, Model, Tree, data, join, name_of, parent_of}; use super::rng::Rng; use super::watchdog::{Hung, Watchdog}; @@ -52,9 +56,9 @@ use super::watchdog::{Hung, Watchdog}; struct Known { bug: &'static str, /// Whether a call that answered `errno` against the model's word, with - /// lines `logged` at `INFO` and above meanwhile, is this bug. `model` is from - /// before the call. - is: fn(model: &Model, op: &FsOp, errno: i32, logged: &[String]) -> bool, + /// lines `logged` at `INFO` and above meanwhile and `before` it, is this + /// bug. `model` is from before the call. + is: fn(model: &Model, op: &FsOp, errno: i32, logged: &[String], before: &[String]) -> bool, /// Whether it also left the mount where the model cannot follow, so the /// seed ends there. ends_seed: fn(model: &Model, op: &FsOp) -> bool, @@ -65,7 +69,7 @@ const KNOWN: &[Known] = &[ // sibling there with the new name refuses it. Known { bug: "B122", - is: |model, op, errno, _| { + is: |model, op, errno, _, _| { let FsOp::Rename { from, to } = op else { return false; }; @@ -85,7 +89,7 @@ const KNOWN: &[Known] = &[ // the replaced file landed after its op was dropped. Known { bug: "B129", - is: |model, op, errno, logged| { + is: |model, op, errno, logged, _| { let FsOp::Rename { to, .. } = op else { return false; }; @@ -101,7 +105,7 @@ const KNOWN: &[Known] = &[ // A write over a queued revision whose gaps are not filled yet is refused. Known { bug: "B123", - is: |_, _, errno, logged| { + is: |_, _, errno, logged, _| { errno == libc::EIO && logged .iter() @@ -115,7 +119,7 @@ const KNOWN: &[Known] = &[ // gone. Known { bug: "B125", - is: |_, _, errno, logged| { + is: |_, _, errno, logged, _| { matches!(errno, libc::EIO | libc::ENOENT) && logged.iter().any(|line| { line.starts_with("enumerate folder children failed") @@ -126,6 +130,23 @@ const KNOWN: &[Known] = &[ // as gone, so the mount no longer follows the model. ends_seed: |_, _| true, }, + // A restart forgets the listings served offline, so the first lookup + // lists a folder on Drive, which does not hold what is still queued for + // it. + Known { + bug: "B132", + is: |_, _, errno, _, before| { + errno == libc::ENOENT + && before + .iter() + .any(|line| line.starts_with("lost the connection to Proton")) + && before + .iter() + .any(|line| line.starts_with("restored pending ops")) + }, + // The listing stays without it until something relists the folder. + ends_seed: |_, _| true, + }, ]; /// An open bug in `docs/BUGS.md` that hangs a call, which the runs tolerate. @@ -159,6 +180,8 @@ pub(crate) struct Profile { /// Whether links go down and come back during the sequence. pub(crate) link_flaps: bool, pub(crate) restarts: bool, + /// Whether files are deleted while the drain holds a write to them. + pub(crate) deletes_mid_upload: bool, /// How long a settle may take; `None` leaves settles out. pub(crate) settle_budget: Option, /// How long one syscall may take. @@ -171,6 +194,10 @@ pub(crate) struct Profile { /// poll (`POLL_INTERVAL`) and change. const CONVERGE_BUDGET: Duration = Duration::from_secs(30); +/// How long the drain may take to pick up a write: its debounce, and then +/// some. +const PICKUP_BUDGET: Duration = Duration::from_secs(8); + /// Steps shown before a failure. All of them are written to `steps.log` in /// the state the failure keeps. const LOG_TAIL: usize = 40; @@ -221,6 +248,10 @@ struct Run { ended_by: Option<&'static str>, /// What the daemons logged at `INFO` and above so far. logged: Vec, + /// Files deleted while the drain held a write to them. + held: Vec, + /// Those of them made again before it went on. + remade: Vec, watchdog: Watchdog, } @@ -243,6 +274,8 @@ pub(crate) fn run( known: BTreeMap::new(), ended_by: None, logged: Vec::new(), + held: Vec::new(), + remade: Vec::new(), watchdog: Watchdog::new(&dir), }; let outcome = run.execute(&dir, &mut rng); @@ -343,17 +376,10 @@ impl Run { if let Some(budget) = self.profile.settle_budget && self.clients.iter().all(|client| client.online) { - self.note("settle".into()); - let start = Instant::now(); - if !wait_until(budget, || self.landed()) { - return Err(format!( - "liveness: the queues did not drain within {budget:?}\n{}", - self.difference() - )); - } - self.note(format!("settled in {:?}", start.elapsed())); + self.settle(budget)?; } } + 15..25 if self.profile.deletes_mid_upload => self.delete_mid_upload(index, rng)?, _ => { let op = self.clients[index].model.next_op(rng); self.syscall(index, op)?; @@ -362,6 +388,88 @@ impl Run { Ok(()) } + /// Wait until every queue is empty and Drive holds what the models say. + fn settle(&mut self, budget: Duration) -> Result<(), String> { + self.note("settle".into()); + let start = Instant::now(); + if !wait_until(budget, || self.landed()) { + return Err(format!( + "liveness: the queues did not drain within {budget:?}\n{}", + self.difference() + )); + } + self.note(format!("settled in {:?}", start.elapsed())); + Ok(()) + } + + /// Write a file Drive holds and delete it once the drain has picked the + /// write up. The drain's read of the node is held until the delete has + /// landed and a read shows it, so the drain finds the file trashed while + /// it holds a write that is no longer queued (B105). Half the time the + /// file is made again meanwhile, which queues a write under the op id the + /// delete freed (B133). + fn delete_mid_upload(&mut self, index: usize, rng: &mut Rng) -> Result<(), String> { + let file = rng + .pick(&self.clients[index].model.files()) + .map(|path| (*path).clone()); + let (Some(path), Some(budget), true) = ( + file, + self.profile.settle_budget, + self.clients.iter().all(|client| client.online), + ) else { + let op = self.clients[index].model.next_op(rng); + return self.syscall(index, op); + }; + self.settle(budget)?; + let Some(uid) = self.drive.lookup(&join(&owned_folder(index), &path)) else { + return Err(format!( + "no loss: client {index}'s {path} is not on Drive after a settle" + )); + }; + let held = self.clients[index].daemon().client.hold_next_read(&uid); + self.note(format!("client {index}: hold the next read of {path}")); + let mut bytes = data(rng); + bytes.push(b'.'); + self.syscall( + index, + FsOp::Write { + path: path.clone(), + offset: 0, + data: bytes, + }, + )?; + if self.ended_by.is_some() { + return Ok(()); + } + if !wait_until(PICKUP_BUDGET, || held.reached()) { + self.note(format!("client {index}: nothing read {path} back")); + return Ok(()); + } + self.syscall(index, FsOp::Unlink { path: path.clone() })?; + if self.ended_by.is_some() { + return Ok(()); + } + self.held.push(uid.clone()); + if rng.chance(0.5) { + self.syscall( + index, + FsOp::Create { + path: path.clone(), + data: data(rng), + }, + )?; + self.remade.push(uid); + } + let faults = (self.profile.faults)(); + let latency = Duration::from_millis(faults.latency_ms.1); + std::thread::sleep(faults.listing_lag + latency); + drop(held); + self.note(format!("client {index}: the read of {path} goes on")); + // Nothing else is queued before the drain has answered. + std::thread::sleep(2 * latency + Duration::from_millis(50)); + Ok(()) + } + fn syscall(&mut self, index: usize, op: FsOp) -> Result<(), String> { let client = &mut self.clients[index]; let root = client.root(index); @@ -369,6 +477,7 @@ impl Run { let before = client.model.content(target(&op)).cloned(); let expected = client.model.apply(&op); self.logged.extend(take_logged()); + let logged_before = self.logged.len(); let start = Instant::now(); let actual = perform(&root, &op, before.as_deref()); let took = start.elapsed(); @@ -383,9 +492,9 @@ impl Run { if actual != expected && let Err(errno) = actual && tolerate_known() - && let Some(known) = KNOWN - .iter() - .find(|known| (known.is)(&model, &op, errno, &logged)) + && let Some(known) = KNOWN.iter().find(|known| { + (known.is)(&model, &op, errno, &logged, &self.logged[..logged_before]) + }) { *self.known.entry(known.bug).or_default() += 1; if (known.ends_seed)(&model, &op) { @@ -431,6 +540,45 @@ impl Run { self.difference() )); } + // Each client's folder is its own, so a file of it gone from Drive was + // deleted by the client. + let own_deletes: Vec = self + .logged + .iter() + .filter(|line| is_own_delete_kept(line)) + .cloned() + .collect(); + for line in own_deletes { + let bug = if self + .remade + .iter() + .any(|uid| line.contains(&format!(" uid={uid} "))) + { + // Made again while the drain held its write: the new file's + // write took the freed op id. + "B133" + } else if !self + .held + .iter() + .any(|uid| line.contains(&format!(" uid={uid} "))) + && !line.contains(" name=\"recovered-") + { + // The tree still placed the file, so the drain looked before + // the unlink had dropped its write. + "B134" + } else { + return Err(format!( + "no false conflicts: the drain took a client's own delete for another device's: {line}" + )); + }; + if !tolerate_known() { + return Err(format!( + "no false conflicts: the drain took a client's own delete for another device's: {line}" + )); + } + *self.known.entry(bug).or_default() += 1; + self.note(format!("known bug {bug}: {line}")); + } let drive = self.drive_tree(); let mut explained = Vec::new(); for index in 0..self.clients.len() { @@ -507,6 +655,13 @@ impl Run { /// /// - B131: the tree kept the create's placeholder, so a write based on it /// found Drive's copy changed. + /// + /// A conflict copy of a file the client deleted takes the drain saying it + /// kept that file's write as the copy, because Drive had the file in the + /// trash while the tree still placed it: + /// + /// - B134: the unlink trashed the file on Drive before it dropped the + /// write. fn known_damage(&self, index: usize, drive: &Tree) -> Vec { if !tolerate_known() { return Vec::new(); @@ -548,24 +703,43 @@ impl Run { line.starts_with("queued write landed as a conflict copy") && line.ends_with(&alt) }) }; + let kept_deleted = |copy: &str| { + let alt = format!(" alt={:?}", name_of(copy)); + self.logged.iter().any(|line| { + line.starts_with("queued write landed as a conflict copy") + && line.ends_with(&alt) + && line.split(' ').any(|field| { + field.starts_with("uid=") + && self.logged.iter().any(|kept| { + is_own_delete_kept(kept) + && !kept.contains(" name=\"recovered-") + && kept.split(' ').any(|other| other == field) + }) + }) + }) + }; let mut damage = Vec::new(); for (copy, content) in &drive { - if let Some(content) = content - && is_conflict_copy(copy) - && !model.contains_key(copy) - && unread_create - && kept_write(copy) - { - damage.push(KnownDamage { - bug: "B131", - path: copy.clone(), - copy: Some(copy.clone()), - change: FsOp::Create { - path: copy.clone(), - data: content.clone(), - }, - }); + let Some(content) = content else { continue }; + if !is_conflict_copy(copy) || model.contains_key(copy) { + continue; } + let bug = if unread_create && kept_write(copy) { + "B131" + } else if kept_deleted(copy) { + "B134" + } else { + continue; + }; + damage.push(KnownDamage { + bug, + path: copy.clone(), + copy: Some(copy.clone()), + change: FsOp::Create { + path: copy.clone(), + data: content.clone(), + }, + }); } for (path, content) in model { let Some(content) = content else { continue }; @@ -775,6 +949,14 @@ fn owned_folder(index: usize) -> String { format!("c{index}") } +/// Whether `line` is the drain keeping a queued write as a conflict copy +/// because Drive no longer holds the file, or holds it in the trash. +fn is_own_delete_kept(line: &str) -> bool { + line.starts_with("queued write conflicts; keeping a conflict copy") + && (line.contains(" reason=\"the file was trashed remotely\"") + || line.contains(" reason=\"the file no longer exists remotely\"")) +} + fn is_conflict_copy(path: &str) -> bool { path.split('/') .any(|name| name.contains("(sync-conflict") || name.starts_with("recovered-")) @@ -867,6 +1049,20 @@ mod tests { faults: Faults::lan, link_flaps: false, restarts: false, + deletes_mid_upload: false, + settle_budget: Some(Duration::from_secs(4)), + syscall_budget: Duration::from_secs(2), + drain_budget: Duration::from_secs(20), + }; + + static ONE_CLIENT_DELETES: Profile = Profile { + name: "one-client-deletes", + clients: 1, + steps: 30, + faults: Faults::lan, + link_flaps: false, + restarts: false, + deletes_mid_upload: true, settle_budget: Some(Duration::from_secs(4)), syscall_budget: Duration::from_secs(2), drain_budget: Duration::from_secs(20), @@ -879,6 +1075,7 @@ mod tests { faults: Faults::wifi, link_flaps: true, restarts: true, + deletes_mid_upload: false, settle_budget: None, syscall_budget: Duration::from_secs(10), drain_budget: Duration::from_secs(120), @@ -891,6 +1088,7 @@ mod tests { faults: Faults::lan, link_flaps: true, restarts: true, + deletes_mid_upload: false, settle_budget: Some(Duration::from_secs(4)), syscall_budget: Duration::from_secs(2), drain_budget: Duration::from_secs(60), @@ -902,6 +1100,12 @@ mod tests { run_seeds(&ONE_CLIENT_LAN, 12); } + #[test] + #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse --lib sim::run -- --ignored`"] + fn one_client_deleting_files_as_they_upload() { + run_seeds(&ONE_CLIENT_DELETES, 3); + } + #[test] #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse --lib sim::run -- --ignored`"] fn one_client_on_a_flaky_link() { @@ -921,6 +1125,19 @@ mod tests { assert!(!is_conflict_copy("c0/x/a.txt")); } + #[test] + fn a_write_kept_because_its_file_is_gone_is_recognised() { + let kept = "queued write conflicts; keeping a conflict copy uid=simvol~L10 \ + name=\"f.txt\" reason=\"the file was trashed remotely\""; + assert!(is_own_delete_kept(kept)); + assert!(is_own_delete_kept( + &kept.replace("was trashed", "no longer exists") + )); + assert!(!is_own_delete_kept( + &kept.replace("the file was trashed remotely", "revision changed") + )); + } + #[test] fn an_invalidation_stuck_on_a_page_lock_is_b130() { let b130 = |stacks: &str| KNOWN_HANGS.iter().any(|known| (known.is)(stacks));