From 2ff30ecf703cdf76cf301836dcac04b3055b92b9 Mon Sep 17 00:00:00 2001 From: Sausage Agent Date: Fri, 16 Oct 2026 01:25:11 +0000 Subject: [PATCH] Actor-served files: /actor//fs/ File operations under /actor//fs/ become messages from `fs` to (fs_read, fs_write, fs_mkdir, fs_remove, fs_rename) with a 10 s timeout. vfs.rs wraps the wasi:filesystem implementation of wasmtime-wasi. --- PROTOCOL.md | 27 + cli/src/broker.rs | 33 +- cli/src/lib.rs | 1 + cli/src/runtime.rs | 22 +- cli/src/vfs.rs | 716 +++++++++++++++++++++++++++ cli/src/world.rs | 5 +- cli/tests/actors/toolbox/src/main.rs | 36 ++ cli/tests/host.rs | 30 ++ 8 files changed, 866 insertions(+), 4 deletions(-) create mode 100644 cli/src/vfs.rs diff --git a/PROTOCOL.md b/PROTOCOL.md index 752ac19..1fd6f83 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -78,6 +78,33 @@ so that readers never see half of it, write a temporary file and `rename` it Send large or binary data as a path under `/actor//public/`, not inside a message. +### Actor-served files: `/actor//fs/` + +Files under `/actor//fs/` are not on disk: the actor `` serves them. +Every actor can read and write them. The host turns each file operation into a +message from `fs` to `` and waits up to 10 seconds for the reply. `path` +is relative to `fs/` and starts with `/` (the folder itself is `/`). + +| Request | Reply | +| --- | --- | +| `{"type": "fs_read", "path"}` | `{"type": "fs_file", "text"}` (or `"base64"` instead of `"text"`), or `{"type": "fs_dir", "entries": [{"name", "type": "file" \| "dir"}]}` | +| `{"type": "fs_write", "path", "text"}` (or `"base64"` for non-UTF-8 data) | `{"type": "fs_done"}` | +| `{"type": "fs_mkdir", "path"}`, `{"type": "fs_remove", "path"}` | `{"type": "fs_done"}` | +| `{"type": "fs_rename", "path", "to"}` | `{"type": "fs_done"}` | + +Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`. +`errno` is a `wasi:filesystem` error code: `no-entry`, `exist`, +`not-directory`, `is-directory`, `not-empty`, `access`, `not-permitted`, +`invalid`, `unsupported`, `busy`, or `io` (also for anything else). + +- Opening a file sends `fs_read` once. Reads come from that copy. +- After a change, closing (or syncing) the file sends the whole file in one + `fs_write`. +- No reply within 10 seconds, or an actor that does not exist, makes the + operation fail (`io`, or `no-entry`). An actor that opens its own `fs/` + waits for itself and fails after the timeout. +- `fs` is reserved: no actor can have that ID. + ## The broker Requests to `broker`. The answer is a message whose `reply_to` is the ID of diff --git a/cli/src/broker.rs b/cli/src/broker.rs index 918640a..c90f6e1 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -20,6 +20,7 @@ use wasmtime::component::Component; use crate::candidate; use crate::runtime::{InstanceConfig, Mount, Runtime}; use crate::supervisor::{Control, Reply, Supervisor}; +use crate::vfs; use crate::world::{self, ActorMeta, NATIVE, VersionMeta, World}; pub const MAILBOX_SIZE: usize = 1000; @@ -60,12 +61,15 @@ pub struct Broker { seq: AtomicU64, /// Set by the `exit` request: the host process should stop. exit: tokio::sync::Notify, + /// Requests of the host from `fs` (see `vfs.rs`) that wait for a reply, by message ID. + asks: Mutex>>, + me: std::sync::Weak, } impl Broker { pub fn new(world: World, runtime: Arc) -> Result> { let metas = world.actor_ids()?.into_iter().map(|id| (id.clone(), world.read_meta(&id))).collect(); - Ok(Arc::new(Self { + Ok(Arc::new_cyclic(|me| Self { world, runtime, mailboxes: Mutex::new(HashMap::new()), @@ -74,9 +78,27 @@ impl Broker { op_locks: Mutex::new(HashMap::new()), seq: AtomicU64::new(1), exit: tokio::sync::Notify::new(), + asks: Mutex::new(HashMap::new()), + me: me.clone(), })) } + /// Send `body` from `fs` to `to` now. The future resolves to the reply + /// body, or `None` after `vfs::ASK_TIMEOUT`. + pub fn ask(self: &Arc, to: &str, body: Value) -> impl Future> + Send + 'static { + let own = format!("q{}", self.seq.fetch_add(1, Ordering::Relaxed)); + let id = format!("{}/{own}", vfs::ID); + let (tx, rx) = oneshot::channel(); + self.asks.lock().unwrap().insert(id.clone(), tx); + self.send(vfs::ID, Some(to), Some(&own), None, body); + let this = self.clone(); + async move { + let reply = tokio::time::timeout(vfs::ASK_TIMEOUT, rx).await; + this.asks.lock().unwrap().remove(&id); + reply.ok()?.ok() + } + } + /// Wait until an actor sends `{"type": "exit"}` to `broker`. The `sausage` /// binary then stops all actors and exits, so that a wrapper such as /// `nix run .#loop` can start a fresh (updated) host. @@ -155,6 +177,13 @@ impl Broker { } fn deliver(&self, env: Envelope) { + if env.to == vfs::ID { + let waiting = env.reply_to.as_ref().and_then(|r| self.asks.lock().unwrap().remove(r)); + if let Some(tx) = waiting { + let _ = tx.send(env.body); + } + return; + } let tx = self.mailboxes.lock().unwrap().get(&env.to).cloned(); let error = match tx { None => format!("no actor with ID {}", env.to), @@ -201,6 +230,8 @@ impl Broker { network: scratch.is_none(), stderr_log: own.join("stderr.log"), env, + actor: id.to_string(), + broker: self.me.clone(), } } diff --git a/cli/src/lib.rs b/cli/src/lib.rs index 6fcd102..3ca15ab 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -9,6 +9,7 @@ pub mod cargo_actor; pub mod console; pub mod runtime; pub mod supervisor; +pub mod vfs; pub mod world; use std::path::Path; diff --git a/cli/src/runtime.rs b/cli/src/runtime.rs index eab6d1f..ce0c9ad 100644 --- a/cli/src/runtime.rs +++ b/cli/src/runtime.rs @@ -4,7 +4,7 @@ //! NDJSON lines to its stdin and reads NDJSON lines from its stdout. use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::{Arc, Weak}; use std::time::Duration; use anyhow::{Context, Result}; @@ -14,10 +14,15 @@ use tokio::task::JoinHandle; use wasmtime::component::{Component, Linker, ResourceTable}; use wasmtime::{Config, Engine, Store}; use wasmtime_wasi::cli::{AsyncStdinStream, AsyncStdoutStream}; +use wasmtime_wasi::filesystem::WasiFilesystemCtxView; use wasmtime_wasi::p2::bindings::Command; +use wasmtime_wasi::p2::bindings::filesystem; use wasmtime_wasi::{FsPerms, I32Exit, WasiCtx, WasiCtxView, WasiView}; use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpHooks, WasiHttpView}; +use crate::broker::Broker; +use crate::vfs::{Vfs, VfsState, VfsView}; + /// A host folder that the guest sees at `guest`. #[derive(Clone, Debug)] pub struct Mount { @@ -65,6 +70,11 @@ pub struct ActorState { table: ResourceTable, http: WasiHttpCtx, hooks: HttpHooks, + vfs: VfsState, +} + +fn vfs_view(s: &mut ActorState) -> VfsView<'_> { + VfsView { inner: WasiFilesystemCtxView { ctx: s.ctx.filesystem(), table: &mut s.table }, state: &mut s.vfs } } impl WasiView for ActorState { @@ -111,6 +121,10 @@ pub struct InstanceConfig { pub network: bool, pub stderr_log: PathBuf, pub env: Vec<(String, String)>, + /// The actor that runs in the instance. + pub actor: String, + /// Asks the actors that serve `/actor//fs/` (see `vfs.rs`). + pub broker: Weak, } /// A running guest. Dropping it kills the guest. @@ -224,6 +238,11 @@ 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 `wasi:filesystem` with the wrapper that serves `/actor//fs/`. + linker.allow_shadowing(true); + filesystem::types::add_to_linker::(&mut linker, vfs_view)?; + filesystem::preopens::add_to_linker::(&mut linker, vfs_view)?; + linker.allow_shadowing(false); // Time slices: every 10 ms each running guest yields to the scheduler. let ticker = engine.clone(); @@ -274,6 +293,7 @@ impl Runtime { table: ResourceTable::new(), http: WasiHttpCtx::new(), hooks: HttpHooks { allow: cfg.network }, + vfs: VfsState::new(cfg.broker.clone(), cfg.actor.clone()), }; let (line_tx, line_rx) = mpsc::channel(1024); diff --git a/cli/src/vfs.rs b/cli/src/vfs.rs new file mode 100644 index 0000000..21fc023 --- /dev/null +++ b/cli/src/vfs.rs @@ -0,0 +1,716 @@ +//! Actor-served files: `/actor//fs/...` (see PROTOCOL.md, "Actor-served +//! files"). +//! +//! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`. +//! A path under `/actor//fs/` does not reach the disk: the host asks the +//! actor `` with a message from `fs` and waits up to `ASK_TIMEOUT` for +//! the reply. Every other path goes to `wasmtime-wasi` unchanged. +//! +//! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}` +//! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`, +//! `fs_rename {path, to}` -> `fs_done` +//! - any of them -> `fs_error {errno}` (a `wasi:filesystem` error-code name) +//! +//! A file is read once when it is opened. After a change, it is written back +//! in one `fs_write` when it is synced or closed. + +use std::collections::HashSet; +use std::hash::{Hash, Hasher}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, Weak}; + +use bytes::Bytes; +use serde_json::{Value, json}; +use wasmtime::component::{HasData, Resource}; +use wasmtime_wasi::filesystem::WasiFilesystemCtxView; +use wasmtime_wasi::p2::bindings::filesystem::types::{ + self, DescriptorFlags, DescriptorType, DirectoryEntry, ErrorCode, HostDescriptor, HostDirectoryEntryStream, OpenFlags, +}; +use wasmtime_wasi::p2::bindings::filesystem::preopens; +use wasmtime_wasi::p2::pipe::MemoryInputPipe; +use wasmtime_wasi::p2::{DynInputStream, DynOutputStream, FsError, FsResult, OutputStream, Pollable, StreamResult}; + +use crate::broker::Broker; + +/// The `from` of the host's requests. Replies to it go back to the waiting call. +pub const ID: &str = "fs"; +pub const ASK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +/// Per-instance state. +pub struct VfsState { + broker: Weak, + /// The actor of this instance, for notes in its `stderr.log`. + me: String, + /// Descriptors of the `/actor` preopen. + roots: HashSet, +} + +impl VfsState { + pub fn new(broker: Weak, me: String) -> Self { + Self { broker, me, roots: HashSet::new() } + } +} + +/// [`HasData`] for the linker: the filesystem functions get a [`VfsView`]. +pub struct Vfs; + +impl HasData for Vfs { + type Data<'a> = VfsView<'a>; +} + +pub struct VfsView<'a> { + pub inner: WasiFilesystemCtxView<'a>, + pub state: &'a mut VfsState, +} + +/// A descriptor of an actor-served file or folder (in the resource table in +/// place of a `wasmtime-wasi` descriptor). +#[derive(Clone)] +enum Virt { + Dir { actor: String, path: String }, + File(VFile), +} + +#[derive(Clone)] +struct VFile { + actor: String, + path: String, + data: Arc>>, + dirty: Arc, + write: bool, +} + +/// A directory-entry-stream of an actor-served folder. +struct VDir(std::vec::IntoIter); + +enum Node { + File(Vec), + Dir(Vec), +} + +/// Sends the requests. Owned, so that no borrow of the view lives across an await. +struct Asker { + broker: Option>, + me: String, +} + +impl Asker { + async fn ask(&self, actor: &str, body: Value) -> Result { + let broker = self.broker.as_ref().ok_or(ErrorCode::Io)?; + let what = format!("{} {} to {actor}", body["type"].as_str().unwrap_or(""), body["path"].as_str().unwrap_or("")); + match broker.ask(actor, body).await { + None => { + broker.log(&self.me, &format!("{what}: no reply within {} s", ASK_TIMEOUT.as_secs())); + Err(ErrorCode::Io) + } + Some(reply) => match reply["type"].as_str() { + Some("fs_error") => Err(errno(&reply["errno"])), + Some("undeliverable") => Err(ErrorCode::NoEntry), + _ => Ok(reply), + }, + } + } + + async fn read(&self, actor: &str, path: &str) -> Result { + let reply = self.ask(actor, json!({"type": "fs_read", "path": path})).await?; + match reply["type"].as_str() { + Some("fs_file") => content(&reply).map(Node::File), + Some("fs_dir") => Ok(Node::Dir( + reply["entries"] + .as_array() + .into_iter() + .flatten() + .filter_map(|e| { + let type_ = if e["type"] == "dir" { DescriptorType::Directory } else { DescriptorType::RegularFile }; + Some(DirectoryEntry { type_, name: e["name"].as_str()?.to_string() }) + }) + .collect(), + )), + _ => { + if let Some(b) = &self.broker { + b.log(&self.me, &format!("fs_read {path} to {actor}: unexpected reply {reply}")); + } + Err(ErrorCode::Io) + } + } + } + + async fn change(&self, actor: &str, body: Value) -> FsResult<()> { + self.ask(actor, body).await.map(|_| ()).map_err(FsError::from) + } + + async fn flush(&self, f: &VFile) -> FsResult<()> { + if f.dirty.swap(false, Ordering::SeqCst) + && let Err(e) = self.ask(&f.actor, write_body(f)).await + { + f.dirty.store(true, Ordering::SeqCst); + return Err(e.into()); + } + Ok(()) + } +} + +fn errno(v: &Value) -> ErrorCode { + match v.as_str().unwrap_or("") { + "no-entry" => ErrorCode::NoEntry, + "exist" => ErrorCode::Exist, + "not-directory" => ErrorCode::NotDirectory, + "is-directory" => ErrorCode::IsDirectory, + "not-empty" => ErrorCode::NotEmpty, + "access" => ErrorCode::Access, + "not-permitted" => ErrorCode::NotPermitted, + "invalid" => ErrorCode::Invalid, + "unsupported" => ErrorCode::Unsupported, + "busy" => ErrorCode::Busy, + _ => ErrorCode::Io, + } +} + +fn content(v: &Value) -> Result, ErrorCode> { + if let Some(text) = v["text"].as_str() { + return Ok(text.as_bytes().to_vec()); + } + match v["base64"].as_str() { + Some(b) => base64_decode(b).ok_or(ErrorCode::IllegalByteSequence), + None => Ok(Vec::new()), + } +} + +fn write_body(f: &VFile) -> Value { + let data = f.data.lock().unwrap().clone(); + match String::from_utf8(data) { + Ok(text) => json!({"type": "fs_write", "path": f.path, "text": text}), + Err(e) => json!({"type": "fs_write", "path": f.path, "base64": base64_encode(e.as_bytes())}), + } +} + +/// `rel` resolved against the folder `base`, as an absolute path. +fn join(base: &str, rel: &str) -> String { + let mut parts: Vec<&str> = base.split('/').filter(|p| !p.is_empty()).collect(); + for p in rel.split('/') { + match p { + "" | "." => {} + ".." => { + parts.pop(); + } + p => parts.push(p), + } + } + format!("/{}", parts.join("/")) +} + +fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat { + types::DescriptorStat { + type_, + link_count: 1, + size, + data_access_timestamp: None, + data_modification_timestamp: None, + status_change_timestamp: None, + } +} + +fn node_stat(node: &Node) -> types::DescriptorStat { + match node { + Node::File(data) => stat(DescriptorType::RegularFile, data.len() as u64), + Node::Dir(_) => stat(DescriptorType::Directory, 0), + } +} + +fn hash(actor: &str, path: &str) -> types::MetadataHashValue { + let mut h = std::collections::hash_map::DefaultHasher::new(); + (actor, path).hash(&mut h); + types::MetadataHashValue { lower: h.finish(), upper: 0 } +} + +fn write_at(data: &mut Vec, at: usize, bytes: &[u8]) { + if data.len() < at + bytes.len() { + data.resize(at + bytes.len(), 0); + } + data[at..at + bytes.len()].copy_from_slice(bytes); +} + +/// An output stream into an actor-served file (`pos: None` appends). +struct VWriter { + file: VFile, + pos: Option, +} + +#[wasmtime_wasi::async_trait] +impl Pollable for VWriter { + async fn ready(&mut self) {} +} + +#[wasmtime_wasi::async_trait] +impl OutputStream for VWriter { + fn write(&mut self, bytes: Bytes) -> StreamResult<()> { + let mut data = self.file.data.lock().unwrap(); + let at = self.pos.unwrap_or(data.len()); + write_at(&mut data, at, &bytes); + if let Some(p) = &mut self.pos { + *p += bytes.len(); + } + self.file.dirty.store(true, Ordering::SeqCst); + Ok(()) + } + fn flush(&mut self) -> StreamResult<()> { + Ok(()) + } + fn check_write(&mut self) -> StreamResult { + Ok(64 * 1024) + } +} + +impl VfsView<'_> { + fn asker(&self) -> Asker { + Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() } + } + + fn virt(&mut self, fd: &Resource) -> Option { + self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::().cloned() + } + + fn file(&mut self, fd: &Resource) -> Option> { + match self.virt(fd)? { + Virt::File(f) => Some(Ok(f)), + Virt::Dir { .. } => Some(Err(ErrorCode::IsDirectory.into())), + } + } + + /// The actor and its path, if `path` relative to `fd` is actor-served. + fn target(&mut self, fd: &Resource, path: &str) -> Option<(String, String)> { + match self.virt(fd) { + Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))), + Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))), + None if self.state.roots.contains(&fd.rep()) => { + let full = join("/", path); + let mut parts = full.splitn(4, '/').skip(1); + let (actor, fs, rest) = (parts.next()?, parts.next()?, parts.next().unwrap_or("")); + (fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{rest}"))) + } + None => None, + } + } + + fn push(&mut self, v: Virt) -> FsResult> { + let r = self.inner.table.push(v)?; + Ok(Resource::new_own(r.rep())) + } + + async fn open(&mut self, actor: String, path: String, oflags: OpenFlags, flags: DescriptorFlags) -> FsResult> { + let write = flags.contains(DescriptorFlags::WRITE); + let create = oflags.contains(OpenFlags::CREATE); + let exclusive = create && oflags.contains(OpenFlags::EXCLUSIVE); + let virt = match self.asker().read(&actor, &path).await { + Ok(_) if exclusive => return Err(ErrorCode::Exist.into()), + Ok(Node::Dir(_)) if write => return Err(ErrorCode::IsDirectory.into()), + Ok(Node::Dir(_)) => Virt::Dir { actor, path }, + Ok(Node::File(_)) if oflags.contains(OpenFlags::DIRECTORY) => return Err(ErrorCode::NotDirectory.into()), + Ok(Node::File(data)) => { + let truncate = oflags.contains(OpenFlags::TRUNCATE); + let data = if truncate { Vec::new() } else { data }; + Virt::File(VFile { actor, path, data: Arc::new(Mutex::new(data)), dirty: Arc::new(AtomicBool::new(truncate)), write }) + } + Err(ErrorCode::NoEntry) if create && !oflags.contains(OpenFlags::DIRECTORY) => Virt::File(VFile { + actor, + path, + data: Arc::default(), + dirty: Arc::new(AtomicBool::new(true)), + write, + }), + Err(e) => return Err(e.into()), + }; + self.push(virt) + } + + /// Send the changed file without waiting (the descriptor is gone). The + /// request leaves now, so later requests of this instance come after it. + fn write_back_later(&self, f: &VFile) { + let Some(broker) = self.state.broker.upgrade() else { return }; + let reply = broker.ask(&f.actor, write_body(f)); + let (me, what) = (self.state.me.clone(), format!("fs_write {} to {}", f.path, f.actor)); + tokio::spawn(async move { + match reply.await { + Some(r) if r["type"] != "fs_error" && r["type"] != "undeliverable" => {} + r => broker.log(&me, &format!("{what} failed: {}", r.map_or("no reply".to_string(), |r| r.to_string()))), + } + }); + } +} + +impl preopens::Host for VfsView<'_> { + fn get_directories(&mut self) -> wasmtime::Result, String)>> { + let dirs = preopens::Host::get_directories(&mut self.inner)?; + for (fd, name) in &dirs { + if name == "/actor" { + self.state.roots.insert(fd.rep()); + } + } + Ok(dirs) + } +} + +impl types::Host for VfsView<'_> { + 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 VfsView<'_> { + async fn advise(&mut self, fd: Resource, offset: types::Filesize, len: types::Filesize, advice: types::Advice) -> FsResult<()> { + if self.virt(&fd).is_some() { + return Ok(()); + } + HostDescriptor::advise(&mut self.inner, fd, offset, len, advice).await + } + + async fn sync_data(&mut self, fd: Resource) -> FsResult<()> { + match self.virt(&fd) { + Some(Virt::File(f)) => self.asker().flush(&f).await, + Some(Virt::Dir { .. }) => Ok(()), + None => HostDescriptor::sync_data(&mut self.inner, fd).await, + } + } + + async fn get_flags(&mut self, fd: Resource) -> FsResult { + match self.virt(&fd) { + Some(Virt::File(f)) if f.write => Ok(DescriptorFlags::READ | DescriptorFlags::WRITE), + Some(Virt::File(_)) => Ok(DescriptorFlags::READ), + Some(Virt::Dir { .. }) => Ok(DescriptorFlags::READ | DescriptorFlags::MUTATE_DIRECTORY), + None => HostDescriptor::get_flags(&mut self.inner, fd).await, + } + } + + async fn get_type(&mut self, fd: Resource) -> FsResult { + match self.virt(&fd) { + Some(Virt::File(_)) => Ok(DescriptorType::RegularFile), + Some(Virt::Dir { .. }) => Ok(DescriptorType::Directory), + None => HostDescriptor::get_type(&mut self.inner, fd).await, + } + } + + async fn set_size(&mut self, fd: Resource, size: types::Filesize) -> FsResult<()> { + match self.file(&fd) { + Some(f) => { + let f = f?; + if !f.write { + return Err(ErrorCode::NotPermitted.into()); + } + f.data.lock().unwrap().resize(size as usize, 0); + f.dirty.store(true, Ordering::SeqCst); + Ok(()) + } + None => HostDescriptor::set_size(&mut self.inner, fd, size).await, + } + } + + async fn set_times(&mut self, fd: Resource, atim: types::NewTimestamp, mtim: types::NewTimestamp) -> FsResult<()> { + if self.virt(&fd).is_some() { + return Ok(()); + } + 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)> { + match self.file(&fd) { + Some(f) => { + let f = f?; + let data = f.data.lock().unwrap(); + let start = usize::try_from(offset).unwrap_or(usize::MAX).min(data.len()); + let end = start.saturating_add(usize::try_from(len).unwrap_or(usize::MAX)).min(data.len()); + Ok((data[start..end].to_vec(), start == end)) + } + None => HostDescriptor::read(&mut self.inner, fd, len, offset).await, + } + } + + async fn write(&mut self, fd: Resource, buf: Vec, offset: types::Filesize) -> FsResult { + match self.file(&fd) { + Some(f) => { + let f = f?; + if !f.write { + return Err(ErrorCode::NotPermitted.into()); + } + write_at(&mut f.data.lock().unwrap(), offset as usize, &buf); + f.dirty.store(true, Ordering::SeqCst); + Ok(buf.len() as u64) + } + None => HostDescriptor::write(&mut self.inner, fd, buf, offset).await, + } + } + + async fn read_directory(&mut self, fd: Resource) -> FsResult> { + match self.virt(&fd) { + Some(Virt::Dir { actor, path }) => match self.asker().read(&actor, &path).await? { + Node::Dir(entries) => { + let r = self.inner.table.push(VDir(entries.into_iter()))?; + Ok(Resource::new_own(r.rep())) + } + Node::File(_) => Err(ErrorCode::NotDirectory.into()), + }, + Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()), + None => HostDescriptor::read_directory(&mut self.inner, fd).await, + } + } + + async fn sync(&mut self, fd: Resource) -> FsResult<()> { + match self.virt(&fd) { + Some(Virt::File(f)) => self.asker().flush(&f).await, + Some(Virt::Dir { .. }) => Ok(()), + None => HostDescriptor::sync(&mut self.inner, fd).await, + } + } + + async fn create_directory_at(&mut self, fd: Resource, path: String) -> FsResult<()> { + match self.target(&fd, &path) { + Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_mkdir", "path": path})).await, + None => HostDescriptor::create_directory_at(&mut self.inner, fd, path).await, + } + } + + async fn stat(&mut self, fd: Resource) -> FsResult { + match self.virt(&fd) { + Some(Virt::File(f)) => Ok(stat(DescriptorType::RegularFile, f.data.lock().unwrap().len() as u64)), + Some(Virt::Dir { .. }) => Ok(stat(DescriptorType::Directory, 0)), + None => HostDescriptor::stat(&mut self.inner, fd).await, + } + } + + async fn stat_at(&mut self, fd: Resource, path_flags: types::PathFlags, path: String) -> FsResult { + match self.target(&fd, &path) { + Some((actor, path)) => Ok(node_stat(&self.asker().read(&actor, &path).await?)), + None => 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<()> { + if self.target(&fd, &path).is_some() { + return Ok(()); + } + 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<()> { + if self.target(&fd, &old_path).is_some() || self.target(&new_descriptor, &new_path).is_some() { + return Err(ErrorCode::Unsupported.into()); + } + 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: OpenFlags, + flags: DescriptorFlags, + ) -> FsResult> { + match self.target(&fd, &path) { + Some((actor, path)) => self.open(actor, path, oflags, flags).await, + None => HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await, + } + } + + fn drop(&mut self, fd: Resource) -> wasmtime::Result<()> { + if self.virt(&fd).is_some() { + if let Virt::File(f) = self.inner.table.delete(Resource::::new_own(fd.rep()))? + && f.dirty.load(Ordering::SeqCst) + { + self.write_back_later(&f); + } + return Ok(()); + } + self.state.roots.remove(&fd.rep()); + HostDescriptor::drop(&mut self.inner, fd) + } + + async fn readlink_at(&mut self, fd: Resource, path: String) -> FsResult { + if self.target(&fd, &path).is_some() { + return Err(ErrorCode::Invalid.into()); + } + HostDescriptor::readlink_at(&mut self.inner, fd, path).await + } + + async fn remove_directory_at(&mut self, fd: Resource, path: String) -> FsResult<()> { + match self.target(&fd, &path) { + Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_remove", "path": path})).await, + None => 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<()> { + match (self.target(&fd, &old_path), self.target(&new_fd, &new_path)) { + (None, None) => HostDescriptor::rename_at(&mut self.inner, fd, old_path, new_fd, new_path).await, + (Some((a, from)), Some((b, to))) if a == b => { + self.asker().change(&a, json!({"type": "fs_rename", "path": from, "to": to})).await + } + _ => Err(ErrorCode::Unsupported.into()), + } + } + + async fn symlink_at(&mut self, fd: Resource, src_path: String, dest_path: String) -> FsResult<()> { + if self.target(&fd, &dest_path).is_some() { + return Err(ErrorCode::Unsupported.into()); + } + HostDescriptor::symlink_at(&mut self.inner, fd, src_path, dest_path).await + } + + async fn unlink_file_at(&mut self, fd: Resource, path: String) -> FsResult<()> { + match self.target(&fd, &path) { + Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_remove", "path": path})).await, + None => HostDescriptor::unlink_file_at(&mut self.inner, fd, path).await, + } + } + + fn read_via_stream(&mut self, fd: Resource, offset: types::Filesize) -> FsResult> { + match self.file(&fd) { + Some(f) => { + let f = f?; + let data = f.data.lock().unwrap(); + let start = usize::try_from(offset).unwrap_or(usize::MAX).min(data.len()); + let stream: DynInputStream = Box::new(MemoryInputPipe::new(data[start..].to_vec())); + std::mem::drop(data); + Ok(self.inner.table.push(stream)?) + } + None => HostDescriptor::read_via_stream(&mut self.inner, fd, offset), + } + } + + fn write_via_stream(&mut self, fd: Resource, offset: types::Filesize) -> FsResult> { + match self.file(&fd) { + Some(f) => { + let f = f?; + if !f.write { + return Err(ErrorCode::NotPermitted.into()); + } + let stream: DynOutputStream = Box::new(VWriter { file: f, pos: Some(offset as usize) }); + Ok(self.inner.table.push(stream)?) + } + None => HostDescriptor::write_via_stream(&mut self.inner, fd, offset), + } + } + + fn append_via_stream(&mut self, fd: Resource) -> FsResult> { + match self.file(&fd) { + Some(f) => { + let f = f?; + if !f.write { + return Err(ErrorCode::NotPermitted.into()); + } + let stream: DynOutputStream = Box::new(VWriter { file: f, pos: None }); + Ok(self.inner.table.push(stream)?) + } + None => HostDescriptor::append_via_stream(&mut self.inner, fd), + } + } + + async fn is_same_object(&mut self, a: Resource, b: Resource) -> wasmtime::Result { + let key = |v: Virt| match v { + Virt::Dir { actor, path } => (actor, path), + Virt::File(f) => (f.actor, f.path), + }; + match (self.virt(&a), self.virt(&b)) { + (None, None) => HostDescriptor::is_same_object(&mut self.inner, a, b).await, + (Some(x), Some(y)) => Ok(key(x) == key(y)), + _ => Ok(false), + } + } + + async fn metadata_hash(&mut self, fd: Resource) -> FsResult { + match self.virt(&fd) { + Some(Virt::Dir { actor, path }) => Ok(hash(&actor, &path)), + Some(Virt::File(f)) => Ok(hash(&f.actor, &f.path)), + None => HostDescriptor::metadata_hash(&mut self.inner, fd).await, + } + } + + async fn metadata_hash_at(&mut self, fd: Resource, path_flags: types::PathFlags, path: String) -> FsResult { + match self.target(&fd, &path) { + Some((actor, path)) => Ok(hash(&actor, &path)), + None => HostDescriptor::metadata_hash_at(&mut self.inner, fd, path_flags, path).await, + } + } +} + +impl HostDirectoryEntryStream for VfsView<'_> { + async fn read_directory_entry(&mut self, stream: Resource) -> FsResult> { + if let Some(dir) = self.inner.table.get_any_mut(stream.rep()).ok().and_then(|a| a.downcast_mut::()) { + return Ok(dir.0.next()); + } + HostDirectoryEntryStream::read_directory_entry(&mut self.inner, stream).await + } + + fn drop(&mut self, stream: Resource) -> wasmtime::Result<()> { + let ours = self.inner.table.get_any_mut(stream.rep()).ok().is_some_and(|a| a.is::()); + if ours { + self.inner.table.delete(Resource::::new_own(stream.rep()))?; + return Ok(()); + } + HostDirectoryEntryStream::drop(&mut self.inner, stream) + } +} + +const B64: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + +fn base64_encode(bytes: &[u8]) -> String { + let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4); + for chunk in bytes.chunks(3) { + let n = chunk.iter().enumerate().fold(0u32, |n, (i, b)| n | (*b as u32) << (16 - 8 * i)); + for i in 0..4 { + out.push(if i <= chunk.len() { B64[(n >> (18 - 6 * i) & 63) as usize] as char } else { '=' }); + } + } + out +} + +fn base64_decode(text: &str) -> Option> { + let (mut out, mut n, mut bits) = (Vec::new(), 0u32, 0); + for c in text.bytes().filter(|c| !c.is_ascii_whitespace() && *c != b'=') { + n = (n << 6 | B64.iter().position(|b| *b == c)? as u32) & 0xFF_FFFF; + bits += 6; + if bits >= 8 { + bits -= 8; + out.push((n >> bits) as u8); + } + } + Some(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn base64_round_trip() { + for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] { + let enc = base64_encode(s.as_bytes()); + assert_eq!(base64_decode(&enc).unwrap(), s.as_bytes()); + } + assert_eq!(base64_encode(b"hello"), "aGVsbG8="); + } + + #[test] + fn join_resolves_dots() { + assert_eq!(join("/a/b", "../c/./d"), "/a/c/d"); + assert_eq!(join("/", "x/fs"), "/x/fs"); + assert_eq!(join("/", ".."), "/"); + } +} diff --git a/cli/src/world.rs b/cli/src/world.rs index d6ef120..e7d78af 100644 --- a/cli/src/world.rs +++ b/cli/src/world.rs @@ -22,8 +22,9 @@ use serde::{Deserialize, Serialize}; /// 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"]; +/// IDs of the actors that the host implements natively. `fs` sends the +/// requests for actor-served files (`vfs.rs`). +pub const NATIVE: &[&str] = &["broker", "console", "cargo", "fs"]; #[derive(Clone)] pub struct World { diff --git a/cli/tests/actors/toolbox/src/main.rs b/cli/tests/actors/toolbox/src/main.rs index 08dc4c1..6672db7 100644 --- a/cli/tests/actors/toolbox/src/main.rs +++ b/cli/tests/actors/toolbox/src/main.rs @@ -39,6 +39,8 @@ fn main() { } let mut open = HashMap::::new(); + // Files that this actor serves at /actor//fs/ (path -> text). + let mut served = std::collections::BTreeMap::::new(); let stdin = io::stdin(); let mut out = io::stdout().lock(); let mut emit_shutdown = false; @@ -73,6 +75,40 @@ fn main() { break; } Some("echo") => json!({"type": "echo", "value": body["value"]}), + Some("fs_read") => { + let dir = if path == "/" { "/".to_string() } else { format!("{path}/") }; + let mut names: Vec = served + .keys() + .filter_map(|k| k.strip_prefix(&dir)) + .map(|rest| match rest.split_once('/') { + Some((first, _)) => format!("{first}/"), + None => rest.to_string(), + }) + .collect(); + names.dedup(); + if let Some(text) = served.get(&path) { + json!({"type": "fs_file", "text": text}) + } else if path == "/" || !names.is_empty() { + let entries: Vec = names + .iter() + .map(|n| match n.strip_suffix('/') { + Some(d) => json!({"name": d, "type": "dir"}), + None => json!({"name": n, "type": "file"}), + }) + .collect(); + json!({"type": "fs_dir", "entries": entries}) + } else { + json!({"type": "fs_error", "errno": "no-entry"}) + } + } + Some("fs_write") => { + served.insert(path.clone(), text.to_string()); + json!({"type": "fs_done"}) + } + Some("fs_remove") => match served.remove(&path) { + Some(_) => json!({"type": "fs_done"}), + None => json!({"type": "fs_error", "errno": "no-entry"}), + }, Some("publish") => { send( &mut out, diff --git a/cli/tests/host.rs b/cli/tests/host.rs index c417034..81ee5e6 100644 --- a/cli/tests/host.rs +++ b/cli/tests/host.rs @@ -216,6 +216,36 @@ async fn actors_can_read_but_not_write_other_actor_data() { host.stop().await; } +#[tokio::test] +async fn actors_serve_files_under_fs() { + // "writer" serves its fs/ from memory (toolbox fs_read / fs_write / fs_remove). + let (_, host, mut mailbox) = filesystem_world("actor-fs").await; + let file = "/actor/writer/fs/notes/a.txt"; + let write = json!({"type": "write", "path": file, "text": "hello"}); + ask_type(&host, &mut mailbox, "reader", write, "ok").await; + // The write reached the owner as one fs_write message. + let direct = json!({"type": "fs_read", "path": "/notes/a.txt"}); + let served = ask_type(&host, &mut mailbox, "writer", direct, "fs_file").await; + assert_eq!(served.body["text"], "hello"); + + let read = json!({"type": "read", "path": file}); + assert_eq!(ask_type(&host, &mut mailbox, "reader", read, "content").await.body["text"], "hello"); + let list = json!({"type": "list", "path": "/actor/writer/fs/notes"}); + assert_eq!(ask_type(&host, &mut mailbox, "reader", list, "entries").await.body["names"], json!(["a.txt"])); + let root = json!({"type": "list", "path": "/actor/writer/fs"}); + assert_eq!(ask_type(&host, &mut mailbox, "reader", root, "entries").await.body["names"], json!(["notes"])); + + // fs_error and undeliverable become errors of the file operation. + for path in ["/actor/writer/fs/missing.txt", "/actor/nobody/fs/x"] { + ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": path}), "error").await; + } + // An actor that does not answer (here: it is busy reading itself) fails after the timeout. + let id = host.broker.send("console", Some("reader"), None, None, json!({"type": "read", "path": "/actor/reader/fs/x"})); + let timed_out = receive_reply(&mut mailbox, &id, Duration::from_secs(20)).await; + assert_eq!(timed_out.body["type"], "error"); + host.stop().await; +} + #[tokio::test] async fn files_are_plain_posix_files() { let (_, host, mut mailbox) = filesystem_world("posix").await;