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::{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    /// Release process-local accounting after a successful explicit close,
73    /// including when execution history skipped the backend call.
74    fn release_buffer(&mut self);
75    fn id(&self) -> EffectId;
76    async fn close(
77        &mut self,
78        filesystem: &dyn Filesystem,
79        effect_id: &EffectId,
80        work: Option<Arc<WorkBudget>>,
81        limits: &TenantLimits,
82    ) -> Result<(), FilesystemError>;
83}
84
85pub struct BackendReader {
86    pub reader: crate::runtime::filesystem::FileReader,
87    charge: ByteCharge,
88}
89impl BackendReader {
90    pub fn new(
91        reader: crate::runtime::filesystem::FileReader,
92        limits: &TenantLimits,
93    ) -> Result<Self, MemoryCapExceeded> {
94        Ok(Self {
95            reader,
96            charge: ByteCharge::new(limits, HANDLE_BUF_BYTES)?,
97        })
98    }
99    pub fn id(&self) -> EffectId {
100        self.reader.id.clone()
101    }
102}
103#[async_trait::async_trait]
104impl Closable for BackendReader {
105    fn release_buffer(&mut self) {
106        self.charge.release();
107    }
108    fn id(&self) -> EffectId {
109        self.reader.id.clone()
110    }
111    async fn close(
112        &mut self,
113        filesystem: &dyn Filesystem,
114        effect_id: &EffectId,
115        work: Option<Arc<WorkBudget>>,
116        _limits: &TenantLimits,
117    ) -> Result<(), FilesystemError> {
118        filesystem
119            .close_reader(effect_id, work, &mut self.reader)
120            .await?;
121        self.charge.release();
122        Ok(())
123    }
124}
125
126pub struct ChargedDirIter {
127    pub walk: crate::runtime::filesystem::DirectoryReader,
128    charge: ByteCharge,
129}
130impl ChargedDirIter {
131    pub fn new(
132        walk: crate::runtime::filesystem::DirectoryReader,
133        limits: &TenantLimits,
134    ) -> Result<Self, MemoryCapExceeded> {
135        Ok(Self {
136            charge: ByteCharge::new(limits, HANDLE_BUF_BYTES)?,
137            walk,
138        })
139    }
140    pub fn id(&self) -> EffectId {
141        self.walk.id.clone()
142    }
143}
144#[async_trait::async_trait]
145impl Closable for ChargedDirIter {
146    fn release_buffer(&mut self) {
147        self.charge.release();
148    }
149    fn id(&self) -> EffectId {
150        self.walk.id.clone()
151    }
152    async fn close(
153        &mut self,
154        filesystem: &dyn Filesystem,
155        effect_id: &EffectId,
156        work: Option<Arc<WorkBudget>>,
157        _limits: &TenantLimits,
158    ) -> Result<(), FilesystemError> {
159        filesystem
160            .close_directory(effect_id, work, &mut self.walk, _limits)
161            .await?;
162        self.charge.release();
163        Ok(())
164    }
165}
166
167/// A guest writer owns its resource state and accounts for its bounded buffer.
168pub struct BackendFileWriter {
169    writer: crate::runtime::filesystem::FileWriter,
170    charge: ByteCharge,
171}
172impl BackendFileWriter {
173    pub fn release_buffer(&mut self) {
174        self.charge.release();
175    }
176    pub fn id(&self) -> EffectId {
177        self.writer.id.clone()
178    }
179    pub fn new(
180        writer: crate::runtime::filesystem::FileWriter,
181        limits: &TenantLimits,
182    ) -> Result<Self, MemoryCapExceeded> {
183        Ok(Self {
184            writer,
185            charge: ByteCharge::new(limits, HANDLE_BUF_BYTES)?,
186        })
187    }
188    pub async fn write(
189        &mut self,
190        filesystem: &dyn Filesystem,
191        effect_id: &EffectId,
192        work: Option<Arc<WorkBudget>>,
193        bytes: Vec<u8>,
194    ) -> Result<(), FilesystemError> {
195        filesystem
196            .write_chunk(effect_id, work, &mut self.writer, bytes)
197            .await
198    }
199    pub async fn close(
200        &mut self,
201        filesystem: &dyn Filesystem,
202        effect_id: &EffectId,
203        work: Option<Arc<WorkBudget>>,
204    ) -> Result<(), FilesystemError> {
205        let result = filesystem
206            .close_writer(effect_id, work, &mut self.writer)
207            .await;
208        if self.writer.closed {
209            self.charge.release();
210        }
211        result
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218    use crate::runtime::default_filesystem::DefaultFilesystem;
219
220    #[tokio::test]
221    async fn failed_native_backend_close_refunds_a_discarded_buffer() {
222        use crate::runtime::default_filesystem::DefaultFilesystem;
223        use crate::runtime::filesystem::Filesystem;
224        let root = tempfile::tempdir().unwrap();
225        let filesystem = DefaultFilesystem::new(
226            crate::runtime::Vfs::external(root.path().to_path_buf()).unwrap(),
227            "run".into(),
228        );
229        let effect_id = EffectId::stub().unwrap();
230        let limits = TenantLimits::new(64 * 1024);
231        let writer = filesystem
232            .open_writer(&effect_id, None, "/target")
233            .await
234            .unwrap();
235        let mut handle = BackendFileWriter::new(writer, &limits).unwrap();
236        handle
237            .write(&filesystem, &effect_id, None, b"data".to_vec())
238            .await
239            .unwrap();
240        std::fs::create_dir(root.path().join("target")).unwrap();
241        assert!(handle.close(&filesystem, &effect_id, None).await.is_err());
242        assert_eq!(limits.host_attached_bytes(), 0);
243        assert!(handle.close(&filesystem, &effect_id, None).await.is_ok());
244    }
245
246    async fn writer_over(vfs: &crate::runtime::Vfs, limits: &TenantLimits) -> BackendFileWriter {
247        use crate::runtime::default_filesystem::DefaultFilesystem;
248        use crate::runtime::filesystem::Filesystem;
249        let writer = DefaultFilesystem::new(vfs.clone(), "run".into())
250            .open_writer(&EffectId::stub().unwrap(), None, "/out.txt")
251            .await
252            .unwrap();
253        BackendFileWriter::new(writer, limits).unwrap()
254    }
255
256    fn temp_leftovers(root: &std::path::Path) -> Vec<std::ffi::OsString> {
257        std::fs::read_dir(root)
258            .unwrap()
259            .map(|e| e.unwrap().file_name())
260            .filter(|n| n.to_string_lossy().starts_with("out.txt."))
261            .collect()
262    }
263
264    #[tokio::test]
265    async fn unclosed_writer_drop_removes_temp_and_skips_rename() {
266        let limits = TenantLimits::new(10 * 1024 * 1024);
267        let dir = tempfile::tempdir().unwrap();
268        let vfs = crate::runtime::Vfs::external(dir.path().to_path_buf()).unwrap();
269        let mut w = writer_over(&vfs, &limits).await;
270        w.write(
271            &DefaultFilesystem::new(vfs.clone(), "run".into()),
272            &EffectId::stub().unwrap(),
273            None,
274            b"hello\n".to_vec(),
275        )
276        .await
277        .unwrap();
278        assert!(limits.host_attached_bytes() > 0);
279        drop(w);
280        assert!(
281            temp_leftovers(dir.path()).is_empty(),
282            "temp file removed on drop"
283        );
284        assert!(
285            !dir.path().join("out.txt").exists(),
286            "no rename without explicit close"
287        );
288        assert_eq!(limits.host_attached_bytes(), 0);
289    }
290
291    #[tokio::test]
292    async fn writer_close_renames_and_refunds() {
293        let limits = TenantLimits::new(10 * 1024 * 1024);
294        let dir = tempfile::tempdir().unwrap();
295        let vfs = crate::runtime::Vfs::external(dir.path().to_path_buf()).unwrap();
296        let final_path = dir.path().join("out.txt");
297        let mut w = writer_over(&vfs, &limits).await;
298        w.write(
299            &DefaultFilesystem::new(vfs.clone(), "run".into()),
300            &EffectId::stub().unwrap(),
301            None,
302            b"hello\n".to_vec(),
303        )
304        .await
305        .unwrap();
306        w.close(
307            &DefaultFilesystem::new(vfs.clone(), "run".into()),
308            &EffectId::stub().unwrap(),
309            None,
310        )
311        .await
312        .unwrap();
313        assert!(final_path.exists());
314        assert!(temp_leftovers(dir.path()).is_empty());
315        assert_eq!(limits.host_attached_bytes(), 0);
316        assert_eq!(std::fs::read_to_string(&final_path).unwrap(), "hello\n");
317    }
318}
submilli_engine/stdlib/fs/handles.rs

submilli_engine/stdlib/fs/
handles.rs
