diff --git a/crates/pdfs-fuse/src/sim/daemon.rs b/crates/pdfs-fuse/src/sim/daemon.rs index 0f84204..05f9ca2 100644 --- a/crates/pdfs-fuse/src/sim/daemon.rs +++ b/crates/pdfs-fuse/src/sim/daemon.rs @@ -15,6 +15,7 @@ use std::time::{Duration, Instant}; use pdfs_core::cache::ContentCache; use pdfs_core::config::AppConfig; +use pdfs_core::control::{self, PendingOpInfo, Request, Response}; use pdfs_core::db::Db; use proton_drive_rs::ProtonDriveClient; use proton_drive_rs::proton_sdk::config::ProtonClientConfiguration; @@ -26,9 +27,14 @@ 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); +/// How long a restart may wait for the last run to let go of its database. +const LAST_RUN_DEADLINE: Duration = Duration::from_secs(30); + pub(crate) struct Daemon { pub(crate) client: FakeClient, pub(crate) mountpoint: PathBuf, + dir: PathBuf, + socket: PathBuf, stop: SyncSender<()>, mount: Option>>, rt: Option, @@ -41,8 +47,22 @@ impl Daemon { pub(crate) fn start(dir: &Path, client: FakeClient) -> io::Result { log_to_test_output(); let mountpoint = dir.join("mnt"); + let socket = dir.join("control.sock"); std::fs::create_dir_all(&mountpoint)?; - let db = Arc::new(Db::open(&dir.join("pdfs.db")).map_err(io::Error::other)?); + // A thread of the last run that is still in a request holds the + // database, and with it the single-writer lock, until it returns. A + // real restart waits for the old process to exit in the same way. + let mut db = Db::open(&dir.join("pdfs.db")); + let deadline = Instant::now() + LAST_RUN_DEADLINE; + while db + .as_ref() + .is_err_and(|error| error.to_string().contains("already using")) + && Instant::now() < deadline + { + std::thread::sleep(Duration::from_millis(100)); + db = Db::open(&dir.join("pdfs.db")); + } + let db = Arc::new(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() @@ -54,7 +74,7 @@ impl Daemon { let drive = Arc::new(client.clone()); let handle = rt.handle().clone(); let mountpoint = mountpoint.clone(); - let socket = dir.join("control.sock"); + let socket = socket.clone(); std::thread::Builder::new() .name("sim-mount".into()) .spawn(move || { @@ -92,12 +112,28 @@ impl Daemon { Ok(Self { client, mountpoint, + dir: dir.to_path_buf(), + socket, stop, mount: Some(mount), rt: Some(rt), }) } + /// What the daemon still owes Drive, as `pdfs sync queue` shows it. + pub(crate) fn pending(&self) -> io::Result> { + match control::send(&self.socket, &Request::ListPendingOps).map_err(io::Error::other)? { + Response::PendingOps { items } => Ok(items), + other => Err(io::Error::other(format!("unexpected answer {other:?}"))), + } + } + + /// Stop the daemon and start it again on the same state and link. + pub(crate) fn restart(mut self) -> io::Result { + self.shut_down()?; + Self::start(&self.dir.clone(), self.client.clone()) + } + /// Stop the daemon the way `systemctl --user stop` does. pub(crate) fn stop(mut self) -> io::Result { self.shut_down() @@ -134,13 +170,50 @@ fn is_mounted(path: &Path) -> bool { } /// The daemons' logs, filtered by `RUST_LOG` and shown with `--nocapture`. +/// What they log at `INFO` and above is also kept for [`take_logged`], +/// whatever the filter. fn log_to_test_output() { - let _ = tracing_subscriber::fmt() - .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) - .with_test_writer() + use tracing_subscriber::prelude::*; + let _ = tracing_subscriber::registry() + .with( + tracing_subscriber::fmt::layer() + .with_test_writer() + .with_filter(tracing_subscriber::EnvFilter::from_default_env()), + ) + .with(KeepLogged.with_filter(tracing_subscriber::filter::LevelFilter::INFO)) .try_init(); } +/// What the daemons logged at `INFO` and above since the last call, each as +/// its message followed by its fields: `message key=value ...`. +pub(crate) fn take_logged() -> Vec { + std::mem::take(&mut *LOGGED.lock()) +} + +static LOGGED: parking_lot::Mutex> = parking_lot::Mutex::new(Vec::new()); + +struct KeepLogged; + +impl tracing_subscriber::Layer for KeepLogged { + fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) { + let mut message = Message(String::new()); + event.record(&mut message); + LOGGED.lock().push(message.0); + } +} + +struct Message(String); + +impl tracing::field::Visit for Message { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + self.0.insert_str(0, &format!("{value:?}")); + } else { + self.0.push_str(&format!(" {}={value:?}", field.name())); + } + } +} + /// 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 { diff --git a/crates/pdfs-fuse/src/sim/fake_drive.rs b/crates/pdfs-fuse/src/sim/fake_drive.rs index 2c98cfb..3288f4b 100644 --- a/crates/pdfs-fuse/src/sim/fake_drive.rs +++ b/crates/pdfs-fuse/src/sim/fake_drive.rs @@ -9,12 +9,17 @@ //! 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). +//! ago, so a create is not listed at once and a trashed node still is (B118, +//! B48). A rename, move or new revision can lag by [`Faults::update_lag`], +//! which Drive has not been seen doing; the presets leave it at zero. //! - **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). +//! - **Re-stamped modification time.** A rename or move sets a node's +//! modification time to the time it happened, though its revision stays +//! (B113). //! - **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 @@ -104,13 +109,23 @@ impl Server { } /// The node as a read `lag` behind the present sees it. - fn visible(&self, uid: &NodeUid, lag: Duration, now: Instant) -> Option { + fn visible(&self, uid: &NodeUid, lag: Lag, now: Instant) -> Option { let versions = self.nodes.get(uid)?; - versions - .iter() + let live = |remote: &Option| remote.as_ref().is_some_and(|r| !r.trashed); + let update = |i: usize| i > 0 && live(&versions[i - 1].1) && live(&versions[i].1); + let passed = + |i: usize| versions[i].0 + if update(i) { lag.update } else { lag.appear } <= now; + // An update shows only once the node it updates has appeared. + let appeared = |mut i: usize| { + while update(i) { + i -= 1; + } + passed(i) + }; + (0..versions.len()) .rev() - .find(|(at, _)| *at + lag <= now) - .and_then(|(_, remote)| remote.clone()) + .find(|&i| passed(i) && appeared(i)) + .and_then(|i| versions[i].1.clone()) } fn live_child_named(&self, parent: &NodeUid, name: &str) -> Option { @@ -449,17 +464,29 @@ impl Device<'_> { let mut server = self.0.server.lock(); let mut remote = server.current(&uid).cloned().expect("live node"); remote.name = new_name.to_owned(); + remote.modified = now_secs(); server.put(&uid, Some(remote)); } } +/// How far behind the present a read is, by the kind of change. +#[derive(Clone, Copy, Debug)] +struct Lag { + appear: Duration, + update: Duration, +} + /// 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. + /// How far behind the present reads are about a node appearing or going + /// to the trash. pub(crate) listing_lag: Duration, + /// How far behind the present reads are about a live node's name, folder + /// or revision. + pub(crate) update_lag: Duration, /// How long a change takes to reach the event feed. pub(crate) event_delay: Duration, /// Chance that an event is delivered twice. @@ -483,6 +510,7 @@ impl Faults { Self { latency_ms: (0, 0), listing_lag: Duration::ZERO, + update_lag: Duration::ZERO, event_delay: Duration::ZERO, duplicate_events: 0.0, reorder_events: 0.0, @@ -590,8 +618,12 @@ impl FakeClient { outcome } - fn lag(&self) -> Duration { - self.0.faults.lock().listing_lag + fn lag(&self) -> Lag { + let faults = self.0.faults.lock(); + Lag { + appear: faults.listing_lag, + update: faults.update_lag, + } } /// The `(parent, name)` a rename or move of `uid` sends as its original @@ -664,6 +696,7 @@ impl FakeClient { } remote.parent = parent; remote.name = name; + remote.modified = now_secs(); server.put(uid, Some(remote)); Ok(()) } @@ -1240,6 +1273,7 @@ fn io_error(error: std::io::Error) -> ProtonError { } fn api(code: ResponseCode, message: &str) -> ProtonError { + tracing::debug!(?code, message, "fake drive refused a request"); let http_status = match code { ResponseCode::DoesNotExist => 404, ResponseCode::AlreadyExists | ResponseCode::InvalidRequirements => 422, @@ -1425,6 +1459,35 @@ mod tests { }); } + #[test] + fn a_rename_lags_only_by_the_update_lag() { + rt().block_on(async { + let drive = FakeDrive::new(); + let client = drive.client(1, Faults::none()); + let file = drive.device().write("a", b"1"); + client.rename_node(&file, "b", None).await.unwrap(); + let names = || async { + let nodes = client + .enumerate_nodes_light(std::slice::from_ref(&file)) + .await + .unwrap(); + nodes.into_iter().map(|n| n.name).collect::>() + }; + let lagging = |listing_lag, update_lag| Faults { + listing_lag, + update_lag, + ..Faults::none() + }; + let minute = Duration::from_secs(60); + client.set_faults(lagging(minute, Duration::ZERO)); + assert!(names().await.is_empty()); + client.set_faults(lagging(Duration::ZERO, Duration::ZERO)); + assert_eq!(names().await, ["b"]); + client.set_faults(lagging(Duration::ZERO, minute)); + assert_eq!(names().await, ["a"]); + }); + } + #[test] fn every_client_sees_every_change_on_the_feed() { rt().block_on(async { diff --git a/crates/pdfs-fuse/src/sim/mod.rs b/crates/pdfs-fuse/src/sim/mod.rs index 3224e25..b898582 100644 --- a/crates/pdfs-fuse/src/sim/mod.rs +++ b/crates/pdfs-fuse/src/sim/mod.rs @@ -3,4 +3,7 @@ pub(crate) mod daemon; pub(crate) mod fake_drive; +pub(crate) mod model; pub(crate) mod rng; +pub(crate) mod run; +pub(crate) mod watchdog; diff --git a/crates/pdfs-fuse/src/sim/model.rs b/crates/pdfs-fuse/src/sim/model.rs new file mode 100644 index 0000000..588f5f4 --- /dev/null +++ b/crates/pdfs-fuse/src/sim/model.rs @@ -0,0 +1,445 @@ +//! The reference file system of the simulation (`docs/MILESTONE-3.0.0.md` +//! §8.2, property 4): what each syscall a simulated client makes should +//! answer, and the tree it should leave, on a POSIX file system. +//! +//! A [`Model`] covers one client's own folder; paths are relative to it. The +//! same type is used for the trees read back from a mount and from Drive. + +use std::collections::BTreeMap; + +use super::rng::Rng; + +/// A tree by path: `None` is a folder, `Some` a file's content. +pub(crate) type Tree = BTreeMap>>; + +/// One file system call, or the few a program makes in a row. +#[derive(Clone, Debug)] +pub(crate) enum FsOp { + /// `open(O_CREAT | O_TRUNC)`, write `data`, close. + Create { + path: String, + data: Vec, + }, + /// Open an existing file, write `data` at `offset` (past the end leaves + /// a hole), close. + Write { + path: String, + offset: u64, + data: Vec, + }, + /// Open an existing file and `ftruncate` it to `len`. + Truncate { + path: String, + len: u64, + }, + Mkdir { + path: String, + }, + Rename { + from: String, + to: String, + }, + Unlink { + path: String, + }, + /// Open a file, unlink it, append `data`, read it all back, close. + UnlinkOpen { + path: String, + data: Vec, + }, + Rmdir { + path: String, + }, +} + +impl FsOp { + /// The paths the op names, for the log. + pub(crate) fn describe(&self) -> String { + match self { + FsOp::Create { path, data } => format!("create {path} ({} bytes)", data.len()), + FsOp::Write { path, offset, data } => { + format!("write {path} at {offset} ({} bytes)", data.len()) + } + FsOp::Truncate { path, len } => format!("truncate {path} to {len}"), + FsOp::Mkdir { path } => format!("mkdir {path}"), + FsOp::Rename { from, to } => format!("rename {from} -> {to}"), + FsOp::Unlink { path } => format!("unlink {path}"), + FsOp::UnlinkOpen { path, data } => { + format!("unlink {path} while open, then append {} bytes", data.len()) + } + FsOp::Rmdir { path } => format!("rmdir {path}"), + } + } +} + +/// The tree one client should have, and what its syscalls should answer. +#[derive(Clone, Debug, Default)] +pub(crate) struct Model { + tree: Tree, +} + +/// Names are drawn from small pools, so ops collide: a create over an existing +/// file, a rename over another, a mkdir of a name that is taken. +const FILE_NAMES: [&str; 6] = ["a.txt", "b.txt", "c.bin", "d", "e.md", "f.txt"]; +const DIR_NAMES: [&str; 3] = ["x", "y", "z"]; + +impl Model { + pub(crate) fn tree(&self) -> &Tree { + &self.tree + } + + pub(crate) fn content(&self, path: &str) -> Option<&Vec> { + self.tree.get(path)?.as_ref() + } + + /// What `op` should answer on a POSIX file system, applying it on success. + pub(crate) fn apply(&mut self, op: &FsOp) -> Result<(), i32> { + match op { + FsOp::Create { path, data } => { + self.check_parent(path)?; + match self.tree.get(path) { + Some(None) => Err(libc::EISDIR), + _ => { + self.tree.insert(path.clone(), Some(data.clone())); + Ok(()) + } + } + } + FsOp::Write { path, offset, data } => { + let content = self.file_mut(path)?; + // Writing nothing changes nothing, even past the end. + if data.is_empty() { + return Ok(()); + } + let end = *offset as usize + data.len(); + if content.len() < end { + content.resize(end, 0); + } + content[*offset as usize..end].copy_from_slice(data); + Ok(()) + } + FsOp::Truncate { path, len } => { + self.file_mut(path)?.resize(*len as usize, 0); + Ok(()) + } + FsOp::Mkdir { path } => { + self.check_parent(path)?; + if self.tree.contains_key(path) { + return Err(libc::EEXIST); + } + self.tree.insert(path.clone(), None); + Ok(()) + } + FsOp::Rename { from, to } => self.rename(from, to), + FsOp::Unlink { path } | FsOp::UnlinkOpen { path, .. } => { + self.check_parent(path)?; + match self.tree.get(path) { + None => Err(libc::ENOENT), + Some(None) => Err(libc::EISDIR), + Some(Some(_)) => { + self.tree.remove(path); + Ok(()) + } + } + } + FsOp::Rmdir { path } => { + self.check_parent(path)?; + match self.tree.get(path) { + None => Err(libc::ENOENT), + Some(Some(_)) => Err(libc::ENOTDIR), + Some(None) if self.has_children(path) => Err(libc::ENOTEMPTY), + Some(None) => { + self.tree.remove(path); + Ok(()) + } + } + } + } + } + + fn rename(&mut self, from: &str, to: &str) -> Result<(), i32> { + self.check_parent(from)?; + let Some(moving) = self.tree.get(from).cloned() else { + return Err(libc::ENOENT); + }; + self.check_parent(to)?; + if from == to { + return Ok(()); + } + if is_below(to, from) { + return Err(libc::EINVAL); + } + match (&moving, self.tree.get(to)) { + (None, Some(Some(_))) => return Err(libc::ENOTDIR), + (Some(_), Some(None)) => return Err(libc::EISDIR), + (None, Some(None)) if self.has_children(to) => return Err(libc::ENOTEMPTY), + _ => {} + } + let below: Vec<(String, Option>)> = self + .tree + .iter() + .filter(|(path, _)| is_below(path, from)) + .map(|(path, entry)| (path.clone(), entry.clone())) + .collect(); + self.tree.remove(from); + self.tree.insert(to.to_owned(), moving); + for (path, entry) in below { + self.tree.remove(&path); + self.tree + .insert(format!("{to}{}", &path[from.len()..]), entry); + } + Ok(()) + } + + fn file_mut(&mut self, path: &str) -> Result<&mut Vec, i32> { + self.check_parent(path)?; + match self.tree.get_mut(path) { + None => Err(libc::ENOENT), + Some(None) => Err(libc::EISDIR), + Some(Some(content)) => Ok(content), + } + } + + fn check_parent(&self, path: &str) -> Result<(), i32> { + let mut at = String::new(); + let mut names = path.split('/').peekable(); + while let Some(name) = names.next() { + if names.peek().is_none() { + break; + } + if !at.is_empty() { + at.push('/'); + } + at.push_str(name); + match self.tree.get(&at) { + None => return Err(libc::ENOENT), + Some(Some(_)) => return Err(libc::ENOTDIR), + Some(None) => {} + } + } + Ok(()) + } + + fn has_children(&self, path: &str) -> bool { + self.tree.keys().any(|other| is_below(other, path)) + } + + fn files(&self) -> Vec<&String> { + self.tree + .iter() + .filter(|(_, entry)| entry.is_some()) + .map(|(path, _)| path) + .collect() + } + + /// Folders, the client's own root (`""`) first. + fn dirs(&self) -> Vec { + std::iter::once(String::new()) + .chain( + self.tree + .iter() + .filter(|(_, entry)| entry.is_none()) + .map(|(path, _)| path.clone()), + ) + .collect() + } + + /// A random op, mostly one that should succeed. + pub(crate) fn next_op(&self, rng: &mut Rng) -> FsOp { + let dirs = self.dirs(); + let files = self.files(); + let in_dir = |rng: &mut Rng, names: &[&str]| { + let parent = rng.pick(&dirs).expect("the root is always there"); + join(parent, rng.pick(names).expect("names")) + }; + let a_file = |rng: &mut Rng| rng.pick(&files).map(|path| (*path).clone()); + let roll = rng.below(100); + let op = match roll { + 0..30 => Some(FsOp::Create { + path: in_dir(rng, &FILE_NAMES), + data: data(rng), + }), + 30..45 => a_file(rng).map(|path| { + let len = self.content(&path).map_or(0, Vec::len) as u64; + FsOp::Write { + offset: rng.below(len + 8), + path, + data: data(rng), + } + }), + 45..50 => a_file(rng).map(|path| FsOp::Truncate { + len: rng.below(32), + path, + }), + 50..60 => Some(FsOp::Mkdir { + path: in_dir(rng, &DIR_NAMES), + }), + // A rename in place, over another file as often as not. + 60..72 => a_file(rng).map(|from| FsOp::Rename { + to: join(parent_of(&from), rng.pick(&FILE_NAMES).expect("names")), + from, + }), + // A move to another folder, renamed or not. + 72..82 => a_file(rng).map(|from| { + let name = if rng.chance(0.5) { + name_of(&from) + } else { + rng.pick(&FILE_NAMES).expect("names") + }; + let parent = rng.pick(&dirs).expect("the root is always there"); + FsOp::Rename { + to: join(parent, name), + from, + } + }), + // A folder renamed or moved, with what is in it. + 82..86 => { + let from = rng.pick(&dirs[1..]).cloned(); + from.map(|from| FsOp::Rename { + to: in_dir(rng, &DIR_NAMES), + from, + }) + } + 86..94 => a_file(rng).map(|path| FsOp::Unlink { path }), + 94..97 => a_file(rng).map(|path| FsOp::UnlinkOpen { + path, + data: data(rng), + }), + _ => rng + .pick(&dirs[1..]) + .cloned() + .map(|path| FsOp::Rmdir { path }), + }; + op.unwrap_or_else(|| FsOp::Create { + path: in_dir(rng, &FILE_NAMES), + data: data(rng), + }) + } +} + +/// Content for a write: short and recognisable, now and then empty. +fn data(rng: &mut Rng) -> Vec { + let len = if rng.chance(0.1) { + 0 + } else { + rng.between(1, 48) + }; + let tag = rng.next_u64(); + (0..len).map(|i| b'a' + ((tag + i) % 26) as u8).collect() +} + +pub(crate) fn join(parent: &str, name: &str) -> String { + if parent.is_empty() { + name.to_owned() + } else { + format!("{parent}/{name}") + } +} + +pub(crate) fn parent_of(path: &str) -> &str { + path.rsplit_once('/').map_or("", |(parent, _)| parent) +} + +pub(crate) fn name_of(path: &str) -> &str { + path.rsplit_once('/').map_or(path, |(_, name)| name) +} + +fn is_below(path: &str, ancestor: &str) -> bool { + path.len() > ancestor.len() + 1 + && path.starts_with(ancestor) + && path.as_bytes()[ancestor.len()] == b'/' +} + +#[cfg(test)] +mod tests { + use super::*; + + fn create(path: &str, data: &[u8]) -> FsOp { + FsOp::Create { + path: path.into(), + data: data.to_vec(), + } + } + + #[test] + fn a_rename_over_a_file_replaces_it_and_a_folder_moves_with_its_contents() { + let mut model = Model::default(); + model.apply(&FsOp::Mkdir { path: "x".into() }).unwrap(); + model.apply(&create("x/a", b"1")).unwrap(); + model.apply(&create("b", b"2")).unwrap(); + model + .apply(&FsOp::Rename { + from: "b".into(), + to: "x/a".into(), + }) + .unwrap(); + assert_eq!(model.content("x/a"), Some(&b"2".to_vec())); + model + .apply(&FsOp::Rename { + from: "x".into(), + to: "y".into(), + }) + .unwrap(); + assert_eq!(model.tree().keys().collect::>(), vec!["y", "y/a"]); + } + + #[test] + fn calls_fail_the_way_posix_says() { + let mut model = Model::default(); + model.apply(&FsOp::Mkdir { path: "x".into() }).unwrap(); + model.apply(&create("x/a", b"1")).unwrap(); + assert_eq!( + model.apply(&FsOp::Mkdir { path: "x".into() }), + Err(libc::EEXIST) + ); + assert_eq!(model.apply(&create("x", b"")), Err(libc::EISDIR)); + assert_eq!(model.apply(&create("x/a/b", b"")), Err(libc::ENOTDIR)); + assert_eq!(model.apply(&create("q/a", b"")), Err(libc::ENOENT)); + assert_eq!( + model.apply(&FsOp::Rmdir { path: "x".into() }), + Err(libc::ENOTEMPTY) + ); + assert_eq!( + model.apply(&FsOp::Unlink { path: "x".into() }), + Err(libc::EISDIR) + ); + let into_itself = FsOp::Rename { + from: "x".into(), + to: "x/y".into(), + }; + assert_eq!(model.apply(&into_itself), Err(libc::EINVAL)); + } + + #[test] + fn a_write_past_the_end_leaves_a_hole() { + let mut model = Model::default(); + model.apply(&create("a", b"ab")).unwrap(); + let write = FsOp::Write { + path: "a".into(), + offset: 4, + data: b"z".to_vec(), + }; + model.apply(&write).unwrap(); + assert_eq!(model.content("a"), Some(&b"ab\0\0z".to_vec())); + let nothing = FsOp::Write { + path: "a".into(), + offset: 9, + data: Vec::new(), + }; + model.apply(¬hing).unwrap(); + assert_eq!(model.content("a"), Some(&b"ab\0\0z".to_vec())); + } + + #[test] + fn generated_ops_mostly_succeed() { + let mut rng = Rng::new(9); + let mut model = Model::default(); + let ok = (0..500) + .filter(|_| { + let op = model.next_op(&mut rng); + model.apply(&op).is_ok() + }) + .count(); + assert!(ok > 350, "{ok} of 500 succeeded"); + } +} diff --git a/crates/pdfs-fuse/src/sim/run.rs b/crates/pdfs-fuse/src/sim/run.rs new file mode 100644 index 0000000..38cae0d --- /dev/null +++ b/crates/pdfs-fuse/src/sim/run.rs @@ -0,0 +1,934 @@ +//! The simulation runs of `docs/MILESTONE-3.0.0.md` §8.2. +//! +//! 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'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. +//! 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 +//! never returns is caught by the [`Watchdog`], which keeps the stacks. +//! +//! A restart here is a stop and a start: the next run finds what the last +//! one wrote down, but in-flight work is let finish rather than cut. A kill +//! at an arbitrary point needs the daemon in a process of its own. +//! +//! A call that answers otherwise than the model says fails the run, unless it +//! is one of the [`KNOWN`] open bugs. Those are counted instead and the model +//! takes the call back, or the seed ends there when the bug leaves the mount +//! where the model cannot follow, so the runs stay a regression net until +//! they are fixed. A file gone, or a conflict copy the drain made of one of +//! the client's own, is counted too when a known bug explains it +//! (`Run::known_damage`), and so is a hang the stacks show is one of the +//! [`KNOWN_HANGS`]. `PDFS_SIM_KNOWN=fail` fails on all of them. +//! +//! The runs are ignored by default because they mount FUSE; run them with +//! `cargo test -p pdfs-fuse --lib sim::run -- --ignored --test-threads=1`. +//! `PDFS_SIM_SEEDS=n` sets how many seeds each runs, `PDFS_SIM_SEED=s` replays +//! one, and `RUST_LOG=pdfs_fuse=debug` with `--nocapture` shows the daemons' +//! logs. + +use std::fs::OpenOptions; +use std::io::{self, Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +use std::collections::BTreeMap; + +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::rng::Rng; +use super::watchdog::{Hung, Watchdog}; + +/// An open bug in `docs/BUGS.md` the runs tolerate. +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, + /// Whether it also left the mount where the model cannot follow, so the + /// seed ends there. + ends_seed: fn(model: &Model, op: &FsOp) -> bool, +} + +const KNOWN: &[Known] = &[ + // A rename across folders renames in the source folder first, so a + // sibling there with the new name refuses it. + Known { + bug: "B122", + is: |model, op, errno, _| { + let FsOp::Rename { from, to } = op else { + return false; + }; + errno == libc::EIO + && parent_of(from) != parent_of(to) + && name_of(from) != name_of(to) + && model + .tree() + .contains_key(&join(parent_of(from), name_of(to))) + }, + // The destination it was replacing is restored on Drive but stays + // gone from the mount (B124). + ends_seed: |model, op| matches!(op, FsOp::Rename { to, .. } if model.tree().contains_key(to)), + }, + // A replacing rename trashes what it replaces first, so Drive refuses the + // name only when it holds a node the mount has forgotten: the create of + // the replaced file landed after its op was dropped. + Known { + bug: "B129", + is: |model, op, errno, logged| { + let FsOp::Rename { to, .. } = op else { + return false; + }; + errno == libc::EIO + && model.tree().contains_key(to) + && logged + .iter() + .any(|line| line.starts_with("rename failed") && line.contains("AlreadyExists")) + }, + // The file it was replacing is gone from the mount (B124). + ends_seed: |_, _| true, + }, + // A write over a queued revision whose gaps are not filled yet is refused. + Known { + bug: "B123", + is: |_, _, errno, logged| { + errno == libc::EIO + && logged + .iter() + .any(|line| line.starts_with("refusing write over incomplete queued revision")) + }, + ends_seed: |_, _| false, + }, + // A folder made offline is listed on Drive once its listing has been + // invalidated, and Drive does not know it, or does not list it yet. Only + // the client writes its folder, so a folder of its own is never really + // gone. + Known { + bug: "B125", + is: |_, _, errno, logged| { + matches!(errno, libc::EIO | libc::ENOENT) + && logged.iter().any(|line| { + line.starts_with("enumerate folder children failed") + && line.contains("DoesNotExist") + }) + }, + // The folder keeps failing to list, and what is in it can come back + // as gone, so the mount no longer follows the model. + ends_seed: |_, _| true, + }, +]; + +/// An open bug in `docs/BUGS.md` that hangs a call, which the runs tolerate. +/// The seed ends there, since the watchdog aborted the mounts. +struct KnownHang { + bug: &'static str, + /// Whether the stacks the watchdog kept show this bug. + is: fn(stacks: &str) -> bool, +} + +const KNOWN_HANGS: &[KnownHang] = &[ + // A runtime worker invalidating a file's pages waits for the page lock + // of a read, whose fetch waits for a timer that no worker drives. + KnownHang { + bug: "B130", + is: |stacks| { + stacks.split("\nthread ").any(|thread| { + thread.contains("\"tokio-rt-worker\", state D") + && thread.contains("NotifyBatch>::flush") + }) + }, + }, +]; + +/// What a run looks like. +pub(crate) struct Profile { + pub(crate) name: &'static str, + pub(crate) clients: usize, + pub(crate) steps: usize, + pub(crate) faults: fn() -> Faults, + /// Whether links go down and come back during the sequence. + pub(crate) link_flaps: bool, + pub(crate) restarts: bool, + /// How long a settle may take; `None` leaves settles out. + pub(crate) settle_budget: Option, + /// How long one syscall may take. + pub(crate) syscall_budget: Duration, + /// How long the queues may take to drain at the end. + pub(crate) drain_budget: Duration, +} + +/// How long the mounts may take to show Drive's tree at the end: one event +/// poll (`POLL_INTERVAL`) and change. +const CONVERGE_BUDGET: Duration = Duration::from_secs(30); + +/// Steps shown before a failure. All of them are written to `steps.log` in +/// the state the failure keeps. +const LOG_TAIL: usize = 40; + +/// How long past its own budget a step, or the checks after the last one, may +/// run before the watchdog takes it for hung. Longer than a restart may wait +/// for the last run and the mount. +const STALL: Duration = Duration::from_secs(120); + +/// What a known bug left on Drive in place of a file, and how the model +/// takes it in. +struct KnownDamage { + bug: &'static str, + path: String, + /// The conflict copy it made, if it made one. + copy: Option, + change: FsOp, +} + +struct Client { + daemon: Option, + model: Model, + online: bool, +} + +impl Client { + fn daemon(&self) -> &Daemon { + self.daemon + .as_ref() + .expect("a client always has a daemon between steps") + } + + fn root(&self, index: usize) -> PathBuf { + self.daemon().mountpoint.join(owned_folder(index)) + } +} + +struct Run { + seed: u64, + profile: &'static Profile, + drive: std::sync::Arc, + clients: Vec, + log: Vec, + started: Instant, + /// How often each [`KNOWN`] bug turned up. + known: BTreeMap<&'static str, usize>, + /// The known bug that ended the seed early, if one did. + ended_by: Option<&'static str>, + /// What the daemons logged at `INFO` and above so far. + logged: Vec, + watchdog: Watchdog, +} + +/// Run `profile` on `seed`. It answers how often each known bug turned up, +/// or a report of what went wrong. +pub(crate) fn run( + seed: u64, + profile: &'static Profile, +) -> Result, String> { + let dir = scratch(&format!("{}-{seed}", profile.name)); + let mut rng = Rng::new(seed); + let drive = FakeDrive::new(); + let mut run = Run { + seed, + profile, + drive: drive.clone(), + clients: Vec::new(), + log: Vec::new(), + started: Instant::now(), + known: BTreeMap::new(), + ended_by: None, + logged: Vec::new(), + watchdog: Watchdog::new(&dir), + }; + let outcome = run.execute(&dir, &mut rng); + run.watchdog.stop(); + // A hang is what failed, whatever the aborted call answered. + let outcome = match run.watchdog.fired() { + Some(hung) => run.known_hang(&hung), + None => outcome, + }; + for client in &mut run.clients { + if let Some(daemon) = client.daemon.take() { + let _ = daemon.stop(); + } + } + match outcome { + Ok(()) => { + let _ = std::fs::remove_dir_all(&dir); + Ok(run.known) + } + Err(failure) => Err(run.report(&failure, &dir)), + } +} + +impl Run { + /// Count `hung` if it is one of the [`KNOWN_HANGS`], or fail on it. + fn known_hang(&mut self, hung: &Hung) -> Result<(), String> { + let known = KNOWN_HANGS.iter().find(|known| (known.is)(&hung.stacks)); + match known { + Some(known) if tolerate_known() => { + *self.known.entry(known.bug).or_default() += 1; + self.note(format!("known bug {}; the seed ends here", known.bug)); + self.ended_by = Some(known.bug); + Ok(()) + } + _ => Err(format!("liveness: {}", hung.what)), + } + } + + fn execute(&mut self, dir: &Path, rng: &mut Rng) -> Result<(), String> { + for index in 0..self.profile.clients { + let client = self.drive.client(rng.next_u64(), (self.profile.faults)()); + let daemon = Daemon::start(&dir.join(format!("client{index}")), client) + .map_err(|error| format!("client {index} did not mount: {error}"))?; + let folder = daemon.mountpoint.join(owned_folder(index)); + std::fs::create_dir(&folder) + .map_err(|error| format!("client {index}: mkdir {}: {error}", folder.display()))?; + self.clients.push(Client { + daemon: Some(daemon), + model: Model::default(), + online: true, + }); + } + for step in 0..self.profile.steps { + let _watching = self.watchdog.arm(self.step_limit(), format!("step {step}")); + self.step(rng)?; + if self.ended_by.is_some() { + return Ok(()); + } + } + let _watching = self.watchdog.arm( + self.profile.drain_budget + STALL, + "the checks after the last step", + ); + self.finish() + } + + /// How long a step may run before the watchdog takes it for hung. + fn step_limit(&self) -> Duration { + let budget = self.profile.settle_budget.unwrap_or_default(); + budget.max(self.profile.syscall_budget) + STALL + } + + fn step(&mut self, rng: &mut Rng) -> Result<(), String> { + let index = rng.below(self.clients.len() as u64) as usize; + let roll = rng.below(100); + match roll { + 0..4 if self.profile.link_flaps => { + let client = &mut self.clients[index]; + client.online = !client.online; + client.daemon().client.set_online(client.online); + let state = if client.online { "up" } else { "down" }; + self.note(format!("client {index}: link {state}")); + } + 4..6 if self.profile.restarts => { + self.note(format!("client {index}: restart")); + let daemon = self.clients[index].daemon.take().expect("daemon"); + let daemon = daemon + .restart() + .map_err(|error| format!("client {index} did not come back: {error}"))?; + self.clients[index].daemon = Some(daemon); + } + 6..12 => { + let ms = rng.below(400); + self.note(format!("pause {ms} ms")); + std::thread::sleep(Duration::from_millis(ms)); + } + 12..15 => { + 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())); + } + } + _ => { + let op = self.clients[index].model.next_op(rng); + self.syscall(index, op)?; + } + } + Ok(()) + } + + fn syscall(&mut self, index: usize, op: FsOp) -> Result<(), String> { + let client = &mut self.clients[index]; + let root = client.root(index); + let model = client.model.clone(); + let before = client.model.content(target(&op)).cloned(); + let expected = client.model.apply(&op); + self.logged.extend(take_logged()); + let start = Instant::now(); + let actual = perform(&root, &op, before.as_deref()); + let took = start.elapsed(); + self.note(format!( + "client {index}: {} = {} in {took:?}", + op.describe(), + outcome(&actual) + )); + let actual = actual?; + let logged = take_logged(); + self.logged.extend(logged.iter().cloned()); + if actual != expected + && let Err(errno) = actual + && tolerate_known() + && let Some(known) = KNOWN + .iter() + .find(|known| (known.is)(&model, &op, errno, &logged)) + { + *self.known.entry(known.bug).or_default() += 1; + if (known.ends_seed)(&model, &op) { + self.note(format!("known bug {}; the seed ends here", known.bug)); + self.ended_by = Some(known.bug); + } else { + self.note(format!("known bug {}; taken back", known.bug)); + } + self.clients[index].model = model; + return Ok(()); + } + if actual != expected { + return Err(format!( + "posix: client {index}: {} answered {}, expected {}", + op.describe(), + errno(actual), + errno(expected) + )); + } + if took > self.profile.syscall_budget { + return Err(format!( + "liveness: client {index}: {} took {took:?}, over {:?}", + op.describe(), + self.profile.syscall_budget + )); + } + Ok(()) + } + + fn finish(&mut self) -> Result<(), String> { + for client in &mut self.clients { + client.online = true; + client.daemon().client.set_online(true); + } + self.note("every link up; waiting for the queues".into()); + if !wait_until(self.profile.drain_budget, || { + self.logged.extend(take_logged()); + self.landed() + }) { + return Err(format!( + "no loss: Drive does not hold what the clients wrote, {:?} after the links came up\n{}", + self.profile.drain_budget, + self.difference() + )); + } + let drive = self.drive_tree(); + let mut explained = Vec::new(); + for index in 0..self.clients.len() { + for damage in self.known_damage(index, &drive) { + *self.known.entry(damage.bug).or_default() += 1; + self.note(format!( + "known bug {}: client {index}'s {} landed as {:?}", + damage.bug, damage.path, damage.copy + )); + let _ = self.clients[index].model.apply(&damage.change); + if let Some(copy) = damage.copy { + explained.push(join(&owned_folder(index), ©)); + } + } + } + let conflicts: Vec = drive + .into_keys() + .filter(|path| is_conflict_copy(path) && !explained.contains(path)) + .collect(); + if !conflicts.is_empty() { + return Err(format!("no false conflicts: {conflicts:?}")); + } + for index in 0..self.clients.len() { + let mut last = Tree::new(); + let converged = wait_until(CONVERGE_BUDGET, || { + last = read_tree(&self.clients[index].daemon().mountpoint); + last == self.drive_tree() + }); + if !converged { + return Err(format!( + "convergence: client {index}'s mount differs from Drive\n{}", + diff(&last, &self.drive_tree(), "mount", "Drive") + )); + } + } + Ok(()) + } + + /// Whether every client's queue is empty and Drive holds what its model + /// says, but for what known bugs explain. + fn landed(&self) -> bool { + let drive = self.drive_tree(); + self.clients.iter().enumerate().all(|(index, client)| { + let mut model = client.model.clone(); + for damage in self.known_damage(index, &drive) { + let _ = model.apply(&damage.change); + } + client + .daemon() + .pending() + .is_ok_and(|items| items.is_empty()) + && subtree(&drive, &owned_folder(index)) == *model.tree() + }) + } + + /// The files of client `index` that a known bug left otherwise on Drive + /// than the model says. A conflict copy takes Drive holding it with the + /// model's bytes, and the drain saying it made that copy for that name: + /// + /// - B126: the file's queued create ran into its name while a queued + /// rename was about to free it, so it is there only as the copy. + /// - B127: the file's queued create landed but its answer was lost, so + /// the retry found it and made the copy next to it. + /// + /// A lost file takes the drain saying it adopted a node for that name + /// whose queued trash then landed: + /// + /// - B128: the file's queued create ran into the name of the file it + /// replaced and took that over, trash and all. + /// + /// A conflict copy of an earlier version, which the model no longer + /// holds, takes the drain saying it kept a queued write as that copy + /// after a landed create could not be read back: + /// + /// - B131: the tree kept the create's placeholder, so a write based on it + /// found Drive's copy changed. + fn known_damage(&self, index: usize, drive: &Tree) -> Vec { + if !tolerate_known() { + return Vec::new(); + } + let drive = subtree(drive, &owned_folder(index)); + let model = self.clients[index].model.tree(); + let made_copy = |path: &str, copy: &str| { + let wanted = format!("wanted={:?}", name_of(path)); + let name = format!("name={:?}", name_of(copy)); + self.logged.iter().any(|line| { + line.starts_with("name is taken remotely; creating under a conflict name") + && line.contains(&wanted) + && line.contains(&name) + }) + }; + let adopted_trashed = |path: &str| { + let wanted = format!("wanted={:?}", name_of(path)); + self.logged.iter().any(|line| { + line.starts_with("the name is held by our own unanswered create; adopting it") + && line.contains(&wanted) + && line.split(' ').any(|field| { + field.strip_prefix("twin=").is_some_and(|twin| { + let uid = format!("uid={twin}"); + self.logged.iter().any(|line| { + line.starts_with("pending trash landed") + && line.split(' ').any(|field| field == uid) + }) + }) + }) + }) + }; + let unread_create = self.logged.iter().any(|line| { + line.starts_with("pending upload failed; will retry uid=local~") + && line.ends_with("error=fetch node: no such file or folder") + }); + let kept_write = |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) + }) + }; + 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(), + }, + }); + } + } + for (path, content) in model { + let Some(content) = content else { continue }; + let copy = drive.iter().find(|(copy, other)| { + is_conflict_copy(copy) + && !model.contains_key(*copy) + && parent_of(copy) == parent_of(path) + && other.as_ref() == Some(content) + && made_copy(path, copy) + }); + let Some((copy, _)) = copy else { + if !drive.contains_key(path) && adopted_trashed(path) { + damage.push(KnownDamage { + bug: "B128", + path: path.clone(), + copy: None, + change: FsOp::Unlink { path: path.clone() }, + }); + } + continue; + }; + let (bug, change) = match drive.get(path) { + None => ( + "B126", + FsOp::Rename { + from: path.clone(), + to: copy.clone(), + }, + ), + Some(Some(landed)) if landed == content => ( + "B127", + FsOp::Create { + path: copy.clone(), + data: content.clone(), + }, + ), + Some(_) => continue, + }; + damage.push(KnownDamage { + bug, + path: path.clone(), + copy: Some(copy.clone()), + change, + }); + } + damage + } + + fn difference(&self) -> String { + let drive = self.drive_tree(); + let mut out = String::new(); + for (index, client) in self.clients.iter().enumerate() { + let pending = client.daemon().pending().unwrap_or_default(); + for item in pending { + out.push_str(&format!( + " client {index} still owes: {} {} (attempts {}, last error {:?})\n", + item.kind, item.path, item.attempts, item.last_error + )); + } + out.push_str(&diff( + client.model.tree(), + &subtree(&drive, &owned_folder(index)), + &format!("client {index}'s model"), + "Drive", + )); + } + out + } + + fn drive_tree(&self) -> Tree { + self.drive + .tree() + .into_iter() + .map(|(path, entry)| { + let content = match entry { + Entry::Folder => None, + Entry::File(content) => Some(content.as_ref().clone()), + }; + (path, content) + }) + .collect() + } + + fn note(&mut self, line: String) { + let line = format!("{:>8.3}s {line}", self.started.elapsed().as_secs_f64()); + self.log.push(line); + } + + fn report(&self, failure: &str, dir: &Path) -> String { + let _ = std::fs::write(dir.join("steps.log"), self.log.join("\n") + "\n"); + let tail = &self.log[self.log.len().saturating_sub(LOG_TAIL)..]; + format!( + "{} seed {} failed: {failure}\nreplay with PDFS_SIM_SEED={}; state kept in {}\nlast steps:\n{}", + self.profile.name, + self.seed, + self.seed, + dir.display(), + tail.join("\n") + ) + } +} + +/// Make `op` under `root`. The outer error is a check that failed on the +/// way (the content an open file reads back), the inner one the errno. +fn perform(root: &Path, op: &FsOp, before: Option<&[u8]>) -> Result, String> { + let at = |path: &str| root.join(path); + let result: io::Result<()> = match op { + FsOp::Create { path, data } => std::fs::write(at(path), data), + FsOp::Write { path, offset, data } => OpenOptions::new() + .write(true) + .open(at(path)) + .and_then(|mut file| { + file.seek(SeekFrom::Start(*offset))?; + file.write_all(data) + }), + FsOp::Truncate { path, len } => OpenOptions::new() + .write(true) + .open(at(path)) + .and_then(|file| file.set_len(*len)), + FsOp::Mkdir { path } => std::fs::create_dir(at(path)), + FsOp::Rename { from, to } => std::fs::rename(at(from), at(to)), + FsOp::Unlink { path } => std::fs::remove_file(at(path)), + FsOp::Rmdir { path } => std::fs::remove_dir(at(path)), + FsOp::UnlinkOpen { path, data } => { + let mut read = Vec::new(); + let result = OpenOptions::new() + .read(true) + .write(true) + .open(at(path)) + .and_then(|mut file| { + std::fs::remove_file(at(path))?; + file.seek(SeekFrom::End(0))?; + file.write_all(data)?; + file.seek(SeekFrom::Start(0))?; + file.read_to_end(&mut read).map(|_| ()) + }); + if result.is_ok() { + let expected = [before.unwrap_or_default(), data].concat(); + if read != expected { + return Err(format!( + "posix: {} read back {:?}, expected {:?}", + op.describe(), + String::from_utf8_lossy(&read), + String::from_utf8_lossy(&expected) + )); + } + } + result + } + }; + Ok(result.map_err(|error| error.raw_os_error().unwrap_or(libc::EIO))) +} + +fn target(op: &FsOp) -> &str { + match op { + FsOp::Create { path, .. } + | FsOp::Write { path, .. } + | FsOp::Truncate { path, .. } + | FsOp::Mkdir { path } + | FsOp::Unlink { path } + | FsOp::UnlinkOpen { path, .. } + | FsOp::Rmdir { path } => path, + FsOp::Rename { from, .. } => from, + } +} + +/// The tree a mount shows, read through it, without the folders it makes up +/// that are not on Drive. +fn read_tree(mountpoint: &Path) -> Tree { + fn walk(dir: &Path, prefix: &str, tree: &mut Tree) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let name = entry.file_name().to_string_lossy().into_owned(); + if prefix.is_empty() && name == crate::r#virtual::SHARED_WITH_ME_BASE { + continue; + } + let path = super::model::join(prefix, &name); + if entry.file_type().is_ok_and(|kind| kind.is_dir()) { + tree.insert(path.clone(), None); + walk(&entry.path(), &path, tree); + } else { + tree.insert(path, std::fs::read(entry.path()).ok()); + } + } + } + let mut tree = Tree::new(); + walk(mountpoint, "", &mut tree); + tree +} + +/// The part of `tree` below `folder`, relative to it. +fn subtree(tree: &Tree, folder: &str) -> Tree { + let prefix = format!("{folder}/"); + tree.iter() + .filter_map(|(path, entry)| Some((path.strip_prefix(&prefix)?.to_owned(), entry.clone()))) + .collect() +} + +/// Whether a known bug is counted rather than failing the run. +fn tolerate_known() -> bool { + std::env::var("PDFS_SIM_KNOWN").as_deref() != Ok("fail") +} + +fn owned_folder(index: usize) -> String { + format!("c{index}") +} + +fn is_conflict_copy(path: &str) -> bool { + path.split('/') + .any(|name| name.contains("(sync-conflict") || name.starts_with("recovered-")) +} + +fn diff(left: &Tree, right: &Tree, left_name: &str, right_name: &str) -> String { + let show = |entry: &Option>| match entry { + None => "folder".to_owned(), + Some(content) => format!("{:?}", String::from_utf8_lossy(content)), + }; + let mut out = String::new(); + for (path, entry) in left { + match right.get(path) { + None => out.push_str(&format!( + " {path}: only in {left_name} ({})\n", + show(entry) + )), + Some(other) if other != entry => out.push_str(&format!( + " {path}: {left_name} has {}, {right_name} has {}\n", + show(entry), + show(other) + )), + Some(_) => {} + } + } + for (path, entry) in right { + if !left.contains_key(path) { + out.push_str(&format!( + " {path}: only in {right_name} ({})\n", + show(entry) + )); + } + } + out +} + +fn errno(outcome: Result<(), i32>) -> String { + match outcome { + Ok(()) => "ok".into(), + Err(code) => io::Error::from_raw_os_error(code).to_string(), + } +} + +fn outcome(actual: &Result, String>) -> String { + match actual { + Ok(result) => errno(*result), + Err(check) => check.clone(), + } +} + +/// Run `profile` over the seeds the environment asks for: `PDFS_SIM_SEED` +/// alone, or `PDFS_SIM_SEEDS` of them (default `default_seeds`) from 1. +pub(crate) fn run_seeds(profile: &'static Profile, default_seeds: u64) { + let seeds: Vec = match std::env::var("PDFS_SIM_SEED") { + Ok(seed) => vec![seed.parse().expect("PDFS_SIM_SEED is a number")], + Err(_) => { + let count = std::env::var("PDFS_SIM_SEEDS") + .ok() + .and_then(|count| count.parse().ok()) + .unwrap_or(default_seeds); + (1..=count).collect() + } + }; + let mut failures = Vec::new(); + let mut known = BTreeMap::<&str, usize>::new(); + for seed in seeds { + match run(seed, profile) { + Ok(hits) => { + for (bug, count) in hits { + *known.entry(bug).or_default() += count; + } + } + Err(failure) => failures.push(failure), + } + } + if !known.is_empty() { + eprintln!("{}: known bugs turned up: {known:?}", profile.name); + } + assert!(failures.is_empty(), "\n{}", failures.join("\n\n")); +} + +#[cfg(test)] +mod tests { + use super::*; + + static ONE_CLIENT_LAN: Profile = Profile { + name: "one-client-lan", + clients: 1, + steps: 80, + faults: Faults::lan, + link_flaps: false, + restarts: false, + settle_budget: Some(Duration::from_secs(4)), + syscall_budget: Duration::from_secs(2), + drain_budget: Duration::from_secs(20), + }; + + static ONE_CLIENT_FLAKY: Profile = Profile { + name: "one-client-flaky", + clients: 1, + steps: 80, + faults: Faults::wifi, + link_flaps: true, + restarts: true, + settle_budget: None, + syscall_budget: Duration::from_secs(10), + drain_budget: Duration::from_secs(120), + }; + + static THREE_CLIENTS: Profile = Profile { + name: "three-clients", + clients: 3, + steps: 90, + faults: Faults::lan, + link_flaps: true, + restarts: true, + settle_budget: Some(Duration::from_secs(4)), + syscall_budget: Duration::from_secs(2), + drain_budget: Duration::from_secs(60), + }; + + #[test] + #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse --lib sim::run -- --ignored`"] + fn one_client_on_a_good_link() { + run_seeds(&ONE_CLIENT_LAN, 12); + } + + #[test] + #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse --lib sim::run -- --ignored`"] + fn one_client_on_a_flaky_link() { + run_seeds(&ONE_CLIENT_FLAKY, 4); + } + + #[test] + #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse --lib sim::run -- --ignored`"] + fn three_clients_one_writer_each() { + run_seeds(&THREE_CLIENTS, 3); + } + + #[test] + fn conflict_copies_are_recognised_by_name() { + assert!(is_conflict_copy("c0/x/a (sync-conflict 1790000000).txt")); + assert!(is_conflict_copy("recovered-XrPUSW5W")); + assert!(!is_conflict_copy("c0/x/a.txt")); + } + + #[test] + fn an_invalidation_stuck_on_a_page_lock_is_b130() { + let b130 = |stacks: &str| KNOWN_HANGS.iter().any(|known| (known.is)(stacks)); + let stuck = "thread 1 \"fuser-0\", state S, waiting in fuse_dev_do_read\n\ + \nthread 2 \"tokio-rt-worker\", state D, waiting in __folio_lock\n\ + \x20 11: ::flush\n"; + assert!(b130(stuck)); + assert!(!b130(&stuck.replace("state D", "state S"))); + assert!(!b130(&stuck.replace("NotifyBatch", "Other"))); + } +} diff --git a/crates/pdfs-fuse/src/sim/watchdog.rs b/crates/pdfs-fuse/src/sim/watchdog.rs new file mode 100644 index 0000000..d802d10 --- /dev/null +++ b/crates/pdfs-fuse/src/sim/watchdog.rs @@ -0,0 +1,322 @@ +//! Turns a call through a mount that never returns into a failed seed. +//! +//! A deadlock in a daemon leaves the run's own thread blocked in the kernel, +//! where no budget check is reached, and a CI job then waits for its timeout. +//! The watchdog runs beside the run. When what it was armed for outlives its +//! limit, it aborts the FUSE connections of the run's mounts, so the blocked +//! call fails and the run reports the hang, and writes where every thread was +//! to `stacks.txt`. +//! +//! The stacks are taken in the process: no tracer can stop a thread that +//! waits on a FUSE request, so `gdb` and `eu-stack` wait as long as the hang. +//! Each thread is sent [`stack_signal`] and captures its own backtrace in the +//! handler. A thread in the kernel takes the signal only once it leaves, so +//! the mounts are aborted after the others have answered, and its backtrace +//! is the call it was blocked in. Its kernel wait channel is recorded first. + +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +use parking_lot::Mutex; + +use super::daemon::wait_until; + +/// How long the threads that are not blocked in the kernel have to answer, +/// and the blocked ones once the mounts are aborted. +const ANSWER: Duration = Duration::from_secs(5); + +pub(crate) struct Watchdog { + shared: Arc, + thread: Option>, +} + +struct Shared { + /// The run's directory: its mounts are below it, and the stacks go in it. + dir: PathBuf, + armed: Mutex>, + /// What hung, once something has. + fired: Mutex>, + stop: AtomicBool, +} + +struct Armed { + what: String, + limit: Duration, + since: Instant, +} + +/// What hung, and where every thread was then. +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct Hung { + pub(crate) what: String, + pub(crate) stacks: String, +} + +/// Disarms the watchdog when dropped. +pub(crate) struct Watching(Arc); + +impl Watchdog { + pub(crate) fn new(dir: &Path) -> Self { + let shared = Arc::new(Shared { + dir: dir.to_path_buf(), + armed: Mutex::new(None), + fired: Mutex::new(None), + stop: AtomicBool::new(false), + }); + let thread = { + let shared = shared.clone(); + std::thread::Builder::new() + .name("sim-watchdog".into()) + .spawn(move || watch(&shared)) + .expect("spawn the simulation watchdog") + }; + Self { + shared, + thread: Some(thread), + } + } + + /// Watch `what` until the guard is dropped, and abort the mounts if it is + /// still going after `limit`. + pub(crate) fn arm(&self, limit: Duration, what: impl Into) -> Watching { + *self.shared.armed.lock() = Some(Armed { + what: what.into(), + limit, + since: Instant::now(), + }); + Watching(self.shared.clone()) + } + + /// What hung, if something did. Call [`Watchdog::stop`] first, or the + /// stacks may still be on their way. + pub(crate) fn fired(&self) -> Option { + self.shared.fired.lock().clone() + } + + /// Stop watching, once the stacks of a hang that was caught are taken. + pub(crate) fn stop(&mut self) { + self.shared.stop.store(true, Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +impl Drop for Watching { + fn drop(&mut self) { + *self.0.armed.lock() = None; + } +} + +impl Drop for Watchdog { + fn drop(&mut self) { + self.stop(); + } +} + +fn watch(shared: &Shared) { + while !shared.stop.load(Ordering::Relaxed) { + std::thread::sleep(Duration::from_millis(200)); + let hung = shared + .armed + .lock() + .as_ref() + .filter(|armed| armed.since.elapsed() > armed.limit) + .map(|armed| format!("{} has not returned after {:?}", armed.what, armed.limit)); + let Some(hung) = hung else { continue }; + *shared.armed.lock() = None; + let path = shared.dir.join("stacks.txt"); + // Before anything is let go: the run reports the hang, not whatever + // the aborted call answers. + *shared.fired.lock() = Some(Hung { + what: format!("{hung}; the stacks are in {}", path.display()), + stacks: String::new(), + }); + let stacks = take_stacks(|| abort_mounts_below(&shared.dir)); + let what = match std::fs::write(&path, &stacks) { + Ok(()) => format!("{hung}; the stacks are in {}", path.display()), + Err(error) => format!("{hung}; the stacks could not be kept: {error}"), + }; + *shared.fired.lock() = Some(Hung { what, stacks }); + } +} + +/// One thread of this process as the kernel sees it. +struct Task { + tid: i32, + name: String, + /// `D` while it waits in the kernel uninterruptibly. + state: String, + wchan: String, +} + +/// Where every other thread of this process is, with `release` called once +/// those not blocked in the kernel have answered. +fn take_stacks(release: impl FnOnce()) -> String { + let own = gettid(); + let tasks = tasks(); + STACKS.lock().clear(); + install_stack_handler(); + for task in &tasks { + if task.tid != own { + // SAFETY: tgkill only sends a signal, whose handler is installed. + unsafe { + libc::syscall( + libc::SYS_tgkill, + std::process::id() as i32, + task.tid, + stack_signal(), + ) + }; + } + } + let answered = |blocked: bool| { + let stacks = STACKS.lock(); + tasks + .iter() + .filter(|task| task.tid != own && (blocked || task.state != "D")) + .all(|task| stacks.contains_key(&task.tid)) + }; + wait_until(ANSWER, || answered(false)); + release(); + wait_until(ANSWER, || answered(true)); + let stacks = std::mem::take(&mut *STACKS.lock()); + let mut out = String::new(); + for task in tasks.iter().filter(|task| task.tid != own) { + out.push_str(&format!( + "thread {} {:?}, state {}, waiting in {}\n", + task.tid, task.name, task.state, task.wchan + )); + match stacks.get(&task.tid) { + Some(stack) => out.push_str(&format!("{stack}\n")), + None => out.push_str(" (no answer)\n\n"), + } + } + out +} + +fn tasks() -> Vec { + let Ok(entries) = std::fs::read_dir("/proc/self/task") else { + return Vec::new(); + }; + let read = |dir: &Path, file: &str| { + std::fs::read_to_string(dir.join(file)) + .unwrap_or_default() + .trim() + .to_owned() + }; + entries + .flatten() + .filter_map(|entry| { + let tid = entry.file_name().to_str()?.parse().ok()?; + let dir = entry.path(); + // The state follows the name in parentheses, which may hold spaces. + let stat = read(&dir, "stat"); + let state = stat.rsplit_once(") ")?.1.split(' ').next()?.to_owned(); + Some(Task { + tid, + name: read(&dir, "comm"), + state, + wchan: read(&dir, "wchan"), + }) + }) + .collect() +} + +/// The backtraces the threads captured, by thread id. +static STACKS: Mutex> = Mutex::new(BTreeMap::new()); + +/// The signal that asks a thread for its stack: one of the real-time signals +/// that nothing in the daemon uses. +fn stack_signal() -> i32 { + libc::SIGRTMIN() + 7 +} + +fn install_stack_handler() { + static INSTALLED: std::sync::Once = std::sync::Once::new(); + extern "C" fn handler(_: i32) { + // Not async-signal-safe: the capture allocates. The process is being + // taken down for a hang anyway, and a thread interrupted inside the + // allocator only fails to answer. + let stack = std::backtrace::Backtrace::force_capture(); + STACKS.lock().insert(gettid(), stack); + } + INSTALLED.call_once(|| { + // SAFETY: a zeroed sigaction with a handler and SA_RESTART is valid, + // and the handler only touches `STACKS`. + unsafe { + let mut action: libc::sigaction = std::mem::zeroed(); + action.sa_sigaction = handler as extern "C" fn(i32) as usize; + action.sa_flags = libc::SA_RESTART; + libc::sigemptyset(&mut action.sa_mask); + libc::sigaction(stack_signal(), &action, std::ptr::null_mut()); + } + }); +} + +fn gettid() -> i32 { + // SAFETY: gettid has no arguments and cannot fail. + unsafe { libc::gettid() } +} + +/// Abort the FUSE connection of every mount below `dir`. Its connection is +/// named by the mount's device minor, and the mount's owner may abort it. +fn abort_mounts_below(dir: &Path) { + let Ok(mountinfo) = std::fs::read_to_string("/proc/self/mountinfo") else { + return; + }; + for line in mountinfo.lines() { + let fields: Vec<&str> = line.split(' ').collect(); + let (Some(device), Some(mountpoint)) = (fields.get(2), fields.get(4)) else { + continue; + }; + if !Path::new(mountpoint).starts_with(dir) { + continue; + } + if let Some((_, minor)) = device.split_once(':') { + let _ = std::fs::write(format!("/sys/fs/fuse/connections/{minor}/abort"), "1"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sim::daemon::scratch; + + #[test] + #[ignore = "signals every thread of the test process: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"] + fn a_call_past_its_limit_is_reported_with_the_stacks() { + let dir = scratch("watchdog"); + let mut watchdog = Watchdog::new(&dir); + { + let _watching = watchdog.arm(Duration::from_millis(100), "a nap"); + std::thread::sleep(Duration::from_secs(1)); + } + watchdog.stop(); + let hung = watchdog.fired().expect("the watchdog fired"); + assert!( + hung.what.starts_with("a nap has not returned after"), + "{hung:?}" + ); + assert!(hung.stacks.contains("a_call_past_its_limit"), "{hung:?}"); + let kept = std::fs::read_to_string(dir.join("stacks.txt")).unwrap_or_default(); + assert_eq!(kept, hung.stacks); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn a_call_within_its_limit_is_not() { + let dir = scratch("watchdog-quiet"); + let watchdog = Watchdog::new(&dir); + drop(watchdog.arm(Duration::from_millis(100), "a quick call")); + std::thread::sleep(Duration::from_millis(500)); + assert_eq!(watchdog.fired(), None); + drop(watchdog); + let _ = std::fs::remove_dir_all(&dir); + } +}