Skip to main content

submilli_engine/runtime/
default_filesystem.rs

1//! Default filesystem implementation over the contained VFS.
2mod download;
3use super::durable::{EffectId, RunId};
4use super::filesystem::{
5    DirectoryEntry, DirectoryHandle, DirectoryReader, FileReader, FileWriter, Filesystem,
6    FilesystemError, FilesystemInfo, MutationBudget, OpenWrite, ReadItem, ReadOutput, ReadPosition,
7    ReaderHandle, ReaderOptions, TempFile, WriterHandle, bounded_read,
8};
9use super::filesystem::{
10    DirectoryNames, FileContents, FileUpdate, FileUpdateOutput, PreparedFileUpdate,
11};
12use super::fs::{self, ContainError, ContentPath, FileIdentity, LinkPath, MAX_REMOVE_ENTRIES};
13use super::fuel::WorkBudget;
14use super::limits::{HostBytes, TenantLimits};
15use super::vfs::Placement;
16use super::{DiskQuota, Holder, QuotaCharge, QuotaExceeded, Vfs};
17use crate::stdlib::shared::{contain_trap, quota_refusal, write_target_trap};
18use cap_std::fs::{Dir, DirEntry, ReadDir};
19use std::collections::VecDeque;
20use std::ffi::{OsStr, OsString};
21use std::fs::File;
22use std::io::{BufWriter, Read, Seek, SeekFrom, Write};
23use std::path::{Path, PathBuf};
24use std::sync::Arc;
25
26pub struct DefaultFilesystem {
27    vfs: Vfs,
28    run_id: RunId,
29}
30impl DefaultFilesystem {
31    pub fn new(vfs: Vfs, run_id: RunId) -> Self {
32        Self { vfs, run_id }
33    }
34}
35#[async_trait::async_trait]
36impl Filesystem for DefaultFilesystem {
37    fn info(&self) -> FilesystemInfo {
38        FilesystemInfo::from_vfs(&self.vfs)
39    }
40    fn cwd(&self) -> &str {
41        self.vfs.cwd()
42    }
43    async fn placement(&self, path: &str) -> Result<Placement, FilesystemError> {
44        let vfs = self.vfs.clone();
45        let path = path.to_owned();
46        blocking(move || {
47            Ok(fs::resolve_content(&vfs, vfs.cwd(), &path)?
48                .placement()
49                .clone())
50        })
51        .await
52    }
53    async fn read_regular(
54        &self,
55        work: Option<Arc<WorkBudget>>,
56        path: &str,
57        maximum_bytes: u64,
58        limits: &TenantLimits,
59    ) -> Result<FileContents, FilesystemError> {
60        let vfs = self.vfs.clone();
61        let path = path.to_owned();
62        read_regular(vfs, work, path, maximum_bytes, limits).await
63    }
64    async fn directory_names(
65        &self,
66        work: Option<Arc<WorkBudget>>,
67        path: &str,
68        maximum_entries: usize,
69        limits: &TenantLimits,
70    ) -> Result<DirectoryNames, FilesystemError> {
71        let vfs = self.vfs.clone();
72        let path = path.to_owned();
73        directory_names(vfs, work, path, maximum_entries, limits).await
74    }
75    async fn begin_download(
76        &self,
77        _effect: &EffectId,
78        path: &str,
79        options: super::filesystem::DownloadOptions,
80        _work: Arc<WorkBudget>,
81    ) -> Result<Box<dyn super::filesystem::Download>, FilesystemError> {
82        download::begin(self.vfs.clone(), path.to_owned(), options).await
83    }
84    async fn begin_file_update(
85        &self,
86        _effect_id: &EffectId,
87        path: &str,
88        _request: [u8; 32],
89        _limits: &TenantLimits,
90    ) -> Result<Box<dyn FileUpdate>, FilesystemError> {
91        Ok(Box::new(FileReplacement {
92            vfs: self.vfs.clone(),
93            path: path.to_owned(),
94        }))
95    }
96    async fn metadata(
97        &self,
98        _effect_id: &EffectId,
99        _work: Option<Arc<WorkBudget>>,
100        path: &str,
101        follow: bool,
102    ) -> Result<cap_std::fs::Metadata, FilesystemError> {
103        let vfs = self.vfs.clone();
104        let path = path.to_owned();
105        blocking(move || {
106            if follow {
107                Ok(fs::resolve_content(&vfs, vfs.cwd(), &path)?.metadata()?)
108            } else {
109                Ok(fs::resolve_link(&vfs, vfs.cwd(), &path)?.symlink_metadata()?)
110            }
111        })
112        .await
113    }
114    async fn read(
115        &self,
116        _effect_id: &EffectId,
117        _work: Option<Arc<WorkBudget>>,
118        path: &str,
119        offset: u64,
120        maximum_bytes: u64,
121    ) -> Result<Vec<u8>, FilesystemError> {
122        let file = fs::resolve_content(&self.vfs, self.vfs.cwd(), path)?.open()?;
123        bounded_read(file, offset, maximum_bytes)
124    }
125    async fn write(
126        &self,
127        _effect_id: &EffectId,
128        _work: Option<Arc<WorkBudget>>,
129        path: &str,
130        bytes: Vec<u8>,
131    ) -> Result<(), FilesystemError> {
132        let path = fs::resolve_content(&self.vfs, self.vfs.cwd(), path)?;
133        path.writable()?;
134        if path.is_root() {
135            return Err(FilesystemError::Unsupported(
136                "cannot write to the VFS root or a mount point",
137            ));
138        }
139        let mut quota = QuotaCharge::new(path.placement().quota().cloned(), path.regular_file());
140        quota.reserve(bytes.len() as u64)?;
141        let temp = path.temp_sibling();
142        let mut file = temp.create_new()?;
143        let result = (|| {
144            file.write_all(&bytes)?;
145            file.sync_all()?;
146            drop(file);
147            temp.rename_to(&path)?;
148            Ok(())
149        })();
150        if result.is_err() {
151            let _ = temp.remove_file();
152        } else {
153            quota.commit();
154        }
155        result
156    }
157    async fn append(
158        &self,
159        _effect_id: &EffectId,
160        _work: Option<Arc<WorkBudget>>,
161        path: &str,
162        bytes: Vec<u8>,
163    ) -> Result<(), FilesystemError> {
164        let path = fs::resolve_content(&self.vfs, self.vfs.cwd(), path)?;
165        path.writable()?;
166        if path.is_root() {
167            return Err(FilesystemError::Unsupported(
168                "cannot write to the VFS root or a mount point",
169            ));
170        }
171        let count = bytes.len() as u64;
172        let before = path.file_len();
173        let mut quota = QuotaCharge::new(path.placement().quota().cloned(), None);
174        quota.reserve(count)?;
175        let result = path
176            .append()
177            .map_err(FilesystemError::from)
178            .and_then(|mut file| file.write_all(&bytes).map_err(Into::into));
179        if result.is_err() {
180            quota
181                .unreserve(count.saturating_sub(path.file_len().saturating_sub(before).min(count)));
182            quota.keep();
183        } else {
184            quota.commit();
185        }
186        result
187    }
188    async fn mkdir(
189        &self,
190        _effect_id: &EffectId,
191        _work: Option<Arc<WorkBudget>>,
192        path: &str,
193        recursive: bool,
194    ) -> Result<(), FilesystemError> {
195        let path = fs::resolve_content(&self.vfs, self.vfs.cwd(), path)?;
196        if recursive {
197            path.create_dir_all()?;
198        } else {
199            path.create_dir()?;
200        }
201        Ok(())
202    }
203    async fn remove(
204        &self,
205        _effect_id: &EffectId,
206        _work: Option<Arc<WorkBudget>>,
207        path: &str,
208        recursive: bool,
209        budget: &mut dyn MutationBudget,
210    ) -> Result<(), FilesystemError> {
211        default_remove(&self.vfs, path, recursive, budget).map_err(FilesystemError::Execution)
212    }
213    async fn move_entry(
214        &self,
215        _effect_id: &EffectId,
216        _work: Option<Arc<WorkBudget>>,
217        from: &str,
218        to: &str,
219        budget: &mut dyn MutationBudget,
220    ) -> Result<(), FilesystemError> {
221        default_move(&self.vfs, from, to, budget).map_err(FilesystemError::Execution)
222    }
223    async fn copy(
224        &self,
225        _effect_id: &EffectId,
226        _work: Option<Arc<WorkBudget>>,
227        from: &str,
228        to: &str,
229        recursive: bool,
230        budget: &mut dyn MutationBudget,
231    ) -> Result<(), FilesystemError> {
232        default_copy(&self.vfs, from, to, recursive, budget).map_err(FilesystemError::Execution)
233    }
234    async fn next_reader(
235        &self,
236        _: &EffectId,
237        work: Option<Arc<WorkBudget>>,
238        reader: &mut FileReader,
239        limits: &super::limits::TenantLimits,
240    ) -> Result<ReadOutput, FilesystemError> {
241        let native = reader.local.as_mut().ok_or(FilesystemError::Unsupported(
242            "default reader is not available in this process",
243        ))?;
244        let Some(file) = &mut native.file else {
245            return empty_read(limits);
246        };
247        read_item(
248            file,
249            &reader.options,
250            &mut native.position,
251            work.as_deref(),
252            limits,
253        )
254    }
255    async fn close_reader(
256        &self,
257        _: &EffectId,
258        _: Option<Arc<WorkBudget>>,
259        reader: &mut FileReader,
260    ) -> Result<(), FilesystemError> {
261        let native = reader.local.as_mut().ok_or(FilesystemError::Unsupported(
262            "default reader is not available in this process",
263        ))?;
264        native.file = None;
265        native.held = None;
266        native.position.closed = true;
267        Ok(())
268    }
269    async fn next_directory(
270        &self,
271        _: &EffectId,
272        _: Option<Arc<WorkBudget>>,
273        directory: &mut DirectoryReader,
274        _: &super::limits::TenantLimits,
275    ) -> Result<Option<DirectoryEntry>, FilesystemError> {
276        let native = directory
277            .local
278            .as_mut()
279            .ok_or(FilesystemError::Unsupported(
280                "default listing is not available in this process",
281            ))?;
282        Ok(native.entries.as_mut().and_then(Iterator::next))
283    }
284    async fn close_directory(
285        &self,
286        _: &EffectId,
287        _: Option<Arc<WorkBudget>>,
288        directory: &mut DirectoryReader,
289        _: &super::limits::TenantLimits,
290    ) -> Result<(), FilesystemError> {
291        let native = directory
292            .local
293            .as_mut()
294            .ok_or(FilesystemError::Unsupported(
295                "default listing is not available in this process",
296            ))?;
297        native.entries = None;
298        Ok(())
299    }
300    async fn write_chunk(
301        &self,
302        _: &EffectId,
303        _: Option<Arc<WorkBudget>>,
304        writer: &mut FileWriter,
305        bytes: Vec<u8>,
306    ) -> Result<(), FilesystemError> {
307        let native = writer.local.as_mut().ok_or(FilesystemError::Unsupported(
308            "default writer is not available in this process",
309        ))?;
310        native.write_bytes(&bytes).map_err(Into::into)
311    }
312    async fn close_writer(
313        &self,
314        _: &EffectId,
315        _: Option<Arc<WorkBudget>>,
316        writer: &mut FileWriter,
317    ) -> Result<(), FilesystemError> {
318        let native = writer.local.as_mut().ok_or(FilesystemError::Unsupported(
319            "default writer is not available in this process",
320        ))?;
321        let result = native.close().map_err(Into::into);
322        writer.closed = native.open.is_none();
323        result
324    }
325    async fn open_reader(
326        &self,
327        _effect_id: &EffectId,
328        _work: Option<Arc<WorkBudget>>,
329        path: &str,
330        options: ReaderOptions,
331    ) -> Result<FileReader, FilesystemError> {
332        let path = fs::resolve_content(&self.vfs, self.vfs.cwd(), path)?;
333        let file = path.open()?;
334        let held = path.placement().quota().and_then(|quota| {
335            let metadata = file.metadata().ok()?;
336            Some(quota.hold(fs::FileIdentity::of(&metadata).ok()?, super::Holder::Reader))
337        });
338        Ok(FileReader {
339            run_id: self.run_id.clone(),
340            id: _effect_id.clone(),
341            options,
342            local: Some(ReaderHandle {
343                file: Some(file.into_std()),
344                position: ReadPosition::default(),
345                held,
346            }),
347        })
348    }
349    async fn open_directory(
350        &self,
351        _effect_id: &EffectId,
352        _work: Option<Arc<WorkBudget>>,
353        path: &str,
354        recursive: bool,
355        _: &super::limits::TenantLimits,
356    ) -> Result<DirectoryReader, FilesystemError> {
357        let resolved = fs::resolve_content(&self.vfs, self.vfs.cwd(), path)?;
358        if !resolved.metadata()?.is_dir() {
359            return Err(std::io::Error::from(std::io::ErrorKind::NotADirectory).into());
360        }
361        let base = std::sync::Arc::new(resolved.open_dir()?);
362        let mounts = if recursive {
363            resolved.mounts_below()
364        } else {
365            Vec::new()
366        };
367        let normalized = fs::guest_normalize(self.vfs.cwd(), path).map_err(ContainError::from)?;
368        let prefix = if normalized == "/" {
369            String::new()
370        } else {
371            format!("{}/", normalized.trim_start_matches('/'))
372        };
373        let mut walk = ContainedWalk::new(base, prefix, recursive, mounts)?;
374        Ok(DirectoryReader {
375            run_id: self.run_id.clone(),
376            id: _effect_id.clone(),
377            local: Some(DirectoryHandle {
378                entries: Some(Box::new(std::iter::from_fn(move || walk.next_entry()))),
379            }),
380        })
381    }
382    async fn open_writer(
383        &self,
384        _effect_id: &EffectId,
385        _work: Option<Arc<WorkBudget>>,
386        path: &str,
387    ) -> Result<FileWriter, FilesystemError> {
388        let path = fs::resolve_content(&self.vfs, self.vfs.cwd(), path)?;
389        path.writable()?;
390        if path.is_root() {
391            return Err(FilesystemError::Unsupported(
392                "cannot write to the VFS root or a mount point",
393            ));
394        }
395        let temp_path = path.temp_sibling();
396        let file = temp_path.create_new()?;
397        let temp = TempFile::created(temp_path, &file)?;
398        let quota = path.placement().quota().cloned();
399        Ok(FileWriter {
400            run_id: self.run_id.clone(),
401            id: _effect_id.clone(),
402            closed: false,
403            local: Some(WriterHandle::new(file.into_std(), temp, path, quota)),
404        })
405    }
406}
407
408/// Shared bounded observations for native filesystem implementations. They do
409/// not record history: the enclosing host observation commits before exposure.
410pub async fn read_regular(
411    vfs: Vfs,
412    work: Option<Arc<WorkBudget>>,
413    path: String,
414    maximum_bytes: u64,
415    limits: &TenantLimits,
416) -> Result<FileContents, FilesystemError> {
417    let (file, metadata) = blocking(move || {
418        let resolved = fs::resolve_content(&vfs, vfs.cwd(), &path)?;
419        Ok(resolved.open_regular()?)
420    })
421    .await?;
422    let length = metadata.len();
423    if length > maximum_bytes {
424        return Err(FilesystemError::Rejected(
425            "code: result or input exceeds maxReadSize; use a smaller file or query".into(),
426        ));
427    }
428    let bound = length
429        .checked_add(1)
430        .ok_or(FilesystemError::Unsupported("file is too large"))?;
431    let memory = HostBytes::new(limits, bound).map_err(|e| FilesystemError::Execution(e.into()))?;
432    if let Some(work) = work {
433        work.charge(super::fuel::IO, bound)?;
434    }
435    blocking(move || {
436        let capacity = usize::try_from(bound)
437            .map_err(|_| FilesystemError::Unsupported("file is too large"))?;
438        let mut bytes = Vec::new();
439        bytes
440            .try_reserve_exact(capacity)
441            .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
442        // read_to_end may reserve speculative capacity at EOF. A fixed slice keeps
443        // the admitted allocation exact, including the single growth-detection byte.
444        bytes.resize(capacity, 0);
445        let mut file = file;
446        let mut filled = 0;
447        while filled < capacity {
448            let count = file.read(&mut bytes[filled..])?;
449            if count == 0 {
450                break;
451            }
452            filled += count;
453        }
454        bytes.truncate(filled);
455        if bytes.len() as u64 > length {
456            return Err(FilesystemError::Rejected(
457                "file grew during read; retry".into(),
458            ));
459        }
460        Ok(FileContents {
461            bytes,
462            _memory: memory,
463        })
464    })
465    .await
466}
467
468pub async fn directory_names(
469    vfs: Vfs,
470    work: Option<Arc<WorkBudget>>,
471    path: String,
472    maximum_entries: usize,
473    limits: &TenantLimits,
474) -> Result<DirectoryNames, FilesystemError> {
475    let worker_path = path.clone();
476    let mut entries = blocking(move || {
477        let resolved = fs::resolve_content(&vfs, vfs.cwd(), &worker_path)?;
478        Ok(resolved.entries()?)
479    })
480    .await?;
481    let mut memory = HostBytes::new(limits, 0).map_err(|e| FilesystemError::Execution(e.into()))?;
482    let mut names = Vec::new();
483    loop {
484        let (entry, remaining) = blocking(move || Ok((entries.next(), entries))).await?;
485        entries = remaining;
486        let Some(entry) = entry else { break };
487        let entry = entry?;
488        if names.len() == maximum_entries {
489            return Err(FilesystemError::Rejected(
490                "traversal exceeds 20000 entries; choose a smaller root".into(),
491            ));
492        }
493        if let Some(work) = &work {
494            work.charge(super::fuel::SYSCALL, 1)?;
495        }
496        let raw = entry.file_name();
497        let name = raw
498            .to_str()
499            .ok_or_else(|| FilesystemError::Rejected(format!("non-UTF-8 filename in {path}")))?;
500        memory
501            .grow(limits, name.len() as u64)
502            .map_err(|e| FilesystemError::Execution(e.into()))?;
503        if names.len() == names.capacity() {
504            let capacity = names
505                .capacity()
506                .saturating_mul(2)
507                .max(16)
508                .min(maximum_entries);
509            memory
510                .grow(
511                    limits,
512                    ((capacity - names.capacity()) * std::mem::size_of::<String>()) as u64,
513                )
514                .map_err(|e| FilesystemError::Execution(e.into()))?;
515            names
516                .try_reserve_exact(capacity - names.len())
517                .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
518        }
519        names.push(name.to_owned());
520    }
521    Ok(DirectoryNames {
522        names,
523        _memory: memory,
524    })
525}
526
527struct FileReplacement {
528    vfs: Vfs,
529    path: String,
530}
531#[async_trait::async_trait]
532impl FileUpdate for FileReplacement {
533    fn saved_output(&self) -> Option<(&[u8], u64)> {
534        None
535    }
536    async fn publish(
537        self: Box<Self>,
538        work: Arc<WorkBudget>,
539        prepared: Option<PreparedFileUpdate>,
540        limits: &TenantLimits,
541    ) -> Result<FileUpdateOutput, FilesystemError> {
542        let prepared =
543            prepared.ok_or(FilesystemError::Unsupported("missing prepared file update"))?;
544        let Some(bytes) = prepared.contents else {
545            return Ok(prepared.output);
546        };
547        let memory = HostBytes::new(limits, bytes.capacity() as u64)
548            .map_err(|e| FilesystemError::Execution(e.into()))?;
549        blocking(move || {
550            let _memory = memory;
551            replace_existing(&self.vfs, &self.path, &bytes, &work)
552        })
553        .await?;
554        Ok(prepared.output)
555    }
556}
557
558async fn blocking<T: Send + 'static>(
559    operation: impl FnOnce() -> Result<T, FilesystemError> + Send + 'static,
560) -> Result<T, FilesystemError> {
561    tokio::task::spawn_blocking(operation)
562        .await
563        .map_err(|e| FilesystemError::Backend(Box::new(e)))?
564}
565
566fn replace_existing(
567    vfs: &Vfs,
568    path: &str,
569    bytes: &[u8],
570    work: &WorkBudget,
571) -> Result<(), FilesystemError> {
572    let destination = fs::resolve_content(vfs, vfs.cwd(), path)?;
573    destination.check_rename_end()?;
574    let (_, metadata) = destination.open_regular()?;
575    let mut quota = QuotaCharge::new(
576        destination.placement().quota().cloned(),
577        destination.regular_file(),
578    );
579    quota.reserve(bytes.len() as u64)?;
580    work.charge(super::fuel::IO, bytes.len() as u64)?;
581    work.charge(super::fuel::SYSCALL, 8)?;
582    let temporary = destination.temp_sibling();
583    temporary.check_rename_end()?;
584    let mut file = temporary.create_new()?;
585    let result = (|| {
586        file.write_all(bytes)?;
587        file.set_permissions(metadata.permissions())?;
588        file.sync_all()?;
589        work.before_effect();
590        temporary
591            .rename_to(&destination)
592            .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
593        quota.commit();
594        // The default backend is nonpersistent. Windows cannot fsync a
595        // directory through this handle; Unix also persists the rename here.
596        #[cfg(unix)]
597        destination
598            .sync_parent()
599            .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
600        Ok(())
601    })();
602    if result.is_err() {
603        let _ = temporary.remove_file();
604    }
605    result
606}
607
608/// Read one logical item, anchoring every attempt at the supplied position.
609/// Read-ahead is local only. Errors leave the logical position unchanged.
610/// Durable backends commit the returned position before exposing the result.
611pub fn read_item(
612    file: &mut (impl Read + Seek),
613    options: &ReaderOptions,
614    position: &mut ReadPosition,
615    work: Option<&WorkBudget>,
616    limits: &super::limits::TenantLimits,
617) -> Result<ReadOutput, FilesystemError> {
618    if position.closed {
619        return empty_read(limits);
620    }
621    file.seek(SeekFrom::Start(position.offset))?;
622    let mut next = position.clone();
623    let result = match options {
624        ReaderOptions::Bytes { chunk_size } => {
625            read_chunk(file, *chunk_size, &mut next, work, limits)
626        }
627        ReaderOptions::Lines { maximum_bytes } => {
628            read_line(file, *maximum_bytes, &mut next, work, limits)
629        }
630    }?;
631    *position = next;
632    Ok(result)
633}
634fn empty_read(limits: &super::limits::TenantLimits) -> Result<ReadOutput, FilesystemError> {
635    Ok(ReadOutput {
636        item: None,
637        _charge: Some(memory(limits, 0)?),
638    })
639}
640fn memory(
641    limits: &super::limits::TenantLimits,
642    bytes: u64,
643) -> Result<super::limits::HostBytes, FilesystemError> {
644    super::limits::HostBytes::new(limits, bytes)
645        .map_err(|e| FilesystemError::Execution(wasmtime::Error::new(e)))
646}
647fn read_cost(
648    work: Option<&WorkBudget>,
649    rate: super::fuel::Rate,
650    bytes: usize,
651) -> Result<(), FilesystemError> {
652    if let Some(work) = work {
653        // The read has happened; exhaustion cannot discard its result.
654        work.before_effect();
655        work.charge(rate, bytes as u64)?;
656    }
657    Ok(())
658}
659fn read_chunk(
660    file: &mut impl Read,
661    size: usize,
662    position: &mut ReadPosition,
663    work: Option<&WorkBudget>,
664    limits: &super::limits::TenantLimits,
665) -> Result<ReadOutput, FilesystemError> {
666    if size == 0 {
667        return Err(FilesystemError::Rejected(
668            "chunk size must be positive".into(),
669        ));
670    }
671    if let Some(work) = work {
672        work.charge(super::fuel::COPY, size as u64)?;
673    }
674    let charge = memory(limits, size as u64)?;
675    let mut bytes = Vec::new();
676    bytes
677        .try_reserve_exact(size)
678        .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
679    bytes.resize(size, 0);
680    let mut filled = 0;
681    while filled < size {
682        let count = file.read(&mut bytes[filled..])?;
683        read_cost(work, super::fuel::IO, count)?;
684        if count == 0 {
685            break;
686        }
687        filled += count;
688    }
689    position.offset = position
690        .offset
691        .checked_add(filled as u64)
692        .ok_or_else(|| FilesystemError::Rejected("reader offset overflow".into()))?;
693    bytes.truncate(filled);
694    Ok(ReadOutput {
695        item: (filled > 0).then_some(ReadItem::Bytes(bytes)),
696        _charge: Some(charge),
697    })
698}
699fn read_line(
700    file: &mut impl Read,
701    maximum: u64,
702    position: &mut ReadPosition,
703    work: Option<&WorkBudget>,
704    limits: &super::limits::TenantLimits,
705) -> Result<ReadOutput, FilesystemError> {
706    use std::io::BufRead;
707    let mut charge = memory(limits, 8192)?;
708    let mut reader = std::io::BufReader::new(file);
709    let mut bytes = Vec::new();
710    loop {
711        let needs_read = reader.buffer().is_empty();
712        let available = reader.fill_buf()?;
713        if needs_read {
714            read_cost(work, super::fuel::IO, available.len())?;
715        }
716        if available.is_empty() {
717            break;
718        }
719        let newline = available.iter().position(|&b| b == b'\n');
720        let count = newline.map_or(available.len(), |i| i + 1);
721        read_cost(work, super::fuel::SCAN, count)?;
722        let length = bytes.len().checked_add(count).ok_or_else(|| {
723            FilesystemError::Execution(super::host::range_error(
724                "fs.lines: line length exceeds host capacity",
725            ))
726        })?;
727        if length as u64 > maximum {
728            return Err(FilesystemError::Execution(super::host::range_error(
729                format!(
730                    "fs.lines: line exceeds maxReadSize of {maximum} bytes; read smaller chunks with fs.bytes"
731                ),
732            )));
733        }
734        if length > bytes.capacity() {
735            let capacity = length
736                .max(bytes.capacity().saturating_mul(2))
737                .min(usize::try_from(maximum).unwrap_or(usize::MAX));
738            charge
739                .grow(limits, (capacity - bytes.capacity()) as u64)
740                .map_err(|e| FilesystemError::Execution(wasmtime::Error::new(e)))?;
741            bytes
742                .try_reserve_exact(capacity - bytes.len())
743                .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
744        }
745        bytes.extend_from_slice(&available[..count]);
746        reader.consume(count);
747        if newline.is_some() {
748            break;
749        }
750    }
751    if bytes.is_empty() {
752        return empty_read(limits);
753    }
754    position.offset = position
755        .offset
756        .checked_add(bytes.len() as u64)
757        .ok_or_else(|| FilesystemError::Rejected("reader offset overflow".into()))?;
758    if bytes.last() == Some(&b'\n') {
759        bytes.pop();
760        if bytes.last() == Some(&b'\r') {
761            bytes.pop();
762        }
763    }
764    let capacity = bytes
765        .len()
766        .checked_mul(3)
767        .ok_or_else(|| FilesystemError::Rejected("line capacity overflow".into()))?;
768    let output_charge = memory(limits, capacity as u64)?;
769    let mut text = String::new();
770    text.try_reserve_exact(capacity)
771        .map_err(|e| FilesystemError::Backend(Box::new(e)))?;
772    for part in bytes.utf8_chunks() {
773        text.push_str(part.valid());
774        if !part.invalid().is_empty() {
775            text.push('\u{fffd}');
776        }
777    }
778    if !position.bom_handled {
779        position.bom_handled = true;
780        if text.starts_with('\u{feff}') {
781            text.drain(..'\u{feff}'.len_utf8());
782        }
783    }
784    Ok(ReadOutput {
785        item: Some(ReadItem::Line(text)),
786        _charge: Some(output_charge),
787    })
788}
789
790/// Concurrently open directory handles per walk. A deep tree times the number of list
791/// handles a guest holds open consumes process-wide descriptors that the tenant limiter
792/// does not account for, since it charges a byte budget only. Subdirectories found past
793/// the cap are postponed and reopened from the base handle once a handle frees up.
794///
795/// Higher than `walkdir`'s `max_open` default of 10: `walkdir` pays only extra syscalls
796/// when it hits the cap (it closes the *shallowest* level and reopens it by path,
797/// preserving depth-first order), whereas postponing costs ordering as well — a postponed
798/// subtree lands after the rest of its parent's entries. Real checkouts (nested
799/// `node_modules`, deep monorepos) routinely pass 10 levels and rarely pass 32.
800///
801/// The `list` doc in `declaration.rs` states this depth as the guest-facing ordering
802/// contract — change both together.
803const MAX_OPEN_DIRS: usize = 32;
804
805/// Postponed subdirectory names a single walk will hold at once. Only levels past
806/// [`MAX_OPEN_DIRS`] postpone at all, so reaching this needs a directory 32 deep with a
807/// fan-out in the thousands; past it a subdirectory is skipped rather than remembered,
808/// which is the same best-effort outcome an unreadable one already gets. Without a
809/// ceiling the queue grows with the width of a directory the guest did not have to pay
810/// for in the byte cap.
811const MAX_DEFERRED: usize = 16_384;
812
813struct Level {
814    /// `None` once the directory is fully read: the handle is released at that point,
815    /// while the frame stays on the stack until [`Level::deferred`] drains.
816    iter: Option<ReadDir>,
817    /// Path of this level's directory relative to the listed directory — empty at the top.
818    /// Base-relative rather than guest-absolute so a postponed level can be reopened
819    /// against the base handle, and a `PathBuf` rather than a `String` so a name the
820    /// platform allows but UTF-8 does not still reopens — the lossy [`Level::prefix`]
821    /// would have replaced it with U+FFFD and lost the subtree.
822    path: PathBuf,
823    /// Lossy rendering of [`Level::path`] with guest `/` separators: `""` at the top,
824    /// `"sub/"` one down. Guest paths are built from this.
825    prefix: String,
826    /// Names of subdirectories of this level whose descent was postponed — because the
827    /// open-handle cap was reached, or because opening them from this level's handle
828    /// failed (most likely descriptor exhaustion, which is exactly what the cap manages).
829    /// Drained in order once this level is fully read, each rejoined to [`Level::path`]
830    /// and reopened from the base handle.
831    ///
832    /// Bare names, not paths: at the depth where postponement starts, the level's own
833    /// path is 32 components long, and storing it again per postponed child would make
834    /// the queue tens of times larger than the entry names it stands for.
835    deferred: VecDeque<OsString>,
836}
837
838impl Level {
839    fn new(path: PathBuf, iter: ReadDir) -> Self {
840        let mut prefix = String::new();
841        for component in &path {
842            prefix.push_str(&component.to_string_lossy());
843            prefix.push('/');
844        }
845        Self {
846            iter: Some(iter),
847            path,
848            prefix,
849            deferred: VecDeque::new(),
850        }
851    }
852}
853
854/// Directory walker (flat or recursive), rooted at a contained handle.
855///
856/// Lazy, pre-order, and non-following — matching `walkdir`'s default, so a checkout's
857/// internal links surface as entries rather than being descended and double-walked.
858/// Per-entry errors skip and continue rather than trapping the whole iteration: a
859/// recursive walk over a real checkout is the flagship use case, and one unreadable
860/// subtree should not end it.
861///
862/// Ordering: every directory is yielded before its descendants, and every subtree is
863/// yielded as one contiguous run — nothing outside a directory's subtree is interleaved
864/// into it. Strictly depth-first down to [`MAX_OPEN_DIRS`] levels of nesting; past that,
865/// a subdirectory's own contents follow the rest of its parent's entries instead of
866/// coming immediately after it, because the descent has to wait for a directory handle.
867struct ContainedWalk {
868    base: Arc<Dir>,
869    /// Mount points below the base, relative to it, with the volume handle a
870    /// descent opens in place of the empty directory the root holds there.
871    mounts: Vec<(PathBuf, Arc<Dir>)>,
872    /// Guest-relative path of the listed directory, `""` for the root, else `"tree/"`.
873    base_prefix: String,
874    stack: Vec<Level>,
875    /// Levels on `stack` still holding a `ReadDir`; the frames that have released theirs
876    /// stay on the stack while their postponed subdirectories drain.
877    open_dirs: usize,
878    /// Postponed names held across every level, against [`MAX_DEFERRED`].
879    deferred: usize,
880    recursive: bool,
881}
882
883impl ContainedWalk {
884    /// `base_prefix` is the guest-relative path of the listed directory, `""` for the root.
885    /// `mounts` are the mount points below it; see [`ContentPath::mounts_below`].
886    pub fn new(
887        base: Arc<Dir>,
888        base_prefix: String,
889        recursive: bool,
890        mounts: Vec<(PathBuf, Arc<Dir>)>,
891    ) -> std::io::Result<Self> {
892        let iter = base.entries()?;
893        Ok(Self {
894            base,
895            mounts,
896            base_prefix,
897            stack: vec![Level::new(PathBuf::new(), iter)],
898            open_dirs: 1,
899            deferred: 0,
900            recursive,
901        })
902    }
903
904    fn next_entry(&mut self) -> Option<DirectoryEntry> {
905        loop {
906            let level = self.stack.last_mut()?;
907            match level.iter.as_mut().map(Iterator::next) {
908                Some(Some(Ok(entry))) => {
909                    if let Some(found) = self.visit(entry) {
910                        return Some(found);
911                    }
912                }
913                Some(Some(Err(_))) => {}
914                Some(None) => {
915                    level.iter = None;
916                    self.open_dirs -= 1;
917                }
918                None => self.resume_deferred(),
919            }
920        }
921    }
922
923    /// Yields `entry`, descending into it first when it is a directory a recursive walk
924    /// should enter. `None` for an entry whose type cannot be determined at all.
925    fn visit(&mut self, entry: DirEntry) -> Option<DirectoryEntry> {
926        let raw_name = entry.file_name();
927        // `file_type` can fail where the directory entry carries no type; the entry's
928        // own metadata is the fallback before giving up on it.
929        let ft = entry
930            .file_type()
931            .ok()
932            .or_else(|| entry.metadata().ok().map(|m| m.file_type()))?;
933        let size = if ft.is_file() {
934            entry.metadata().map_or(0, |m| m.len())
935        } else {
936            0
937        };
938        let level = self.stack.last()?;
939        let name = raw_name.to_string_lossy().into_owned();
940        let guest_path = format!("/{}{}{}", self.base_prefix, level.prefix, name);
941        if self.recursive && ft.is_dir() && !ft.is_symlink() {
942            self.descend(&raw_name, &entry);
943        }
944        Some(DirectoryEntry {
945            name,
946            path: guest_path,
947            kind: kind_of(&ft),
948            bytes: size,
949        })
950    }
951
952    /// Opens the subdirectory named `name` as the new innermost level, or postpones it on
953    /// the current level — when there is no handle to spare, and equally when the open
954    /// fails, since a failure here is most likely descriptor exhaustion and dropping the
955    /// subtree over that would silently truncate the walk. Past [`MAX_DEFERRED`] it is
956    /// dropped after all, the walk having run out of room to remember it.
957    fn descend(&mut self, name: &OsStr, entry: &DirEntry) {
958        if self.open_dirs < MAX_OPEN_DIRS
959            && let Some(child) = self.stack.last().map(|level| level.path.join(name))
960            && let Ok(dir) = self.open_child(&child, entry)
961            && let Ok(iter) = dir.entries()
962        {
963            self.stack.push(Level::new(child, iter));
964            self.open_dirs += 1;
965            return;
966        }
967        if self.deferred < MAX_DEFERRED
968            && let Some(level) = self.stack.last_mut()
969        {
970            level.deferred.push_back(name.to_os_string());
971            self.deferred += 1;
972        }
973    }
974
975    /// The innermost level is fully read and has released its handle: descend into the
976    /// next subdirectory it postponed, or pop it once none are left. A postponed directory
977    /// that still won't open is skipped — same best-effort contract as an unreadable one,
978    /// and one retry keeps the walk finite.
979    fn resume_deferred(&mut self) {
980        let Some(level) = self.stack.last_mut() else {
981            return;
982        };
983        let Some(name) = level.deferred.pop_front() else {
984            self.stack.pop();
985            return;
986        };
987        self.deferred -= 1;
988        let path = level.path.join(name);
989        // The level draining its deferrals released its own handle first, so `open_dirs`
990        // is always below the cap here.
991        if let Ok(dir) = self.open_path(&path)
992            && let Ok(iter) = dir.entries()
993        {
994            self.stack.push(Level::new(path, iter));
995            self.open_dirs += 1;
996        }
997    }
998
999    /// The directory `entry` names at `path`, or the mounted volume when `path`
1000    /// is a mount point.
1001    fn open_child(&self, path: &Path, entry: &DirEntry) -> std::io::Result<Dir> {
1002        match self.mounts.iter().find(|(rel, _)| rel == path) {
1003            Some((_, volume)) => volume.try_clone(),
1004            None => entry.open_dir(),
1005        }
1006    }
1007
1008    /// The directory at `path` below the base, through the volume mounted over
1009    /// it if any.
1010    fn open_path(&self, path: &Path) -> std::io::Result<Dir> {
1011        for (rel, volume) in &self.mounts {
1012            if let Ok(rest) = path.strip_prefix(rel) {
1013                if rest.as_os_str().is_empty() {
1014                    return volume.try_clone();
1015                }
1016                return volume.open_dir(rest);
1017            }
1018        }
1019        self.base.open_dir(path)
1020    }
1021}
1022
1023pub fn kind_of(ft: &cap_std::fs::FileType) -> &'static str {
1024    if ft.is_symlink() {
1025        "symlink"
1026    } else if ft.is_dir() {
1027        "directory"
1028    } else if ft.is_file() {
1029        "file"
1030    } else {
1031        "other"
1032    }
1033}
1034
1035impl TempFile {
1036    /// The temp file just created at `path` and opened as `file`.
1037    pub fn created(path: ContentPath, file: &cap_std::fs::File) -> std::io::Result<Self> {
1038        Ok(Self {
1039            path,
1040            identity: FileIdentity::of(&file.metadata()?)?,
1041        })
1042    }
1043
1044    fn is_intact(&self) -> Result<bool, ContainError> {
1045        self.path.refers_to(self.identity)
1046    }
1047}
1048
1049/// Why a write through a [`WriterHandle`] failed.
1050#[derive(Debug)]
1051enum WriteError {
1052    Io(std::io::Error),
1053    Contain(ContainError),
1054    Full(QuotaExceeded),
1055}
1056
1057impl std::fmt::Display for WriteError {
1058    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1059        match self {
1060            WriteError::Io(e) => e.fmt(f),
1061            WriteError::Contain(e) => e.fmt(f),
1062            WriteError::Full(e) => e.fmt(f),
1063        }
1064    }
1065}
1066
1067impl WriterHandle {
1068    fn new(
1069        file: File,
1070        temp: TempFile,
1071        final_path: ContentPath,
1072        quota: Option<Arc<DiskQuota>>,
1073    ) -> Self {
1074        let held = quota
1075            .as_ref()
1076            .map(|quota| quota.hold(temp.identity, Holder::Writer));
1077        Self {
1078            open: Some(OpenWrite {
1079                writer: BufWriter::new(file),
1080                disk_charge: QuotaCharge::new(quota, None),
1081                held,
1082            }),
1083            temp,
1084            final_path,
1085        }
1086    }
1087
1088    fn write_bytes(&mut self, bytes: &[u8]) -> Result<(), WriteError> {
1089        let open = self.reserve(bytes.len() as u64)?;
1090        open.writer.write_all(bytes).map_err(WriteError::Io)
1091    }
1092
1093    fn reserve(&mut self, bytes: u64) -> Result<&mut OpenWrite, WriteError> {
1094        let open = self
1095            .open
1096            .as_mut()
1097            .ok_or_else(|| WriteError::Io(std::io::Error::other("writer already closed")))?;
1098        open.disk_charge.reserve(bytes).map_err(WriteError::Full)?;
1099        Ok(open)
1100    }
1101
1102    /// Flush, fsync, and atomically rename temp→final. Idempotent: a second call after a
1103    /// successful close is a no-op.
1104    fn close(&mut self) -> Result<(), WriteError> {
1105        let Some(open) = self.open.take() else {
1106            return Ok(());
1107        };
1108        finalize(open, &self.temp, &self.final_path)
1109    }
1110}
1111
1112impl Drop for WriterHandle {
1113    fn drop(&mut self) {
1114        if let Some(open) = self.open.take() {
1115            // Discard the temp file rather than rename it: a GC-timed rename would be
1116            // non-deterministic.
1117            let unlinked = open
1118                .writer
1119                .into_inner()
1120                .ok()
1121                .is_some_and(|file| is_unlinked(&file));
1122            discard_temp(&self.temp, open.disk_charge, unlinked);
1123            drop(open.held);
1124        }
1125    }
1126}
1127
1128/// Whether `file` has no name left, so its data goes once every handle on it
1129/// closes.
1130fn is_unlinked(file: &File) -> bool {
1131    cap_std::fs::Metadata::from_file(file)
1132        .is_ok_and(|meta| crate::runtime::fs::link_count(&meta) == Some(0))
1133}
1134
1135/// Remove an uncommitted temp file and settle its disk charge. The program can reach
1136/// the temp file by name while the writer is open. If the name no longer refers to
1137/// the writer's file, it was moved or replaced: nothing is removed, and its bytes
1138/// stay counted while they may still be on disk under another name. Once the file
1139/// has no name at all, whether removed here or by the program, its bytes go when
1140/// the last handle on it closes.
1141fn discard_temp(temp: &TempFile, disk_charge: QuotaCharge, unlinked: bool) {
1142    let removed = temp.is_intact().unwrap_or(false) && temp.path.remove_file().is_ok();
1143    if removed || unlinked {
1144        disk_charge.release_when_closed(temp.identity);
1145    } else {
1146        disk_charge.keep();
1147    }
1148}
1149
1150/// Commit the temp file over the final one. Any failure discards the temp file.
1151fn finalize(open: OpenWrite, temp: &TempFile, final_path: &ContentPath) -> Result<(), WriteError> {
1152    let OpenWrite {
1153        writer,
1154        mut disk_charge,
1155        held,
1156    } = open;
1157    let committed = match flush_and_sync(writer) {
1158        Ok(unlinked) => {
1159            let renamed = rename_into_place(temp, final_path, &mut disk_charge);
1160            match renamed {
1161                Ok(()) => disk_charge.commit(),
1162                Err(_) => discard_temp(temp, disk_charge, unlinked),
1163            }
1164            renamed
1165        }
1166        Err(err) => {
1167            discard_temp(temp, disk_charge, false);
1168            Err(err)
1169        }
1170    };
1171    drop(held);
1172    committed
1173}
1174
1175/// Flush and sync the writer's file and close it, reporting whether the file has no
1176/// name left. Closing before the rename lets Windows handles release; unix `rename`
1177/// over an open file works either way.
1178fn flush_and_sync(writer: BufWriter<File>) -> Result<bool, WriteError> {
1179    let mut file = writer
1180        .into_inner()
1181        .map_err(|e| WriteError::Io(e.into_error()))?;
1182    file.flush().map_err(WriteError::Io)?;
1183    file.sync_all().map_err(WriteError::Io)?;
1184    Ok(is_unlinked(&file))
1185}
1186
1187/// Rename the temp file over the final one, first covering what it replaces.
1188fn rename_into_place(
1189    temp: &TempFile,
1190    final_path: &ContentPath,
1191    disk_charge: &mut QuotaCharge,
1192) -> Result<(), WriteError> {
1193    if !temp.is_intact().map_err(WriteError::Contain)? {
1194        return Err(WriteError::Io(std::io::Error::other(
1195            "the writer's temporary file was moved or replaced",
1196        )));
1197    }
1198    disk_charge
1199        .cover(final_path.regular_file())
1200        .map_err(WriteError::Full)?;
1201    temp.path.rename_to(final_path).map_err(WriteError::Contain)
1202}
1203
1204impl From<WriteError> for FilesystemError {
1205    fn from(error: WriteError) -> Self {
1206        match error {
1207            WriteError::Io(e) => Self::Io(e),
1208            WriteError::Contain(e) => Self::Containment(e),
1209            WriteError::Full(e) => Self::Quota(e),
1210        }
1211    }
1212}
1213
1214fn default_remove(
1215    vfs: &Vfs,
1216    path: &str,
1217    recursive: bool,
1218    budget: &mut dyn MutationBudget,
1219) -> wasmtime::Result<()> {
1220    let resolved =
1221        fs::resolve_link(vfs, vfs.cwd(), path).map_err(|e| contain_trap("fs.remove", path, &e))?;
1222    let context = MutationContext {
1223        op: "fs.remove",
1224        path,
1225        quota: resolved.placement().quota().cloned(),
1226        changed: false,
1227    };
1228    meter_mutation(budget, context, |step| {
1229        fs::removal::remove(&resolved, recursive, step)
1230    })
1231}
1232fn default_move(
1233    vfs: &Vfs,
1234    from: &str,
1235    to: &str,
1236    budget: &mut dyn MutationBudget,
1237) -> wasmtime::Result<()> {
1238    let from_resolved =
1239        fs::resolve_link(vfs, vfs.cwd(), from).map_err(|e| contain_trap("fs.move", from, &e))?;
1240    let to_resolved =
1241        fs::resolve_link(vfs, vfs.cwd(), to).map_err(|e| contain_trap("fs.move", to, &e))?;
1242    let pair = format!("{from} -> {to}");
1243    if from_resolved.is_root() || to_resolved.is_root() {
1244        wasmtime::bail!(
1245            "fs.move {pair}: the VFS root and mount points cannot be moved or replaced"
1246        );
1247    }
1248    if from_resolved
1249        .same_entry(&to_resolved)
1250        .map_err(|e| contain_trap("fs.move", &pair, &e))?
1251    {
1252        return Ok(());
1253    }
1254    if !from_resolved
1255        .placement()
1256        .same_volume(to_resolved.placement())
1257    {
1258        let mut work = CopyWork::default();
1259        let result = move_across(budget, &from_resolved, &to_resolved, &pair, &mut work);
1260        work.settle(budget)?;
1261        return result;
1262    }
1263    // A file moved over another frees the one it replaced; a move onto the same
1264    // file, under any spelling, frees nothing.
1265    let moved = from_resolved.regular_file().map(|(file, _)| file);
1266    let replaced = to_resolved
1267        .regular_file()
1268        .filter(|(file, _)| Some(*file) != moved);
1269    let context = MutationContext {
1270        op: "fs.move",
1271        path: &pair,
1272        quota: None,
1273        changed: false,
1274    };
1275    meter_mutation(budget, context, |step| {
1276        crate::runtime::fs::removal::rename(&from_resolved, &to_resolved, step)
1277    })?;
1278    if let (Some(quota), Some((file, bytes))) = (to_resolved.placement().quota(), replaced) {
1279        quota.release_file(file, bytes);
1280    }
1281    Ok(())
1282}
1283fn default_copy(
1284    vfs: &Vfs,
1285    from: &str,
1286    to: &str,
1287    recursive: bool,
1288    budget: &mut dyn MutationBudget,
1289) -> wasmtime::Result<()> {
1290    let from_resolved =
1291        fs::resolve_link(vfs, vfs.cwd(), from).map_err(|e| contain_trap("fs.copy", from, &e))?;
1292    let pair = format!("{from} -> {to}");
1293    let mount = from_resolved
1294        .contains_mount_point()
1295        .map_err(|e| contain_trap("fs.copy", &pair, &e))?;
1296    if let Some(mount) = mount {
1297        wasmtime::bail!(
1298            "fs.copy {pair}: {from} contains the mount point {mount}; copy the \
1299                     mount's contents separately"
1300        );
1301    }
1302    // The content path routes to the same volume as the link path, and resolves
1303    // even when the destination's parent does not exist yet.
1304    let to_volume =
1305        fs::resolve_content(vfs, vfs.cwd(), to).map_err(|e| contain_trap("fs.copy", to, &e))?;
1306    from_resolved
1307        .check_copy_destination(&to_volume)
1308        .map_err(|e| contain_trap("fs.copy", &pair, &e))?;
1309    // A recursive copy used to reach `create_dir_all(to)`, which built the whole
1310    // destination chain. Resolving the destination now opens its parent, so that
1311    // chain has to exist first or a working call starts failing.
1312    if recursive
1313        && from_resolved
1314            .symlink_metadata()
1315            .is_ok_and(|meta| meta.is_dir())
1316    {
1317        to_volume
1318            .create_dir_all()
1319            .map_err(|e| write_target_trap("fs.copy", to, &e))?;
1320    }
1321    let to_resolved =
1322        fs::resolve_link(vfs, vfs.cwd(), to).map_err(|e| contain_trap("fs.copy", to, &e))?;
1323    let quota = to_resolved.placement().quota().cloned();
1324    let mut work = CopyWork::default();
1325    let result = copy_bounded(
1326        &from_resolved,
1327        &to_resolved,
1328        &CopyRun {
1329            recursive,
1330            links: LinkCopies::BestEffort,
1331            op: "fs.copy",
1332            pair: &pair,
1333            quota: quota.as_ref(),
1334        },
1335        &mut work,
1336    );
1337    work.settle(budget)?;
1338    result
1339}
1340/// Reproduce a symlink at `to` with `from`'s target stored verbatim.
1341///
1342/// Verbatim is the point: the target is never resolved, so a link that escapes the root is
1343/// reproduced as one that still refuses rather than as a working door out of it.
1344fn copy_link(from: &LinkPath, to: &LinkPath) -> Result<(), ContainError> {
1345    to.symlink(&from.read_link_contents()?, from.link_kind()?)
1346}
1347
1348/// A move between two volumes, which no rename can make: copy into a temporary
1349/// sibling of the destination (charged to the destination's limit), rename it into
1350/// place, then remove the source (released from the source's limit).
1351///
1352/// Until the rename, a failure removes the temporary copy and leaves the source
1353/// untouched. Once the destination is in place it is kept; if the source then
1354/// cannot be removed, the call fails saying both copies may exist. A link the copy
1355/// cannot reproduce fails the move, since the source is removed after.
1356fn move_across(
1357    budget: &mut dyn MutationBudget,
1358    from: &LinkPath,
1359    to: &LinkPath,
1360    pair: &str,
1361    work: &mut CopyWork,
1362) -> wasmtime::Result<()> {
1363    let meta = from
1364        .symlink_metadata()
1365        .map_err(|e| contain_trap("fs.move", pair, &e))?;
1366    let is_dir = meta.is_dir();
1367    // The source is removed last, so a tree too deep or too large to remove, or
1368    // one holding a mount point, is refused now, before a copy of it lands in the
1369    // destination. The destination gets the checks its rename will run, so a
1370    // move that must fail does no copying first.
1371    meter_mutation(
1372        budget,
1373        MutationContext {
1374            op: "fs.move",
1375            path: pair,
1376            quota: None,
1377            changed: false,
1378        },
1379        |step| {
1380            crate::runtime::fs::removal::validate(from, &mut *step)?;
1381            crate::runtime::fs::removal::validate(to, step)
1382        },
1383    )?;
1384    refuse_unfitting_destination(to, is_dir, pair)?;
1385    let to_quota = to.placement().quota().cloned();
1386    let staged = to.temp_sibling();
1387    let run = CopyRun {
1388        recursive: true,
1389        links: LinkCopies::Required,
1390        op: "fs.move",
1391        pair,
1392        quota: to_quota.as_ref(),
1393    };
1394    let copied = copy_bounded(from, &staged, &run, work).and_then(|()| {
1395        let replaced = to.regular_file();
1396        meter_mutation(
1397            budget,
1398            MutationContext {
1399                op: "fs.move",
1400                path: pair,
1401                quota: None,
1402                changed: true,
1403            },
1404            |step| crate::runtime::fs::removal::rename(&staged, to, step),
1405        )?;
1406        if let (Some(quota), Some((file, bytes))) = (to_quota.as_ref(), replaced) {
1407            quota.release_file(file, bytes);
1408        }
1409        Ok(())
1410    });
1411    if let Err(err) = copied {
1412        // Best effort: what cannot be removed stays charged, so the count errs
1413        // toward refusing a later write.
1414        let _ = meter_mutation(
1415            budget,
1416            MutationContext {
1417                op: "fs.move",
1418                path: pair,
1419                quota: to_quota,
1420                changed: true,
1421            },
1422            |step| crate::runtime::fs::removal::remove(&staged, true, step),
1423        );
1424        return Err(err);
1425    }
1426    meter_mutation(
1427        budget,
1428        MutationContext {
1429            op: "fs.move",
1430            path: pair,
1431            quota: from.placement().quota().cloned(),
1432            changed: true,
1433        },
1434        |step| {
1435            crate::runtime::fs::removal::remove(from, true, step).map_err(|error| {
1436            if crate::runtime::host::ends_the_run(&error) { return error; }
1437            wasmtime::Error::msg(format!(
1438                "fs.move {pair}: moved to the destination but could not remove the source, so both may now exist: {error}"
1439            ))
1440        })
1441        },
1442    )
1443}
1444
1445/// Refuse a destination a rename of the source could not replace — a directory
1446/// in place of a file, a file in place of a directory, or a directory that isn't
1447/// empty — as a rename within one volume would, before any copying.
1448fn refuse_unfitting_destination(
1449    to: &LinkPath,
1450    moving_dir: bool,
1451    pair: &str,
1452) -> wasmtime::Result<()> {
1453    let existing = match to.symlink_metadata() {
1454        Ok(existing) => existing,
1455        Err(ContainError::Io(err)) if err.kind() == std::io::ErrorKind::NotFound => {
1456            return Ok(());
1457        }
1458        Err(err) => return Err(contain_trap("fs.move", pair, &err)),
1459    };
1460    let replacing_dir = existing.is_dir();
1461    if replacing_dir != moving_dir {
1462        let (what, onto) = if moving_dir {
1463            ("a directory", "something that is not a directory")
1464        } else {
1465            ("a file", "a directory")
1466        };
1467        wasmtime::bail!("fs.move {pair}: cannot move {what} onto {onto}");
1468    }
1469    if !replacing_dir {
1470        return Ok(());
1471    }
1472    let has_entries = to
1473        .entries()
1474        .map_err(|e| contain_trap("fs.move", pair, &e))?
1475        .next()
1476        .is_some();
1477    if has_entries {
1478        wasmtime::bail!("fs.move {pair}: the destination directory is not empty");
1479    }
1480    Ok(())
1481}
1482
1483struct MutationContext<'a> {
1484    op: &'static str,
1485    path: &'a str,
1486    quota: Option<Arc<DiskQuota>>,
1487    changed: bool,
1488}
1489
1490fn meter_mutation(
1491    budget: &mut dyn MutationBudget,
1492    context: MutationContext<'_>,
1493    run: impl FnOnce(
1494        &mut dyn FnMut(crate::runtime::fs::removal::Step) -> wasmtime::Result<()>,
1495    ) -> wasmtime::Result<()>,
1496) -> wasmtime::Result<()> {
1497    use crate::runtime::fs::removal::Step;
1498    budget.begin()?;
1499    let mut changed = context.changed;
1500    let result = run(&mut |step| {
1501        match step {
1502            Step::Inspect(units) | Step::Mutate(units) if changed => {
1503                budget.syscalls(units, true)?;
1504            }
1505            Step::Inspect(units) | Step::Mutate(units) => {
1506                budget.syscalls(units, false)?;
1507            }
1508            Step::Reserve(bytes) => budget.reserve(bytes)?,
1509            Step::Unlinked(file) => {
1510                changed = true;
1511                budget.effect();
1512                if let (Some(quota), Some((file, bytes))) = (&context.quota, file) {
1513                    quota.release_file(file, bytes);
1514                }
1515            }
1516        }
1517        Ok(())
1518    });
1519    budget.end();
1520    result.map_err(|error| match error.downcast::<ContainError>() {
1521        Ok(error) => contain_trap(context.op, context.path, &error),
1522        Err(error) => error,
1523    })
1524}
1525
1526/// How one copy walk treats its tree.
1527#[derive(Clone, Copy)]
1528struct CopyRun<'a> {
1529    recursive: bool,
1530    links: LinkCopies,
1531    /// `fs.copy` or `fs.move`, for diagnostics.
1532    op: &'a str,
1533    /// `from -> to` as the program wrote them, for diagnostics.
1534    pair: &'a str,
1535    /// The destination volume's limit.
1536    quota: Option<&'a Arc<DiskQuota>>,
1537}
1538
1539/// What a copy does with a link inside the tree that the platform will not
1540/// reproduce, such as an absolute one.
1541#[derive(Clone, Copy, PartialEq, Eq)]
1542enum LinkCopies {
1543    /// Skip it: a copy leaves the source in place, so nothing is lost.
1544    BestEffort,
1545    /// Fail: a move removes the source afterwards, which would lose the link.
1546    Required,
1547}
1548
1549/// Matches the entry-count limit used by removal.
1550const MAX_COPY_ENTRIES: u64 = MAX_REMOVE_ENTRIES as u64;
1551const MAX_COPY_DEPTH: usize = 64;
1552
1553#[derive(Default)]
1554struct CopyWork {
1555    entries: u64,
1556    bytes: u64,
1557}
1558
1559impl CopyWork {
1560    fn visit(&mut self, op: &str, pair: &str) -> wasmtime::Result<()> {
1561        if self.entries >= MAX_COPY_ENTRIES {
1562            return Err(crate::runtime::host::range_error(format!(
1563                "{op} {pair}: exceeds {MAX_COPY_ENTRIES} entries; copy smaller subtrees"
1564            )));
1565        }
1566        self.entries += 1;
1567        Ok(())
1568    }
1569
1570    fn settle(&self, budget: &mut dyn MutationBudget) -> wasmtime::Result<()> {
1571        budget.copied(self.entries, self.bytes)
1572    }
1573}
1574
1575struct CopyFrame {
1576    from: LinkPath,
1577    to: LinkPath,
1578    from_dir: Arc<cap_std::fs::Dir>,
1579    to_dir: Arc<cap_std::fs::Dir>,
1580    entries: cap_std::fs::ReadDir,
1581}
1582
1583/// Depth-first traversal retains directory iterators instead of recursing or
1584/// collecting every sibling path. Symlinks are copied as links, never followed.
1585fn copy_bounded(
1586    from: &LinkPath,
1587    to: &LinkPath,
1588    run: &CopyRun<'_>,
1589    work: &mut CopyWork,
1590) -> wasmtime::Result<()> {
1591    let mut frames = Vec::new();
1592    if let Some(frame) = copy_entry(from, to, run, work, 0)? {
1593        frames
1594            .try_reserve(1)
1595            .map_err(crate::runtime::host::fatal_host_error)?;
1596        frames.push(frame);
1597    }
1598    while let Some(frame) = frames.last_mut() {
1599        let Some(entry) = frame.entries.next() else {
1600            frames.pop();
1601            continue;
1602        };
1603        let entry = entry.map_err(|e| contain_trap(run.op, run.pair, &ContainError::from(e)))?;
1604        let name = entry.file_name();
1605        let from = frame.from.child(Arc::clone(&frame.from_dir), name.clone());
1606        let to = frame.to.child(Arc::clone(&frame.to_dir), name);
1607        if let Some(child) = copy_entry(&from, &to, run, work, frames.len())? {
1608            frames
1609                .try_reserve(1)
1610                .map_err(crate::runtime::host::fatal_host_error)?;
1611            frames.push(child);
1612        }
1613    }
1614    Ok(())
1615}
1616
1617fn copy_entry(
1618    from: &LinkPath,
1619    to: &LinkPath,
1620    run: &CopyRun<'_>,
1621    work: &mut CopyWork,
1622    depth: usize,
1623) -> wasmtime::Result<Option<CopyFrame>> {
1624    let CopyRun {
1625        recursive,
1626        links,
1627        op,
1628        pair,
1629        quota,
1630    } = *run;
1631    work.visit(op, pair)?;
1632    let meta = from
1633        .symlink_metadata()
1634        .map_err(|e| contain_trap(op, pair, &e))?;
1635    let ft = meta.file_type();
1636    if ft.is_symlink() {
1637        let result = copy_link(from, to);
1638        // Only nested unrepresentable links are best-effort. Root failures and
1639        // every move failure retain their existing error behavior.
1640        if depth == 0 || links == LinkCopies::Required {
1641            result.map_err(|e| contain_trap(op, pair, &e))?;
1642        }
1643        return Ok(None);
1644    }
1645    if ft.is_dir() {
1646        if !recursive {
1647            wasmtime::bail!("{op} {pair}: source is a directory (use {{ recursive: true }})");
1648        }
1649        if depth >= MAX_COPY_DEPTH {
1650            return Err(crate::runtime::host::range_error(format!(
1651                "{op} {pair}: exceeds {MAX_COPY_DEPTH} directory levels; copy smaller subtrees"
1652            )));
1653        }
1654        to.create_dir_all()
1655            .map_err(|e| contain_trap(op, pair, &e))?;
1656        let from_dir = Arc::new(from.open_dir().map_err(|e| contain_trap(op, pair, &e))?);
1657        let to_dir = Arc::new(to.open_dir().map_err(|e| contain_trap(op, pair, &e))?);
1658        let entries = from_dir
1659            .entries()
1660            .map_err(|e| contain_trap(op, pair, &ContainError::from(e)))?;
1661        return Ok(Some(CopyFrame {
1662            from: from.clone(),
1663            to: to.clone(),
1664            from_dir,
1665            to_dir,
1666            entries,
1667        }));
1668    }
1669    // A copy may leave earlier files in place on error. Preserve quota claims
1670    // for whatever landed, and settle known completed bytes on either outcome.
1671    let mut disk_charge = QuotaCharge::in_place(quota.cloned(), to.copied_onto_file());
1672    disk_charge
1673        .reserve(meta.len())
1674        .map_err(|exceeded| quota_refusal(op, pair, exceeded))?;
1675    match from.copy_to_counted(to) {
1676        Ok(bytes) => {
1677            work.bytes = work.bytes.saturating_add(bytes);
1678            disk_charge.commit();
1679        }
1680        Err(e) => {
1681            disk_charge.settle_at(to.copied_onto_file().map_or(0, |(_, bytes)| bytes));
1682            return Err(contain_trap(op, pair, &e));
1683        }
1684    }
1685    Ok(None)
1686}
1687
1688#[cfg(test)]
1689mod tests {
1690    use super::*;
1691
1692    #[test]
1693    fn code_io_waits_for_the_blocking_pool() {
1694        single_worker_runtime().block_on(async {
1695            let root = tempfile::tempdir().unwrap();
1696            std::fs::write(root.path().join("input"), b"abc").unwrap();
1697            let filesystem = DefaultFilesystem::new(
1698                Vfs::external(root.path().to_owned()).unwrap(),
1699                "run".into(),
1700            );
1701            let limits = TenantLimits::new(1024);
1702            queued(filesystem.placement("/input")).await.unwrap();
1703            let contents = queued(filesystem.read_regular(None, "/input", 10, &limits))
1704                .await
1705                .unwrap();
1706            assert_eq!(contents.bytes, b"abc");
1707            assert_eq!(limits.host_attached_bytes(), 4);
1708            drop(contents);
1709            let names = queued(filesystem.directory_names(None, "/", 10, &limits))
1710                .await
1711                .unwrap();
1712            assert_eq!(names.names, ["input"]);
1713            drop(names);
1714            let effect = "stat".into();
1715            for follow in [true, false] {
1716                let metadata = queued(filesystem.metadata(&effect, None, "/input", follow))
1717                    .await
1718                    .unwrap();
1719                assert_eq!(metadata.len(), 3);
1720            }
1721            let update = filesystem
1722                .begin_file_update(&"update".into(), "/input", [0; 32], &limits)
1723                .await
1724                .unwrap();
1725            let prepared = PreparedFileUpdate {
1726                contents: Some(b"changed".to_vec()),
1727                output: FileUpdateOutput {
1728                    bytes: Vec::new(),
1729                    memory: HostBytes::new(&limits, 0).unwrap(),
1730                },
1731            };
1732            queued(update.publish(Arc::new(WorkBudget::new(u64::MAX)), Some(prepared), &limits))
1733                .await
1734                .unwrap();
1735            assert_eq!(
1736                std::fs::read(root.path().join("input")).unwrap(),
1737                b"changed"
1738            );
1739            assert_eq!(limits.host_attached_bytes(), 0);
1740            assert!(limits.peak_bytes() > 0);
1741        });
1742    }
1743
1744    fn single_worker_runtime() -> tokio::runtime::Runtime {
1745        tokio::runtime::Builder::new_current_thread()
1746            .enable_all()
1747            .max_blocking_threads(1)
1748            .build()
1749            .unwrap()
1750    }
1751    async fn queued<F: std::future::Future>(future: F) -> F::Output {
1752        let (release, wait) = tokio::sync::oneshot::channel();
1753        let blocker = tokio::task::spawn_blocking(move || wait.blocking_recv());
1754        let mut future = Box::pin(future);
1755        assert_pending(future.as_mut());
1756        release.send(()).unwrap();
1757        blocker.await.unwrap().unwrap();
1758        future.await
1759    }
1760    fn assert_pending<F: std::future::Future>(future: std::pin::Pin<&mut F>) {
1761        let mut context = std::task::Context::from_waker(std::task::Waker::noop());
1762        assert!(
1763            future.poll(&mut context).is_pending(),
1764            "filesystem I/O ran while blocking pool was occupied"
1765        );
1766    }
1767
1768    #[tokio::test]
1769    async fn resource_serialization_keeps_identity_without_local_handles() {
1770        use super::*;
1771        use crate::runtime::limits::TenantLimits;
1772
1773        let root = tempfile::tempdir().unwrap();
1774        std::fs::write(root.path().join("input"), b"abc").unwrap();
1775        let data = crate::runtime::StoreData::with_vfs(
1776            "run".into(),
1777            Vfs::external(root.path().to_owned()).unwrap(),
1778        );
1779        let filesystem = data.filesystem;
1780        let limits = TenantLimits::new(1 << 20);
1781        let mut reader = filesystem
1782            .open_reader(
1783                &"open-reader".into(),
1784                None,
1785                "/input",
1786                ReaderOptions::Bytes { chunk_size: 3 },
1787            )
1788            .await
1789            .unwrap();
1790        let mut writer = filesystem
1791            .open_writer(&"open-writer".into(), None, "/output")
1792            .await
1793            .unwrap();
1794        let mut directory = filesystem
1795            .open_directory(&"open-directory".into(), None, "/", false, &limits)
1796            .await
1797            .unwrap();
1798
1799        // Encoding live resources neither includes nor closes their OS handles.
1800        let encoded_reader = serde_json::to_value(&reader).unwrap();
1801        let encoded_writer = serde_json::to_value(&writer).unwrap();
1802        let encoded_directory = serde_json::to_value(&directory).unwrap();
1803        for encoded in [&encoded_reader, &encoded_writer, &encoded_directory] {
1804            assert_eq!(encoded["run_id"], "run");
1805            assert!(encoded.get("local").is_none());
1806        }
1807        let restored_reader: FileReader = serde_json::from_value(encoded_reader).unwrap();
1808        let restored_writer: FileWriter = serde_json::from_value(encoded_writer).unwrap();
1809        let restored_directory: DirectoryReader =
1810            serde_json::from_value(encoded_directory).unwrap();
1811        assert!(restored_reader.local.is_none());
1812        assert!(restored_writer.local.is_none());
1813        assert!(restored_directory.local.is_none());
1814        assert_eq!(restored_reader.id, reader.id);
1815        assert_eq!(restored_writer.id, writer.id);
1816        assert_eq!(restored_directory.id, directory.id);
1817        drop((restored_reader, restored_writer, restored_directory));
1818
1819        let output = filesystem
1820            .next_reader(&"read".into(), None, &mut reader, &limits)
1821            .await
1822            .unwrap();
1823        assert!(matches!(output.item, Some(ReadItem::Bytes(bytes)) if bytes == b"abc"));
1824        assert!(
1825            filesystem
1826                .next_directory(&"next".into(), None, &mut directory, &limits)
1827                .await
1828                .unwrap()
1829                .is_some()
1830        );
1831        filesystem
1832            .write_chunk(&"write".into(), None, &mut writer, b"saved".to_vec())
1833            .await
1834            .unwrap();
1835        filesystem
1836            .close_writer(&"close".into(), None, &mut writer)
1837            .await
1838            .unwrap();
1839        assert_eq!(std::fs::read(root.path().join("output")).unwrap(), b"saved");
1840    }
1841
1842    #[test]
1843    fn logical_lines_bound_memory_and_restore_without_buffered_read_ahead() {
1844        use super::*;
1845        use crate::runtime::limits::TenantLimits;
1846        let limits = TenantLimits::new(1 << 20);
1847        let work = Arc::new(WorkBudget::new(u64::MAX));
1848        let mut file = std::io::Cursor::new(b"\xef\xbb\xbfhi\r\n\xff\n");
1849        let options = ReaderOptions::Lines { maximum_bytes: 32 };
1850        let mut position = ReadPosition::default();
1851        let output = read_item(
1852            &mut file,
1853            &options,
1854            &mut position,
1855            Some(work.as_ref()),
1856            &limits,
1857        )
1858        .unwrap();
1859        assert!(matches!(&output.item, Some(ReadItem::Line(line)) if line == "hi"));
1860        assert_eq!(position.offset, 7);
1861        assert!(limits.host_attached_bytes() > 0);
1862        drop(output);
1863        let output = read_item(
1864            &mut file,
1865            &options,
1866            &mut position,
1867            Some(work.as_ref()),
1868            &limits,
1869        )
1870        .unwrap();
1871        assert!(matches!(&output.item, Some(ReadItem::Line(line)) if line == "\u{fffd}"));
1872        drop(output);
1873        assert!(
1874            read_item(
1875                &mut file,
1876                &options,
1877                &mut position,
1878                Some(work.as_ref()),
1879                &limits
1880            )
1881            .unwrap()
1882            .item
1883            .is_none()
1884        );
1885        assert_eq!(limits.host_attached_bytes(), 0);
1886    }
1887
1888    #[test]
1889    fn line_limit_and_memory_refusal_leave_progress_retryable() {
1890        use super::*;
1891        use crate::runtime::limits::TenantLimits;
1892        let limits = TenantLimits::new(1 << 20);
1893        let mut position = ReadPosition::default();
1894        let mut file = std::io::Cursor::new(b"abcde");
1895        let options = ReaderOptions::Lines { maximum_bytes: 4 };
1896        for _ in 0..2 {
1897            let error = read_item(&mut file, &options, &mut position, None, &limits)
1898                .err()
1899                .unwrap();
1900            assert!(error.to_string().contains("maxReadSize"));
1901            assert_eq!(position.offset, 0);
1902            assert_eq!(limits.host_attached_bytes(), 0);
1903        }
1904        let limits = TenantLimits::new(8192 + 4);
1905        let mut file = std::io::Cursor::new(b"abcd");
1906        let error = read_item(&mut file, &options, &mut position, None, &limits)
1907            .err()
1908            .unwrap();
1909        assert!(
1910            matches!(error, FilesystemError::Execution(ref e) if e.is::<crate::runtime::limits::MemoryCapExceeded>())
1911        );
1912        assert_eq!(position.offset, 0);
1913        assert_eq!(limits.host_attached_bytes(), 0);
1914    }
1915
1916    #[test]
1917    fn read_failures_refund_buffers_and_charge_consumed_io_without_advancing_progress() {
1918        use super::*;
1919        use crate::runtime::limits::TenantLimits;
1920        struct FailsAfterOne(std::io::Cursor<[u8; 2]>);
1921        impl Read for FailsAfterOne {
1922            fn read(&mut self, bytes: &mut [u8]) -> std::io::Result<usize> {
1923                if self.0.position() > 0 {
1924                    return Err(std::io::Error::other("injected read failure"));
1925                }
1926                self.0.read(&mut bytes[..1])
1927            }
1928        }
1929        impl Seek for FailsAfterOne {
1930            fn seek(&mut self, position: SeekFrom) -> std::io::Result<u64> {
1931                self.0.seek(position)
1932            }
1933        }
1934        let limits = TenantLimits::new(1 << 20);
1935        let work = Arc::new(crate::runtime::fuel::WorkBudget::new(u64::MAX));
1936        let mut file = FailsAfterOne(std::io::Cursor::new(*b"ab"));
1937        let mut position = ReadPosition::default();
1938        assert!(
1939            read_item(
1940                &mut file,
1941                &ReaderOptions::Bytes { chunk_size: 2 },
1942                &mut position,
1943                Some(work.as_ref()),
1944                &limits
1945            )
1946            .is_err()
1947        );
1948        assert_eq!(position.offset, 0);
1949        assert_eq!(limits.host_attached_bytes(), 0);
1950        assert!(work.spent() >= crate::runtime::fuel::IO.cost(1));
1951    }
1952
1953    #[test]
1954    fn unknown_read_cost_settles_after_read_even_when_fuel_is_spent() {
1955        use super::*;
1956        let limits = crate::runtime::limits::TenantLimits::new(1 << 20);
1957        let work = Arc::new(crate::runtime::fuel::WorkBudget::new(0));
1958        let mut file = std::io::Cursor::new(b"line\n");
1959        let mut position = ReadPosition::default();
1960        assert!(
1961            read_item(
1962                &mut file,
1963                &ReaderOptions::Lines { maximum_bytes: 32 },
1964                &mut position,
1965                Some(work.as_ref()),
1966                &limits
1967            )
1968            .unwrap()
1969            .item
1970            .is_some()
1971        );
1972        assert_eq!(position.offset, 5);
1973        assert!(work.spent() > 0);
1974    }
1975
1976    use crate::runtime::fs::resolve_link;
1977    #[test]
1978    fn copy_entry_and_depth_limits_preserve_completed_work() {
1979        let root = tempfile::tempdir().unwrap();
1980        std::fs::create_dir(root.path().join("source")).unwrap();
1981        std::fs::write(root.path().join("source/file"), "content").unwrap();
1982        let vfs = Vfs::external(root.path().to_path_buf()).unwrap();
1983        let from = resolve_link(&vfs, "/", "/source").unwrap();
1984        let to = resolve_link(&vfs, "/", "/target").unwrap();
1985        let run = CopyRun {
1986            recursive: true,
1987            links: LinkCopies::Required,
1988            op: "fs.copy",
1989            pair: "source -> target",
1990            quota: None,
1991        };
1992        let mut work = CopyWork {
1993            entries: MAX_COPY_ENTRIES - 1,
1994            bytes: 0,
1995        };
1996        let error = copy_bounded(&from, &to, &run, &mut work).unwrap_err();
1997        assert!(error.to_string().contains("entries"));
1998        assert!(root.path().join("target").is_dir());
1999        assert!(!root.path().join("target/file").exists());
2000        assert_eq!(
2001            std::fs::read(root.path().join("source/file")).unwrap(),
2002            b"content"
2003        );
2004        let to = resolve_link(&vfs, "/", "/too-deep").unwrap();
2005        let error = copy_entry(&from, &to, &run, &mut CopyWork::default(), MAX_COPY_DEPTH)
2006            .err()
2007            .unwrap();
2008        assert!(error.to_string().contains("directory levels"));
2009        assert!(!root.path().join("too-deep").exists());
2010    }
2011
2012    /// Builds `root/<branch>` as a chain `l1/l2/…/l{depth}` with an `f.txt` in every
2013    /// directory. Returns every path the walk should yield, relative to `root`.
2014    fn build_chain(root: &std::path::Path, branch: &str, depth: usize) -> Vec<String> {
2015        let mut path = root.join(branch);
2016        let mut rel = branch.to_string();
2017        std::fs::create_dir_all(&path).unwrap();
2018        std::fs::write(path.join("f.txt"), "x").unwrap();
2019        let mut expected = vec![rel.clone(), format!("{rel}/f.txt")];
2020        for level in 1..=depth {
2021            path = path.join(format!("l{level}"));
2022            rel = format!("{rel}/l{level}");
2023            std::fs::create_dir(&path).unwrap();
2024            std::fs::write(path.join("f.txt"), "x").unwrap();
2025            expected.push(rel.clone());
2026            expected.push(format!("{rel}/f.txt"));
2027        }
2028        expected
2029    }
2030
2031    fn walk_over(root: &std::path::Path, listed: &str, recursive: bool) -> ContainedWalk {
2032        let vfs = crate::runtime::Vfs::external(root.to_path_buf()).unwrap();
2033        let resolved = crate::runtime::fs::resolve_content(&vfs, "/", listed).unwrap();
2034        let base = Arc::new(resolved.open_dir().unwrap());
2035        let base_prefix = if listed == "/" {
2036            String::new()
2037        } else {
2038            format!("{}/", listed.trim_start_matches('/'))
2039        };
2040        ContainedWalk::new(base, base_prefix, recursive, Vec::new()).unwrap()
2041    }
2042
2043    fn walk_all(root: &std::path::Path, listed: &str) -> Vec<DirectoryEntry> {
2044        let mut walk = walk_over(root, listed, true);
2045        std::iter::from_fn(|| walk.next_entry()).collect()
2046    }
2047
2048    /// Every subtree is one consecutive run: for each directory, its descendants are
2049    /// yielded after it with nothing from outside the subtree interleaved.
2050    fn assert_subtrees_contiguous(entries: &[DirectoryEntry]) {
2051        for (i, dir) in entries.iter().enumerate() {
2052            if dir.kind != "directory" {
2053                continue;
2054            }
2055            let prefix = format!("{}/", dir.path);
2056            let inside: Vec<usize> = entries
2057                .iter()
2058                .enumerate()
2059                .filter(|(_, e)| e.path.starts_with(&prefix))
2060                .map(|(j, _)| j)
2061                .collect();
2062            let (Some(&first), Some(&last)) = (inside.first(), inside.last()) else {
2063                continue;
2064            };
2065            assert!(
2066                first > i,
2067                "{} yielded before its parent",
2068                entries[first].path
2069            );
2070            assert_eq!(
2071                last - first + 1,
2072                inside.len(),
2073                "entries from outside {} are interleaved into its subtree",
2074                dir.path,
2075            );
2076        }
2077    }
2078
2079    #[test]
2080    fn deep_tree_yields_every_entry_exactly_once() {
2081        let dir = tempfile::tempdir().unwrap();
2082        let mut expected = build_chain(dir.path(), "a", MAX_OPEN_DIRS + 5);
2083        let mut got: Vec<String> = walk_all(dir.path(), "/")
2084            .into_iter()
2085            .map(|e| e.path.trim_start_matches('/').to_string())
2086            .collect();
2087        got.sort();
2088        expected.sort();
2089        assert_eq!(
2090            got, expected,
2091            "deep walk must yield every entry exactly once"
2092        );
2093    }
2094
2095    /// Past the open-handle cap the queue holds bare names, rejoined to the level's
2096    /// own path when the descent resumes. A wide level down there is where a mistake
2097    /// in that rejoin shows up — either as a doubled prefix or as a lost subtree.
2098    #[test]
2099    fn a_wide_level_past_the_open_dir_cap_walks_completely() {
2100        let dir = tempfile::tempdir().unwrap();
2101        let mut expected = build_chain(dir.path(), "a", MAX_OPEN_DIRS + 2);
2102        let deep = expected
2103            .iter()
2104            .filter(|p| !p.ends_with("f.txt"))
2105            .max_by_key(|p| p.matches('/').count())
2106            .expect("the deepest directory")
2107            .clone();
2108        for i in 0..64 {
2109            let child = format!("{deep}/w{i}");
2110            std::fs::create_dir(dir.path().join(&child)).unwrap();
2111            std::fs::write(dir.path().join(&child).join("f.txt"), "x").unwrap();
2112            expected.push(child.clone());
2113            expected.push(format!("{child}/f.txt"));
2114        }
2115
2116        let mut got: Vec<String> = walk_all(dir.path(), "/")
2117            .into_iter()
2118            .map(|e| e.path.trim_start_matches('/').to_string())
2119            .collect();
2120        got.sort();
2121        expected.sort();
2122        assert_eq!(
2123            got, expected,
2124            "every postponed subtree must still be walked"
2125        );
2126    }
2127
2128    #[test]
2129    fn deep_tree_under_a_nested_base_keeps_the_base_prefix() {
2130        let dir = tempfile::tempdir().unwrap();
2131        let expected: Vec<String> = build_chain(dir.path(), "a", MAX_OPEN_DIRS + 5)
2132            .into_iter()
2133            .filter(|p| p != "a")
2134            .map(|p| format!("/{p}"))
2135            .collect();
2136        let mut got: Vec<String> = walk_all(dir.path(), "/a")
2137            .into_iter()
2138            .map(|e| e.path)
2139            .collect();
2140        let mut expected = expected;
2141        got.sort();
2142        expected.sort();
2143        assert_eq!(
2144            got, expected,
2145            "guest paths keep the listed directory's prefix"
2146        );
2147    }
2148
2149    #[test]
2150    fn deep_subtrees_stay_contiguous_past_the_open_dir_cap() {
2151        let dir = tempfile::tempdir().unwrap();
2152        build_chain(dir.path(), "a", MAX_OPEN_DIRS + 5);
2153        build_chain(dir.path(), "b", MAX_OPEN_DIRS + 5);
2154        assert_subtrees_contiguous(&walk_all(dir.path(), "/"));
2155    }
2156
2157    #[test]
2158    fn open_directory_handles_stay_within_the_cap() {
2159        let dir = tempfile::tempdir().unwrap();
2160        build_chain(dir.path(), "a", MAX_OPEN_DIRS + 5);
2161        let mut walk = walk_over(dir.path(), "/", true);
2162        let mut deepest = 0;
2163        while walk.next_entry().is_some() {
2164            deepest = deepest.max(walk.open_dirs);
2165            assert!(
2166                walk.open_dirs <= MAX_OPEN_DIRS,
2167                "walk held {} directory handles",
2168                walk.open_dirs,
2169            );
2170        }
2171        assert_eq!(deepest, MAX_OPEN_DIRS, "a chain past the cap must reach it");
2172    }
2173
2174    #[cfg(unix)]
2175    #[test]
2176    fn a_subdirectory_that_would_not_open_is_retried_from_the_base() {
2177        use std::os::unix::fs::PermissionsExt;
2178
2179        let dir = tempfile::tempdir().unwrap();
2180        let blocked = dir.path().join("w");
2181        std::fs::create_dir(&blocked).unwrap();
2182        std::fs::write(blocked.join("inside.txt"), "x").unwrap();
2183        std::fs::write(dir.path().join("sibling.txt"), "x").unwrap();
2184        std::fs::set_permissions(&blocked, std::fs::Permissions::from_mode(0o000)).unwrap();
2185        if std::fs::read_dir(&blocked).is_ok() {
2186            eprintln!("skipping: this user can read a 0o000 directory");
2187            return;
2188        }
2189
2190        let mut walk = walk_over(dir.path(), "/", true);
2191        let mut paths = Vec::new();
2192        while let Some(entry) = walk.next_entry() {
2193            paths.push(entry.path);
2194            // The descent into `/w` has already failed by the time it is yielded; the walk
2195            // must retry it when it drains, not drop the subtree.
2196            if entry.name == "w" {
2197                std::fs::set_permissions(&blocked, std::fs::Permissions::from_mode(0o755)).unwrap();
2198            }
2199        }
2200        assert!(
2201            paths.iter().any(|p| p == "/w/inside.txt"),
2202            "a subdirectory that failed to open must be retried, got {paths:?}",
2203        );
2204    }
2205
2206    #[cfg(unix)]
2207    #[test]
2208    fn a_deferred_directory_with_a_non_utf8_name_still_walks() {
2209        use std::os::unix::ffi::OsStrExt;
2210
2211        let dir = tempfile::tempdir().unwrap();
2212        build_chain(dir.path(), "a", MAX_OPEN_DIRS + 3);
2213        let mut deepest = dir.path().join("a");
2214        for level in 1..=(MAX_OPEN_DIRS + 3) {
2215            deepest = deepest.join(format!("l{level}"));
2216        }
2217        let odd = deepest.join(std::ffi::OsStr::from_bytes(b"bad\xffname"));
2218        // APFS (and any other filesystem enforcing UTF-8 names) rejects this outright,
2219        // which makes the case unreachable there rather than untested.
2220        if std::fs::create_dir(&odd).is_err() {
2221            eprintln!("skipping: this filesystem rejects non-UTF-8 names");
2222            return;
2223        }
2224        std::fs::write(odd.join("in.txt"), "x").unwrap();
2225
2226        let entries = walk_all(dir.path(), "/");
2227        assert_eq!(
2228            entries.iter().filter(|e| e.name == "in.txt").count(),
2229            1,
2230            "a deferred directory whose name is not UTF-8 must still be reopened",
2231        );
2232    }
2233}