redevplugin_worker_sdk/
fs.rs

1use crate::api;
2use crate::error::{Error, ErrorCode, Result};
3use crate::resource::{Handle, IO_FLAG_EOF, MAX_IO_CHUNK_BYTES};
4use serde::{Deserialize, Serialize};
5use std::sync::atomic::{AtomicU64, Ordering};
6
7static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(1);
8const ATOMIC_TEMP_ATTEMPTS: usize = 16;
9
10#[derive(Debug, Clone, Default, Serialize)]
11pub struct OpenOptions {
12    pub read: bool,
13    pub write: bool,
14    pub create: bool,
15    pub create_new: bool,
16    pub truncate: bool,
17    pub append: bool,
18    #[serde(skip_serializing_if = "Option::is_none")]
19    pub mode: Option<u32>,
20}
21
22#[derive(Debug, Clone, Deserialize)]
23pub struct FileStat {
24    pub uri: String,
25    pub kind: FileKind,
26    pub size: u64,
27    pub mode: u32,
28    pub modified_unix_ms: i64,
29    #[serde(default)]
30    pub created_unix_ms: Option<i64>,
31}
32
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
34#[serde(rename_all = "snake_case")]
35pub enum FileKind {
36    File,
37    Directory,
38    Symlink,
39    Other,
40    #[serde(other)]
41    Unknown,
42}
43
44#[derive(Debug, Clone, Deserialize)]
45pub struct DirectoryEntry {
46    pub name: String,
47    pub uri: String,
48    pub kind: FileKind,
49}
50
51#[derive(Debug, Clone, Deserialize)]
52pub struct DirectoryPage {
53    #[serde(default)]
54    pub entries: Vec<DirectoryEntry>,
55    pub eof: bool,
56}
57
58#[derive(Debug, Clone, Deserialize)]
59pub struct WatchEvent {
60    pub sequence: u64,
61    pub kind: WatchKind,
62    pub uri: String,
63    #[serde(default)]
64    pub previous_uri: Option<String>,
65}
66
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
68#[serde(rename_all = "snake_case")]
69pub enum WatchKind {
70    Create,
71    Change,
72    Delete,
73    Rename,
74    Overflow,
75    #[serde(other)]
76    Unknown,
77}
78
79#[derive(Debug, Clone, Deserialize)]
80pub struct MountInfo {
81    pub id: String,
82    pub uri: String,
83    pub read_only: bool,
84}
85
86#[derive(Debug, Clone, Deserialize)]
87pub struct Mounts {
88    #[serde(default)]
89    pub mounts: Vec<MountInfo>,
90}
91
92#[derive(Deserialize)]
93struct HandleResult {
94    handle: u64,
95}
96
97#[derive(Serialize)]
98struct URIArguments<'a> {
99    uri: &'a str,
100}
101
102pub struct File {
103    handle: Handle,
104}
105
106impl File {
107    pub fn open(uri: &str, options: OpenOptions) -> Result<Self> {
108        #[derive(Serialize)]
109        struct Arguments<'a> {
110            uri: &'a str,
111            options: OpenOptions,
112        }
113        let opened: HandleResult = api::call("fs.open", &Arguments { uri, options })?;
114        Ok(Self {
115            handle: Handle::new(opened.handle)?,
116        })
117    }
118
119    pub fn read(&mut self, capacity: usize) -> Result<(Vec<u8>, u32)> {
120        self.handle.read(capacity)
121    }
122
123    pub fn write_all(&mut self, bytes: &[u8]) -> Result<()> {
124        write_file_chunks(bytes, |chunk| self.handle.write(chunk, 0))
125    }
126
127    pub fn seek(&mut self, offset: i64, whence: u32) -> Result<u64> {
128        self.handle.seek(offset, whence)
129    }
130
131    pub fn sync(&mut self) -> Result<()> {
132        #[derive(Serialize)]
133        struct Arguments {
134            handle: u64,
135        }
136        let _: serde_json::Value = api::call(
137            "fs.sync",
138            &Arguments {
139                handle: self.handle.id(),
140            },
141        )?;
142        Ok(())
143    }
144
145    pub fn close(mut self) -> Result<()> {
146        self.handle.close()
147    }
148}
149
150fn write_file_chunks(bytes: &[u8], mut write: impl FnMut(&[u8]) -> Result<()>) -> Result<()> {
151    let mut acknowledged = 0;
152    for chunk in bytes.chunks(MAX_IO_CHUNK_BYTES) {
153        if let Err(mut error) = write(chunk) {
154            // A later failure cannot undo earlier chunks. The failing chunk may
155            // also have changed bytes; only prior acknowledgements are counted.
156            if !error.details.is_object() {
157                error.details = serde_json::json!({});
158            }
159            error.details["mutation_outcome"] = serde_json::json!("unknown");
160            error.details["acknowledged_bytes"] = serde_json::json!(acknowledged);
161            error.retryable = false;
162            return Err(error);
163        }
164        acknowledged += chunk.len();
165    }
166    Ok(())
167}
168
169pub struct Directory {
170    handle: Handle,
171}
172
173impl Directory {
174    pub fn open(uri: &str) -> Result<Self> {
175        let opened: HandleResult = api::call("fs.read_dir.open", &URIArguments { uri })?;
176        Ok(Self {
177            handle: Handle::new(opened.handle)?,
178        })
179    }
180
181    pub fn next(&mut self, limit: u16) -> Result<DirectoryPage> {
182        #[derive(Serialize)]
183        struct Arguments {
184            handle: u64,
185            limit: u16,
186        }
187        api::call(
188            "fs.read_dir.next",
189            &Arguments {
190                handle: self.handle.id(),
191                limit,
192            },
193        )
194    }
195
196    pub fn close(mut self) -> Result<()> {
197        self.handle.close()
198    }
199}
200
201pub struct Watch {
202    handle: Handle,
203}
204
205impl Watch {
206    pub fn open(uri: &str) -> Result<Self> {
207        #[derive(Serialize)]
208        struct Arguments<'a> {
209            uri: &'a str,
210            recursive: bool,
211        }
212        let opened: HandleResult = api::call(
213            "fs.watch",
214            &Arguments {
215                uri,
216                recursive: false,
217            },
218        )?;
219        Ok(Self {
220            handle: Handle::new(opened.handle)?,
221        })
222    }
223
224    pub fn next(&mut self, timeout_ms: u32) -> Result<WatchEvent> {
225        #[derive(Serialize)]
226        struct Arguments {
227            handle: u64,
228            timeout_ms: u32,
229        }
230        api::call(
231            "fs.watch_next",
232            &Arguments {
233                handle: self.handle.id(),
234                timeout_ms,
235            },
236        )
237    }
238
239    pub fn close(mut self) -> Result<()> {
240        self.handle.close()
241    }
242}
243
244pub fn mounts() -> Result<Mounts> {
245    api::call("fs.mounts", &serde_json::json!({}))
246}
247
248pub fn stat(uri: &str, follow_symlinks: bool) -> Result<FileStat> {
249    #[derive(Serialize)]
250    struct Arguments<'a> {
251        uri: &'a str,
252        follow_symlinks: bool,
253    }
254    api::call(
255        "fs.stat",
256        &Arguments {
257            uri,
258            follow_symlinks,
259        },
260    )
261}
262
263pub fn read_file(uri: &str) -> Result<Vec<u8>> {
264    let mut file = File::open(
265        uri,
266        OpenOptions {
267            read: true,
268            ..OpenOptions::default()
269        },
270    )?;
271    let mut result = Vec::new();
272    loop {
273        let (chunk, flags) = file.read(MAX_IO_CHUNK_BYTES)?;
274        if chunk.is_empty() && flags & IO_FLAG_EOF == 0 {
275            return Err(Error::internal("file read made no progress"));
276        }
277        result.extend_from_slice(&chunk);
278        if flags & IO_FLAG_EOF != 0 {
279            file.close()?;
280            return Ok(result);
281        }
282    }
283}
284
285pub fn read_text(uri: &str) -> Result<String> {
286    String::from_utf8(read_file(uri)?)
287        .map_err(|_| Error::internal("file content is not valid UTF-8"))
288}
289
290pub fn write_file(uri: &str, bytes: &[u8], atomic: bool) -> Result<()> {
291    if !atomic {
292        return write_direct(uri, bytes, false);
293    }
294    for _ in 0..ATOMIC_TEMP_ATTEMPTS {
295        let temporary = atomic_temporary_uri(uri)?;
296        match write_direct(&temporary, bytes, true) {
297            Ok(()) => {
298                let result = rename(&temporary, uri, true);
299                if result.is_err() {
300                    let _ = remove(&temporary, false);
301                }
302                return result;
303            }
304            Err(error) if error.code == ErrorCode::AlreadyExists => continue,
305            Err(error) => {
306                let _ = remove(&temporary, false);
307                return Err(error);
308            }
309        }
310    }
311    Err(Error::internal(
312        "could not allocate a unique same-directory atomic write file",
313    ))
314}
315
316pub fn write_text(uri: &str, text: &str, atomic: bool) -> Result<()> {
317    write_file(uri, text.as_bytes(), atomic)
318}
319
320fn write_direct(uri: &str, bytes: &[u8], create_new: bool) -> Result<()> {
321    let mut file = File::open(
322        uri,
323        OpenOptions {
324            write: true,
325            create: !create_new,
326            create_new,
327            truncate: !create_new,
328            mode: Some(0o600),
329            ..OpenOptions::default()
330        },
331    )?;
332    file.write_all(bytes)?;
333    file.sync()?;
334    file.close()
335}
336
337pub fn remove(uri: &str, recursive: bool) -> Result<()> {
338    #[derive(Serialize)]
339    struct Arguments<'a> {
340        uri: &'a str,
341        recursive: bool,
342    }
343    let _: serde_json::Value = api::call("fs.remove", &Arguments { uri, recursive })?;
344    Ok(())
345}
346
347pub fn mkdir(uri: &str, recursive: bool, mode: u32) -> Result<()> {
348    #[derive(Serialize)]
349    struct Arguments<'a> {
350        uri: &'a str,
351        recursive: bool,
352        mode: u32,
353    }
354    let _: serde_json::Value = api::call(
355        "fs.mkdir",
356        &Arguments {
357            uri,
358            recursive,
359            mode,
360        },
361    )?;
362    Ok(())
363}
364
365pub fn rename(from: &str, to: &str, overwrite: bool) -> Result<()> {
366    transfer("fs.rename", from, to, overwrite)
367}
368
369pub fn copy(from: &str, to: &str, overwrite: bool) -> Result<()> {
370    transfer("fs.copy", from, to, overwrite)
371}
372
373fn transfer(operation: &str, from: &str, to: &str, overwrite: bool) -> Result<()> {
374    #[derive(Serialize)]
375    struct Arguments<'a> {
376        from: &'a str,
377        to: &'a str,
378        overwrite: bool,
379    }
380    let _: serde_json::Value = api::call(
381        operation,
382        &Arguments {
383            from,
384            to,
385            overwrite,
386        },
387    )?;
388    Ok(())
389}
390
391pub fn set_times(uri: &str, accessed_unix_ms: i64, modified_unix_ms: i64) -> Result<()> {
392    #[derive(Serialize)]
393    struct Arguments<'a> {
394        uri: &'a str,
395        accessed_unix_ms: i64,
396        modified_unix_ms: i64,
397    }
398    let _: serde_json::Value = api::call(
399        "fs.set_times",
400        &Arguments {
401            uri,
402            accessed_unix_ms,
403            modified_unix_ms,
404        },
405    )?;
406    Ok(())
407}
408
409fn atomic_temporary_uri(uri: &str) -> Result<String> {
410    let (directory, name) = uri
411        .rsplit_once('/')
412        .filter(|(_, name)| !name.is_empty())
413        .ok_or_else(|| Error::from_abi_status(-1))?;
414    let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
415    Ok(format!(
416        "{directory}/.{name}.redevplugin-{sequence:016x}.tmp"
417    ))
418}
419
420#[cfg(test)]
421mod tests {
422    use super::*;
423
424    #[test]
425    fn chunk_failure_preserves_acknowledged_bytes_and_never_replays() {
426        let bytes = vec![0; MAX_IO_CHUNK_BYTES + 1];
427        let mut calls = 0;
428        let error = write_file_chunks(&bytes, |_| {
429            calls += 1;
430            if calls == 1 {
431                Ok(())
432            } else {
433                Err(Error::from_abi_status(-2))
434            }
435        })
436        .unwrap_err();
437        assert_eq!(calls, 2);
438        assert_eq!(error.code, ErrorCode::PermissionDenied);
439        assert_eq!(error.details["mutation_outcome"], "unknown");
440        assert_eq!(error.details["acknowledged_bytes"], MAX_IO_CHUNK_BYTES);
441        assert!(!error.retryable);
442    }
443
444    #[test]
445    fn atomic_temporary_file_stays_in_the_same_directory() {
446        let temporary = atomic_temporary_uri("redevfs://workspace/src/data.bin").unwrap();
447        assert!(temporary.starts_with("redevfs://workspace/src/.data.bin.redevplugin-"));
448        assert!(temporary.ends_with(".tmp"));
449    }
450}
redevplugin_worker_sdk/fs.rs

redevplugin_worker_sdk/
fs.rs
