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.data.lock().await.get(dirid).unwrap().file.clone();
86
87        if dir.is_document() {
88            info!("NOTDIR");
89            return Err(nfsstat3::NFS3ERR_NOTDIR);
90        }
91
92        // if looking for dir/. its the current directory
93        if filename.as_ref() == [b'.'] {
94            info!(". == {dirid}");
95            return Ok(*dirid);
96        }
97
98        // if looking for dir/.. its the parent directory
99        if filename.as_ref() == [b'.', b'.'] {
100            info!(".. == {}", dir.parent);
101            return Ok(dir.parent.into());
102        }
103
104        let children = self.lb.get_children(&dir.id).await.unwrap();
105        let file_name = get_string(filename);
106
107        for child in children {
108            if file_name == child.name {
109                info!("{}", child.id);
110                return Ok(child.id.into());
111            }
112        }
113
114        info!("NOENT");
115        Err(nfsstat3::NFS3ERR_NOENT)
116    }
117
118    #[instrument(skip(self), fields(id = id.to_string()))]
119    async fn getattr(&self, id: &Self::Handle) -> Result<fattr3, nfsstat3> {
120        let file = self.data.lock().await.get(id).unwrap().fattr.clone();
121        info!("fattr = {:?}", file);
122        Ok(file)
123    }
124
125    #[instrument(skip(self), fields(id = id.to_string(), offset, count))]
126    async fn read(
127        &self, id: &Self::Handle, offset: u64, count: u32,
128    ) -> Result<(Vec<u8>, bool), nfsstat3> {
129        let offset = offset as usize;
130        let count = count as usize;
131
132        let doc = self.lb.read_document(*id.as_uuid(), false).await.unwrap();
133
134        if offset >= doc.len() {
135            info!("[] EOF");
136            return Ok((vec![], true));
137        }
138
139        if offset + count >= doc.len() {
140            info!("|{}| EOF", doc[offset..].len());
141            return Ok((doc[offset..].to_vec(), true));
142        }
143
144        info!("|{}|", count);
145        return Ok((doc[offset..offset + count].to_vec(), false));
146    }
147
148    #[instrument(skip(self), fields(dirid = dirid.to_string(), start_after = cookie))]
149    async fn readdir(
150        &self, dirid: &Self::Handle, cookie: u64,
151    ) -> Result<impl nfs3_server::vfs::ReadDirIterator, nfsstat3> {
152        let iter = self.load_children(dirid, cookie).await;
153        Ok(Iterator { inner: iter })
154    }
155
156    #[instrument(skip(self), fields(dirid = dirid.to_string(), start_after = cookie))]
157    async fn readdirplus(
158        &self, dirid: &Self::Handle, cookie: u64,
159    ) -> Result<impl ReadDirPlusIterator<UuidFileHandle>, nfsstat3> {
160        let iter = self.load_children(dirid, cookie).await;
161        let data = self.data.lock().await;
162
163        let iter = iter
164            .map(move |file| {
165                let id: UuidFileHandle = file.id.into();
166                let name = file.name.as_bytes().to_vec().into();
167                let fattr = data.get(&id).map(|entry| entry.fattr.clone());
168                DirEntryPlus {
169                    fileid: id.fileid(),
170                    name,
171                    cookie: id.fileid(),
172                    name_attributes: fattr,
173                    name_handle: Some(id),
174                }
175            })
176            .collect::<Vec<_>>()
177            .into_iter();
178        Ok(IteratorPlus { inner: iter })
179    }
180
181    async fn readlink(&self, _id: &Self::Handle) -> Result<nfspath3<'_>, nfsstat3> {
182        info!("readlink NOTSUPP");
183        Err(nfsstat3::NFS3ERR_NOTSUPP)
184    }
185}
186
187impl NfsFileSystem for Drive {
188    #[instrument(skip(self), fields(id = id.to_string()))]
189    async fn setattr(&self, id: &Self::Handle, setattr: sattr3) -> Result<fattr3, nfsstat3> {
190        let mut data = self.data.lock().await;
191        let now = FileEntry::now();
192        let entry = data.get_mut(id).unwrap();
193
194        if let Nfs3Option::Some(new) = setattr.size
195            && entry.fattr.size != new
196        {
197            let mut doc = self.lb.read_document(*id.as_uuid(), false).await.unwrap();
198            doc.resize(new as usize, 0);
199            self.lb.write_document(*id.as_uuid(), &doc).await.unwrap();
200            entry.fattr.mtime = now;
201            entry.fattr.ctime = now;
202        }
203
204        match setattr.atime {
205            set_atime::DONT_CHANGE => {}
206            set_atime::SET_TO_SERVER_TIME => {
207                entry.fattr.atime = now;
208            }
209            set_atime::SET_TO_CLIENT_TIME(ts) => {
210                entry.fattr.atime = ts;
211            }
212        }
213
214        match setattr.mtime {
215            set_mtime::DONT_CHANGE => {}
216            set_mtime::SET_TO_SERVER_TIME => {
217                entry.fattr.mtime = now;
218                entry.fattr.ctime = now;
219            }
220            set_mtime::SET_TO_CLIENT_TIME(ts) => {
221                entry.fattr.mtime = ts;
222                entry.fattr.ctime = ts;
223            }
224        }
225
226        if let Nfs3Option::Some(uid) = setattr.uid {
227            entry.fattr.uid = uid;
228            entry.fattr.ctime = now;
229        }
230
231        if let Nfs3Option::Some(gid) = setattr.gid {
232            entry.fattr.gid = gid;
233            entry.fattr.ctime = now;
234        }
235
236        if let Nfs3Option::Some(mode) = setattr.mode {
237            entry.fattr.mode = mode;
238            entry.fattr.ctime = now;
239        }
240
241        info!("fattr = {:?}", entry.fattr);
242        Ok(entry.fattr.clone())
243    }
244
245    #[instrument(skip(self), fields(id = id.to_string(), buffer = buffer.len()))]
246    async fn write(
247        &self, id: &Self::Handle, offset: u64, buffer: &[u8],
248    ) -> Result<fattr3, nfsstat3> {
249        let offset = offset as usize;
250
251        let mut data = self.data.lock().await;
252        let entry = data.get_mut(id).unwrap();
253
254        let mut doc = self.lb.read_document(*id.as_uuid(), false).await.unwrap();
255        let mut expanded = false;
256        if offset + buffer.len() > doc.len() {
257            doc.resize(offset + buffer.len(), 0);
258            doc[offset..].copy_from_slice(buffer);
259            expanded = true;
260        } else {
261            for (idx, datum) in buffer.iter().enumerate() {
262                doc[offset + idx] = *datum;
263            }
264        }
265        let doc_size = doc.len();
266        self.lb.write_document(*id.as_uuid(), &doc).await.unwrap();
267
268        entry.fattr.size = doc_size as u64;
269
270        info!("expanded={expanded}, fattr.size = {}", doc_size);
271
272        Ok(entry.fattr.clone())
273    }
274
275    // todo this should create a file regardless of whether it exists
276    #[instrument(skip(self), fields(dirid = dirid.to_string(), filename = get_string(filename)))]
277    async fn create(
278        &self, dirid: &Self::Handle, filename: &filename3<'_>, attr: sattr3,
279    ) -> Result<(Self::Handle, fattr3), nfsstat3> {
280        let filename = get_string(filename);
281        let file = self
282            .lb
283            .create_file(&filename, dirid.as_uuid(), FileType::Document)
284            .await
285            .unwrap();
286
287        let id = file.id.into();
288        let entry = FileEntry::from_file(file, 0);
289        self.data.lock().await.insert(id, entry);
290
291        let file = self.setattr(&id, attr).await.unwrap();
292
293        info!("({id}, size={})", file.size);
294        Ok((id, file))
295    }
296
297    #[instrument(skip(self), fields(dirid = dirid.to_string(), filename = get_string(filename)))]
298    async fn create_exclusive(
299        &self, dirid: &Self::Handle, filename: &filename3<'_>,
300        createverf: nfs3_server::nfs3_types::nfs3::createverf3,
301    ) -> Result<Self::Handle, nfsstat3> {
302        let filename = get_string(filename);
303        let children = self.lb.get_children(dirid.as_uuid()).await.unwrap();
304        for child in children {
305            if child.name == filename {
306                warn!("exists already");
307                return Err(nfsstat3::NFS3ERR_EXIST);
308            }
309        }
310
311        let file = self
312            .lb
313            .create_file(&filename, dirid.as_uuid(), FileType::Document)
314            .await
315            .unwrap();
316
317        let id = file.id.into();
318        let entry = FileEntry::from_file(file, 0);
319        info!("({id}, size={})", entry.fattr.size);
320        self.data.lock().await.insert(id, entry);
321
322        Ok(id)
323    }
324
325    #[instrument(skip(self), fields(dirid = dirid.to_string(), dirname = get_string(dirname)))]
326    async fn mkdir(
327        &self, dirid: &Self::Handle, dirname: &filename3<'_>,
328    ) -> Result<(Self::Handle, fattr3), nfsstat3> {
329        let filename = get_string(dirname);
330        let file = self
331            .lb
332            .create_file(&filename, dirid.as_uuid(), FileType::Folder)
333            .await
334            .unwrap();
335
336        let id = file.id.into();
337        let entry = FileEntry::from_file(file, 0);
338        let fattr = entry.fattr.clone();
339        self.data.lock().await.insert(id, entry);
340
341        info!("({id}, fattr={fattr:?})");
342        Ok((id, fattr))
343    }
344
345    /// Removes a file.
346    /// If not supported dur to readonly file system
347    /// this should return Err(nfsstat3::NFS3ERR_ROFS)
348    #[instrument(skip(self), fields(dirid = dirid.to_string(), filename = get_string(filename)))]
349    async fn remove(&self, dirid: &Self::Handle, filename: &filename3<'_>) -> Result<(), nfsstat3> {
350        let mut data = self.data.lock().await;
351
352        let children = self.lb.get_children(dirid.as_uuid()).await.unwrap();
353        let file_name = get_string(filename);
354
355        for child in children {
356            if file_name == child.name {
357                info!("deleted");
358                let _ = self.lb.delete(&child.id).await; // ignore errors
359                data.remove(&child.id.into());
360                return Ok(());
361            }
362        }
363
364        info!("NOENT");
365        Err(nfsstat3::NFS3ERR_NOENT)
366    }
367
368    /// either an overwrite rename or move
369    #[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)))]
370    async fn rename<'a>(
371        &self, from_dirid: &Self::Handle, from_filename: &filename3<'a>, to_dirid: &Self::Handle,
372        to_filename: &filename3<'a>,
373    ) -> Result<(), nfsstat3> {
374        let mut data = self.data.lock().await;
375
376        let from_filename = get_string(from_filename);
377        let to_filename = get_string(to_filename);
378
379        let src_children = self.lb.get_children(from_dirid.as_uuid()).await.unwrap();
380
381        let mut from_id = None;
382        let mut to_id = None;
383        for child in src_children {
384            if child.name == from_filename {
385                from_id = Some(child.id);
386            }
387
388            if to_dirid == from_dirid && child.name == to_filename {
389                to_id = Some(child.id);
390            }
391        }
392
393        if to_dirid != from_dirid {
394            let dst_children = self.lb.get_children(to_dirid.as_uuid()).await.unwrap();
395            for child in dst_children {
396                if child.name == to_filename {
397                    to_id = Some(child.id);
398                }
399            }
400        }
401
402        let from_id = from_id.unwrap();
403
404        match to_id {
405            // we are overwriting a file
406            Some(id) => {
407                info!("overwrite {from_id} -> {id}");
408                let from_doc = self.lb.read_document(from_id, false).await.unwrap();
409                info!("|{}|", from_doc.len());
410                let doc_len = from_doc.len() as u64;
411                self.lb.write_document(id, &from_doc).await.unwrap();
412                self.lb.delete(&from_id).await.unwrap();
413
414                let entry = data.get_mut(&id.into()).unwrap();
415                entry.fattr.size = doc_len;
416
417                data.remove(&from_id.into());
418            }
419
420            // we are doing a move and/or rename
421            None => {
422                if from_dirid != to_dirid {
423                    info!("move {} -> {}\t", from_id, to_dirid);
424                    self.lb
425                        .move_file(&from_id, to_dirid.as_uuid())
426                        .await
427                        .unwrap();
428                }
429
430                if from_filename != to_filename {
431                    info!("rename {} -> {}\t", from_id, to_filename);
432                    self.lb.rename_file(&from_id, &to_filename).await.unwrap();
433                }
434
435                let entry = data.get_mut(&from_id.into()).unwrap();
436
437                let file = self.lb.get_file_by_id(from_id).await.unwrap();
438                entry.file = file;
439
440                info!("ok");
441            }
442        }
443
444        Ok(())
445    }
446
447    async fn symlink<'a>(
448        &self, _dirid: &Self::Handle, _linkname: &filename3<'a>, _symlink: &nfspath3<'a>,
449        _attr: &sattr3,
450    ) -> Result<(Self::Handle, fattr3), nfsstat3> {
451        info!("symlink NOTSUPP");
452        Err(nfsstat3::NFS3ERR_NOTSUPP)
453    }
454}
455
456pub struct Iterator<I>
457where
458    I: StdIterator<Item = File> + Send + Sync + 'static,
459{
460    inner: I,
461}
462
463impl<I> ReadDirIterator for Iterator<I>
464where
465    I: StdIterator<Item = File> + Send + Sync + 'static,
466{
467    async fn next(&mut self) -> nfs3_server::vfs::NextResult<DirEntry> {
468        match self.inner.next() {
469            Some(entry) => nfs3_server::vfs::NextResult::Ok(DirEntry {
470                fileid: file_id(&entry),
471                name: entry.name.as_bytes().to_vec().into(),
472                cookie: 0,
473            }),
474            None => nfs3_server::vfs::NextResult::Eof,
475        }
476    }
477}
478
479pub struct IteratorPlus<I>
480where
481    I: StdIterator<Item = DirEntryPlus<UuidFileHandle>> + Send + Sync + 'static,
482{
483    inner: I,
484}
485
486impl<I> ReadDirPlusIterator<UuidFileHandle> for IteratorPlus<I>
487where
488    I: StdIterator<Item = DirEntryPlus<UuidFileHandle>> + Send + Sync + 'static,
489{
490    async fn next(&mut self) -> nfs3_server::vfs::NextResult<DirEntryPlus<UuidFileHandle>> {
491        match self.inner.next() {
492            Some(entry) => nfs3_server::vfs::NextResult::Ok(entry),
493            None => nfs3_server::vfs::NextResult::Eof,
494        }
495    }
496}
lb_fs/fs_impl.rs

lb_fs/
fs_impl.rs
