vfs/adapter/
mounted_fs.rs

1use crate::posix::{
2    MountedFileSystem, VfsError as PosixVfsError, VfsResult as PosixVfsResult, VirtualDirEntry,
3    VirtualStat,
4};
5use agentos_runtime::{BlockingJobError, RuntimeContext};
6use std::any::Any;
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::Arc;
9
10static NEXT_ENGINE_DEVICE_ID: AtomicU64 = AtomicU64::new(4096);
11
12pub struct MountedEngineFileSystem<F> {
13    inner: Arc<F>,
14    runtime: RuntimeContext,
15    device_id: u64,
16}
17
18impl<F> MountedEngineFileSystem<F> {
19    pub fn with_runtime_context(inner: F, runtime: RuntimeContext) -> Self {
20        Self {
21            inner: Arc::new(inner),
22            runtime,
23            device_id: NEXT_ENGINE_DEVICE_ID.fetch_add(1, Ordering::Relaxed),
24        }
25    }
26
27    fn run<T>(
28        &self,
29        reserved_bytes: usize,
30        future: impl std::future::Future<Output = crate::engine::VfsResult<T>> + Send + 'static,
31    ) -> PosixVfsResult<T>
32    where
33        T: Send + 'static,
34    {
35        if agentos_runtime::is_runtime_worker_thread() {
36            return Err(PosixVfsError::new(
37                "EDEADLK",
38                "ERR_AGENTOS_VFS_RUNTIME_WORKER_WAIT: synchronous mounted filesystem calls must run outside an AgentOS Tokio worker",
39            ));
40        }
41        let handle = self.runtime.handle().clone();
42        let runtime = self.runtime.clone();
43        let cancel = Arc::new(tokio::sync::Notify::new());
44        let worker_cancel = Arc::clone(&cancel);
45        let result = self.runtime.blocking().run_sync(
46            reserved_bytes,
47            self.runtime.blocking_job_timeout(),
48            move || {
49                handle.block_on(async move {
50                    tokio::select! {
51                        result = future => Some(result),
52                        () = runtime.admission_closed() => None,
53                        () = worker_cancel.notified() => None,
54                    }
55                })
56            },
57        );
58        match result {
59            Ok(Some(result)) => result.map_err(convert_error),
60            Ok(None) => Err(PosixVfsError::new(
61                "ECANCELED",
62                "ERR_AGENTOS_VFS_CANCELLED: mounted filesystem runtime admission closed",
63            )),
64            Err(error) => {
65                // `run_sync` stops waiting at the configured deadline. Wake the
66                // still-admitted worker so the engine future is dropped and the
67                // fixed blocking-executor slot cannot remain stranded.
68                cancel.notify_one();
69                Err(blocking_job_error(error))
70            }
71        }
72    }
73}
74
75impl<F> MountedFileSystem for MountedEngineFileSystem<F>
76where
77    F: crate::engine::VirtualFileSystem + 'static,
78{
79    fn as_any(&self) -> &dyn Any {
80        self
81    }
82
83    fn as_any_mut(&mut self) -> &mut dyn Any {
84        self
85    }
86
87    fn read_file(&mut self, path: &str) -> PosixVfsResult<Vec<u8>> {
88        let inner = Arc::clone(&self.inner);
89        let path = path.to_owned();
90        let reserved_bytes = path.len();
91        self.run(reserved_bytes, async move { inner.read_file(&path).await })
92    }
93
94    fn read_dir(&mut self, path: &str) -> PosixVfsResult<Vec<String>> {
95        let inner = Arc::clone(&self.inner);
96        let path = path.to_owned();
97        let reserved_bytes = path.len();
98        self.run(reserved_bytes, async move { inner.read_dir(&path).await })
99    }
100
101    fn read_dir_with_types(&mut self, path: &str) -> PosixVfsResult<Vec<VirtualDirEntry>> {
102        let inner = Arc::clone(&self.inner);
103        let path = path.to_owned();
104        let reserved_bytes = path.len();
105        self.run(reserved_bytes, async move {
106            inner.read_dir_with_types(&path).await
107        })
108        .map(|entries| {
109            entries
110                .into_iter()
111                .map(|entry| VirtualDirEntry {
112                    name: entry.name,
113                    is_directory: entry.kind == crate::engine::InodeType::Directory,
114                    is_symbolic_link: entry.kind == crate::engine::InodeType::Symlink,
115                })
116                .collect()
117        })
118    }
119
120    fn write_file(&mut self, path: &str, content: Vec<u8>) -> PosixVfsResult<()> {
121        let inner = Arc::clone(&self.inner);
122        let path = path.to_owned();
123        let reserved_bytes = path.len().saturating_add(content.len());
124        self.run(reserved_bytes, async move {
125            inner.write_file(&path, &content).await
126        })
127    }
128
129    fn write_file_with_mode(
130        &mut self,
131        path: &str,
132        content: Vec<u8>,
133        mode: Option<u32>,
134    ) -> PosixVfsResult<()> {
135        self.write_file(path, content)?;
136        if let Some(mode) = mode {
137            self.chmod(path, mode)?;
138        }
139        Ok(())
140    }
141
142    fn create_file_exclusive(&mut self, path: &str, content: Vec<u8>) -> PosixVfsResult<()> {
143        if self.exists(path) {
144            return Err(PosixVfsError::new(
145                "EEXIST",
146                format!("file already exists, open '{path}'"),
147            ));
148        }
149        self.write_file(path, content)
150    }
151
152    fn create_file_exclusive_with_mode(
153        &mut self,
154        path: &str,
155        content: Vec<u8>,
156        mode: Option<u32>,
157    ) -> PosixVfsResult<()> {
158        self.create_file_exclusive(path, content)?;
159        if let Some(mode) = mode {
160            self.chmod(path, mode)?;
161        }
162        Ok(())
163    }
164
165    fn append_file(&mut self, path: &str, content: Vec<u8>) -> PosixVfsResult<u64> {
166        let inner = Arc::clone(&self.inner);
167        let path = path.to_owned();
168        let reserved_bytes = path.len().saturating_add(content.len());
169        self.run(reserved_bytes, async move {
170            inner.append(&path, &content).await
171        })
172    }
173
174    fn create_dir(&mut self, path: &str) -> PosixVfsResult<()> {
175        let inner = Arc::clone(&self.inner);
176        let path = path.to_owned();
177        let reserved_bytes = path.len();
178        self.run(reserved_bytes, async move { inner.create_dir(&path).await })
179    }
180
181    fn create_dir_with_mode(&mut self, path: &str, mode: Option<u32>) -> PosixVfsResult<()> {
182        self.create_dir(path)?;
183        if let Some(mode) = mode {
184            self.chmod(path, mode)?;
185        }
186        Ok(())
187    }
188
189    fn mkdir(&mut self, path: &str, recursive: bool) -> PosixVfsResult<()> {
190        let inner = Arc::clone(&self.inner);
191        let path = path.to_owned();
192        let reserved_bytes = path.len();
193        self.run(reserved_bytes, async move {
194            inner.mkdir(&path, recursive).await
195        })
196    }
197
198    fn mkdir_with_mode(
199        &mut self,
200        path: &str,
201        recursive: bool,
202        mode: Option<u32>,
203    ) -> PosixVfsResult<()> {
204        self.mkdir(path, recursive)?;
205        if let Some(mode) = mode {
206            self.chmod(path, mode)?;
207        }
208        Ok(())
209    }
210
211    fn mknod(&mut self, path: &str, mode: u32, rdev: u64) -> PosixVfsResult<()> {
212        let inner = Arc::clone(&self.inner);
213        let path = path.to_owned();
214        let reserved_bytes = path.len();
215        self.run(reserved_bytes, async move {
216            inner.mknod(&path, mode, rdev).await
217        })
218    }
219
220    fn exists(&self, path: &str) -> bool {
221        let inner = Arc::clone(&self.inner);
222        let path = path.to_owned();
223        let reserved_bytes = path.len();
224        match self.run(reserved_bytes, async move { Ok(inner.exists(&path).await) }) {
225            Ok(exists) => exists,
226            Err(error) => {
227                eprintln!("ERR_AGENTOS_VFS_EXISTS: {error}");
228                false
229            }
230        }
231    }
232
233    fn stat(&mut self, path: &str) -> PosixVfsResult<VirtualStat> {
234        let inner = Arc::clone(&self.inner);
235        let path = path.to_owned();
236        let reserved_bytes = path.len();
237        let stat = self.run(reserved_bytes, async move { inner.stat(&path).await })?;
238        Ok(convert_stat(stat, self.device_id))
239    }
240
241    fn remove_file(&mut self, path: &str) -> PosixVfsResult<()> {
242        let inner = Arc::clone(&self.inner);
243        let path = path.to_owned();
244        let reserved_bytes = path.len();
245        self.run(
246            reserved_bytes,
247            async move { inner.remove_file(&path).await },
248        )
249    }
250
251    fn remove_dir(&mut self, path: &str) -> PosixVfsResult<()> {
252        let inner = Arc::clone(&self.inner);
253        let path = path.to_owned();
254        let reserved_bytes = path.len();
255        self.run(reserved_bytes, async move { inner.remove_dir(&path).await })
256    }
257
258    fn rename(&mut self, old_path: &str, new_path: &str) -> PosixVfsResult<()> {
259        let inner = Arc::clone(&self.inner);
260        let old_path = old_path.to_owned();
261        let new_path = new_path.to_owned();
262        let reserved_bytes = old_path.len().saturating_add(new_path.len());
263        self.run(reserved_bytes, async move {
264            inner.rename(&old_path, &new_path).await
265        })
266    }
267
268    fn realpath(&self, path: &str) -> PosixVfsResult<String> {
269        let inner = Arc::clone(&self.inner);
270        let path = path.to_owned();
271        let reserved_bytes = path.len();
272        self.run(reserved_bytes, async move { inner.realpath(&path).await })
273    }
274
275    fn symlink(&mut self, target: &str, link_path: &str) -> PosixVfsResult<()> {
276        let inner = Arc::clone(&self.inner);
277        let target = target.to_owned();
278        let link_path = link_path.to_owned();
279        let reserved_bytes = target.len().saturating_add(link_path.len());
280        self.run(reserved_bytes, async move {
281            inner.symlink(&target, &link_path).await
282        })
283    }
284
285    fn read_link(&self, path: &str) -> PosixVfsResult<String> {
286        let inner = Arc::clone(&self.inner);
287        let path = path.to_owned();
288        let reserved_bytes = path.len();
289        self.run(reserved_bytes, async move { inner.readlink(&path).await })
290    }
291
292    fn lstat(&self, path: &str) -> PosixVfsResult<VirtualStat> {
293        let inner = Arc::clone(&self.inner);
294        let path = path.to_owned();
295        let reserved_bytes = path.len();
296        let stat = self.run(reserved_bytes, async move { inner.lstat(&path).await })?;
297        Ok(convert_stat(stat, self.device_id))
298    }
299
300    fn link(&mut self, old_path: &str, new_path: &str) -> PosixVfsResult<()> {
301        let inner = Arc::clone(&self.inner);
302        let old_path = old_path.to_owned();
303        let new_path = new_path.to_owned();
304        let reserved_bytes = old_path.len().saturating_add(new_path.len());
305        self.run(reserved_bytes, async move {
306            inner.link(&old_path, &new_path).await
307        })
308    }
309
310    fn chmod(&mut self, path: &str, mode: u32) -> PosixVfsResult<()> {
311        let inner = Arc::clone(&self.inner);
312        let path = path.to_owned();
313        let reserved_bytes = path.len();
314        self.run(
315            reserved_bytes,
316            async move { inner.chmod(&path, mode).await },
317        )
318    }
319
320    fn chown(&mut self, path: &str, uid: u32, gid: u32) -> PosixVfsResult<()> {
321        let inner = Arc::clone(&self.inner);
322        let path = path.to_owned();
323        let reserved_bytes = path.len();
324        self.run(
325            reserved_bytes,
326            async move { inner.chown(&path, uid, gid).await },
327        )
328    }
329
330    fn chown_spec(
331        &mut self,
332        path: &str,
333        uid: u32,
334        gid: u32,
335        follow_symlinks: bool,
336    ) -> PosixVfsResult<()> {
337        let inner = Arc::clone(&self.inner);
338        let path = path.to_owned();
339        let reserved_bytes = path.len();
340        self.run(reserved_bytes, async move {
341            if follow_symlinks {
342                inner.chown(&path, uid, gid).await
343            } else {
344                inner.lchown(&path, uid, gid).await
345            }
346        })
347    }
348
349    fn lchown(&mut self, path: &str, uid: u32, gid: u32) -> PosixVfsResult<()> {
350        self.chown_spec(path, uid, gid, false)
351    }
352
353    fn get_xattr(
354        &mut self,
355        path: &str,
356        name: &str,
357        follow_symlinks: bool,
358    ) -> PosixVfsResult<Vec<u8>> {
359        let inner = Arc::clone(&self.inner);
360        let path = path.to_owned();
361        let name = name.to_owned();
362        let reserved_bytes = path.len().saturating_add(name.len());
363        self.run(reserved_bytes, async move {
364            inner.get_xattr(&path, &name, follow_symlinks).await
365        })
366    }
367
368    fn list_xattrs(&mut self, path: &str, follow_symlinks: bool) -> PosixVfsResult<Vec<String>> {
369        let inner = Arc::clone(&self.inner);
370        let path = path.to_owned();
371        let reserved_bytes = path.len();
372        self.run(reserved_bytes, async move {
373            inner.list_xattrs(&path, follow_symlinks).await
374        })
375    }
376
377    fn set_xattr(
378        &mut self,
379        path: &str,
380        name: &str,
381        value: Vec<u8>,
382        flags: u32,
383        follow_symlinks: bool,
384    ) -> PosixVfsResult<()> {
385        let inner = Arc::clone(&self.inner);
386        let path = path.to_owned();
387        let name = name.to_owned();
388        let reserved_bytes = path
389            .len()
390            .saturating_add(name.len())
391            .saturating_add(value.len());
392        self.run(reserved_bytes, async move {
393            inner
394                .set_xattr(&path, &name, &value, flags, follow_symlinks)
395                .await
396        })
397    }
398
399    fn remove_xattr(
400        &mut self,
401        path: &str,
402        name: &str,
403        follow_symlinks: bool,
404    ) -> PosixVfsResult<()> {
405        let inner = Arc::clone(&self.inner);
406        let path = path.to_owned();
407        let name = name.to_owned();
408        let reserved_bytes = path.len().saturating_add(name.len());
409        self.run(reserved_bytes, async move {
410            inner.remove_xattr(&path, &name, follow_symlinks).await
411        })
412    }
413
414    fn utimes(&mut self, path: &str, atime_ms: u64, mtime_ms: u64) -> PosixVfsResult<()> {
415        let inner = Arc::clone(&self.inner);
416        let path = path.to_owned();
417        let reserved_bytes = path.len();
418        self.run(reserved_bytes, async move {
419            inner.utimes(&path, atime_ms, mtime_ms).await
420        })
421    }
422
423    fn set_atime(&mut self, path: &str, atime_ms: u64) -> PosixVfsResult<()> {
424        let inner = Arc::clone(&self.inner);
425        let path = path.to_owned();
426        let reserved_bytes = path.len();
427        self.run(reserved_bytes, async move {
428            inner.set_atime(&path, atime_ms).await
429        })
430    }
431
432    fn truncate(&mut self, path: &str, length: u64) -> PosixVfsResult<()> {
433        let inner = Arc::clone(&self.inner);
434        let path = path.to_owned();
435        let reserved_bytes = path.len();
436        self.run(reserved_bytes, async move {
437            inner.truncate(&path, length).await
438        })
439    }
440
441    fn sync(&mut self, path: &str) -> PosixVfsResult<()> {
442        let inner = Arc::clone(&self.inner);
443        let path = path.to_owned();
444        let reserved_bytes = path.len();
445        self.run(reserved_bytes, async move { inner.sync(&path).await })
446    }
447
448    fn allocate(&mut self, path: &str, offset: u64, length: u64) -> PosixVfsResult<()> {
449        let inner = Arc::clone(&self.inner);
450        let path = path.to_owned();
451        let reserved_bytes = path.len();
452        self.run(reserved_bytes, async move {
453            inner.allocate(&path, offset, length).await
454        })
455    }
456
457    fn insert_range(&mut self, path: &str, offset: u64, length: u64) -> PosixVfsResult<()> {
458        let inner = Arc::clone(&self.inner);
459        let path = path.to_owned();
460        let reserved_bytes = path.len();
461        self.run(reserved_bytes, async move {
462            inner.insert_range(&path, offset, length).await
463        })
464    }
465
466    fn collapse_range(&mut self, path: &str, offset: u64, length: u64) -> PosixVfsResult<()> {
467        let inner = Arc::clone(&self.inner);
468        let path = path.to_owned();
469        let reserved_bytes = path.len();
470        self.run(reserved_bytes, async move {
471            inner.collapse_range(&path, offset, length).await
472        })
473    }
474
475    fn zero_range(
476        &mut self,
477        path: &str,
478        offset: u64,
479        length: u64,
480        keep_size: bool,
481    ) -> PosixVfsResult<()> {
482        let inner = Arc::clone(&self.inner);
483        let path = path.to_owned();
484        let reserved_bytes = path.len();
485        self.run(reserved_bytes, async move {
486            inner.zero_range(&path, offset, length, keep_size).await
487        })
488    }
489
490    fn punch_hole(&mut self, path: &str, offset: u64, length: u64) -> PosixVfsResult<()> {
491        let inner = Arc::clone(&self.inner);
492        let path = path.to_owned();
493        let reserved_bytes = path.len();
494        self.run(reserved_bytes, async move {
495            inner.punch_hole(&path, offset, length).await
496        })
497    }
498
499    fn allocated_ranges(&mut self, path: &str) -> PosixVfsResult<Vec<(u64, u64)>> {
500        let inner = Arc::clone(&self.inner);
501        let path = path.to_owned();
502        let reserved_bytes = path.len();
503        self.run(reserved_bytes, async move {
504            inner.allocated_ranges(&path).await
505        })
506    }
507
508    fn unwritten_ranges(&mut self, path: &str) -> PosixVfsResult<Vec<(u64, u64)>> {
509        let inner = Arc::clone(&self.inner);
510        let path = path.to_owned();
511        let reserved_bytes = path.len();
512        self.run(reserved_bytes, async move {
513            inner.unwritten_ranges(&path).await
514        })
515    }
516
517    fn pread(&mut self, path: &str, offset: u64, length: usize) -> PosixVfsResult<Vec<u8>> {
518        let inner = Arc::clone(&self.inner);
519        let path = path.to_owned();
520        let reserved_bytes = path.len().saturating_add(length);
521        self.run(reserved_bytes, async move {
522            inner.pread(&path, offset, length).await
523        })
524    }
525
526    fn pwrite(&mut self, path: &str, content: Vec<u8>, offset: u64) -> PosixVfsResult<()> {
527        let inner = Arc::clone(&self.inner);
528        let path = path.to_owned();
529        let reserved_bytes = path.len().saturating_add(content.len());
530        self.run(reserved_bytes, async move {
531            inner.pwrite(&path, &content, offset).await
532        })
533    }
534}
535
536fn blocking_job_error(error: BlockingJobError) -> PosixVfsError {
537    let code = match error {
538        BlockingJobError::ResourceLimit(_) | BlockingJobError::Capacity { .. } => "EAGAIN",
539        BlockingJobError::ShuttingDown => "ECANCELED",
540        BlockingJobError::TimedOut { .. } => "ETIMEDOUT",
541        BlockingJobError::WorkerDropped => "EIO",
542    };
543    PosixVfsError::new(code, error.to_string())
544}
545
546fn convert_error(error: crate::engine::VfsError) -> PosixVfsError {
547    PosixVfsError::new(error.code(), error.message().to_owned())
548}
549
550fn convert_stat(stat: crate::engine::VirtualStat, device_id: u64) -> VirtualStat {
551    VirtualStat {
552        mode: stat.mode,
553        size: stat.size,
554        blocks: stat.blocks,
555        dev: device_id,
556        rdev: stat.rdev,
557        is_directory: stat.is_directory,
558        is_symbolic_link: stat.is_symbolic_link,
559        atime_ms: timespec_ms(stat.atime),
560        atime_nsec: stat.atime.nsec,
561        mtime_ms: timespec_ms(stat.mtime),
562        mtime_nsec: stat.mtime.nsec,
563        ctime_ms: timespec_ms(stat.ctime),
564        ctime_nsec: stat.ctime.nsec,
565        birthtime_ms: timespec_ms(stat.birthtime),
566        ino: stat.ino,
567        nlink: stat.nlink,
568        uid: stat.uid,
569        gid: stat.gid,
570    }
571}
572
573fn timespec_ms(time: crate::engine::Timespec) -> u64 {
574    if time.sec < 0 {
575        return 0;
576    }
577    (time.sec as u64).saturating_mul(1_000) + u64::from(time.nsec / 1_000_000)
578}
579
580#[cfg(test)]
581mod tests {
582    use super::*;
583    use crate::engine::engines::{ChunkedFs, ChunkedFsOptions};
584    use crate::engine::mem::{InMemoryMetadataStore, MemoryBlockStore};
585    use crate::posix::S_IFREG;
586
587    #[test]
588    fn mounted_engine_filesystem_bridges_sync_posix_calls() {
589        let runtime =
590            agentos_runtime::SidecarRuntime::process(&agentos_runtime::RuntimeConfig::default())
591                .expect("create test runtime");
592        let fs = ChunkedFs::with_options(
593            InMemoryMetadataStore::new(),
594            MemoryBlockStore::new(),
595            ChunkedFsOptions {
596                inline_threshold: 2,
597                chunk_size: 4,
598                ..ChunkedFsOptions::default()
599            },
600        );
601        let mut mounted = MountedEngineFileSystem::with_runtime_context(fs, runtime.context());
602
603        mounted
604            .mkdir("/work/nested", true)
605            .expect("create nested dir");
606        mounted
607            .write_file_with_mode("/work/nested/file.txt", b"hello".to_vec(), Some(0o600))
608            .expect("write file");
609        assert_eq!(
610            mounted
611                .pread("/work/nested/file.txt", 1, 3)
612                .expect("pread file"),
613            b"ell"
614        );
615        let entries = mounted
616            .read_dir_with_types("/work/nested")
617            .expect("read typed dir");
618        assert_eq!(entries.len(), 1);
619        assert_eq!(entries[0].name, "file.txt");
620        assert!(!entries[0].is_directory);
621
622        let stat = mounted.stat("/work/nested/file.txt").expect("stat file");
623        assert_eq!(stat.mode & 0o777, 0o600);
624        assert_eq!(stat.mode & S_IFREG, S_IFREG);
625        assert_eq!(stat.size, 5);
626    }
627
628    #[test]
629    fn mounted_engine_filesystem_rejects_waits_on_agentos_runtime_workers() {
630        let runtime =
631            agentos_runtime::SidecarRuntime::process(&agentos_runtime::RuntimeConfig::default())
632                .expect("create test runtime");
633        let context = runtime.context();
634        let task_context = context.clone();
635        let task = context
636            .spawn(agentos_runtime::TaskClass::Plugin, async move {
637                let fs = ChunkedFs::with_options(
638                    InMemoryMetadataStore::new(),
639                    MemoryBlockStore::new(),
640                    ChunkedFsOptions::default(),
641                );
642                let mut mounted = MountedEngineFileSystem::with_runtime_context(fs, task_context);
643                mounted
644                    .mkdir("/must-not-block", true)
645                    .expect_err("runtime workers must not synchronously wait")
646            })
647            .expect("spawn worker regression");
648        let error = runtime.block_on(task).expect("worker regression join");
649        assert_eq!(error.code(), "EDEADLK");
650        assert!(error
651            .message()
652            .contains("ERR_AGENTOS_VFS_RUNTIME_WORKER_WAIT"));
653    }
654}
vfs/adapter/mounted_fs.rs

vfs/adapter/
mounted_fs.rs
