submilli_engine/runtime/
filesystem.rs

1//! Filesystem services supplied by the embedding runtime.
2//!
3//! Capability checks and fuel admission belong to the host caller. Embedders
4//! supply an effect ID and optional work budget with each call.
5use super::durable::{EffectId, RunId};
6use super::fs::{ContainError, ContentPath, FileIdentity};
7use super::fuel::WorkBudget;
8use super::vfs::{Access, Placement, VfsMode};
9use super::{OpenFileGuard, QuotaCharge, QuotaExceeded, Vfs};
10use std::fs::File;
11use std::io::{BufWriter, Read, Seek, SeekFrom};
12use std::sync::Arc;
13
14#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
15pub enum ReaderOptions {
16    Bytes { chunk_size: usize },
17    Lines { maximum_bytes: u64 },
18}
19#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
20pub struct ReadPosition {
21    pub offset: u64,
22    pub bom_handled: bool,
23    pub closed: bool,
24}
25#[derive(serde::Serialize, serde::Deserialize)]
26pub enum ReadItem {
27    Bytes(#[serde(with = "super::effects::binary")] Vec<u8>),
28    Line(String),
29}
30#[derive(serde::Serialize, serde::Deserialize)]
31pub struct ReadOutput {
32    pub item: Option<ReadItem>,
33    #[serde(skip)]
34    pub(crate) _charge: Option<super::limits::HostBytes>,
35}
36impl ReadOutput {
37    /// Keep a pre-admitted allocation charged until output marshalling finishes.
38    pub fn new(item: Option<ReadItem>, reservation: super::limits::HostBytes) -> Self {
39        Self {
40            item,
41            _charge: Some(reservation),
42        }
43    }
44}
45
46#[derive(Debug, thiserror::Error)]
47pub enum FilesystemError {
48    #[error(transparent)]
49    Containment(#[from] ContainError),
50    #[error(transparent)]
51    Io(#[from] std::io::Error),
52    #[error(transparent)]
53    Quota(#[from] QuotaExceeded),
54    #[error("unsupported filesystem operation: {0}")]
55    Unsupported(&'static str),
56    #[error("{0}")]
57    Execution(wasmtime::Error),
58    #[error("{0}")]
59    Rejected(String),
60    #[error(transparent)]
61    Exhausted(#[from] super::fuel::Exhausted),
62    #[error("filesystem backend failed: {0}")]
63    Backend(Box<dyn std::error::Error + Send + Sync>),
64}
65
66#[derive(Clone)]
67pub struct MountInfo {
68    pub path: String,
69    pub volume: String,
70    pub access: Access,
71    pub maximum_bytes: Option<u64>,
72}
73#[derive(Clone)]
74pub struct FilesystemInfo {
75    pub cwd: String,
76    pub mode: VfsMode,
77    pub access: Access,
78    pub volume: Option<String>,
79    pub maximum_bytes: Option<u64>,
80    pub mounts: Vec<MountInfo>,
81}
82impl FilesystemInfo {
83    pub fn from_vfs(vfs: &Vfs) -> Self {
84        Self {
85            cwd: vfs.cwd().into(),
86            mode: vfs.mode(),
87            access: vfs.access(),
88            volume: vfs.volume().map(str::to_owned),
89            maximum_bytes: vfs.quota().map(|q| q.limit()),
90            mounts: vfs
91                .mounts()
92                .iter()
93                .map(|m| MountInfo {
94                    path: m.guest_path().into(),
95                    volume: m.volume().into(),
96                    access: m.access(),
97                    maximum_bytes: m.quota().map(|q| q.limit()),
98                })
99                .collect(),
100        }
101    }
102}
103
104/// Resource values contain persistent identity and optional process-local handles.
105/// They never retain a backend or persistence services. Every resource belongs to
106/// a run; process-local handles are omitted from history and absent after decoding.
107#[derive(serde::Serialize, serde::Deserialize)]
108pub struct FileReader {
109    pub run_id: RunId,
110    pub id: EffectId,
111    pub options: ReaderOptions,
112    #[serde(skip)]
113    pub(crate) local: Option<ReaderHandle>,
114}
115impl FileReader {
116    pub fn new(run_id: RunId, id: EffectId, options: ReaderOptions) -> Self {
117        Self {
118            run_id,
119            id,
120            options,
121            local: None,
122        }
123    }
124}
125
126#[derive(serde::Serialize, serde::Deserialize)]
127pub struct FileWriter {
128    pub run_id: RunId,
129    pub id: EffectId,
130    pub closed: bool,
131    #[serde(skip)]
132    pub(crate) local: Option<WriterHandle>,
133}
134impl FileWriter {
135    pub fn new(run_id: RunId, id: EffectId) -> Self {
136        Self {
137            run_id,
138            id,
139            closed: false,
140            local: None,
141        }
142    }
143}
144
145#[derive(serde::Serialize, serde::Deserialize)]
146pub struct DirectoryReader {
147    pub run_id: RunId,
148    pub id: EffectId,
149    #[serde(skip)]
150    pub(crate) local: Option<DirectoryHandle>,
151}
152impl DirectoryReader {
153    pub fn new(run_id: RunId, id: EffectId) -> Self {
154        Self {
155            run_id,
156            id,
157            local: None,
158        }
159    }
160}
161
162/// An open reader's process-local file, cursor and disk-quota hold.
163pub(crate) struct ReaderHandle {
164    pub(crate) file: Option<File>,
165    pub(crate) position: ReadPosition,
166    pub(crate) held: Option<OpenFileGuard>,
167}
168
169/// A directory iterator is process-local and closes when dropped.
170pub(crate) struct DirectoryHandle {
171    pub(crate) entries: Option<Box<dyn Iterator<Item = DirectoryEntry> + Send + Sync>>,
172}
173
174/// Buffered writer to a temp sibling file. Explicit [`close`](Self::close) fsyncs and
175/// atomically renames it onto `final_path`; `Drop` (the safety net for an un-`close`d
176/// writer) discards the temp file without renaming, removing it only while its name
177/// still refers to the writer's file.
178pub(crate) struct WriterHandle {
179    /// `None` once the writer is closed.
180    pub(crate) open: Option<OpenWrite>,
181    /// Both sides stay contained paths for the whole lifetime of the writer. The commit
182    /// happens through the same handle the open resolved against, so a link swapped over
183    /// the parent between `writer()` and `close()` cannot redirect it — and the drop path
184    /// cannot unlink something under the server's working directory.
185    pub(crate) temp: TempFile,
186    pub(crate) final_path: ContentPath,
187}
188
189/// The temp file a writer streams into. The program can reach it by name while the
190/// writer is open, so before committing or discarding it, the writer checks that the
191/// name still refers to the file it created.
192pub(crate) struct TempFile {
193    pub(crate) path: ContentPath,
194    pub(crate) identity: FileIdentity,
195}
196
197/// An open writer's buffer and its claim on the VFS's size limit. Program code runs
198/// between writes, so the claim replaces nothing up front: every byte written counts
199/// until `close` covers the file it replaces.
200pub(crate) struct OpenWrite {
201    pub(crate) writer: BufWriter<File>,
202    pub(crate) disk_charge: QuotaCharge,
203    /// Registers the temp file as the writer's own, so a removal of it by name
204    /// leaves its bytes for the writer to settle.
205    pub(crate) held: Option<OpenFileGuard>,
206}
207
208#[derive(Clone, serde::Serialize)]
209pub struct DirectoryEntry {
210    pub name: String,
211    pub path: String,
212    pub kind: &'static str,
213    pub bytes: u64,
214}
215impl<'de> serde::Deserialize<'de> for DirectoryEntry {
216    fn deserialize<D: serde::Deserializer<'de>>(decoder: D) -> Result<Self, D::Error> {
217        #[derive(serde::Deserialize)]
218        struct Entry {
219            name: String,
220            path: String,
221            kind: String,
222            bytes: u64,
223        }
224        let entry = Entry::deserialize(decoder)?;
225        let kind = match entry.kind.as_str() {
226            "file" => "file",
227            "directory" => "directory",
228            "symlink" => "symlink",
229            "other" => "other",
230            _ => return Err(serde::de::Error::custom("invalid filesystem entry kind")),
231        };
232        Ok(Self {
233            name: entry.name,
234            path: entry.path,
235            kind,
236            bytes: entry.bytes,
237        })
238    }
239}
240#[async_trait::async_trait]
241pub trait Filesystem: Send + Sync {
242    /// Whether resources can be used after decoding without their local handles.
243    fn supports_resource_replay(&self) -> bool {
244        false
245    }
246
247    fn info(&self) -> FilesystemInfo;
248    fn cwd(&self) -> &str;
249    /// Additional fuel to admit before resolving a guest path.
250    fn path_cost(&self, _path: &str) -> u64 {
251        0
252    }
253    fn placement(&self, path: &str) -> Result<Placement, FilesystemError>;
254    async fn metadata(
255        &self,
256        effect_id: &EffectId,
257        work: Option<Arc<WorkBudget>>,
258        path: &str,
259        follow: bool,
260    ) -> Result<cap_std::fs::Metadata, FilesystemError>;
261    async fn read(
262        &self,
263        effect_id: &EffectId,
264        work: Option<Arc<WorkBudget>>,
265        path: &str,
266        offset: u64,
267        maximum_bytes: u64,
268    ) -> Result<Vec<u8>, FilesystemError>;
269    /// Atomically replace the file contents, creating the file if it is missing.
270    async fn write(
271        &self,
272        effect_id: &EffectId,
273        work: Option<Arc<WorkBudget>>,
274        path: &str,
275        bytes: Vec<u8>,
276    ) -> Result<(), FilesystemError>;
277    /// Append bytes to the file, creating the file if it is missing.
278    async fn append(
279        &self,
280        effect_id: &EffectId,
281        work: Option<Arc<WorkBudget>>,
282        path: &str,
283        bytes: Vec<u8>,
284    ) -> Result<(), FilesystemError>;
285    async fn mkdir(
286        &self,
287        effect_id: &EffectId,
288        work: Option<Arc<WorkBudget>>,
289        path: &str,
290        recursive: bool,
291    ) -> Result<(), FilesystemError>;
292    async fn remove(
293        &self,
294        effect_id: &EffectId,
295        work: Option<Arc<WorkBudget>>,
296        path: &str,
297        recursive: bool,
298        budget: &mut dyn MutationBudget,
299    ) -> Result<(), FilesystemError>;
300    async fn move_entry(
301        &self,
302        effect_id: &EffectId,
303        work: Option<Arc<WorkBudget>>,
304        from: &str,
305        to: &str,
306        budget: &mut dyn MutationBudget,
307    ) -> Result<(), FilesystemError>;
308    async fn copy(
309        &self,
310        effect_id: &EffectId,
311        work: Option<Arc<WorkBudget>>,
312        from: &str,
313        to: &str,
314        recursive: bool,
315        budget: &mut dyn MutationBudget,
316    ) -> Result<(), FilesystemError>;
317    async fn next_reader(
318        &self,
319        effect_id: &EffectId,
320        work: Option<Arc<WorkBudget>>,
321        reader: &mut FileReader,
322        limits: &super::limits::TenantLimits,
323    ) -> Result<ReadOutput, FilesystemError>;
324    async fn close_reader(
325        &self,
326        effect_id: &EffectId,
327        work: Option<Arc<WorkBudget>>,
328        reader: &mut FileReader,
329    ) -> Result<(), FilesystemError>;
330    async fn next_directory(
331        &self,
332        effect_id: &EffectId,
333        work: Option<Arc<WorkBudget>>,
334        directory: &mut DirectoryReader,
335        limits: &super::limits::TenantLimits,
336    ) -> Result<Option<DirectoryEntry>, FilesystemError>;
337    async fn close_directory(
338        &self,
339        effect_id: &EffectId,
340        work: Option<Arc<WorkBudget>>,
341        directory: &mut DirectoryReader,
342        limits: &super::limits::TenantLimits,
343    ) -> Result<(), FilesystemError>;
344    async fn write_chunk(
345        &self,
346        effect_id: &EffectId,
347        work: Option<Arc<WorkBudget>>,
348        writer: &mut FileWriter,
349        bytes: Vec<u8>,
350    ) -> Result<(), FilesystemError>;
351    /// Only explicit close publishes. Dropping a resource never publishes it.
352    async fn close_writer(
353        &self,
354        effect_id: &EffectId,
355        work: Option<Arc<WorkBudget>>,
356        writer: &mut FileWriter,
357    ) -> Result<(), FilesystemError>;
358    async fn open_reader(
359        &self,
360        effect_id: &EffectId,
361        work: Option<Arc<WorkBudget>>,
362        path: &str,
363        options: ReaderOptions,
364    ) -> Result<FileReader, FilesystemError>;
365    async fn open_directory(
366        &self,
367        effect_id: &EffectId,
368        work: Option<Arc<WorkBudget>>,
369        path: &str,
370        recursive: bool,
371        limits: &super::limits::TenantLimits,
372    ) -> Result<DirectoryReader, FilesystemError>;
373    async fn open_writer(
374        &self,
375        effect_id: &EffectId,
376        work: Option<Arc<WorkBudget>>,
377        path: &str,
378    ) -> Result<FileWriter, FilesystemError>;
379}
380
381pub fn bounded_read(
382    mut reader: impl Read + Seek,
383    offset: u64,
384    maximum_bytes: u64,
385) -> Result<Vec<u8>, FilesystemError> {
386    reader.seek(SeekFrom::Start(offset))?;
387    let mut bytes = Vec::new();
388    let mut bounded = reader.take(maximum_bytes);
389    let mut chunk = [0; 8192];
390    loop {
391        let read = bounded.read(&mut chunk)?;
392        if read == 0 {
393            return Ok(bytes);
394        }
395        bytes
396            .try_reserve(read)
397            .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
398        bytes.extend_from_slice(&chunk[..read]);
399    }
400}
401/// Resource accounting for native tree operations. Once a mutation succeeds,
402/// subsequent work must settle fuel rather than refuse. Heap admission stays fallible.
403pub trait MutationBudget: Send {
404    fn begin(&mut self) -> wasmtime::Result<()>;
405    fn reserve(&mut self, bytes: u64) -> wasmtime::Result<()>;
406    fn end(&mut self);
407    fn effect(&mut self);
408    fn syscalls(&mut self, count: u64, settled: bool) -> wasmtime::Result<()>;
409    fn copied(&mut self, entries: u64, bytes: u64) -> wasmtime::Result<()>;
410}
submilli_engine/runtime/filesystem.rs

submilli_engine/runtime/
filesystem.rs
