submilli_engine/stdlib/fs/
handles.rs

1//! Externref payloads for `submilli:fs` open resources.
2//!
3//! Each payload owns a backend resource plus a clone of
4//! `TenantLimits.host_attached_bytes` (via [`ByteCharge`]). It charges its buffer
5//! footprint at creation and refunds it when the handle closes or is reclaimed.
6//!
7//! Readers close eagerly. Writers publish only through explicit backend close;
8//! dropping a guest handle releases resources without publishing output.
9
10use std::sync::Arc;
11use std::sync::atomic::{AtomicU64, Ordering};
12
13use crate::runtime::durable::EffectId;
14use crate::runtime::filesystem::FilesystemError;
15use crate::runtime::fuel::WorkBudget;
16use crate::runtime::limits::{MemoryCapExceeded, TenantLimits};
17
18pub use crate::runtime::default_filesystem::kind_of;
19
20/// Footprint charged per reader/writer, matching the default `BufReader`/`BufWriter`
21/// capacity. Byte readers add their `chunk_size` working buffer on top.
22const HANDLE_BUF_BYTES: u64 = 8 * 1024;
23
24/// Bytes charged against the store's host-attached counter, refunded on `release`/`drop`.
25pub(super) struct ByteCharge {
26    bytes: u64,
27    counter: Arc<AtomicU64>,
28}
29
30impl ByteCharge {
31    pub(super) fn new(limits: &TenantLimits, bytes: u64) -> Result<Self, MemoryCapExceeded> {
32        limits.charge_host_bytes(bytes)?;
33        Ok(Self {
34            bytes,
35            counter: limits.host_attached_counter(),
36        })
37    }
38
39    pub(super) fn grow(
40        &mut self,
41        limits: &TenantLimits,
42        bytes: u64,
43    ) -> Result<(), MemoryCapExceeded> {
44        limits.charge_host_bytes(bytes)?;
45        self.bytes = self.bytes.saturating_add(bytes);
46        Ok(())
47    }
48
49    /// Refunds the charged bytes once; subsequent calls are no-ops.
50    fn release(&mut self) {
51        let n = std::mem::take(&mut self.bytes);
52        if n == 0 {
53            return;
54        }
55        let _ = self
56            .counter
57            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
58                Some(current.saturating_sub(n))
59            });
60    }
61}
62
63impl Drop for ByteCharge {
64    fn drop(&mut self) {
65        self.release();
66    }
67}
68
69/// Explicit guest close is effect-aware. Drop only releases local resources.
70#[async_trait::async_trait]
71pub trait Closable: Send + Sync {
72    async fn close(
73        &mut self,
74        effect_id: &EffectId,
75        work: Option<Arc<WorkBudget>>,
76        limits: &TenantLimits,
77    ) -> Result<(), FilesystemError>;
78}
79
80pub struct BackendReader {
81    pub reader: Box<dyn crate::runtime::filesystem::FileReader>,
82    charge: ByteCharge,
83}
84impl BackendReader {
85    pub fn new(
86        reader: Box<dyn crate::runtime::filesystem::FileReader>,
87        limits: &TenantLimits,
88    ) -> Result<Self, MemoryCapExceeded> {
89        Ok(Self {
90            reader,
91            charge: ByteCharge::new(limits, HANDLE_BUF_BYTES)?,
92        })
93    }
94    pub fn descriptor(&self) -> Option<crate::runtime::filesystem::ResourceDescriptor> {
95        self.reader.descriptor()
96    }
97}
98#[async_trait::async_trait]
99impl Closable for BackendReader {
100    async fn close(
101        &mut self,
102        effect_id: &EffectId,
103        work: Option<Arc<WorkBudget>>,
104        _limits: &TenantLimits,
105    ) -> Result<(), FilesystemError> {
106        self.reader.close(effect_id, work).await?;
107        self.charge.release();
108        Ok(())
109    }
110}
111
112pub struct ChargedDirIter {
113    pub walk: Box<dyn crate::runtime::filesystem::DirectoryReader>,
114    charge: ByteCharge,
115}
116impl ChargedDirIter {
117    pub fn new(
118        walk: Box<dyn crate::runtime::filesystem::DirectoryReader>,
119        limits: &TenantLimits,
120    ) -> Result<Self, MemoryCapExceeded> {
121        Ok(Self {
122            charge: ByteCharge::new(limits, HANDLE_BUF_BYTES.max(walk.retained_bytes()))?,
123            walk,
124        })
125    }
126    pub fn descriptor(&self) -> Option<crate::runtime::filesystem::ResourceDescriptor> {
127        self.walk.descriptor()
128    }
129}
130#[async_trait::async_trait]
131impl Closable for ChargedDirIter {
132    async fn close(
133        &mut self,
134        effect_id: &EffectId,
135        work: Option<Arc<WorkBudget>>,
136        _limits: &TenantLimits,
137    ) -> Result<(), FilesystemError> {
138        self.walk.close(effect_id, work, _limits).await?;
139        self.charge.release();
140        Ok(())
141    }
142}
143
144/// A guest writer holds only the selected backend and its bounded host buffer.
145pub struct BackendFileWriter {
146    writer: Box<dyn crate::runtime::filesystem::FileWriter>,
147    charge: ByteCharge,
148}
149impl BackendFileWriter {
150    pub fn descriptor(&self) -> Option<crate::runtime::filesystem::ResourceDescriptor> {
151        self.writer.descriptor()
152    }
153    pub fn new(
154        writer: Box<dyn crate::runtime::filesystem::FileWriter>,
155        limits: &TenantLimits,
156    ) -> Result<Self, MemoryCapExceeded> {
157        Ok(Self {
158            writer,
159            charge: ByteCharge::new(limits, HANDLE_BUF_BYTES)?,
160        })
161    }
162    pub async fn write(
163        &mut self,
164        effect_id: &EffectId,
165        work: Option<Arc<WorkBudget>>,
166        bytes: Vec<u8>,
167    ) -> Result<(), FilesystemError> {
168        self.writer.write(effect_id, work, bytes).await
169    }
170    pub async fn close(
171        &mut self,
172        effect_id: &EffectId,
173        work: Option<Arc<WorkBudget>>,
174    ) -> Result<(), FilesystemError> {
175        let result = self.writer.close(effect_id, work).await;
176        if self.writer.is_closed() {
177            self.charge.release();
178        }
179        result
180    }
181}
182
183#[cfg(test)]
184mod tests {
185    use super::*;
186
187    #[tokio::test]
188    async fn failed_native_backend_close_refunds_a_discarded_buffer() {
189        use crate::runtime::default_filesystem::DefaultFilesystem;
190        use crate::runtime::filesystem::Filesystem;
191        let root = tempfile::tempdir().unwrap();
192        let filesystem = DefaultFilesystem::new(
193            crate::runtime::Vfs::external(root.path().to_path_buf()).unwrap(),
194        );
195        let effect_id = EffectId::stub().unwrap();
196        let limits = TenantLimits::new(64 * 1024);
197        let writer = filesystem
198            .open_writer(&effect_id, None, "/target")
199            .await
200            .unwrap();
201        let mut handle = BackendFileWriter::new(writer, &limits).unwrap();
202        handle
203            .write(&effect_id, None, b"data".to_vec())
204            .await
205            .unwrap();
206        std::fs::create_dir(root.path().join("target")).unwrap();
207        assert!(handle.close(&effect_id, None).await.is_err());
208        assert_eq!(limits.host_attached_bytes(), 0);
209        assert!(handle.close(&effect_id, None).await.is_ok());
210    }
211
212    async fn writer_over(vfs: &crate::runtime::Vfs, limits: &TenantLimits) -> BackendFileWriter {
213        use crate::runtime::default_filesystem::DefaultFilesystem;
214        use crate::runtime::filesystem::Filesystem;
215        let writer = DefaultFilesystem::new(vfs.clone())
216            .open_writer(&EffectId::stub().unwrap(), None, "/out.txt")
217            .await
218            .unwrap();
219        BackendFileWriter::new(writer, limits).unwrap()
220    }
221
222    fn temp_leftovers(root: &std::path::Path) -> Vec<std::ffi::OsString> {
223        std::fs::read_dir(root)
224            .unwrap()
225            .map(|e| e.unwrap().file_name())
226            .filter(|n| n.to_string_lossy().starts_with("out.txt."))
227            .collect()
228    }
229
230    #[tokio::test]
231    async fn unclosed_writer_drop_removes_temp_and_skips_rename() {
232        let limits = TenantLimits::new(10 * 1024 * 1024);
233        let dir = tempfile::tempdir().unwrap();
234        let vfs = crate::runtime::Vfs::external(dir.path().to_path_buf()).unwrap();
235        let mut w = writer_over(&vfs, &limits).await;
236        w.write(&EffectId::stub().unwrap(), None, b"hello\n".to_vec())
237            .await
238            .unwrap();
239        assert!(limits.host_attached_bytes() > 0);
240        drop(w);
241        assert!(
242            temp_leftovers(dir.path()).is_empty(),
243            "temp file removed on drop"
244        );
245        assert!(
246            !dir.path().join("out.txt").exists(),
247            "no rename without explicit close"
248        );
249        assert_eq!(limits.host_attached_bytes(), 0);
250    }
251
252    #[tokio::test]
253    async fn writer_close_renames_and_refunds() {
254        let limits = TenantLimits::new(10 * 1024 * 1024);
255        let dir = tempfile::tempdir().unwrap();
256        let vfs = crate::runtime::Vfs::external(dir.path().to_path_buf()).unwrap();
257        let final_path = dir.path().join("out.txt");
258        let mut w = writer_over(&vfs, &limits).await;
259        w.write(&EffectId::stub().unwrap(), None, b"hello\n".to_vec())
260            .await
261            .unwrap();
262        w.close(&EffectId::stub().unwrap(), None).await.unwrap();
263        assert!(final_path.exists());
264        assert!(temp_leftovers(dir.path()).is_empty());
265        assert_eq!(limits.host_attached_bytes(), 0);
266        assert_eq!(std::fs::read_to_string(&final_path).unwrap(), "hello\n");
267    }
268}
submilli_engine/stdlib/fs/handles.rs

submilli_engine/stdlib/fs/
handles.rs
