Plain POSIX files: remove commit on close (fs.rs)

Staging every write until close broke tempfile, git, and SQLite, which
expect normal files. Actors that need an atomic publish write a temporary
file and rename it; the SDK has write_atomic for that.
This commit is contained in:
2026-10-15 23:33:20 +01:00
parent 599ee510cb
commit bcde6ba325
11 changed files with 59 additions and 718 deletions
+1 -2
View File
@@ -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};
-446
View File
@@ -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<Mount>,
dirs: HashMap<u32, DirInfo>,
pending: HashMap<u32, Pending>,
}
impl FsExt {
pub fn new(mounts: Vec<Mount>) -> 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<PathBuf> {
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<Vec<(Resource<Descriptor>, 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<ErrorCode> {
types::Host::convert_error_code(&mut self.inner, err)
}
fn filesystem_error_code(
&mut self,
err: Resource<wasmtime::Error>,
) -> wasmtime::Result<Option<ErrorCode>> {
types::Host::filesystem_error_code(&mut self.inner, err)
}
}
impl HostDescriptor for ActorFs<'_> {
async fn advise(
&mut self,
fd: Resource<types::Descriptor>,
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<types::Descriptor>) -> FsResult<()> {
HostDescriptor::sync_data(&mut self.inner, fd).await
}
async fn get_flags(&mut self, fd: Resource<types::Descriptor>) -> FsResult<types::DescriptorFlags> {
HostDescriptor::get_flags(&mut self.inner, fd).await
}
async fn get_type(&mut self, fd: Resource<types::Descriptor>) -> FsResult<types::DescriptorType> {
HostDescriptor::get_type(&mut self.inner, fd).await
}
async fn set_size(&mut self, fd: Resource<types::Descriptor>, size: types::Filesize) -> FsResult<()> {
HostDescriptor::set_size(&mut self.inner, fd, size).await
}
async fn set_times(
&mut self,
fd: Resource<types::Descriptor>,
atim: types::NewTimestamp,
mtim: types::NewTimestamp,
) -> FsResult<()> {
HostDescriptor::set_times(&mut self.inner, fd, atim, mtim).await
}
async fn read(
&mut self,
fd: Resource<types::Descriptor>,
len: types::Filesize,
offset: types::Filesize,
) -> FsResult<(Vec<u8>, bool)> {
HostDescriptor::read(&mut self.inner, fd, len, offset).await
}
async fn write(
&mut self,
fd: Resource<types::Descriptor>,
buf: Vec<u8>,
offset: types::Filesize,
) -> FsResult<types::Filesize> {
HostDescriptor::write(&mut self.inner, fd, buf, offset).await
}
async fn read_directory(
&mut self,
fd: Resource<types::Descriptor>,
) -> FsResult<Resource<types::DirectoryEntryStream>> {
HostDescriptor::read_directory(&mut self.inner, fd).await
}
async fn sync(&mut self, fd: Resource<types::Descriptor>) -> FsResult<()> {
HostDescriptor::sync(&mut self.inner, fd).await
}
async fn create_directory_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<()> {
HostDescriptor::create_directory_at(&mut self.inner, fd, path).await
}
async fn stat(&mut self, fd: Resource<types::Descriptor>) -> FsResult<types::DescriptorStat> {
HostDescriptor::stat(&mut self.inner, fd).await
}
async fn stat_at(
&mut self,
fd: Resource<types::Descriptor>,
path_flags: types::PathFlags,
path: String,
) -> FsResult<types::DescriptorStat> {
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<types::Descriptor>,
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<types::Descriptor>,
old_path_flags: types::PathFlags,
old_path: String,
new_descriptor: Resource<types::Descriptor>,
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<types::Descriptor>,
path_flags: types::PathFlags,
path: String,
oflags: types::OpenFlags,
flags: types::DescriptorFlags,
) -> FsResult<Resource<types::Descriptor>> {
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<types::Descriptor>) -> 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<types::Descriptor>, path: String) -> FsResult<String> {
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<types::Descriptor>, path: String) -> FsResult<()> {
HostDescriptor::remove_directory_at(&mut self.inner, fd, path).await
}
async fn rename_at(
&mut self,
fd: Resource<types::Descriptor>,
old_path: String,
new_fd: Resource<types::Descriptor>,
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<types::Descriptor>,
_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<types::Descriptor>, 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<types::Descriptor>,
offset: types::Filesize,
) -> FsResult<Resource<DynInputStream>> {
HostDescriptor::read_via_stream(&mut self.inner, fd, offset)
}
fn write_via_stream(
&mut self,
fd: Resource<types::Descriptor>,
offset: types::Filesize,
) -> FsResult<Resource<DynOutputStream>> {
HostDescriptor::write_via_stream(&mut self.inner, fd, offset)
}
fn append_via_stream(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Resource<DynOutputStream>> {
HostDescriptor::append_via_stream(&mut self.inner, fd)
}
async fn is_same_object(
&mut self,
a: Resource<types::Descriptor>,
b: Resource<types::Descriptor>,
) -> wasmtime::Result<bool> {
HostDescriptor::is_same_object(&mut self.inner, a, b).await
}
async fn metadata_hash(&mut self, fd: Resource<types::Descriptor>) -> FsResult<types::MetadataHashValue> {
HostDescriptor::metadata_hash(&mut self.inner, fd).await
}
async fn metadata_hash_at(
&mut self,
fd: Resource<types::Descriptor>,
path_flags: types::PathFlags,
path: String,
) -> FsResult<types::MetadataHashValue> {
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<types::DirectoryEntryStream>,
) -> FsResult<Option<types::DirectoryEntry>> {
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<types::DirectoryEntryStream>) -> wasmtime::Result<()> {
HostDirectoryEntryStream::drop(&mut self.inner, stream)
}
}
-1
View File
@@ -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;
+13 -42
View File
@@ -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<ActorState> = 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::<ActorState, ActorFsData>(&mut linker, actor_fs)?;
preopens::add_to_linker::<ActorState, ActorFsData>(&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::<I32Exit>() {
@@ -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 })
+2 -1
View File
@@ -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"];