Actor-served files: /actor/<id>/fs/

File operations under /actor/<id>/fs/ become messages from `fs` to <id>
(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

@@ -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/<id>/public/`, not inside a
 message.
 
+### Actor-served files: `/actor/<id>/fs/`
+
+Files under `/actor/<id>/fs/` are not on disk: the actor `<id>` serves them.
+Every actor can read and write them. The host turns each file operation into a
+message from `fs` to `<id>` 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

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<HashMap<String, oneshot::Sender<Value>>>,
+    me: std::sync::Weak<Broker>,
 }
 
 impl Broker {
     pub fn new(world: World, runtime: Arc<Runtime>) -> Result<Arc<Self>> {
         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<Self>, to: &str, body: Value) -> impl Future<Output = Option<Value>> + 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(),
         }
     }
 

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;

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/<id>/fs/` (see `vfs.rs`).
+    pub broker: Weak<Broker>,
 }
 
 /// A running guest. Dropping it kills the guest.
@@ -224,6 +238,11 @@ 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 `wasi:filesystem` with the wrapper that serves `/actor/<id>/fs/`.
+        linker.allow_shadowing(true);
+        filesystem::types::add_to_linker::<ActorState, Vfs>(&mut linker, vfs_view)?;
+        filesystem::preopens::add_to_linker::<ActorState, Vfs>(&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);

cli/src/vfs.rs

@@ -0,0 +1,716 @@
+//! Actor-served files: `/actor/<id>/fs/...` (see PROTOCOL.md, "Actor-served
+//! files").
+//!
+//! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`.
+//! A path under `/actor/<id>/fs/` does not reach the disk: the host asks the
+//! actor `<id>` 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<Broker>,
+    /// The actor of this instance, for notes in its `stderr.log`.
+    me: String,
+    /// Descriptors of the `/actor` preopen.
+    roots: HashSet<u32>,
+}
+
+impl VfsState {
+    pub fn new(broker: Weak<Broker>, 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<Mutex<Vec<u8>>>,
+    dirty: Arc<AtomicBool>,
+    write: bool,
+}
+
+/// A directory-entry-stream of an actor-served folder.
+struct VDir(std::vec::IntoIter<DirectoryEntry>);
+
+enum Node {
+    File(Vec<u8>),
+    Dir(Vec<DirectoryEntry>),
+}
+
+/// Sends the requests. Owned, so that no borrow of the view lives across an await.
+struct Asker {
+    broker: Option<Arc<Broker>>,
+    me: String,
+}
+
+impl Asker {
+    async fn ask(&self, actor: &str, body: Value) -> Result<Value, ErrorCode> {
+        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<Node, ErrorCode> {
+        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<Vec<u8>, 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<u8>, 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<usize>,
+}
+
+#[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<usize> {
+        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<types::Descriptor>) -> Option<Virt> {
+        self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned()
+    }
+
+    fn file(&mut self, fd: &Resource<types::Descriptor>) -> Option<FsResult<VFile>> {
+        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<types::Descriptor>, 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<Resource<types::Descriptor>> {
+        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<Resource<types::Descriptor>> {
+        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<Vec<(Resource<types::Descriptor>, 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<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 VfsView<'_> {
+    async fn advise(&mut self, fd: Resource<types::Descriptor>, 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<types::Descriptor>) -> 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<types::Descriptor>) -> FsResult<DescriptorFlags> {
+        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<types::Descriptor>) -> FsResult<DescriptorType> {
+        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<types::Descriptor>, 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<types::Descriptor>, 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<types::Descriptor>, len: types::Filesize, offset: types::Filesize) -> FsResult<(Vec<u8>, 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<types::Descriptor>, buf: Vec<u8>, offset: types::Filesize) -> FsResult<types::Filesize> {
+        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<types::Descriptor>) -> FsResult<Resource<types::DirectoryEntryStream>> {
+        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<types::Descriptor>) -> 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<types::Descriptor>, 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<types::Descriptor>) -> FsResult<types::DescriptorStat> {
+        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<types::Descriptor>, path_flags: types::PathFlags, path: String) -> FsResult<types::DescriptorStat> {
+        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<types::Descriptor>,
+        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<types::Descriptor>,
+        old_path_flags: types::PathFlags,
+        old_path: String,
+        new_descriptor: Resource<types::Descriptor>,
+        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<types::Descriptor>,
+        path_flags: types::PathFlags,
+        path: String,
+        oflags: OpenFlags,
+        flags: DescriptorFlags,
+    ) -> FsResult<Resource<types::Descriptor>> {
+        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<types::Descriptor>) -> wasmtime::Result<()> {
+        if self.virt(&fd).is_some() {
+            if let Virt::File(f) = self.inner.table.delete(Resource::<Virt>::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<types::Descriptor>, path: String) -> FsResult<String> {
+        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<types::Descriptor>, 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<types::Descriptor>, old_path: String, new_fd: Resource<types::Descriptor>, 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<types::Descriptor>, 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<types::Descriptor>, 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<types::Descriptor>, offset: types::Filesize) -> FsResult<Resource<DynInputStream>> {
+        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<types::Descriptor>, offset: types::Filesize) -> FsResult<Resource<DynOutputStream>> {
+        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<types::Descriptor>) -> FsResult<Resource<DynOutputStream>> {
+        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<types::Descriptor>, b: Resource<types::Descriptor>) -> wasmtime::Result<bool> {
+        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<types::Descriptor>) -> FsResult<types::MetadataHashValue> {
+        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<types::Descriptor>, path_flags: types::PathFlags, path: String) -> FsResult<types::MetadataHashValue> {
+        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<types::DirectoryEntryStream>) -> FsResult<Option<DirectoryEntry>> {
+        if let Some(dir) = self.inner.table.get_any_mut(stream.rep()).ok().and_then(|a| a.downcast_mut::<VDir>()) {
+            return Ok(dir.0.next());
+        }
+        HostDirectoryEntryStream::read_directory_entry(&mut self.inner, stream).await
+    }
+
+    fn drop(&mut self, stream: Resource<types::DirectoryEntryStream>) -> wasmtime::Result<()> {
+        let ours = self.inner.table.get_any_mut(stream.rep()).ok().is_some_and(|a| a.is::<VDir>());
+        if ours {
+            self.inner.table.delete(Resource::<VDir>::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<Vec<u8>> {
+    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("/", ".."), "/");
+    }
+}

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 {

cli/tests/actors/toolbox/src/main.rs

@@ -39,6 +39,8 @@ fn main() {
     }
 
     let mut open = HashMap::<String, std::fs::File>::new();
+    // Files that this actor serves at /actor/<id>/fs/ (path -> text).
+    let mut served = std::collections::BTreeMap::<String, String>::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<String> = 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<Value> = 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,

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;