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
66impl FilesystemError {
67    /// A mutation may already have happened. Preserve terminating limit/trap
68    /// errors, but never let an ordinary refusal become a catchable completion.
69    pub fn after_effect(self) -> Self {
70        match self {
71            Self::Execution(error) if super::host::ends_the_run(&error) => Self::Execution(error),
72            Self::Exhausted(_) | Self::Backend(_) => self,
73            other => Self::Backend(Box::new(other)),
74        }
75    }
76}
77
78#[derive(Clone)]
79pub struct MountInfo {
80    pub path: String,
81    pub volume: String,
82    pub access: Access,
83    pub maximum_bytes: Option<u64>,
84}
85#[derive(Clone)]
86pub struct FilesystemInfo {
87    pub cwd: String,
88    pub mode: VfsMode,
89    pub access: Access,
90    pub volume: Option<String>,
91    pub maximum_bytes: Option<u64>,
92    pub mounts: Vec<MountInfo>,
93}
94
95/// Immutable admission arguments; response bytes and metadata never enter this record.
96#[derive(Clone, serde::Serialize)]
97pub struct DownloadOptions {
98    pub overwrite: bool,
99    pub maximum_bytes: u64,
100    pub request: [u8; 32],
101}
102
103/// Disposable streaming storage owned by the filesystem. Drop cleans up staging.
104/// `publish` consumes ownership; completion belongs to the enclosing effect history.
105#[async_trait::async_trait]
106pub trait Download: std::io::Write + Send {
107    fn bytes_written(&self) -> u64;
108    /// Recover the typed filesystem failure behind an io::Write error.
109    fn take_error(&mut self) -> Option<FilesystemError>;
110    async fn publish(self: Box<Self>, work: Arc<WorkBudget>) -> Result<(), FilesystemError>;
111}
112
113/// Bounded native data stays charged until its consumer finishes with it.
114pub struct FileContents {
115    pub bytes: Vec<u8>,
116    pub _memory: super::limits::HostBytes,
117}
118
119pub struct DirectoryNames {
120    pub names: Vec<String>,
121    pub _memory: super::limits::HostBytes,
122}
123
124/// A native transformation has finished, but nothing has been published yet.
125/// The filesystem does not interpret the caller's encoded result.
126pub struct PreparedFileUpdate {
127    /// None records an unchanged/rejected native result without publishing a file.
128    pub contents: Option<Vec<u8>>,
129    pub output: FileUpdateOutput,
130}
131
132pub struct FileUpdateOutput {
133    pub bytes: Vec<u8>,
134    pub memory: super::limits::HostBytes,
135}
136
137/// A replacement opened under the backend's existing serialization boundary.
138/// The caller prepares fresh contents or authorizes the saved output before
139/// supplying its normal immutable worker budget to `publish`.
140#[async_trait::async_trait]
141pub trait FileUpdate: Send {
142    /// Captured native result and replacement-file byte length, when recovering.
143    /// Authorize this result without rereading or transforming the public file.
144    fn saved_output(&self) -> Option<(&[u8], u64)>;
145    /// Supply freshly prepared contents only when `saved_output` is absent.
146    /// Create `work` after preparation or authorization and settle it after this
147    /// call, including failures. No refusing fuel charge may follow publication.
148    async fn publish(
149        self: Box<Self>,
150        work: Arc<WorkBudget>,
151        prepared: Option<PreparedFileUpdate>,
152        limits: &super::limits::TenantLimits,
153    ) -> Result<FileUpdateOutput, FilesystemError>;
154}
155impl FilesystemInfo {
156    pub fn from_vfs(vfs: &Vfs) -> Self {
157        Self {
158            cwd: vfs.cwd().into(),
159            mode: vfs.mode(),
160            access: vfs.access(),
161            volume: vfs.volume().map(str::to_owned),
162            maximum_bytes: vfs.quota().map(|q| q.limit()),
163            mounts: vfs
164                .mounts()
165                .iter()
166                .map(|m| MountInfo {
167                    path: m.guest_path().into(),
168                    volume: m.volume().into(),
169                    access: m.access(),
170                    maximum_bytes: m.quota().map(|q| q.limit()),
171                })
172                .collect(),
173        }
174    }
175}
176
177/// Resource values contain persistent identity and optional process-local handles.
178/// They never retain a backend or persistence services. Every resource belongs to
179/// a run; process-local handles are omitted from history and absent after decoding.
180#[derive(serde::Serialize, serde::Deserialize)]
181pub struct FileReader {
182    pub run_id: RunId,
183    pub id: EffectId,
184    pub options: ReaderOptions,
185    #[serde(skip)]
186    pub(crate) local: Option<ReaderHandle>,
187}
188impl FileReader {
189    pub fn new(run_id: RunId, id: EffectId, options: ReaderOptions) -> Self {
190        Self {
191            run_id,
192            id,
193            options,
194            local: None,
195        }
196    }
197}
198
199#[derive(serde::Serialize, serde::Deserialize)]
200pub struct FileWriter {
201    pub run_id: RunId,
202    pub id: EffectId,
203    pub closed: bool,
204    #[serde(skip)]
205    pub(crate) local: Option<WriterHandle>,
206}
207impl FileWriter {
208    pub fn new(run_id: RunId, id: EffectId) -> Self {
209        Self {
210            run_id,
211            id,
212            closed: false,
213            local: None,
214        }
215    }
216}
217
218#[derive(serde::Serialize, serde::Deserialize)]
219pub struct DirectoryReader {
220    pub run_id: RunId,
221    pub id: EffectId,
222    #[serde(skip)]
223    pub(crate) local: Option<DirectoryHandle>,
224}
225impl DirectoryReader {
226    pub fn new(run_id: RunId, id: EffectId) -> Self {
227        Self {
228            run_id,
229            id,
230            local: None,
231        }
232    }
233}
234
235/// An open reader's process-local file, cursor and disk-quota hold.
236pub(crate) struct ReaderHandle {
237    pub(crate) file: Option<File>,
238    pub(crate) position: ReadPosition,
239    pub(crate) held: Option<OpenFileGuard>,
240}
241
242/// A directory iterator is process-local and closes when dropped.
243pub(crate) struct DirectoryHandle {
244    pub(crate) entries: Option<Box<dyn Iterator<Item = DirectoryEntry> + Send + Sync>>,
245}
246
247/// Buffered writer to a temp sibling file. Explicit [`close`](Self::close) fsyncs and
248/// atomically renames it onto `final_path`; `Drop` (the safety net for an un-`close`d
249/// writer) discards the temp file without renaming, removing it only while its name
250/// still refers to the writer's file.
251pub(crate) struct WriterHandle {
252    /// `None` once the writer is closed.
253    pub(crate) open: Option<OpenWrite>,
254    /// Both sides stay contained paths for the whole lifetime of the writer. The commit
255    /// happens through the same handle the open resolved against, so a link swapped over
256    /// the parent between `writer()` and `close()` cannot redirect it — and the drop path
257    /// cannot unlink something under the server's working directory.
258    pub(crate) temp: TempFile,
259    pub(crate) final_path: ContentPath,
260}
261
262/// The temp file a writer streams into. The program can reach it by name while the
263/// writer is open, so before committing or discarding it, the writer checks that the
264/// name still refers to the file it created.
265pub(crate) struct TempFile {
266    pub(crate) path: ContentPath,
267    pub(crate) identity: FileIdentity,
268}
269
270/// An open writer's buffer and its claim on the VFS's size limit. Program code runs
271/// between writes, so the claim replaces nothing up front: every byte written counts
272/// until `close` covers the file it replaces.
273pub(crate) struct OpenWrite {
274    pub(crate) writer: BufWriter<File>,
275    pub(crate) disk_charge: QuotaCharge,
276    /// Registers the temp file as the writer's own, so a removal of it by name
277    /// leaves its bytes for the writer to settle.
278    pub(crate) held: Option<OpenFileGuard>,
279}
280
281#[derive(Clone, serde::Serialize)]
282pub struct DirectoryEntry {
283    pub name: String,
284    pub path: String,
285    pub kind: &'static str,
286    pub bytes: u64,
287}
288impl<'de> serde::Deserialize<'de> for DirectoryEntry {
289    fn deserialize<D: serde::Deserializer<'de>>(decoder: D) -> Result<Self, D::Error> {
290        #[derive(serde::Deserialize)]
291        struct Entry {
292            name: String,
293            path: String,
294            kind: String,
295            bytes: u64,
296        }
297        let entry = Entry::deserialize(decoder)?;
298        let kind = match entry.kind.as_str() {
299            "file" => "file",
300            "directory" => "directory",
301            "symlink" => "symlink",
302            "other" => "other",
303            _ => return Err(serde::de::Error::custom("invalid filesystem entry kind")),
304        };
305        Ok(Self {
306            name: entry.name,
307            path: entry.path,
308            kind,
309            bytes: entry.bytes,
310        })
311    }
312}
313#[async_trait::async_trait]
314pub trait Filesystem: Send + Sync {
315    /// Download admission survives retry with the same effect identity.
316    fn supports_download_replay(&self) -> bool {
317        false
318    }
319
320    /// File updates retain prepared bytes/results and completed receipts across retries.
321    fn supports_update_replay(&self) -> bool {
322        false
323    }
324    /// Whether resources can be used after decoding without their local handles.
325    fn supports_resource_replay(&self) -> bool {
326        false
327    }
328
329    fn info(&self) -> FilesystemInfo;
330    fn cwd(&self) -> &str;
331    /// Additional fuel to admit before resolving a guest path.
332    fn path_cost(&self, _path: &str) -> u64 {
333        0
334    }
335    /// Resolving placement may follow filesystem links and must not block the host.
336    async fn placement(&self, path: &str) -> Result<Placement, FilesystemError>;
337    /// Whole regular-file observation, with nonblocking special-file refusal.
338    async fn read_regular(
339        &self,
340        work: Option<Arc<WorkBudget>>,
341        path: &str,
342        maximum_bytes: u64,
343        limits: &super::limits::TenantLimits,
344    ) -> Result<FileContents, FilesystemError>;
345    /// Names only: the caller checks each child's authority before its metadata.
346    /// This transient observation is not a guest-visible persistent iterator.
347    async fn directory_names(
348        &self,
349        work: Option<Arc<WorkBudget>>,
350        path: &str,
351        maximum_entries: usize,
352        limits: &super::limits::TenantLimits,
353    ) -> Result<DirectoryNames, FilesystemError>;
354    /// Admit the destination and return fresh disposable staging, retaining native
355    /// serialization until publication or drop. Completion belongs to effect history.
356    async fn begin_download(
357        &self,
358        effect: &EffectId,
359        path: &str,
360        options: DownloadOptions,
361        work: Arc<WorkBudget>,
362    ) -> Result<Box<dyn Download>, FilesystemError>;
363    /// Open a replacement and return any captured result before native preparation.
364    /// The returned operation retains the backend's serialization until publication
365    /// finishes or the caller drops it without publishing.
366    async fn begin_file_update(
367        &self,
368        effect_id: &EffectId,
369        path: &str,
370        request: [u8; 32],
371        limits: &super::limits::TenantLimits,
372    ) -> Result<Box<dyn FileUpdate>, FilesystemError>;
373    async fn metadata(
374        &self,
375        effect_id: &EffectId,
376        work: Option<Arc<WorkBudget>>,
377        path: &str,
378        follow: bool,
379    ) -> Result<cap_std::fs::Metadata, FilesystemError>;
380    async fn read(
381        &self,
382        effect_id: &EffectId,
383        work: Option<Arc<WorkBudget>>,
384        path: &str,
385        offset: u64,
386        maximum_bytes: u64,
387    ) -> Result<Vec<u8>, FilesystemError>;
388    /// Atomically replace the file contents, creating the file if it is missing.
389    async fn write(
390        &self,
391        effect_id: &EffectId,
392        work: Option<Arc<WorkBudget>>,
393        path: &str,
394        bytes: Vec<u8>,
395    ) -> Result<(), FilesystemError>;
396    /// Append bytes to the file, creating the file if it is missing.
397    async fn append(
398        &self,
399        effect_id: &EffectId,
400        work: Option<Arc<WorkBudget>>,
401        path: &str,
402        bytes: Vec<u8>,
403    ) -> Result<(), FilesystemError>;
404    async fn mkdir(
405        &self,
406        effect_id: &EffectId,
407        work: Option<Arc<WorkBudget>>,
408        path: &str,
409        recursive: bool,
410    ) -> Result<(), FilesystemError>;
411    async fn remove(
412        &self,
413        effect_id: &EffectId,
414        work: Option<Arc<WorkBudget>>,
415        path: &str,
416        recursive: bool,
417        budget: &mut dyn MutationBudget,
418    ) -> Result<(), FilesystemError>;
419    async fn move_entry(
420        &self,
421        effect_id: &EffectId,
422        work: Option<Arc<WorkBudget>>,
423        from: &str,
424        to: &str,
425        budget: &mut dyn MutationBudget,
426    ) -> Result<(), FilesystemError>;
427    async fn copy(
428        &self,
429        effect_id: &EffectId,
430        work: Option<Arc<WorkBudget>>,
431        from: &str,
432        to: &str,
433        recursive: bool,
434        budget: &mut dyn MutationBudget,
435    ) -> Result<(), FilesystemError>;
436    async fn next_reader(
437        &self,
438        effect_id: &EffectId,
439        work: Option<Arc<WorkBudget>>,
440        reader: &mut FileReader,
441        limits: &super::limits::TenantLimits,
442    ) -> Result<ReadOutput, FilesystemError>;
443    async fn close_reader(
444        &self,
445        effect_id: &EffectId,
446        work: Option<Arc<WorkBudget>>,
447        reader: &mut FileReader,
448    ) -> Result<(), FilesystemError>;
449    async fn next_directory(
450        &self,
451        effect_id: &EffectId,
452        work: Option<Arc<WorkBudget>>,
453        directory: &mut DirectoryReader,
454        limits: &super::limits::TenantLimits,
455    ) -> Result<Option<DirectoryEntry>, FilesystemError>;
456    async fn close_directory(
457        &self,
458        effect_id: &EffectId,
459        work: Option<Arc<WorkBudget>>,
460        directory: &mut DirectoryReader,
461        limits: &super::limits::TenantLimits,
462    ) -> Result<(), FilesystemError>;
463    async fn write_chunk(
464        &self,
465        effect_id: &EffectId,
466        work: Option<Arc<WorkBudget>>,
467        writer: &mut FileWriter,
468        bytes: Vec<u8>,
469    ) -> Result<(), FilesystemError>;
470    /// Only explicit close publishes. Dropping a resource never publishes it.
471    async fn close_writer(
472        &self,
473        effect_id: &EffectId,
474        work: Option<Arc<WorkBudget>>,
475        writer: &mut FileWriter,
476    ) -> Result<(), FilesystemError>;
477    async fn open_reader(
478        &self,
479        effect_id: &EffectId,
480        work: Option<Arc<WorkBudget>>,
481        path: &str,
482        options: ReaderOptions,
483    ) -> Result<FileReader, FilesystemError>;
484    async fn open_directory(
485        &self,
486        effect_id: &EffectId,
487        work: Option<Arc<WorkBudget>>,
488        path: &str,
489        recursive: bool,
490        limits: &super::limits::TenantLimits,
491    ) -> Result<DirectoryReader, FilesystemError>;
492    async fn open_writer(
493        &self,
494        effect_id: &EffectId,
495        work: Option<Arc<WorkBudget>>,
496        path: &str,
497    ) -> Result<FileWriter, FilesystemError>;
498}
499
500pub fn bounded_read(
501    mut reader: impl Read + Seek,
502    offset: u64,
503    maximum_bytes: u64,
504) -> Result<Vec<u8>, FilesystemError> {
505    reader.seek(SeekFrom::Start(offset))?;
506    let mut bytes = Vec::new();
507    let mut bounded = reader.take(maximum_bytes);
508    let mut chunk = [0; 8192];
509    loop {
510        let read = bounded.read(&mut chunk)?;
511        if read == 0 {
512            return Ok(bytes);
513        }
514        bytes
515            .try_reserve(read)
516            .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
517        bytes.extend_from_slice(&chunk[..read]);
518    }
519}
520/// Resource accounting for native tree operations. Once a mutation succeeds,
521/// subsequent work must settle fuel rather than refuse. Heap admission stays fallible.
522pub trait MutationBudget: Send {
523    fn begin(&mut self) -> wasmtime::Result<()>;
524    fn reserve(&mut self, bytes: u64) -> wasmtime::Result<()>;
525    fn end(&mut self);
526    fn effect(&mut self);
527    fn syscalls(&mut self, count: u64, settled: bool) -> wasmtime::Result<()>;
528    fn copied(&mut self, entries: u64, bytes: u64) -> wasmtime::Result<()>;
529}
submilli_engine/runtime/filesystem.rs

submilli_engine/runtime/
filesystem.rs
