nvrl/proton-drive-linux
1
0 Fork 0
mirror of https://github.com/narrrl/proton-drive-linux.git synced 2026-11-03 16:41:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

test(fuse): add seeded simulation runs over the FakeDrive

Browse Source 
The second half of the simulation harness in docs/MILESTONE-3.0.0.md
section 8.2. A run starts one or more real FUSE daemons on a shared
FakeDrive. Each client owns a folder `c<n>` that only it writes. A seed
then draws syscalls through the mounts, link flaps, restarts, pauses and
settles. Every syscall is checked against a model of the client's tree as
it happens. After the last step every link comes up and the run checks
no loss, POSIX behaviour, no false conflicts, convergence and liveness.

- sim::model is the POSIX model: the tree, what each call should answer,
  and how it changes the tree.
- sim::run has the profiles: one client on a good link, one on a flaky
  link with flaps and restarts, and three clients. PDFS_SIM_SEEDS sets
  how many seeds each runs, and PDFS_SIM_SEED replays one.
- sim::watchdog catches a call that never returns. It takes every
  thread's stack in the process, since no tracer can stop a thread
  blocked on a FUSE request. It then aborts the run's FUSE connections
  and writes the stacks next to the run's state.
- The daemons' logs at INFO and above are kept, so a known open bug is
  told apart from a new failure. Known bugs are counted, and the seed ends
  where the mount leaves the model behind. PDFS_SIM_KNOWN=fail fails on
  them instead.
- A restart retries opening the database while the last run still holds
  it.
- The FakeDrive restamps a node's modification time on a rename or move,
  as Drive does (B113).

The runs mount FUSE, so they are ignored by default. Run them with
`cargo test -p pdfs-fuse --lib sim:: -- --ignored --test-threads=1`.
Test-only; no behaviour change.
This commit is contained in:
nvrl committed 2026-10-03 07:47:01 +02:00
1 parent 3d10338807
commit 97743f17e3
6 files changed
+1854 -14
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/pdfs-fuse/src/sim/daemon.rs
+78 -5
View File
@@ -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<JoinHandle<io::Result<MountOutcome>>>,
rt: Option<tokio::runtime::Runtime>,
@@ -41,8 +47,22 @@ impl Daemon {
pub(crate) fn start(dir: &Path, client: FakeClient) -> io::Result<Self> {
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<Vec<PendingOpInfo>> {
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> {
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<MountOutcome> {
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<String> {
std::mem::take(&mut *LOGGED.lock())
}
static LOGGED: parking_lot::Mutex<Vec<String>> = parking_lot::Mutex::new(Vec::new());
struct KeepLogged;
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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 {
crates/pdfs-fuse/src/sim/fake_drive.rs
+72 -9
View File
@@ -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<Remote> {
fn visible(&self, uid: &NodeUid, lag: Lag, now: Instant) -> Option<Remote> {
let versions = self.nodes.get(uid)?;
versions
.iter()
let live = |remote: &Option<Remote>| 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<NodeUid> {
@@ -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::<Vec<_>>()
};
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 {
crates/pdfs-fuse/src/sim/mod.rs
+3
View File
@@ -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;
crates/pdfs-fuse/src/sim/model.rs
+445
View File
@@ -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<String, Option<Vec<u8>>>;
/// 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<u8>,
},
/// Open an existing file, write `data` at `offset` (past the end leaves
/// a hole), close.
Write {
path: String,
offset: u64,
data: Vec<u8>,
},
/// 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<u8>,
},
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<u8>> {
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<Vec<u8>>)> = 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<u8>, 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<String> {
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<u8> {
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<_>>(), 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(&nothing).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");
}
}
crates/pdfs-fuse/src/sim/run.rs
+934
View File
@@ -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<n>` 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<Duration>,
/// 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<String>,
change: FsOp,
}
struct Client {
daemon: Option<Daemon>,
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<FakeDrive>,
clients: Vec<Client>,
log: Vec<String>,
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<String>,
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<BTreeMap<&'static str, usize>, 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), &copy));
}
}
}
let conflicts: Vec<String> = 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<KnownDamage> {
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<Result<(), i32>, 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<Vec<u8>>| 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<Result<(), i32>, 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<u64> = 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: <pdfs_fuse::NotifyBatch>::flush\n";
assert!(b130(stuck));
assert!(!b130(&stuck.replace("state D", "state S")));
assert!(!b130(&stuck.replace("NotifyBatch", "Other")));
}
}
crates/pdfs-fuse/src/sim/watchdog.rs
+322
View File
@@ -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<Shared>,
thread: Option<JoinHandle<()>>,
}
struct Shared {
/// The run's directory: its mounts are below it, and the stacks go in it.
dir: PathBuf,
armed: Mutex<Option<Armed>>,
/// What hung, once something has.
fired: Mutex<Option<Hung>>,
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<Shared>);
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<String>) -> 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<Hung> {
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<Task> {
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<BTreeMap<i32, std::backtrace::Backtrace>> = 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);
}
}