Compare commits
2
Commits
plugins
...
b4274f9fc5
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b4274f9fc5 | ||
|
|
842fe62c87 |
@@ -85,6 +85,13 @@ 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`
|
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 `/`).
|
is relative to `fs/` and starts with `/` (the folder itself is `/`).
|
||||||
|
|
||||||
|
To make this virtual directory discoverable, an actor that serves it declares
|
||||||
|
`Actor-served files` and `fs_read` in its leading `//!` documentation in
|
||||||
|
`source/src/main.rs`. The host uses that declaration when listing
|
||||||
|
`/actor/<id>/`; it does not probe every actor (which could stall on actors that
|
||||||
|
do not handle filesystem messages). This is discovery metadata only: opening
|
||||||
|
and reading the path still requires a valid `fs_read` reply.
|
||||||
|
|
||||||
| Request | Reply |
|
| 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_read", "path"}` | `{"type": "fs_file", "text"}` (or `"base64"` instead of `"text"`), or `{"type": "fs_dir", "entries": [{"name", "type": "file" \| "dir"}]}` |
|
||||||
|
|||||||
+94
-15
@@ -4,7 +4,8 @@
|
|||||||
//! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`.
|
//! 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
|
//! 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
|
//! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for
|
||||||
//! the reply. Every other path goes to `wasmtime-wasi` unchanged.
|
//! the reply. Other paths go to `wasmtime-wasi` unchanged except that an
|
||||||
|
//! actor directory listing advertises `fs/` when its source docs declare it.
|
||||||
//!
|
//!
|
||||||
//! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}`
|
//! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}`
|
||||||
//! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`,
|
//! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`,
|
||||||
@@ -14,7 +15,7 @@
|
|||||||
//! A file is read once when it is opened. After a change, it is written back
|
//! 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.
|
//! in one `fs_write` when it is synced or closed.
|
||||||
|
|
||||||
use std::collections::HashSet;
|
use std::collections::HashMap;
|
||||||
use std::hash::{Hash, Hasher};
|
use std::hash::{Hash, Hasher};
|
||||||
use std::sync::atomic::{AtomicBool, Ordering};
|
use std::sync::atomic::{AtomicBool, Ordering};
|
||||||
use std::sync::{Arc, Mutex, Weak};
|
use std::sync::{Arc, Mutex, Weak};
|
||||||
@@ -41,13 +42,14 @@ pub struct VfsState {
|
|||||||
broker: Weak<Broker>,
|
broker: Weak<Broker>,
|
||||||
/// The actor of this instance, for notes in its `stderr.log`.
|
/// The actor of this instance, for notes in its `stderr.log`.
|
||||||
me: String,
|
me: String,
|
||||||
/// Descriptors of the `/actor` preopen.
|
/// Native directory descriptors and their absolute paths. Keeping every
|
||||||
roots: HashSet<u32>,
|
/// `/actor` child (not only the preopen) lets `fs/` work after traversal.
|
||||||
|
dirs: HashMap<u32, String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl VfsState {
|
impl VfsState {
|
||||||
pub fn new(broker: Weak<Broker>, me: String) -> Self {
|
pub fn new(broker: Weak<Broker>, me: String) -> Self {
|
||||||
Self { broker, me, roots: HashSet::new() }
|
Self { broker, me, dirs: HashMap::new() }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -199,6 +201,33 @@ fn join(base: &str, rel: &str) -> String {
|
|||||||
format!("/{}", parts.join("/"))
|
format!("/{}", parts.join("/"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Split a global path under an actor's virtual `fs/` into its owner and
|
||||||
|
/// actor-relative path.
|
||||||
|
fn actor_fs_path(full: &str) -> Option<(String, String)> {
|
||||||
|
let rest = full.strip_prefix("/actor/")?;
|
||||||
|
let (actor, rest) = rest.split_once('/')?;
|
||||||
|
let (fs, path) = rest.split_once('/').unwrap_or((rest, ""));
|
||||||
|
(fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{path}")))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn actor_root(path: &str) -> Option<&str> {
|
||||||
|
let actor = path.strip_prefix("/actor/")?;
|
||||||
|
(!actor.is_empty() && !actor.contains('/')).then_some(actor)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// An actor advertises its virtual filesystem in its leading `//!` docs.
|
||||||
|
/// This is discovery metadata; actual reads still go through `fs_read`.
|
||||||
|
fn declares_actor_fs(source: &str) -> bool {
|
||||||
|
let docs = source
|
||||||
|
.lines()
|
||||||
|
.take_while(|line| line.trim().starts_with("//!"))
|
||||||
|
.map(|line| line.trim().trim_start_matches("//!").trim())
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join(" ")
|
||||||
|
.to_ascii_lowercase();
|
||||||
|
docs.contains("actor-served files") && docs.contains("fs_read")
|
||||||
|
}
|
||||||
|
|
||||||
fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat {
|
fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat {
|
||||||
types::DescriptorStat {
|
types::DescriptorStat {
|
||||||
type_,
|
type_,
|
||||||
@@ -266,6 +295,14 @@ impl VfsView<'_> {
|
|||||||
Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() }
|
Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn actor_serves_fs(&self, actor: &str) -> bool {
|
||||||
|
self.state
|
||||||
|
.broker
|
||||||
|
.upgrade()
|
||||||
|
.and_then(|broker| std::fs::read_to_string(broker.world.source_dir(actor).join("src/main.rs")).ok())
|
||||||
|
.is_some_and(|source| declares_actor_fs(&source))
|
||||||
|
}
|
||||||
|
|
||||||
fn virt(&mut self, fd: &Resource<types::Descriptor>) -> Option<Virt> {
|
fn virt(&mut self, fd: &Resource<types::Descriptor>) -> Option<Virt> {
|
||||||
self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned()
|
self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned()
|
||||||
}
|
}
|
||||||
@@ -282,16 +319,26 @@ impl VfsView<'_> {
|
|||||||
match self.virt(fd) {
|
match self.virt(fd) {
|
||||||
Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))),
|
Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))),
|
||||||
Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))),
|
Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))),
|
||||||
None if self.state.roots.contains(&fd.rep()) => {
|
None => {
|
||||||
let full = join("/", path);
|
let base = self.state.dirs.get(&fd.rep())?;
|
||||||
let mut parts = full.splitn(4, '/').skip(1);
|
actor_fs_path(&join(base, path))
|
||||||
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,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn native_directory_entries(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Vec<DirectoryEntry>> {
|
||||||
|
let stream = HostDescriptor::read_directory(&mut self.inner, fd).await?;
|
||||||
|
let mut entries = Vec::new();
|
||||||
|
loop {
|
||||||
|
match HostDirectoryEntryStream::read_directory_entry(&mut self.inner, stream.clone()).await? {
|
||||||
|
Some(entry) => entries.push(entry),
|
||||||
|
None => break,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
HostDirectoryEntryStream::drop(&mut self.inner, stream)?;
|
||||||
|
Ok(entries)
|
||||||
|
}
|
||||||
|
|
||||||
fn push(&mut self, v: Virt) -> FsResult<Resource<types::Descriptor>> {
|
fn push(&mut self, v: Virt) -> FsResult<Resource<types::Descriptor>> {
|
||||||
let r = self.inner.table.push(v)?;
|
let r = self.inner.table.push(v)?;
|
||||||
Ok(Resource::new_own(r.rep()))
|
Ok(Resource::new_own(r.rep()))
|
||||||
@@ -343,7 +390,7 @@ impl preopens::Host for VfsView<'_> {
|
|||||||
let dirs = preopens::Host::get_directories(&mut self.inner)?;
|
let dirs = preopens::Host::get_directories(&mut self.inner)?;
|
||||||
for (fd, name) in &dirs {
|
for (fd, name) in &dirs {
|
||||||
if name == "/actor" {
|
if name == "/actor" {
|
||||||
self.state.roots.insert(fd.rep());
|
self.state.dirs.insert(fd.rep(), "/actor".to_string());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Ok(dirs)
|
Ok(dirs)
|
||||||
@@ -453,7 +500,19 @@ impl HostDescriptor for VfsView<'_> {
|
|||||||
Node::File(_) => Err(ErrorCode::NotDirectory.into()),
|
Node::File(_) => Err(ErrorCode::NotDirectory.into()),
|
||||||
},
|
},
|
||||||
Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()),
|
Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()),
|
||||||
None => HostDescriptor::read_directory(&mut self.inner, fd).await,
|
None => {
|
||||||
|
let actor = self.state.dirs.get(&fd.rep()).and_then(|path| actor_root(path)).map(str::to_string);
|
||||||
|
if actor.as_deref().is_some_and(|id| self.actor_serves_fs(id)) {
|
||||||
|
let mut entries = self.native_directory_entries(fd).await?;
|
||||||
|
if !entries.iter().any(|entry| entry.name == "fs") {
|
||||||
|
entries.push(DirectoryEntry { type_: DescriptorType::Directory, name: "fs".to_string() });
|
||||||
|
}
|
||||||
|
let r = self.inner.table.push(VDir(entries.into_iter()))?;
|
||||||
|
Ok(Resource::new_own(r.rep()))
|
||||||
|
} else {
|
||||||
|
HostDescriptor::read_directory(&mut self.inner, fd).await
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -525,7 +584,16 @@ impl HostDescriptor for VfsView<'_> {
|
|||||||
) -> FsResult<Resource<types::Descriptor>> {
|
) -> FsResult<Resource<types::Descriptor>> {
|
||||||
match self.target(&fd, &path) {
|
match self.target(&fd, &path) {
|
||||||
Some((actor, path)) => self.open(actor, path, oflags, flags).await,
|
Some((actor, path)) => self.open(actor, path, oflags, flags).await,
|
||||||
None => HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await,
|
None => {
|
||||||
|
let child_path = self.state.dirs.get(&fd.rep()).map(|base| join(base, &path));
|
||||||
|
let opened = HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await?;
|
||||||
|
if let Some(path) = child_path
|
||||||
|
&& HostDescriptor::get_type(&mut self.inner, opened.clone()).await? == DescriptorType::Directory
|
||||||
|
{
|
||||||
|
self.state.dirs.insert(opened.rep(), path);
|
||||||
|
}
|
||||||
|
Ok(opened)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -538,7 +606,7 @@ impl HostDescriptor for VfsView<'_> {
|
|||||||
}
|
}
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
self.state.roots.remove(&fd.rep());
|
self.state.dirs.remove(&fd.rep());
|
||||||
HostDescriptor::drop(&mut self.inner, fd)
|
HostDescriptor::drop(&mut self.inner, fd)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -698,6 +766,17 @@ fn base64_decode(text: &str) -> Option<Vec<u8>> {
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn actor_fs_paths_and_discovery_docs() {
|
||||||
|
assert_eq!(actor_fs_path("/actor/brain/fs"), Some(("brain".to_string(), "/".to_string())));
|
||||||
|
assert_eq!(actor_fs_path("/actor/brain/fs/docs/readme.md"), Some(("brain".to_string(), "/docs/readme.md".to_string())));
|
||||||
|
assert_eq!(actor_fs_path("/actor/brain/public/fs/file"), None);
|
||||||
|
assert_eq!(actor_root("/actor/brain"), Some("brain"));
|
||||||
|
assert_eq!(actor_root("/actor/brain/fs"), None);
|
||||||
|
assert!(declares_actor_fs("//! brain\n//! Actor-served files include fs_read.\nfn main() {}"));
|
||||||
|
assert!(!declares_actor_fs("//! actor without fs support\nfn main() {}"));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn base64_round_trip() {
|
fn base64_round_trip() {
|
||||||
for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] {
|
for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] {
|
||||||
|
|||||||
@@ -246,6 +246,71 @@ async fn actors_serve_files_under_fs() {
|
|||||||
host.stop().await;
|
host.stop().await;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn actor_fs_is_discoverable_only_when_its_source_docs_advertise_it() {
|
||||||
|
let world = create_world("actor-fs-discovery");
|
||||||
|
let wasm = toolbox_wasm().clone();
|
||||||
|
let (host, mut mailbox) = start(&world).await;
|
||||||
|
|
||||||
|
let advertised = passing_source(&world, "advertised-fs");
|
||||||
|
std::fs::create_dir_all(advertised.join("src")).unwrap();
|
||||||
|
std::fs::write(
|
||||||
|
advertised.join("src/main.rs"),
|
||||||
|
"//! test actor\n//! Actor-served files: implements fs_read and fs_dir.\n",
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
spawn_type(&host, &mut mailbox, "served", &wasm, &advertised).await;
|
||||||
|
|
||||||
|
let undocumented = passing_source(&world, "undocumented-fs");
|
||||||
|
std::fs::create_dir_all(undocumented.join("src")).unwrap();
|
||||||
|
std::fs::write(undocumented.join("src/main.rs"), "//! test actor without a virtual filesystem\n").unwrap();
|
||||||
|
spawn_type(&host, &mut mailbox, "plain", &wasm, &undocumented).await;
|
||||||
|
|
||||||
|
let reader_source = passing_source(&world, "fs-reader");
|
||||||
|
spawn_type(&host, &mut mailbox, "reader", &wasm, &reader_source).await;
|
||||||
|
|
||||||
|
let served_entries = ask_type(
|
||||||
|
&host,
|
||||||
|
&mut mailbox,
|
||||||
|
"reader",
|
||||||
|
json!({"type": "list", "path": "/actor/served"}),
|
||||||
|
"entries",
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert!(served_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
|
||||||
|
|
||||||
|
let plain_entries = ask_type(
|
||||||
|
&host,
|
||||||
|
&mut mailbox,
|
||||||
|
"reader",
|
||||||
|
json!({"type": "list", "path": "/actor/plain"}),
|
||||||
|
"entries",
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert!(!plain_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
|
||||||
|
|
||||||
|
let path = "/actor/served/fs/readme.txt";
|
||||||
|
ask_type(
|
||||||
|
&host,
|
||||||
|
&mut mailbox,
|
||||||
|
"reader",
|
||||||
|
json!({"type": "write", "path": path, "text": "virtual document"}),
|
||||||
|
"ok",
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let document = ask_type(
|
||||||
|
&host,
|
||||||
|
&mut mailbox,
|
||||||
|
"reader",
|
||||||
|
json!({"type": "read", "path": path}),
|
||||||
|
"content",
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert_eq!(document.body["text"], "virtual document");
|
||||||
|
|
||||||
|
host.stop().await;
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn files_are_plain_posix_files() {
|
async fn files_are_plain_posix_files() {
|
||||||
let (_, host, mut mailbox) = filesystem_world("posix").await;
|
let (_, host, mut mailbox) = filesystem_world("posix").await;
|
||||||
|
|||||||
Reference in New Issue
Block a user