lb_fs/
fs_impl.rs

1use crate::cache::FileEntry;
2use crate::file_handle::UuidFileHandle;
3use crate::utils::{file_id, get_string};
4use lb_rs::model::file::File;
5use lb_rs::model::file_metadata::FileType;
6use lb_rs::{Lb, Uuid};
7use nfs3_server::nfs3_types::nfs3::{
8    Nfs3Option, fattr3, filename3, nfspath3, nfsstat3, sattr3, set_atime, set_mtime,
9};
10use nfs3_server::vfs::{
11    DirEntry, DirEntryPlus, NfsFileSystem, NfsReadFileSystem, ReadDirIterator, ReadDirPlusIterator,
12};
13use std::collections::HashMap;
14use std::iter::Iterator as StdIterator;
15use std::sync::Arc;
16use tokio::sync::Mutex;
17use tracing::{info, instrument, warn};
18
19type EntriesMap = Arc<Mutex<HashMap<UuidFileHandle, FileEntry>>>;
20
21#[derive(Clone)]
22pub struct Drive {
23    pub lb: Lb,
24
25    /// must be not-nil before NFSFIlesSystem is mounted
26    pub root: Uuid,
27
28    /// probably this doesn't need to have a global lock, but interactions here are generally
29    /// speedy, and for now we'll go for robustness over performance. Hopefully this accomplishes
30    /// that and not deadlock. TBD.
31    ///
32    /// this is stored in memory as it's own entity and not stored in core for two reasons:
33    /// 1. size computations are expensive in core
34    /// 2. nfs needs to update timestamps to specified values
35    /// 3. nfs models properties we don't, like file permission bits
36    pub data: EntriesMap,
37}
38
39impl Drive {
40    /// Loads the child entries of a directory, beginning after the specified cookie.
41    ///
42    /// The cookie corresponds to the file ID of the last entry returned by a previous `readdir` or
43    /// `readdirplus` call. A cookie value of `0` indicates that iteration should begin from the
44    /// start of the directory.
45    ///
46    /// Note: The file ID used as a cookie represents half of the file’s UUID.
47    /// While rare, collisions between file IDs can occur, meaning two distinct files may share
48    /// the same ID. In such cases, some entries might be skipped. This issue affects only large
49    /// datasets when `readdir` or `readdirplus` is invoked multiple times. The possible solution
50    /// might be to use an index as the cookie instead of the file ID.
51    async fn load_children(
52        &self, dirid: &UuidFileHandle, cookie: u64,
53    ) -> impl StdIterator<Item = File> + 'static {
54        let mut children = self.lb.get_children(dirid.as_uuid()).await.unwrap();
55
56        children.sort_by(|a, b| a.id.cmp(&b.id));
57
58        let mut start_index = 0;
59        if cookie > 0 {
60            start_index = children
61                .iter()
62                .position(|child| file_id(child) > cookie)
63                .unwrap_or_else(|| {
64                    warn!("cookie {cookie} not found");
65                    children.len()
66                });
67        }
68
69        children.into_iter().skip(start_index)
70    }
71}
72
73impl NfsReadFileSystem for Drive {
74    type Handle = UuidFileHandle;
75
76    #[instrument(skip(self))]
77    fn root_dir(&self) -> Self::Handle {
78        self.root.into()
79    }
80
81    #[instrument(skip(self), fields(dirid = dirid.to_string(), filename = get_string(filename)))]
82    async fn lookup(
83        &self, dirid: &Self::Handle, filename: &filename3<'_>,
84    ) -> Result<Self::Handle, nfsstat3> {
85        let dir = self
86            .data
87            .lock()
88            .await
89            .get(dirid)
90            .ok_or(nfsstat3::NFS3ERR_STALE)?
91            .file
92            .clone();
93
94        if dir.is_document() {
95            info!("NOTDIR");
96            return Err(nfsstat3::NFS3ERR_NOTDIR);
97        }
98
99        // if looking for dir/. its the current directory
100        if filename.as_ref() == [b'.'] {
101            info!(". == {dirid}");
102            return Ok(*dirid);
103        }
104
105        // if looking for dir/.. its the parent directory
106        if filename.as_ref() == [b'.', b'.'] {
107            info!(".. == {}", dir.parent);
108            return Ok(dir.parent.into());
109        }
110
111        // todo this should almost certainly just operate on the cache
112        let children = self.lb.get_children(&dir.id).await.unwrap();
113        let file_name = get_string(filename);
114
115        for child in children {
116            if file_name == child.name {
117                info!("{}", child.id);
118                return Ok(child.id.into());
119            }
120        }
121
122        info!("NOENT");
123        Err(nfsstat3::NFS3ERR_NOENT)
124    }
125
126    #[instrument(skip(self), fields(id = id.to_string()))]
127    async fn getattr(&self, id: &Self::Handle) -> Result<fattr3, nfsstat3> {
128        let file = self
129            .data
130            .lock()
131            .await
132            .get(id)
133            .ok_or(nfsstat3::NFS3ERR_STALE)?
134            .fattr
135            .clone();
136        info!("fattr = {:?}", file);
137        Ok(file)
138    }
139
140    #[instrument(skip(self), fields(id = id.to_string(), offset, count))]
141    async fn read(
142        &self, id: &Self::Handle, offset: u64, count: u32,
143    ) -> Result<(Vec<u8>, bool), nfsstat3> {
144        let offset = offset as usize;
145        let count = count as usize;
146
147        let doc = self.lb.read_document(*id.as_uuid(), false).await.unwrap();
148
149        if offset >= doc.len() {
150            info!("[] EOF");
151            return Ok((vec![], true));
152        }
153
154        if offset + count >= doc.len() {
155            info!("|{}| EOF", doc[offset..].len());
156            return Ok((doc[offset..].to_vec(), true));
157        }
158
159        info!("|{}|", count);
160        return Ok((doc[offset..offset + count].to_vec(), false));
161    }
162
163    #[instrument(skip(self), fields(dirid = dirid.to_string(), start_after = cookie))]
164    async fn readdir(
165        &self, dirid: &Self::Handle, cookie: u64,
166    ) -> Result<impl nfs3_server::vfs::ReadDirIterator, nfsstat3> {
167        let iter = self.load_children(dirid, cookie).await;
168        Ok(Iterator { inner: iter })
169    }
170
171    #[instrument(skip(self), fields(dirid = dirid.to_string(), start_after = cookie))]
172    async fn readdirplus(
173        &self, dirid: &Self::Handle, cookie: u64,
174    ) -> Result<impl ReadDirPlusIterator<UuidFileHandle>, nfsstat3> {
175        let iter = self.load_children(dirid, cookie).await;
176        let data = self.data.lock().await;
177
178        let iter = iter
179            .map(move |file| {
180                let id: UuidFileHandle = file.id.into();
181                let name = file.name.as_bytes().to_vec().into();
182                let fattr = data.get(&id).map(|entry| entry.fattr.clone());
183                DirEntryPlus {
184                    fileid: id.fileid(),
185                    name,
186                    cookie: id.fileid(),
187                    name_attributes: fattr,
188                    name_handle: Some(id),
189                }
190            })
191            .collect::<Vec<_>>()
192            .into_iter();
193        Ok(IteratorPlus { inner: iter })
194    }
195
196    async fn readlink(&self, _id: &Self::Handle) -> Result<nfspath3<'_>, nfsstat3> {
197        info!("readlink NOTSUPP");
198        Err(nfsstat3::NFS3ERR_NOTSUPP)
199    }
200}
201
202impl NfsFileSystem for Drive {
203    #[instrument(skip(self), fields(id = id.to_string()))]
204    async fn setattr(&self, id: &Self::Handle, setattr: sattr3) -> Result<fattr3, nfsstat3> {
205        let mut data = self.data.lock().await;
206        let now = FileEntry::now();
207        let entry = data.get_mut(id).unwrap();
208
209        if let Nfs3Option::Some(new) = setattr.size
210            && entry.fattr.size != new
211        {
212            let mut doc = self.lb.read_document(*id.as_uuid(), false).await.unwrap();
213            doc.resize(new as usize, 0);
214            self.lb.write_document(*id.as_uuid(), &doc).await.unwrap();
215            entry.fattr.mtime = now;
216            entry.fattr.ctime = now;
217        }
218
219        match setattr.atime {
220            set_atime::DONT_CHANGE => {}
221            set_atime::SET_TO_SERVER_TIME => {
222                entry.fattr.atime = now;
223            }
224            set_atime::SET_TO_CLIENT_TIME(ts) => {
225                entry.fattr.atime = ts;
226            }
227        }
228
229        match setattr.mtime {
230            set_mtime::DONT_CHANGE => {}
231            set_mtime::SET_TO_SERVER_TIME => {
232                entry.fattr.mtime = now;
233                entry.fattr.ctime = now;
234            }
235            set_mtime::SET_TO_CLIENT_TIME(ts) => {
236                entry.fattr.mtime = ts;
237                entry.fattr.ctime = ts;
238            }
239        }
240
241        if let Nfs3Option::Some(uid) = setattr.uid {
242            entry.fattr.uid = uid;
243            entry.fattr.ctime = now;
244        }
245
246        if let Nfs3Option::Some(gid) = setattr.gid {
247            entry.fattr.gid = gid;
248            entry.fattr.ctime = now;
249        }
250
251        if let Nfs3Option::Some(mode) = setattr.mode {
252            entry.fattr.mode = mode;
253            entry.fattr.ctime = now;
254        }
255
256        info!("fattr = {:?}", entry.fattr);
257        Ok(entry.fattr.clone())
258    }
259
260    #[instrument(skip(self), fields(id = id.to_string(), buffer = buffer.len()))]
261    async fn write(
262        &self, id: &Self::Handle, offset: u64, buffer: &[u8],
263    ) -> Result<fattr3, nfsstat3> {
264        let offset = offset as usize;
265
266        let mut data = self.data.lock().await;
267        let entry = data.get_mut(id).unwrap();
268
269        let mut doc = self.lb.read_document(*id.as_uuid(), false).await.unwrap();
270        let mut expanded = false;
271        if offset + buffer.len() > doc.len() {
272            doc.resize(offset + buffer.len(), 0);
273            doc[offset..].copy_from_slice(buffer);
274            expanded = true;
275        } else {
276            for (idx, datum) in buffer.iter().enumerate() {
277                doc[offset + idx] = *datum;
278            }
279        }
280        let doc_size = doc.len();
281        self.lb.write_document(*id.as_uuid(), &doc).await.unwrap();
282
283        entry.fattr.size = doc_size as u64;
284
285        info!("expanded={expanded}, fattr.size = {}", doc_size);
286
287        Ok(entry.fattr.clone())
288    }
289
290    // todo this should create a file regardless of whether it exists
291    #[instrument(skip(self), fields(dirid = dirid.to_string(), filename = get_string(filename)))]
292    async fn create(
293        &self, dirid: &Self::Handle, filename: &filename3<'_>, attr: sattr3,
294    ) -> Result<(Self::Handle, fattr3), nfsstat3> {
295        let filename = get_string(filename);
296        let file = self
297            .lb
298            .create_file(&filename, dirid.as_uuid(), FileType::Document)
299            .await
300            .unwrap();
301
302        let id = file.id.into();
303        let entry = FileEntry::from_file(file, 0);
304        self.data.lock().await.insert(id, entry);
305
306        let file = self.setattr(&id, attr).await.unwrap();
307
308        info!("({id}, size={})", file.size);
309        Ok((id, file))
310    }
311
312    #[instrument(skip(self), fields(dirid = dirid.to_string(), filename = get_string(filename)))]
313    async fn create_exclusive(
314        &self, dirid: &Self::Handle, filename: &filename3<'_>,
315        createverf: nfs3_server::nfs3_types::nfs3::createverf3,
316    ) -> Result<Self::Handle, nfsstat3> {
317        let filename = get_string(filename);
318        let children = self.lb.get_children(dirid.as_uuid()).await.unwrap();
319        for child in children {
320            if child.name == filename {
321                warn!("exists already");
322                return Err(nfsstat3::NFS3ERR_EXIST);
323            }
324        }
325
326        let file = self
327            .lb
328            .create_file(&filename, dirid.as_uuid(), FileType::Document)
329            .await
330            .unwrap();
331
332        let id = file.id.into();
333        let entry = FileEntry::from_file(file, 0);
334        info!("({id}, size={})", entry.fattr.size);
335        self.data.lock().await.insert(id, entry);
336
337        Ok(id)
338    }
339
340    #[instrument(skip(self), fields(dirid = dirid.to_string(), dirname = get_string(dirname)))]
341    async fn mkdir(
342        &self, dirid: &Self::Handle, dirname: &filename3<'_>,
343    ) -> Result<(Self::Handle, fattr3), nfsstat3> {
344        let filename = get_string(dirname);
345        let file = self
346            .lb
347            .create_file(&filename, dirid.as_uuid(), FileType::Folder)
348            .await
349            .unwrap();
350
351        let id = file.id.into();
352        let entry = FileEntry::from_file(file, 0);
353        let fattr = entry.fattr.clone();
354        self.data.lock().await.insert(id, entry);
355
356        info!("({id}, fattr={fattr:?})");
357        Ok((id, fattr))
358    }
359
360    /// Removes a file.
361    /// If not supported dur to readonly file system
362    /// this should return Err(nfsstat3::NFS3ERR_ROFS)
363    #[instrument(skip(self), fields(dirid = dirid.to_string(), filename = get_string(filename)))]
364    async fn remove(&self, dirid: &Self::Handle, filename: &filename3<'_>) -> Result<(), nfsstat3> {
365        let mut data = self.data.lock().await;
366
367        let children = self.lb.get_children(dirid.as_uuid()).await.unwrap();
368        let file_name = get_string(filename);
369
370        for child in children {
371            if file_name == child.name {
372                info!("deleted");
373                let _ = self.lb.delete(&child.id).await; // ignore errors
374                data.remove(&child.id.into());
375                return Ok(());
376            }
377        }
378
379        info!("NOENT");
380        Err(nfsstat3::NFS3ERR_NOENT)
381    }
382
383    /// either an overwrite rename or move
384    #[instrument(skip(self), fields(from_dirid = from_dirid.to_string(), from_filename = get_string(from_filename), to_dirid = to_dirid.to_string(), to_filename = get_string(to_filename)))]
385    async fn rename<'a>(
386        &self, from_dirid: &Self::Handle, from_filename: &filename3<'a>, to_dirid: &Self::Handle,
387        to_filename: &filename3<'a>,
388    ) -> Result<(), nfsstat3> {
389        let mut data = self.data.lock().await;
390
391        let from_filename = get_string(from_filename);
392        let to_filename = get_string(to_filename);
393
394        let src_children = self.lb.get_children(from_dirid.as_uuid()).await.unwrap();
395
396        let mut from_id = None;
397        let mut to_id = None;
398        for child in src_children {
399            if child.name == from_filename {
400                from_id = Some(child.id);
401            }
402
403            if to_dirid == from_dirid && child.name == to_filename {
404                to_id = Some(child.id);
405            }
406        }
407
408        if to_dirid != from_dirid {
409            let dst_children = self.lb.get_children(to_dirid.as_uuid()).await.unwrap();
410            for child in dst_children {
411                if child.name == to_filename {
412                    to_id = Some(child.id);
413                }
414            }
415        }
416
417        let from_id = from_id.unwrap();
418
419        match to_id {
420            // we are overwriting a file
421            Some(id) => {
422                info!("overwrite {from_id} -> {id}");
423                let from_doc = self.lb.read_document(from_id, false).await.unwrap();
424                info!("|{}|", from_doc.len());
425                let doc_len = from_doc.len() as u64;
426                self.lb.write_document(id, &from_doc).await.unwrap();
427                self.lb.delete(&from_id).await.unwrap();
428
429                let entry = data.get_mut(&id.into()).unwrap();
430                entry.fattr.size = doc_len;
431
432                data.remove(&from_id.into());
433            }
434
435            // we are doing a move and/or rename
436            None => {
437                if from_dirid != to_dirid {
438                    info!("move {} -> {}\t", from_id, to_dirid);
439                    self.lb
440                        .move_file(&from_id, to_dirid.as_uuid())
441                        .await
442                        .unwrap();
443                }
444
445                if from_filename != to_filename {
446                    info!("rename {} -> {}\t", from_id, to_filename);
447                    self.lb.rename_file(&from_id, &to_filename).await.unwrap();
448                }
449
450                let entry = data.get_mut(&from_id.into()).unwrap();
451
452                let file = self.lb.get_file_by_id(from_id).await.unwrap();
453                entry.file = file;
454
455                info!("ok");
456            }
457        }
458
459        Ok(())
460    }
461
462    async fn symlink<'a>(
463        &self, _dirid: &Self::Handle, _linkname: &filename3<'a>, _symlink: &nfspath3<'a>,
464        _attr: &sattr3,
465    ) -> Result<(Self::Handle, fattr3), nfsstat3> {
466        info!("symlink NOTSUPP");
467        Err(nfsstat3::NFS3ERR_NOTSUPP)
468    }
469}
470
471pub struct Iterator<I>
472where
473    I: StdIterator<Item = File> + Send + Sync + 'static,
474{
475    inner: I,
476}
477
478impl<I> ReadDirIterator for Iterator<I>
479where
480    I: StdIterator<Item = File> + Send + Sync + 'static,
481{
482    async fn next(&mut self) -> nfs3_server::vfs::NextResult<DirEntry> {
483        match self.inner.next() {
484            Some(entry) => nfs3_server::vfs::NextResult::Ok(DirEntry {
485                fileid: file_id(&entry),
486                name: entry.name.as_bytes().to_vec().into(),
487                cookie: 0,
488            }),
489            None => nfs3_server::vfs::NextResult::Eof,
490        }
491    }
492}
493
494pub struct IteratorPlus<I>
495where
496    I: StdIterator<Item = DirEntryPlus<UuidFileHandle>> + Send + Sync + 'static,
497{
498    inner: I,
499}
500
501impl<I> ReadDirPlusIterator<UuidFileHandle> for IteratorPlus<I>
502where
503    I: StdIterator<Item = DirEntryPlus<UuidFileHandle>> + Send + Sync + 'static,
504{
505    async fn next(&mut self) -> nfs3_server::vfs::NextResult<DirEntryPlus<UuidFileHandle>> {
506        match self.inner.next() {
507            Some(entry) => nfs3_server::vfs::NextResult::Ok(entry),
508            None => nfs3_server::vfs::NextResult::Eof,
509        }
510    }
511}
lb_fs/fs_impl.rs

lb_fs/
fs_impl.rs
