diff --git a/AGENTS.md b/AGENTS.md index d1d20b6..e3b4225 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -21,7 +21,7 @@ Keep it that way. - `lib.rs`: `Host`, which starts and stops a world. - `broker.rs`: routing, mailboxes, the version store, broker requests. - `supervisor.rs`: runs one WASM actor (restarts, swaps, rollbacks). - - `runtime.rs`: Wasmtime instances. `fs.rs`: commit on close. + - `runtime.rs`: Wasmtime instances and what they get (mounts, network). - `candidate.rs`: tests a new version. `cargo_actor.rs`: builds Rust. - `world.rs`: the on-disk layout. `console.rs`: the terminal. - `sdk/`: `sausage-actor`, the SDK for Rust actors. Never published: the diff --git a/DECISIONS.md b/DECISIONS.md index fa9e38d..be3e965 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -39,8 +39,9 @@ actor, or a change that the agent makes to itself. ## Small choices -- **Commit on close** for new and truncated files, so readers never see half - a file, in any language. Appends write in place, so logs stay cheap. +- **Plain POSIX files.** An earlier version staged every write until close. + That broke `tempfile`, git, and SQLite, which expect normal files. Atomic + publishing is the usual temporary file + `rename`. - **Test first, then switch.** A new version is tested where it lies. Only a version that passes goes to `versions/`, and the old one runs until then. `versions/` keeps every version that was installed, so a rollback diff --git a/PROTOCOL.md b/PROTOCOL.md index e90f446..56a36c1 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -71,10 +71,9 @@ Each instance sees three folders: (installed version, running, subscriptions), `source/`, `public/`, `data/`, `stderr.log`, and `versions//`. `ls /actor` lists all actors. -A new or truncated file becomes visible when the writer closes it. Before -that, readers see the old file or no file. If the instance crashes, the write -is lost. A write to an existing file without truncate (append, edit) goes in -place. +Files are plain POSIX files: a write is visible at once. To publish a file +so that readers never see half of it, write a temporary file and `rename` it +(the Rust SDK has `write_atomic`). Send large or binary data as a path under `/actor//public/`, not inside a message. diff --git a/cli/src/broker.rs b/cli/src/broker.rs index eb2d682..f51dccf 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -18,8 +18,7 @@ use tokio::sync::{mpsc, oneshot}; use wasmtime::component::Component; use crate::candidate; -use crate::fs::Mount; -use crate::runtime::{InstanceConfig, Runtime}; +use crate::runtime::{InstanceConfig, Mount, Runtime}; use crate::supervisor::{Control, Reply, Supervisor}; use crate::world::{self, ActorMeta, NATIVE, VersionMeta, World}; diff --git a/cli/src/fs.rs b/cli/src/fs.rs deleted file mode 100644 index 67a9db6..0000000 --- a/cli/src/fs.rs +++ /dev/null @@ -1,446 +0,0 @@ -//! Filesystem layer with commit-on-close semantics. -//! -//! The layer wraps the wasmtime-wasi `wasi:filesystem` implementation. In a -//! writable mount, an open that creates or truncates a file goes to a hidden -//! temporary file next to the target. When the guest drops the descriptor -//! (close), the layer renames the temporary file into place. Until then, no -//! other reader sees the change. If the instance traps or is killed, the -//! changes are lost. -//! -//! An open of an existing file without truncate (append, in-place edit) -//! writes in place, so that appending to a log does not copy the log. - - -use std::collections::HashMap; -use std::path::{Component, Path, PathBuf}; - -use wasmtime::component::{HasData, Resource}; -use wasmtime_wasi::filesystem::{Descriptor, WasiFilesystemCtxView}; -use wasmtime_wasi::p2::bindings::filesystem::types::{ - self, ErrorCode, HostDescriptor, HostDirectoryEntryStream, -}; -use wasmtime_wasi::p2::bindings::filesystem::preopens; -use wasmtime_wasi::p2::{DynInputStream, DynOutputStream, FsError, FsResult}; - -/// Name prefix of uncommitted files. The layer hides these names from guests. -pub const TMP_PREFIX: &str = ".sausage-tmp-"; - -static TMP_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); - -#[derive(Clone, Debug)] -pub struct Mount { - pub guest: String, - pub host: PathBuf, - pub writable: bool, -} - -#[derive(Clone)] -struct DirInfo { - host: PathBuf, - root: PathBuf, - /// Writable mount: new and truncated files commit on close. - commit: bool, -} - -struct Pending { - tmp: PathBuf, - target: PathBuf, -} - -/// Per-instance state of the layer. -pub struct FsExt { - mounts: Vec, - dirs: HashMap, - pending: HashMap, -} - -impl FsExt { - pub fn new(mounts: Vec) -> Self { - Self { mounts, dirs: HashMap::new(), pending: HashMap::new() } - } - - /// Commit every open write. Use this when the guest exits normally, - /// because process exit closes all files. - pub fn commit_all(&mut self) { - for (_, p) in self.pending.drain() { - commit(p); - } - } -} - -impl Drop for FsExt { - /// Discard writes that were not committed (trap, kill). - fn drop(&mut self) { - for (_, p) in self.pending.drain() { - let _ = std::fs::remove_file(&p.tmp); - } - } -} - -fn commit(p: Pending) { - if let Err(e) = std::fs::rename(&p.tmp, &p.target) { - eprintln!("sausage: commit of {} failed: {e}", p.target.display()); - let _ = std::fs::remove_file(&p.tmp); - } -} - -fn is_tmp_name(path: &str) -> bool { - path.rsplit('/').next().is_some_and(|n| n.starts_with(TMP_PREFIX)) -} - -/// Join `rel` to the directory `dir` lexically. Return `None` if the result -/// leaves the mount root. -fn resolve(dir: &DirInfo, rel: &str) -> Option { - let mut out = dir.host.clone(); - for c in Path::new(rel).components() { - match c { - Component::Normal(n) => out.push(n), - Component::CurDir => {} - Component::ParentDir => { - if !out.pop() { - return None; - } - } - Component::RootDir | Component::Prefix(_) => return None, - } - } - out.starts_with(&dir.root).then_some(out) -} - -pub struct ActorFs<'a> { - pub inner: WasiFilesystemCtxView<'a>, - pub ext: &'a mut FsExt, -} - -pub struct ActorFsData; - -impl HasData for ActorFsData { - type Data<'a> = ActorFs<'a>; -} - -impl preopens::Host for ActorFs<'_> { - fn get_directories(&mut self) -> wasmtime::Result, String)>> { - let dirs = preopens::Host::get_directories(&mut self.inner)?; - for (r, name) in &dirs { - if let Some(m) = self.ext.mounts.iter().find(|m| &m.guest == name) { - self.ext.dirs.insert( - r.rep(), - DirInfo { host: m.host.clone(), root: m.host.clone(), commit: m.writable }, - ); - } - } - Ok(dirs) - } -} - -impl types::Host for ActorFs<'_> { - fn convert_error_code(&mut self, err: FsError) -> wasmtime::Result { - types::Host::convert_error_code(&mut self.inner, err) - } - - fn filesystem_error_code( - &mut self, - err: Resource, - ) -> wasmtime::Result> { - types::Host::filesystem_error_code(&mut self.inner, err) - } -} - -impl HostDescriptor for ActorFs<'_> { - async fn advise( - &mut self, - fd: Resource, - offset: types::Filesize, - len: types::Filesize, - advice: types::Advice, - ) -> FsResult<()> { - HostDescriptor::advise(&mut self.inner, fd, offset, len, advice).await - } - - async fn sync_data(&mut self, fd: Resource) -> FsResult<()> { - HostDescriptor::sync_data(&mut self.inner, fd).await - } - - async fn get_flags(&mut self, fd: Resource) -> FsResult { - HostDescriptor::get_flags(&mut self.inner, fd).await - } - - async fn get_type(&mut self, fd: Resource) -> FsResult { - HostDescriptor::get_type(&mut self.inner, fd).await - } - - async fn set_size(&mut self, fd: Resource, size: types::Filesize) -> FsResult<()> { - HostDescriptor::set_size(&mut self.inner, fd, size).await - } - - async fn set_times( - &mut self, - fd: Resource, - atim: types::NewTimestamp, - mtim: types::NewTimestamp, - ) -> FsResult<()> { - HostDescriptor::set_times(&mut self.inner, fd, atim, mtim).await - } - - async fn read( - &mut self, - fd: Resource, - len: types::Filesize, - offset: types::Filesize, - ) -> FsResult<(Vec, bool)> { - HostDescriptor::read(&mut self.inner, fd, len, offset).await - } - - async fn write( - &mut self, - fd: Resource, - buf: Vec, - offset: types::Filesize, - ) -> FsResult { - HostDescriptor::write(&mut self.inner, fd, buf, offset).await - } - - async fn read_directory( - &mut self, - fd: Resource, - ) -> FsResult> { - HostDescriptor::read_directory(&mut self.inner, fd).await - } - - async fn sync(&mut self, fd: Resource) -> FsResult<()> { - HostDescriptor::sync(&mut self.inner, fd).await - } - - async fn create_directory_at(&mut self, fd: Resource, path: String) -> FsResult<()> { - HostDescriptor::create_directory_at(&mut self.inner, fd, path).await - } - - async fn stat(&mut self, fd: Resource) -> FsResult { - HostDescriptor::stat(&mut self.inner, fd).await - } - - async fn stat_at( - &mut self, - fd: Resource, - path_flags: types::PathFlags, - path: String, - ) -> FsResult { - if is_tmp_name(&path) { - return Err(ErrorCode::NoEntry.into()); - } - HostDescriptor::stat_at(&mut self.inner, fd, path_flags, path).await - } - - async fn set_times_at( - &mut self, - fd: Resource, - path_flags: types::PathFlags, - path: String, - atim: types::NewTimestamp, - mtim: types::NewTimestamp, - ) -> FsResult<()> { - HostDescriptor::set_times_at(&mut self.inner, fd, path_flags, path, atim, mtim).await - } - - async fn link_at( - &mut self, - fd: Resource, - old_path_flags: types::PathFlags, - old_path: String, - new_descriptor: Resource, - new_path: String, - ) -> FsResult<()> { - HostDescriptor::link_at(&mut self.inner, fd, old_path_flags, old_path, new_descriptor, new_path).await - } - - async fn open_at( - &mut self, - fd: Resource, - path_flags: types::PathFlags, - path: String, - oflags: types::OpenFlags, - flags: types::DescriptorFlags, - ) -> FsResult> { - if is_tmp_name(&path) { - return Err(ErrorCode::NoEntry.into()); - } - let parent = self.ext.dirs.get(&fd.rep()).cloned(); - - // Commit-on-close: a new or truncated file goes to a hidden temporary file. - if let Some(dir) = parent.as_ref().filter(|d| d.commit) - && flags.contains(types::DescriptorFlags::WRITE) - && !oflags.contains(types::OpenFlags::DIRECTORY) - && let Some(target) = resolve(dir, &path) - { - let meta = std::fs::metadata(&target).ok(); - let new_file = meta.is_none() && oflags.contains(types::OpenFlags::CREATE); - let truncated = meta.as_ref().is_some_and(|m| m.is_file()) - && oflags.contains(types::OpenFlags::TRUNCATE) - && !oflags.contains(types::OpenFlags::EXCLUSIVE); - if new_file || truncated { - let n = TMP_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - let tmp_name = format!("{TMP_PREFIX}{}-{n}", std::process::id()); - let tmp_rel = match path.rsplit_once('/') { - Some((d, _)) => format!("{d}/{tmp_name}"), - None => tmp_name.clone(), - }; - let tmp = target.with_file_name(&tmp_name); - // The temporary file is always new, also for `truncate` without `create`. - let oflags = oflags | types::OpenFlags::CREATE; - let r = HostDescriptor::open_at(&mut self.inner, fd, path_flags, tmp_rel, oflags, flags).await?; - self.ext.pending.insert(r.rep(), Pending { tmp, target }); - return Ok(r); - } - } - - let r = HostDescriptor::open_at(&mut self.inner, fd, path_flags, path.clone(), oflags, flags).await?; - if let Some(dir) = parent - && matches!(self.inner.table.get(&r)?, Descriptor::Dir(_)) - && let Some(host) = resolve(&dir, &path) - { - self.ext.dirs.insert(r.rep(), DirInfo { host, ..dir }); - } - Ok(r) - } - - fn drop(&mut self, fd: Resource) -> wasmtime::Result<()> { - let rep = fd.rep(); - HostDescriptor::drop(&mut self.inner, fd)?; - self.ext.dirs.remove(&rep); - if let Some(p) = self.ext.pending.remove(&rep) { - commit(p); - } - Ok(()) - } - - async fn readlink_at(&mut self, fd: Resource, path: String) -> FsResult { - if is_tmp_name(&path) { - return Err(ErrorCode::NoEntry.into()); - } - HostDescriptor::readlink_at(&mut self.inner, fd, path).await - } - - async fn remove_directory_at(&mut self, fd: Resource, path: String) -> FsResult<()> { - HostDescriptor::remove_directory_at(&mut self.inner, fd, path).await - } - - async fn rename_at( - &mut self, - fd: Resource, - old_path: String, - new_fd: Resource, - new_path: String, - ) -> FsResult<()> { - if is_tmp_name(&old_path) || is_tmp_name(&new_path) { - return Err(ErrorCode::NoEntry.into()); - } - let host = |ext: &FsExt, fd: u32, path: &str| ext.dirs.get(&fd).and_then(|d| resolve(d, path)); - let moved = host(self.ext, fd.rep(), &old_path).zip(host(self.ext, new_fd.rep(), &new_path)); - HostDescriptor::rename_at(&mut self.inner, fd, old_path, new_fd, new_path).await?; - // Open writes and directories below a renamed folder move with it. - if let Some((from, to)) = moved { - // `join("")` would add a trailing slash, so a renamed file itself - // needs its own case. - let rebase = |p: &mut PathBuf| { - if *p == from { - *p = to.clone(); - } else if let Ok(rest) = p.strip_prefix(&from) { - *p = to.join(rest); - } - }; - for pending in self.ext.pending.values_mut() { - rebase(&mut pending.tmp); - rebase(&mut pending.target); - } - for dir in self.ext.dirs.values_mut() { - rebase(&mut dir.host); - } - } - Ok(()) - } - - async fn symlink_at( - &mut self, - _fd: Resource, - _src_path: String, - _dest_path: String, - ) -> FsResult<()> { - // Symlinks would break the lexical path tracking of commit-on-close. - Err(ErrorCode::NotPermitted.into()) - } - - async fn unlink_file_at(&mut self, fd: Resource, path: String) -> FsResult<()> { - if is_tmp_name(&path) { - return Err(ErrorCode::NoEntry.into()); - } - HostDescriptor::unlink_file_at(&mut self.inner, fd, path).await - } - - fn read_via_stream( - &mut self, - fd: Resource, - offset: types::Filesize, - ) -> FsResult> { - HostDescriptor::read_via_stream(&mut self.inner, fd, offset) - } - - fn write_via_stream( - &mut self, - fd: Resource, - offset: types::Filesize, - ) -> FsResult> { - HostDescriptor::write_via_stream(&mut self.inner, fd, offset) - } - - fn append_via_stream(&mut self, fd: Resource) -> FsResult> { - HostDescriptor::append_via_stream(&mut self.inner, fd) - } - - async fn is_same_object( - &mut self, - a: Resource, - b: Resource, - ) -> wasmtime::Result { - HostDescriptor::is_same_object(&mut self.inner, a, b).await - } - - async fn metadata_hash(&mut self, fd: Resource) -> FsResult { - HostDescriptor::metadata_hash(&mut self.inner, fd).await - } - - async fn metadata_hash_at( - &mut self, - fd: Resource, - path_flags: types::PathFlags, - path: String, - ) -> FsResult { - if is_tmp_name(&path) { - return Err(ErrorCode::NoEntry.into()); - } - HostDescriptor::metadata_hash_at(&mut self.inner, fd, path_flags, path).await - } -} - -impl HostDirectoryEntryStream for ActorFs<'_> { - async fn read_directory_entry( - &mut self, - stream: Resource, - ) -> FsResult> { - let rep = stream.rep(); - loop { - let entry = HostDirectoryEntryStream::read_directory_entry( - &mut self.inner, - Resource::new_borrow(rep), - ) - .await?; - match entry { - Some(e) if e.name.starts_with(TMP_PREFIX) => continue, - other => return Ok(other), - } - } - } - - fn drop(&mut self, stream: Resource) -> wasmtime::Result<()> { - HostDirectoryEntryStream::drop(&mut self.inner, stream) - } -} diff --git a/cli/src/lib.rs b/cli/src/lib.rs index 8489657..6fcd102 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -7,7 +7,6 @@ pub mod broker; pub mod candidate; pub mod cargo_actor; pub mod console; -pub mod fs; pub mod runtime; pub mod supervisor; pub mod world; diff --git a/cli/src/runtime.rs b/cli/src/runtime.rs index 7cc76c9..eab6d1f 100644 --- a/cli/src/runtime.rs +++ b/cli/src/runtime.rs @@ -15,21 +15,15 @@ use wasmtime::component::{Component, Linker, ResourceTable}; use wasmtime::{Config, Engine, Store}; use wasmtime_wasi::cli::{AsyncStdinStream, AsyncStdoutStream}; use wasmtime_wasi::p2::bindings::Command; -use wasmtime_wasi::p2::bindings::filesystem::{preopens, types}; use wasmtime_wasi::{FsPerms, I32Exit, WasiCtx, WasiCtxView, WasiView}; use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpHooks, WasiHttpView}; -use crate::fs::{ActorFs, ActorFsData, FsExt, Mount}; - -pub struct WasiPart { - ctx: WasiCtx, - table: ResourceTable, -} - -impl WasiView for WasiPart { - fn ctx(&mut self) -> WasiCtxView<'_> { - WasiCtxView { ctx: &mut self.ctx, table: &mut self.table } - } +/// A host folder that the guest sees at `guest`. +#[derive(Clone, Debug)] +pub struct Mount { + pub guest: String, + pub host: PathBuf, + pub writable: bool, } pub struct HttpHooks { @@ -67,28 +61,21 @@ impl WasiHttpHooks for HttpHooks { } pub struct ActorState { - wasi: WasiPart, - fsx: FsExt, + ctx: WasiCtx, + table: ResourceTable, http: WasiHttpCtx, hooks: HttpHooks, } impl WasiView for ActorState { fn ctx(&mut self) -> WasiCtxView<'_> { - self.wasi.ctx() + WasiCtxView { ctx: &mut self.ctx, table: &mut self.table } } } impl WasiHttpView for ActorState { fn http(&mut self) -> WasiHttpCtxView<'_> { - WasiHttpCtxView { ctx: &mut self.http, table: &mut self.wasi.table, hooks: &mut self.hooks } - } -} - -fn actor_fs(s: &mut ActorState) -> ActorFs<'_> { - ActorFs { - inner: wasmtime_wasi::filesystem::WasiFilesystemView::filesystem(&mut s.wasi), - ext: &mut s.fsx, + WasiHttpCtxView { ctx: &mut self.http, table: &mut self.table, hooks: &mut self.hooks } } } @@ -108,12 +95,6 @@ pub enum Exit { Trap(String), } -impl Exit { - pub fn is_clean(&self) -> bool { - matches!(self, Exit::Clean) - } -} - impl std::fmt::Display for Exit { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { @@ -243,11 +224,6 @@ impl Runtime { let mut linker: Linker = Linker::new(&engine); wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; wasmtime_wasi_http::p2::add_only_http_to_linker_async(&mut linker)?; - // Replace the filesystem with the commit-on-close layer. - linker.allow_shadowing(true); - types::add_to_linker::(&mut linker, actor_fs)?; - preopens::add_to_linker::(&mut linker, actor_fs)?; - linker.allow_shadowing(false); // Time slices: every 10 ms each running guest yields to the scheduler. let ticker = engine.clone(); @@ -294,8 +270,8 @@ impl Runtime { .with_context(|| format!("mount {} at {}", m.host.display(), m.guest))?; } let state = ActorState { - wasi: WasiPart { ctx: b.build(), table: ResourceTable::new() }, - fsx: FsExt::new(cfg.mounts), + ctx: b.build(), + table: ResourceTable::new(), http: WasiHttpCtx::new(), hooks: HttpHooks { allow: cfg.network }, }; @@ -321,7 +297,7 @@ impl Runtime { cmd.wasi_cli_run().call_run(&mut store).await } .await; - let exit = match result { + match result { Ok(Ok(())) => Exit::Clean, Ok(Err(())) => Exit::Code(1), Err(e) => match e.downcast_ref::() { @@ -334,12 +310,7 @@ impl Runtime { Exit::Trap(short_trap(&full)) } }, - }; - // Process exit closes every file, so a clean exit commits. - if exit.is_clean() { - store.data_mut().fsx.commit_all(); } - exit }); Ok(Instance { stdin: Some(host_in), stdout: line_rx, task }) diff --git a/cli/src/world.rs b/cli/src/world.rs index 004fbdd..d6ef120 100644 --- a/cli/src/world.rs +++ b/cli/src/world.rs @@ -19,7 +19,8 @@ use std::path::{Component, Path, PathBuf}; use anyhow::{Context, Result, bail}; use serde::{Deserialize, Serialize}; -use crate::fs::TMP_PREFIX; +/// Name prefix of the host's temporary files and folders. Copies skip them. +pub const TMP_PREFIX: &str = ".sausage-tmp-"; /// IDs of the actors that the host implements natively. pub const NATIVE: &[&str] = &["broker", "console", "cargo"]; diff --git a/cli/tests/actors/toolbox/src/main.rs b/cli/tests/actors/toolbox/src/main.rs index 2e97e88..08dc4c1 100644 --- a/cli/tests/actors/toolbox/src/main.rs +++ b/cli/tests/actors/toolbox/src/main.rs @@ -107,33 +107,6 @@ fn main() { Err(error) => io_error(error), } } - Some("open-truncate") => { - let result = (|| -> io::Result<()> { - let mut file = std::fs::OpenOptions::new() - .write(true) - .truncate(true) - .open(&path)?; - file.write_all(text.as_bytes())?; - open.insert(path.clone(), file); - Ok(()) - })(); - match result { - Ok(()) => json!({"type": "opened"}), - Err(error) => io_error(error), - } - } - Some("append") => { - let result = (|| -> io::Result<()> { - let mut file = std::fs::OpenOptions::new().append(true).open(&path)?; - file.write_all(text.as_bytes())?; - open.insert(path.clone(), file); - Ok(()) - })(); - match result { - Ok(()) => json!({"type": "opened"}), - Err(error) => io_error(error), - } - } Some("close") => json!({"type": "closed", "closed": open.remove(&path).is_some()}), Some("arm-shutdown") => { emit_shutdown = true; diff --git a/cli/tests/host.rs b/cli/tests/host.rs index 6108bf0..eb57ed9 100644 --- a/cli/tests/host.rs +++ b/cli/tests/host.rs @@ -3,7 +3,6 @@ mod common; use std::time::Duration; use common::*; -use sausage::fs::TMP_PREFIX; use sausage::{Envelope, Host, World}; use serde_json::{Value, json}; use tokio::sync::mpsc; @@ -206,128 +205,21 @@ async fn actors_can_read_but_not_write_other_actor_data() { } #[tokio::test] -async fn filesystem_hides_new_and_truncated_writes_until_close() { - let (_, host, mut mailbox) = filesystem_world("filesystem-commit").await; - let opened = ask( - &host, - &mut mailbox, - "writer", - json!({"type": "open", "path": "/public/new.txt", "text": "new"}), - ) - .await; - assert_eq!(opened.body["type"], "opened"); - let invisible = ask( - &host, - &mut mailbox, - "reader", - json!({"type": "read", "path": "/actor/writer/public/new.txt"}), - ) - .await; - assert_eq!(invisible.body["type"], "error"); - let listing = ask( - &host, - &mut mailbox, - "writer", - json!({"type": "list", "path": "/public"}), - ) - .await; - let names = listing.body["names"].as_array().unwrap(); - assert!( - names - .iter() - .all(|name| !name.as_str().unwrap().starts_with(TMP_PREFIX)) - ); - assert!(names.iter().all(|name| name.as_str() != Some("new.txt"))); - let closed = ask( - &host, - &mut mailbox, - "writer", - json!({"type": "close", "path": "/public/new.txt"}), - ) - .await; - assert_eq!(closed.body["closed"], true); - let visible = ask( - &host, - &mut mailbox, - "reader", - json!({"type": "read", "path": "/actor/writer/public/new.txt"}), - ) - .await; - assert_eq!(visible.body["text"], "new"); - - ask( - &host, - &mut mailbox, - "writer", - json!({"type": "write", "path": "/public/truncated.txt", "text": "old"}), - ) - .await; - ask( - &host, - &mut mailbox, - "writer", - json!({"type": "open", "path": "/public/truncated.txt", "text": "new"}), - ) - .await; - let old_during_truncate = ask( - &host, - &mut mailbox, - "reader", - json!({"type": "read", "path": "/actor/writer/public/truncated.txt"}), - ) - .await; - assert_eq!(old_during_truncate.body["text"], "old"); - ask( - &host, - &mut mailbox, - "writer", - json!({"type": "close", "path": "/public/truncated.txt"}), - ) - .await; - let new_after_truncate = ask( - &host, - &mut mailbox, - "reader", - json!({"type": "read", "path": "/actor/writer/public/truncated.txt"}), - ) - .await; - assert_eq!(new_after_truncate.body["text"], "new"); - host.stop().await; -} - -#[tokio::test] -async fn filesystem_append_is_visible_before_close() { - let (_, host, mut mailbox) = filesystem_world("filesystem-append").await; - ask( - &host, - &mut mailbox, - "writer", - json!({"type": "write", "path": "/public/append.txt", "text": "base"}), - ) - .await; - ask( - &host, - &mut mailbox, - "writer", - json!({"type": "append", "path": "/public/append.txt", "text": "+tail"}), - ) - .await; - let appended = ask( - &host, - &mut mailbox, - "reader", - json!({"type": "read", "path": "/actor/writer/public/append.txt"}), - ) - .await; - assert_eq!(appended.body["text"], "base+tail"); - let closed = ask( - &host, - &mut mailbox, - "writer", - json!({"type": "close", "path": "/public/append.txt"}), - ) - .await; - assert_eq!(closed.body["closed"], true); +async fn files_are_plain_posix_files() { + let (_, host, mut mailbox) = filesystem_world("posix").await; + // An open file is visible by path, to the writer and to every reader. + let open = json!({"type": "open", "path": "/public/new.txt", "text": "new"}); + ask_type(&host, &mut mailbox, "writer", open, "opened").await; + let own = json!({"type": "read", "path": "/public/new.txt"}); + assert_eq!(ask_type(&host, &mut mailbox, "writer", own, "content").await.body["text"], "new"); + let other = json!({"type": "read", "path": "/actor/writer/public/new.txt"}); + assert_eq!(ask_type(&host, &mut mailbox, "reader", other, "content").await.body["text"], "new"); + // It can be renamed while open. + let rename = json!({"type": "rename", "path": "/public/new.txt", "to": "/public/moved.txt"}); + ask_type(&host, &mut mailbox, "writer", rename, "ok").await; + ask_type(&host, &mut mailbox, "writer", json!({"type": "close", "path": "/public/new.txt"}), "closed").await; + let moved = json!({"type": "read", "path": "/actor/writer/public/moved.txt"}); + assert_eq!(ask_type(&host, &mut mailbox, "reader", moved, "content").await.body["text"], "new"); host.stop().await; } @@ -752,75 +644,6 @@ async fn shutdown_reply_and_publish_survive_stop_and_upgrade() { host.stop().await; } -#[tokio::test] -async fn close_commits_open_file_after_parent_directory_rename() { - let (_, host, mut mailbox) = filesystem_world("rename-open-file").await; - let mkdir = json!({"type": "mkdir", "path": "/public/a"}); - ask_type(&host, &mut mailbox, "writer", mkdir, "ok").await; - let open = json!({"type": "open", "path": "/public/a/file", "text": "content"}); - ask_type(&host, &mut mailbox, "writer", open, "opened").await; - let rename = json!({"type": "rename", "path": "/public/a", "to": "/public/b"}); - ask_type(&host, &mut mailbox, "writer", rename, "ok").await; - let close = json!({"type": "close", "path": "/public/a/file"}); - assert_eq!( - ask(&host, &mut mailbox, "writer", close).await.body["closed"], - true - ); - let read = json!({"type": "read", "path": "/public/b/file"}); - assert_eq!( - ask_type(&host, &mut mailbox, "writer", read, "content") - .await - .body["text"], - "content" - ); - host.stop().await; -} - -#[tokio::test] -async fn close_commits_a_staged_write_after_the_file_itself_is_renamed() { - let (_, host, mut mailbox) = filesystem_world("rename-staged-file").await; - let write = json!({"type": "write", "path": "/data/x", "text": "old"}); - ask_type(&host, &mut mailbox, "writer", write, "ok").await; - let open = json!({"type": "open-truncate", "path": "/data/x", "text": "new"}); - ask_type(&host, &mut mailbox, "writer", open, "opened").await; - let rename = json!({"type": "rename", "path": "/data/x", "to": "/data/x.renamed"}); - ask_type(&host, &mut mailbox, "writer", rename, "ok").await; - ask_type(&host, &mut mailbox, "writer", json!({"type": "close", "path": "/data/x"}), "closed").await; - let read = json!({"type": "read", "path": "/data/x.renamed"}); - assert_eq!(ask_type(&host, &mut mailbox, "writer", read, "content").await.body["text"], "new"); - host.stop().await; -} - -#[tokio::test] -async fn truncate_without_create_commits_existing_file_on_close() { - let (_, host, mut mailbox) = filesystem_world("truncate-without-create").await; - let write = json!({"type": "write", "path": "/public/file", "text": "old"}); - ask_type(&host, &mut mailbox, "writer", write, "ok").await; - let open = json!({"type": "open-truncate", "path": "/public/file", "text": "new"}); - ask_type(&host, &mut mailbox, "writer", open, "opened").await; - let path = "/actor/writer/public/file"; - let read = json!({"type": "read", "path": path}); - assert_eq!( - ask_type(&host, &mut mailbox, "reader", read, "content") - .await - .body["text"], - "old" - ); - let close = json!({"type": "close", "path": "/public/file"}); - assert_eq!( - ask(&host, &mut mailbox, "writer", close).await.body["closed"], - true - ); - let read = json!({"type": "read", "path": path}); - assert_eq!( - ask_type(&host, &mut mailbox, "reader", read, "content") - .await - .body["text"], - "new" - ); - host.stop().await; -} - #[tokio::test] async fn candidates_receive_version_and_preexisting_spawn_state() { let world = create_world("candidate-context"); diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index 844023b..3acb758 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -77,6 +77,15 @@ pub fn reply(message: &Message, body: Value) -> String { format!("{}/{id}", me()) } +/// Write a file so that readers see either the old or the new content, +/// never half of it: write a temporary file next to `path`, then rename it. +pub fn write_atomic(path: impl AsRef, contents: impl AsRef<[u8]>) -> std::io::Result<()> { + let path = path.as_ref(); + let tmp = path.with_file_name(format!(".{}.tmp-{}", path.file_name().unwrap_or_default().to_string_lossy(), next_id())); + std::fs::write(&tmp, contents)?; + std::fs::rename(&tmp, path) +} + /// Publish a message to this actor's subscribers and return its full ID. pub fn publish(body: Value) -> String { let id = next_id(); @@ -201,6 +210,18 @@ mod tests { assert_eq!(message.kind(), ""); } + #[test] + fn write_atomic_replaces_the_file() { + let dir = std::env::temp_dir().join(format!("sausage-actor-{}", next_id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("state.json"); + write_atomic(&path, "old").unwrap(); + write_atomic(&path, "new").unwrap(); + assert_eq!(std::fs::read_to_string(&path).unwrap(), "new"); + assert_eq!(std::fs::read_dir(&dir).unwrap().count(), 1); + std::fs::remove_dir_all(dir).unwrap(); + } + #[test] fn generated_ids_differ() { assert_ne!(next_id(), next_id());