Files
sausage/cli/src/vfs.rs
T
Sausage Agent 2ff30ecf70
ci / check (push) Successful in 10s
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.
2026-10-16 01:25:11 +00:00

717 lines
27 KiB
Rust

//! 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("/", ".."), "/");
}
}