clt_database/io/
vfs.rs

1use super::{Buffer, Completion, File, FileSyncType, OpenFlags, IO};
2use crate::ext::VfsMod;
3use crate::io::clock::{Clock, DefaultClock, MonotonicInstant, WallClockInstant};
4use crate::io::CompletionInner;
5use crate::sync::Arc;
6use crate::{LimboError, Result};
7use std::ffi::{c_void, CString};
8use std::ptr::NonNull;
9use turso_ext::{BufferRef, IOCallback, SendPtr, VfsFileImpl, VfsImpl};
10
11impl Clock for VfsMod {
12    fn current_time_monotonic(&self) -> MonotonicInstant {
13        DefaultClock.current_time_monotonic()
14    }
15
16    fn current_time_wall_clock(&self) -> WallClockInstant {
17        DefaultClock.current_time_wall_clock()
18    }
19}
20
21impl IO for VfsMod {
22    fn open_file(&self, path: &str, flags: OpenFlags, direct: bool) -> Result<Arc<dyn File>> {
23        let c_path = CString::new(path).map_err(|_| {
24            LimboError::ExtensionError("Failed to convert path to CString".to_string())
25        })?;
26        let ctx = self.ctx as *mut c_void;
27        let vfs = unsafe { &*self.ctx };
28        let file = unsafe { (vfs.open)(ctx, c_path.as_ptr(), flags.0, direct) };
29        if file.is_null() {
30            return Err(LimboError::ExtensionError("File not found".to_string()));
31        }
32        Ok(Arc::new(turso_ext::VfsFileImpl::new(file, self.ctx)?))
33    }
34
35    fn remove_file(&self, path: &str) -> Result<()> {
36        let c_path = CString::new(path).map_err(|_| {
37            LimboError::ExtensionError("Failed to convert path to CString".to_string())
38        })?;
39        let ctx = self.ctx as *mut c_void;
40        let vfs = unsafe { &*self.ctx };
41        let result = unsafe { (vfs.remove)(ctx, c_path.as_ptr()) };
42        if !result.is_ok() {
43            return Err(LimboError::ExtensionError(result.to_string()));
44        }
45        Ok(())
46    }
47
48    fn step(&self) -> Result<()> {
49        if self.ctx.is_null() {
50            return Err(LimboError::ExtensionError("VFS is null".to_string()));
51        }
52        let vfs = unsafe { &*self.ctx };
53        let result = unsafe { (vfs.run_once)(vfs.vfs) };
54        if !result.is_ok() {
55            return Err(LimboError::ExtensionError(result.to_string()));
56        }
57        Ok(())
58    }
59
60    fn generate_random_number(&self) -> i64 {
61        if self.ctx.is_null() {
62            return -1;
63        }
64        let vfs = unsafe { &*self.ctx };
65        unsafe { (vfs.gen_random_number)() }
66    }
67}
68
69impl VfsMod {
70    #[allow(dead_code)] // used in FFI call
71    fn get_current_time(&self) -> String {
72        if self.ctx.is_null() {
73            return "".to_string();
74        }
75        unsafe {
76            let vfs = &*self.ctx;
77            let chars = (vfs.current_time)();
78            let cstr = CString::from_raw(chars as *mut _);
79            cstr.to_string_lossy().into_owned()
80        }
81    }
82}
83
84/// # Safety
85/// the callback wrapper in the extension library is FnOnce, so we know
86/// that the into_raw/from_raw contract will hold
87unsafe extern "C" fn callback_fn(result: i32, ctx: SendPtr) {
88    let completion = Completion {
89        inner: (Some(Arc::from_raw(ctx.inner().as_ptr() as *mut CompletionInner))),
90    };
91    completion.complete(result);
92}
93
94fn to_callback(c: Completion) -> IOCallback {
95    IOCallback::new(callback_fn, unsafe {
96        NonNull::new_unchecked(Arc::into_raw(c.get_inner().clone()) as *mut c_void)
97    })
98}
99
100impl File for VfsFileImpl {
101    fn lock_file(&self, exclusive: bool) -> Result<()> {
102        let vfs = unsafe { &*self.vfs };
103        let result = unsafe { (vfs.lock)(self.file, exclusive) };
104        if result.is_ok() {
105            return Err(LimboError::ExtensionError(result.to_string()));
106        }
107        Ok(())
108    }
109
110    fn unlock_file(&self) -> Result<()> {
111        if self.vfs.is_null() {
112            return Err(LimboError::ExtensionError("VFS is null".to_string()));
113        }
114        let vfs = unsafe { &*self.vfs };
115        let result = unsafe { (vfs.unlock)(self.file) };
116        if result.is_ok() {
117            return Err(LimboError::ExtensionError(result.to_string()));
118        }
119        Ok(())
120    }
121
122    fn pread(&self, pos: u64, c: Completion) -> Result<Completion> {
123        if self.vfs.is_null() {
124            c.complete(-1);
125            return Err(LimboError::ExtensionError("VFS is null".to_string()));
126        }
127        let r = c.as_read();
128        let buf = r.buf();
129        let len = buf.len();
130        let cb = to_callback(c.clone());
131        let vfs = unsafe { &*self.vfs };
132        let res = unsafe {
133            (vfs.read)(
134                self.file,
135                BufferRef::new(buf.as_mut_ptr(), len),
136                pos as i64,
137                cb,
138            )
139        };
140        if res.is_error() {
141            return Err(LimboError::ExtensionError("pread failed".to_string()));
142        }
143        Ok(c)
144    }
145
146    fn pwrite(&self, pos: u64, buffer: Arc<Buffer>, c: Completion) -> Result<Completion> {
147        if self.vfs.is_null() {
148            c.complete(-1);
149            return Err(LimboError::ExtensionError("VFS is null".to_string()));
150        }
151        let vfs = unsafe { &*self.vfs };
152        let res = unsafe {
153            let len = buffer.len();
154            let cb = to_callback(c.clone());
155            (vfs.write)(
156                self.file,
157                BufferRef::new(buffer.as_ptr() as *mut u8, len),
158                pos as i64,
159                cb,
160            )
161        };
162        if res.is_error() {
163            return Err(LimboError::ExtensionError("pwrite failed".to_string()));
164        }
165        // Keep the buffer alive until the VFS completion fires — the extension
166        // may process the write asynchronously after this function returns.
167        c.keep_write_buffer_alive(buffer);
168        Ok(c)
169    }
170
171    fn sync(&self, c: Completion, _sync_type: FileSyncType) -> Result<Completion> {
172        if self.vfs.is_null() {
173            c.complete(-1);
174            return Err(LimboError::ExtensionError("VFS is null".to_string()));
175        }
176        let vfs = unsafe { &*self.vfs };
177        let cb = to_callback(c.clone());
178        let res = unsafe { (vfs.sync)(self.file, cb) };
179        if res.is_error() {
180            return Err(LimboError::ExtensionError("sync failed".to_string()));
181        }
182        Ok(c)
183    }
184
185    fn size(&self) -> Result<u64> {
186        let vfs = unsafe { &*self.vfs };
187        let result = unsafe { (vfs.size)(self.file) };
188        if result < 0 {
189            Err(LimboError::ExtensionError("size failed".to_string()))
190        } else {
191            Ok(result as u64)
192        }
193    }
194
195    fn truncate(&self, len: u64, c: Completion) -> Result<Completion> {
196        if self.vfs.is_null() {
197            c.complete(-1);
198            return Err(LimboError::ExtensionError("VFS is null".to_string()));
199        }
200        let vfs = unsafe { &*self.vfs };
201        let cb = to_callback(c.clone());
202        let res = unsafe { (vfs.truncate)(self.file, len as i64, cb) };
203        if res.is_error() {
204            return Err(LimboError::ExtensionError("truncate failed".to_string()));
205        }
206        Ok(c)
207    }
208}
209
210impl Drop for VfsMod {
211    fn drop(&mut self) {
212        if self.ctx.is_null() {
213            return;
214        }
215        unsafe {
216            let _ = Box::from_raw(self.ctx as *mut VfsImpl);
217        }
218    }
219}
clt_database/io/vfs.rs

clt_database/io/
vfs.rs
